LEDMatrix/scripts/audit_plugins.py

#!/usr/bin/env python3
"""
LEDMatrix Plugin Security Auditor

Performs AST-based security analysis of all Python files in plugin directories.
Designed to run in CI — exits non-zero on CRITICAL findings only.

Usage:
    python scripts/audit_plugins.py
    python scripts/audit_plugins.py --verbose
    python scripts/audit_plugins.py --plugin hello-world
    python scripts/audit_plugins.py --output results.json
"""

import ast
import argparse
import json
import sys
from dataclasses import dataclass, asdict
from pathlib import Path
from datetime import datetime, timezone

PROJECT_ROOT = Path(__file__).resolve().parent.parent

PLUGIN_BASE_DIRS = [
    PROJECT_ROOT / "plugins",
    PROJECT_ROOT / "plugin-repos",
]


# ─────────────────────────────────────────────────────────────────────────────
# Finding dataclass
# ─────────────────────────────────────────────────────────────────────────────

@dataclass
class Finding:
    plugin_id: str
    file: str
    line: int
    severity: str   # CRITICAL | WARNING | INFO
    rule: str
    message: str

    def to_dict(self) -> dict:
        return asdict(self)


# ─────────────────────────────────────────────────────────────────────────────
# AST visitor
# ─────────────────────────────────────────────────────────────────────────────

class _PluginVisitor(ast.NodeVisitor):
    """Collect security findings from a single plugin Python file."""

    def __init__(self, filepath: Path, plugin_id: str):
        self.filepath = filepath
        self.plugin_id = plugin_id
        self.findings: list[Finding] = []
        # Local name -> real dotted path, so aliased imports and from-imports
        # of dangerous APIs (import subprocess as sp; from builtins import
        # eval as e) are still recognized in visit_Call below.
        self._aliases: dict[str, str] = {}

    def _add(self, node: ast.AST, severity: str, rule: str, message: str) -> None:
        self.findings.append(Finding(
            plugin_id=self.plugin_id,
            file=str(self.filepath.relative_to(PROJECT_ROOT)),
            line=getattr(node, "lineno", 0),
            severity=severity,
            rule=rule,
            message=message,
        ))

    def _resolve(self, local_name: str) -> str:
        """Resolve a local name through recorded import aliases to its real
        dotted path (e.g. "sp" -> "subprocess"); unresolved names pass through
        unchanged."""
        return self._aliases.get(local_name, local_name)

    def _resolve_call_target(self, func: ast.expr) -> str | None:
        """Resolve a Call's func node to a fully-qualified dotted target,
        covering a direct name (bare builtin, aliased import, or
        from-import: from builtins import eval as e; from subprocess
        import run; from os import system as s) and module-attribute
        access (subprocess.run, sp.run, os.system, o.system) uniformly.
        Returns None for call shapes this doesn't attempt to resolve."""
        if isinstance(func, ast.Name):
            return self._resolve(func.id)
        if isinstance(func, ast.Attribute) and isinstance(func.value, ast.Name):
            base = self._resolve(func.value.id)
            return f"{base}.{func.attr}"
        return None

    def visit_Call(self, node: ast.Call) -> None:
        target = self._resolve_call_target(node.func)
        if target is None:
            self.generic_visit(node)
            return

        leaf = target.rsplit(".", 1)[-1]

        # eval() / exec() / compile() — arbitrary code execution, whether a
        # bare call, an aliased import, or a from-import
        # (from builtins import eval as e; e(...))
        if leaf == "eval":
            self._add(node, "CRITICAL", "PLUGIN-001",
                      "eval() call — arbitrary code execution risk")
        elif leaf == "exec":
            self._add(node, "CRITICAL", "PLUGIN-002",
                      "exec() call — arbitrary code execution risk")
        elif leaf == "compile":
            self._add(node, "WARNING", "PLUGIN-003",
                      "compile() call — dynamic code compilation")

        # subprocess.*(shell=True), whether subprocess.run(...), sp.run(...),
        # or a from-import (from subprocess import run; run(..., shell=True))
        if target in {
            "subprocess.run", "subprocess.call", "subprocess.Popen",
            "subprocess.check_call", "subprocess.check_output",
        }:
            for kw in node.keywords:
                if (kw.arg == "shell" and
                        isinstance(kw.value, ast.Constant) and
                        kw.value.value is True):
                    self._add(node, "WARNING", "PLUGIN-004",
                              f"subprocess.{leaf}(shell=True) — "
                              f"shell injection risk if args include user input")

        # os.system(), whether os.system(...), o.system(...), or a
        # from-import (from os import system as s; s(...))
        if target == "os.system":
            self._add(node, "WARNING", "PLUGIN-005",
                      "os.system() call — prefer subprocess with list args")

        self.generic_visit(node)

    def visit_Import(self, node: ast.Import) -> None:
        for alias in node.names:
            if alias.asname:
                local, real = alias.asname, alias.name
            else:
                # `import os.path` binds the top-level name `os`, not `os.path`
                local = real = alias.name.split(".")[0]
            self._aliases[local] = real
            self._check_import(node, alias.name)
        self.generic_visit(node)

    def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
        if node.module:
            for alias in node.names:
                local = alias.asname or alias.name
                self._aliases[local] = f"{node.module}.{alias.name}"
            self._check_import(node, node.module)
        self.generic_visit(node)

    def _check_import(self, node: ast.AST, module_name: str) -> None:
        dangerous = {
            "ctypes": ("WARNING", "PLUGIN-010", "ctypes import — native code execution"),
            "cffi": ("WARNING", "PLUGIN-011", "cffi import — native code execution"),
            "pickle": ("WARNING", "PLUGIN-012",
                       "pickle import — deserialization can execute arbitrary code"),
            "marshal": ("WARNING", "PLUGIN-013",
                        "marshal import — deserialization risk"),
        }
        for mod, (severity, rule, msg) in dangerous.items():
            if module_name == mod or module_name.startswith(mod + "."):
                self._add(node, severity, rule, msg)


