diff --git a/gateway/run.py b/gateway/run.py index 17e28e710..bf8c67002 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -17835,6 +17835,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew 3. The active profile (the multiplexer's own home). """ from hermes_cli.profiles import ( + create_profile, get_active_profile_name, get_profile_dir, profile_exists, @@ -17855,8 +17856,27 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew name = get_active_profile_name() or "default" profile_dir = get_profile_dir(name) - # Warn if an explicit profile doesn't exist on disk + # Principal-routed profiles are part of the isolation boundary. If a + # verified principal resolves to a profile name that has not been + # materialized on disk yet, create it now instead of collapsing the + # request back into the global/default HERMES_HOME. if explicit_profile and not profile_exists(name): + if name.startswith("principal_"): + try: + create_profile(name, no_alias=True) + except FileExistsError: + pass + except Exception: + logger.warning( + "Failed to auto-create principal profile %r for source %s/%s (guild_id=%s)", + explicit_profile, + source.platform.value, + source.chat_id, + getattr(source, "guild_id", None), + exc_info=True, + ) + if profile_exists(name): + return profile_dir logger.warning( "Profile %r does not exist for source %s/%s (guild_id=%s), " "falling back to global HERMES_HOME", diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index b88da647b..85c11ad09 100644 --- a/hermes_cli/web_server.py +++ b/hermes_cli/web_server.py @@ -665,10 +665,26 @@ def _dashboard_identity_scope(request: Request) -> dict[str, Any]: return scope +def _allowed_profile_names_for_scope(scope: dict[str, Any]) -> set[str]: + allowed = {"", "default"} + try: + from gateway.principal_profiles import principal_profile_name + + for principal_id in scope.get("principal_ids") or []: + try: + allowed.add(principal_profile_name(str(principal_id))) + except Exception: + continue + except Exception: + pass + return allowed + + def _enforce_dashboard_profile_access(scope: dict[str, Any], profile: Optional[str]) -> None: if scope.get("admin"): return - if profile and str(profile).strip() not in {"", "default"}: + profile_name = str(profile or "").strip() + if profile_name not in _allowed_profile_names_for_scope(scope): raise HTTPException(status_code=403, detail="forbidden") diff --git a/tests/hermes_cli/test_web_server_session_access.py b/tests/hermes_cli/test_web_server_session_access.py index 7d720a5da..3dac71e4a 100644 --- a/tests/hermes_cli/test_web_server_session_access.py +++ b/tests/hermes_cli/test_web_server_session_access.py @@ -20,6 +20,16 @@ class TestProfilesSessionsAccess: web_server.get_profiles_sessions(_request("user@bremen.com.tw")) assert exc.value.status_code == 403 + def test_non_admin_may_access_own_principal_profile(self): + scope = {"admin": False, "principal_ids": ["p-123"]} + web_server._enforce_dashboard_profile_access(scope, "principal_p123") + + def test_non_admin_cannot_access_other_principal_profile(self): + scope = {"admin": False, "principal_ids": ["p-123"]} + with pytest.raises(HTTPException) as exc: + web_server._enforce_dashboard_profile_access(scope, "principal_other999") + assert exc.value.status_code == 403 + def test_admin_allowed_to_reach_listing(self, monkeypatch, tmp_path): monkeypatch.setattr(web_server, "_dashboard_identity_scope", lambda request: {"admin": True}) monkeypatch.setattr(web_server, "_cron_profile_home", lambda profile: (profile or "default", tmp_path)) diff --git a/tests/tools/test_session_search.py b/tests/tools/test_session_search.py index 4defb9522..9a5633628 100644 --- a/tests/tools/test_session_search.py +++ b/tests/tools/test_session_search.py @@ -20,6 +20,7 @@ from tools.session_search_tool import ( _format_timestamp, session_search, ) +import tools.session_search_tool as session_search_tool @pytest.fixture @@ -338,6 +339,36 @@ class TestAccessControl: assert result["success"] is False assert "只有 admin 權限可跨 profile" in result.get("error", "") + def test_non_admin_may_read_own_principal_profile(self, db, monkeypatch): + class _FakeStore: + def get_principal_context(self, source): + return {"principal_id": "p-123"} + + monkeypatch.setattr("gateway.user_verification.GatewayUserStore", lambda: _FakeStore()) + tokens = set_session_vars(platform="telegram", source="telegram", user_id="alice", chat_id="chat-a") + try: + result = json.loads(session_search(session_id="some-session", profile="principal_p123", db=db)) + finally: + clear_session_vars(tokens) + + assert result["success"] is False + assert "只有 admin 權限可跨 profile" not in result.get("error", "") + + def test_non_admin_cannot_read_other_principal_profile(self, db, monkeypatch): + class _FakeStore: + def get_principal_context(self, source): + return {"principal_id": "p-123"} + + monkeypatch.setattr("gateway.user_verification.GatewayUserStore", lambda: _FakeStore()) + tokens = set_session_vars(platform="telegram", source="telegram", user_id="alice", chat_id="chat-a") + try: + result = json.loads(session_search(session_id="some-session", profile="principal_other999", db=db)) + finally: + clear_session_vars(tokens) + + assert result["success"] is False + assert "只有 admin 權限可跨 profile" in result.get("error", "") + def test_admin_can_read_other_users_session(self, db): db.create_session("alice", source="telegram", user_id="alice", chat_id="chat-a") db.append_message("alice", role="user", content="Alice private note") diff --git a/tools/session_search_tool.py b/tools/session_search_tool.py index ff088f6be..fd7410463 100644 --- a/tools/session_search_tool.py +++ b/tools/session_search_tool.py @@ -120,11 +120,43 @@ def _session_visible_to_requester(meta: Dict[str, Any], ctx: Dict[str, str]) -> return True +def _allowed_profile_names_for_context(ctx: Dict[str, str]) -> set[str]: + allowed = {"", "default"} + try: + from gateway.principal_profiles import principal_profile_name + from gateway.session import SessionSource + from gateway.user_verification import GatewayUserStore + + platform = str(ctx.get("platform") or ctx.get("source") or "").strip() + user_id = str(ctx.get("user_id") or ctx.get("chat_id") or "").strip() + chat_id = str(ctx.get("chat_id") or user_id).strip() + if not platform or not chat_id: + return allowed + source = SessionSource( + platform=platform, + chat_id=chat_id, + user_id=user_id or chat_id, + user_name="", + chat_type="dm", + thread_id=str(ctx.get("thread_id") or "").strip() or None, + ) + context = GatewayUserStore().get_principal_context(source) + principal_id = str((context or {}).get("principal_id") or "").strip() + if principal_id: + allowed.add(principal_profile_name(principal_id)) + except Exception: + pass + return allowed + + def _authorize_profile_access(profile: Optional[str], ctx: Dict[str, str]) -> Optional[str]: if profile is None or not str(profile).strip(): return None if _is_admin_access(ctx) or not _is_scoped_gateway_session(ctx): return None + requested = str(profile).strip() + if requested in _allowed_profile_names_for_context(ctx): + return None return "只有 admin 權限可跨 profile 讀取對話／工作紀錄。"