hermes-agent/plugins/security-guidance/__init__.py

"""security-guidance plugin — fast pattern-matched security warnings on file writes.

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

import json
import logging
import os
import re
from typing import Any, Dict, List, Optional, Tuple

from . import patterns as _patterns

logger = logging.getLogger(__name__)

# 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": ("file_path", ("file_content", "new_string")),
}

# Above this we skip: matching a multi-MB blob has poor signal and slows the agent loop.
_MAX_SCAN_BYTES = 256 * 1024

_TRUTHY = {"1", "true", "yes", "on"}


def _env_flag(name: str) -> bool:
    return os.environ.get(name, "").lower() in _TRUTHY


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


_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), ...]; each rule fires at most once per call."""
    if not content or len(content.encode("utf-8", errors="ignore")) > _MAX_SCAN_BYTES:
        return []
    path = path or ""
    return [(e["ruleName"], e["reminder"]) for e in _COMPILED if _rule_matches(e, path, content)]


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 _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 = ""
    findings: List[Tuple[str, str]] = []
    for val in (args.get(ck) for ck in content_keys):
        if isinstance(val, str) and val:
            findings.extend(_scan_content(path, val))
    return findings


def _format_warning_block(findings: List[Tuple[str, str]]) -> str:
    """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})", ""]
    for _, reminder in findings:
        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, "
        "fix the code before moving on."
    )
    return "\n".join(lines)


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:
        return None
    return {
        "action": "block",
        "message": (
            "security-guidance refused this write: "
            + _format_warning_block(findings)
            + "\n\nTo override, unset SECURITY_GUIDANCE_BLOCK and retry."
        ),
    }


def _on_transform_tool_result(
    tool_name: str = "", args: Any = None, result: Any = None, **_: Any,
) -> Optional[str]:
    """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
    # Don't decorate error results — the model already has bigger problems.
    try:
        parsed = json.loads(result)
        if isinstance(parsed, dict) and "error" in parsed and len(parsed) <= 2:
            return None
    except (ValueError, TypeError):
        pass
    return result + "\n\n" + _format_warning_block(findings)


def register(ctx) -> None:
    ctx.register_hook("pre_tool_call", _on_pre_tool_call)
    ctx.register_hook("transform_tool_result", _on_transform_tool_result)