# ─────────────────────────────────────────────────────────────────────────────
# Per-plugin audit
# ─────────────────────────────────────────────────────────────────────────────

def audit_plugin(plugin_dir: Path) -> list[Finding]:
    """Audit a single plugin directory. Returns all findings."""
    findings: list[Finding] = []
    plugin_id = plugin_dir.name

    # Check for required files
    for required_file, rule, msg in [
        ("manifest.json", "PLUGIN-020",
         "manifest.json missing — plugin may be incomplete"),
        ("config_schema.json", "PLUGIN-021",
         "config_schema.json missing — no input validation schema declared"),
    ]:
        if not (plugin_dir / required_file).exists():
            findings.append(Finding(
                plugin_id=plugin_id,
                file=str((plugin_dir / required_file).relative_to(PROJECT_ROOT)),
                line=0,
                severity="WARNING",
                rule=rule,
                message=msg,
            ))

    # AST analysis of all Python files
    for py_file in sorted(plugin_dir.rglob("*.py")):
        try:
            source = py_file.read_text(encoding="utf-8")
            tree = ast.parse(source, filename=str(py_file))
            visitor = _PluginVisitor(py_file, plugin_id)
            visitor.visit(tree)
            findings.extend(visitor.findings)
        except SyntaxError as exc:
            # A file the visitor can't even parse is a file we can't verify
            # is safe -- this must block the audit, not just warn.
            findings.append(Finding(
                plugin_id=plugin_id,
                file=str(py_file.relative_to(PROJECT_ROOT)),
                line=getattr(exc, "lineno", 0) or 0,
                severity="CRITICAL",
                rule="PLUGIN-030",
                message=f"Python syntax error — cannot be parsed: {exc}",
            ))
        except OSError as exc:
            # Same reasoning as SyntaxError: an unreadable file was never
            # actually scanned, so it must block rather than pass silently.
            findings.append(Finding(
                plugin_id=plugin_id,
                file=str(py_file.relative_to(PROJECT_ROOT)),
                line=0,
                severity="CRITICAL",
                rule="PLUGIN-031",
                message=f"Could not read file: {exc}",
            ))

    return findings


