chore(ci): add security-audit tooling scripts (workflow files pushed separately)

bandit.yaml

@@ -0,0 +1,29 @@
+# bandit.yaml — LEDMatrix bandit configuration
+# https://bandit.readthedocs.io/en/latest/config.html
+#
+# Skips are justified by the specific codebase context documented below.
+# Do not remove skips without updating the justification comment.
+
+skips:
+  # B104: Binding to all interfaces (0.0.0.0)
+  # Intentional — the Flask server binds 0.0.0.0 for LAN access on a Raspberry Pi.
+  # This is not internet-facing and is documented in web_interface/app.py.
+  - B104
+
+  # B603: subprocess call without shell=True
+  # All subprocess.run() calls in this codebase use list arguments (confirmed by
+  # grep — zero uses of shell=True in src/ or web_interface/). List args prevent
+  # shell injection. See src/common/permission_utils.py for the primary usage.
+  - B603
+
+  # B607: Starting a process with a partial executable path
+  # The subprocess calls invoke system utilities (systemctl, sudo, git) by name.
+  # These are fixed-list invocations, not user-controlled, and rely on PATH.
+  - B607
+
+exclude_dirs:
+  - tests
+  - test
+  - venv
+  - .venv
+  - rpi-rgb-led-matrix-master

scripts/audit_plugins.py

@@ -0,0 +1,344 @@
+#!/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())