diff --git a/hermes_cli/commands.py b/hermes_cli/commands.py index 815fb3caa007f..f589248621c56 100644 --- a/hermes_cli/commands.py +++ b/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 / for multiple skills)", "Tools & Skills"), CommandDef("cron", "Manage scheduled tasks", "Tools & Skills", diff --git a/hermes_cli/main.py b/hermes_cli/main.py index dbd80b5d40733..6af77204478b0 100644 --- a/hermes_cli/main.py +++ b/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" diff --git a/hermes_cli/skills_hub.py b/hermes_cli/skills_hub.py index b054070516560..5d39b5202f4da 100644 --- a/hermes_cli/skills_hub.py +++ b/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: diff --git a/tests/tools/test_skills_ast_audit.py b/tests/tools/test_skills_ast_audit.py new file mode 100644 index 0000000000000..c70d6a1f41c47 --- /dev/null +++ b/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 diff --git a/tools/skills_ast_audit.py b/tools/skills_ast_audit.py new file mode 100644 index 0000000000000..e127556c1d9ee --- /dev/null +++ b/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__() + 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, ) + 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__[] + 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__[] — 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)