feat(security): protected agent-instruction files always require write approval
write_file/patch targeting AGENTS.md, CLAUDE.md, SOUL.md, .cursorrules, or a project-local .hermes config dir now ALWAYS prompt the human for approval — even under --yolo/auto-approve — and fail closed when no human channel exists. These files steer future agent behavior, so an injected write to them is a prompt-injection persistence vector. Design: - New _check_protected_instruction_write() in tools/file_tools.py, a sibling of _check_sensitive_path that returns approval-required rather than a hard error. It realpaths before matching (symlink lesson from #41351), matches basenames case-insensitively in ANY directory, rejects './x/../AGENTS.md' traversal via normpath, and gates files whose immediate parent dir is `.hermes` (project-local config) while exempting the authoritative ~/.hermes home (governed by its own guards). - Approval is ONE-OPERATION only: no session/permanent persistence, no yolo bypass — intentionally does not route through _run_approval_gate. Gateway sessions get the button round-trip with allow_permanent and allow_session both False; CLI uses the per-thread approval callback; no channel at all = BLOCKED (fail closed). - Multi-file V4A patches: ONE protected file gates the ENTIRE patch (a single prompt lists all protected targets; deny applies nothing). - Config: security.protected_instruction_files (default true) and security.protected_instruction_extra_patterns (fnmatch on basename). Config read failure keeps the gate ON. Tests: 22 new cases in tests/tools/test_file_write_safety.py covering the adversarial checklist — deny/approve/yolo-bypass attempt, symlink at a protected target, case variants, relative traversal, arbitrary-directory basenames, project-local .hermes, checkout-nested-under-~/.hermes non-gating, patch replace + V4A multi-file atomicity, gateway round-trip, fail-closed with no human, config off/extra patterns. Ported from: RooCodeInc/Roo-Code RooProtectedController (Apache-2.0); companion: #58631 (terminal vector), symlink lesson from #41351.
This commit is contained in:
Teknium
parent
c8369e37f4
commit
fe66596df3
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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"])
|
||||
|
|
|
|||
|
|
@ -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. <repo>/.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"<write to {targets}>"
|
||||
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
|
||||
|
|
|
|||
Loading…
Reference in New Issue