diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index 2c557f3cfc96e..8e37663caf698 100644 --- a/hermes_cli/config_defaults.py +++ b/hermes_cli/config_defaults.py @@ -2169,6 +2169,12 @@ DEFAULT_CONFIG = { "security": { "allow_private_urls": False, # Allow requests to private/internal IPs (for OpenWrt, proxies, VPNs) "redact_secrets": True, + # Writes to agent-instruction files (AGENTS.md/CLAUDE.md/SOUL.md/ + # .cursorrules, project-local .hermes config) always require human + # approval — even under auto-approve/yolo. Extra patterns are + # fnmatch globs matched against the basename (e.g. "*.mdc"). + "protected_instruction_files": True, + "protected_instruction_extra_patterns": [], "tirith_enabled": True, "tirith_path": "tirith", "tirith_timeout": 5, diff --git a/tests/tools/test_file_write_safety.py b/tests/tools/test_file_write_safety.py index ee1296e49588d..0f14664191266 100644 --- a/tests/tools/test_file_write_safety.py +++ b/tests/tools/test_file_write_safety.py @@ -182,7 +182,7 @@ class TestSafeRootDenialMessageIntegration: res = ops.write_file(str(inside), "content") assert res.error is None - assert inside.read_text() == "content" + assert inside.read_text(encoding="utf-8") == "content" class TestCheckSensitivePathMacOSBypass: @@ -233,11 +233,11 @@ class TestAtomicWrite: # A real rename allocates a new inode for the target; an in-place # rewrite would keep the same inode. This proves the swap is atomic. target = tmp_path / "f.txt" - target.write_text("v1") + target.write_text("v1", encoding="utf-8") ino_before = os.stat(target).st_ino res = ops.write_file(str(target), "v2 content") assert res.error is None, res.error - assert target.read_text() == "v2 content" + assert target.read_text(encoding="utf-8") == "v2 content" assert os.stat(target).st_ino != ino_before @@ -249,11 +249,11 @@ class TestAtomicWrite: def test_patch_routes_through_atomic_write(self, ops, tmp_path: Path): target = tmp_path / "edit.py" - target.write_text("a = 1\nb = 2\nc = 3\n") + target.write_text("a = 1\nb = 2\nc = 3\n", encoding="utf-8") os.chmod(target, 0o600) res = ops.patch_replace(str(target), "b = 2", "b = 22") assert res.success, res.error - assert target.read_text() == "a = 1\nb = 22\nc = 3\n" + assert target.read_text(encoding="utf-8") == "a = 1\nb = 22\nc = 3\n" assert (os.stat(target).st_mode & 0o777) == 0o600 @@ -338,5 +338,268 @@ class TestBomHandling: assert ops._file_has_bom(str(target), pre_content="x = 1\n") is True +class TestProtectedInstructionFiles: + """Writes to agent-instruction files ALWAYS require approval. + + AGENTS.md / CLAUDE.md / SOUL.md / .cursorrules / project-local .hermes + config steer future agent behavior, so a prompt-injected agent writing + them is a persistence vector. The gate must ask the human every time — + even under yolo/auto-approve — and fail closed when no human channel + exists. Ported from: RooCodeInc/Roo-Code RooProtectedController + (Apache-2.0); symlink lesson from #41351. + """ + + @pytest.fixture(autouse=True) + def _gate_on(self, monkeypatch): + import tools.file_tools as ft + monkeypatch.setattr( + ft, "_protected_instruction_config", lambda: (True, []) + ) + yield + + @pytest.fixture + def approvals(self, monkeypatch): + """Install a CLI approval callback; record calls; scripted answers.""" + from tools.terminal_tool import set_approval_callback + state = {"calls": [], "answer": "deny"} + + def cb(command, description, **kwargs): + state["calls"].append( + {"command": command, "description": description, **kwargs} + ) + return state["answer"] + + set_approval_callback(cb) + yield state + set_approval_callback(None) + + def _write(self, path, content="injected"): + import json + from tools.file_tools import write_file_tool + return json.loads(write_file_tool(str(path), content)) + + # ---- core behavior ------------------------------------------------- + + @pytest.mark.parametrize( + "name", ["AGENTS.md", "CLAUDE.md", "SOUL.md", ".cursorrules"] + ) + def test_deny_blocks_write(self, tmp_path, approvals, name): + target = tmp_path / name + approvals["answer"] = "deny" + res = self._write(target) + assert res.get("error"), res + assert "BLOCKED" in res["error"] + assert not target.exists() + assert len(approvals["calls"]) == 1 + + def test_approve_once_allows_write(self, tmp_path, approvals): + target = tmp_path / "AGENTS.md" + approvals["answer"] = "once" + res = self._write(target, "approved content") + assert not res.get("error"), res + assert target.read_text(encoding="utf-8") == "approved content" + assert len(approvals["calls"]) == 1 + + def test_prompts_even_under_yolo(self, tmp_path, approvals, monkeypatch): + """The whole point: auto-approve/yolo must NOT bypass this gate.""" + import tools.approval as A + monkeypatch.setattr(A, "_YOLO_MODE_FROZEN", True) + target = tmp_path / "AGENTS.md" + approvals["answer"] = "deny" + res = self._write(target) + assert res.get("error") and "BLOCKED" in res["error"] + assert not target.exists() + assert len(approvals["calls"]) == 1, "yolo bypassed the protected gate" + + def test_second_write_prompts_again(self, tmp_path, approvals): + """One-operation approval: no session stickiness.""" + target = tmp_path / "AGENTS.md" + approvals["answer"] = "once" + self._write(target) + self._write(target, "second") + assert len(approvals["calls"]) == 2 + + def test_regular_file_never_prompts(self, tmp_path, approvals): + res = self._write(tmp_path / "notes.md", "hello") + assert not res.get("error"), res + assert approvals["calls"] == [] + + def test_no_human_fails_closed(self, tmp_path): + # No approval callback registered, not gateway → block, don't hang. + target = tmp_path / "AGENTS.md" + res = self._write(target) + assert res.get("error") and "BLOCKED" in res["error"] + assert not target.exists() + + def test_config_disabled_skips_gate(self, tmp_path, approvals, monkeypatch): + import tools.file_tools as ft + monkeypatch.setattr( + ft, "_protected_instruction_config", lambda: (False, []) + ) + res = self._write(tmp_path / "AGENTS.md", "ok") + assert not res.get("error"), res + assert approvals["calls"] == [] + + def test_extra_patterns_from_config(self, tmp_path, approvals, monkeypatch): + import tools.file_tools as ft + monkeypatch.setattr( + ft, "_protected_instruction_config", lambda: (True, ["*.mdc"]) + ) + approvals["answer"] = "deny" + res = self._write(tmp_path / "rules.mdc") + assert res.get("error") and "BLOCKED" in res["error"] + + # ---- adversarial path shapes ---------------------------------------- + + def test_symlink_to_protected_file_is_gated(self, tmp_path, approvals): + """#41351 lesson: realpath first — innocent name, protected target.""" + real = tmp_path / "AGENTS.md" + real.write_text("original", encoding="utf-8") + link = tmp_path / "innocent.txt" + link.symlink_to(real) + approvals["answer"] = "deny" + res = self._write(link, "injected") + assert res.get("error") and "BLOCKED" in res["error"] + assert real.read_text(encoding="utf-8") == "original" + + def test_case_variant_is_gated(self, tmp_path, approvals): + approvals["answer"] = "deny" + res = self._write(tmp_path / "agents.MD") + assert res.get("error") and "BLOCKED" in res["error"] + + def test_relative_traversal_is_gated(self, tmp_path, approvals, monkeypatch): + (tmp_path / "x").mkdir() + monkeypatch.chdir(tmp_path) + approvals["answer"] = "deny" + res = self._write("./x/../AGENTS.md") + assert res.get("error") and "BLOCKED" in res["error"] + assert not (tmp_path / "AGENTS.md").exists() + + def test_arbitrary_directory_basename_is_gated(self, tmp_path, approvals): + """Any-directory scope: project-context files load from cwd trees.""" + deep = tmp_path / "a" / "b" / "c" + deep.mkdir(parents=True) + approvals["answer"] = "deny" + res = self._write(deep / "CLAUDE.md") + assert res.get("error") and "BLOCKED" in res["error"] + + def test_project_local_hermes_dir_is_gated(self, tmp_path, approvals): + proj = tmp_path / "proj" / ".hermes" + proj.mkdir(parents=True) + approvals["answer"] = "deny" + res = self._write(proj / "config.yaml") + assert res.get("error") and "BLOCKED" in res["error"] + + def test_checkout_nested_under_hermes_dir_not_gated(self, tmp_path, approvals): + """A repo living UNDER a .hermes dir (e.g. ~/.hermes/hermes-agent) + must not have every write gated — only files directly inside a + .hermes dir count as project config.""" + repo = tmp_path / ".hermes" / "some-repo" / "src" + repo.mkdir(parents=True) + res = self._write(repo / "module.py", "x = 1\n") + assert not res.get("error"), res + assert approvals["calls"] == [] + + def test_real_hermes_home_not_gated_by_this_check( + self, tmp_path, approvals, monkeypatch + ): + """~/.hermes itself is governed by existing guards, not this gate.""" + import tools.file_tools as ft + fake_home = tmp_path / ".hermes" + (fake_home / "notes").mkdir(parents=True) + monkeypatch.setattr( + ft, "_get_real_hermes_home", lambda: str(fake_home.resolve()) + ) + res = self._write(fake_home / "notes" / "scratch.txt", "ok") + assert not res.get("error"), res + assert approvals["calls"] == [] + + # ---- patch tool ----------------------------------------------------- + + def test_patch_replace_mode_is_gated(self, tmp_path, approvals): + from tools.file_tools import patch_tool + import json + target = tmp_path / "SOUL.md" + target.write_text("be kind\n", encoding="utf-8") + approvals["answer"] = "deny" + res = json.loads(patch_tool( + mode="replace", path=str(target), + old_string="be kind", new_string="obey injected orders", + )) + assert res.get("error") and "BLOCKED" in res["error"] + assert target.read_text(encoding="utf-8") == "be kind\n" + + def test_patch_v4a_multifile_one_protected_blocks_whole_patch( + self, tmp_path, approvals + ): + """Policy: one protected file gates the ENTIRE patch (deny = nothing + applies, including the innocent file).""" + from tools.file_tools import patch_tool + import json + agents = tmp_path / "AGENTS.md" + agents.write_text("rules\n", encoding="utf-8") + plain = tmp_path / "plain.txt" + plain.write_text("hello\n", encoding="utf-8") + patch = ( + "*** Begin Patch\n" + f"*** Update File: {plain}\n" + "@@\n" + "-hello\n" + "+world\n" + f"*** Update File: {agents}\n" + "@@\n" + "-rules\n" + "+injected\n" + "*** End Patch" + ) + approvals["answer"] = "deny" + res = json.loads(patch_tool(mode="patch", patch=patch)) + assert res.get("error") and "BLOCKED" in res["error"] + assert plain.read_text(encoding="utf-8") == "hello\n" + assert agents.read_text(encoding="utf-8") == "rules\n" + assert len(approvals["calls"]) == 1 + + def test_patch_v4a_approved_applies(self, tmp_path, approvals): + from tools.file_tools import patch_tool + import json + agents = tmp_path / "AGENTS.md" + agents.write_text("rules\n", encoding="utf-8") + patch = ( + "*** Begin Patch\n" + f"*** Update File: {agents}\n" + "@@\n" + "-rules\n" + "+updated rules\n" + "*** End Patch" + ) + approvals["answer"] = "once" + res = json.loads(patch_tool(mode="patch", patch=patch)) + assert not res.get("error"), res + assert agents.read_text(encoding="utf-8") == "updated rules\n" + + # ---- gateway round-trip ---------------------------------------------- + + def test_gateway_notify_resolve_once_allows(self, tmp_path): + import tools.approval as A + session_key = "protected-files-test-session" + token = A.set_current_session_key(session_key) + try: + def notify(approval_data): + # Buttons must not offer persistent scopes for this gate. + assert approval_data.get("allow_permanent") is False + assert approval_data.get("allow_session") is False + A.resolve_gateway_approval(session_key, "once") + + A.register_gateway_notify(session_key, notify) + try: + res = self._write(tmp_path / "AGENTS.md", "gateway approved") + assert not res.get("error"), res + assert (tmp_path / "AGENTS.md").read_text(encoding="utf-8") == "gateway approved" + finally: + A.unregister_gateway_notify(session_key) + finally: + A.reset_current_session_key(token) + + if __name__ == "__main__": pytest.main([__file__, "-v"]) diff --git a/tools/file_tools.py b/tools/file_tools.py index db3bad64b5913..35782cdf7fd13 100644 --- a/tools/file_tools.py +++ b/tools/file_tools.py @@ -702,6 +702,258 @@ def _check_sensitive_path(filepath: str, task_id: str = "default") -> str | None return None +# --------------------------------------------------------------------------- +# Protected agent-instruction files (always-ask approval gate) +# --------------------------------------------------------------------------- +# Files that steer FUTURE agent behavior are a prompt-injection persistence +# vector: an injected instruction that edits AGENTS.md / CLAUDE.md / SOUL.md / +# .cursorrules (or a project-local .hermes config tree) outlives the current +# turn and poisons every later session that loads it. Writes to these files +# therefore ALWAYS require human approval — even under --yolo / auto-approve — +# and fail closed when no human channel exists. +# +# Ported from: RooCodeInc/Roo-Code RooProtectedController (Apache-2.0). +# Companion: the terminal-tool vector is covered separately (#58631); this +# gate covers the write_file/patch vector. Symlink lesson from #41351: +# always realpath before matching. +# +# Scope decision (documented): basenames match in ANY directory, because +# project-context instruction files are loaded from cwd trees — an +# AGENTS.md anywhere the agent might later run from is a live target. +# Basenames match case-insensitively so case-variant spellings on +# case-insensitive filesystems (macOS/Windows) cannot slip past; on +# case-sensitive filesystems most loaders probe common case variants too, +# so the stricter behavior is kept uniform. +_PROTECTED_INSTRUCTION_BASENAMES = frozenset({ + "agents.md", "claude.md", "soul.md", ".cursorrules", +}) + +_real_hermes_home_cached: str | None = None +_real_hermes_home_loaded = False + + +def _get_real_hermes_home() -> str | None: + """Return the realpath of the authoritative Hermes home (cached).""" + global _real_hermes_home_cached, _real_hermes_home_loaded + if _real_hermes_home_loaded: + return _real_hermes_home_cached + _real_hermes_home_loaded = True + try: + from hermes_constants import get_hermes_home + _real_hermes_home_cached = os.path.realpath(str(get_hermes_home())) + except Exception: + try: + _real_hermes_home_cached = os.path.realpath(_expand_tilde("~/.hermes")) + except Exception: + _real_hermes_home_cached = None + return _real_hermes_home_cached + + +def _protected_instruction_config() -> tuple[bool, list[str]]: + """Read the protected-instruction-files gate config. + + Returns ``(enabled, extra_patterns)``. Defaults to enabled with no extra + patterns; config read failures keep the gate ON (fail-safe for a + security boundary). + + Config keys (config.yaml):: + + security: + protected_instruction_files: true # default + protected_instruction_extra_patterns: [] # fnmatch on basename + """ + try: + from hermes_cli.config import load_config, cfg_get + cfg = load_config() + enabled = cfg_get(cfg, "security", "protected_instruction_files", + default=True) + extra = cfg_get(cfg, "security", "protected_instruction_extra_patterns", + default=[]) + except Exception: + return True, [] + if not isinstance(enabled, bool): + enabled = True + if not isinstance(extra, list): + extra = [] + return enabled, [str(p) for p in extra if p] + + +def _protected_instruction_reason(filepath: str, task_id: str = "default", + *, enabled: bool | None = None, + extra_patterns: list[str] | None = None) -> str | None: + """Return a short label when ``filepath`` targets a protected + agent-instruction file, else ``None``. + + Matching runs on BOTH the normalized input path and its realpath so + neither a symlink pointing AT a protected file (#41351) nor a protected + name that is itself a symlink escapes the gate. ``..`` traversal is + neutralized by normpath/realpath before the basename compare. + """ + if enabled is None or extra_patterns is None: + enabled, extra_patterns = _protected_instruction_config() + if not enabled: + return None + + normalized = os.path.normpath(_expand_tilde(filepath)) + try: + resolved = os.path.realpath(str(_resolve_path_for_task(filepath, task_id))) + except (OSError, ValueError, RuntimeError): + resolved = os.path.realpath(normalized) + + # The authoritative ~/.hermes home is governed by its own guards + # (config.yaml hard-block, cross-profile guard, write_approval); this + # gate targets PROJECT-LOCAL instruction files only. Checked before the + # ``.hermes`` component rule below, which would otherwise match the + # home directory itself. + real_home = _get_real_hermes_home() + if real_home and (resolved == real_home + or resolved.startswith(real_home + os.sep)): + return None + + import fnmatch + for candidate in (normalized, resolved): + base = os.path.basename(candidate) + base_lower = base.lower() + if base_lower in _PROTECTED_INSTRUCTION_BASENAMES: + return base + for pattern in extra_patterns: + if fnmatch.fnmatch(base_lower, pattern.lower()): + return base + # Project-local .hermes config dirs (e.g. /.hermes/config.yaml) + # are loaded as project context and steer behavior the same way. + # Scope: the file's IMMEDIATE parent must be ``.hermes`` — matching + # any ancestor named .hermes would gate every write inside a + # checkout that happens to live under ~/.hermes (e.g. the + # hermes-agent repo itself at ~/.hermes/hermes-agent). + parts = candidate.replace("\\", "/").rstrip("/").split("/") + if len(parts) >= 2 and parts[-2] == ".hermes": + return candidate + return None + + +def _request_protected_instruction_approval( + reasons: list[str], task_id: str = "default") -> str | None: + """Ask the human to approve a write to protected instruction file(s). + + Returns ``None`` when approved, or a BLOCKED error string. This gate + intentionally does NOT route through ``_run_approval_gate``: that gate + honors --yolo and session/permanent allowlists, and the entire point + here is one-operation approval EVERY time, with no persistent scope + and no yolo bypass. Fail-closed when no human channel exists. + """ + targets = ", ".join(dict.fromkeys(reasons)) + description = ( + f"Write to protected agent-instruction file(s): {targets}. " + "These files steer future agent behavior; approval is always " + "required (not bypassed by auto-approve)." + ) + display = f"" + blocked = ( + f"BLOCKED: write to protected agent-instruction file(s) ({targets}) " + "{why} The user has NOT consented to this write. Do NOT retry it or " + "attempt the same edit via another path (terminal, execute_code, " + "etc.)." + ) + + try: + import tools.approval as _approval + except Exception: + return blocked.format(why="requires approval but the approval " + "subsystem is unavailable.") + + # Gateway surface: block on the button round-trip when a notify callback + # is registered for this session (Telegram/Discord/Slack). One-operation + # only — no session/permanent buttons are offered. + session_key = _approval.get_current_session_key() + notify_cb = None + try: + with _approval._lock: + notify_cb = _approval._gateway_notify_cbs.get(session_key) + except Exception: + notify_cb = None + + if notify_cb is not None: + approval_data = { + "command": display, + "pattern_key": "protected_instruction_file", + "pattern_keys": ["protected_instruction_file"], + "description": description, + "allow_permanent": False, + "allow_session": False, + } + decision = _approval._await_gateway_decision( + session_key, notify_cb, approval_data, surface="gateway", + ) + if decision.get("notify_failed"): + return blocked.format( + why="requires approval but the approval request could not " + "be delivered.") + choice = decision.get("choice") + if decision.get("resolved") and choice in {"once", "session", "always"}: + # One-operation grant regardless of the tapped scope — nothing + # is persisted for this gate. + return None + if not decision.get("resolved"): + return blocked.format( + why="approval prompt timed out without a user response. " + "Silence is not consent.") + return blocked.format(why="was denied by the user.") + + # CLI surface: per-thread approval callback (prompt_toolkit panel). + callback = None + try: + from tools.terminal_tool import _get_approval_callback + callback = _get_approval_callback() + except Exception: + callback = None + + if callback is not None: + choice = _approval.prompt_dangerous_approval( + display, description, + allow_permanent=False, + approval_callback=callback, + ) + if choice in {"once", "session", "always"}: + # One-operation grant; never persisted (see docstring). + return None + if choice == "timeout": + return blocked.format( + why="approval prompt timed out without a user response. " + "Silence is not consent.") + return blocked.format(why="was denied by the user.") + + # No human channel at all (script, cron, background thread): fail + # closed. Auto-approving here would recreate the persistence vector. + return blocked.format( + why="requires approval but no interactive user or gateway is " + "present to approve it.") + + +def _check_protected_instruction_write(paths: list[str], + task_id: str = "default") -> str | None: + """Gate a write/patch touching protected instruction files. + + Returns ``None`` when no target is protected or the human approved; + otherwise a BLOCKED error string. For multi-file V4A patches, ONE + protected file gates the ENTIRE patch: a single prompt lists every + protected target, and a deny applies nothing (including innocent + files) — partial application of an approved-in-part patch would be + more surprising than an atomic all-or-nothing outcome. + """ + enabled, extra = _protected_instruction_config() + if not enabled: + return None + reasons: list[str] = [] + for p in paths: + reason = _protected_instruction_reason( + p, task_id, enabled=enabled, extra_patterns=extra) + if reason: + reasons.append(reason) + if not reasons: + return None + return _request_protected_instruction_approval(reasons, task_id) + + def _get_container_mirror_prefix_for_task(task_id: str = "default") -> str | None: """Return the container-side Hermes mirror prefix for Docker file tools.""" try: @@ -1768,6 +2020,9 @@ def write_file_tool(path: str, content: str, task_id: str = "default", sensitive_err = _check_sensitive_path(path, task_id) if sensitive_err: return tool_error(sensitive_err) + protected_err = _check_protected_instruction_write([path], task_id) + if protected_err: + return tool_error(protected_err) if not cross_profile: cross_warning = _check_cross_profile_path(path, task_id) if cross_warning: @@ -1900,6 +2155,11 @@ def patch_tool(mode: str = "replace", path: str = None, old_string: str = None, cross_warning = _check_cross_profile_path(_p, task_id) if cross_warning: return tool_error(cross_warning) + # One approval prompt for the whole patch: a single protected file gates + # the ENTIRE patch (deny applies nothing — see the helper's docstring). + protected_err = _check_protected_instruction_write(_paths_to_check, task_id) + if protected_err: + return tool_error(protected_err) try: # Resolve paths for locking. Ordered + deduplicated so concurrent # callers lock in the same order — prevents deadlock on overlapping