From fe6cefd2d981302d795ac64ca64b66c6f3f87c3f Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 20:45:46 -0700 Subject: [PATCH] =?UTF-8?q?refactor(hermes=5Fcli):=20approvals=5Fsuggest/t?= =?UTF-8?q?est,=20approval=5Fmode/transport=20=E2=80=94=20json=20helper,?= =?UTF-8?q?=20=5Fdeny=20helper,=20packed=20layout?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- hermes_cli/approval_mode.py | 38 +++------ hermes_cli/approval_transport.py | 33 ++++---- hermes_cli/approvals_suggest.py | 130 ++++++++++++------------------- hermes_cli/approvals_test.py | 28 ++----- 4 files changed, 79 insertions(+), 150 deletions(-) diff --git a/hermes_cli/approval_mode.py b/hermes_cli/approval_mode.py index fe5d5c24fe..8e835e0b3d 100644 --- a/hermes_cli/approval_mode.py +++ b/hermes_cli/approval_mode.py @@ -37,24 +37,14 @@ def run_approval_mode_command(requested_mode: Optional[str]) -> ApprovalModeResu requested = (requested_mode or "").strip().lower() if not requested: - return ApprovalModeResult( - True, - current, - False, - f"Approval mode: {current} (persistent profile setting).", - ) + return ApprovalModeResult(True, current, False, f"Approval mode: {current} (persistent profile setting).") if requested not in VALID_APPROVAL_MODES: - return ApprovalModeResult( - False, - current, - False, - "Usage: /approvals [manual|smart|off]", - ) + return ApprovalModeResult(False, current, False, "Usage: /approvals [manual|smart|off]") - # set_config_value is the canonical managed-scope/write-safety chokepoint. - # It reports managed policy through stderr + SystemExit, and the fail-closed - # write guard raises RuntimeError on an unparseable config.yaml; capture both - # for slash-command output instead of terminating the interactive worker. + # set_config_value is the canonical managed-scope/write-safety chokepoint. It reports managed + # policy through stderr + SystemExit, and the fail-closed write guard raises RuntimeError on an + # unparseable config.yaml; capture both for slash-command output instead of terminating the + # interactive worker. from hermes_cli.config import set_config_value output = StringIO() @@ -65,24 +55,14 @@ def run_approval_mode_command(requested_mode: Optional[str]) -> ApprovalModeResu detail = output.getvalue().strip() or "Approval mode is managed and cannot be changed." return ApprovalModeResult(False, current, False, detail) except Exception as exc: - return ApprovalModeResult( - False, - current, - False, - f"Failed to save approval mode: {exc}", - ) + return ApprovalModeResult(False, current, False, f"Failed to save approval mode: {exc}") effective = _effective_mode() if effective != requested: return ApprovalModeResult( - False, - effective, - False, + False, effective, False, f"Approval mode remains {effective}; the requested value did not become effective.", ) return ApprovalModeResult( - True, - effective, - effective != current, - f"Approval mode: {effective} (persistent profile setting).", + True, effective, effective != current, f"Approval mode: {effective} (persistent profile setting).", ) diff --git a/hermes_cli/approval_transport.py b/hermes_cli/approval_transport.py index eafa6ee873..64b7a75520 100644 --- a/hermes_cli/approval_transport.py +++ b/hermes_cli/approval_transport.py @@ -25,9 +25,7 @@ _MAX_ACTIVE_TRANSPORT_WORKERS = 8 _transport_worker_slots = threading.BoundedSemaphore(_MAX_ACTIVE_TRANSPORT_WORKERS) ApprovalChoice = Literal["once", "session", "always", "deny"] -ApprovalPresentFn = Callable[ - ["ApprovalRequest"], "ApprovalDecision | Awaitable[ApprovalDecision]" -] +ApprovalPresentFn = Callable[["ApprovalRequest"], "ApprovalDecision | Awaitable[ApprovalDecision]"] @dataclass(frozen=True) @@ -81,9 +79,7 @@ class ApprovalRequest: ) canonical = json.dumps({**fields, "session_key": session_key}, sort_keys=True, separators=(",", ":")) digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest() - return cls( - **{**fields, "pattern_keys": pattern_keys, "allowed_choices": tuple(choices)}, digest=digest - ) + return cls(**{**fields, "pattern_keys": pattern_keys, "allowed_choices": tuple(choices)}, digest=digest) def respond(self, choice: ApprovalChoice | str) -> ApprovalDecision: """Build the correlated response a transport should return.""" @@ -108,6 +104,10 @@ class RegisteredApprovalTransport: profile_home: str +def _deny(failure: str) -> ApprovalTransportResult: + return ApprovalTransportResult("deny", failure) + + def invoke_approval_transport( present: ApprovalPresentFn, request: ApprovalRequest, @@ -123,10 +123,9 @@ def invoke_approval_transport( loop. A callback must return before the host timeout; late results are discarded and cannot authorize another request. """ - if not _transport_worker_slots.acquire(blocking=False): logger.warning("Approval transport worker capacity exhausted") - return ApprovalTransportResult("deny", "busy") + return _deny("busy") results: queue.Queue[tuple[str, object, float]] = queue.Queue(maxsize=1) deadline = time.monotonic() + max(float(timeout_seconds), 0.0) @@ -148,27 +147,23 @@ def invoke_approval_transport( finally: _transport_worker_slots.release() - worker = threading.Thread( - target=_run, name=f"approval-transport-{request.request_id[:8]}", daemon=True - ) + worker = threading.Thread(target=_run, name=f"approval-transport-{request.request_id[:8]}", daemon=True) try: worker.start() except BaseException: _transport_worker_slots.release() logger.warning("Could not start approval transport worker") - return ApprovalTransportResult("deny", "error") + return _deny("error") while True: if is_interrupted is not None and is_interrupted(): logger.info("Approval transport wait interrupted for %s", request.request_id) - return ApprovalTransportResult("deny", "interrupted") + return _deny("interrupted") remaining = deadline - time.monotonic() if remaining <= 0: logger.warning("Approval transport timed out for request %s", request.request_id) - return ApprovalTransportResult("deny", "timeout") + return _deny("timeout") try: - kind, value, completed_at = results.get( - timeout=min(max(float(poll_interval), 0.001), remaining) - ) + kind, value, completed_at = results.get(timeout=min(max(float(poll_interval), 0.001), remaining)) break except queue.Empty: if on_poll is not None: @@ -178,9 +173,7 @@ def invoke_approval_transport( logger.debug("Approval transport poll callback failed", exc_info=True) failure = _validate_decision(kind, value, completed_at, deadline, request) - if failure is not None: - return ApprovalTransportResult("deny", failure) - return ApprovalTransportResult(value.choice) + return _deny(failure) if failure is not None else ApprovalTransportResult(value.choice) def _validate_decision(kind, value, completed_at, deadline, request) -> str | None: diff --git a/hermes_cli/approvals_suggest.py b/hermes_cli/approvals_suggest.py index af92627eef..e0a432d9ed 100644 --- a/hermes_cli/approvals_suggest.py +++ b/hermes_cli/approvals_suggest.py @@ -2,12 +2,10 @@ Hermes has no dedicated approval-decision ledger: ``always`` answers land in ``command_allowlist`` (config.yaml) via :func:`tools.approval.save_permanent_allowlist`, while ``once``/``session`` -approvals are in-memory only. - -So this module mines *implied approvals*: a command that matches a dangerous-command class (the same -:func:`tools.approval.detect_dangerous_command` classifier that triggers the prompt) AND whose tool -result is not a block/denial marker must have been approved by the user (once, session, always, -smart-approve, or yolo) before it ran. +approvals are in-memory only. So this module mines *implied approvals*: a command that matches a +dangerous-command class (the same :func:`tools.approval.detect_dangerous_command` classifier that +triggers the prompt) AND whose tool result is not a block/denial marker must have been approved by +the user (once, session, always, smart-approve, or yolo) before it ran. """ from __future__ import annotations @@ -25,12 +23,10 @@ from typing import Iterable, Iterator, Optional # Safety exclusions # --------------------------------------------------------------------------- -# Dangerous-class descriptions matching ANY of these are never proposed, -# regardless of approval frequency. Matched case-insensitively against the -# pattern-key/description strings produced by tools.approval's -# DANGEROUS_PATTERNS / execution-flag findings. Deliberately conservative: -# a benign class accidentally excluded costs the user one manual config edit; -# a destructive class accidentally proposed costs them data. +# Dangerous-class descriptions matching ANY of these are never proposed, regardless of approval +# frequency (matched case-insensitively against tools.approval's DANGEROUS_PATTERNS / execution-flag +# descriptions). Deliberately conservative: a benign class accidentally excluded costs the user +# one manual config edit; a destructive class accidentally proposed costs them data. _UNSAFE_CLASS_PATTERNS = [ r"delete", # recursive delete, find -delete, branch force delete, ... r"\brm\b", # xargs with rm, find -exec rm @@ -71,8 +67,8 @@ _UNSAFE_CLASS_PATTERNS = [ ] _UNSAFE_CLASS_RE = re.compile("|".join(_UNSAFE_CLASS_PATTERNS), re.IGNORECASE) -# Root binaries that must never anchor a proposed command glob, even if the -# class survived the description filter. Prefix match for the mkfs family. +# Root binaries that must never anchor a proposed command glob, even if the class survived the +# description filter. Prefix match for the mkfs family. _UNSAFE_ROOT_BINARIES = { "rm", "rmdir", "unlink", "shred", "dd", "fdisk", "parted", "wipefs", "sudo", "doas", "su", "chmod", "chown", "chgrp", @@ -82,9 +78,8 @@ _UNSAFE_ROOT_BINARIES = { } _UNSAFE_ROOT_PREFIXES = ("mkfs",) -# Substrings in a role='tool' result that mean the command did NOT execute -# with user consent (blocked, denied, timed out, or still pending). Kept in -# sync with the message templates in tools/approval.py. +# Substrings in a role='tool' result that mean the command did NOT execute with user consent +# (blocked, denied, timed out, or still pending). Kept in sync with tools/approval.py templates. _BLOCK_MARKERS = ( "BLOCKED (hardline)", "BLOCKED: User denied", @@ -128,22 +123,23 @@ def default_db_path() -> Path: def _connect_readonly(db_path: Path) -> sqlite3.Connection: - uri = f"file:{db_path}?mode=ro" - return sqlite3.connect(uri, uri=True) + return sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) def _fetch_rows(cur) -> Iterator[tuple]: """Stream cursor rows in 2000-row batches.""" - while True: - rows = cur.fetchmany(2000) - if not rows: - return + while rows := cur.fetchmany(2000): yield from rows -def _iter_terminal_calls( - con: sqlite3.Connection, since_ts: float -) -> Iterator[tuple[str, str]]: +def _json_or_none(raw): + try: + return json.loads(raw) + except (TypeError, ValueError): + return None + + +def _iter_terminal_calls(con: sqlite3.Connection, since_ts: float) -> Iterator[tuple[str, str]]: """Yield ``(tool_call_id, command)`` for every terminal tool call.""" cur = con.execute( "SELECT tool_calls FROM messages " @@ -152,21 +148,13 @@ def _iter_terminal_calls( (since_ts,), ) for (raw,) in _fetch_rows(cur): - try: - calls = json.loads(raw) - except (TypeError, ValueError): - continue - if not isinstance(calls, list): - continue - for call in calls: - if not isinstance(call, dict): - continue - fn = call.get("function") or {} + calls = _json_or_none(raw) + for call in calls if isinstance(calls, list) else (): + fn = call.get("function") or {} if isinstance(call, dict) else {} if fn.get("name") != "terminal": continue - try: - args = json.loads(fn.get("arguments") or "{}") - except (TypeError, ValueError): + args = _json_or_none(fn.get("arguments") or "{}") + if args is None: continue command = args.get("command") if isinstance(command, str) and command.strip(): @@ -188,12 +176,9 @@ def _blocked_tool_call_ids(con: sqlite3.Connection, since_ts: float) -> set: } -def scan_approval_history( - db_path: Optional[Path] = None, days: int = 90 -) -> list[tuple[str, str]]: - """Return ``(command, dangerous_class_description)`` records mined from the session DB — - dangerous-classified terminal commands that actually executed (i.e. carried an implied user - approval). +def scan_approval_history(db_path: Optional[Path] = None, days: int = 90) -> list[tuple[str, str]]: + """``(command, dangerous_class_description)`` records for dangerous-classified terminal commands + that actually executed (i.e. carried an implied user approval). """ from tools.approval import detect_dangerous_command, detect_hardline_command @@ -210,15 +195,13 @@ def scan_approval_history( for tool_call_id, command in _iter_terminal_calls(con, since_ts): if tool_call_id in blocked: continue - is_hardline, _desc = detect_hardline_command(command) - if is_hardline: - # Hardline commands are unconditionally blocked at runtime; - # never mine them (defense in depth against stale DB rows). + # Hardline commands are unconditionally blocked at runtime; never mine them (defense in + # depth against stale DB rows). + if detect_hardline_command(command)[0]: continue is_dangerous, _key, description = detect_dangerous_command(command) - if not is_dangerous: - continue - records.append((command, description)) + if is_dangerous: + records.append((command, description)) finally: con.close() return records @@ -230,13 +213,9 @@ def scan_approval_history( def normalize_command(command: str) -> str: """Fold user/hermes home prefixes and collapse whitespace.""" - from tools.approval import ( - _rewrite_resolved_hermes_home, - _rewrite_resolved_user_home, - ) + from tools.approval import _rewrite_resolved_hermes_home, _rewrite_resolved_user_home - folded = _rewrite_resolved_user_home(_rewrite_resolved_hermes_home(command)) - return " ".join(folded.split()) + return " ".join(_rewrite_resolved_user_home(_rewrite_resolved_hermes_home(command)).split()) def is_unsafe_class(description: str) -> bool: @@ -257,12 +236,8 @@ def derive_glob(normalized: str) -> Optional[str]: """ from tools.approval import _has_allowlist_shell_operator - if _has_allowlist_shell_operator(normalized): - return None tokens = normalized.split() - if not tokens: - return None - if _unsafe_root_binary(tokens[0]): + if _has_allowlist_shell_operator(normalized) or not tokens or _unsafe_root_binary(tokens[0]): return None if len(tokens) == 1: return tokens[0] @@ -292,19 +267,15 @@ def build_proposals( continue normalized = normalize_command(command) glob = derive_glob(normalized) - key = (glob, "glob") if glob is not None else (description, "class") - pattern, kind = key + pattern, kind = (glob, "glob") if glob is not None else (description, "class") if pattern in existing: continue - proposal = by_pattern.get(key) - if proposal is None: - proposal = by_pattern[key] = Proposal(pattern=pattern, kind=kind) + proposal = by_pattern.setdefault((pattern, kind), Proposal(pattern=pattern, kind=kind)) proposal.count += 1 proposal.classes.add(description) proposal.add_example(normalized) - ranked = [p for p in by_pattern.values() if p.count >= max(min_count, 1)] - ranked.sort(key=lambda p: (-p.count, p.pattern)) + ranked = sorted((p for p in by_pattern.values() if p.count >= max(min_count, 1)), key=lambda p: (-p.count, p.pattern)) return ranked[: max(limit, 1)] @@ -336,12 +307,10 @@ def apply_proposals(proposals: list[Proposal], indices: list[int]) -> set: """Merge chosen proposal patterns into command_allowlist and persist.""" import tools.approval as approval_module - merged = set(approval_module.load_permanent_allowlist()) - for idx in indices: - merged.add(proposals[idx].pattern) + merged = set(approval_module.load_permanent_allowlist()) | {proposals[idx].pattern for idx in indices} approval_module.save_permanent_allowlist(merged) - # Keep the in-process allowlist consistent so a long-lived process sees - # the new entries immediately (mirrors the interactive 'always' path). + # Keep the in-process allowlist consistent so a long-lived process sees the new entries + # immediately (mirrors the interactive 'always' path). approval_module.load_permanent(merged) return merged @@ -382,13 +351,13 @@ def suggest_command(args) -> int: import tools.approval as approval_module existing = set(approval_module.load_permanent_allowlist()) - records = scan_approval_history(db_path, days=days) proposals = build_proposals( - records, + scan_approval_history(db_path, days=days), existing=existing, min_count=getattr(args, "min_count", 2), limit=getattr(args, "limit", 20), ) + as_json = getattr(args, "json", False) apply_spec = getattr(args, "apply_indices", None) if apply_spec: @@ -399,17 +368,16 @@ def suggest_command(args) -> int: return 1 merged = apply_proposals(proposals, indices) applied = [proposals[i].pattern for i in indices] - if getattr(args, "json", False): + if as_json: print(json.dumps({"applied": applied, "allowlist_size": len(merged)})) else: print("Added to command_allowlist:") for pattern in applied: print(f" + {pattern}") - print(f"\ncommand_allowlist now has {len(merged)} entries " - "(~/.hermes/config.yaml).") + print(f"\ncommand_allowlist now has {len(merged)} entries (~/.hermes/config.yaml).") return 0 - if getattr(args, "json", False): + if as_json: payload = { "db": str(db_path), "days": days, diff --git a/hermes_cli/approvals_test.py b/hermes_cli/approvals_test.py index c9bca6d57a..6400c61cab 100644 --- a/hermes_cli/approvals_test.py +++ b/hermes_cli/approvals_test.py @@ -7,7 +7,7 @@ order the runtime guard (``check_all_command_guards``) applies them: 1. container-skip gate (isolated backends bypass all guards), 2. hardline blocklist (never bypassable, fires before yolo/off), 3. sudo-stdin guard (unconditional), 4. user ``approvals.deny`` rules (fire before yolo/off), 5. yolo / ``approvals.mode: off`` bypass, 6. permanent -``command_allowlist``, 7. +``command_allowlist``, 7. dangerous-pattern detection (would prompt). """ from __future__ import annotations @@ -36,8 +36,8 @@ def evaluate_command(command: str, env_type: str = "local") -> dict: """ import tools.approval as approval - # Sync config-persisted "always" patterns so the allowlist check below - # sees what the runtime would see (load is read-only). + # Sync config-persisted "always" patterns so the allowlist check below sees what the runtime + # would see (load is read-only). try: approval.load_permanent_allowlist() except Exception: @@ -56,8 +56,7 @@ def evaluate_command(command: str, env_type: str = "local") -> dict: "normalized_variants": variants, } - # 1. Isolated container backends skip every guard (runtime parity: - # this fires BEFORE the hardline floor in check_all_command_guards). + # 1. Isolated container backends skip every guard (fires BEFORE the hardline floor at runtime). if approval._should_skip_container_guards(env_type): return result( "allow", @@ -77,10 +76,7 @@ def evaluate_command(command: str, env_type: str = "local") -> dict: # 3. Sudo stdin guard — unconditional, like the hardline floor. is_sudo_guess, sudo_desc = approval._check_sudo_stdin_guard(command) if is_sudo_guess: - return result( - "hardline-deny", rule=sudo_desc, - detail="sudo stdin guard (unconditional block)", - ) + return result("hardline-deny", rule=sudo_desc, detail="sudo stdin guard (unconditional block)") # 4. User-defined approvals.deny rules — fire before yolo/off. deny_pattern = approval._match_user_deny_rule(command) @@ -103,11 +99,7 @@ def evaluate_command(command: str, env_type: str = "local") -> dict: # 6. Permanent command_allowlist. if approval._command_matches_permanent_allowlist(command): - return result( - "allow", - detail="matches command_allowlist in config.yaml " - "(permanently approved)", - ) + return result("allow", detail="matches command_allowlist in config.yaml (permanently approved)") # 7. Dangerous-pattern detection → would prompt. is_dangerous, pattern_key, description = approval.detect_dangerous_command(command) @@ -138,18 +130,14 @@ def _render_text(verdict: dict) -> None: def approvals_test_command(args) -> int: """Handle ``hermes approvals test ``. Returns the exit code.""" words = list(getattr(args, "command_words", None) or []) - # argparse REMAINDER keeps a leading "--" separator; it is not part of - # the command being evaluated. + # argparse REMAINDER keeps a leading "--" separator; it is not part of the command. if words and words[0] == "--": words = words[1:] if not words: print("usage: hermes approvals test [--env-type TYPE] [--json] -- ") return EXIT_USAGE - command = " ".join(words) - env_type = getattr(args, "env_type", None) or "local" - - verdict = evaluate_command(command, env_type=env_type) + verdict = evaluate_command(" ".join(words), env_type=getattr(args, "env_type", None) or "local") if getattr(args, "json", False): print(json.dumps(verdict, indent=2)) else: