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

refactor(plugins): disk-cleanup and security-guidance — table-driven scans/patterns, safety lists unchanged

Browse Source
This commit is contained in:
Teknium
2026-09-02 11:23:23 -07:00
parent 517f593af9
commit 5d78bd817d
5 changed files with 481 additions and 968 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/disk-cleanup/__init__.py
+111 -196
View File
@@ -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 <path>` 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> <category>"
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 <path>"
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 <path>` to skip, or delete "
"manually via terminal."
)
return "\n".join(lines)
if sub == "track":
if len(argv) < 3:
return "Usage: /disk-cleanup track <path> <category>"
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 <path>"
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)
plugins/disk-cleanup/disk_cleanup.py
+182 -343
View File
@@ -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/<name>`` — 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
plugins/security-guidance/__init__.py
+75 -169
View File
@@ -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)
plugins/security-guidance/patterns.py
+111 -258
View File
@@ -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"(?<![a-zA-Z0-9_\.])exec\(",
"reminder": """⚠️ Security Warning: Using child_process.exec() can lead to command injection vulnerabilities.
- github.head_ref"""
_CHILD_PROCESS_EXEC_REMINDER = """⚠️ Security Warning: Using child_process.exec() can lead to command injection vulnerabilities.
exec() runs the command string through a shell, so any user input interpolated into it can inject arbitrary commands. Prefer child_process.execFile() (or spawn()) with an argument array instead of building a shell string.
@@ -113,56 +100,9 @@ Why execFile/spawn with an argument array is safer:
- No shell is involved, so shell metacharacters in arguments are not interpreted
- Arguments are passed directly to the program rather than interpolated into a command string
Only use exec() if you absolutely need shell features and the input is guaranteed to be safe.""",
},
{
"ruleName": "new_function_injection",
"substrings": ["new Function"],
"reminder": "\u26a0\ufe0f Security Warning: Using new Function() with string interpolation is a CODE INJECTION vulnerability. If any variable is concatenated or interpolated into the function body string, an attacker controlling that variable can execute arbitrary code. Use safe alternatives: for property access use obj[key] or array.reduce((o, k) => 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"(?<![a-zA-Z0-9_\.])eval\(",
"reminder": "⚠️ 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.",
},
{
"ruleName": "react_dangerously_set_html",
"substrings": ["dangerouslySetInnerHTML"],
"reminder": "⚠️ Security Warning: dangerouslySetInnerHTML can lead to XSS vulnerabilities if used with untrusted content. Ensure all content is properly sanitized using an HTML sanitizer library like DOMPurify, or use safe alternatives.",
},
{
"ruleName": "document_write_xss",
"substrings": ["document.write"],
"reminder": "⚠️ Security Warning: document.write() can be exploited for XSS attacks and has performance issues. Use DOM manipulation methods like createElement() and appendChild() instead.",
},
{
"ruleName": "innerHTML_xss",
"substrings": [".innerHTML =", ".innerHTML="],
"reminder": "⚠️ Security Warning: Setting innerHTML with untrusted content can lead to XSS vulnerabilities. Use textContent for plain text or safe DOM methods for HTML content. If you need HTML support, consider using an HTML sanitizer library such as DOMPurify.",
},
{
"ruleName": "pickle_deserialization",
# Match deserialization only (load/loads/Unpickler). pickle.dump is
# not the RCE surface. `pkl_load` needs a word boundary so similarly
# named safe loaders don't match.
"path_filter": lambda p: p.endswith(_PY_EXTS),
"regex": r"(?<![a-zA-Z0-9_])pickle\.(loads?|Unpickler)\b|(?<![a-zA-Z0-9_])pkl_load\(",
"reminder": _UNSAFE_DESERIALIZATION_REMINDER,
},
{
"ruleName": "os_system_injection",
"path_filter": lambda p: p.endswith(_PY_EXTS),
"regex": r"\bos\.system\s*\(",
"substrings": ["from os import system"],
"reminder": "⚠️ Security Warning: os.system() runs a shell and is a command-injection sink. Use subprocess.run([...]) with a list of arguments instead. If this is safe or is explicitly needed, briefly document that in a comment before continuing.",
},
{
"ruleName": "python_subprocess_shell",
"regex": r"subprocess\.(?:run|call|Popen|check_output|check_call)\(.*shell\s*=\s*True",
"reminder": """⚠️ Security Warning: Using subprocess with shell=True enables command injection.
Only use exec() if you absolutely need shell features and the input is guaranteed to be safe."""
_SUBPROCESS_SHELL_REMINDER = """⚠️ Security Warning: Using subprocess with shell=True enables command injection.
UNSAFE:
subprocess.run(f"ls {user_input}", shell=True)
@@ -172,16 +112,9 @@ SAFE - pass arguments as a list without shell:
subprocess.run(["ls", user_input])
subprocess.call(["grep", pattern])
When arguments are passed as a list without shell=True, special characters cannot be interpreted as shell metacharacters.""",
},
# =====================================================================
# Go-specific security patterns
# =====================================================================
{
"ruleName": "go_exec_shell_injection",
# Detect exec.Command with shell invocation (sh, bash, /bin/sh, /bin/bash)
"regex": r'exec\.Command\(\s*"(?:sh|bash|/bin/sh|/bin/bash)"',
"reminder": """⚠️ Security Warning: Using exec.Command with a shell interpreter (sh/bash) enables command injection.
When arguments are passed as a list without shell=True, special characters cannot be interpreted as shell metacharacters."""
_GO_EXEC_SHELL_REMINDER = """⚠️ Security Warning: Using exec.Command with a shell interpreter (sh/bash) enables command injection.
UNSAFE:
exec.Command("sh", "-c", "ping -c 1 " + host)
@@ -196,173 +129,93 @@ When arguments are passed directly (not through a shell), special characters in
Additionally, validate user inputs:
- For hostnames/IPs: use net.ParseIP() or a hostname regex
- For file paths: use filepath.Clean() and verify the result is within an allowed directory
- For numeric values: parse to int/float first""",
},
{
"ruleName": "unsafe_yaml_load",
"regex": r"\byaml\.load\s*\((?![^)\n]{0,80}\bSafe)",
"reminder": _UNSAFE_YAML_LOAD_REMINDER,
},
{
"ruleName": "node_createcipher_no_iv",
"regex": r"\bcrypto\.(createCipher|createDecipher)\b",
"reminder": "⚠️ Security Warning: Use crypto.createCipheriv() / createDecipheriv(). createCipher was removed in Node 22 and derives the key insecurely (no IV, MD5-based KDF).",
},
{
"ruleName": "aes_ecb_mode",
"regex": r"\bAES\.MODE_ECB\b|\bmodes\.ECB\s*\(|[\x22\x27]aes-\d+-ecb[\x22\x27]",
"reminder": "⚠️ Security Warning: Use AES-GCM or AES-CBC with HMAC. ECB mode leaks plaintext structure (identical blocks encrypt to identical ciphertext).",
},
{
"ruleName": "tls_verification_disabled",
"regex": r"\bverify\s*=\s*False\b|rejectUnauthorized\s*:\s*false|InsecureSkipVerify\s*:\s*true|NODE_TLS_REJECT_UNAUTHORIZED\s*=\s*[\x22\x27]?0|ssl\._create_unverified_context|check_hostname\s*=\s*False",
"reminder": "⚠️ Security Warning: Don't disable TLS verification. This allows MITM attacks. For self-signed dev certs, add the CA to your trust store or use a properly-issued cert.",
},
{
"ruleName": "marshal_loads",
"regex": r"\bmarshal\.loads?\s*\(",
"reminder": _UNSAFE_DESERIALIZATION_REMINDER,
},
{
"ruleName": "shelve_open",
"regex": r"\bshelve\.open\s*\(",
"reminder": _UNSAFE_DESERIALIZATION_REMINDER,
},
{
"ruleName": "xml_unsafe_parse",
"regex": r"\b(xml\.etree\.ElementTree|ElementTree|ET)\.(parse|fromstring|XML)\s*\(|\bminidom\.(parse|parseString)\s*\(|\bxml\.sax\.(parse|make_parser)\b",
"reminder": "⚠️ Security Warning: Use defusedxml.ElementTree. Python's stdlib XML parsers are vulnerable to XXE (external entity) and billion-laughs attacks by default.",
},
{
"ruleName": "pickle_variants_load",
"regex": r"\b(cPickle|cloudpickle|dill)\.(load|loads)\s*\(",
"reminder": _UNSAFE_DESERIALIZATION_REMINDER,
},
{
"ruleName": "outerHTML_xss",
"substrings": [".outerHTML =", ".outerHTML="],
"reminder": "⚠️ Security Warning: Use textContent or sanitize with DOMPurify. outerHTML assignment is an XSS sink equivalent to innerHTML.",
},
{
"ruleName": "insertAdjacentHTML_xss",
"substrings": [".insertAdjacentHTML("],
"reminder": "⚠️ Security Warning: Use insertAdjacentText() or sanitize with DOMPurify. insertAdjacentHTML is an XSS sink.",
},
{
"ruleName": "script_src_without_sri",
# Detect remote code execution via dynamic import/eval of fetched content.
# Negative lookahead after src checks for integrity= anywhere in the remaining tag.
"regex": (
r"<script\s+(?![^>]{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 `<script` checks for integrity= anywhere
# in the remaining tag.
# - torch_unsafe_load is 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.
# - yaml_unsafe_load_variants covers yaml.unsafe_load plus unsafe wrapper method names seen in
# the wild; bare yaml.load() is unsafe_yaml_load's job.
# - 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).
SECURITY_PATTERNS = [
_rule("github_actions_workflow", _GITHUB_ACTIONS_REMINDER,
path_check=lambda path: ".github/workflows/" in path
and (path.endswith(".yml") or path.endswith(".yaml"))),
_rule("child_process_exec", _CHILD_PROCESS_EXEC_REMINDER, path_filter=_JS_ONLY,
substrings=["child_process.exec", "execSync("], regex=r"(?<![a-zA-Z0-9_\.])exec\("),
_rule("new_function_injection",
"\u26a0\ufe0f Security Warning: Using new Function() with string interpolation is a CODE INJECTION vulnerability. If any variable is concatenated or interpolated into the function body string, an attacker controlling that variable can execute arbitrary code. Use safe alternatives: for property access use obj[key] or array.reduce((o, k) => 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"(?<![a-zA-Z0-9_\.])eval\("),
_rule("react_dangerously_set_html",
"⚠️ Security Warning: dangerouslySetInnerHTML can lead to XSS vulnerabilities if used with untrusted content. Ensure all content is properly sanitized using an HTML sanitizer library like DOMPurify, or use safe alternatives.",
substrings=["dangerouslySetInnerHTML"]),
_rule("document_write_xss",
"⚠️ Security Warning: document.write() can be exploited for XSS attacks and has performance issues. Use DOM manipulation methods like createElement() and appendChild() instead.",
substrings=["document.write"]),
_rule("innerHTML_xss",
"⚠️ Security Warning: Setting innerHTML with untrusted content can lead to XSS vulnerabilities. Use textContent for plain text or safe DOM methods for HTML content. If you need HTML support, consider using an HTML sanitizer library such as DOMPurify.",
substrings=[".innerHTML =", ".innerHTML="]),
_rule("pickle_deserialization", _UNSAFE_DESERIALIZATION_REMINDER, path_filter=_PY_ONLY,
regex=r"(?<![a-zA-Z0-9_])pickle\.(loads?|Unpickler)\b|(?<![a-zA-Z0-9_])pkl_load\("),
_rule("os_system_injection",
"⚠️ Security Warning: os.system() runs a shell and is a command-injection sink. Use subprocess.run([...]) with a list of arguments instead. If this is safe or is explicitly needed, briefly document that in a comment before continuing.",
path_filter=_PY_ONLY, regex=r"\bos\.system\s*\(", substrings=["from os import system"]),
_rule("python_subprocess_shell", _SUBPROCESS_SHELL_REMINDER,
regex=r"subprocess\.(?:run|call|Popen|check_output|check_call)\(.*shell\s*=\s*True"),
# Go: exec.Command with a shell invocation (sh, bash, /bin/sh, /bin/bash)
_rule("go_exec_shell_injection", _GO_EXEC_SHELL_REMINDER,
regex=r'exec\.Command\(\s*"(?:sh|bash|/bin/sh|/bin/bash)"'),
_rule("unsafe_yaml_load", _UNSAFE_YAML_LOAD_REMINDER,
regex=r"\byaml\.load\s*\((?![^)\n]{0,80}\bSafe)"),
_rule("node_createcipher_no_iv",
"⚠️ Security Warning: Use crypto.createCipheriv() / createDecipheriv(). createCipher was removed in Node 22 and derives the key insecurely (no IV, MD5-based KDF).",
regex=r"\bcrypto\.(createCipher|createDecipher)\b"),
_rule("aes_ecb_mode",
"⚠️ Security Warning: Use AES-GCM or AES-CBC with HMAC. ECB mode leaks plaintext structure (identical blocks encrypt to identical ciphertext).",
regex=r"\bAES\.MODE_ECB\b|\bmodes\.ECB\s*\(|[\x22\x27]aes-\d+-ecb[\x22\x27]"),
_rule("tls_verification_disabled",
"⚠️ Security Warning: Don't disable TLS verification. This allows MITM attacks. For self-signed dev certs, add the CA to your trust store or use a properly-issued cert.",
regex=r"\bverify\s*=\s*False\b|rejectUnauthorized\s*:\s*false|InsecureSkipVerify\s*:\s*true|NODE_TLS_REJECT_UNAUTHORIZED\s*=\s*[\x22\x27]?0|ssl\._create_unverified_context|check_hostname\s*=\s*False"),
_rule("marshal_loads", _UNSAFE_DESERIALIZATION_REMINDER, regex=r"\bmarshal\.loads?\s*\("),
_rule("shelve_open", _UNSAFE_DESERIALIZATION_REMINDER, regex=r"\bshelve\.open\s*\("),
_rule("xml_unsafe_parse",
"⚠️ Security Warning: Use defusedxml.ElementTree. Python's stdlib XML parsers are vulnerable to XXE (external entity) and billion-laughs attacks by default.",
regex=r"\b(xml\.etree\.ElementTree|ElementTree|ET)\.(parse|fromstring|XML)\s*\(|\bminidom\.(parse|parseString)\s*\(|\bxml\.sax\.(parse|make_parser)\b"),
_rule("pickle_variants_load", _UNSAFE_DESERIALIZATION_REMINDER,
regex=r"\b(cPickle|cloudpickle|dill)\.(load|loads)\s*\("),
_rule("outerHTML_xss",
"⚠️ Security Warning: Use textContent or sanitize with DOMPurify. outerHTML assignment is an XSS sink equivalent to innerHTML.",
substrings=[".outerHTML =", ".outerHTML="]),
_rule("insertAdjacentHTML_xss",
"⚠️ Security Warning: Use insertAdjacentText() or sanitize with DOMPurify. insertAdjacentHTML is an XSS sink.",
substrings=[".insertAdjacentHTML("]),
_rule("script_src_without_sri",
'⚠️ Security Warning: Add integrity="sha384-..." crossorigin="anonymous" to external script tags. Loading scripts without Subresource Integrity exposes you to CDN compromise.',
regex=(r"<script\s+(?![^>]{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
tests/plugins/test_security_guidance_plugin.py
+2 -2
View File
@@ -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),