From 5d78bd817d6c3c7ffb22019d7e9df35c8052cde1 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 11:23:23 -0700 Subject: [PATCH] =?UTF-8?q?refactor(plugins):=20disk-cleanup=20and=20secur?= =?UTF-8?q?ity-guidance=20=E2=80=94=20table-driven=20scans/patterns,=20saf?= =?UTF-8?q?ety=20lists=20unchanged?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- plugins/disk-cleanup/__init__.py | 307 ++++------ plugins/disk-cleanup/disk_cleanup.py | 525 ++++++------------ plugins/security-guidance/__init__.py | 244 +++----- plugins/security-guidance/patterns.py | 369 ++++-------- .../plugins/test_security_guidance_plugin.py | 4 +- 5 files changed, 481 insertions(+), 968 deletions(-) diff --git a/plugins/disk-cleanup/__init__.py b/plugins/disk-cleanup/__init__.py index 71d44b1c89..1384a8df57 100644 --- a/plugins/disk-cleanup/__init__.py +++ b/plugins/disk-cleanup/__init__.py @@ -2,20 +2,14 @@ Wires three behaviours: -1. ``post_tool_call`` hook — inspects ``write_file`` and ``terminal`` - tool results for newly-created paths matching test/temp patterns - under ``HERMES_HOME`` and tracks them silently. Zero agent - compliance required. - -2. ``on_session_end`` hook — when any test files were auto-tracked - during the just-finished turn, runs :func:`disk_cleanup.quick` and - logs a single line to ``$HERMES_HOME/disk-cleanup/cleanup.log``. - -3. ``/disk-cleanup`` slash command — manual ``status``, ``dry-run``, - ``quick``, ``deep``, ``track``, ``forget``. - -Replaces PR #12212's skill-plus-script design: the agent no longer -needs to remember to run commands. +1. ``post_tool_call`` hook — inspects ``write_file`` / ``patch`` / ``terminal`` + tool calls for newly-created paths matching test/temp patterns under + ``HERMES_HOME`` and tracks them silently. Zero agent compliance required. +2. ``on_session_end`` hook — when any test files were auto-tracked during the + just-finished turn, runs :func:`disk_cleanup.quick` and logs one line to + ``$HERMES_HOME/disk-cleanup/cleanup.log``. +3. ``/disk-cleanup`` slash command — manual ``status``, ``dry-run``, ``quick``, + ``deep``, ``track``, ``forget``. """ from __future__ import annotations @@ -25,105 +19,55 @@ import re import shlex import threading from pathlib import Path -from typing import Any, Dict, Optional, Set +from typing import Any, Callable, Dict, List, Optional, Set from . import disk_cleanup as dg logger = logging.getLogger(__name__) -# Per-task set of "test files newly tracked this turn". Keyed by task_id -# (or session_id as fallback) so on_session_end can decide whether to run -# cleanup. Guarded by a lock — post_tool_call can fire concurrently on -# parallel tool calls. +# Per-task set of test files newly tracked this turn, keyed by task_id (or +# session_id as fallback) so on_session_end can decide whether to run cleanup. +# Locked: post_tool_call can fire concurrently on parallel tool calls. _recent_test_tracks: Dict[str, Set[str]] = {} _lock = threading.Lock() - -# Tool-call result shapes we can parse -_WRITE_FILE_PATH_KEY = "path" _TERMINAL_PATH_REGEX = re.compile(r"(?:^|\s)(/[^\s'\"`]+|\~/[^\s'\"`]+)") -# --------------------------------------------------------------------------- -# Helpers -# --------------------------------------------------------------------------- +# --- Path extraction from tool calls ---------------------------------------- -def _tracker_key(task_id: str, session_id: str) -> str: - return task_id or session_id or "default" - - -def _record_track(task_id: str, session_id: str, path: Path, category: str) -> None: - """Record that we tracked *path* as *category* during this turn.""" - if category != "test": - return - key = _tracker_key(task_id, session_id) - with _lock: - _recent_test_tracks.setdefault(key, set()).add(str(path)) - - -def _drain(task_id: str, session_id: str) -> Set[str]: - """Pop the set of test paths tracked during this turn.""" - key = _tracker_key(task_id, session_id) - with _lock: - return _recent_test_tracks.pop(key, set()) - - -def _attempt_track(path_str: str, task_id: str, session_id: str) -> None: - """Best-effort auto-track. Never raises.""" - try: - p = Path(path_str).expanduser() - except Exception: - return - if not p.exists(): - return - category = dg.guess_category(p) - if category is None: - return - newly = dg.track(str(p), category, silent=True) - if newly: - _record_track(task_id, session_id, p, category) - - -def _extract_paths_from_write_file(args: Dict[str, Any]) -> Set[str]: - path = args.get(_WRITE_FILE_PATH_KEY) - return {path} if isinstance(path, str) and path else set() - - -def _extract_paths_from_patch(args: Dict[str, Any]) -> Set[str]: - # The patch tool creates new files via the `mode="patch"` path too, but - # most of its use is editing existing files — we only care about new - # ephemeral creations, so treat patch conservatively and only pick up - # the single-file `path` arg. Track-then-cleanup is idempotent, so - # re-tracking an already-tracked file is a no-op (dedup in track()). +def _extract_path_arg(args: Dict[str, Any], result: str) -> Set[str]: + """write_file and patch: only the single-file ``path`` arg. patch mostly edits + existing files; re-tracking is a no-op (track() dedups), so this is safe.""" path = args.get("path") return {path} if isinstance(path, str) and path else set() def _extract_paths_from_terminal(args: Dict[str, Any], result: str) -> Set[str]: - """Best-effort: pull candidate filesystem paths from a terminal command - and its output, then let ``guess_category`` / ``is_safe_path`` filter. - """ + """Best-effort: pull candidate filesystem paths from a terminal command and + its output; ``guess_category`` / ``is_safe_path`` filter them afterwards.""" paths: Set[str] = set() cmd = args.get("command") or "" if isinstance(cmd, str) and cmd: - # Tokenise the command — catches `touch /tmp/hermes-x/test_foo.py` - try: - for tok in shlex.split(cmd, posix=True): - if tok.startswith(("/", "~")): - paths.add(tok) + try: # tokenise — catches `touch /tmp/hermes-x/test_foo.py` + paths.update(tok for tok in shlex.split(cmd, posix=True) if tok.startswith(("/", "~"))) except ValueError: pass # Only scan the result text if it's a reasonable size (avoid 50KB dumps). if isinstance(result, str) and len(result) < 4096: - for match in _TERMINAL_PATH_REGEX.findall(result): - paths.add(match) + paths.update(_TERMINAL_PATH_REGEX.findall(result)) return paths -# --------------------------------------------------------------------------- -# Hooks -# --------------------------------------------------------------------------- +_PATH_EXTRACTORS: Dict[str, Callable[[Dict[str, Any], str], Set[str]]] = { + "write_file": _extract_path_arg, + "patch": _extract_path_arg, + "terminal": _extract_paths_from_terminal, +} + + +# --- Hooks ------------------------------------------------------------------ def _on_post_tool_call( tool_name: str = "", @@ -134,22 +78,21 @@ def _on_post_tool_call( tool_call_id: str = "", **_: Any, ) -> None: - """Auto-track ephemeral files created by recent tool calls.""" - if not isinstance(args, dict): + """Auto-track ephemeral files created by recent tool calls. Best-effort, never raises.""" + extractor = _PATH_EXTRACTORS.get(tool_name) + if not isinstance(args, dict) or extractor is None: return - - candidates: Set[str] = set() - if tool_name == "write_file": - candidates = _extract_paths_from_write_file(args) - elif tool_name == "patch": - candidates = _extract_paths_from_patch(args) - elif tool_name == "terminal": - candidates = _extract_paths_from_terminal(args, result if isinstance(result, str) else "") - else: - return - - for path_str in candidates: - _attempt_track(path_str, task_id, session_id) + for path_str in extractor(args, result if isinstance(result, str) else ""): + try: + p = Path(path_str).expanduser() + except Exception: + continue + if not p.exists(): + continue + category = dg.guess_category(p) + if category is not None and dg.track(str(p), category, silent=True) and category == "test": + with _lock: + _recent_test_tracks.setdefault(task_id or session_id or "default", set()).add(str(p)) def _on_session_end( @@ -159,20 +102,12 @@ def _on_session_end( **_: Any, ) -> None: """Run quick cleanup if any test files were tracked during this turn.""" - # Drain both task-level and session-level buckets. In practice only one - # is populated per turn; the other is empty. - drained_session = _drain("", session_id) - # Also drain any task-scoped buckets that happen to exist. This is a - # cheap sweep: if an agent spawned subagents (each with their own - # task_id) they'll have recorded into separate buckets; we want to - # cleanup them all at session end. + # Drain the session bucket plus every task-scoped bucket: subagents record + # into their own task_id buckets and should be cleaned up at session end too. with _lock: - task_buckets = list(_recent_test_tracks.keys()) - for key in task_buckets: - if key and key != session_id: - _recent_test_tracks.pop(key, None) - - if not drained_session and not task_buckets: + had_tracks = bool(_recent_test_tracks.pop(session_id or "default", None) or _recent_test_tracks) + _recent_test_tracks.clear() + if not had_tracks: return try: @@ -188,9 +123,7 @@ def _on_session_end( ) -# --------------------------------------------------------------------------- -# Slash command -# --------------------------------------------------------------------------- +# --- Slash command ---------------------------------------------------------- _HELP_TEXT = """\ /disk-cleanup — ephemeral-file cleanup @@ -220,91 +153,73 @@ def _fmt_summary(summary: Dict[str, Any]) -> str: return base +def _item_block(header: str, items: List[Dict], indent: str) -> List[str]: + """``header`` formatted with ``{n}`` / ``{size}``, followed by one line per item.""" + size = dg.fmt_size(sum(i["size"] for i in items)) + return [header.format(n=len(items), size=size)] + [f"{indent}[{item['category']}] {item['path']}" for item in items] + + +def _cmd_dry_run(argv: List[str]) -> str: + auto, prompt = dg.dry_run() + lines = ["Dry-run preview (nothing deleted):"] + lines += _item_block(" Auto-delete : {n} files ({size})", auto, " ") + lines += _item_block(" Needs prompt: {n} files ({size})", prompt, " ") + lines.append(f"\n Total potential: {dg.fmt_size(sum(i['size'] for i in auto + prompt))}") + return "\n".join(lines) + + +def _cmd_deep(argv: List[str]) -> str: + # In-session deep can't prompt interactively — show what quick cleaned plus + # the items that WOULD need confirmation. + quick_summary = dg.quick() + _auto, prompt_items = dg.dry_run() + lines = [_fmt_summary(quick_summary)] + if prompt_items: + lines += _item_block("\n{n} item(s) need confirmation ({size}):", prompt_items, " ") + lines.append("\nRun `/disk-cleanup forget ` to skip, or delete manually via terminal.") + return "\n".join(lines) + + +def _cmd_track(argv: List[str]) -> str: + if len(argv) < 3: + return "Usage: /disk-cleanup track " + path_arg, category = argv[1], argv[2] + if category not in dg.ALLOWED_CATEGORIES: + return f"Unknown category '{category}'. Allowed: {sorted(dg.ALLOWED_CATEGORIES)}" + if dg.track(path_arg, category, silent=True): + return f"Tracked {path_arg} as '{category}'." + return f"Not tracked (already present, missing, or outside HERMES_HOME): {path_arg}" + + +def _cmd_forget(argv: List[str]) -> str: + if len(argv) < 2: + return "Usage: /disk-cleanup forget " + n = dg.forget(argv[1]) + return ( + f"Removed {n} tracking entr{'y' if n == 1 else 'ies'} for {argv[1]}." + if n else f"Not found in tracking: {argv[1]}" + ) + + +_SUBCOMMANDS: Dict[str, Callable[[List[str]], str]] = { + "status": lambda argv: dg.format_status(dg.status()), + "dry-run": _cmd_dry_run, + "quick": lambda argv: _fmt_summary(dg.quick()), + "deep": _cmd_deep, + "track": _cmd_track, + "forget": _cmd_forget, +} + + def _handle_slash(raw_args: str) -> Optional[str]: argv = raw_args.strip().split() if not argv or argv[0] in {"help", "-h", "--help"}: return _HELP_TEXT + handler = _SUBCOMMANDS.get(argv[0]) + if handler is None: + return f"Unknown subcommand: {argv[0]}\n\n{_HELP_TEXT}" + return handler(argv) - sub = argv[0] - - if sub == "status": - return dg.format_status(dg.status()) - - if sub == "dry-run": - auto, prompt = dg.dry_run() - auto_size = sum(i["size"] for i in auto) - prompt_size = sum(i["size"] for i in prompt) - lines = [ - "Dry-run preview (nothing deleted):", - f" Auto-delete : {len(auto)} files ({dg.fmt_size(auto_size)})", - ] - for item in auto: - lines.append(f" [{item['category']}] {item['path']}") - lines.append( - f" Needs prompt: {len(prompt)} files ({dg.fmt_size(prompt_size)})" - ) - for item in prompt: - lines.append(f" [{item['category']}] {item['path']}") - lines.append( - f"\n Total potential: {dg.fmt_size(auto_size + prompt_size)}" - ) - return "\n".join(lines) - - if sub == "quick": - return _fmt_summary(dg.quick()) - - if sub == "deep": - # In-session deep can't prompt the user interactively — show what - # quick cleaned plus the items that WOULD need confirmation. - quick_summary = dg.quick() - _auto, prompt_items = dg.dry_run() - lines = [_fmt_summary(quick_summary)] - if prompt_items: - size = sum(i["size"] for i in prompt_items) - lines.append( - f"\n{len(prompt_items)} item(s) need confirmation " - f"({dg.fmt_size(size)}):" - ) - for item in prompt_items: - lines.append(f" [{item['category']}] {item['path']}") - lines.append( - "\nRun `/disk-cleanup forget ` to skip, or delete " - "manually via terminal." - ) - return "\n".join(lines) - - if sub == "track": - if len(argv) < 3: - return "Usage: /disk-cleanup track " - path_arg = argv[1] - category = argv[2] - if category not in dg.ALLOWED_CATEGORIES: - return ( - f"Unknown category '{category}'. " - f"Allowed: {sorted(dg.ALLOWED_CATEGORIES)}" - ) - if dg.track(path_arg, category, silent=True): - return f"Tracked {path_arg} as '{category}'." - return ( - f"Not tracked (already present, missing, or outside HERMES_HOME): " - f"{path_arg}" - ) - - if sub == "forget": - if len(argv) < 2: - return "Usage: /disk-cleanup forget " - n = dg.forget(argv[1]) - return ( - f"Removed {n} tracking entr{'y' if n == 1 else 'ies'} for {argv[1]}." - if n else f"Not found in tracking: {argv[1]}" - ) - - return f"Unknown subcommand: {sub}\n\n{_HELP_TEXT}" - - -# --------------------------------------------------------------------------- -# Plugin registration -# --------------------------------------------------------------------------- def register(ctx) -> None: ctx.register_hook("post_tool_call", _on_post_tool_call) diff --git a/plugins/disk-cleanup/disk_cleanup.py b/plugins/disk-cleanup/disk_cleanup.py index 8432c5a7ba..66b6126e81 100755 --- a/plugins/disk-cleanup/disk_cleanup.py +++ b/plugins/disk-cleanup/disk_cleanup.py @@ -1,22 +1,15 @@ """disk_cleanup — ephemeral file cleanup for Hermes Agent. -Library module wrapping the deterministic cleanup rules written by -@LVT382009 in PR #12212. The plugin ``__init__.py`` wires these -functions into ``post_tool_call`` and ``on_session_end`` hooks so -tracking and cleanup happen automatically — the agent never needs to -call a tool or remember a skill. +Library module behind the disk-cleanup plugin; ``__init__.py`` wires these +functions into ``post_tool_call`` / ``on_session_end`` hooks so tracking and +cleanup happen without the agent calling a tool or remembering a skill. -Rules: - - test files → delete immediately at task end (age >= 0) - - temp files → delete after 7 days - - cron-output → delete after 14 days - - empty dirs → always delete (under HERMES_HOME) - - research → keep 10 newest, prompt for older (deep only) - - chrome-profile→ prompt after 14 days (deep only) - - >500 MB files → prompt always (deep only) +Rules: test files delete at task end (age >= 0); temp after 7 days; cron-output +after 14 days; empty dirs under HERMES_HOME always. Deep-only prompts: research +(keep 10 newest, > 30 days), chrome-profile > 14 days, any file > 500 MB. -Scope: strictly HERMES_HOME and /tmp/hermes-* -Never touches: ~/.hermes/logs/ or any system directory. +Scope: strictly HERMES_HOME and /tmp/hermes-*. Never touches ~/.hermes/logs/ or +any system directory. """ from __future__ import annotations @@ -26,7 +19,7 @@ import logging import shutil from datetime import datetime, timezone from pathlib import Path -from typing import Any, Dict, List, Optional, Tuple +from typing import Any, Callable, Dict, Iterator, List, Optional, Tuple try: from hermes_constants import get_hermes_home @@ -40,75 +33,51 @@ except Exception: # pragma: no cover — plugin may load before constants resol logger = logging.getLogger(__name__) - -# --------------------------------------------------------------------------- -# Paths -# --------------------------------------------------------------------------- - -def get_state_dir() -> Path: - """State dir — separate from ``$HERMES_HOME/logs/``.""" - return get_hermes_home() / "disk-cleanup" +_LARGE_FILE_BYTES = 500 * 1024 * 1024 -def get_tracked_file() -> Path: - return get_state_dir() / "tracked.json" +# --- Paths / safety --------------------------------------------------------- +def _state_file(name: str) -> Path: + """``$HERMES_HOME/disk-cleanup/`` — state and audit log deliberately + live outside ``$HERMES_HOME/logs/``.""" + return get_hermes_home() / "disk-cleanup" / name -def get_log_file() -> Path: - """Audit log — intentionally NOT under ``$HERMES_HOME/logs/``.""" - return get_state_dir() / "cleanup.log" - - -# --------------------------------------------------------------------------- -# Path safety -# --------------------------------------------------------------------------- def is_safe_path(path: Path) -> bool: """Accept only paths under HERMES_HOME or ``/tmp/hermes-*``. Rejects Windows mounts (``/mnt/c`` etc.) and any system directory. """ - hermes_home = get_hermes_home() try: - path.resolve().relative_to(hermes_home) + path.resolve().relative_to(get_hermes_home()) return True except (ValueError, OSError): pass - # Allow /tmp/hermes-* explicitly parts = path.parts - if len(parts) >= 3 and parts[1] == "tmp" and parts[2].startswith("hermes-"): - return True - return False + return len(parts) >= 3 and parts[1] == "tmp" and parts[2].startswith("hermes-") -# --------------------------------------------------------------------------- -# Audit log -# --------------------------------------------------------------------------- - def _log(message: str) -> None: + """Append to the audit log; never let it break the agent loop.""" try: - log_file = get_log_file() + log_file = _state_file("cleanup.log") log_file.parent.mkdir(parents=True, exist_ok=True) ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S") with open(log_file, "a", encoding="utf-8") as f: f.write(f"[{ts}] {message}\n") except OSError: - # Never let the audit log break the agent loop. pass -# --------------------------------------------------------------------------- -# tracked.json — atomic read/write, backup scoped to tracked.json only -# --------------------------------------------------------------------------- +# --- tracked.json — atomic read/write, backup scoped to tracked.json only ---- def load_tracked() -> List[Dict[str, Any]]: """Load tracked.json. Restores from ``.bak`` on corruption.""" - tf = get_tracked_file() + tf = _state_file("tracked.json") tf.parent.mkdir(parents=True, exist_ok=True) - if not tf.exists(): return [] - try: return json.loads(tf.read_text(encoding="utf-8")) except (json.JSONDecodeError, ValueError): @@ -126,7 +95,7 @@ def load_tracked() -> List[Dict[str, Any]]: def save_tracked(tracked: List[Dict[str, Any]]) -> None: """Atomic write: ``.tmp`` → backup old → rename.""" - tf = get_tracked_file() + tf = _state_file("tracked.json") tf.parent.mkdir(parents=True, exist_ok=True) tmp = tf.with_suffix(".json.tmp") tmp.write_text(json.dumps(tracked, indent=2), encoding="utf-8") @@ -135,21 +104,19 @@ def save_tracked(tracked: List[Dict[str, Any]]) -> None: tmp.replace(tf) -# --------------------------------------------------------------------------- -# Categories -# --------------------------------------------------------------------------- +# --- Categories / protected trees ------------------------------------------- ALLOWED_CATEGORIES = { "temp", "test", "research", "download", "chrome-profile", "cron-output", "other", } +# Top-level HERMES_HOME dirs whose empty subdirs are never swept. The last row +# is user-authored project trees (patches/, projects/, ...) — never sweep inside. _EMPTY_DIR_PROTECTED_TOP_LEVEL = frozenset({ "logs", "memories", "sessions", "cron", "cronjobs", "cache", "skills", "plugins", "disk-cleanup", "optional-skills", "hermes-agent", "backups", "profiles", ".worktrees", - # User-authored project trees — never sweep empty directories - # inside these (#75403). "patches", "projects", "skins", "themes", "contributors", }) @@ -158,37 +125,39 @@ _EMPTY_DIR_SWEEP_PRUNE_DIRS = frozenset({ "site-packages", "__pycache__", }) +# Top-level entries under HERMES_HOME that guess_category() never auto-tracks: +# state dir, logs, memory, sessions, config/secrets, and user-authored project +# trees (a file named test_*/tmp_* inside patches/ or projects/ is not disposable). +_NEVER_TRACK_TOP_LEVEL = frozenset({ + "disk-cleanup", "logs", "memories", "sessions", "config.yaml", + "skills", "plugins", ".env", "USER.md", "MEMORY.md", "SOUL.md", + "auth.json", "hermes-agent", + "patches", "projects", "skins", "themes", "contributors", + "profiles", "backups", "optional-skills", +}) -# Paths under $HERMES_HOME that must NEVER be deleted by quick(), -# regardless of what the stored category says. This is a defense-in-depth -# guard against stale tracked.json entries from before #34840. +# Defense-in-depth for quick(): exact cron control-plane paths never deleted, +# regardless of stored category (guards stale tracked.json entries). _PROTECTED_CRON_PATHS: set[str] = set() def _is_protected_cron_path(p: Path) -> bool: - """Return True if *p* is a cron control-plane file/directory that must - never be deleted. + """True if *p* is cron control-plane state that must never be deleted. - This matches, by EXACT path only, the ``cron/`` directory itself, known - control-plane files (``jobs.json``, ``.tick.lock``), and the ``output/`` - root directory. It does NOT (and must not be "simplified" to) blanket-match - everything under ``cron/output/`` — those run artifacts are disposable and - are cleaned by retention policy; only the ``output/`` root itself is - protected, because deleting it wholesale erases every job's retained run - history at once. + Matches by EXACT path only: the ``cron/`` dir itself, ``jobs.json``, + ``.tick.lock``, and the ``output/`` root. It must NOT be widened to + everything under ``cron/output/`` — run artifacts there are disposable and + cleaned by retention; only the ``output/`` root is protected because + deleting it wholesale erases every job's retained run history. """ - # Lazily build the set once per process so HERMES_HOME is resolved - # exactly once. - if not _PROTECTED_CRON_PATHS: + if not _PROTECTED_CRON_PATHS: # built lazily so HERMES_HOME resolves once hermes_home = get_hermes_home() for parent in ("cron", "cronjobs"): base = hermes_home / parent - _PROTECTED_CRON_PATHS.add(str(base)) - _PROTECTED_CRON_PATHS.add(str(base / "output")) - _PROTECTED_CRON_PATHS.add(str(base / "jobs.json")) - _PROTECTED_CRON_PATHS.add(str(base / ".tick.lock")) - resolved = str(p.resolve()) - return resolved in _PROTECTED_CRON_PATHS + _PROTECTED_CRON_PATHS.update( + str(x) for x in (base, base / "output", base / "jobs.json", base / ".tick.lock") + ) + return str(p.resolve()) in _PROTECTED_CRON_PATHS def fmt_size(n: float) -> str: @@ -199,9 +168,7 @@ def fmt_size(n: float) -> str: return f"{n:.1f} PB" -# --------------------------------------------------------------------------- -# Track / forget -# --------------------------------------------------------------------------- +# --- Track / forget --------------------------------------------------------- def track(path_str: str, category: str, silent: bool = False) -> bool: """Register a file for tracking. Returns True if newly tracked.""" @@ -210,19 +177,15 @@ def track(path_str: str, category: str, silent: bool = False) -> bool: category = "other" path = Path(path_str).resolve() - if not path.exists(): _log(f"SKIP: {path} (does not exist)") return False - if not is_safe_path(path): _log(f"REJECT: {path} (outside HERMES_HOME)") return False size = path.stat().st_size if path.is_file() else 0 tracked = load_tracked() - - # Deduplicate if any(item["path"] == str(path) for item in tracked): return False @@ -252,288 +215,185 @@ def forget(path_str: str) -> int: return removed -# --------------------------------------------------------------------------- -# Dry run -# --------------------------------------------------------------------------- - -def dry_run() -> Tuple[List[Dict], List[Dict]]: - """Return (auto_delete_list, needs_prompt_list) without touching files.""" - tracked = load_tracked() - now = datetime.now(timezone.utc) - - auto: List[Dict] = [] - prompt: List[Dict] = [] +# --- Rules shared by dry_run / quick / deep --------------------------------- +def _live_items(tracked: List[Dict], now: datetime, *, log_stale: bool = False) -> Iterator[Tuple[Dict, Path, int]]: + """Yield ``(item, path, age_days)`` for entries whose path still exists.""" for item in tracked: p = Path(item["path"]) if not p.exists(): + if log_stale: + _log(f"STALE: {p} (removed from tracking)") continue - age = (now - datetime.fromisoformat(item["timestamp"])).days + yield item, p, (now - datetime.fromisoformat(item["timestamp"])).days + + +def _is_auto_delete(cat: str, age: int) -> bool: + return cat == "test" or (cat == "temp" and age > 7) or (cat == "cron-output" and age > 14) + + +def _prompt_group(item: Dict, age: int) -> Optional[str]: + """Deep-only bucket: ``research`` / ``chrome`` / ``large`` or None.""" + cat = item["category"] + if cat == "research" and age > 30: + return "research" + if cat == "chrome-profile" and age > 14: + return "chrome" + if item["size"] > _LARGE_FILE_BYTES: + return "large" + return None + + +def _delete_item(item: Dict) -> Optional[str]: + """Delete a tracked file/dir and audit-log it. Returns an error string on OSError, else None.""" + p = Path(item["path"]) + try: + if p.is_file(): + p.unlink() + elif p.is_dir(): + shutil.rmtree(p) + except OSError as e: + _log(f"ERROR deleting {p}: {e}") + return f"{p}: {e}" + _log(f"DELETED: {p} ({item['category']}, {fmt_size(item['size'])})") + return None + + +# Stored categories that are re-validated against guess_category() before use. +# Old tracked.json entries can carry "cron-output" for control-plane files +# (cron/jobs.json) or "test" for files now under protected project trees; +# guess_category() was tightened later but existing entries were never re-checked. +_STALE_SKIP_NOTE = {"cron-output": "", "test": " — under protected tree"} + + +# --- Dry run / quick / deep ------------------------------------------------- + +def dry_run() -> Tuple[List[Dict], List[Dict]]: + """Return (auto_delete_list, needs_prompt_list) without touching files.""" + auto: List[Dict] = [] + prompt: List[Dict] = [] + for item, p, age in _live_items(load_tracked(), datetime.now(timezone.utc)): cat = item["category"] - size = item["size"] - - # Re-validate stale "cron-output" entries (fixes #37721). - if cat == "cron-output": - re_cat = guess_category(p) - if re_cat != "cron-output": - # Stale entry — would be skipped by quick(); omit from - # dry-run output too. - continue - - if cat == "test": + # Stale cron-output entries are skipped by quick(); omit them here too. + if cat == "cron-output" and guess_category(p) != "cron-output": + continue + if _is_auto_delete(cat, age): auto.append(item) - elif cat == "temp" and age > 7: - auto.append(item) - elif cat == "cron-output" and age > 14: - auto.append(item) - elif cat == "research" and age > 30: + elif _prompt_group(item, age): prompt.append(item) - elif cat == "chrome-profile" and age > 14: - prompt.append(item) - elif size > 500 * 1024 * 1024: - prompt.append(item) - return auto, prompt -# --------------------------------------------------------------------------- -# Quick cleanup -# --------------------------------------------------------------------------- - def quick() -> Dict[str, Any]: """Safe deterministic cleanup — no prompts. - Returns: ``{"deleted": N, "empty_dirs": N, "freed": bytes, - "errors": [str, ...]}``. + Returns: ``{"deleted": N, "empty_dirs": N, "freed": bytes, "errors": [str, ...]}``. """ - tracked = load_tracked() - now = datetime.now(timezone.utc) - deleted = 0 - freed = 0 + deleted = freed = 0 new_tracked: List[Dict] = [] errors: List[str] = [] - for item in tracked: - p = Path(item["path"]) + for item, p, age in _live_items(load_tracked(), datetime.now(timezone.utc), log_stale=True): cat = item["category"] - - if not p.exists(): - _log(f"STALE: {p} (removed from tracking)") + if cat in _STALE_SKIP_NOTE and (re_cat := guess_category(p)) != cat: + # Misclassified stale entry — drop it rather than delete the file. + _log(f"SKIP stale {cat} entry: {p} (re-classified as {re_cat!r}{_STALE_SKIP_NOTE[cat]})") continue - - age = (now - datetime.fromisoformat(item["timestamp"])).days - - # ---- stale-state migration (fixes #37721) ---- - # Old tracked.json entries may carry a "cron-output" category for - # paths that are NOT under cron/output/ (e.g. cron/jobs.json). - # guess_category() was fixed in #34840, but existing entries are - # never re-validated. Re-classify here so stale entries for cron - # control-plane state are not deleted. - if cat == "cron-output": - re_cat = guess_category(p) - if re_cat != "cron-output": - _log( - f"SKIP stale cron-output entry: {p} " - f"(re-classified as {re_cat!r})" - ) - # Drop the stale entry — it was misclassified. - continue - - # ---- stale-state migration for 'test' category (fixes #75403) ---- - # Old tracked.json entries may carry a "test" category for paths - # that are now under protected project directories (patches/, - # projects/, etc.). guess_category() was tightened in the fix for - # #75403, but existing entries are never re-validated. Re-classify - # here so stale entries for protected paths are not deleted. - if cat == "test": - re_cat = guess_category(p) - if re_cat != "test": - _log( - f"SKIP stale test entry: {p} " - f"(re-classified as {re_cat!r} — under protected tree)" - ) - continue - - # Hard safety net: never delete cron control-plane state even if - # the category somehow slipped through re-validation above. + # Hard safety net even if re-validation above somehow let it through. if _is_protected_cron_path(p): _log(f"SKIP protected cron path: {p}") continue - - should_delete = ( - cat == "test" - or (cat == "temp" and age > 7) - or (cat == "cron-output" and age > 14) - ) - - if should_delete: - try: - if p.is_file(): - p.unlink() - elif p.is_dir(): - shutil.rmtree(p) - freed += item["size"] - deleted += 1 - _log(f"DELETED: {p} ({cat}, {fmt_size(item['size'])})") - except OSError as e: - _log(f"ERROR deleting {p}: {e}") - errors.append(f"{p}: {e}") - new_tracked.append(item) + if not _is_auto_delete(cat, age): + new_tracked.append(item) + continue + err = _delete_item(item) + if err is None: + freed += item["size"] + deleted += 1 else: + errors.append(err) new_tracked.append(item) - # Remove empty dirs under HERMES_HOME, but never recurse into known - # durable state trees. Some installs place the Hermes checkout, venv, - # and desktop build under HERMES_HOME; a full rglob over that tree can - # stall the gateway event loop for minutes. - hermes_home = get_hermes_home() - empty_removed = 0 - sweep_stack: List[Tuple[Path, bool]] = [] - try: - for top in hermes_home.iterdir(): - if ( - top.is_dir() - and not top.is_symlink() - and top.name not in _EMPTY_DIR_PROTECTED_TOP_LEVEL - and top.name not in _EMPTY_DIR_SWEEP_PRUNE_DIRS - ): - sweep_stack.append((top, False)) - except OSError: - sweep_stack = [] + empty_removed = _sweep_empty_dirs(get_hermes_home()) + save_tracked(new_tracked) + _log(f"QUICK_SUMMARY: {deleted} files, {empty_removed} dirs, {fmt_size(freed)}") + return {"deleted": deleted, "empty_dirs": empty_removed, "freed": freed, "errors": errors} - while sweep_stack: - dirpath, visited = sweep_stack.pop() + +def _subdirs(dirpath: Path, exclude: frozenset) -> List[Path]: + try: + return [c for c in dirpath.iterdir() if c.is_dir() and not c.is_symlink() and c.name not in exclude] + except OSError: + return [] + + +def _sweep_empty_dirs(hermes_home: Path) -> int: + """Remove empty dirs under HERMES_HOME, never recursing into durable state + trees. Some installs keep the Hermes checkout, venv, and desktop build under + HERMES_HOME; a full rglob there can stall the gateway event loop for minutes. + Iterative post-order so parents emptied by child removal are caught.""" + removed = 0 + stack: List[Tuple[Path, bool]] = [ + (top, False) for top in _subdirs(hermes_home, _EMPTY_DIR_PROTECTED_TOP_LEVEL | _EMPTY_DIR_SWEEP_PRUNE_DIRS) + ] + while stack: + dirpath, visited = stack.pop() if visited: try: if not any(dirpath.iterdir()): dirpath.rmdir() - empty_removed += 1 + removed += 1 _log(f"DELETED: {dirpath} (empty dir)") except OSError: pass continue - - sweep_stack.append((dirpath, True)) - try: - for child in dirpath.iterdir(): - if ( - child.is_dir() - and not child.is_symlink() - and child.name not in _EMPTY_DIR_SWEEP_PRUNE_DIRS - ): - sweep_stack.append((child, False)) - except OSError: - pass - - save_tracked(new_tracked) - _log( - f"QUICK_SUMMARY: {deleted} files, {empty_removed} dirs, " - f"{fmt_size(freed)}" - ) - return { - "deleted": deleted, - "empty_dirs": empty_removed, - "freed": freed, - "errors": errors, - } + stack.append((dirpath, True)) + stack.extend((child, False) for child in _subdirs(dirpath, _EMPTY_DIR_SWEEP_PRUNE_DIRS)) + return removed -# --------------------------------------------------------------------------- -# Deep cleanup (interactive — not called from plugin hooks) -# --------------------------------------------------------------------------- - -def deep( - confirm: Optional[callable] = None, -) -> Dict[str, Any]: - """Deep cleanup. - - Runs :func:`quick` first, then asks the *confirm* callable for each - risky item (research > 30d beyond 10 newest, chrome-profile > 14d, - any file > 500 MB). *confirm(item)* must return True to delete. +def deep(confirm: Optional[Callable[[Dict], bool]] = None) -> Dict[str, Any]: + """Deep cleanup: :func:`quick`, then ask *confirm(item)* for each risky item + (research > 30d beyond the 10 newest, chrome-profile > 14d, any file > 500 MB). Returns: ``{"quick": {...}, "deep_deleted": N, "deep_freed": bytes}``. """ quick_result = quick() - - if confirm is None: - # No interactive confirmer — deep stops after the quick pass. + if confirm is None: # no interactive confirmer — stop after the quick pass return {"quick": quick_result, "deep_deleted": 0, "deep_freed": 0} tracked = load_tracked() - now = datetime.now(timezone.utc) - research, chrome, large = [], [], [] + groups: Dict[str, List[Dict]] = {"research": [], "chrome": [], "large": []} + for item, _p, age in _live_items(tracked, datetime.now(timezone.utc)): + group = _prompt_group(item, age) + if group: + groups[group].append(item) - for item in tracked: - p = Path(item["path"]) - if not p.exists(): - continue - age = (now - datetime.fromisoformat(item["timestamp"])).days - cat = item["category"] + groups["research"].sort(key=lambda x: x["timestamp"], reverse=True) + del groups["research"][:10] # keep the 10 newest research items - if cat == "research" and age > 30: - research.append(item) - elif cat == "chrome-profile" and age > 14: - chrome.append(item) - elif item["size"] > 500 * 1024 * 1024: - large.append(item) - - research.sort(key=lambda x: x["timestamp"], reverse=True) - old_research = research[10:] - - freed, count = 0, 0 - to_remove: List[Dict] = [] - - for group in (old_research, chrome, large): - for item in group: - if confirm(item): - try: - p = Path(item["path"]) - if p.is_file(): - p.unlink() - elif p.is_dir(): - shutil.rmtree(p) - to_remove.append(item) - freed += item["size"] - count += 1 - _log( - f"DELETED: {p} ({item['category']}, " - f"{fmt_size(item['size'])})" - ) - except OSError as e: - _log(f"ERROR deleting {item['path']}: {e}") - - if to_remove: - remove_paths = {i["path"] for i in to_remove} + removed = [item for group in groups.values() for item in group if confirm(item) and _delete_item(item) is None] + if removed: + remove_paths = {i["path"] for i in removed} save_tracked([i for i in tracked if i["path"] not in remove_paths]) - return {"quick": quick_result, "deep_deleted": count, "deep_freed": freed} + return {"quick": quick_result, "deep_deleted": len(removed), "deep_freed": sum(i["size"] for i in removed)} -# --------------------------------------------------------------------------- -# Status -# --------------------------------------------------------------------------- +# --- Status ----------------------------------------------------------------- def status() -> Dict[str, Any]: """Return per-category breakdown and top 10 largest tracked files.""" tracked = load_tracked() cats: Dict[str, Dict] = {} for item in tracked: - c = item["category"] - cats.setdefault(c, {"count": 0, "size": 0}) - cats[c]["count"] += 1 - cats[c]["size"] += item["size"] + c = cats.setdefault(item["category"], {"count": 0, "size": 0}) + c["count"] += 1 + c["size"] += item["size"] - existing = [ - (i["path"], i["size"], i["category"]) - for i in tracked if Path(i["path"]).exists() - ] + existing = [(i["path"], i["size"], i["category"]) for i in tracked if Path(i["path"]).exists()] existing.sort(key=lambda x: x[1], reverse=True) - - return { - "categories": cats, - "top10": existing[:10], - "total_tracked": len(tracked), - } + return {"categories": cats, "top10": existing[:10], "total_tracked": len(tracked)} def format_status(s: Dict[str, Any]) -> str: @@ -542,23 +402,18 @@ def format_status(s: Dict[str, Any]) -> str: cats = s["categories"] for cat, d in sorted(cats.items(), key=lambda x: x[1]["size"], reverse=True): lines.append(f"{cat:<20} {d['count']:>6} {fmt_size(d['size']):>10}") - if not cats: lines.append("(nothing tracked yet)") - lines.append("") - lines.append("Top 10 largest tracked files:") + lines += ["", "Top 10 largest tracked files:"] if not s["top10"]: lines.append(" (none)") - else: - for rank, (path, size, cat) in enumerate(s["top10"], 1): - lines.append(f" {rank:>2}. {fmt_size(size):>8} [{cat}] {path}") + for rank, (path, size, cat) in enumerate(s["top10"], 1): + lines.append(f" {rank:>2}. {fmt_size(size):>8} [{cat}] {path}") return "\n".join(lines) -# --------------------------------------------------------------------------- -# Auto-categorisation from tool-call inspection -# --------------------------------------------------------------------------- +# --- Auto-categorisation from tool-call inspection -------------------------- _TEST_PATTERNS = ("test_", "tmp_") _TEST_SUFFIXES = (".test.py", ".test.js", ".test.ts", ".test.md") @@ -572,40 +427,24 @@ def guess_category(path: Path) -> Optional[str]: if not is_safe_path(path): return None - # Skip the state dir itself, logs, memory files, sessions, config. - hermes_home = get_hermes_home() try: - rel = path.resolve().relative_to(hermes_home) + rel = path.resolve().relative_to(get_hermes_home()) top = rel.parts[0] if rel.parts else "" - if top in { - "disk-cleanup", "logs", "memories", "sessions", "config.yaml", - "skills", "plugins", ".env", "USER.md", "MEMORY.md", "SOUL.md", - "auth.json", "hermes-agent", - # User-authored and project trees — never auto-delete files - # inside these just because they happen to be named test_* or - # tmp_* (#75403, also #32164, #37721). - "patches", "projects", "skins", "themes", "contributors", - "profiles", "backups", "optional-skills", - }: + if top in _NEVER_TRACK_TOP_LEVEL: return None - if top == "cron" or top == "cronjobs": - # Only files under the disposable ``output/`` subtree are - # cleanup candidates. Top-level cron control-plane state - # (e.g. ``jobs.json``, ``.tick.lock``) must never be - # auto-tracked — deleting it wipes the live scheduler - # registry. See issue #32164. + if top in ("cron", "cronjobs"): + # Only the disposable ``output/`` subtree is a candidate. Top-level + # control-plane state (jobs.json, .tick.lock) must never be tracked — + # deleting it wipes the live scheduler registry. if len(rel.parts) >= 3 and rel.parts[1] == "output": return "cron-output" return None if top == "cache": return "temp" except ValueError: - # Path isn't under HERMES_HOME (e.g. /tmp/hermes-*) — fall through. - pass + pass # not under HERMES_HOME (e.g. /tmp/hermes-*) — fall through to name rules name = path.name - if name.startswith(_TEST_PATTERNS): - return "test" - if any(name.endswith(sfx) for sfx in _TEST_SUFFIXES): + if name.startswith(_TEST_PATTERNS) or name.endswith(_TEST_SUFFIXES): return "test" return None diff --git a/plugins/security-guidance/__init__.py b/plugins/security-guidance/__init__.py index 99cc6f725e..387a008524 100644 --- a/plugins/security-guidance/__init__.py +++ b/plugins/security-guidance/__init__.py @@ -1,32 +1,11 @@ """security-guidance plugin — fast pattern-matched security warnings on file writes. -Wires one behaviour: - -* ``transform_tool_result`` hook — scans the *content being written* by - ``write_file`` / ``patch`` / ``skill_manage`` (write/patch modes) for known - dangerous code patterns (eval(, pickle.load, yaml.load, os.system, - subprocess(shell=True), dangerouslySetInnerHTML, verify=False, ECB, - XXE-prone XML parsers, GitHub Actions ``${{ github.event.* }}`` injection, - torch.load without ``weights_only=True``, ...). When any pattern matches, - the plugin appends a ``⚠️ Security warning`` block to the JSON tool-result - string. The file is still written; the model sees the warning in the next - turn's tool message and can self-correct. - -Why not block? Patterns have a non-trivial false-positive rate (``eval(`` in -a tokenizer, ``yaml.load`` already wrapped in ``yaml.SafeLoader``, ECB inside -a test fixture). Blocking would force every false positive into an approval -prompt or an interrupted workflow. Warning is the right severity for layer -1 — the agent reads the warning and either fixes the code or briefly -documents why the construct is safe. - -For block-mode (refuse the write entirely), set -``SECURITY_GUIDANCE_BLOCK=1``. This trades convenience for strictness and -is intended for shared dev environments where unsafe-by-default patterns -are policy violations. - -Pattern data lives in ``patterns.py``, forked verbatim from Anthropic's -``claude-plugins-official`` under Apache-2.0. See ``LICENSE`` and ``NOTICE`` -in this directory. +Scans content written by ``write_file`` / ``patch`` / ``skill_manage`` for known dangerous +code patterns and appends a ``⚠️ Security guidance`` block to the tool result; the file is +still written and the model self-corrects next turn. Warn (not block) by default because +patterns have a real false-positive rate (``eval(`` in a tokenizer, ECB in a test fixture); +``SECURITY_GUIDANCE_BLOCK=1`` refuses the write instead, ``SECURITY_GUIDANCE_DISABLE=1`` is +a kill switch. Pattern data is ``patterns.py`` (Apache-2.0 fork, see LICENSE / NOTICE). """ from __future__ import annotations @@ -41,140 +20,101 @@ from . import patterns as _patterns logger = logging.getLogger(__name__) - -# --------------------------------------------------------------------------- -# Configuration -# --------------------------------------------------------------------------- - -# Tool names whose args carry "code being written to disk" we want to scan. -# Maps tool name -> (path_arg_name, content_arg_names). For tools with multiple -# possible content fields (patch's old/new_string vs raw patch text), we scan -# every populated string field. +# tool name -> (path_arg_name, content_arg_names). Every populated content field is scanned +# (patch's new_string vs raw patch text; skill_manage's file_path is the path inside the skill dir). _TARGET_TOOLS: Dict[str, Tuple[str, Tuple[str, ...]]] = { "write_file": ("path", ("content",)), "patch": ("path", ("new_string", "patch")), - # skill_manage write_file / patch sub-actions land here. file_path holds - # the relative path inside the skill dir; we scan it the same way. "skill_manage": ("file_path", ("file_content", "new_string")), } -# Cap on how much content we scan. Above this we skip — pattern matching a -# 10 MB blob has poor signal-to-noise and would slow down the agent loop. +# Above this we skip: matching a multi-MB blob has poor signal and slows the agent loop. _MAX_SCAN_BYTES = 256 * 1024 - -def _block_mode_enabled() -> bool: - return os.environ.get("SECURITY_GUIDANCE_BLOCK", "").lower() in {"1", "true", "yes", "on"} +_TRUTHY = {"1", "true", "yes", "on"} -def _plugin_disabled() -> bool: - return os.environ.get("SECURITY_GUIDANCE_DISABLE", "").lower() in {"1", "true", "yes", "on"} +def _env_flag(name: str) -> bool: + return os.environ.get(name, "").lower() in _TRUTHY -# --------------------------------------------------------------------------- -# Scanning -# --------------------------------------------------------------------------- +def _compile_rules() -> List[Dict[str, Any]]: + """Pre-compile regexes once; substrings stay plain (``in`` beats a literal regex).""" + compiled: List[Dict[str, Any]] = [] + for rule in _patterns.SECURITY_PATTERNS: + regex = None + re_src = rule.get("regex") + if re_src: + try: + regex = re.compile(re_src) + except re.error as err: + logger.warning( + "security-guidance: skipping rule %s — invalid regex %r: %s", + rule["ruleName"], re_src, err, + ) + continue + compiled.append({ + "ruleName": rule["ruleName"], + "reminder": rule["reminder"], + "path_filter": rule.get("path_filter"), + "path_check": rule.get("path_check"), + "substrings": tuple(rule.get("substrings", ())), + "regex": regex, + }) + return compiled -# Pre-compile the regex patterns once. Substring patterns stay as plain -# strings — ``str.__contains__`` is faster than a regex of literal chars. -_COMPILED: List[Dict[str, Any]] = [] -for _rule in _patterns.SECURITY_PATTERNS: - _entry: Dict[str, Any] = { - "ruleName": _rule["ruleName"], - "reminder": _rule["reminder"], - "path_filter": _rule.get("path_filter"), - "path_check": _rule.get("path_check"), - "substrings": tuple(_rule.get("substrings", ())), - "regex": None, - } - _re_src = _rule.get("regex") - if _re_src: - try: - _entry["regex"] = re.compile(_re_src) - except re.error as _err: - logger.warning( - "security-guidance: skipping rule %s — invalid regex %r: %s", - _rule["ruleName"], _re_src, _err, - ) - continue - _COMPILED.append(_entry) +_COMPILED: List[Dict[str, Any]] = _compile_rules() + + +def _rule_matches(entry: Dict[str, Any], path: str, content: str) -> bool: + """One rule against one write. Path predicates are best-effort: an exception is a non-match. + + path_check rules fire on the path ALONE (e.g. "you're editing a workflow file") and never + pattern-match content; path_filter gates content rules to relevant file types. + """ + try: + if entry["path_check"] is not None: + return bool(entry["path_check"](path)) + if entry["path_filter"] is not None and not entry["path_filter"](path): + return False + except Exception: + return False + return any(sub in content for sub in entry["substrings"]) or ( + entry["regex"] is not None and bool(entry["regex"].search(content)) + ) def _scan_content(path: str, content: str) -> List[Tuple[str, str]]: - """Return [(ruleName, reminder), ...] for every pattern that matches. - - ``path`` is used by per-rule path filters (path_filter / path_check). - Each rule fires at most once per call — multiple matches of the same - rule collapse into a single warning entry. - """ + """Return [(ruleName, reminder), ...]; each rule fires at most once per call.""" if not content or len(content.encode("utf-8", errors="ignore")) > _MAX_SCAN_BYTES: return [] - hits: List[Tuple[str, str]] = [] - for entry in _COMPILED: - # path_check: rule fires PURELY on path match (no content regex). Used - # for blanket "you're editing a sensitive file, here are reminders" - # warnings — github_actions_workflow is the canonical example. - path_check = entry.get("path_check") - if path_check is not None: - try: - if path_check(path or ""): - hits.append((entry["ruleName"], entry["reminder"])) - except Exception: - pass - # Path-check rules don't also pattern-match content; move on. - continue - # path_filter: rule is skipped when the path filter returns False - # (e.g. Python-only rules skip .js files; eval_injection skips .md) - path_filter = entry.get("path_filter") - if path_filter is not None: - try: - if not path_filter(path or ""): - continue - except Exception: - continue - matched = False - for sub in entry["substrings"]: - if sub in content: - matched = True - break - if not matched and entry["regex"] is not None: - if entry["regex"].search(content): - matched = True - if matched: - hits.append((entry["ruleName"], entry["reminder"])) - return hits + path = path or "" + return [(e["ruleName"], e["reminder"]) for e in _COMPILED if _rule_matches(e, path, content)] -def _extract_path_and_content(tool_name: str, args: Any) -> List[Tuple[str, str]]: - """Return [(path, content), ...] for a tool call. Empty if nothing to scan.""" +def _scan_args(tool_name: str, args: Any) -> List[Tuple[str, str]]: + """Shared scan for both hooks (block mode via pre_tool_call, warn mode via transform).""" spec = _TARGET_TOOLS.get(tool_name) - if spec is None or not isinstance(args, dict): + if _env_flag("SECURITY_GUIDANCE_DISABLE") or spec is None or not isinstance(args, dict): return [] path_key, content_keys = spec path = args.get(path_key) or "" if not isinstance(path, str): path = "" - out: List[Tuple[str, str]] = [] - for ck in content_keys: - val = args.get(ck) + findings: List[Tuple[str, str]] = [] + for val in (args.get(ck) for ck in content_keys): if isinstance(val, str) and val: - out.append((path, val)) - return out + findings.extend(_scan_content(path, val)) + return findings def _format_warning_block(findings: List[Tuple[str, str]]) -> str: - """Render findings into a Markdown block appended to the tool result.""" + """Render findings into the Markdown block appended to the tool result.""" names = ", ".join(name for name, _ in findings) - lines = [ - "", - "---", - f"⚠️ Security guidance — {len(findings)} pattern{'s' if len(findings) != 1 else ''} matched ({names})", - "", - ] + lines = ["", "---", f"⚠️ Security guidance — {len(findings)} pattern{'s' if len(findings) != 1 else ''} matched ({names})", ""] for _, reminder in findings: - lines.append(reminder) - lines.append("") + lines += [reminder, ""] lines.append( "Pattern matches can be false positives. If the construct is safe in this " "context, briefly document why in a code comment and continue. Otherwise, " @@ -183,33 +123,9 @@ def _format_warning_block(findings: List[Tuple[str, str]]) -> str: return "\n".join(lines) -# --------------------------------------------------------------------------- -# Hooks -# --------------------------------------------------------------------------- - - -def _scan_args(tool_name: str, args: Any) -> List[Tuple[str, str]]: - """Common scan path used by both pre_tool_call (block mode) and - transform_tool_result (warn mode).""" - if _plugin_disabled(): - return [] - findings: List[Tuple[str, str]] = [] - for path, content in _extract_path_and_content(tool_name, args): - findings.extend(_scan_content(path, content)) - return findings - - -def _on_pre_tool_call( - tool_name: str = "", - args: Any = None, - **_: Any, -) -> Optional[Dict[str, str]]: - """In block mode, refuse the write if any pattern matches. - - Default mode is non-blocking — we return None here and let - ``transform_tool_result`` append a warning to the result instead. - """ - if not _block_mode_enabled(): +def _on_pre_tool_call(tool_name: str = "", args: Any = None, **_: Any) -> Optional[Dict[str, str]]: + """Block mode only: refuse the write if any pattern matches (None = let it through).""" + if not _env_flag("SECURITY_GUIDANCE_BLOCK"): return None findings = _scan_args(tool_name, args) if not findings: @@ -225,25 +141,15 @@ def _on_pre_tool_call( def _on_transform_tool_result( - tool_name: str = "", - args: Any = None, - result: Any = None, - **_: Any, + tool_name: str = "", args: Any = None, result: Any = None, **_: Any, ) -> Optional[str]: - """Warn-mode hook: append a security-warning block to the tool result. - - Returning a string replaces the result that the model sees in the next - turn. Returning None leaves the result unchanged. - """ - # Block mode handles findings via pre_tool_call; nothing for this hook - # to do in that case (the tool didn't run, so there's no result to wrap). - if _block_mode_enabled(): + """Warn mode: append the warning block to the result string (None = unchanged).""" + # In block mode pre_tool_call already handled it — the tool didn't run, no result to wrap. + if _env_flag("SECURITY_GUIDANCE_BLOCK") or not isinstance(result, str): return None findings = _scan_args(tool_name, args) if not findings: return None - if not isinstance(result, str): - return None # Don't decorate error results — the model already has bigger problems. try: parsed = json.loads(result) diff --git a/plugins/security-guidance/patterns.py b/plugins/security-guidance/patterns.py index 6980888733..296a947f99 100644 --- a/plugins/security-guidance/patterns.py +++ b/plugins/security-guidance/patterns.py @@ -1,10 +1,8 @@ -""" -Regex-based security pattern definitions for the security-guidance plugin. +"""Regex-based security pattern definitions for the security-guidance plugin. -Pure data + one pure helper. No env-var reads, no I/O — kept side-effect-free -so it can be imported in isolation. +Pure data. No env-var reads, no I/O — importable in isolation. -Forked verbatim from Anthropic's claude-plugins-official repository +Forked from Anthropic's claude-plugins-official repository (plugins/security-guidance/hooks/patterns.py) under the Apache License 2.0: https://github.com/anthropics/claude-plugins-official @@ -22,18 +20,20 @@ Forked verbatim from Anthropic's claude-plugins-official repository See the License for the specific language governing permissions and limitations under the License. -Modifications by NousResearch for the Hermes Agent plugin port: - - none to the pattern data itself; this file is byte-for-byte the upstream - patterns.py at commit 0bde168 (2026-05-26). Hermes-side wiring lives in - __init__.py. +NousResearch modifications: pattern data unchanged from upstream; the upstream RuleId +telemetry table (Claude Code PostToolUse metrics) is dropped — Hermes has no consumer. +Hermes-side wiring lives in __init__.py. """ -from enum import IntEnum - - _JS_EXTS = (".js", ".jsx", ".ts", ".tsx", ".mjs", ".cjs", ".mts", ".cts", ".vue", ".svelte") _PY_EXTS = (".py", ".pyi", ".ipynb") _DOC_EXTS = (".md", ".mdx", ".txt", ".rst", ".json", ".yaml", ".yml") +# Shared path_filter predicates. JS-only gating keeps bare `exec(` off Python's exec() +# and prose; Python-only gating keeps pickle/os.system rules off other languages; +# the eval rule skips doc/prose files entirely. +_JS_ONLY = lambda p: p.endswith(_JS_EXTS) # noqa: E731 +_PY_ONLY = lambda p: p.endswith(_PY_EXTS) # noqa: E731 +_NOT_DOCS = lambda p: not p.endswith(_DOC_EXTS) # noqa: E731 _UNSAFE_DESERIALIZATION_REMINDER = """⚠️ Security Warning: Loading pickle data (or equivalents: cPickle, cloudpickle, dill, marshal, shelve, joblib, pandas.read_pickle, numpy with allow_pickle=True) from untrusted sources allows arbitrary code execution. @@ -49,13 +49,7 @@ _UNSAFE_TORCH_LOAD_REMINDER = """⚠️ Security Warning: torch.load() defaults If the file only contains tensors and simple data structures, pass weights_only=True (or set TORCH_FORCE_WEIGHTS_ONLY_LOAD=1).""" -# Security patterns configuration -SECURITY_PATTERNS = [ - { - "ruleName": "github_actions_workflow", - "path_check": lambda path: ".github/workflows/" in path - and (path.endswith(".yml") or path.endswith(".yaml")), - "reminder": """⚠️ Security Warning: You are editing a GitHub Actions workflow file. Be aware of these security risks: +_GITHUB_ACTIONS_REMINDER = """⚠️ Security Warning: You are editing a GitHub Actions workflow file. Be aware of these security risks: 1. **Command Injection**: Never use untrusted input (like issue titles, PR descriptions, commit messages) directly in run: commands without proper escaping 2. **Use environment variables**: Instead of ${{ github.event.issue.title }}, use env: with proper quoting @@ -89,16 +83,9 @@ Other risky inputs to be careful with: - github.event.client_payload.* (repository_dispatch events — attacker can set any field) 4. **Ref injection**: Never use untrusted input in `ref:` parameters of `actions/checkout`. For `client_payload.pr_number`, validate it matches `^[0-9]+$` before using in `ref: refs/pull/${{ ... }}/head` -- github.head_ref""", - }, - { - "ruleName": "child_process_exec", - # Gate to JS/TS files — bare `exec(` otherwise fires on Python's - # exec() and on prose/docstrings mentioning exec. - "path_filter": lambda p: p.endswith(_JS_EXTS), - "substrings": ["child_process.exec", "execSync("], - "regex": r"(? o[k], root); for computation use a safe expression parser. NEVER interpolate untrusted strings into new Function() bodies.", - }, - { - "ruleName": "eval_injection", - # Lookbehind excludes `.` so method calls like PyTorch model.eval(), - # redis.eval(), spec.eval() don't match. Skip doc/prose files. - "path_filter": lambda p: not p.endswith(_DOC_EXTS), - "regex": r"(?]{0,400}integrity\s*=)" - r"[^>]{0,200}src\s*=\s*[\x22\x27](?:https?:)?//" - r"[^\x22\x27]{1,300}[\x22\x27]" - r"[^>]{0,100}>" - ), - "reminder": '⚠️ Security Warning: Add integrity="sha384-..." crossorigin="anonymous" to external script tags. Loading scripts without Subresource Integrity exposes you to CDN compromise.', - }, - { - "ruleName": "torch_unsafe_load", - # Suppressed by weights_only=True on the same line (within 200 chars). weights_only=False - # still triggers. Multi-line calls false-positive — same known limitation as unsafe_yaml_load. - "regex": r"(?:\btorch\.load|\.torch_load)\s*\((?![^)\n]{0,200}weights_only\s*=\s*True)", - "reminder": _UNSAFE_TORCH_LOAD_REMINDER, - }, - { - "ruleName": "yaml_unsafe_load_variants", - # yaml.unsafe_load (stdlib alias) plus unsafe wrapper method names seen in the wild. - # Bare yaml.load() is unsafe_yaml_load's job (RuleId 12). - "regex": r"(?:\byaml\.unsafe_load|\.yaml_unsafe_load)\s*\(", - "reminder": _UNSAFE_YAML_LOAD_REMINDER, - }, - { - "ruleName": "pickle_wrapper_load", - # Library APIs that unpickle without saying "pickle". numpy.load only triggers - # when allow_pickle=True is explicit (defaults to False since numpy 1.16.3). - "regex": r"\bjoblib\.load\s*\(|\b(?:pd|pandas)\.read_pickle\s*\(|\.cloudpickle_load\s*\(|\b(?:np|numpy)\.load\s*\([^)\n]{0,200}allow_pickle\s*=\s*True", - "reminder": _UNSAFE_DESERIALIZATION_REMINDER, - }, +- For numeric values: parse to int/float first""" + + +def _rule(name, reminder, **triggers): + """Build one rule dict; only the trigger keys given (regex / substrings / + path_filter / path_check) are present, so consumers can test with ``in``.""" + return {"ruleName": name, "reminder": reminder, **triggers} + + +# Security patterns configuration. Regex notes: +# - eval / exec lookbehinds exclude `.` so method calls (model.eval(), redis.eval()) don't match. +# - pickle matches deserialization only (load/loads/Unpickler); pickle.dump is not the RCE +# surface, and `pkl_load` needs a word boundary so similarly named safe loaders don't match. +# - script_src_without_sri: negative lookahead after ` o[k], root); for computation use a safe expression parser. NEVER interpolate untrusted strings into new Function() bodies.", + substrings=["new Function"]), + _rule("eval_injection", + "⚠️ Security Warning: eval() executes arbitrary code and is a major security risk. Use JSON.parse() for data, ast.literal_eval() for Python literals, or a safe expression parser. If this is safe or is explicitly needed, briefly document that in a comment before continuing.", + path_filter=_NOT_DOCS, regex=r"(?]{0,400}integrity\s*=)" + r"[^>]{0,200}src\s*=\s*[\x22\x27](?:https?:)?//" + r"[^\x22\x27]{1,300}[\x22\x27]" + r"[^>]{0,100}>")), + _rule("torch_unsafe_load", _UNSAFE_TORCH_LOAD_REMINDER, + regex=r"(?:\btorch\.load|\.torch_load)\s*\((?![^)\n]{0,200}weights_only\s*=\s*True)"), + _rule("yaml_unsafe_load_variants", _UNSAFE_YAML_LOAD_REMINDER, + regex=r"(?:\byaml\.unsafe_load|\.yaml_unsafe_load)\s*\("), + _rule("pickle_wrapper_load", _UNSAFE_DESERIALIZATION_REMINDER, + regex=r"\bjoblib\.load\s*\(|\b(?:pd|pandas)\.read_pickle\s*\(|\.cloudpickle_load\s*\(|\b(?:np|numpy)\.load\s*\([^)\n]{0,200}allow_pickle\s*=\s*True"), ] - - -class RuleId(IntEnum): - """ - Stable numeric IDs for SECURITY_PATTERNS rules, emitted via the PostToolUse - metrics field so telemetry can attribute pattern-warning events to - specific checks. The metrics schema only allows bool|number values (no - strings), so rule names can't be sent directly. - - Values are frozen: do not renumber existing entries. Append new ones. - """ - GITHUB_ACTIONS_WORKFLOW = 1 - CHILD_PROCESS_EXEC = 2 - NEW_FUNCTION_INJECTION = 3 - EVAL_INJECTION = 4 - REACT_DANGEROUSLY_SET_HTML = 5 - DOCUMENT_WRITE_XSS = 6 - INNERHTML_XSS = 7 - PICKLE_DESERIALIZATION = 8 - OS_SYSTEM_INJECTION = 9 - PYTHON_SUBPROCESS_SHELL = 10 - GO_EXEC_SHELL_INJECTION = 11 - UNSAFE_YAML_LOAD = 12 - NODE_CREATECIPHER_NO_IV = 13 - AES_ECB_MODE = 14 - TLS_VERIFICATION_DISABLED = 15 - MARSHAL_LOADS = 16 - SHELVE_OPEN = 17 - XML_UNSAFE_PARSE = 18 - PICKLE_VARIANTS_LOAD = 19 - OUTERHTML_XSS = 20 - INSERTADJACENTHTML_XSS = 21 - SCRIPT_SRC_WITHOUT_SRI = 22 - TORCH_UNSAFE_LOAD = 23 - YAML_UNSAFE_LOAD_VARIANTS = 24 - PICKLE_WRAPPER_LOAD = 25 - - -_RULE_NAME_TO_ID = { - "github_actions_workflow": RuleId.GITHUB_ACTIONS_WORKFLOW, - "child_process_exec": RuleId.CHILD_PROCESS_EXEC, - "new_function_injection": RuleId.NEW_FUNCTION_INJECTION, - "eval_injection": RuleId.EVAL_INJECTION, - "react_dangerously_set_html": RuleId.REACT_DANGEROUSLY_SET_HTML, - "document_write_xss": RuleId.DOCUMENT_WRITE_XSS, - "innerHTML_xss": RuleId.INNERHTML_XSS, - "pickle_deserialization": RuleId.PICKLE_DESERIALIZATION, - "os_system_injection": RuleId.OS_SYSTEM_INJECTION, - "python_subprocess_shell": RuleId.PYTHON_SUBPROCESS_SHELL, - "go_exec_shell_injection": RuleId.GO_EXEC_SHELL_INJECTION, - "unsafe_yaml_load": RuleId.UNSAFE_YAML_LOAD, - "node_createcipher_no_iv": RuleId.NODE_CREATECIPHER_NO_IV, - "aes_ecb_mode": RuleId.AES_ECB_MODE, - "tls_verification_disabled": RuleId.TLS_VERIFICATION_DISABLED, - "marshal_loads": RuleId.MARSHAL_LOADS, - "shelve_open": RuleId.SHELVE_OPEN, - "xml_unsafe_parse": RuleId.XML_UNSAFE_PARSE, - "pickle_variants_load": RuleId.PICKLE_VARIANTS_LOAD, - "outerHTML_xss": RuleId.OUTERHTML_XSS, - "insertAdjacentHTML_xss": RuleId.INSERTADJACENTHTML_XSS, - "script_src_without_sri": RuleId.SCRIPT_SRC_WITHOUT_SRI, - "torch_unsafe_load": RuleId.TORCH_UNSAFE_LOAD, - "yaml_unsafe_load_variants": RuleId.YAML_UNSAFE_LOAD_VARIANTS, - "pickle_wrapper_load": RuleId.PICKLE_WRAPPER_LOAD, -} - -# Fail loudly at import time if a pattern is added without a RuleId. -# This fires in pytest on every PR, so desync is caught before merge. -assert set(_RULE_NAME_TO_ID) == {p["ruleName"] for p in SECURITY_PATTERNS}, ( - f"RuleId enum out of sync with SECURITY_PATTERNS: " - f"missing={set(p['ruleName'] for p in SECURITY_PATTERNS) - set(_RULE_NAME_TO_ID)}, " - f"extra={set(_RULE_NAME_TO_ID) - set(p['ruleName'] for p in SECURITY_PATTERNS)}" -) - - -def rule_names_to_mask(rule_names): - """Pack a set of rule names into a bitmask. Bit N set means RuleId(N) matched. - User-defined patterns (rule_name starting with "user:") have no static - RuleId and are excluded from the mask.""" - mask = 0 - for name in rule_names: - if name in _RULE_NAME_TO_ID: - mask |= 1 << _RULE_NAME_TO_ID[name] - return mask diff --git a/tests/plugins/test_security_guidance_plugin.py b/tests/plugins/test_security_guidance_plugin.py index 6be99c6b3a..a00bafddcd 100644 --- a/tests/plugins/test_security_guidance_plugin.py +++ b/tests/plugins/test_security_guidance_plugin.py @@ -2,8 +2,8 @@ Covers ``plugins/security-guidance/``: - * ``patterns.py`` data integrity — every rule has a ``RuleId``, the - fail-loud import assertion is wired. + * ``patterns.py`` data integrity — every rule has a name, reminder, and at + least one trigger; names are unique. * ``_scan_content`` — true positives (pickle.load, yaml.load, eval, dangerouslySetInnerHTML, GitHub Actions workflow), true negatives (.md skips Python rules, ``model.eval()`` doesn't trip eval),