diff --git a/bandit.yaml b/bandit.yaml new file mode 100644 index 00000000..14a6772d --- /dev/null +++ b/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 diff --git a/scripts/audit_plugins.py b/scripts/audit_plugins.py new file mode 100644 index 00000000..79a61acf --- /dev/null +++ b/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()) diff --git a/scripts/generate_report.py b/scripts/generate_report.py new file mode 100644 index 00000000..753ea35b --- /dev/null +++ b/scripts/generate_report.py @@ -0,0 +1,356 @@ +#!/usr/bin/env python3 +""" +Security Report Generator + +Aggregates JSON output from all CI security audit jobs into a single +Markdown report suitable for PR comments and artifact storage. + +Expected artifact layout (from actions/download-artifact@v4): + / + sast-results/ + bandit-results.json + semgrep-results.json + dependency-audit-results/ + pip-audit-results.json + safety-results.json + secrets-scan-results/ + gitleaks-results.json + security-proofs-results/ + security-proofs-results.json + plugin-audit-results/ + plugin-audit-results.json + +Usage: + python scripts/generate_report.py --artifact-dir audit-artifacts/ --output report.md + python scripts/generate_report.py --artifact-dir audit-artifacts/ --output report.md --verbose +""" + +import argparse +import json +import sys +from pathlib import Path +from datetime import datetime, timezone + +PROJECT_ROOT = Path(__file__).resolve().parent.parent + +# Gitleaks matches exactly equal to one of these (not a substring match -- a +# real secret that merely contains one of these words as part of its actual +# value must still be reported) are known template placeholders. +_GITLEAKS_SUPPRESS_EXACT_VALUES = { + "YOUR_YOUTUBE_API_KEY", + "YOUR_YOUTUBE_CHANNEL_ID", + "YOUR_GITHUB_PERSONAL_ACCESS_TOKEN", +} + +# Findings in these files are suppressed regardless of value -- they are +# template/example files that are expected to only ever contain placeholders. +_GITLEAKS_SUPPRESS_PATHS = [ + "config_secrets.template.json", + "config.template.json", +] + + +# ───────────────────────────────────────────────────────────────────────────── +# Helpers +# ───────────────────────────────────────────────────────────────────────────── + +def _load(path: Path) -> tuple[dict | list | None, str | None]: + """Load a JSON artifact file. + + Returns (data, error): error is None on success (data is whatever was + parsed, which may legitimately be an empty list/dict for a clean scan); + otherwise error is a human-readable reason the artifact is unavailable, + distinguishing "missing/malformed artifact" from "valid empty result" so + callers don't silently treat a broken CI job as a clean pass. + """ + if not path.exists(): + return None, f"artifact not found: {path}" + try: + return json.loads(path.read_text(encoding="utf-8")), None + except (json.JSONDecodeError, OSError) as exc: + return None, f"could not read/parse {path}: {exc}" + + +def _md_sanitize_cell(value: object) -> str: + """Escape/normalize a value so scanner-controlled content (a matched + secret, a bandit issue_text, a file path) can't alter the Markdown + table's structure: pipes would add bogus columns, newlines would break + out of the row (or forge a fake header/separator line).""" + text = str(value) + text = text.replace("\\", "\\\\").replace("|", "\\|") + text = text.replace("\r\n", " ").replace("\n", " ").replace("\r", " ") + return text + + +def _md_table_row(*cells: str) -> str: + return "| " + " | ".join(_md_sanitize_cell(c) for c in cells) + " |" + + +# ───────────────────────────────────────────────────────────────────────────── +# Per-tool summarizers +# Returns: (markdown_lines: list[str], critical_count: int, available: bool) +# `available=False` means the artifact was missing or malformed -- distinct +# from a valid scan that simply found nothing -- so the caller can report +# INCOMPLETE instead of silently counting it as a clean pass. +# ───────────────────────────────────────────────────────────────────────────── + +def _summarize_bandit(artifact_dir: Path) -> tuple[list[str], int, bool]: + data, error = _load(artifact_dir / "sast-results" / "bandit-results.json") + if error: + return [f"_bandit results unavailable: {error}_"], 0, False + + results = data.get("results", []) + high = [r for r in results if r.get("issue_severity") == "HIGH"] + medium = [r for r in results if r.get("issue_severity") == "MEDIUM"] + low = [r for r in results if r.get("issue_severity") == "LOW"] + + lines = [ + f"**Bandit**: {len(high)} HIGH · {len(medium)} MEDIUM · {len(low)} LOW" + ] + + if high: + lines += [ + "", + "| Severity | File | Line | Issue |", + "| --- | --- | --- | --- |", + ] + for r in high[:10]: + fname = Path(r.get("filename", "")).name + lines.append(_md_table_row( + "HIGH", f"`{fname}`", + str(r.get("line_number", "?")), + r.get("issue_text", "") + )) + if len(high) > 10: + lines.append(f"_… and {len(high) - 10} more HIGH findings_") + + return lines, len(high), True + + +def _summarize_pip_audit(artifact_dir: Path) -> tuple[list[str], int, bool]: + data, error = _load(artifact_dir / "dependency-audit-results" / "pip-audit-results.json") + if error: + return [f"_pip-audit results unavailable: {error}_"], 0, False + + # pip-audit JSON format: {"dependencies": [{"name": ..., "vulns": [...]}]} + vulns: list[dict] = [] + for dep in data.get("dependencies", []): + for v in dep.get("vulns", []): + vulns.append({"package": dep.get("name", "?"), **v}) + + lines = [f"**pip-audit**: {len(vulns)} vulnerabilities found"] + + if vulns: + lines += ["", "| Package | ID | Fix |", "| --- | --- | --- |"] + for v in vulns[:10]: + fix = v.get("fix_versions", ["none"]) + fix_str = ", ".join(fix) if fix else "none" + lines.append(_md_table_row( + v.get("package", "?"), + v.get("id", "?"), + fix_str, + )) + + # Treat known vulnerabilities as warnings, not critical (they may be unavoidable) + return lines, 0, True + + +def _summarize_gitleaks(artifact_dir: Path) -> tuple[list[str], int, bool]: + data, error = _load(artifact_dir / "secrets-scan-results" / "gitleaks-results.json") + if error: + return [f"_gitleaks results unavailable: {error}_"], 0, False + + if not isinstance(data, list): + data = [] + + real_findings = [] + suppressed = 0 + for finding in data: + secret_val = str(finding.get("Secret", "") or finding.get("Match", "")) + file_name = Path(finding.get("File", "")).name + if (secret_val in _GITLEAKS_SUPPRESS_EXACT_VALUES + or file_name in _GITLEAKS_SUPPRESS_PATHS): + suppressed += 1 + else: + real_findings.append(finding) + + lines = [ + f"**Gitleaks**: {len(real_findings)} finding(s) " + f"({suppressed} suppressed as template placeholders)" + ] + + if real_findings: + lines += ["", "| Rule | File | Line | Description |", "| --- | --- | --- | --- |"] + for f in real_findings[:10]: + fname = Path(f.get("File", "")).name + lines.append(_md_table_row( + f.get("RuleID", "?"), + f"`{fname}`", + str(f.get("StartLine", "?")), + f.get("Description", ""), + )) + + critical = len(real_findings) # any real secret is critical + return lines, critical, True + + +def _summarize_security_proofs(artifact_dir: Path) -> tuple[list[str], int, bool]: + data, error = _load(artifact_dir / "security-proofs-results" / "security-proofs-results.json") + if error: + return [f"_security proofs results unavailable: {error}_"], 0, False + + if not isinstance(data, list): + data = [] + + critical = [r for r in data if r.get("severity") == "CRITICAL"] + warnings = [r for r in data if r.get("severity") == "WARNING"] + passed = [r for r in data if r.get("severity") == "PASS"] + skipped = [r for r in data if r.get("severity") == "SKIP"] + + lines = [ + f"**Security Proofs**: " + f"{len(passed)} PASS · {len(warnings)} WARN · " + f"{len(critical)} CRITICAL · {len(skipped)} SKIP", + "", + ] + + _icon = {"PASS": "✅", "INFO": "ℹ️", "WARNING": "⚠️", # nosec B105 - severity labels, not credentials + "CRITICAL": "🚨", "SKIP": "⏭️"} + for r in data: + icon = _icon.get(r.get("severity", ""), "❓") + lines.append( + f"- {icon} **{r.get('test_id', '?')}**: {r.get('message', '')}" + ) + if r.get("details") and r.get("severity") in ("CRITICAL", "WARNING"): + lines.append(f" - _{r['details']}_") + + return lines, len(critical), True + + +def _summarize_plugin_audit(artifact_dir: Path) -> tuple[list[str], int, bool]: + data, error = _load(artifact_dir / "plugin-audit-results" / "plugin-audit-results.json") + if error: + return [f"_plugin audit results unavailable: {error}_"], 0, False + + summary = data.get("summary", {}) + findings = data.get("findings", []) + critical_findings = [f for f in findings if f.get("severity") == "CRITICAL"] + warning_findings = [f for f in findings if f.get("severity") == "WARNING"] + + lines = [ + f"**Plugin Audit**: {data.get('plugins_scanned', '?')} plugins scanned — " + f"{summary.get('critical', 0)} CRITICAL · {summary.get('warnings', 0)} WARNINGS" + ] + + if critical_findings: + lines += ["", "| Plugin | File | Line | Rule | Message |", + "| --- | --- | --- | --- | --- |"] + for f in critical_findings[:10]: + fname = Path(f.get("file", "")).name + lines.append(_md_table_row( + f.get("plugin_id", "?"), + f"`{fname}`", + str(f.get("line", "?")), + f.get("rule", "?"), + f.get("message", ""), + )) + + if warning_findings and not critical_findings: + lines.append(f"\n_{len(warning_findings)} warning(s) found — see artifact for details_") + + return lines, summary.get("critical", 0), True + + +# ───────────────────────────────────────────────────────────────────────────── +# Main +# ───────────────────────────────────────────────────────────────────────────── + +def main() -> int: + parser = argparse.ArgumentParser( + description="Generate consolidated security audit report", + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("--artifact-dir", required=True, + help="Directory containing downloaded CI artifacts") + parser.add_argument("--output", "-o", required=True, + help="Output Markdown file path") + parser.add_argument("--verbose", "-v", action="store_true") + args = parser.parse_args() + + artifact_dir = Path(args.artifact_dir) + timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") + + bandit_lines, bandit_crit, bandit_ok = _summarize_bandit(artifact_dir) + pip_audit_lines, pip_audit_crit, pip_audit_ok = _summarize_pip_audit(artifact_dir) + gitleaks_lines, gitleaks_crit, gitleaks_ok = _summarize_gitleaks(artifact_dir) + proofs_lines, proofs_crit, proofs_ok = _summarize_security_proofs(artifact_dir) + plugins_lines, plugins_crit, plugins_ok = _summarize_plugin_audit(artifact_dir) + + unavailable_tools = [ + name for name, ok in [ + ("bandit", bandit_ok), ("pip-audit", pip_audit_ok), + ("gitleaks", gitleaks_ok), ("security-proofs", proofs_ok), + ("plugin-audit", plugins_ok), + ] if not ok + ] + + total_critical = bandit_crit + pip_audit_crit + gitleaks_crit + proofs_crit + plugins_crit + if unavailable_tools: + # A missing/malformed artifact means that tool's checks never + # actually ran -- this must not be reported as a clean PASS just + # because the *artifacts that did load* found nothing. + overall = "INCOMPLETE ⚠️" + elif total_critical > 0: + overall = "ACTION REQUIRED 🚨" + else: + overall = "PASSED ✅" + + def section(title: str, lines: list[str]) -> str: + return f"### {title}\n\n" + "\n".join(lines) + "\n" + + incomplete_note = ( + f"\n_⚠️ Incomplete: results unavailable for {', '.join(unavailable_tools)} " + f"— see the corresponding section(s) below for details_\n" + if unavailable_tools else "" + ) + + report = f"""## 🔒 Security Audit — {overall} + +_Generated: {timestamp}_ +{incomplete_note} +| Critical | High/Warn | Overall | +| :---: | :---: | :---: | +| {'🚨 ' + str(total_critical) if total_critical else '✅ 0'} | ⚠️ see below | {overall} | + +--- + +{section('SAST — Bandit', bandit_lines)} +{section('Dependencies — pip-audit', pip_audit_lines)} +{section('Secrets — Gitleaks', gitleaks_lines)} +{section('LEDMatrix Security Proofs', proofs_lines)} +{section('Plugin Security Audit', plugins_lines)} +--- + +_Total critical findings: **{total_critical}**_ +""" + + output_path = Path(args.output) + output_path.write_text(report, encoding="utf-8") + + if args.verbose: + print(f" Report written to: {output_path}") + print(f" Status: {overall}") + print(f" Critical findings: {total_critical}") + print(f" bandit={bandit_crit} pip-audit={pip_audit_crit} " + f"gitleaks={gitleaks_crit} proofs={proofs_crit} plugins={plugins_crit}") + if unavailable_tools: + print(f" Unavailable: {', '.join(unavailable_tools)}") + + if unavailable_tools: + return 1 + + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/prove_security.py b/scripts/prove_security.py new file mode 100644 index 00000000..9a5fbc88 --- /dev/null +++ b/scripts/prove_security.py @@ -0,0 +1,593 @@ +#!/usr/bin/env python3 +""" +LEDMatrix Security Proof Tests + +Automated proofs that run in CI to verify security properties hold on every +commit. Inspired by the Huntarr security review approach of using standard +tooling to confirm specific vulnerability classes are absent. + +Usage: + python scripts/prove_security.py + python scripts/prove_security.py --verbose + python scripts/prove_security.py --output results.json + +Exit code: 1 only if CRITICAL findings are detected. Warnings are reported +but do not block CI. +""" + +import ast +import argparse +import hashlib +import json +import re +import sys +from dataclasses import dataclass, asdict +from pathlib import Path + +PROJECT_ROOT = Path(__file__).resolve().parent.parent + + +# ───────────────────────────────────────────────────────────────────────────── +# Result dataclass +# ───────────────────────────────────────────────────────────────────────────── + +@dataclass +class TestResult: + test_id: str + severity: str # PASS | INFO | WARNING | CRITICAL | SKIP + message: str + details: str = "" + + def to_dict(self) -> dict: + return asdict(self) + + @property + def icon(self) -> str: + return { + "PASS": "✅", # nosec B105 - severity label, not a credential + "INFO": "ℹ️ ", + "WARNING": "⚠️ ", + "CRITICAL": "🚨", + "SKIP": "⏭️ ", + }.get(self.severity, "❓") + + +# ───────────────────────────────────────────────────────────────────────────── +# T1: Plugin Loading / Zip Slip +# ───────────────────────────────────────────────────────────────────────────── + +def test_t1a_zip_slip_protection() -> TestResult: + """ + Verify that zip-slip protection actually guards zip extraction in + store_manager.py. + + A whole-file substring check for "is_relative_to"/"Zip-slip detected" + would pass even if the guard existed somewhere unrelated, or covered + only one of several extract()/extractall() call sites. Instead, this + walks the AST: for every extract()/extractall() call, it confirms an + is_relative_to() check (and the "Zip-slip detected" log) appears + earlier in that same enclosing function -- validate-then-bulk-extract + (validate every member, then call extractall() only after all passed) + counts as protecting the call, since it covers the same member list. + """ + store_manager = PROJECT_ROOT / "src" / "plugin_system" / "store_manager.py" + if not store_manager.exists(): + return TestResult("T1a", "CRITICAL", + "store_manager.py not found", + f"Expected at {store_manager}") + + content = store_manager.read_text(encoding="utf-8") + try: + tree = ast.parse(content, filename=str(store_manager)) + except SyntaxError as exc: + return TestResult("T1a", "CRITICAL", + "store_manager.py could not be parsed", + str(exc)) + + extraction_sites = 0 + unprotected: list[str] = [] + + for func in ast.walk(tree): + if not isinstance(func, (ast.FunctionDef, ast.AsyncFunctionDef)): + continue + + extract_calls = [ + node for node in ast.walk(func) + if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) + and node.func.attr in ("extract", "extractall") + ] + if not extract_calls: + continue + extraction_sites += len(extract_calls) + + guard_lines = [ + n.lineno for n in ast.walk(func) + if isinstance(n, ast.Attribute) and n.attr == "is_relative_to" + ] + has_zip_slip_log = any( + isinstance(n, ast.Constant) and isinstance(n.value, str) + and "Zip-slip detected" in n.value + for n in ast.walk(func) + ) + + for call in extract_calls: + guarded = has_zip_slip_log and any(g < call.lineno for g in guard_lines) + if not guarded: + unprotected.append( + f"{func.name}() line {call.lineno}: {call.func.attr}() call not " + f"clearly preceded by an is_relative_to() guard + Zip-slip log " + f"in the same function" + ) + + if extraction_sites == 0: + return TestResult("T1a", "WARNING", + "No zipfile extract()/extractall() calls found in store_manager.py", + "Verify plugin installation no longer extracts zip archives, " + "or that this check still targets the right file") + + if unprotected: + return TestResult("T1a", "CRITICAL", + f"{len(unprotected)} of {extraction_sites} zip extraction " + f"call(s) not clearly guarded", + "; ".join(unprotected)) + + return TestResult("T1a", "PASS", + "Zip-slip protection verified", + f"All {extraction_sites} extract()/extractall() call(s) in " + f"store_manager.py are preceded by an is_relative_to() guard " + f"with a Zip-slip log in the same function") + + +def test_t1b_dangerous_plugin_calls() -> list[TestResult]: + """ + Scan plugin directories for dangerous function calls (eval, exec). + These represent arbitrary code execution risks in plugin code. + """ + results = [] + plugin_dirs = [ + PROJECT_ROOT / "plugins", + PROJECT_ROOT / "plugin-repos", + ] + + violations: list[str] = [] + files_scanned = 0 + + scan_errors: list[str] = [] + + for base in plugin_dirs: + if not base.exists(): + continue + for plugin_dir in sorted(base.iterdir()): + if not plugin_dir.is_dir() or plugin_dir.name.startswith(('.', '_')): + continue + for py_file in plugin_dir.rglob("*.py"): + files_scanned += 1 + try: + source = py_file.read_text(encoding="utf-8") + tree = ast.parse(source, filename=str(py_file)) + for node in ast.walk(tree): + if isinstance(node, ast.Call) and isinstance(node.func, ast.Name): + if node.func.id in ("eval", "exec"): + rel = py_file.relative_to(PROJECT_ROOT) + violations.append( + f"{rel}:{node.lineno} — {node.func.id}() call") + except (SyntaxError, OSError) as exc: + # A file we couldn't parse/read was never actually + # scanned for eval()/exec() -- that must block this + # test, not silently pass as if it were clean. + rel = py_file.relative_to(PROJECT_ROOT) + scan_errors.append(f"{rel} — {type(exc).__name__}: {exc}") + + if scan_errors: + results.append(TestResult( + "T1b", "CRITICAL", + f"{len(scan_errors)} plugin file(s) could not be scanned for eval()/exec()", + "; ".join(scan_errors[:10]) + )) + + if violations: + results.append(TestResult( + "T1b", "CRITICAL", + f"Dangerous function calls found in plugins ({len(violations)} instance(s))", + "; ".join(violations[:10]) + )) + elif not scan_errors: + results.append(TestResult( + "T1b", "PASS", + "No eval()/exec() calls found in plugins", + f"{files_scanned} plugin Python files scanned" + )) + + return results + + +# ───────────────────────────────────────────────────────────────────────────── +# T2: API Surface Inventory +# ───────────────────────────────────────────────────────────────────────────── + +def test_t2a_api_surface_inventory() -> TestResult: + """ + Document the API surface area. + + This app intentionally has no authentication (local-only Raspberry Pi + design, documented in web_interface/app.py). This test produces an + inventory for audit purposes and warns only if the design-intent comment + is removed from app.py (which would indicate someone deleted the rationale + without adding auth, rather than a deliberate undocumented change). + """ + api_file = PROJECT_ROOT / "web_interface" / "blueprints" / "api_v3.py" + app_file = PROJECT_ROOT / "web_interface" / "app.py" + + if not api_file.exists(): + return TestResult("T2a", "WARNING", "api_v3.py not found", str(api_file)) + + api_content = api_file.read_text(encoding="utf-8") + routes = re.findall(r"@api_v3\.route\('([^']+)'", api_content) + + csrf_documented = False + if app_file.exists(): + app_content = app_file.read_text(encoding="utf-8") + csrf_documented = "CSRF protection disabled for local-only" in app_content + + summary = ( + f"{len(routes)} API routes in api_v3.py. " + f"No auth decorators (intentional local-only design). " + f"CSRF disabled: {'YES — design intent documented in app.py' if csrf_documented else 'YES — but design intent comment NOT found in app.py'}. " + f"Rate limiting: 1000/min." + ) + + if not csrf_documented: + return TestResult( + "T2a", "WARNING", + "CSRF is disabled but the design-intent comment is missing from app.py", + "Add the rationale comment back, or add proper CSRF protection if " + "the app is now internet-facing" + ) + + # There is currently no config mechanism that actually enforces the + # local-only boundary the design-intent comment describes -- app.py + # hardcodes host='0.0.0.0' unconditionally, so nothing here can confirm + # this deployment is in fact LAN-only. Reporting this as mere INFO + # understates that: an unauthenticated, CSRF-disabled API surface is a + # real risk the moment this ever runs somewhere other than a home LAN, + # documented rationale or not. + return TestResult( + "T2a", "WARNING", + "API surface has no auth and CSRF disabled; enforcement of the " + "documented local-only boundary cannot be confirmed", + summary + ) + + +# ───────────────────────────────────────────────────────────────────────────── +# T3: Secrets & Credential Handling +# ───────────────────────────────────────────────────────────────────────────── + +# Patterns that suggest real credentials (must be >8 chars, not placeholders) +_SECRET_PATTERNS = [ + (r'(?i)password\s*=\s*["\'](?!none|empty|placeholder|example|test|default|""|'')[^"\']{8,}["\']', "WARNING", "password"), + (r'(?i)api[_-]?key\s*=\s*["\'](?!none|empty|placeholder|YOUR_|example|test)[^"\']{16,}["\']', "WARNING", "api_key"), + (r'(?i)secret\s*=\s*["\'](?!none|empty|placeholder|YOUR_|example|test)[^"\']{16,}["\']', "WARNING", "secret"), + # Real GitHub token pattern + (r'ghp_[a-zA-Z0-9]{36}', "CRITICAL", "github_token"), + # Generic long bearer tokens + (r'Bearer\s+[a-zA-Z0-9\-_\.]{32,}', "WARNING", "bearer_token"), +] + +_TEMPLATE_SKIP_STRINGS = [ + "YOUR_", "PLACEHOLDER", "_HERE", "example.com", "config_secrets.template", + "prove_security", # this file itself +] + +_SCAN_DIRS = ["src", "web_interface", "scripts"] + + +def test_t3a_hardcoded_secrets() -> TestResult: + """Scan source code for hardcoded credentials.""" + violations: list[str] = [] + + for dir_name in _SCAN_DIRS: + scan_dir = PROJECT_ROOT / dir_name + if not scan_dir.exists(): + continue + for py_file in scan_dir.rglob("*.py"): + # Skip test files and this script + if "test" in str(py_file).lower() or "prove_security" in str(py_file): + continue + try: + content = py_file.read_text(encoding="utf-8") + except OSError: + continue + + for pattern, severity, pattern_type in _SECRET_PATTERNS: + for match in re.finditer(pattern, content): + line_content = match.group(0) + # Skip lines containing template placeholder strings. + # line_content is only used for this in-memory check -- + # it must never be stored or included in output below. + if any(skip in line_content for skip in _TEMPLATE_SKIP_STRINGS): + continue + rel = py_file.relative_to(PROJECT_ROOT) + line_no = content[: match.start()].count("\n") + 1 + # Redacted fingerprint lets the same finding be recognized + # across scans without ever reporting the matched + # credential itself (which would otherwise get published + # into CI logs, JSON artifacts, and PR comments -- wider + # exposure than the original leak). + fingerprint = hashlib.sha256(line_content.encode()).hexdigest()[:12] + violations.append( + f"[{severity}] {rel}:{line_no} — {pattern_type} " + f"(fingerprint {fingerprint})" + ) + + critical_violations = [v for v in violations if "[CRITICAL]" in v] + if critical_violations: + return TestResult( + "T3a", "CRITICAL", + f"Hardcoded secrets found ({len(critical_violations)} critical)", + "; ".join(critical_violations[:5]) + ) + if violations: + return TestResult( + "T3a", "WARNING", + f"Potential hardcoded secrets found ({len(violations)} instance(s))", + "; ".join(violations[:5]) + ) + + return TestResult("T3a", "PASS", "No hardcoded secrets detected", + f"Scanned {', '.join(_SCAN_DIRS)}") + + +def test_t3b_plaintext_password_storage() -> TestResult: + """ + Check for user account password storage without hashing. + + The LEDMatrix app has no user account system, so this should produce INFO. + It would only CRITICAL if someone added user auth and stored passwords without hashing. + + We require all three of: a password *variable assignment or DB operation*, + a clear storage call (INSERT / db commit / ORM save), and no hashing lib present + — to avoid false positives from files that contain 'password' for WiFi handling + and '.save()' for image/file saving in unrelated functions. + """ + hashing_libs = ["bcrypt", "argon2", "pbkdf2", "scrypt", + "generate_password_hash", "hashpw", "make_password"] + # Patterns that indicate password being stored in a database / ORM context. + # Must be specific enough to avoid matching set.add(), file.save(), etc. + db_storage_patterns = ["INSERT INTO", "db.session", "session.add(", "session.commit(", "orm.save"] + + password_storage_found = False + + for dir_name in _SCAN_DIRS: + scan_dir = PROJECT_ROOT / dir_name + if not scan_dir.exists(): + continue + for py_file in scan_dir.rglob("*.py"): + try: + content = py_file.read_text(encoding="utf-8") + except OSError: + continue + # Require DB/ORM context specifically — not just any .save() call + if ("password" in content.lower() and + any(store in content for store in db_storage_patterns) and + not any(h in content for h in hashing_libs)): + password_storage_found = True + + if password_storage_found: + return TestResult( + "T3b", "CRITICAL", + "Potential plaintext password storage in database/ORM detected", + "Found password + database storage operations without a recognized hashing library" + ) + + return TestResult("T3b", "INFO", + "No plaintext password storage detected", + "App has no user account system — expected result") + + +# ───────────────────────────────────────────────────────────────────────────── +# T4: Path Traversal +# ───────────────────────────────────────────────────────────────────────────── + +def test_t4a_path_traversal() -> TestResult: + """ + Verify static file serving uses send_from_directory (safe) rather than + open() with user-supplied paths. Also checks for extractall() calls that + lack the is_relative_to() guard. + """ + issues: list[str] = [] + + app_file = PROJECT_ROOT / "web_interface" / "app.py" + if app_file.exists(): + content = app_file.read_text(encoding="utf-8") + # The file-serve route should use send_from_directory or commonpath + if "send_from_directory" not in content and "commonpath" not in content: + issues.append("app.py: file-serve routes may not use send_from_directory/commonpath") + + # Check all extractall() calls have a preceding is_relative_to guard + for py_file in (PROJECT_ROOT / "src").rglob("*.py"): + try: + content = py_file.read_text(encoding="utf-8") + except OSError: + continue + if "extractall(" in content and "is_relative_to" not in content: + rel = py_file.relative_to(PROJECT_ROOT) + issues.append(f"{rel}: extractall() without is_relative_to() guard") + + if issues: + return TestResult( + "T4a", "WARNING", + f"Potential path traversal patterns found ({len(issues)})", + "; ".join(issues) + ) + + return TestResult("T4a", "PASS", + "Path traversal mitigations verified", + "send_from_directory/commonpath used for file serving; " + "extractall() calls have is_relative_to() guards") + + +# ───────────────────────────────────────────────────────────────────────────── +# T5: Auth Bypass Patterns +# ───────────────────────────────────────────────────────────────────────────── + +def test_t5a_auth_bypass_patterns() -> TestResult: + """ + Look for broken auth bypass patterns — not the intentional no-auth design + (T2a covers that), but patterns that suggest auth was INTENDED to exist + but has an exploitable bypass: broad substring matching, debug-mode skips, + or if-True conditions. + """ + bypass_signals = [ + (r'if\s+True\s*:', "if True: bypass"), + (r'if\s+debug\s*:', "debug-mode auth skip"), + (r'request\.path\s+in\s+', "substring path matching in auth (Huntarr pattern)"), + (r'EXEMPT_ROUTES\s*=', "exempt routes list"), + ] + + findings: list[str] = [] + + for dir_name in ["src", "web_interface"]: + scan_dir = PROJECT_ROOT / dir_name + if not scan_dir.exists(): + continue + for py_file in scan_dir.rglob("*.py"): + try: + content = py_file.read_text(encoding="utf-8") + except OSError: + continue + for pattern, label in bypass_signals: + if re.search(pattern, content): + # Only flag if the file also contains auth-related terms + if any(auth in content.lower() for auth in + ["auth", "login", "authenticate", "token", "permission"]): + rel = py_file.relative_to(PROJECT_ROOT) + findings.append(f"{rel}: {label}") + + if findings: + return TestResult( + "T5a", "WARNING", + f"Potential auth bypass patterns found ({len(findings)})", + "; ".join(findings[:5]) + ) + + return TestResult("T5a", "PASS", + "No auth bypass patterns detected", + "Checked src/ and web_interface/ for bypass signals") + + +# ───────────────────────────────────────────────────────────────────────────── +# T6: Docker / Container Hardening +# ───────────────────────────────────────────────────────────────────────────── + +def test_t6_docker_hardening() -> TestResult: + """Container security — skipped if no Dockerfile exists.""" + dockerfile = PROJECT_ROOT / "Dockerfile" + if not dockerfile.exists(): + return TestResult("T6", "SKIP", + "No Dockerfile found — container security scan not applicable", + "If Docker support is added in future, enable hadolint/trivy scanning " + "in .github/workflows/security-audit.yml") + + content = dockerfile.read_text(encoding="utf-8") + issues: list[str] = [] + + # Check for non-root USER directive + user_lines = [l for l in content.splitlines() if l.strip().startswith("USER")] + if not user_lines or user_lines[-1].strip() == "USER root": + issues.append("Container runs as root — use USER directive to drop privileges") + + # Check for pinned base image tags. A tag (even a specific version, not + # just :latest) is mutable -- the same tag can point to a different + # image later. Only a @sha256 digest is truly immutable/reproducible. + from_lines = [line for line in content.splitlines() if line.strip().startswith("FROM")] + for from_line in from_lines: + parts = from_line.split() + # FROM [--platform=] [AS ] -- skip an + # optional --platform= flag so it's never mistaken for the image + # token itself (which would falsely report it as unpinned). + image_parts = [p for p in parts[1:] if not p.startswith("--platform=")] + if image_parts: + image = image_parts[0] + if "@sha256:" not in image: + issues.append(f"Base image not pinned to a digest: {image}") + + if issues: + return TestResult("T6", "WARNING", + f"Dockerfile hardening issues ({len(issues)})", + "; ".join(issues)) + + return TestResult("T6", "PASS", "Dockerfile hardening checks passed", "") + + +# ───────────────────────────────────────────────────────────────────────────── +# Runner +# ───────────────────────────────────────────────────────────────────────────── + +def main() -> int: + parser = argparse.ArgumentParser( + description="LEDMatrix security proof tests", + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("--output", "-o", default=None, + help="Write JSON results to this file") + parser.add_argument("--verbose", "-v", action="store_true", + help="Show details for each check") + args = parser.parse_args() + + print("=" * 60) + print("LEDMatrix Security Proof Tests") + print(f"Project root: {PROJECT_ROOT}") + print("=" * 60) + + all_results: list[TestResult] = [] + + # Run all test groups + all_results.append(test_t1a_zip_slip_protection()) + all_results.extend(test_t1b_dangerous_plugin_calls()) + all_results.append(test_t2a_api_surface_inventory()) + all_results.append(test_t3a_hardcoded_secrets()) + all_results.append(test_t3b_plaintext_password_storage()) + all_results.append(test_t4a_path_traversal()) + all_results.append(test_t5a_auth_bypass_patterns()) + all_results.append(test_t6_docker_hardening()) + + # Print results + print() + for r in all_results: + line = f" {r.icon} [{r.severity:<8}] {r.test_id}: {r.message}" + print(line) + if args.verbose and r.details: + print(f" {r.details}") + + # Tally + critical = [r for r in all_results if r.severity == "CRITICAL"] + warnings = [r for r in all_results if r.severity == "WARNING"] + passed = [r for r in all_results if r.severity == "PASS"] + skipped = [r for r in all_results if r.severity == "SKIP"] + + print() + print(f" Results: {len(passed)} PASS {len(warnings)} WARN " + f"{len(critical)} CRITICAL {len(skipped)} SKIP") + + # Write JSON output + if args.output: + output_data = [r.to_dict() for r in all_results] + Path(args.output).write_text( + json.dumps(output_data, indent=2), encoding="utf-8" + ) + print(f" Results written to: {args.output}") + + if critical: + print(f"\n 🚨 {len(critical)} CRITICAL issue(s) found — blocking") + return 1 + + if warnings: + print(f"\n ⚠️ {len(warnings)} warning(s) found — non-blocking") + + print("\n ✅ All checks passed (warnings are non-blocking)") + return 0 + + +if __name__ == "__main__": + sys.exit(main())