ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

refactor(approval): share hardline RECOVERY prefix, fold allowlist scanner branches, inline smart observer hooks (249+211->228+164)

Browse Source
This commit is contained in:
Teknium
2026-09-02 21:06:19 -07:00
parent db5fef39cd
commit aad348ce5e
2 changed files with 49 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/approval_floors.py
+24 -45
View File
@@ -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:
tools/approval_smart.py
+25 -72
View File
@@ -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 <command> 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"<command>\n{sanitized_command}\n</command>\n\n"
f"<command>\n{_strip_shell_comments(command)}\n</command>\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