mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-27 17:58:07 +00:00
feat(approvals): hermes approvals suggest — mine approval history into allowlist proposals
This commit is contained in:
teknium1
Teknium
parent
d6fdae6746
commit
db90e36202
5 changed files with 1021 additions and 0 deletions
482
hermes_cli/approvals_suggest.py
Normal file
482
hermes_cli/approvals_suggest.py
Normal file
|
|
@ -0,0 +1,482 @@
|
|||
"""``hermes approvals suggest`` — mine approval history into allowlist proposals.
|
||||
|
||||
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. What *does* persist
|
||||
is the session DB (``~/.hermes/state.db``): every assistant ``terminal`` tool
|
||||
call is stored with its arguments, and the paired ``role='tool'`` result
|
||||
records whether the command was blocked/denied ("BLOCKED: User denied …",
|
||||
"Asking the user for approval") or actually executed.
|
||||
|
||||
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. Frequently re-approved
|
||||
patterns are exactly the prompts worth turning into one-time allowlist
|
||||
policy — the port of Claude Code's ``/fewer-permission-prompts``.
|
||||
|
||||
Safety posture:
|
||||
|
||||
* **Never auto-applies.** The default run is a dry proposal; only an explicit
|
||||
``--apply N[,M...]`` merges the chosen patterns into ``command_allowlist``
|
||||
via the existing :func:`tools.approval.save_permanent_allowlist` path.
|
||||
* **Hardline commands are never proposed** — anything matched by
|
||||
:func:`tools.approval.detect_hardline_command` is dropped outright.
|
||||
* **Destructive / privilege / credential / obfuscation classes are never
|
||||
proposed**, no matter how often they were approved. ``rm -rf build/``
|
||||
approved 100 times still never yields an ``rm`` allowlist entry. Only
|
||||
benign, recoverable classes (container lifecycle, git force push, service
|
||||
restarts, hermes self-management, …) are eligible.
|
||||
* **Dangerous root binaries never become globs** (``rm *``, ``sudo *`` …).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
import sqlite3
|
||||
import time
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
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.
|
||||
_UNSAFE_CLASS_PATTERNS = [
|
||||
r"delete", # recursive delete, find -delete, branch force delete, ...
|
||||
r"\brm\b", # xargs with rm, find -exec rm
|
||||
r"destro", # git reset --hard (destroys ...), destructive
|
||||
r"wipe",
|
||||
r"format", # format filesystem
|
||||
r"\bdisk\b",
|
||||
r"block device",
|
||||
r"fork bomb",
|
||||
r"kill (?:all )?process", # force/regex/all process kills
|
||||
r"kill all",
|
||||
r"self-termination",
|
||||
r"\bsudo\b",
|
||||
r"privilege",
|
||||
r"credential",
|
||||
r"\bssh\b",
|
||||
r"shell.rc",
|
||||
r"system config",
|
||||
r"system file",
|
||||
r"\bsql\b", # SQL DROP / TRUNCATE / DELETE without WHERE
|
||||
r"\bchown\b",
|
||||
r"\bchmod\b",
|
||||
r"writable", # world/other-writable permissions
|
||||
r"overwrite",
|
||||
r"in-place edit",
|
||||
r"pipe", # pipe remote/decoded content to shell
|
||||
r"obfuscation",
|
||||
r"remote content",
|
||||
r"remote script",
|
||||
r"heredoc",
|
||||
r"encoded", # PowerShell encoded command execution
|
||||
r"command substitution",
|
||||
r"process substitution",
|
||||
r"\bdd\b",
|
||||
r"shutdown",
|
||||
r"reboot",
|
||||
r"hardline",
|
||||
]
|
||||
_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.
|
||||
_UNSAFE_ROOT_BINARIES = {
|
||||
"rm", "rmdir", "unlink", "shred", "dd", "fdisk", "parted", "wipefs",
|
||||
"sudo", "doas", "su", "chmod", "chown", "chgrp",
|
||||
"kill", "killall", "pkill",
|
||||
"halt", "shutdown", "reboot", "poweroff", "init",
|
||||
"del", "format", "truncate", "mkswap",
|
||||
}
|
||||
_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.
|
||||
_BLOCK_MARKERS = (
|
||||
"BLOCKED (hardline)",
|
||||
"BLOCKED: User denied",
|
||||
"BLOCKED: Action ",
|
||||
"BLOCKED: Command flagged as dangerous",
|
||||
"BLOCKED: approval required",
|
||||
"BLOCKED: Failed to send approval request",
|
||||
"The user has NOT consented",
|
||||
"Asking the user for approval",
|
||||
"approval_required",
|
||||
"BLOCKED by user deny rule",
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Proposal:
|
||||
"""One ranked allowlist proposal."""
|
||||
|
||||
pattern: str # command glob ("git push *") or class key
|
||||
kind: str # "glob" | "class"
|
||||
count: int = 0
|
||||
classes: set = field(default_factory=set)
|
||||
examples: list = field(default_factory=list)
|
||||
|
||||
def add_example(self, command: str) -> None:
|
||||
short = command.strip()
|
||||
if len(short) > 100:
|
||||
short = short[:97] + "..."
|
||||
if short not in self.examples and len(self.examples) < 3:
|
||||
self.examples.append(short)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Scan: session DB -> (command, class description) records
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def default_db_path() -> Path:
|
||||
from hermes_constants import get_hermes_home
|
||||
|
||||
return get_hermes_home() / "state.db"
|
||||
|
||||
|
||||
def _connect_readonly(db_path: Path) -> sqlite3.Connection:
|
||||
uri = f"file:{db_path}?mode=ro"
|
||||
return sqlite3.connect(uri, uri=True)
|
||||
|
||||
|
||||
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 "
|
||||
"WHERE role='assistant' AND tool_calls IS NOT NULL "
|
||||
"AND tool_calls LIKE '%terminal%' AND timestamp >= ?",
|
||||
(since_ts,),
|
||||
)
|
||||
while True:
|
||||
rows = cur.fetchmany(2000)
|
||||
if not rows:
|
||||
break
|
||||
for (raw,) in rows:
|
||||
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 {}
|
||||
if fn.get("name") != "terminal":
|
||||
continue
|
||||
try:
|
||||
args = json.loads(fn.get("arguments") or "{}")
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
command = args.get("command")
|
||||
if isinstance(command, str) and command.strip():
|
||||
yield (call.get("id") or "", command)
|
||||
|
||||
|
||||
def _blocked_tool_call_ids(con: sqlite3.Connection, since_ts: float) -> set:
|
||||
"""Collect tool_call_ids whose result shows the command never ran freely."""
|
||||
blocked: set = set()
|
||||
cur = con.execute(
|
||||
"SELECT tool_call_id, content FROM messages "
|
||||
"WHERE role='tool' AND tool_call_id IS NOT NULL AND timestamp >= ? "
|
||||
"AND (content LIKE '%BLOCKED%' OR content LIKE '%approval%')",
|
||||
(since_ts,),
|
||||
)
|
||||
while True:
|
||||
rows = cur.fetchmany(2000)
|
||||
if not rows:
|
||||
break
|
||||
for tool_call_id, content in rows:
|
||||
if not content:
|
||||
continue
|
||||
if any(marker in content for marker in _BLOCK_MARKERS):
|
||||
blocked.add(tool_call_id)
|
||||
return blocked
|
||||
|
||||
|
||||
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).
|
||||
"""
|
||||
from tools.approval import detect_dangerous_command, detect_hardline_command
|
||||
|
||||
path = Path(db_path) if db_path else default_db_path()
|
||||
if not path.exists():
|
||||
return []
|
||||
|
||||
since_ts = 0.0 if days <= 0 else time.time() - days * 86400
|
||||
|
||||
records: list[tuple[str, str]] = []
|
||||
con = _connect_readonly(path)
|
||||
try:
|
||||
blocked = _blocked_tool_call_ids(con, since_ts)
|
||||
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).
|
||||
continue
|
||||
is_dangerous, _key, description = detect_dangerous_command(command)
|
||||
if not is_dangerous:
|
||||
continue
|
||||
records.append((command, description))
|
||||
finally:
|
||||
con.close()
|
||||
return records
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Normalize -> aggregate -> rank -> exclude
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def normalize_command(command: str) -> str:
|
||||
"""Fold user/hermes home prefixes and collapse whitespace.
|
||||
|
||||
Reuses tools.approval's home-folding machinery so proposals are portable
|
||||
across machines/users (``/home/alice/x`` -> ``~/x``).
|
||||
"""
|
||||
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())
|
||||
|
||||
|
||||
def is_unsafe_class(description: str) -> bool:
|
||||
"""True when a dangerous-class description must never be proposed."""
|
||||
return bool(_UNSAFE_CLASS_RE.search(description or ""))
|
||||
|
||||
|
||||
def _unsafe_root_binary(token: str) -> bool:
|
||||
tok = token.lower().rsplit("/", 1)[-1]
|
||||
if tok in _UNSAFE_ROOT_BINARIES:
|
||||
return True
|
||||
return any(tok.startswith(p) for p in _UNSAFE_ROOT_PREFIXES)
|
||||
|
||||
|
||||
def derive_glob(normalized: str) -> Optional[str]:
|
||||
"""Derive a narrow command glob (``git push *``) from a simple command.
|
||||
|
||||
Returns None for compound commands (shell operators — the runtime
|
||||
allowlist matcher refuses those anyway) and for commands anchored on an
|
||||
unsafe root binary.
|
||||
"""
|
||||
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]):
|
||||
return None
|
||||
if len(tokens) == 1:
|
||||
return tokens[0]
|
||||
second = tokens[1]
|
||||
if second.startswith("-") or any(ch in second for ch in "*?[$"):
|
||||
return f"{tokens[0]} *"
|
||||
return f"{tokens[0]} {second} *"
|
||||
|
||||
|
||||
def build_proposals(
|
||||
records: Iterable[tuple[str, str]],
|
||||
existing: Optional[set] = None,
|
||||
min_count: int = 2,
|
||||
limit: int = 20,
|
||||
) -> list[Proposal]:
|
||||
"""Aggregate scan records into a ranked, safety-filtered proposal list.
|
||||
|
||||
Grain: a command glob (``git push *``) for simple commands; the
|
||||
dangerous-class description itself (the same key an interactive
|
||||
``[a]lways`` answer persists) for compound commands where no safe glob
|
||||
can be derived.
|
||||
"""
|
||||
existing = existing or set()
|
||||
by_pattern: dict[tuple[str, str], Proposal] = {}
|
||||
|
||||
for command, description in records:
|
||||
if is_unsafe_class(description):
|
||||
continue
|
||||
normalized = normalize_command(command)
|
||||
glob = derive_glob(normalized)
|
||||
if glob is not None:
|
||||
key = (glob, "glob")
|
||||
else:
|
||||
key = (description, "class")
|
||||
pattern, kind = key
|
||||
if pattern in existing:
|
||||
continue
|
||||
proposal = by_pattern.get(key)
|
||||
if proposal is None:
|
||||
proposal = by_pattern[key] = 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))
|
||||
return ranked[: max(limit, 1)]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Apply / render
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def parse_apply_indices(spec: str, total: int) -> list[int]:
|
||||
"""Parse ``"1,3"`` into validated zero-based indices."""
|
||||
indices: list[int] = []
|
||||
for part in (spec or "").split(","):
|
||||
part = part.strip()
|
||||
if not part:
|
||||
continue
|
||||
try:
|
||||
n = int(part)
|
||||
except ValueError:
|
||||
raise ValueError(f"invalid selection {part!r} — expected numbers like 1,3")
|
||||
if n < 1 or n > total:
|
||||
raise ValueError(f"selection {n} out of range (1..{total})")
|
||||
if (n - 1) not in indices:
|
||||
indices.append(n - 1)
|
||||
if not indices:
|
||||
raise ValueError("no valid selections in --apply")
|
||||
return indices
|
||||
|
||||
|
||||
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)
|
||||
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).
|
||||
approval_module.load_permanent(merged)
|
||||
return merged
|
||||
|
||||
|
||||
def _render_text(proposals: list[Proposal], days: int) -> None:
|
||||
window = "all history" if days <= 0 else f"last {days} days"
|
||||
if not proposals:
|
||||
print(
|
||||
f"No allowlist candidates found in approval history ({window}).\n"
|
||||
"Either nothing dangerous was approved often enough "
|
||||
"(see --min-count/--days), or the approved classes are "
|
||||
"excluded for safety."
|
||||
)
|
||||
return
|
||||
print(f"Proposed command_allowlist additions (from approval history, {window}):\n")
|
||||
for i, p in enumerate(proposals, 1):
|
||||
kind = " (class key)" if p.kind == "class" else ""
|
||||
print(f" {i}. {p.pattern} — approved {p.count}x{kind}")
|
||||
for cls in sorted(p.classes):
|
||||
print(f" class: {cls}")
|
||||
for ex in p.examples:
|
||||
print(f" e.g. {ex}")
|
||||
print(
|
||||
"\nNothing has been changed. Apply selected entries with:\n"
|
||||
" hermes approvals suggest --apply 1,3\n"
|
||||
"Entries are merged into command_allowlist in ~/.hermes/config.yaml."
|
||||
)
|
||||
|
||||
|
||||
def suggest_command(args) -> int:
|
||||
"""Entry point for ``hermes approvals suggest``."""
|
||||
db_path = Path(args.db) if getattr(args, "db", None) else default_db_path()
|
||||
days = getattr(args, "days", 90)
|
||||
if not db_path.exists():
|
||||
print(f"Session database not found: {db_path}")
|
||||
return 1
|
||||
|
||||
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,
|
||||
existing=existing,
|
||||
min_count=getattr(args, "min_count", 2),
|
||||
limit=getattr(args, "limit", 20),
|
||||
)
|
||||
|
||||
apply_spec = getattr(args, "apply_indices", None)
|
||||
if apply_spec:
|
||||
try:
|
||||
indices = parse_apply_indices(apply_spec, len(proposals))
|
||||
except ValueError as exc:
|
||||
print(f"--apply error: {exc}")
|
||||
return 1
|
||||
merged = apply_proposals(proposals, indices)
|
||||
applied = [proposals[i].pattern for i in indices]
|
||||
if getattr(args, "json", False):
|
||||
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).")
|
||||
return 0
|
||||
|
||||
if getattr(args, "json", False):
|
||||
payload = {
|
||||
"db": str(db_path),
|
||||
"days": days,
|
||||
"proposals": [
|
||||
{
|
||||
"n": i,
|
||||
"pattern": p.pattern,
|
||||
"kind": p.kind,
|
||||
"count": p.count,
|
||||
"classes": sorted(p.classes),
|
||||
"examples": p.examples,
|
||||
}
|
||||
for i, p in enumerate(proposals, 1)
|
||||
],
|
||||
}
|
||||
print(json.dumps(payload, indent=2))
|
||||
return 0
|
||||
|
||||
_render_text(proposals, days)
|
||||
return 0
|
||||
|
||||
|
||||
def approvals_command(args) -> int:
|
||||
"""Dispatch ``hermes approvals <subcommand>``."""
|
||||
sub = getattr(args, "approvals_command", None)
|
||||
if sub == "suggest":
|
||||
return suggest_command(args)
|
||||
print(
|
||||
"usage: hermes approvals <subcommand>\n"
|
||||
"\n"
|
||||
"subcommands:\n"
|
||||
" suggest Mine past approval decisions into a proposed\n"
|
||||
" command_allowlist (dry by default; --apply N,M to merge)\n"
|
||||
"\n"
|
||||
"Run `hermes approvals suggest -h` for details."
|
||||
)
|
||||
return 1
|
||||
|
|
@ -445,6 +445,7 @@ from hermes_cli.subcommands.webhook import build_webhook_parser
|
|||
from hermes_cli.subcommands.hooks import build_hooks_parser
|
||||
from hermes_cli.subcommands.doctor import build_doctor_parser
|
||||
from hermes_cli.subcommands.security import build_security_parser
|
||||
from hermes_cli.subcommands.approvals import build_approvals_parser
|
||||
from hermes_cli.subcommands.dump import build_dump_parser
|
||||
from hermes_cli.subcommands.debug import build_debug_parser
|
||||
from hermes_cli.subcommands.backup import build_backup_parser
|
||||
|
|
@ -4558,6 +4559,16 @@ def cmd_security(args):
|
|||
sys.exit(2)
|
||||
|
||||
|
||||
def cmd_approvals(args):
|
||||
"""Dispatch `hermes approvals <subcmd>`."""
|
||||
from hermes_cli.approvals_suggest import approvals_command
|
||||
|
||||
status = approvals_command(args)
|
||||
if status:
|
||||
sys.exit(status)
|
||||
return status
|
||||
|
||||
|
||||
def cmd_dump(args):
|
||||
"""Dump setup summary for support/debugging."""
|
||||
from hermes_cli.dump import run_dump
|
||||
|
|
@ -15657,6 +15668,11 @@ def main():
|
|||
# =========================================================================
|
||||
build_security_parser(subparsers, cmd_security=cmd_security)
|
||||
|
||||
# =========================================================================
|
||||
# approvals command (parser built in hermes_cli/subcommands/approvals.py)
|
||||
# =========================================================================
|
||||
build_approvals_parser(subparsers, cmd_approvals=cmd_approvals)
|
||||
|
||||
# =========================================================================
|
||||
# dump command (parser built in hermes_cli/subcommands/dump.py)
|
||||
# =========================================================================
|
||||
|
|
|
|||
77
hermes_cli/subcommands/approvals.py
Normal file
77
hermes_cli/subcommands/approvals.py
Normal file
|
|
@ -0,0 +1,77 @@
|
|||
"""``hermes approvals`` subcommand parser.
|
||||
|
||||
Follows the cron/security pattern: parser construction lives here, the
|
||||
handler is injected by ``main.py`` so this module never imports ``main``
|
||||
(cycle avoidance).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Callable
|
||||
|
||||
|
||||
def build_approvals_parser(subparsers, *, cmd_approvals: Callable) -> None:
|
||||
"""Attach the ``approvals`` subcommand to ``subparsers``."""
|
||||
approvals_parser = subparsers.add_parser(
|
||||
"approvals",
|
||||
help="Approval-prompt tools (mine history into allowlist proposals)",
|
||||
description=(
|
||||
"Tools for the dangerous-command approval system. "
|
||||
"`hermes approvals suggest` mines past approval decisions from "
|
||||
"the session database and proposes command_allowlist entries so "
|
||||
"repeatedly-approved commands stop prompting."
|
||||
),
|
||||
)
|
||||
approvals_subparsers = approvals_parser.add_subparsers(
|
||||
dest="approvals_command",
|
||||
metavar="<subcommand>",
|
||||
)
|
||||
|
||||
suggest_parser = approvals_subparsers.add_parser(
|
||||
"suggest",
|
||||
help="Propose command_allowlist entries from past approvals",
|
||||
description=(
|
||||
"Scan the session database for dangerous-classified commands "
|
||||
"that ran with user approval, rank the recurring patterns, and "
|
||||
"print a numbered allowlist proposal. Nothing is written unless "
|
||||
"--apply is given. Destructive classes (recursive delete, sudo, "
|
||||
"disk writes, credential edits, ...) are never proposed."
|
||||
),
|
||||
)
|
||||
suggest_parser.add_argument(
|
||||
"--apply",
|
||||
dest="apply_indices",
|
||||
metavar="N[,M...]",
|
||||
help="Merge the numbered proposals (from a prior run) into "
|
||||
"command_allowlist in config.yaml",
|
||||
)
|
||||
suggest_parser.add_argument(
|
||||
"--json",
|
||||
action="store_true",
|
||||
help="Emit machine-readable JSON instead of human-readable text",
|
||||
)
|
||||
suggest_parser.add_argument(
|
||||
"--days",
|
||||
type=int,
|
||||
default=90,
|
||||
help="How far back to scan session history (default: 90; 0 = all)",
|
||||
)
|
||||
suggest_parser.add_argument(
|
||||
"--min-count",
|
||||
dest="min_count",
|
||||
type=int,
|
||||
default=2,
|
||||
help="Minimum approval count for a pattern to be proposed (default: 2)",
|
||||
)
|
||||
suggest_parser.add_argument(
|
||||
"--limit",
|
||||
type=int,
|
||||
default=20,
|
||||
help="Maximum number of proposals to show (default: 20)",
|
||||
)
|
||||
suggest_parser.add_argument(
|
||||
"--db",
|
||||
help="Path to an alternate session database (default: ~/.hermes/state.db)",
|
||||
)
|
||||
suggest_parser.set_defaults(func=cmd_approvals)
|
||||
approvals_parser.set_defaults(func=cmd_approvals)
|
||||
409
tests/hermes_cli/test_approvals_suggest.py
Normal file
409
tests/hermes_cli/test_approvals_suggest.py
Normal file
|
|
@ -0,0 +1,409 @@
|
|||
"""Tests for ``hermes approvals suggest`` (hermes_cli/approvals_suggest.py).
|
||||
|
||||
Approval history in Hermes is implied, not ledgered: the session DB
|
||||
(state.db) stores every assistant ``terminal`` tool call plus its paired
|
||||
``role='tool'`` result. A dangerous-classified command whose result is not a
|
||||
BLOCKED/denied/pending marker ran with user consent. These tests build
|
||||
synthetic state.db fixtures and verify scanning, ranking, safety exclusions,
|
||||
and the --apply merge path.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sqlite3
|
||||
import time
|
||||
from argparse import Namespace
|
||||
|
||||
import pytest
|
||||
|
||||
import tools.approval as approval_module
|
||||
from hermes_cli.approvals_suggest import (
|
||||
Proposal,
|
||||
apply_proposals,
|
||||
build_proposals,
|
||||
derive_glob,
|
||||
is_unsafe_class,
|
||||
normalize_command,
|
||||
parse_apply_indices,
|
||||
scan_approval_history,
|
||||
suggest_command,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixture helpers: synthetic session DB
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _make_db(path):
|
||||
con = sqlite3.connect(path)
|
||||
con.executescript(
|
||||
"""
|
||||
CREATE TABLE messages (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
session_id TEXT NOT NULL,
|
||||
role TEXT NOT NULL,
|
||||
content TEXT,
|
||||
tool_call_id TEXT,
|
||||
tool_calls TEXT,
|
||||
timestamp REAL NOT NULL
|
||||
);
|
||||
"""
|
||||
)
|
||||
con.commit()
|
||||
return con
|
||||
|
||||
|
||||
_ID_COUNTER = [0]
|
||||
|
||||
|
||||
def _add_terminal_call(con, command, result="ok: done", ts=None):
|
||||
"""Insert an assistant terminal tool call + its paired tool result."""
|
||||
_ID_COUNTER[0] += 1
|
||||
call_id = f"call_{_ID_COUNTER[0]}"
|
||||
ts = ts if ts is not None else time.time()
|
||||
tool_calls = json.dumps(
|
||||
[
|
||||
{
|
||||
"id": call_id,
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "terminal",
|
||||
"arguments": json.dumps({"command": command}),
|
||||
},
|
||||
}
|
||||
]
|
||||
)
|
||||
con.execute(
|
||||
"INSERT INTO messages (session_id, role, content, tool_calls, timestamp) "
|
||||
"VALUES ('s1', 'assistant', '', ?, ?)",
|
||||
(tool_calls, ts),
|
||||
)
|
||||
con.execute(
|
||||
"INSERT INTO messages (session_id, role, content, tool_call_id, timestamp) "
|
||||
"VALUES ('s1', 'tool', ?, ?, ?)",
|
||||
(result, call_id, ts + 1),
|
||||
)
|
||||
con.commit()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def db_path(tmp_path):
|
||||
path = tmp_path / "state.db"
|
||||
con = _make_db(path)
|
||||
yield path, con
|
||||
con.close()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def isolated_allowlist(monkeypatch):
|
||||
"""Fake config-backed allowlist store so no real config.yaml is touched."""
|
||||
store = {"patterns": set(), "saves": 0}
|
||||
|
||||
def fake_load():
|
||||
return set(store["patterns"])
|
||||
|
||||
def fake_save(patterns):
|
||||
store["patterns"] = set(patterns)
|
||||
store["saves"] += 1
|
||||
|
||||
monkeypatch.setattr(approval_module, "load_permanent_allowlist", fake_load)
|
||||
monkeypatch.setattr(approval_module, "save_permanent_allowlist", fake_save)
|
||||
# apply_proposals also syncs into the in-process set; keep it isolated.
|
||||
saved = approval_module._permanent_approved.copy()
|
||||
approval_module._permanent_approved.clear()
|
||||
yield store
|
||||
approval_module._permanent_approved.clear()
|
||||
approval_module._permanent_approved.update(saved)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Scan
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestScan:
|
||||
def test_scan_finds_approved_dangerous_commands(self, db_path):
|
||||
path, con = db_path
|
||||
for _ in range(3):
|
||||
_add_terminal_call(con, "git push --force origin main")
|
||||
# A benign command never enters approval history.
|
||||
_add_terminal_call(con, "ls -la")
|
||||
records = scan_approval_history(path, days=0)
|
||||
commands = [c for c, _ in records]
|
||||
assert len(commands) == 3
|
||||
assert all("git push" in c for c in commands)
|
||||
|
||||
def test_blocked_and_denied_results_are_not_approvals(self, db_path):
|
||||
path, con = db_path
|
||||
_add_terminal_call(
|
||||
con,
|
||||
"git push --force origin main",
|
||||
result="BLOCKED: User denied this potentially dangerous action",
|
||||
)
|
||||
_add_terminal_call(
|
||||
con,
|
||||
"docker restart web",
|
||||
result=(
|
||||
"⚠️ This action is potentially dangerous. "
|
||||
"Asking the user for approval."
|
||||
),
|
||||
)
|
||||
assert scan_approval_history(path, days=0) == []
|
||||
|
||||
def test_days_window_filters_old_history(self, db_path):
|
||||
path, con = db_path
|
||||
old_ts = time.time() - 200 * 86400
|
||||
_add_terminal_call(con, "git push --force origin main", ts=old_ts)
|
||||
_add_terminal_call(con, "git push --force origin main")
|
||||
assert len(scan_approval_history(path, days=90)) == 1
|
||||
assert len(scan_approval_history(path, days=0)) == 2
|
||||
|
||||
def test_missing_db_returns_empty(self, tmp_path):
|
||||
assert scan_approval_history(tmp_path / "nope.db", days=0) == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Normalize / glob derivation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestNormalizeAndGlob:
|
||||
def test_normalize_folds_home_prefix(self):
|
||||
import os
|
||||
|
||||
home = os.path.expanduser("~")
|
||||
normalized = normalize_command(f"git checkout -- {home}/project/file.txt")
|
||||
assert "~/project/file.txt" in normalized
|
||||
assert home not in normalized
|
||||
|
||||
def test_derive_glob_uses_first_two_tokens(self):
|
||||
assert derive_glob("git push --force origin main") == "git push *"
|
||||
assert derive_glob("docker restart web") == "docker restart *"
|
||||
|
||||
def test_derive_glob_flag_second_token_falls_back_to_root(self):
|
||||
assert derive_glob("hermes --yolo update") == "hermes --yolo".split()[0] + " *"
|
||||
|
||||
def test_derive_glob_rejects_compound_commands(self):
|
||||
assert derive_glob("git push --force && rm -rf /tmp/x") is None
|
||||
assert derive_glob("echo hi; docker restart web") is None
|
||||
|
||||
def test_derive_glob_never_anchors_unsafe_binaries(self):
|
||||
assert derive_glob("rm -rf ./build") is None
|
||||
assert derive_glob("sudo apt install foo") is None
|
||||
assert derive_glob("/bin/rm -rf ./build") is None
|
||||
assert derive_glob("mkfs.ext4 /dev/sdb1") is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Ranking + safety exclusion
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestRankingAndSafety:
|
||||
def test_ranking_orders_by_frequency(self, db_path):
|
||||
path, con = db_path
|
||||
for _ in range(5):
|
||||
_add_terminal_call(con, "docker restart web")
|
||||
for _ in range(2):
|
||||
_add_terminal_call(con, "git push --force origin main")
|
||||
records = scan_approval_history(path, days=0)
|
||||
proposals = build_proposals(records, min_count=2)
|
||||
assert [p.pattern for p in proposals] == ["docker restart *", "git push *"]
|
||||
assert proposals[0].count == 5
|
||||
assert proposals[1].count == 2
|
||||
|
||||
def test_min_count_threshold(self, db_path):
|
||||
path, con = db_path
|
||||
_add_terminal_call(con, "git push --force origin main")
|
||||
records = scan_approval_history(path, days=0)
|
||||
assert build_proposals(records, min_count=2) == []
|
||||
assert len(build_proposals(records, min_count=1)) == 1
|
||||
|
||||
def test_rm_rf_never_proposed_even_after_100_approvals(self, db_path):
|
||||
path, con = db_path
|
||||
for _ in range(100):
|
||||
_add_terminal_call(con, "rm -rf ./build")
|
||||
records = scan_approval_history(path, days=0)
|
||||
# The commands ARE mined (they ran with approval) ...
|
||||
assert len(records) == 100
|
||||
# ... but the destructive class is unconditionally excluded.
|
||||
proposals = build_proposals(records, min_count=1)
|
||||
assert proposals == []
|
||||
|
||||
def test_unsafe_classes_are_excluded(self):
|
||||
for desc in (
|
||||
"recursive delete",
|
||||
"git reset --hard (destroys uncommitted changes)",
|
||||
"sudo with privilege flag (stdin/askpass/shell/list)",
|
||||
"pipe remote content to shell",
|
||||
"overwrite system config",
|
||||
"SQL DROP",
|
||||
"in-place edit of sensitive credential/SSH/shell-rc path",
|
||||
"format filesystem",
|
||||
"kill all processes",
|
||||
"write to block device",
|
||||
):
|
||||
assert is_unsafe_class(desc), desc
|
||||
|
||||
def test_benign_classes_are_not_excluded(self):
|
||||
for desc in (
|
||||
"git force push (rewrites remote history)",
|
||||
"docker restart/stop/kill (container lifecycle)",
|
||||
"hermes update (restarts gateway, kills running agents)",
|
||||
"stop/restart system service",
|
||||
):
|
||||
assert not is_unsafe_class(desc), desc
|
||||
|
||||
def test_existing_allowlist_entries_are_skipped(self, db_path):
|
||||
path, con = db_path
|
||||
for _ in range(3):
|
||||
_add_terminal_call(con, "git push --force origin main")
|
||||
records = scan_approval_history(path, days=0)
|
||||
proposals = build_proposals(records, existing={"git push *"}, min_count=1)
|
||||
assert proposals == []
|
||||
|
||||
def test_hardline_commands_never_mined(self, db_path):
|
||||
path, con = db_path
|
||||
# Even if a hardline command somehow shows an executed result in the
|
||||
# DB, the miner refuses it (defense in depth).
|
||||
for _ in range(5):
|
||||
_add_terminal_call(con, "rm -rf /")
|
||||
assert scan_approval_history(path, days=0) == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# --apply / dry-run
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _args(db, **kw):
|
||||
base = dict(
|
||||
db=str(db), days=0, min_count=1, limit=20,
|
||||
apply_indices=None, json=False,
|
||||
)
|
||||
base.update(kw)
|
||||
return Namespace(**base)
|
||||
|
||||
|
||||
class TestApply:
|
||||
def test_parse_apply_indices(self):
|
||||
assert parse_apply_indices("1,3", 5) == [0, 2]
|
||||
assert parse_apply_indices(" 2 ", 2) == [1]
|
||||
with pytest.raises(ValueError):
|
||||
parse_apply_indices("0", 3)
|
||||
with pytest.raises(ValueError):
|
||||
parse_apply_indices("4", 3)
|
||||
with pytest.raises(ValueError):
|
||||
parse_apply_indices("a,b", 3)
|
||||
with pytest.raises(ValueError):
|
||||
parse_apply_indices("", 3)
|
||||
|
||||
def test_apply_merges_and_persists(self, isolated_allowlist):
|
||||
isolated_allowlist["patterns"] = {"podman *"}
|
||||
proposals = [
|
||||
Proposal(pattern="git push *", kind="glob", count=5),
|
||||
Proposal(pattern="docker restart *", kind="glob", count=3),
|
||||
]
|
||||
merged = apply_proposals(proposals, [0])
|
||||
assert merged == {"podman *", "git push *"}
|
||||
assert isolated_allowlist["patterns"] == {"podman *", "git push *"}
|
||||
assert isolated_allowlist["saves"] == 1
|
||||
# In-process allowlist synced too (long-lived process consistency).
|
||||
assert "git push *" in approval_module._permanent_approved
|
||||
|
||||
def test_suggest_apply_end_to_end(self, db_path, isolated_allowlist, capsys):
|
||||
path, con = db_path
|
||||
for _ in range(4):
|
||||
_add_terminal_call(con, "git push --force origin main")
|
||||
for _ in range(3):
|
||||
_add_terminal_call(con, "docker restart web")
|
||||
rc = suggest_command(_args(path, apply_indices="1,2"))
|
||||
assert rc == 0
|
||||
assert isolated_allowlist["patterns"] == {"git push *", "docker restart *"}
|
||||
out = capsys.readouterr().out
|
||||
assert "git push *" in out and "docker restart *" in out
|
||||
|
||||
def test_dry_default_writes_nothing(self, db_path, isolated_allowlist, capsys):
|
||||
path, con = db_path
|
||||
for _ in range(4):
|
||||
_add_terminal_call(con, "git push --force origin main")
|
||||
rc = suggest_command(_args(path))
|
||||
assert rc == 0
|
||||
assert isolated_allowlist["saves"] == 0
|
||||
assert isolated_allowlist["patterns"] == set()
|
||||
out = capsys.readouterr().out
|
||||
assert "git push *" in out
|
||||
assert "Nothing has been changed" in out
|
||||
|
||||
def test_apply_out_of_range_errors_without_writing(
|
||||
self, db_path, isolated_allowlist, capsys
|
||||
):
|
||||
path, con = db_path
|
||||
for _ in range(4):
|
||||
_add_terminal_call(con, "git push --force origin main")
|
||||
rc = suggest_command(_args(path, apply_indices="7"))
|
||||
assert rc == 1
|
||||
assert isolated_allowlist["saves"] == 0
|
||||
|
||||
|
||||
class TestJsonOutput:
|
||||
def test_json_proposal_output(self, db_path, isolated_allowlist, capsys):
|
||||
path, con = db_path
|
||||
for _ in range(4):
|
||||
_add_terminal_call(con, "git push --force origin main")
|
||||
rc = suggest_command(_args(path, json=True))
|
||||
assert rc == 0
|
||||
payload = json.loads(capsys.readouterr().out)
|
||||
assert payload["proposals"][0]["pattern"] == "git push *"
|
||||
assert payload["proposals"][0]["count"] == 4
|
||||
assert payload["proposals"][0]["n"] == 1
|
||||
assert isolated_allowlist["saves"] == 0
|
||||
|
||||
def test_json_apply_output(self, db_path, isolated_allowlist, capsys):
|
||||
path, con = db_path
|
||||
for _ in range(4):
|
||||
_add_terminal_call(con, "git push --force origin main")
|
||||
rc = suggest_command(_args(path, apply_indices="1", json=True))
|
||||
assert rc == 0
|
||||
payload = json.loads(capsys.readouterr().out)
|
||||
assert payload["applied"] == ["git push *"]
|
||||
assert isolated_allowlist["patterns"] == {"git push *"}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Parser wiring
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestParserWiring:
|
||||
def test_build_approvals_parser_wires_suggest(self):
|
||||
import argparse
|
||||
|
||||
from hermes_cli.subcommands.approvals import build_approvals_parser
|
||||
|
||||
sentinel = object()
|
||||
parser = argparse.ArgumentParser()
|
||||
subparsers = parser.add_subparsers(dest="command")
|
||||
build_approvals_parser(subparsers, cmd_approvals=lambda a: sentinel)
|
||||
|
||||
args = parser.parse_args(
|
||||
["approvals", "suggest", "--apply", "1,3", "--json",
|
||||
"--days", "30", "--min-count", "5", "--limit", "10"]
|
||||
)
|
||||
assert args.approvals_command == "suggest"
|
||||
assert args.apply_indices == "1,3"
|
||||
assert args.json is True
|
||||
assert args.days == 30
|
||||
assert args.min_count == 5
|
||||
assert args.limit == 10
|
||||
assert args.func(args) is sentinel
|
||||
|
||||
def test_suggest_defaults_are_dry_and_bounded(self):
|
||||
import argparse
|
||||
|
||||
from hermes_cli.subcommands.approvals import build_approvals_parser
|
||||
|
||||
parser = argparse.ArgumentParser()
|
||||
subparsers = parser.add_subparsers(dest="command")
|
||||
build_approvals_parser(subparsers, cmd_approvals=lambda a: None)
|
||||
args = parser.parse_args(["approvals", "suggest"])
|
||||
assert args.apply_indices is None
|
||||
assert args.json is False
|
||||
assert args.days == 90
|
||||
assert args.min_count == 2
|
||||
|
|
@ -233,6 +233,43 @@ These patterns are loaded at startup and silently approved in all future session
|
|||
Use `hermes config edit` to review or remove patterns from your permanent allowlist.
|
||||
:::
|
||||
|
||||
### Mining Approval History (`hermes approvals suggest`)
|
||||
|
||||
Instead of answering the same prompt session after session, you can mine your
|
||||
past approval decisions into allowlist proposals:
|
||||
|
||||
```bash
|
||||
hermes approvals suggest # dry run — prints a numbered proposal
|
||||
hermes approvals suggest --apply 1,3 # merge picks into command_allowlist
|
||||
hermes approvals suggest --json # machine-readable output
|
||||
```
|
||||
|
||||
The command scans the session database (`~/.hermes/state.db`) for
|
||||
dangerous-classified commands that actually executed — i.e. commands you
|
||||
approved — aggregates them into patterns (`git push *`, or the dangerous-class
|
||||
key for compound commands), and ranks them by approval frequency:
|
||||
|
||||
```
|
||||
Proposed command_allowlist additions (from approval history, last 90 days):
|
||||
|
||||
1. git push * — approved 14x
|
||||
2. docker restart/stop/kill (container lifecycle) — approved 9x (class key)
|
||||
```
|
||||
|
||||
Safety rules:
|
||||
|
||||
- **Nothing is ever applied automatically** — the default run is read-only;
|
||||
only an explicit `--apply N[,M...]` writes to `config.yaml`.
|
||||
- **Destructive classes are never proposed**, no matter how often they were
|
||||
approved: recursive deletes, `sudo`, disk/device writes, credential and
|
||||
system-config edits, pipe-to-shell, SQL DROP/TRUNCATE, process kills, and
|
||||
every hardline class are excluded outright. `rm -rf build/` approved 100
|
||||
times still never yields an `rm` entry.
|
||||
- Proposals already covered by your existing `command_allowlist` are skipped.
|
||||
|
||||
Useful flags: `--days N` (history window, default 90), `--min-count N`
|
||||
(minimum approvals to qualify, default 2), `--limit N`, and `--db PATH`.
|
||||
|
||||
## File Write Safety {#file-write-safety}
|
||||
|
||||
Before `write_file` or `patch` touches disk, Hermes checks the target path against a denylist and an optional sandbox. Blocked writes return an error to the agent immediately — **there is no approval prompt** and no way to override from the chat UI. The model may still claim the edit succeeded; when `display.file_mutation_verifier` is on (default), trust the [file-mutation verifier footer](./configuration.md#file-mutation-verifier) over the assistant's closing summary.
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue