diff --git a/agent/delegation_context.py b/agent/delegation_context.py index fac7fb638cbe..d65cb4894425 100644 --- a/agent/delegation_context.py +++ b/agent/delegation_context.py @@ -22,6 +22,7 @@ KANBAN_ENV_KEYS: tuple[str, ...] = ( "HERMES_KANBAN_TASK", "HERMES_KANBAN_RUN_ID", "HERMES_KANBAN_CLAIM_LOCK", "HERMES_KANBAN_GOAL_MODE", "HERMES_KANBAN_GOAL_MAX_TURNS", + # HERMES_KANBAN_SAFE_ROOT_ACTIVE intentionally survives for delegate children. ) diff --git a/agent/file_safety.py b/agent/file_safety.py index a06579802cb0..7b7ab85d2e9d 100644 --- a/agent/file_safety.py +++ b/agent/file_safety.py @@ -191,10 +191,17 @@ def build_write_denied_prefixes(home: str) -> list[str]: return [os.path.realpath(p) + os.sep for p in paths] +def _write_safe_root_raw() -> str: + """Process env, or the routed profile scope when bound (multiplex/desktop turns).""" + from tools.write_safe_root_scope import write_safe_root_env + + return write_safe_root_env() + + def get_safe_write_roots() -> set[str]: """Resolved HERMES_WRITE_SAFE_ROOT paths (``os.pathsep``-separated list).""" roots: set[str] = set() - for path in filter(None, os.getenv("HERMES_WRITE_SAFE_ROOT", "").split(os.pathsep)): + for path in filter(None, _write_safe_root_raw().split(os.pathsep)): with suppress(OSError, ValueError): roots.add(os.path.realpath(os.path.expanduser(path))) return roots @@ -250,6 +257,16 @@ def _classify_write_denial(path: str) -> Optional[str]: if safe_roots and not any(_is_under(resolved, root) for root in safe_roots): return "safe_root" + from tools.write_safe_root_scope import ( + get_process_write_safe_roots, + is_write_safe_root_scope_bound, + ) + if is_write_safe_root_scope_bound(): + inherited = get_process_write_safe_roots() + if inherited and not any(_is_under(resolved, root) for root in safe_roots): + if any(_is_under(resolved, root) for root in inherited): + return "safe_root" + return None diff --git a/gateway/run.py b/gateway/run.py index 0304da3dddcb..cdf4d0f26eb4 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -1726,13 +1726,15 @@ def _profile_runtime_scope( # Without it terminal_tool reads the process-global TERMINAL_* vars a previous profile's turn may have # pinned (first-writer-wins backend leak; #68559). from tools.terminal_scope import install_and_reset_profile_terminal_scope + from tools.write_safe_root_scope import install_and_reset_profile_write_safe_root_scope with install_and_reset_profile_terminal_scope(Path(profile_home)): - try: - yield - finally: - reset_secret_scope(secret_token) - reset_hermes_home_override(home_token) + with install_and_reset_profile_write_safe_root_scope(Path(profile_home)): + try: + yield + finally: + reset_secret_scope(secret_token) + reset_hermes_home_override(home_token) @_asynccontextmanager diff --git a/hermes_cli/env_loader.py b/hermes_cli/env_loader.py index 2f2f895eb549..4bc15ed19d9a 100644 --- a/hermes_cli/env_loader.py +++ b/hermes_cli/env_loader.py @@ -351,6 +351,12 @@ def load_hermes_dotenv( return [] loaded: list[Path] = [] + kanban_worker = ( + "HERMES_KANBAN_TASK" in os.environ + or "HERMES_KANBAN_SAFE_ROOT_ACTIVE" in os.environ + ) + inherited_safe_root = os.environ.get("HERMES_WRITE_SAFE_ROOT") + had_safe_root = "HERMES_WRITE_SAFE_ROOT" in os.environ user_env = home_path / ".env" project_env_path = Path(project_env) if project_env else None @@ -403,6 +409,13 @@ def load_hermes_dotenv( # to the stale .env value mid-session (#29186, #67323). _reapply_terminal_config_bridge(home_path) + # Task-scoped roots must outrank profile and managed env files. + if kanban_worker: + if had_safe_root: + os.environ["HERMES_WRITE_SAFE_ROOT"] = inherited_safe_root or "" + else: + os.environ.pop("HERMES_WRITE_SAFE_ROOT", None) + return loaded diff --git a/hermes_cli/kanban_db_dispatch.py b/hermes_cli/kanban_db_dispatch.py index bb1fd9318683..b4347dda38f0 100644 --- a/hermes_cli/kanban_db_dispatch.py +++ b/hermes_cli/kanban_db_dispatch.py @@ -2572,6 +2572,7 @@ def _default_spawn(task: Task, workspace: str, *, board: Optional[str] = None) - env["HERMES_TENANT"] = task.tenant env["HERMES_KANBAN_TASK"] = task.id env["HERMES_KANBAN_WORKSPACE"] = workspace + env["HERMES_KANBAN_SAFE_ROOT_ACTIVE"] = "1" # Tag the session `kanban` so session-browsing surfaces filter it out by # source instead of rendering one sidebar row per attempt. env["HERMES_SESSION_SOURCE"] = "kanban" @@ -2586,8 +2587,21 @@ def _default_spawn(task: Task, workspace: str, *, board: Optional[str] = None) - # home) and build_context_files_prompt (#34619 — workers loaded the dispatching gateway's AGENTS.md # instead of the task's). Setting it to the workspace fixes both: the workspace is where the task's work # actually happens. + accepted_workspace = None if workspace and os.path.isabs(workspace) and os.path.isdir(workspace): - env["TERMINAL_CWD"] = workspace + try: + normalized_workspace = os.path.realpath(workspace) + if ( + os.path.dirname(normalized_workspace) != normalized_workspace + and os.pathsep not in normalized_workspace + ): + accepted_workspace = normalized_workspace + except (OSError, ValueError): + pass + if accepted_workspace is not None: + # Scope native file-tool writes to this task root; terminal and OS access remain outside it. + env["TERMINAL_CWD"] = accepted_workspace + env["HERMES_WRITE_SAFE_ROOT"] = accepted_workspace if task.branch_name: env["HERMES_KANBAN_BRANCH"] = task.branch_name if task.current_run_id is not None: diff --git a/tests/conftest.py b/tests/conftest.py index f020209ef817..00c6c4949e42 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -322,6 +322,7 @@ def _looks_like_credential(name: str) -> bool: "HERMES_KANBAN_RUN_ID", "HERMES_KANBAN_CLAIM_LOCK", "HERMES_KANBAN_DISPATCH_IN_GATEWAY", + "HERMES_KANBAN_SAFE_ROOT_ACTIVE", # Pytest is routinely launched from a delegated worker. The worker # lineage marker must not make parent-state tests run as delegated # children; tests that exercise child behavior set it explicitly. diff --git a/tests/hermes_cli/test_kanban_worker_terminal_cwd.py b/tests/hermes_cli/test_kanban_worker_terminal_cwd.py index f2561ffbdea1..f6737ac04173 100644 --- a/tests/hermes_cli/test_kanban_worker_terminal_cwd.py +++ b/tests/hermes_cli/test_kanban_worker_terminal_cwd.py @@ -9,11 +9,21 @@ the dispatching gateway's cwd — relative writes landed in the gateway user's home (#41312) and the wrong profile's ``AGENTS.md`` was loaded (#34619). Pinning ``TERMINAL_CWD`` to the workspace fixes both. + +#70688: dispatcher-spawned workers also inherit a deployment-wide +``HERMES_WRITE_SAFE_ROOT``, letting sibling tasks write each other's workspaces. +The spawn path replaces inherited roots with a normalized task workspace and +preserves that scope through dotenv reload and delegated children. """ from __future__ import annotations +import os import subprocess +from pathlib import Path +from unittest import mock + +import pytest def _make_task(kb, *, assignee: str = "w"): @@ -53,11 +63,33 @@ def fake_popen(cmd, *args, **kwargs): captured["cwd"] = kwargs.get("cwd") return FakeProc() - monkeypatch.setattr(subprocess, "Popen", fake_popen) - kbd._default_spawn(_make_task(kb), workspace) + with monkeypatch.context() as capture_patch: + capture_patch.setattr(subprocess, "Popen", fake_popen) + kbd._default_spawn(_make_task(kb), workspace) return captured +def _shell_file_operations(workspace: Path): + from tools.environments.local import LocalEnvironment + from tools.file_operations import ShellFileOperations + + environment = LocalEnvironment(cwd=str(workspace)) + return ShellFileOperations(environment, cwd=str(workspace)) + + +@pytest.fixture +def kanban_home(tmp_path, monkeypatch): + home = tmp_path / ".hermes" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.setattr(Path, "home", lambda: tmp_path) + from hermes_cli import kanban_db as kb + + kb._INITIALIZED_PATHS.clear() + kb.init_db() + return home + + def test_terminal_cwd_pinned_to_workspace(monkeypatch, tmp_path): """A real, absolute workspace dir is pinned as TERMINAL_CWD.""" root = tmp_path / ".hermes" @@ -72,10 +104,322 @@ def test_terminal_cwd_pinned_to_workspace(monkeypatch, tmp_path): workspace.mkdir() captured = _capture_spawn_env(kb, monkeypatch, str(workspace)) + child_root = os.path.realpath(workspace) - assert captured["env"]["TERMINAL_CWD"] == str(workspace) + assert captured["env"]["TERMINAL_CWD"] == child_root + assert captured["env"]["HERMES_WRITE_SAFE_ROOT"] == child_root # The subprocess cwd and TERMINAL_CWD must agree — both anchor the workspace. assert captured["cwd"] == str(workspace) assert captured["env"]["HERMES_KANBAN_WORKSPACE"] == str(workspace) +def test_worker_lineage_marker_isolated_from_test_process(): + assert os.environ.get("HERMES_KANBAN_SAFE_ROOT_ACTIVE") is None + + +def test_narrow_inherited_root_replaced_for_scratch_workspace_write( + monkeypatch, tmp_path +): + root = tmp_path / "board" + workspace = root / "scratch-a" + sibling = root / "scratch-b" + inherited = tmp_path / "deployment-root" + workspace.mkdir(parents=True) + sibling.mkdir() + inherited.mkdir() + (tmp_path / ".hermes" / "profiles" / "w").mkdir(parents=True) + monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) + monkeypatch.setenv("TERMINAL_CWD", str(inherited)) + monkeypatch.setenv("HERMES_WRITE_SAFE_ROOT", str(inherited)) + + from hermes_cli import kanban_db as kb + + captured = _capture_spawn_env(kb, monkeypatch, str(workspace)) + child_root = os.path.realpath(workspace) + assert captured["env"]["TERMINAL_CWD"] == child_root + + target = workspace / "own.txt" + with mock.patch.dict(os.environ, captured["env"], clear=True): + result = _shell_file_operations(workspace).write_file(str(target), "own") + + assert result.error is None + assert captured["env"]["HERMES_WRITE_SAFE_ROOT"] == child_root + assert target.read_text(encoding="utf-8") == "own" + + +def test_task_safe_root_survives_profile_dotenv_override(monkeypatch, tmp_path): + root = tmp_path / "board" + workspace = root / "scratch-a" + inherited = tmp_path / "deployment-root" + workspace.mkdir(parents=True) + inherited.mkdir() + profile_home = tmp_path / ".hermes" / "profiles" / "w" + profile_home.mkdir(parents=True) + (profile_home / ".env").write_text( + f"HERMES_WRITE_SAFE_ROOT={inherited}\n", encoding="utf-8" + ) + monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) + monkeypatch.setenv("HERMES_WRITE_SAFE_ROOT", str(inherited)) + + from hermes_cli import kanban_db as kb + from hermes_cli.env_loader import load_hermes_dotenv + + captured = _capture_spawn_env(kb, monkeypatch, str(workspace)) + child_root = os.path.realpath(workspace) + with mock.patch.dict(os.environ, captured["env"], clear=True): + load_hermes_dotenv(hermes_home=captured["env"]["HERMES_HOME"]) + target = workspace / "dotenv-own.txt" + result = _shell_file_operations(workspace).write_file(str(target), "own") + + assert result.error is None + assert captured["env"]["HERMES_WRITE_SAFE_ROOT"] == child_root + assert target.read_text(encoding="utf-8") == "own" + + +def test_delegated_worker_safe_root_survives_profile_dotenv_override( + monkeypatch, tmp_path +): + root = tmp_path / "board" + workspace = root / "scratch-a" + inherited = tmp_path / "deployment-root" + workspace.mkdir(parents=True) + inherited.mkdir() + profile_home = tmp_path / ".hermes" / "profiles" / "w" + profile_home.mkdir(parents=True) + (profile_home / ".env").write_text( + f"HERMES_WRITE_SAFE_ROOT={inherited}\n", encoding="utf-8" + ) + monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) + monkeypatch.setenv("HERMES_WRITE_SAFE_ROOT", str(inherited)) + + from agent.delegation_context import scrub_kanban_env + from hermes_cli import kanban_db as kb + from hermes_cli.env_loader import load_hermes_dotenv + + captured = _capture_spawn_env(kb, monkeypatch, str(workspace)) + delegated_env = scrub_kanban_env(captured["env"]) + child_root = os.path.realpath(workspace) + assert "HERMES_KANBAN_TASK" not in delegated_env + assert delegated_env["HERMES_KANBAN_SAFE_ROOT_ACTIVE"] == "1" + with mock.patch.dict(os.environ, delegated_env, clear=True): + load_hermes_dotenv(hermes_home=delegated_env["HERMES_HOME"]) + target = workspace / "delegated-own.txt" + result = _shell_file_operations(workspace).write_file(str(target), "own") + + assert result.error is None + assert target.read_text(encoding="utf-8") == "own" + assert delegated_env["HERMES_WRITE_SAFE_ROOT"] == child_root + + +def test_sibling_and_traversal_mutations_are_denied(monkeypatch, tmp_path): + root = tmp_path / "board" + workspace = root / "scratch-a" + sibling = root / "scratch-b" + workspace.mkdir(parents=True) + sibling.mkdir() + (tmp_path / ".hermes" / "profiles" / "w").mkdir(parents=True) + monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) + monkeypatch.setenv("TERMINAL_CWD", str(root)) + monkeypatch.setenv("HERMES_WRITE_SAFE_ROOT", str(root)) + + from hermes_cli import kanban_db as kb + + direct = sibling / "direct.txt" + traversal = sibling / "traversal.txt" + sibling_write = sibling / "sibling-write.txt" + sibling_delete = sibling / "sibling-delete.txt" + own_move = workspace / "own-move.txt" + moved = sibling / "moved.txt" + direct.write_text("before", encoding="utf-8") + traversal.write_text("before", encoding="utf-8") + sibling_delete.write_text("before", encoding="utf-8") + own_move.write_text("before", encoding="utf-8") + captured = _capture_spawn_env(kb, monkeypatch, str(workspace)) + + with mock.patch.dict(os.environ, captured["env"], clear=True): + ops = _shell_file_operations(workspace) + write_result = ops.write_file(str(sibling_write), "blocked") + direct_result = ops.patch_replace(str(direct), "before", "after") + traversal_result = ops.patch_replace( + str(workspace / ".." / "scratch-b" / "traversal.txt"), + "before", + "after", + ) + delete_result = ops.delete_file(str(sibling_delete)) + move_result = ops.move_file(str(own_move), str(moved)) + + assert write_result.error is not None + assert "outside HERMES_WRITE_SAFE_ROOT" in write_result.error + assert direct_result.error is not None + assert "outside HERMES_WRITE_SAFE_ROOT" in direct_result.error + assert traversal_result.error is not None + assert "outside HERMES_WRITE_SAFE_ROOT" in traversal_result.error + assert delete_result.error is not None + assert "outside HERMES_WRITE_SAFE_ROOT" in delete_result.error + assert move_result.error is not None + assert "outside HERMES_WRITE_SAFE_ROOT" in move_result.error + assert not sibling_write.exists() + assert direct.read_text(encoding="utf-8") == "before" + assert traversal.read_text(encoding="utf-8") == "before" + assert sibling_delete.read_text(encoding="utf-8") == "before" + assert own_move.read_text(encoding="utf-8") == "before" + assert not moved.exists() + + +def test_symlink_escape_is_denied(monkeypatch, tmp_path): + if os.name == "nt": + pytest.skip("symlink boundary is covered on POSIX CI") + + root = tmp_path / "board" + workspace = root / "scratch-a" + outside = root / "outside" + workspace.mkdir(parents=True) + outside.mkdir() + try: + (workspace / "escape").symlink_to(outside, target_is_directory=True) + except OSError as exc: + pytest.skip(f"symlink creation unavailable: {exc}") + (tmp_path / ".hermes" / "profiles" / "w").mkdir(parents=True) + monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) + monkeypatch.setenv("TERMINAL_CWD", str(root)) + monkeypatch.setenv("HERMES_WRITE_SAFE_ROOT", str(root)) + + from hermes_cli import kanban_db as kb + + captured = _capture_spawn_env(kb, monkeypatch, str(workspace)) + target = workspace / "escape" / "escaped.txt" + with mock.patch.dict(os.environ, captured["env"], clear=True): + result = _shell_file_operations(workspace).write_file(str(target), "blocked") + + assert result.error is not None + assert "outside HERMES_WRITE_SAFE_ROOT" in result.error + assert not (outside / "escaped.txt").exists() + + +def test_workspace_variables_share_realpath(monkeypatch, tmp_path): + target = tmp_path / "real-workspace" + workspace = tmp_path / "workspace-link" + target.mkdir() + try: + workspace.symlink_to(target, target_is_directory=True) + except OSError as exc: + pytest.skip(f"symlink creation unavailable: {exc}") + inherited = tmp_path / "inherited-root" + inherited.mkdir() + (tmp_path / ".hermes" / "profiles" / "w").mkdir(parents=True) + monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) + monkeypatch.setenv("TERMINAL_CWD", str(inherited)) + monkeypatch.setenv("HERMES_WRITE_SAFE_ROOT", str(inherited)) + + from hermes_cli import kanban_db as kb + + captured = _capture_spawn_env(kb, monkeypatch, str(workspace)) + assert captured["cwd"] == str(workspace) + assert captured["env"]["HERMES_KANBAN_WORKSPACE"] == str(workspace) + assert captured["env"]["TERMINAL_CWD"] == os.path.realpath(workspace) + assert captured["env"]["HERMES_WRITE_SAFE_ROOT"] == os.path.realpath(workspace) + + +def test_invalid_workspace_preserves_inherited_policy(monkeypatch, tmp_path): + inherited_cwd = tmp_path / "inherited-cwd" + inherited_root = tmp_path / "inherited-root" + existing_file = tmp_path / "workspace-file" + separator_dir = tmp_path / f"workspace{os.pathsep}roots" + inherited_cwd.mkdir() + inherited_root.mkdir() + existing_file.write_text("file", encoding="utf-8") + separator_dir.mkdir() + (tmp_path / ".hermes" / "profiles" / "w").mkdir(parents=True) + monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes")) + monkeypatch.setenv("TERMINAL_CWD", str(inherited_cwd)) + monkeypatch.setenv("HERMES_WRITE_SAFE_ROOT", str(inherited_root)) + + candidates = [ + "relative-workspace", + "", + str(tmp_path / "missing-workspace"), + str(existing_file), + os.path.abspath(os.sep), + str(separator_dir), + ] + from hermes_cli import kanban_db as kb + + for candidate in candidates: + captured = _capture_spawn_env(kb, monkeypatch, candidate) + assert captured["env"]["TERMINAL_CWD"] == str(inherited_cwd), candidate + assert captured["env"]["HERMES_WRITE_SAFE_ROOT"] == str(inherited_root), candidate + + +def test_dispatch_scopes_materialized_scratch_workspace(kanban_home, monkeypatch): + from hermes_cli import kanban_db as kb + from hermes_cli import kanban_db_dispatch as kbd + + inherited = kanban_home.parent / "deployment-root" + inherited.mkdir() + monkeypatch.setenv("TERMINAL_CWD", str(inherited)) + monkeypatch.setenv("HERMES_WRITE_SAFE_ROOT", str(inherited)) + monkeypatch.setattr(kbd, "_resolve_hermes_argv", lambda: ["hermes"]) + captured = {} + + class FakeProc: + pid = 4242 + + def fake_popen(cmd, *args, **kwargs): + captured["env"] = dict(kwargs["env"]) + captured["cwd"] = kwargs["cwd"] + return FakeProc() + + monkeypatch.setattr(subprocess, "Popen", fake_popen) + with kb.connect() as conn: + task_id = kb.create_task( + conn, + title="scoped scratch", + assignee="default", + workspace_kind="scratch", + ) + result = kb.dispatch_once(conn, max_spawn=1) + task = kb.get_task(conn, task_id) + + assert [item[0] for item in result.spawned] == [task_id] + assert task is not None + workspace = Path(task.workspace_path) + assert workspace.is_dir() + assert captured["cwd"] == str(workspace) + assert captured["env"]["HERMES_WRITE_SAFE_ROOT"] == os.path.realpath(workspace) + + +def test_scoped_spawn_failure_records_existing_failure_flow(kanban_home, monkeypatch): + from hermes_cli import kanban_db as kb + from hermes_cli import kanban_db_dispatch as kbd + + monkeypatch.setattr(kbd, "_resolve_hermes_argv", lambda: ["hermes"]) + + def failing_popen(*args, **kwargs): + raise OSError("spawn boom") + + monkeypatch.setattr(subprocess, "Popen", failing_popen) + with kb.connect() as conn: + task_id = kb.create_task( + conn, + title="spawn failure", + assignee="default", + workspace_kind="scratch", + ) + first = kb.dispatch_once(conn, failure_limit=2) + after_first = kb.get_task(conn, task_id) + second = kb.dispatch_once(conn, failure_limit=2) + after_second = kb.get_task(conn, task_id) + event_kinds = [event.kind for event in kb.list_events(conn, task_id)] + + assert first.spawned == [] + assert second.spawned == [] + assert after_first is not None + assert after_first.status == "ready" + assert after_first.claim_lock is None + assert after_first.consecutive_failures == 1 + assert after_second is not None + assert after_second.status == "blocked" + assert after_second.claim_lock is None + assert after_second.consecutive_failures == 2 + assert "spawn_failed" in event_kinds + assert "gave_up" in event_kinds diff --git a/tests/tools/test_write_safe_root_profile_isolation.py b/tests/tools/test_write_safe_root_profile_isolation.py new file mode 100644 index 000000000000..2caf3b0a52a1 --- /dev/null +++ b/tests/tools/test_write_safe_root_profile_isolation.py @@ -0,0 +1,289 @@ +"""#70688 (fluxkapacitor): HERMES_WRITE_SAFE_ROOT must not inherit the launch profile. + +When a secondary-profile session runs inside a long-lived host (desktop/TUI gateway, +multiplex gateway), the process ``os.environ`` carries the launch profile's +``HERMES_WRITE_SAFE_ROOT``. Profile turn scopes install secret + terminal policy +but ``agent.file_safety.get_safe_write_roots()`` still reads the ambient env, so +the routed profile's own vault is refused while the launch profile's vault is +permitted. + +These tests model the real desktop turn boundary (``prompt_turn._prepare_turn_input``) +and the multiplex gateway boundary (``gateway.run._profile_runtime_scope`` + +multiplex dotenv skip). They log setup / expected / actual / pass-fail — not just +a pytest name. +""" + +from __future__ import annotations + +import contextlib +import os +from pathlib import Path + +import pytest + +from agent.file_safety import get_safe_write_roots, get_write_denied_error +from agent.secret_scope import ( + build_profile_secret_scope, + reset_secret_scope, + set_multiplex_active, + set_secret_scope, +) +from hermes_cli.env_loader import load_hermes_dotenv +from hermes_constants import reset_hermes_home_override, set_hermes_home_override +from tools.environments.local import LocalEnvironment +from tools.file_operations import ShellFileOperations +from tools.terminal_scope import install_profile_terminal_scope, reset_terminal_scope +from tools.write_safe_root_scope import ( + install_profile_write_safe_root_scope, + reset_write_safe_root_scope, +) + + +def _profile_home(tmp_path: Path, name: str, *, dotenv: str = "") -> Path: + if name == "default": + home = tmp_path / ".hermes" + else: + home = tmp_path / ".hermes" / "profiles" / name + home.mkdir(parents=True, exist_ok=True) + if dotenv: + (home / ".env").write_text(dotenv, encoding="utf-8") + return home + + +def _file_ops(cwd: Path) -> ShellFileOperations: + env = LocalEnvironment(cwd=str(cwd)) + return ShellFileOperations(env, cwd=str(cwd)) + + +def _write_probe(ops: ShellFileOperations, target: Path) -> tuple[bool, str | None]: + result = ops.write_file(str(target), "probe") + return result.error is None, result.error + + +def _log_case( + *, + label: str, + setup: str, + expected: str, + actual: str, + passed: bool, +) -> None: + status = "PASS" if passed else "FAIL" + why = "matches expectation" if passed else "BUG: launch-profile safe root leaked" + print( + f"\n[{label}] {status}\n" + f" setup: {setup}\n" + f" expected: {expected}\n" + f" actual: {actual}\n" + f" why: {why}" + ) + + +@contextlib.contextmanager +def _desktop_turn_scope(profile_home: str | Path): + """Mirror ``tui_gateway/prompt_turn.py::_prepare_turn_input`` scope binding.""" + home_token = set_hermes_home_override(str(profile_home)) + secret_token = set_secret_scope(build_profile_secret_scope(Path(profile_home))) + terminal_token = install_profile_terminal_scope(Path(profile_home)) + wsr_token = install_profile_write_safe_root_scope(Path(profile_home)) + try: + yield + finally: + reset_write_safe_root_scope(wsr_token) + reset_terminal_scope(terminal_token) + reset_secret_scope(secret_token) + reset_hermes_home_override(home_token) + + +@contextlib.contextmanager +def _multiplex_turn_scope(profile_home: str | Path): + """Mirror ``gateway/run.py::_profile_runtime_scope`` (home + secrets + terminal).""" + with _desktop_turn_scope(profile_home): + yield + + +@pytest.fixture +def launch_env(tmp_path, monkeypatch): + """Launch profile loaded into process env; secondary profile only in its .env.""" + default_vault = tmp_path / "default-vault" + secondary_vault = tmp_path / "secondary-vault" + default_vault.mkdir() + secondary_vault.mkdir() + + default_home = _profile_home( + tmp_path, + "default", + dotenv=f"HERMES_WRITE_SAFE_ROOT={default_vault}\n", + ) + secondary_home = _profile_home( + tmp_path, + "app", + dotenv=f"HERMES_WRITE_SAFE_ROOT={secondary_vault}\n", + ) + + monkeypatch.setenv("HERMES_HOME", str(default_home)) + monkeypatch.delenv("HERMES_WRITE_SAFE_ROOT", raising=False) + + # Desktop/TUI gateway startup: launch profile dotenv becomes process-global. + load_hermes_dotenv(hermes_home=default_home) + + return { + "default_home": default_home, + "secondary_home": secondary_home, + "default_vault": default_vault, + "secondary_vault": secondary_vault, + } + + +def test_desktop_turn_scope_uses_launch_safe_root_not_profile_vault(launch_env): + """Models a desktop-hosted turn for a secondary profile after cli -> desktop.""" + secondary = launch_env["secondary_home"] + own_target = launch_env["secondary_vault"] / "own.txt" + default_target = launch_env["default_vault"] / "private.txt" + + with _desktop_turn_scope(secondary): + roots = get_safe_write_roots() + own_allowed, own_err = _write_probe(_file_ops(launch_env["secondary_vault"]), own_target) + default_allowed, default_err = _write_probe( + _file_ops(launch_env["default_vault"]), default_target + ) + + expected_roots = {os.path.realpath(str(launch_env["secondary_vault"]))} + actual_roots = roots + + _log_case( + label="safe-root resolution", + setup=( + f"process env from {launch_env['default_home']}/.env; " + f"desktop turn scope for {secondary}" + ), + expected=f"roots == {sorted(expected_roots)}", + actual=f"roots == {sorted(actual_roots)}", + passed=actual_roots == expected_roots, + ) + + _log_case( + label="write own vault", + setup=f"target={own_target}", + expected="allowed (inside profile vault)", + actual=f"allowed={own_allowed} error={own_err!r}", + passed=own_allowed, + ) + + _log_case( + label="write launch vault", + setup=f"target={default_target}", + expected="denied (outside profile vault)", + actual=f"allowed={default_allowed} error={default_err!r}", + passed=not default_allowed, + ) + + assert actual_roots == expected_roots + assert own_allowed, own_err + assert not default_allowed, "launch vault must not be writable for secondary profile turn" + + +def test_multiplex_gateway_scope_skips_dotenv_and_keeps_launch_safe_root(launch_env): + """Models multiplex gateway: routed profile .env never reloads into os.environ.""" + secondary = launch_env["secondary_home"] + own_target = launch_env["secondary_vault"] / "own.txt" + default_target = launch_env["default_vault"] / "private.txt" + + set_multiplex_active(True) + home_token = set_hermes_home_override(str(secondary)) + try: + loaded = load_hermes_dotenv(hermes_home=secondary) + assert loaded == [], "multiplex must skip process-global dotenv for routed home" + + with _multiplex_turn_scope(secondary): + roots = get_safe_write_roots() + own_err = get_write_denied_error(str(own_target)) + default_err = get_write_denied_error(str(default_target)) + finally: + reset_hermes_home_override(home_token) + set_multiplex_active(False) + + expected_roots = {os.path.realpath(str(launch_env["secondary_vault"]))} + actual_roots = roots + + _log_case( + label="multiplex dotenv skip", + setup=f"load_hermes_dotenv({secondary}) under multiplex", + expected="[] (no process env mutation)", + actual=f"loaded={loaded}", + passed=loaded == [], + ) + + _log_case( + label="multiplex safe-root resolution", + setup=f"launch env + multiplex turn scope for {secondary}", + expected=f"roots == {sorted(expected_roots)}", + actual=f"roots == {sorted(actual_roots)}", + passed=actual_roots == expected_roots, + ) + + _log_case( + label="multiplex write own vault", + setup=f"target={own_target}", + expected="no denial", + actual=f"error={own_err!r}", + passed=own_err is None, + ) + + _log_case( + label="multiplex write launch vault", + setup=f"target={default_target}", + expected="safe_root denial", + actual=f"error={default_err!r}", + passed=default_err is not None and "outside HERMES_WRITE_SAFE_ROOT" in default_err, + ) + + assert actual_roots == expected_roots + assert own_err is None + assert default_err is not None + + +def test_surface_switch_does_not_rebind_write_safe_root(launch_env): + """cli -> desktop surface switch keeps stored prompt; it must not fix safe roots either.""" + from agent.surface_switch import stage_surface_switch_note + + secondary = launch_env["secondary_home"] + own_target = launch_env["secondary_vault"] / "after-switch.txt" + + class _FakeAgent: + platform = "desktop" + provider = "openai" + api_mode = "chat_completions" + session_id = "sess-70688" + _surface_switch_note = "" + + agent = _FakeAgent() + prompt = "...\n\nPlatform: cli\n\n[runtime tail]" + history = [{"role": "user", "content": "prior turn on cli"}] + + with _desktop_turn_scope(secondary): + switched = stage_surface_switch_note(agent, prompt, history) + roots_after_switch = get_safe_write_roots() + allowed, err = _write_probe(_file_ops(launch_env["secondary_vault"]), own_target) + + expected_roots = {os.path.realpath(str(launch_env["secondary_vault"]))} + + _log_case( + label="surface switch note", + setup="platform cli -> desktop under secondary profile scope", + expected="note staged", + actual=f"switched={switched} note={getattr(agent, '_surface_switch_note', '')[:80]!r}...", + passed=switched, + ) + + _log_case( + label="safe root after surface switch", + setup="same turn scope as desktop prompt submit", + expected=f"roots == {sorted(expected_roots)}", + actual=f"roots == {sorted(roots_after_switch)} allowed={allowed} error={err!r}", + passed=roots_after_switch == expected_roots and allowed, + ) + + assert switched + assert roots_after_switch == expected_roots + assert allowed, err diff --git a/tests/tools/test_write_safe_root_profile_matrix.py b/tests/tools/test_write_safe_root_profile_matrix.py new file mode 100644 index 000000000000..209c3a452ce2 --- /dev/null +++ b/tests/tools/test_write_safe_root_profile_matrix.py @@ -0,0 +1,243 @@ +"""HERMES_WRITE_SAFE_ROOT permutation matrix (#70688 coverage extension). + +Models routed desktop/multiplex turn scope (same path as test_write_safe_root_profile_isolation). +Every case prints setup / expected / actual / why. +""" + +from __future__ import annotations + +import os +from pathlib import Path + +from agent.file_safety import get_safe_write_roots +from hermes_cli.env_loader import load_hermes_dotenv +from tests.tools.test_write_safe_root_profile_isolation import ( + _desktop_turn_scope, + _file_ops, + _log_case, + _profile_home, + _write_probe, +) +from tools.write_safe_root_scope import get_write_safe_root_scope + + +def _load_launch(tmp_path, monkeypatch, *, launch_dotenv: str, secondary_dotenv: str): + """Launch profile dotenv → process env; secondary profile optional .env.""" + launch_home = _profile_home(tmp_path, "default", dotenv=launch_dotenv) + secondary_home = _profile_home(tmp_path, "app", dotenv=secondary_dotenv) + + monkeypatch.setenv("HERMES_HOME", str(launch_home)) + monkeypatch.delenv("HERMES_WRITE_SAFE_ROOT", raising=False) + load_hermes_dotenv(hermes_home=launch_home) + + return { + "launch_home": launch_home, + "secondary_home": secondary_home, + "process_wsr": os.environ.get("HERMES_WRITE_SAFE_ROOT"), + } + + +def _assert_write_matrix( + *, + label: str, + env: dict, + own_target: Path, + foreign_target: Path, + expected_own_allowed: bool, + expected_foreign_allowed: bool, + expected_roots: set[str] | None = None, +): + with _desktop_turn_scope(env["secondary_home"]): + roots = get_safe_write_roots() + own_allowed, own_err = _write_probe(_file_ops(own_target.parent), own_target) + foreign_allowed, foreign_err = _write_probe( + _file_ops(foreign_target.parent), foreign_target + ) + scoped = get_write_safe_root_scope() + + if expected_roots is not None: + _log_case( + label=f"{label} roots", + setup=f"scoped={scoped!r} process={env['process_wsr']!r}", + expected=f"roots == {sorted(expected_roots)}", + actual=f"roots == {sorted(roots)}", + passed=roots == expected_roots, + ) + assert roots == expected_roots + + _log_case( + label=f"{label} own vault", + setup=f"target={own_target}", + expected="allowed" if expected_own_allowed else "denied", + actual=f"allowed={own_allowed} error={own_err!r}", + passed=own_allowed == expected_own_allowed, + ) + _log_case( + label=f"{label} foreign/launch vault", + setup=f"target={foreign_target}", + expected="allowed" if expected_foreign_allowed else "denied", + actual=f"allowed={foreign_allowed} error={foreign_err!r}", + passed=foreign_allowed == expected_foreign_allowed, + ) + + assert own_allowed == expected_own_allowed, own_err + assert foreign_allowed == expected_foreign_allowed, foreign_err + + +def test_m1_launch_unset_secondary_set(tmp_path, monkeypatch): + """M1: launch WSR unset; secondary WSR set → own allow, foreign deny.""" + foreign_vault = tmp_path / "secondary-only-vault" + foreign_vault.mkdir() + env = _load_launch( + tmp_path, + monkeypatch, + launch_dotenv="OPENAI_API_KEY=placeholder\n", + secondary_dotenv=f"HERMES_WRITE_SAFE_ROOT={foreign_vault}\n", + ) + own = foreign_vault / "own.txt" + foreign = tmp_path / "outside.txt" + foreign.parent.mkdir(exist_ok=True) + + _assert_write_matrix( + label="M1", + env=env, + own_target=own, + foreign_target=foreign, + expected_own_allowed=True, + expected_foreign_allowed=False, + expected_roots={os.path.realpath(str(foreign_vault))}, + ) + + +def test_m2_launch_set_secondary_unset(tmp_path, monkeypatch): + """M2 hard gate: launch WSR set, secondary unset → launch vault must stay denied.""" + launch_vault = tmp_path / "launch-vault" + launch_vault.mkdir() + env = _load_launch( + tmp_path, + monkeypatch, + launch_dotenv=f"HERMES_WRITE_SAFE_ROOT={launch_vault}\n", + secondary_dotenv="OPENAI_API_KEY=placeholder\n", + ) + own_outside = tmp_path / "neutral.txt" + launch_target = launch_vault / "leak.txt" + + _assert_write_matrix( + label="M2", + env=env, + own_target=own_outside, + foreign_target=launch_target, + expected_own_allowed=True, + expected_foreign_allowed=False, + expected_roots=set(), + ) + + +def test_m3_both_unset_no_process_leak(tmp_path, monkeypatch): + """M3: both unset; no leftover process WSR opens a foreign vault.""" + env = _load_launch( + tmp_path, + monkeypatch, + launch_dotenv="OPENAI_API_KEY=placeholder\n", + secondary_dotenv="OPENAI_API_KEY=other\n", + ) + monkeypatch.delenv("HERMES_WRITE_SAFE_ROOT", raising=False) + assert env["process_wsr"] in (None, "") + + neutral = tmp_path / "neutral.txt" + foreign = tmp_path / "other.txt" + + _assert_write_matrix( + label="M3", + env=env, + own_target=neutral, + foreign_target=foreign, + expected_own_allowed=True, + expected_foreign_allowed=True, + expected_roots=set(), + ) + + +def test_m4_nested_parent_child_vaults(tmp_path, monkeypatch): + """M4: nested parent vs child WSR → child allow; parent-outside-child deny.""" + parent = tmp_path / "Developer" + child = parent / "app" + parent.mkdir() + child.mkdir(parents=True) + launch_home = _profile_home( + tmp_path, "default", dotenv=f"HERMES_WRITE_SAFE_ROOT={parent}\n", + ) + secondary_home = _profile_home( + tmp_path, "app", dotenv=f"HERMES_WRITE_SAFE_ROOT={child}\n", + ) + monkeypatch.setenv("HERMES_HOME", str(launch_home)) + monkeypatch.delenv("HERMES_WRITE_SAFE_ROOT", raising=False) + load_hermes_dotenv(hermes_home=launch_home) + env = { + "launch_home": launch_home, + "secondary_home": secondary_home, + "parent_vault": parent, + "child_vault": child, + "process_wsr": os.environ.get("HERMES_WRITE_SAFE_ROOT"), + } + + child_target = child / "own.txt" + parent_outside = parent / "sibling.txt" + + _assert_write_matrix( + label="M4 child", + env=env, + own_target=child_target, + foreign_target=parent_outside, + expected_own_allowed=True, + expected_foreign_allowed=False, + expected_roots={os.path.realpath(str(child))}, + ) + + +def test_m5_unscoped_reads_process_env(tmp_path, monkeypatch): + """M5: scope not bound → get_safe_write_roots follows os.environ.""" + vault = tmp_path / "env-vault" + vault.mkdir() + monkeypatch.delenv("HERMES_WRITE_SAFE_ROOT", raising=False) + + roots_unset = get_safe_write_roots() + _log_case( + label="M5 unset", + setup="scope unbound, HERMES_WRITE_SAFE_ROOT unset", + expected="roots == set()", + actual=f"roots == {sorted(roots_unset)}", + passed=roots_unset == set(), + ) + + monkeypatch.setenv("HERMES_WRITE_SAFE_ROOT", str(vault)) + roots_set = get_safe_write_roots() + expected = {os.path.realpath(str(vault))} + _log_case( + label="M5 set", + setup=f"scope unbound, env={vault}", + expected=f"roots == {sorted(expected)}", + actual=f"roots == {sorted(roots_set)}", + passed=roots_set == expected, + ) + assert roots_unset == set() + assert roots_set == expected + + +def test_m6_empty_string_vs_absent_key_same_when_scoped(tmp_path, monkeypatch): + """M6: HERMES_WRITE_SAFE_ROOT= vs key absent → same scoped build (empty).""" + from tools.write_safe_root_scope import build_profile_write_safe_root + + absent = _profile_home(tmp_path, "absent", dotenv="FOO=bar\n") + empty = _profile_home(tmp_path, "empty", dotenv="HERMES_WRITE_SAFE_ROOT=\nFOO=bar\n") + + absent_val = build_profile_write_safe_root(absent) + empty_val = build_profile_write_safe_root(empty) + _log_case( + label="M6 build", + setup="profile .env absent key vs empty value", + expected="both == ''", + actual=f"absent={absent_val!r} empty={empty_val!r}", + passed=absent_val == "" == empty_val, + ) + assert absent_val == empty_val == "" diff --git a/tests/tui_gateway/test_write_safe_root_profile_desktop.py b/tests/tui_gateway/test_write_safe_root_profile_desktop.py new file mode 100644 index 000000000000..cf83f057743b --- /dev/null +++ b/tests/tui_gateway/test_write_safe_root_profile_desktop.py @@ -0,0 +1,131 @@ +"""Regression: desktop/TUI routed profile must not inherit launch HERMES_WRITE_SAFE_ROOT (#70688). + +fluxkapacitor (Sep 11 2026): a desktop-hosted session for a non-default profile ran in a +process that still carried the default profile's ``HERMES_WRITE_SAFE_ROOT``. Own-vault writes +were refused; the default profile's vault was writable — inverted permissions. + +This models the real host path: ``tui_gateway/server.py`` loads the launch profile dotenv once at +startup; a routed session binds ``_session_profile_runtime_scope`` (home + secret + terminal +scopes) per turn — but ``agent/file_safety.get_safe_write_roots()`` reads ``os.environ`` only. + +Every check prints expected vs actual (brief requirement); silent asserts alone are insufficient. +""" + +from __future__ import annotations + +import os +from pathlib import Path + +import pytest + +from agent.file_safety import get_safe_write_roots, get_write_denied_error +from agent.secret_scope import set_multiplex_active +from hermes_cli.env_loader import load_hermes_dotenv + + +def _report(check: str, expected, actual) -> bool: + ok = expected == actual + print(f"\n CHECK: {check}") + print(f" expected: {expected!r}") + print(f" actual: {actual!r}") + print(f" result: {'PASS' if ok else 'FAIL'}") + return ok + + +def _profile_home(tmp_path: Path, name: str, *, dotenv: str) -> Path: + home = tmp_path / ".hermes" / "profiles" / name + home.mkdir(parents=True) + (home / ".env").write_text(dotenv, encoding="utf-8") + (home / "config.yaml").write_text("toolsets: []\n", encoding="utf-8") + return home + + +def test_routed_profile_write_safe_root_matches_profile_dotenv_not_launch_process( + tmp_path, monkeypatch, +): + """After launch dotenv + desktop session scope, file writes use the routed profile vault. + + fluxkapacitor used nested ``~/Developer`` vs ``~/Developer/`` on macOS; disjoint + sibling vaults here assert post-fix behavior: own vault allowed, foreign vault denied. + """ + default_vault = tmp_path / "default-vault" + profile_vault = tmp_path / "profile-vault" + default_vault.mkdir() + profile_vault.mkdir() + + launch_home = tmp_path / ".hermes" + launch_home.mkdir() + (launch_home / ".env").write_text( + f"HERMES_WRITE_SAFE_ROOT={default_vault}\n", encoding="utf-8" + ) + (launch_home / "config.yaml").write_text("toolsets: []\n", encoding="utf-8") + + profile_home = _profile_home( + tmp_path, + "bee", + dotenv=f"HERMES_WRITE_SAFE_ROOT={profile_vault}\n", + ) + + print("\n=== SETUP ===") + print(f" launch_home: {launch_home}") + print(f" profile_home: {profile_home}") + print(f" default_vault: {default_vault}") + print(f" profile_vault: {profile_vault}") + print(" host path: load_hermes_dotenv(launch) then _session_profile_runtime_scope") + + monkeypatch.setenv("HERMES_HOME", str(launch_home)) + monkeypatch.setattr(Path, "home", lambda: tmp_path) + set_multiplex_active(True) + + # Mirror tui_gateway/server.py module import: launch profile dotenv → process env. + load_hermes_dotenv(hermes_home=launch_home) + launch_wsr = os.environ.get("HERMES_WRITE_SAFE_ROOT") + _report( + "process env after launch dotenv (default vault)", + str(default_vault), + launch_wsr, + ) + + import tui_gateway.server as server + + own_target = profile_vault / "own.txt" + foreign_target = default_vault / "foreign.txt" + profile_root = os.path.realpath(profile_vault) + + with server._session_profile_runtime_scope({"profile_home": str(profile_home)}): + active_roots = get_safe_write_roots() + own_error = get_write_denied_error(str(own_target)) + foreign_error = get_write_denied_error(str(foreign_target)) + + print("\n=== INSIDE ROUTED DESKTOP SESSION SCOPE ===") + checks = [ + _report( + "get_safe_write_roots()", + {profile_root}, + active_roots, + ), + _report( + "write own profile vault (denial error)", + None, + own_error, + ), + _report( + "write default profile vault blocked", + True, + foreign_error is not None + and "outside HERMES_WRITE_SAFE_ROOT" in foreign_error, + ), + ] + if foreign_error is not None: + print(f" foreign_error text: {foreign_error!r}") + + print("\n=== SUMMARY ===") + failed = [i for i, ok in enumerate(checks, 1) if not ok] + if failed: + print(f" FAILED checks: {failed}") + else: + print(" all checks passed") + + assert checks[0], "active safe roots must be the routed profile vault, not launch process env" + assert checks[1], f"own-vault write must be allowed; got {own_error!r}" + assert checks[2], f"default-vault write must be denied; got {foreign_error!r}" diff --git a/tools/write_safe_root_scope.py b/tools/write_safe_root_scope.py new file mode 100644 index 000000000000..649440875679 --- /dev/null +++ b/tools/write_safe_root_scope.py @@ -0,0 +1,77 @@ +"""Per-turn HERMES_WRITE_SAFE_ROOT scope for multiplexed surfaces (#70688). + +Multiplexed hosts load the launch profile's dotenv into ``os.environ`` once; routed +profile turns must not inherit that vault. Like ``tools/terminal_scope``, a ContextVar +holds the active profile's value; while bound, ``agent.file_safety.get_safe_write_roots()`` +resolves ONLY from it (never ambient env). An empty profile value means no profile vault, +but inherited launch-process roots stay blocked (#70688 M2). +""" + +from __future__ import annotations + +import os +from contextlib import contextmanager, suppress +from contextvars import ContextVar, Token +from pathlib import Path +from typing import Any, Iterator, Optional + +_write_safe_root_scope_var: ContextVar[Optional[str]] = ContextVar( + "hermes_write_safe_root_scope", default=None, +) + + +def set_write_safe_root_scope(value: Optional[str]) -> Token: + return _write_safe_root_scope_var.set(value) + + +def reset_write_safe_root_scope(token: Token) -> None: + _write_safe_root_scope_var.reset(token) + + +def get_write_safe_root_scope() -> Optional[str]: + return _write_safe_root_scope_var.get() + + +def is_write_safe_root_scope_bound() -> bool: + return get_write_safe_root_scope() is not None + + +def get_process_write_safe_roots() -> set[str]: + """Resolved ``HERMES_WRITE_SAFE_ROOT`` from the launch process env only.""" + roots: set[str] = set() + for path in filter(None, os.environ.get("HERMES_WRITE_SAFE_ROOT", "").split(os.pathsep)): + with suppress(OSError, ValueError): + roots.add(os.path.realpath(os.path.expanduser(path))) + return roots + + +def write_safe_root_env() -> str: + """Authoritative read of ``HERMES_WRITE_SAFE_ROOT`` for file write guards.""" + scope = _write_safe_root_scope_var.get() + if scope is None: + return os.environ.get("HERMES_WRITE_SAFE_ROOT", "") + return scope + + +def build_profile_write_safe_root(hermes_home: Any) -> str: + """Read ``HERMES_WRITE_SAFE_ROOT`` from a profile home's ``.env`` only.""" + env_path = Path(hermes_home) / ".env" + if not env_path.is_file(): + return "" + from agent.secret_scope import load_env_file + + value = load_env_file(env_path).get("HERMES_WRITE_SAFE_ROOT") + return "" if value is None else str(value) + + +def install_profile_write_safe_root_scope(hermes_home: Any) -> Token: + return set_write_safe_root_scope(build_profile_write_safe_root(hermes_home)) + + +@contextmanager +def install_and_reset_profile_write_safe_root_scope(hermes_home: Any) -> Iterator[None]: + token = install_profile_write_safe_root_scope(hermes_home) + try: + yield + finally: + reset_write_safe_root_scope(token) diff --git a/tui_gateway/model_switch.py b/tui_gateway/model_switch.py index 0bf80e54c955..58bc723a0c0e 100644 --- a/tui_gateway/model_switch.py +++ b/tui_gateway/model_switch.py @@ -85,21 +85,28 @@ def _profile_runtime_scope_tokens(profile_home) -> "_TurnScopes": # Same terminal policy the gateway binds per turn: a docker-configured profile # must never resolve the launch process's pinned env. Failure → refusal scope. from tools.terminal_scope import install_profile_terminal_scope + from tools.write_safe_root_scope import install_profile_write_safe_root_scope scopes.terminal = install_profile_terminal_scope(home, env_overlay=overlay) + scopes.write_safe_root = install_profile_write_safe_root_scope(home) return scopes def _release_profile_runtime_scope_tokens(scopes: "_TurnScopes | None") -> None: - """Release terminal → secret → home. Each reset is independent: a failing terminal reset must - not leave the previous profile's secrets / HERMES_HOME installed for the next body in this - context (a fail-open scope leak on the teardown path). The first failure is re-raised after - every scope has been released.""" + """Release terminal → write-safe-root → secret → home. Each reset is independent: a failing + terminal reset must not leave the previous profile's secrets / HERMES_HOME installed for the + next body in this context (a fail-open scope leak on the teardown path). The first failure is + re-raised after every scope has been released.""" if scopes is None: return from tools.terminal_scope import reset_terminal_scope + from tools.write_safe_root_scope import reset_write_safe_root_scope first_error: BaseException | None = None - for token, reset in ((scopes.terminal, reset_terminal_scope), (scopes.secret, reset_secret_scope), - (scopes.home, reset_hermes_home_override)): + for token, reset in ( + (scopes.terminal, reset_terminal_scope), + (scopes.write_safe_root, reset_write_safe_root_scope), + (scopes.secret, reset_secret_scope), + (scopes.home, reset_hermes_home_override), + ): if token is None: continue try: diff --git a/tui_gateway/prompt_turn.py b/tui_gateway/prompt_turn.py index d04de58ecf9e..d6cc68bda1be 100644 --- a/tui_gateway/prompt_turn.py +++ b/tui_gateway/prompt_turn.py @@ -190,6 +190,7 @@ class _TurnScopes: home: Any = None # per-turn HERMES_HOME override for a resumed remote profile secret: Any = None terminal: Any = None + write_safe_root: Any = None def _route_turn_images(agent, prompt: Any, images: list[str]) -> Any: @@ -498,6 +499,7 @@ def _prepare_turn_input(sid: str, session: dict, st: _TurnRun, text: Any, images bound = _profile_runtime_scope_tokens(session.get("profile_home")) if bound is not None: scopes.home, scopes.secret, scopes.terminal = bound.home, bound.secret, bound.terminal + scopes.write_safe_root = bound.write_safe_root # The sudo password callback is thread-local: without re-wiring here, sudo prompts # fall through to /dev/tty and hang the headless gateway (re-run is a no-op). _wire_callbacks(sid) @@ -810,6 +812,9 @@ def _finish_turn(sid: str, session: dict, st: _TurnRun) -> None: if scopes.terminal is not None: from tools.terminal_scope import reset_terminal_scope reset_terminal_scope(scopes.terminal) + if scopes.write_safe_root is not None: + from tools.write_safe_root_scope import reset_write_safe_root_scope + reset_write_safe_root_scope(scopes.write_safe_root) _clear_session_context(scopes.session_tokens) diff --git a/website/docs/user-guide/features/kanban-worker-lanes.md b/website/docs/user-guide/features/kanban-worker-lanes.md index 39c7d53c2b19..4a1ea32fefb4 100644 --- a/website/docs/user-guide/features/kanban-worker-lanes.md +++ b/website/docs/user-guide/features/kanban-worker-lanes.md @@ -41,11 +41,19 @@ For Hermes profile lanes, the dispatcher's `_default_spawn` runs `hermes -p ::`) | +| `HERMES_WRITE_SAFE_ROOT` | the normalized absolute task workspace used by native file tools | +| `TERMINAL_CWD` | the same task workspace used as the worker process's current directory | | `HERMES_PROFILE` | the worker's own profile name (for `kanban_comment` author attribution) | | `HERMES_TENANT` | tenant namespace, if the task has one | For non-Hermes lanes (registered via a plugin), the plugin supplies its own `spawn_fn` callable that gets `task`, `workspace`, and `board` and returns an optional pid for crash detection. +The default Hermes lane scopes native `write_file` and `patch` operations to the task workspace through `HERMES_WRITE_SAFE_ROOT`. This is defense in depth, not an operating-system sandbox: terminal commands and subprocesses can still write as the worker's OS user, so integrations that need a hard boundary must provide their own sandbox. + +:::caution Multi-board installs +Setting global `HERMES_KANBAN_DB`, `HERMES_KANBAN_WORKSPACES_ROOT`, or `HERMES_KANBAN_ATTACHMENTS_ROOT` in a shared shell or `.env` collapses board isolation — every worker on that host resolves the same database and workspace tree. Leave these unset for multi-board deployments; the dispatcher injects per-board values at spawn time. +::: + ### Descendant process scope A task assignment belongs to the dispatcher worker, not to every program it starts. diff --git a/website/docs/user-guide/security.md b/website/docs/user-guide/security.md index f06b2571b7c9..a872f795347f 100644 --- a/website/docs/user-guide/security.md +++ b/website/docs/user-guide/security.md @@ -340,6 +340,8 @@ Useful flags: `--days N` (history window, default 90), `--min-count N` Before `write_file` or `patch` touches disk, Hermes checks the target path against a denylist and an optional sandbox. Blocked writes return an error to the agent immediately — **there is no approval prompt** and no way to override from the chat UI. The model may still claim the edit succeeded; when `display.file_mutation_verifier` is on (default), trust the [file-mutation verifier footer](./configuration.md#file-mutation-verifier) over the assistant's closing summary. +Kanban workers receive a task-scoped `HERMES_WRITE_SAFE_ROOT` and `TERMINAL_CWD`. The file-tool root limits native `write_file` and `patch` mutations to that task workspace; terminal commands and child processes retain the worker's OS privileges and are outside this boundary. + ### Protected paths (always blocked) These categories are always denied, even when `HERMES_WRITE_SAFE_ROOT` is unset: