feat(terminal): auto-save parser-limit-blocked payloads as runnable scripts
Follow-up to the recovery-recipe commit on this branch, per review: instead of only TELLING the model to re-author the payload via write_file (2 turns), materialize the blocked command to ~/.hermes/cache/blocked-scripts/blocked-*.sh and point the recovery at it directly: 'saved to <path> - review it, then run terminal(command="bash <path>")' (1 turn). Safety posture is unchanged or better: - Nothing is executed here; the file is only written. - The bash <path> follow-up goes through the normal execution pipeline, including the referenced-script content guard, which inspects script files named in commands - the payload is MORE visible to policy than it was inline. - Genuine hardline blocks (destructive ops) never save anything (test-asserted). - Save failures fall back to the previous manual write_file recipe. - 7-day opportunistic cleanup of saved payloads.
This commit is contained in:
Teknium
parent
b1711c6f2e
commit
6f5d6b1f5b
|
|
@ -7,21 +7,58 @@ from tools.terminal_tool import _foreground_background_guidance
|
|||
|
||||
|
||||
class TestParserLimitRecovery:
|
||||
def test_parser_limit_block_has_recovery_recipe(self):
|
||||
r = _hardline_block_result(_PARSER_LIMIT_DESCRIPTION)
|
||||
def test_parser_limit_block_saves_payload_and_names_it(self, tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
|
||||
cmd = "python3 -c '" + "x = 1; " * 900 + "'"
|
||||
r = _hardline_block_result(_PARSER_LIMIT_DESCRIPTION, cmd)
|
||||
assert r["approved"] is False
|
||||
assert "RECOVERY" in r["message"]
|
||||
assert "blocked-scripts" in r["message"]
|
||||
import re as _re
|
||||
m = _re.search(r"saved to (\S+\.sh)", r["message"])
|
||||
assert m, r["message"]
|
||||
from pathlib import Path
|
||||
saved = Path(m.group(1))
|
||||
assert saved.exists()
|
||||
body = saved.read_text()
|
||||
assert cmd in body
|
||||
assert body.startswith("#!/bin/bash")
|
||||
assert f"bash {saved}" in r["message"]
|
||||
|
||||
def test_save_failure_falls_back_to_manual_recipe(self, monkeypatch):
|
||||
import tools.approval as ap
|
||||
monkeypatch.setattr(ap, "_save_blocked_payload", lambda c: None)
|
||||
r = _hardline_block_result(_PARSER_LIMIT_DESCRIPTION, "python3 -c 'x'")
|
||||
assert "write_file" in r["message"]
|
||||
assert "bash /path/script.sh" in r["message"]
|
||||
|
||||
def test_no_command_falls_back_to_manual_recipe(self):
|
||||
r = _hardline_block_result(_PARSER_LIMIT_DESCRIPTION)
|
||||
assert "RECOVERY" in r["message"]
|
||||
assert "write_file" in r["message"]
|
||||
|
||||
def test_malformed_exec_block_has_recovery_recipe(self):
|
||||
r = _hardline_block_result(_MALFORMED_EXEC_DESCRIPTION)
|
||||
assert "RECOVERY" in r["message"]
|
||||
|
||||
def test_real_hardline_blocks_unchanged(self):
|
||||
r = _hardline_block_result("recursive delete of root filesystem")
|
||||
def test_real_hardline_blocks_unchanged(self, tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
|
||||
r = _hardline_block_result("recursive delete of root filesystem", "rm -rf --no-preserve-root /")
|
||||
assert "RECOVERY" not in r["message"]
|
||||
assert "unconditional blocklist" in r["message"]
|
||||
# And nothing was saved for a genuine hardline block.
|
||||
assert not (tmp_path / ".hermes" / "cache" / "blocked-scripts").exists()
|
||||
|
||||
def test_old_saved_payloads_cleaned(self, tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
|
||||
import os
|
||||
d = tmp_path / ".hermes" / "cache" / "blocked-scripts"
|
||||
d.mkdir(parents=True)
|
||||
stale = d / "blocked-1-dead.sh"
|
||||
stale.write_text("old")
|
||||
os.utime(stale, (1, 1))
|
||||
_hardline_block_result(_PARSER_LIMIT_DESCRIPTION, "python3 -c 'y'")
|
||||
assert not stale.exists()
|
||||
|
||||
|
||||
class TestBackgroundGuidanceRecipes:
|
||||
|
|
|
|||
|
|
@ -585,7 +585,52 @@ def _user_deny_block_result(pattern: str) -> dict:
|
|||
}
|
||||
|
||||
|
||||
def _hardline_block_result(description: str) -> dict:
|
||||
def _save_blocked_payload(command: str) -> Optional[str]:
|
||||
"""Persist a parser-limit-blocked command as a runnable script.
|
||||
|
||||
The parser-limit block fires on payload SIZE/shape, not on the
|
||||
operation — the command itself is usually a legitimate script the
|
||||
model inlined (heredoc, giant one-liner). Materialize it to a file so
|
||||
the recovery is one turn (`bash <file>`) instead of two (re-author via
|
||||
write_file, then run). Saving is strictly safer than the hint-only
|
||||
path: the file goes through the same execution pipeline as any other
|
||||
script (including the referenced-script content guard), and nothing
|
||||
is executed here.
|
||||
|
||||
Returns the saved path, or None on any failure (the hint then falls
|
||||
back to the manual write_file recipe).
|
||||
"""
|
||||
try:
|
||||
from hermes_constants import get_hermes_home
|
||||
import time as _time
|
||||
import uuid as _uuid
|
||||
script_dir = get_hermes_home() / "cache" / "blocked-scripts"
|
||||
script_dir.mkdir(parents=True, exist_ok=True)
|
||||
# Opportunistic cleanup: blocked payloads older than 7 days.
|
||||
cutoff = _time.time() - 7 * 86400
|
||||
for old in script_dir.glob("blocked-*.sh"):
|
||||
try:
|
||||
if old.stat().st_mtime < cutoff:
|
||||
old.unlink()
|
||||
except OSError:
|
||||
pass
|
||||
path = script_dir / f"blocked-{int(_time.time())}-{_uuid.uuid4().hex[:8]}.sh"
|
||||
path.write_text(
|
||||
"#!/bin/bash\n"
|
||||
"# Auto-saved by Hermes: this command exceeded the inline command\n"
|
||||
"# parser limit and was blocked from direct execution. Review it,\n"
|
||||
"# then run it via: bash " + str(path) + "\n"
|
||||
+ command
|
||||
+ ("\n" if not command.endswith("\n") else ""),
|
||||
encoding="utf-8", errors="replace",
|
||||
)
|
||||
return str(path)
|
||||
except Exception:
|
||||
logger.debug("failed to save blocked payload", exc_info=True)
|
||||
return None
|
||||
|
||||
|
||||
def _hardline_block_result(description: str, command: str = "") -> dict:
|
||||
"""Build the standard block result for a hardline match."""
|
||||
message = (
|
||||
f"BLOCKED (hardline): {description}. "
|
||||
|
|
@ -599,15 +644,26 @@ def _hardline_block_result(description: str) -> dict:
|
|||
# (heredoc script, base64 blob, one-line python -c program) — not a
|
||||
# genuinely forbidden operation. 198 occurrences in a 250k-call
|
||||
# production window, typically followed by blind rephrase retries.
|
||||
# Tell the model the working alternative explicitly.
|
||||
# Auto-save the payload as a runnable script and point at it; fall
|
||||
# back to the manual write_file recipe when saving fails.
|
||||
if description in (_PARSER_LIMIT_DESCRIPTION, _MALFORMED_EXEC_DESCRIPTION):
|
||||
message += (
|
||||
" RECOVERY: this block fires on oversized/unparseable inline "
|
||||
"command payloads (heredocs, giant one-liners), not on the "
|
||||
"operation itself. Write the script to a file with write_file, "
|
||||
"then run it: terminal(command=\"bash /path/script.sh\") or "
|
||||
"\"python3 /path/script.py\". Do not retry inline."
|
||||
)
|
||||
saved = _save_blocked_payload(command) if command else None
|
||||
if saved:
|
||||
message += (
|
||||
" RECOVERY: this block fires on oversized/unparseable inline "
|
||||
"command payloads (heredocs, giant one-liners), not on the "
|
||||
f"operation itself. Your command was saved to {saved} — "
|
||||
f"review it, then run: terminal(command=\"bash {saved}\"). "
|
||||
"Do not retry inline."
|
||||
)
|
||||
else:
|
||||
message += (
|
||||
" RECOVERY: this block fires on oversized/unparseable inline "
|
||||
"command payloads (heredocs, giant one-liners), not on the "
|
||||
"operation itself. Write the script to a file with write_file, "
|
||||
"then run it: terminal(command=\"bash /path/script.sh\") or "
|
||||
"\"python3 /path/script.py\". Do not retry inline."
|
||||
)
|
||||
return {
|
||||
"approved": False,
|
||||
"hardline": True,
|
||||
|
|
@ -3122,7 +3178,7 @@ def check_dangerous_command(command: str, env_type: str,
|
|||
is_hardline, hardline_desc = detect_hardline_command(command)
|
||||
if is_hardline:
|
||||
logger.warning("Hardline block: %s (command: %s)", hardline_desc, command[:200])
|
||||
return _hardline_block_result(hardline_desc)
|
||||
return _hardline_block_result(hardline_desc, command)
|
||||
|
||||
# User-defined deny rules (approvals.deny in config.yaml): like the
|
||||
# hardline floor, these fire BEFORE the yolo bypass — a deny rule is the
|
||||
|
|
@ -3422,7 +3478,7 @@ def check_all_command_guards(command: str, env_type: str,
|
|||
is_hardline, hardline_desc = detect_hardline_command(command)
|
||||
if is_hardline:
|
||||
logger.warning("Hardline block: %s (command: %s)", hardline_desc, command[:200])
|
||||
return _hardline_block_result(hardline_desc)
|
||||
return _hardline_block_result(hardline_desc, command)
|
||||
|
||||
# == Sudo stdin guard ==
|
||||
# Like the hardline floor above, this is unconditional: there is never a
|
||||
|
|
|
|||
Loading…
Reference in New Issue