diff --git a/tests/conftest.py b/tests/conftest.py index 645307fb33..23448b80a9 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -510,6 +510,20 @@ def _hermetic_environment(tmp_path, monkeypatch): # reading real sessions into assertions and writing test rows into the # real profile. Re-pin the constant to this test's home. (Several test # files already do this locally; this makes it an invariant.) + # 3c. Multi-profile hosting is a process-global latch (``set_multiplex_active`` and the + # launch-env snapshot flip once and stay). A test that routes one RPC/request to a named + # profile would otherwise leave every later test in the file fail-closed (unscoped + # ``get_env_value`` in a test body raises). Reset the latch per test. + secret_scope_mod = sys.modules.get("agent.secret_scope") + if secret_scope_mod is not None and hasattr(secret_scope_mod, "_MULTIPLEX_ACTIVE"): + monkeypatch.setattr(secret_scope_mod, "_MULTIPLEX_ACTIVE", False) + launch_policy_mod = sys.modules.get("tui_gateway.launch_profile_policy") + if launch_policy_mod is not None and hasattr(launch_policy_mod, "_snapshot"): + monkeypatch.setattr(launch_policy_mod, "_snapshot", None) + tui_server_mod = sys.modules.get("tui_gateway.server") + if tui_server_mod is not None and hasattr(tui_server_mod, "_served_profile_homes"): + monkeypatch.setattr(tui_server_mod, "_served_profile_homes", set()) + hermes_state_mod = sys.modules.get("hermes_state") if hermes_state_mod is not None and hasattr(hermes_state_mod, "DEFAULT_DB_PATH"): monkeypatch.setattr( diff --git a/tests/hermes_cli/test_web_multi_profile_scope.py b/tests/hermes_cli/test_web_multi_profile_scope.py new file mode 100644 index 0000000000..67815e061a --- /dev/null +++ b/tests/hermes_cli/test_web_multi_profile_scope.py @@ -0,0 +1,101 @@ +"""Dashboard routers that read a named profile's config/credentials run under that profile's FULL +scope (home + secrets) and never mutate the dashboard process environment. + +Regression for the cross-profile leak class in ``hermes dashboard`` / ``hermes serve``: +``_config_profile_scope`` bound only HERMES_HOME, so ``GET /api/config?profile=B`` expanded B's +``${VAR}`` refs to the DEFAULT profile's plaintext credentials (its ``os.environ``), and console +``send`` for B (``send_cmd._load_hermes_env``) copied B's ``.env`` into the shared process env with +``override=True``, so every later default-profile read saw B's tokens. +""" + +from __future__ import annotations + +import os +from pathlib import Path + +import pytest + +pytest.importorskip("fastapi") +from starlette.testclient import TestClient # noqa: E402 + +A_VAL = "a-only-secret-0001" +B_VAL = "b-only-secret-0002" + + +@pytest.fixture +def two_homes(tmp_path, monkeypatch): + root = tmp_path / "hermes_home" + b = root / "profiles" / "b" + b.mkdir(parents=True) + (root / ".env").write_text(f"A_ONLY_TOKEN={A_VAL}\n", encoding="utf-8") + (b / ".env").write_text(f"B_ONLY_TOKEN={B_VAL}\nTELEGRAM_BOT_TOKEN=b-telegram-token\n", encoding="utf-8") + for home in (root, b): + (home / "config.yaml").write_text( + "model:\n default: openai/gpt-4o-mini\n api_key: ${A_ONLY_TOKEN}\n" + "custom_probe:\n a_ref: ${A_ONLY_TOKEN}\n b_ref: ${B_ONLY_TOKEN}\n", + encoding="utf-8") + monkeypatch.setenv("HERMES_HOME", str(root)) + monkeypatch.setenv("A_ONLY_TOKEN", A_VAL) # the dashboard process loaded its own .env + monkeypatch.delenv("TELEGRAM_BOT_TOKEN", raising=False) + from agent import secret_scope + from tui_gateway import launch_profile_policy as lpp + monkeypatch.setattr(secret_scope, "_MULTIPLEX_ACTIVE", False) + monkeypatch.setattr(lpp, "_snapshot", None) + from hermes_cli import config as cfg_mod + for attr in ("_CONFIG_CACHE", "_config_cache"): + if hasattr(cfg_mod, attr): + monkeypatch.setattr(cfg_mod, attr, None if not isinstance(getattr(cfg_mod, attr), dict) else {}) + return root, b + + +@pytest.fixture +def client(two_homes): + from hermes_cli.web_server import _SESSION_HEADER_NAME, _SESSION_TOKEN, app + c = TestClient(app) + c.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN + return c + + +def test_get_config_for_named_profile_expands_only_its_own_secrets(client, two_homes): + from agent.secret_scope import is_multiplex_active + + resp = client.get("/api/config?profile=b") + assert resp.status_code == 200, resp.text + probe = resp.json()["custom_probe"] + assert probe["b_ref"] == B_VAL + assert probe["a_ref"] == "${A_ONLY_TOKEN}" # not the dashboard profile's value + # Hosting a second profile flipped the process to fail-closed; environ is untouched. + assert is_multiplex_active() + assert os.environ["A_ONLY_TOKEN"] == A_VAL and "B_ONLY_TOKEN" not in os.environ + + # The dashboard's own profile still resolves its own value (frozen launch env). + probe_a = client.get("/api/config").json()["custom_probe"] + assert probe_a["a_ref"] == A_VAL and probe_a["b_ref"] == "${B_ONLY_TOKEN}" + + +def test_console_send_for_named_profile_does_not_write_process_env(two_homes, monkeypatch): + """``send`` loads the target profile's ``.env`` for the gateway config loader; inside a + multi-profile host that must land in the request's scope, never ``os.environ``.""" + from hermes_cli.web_routers.chat_ws import _execute_console_line + + root, b = two_homes + seen = {} + + def fake_send(args): + import hermes_cli.send_cmd as send_cmd + from gateway.config import _getenv + send_cmd._load_hermes_env() + seen["loader_sees"] = _getenv("TELEGRAM_BOT_TOKEN") + seen["environ_has"] = "TELEGRAM_BOT_TOKEN" in os.environ + seen["home"] = Path(os.environ.get("HERMES_HOME", "")) + return '{"success": true}' + + class Engine: + def execute(self, line, *, confirmed=False): + import argparse + return fake_send(argparse.Namespace()) + + _execute_console_line(Engine(), "send --to telegram hi", confirmed=False, profile="b") + assert seen["loader_sees"] == "b-telegram-token" # the loader gets B's token through the scope + assert seen["environ_has"] is False # and the dashboard process env never learns it + assert "B_ONLY_TOKEN" not in os.environ diff --git a/tests/tui_gateway/test_multi_profile_hosting_fail_closed.py b/tests/tui_gateway/test_multi_profile_hosting_fail_closed.py new file mode 100644 index 0000000000..76991aaec8 --- /dev/null +++ b/tests/tui_gateway/test_multi_profile_hosting_fail_closed.py @@ -0,0 +1,142 @@ +"""Multi-profile hosting in the TUI gateway is fail-closed and every profile-scoped RPC runs under +the FULL runtime scope of the requested profile (home + secrets + terminal), the launch profile +included once the process multiplexes. + +Regression for the silent cross-profile secret leak class: ``hermes serve`` hosted many profile +homes but never called ``set_multiplex_active(True)``, so every unscoped ``get_secret`` read for a +secondary silently returned the LAUNCH profile's ``os.environ`` value; ``@_profile_scoped`` bound +only HERMES_HOME. And the launch-profile asymmetry: a default-member hosted-room turn in a +``multiplex_profiles: true`` gateway died at agent build with ``UnscopedSecretError``. +""" + +from __future__ import annotations + +import os +from pathlib import Path + +import pytest + +import tui_gateway.server as server +from tui_gateway import launch_profile_policy as lpp + +A_VAL = "a-only-secret-0001" +B_VAL = "b-only-secret-0002" +ENV_VAL = "systemd-injected-0003" + + +@pytest.fixture +def two_homes(tmp_path, monkeypatch): + """Launch home (root) + secondary ``profiles/b``; B's config references both tokens.""" + root = tmp_path / "hermes_home" + b = root / "profiles" / "b" + b.mkdir(parents=True) + (root / ".env").write_text(f"A_ONLY_TOKEN={A_VAL}\n", encoding="utf-8") + (b / ".env").write_text(f"B_ONLY_TOKEN={B_VAL}\n", encoding="utf-8") + for home in (root, b): + (home / "config.yaml").write_text( + "probe:\n a_ref: ${A_ONLY_TOKEN}\n b_ref: ${B_ONLY_TOKEN}\n env_ref: ${INJECTED_TOKEN}\n", + encoding="utf-8") + monkeypatch.setenv("HERMES_HOME", str(root)) + monkeypatch.setenv("A_ONLY_TOKEN", A_VAL) # the launch process loaded its own .env + monkeypatch.setenv("INJECTED_TOKEN", ENV_VAL) # systemd / op run credential injection + monkeypatch.setattr(server, "_hermes_home", root) + monkeypatch.setattr(server, "_served_profile_homes", set()) + monkeypatch.setattr(lpp, "_snapshot", None) + from agent import secret_scope + monkeypatch.setattr(secret_scope, "_MULTIPLEX_ACTIVE", False) + server._cfg_cache = server._cfg_mtime = server._cfg_path = None + return root, b + + +def _probe(profile: str | None) -> dict: + params = {"key": "full"} + if profile: + params["profile"] = profile + server._cfg_cache = server._cfg_mtime = server._cfg_path = None + resp = server._methods["config.get"]("rid", params) + assert "error" not in resp, resp + return resp["result"]["config"]["probe"] + + +def test_config_get_for_secondary_resolves_only_its_own_secrets_and_flips_fail_closed(two_homes): + from agent.secret_scope import UnscopedSecretError, get_secret, is_multiplex_active + + root, _b = two_homes + assert not is_multiplex_active() # single-profile so far + + probe_b = _probe("b") + assert probe_b["b_ref"] == B_VAL + assert probe_b["a_ref"] == "${A_ONLY_TOKEN}" # never the launch profile's value + assert probe_b["env_ref"] == "${INJECTED_TOKEN}" # never the launch process env + # Hosting a second home flipped the process: an unscoped read now raises instead of borrowing. + assert is_multiplex_active() + with pytest.raises(UnscopedSecretError): + get_secret("A_ONLY_TOKEN") + assert os.environ["A_ONLY_TOKEN"] == A_VAL # never mutated + + # The launch profile is a profile too: its RPC keeps its own .env AND its injected env. + probe_a = _probe(None) + assert probe_a["a_ref"] == A_VAL + assert probe_a["env_ref"] == ENV_VAL + assert probe_a["b_ref"] == "${B_ONLY_TOKEN}" + + +def test_single_profile_serve_keeps_environ_fallthrough(two_homes): + """Control: with no secondary ever requested the launch profile stays unscoped, so credentials + injected only via the process env (systemd, ``op run``) keep resolving.""" + from agent.secret_scope import get_secret, is_multiplex_active + + probe = _probe(None) + assert probe["env_ref"] == ENV_VAL + assert not is_multiplex_active() + assert get_secret("INJECTED_TOKEN") == ENV_VAL + + +def test_rpc_scope_reaches_llm_oneshot_and_model_options(two_homes, monkeypatch): + """The scope must wrap the body of every credential-reading RPC, not only config.get.""" + from agent.secret_scope import get_secret + + root, b = two_homes + seen = {} + + def fake_oneshot(**kwargs): + seen["oneshot"] = (Path(os.environ.get("HERMES_HOME", "")), get_secret("B_ONLY_TOKEN"), get_secret("A_ONLY_TOKEN")) + from hermes_constants import get_hermes_home + seen["oneshot_home"] = Path(get_hermes_home()) + return "t" + + monkeypatch.setattr("agent.oneshot.run_oneshot", fake_oneshot) + monkeypatch.setattr(server, "_model_picker_context", lambda agent: object()) + + def build_payload(ctx, **kwargs): + from hermes_constants import get_hermes_home + seen["options"] = (Path(get_hermes_home()), get_secret("B_ONLY_TOKEN"), get_secret("A_ONLY_TOKEN")) + return {"providers": []} + + monkeypatch.setattr("hermes_cli.inventory.build_model_options_payload", build_payload) + + r = server._methods["llm.oneshot"]("r1", {"profile": "b", "instructions": "x", "input": "y"}) + assert r["result"]["text"] == "t" + assert seen["oneshot_home"] == b and seen["oneshot"][1:] == (B_VAL, None) + r = server._methods["model.options"]("r2", {"profile": "b"}) + assert r["result"] == {"providers": []} + assert seen["options"] == (b, B_VAL, None) + + +def test_launch_profile_agent_build_is_scoped_once_multiplexing(two_homes, monkeypatch): + """The C6 asymmetry: a default-profile session (``profile_home`` None) in a multiplexing process + must bind the launch profile's own scope for its agent build instead of running unscoped.""" + from agent.secret_scope import current_secret_scope, set_multiplex_active + from hermes_constants import get_hermes_home + + root, _b = two_homes + set_multiplex_active(True) # the messaging gateway's flip (GatewayRunner.__init__) + scopes = server._bind_build_profile_scopes(None) + try: + scope = current_secret_scope() + assert scope is not None and scope["A_ONLY_TOKEN"] == A_VAL and scope["INJECTED_TOKEN"] == ENV_VAL + assert "B_ONLY_TOKEN" not in scope + assert Path(get_hermes_home()) == root + finally: + server._release_build_profile_scopes(scopes) + assert current_secret_scope() is None diff --git a/tests/tui_gateway/test_profile_terminal_scope_entrypoints.py b/tests/tui_gateway/test_profile_terminal_scope_entrypoints.py index 34eaae3dad..86ae812f2c 100644 --- a/tests/tui_gateway/test_profile_terminal_scope_entrypoints.py +++ b/tests/tui_gateway/test_profile_terminal_scope_entrypoints.py @@ -23,7 +23,7 @@ import pytest from tools import terminal_tool as tt from tools.terminal_scope import get_terminal_scope -from tui_gateway import launch_terminal_policy as ltp +from tui_gateway import launch_profile_policy as ltp from tui_gateway import server @@ -33,6 +33,7 @@ def _launch_local_env(monkeypatch): monkeypatch.setenv("TERMINAL_ENV", "local") monkeypatch.setattr(tt, "_terminal_config_bridge_attempted", False) monkeypatch.setattr(ltp, "_snapshot", None) + monkeypatch.setattr("agent.secret_scope._MULTIPLEX_ACTIVE", False) monkeypatch.setattr("agent.secret_scope.build_profile_secret_scope", lambda _h: {}) @@ -132,6 +133,10 @@ def _launch_turn_policy(launch_home): from tools.terminal_scope import reset_terminal_scope if st.scopes.terminal is not None: reset_terminal_scope(st.scopes.terminal) + if st.scopes.secret is not None: + server.reset_secret_scope(st.scopes.secret) + if st.scopes.home is not None: + server.reset_hermes_home_override(st.scopes.home) if st.scopes.approval is not None: reset_current_session_key(st.scopes.approval) server._clear_session_context(st.scopes.session_tokens) @@ -144,7 +149,7 @@ def test_launch_turn_keeps_env_only_ssh_policy_once_multiplexing_is_active(tmp_p monkeypatch.setenv("HERMES_HOME", str(launch)) monkeypatch.setenv("TERMINAL_ENV", "ssh") monkeypatch.setenv("TERMINAL_SSH_HOST", "example.test") - ltp.capture_launch_terminal_env() # multiplex activation: first secondary served + ltp.activate_multi_profile_hosting() # multiplex activation: first secondary served cfg = _launch_turn_policy(launch) assert (cfg["env_type"], cfg["ssh_host"]) == ("ssh", "example.test") @@ -158,7 +163,7 @@ def test_launch_turn_ignores_ambient_terminal_env_written_after_activation(tmp_p monkeypatch.setenv("HERMES_HOME", str(launch)) monkeypatch.setenv("TERMINAL_ENV", "ssh") monkeypatch.setenv("TERMINAL_SSH_HOST", "example.test") - ltp.capture_launch_terminal_env() + ltp.activate_multi_profile_hosting() # A secondary context later poisons the process env (the pre-#108440 latch shape). monkeypatch.setenv("TERMINAL_ENV", "docker") monkeypatch.setenv("TERMINAL_DOCKER_IMAGE", "bee/img:1")