feat(skills): add opt-in AST deep diagnostics (salvage of #30918)

hermes_cli/commands.py

@@ -164,7 +164,7 @@ class CommandDef:
                cli_only=True),
     CommandDef("skills", "Search, install, inspect, or manage skills",
                "Tools & Skills", cli_only=True,
-               subcommands=("search", "browse", "inspect", "install")),
+               subcommands=("search", "browse", "inspect", "install", "audit")),
     CommandDef("bundles", "List skill bundles (aliases /<name> for multiple skills)",
                "Tools & Skills"),
     CommandDef("cron", "Manage scheduled tasks", "Tools & Skills",

hermes_cli/main.py

@@ -12301,6 +12301,11 @@ def cmd_pairing(args):
     skills_audit.add_argument(
         "name", nargs="?", help="Specific skill to audit (default: all)"
     )
+    skills_audit.add_argument(
+        "--deep",
+        action="store_true",
+        help="Run AST-level analysis on Python files (opt-in diagnostic)",
+    )
 
     skills_uninstall = skills_subparsers.add_parser(
         "uninstall", help="Remove a hub-installed skill"

hermes_cli/skills_hub.py

@@ -906,8 +906,14 @@ def do_update(name: Optional[str] = None, console: Optional[Console] = None) ->
     c.print(f"[bold green]Updated {len(updates)} skill(s).[/]\n")
 
 
-def do_audit(name: Optional[str] = None, console: Optional[Console] = None) -> None:
-    """Re-run security scan on installed hub skills."""
+def do_audit(name: Optional[str] = None, console: Optional[Console] = None,
+             deep: bool = False) -> None:
+    """Re-run security scan on installed hub skills.
+
+    When ``deep=True``, also runs an opt-in AST-level diagnostic on Python
+    files (review aid only — not a security gate; skills_guard.py verdicts
+    are unchanged).
+    """
     from tools.skills_hub import HubLockFile, SKILLS_DIR
     from tools.skills_guard import scan_skill, format_scan_report
 
@@ -928,6 +934,9 @@ def do_audit(name: Optional[str] = None, console: Optional[Console] = None) -> N
 
     c.print(f"\n[bold]Auditing {len(targets)} skill(s)...[/]\n")
 
+    if deep:
+        from tools.skills_ast_audit import ast_scan_path, format_ast_report
+
     for entry in targets:
         skill_path = SKILLS_DIR / entry["install_path"]
         if not skill_path.exists():
@@ -936,6 +945,10 @@ def do_audit(name: Optional[str] = None, console: Optional[Console] = None) -> N
 
         result = scan_skill(skill_path, source=entry.get("identifier", entry["source"]))
         c.print(format_scan_report(result))
+
+        if deep:
+            c.print(format_ast_report(ast_scan_path(skill_path), skill_name=entry["name"]))
+
         c.print()
 
 
@@ -1343,7 +1356,8 @@ def skills_command(args) -> None:
     elif action == "update":
         do_update(name=getattr(args, "name", None))
     elif action == "audit":
-        do_audit(name=getattr(args, "name", None))
+        do_audit(name=getattr(args, "name", None),
+                 deep=getattr(args, "deep", False))
     elif action == "uninstall":
         do_uninstall(args.name)
     elif action == "reset":
@@ -1395,6 +1409,8 @@ def handle_skills_slash(cmd: str, console: Optional[Console] = None) -> None:
         /skills update
         /skills audit
         /skills audit my-skill
+        /skills audit --deep
+        /skills audit my-skill --deep
         /skills uninstall my-skill
         /skills tap list
         /skills tap add owner/repo
@@ -1509,8 +1525,9 @@ def handle_skills_slash(cmd: str, console: Optional[Console] = None) -> None:
         do_update(name=name, console=c)
 
     elif action == "audit":
-        name = args[0] if args else None
-        do_audit(name=name, console=c)
+        name = args[0] if args and not args[0].startswith("--") else None
+        deep = "--deep" in args
+        do_audit(name=name, console=c, deep=deep)
 
     elif action == "uninstall":
         if not args:

tests/tools/test_skills_ast_audit.py

@@ -0,0 +1,103 @@
+"""Tests for tools.skills_ast_audit — opt-in AST diagnostic scanner."""
+
+import sys
+from pathlib import Path
+
+from tools.skills_ast_audit import ast_scan_path, format_ast_report
+
+
+def _pids(findings):
+    return [pid for (_f, _l, pid, _d) in findings]
+
+
+def test_bypass_payload_detected(tmp_path):
+    """The exact bypass shape from #7072 is caught."""
+    f = tmp_path / "exfil.py"
+    f.write_text(
+        "import importlib\n"
+        "parts = ['o', 's']\n"
+        "m = importlib.import_module(''.join(parts))\n"
+        "e = m.__dict__[''.join(['e','n','v'])]\n"
+    )
+    pids = _pids(ast_scan_path(f))
+    assert "dynamic_import" in pids
+    assert "importlib_import" in pids
+    assert "dict_access" in pids
+
+
+def test_syntax_error_does_not_crash(tmp_path):
+    f = tmp_path / "bad.py"
+    f.write_text("def broken(\n")
+    assert ast_scan_path(f) == []
+
+
+def test_recursion_error_does_not_crash(tmp_path):
+    f = tmp_path / "deep.py"
+    f.write_text("a" + ".x" * 5000 + "\n")
+    orig = sys.getrecursionlimit()
+    sys.setrecursionlimit(200)
+    try:
+        result = ast_scan_path(f)
+    finally:
+        sys.setrecursionlimit(orig)
+    assert isinstance(result, list)
+
+
+def test_importer_lookalike_not_flagged(tmp_path):
+    """`import importer` must NOT match — dot-bounded prefix."""
+    f = tmp_path / "ok.py"
+    f.write_text("import importer\nfrom importer import x\n")
+    assert _pids(ast_scan_path(f)) == []
+
+
+def test_literal_dunder_import_not_flagged(tmp_path):
+    """__import__('os') with a literal is not flagged (regex catches those)."""
+    f = tmp_path / "ok.py"
+    f.write_text("m = __import__('os')\n")
+    assert "dynamic_import_computed" not in _pids(ast_scan_path(f))
+
+
+def test_non_python_file_returns_empty(tmp_path):
+    f = tmp_path / "script.sh"
+    f.write_text("import importlib\n")
+    assert ast_scan_path(f) == []
+
+
+def test_directory_scans_recursively_and_skips_cache_dirs(tmp_path):
+    skill = tmp_path / "s"
+    skill.mkdir()
+    (skill / "main.py").write_text("import importlib\n")
+    (skill / "sub").mkdir()
+    (skill / "sub" / "u.py").write_text("from importlib.util import find_spec\n")
+    for d in ("__pycache__", ".venv", "venv", "node_modules"):
+        ignored = skill / d
+        ignored.mkdir()
+        (ignored / "junk.py").write_text("import importlib\n")
+    pids = _pids(ast_scan_path(skill))
+    assert pids.count("importlib_import") == 2
+
+
+def test_missing_path_returns_empty(tmp_path):
+    assert ast_scan_path(tmp_path / "does_not_exist") == []
+
+
+def test_dynamic_getattr_and_dict_access_detected(tmp_path):
+    f = tmp_path / "g.py"
+    f.write_text("name = 'x'\nv = getattr(o, name)\nv = o.__dict__[name]\n")
+    pids = _pids(ast_scan_path(f))
+    assert "dynamic_getattr" in pids
+    assert "dict_access" in pids
+
+
+def test_format_report_empty():
+    assert "No dynamic" in format_ast_report([])
+
+
+def test_format_report_with_findings():
+    findings = [
+        ("a.py", 1, "importlib_import", "import importlib — ..."),
+        ("a.py", 3, "dynamic_import", "importlib.import_module() — ..."),
+    ]
+    out = format_ast_report(findings, skill_name="test")
+    assert "test" in out and "a.py" in out and "L1" in out and "L3" in out
+    assert "diagnostic hints" in out

tools/skills_ast_audit.py

@@ -0,0 +1,133 @@
+"""
+AST-level deep audit for skill Python files — opt-in diagnostic, not a security gate.
+
+Per SECURITY.md §2.4, Skills Guard is in-process heuristics ("useful — not
+boundaries"). This module is a separate opt-in diagnostic that flags dynamic
+import / dynamic attribute access patterns operators may want to eyeball when
+reviewing third-party skill code. Every pattern flagged here has legitimate
+uses; findings are hints for human review, not verdicts.
+
+CLI: ``hermes skills audit --deep``
+"""
+
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+from typing import List, Tuple
+
+# (file, line, pattern_id, description)
+Finding = Tuple[str, int, str, str]
+
+_IGNORED_DIRS = {"__pycache__", ".venv", "venv", "node_modules"}
+
+
+def _scan_source(content: str, rel_path: str) -> List[Finding]:
+    try:
+        tree = ast.parse(content)
+    except (SyntaxError, ValueError, RecursionError):
+        return []
+
+    findings: List[Finding] = []
+
+    class V(ast.NodeVisitor):
+        def visit_Call(self, node):
+            f = node.func
+            # importlib.import_module(...)
+            if isinstance(f, ast.Attribute) and f.attr == "import_module":
+                findings.append((rel_path, node.lineno, "dynamic_import",
+                                 "importlib.import_module() — loads arbitrary modules at runtime"))
+            # __import__(<computed>)
+            elif isinstance(f, ast.Name) and f.id == "__import__":
+                if node.args and not isinstance(node.args[0], ast.Constant):
+                    findings.append((rel_path, node.lineno, "dynamic_import_computed",
+                                     "__import__ with non-literal module name"))
+            # getattr(obj, <computed>)
+            elif isinstance(f, ast.Name) and f.id == "getattr":
+                if len(node.args) >= 2 and not isinstance(node.args[1], ast.Constant):
+                    findings.append((rel_path, node.lineno, "dynamic_getattr",
+                                     "getattr with non-literal attribute name"))
+            self.generic_visit(node)
+
+        def visit_Subscript(self, node):
+            # obj.__dict__[<computed>]
+            if (isinstance(node.value, ast.Attribute)
+                    and node.value.attr == "__dict__"
+                    and not isinstance(node.slice, ast.Constant)):
+                findings.append((rel_path, node.lineno, "dict_access",
+                                 "__dict__[<computed>] — dynamic attribute access"))
+            self.generic_visit(node)
+
+        def visit_Import(self, node):
+            for a in node.names:
+                if a.name == "importlib" or a.name.startswith("importlib."):
+                    findings.append((rel_path, node.lineno, "importlib_import",
+                                     f"import {a.name} — enables dynamic module loading"))
+            self.generic_visit(node)
+
+        def visit_ImportFrom(self, node):
+            m = node.module or ""
+            if m == "importlib" or m.startswith("importlib."):
+                findings.append((rel_path, node.lineno, "importlib_import",
+                                 f"from {m} import ... — enables dynamic module loading"))
+            self.generic_visit(node)
+
+    try:
+        V().visit(tree)
+    except (RecursionError, ValueError, RuntimeError):
+        # Hostile/pathological input: return what we collected so far.
+        pass
+
+    return findings
+
+
+def ast_scan_path(path: Path) -> List[Finding]:
+    """Scan a single .py file or recursively scan all .py under a directory.
+
+    Returns a list of (file, line, pattern_id, description) tuples. Empty for
+    non-Python paths, missing paths, or paths with no matching patterns.
+    """
+    if path.is_file():
+        if path.suffix.lower() != ".py":
+            return []
+        try:
+            content = path.read_text(encoding="utf-8", errors="replace")
+        except OSError:
+            return []
+        return _scan_source(content, path.name)
+
+    if not path.is_dir():
+        return []
+
+    out: List[Finding] = []
+    for py in sorted(path.rglob("*.py")):
+        if set(py.parent.parts) & _IGNORED_DIRS:
+            continue
+        try:
+            content = py.read_text(encoding="utf-8", errors="replace")
+        except OSError:
+            continue
+        try:
+            rel = py.relative_to(path).as_posix()
+        except ValueError:
+            rel = py.name
+        out.extend(_scan_source(content, rel))
+    return out
+
+
+def format_ast_report(findings: List[Finding], skill_name: str = "") -> str:
+    """Plain-text report (Rich-markup-free) grouped by file."""
+    header = f"AST deep scan: {skill_name}" if skill_name else "AST deep scan"
+    if not findings:
+        return f"{header}\n  No dynamic import/access patterns detected."
+
+    lines = [header, f"  {len(findings)} finding(s):"]
+    current = None
+    for f, line, pid, desc in sorted(findings):
+        if f != current:
+            current = f
+            lines.append(f"  {f}")
+        lines.append(f"    L{line}  {pid}  — {desc}")
+    lines.append("")
+    lines.append("  Note: diagnostic hints for human review, not security verdicts.")
+    return "\n".join(lines)