From aad348ce5e4ab3ac9c4e9ec14e45ce8358d1fe1f Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 21:06:19 -0700 Subject: [PATCH] refactor(approval): share hardline RECOVERY prefix, fold allowlist scanner branches, inline smart observer hooks (249+211->228+164) --- tools/approval_floors.py | 69 ++++++++++------------------ tools/approval_smart.py | 97 +++++++++++----------------------------- 2 files changed, 49 insertions(+), 117 deletions(-) diff --git a/tools/approval_floors.py b/tools/approval_floors.py index 262c81186b..a25b0295ab 100644 --- a/tools/approval_floors.py +++ b/tools/approval_floors.py @@ -12,7 +12,6 @@ import logging import re import time import uuid -from typing import Optional from tools.approval_detection import _MALFORMED_EXEC_DESCRIPTION, _PARSER_LIMIT_DESCRIPTION logger = logging.getLogger("tools.approval") @@ -32,8 +31,7 @@ def _match_user_deny_rule(command: str) -> str | None: deny_patterns = _a._get_approval_config().get("deny") or [] except Exception: return None - globs = [p.strip() for p in deny_patterns - if isinstance(p, str) and p.strip()] + globs = [p.strip() for p in deny_patterns if isinstance(p, str) and p.strip()] if not globs: return None for command_variant in _a._command_detection_variants(command): @@ -59,7 +57,7 @@ def _user_deny_block_result(pattern: str) -> dict: } -def _save_blocked_payload(command: str) -> Optional[str]: +def _save_blocked_payload(command: str) -> str | None: """Persist a parser-limit-blocked command as a runnable script. The parser-limit block fires on payload SIZE/shape, not the operation — @@ -86,9 +84,8 @@ def _save_blocked_payload(command: str) -> Optional[str]: "#!/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 ""), + f"# then run it via: bash {path}\n" + + command + ("" if command.endswith("\n") else "\n"), encoding="utf-8", errors="replace", ) return str(path) @@ -97,6 +94,13 @@ def _save_blocked_payload(command: str) -> Optional[str]: return None +_RECOVERY_PREFIX = ( + " RECOVERY: this block fires on oversized/unparseable inline " + "command payloads (heredocs, giant one-liners), not on the " + "operation itself. " +) + + def _hardline_block_result(description: str, command: str = "") -> dict: """Build the standard block result for a hardline match.""" from tools import approval as _a @@ -114,26 +118,18 @@ def _hardline_block_result(description: str, command: str = "") -> dict: if description in (_PARSER_LIMIT_DESCRIPTION, _MALFORMED_EXEC_DESCRIPTION): saved = _a._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} — " + message += _RECOVERY_PREFIX + ( + f"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, " + message += _RECOVERY_PREFIX + ( + "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, - "message": message, - } + return {"approved": False, "hardline": True, "message": message} def _sudo_stdin_block_result(description: str) -> dict: @@ -179,40 +175,27 @@ def _has_allowlist_shell_operator(command: str) -> bool: n = len(command) while i < n: ch = command[i] - if quote == "'": - if ch == "'": - quote = None - elif ch in _SHELL_CONTROL_CHARS: - has_reinterpretable = True - i += 1 - continue - if ch == "\\": + if ch == "\\" and quote != "'": nxt = command[i + 1] if i + 1 < n else "" if nxt in _SHELL_CONTROL_CHARS: has_reinterpretable = True i += 2 continue - if quote == '"': - if ch == '"': + if quote is not None: + if ch == quote: quote = None - elif ch in ("`", "$"): + elif quote == '"' and ch in ("`", "$"): return True # expansion is active inside double quotes elif ch in _SHELL_CONTROL_CHARS: has_reinterpretable = True - i += 1 - continue - if ch in ("'", '"'): + elif ch in ("'", '"'): quote = ch - i += 1 - continue - if ch == "$": + elif ch == "$": # Unquoted $ is only compound when it opens a substitution # ("$HOME" stays simple, matching the historical `\$\(` behavior). if i + 1 < n and command[i + 1] == "(": return True - i += 1 - continue - if ch in _SHELL_CONTROL_CHARS and ch not in "()": + elif ch in _SHELL_CONTROL_CHARS and ch not in "()": return True i += 1 # An unterminated quote means we can't reason about the command shape. @@ -232,14 +215,10 @@ def _command_matches_permanent_allowlist(command: str) -> bool: command = (command or "").strip() if not command or _a._has_allowlist_shell_operator(command): return False - with _a._lock: patterns = tuple(_a._permanent_approved) - for pattern in patterns: - if not isinstance(pattern, str): - continue - pattern = pattern.strip() + pattern = pattern.strip() if isinstance(pattern, str) else "" if not pattern: continue if command == pattern: diff --git a/tools/approval_smart.py b/tools/approval_smart.py index 4409d6c5c2..4c675dd142 100644 --- a/tools/approval_smart.py +++ b/tools/approval_smart.py @@ -35,10 +35,8 @@ _VERDICTS = {"APPROVE": "approve", "DENY": "deny"} def _strip_line_comment(line: str) -> str: - """Remove a trailing ``# comment`` from one shell line, quote-aware. - - Tracks single/double quote state so ``echo "hello # world"`` survives. - """ + """Remove a trailing ``# comment`` from one shell line, quote-aware + (``echo "hello # world"`` survives).""" in_single = in_double = False i = 0 while i < len(line): @@ -74,7 +72,6 @@ def _strip_shell_comments(command: str) -> str: def _get_smart_policy() -> str: """Operator rules (``approvals.smart_policy``) appended to the guardian's system prompt.""" from tools.approval import _get_approval_config - policy = _get_approval_config().get("smart_policy", "") return policy.strip() if isinstance(policy, str) else "" @@ -87,14 +84,9 @@ def _smart_approve(command: str, description: str) -> str: # Pass the timeout explicitly AND log call + duration: this synchronous # call gates EVERY flagged command, and a stalled provider once froze - # turns for tens of minutes with zero log output (#82846, #72500). + # turns for tens of minutes with zero log output. smart_timeout = _get_task_timeout("approval") - logger.debug( - "Smart approvals: assessing risk for command (timeout=%ss)", - smart_timeout, - ) - sanitized_command = _strip_shell_comments(command) - + logger.debug("Smart approvals: assessing risk for command (timeout=%ss)", smart_timeout) system_prompt = _SYSTEM_PROMPT # Operator policy goes in the SYSTEM prompt only — the trusted channel. # Never next to the block: that would dilute the trust @@ -107,17 +99,15 @@ def _smart_approve(command: str, description: str) -> str: "TRUSTED instructions, unlike the command text):\n" f"{operator_policy}" ) - user_prompt = ( f"The following command was flagged as: {description}\n\n" - f"\n{sanitized_command}\n\n\n" + f"\n{_strip_shell_comments(command)}\n\n\n" "Assess the ACTUAL risk of the shell operations in this command. " "Many flagged commands are false positives — for example, " '`python -c "print(\'hello\')"` is flagged as "script execution ' 'via -c flag" but is completely harmless.\n\n' "Respond with exactly one word: APPROVE, DENY, or ESCALATE" ) - response = call_llm( task="approval", messages=[ @@ -128,15 +118,12 @@ def _smart_approve(command: str, description: str) -> str: max_tokens=16, timeout=smart_timeout, ) - logger.debug( - "Smart approvals: LLM call completed in %.1fs", - time.monotonic() - _smart_t0, - ) + logger.debug("Smart approvals: LLM call completed in %.1fs", time.monotonic() - _smart_t0) answer = (response.choices[0].message.content or "").strip().upper() return _VERDICTS.get(answer, "escalate") except Exception as e: # WARNING, not DEBUG: a failed/blocked guardian call is a real event - # the operator needs to see (#82846 — the hang was invisible). + # the operator needs to see (the hang was invisible at DEBUG). logger.warning( "Smart approvals: LLM call failed after %.1fs (%s: %s), escalating", time.monotonic() - _smart_t0, @@ -146,15 +133,9 @@ def _smart_approve(command: str, description: str) -> str: return "escalate" -def _prepare_smart_approval_observer( - *, - command: str, - description: str, - pattern_key: str, - pattern_keys: list[str], - session_key: str, -) -> dict | None: - """Redact and emit the pre-decision smart approval observer hook. +def _smart_verdict(command: str, description: str, pattern_key: str, + pattern_keys: list[str], session_key: str) -> str: + """Run the guardian LLM with observer hooks; 'approve' | 'deny' | 'escalate'. Redaction is observer-payload preparation, not approval policy: if it fails, skip observability rather than leak raw data or block the LLM decision. @@ -162,50 +143,22 @@ def _prepare_smart_approval_observer( from tools import approval as _a try: from agent.redact import redact_sensitive_text - - hook_command = redact_sensitive_text(command, force=True) - hook_description = redact_sensitive_text(description, force=True) + payload = { + "command": redact_sensitive_text(command, force=True), + "description": redact_sensitive_text(description, force=True), + "pattern_key": pattern_key, + "pattern_keys": list(pattern_keys), + "session_key": session_key, + "surface": "smart", + } except Exception as exc: logger.debug("Smart approval hook redaction failed: %s", exc) - return None - - payload = { - "command": hook_command, - "description": hook_description, - "pattern_key": pattern_key, - "pattern_keys": list(pattern_keys), - "session_key": session_key, - "surface": "smart", - } - _a._fire_approval_hook("pre_approval_request", **payload) - return payload - - -def _observe_smart_approval_verdict(payload: dict | None, verdict: str) -> None: - """Emit a smart verdict after the auxiliary LLM decision, if safe.""" - from tools import approval as _a - if payload is None or verdict not in {"approve", "deny"}: - return - _a._fire_approval_hook( - "post_approval_response", - **payload, - choice=f"smart_{verdict}", - decided_by="aux_llm", - ) - - -def _smart_verdict(command: str, description: str, pattern_key: str, - pattern_keys: list[str], session_key: str) -> str: - """Run the guardian LLM with observer hooks; 'approve' | 'deny' | 'escalate'.""" - from tools import approval as _a - - observer_payload = _prepare_smart_approval_observer( - command=command, - description=description, - pattern_key=pattern_key, - pattern_keys=pattern_keys, - session_key=session_key, - ) + payload = None + else: + _a._fire_approval_hook("pre_approval_request", **payload) verdict = _a._smart_approve(command, description) - _observe_smart_approval_verdict(observer_payload, verdict) + if payload is not None and verdict in {"approve", "deny"}: + _a._fire_approval_hook( + "post_approval_response", **payload, choice=f"smart_{verdict}", decided_by="aux_llm", + ) return verdict