# ─────────────────────────────────────────────────────────────────────────────
# Main
# ─────────────────────────────────────────────────────────────────────────────

def main() -> int:
    parser = argparse.ArgumentParser(
        description="LEDMatrix plugin security auditor",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("--plugin", "-p", default=None,
                        help="Audit a specific plugin ID only")
    parser.add_argument("--output", "-o", default=None,
                        help="Write JSON results to this file")
    parser.add_argument("--verbose", "-v", action="store_true",
                        help="Show all findings, not just summary")
    args = parser.parse_args()

    print("=" * 60)
    print("LEDMatrix Plugin Security Audit")
    print(f"Project root: {PROJECT_ROOT}")
    print("=" * 60)

    all_findings: list[Finding] = []
    plugins_scanned = 0
    plugin_found = args.plugin is None

    for base_dir in PLUGIN_BASE_DIRS:
        if not base_dir.exists():
            if args.verbose:
                print(f"  ⏭️  Skipping {base_dir.name}/ (directory not found)")
            continue

        base_label = base_dir.relative_to(PROJECT_ROOT)
        print(f"\n  Scanning {base_label}/")

        for plugin_dir in sorted(base_dir.iterdir()):
            if not plugin_dir.is_dir():
                continue
            if plugin_dir.name.startswith((".", "_")):
                continue
            if args.plugin and plugin_dir.name != args.plugin:
                continue
            if args.plugin:
                plugin_found = True

            findings = audit_plugin(plugin_dir)
            all_findings.extend(findings)
            plugins_scanned += 1

            critical = [f for f in findings if f.severity == "CRITICAL"]
            warnings = [f for f in findings if f.severity == "WARNING"]

            if critical:
                icon, label = "🚨", "CRITICAL"
            elif warnings:
                icon, label = "⚠️ ", "WARN   "
            else:
                icon, label = "✅", "PASS   "

            print(f"    {icon} [{label}] {plugin_dir.name}"
                  f" — {len(critical)} critical, {len(warnings)} warnings")

            if args.verbose:
                for f in findings:
                    severity_icon = {"CRITICAL": "🚨", "WARNING": "⚠️ ", "INFO": "ℹ️ "}.get(
                        f.severity, "  "
                    )
                    print(f"           {severity_icon} {f.rule} {f.file}:{f.line} — {f.message}")

    if args.plugin and not plugin_found:
        print(f"\n  🚨 Plugin '{args.plugin}' not found in any of "
              f"{[str(d.relative_to(PROJECT_ROOT)) for d in PLUGIN_BASE_DIRS]} — "
              f"nothing was audited")
        return 1

    # Summary
    critical_findings = [f for f in all_findings if f.severity == "CRITICAL"]
    warning_findings = [f for f in all_findings if f.severity == "WARNING"]

    print(f"\n{'=' * 60}")
    print(f"  Plugins scanned : {plugins_scanned}")
    print(f"  CRITICAL        : {len(critical_findings)}")
    print(f"  WARNING         : {len(warning_findings)}")

    if critical_findings:
        print("\n  🚨 CRITICAL findings:")
        for f in critical_findings:
            print(f"    {f.plugin_id} | {Path(f.file).name}:{f.line} | {f.message}")

    # Write JSON output
    if args.output:
        output_data = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "plugins_scanned": plugins_scanned,
            "summary": {
                "critical": len(critical_findings),
                "warnings": len(warning_findings),
            },
            "findings": [f.to_dict() for f in all_findings],
        }
        Path(args.output).write_text(
            json.dumps(output_data, indent=2), encoding="utf-8"
        )
        print(f"\n  Results written to: {args.output}")

    if critical_findings:
        print("\n  🚨 Blocking — CRITICAL issues must be resolved")
        return 1

    print("\n  ✅ No critical issues found")
    return 0


if __name__ == "__main__":
    sys.exit(main())