From 32b6249d179b45c83f64cd3caf196ea085f26c57 Mon Sep 17 00:00:00 2001 From: ChuckBuilds Date: Tue, 14 Jul 2026 16:30:19 -0400 Subject: [PATCH 1/4] chore(ci): add security-audit workflow and plugin security-proof scripts - scripts/prove_security.py, audit_plugins.py, generate_report.py -- automated checks (dangerous eval()/exec() calls, dependency scanning, report generation) for plugins. - .github/workflows/security-audit.yml + bandit.yaml -- CI wiring for the above plus gitleaks secret scanning and bandit static analysis. - .github/workflows/tests.yml -- pytest matrix across Python 3.10-3.12. Also fixes two Codacy findings while these files are freshly landing: - prove_security.py: dropped a pointless f-string prefix with no placeholders. - security-audit.yml: pinned gitleaks/gitleaks-action to a full commit SHA (matching this repo's existing pinning convention in test.yml) instead of the floating v2 tag. Split out of the original chore/dead-code-removal commit, which had accidentally bundled this in alongside unrelated dead-code deletions. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01KEZK1P1Q1fu5pcuVrkrCFZ --- .github/workflows/security-audit.yml | 277 +++++++++++++++ .github/workflows/tests.yml | 46 +++ bandit.yaml | 33 ++ scripts/audit_plugins.py | 296 ++++++++++++++++ scripts/generate_report.py | 297 ++++++++++++++++ scripts/prove_security.py | 504 +++++++++++++++++++++++++++ 6 files changed, 1453 insertions(+) create mode 100644 .github/workflows/security-audit.yml create mode 100644 .github/workflows/tests.yml create mode 100644 bandit.yaml create mode 100644 scripts/audit_plugins.py create mode 100644 scripts/generate_report.py create mode 100644 scripts/prove_security.py diff --git a/.github/workflows/security-audit.yml b/.github/workflows/security-audit.yml new file mode 100644 index 00000000..bd78ca4a --- /dev/null +++ b/.github/workflows/security-audit.yml @@ -0,0 +1,277 @@ +name: Security Audit + +on: + push: + branches: + - main + - 'feat/**' + pull_request: + branches: + - main + schedule: + # Weekly full scan — catches new CVEs in existing deps + - cron: '0 6 * * 1' + workflow_dispatch: + +permissions: + contents: read + pull-requests: write + security-events: write + +jobs: + + # ───────────────────────────────────────────────────────────────────────── + # SAST — Static Application Security Testing + # ───────────────────────────────────────────────────────────────────────── + sast: + name: Static Analysis (bandit + semgrep) + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: '3.11' + + - name: Install SAST tools + run: pip install bandit==1.8.3 semgrep + + # Bandit — Python-specific security linter + # --exit-zero: findings are warnings, not CI blockers. + # The security-report job interprets severity. + - name: Run bandit + run: | + bandit -r src/ web_interface/ \ + -c bandit.yaml \ + -f json \ + -o bandit-results.json \ + --exit-zero + + # Semgrep — broader pattern-based analysis + # || true: prevents network/rate-limit errors from blocking the workflow + - name: Run semgrep + run: | + semgrep --config "p/python" \ + --config "p/flask" \ + --json \ + --output semgrep-results.json \ + src/ web_interface/ \ + || true + + - name: Upload SAST artifacts + uses: actions/upload-artifact@v4 + if: always() + with: + name: sast-results + path: | + bandit-results.json + semgrep-results.json + retention-days: 30 + + + # ───────────────────────────────────────────────────────────────────────── + # Dependency Vulnerability Scanning + # ───────────────────────────────────────────────────────────────────────── + dependency-audit: + name: Dependency Audit (pip-audit + safety) + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: '3.11' + + - name: Install audit tools + run: pip install pip-audit safety + + # Install project deps. Hardware-specific packages (rgbmatrix) will fail + # to build on Ubuntu runners — || true handles this gracefully. + # pip-audit operates on installed packages; partial install is acceptable. + - name: Install project dependencies + run: | + pip install -r requirements.txt || true + pip install -r web_interface/requirements.txt || true + pip install -r requirements-emulator.txt || true + + - name: Run pip-audit + run: | + pip-audit --format json --output pip-audit-results.json || true + + - name: Run safety check + run: | + safety check --output json > safety-results.json 2>&1 || true + + - name: Upload dependency audit artifacts + uses: actions/upload-artifact@v4 + if: always() + with: + name: dependency-audit-results + path: | + pip-audit-results.json + safety-results.json + retention-days: 30 + + + # ───────────────────────────────────────────────────────────────────────── + # Secrets Detection + # ───────────────────────────────────────────────────────────────────────── + secrets-scan: + name: Secrets Scan (gitleaks) + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + with: + fetch-depth: 0 # Full history for scanning all commits + + # continue-on-error: config/config_secrets.template.json contains + # placeholder strings (YOUR_*) that may trigger gitleaks rules. + # The generate_report.py script suppresses these false positives. + - name: Run gitleaks + uses: gitleaks/gitleaks-action@ff98106e4c7b2bc287b24eaf42907196329070c7 # v2.3.9 + continue-on-error: true + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + + - name: Upload secrets scan artifacts + uses: actions/upload-artifact@v4 + if: always() + with: + name: secrets-scan-results + path: results.sarif + retention-days: 30 + + + # ───────────────────────────────────────────────────────────────────────── + # LEDMatrix-Specific Security Proofs + # ───────────────────────────────────────────────────────────────────────── + ledmatrix-security-proofs: + name: LEDMatrix Security Proofs + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: '3.11' + + - name: Install dependencies + run: pip install -r requirements.txt || true + + # Script exits 1 only on CRITICAL findings. + # Warnings are reported but do not block the workflow. + - name: Run security proofs + run: | + python scripts/prove_security.py \ + --output security-proofs-results.json \ + --verbose + + - name: Upload proofs artifacts + uses: actions/upload-artifact@v4 + if: always() + with: + name: security-proofs-results + path: security-proofs-results.json + retention-days: 30 + + + # ───────────────────────────────────────────────────────────────────────── + # Plugin Security Audit + # ───────────────────────────────────────────────────────────────────────── + plugin-audit: + name: Plugin Security Audit + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: '3.11' + + # Script exits 1 only on CRITICAL findings (eval/exec in plugins). + # Missing manifest.json etc are warnings. + - name: Run plugin audit + run: | + python scripts/audit_plugins.py \ + --output plugin-audit-results.json \ + --verbose + + - name: Upload plugin audit artifacts + uses: actions/upload-artifact@v4 + if: always() + with: + name: plugin-audit-results + path: plugin-audit-results.json + retention-days: 30 + + + # ───────────────────────────────────────────────────────────────────────── + # Aggregate Report + # ───────────────────────────────────────────────────────────────────────── + security-report: + name: Security Report + runs-on: ubuntu-latest + needs: + - sast + - dependency-audit + - secrets-scan + - ledmatrix-security-proofs + - plugin-audit + if: always() # Run even if upstream jobs fail or are skipped + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: '3.11' + + - name: Download all artifacts + uses: actions/download-artifact@v4 + with: + path: audit-artifacts/ + + - name: Generate consolidated report + run: | + python scripts/generate_report.py \ + --artifact-dir audit-artifacts/ \ + --output security-report.md \ + --verbose + + - name: Upload consolidated report + uses: actions/upload-artifact@v4 + with: + name: security-report + path: security-report.md + retention-days: 90 + + - name: Comment on PR + if: github.event_name == 'pull_request' + uses: actions/github-script@v7 + with: + script: | + const fs = require('fs'); + const report = fs.readFileSync('security-report.md', 'utf8'); + // Use sticky comment — update existing comment rather than adding a new one each run + const { data: comments } = await github.rest.issues.listComments({ + issue_number: context.issue.number, + owner: context.repo.owner, + repo: context.repo.repo, + }); + const botComment = comments.find(c => + c.user.type === 'Bot' && c.body.includes('🔒 Security Audit') + ); + if (botComment) { + await github.rest.issues.updateComment({ + comment_id: botComment.id, + owner: context.repo.owner, + repo: context.repo.repo, + body: report, + }); + } else { + await github.rest.issues.createComment({ + issue_number: context.issue.number, + owner: context.repo.owner, + repo: context.repo.repo, + body: report, + }); + } diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 00000000..c59cac7d --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,46 @@ +name: Tests + +on: + push: + branches: [main] + pull_request: + branches: [main] + workflow_dispatch: + +permissions: + contents: read + +jobs: + test: + name: pytest (Python ${{ matrix.python-version }}) + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: ['3.10', '3.11', '3.12'] + steps: + - name: Check out repository + uses: actions/checkout@v4 + with: + submodules: false # rgbmatrix submodule not needed in EMULATOR mode + + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + cache: pip + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -r requirements.txt + # Optional deps that some test modules import + pip install scipy psutil Flask-Limiter + + - name: Run tests + env: + EMULATOR: "true" + run: | + pytest \ + -m "not hardware and not slow" \ + --tb=short diff --git a/bandit.yaml b/bandit.yaml new file mode 100644 index 00000000..abc7b273 --- /dev/null +++ b/bandit.yaml @@ -0,0 +1,33 @@ +# 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 + # prove_security.py intentionally contains detection patterns as string literals + # (e.g. "eval(", "exec(") to search for in other files — bandit would flag + # these as false positives. + - scripts/prove_security.py diff --git a/scripts/audit_plugins.py b/scripts/audit_plugins.py new file mode 100644 index 00000000..b2a92c14 --- /dev/null +++ b/scripts/audit_plugins.py @@ -0,0 +1,296 @@ +#!/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] = [] + + 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 visit_Call(self, node: ast.Call) -> None: + # eval() / exec() — arbitrary code execution + if isinstance(node.func, ast.Name): + if node.func.id == "eval": + self._add(node, "CRITICAL", "PLUGIN-001", + "eval() call — arbitrary code execution risk") + elif node.func.id == "exec": + self._add(node, "CRITICAL", "PLUGIN-002", + "exec() call — arbitrary code execution risk") + elif node.func.id == "compile": + self._add(node, "WARNING", "PLUGIN-003", + "compile() call — dynamic code compilation") + + # subprocess.*(shell=True) + if isinstance(node.func, ast.Attribute): + is_subprocess = ( + isinstance(node.func.value, ast.Name) and + node.func.value.id == "subprocess" and + node.func.attr in ("run", "call", "Popen", "check_call", "check_output") + ) + if is_subprocess: + 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.{node.func.attr}(shell=True) — " + f"shell injection risk if args include user input") + + # os.system() — shell execution + is_os_system = ( + isinstance(node.func.value, ast.Name) and + node.func.value.id == "os" and + node.func.attr == "system" + ) + if is_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: + self._check_import(node, alias.name) + self.generic_visit(node) + + def visit_ImportFrom(self, node: ast.ImportFrom) -> None: + if node.module: + 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: + findings.append(Finding( + plugin_id=plugin_id, + file=str(py_file.relative_to(PROJECT_ROOT)), + line=getattr(exc, "lineno", 0) or 0, + severity="WARNING", + rule="PLUGIN-030", + message=f"Python syntax error — cannot be parsed: {exc}", + )) + except OSError as exc: + findings.append(Finding( + plugin_id=plugin_id, + file=str(py_file.relative_to(PROJECT_ROOT)), + line=0, + severity="INFO", + 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 + + 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 + + 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}") + + # 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..0459c692 --- /dev/null +++ b/scripts/generate_report.py @@ -0,0 +1,297 @@ +#!/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 containing these strings are template placeholders, not real secrets +_GITLEAKS_SUPPRESS = [ + "YOUR_", + "PLACEHOLDER", + "_HERE", + "example.com", + "config_secrets.template", +] + + +# ───────────────────────────────────────────────────────────────────────────── +# Helpers +# ───────────────────────────────────────────────────────────────────────────── + +def _load(path: Path) -> dict | list | None: + """Load JSON file, returning None on any error.""" + try: + return json.loads(path.read_text(encoding="utf-8")) + except (json.JSONDecodeError, FileNotFoundError, OSError): + return None + + +def _md_table_row(*cells: str) -> str: + return "| " + " | ".join(str(c) for c in cells) + " |" + + +# ───────────────────────────────────────────────────────────────────────────── +# Per-tool summarizers +# Returns: (markdown_lines: list[str], critical_count: int) +# ───────────────────────────────────────────────────────────────────────────── + +def _summarize_bandit(artifact_dir: Path) -> tuple[list[str], int]: + data = _load(artifact_dir / "sast-results" / "bandit-results.json") + if data is None: + return ["_bandit results not available_"], 0 + + 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) + + +def _summarize_pip_audit(artifact_dir: Path) -> tuple[list[str], int]: + data = _load(artifact_dir / "dependency-audit-results" / "pip-audit-results.json") + if data is None: + return ["_pip-audit results not available_"], 0 + + # 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 + + +def _summarize_gitleaks(artifact_dir: Path) -> tuple[list[str], int]: + data = _load(artifact_dir / "secrets-scan-results" / "gitleaks-results.json") + if data is None: + return ["_gitleaks results not available_"], 0 + + if not isinstance(data, list): + data = [] + + real_findings = [] + suppressed = 0 + for finding in data: + secret_val = str(finding.get("Secret", "") or finding.get("Match", "")) + if any(p in secret_val for p in _GITLEAKS_SUPPRESS): + 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 + + +def _summarize_security_proofs(artifact_dir: Path) -> tuple[list[str], int]: + data = _load(artifact_dir / "security-proofs-results" / "security-proofs-results.json") + if data is None: + return ["_security proofs results not available_"], 0 + + 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": "⚠️", + "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) + + +def _summarize_plugin_audit(artifact_dir: Path) -> tuple[list[str], int]: + data = _load(artifact_dir / "plugin-audit-results" / "plugin-audit-results.json") + if data is None: + return ["_plugin audit results not available_"], 0 + + 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) + + +# ───────────────────────────────────────────────────────────────────────────── +# 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 = _summarize_bandit(artifact_dir) + pip_audit_lines, pip_audit_crit = _summarize_pip_audit(artifact_dir) + gitleaks_lines, gitleaks_crit = _summarize_gitleaks(artifact_dir) + proofs_lines, proofs_crit = _summarize_security_proofs(artifact_dir) + plugins_lines, plugins_crit = _summarize_plugin_audit(artifact_dir) + + total_critical = bandit_crit + pip_audit_crit + gitleaks_crit + proofs_crit + plugins_crit + overall = "ACTION REQUIRED 🚨" if total_critical > 0 else "PASSED ✅" + + def section(title: str, lines: list[str]) -> str: + return f"### {title}\n\n" + "\n".join(lines) + "\n" + + report = f"""## 🔒 Security Audit — {overall} + +_Generated: {timestamp}_ + +| 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}") + + 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..8e7fdfab --- /dev/null +++ b/scripts/prove_security.py @@ -0,0 +1,504 @@ +#!/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 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": "✅", + "INFO": "ℹ️ ", + "WARNING": "⚠️ ", + "CRITICAL": "🚨", + "SKIP": "⏭️ ", + }.get(self.severity, "❓") + + +# ───────────────────────────────────────────────────────────────────────────── +# T1: Plugin Loading / Zip Slip +# ───────────────────────────────────────────────────────────────────────────── + +def test_t1a_zip_slip_protection() -> TestResult: + """ + Verify that zip-slip protection exists in store_manager.py. + + The protection lives at src/plugin_system/store_manager.py and uses + Path.is_relative_to() to validate each zip member before extraction. + This test confirms the guard is present — it should always pass green. + """ + 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") + + has_relative_to = "is_relative_to" in content + has_log_message = "Zip-slip detected" in content + + if not has_relative_to: + return TestResult("T1a", "CRITICAL", + "Zip-slip protection (is_relative_to) NOT FOUND in store_manager.py", + "The is_relative_to() guard must be present before zipfile.extractall()") + + if not has_log_message: + return TestResult("T1a", "WARNING", + "is_relative_to() found but 'Zip-slip detected' log message missing", + "Verify the protection block is still active and the log was not removed") + + return TestResult("T1a", "PASS", + "Zip-slip protection verified", + "is_relative_to() guard + 'Zip-slip detected' log present in store_manager.py") + + +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 + + 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): + pass + + if violations: + results.append(TestResult( + "T1b", "CRITICAL", + f"Dangerous function calls found in plugins ({len(violations)} instance(s))", + "; ".join(violations[:10]) + )) + else: + 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" + ) + + return TestResult("T2a", "INFO", "API surface documented", 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"), + (r'(?i)api[_-]?key\s*=\s*["\'](?!none|empty|placeholder|YOUR_|example|test)[^"\']{16,}["\']', "WARNING"), + (r'(?i)secret\s*=\s*["\'](?!none|empty|placeholder|YOUR_|example|test)[^"\']{16,}["\']', "WARNING"), + # Real GitHub token pattern + (r'ghp_[a-zA-Z0-9]{36}', "CRITICAL"), + # Generic long bearer tokens + (r'Bearer\s+[a-zA-Z0-9\-_\.]{32,}', "WARNING"), +] + +_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 in _SECRET_PATTERNS: + for match in re.finditer(pattern, content): + line_content = match.group(0) + # Skip lines containing template placeholder strings + 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 + violations.append( + f"[{severity}] {rel}:{line_no} — {line_content[:60]}" + ) + + 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 + from_lines = [l for l in content.splitlines() if l.strip().startswith("FROM")] + for from_line in from_lines: + parts = from_line.split() + if len(parts) >= 2: + image = parts[1] + if ":" not in image or image.endswith(":latest"): + issues.append(f"Unpinned base image: {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()) From acaa11f3f8a4e967fde1afa7e7eaac87e0600583 Mon Sep 17 00:00:00 2001 From: ChuckBuilds Date: Tue, 14 Jul 2026 16:51:55 -0400 Subject: [PATCH 2/4] chore: drop workflow files -- pushed separately (needs workflow OAuth scope) --- .github/workflows/security-audit.yml | 277 --------------------------- .github/workflows/tests.yml | 46 ----- 2 files changed, 323 deletions(-) delete mode 100644 .github/workflows/security-audit.yml delete mode 100644 .github/workflows/tests.yml diff --git a/.github/workflows/security-audit.yml b/.github/workflows/security-audit.yml deleted file mode 100644 index bd78ca4a..00000000 --- a/.github/workflows/security-audit.yml +++ /dev/null @@ -1,277 +0,0 @@ -name: Security Audit - -on: - push: - branches: - - main - - 'feat/**' - pull_request: - branches: - - main - schedule: - # Weekly full scan — catches new CVEs in existing deps - - cron: '0 6 * * 1' - workflow_dispatch: - -permissions: - contents: read - pull-requests: write - security-events: write - -jobs: - - # ───────────────────────────────────────────────────────────────────────── - # SAST — Static Application Security Testing - # ───────────────────────────────────────────────────────────────────────── - sast: - name: Static Analysis (bandit + semgrep) - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v4 - - - uses: actions/setup-python@v5 - with: - python-version: '3.11' - - - name: Install SAST tools - run: pip install bandit==1.8.3 semgrep - - # Bandit — Python-specific security linter - # --exit-zero: findings are warnings, not CI blockers. - # The security-report job interprets severity. - - name: Run bandit - run: | - bandit -r src/ web_interface/ \ - -c bandit.yaml \ - -f json \ - -o bandit-results.json \ - --exit-zero - - # Semgrep — broader pattern-based analysis - # || true: prevents network/rate-limit errors from blocking the workflow - - name: Run semgrep - run: | - semgrep --config "p/python" \ - --config "p/flask" \ - --json \ - --output semgrep-results.json \ - src/ web_interface/ \ - || true - - - name: Upload SAST artifacts - uses: actions/upload-artifact@v4 - if: always() - with: - name: sast-results - path: | - bandit-results.json - semgrep-results.json - retention-days: 30 - - - # ───────────────────────────────────────────────────────────────────────── - # Dependency Vulnerability Scanning - # ───────────────────────────────────────────────────────────────────────── - dependency-audit: - name: Dependency Audit (pip-audit + safety) - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v4 - - - uses: actions/setup-python@v5 - with: - python-version: '3.11' - - - name: Install audit tools - run: pip install pip-audit safety - - # Install project deps. Hardware-specific packages (rgbmatrix) will fail - # to build on Ubuntu runners — || true handles this gracefully. - # pip-audit operates on installed packages; partial install is acceptable. - - name: Install project dependencies - run: | - pip install -r requirements.txt || true - pip install -r web_interface/requirements.txt || true - pip install -r requirements-emulator.txt || true - - - name: Run pip-audit - run: | - pip-audit --format json --output pip-audit-results.json || true - - - name: Run safety check - run: | - safety check --output json > safety-results.json 2>&1 || true - - - name: Upload dependency audit artifacts - uses: actions/upload-artifact@v4 - if: always() - with: - name: dependency-audit-results - path: | - pip-audit-results.json - safety-results.json - retention-days: 30 - - - # ───────────────────────────────────────────────────────────────────────── - # Secrets Detection - # ───────────────────────────────────────────────────────────────────────── - secrets-scan: - name: Secrets Scan (gitleaks) - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v4 - with: - fetch-depth: 0 # Full history for scanning all commits - - # continue-on-error: config/config_secrets.template.json contains - # placeholder strings (YOUR_*) that may trigger gitleaks rules. - # The generate_report.py script suppresses these false positives. - - name: Run gitleaks - uses: gitleaks/gitleaks-action@ff98106e4c7b2bc287b24eaf42907196329070c7 # v2.3.9 - continue-on-error: true - env: - GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} - - - name: Upload secrets scan artifacts - uses: actions/upload-artifact@v4 - if: always() - with: - name: secrets-scan-results - path: results.sarif - retention-days: 30 - - - # ───────────────────────────────────────────────────────────────────────── - # LEDMatrix-Specific Security Proofs - # ───────────────────────────────────────────────────────────────────────── - ledmatrix-security-proofs: - name: LEDMatrix Security Proofs - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v4 - - - uses: actions/setup-python@v5 - with: - python-version: '3.11' - - - name: Install dependencies - run: pip install -r requirements.txt || true - - # Script exits 1 only on CRITICAL findings. - # Warnings are reported but do not block the workflow. - - name: Run security proofs - run: | - python scripts/prove_security.py \ - --output security-proofs-results.json \ - --verbose - - - name: Upload proofs artifacts - uses: actions/upload-artifact@v4 - if: always() - with: - name: security-proofs-results - path: security-proofs-results.json - retention-days: 30 - - - # ───────────────────────────────────────────────────────────────────────── - # Plugin Security Audit - # ───────────────────────────────────────────────────────────────────────── - plugin-audit: - name: Plugin Security Audit - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v4 - - - uses: actions/setup-python@v5 - with: - python-version: '3.11' - - # Script exits 1 only on CRITICAL findings (eval/exec in plugins). - # Missing manifest.json etc are warnings. - - name: Run plugin audit - run: | - python scripts/audit_plugins.py \ - --output plugin-audit-results.json \ - --verbose - - - name: Upload plugin audit artifacts - uses: actions/upload-artifact@v4 - if: always() - with: - name: plugin-audit-results - path: plugin-audit-results.json - retention-days: 30 - - - # ───────────────────────────────────────────────────────────────────────── - # Aggregate Report - # ───────────────────────────────────────────────────────────────────────── - security-report: - name: Security Report - runs-on: ubuntu-latest - needs: - - sast - - dependency-audit - - secrets-scan - - ledmatrix-security-proofs - - plugin-audit - if: always() # Run even if upstream jobs fail or are skipped - steps: - - uses: actions/checkout@v4 - - - uses: actions/setup-python@v5 - with: - python-version: '3.11' - - - name: Download all artifacts - uses: actions/download-artifact@v4 - with: - path: audit-artifacts/ - - - name: Generate consolidated report - run: | - python scripts/generate_report.py \ - --artifact-dir audit-artifacts/ \ - --output security-report.md \ - --verbose - - - name: Upload consolidated report - uses: actions/upload-artifact@v4 - with: - name: security-report - path: security-report.md - retention-days: 90 - - - name: Comment on PR - if: github.event_name == 'pull_request' - uses: actions/github-script@v7 - with: - script: | - const fs = require('fs'); - const report = fs.readFileSync('security-report.md', 'utf8'); - // Use sticky comment — update existing comment rather than adding a new one each run - const { data: comments } = await github.rest.issues.listComments({ - issue_number: context.issue.number, - owner: context.repo.owner, - repo: context.repo.repo, - }); - const botComment = comments.find(c => - c.user.type === 'Bot' && c.body.includes('🔒 Security Audit') - ); - if (botComment) { - await github.rest.issues.updateComment({ - comment_id: botComment.id, - owner: context.repo.owner, - repo: context.repo.repo, - body: report, - }); - } else { - await github.rest.issues.createComment({ - issue_number: context.issue.number, - owner: context.repo.owner, - repo: context.repo.repo, - body: report, - }); - } diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml deleted file mode 100644 index c59cac7d..00000000 --- a/.github/workflows/tests.yml +++ /dev/null @@ -1,46 +0,0 @@ -name: Tests - -on: - push: - branches: [main] - pull_request: - branches: [main] - workflow_dispatch: - -permissions: - contents: read - -jobs: - test: - name: pytest (Python ${{ matrix.python-version }}) - runs-on: ubuntu-latest - strategy: - fail-fast: false - matrix: - python-version: ['3.10', '3.11', '3.12'] - steps: - - name: Check out repository - uses: actions/checkout@v4 - with: - submodules: false # rgbmatrix submodule not needed in EMULATOR mode - - - name: Set up Python ${{ matrix.python-version }} - uses: actions/setup-python@v5 - with: - python-version: ${{ matrix.python-version }} - cache: pip - - - name: Install dependencies - run: | - python -m pip install --upgrade pip - pip install -r requirements.txt - # Optional deps that some test modules import - pip install scipy psutil Flask-Limiter - - - name: Run tests - env: - EMULATOR: "true" - run: | - pytest \ - -m "not hardware and not slow" \ - --tb=short From d2c500e041df3b660b7fb6127a821c609bf3a0eb Mon Sep 17 00:00:00 2001 From: ChuckBuilds Date: Wed, 15 Jul 2026 11:20:31 -0400 Subject: [PATCH 3/4] fix(security-tooling): address PR review findings across bandit.yaml, audit_plugins.py, generate_report.py, prove_security.py bandit.yaml: - Removed scripts/prove_security.py's file-level exclusion. Ran bandit directly to get ground truth: the real false positive is B105 (dict key "PASS" misread as password-like), not the eval/exec pattern the old comment claimed. Added a targeted # nosec B105 there, and found+fixed the identical pattern already present in generate_report.py. - Left the repo-wide B607 skip as-is: confirmed via AST scan that properly narrowing it touches 100+ bare-name subprocess call sites across wifi_manager.py, store_manager.py, permission_utils.py, app.py, and start.py -- none of which are part of this PR. Out of proportion to fix here; flagged as a dedicated follow-up. scripts/audit_plugins.py: - SyntaxError/OSError while scanning a plugin file now report CRITICAL (blocking) instead of WARNING/INFO -- a file that couldn't be parsed or read was never actually checked for danger, so it must not silently pass the audit. - --plugin now tracks whether the requested plugin was found across all PLUGIN_BASE_DIRS and exits 1 with a clear error if not, instead of silently scanning zero plugins and reporting success. - The AST visitor now tracks import aliases (import subprocess as sp; from builtins import eval as e) and resolves them before checking against dangerous APIs, closing a straightforward evasion of every PLUGIN-001 through PLUGIN-005 check. Verified against both aliased and unaliased evasion patterns. scripts/generate_report.py: - _md_table_row now escapes pipe characters and normalizes newlines in every cell, so scanner-controlled content (a matched secret, a bandit issue_text) can't corrupt the Markdown table structure. - _load now distinguishes "artifact missing/malformed" from "valid empty result": each summarizer returns an availability flag, and main() now reports INCOMPLETE (not PASSED) with exit code 1 when any artifact is unavailable, instead of silently folding it in as 0 findings. - Gitleaks suppression now uses exact-match placeholder values (pulled from the actual config_secrets.template.json) plus a template-path allowlist, replacing broad substring checks that could hide a real secret containing something like "example.com" as part of its value. scripts/prove_security.py: - T1b (dangerous plugin calls): a file that fails to parse/read now reports CRITICAL with the exception details instead of being silently swallowed by `except (SyntaxError, OSError): pass`. - T6 (Docker hardening): base images must now be pinned to an @sha256 digest; a specific tag like python:3.12 is mutable and is now correctly flagged as unpinned, not just missing tags or :latest. - T2a (API surface): no config mechanism for enforcing local-only access exists in this codebase today (app.py hardcodes host='0.0.0.0'), so the "environment-aware" check as described isn't buildable without adding new config infrastructure -- out of scope here. Applied the achievable part: upgraded from INFO to WARNING, since enforcement can never currently be confirmed. - T1a (zip-slip): replaced the whole-file substring check with an AST walk that finds every extract()/extractall() call and confirms an is_relative_to() guard + "Zip-slip detected" log precede it in the same function. Verified it still passes on the real store_manager.py (both the per-member and validate-then-bulk-extract call sites) and correctly flags a synthetic unguarded extractall(). - T3a (hardcoded secrets): violation details no longer include the matched credential text -- only file, line, pattern type, and a redacted SHA-256 fingerprint, so a real finding doesn't get published into CI logs/artifacts/PR comments with wider exposure than the original leak. Verified with a synthetic secret that no raw content reaches the output. Validated: all four files compile; bandit scans all three scripts clean (2 legitimate targeted suppressions, 0 unaddressed findings); each new/ changed code path exercised directly (alias evasion, unmatched --plugin, missing/malformed/valid-empty artifacts, digest-pinning, zip-slip guard/no-guard, secret redaction); full audit_plugins.py -> prove_security.py -> generate_report.py pipeline run end-to-end producing a correct report. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01KEZK1P1Q1fu5pcuVrkrCFZ --- bandit.yaml | 4 - scripts/audit_plugins.py | 62 +++++++++++---- scripts/generate_report.py | 155 +++++++++++++++++++++++++------------ scripts/prove_security.py | 147 +++++++++++++++++++++++++++-------- 4 files changed, 270 insertions(+), 98 deletions(-) diff --git a/bandit.yaml b/bandit.yaml index abc7b273..14a6772d 100644 --- a/bandit.yaml +++ b/bandit.yaml @@ -27,7 +27,3 @@ exclude_dirs: - venv - .venv - rpi-rgb-led-matrix-master - # prove_security.py intentionally contains detection patterns as string literals - # (e.g. "eval(", "exec(") to search for in other files — bandit would flag - # these as false positives. - - scripts/prove_security.py diff --git a/scripts/audit_plugins.py b/scripts/audit_plugins.py index b2a92c14..09f3d647 100644 --- a/scripts/audit_plugins.py +++ b/scripts/audit_plugins.py @@ -56,6 +56,10 @@ 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( @@ -67,24 +71,34 @@ def _add(self, node: ast.AST, severity: str, rule: str, message: str) -> None: 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 visit_Call(self, node: ast.Call) -> None: - # eval() / exec() — arbitrary code execution + # eval() / exec() / compile() — arbitrary code execution, including + # aliased or from-imported forms (from builtins import eval as e; e(...)) if isinstance(node.func, ast.Name): - if node.func.id == "eval": + target = self._resolve(node.func.id).rsplit(".", 1)[-1] + if target == "eval": self._add(node, "CRITICAL", "PLUGIN-001", "eval() call — arbitrary code execution risk") - elif node.func.id == "exec": + elif target == "exec": self._add(node, "CRITICAL", "PLUGIN-002", "exec() call — arbitrary code execution risk") - elif node.func.id == "compile": + elif target == "compile": self._add(node, "WARNING", "PLUGIN-003", "compile() call — dynamic code compilation") - # subprocess.*(shell=True) - if isinstance(node.func, ast.Attribute): + # subprocess.*(shell=True) / os.system(), including aliased imports + # (import subprocess as sp; import os as o) + if isinstance(node.func, ast.Attribute) and isinstance(node.func.value, ast.Name): + base = self._resolve(node.func.value.id) + is_subprocess = ( - isinstance(node.func.value, ast.Name) and - node.func.value.id == "subprocess" and + base == "subprocess" and node.func.attr in ("run", "call", "Popen", "check_call", "check_output") ) if is_subprocess: @@ -97,11 +111,7 @@ def visit_Call(self, node: ast.Call) -> None: f"shell injection risk if args include user input") # os.system() — shell execution - is_os_system = ( - isinstance(node.func.value, ast.Name) and - node.func.value.id == "os" and - node.func.attr == "system" - ) + is_os_system = base == "os" and node.func.attr == "system" if is_os_system: self._add(node, "WARNING", "PLUGIN-005", "os.system() call — prefer subprocess with list args") @@ -110,11 +120,20 @@ def visit_Call(self, node: ast.Call) -> None: 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) @@ -167,20 +186,24 @@ def audit_plugin(plugin_dir: Path) -> list[Finding]: 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="WARNING", + 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="INFO", + severity="CRITICAL", rule="PLUGIN-031", message=f"Could not read file: {exc}", )) @@ -212,6 +235,7 @@ def main() -> int: 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(): @@ -229,6 +253,8 @@ def main() -> int: 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) @@ -254,6 +280,12 @@ def main() -> int: ) 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"] diff --git a/scripts/generate_report.py b/scripts/generate_report.py index 0459c692..86091ab8 100644 --- a/scripts/generate_report.py +++ b/scripts/generate_report.py @@ -33,13 +33,20 @@ PROJECT_ROOT = Path(__file__).resolve().parent.parent -# Gitleaks matches containing these strings are template placeholders, not real secrets -_GITLEAKS_SUPPRESS = [ - "YOUR_", - "PLACEHOLDER", - "_HERE", - "example.com", - "config_secrets.template", +# 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", ] @@ -47,27 +54,50 @@ # Helpers # ───────────────────────────────────────────────────────────────────────────── -def _load(path: Path) -> dict | list | None: - """Load JSON file, returning None on any error.""" +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")) - except (json.JSONDecodeError, FileNotFoundError, OSError): - return None + 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) -> 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(str(c) for c in cells) + " |" + return "| " + " | ".join(_md_sanitize_cell(c) for c in cells) + " |" # ───────────────────────────────────────────────────────────────────────────── # Per-tool summarizers -# Returns: (markdown_lines: list[str], critical_count: int) +# 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]: - data = _load(artifact_dir / "sast-results" / "bandit-results.json") - if data is None: - return ["_bandit results not available_"], 0 +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"] @@ -94,13 +124,13 @@ def _summarize_bandit(artifact_dir: Path) -> tuple[list[str], int]: if len(high) > 10: lines.append(f"_… and {len(high) - 10} more HIGH findings_") - return lines, len(high) + return lines, len(high), True -def _summarize_pip_audit(artifact_dir: Path) -> tuple[list[str], int]: - data = _load(artifact_dir / "dependency-audit-results" / "pip-audit-results.json") - if data is None: - return ["_pip-audit results not available_"], 0 +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] = [] @@ -122,13 +152,13 @@ def _summarize_pip_audit(artifact_dir: Path) -> tuple[list[str], int]: )) # Treat known vulnerabilities as warnings, not critical (they may be unavoidable) - return lines, 0 + return lines, 0, True -def _summarize_gitleaks(artifact_dir: Path) -> tuple[list[str], int]: - data = _load(artifact_dir / "secrets-scan-results" / "gitleaks-results.json") - if data is None: - return ["_gitleaks results not available_"], 0 +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 = [] @@ -137,7 +167,9 @@ def _summarize_gitleaks(artifact_dir: Path) -> tuple[list[str], int]: suppressed = 0 for finding in data: secret_val = str(finding.get("Secret", "") or finding.get("Match", "")) - if any(p in secret_val for p in _GITLEAKS_SUPPRESS): + 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) @@ -159,13 +191,13 @@ def _summarize_gitleaks(artifact_dir: Path) -> tuple[list[str], int]: )) critical = len(real_findings) # any real secret is critical - return lines, critical + return lines, critical, True -def _summarize_security_proofs(artifact_dir: Path) -> tuple[list[str], int]: - data = _load(artifact_dir / "security-proofs-results" / "security-proofs-results.json") - if data is None: - return ["_security proofs results not available_"], 0 +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 = [] @@ -182,7 +214,7 @@ def _summarize_security_proofs(artifact_dir: Path) -> tuple[list[str], int]: "", ] - _icon = {"PASS": "✅", "INFO": "ℹ️", "WARNING": "⚠️", + _icon = {"PASS": "✅", "INFO": "ℹ️", "WARNING": "⚠️", # nosec B105 - severity labels, not credentials "CRITICAL": "🚨", "SKIP": "⏭️"} for r in data: icon = _icon.get(r.get("severity", ""), "❓") @@ -192,13 +224,13 @@ def _summarize_security_proofs(artifact_dir: Path) -> tuple[list[str], int]: if r.get("details") and r.get("severity") in ("CRITICAL", "WARNING"): lines.append(f" - _{r['details']}_") - return lines, len(critical) + return lines, len(critical), True -def _summarize_plugin_audit(artifact_dir: Path) -> tuple[list[str], int]: - data = _load(artifact_dir / "plugin-audit-results" / "plugin-audit-results.json") - if data is None: - return ["_plugin audit results not available_"], 0 +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", []) @@ -226,7 +258,7 @@ def _summarize_plugin_audit(artifact_dir: Path) -> tuple[list[str], int]: 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) + return lines, summary.get("critical", 0), True # ───────────────────────────────────────────────────────────────────────────── @@ -248,22 +280,44 @@ def main() -> int: artifact_dir = Path(args.artifact_dir) timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") - bandit_lines, bandit_crit = _summarize_bandit(artifact_dir) - pip_audit_lines, pip_audit_crit = _summarize_pip_audit(artifact_dir) - gitleaks_lines, gitleaks_crit = _summarize_gitleaks(artifact_dir) - proofs_lines, proofs_crit = _summarize_security_proofs(artifact_dir) - plugins_lines, plugins_crit = _summarize_plugin_audit(artifact_dir) + 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 - overall = "ACTION REQUIRED 🚨" if total_critical > 0 else "PASSED ✅" + 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} | @@ -289,6 +343,11 @@ def section(title: str, lines: list[str]) -> str: 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 diff --git a/scripts/prove_security.py b/scripts/prove_security.py index 8e7fdfab..727bcb2b 100644 --- a/scripts/prove_security.py +++ b/scripts/prove_security.py @@ -17,6 +17,7 @@ import ast import argparse +import hashlib import json import re import sys @@ -43,7 +44,7 @@ def to_dict(self) -> dict: @property def icon(self) -> str: return { - "PASS": "✅", + "PASS": "✅", # nosec B105 - severity label, not a credential "INFO": "ℹ️ ", "WARNING": "⚠️ ", "CRITICAL": "🚨", @@ -57,11 +58,17 @@ def icon(self) -> str: def test_t1a_zip_slip_protection() -> TestResult: """ - Verify that zip-slip protection exists in store_manager.py. - - The protection lives at src/plugin_system/store_manager.py and uses - Path.is_relative_to() to validate each zip member before extraction. - This test confirms the guard is present — it should always pass green. + 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(): @@ -70,23 +77,65 @@ def test_t1a_zip_slip_protection() -> TestResult: 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)) - has_relative_to = "is_relative_to" in content - has_log_message = "Zip-slip detected" in content + extraction_sites = 0 + unprotected: list[str] = [] - if not has_relative_to: - return TestResult("T1a", "CRITICAL", - "Zip-slip protection (is_relative_to) NOT FOUND in store_manager.py", - "The is_relative_to() guard must be present before zipfile.extractall()") + for func in ast.walk(tree): + if not isinstance(func, (ast.FunctionDef, ast.AsyncFunctionDef)): + continue - if not has_log_message: + 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", - "is_relative_to() found but 'Zip-slip detected' log message missing", - "Verify the protection block is still active and the log was not removed") + "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", - "is_relative_to() guard + 'Zip-slip detected' log present in store_manager.py") + 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]: @@ -103,6 +152,8 @@ def test_t1b_dangerous_plugin_calls() -> list[TestResult]: violations: list[str] = [] files_scanned = 0 + scan_errors: list[str] = [] + for base in plugin_dirs: if not base.exists(): continue @@ -120,8 +171,19 @@ def test_t1b_dangerous_plugin_calls() -> list[TestResult]: rel = py_file.relative_to(PROJECT_ROOT) violations.append( f"{rel}:{node.lineno} — {node.func.id}() call") - except (SyntaxError, OSError): - pass + 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( @@ -129,7 +191,7 @@ def test_t1b_dangerous_plugin_calls() -> list[TestResult]: f"Dangerous function calls found in plugins ({len(violations)} instance(s))", "; ".join(violations[:10]) )) - else: + elif not scan_errors: results.append(TestResult( "T1b", "PASS", "No eval()/exec() calls found in plugins", @@ -182,7 +244,19 @@ def test_t2a_api_surface_inventory() -> TestResult: "the app is now internet-facing" ) - return TestResult("T2a", "INFO", "API surface documented", summary) + # 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 + ) # ───────────────────────────────────────────────────────────────────────────── @@ -191,13 +265,13 @@ def test_t2a_api_surface_inventory() -> TestResult: # 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"), - (r'(?i)api[_-]?key\s*=\s*["\'](?!none|empty|placeholder|YOUR_|example|test)[^"\']{16,}["\']', "WARNING"), - (r'(?i)secret\s*=\s*["\'](?!none|empty|placeholder|YOUR_|example|test)[^"\']{16,}["\']', "WARNING"), + (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"), + (r'ghp_[a-zA-Z0-9]{36}', "CRITICAL", "github_token"), # Generic long bearer tokens - (r'Bearer\s+[a-zA-Z0-9\-_\.]{32,}', "WARNING"), + (r'Bearer\s+[a-zA-Z0-9\-_\.]{32,}', "WARNING", "bearer_token"), ] _TEMPLATE_SKIP_STRINGS = [ @@ -225,16 +299,25 @@ def test_t3a_hardcoded_secrets() -> TestResult: except OSError: continue - for pattern, severity in _SECRET_PATTERNS: + 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 + # 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} — {line_content[:60]}" + f"[{severity}] {rel}:{line_no} — {pattern_type} " + f"(fingerprint {fingerprint})" ) critical_violations = [v for v in violations if "[CRITICAL]" in v] @@ -414,14 +497,16 @@ def test_t6_docker_hardening() -> TestResult: 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 + # 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 = [l for l in content.splitlines() if l.strip().startswith("FROM")] for from_line in from_lines: parts = from_line.split() if len(parts) >= 2: image = parts[1] - if ":" not in image or image.endswith(":latest"): - issues.append(f"Unpinned base image: {image}") + if "@sha256:" not in image: + issues.append(f"Base image not pinned to a digest: {image}") if issues: return TestResult("T6", "WARNING", From 7aa2ef28cad800224c74fbe2046b656cf7bdb82e Mon Sep 17 00:00:00 2001 From: ChuckBuilds Date: Wed, 15 Jul 2026 11:38:37 -0400 Subject: [PATCH 4/4] fix(security-tooling): address follow-up review findings on PR #414 scripts/generate_report.py: - Added an explicit `object` type annotation to _md_sanitize_cell's value parameter -- it deliberately accepts any stringifiable value (calls str(value) unconditionally), so `object` reflects its actual contract more accurately than leaving it untyped. scripts/prove_security.py: - Dockerfile FROM-line parsing: renamed the comprehension variable `l` to `line` (ambiguous single-letter name). More importantly, fixed a real false-positive: `FROM --platform= ` was reading the --platform= flag itself as the image token, so a properly digest-pinned image behind a platform flag was incorrectly reported as unpinned. Verified against platform+digest, platform+tag-only, and digest+AS-alias Dockerfiles. scripts/audit_plugins.py: - Consolidated visit_Call's dangerous-API detection: previously, alias resolution only covered ast.Name calls for eval/exec/compile and ast.Attribute calls for subprocess/os.system, missing from-imported subprocess/os functions called as bare names (from subprocess import run as prun; prun(cmd, shell=True) or from os import system as s; s(cmd)). Added _resolve_call_target() to resolve both call shapes to a single fully-qualified target, then run all five PLUGIN-00x checks against that one resolved value. Verified against 10 evasion combinations (from-import aliases, direct/attribute calls, aliased module imports) and confirmed zero false positives on benign os/ subprocess usage without shell=True. Validated: all three files compile, bandit scans clean (same 2 legitimate suppressions as before, 0 new findings), audit_plugins.py/prove_security.py re-run against the real repo with no regressions from the prior fix pass. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01KEZK1P1Q1fu5pcuVrkrCFZ --- scripts/audit_plugins.py | 90 ++++++++++++++++++++++---------------- scripts/generate_report.py | 2 +- scripts/prove_security.py | 10 +++-- 3 files changed, 61 insertions(+), 41 deletions(-) diff --git a/scripts/audit_plugins.py b/scripts/audit_plugins.py index 09f3d647..79a61acf 100644 --- a/scripts/audit_plugins.py +++ b/scripts/audit_plugins.py @@ -77,44 +77,60 @@ def _resolve(self, local_name: str) -> str: 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: - # eval() / exec() / compile() — arbitrary code execution, including - # aliased or from-imported forms (from builtins import eval as e; e(...)) - if isinstance(node.func, ast.Name): - target = self._resolve(node.func.id).rsplit(".", 1)[-1] - if target == "eval": - self._add(node, "CRITICAL", "PLUGIN-001", - "eval() call — arbitrary code execution risk") - elif target == "exec": - self._add(node, "CRITICAL", "PLUGIN-002", - "exec() call — arbitrary code execution risk") - elif target == "compile": - self._add(node, "WARNING", "PLUGIN-003", - "compile() call — dynamic code compilation") - - # subprocess.*(shell=True) / os.system(), including aliased imports - # (import subprocess as sp; import os as o) - if isinstance(node.func, ast.Attribute) and isinstance(node.func.value, ast.Name): - base = self._resolve(node.func.value.id) - - is_subprocess = ( - base == "subprocess" and - node.func.attr in ("run", "call", "Popen", "check_call", "check_output") - ) - if is_subprocess: - 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.{node.func.attr}(shell=True) — " - f"shell injection risk if args include user input") - - # os.system() — shell execution - is_os_system = base == "os" and node.func.attr == "system" - if is_os_system: - self._add(node, "WARNING", "PLUGIN-005", - "os.system() call — prefer subprocess with list args") + 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) diff --git a/scripts/generate_report.py b/scripts/generate_report.py index 86091ab8..753ea35b 100644 --- a/scripts/generate_report.py +++ b/scripts/generate_report.py @@ -71,7 +71,7 @@ def _load(path: Path) -> tuple[dict | list | None, str | None]: return None, f"could not read/parse {path}: {exc}" -def _md_sanitize_cell(value) -> str: +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 diff --git a/scripts/prove_security.py b/scripts/prove_security.py index 727bcb2b..9a5fbc88 100644 --- a/scripts/prove_security.py +++ b/scripts/prove_security.py @@ -500,11 +500,15 @@ def test_t6_docker_hardening() -> TestResult: # 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 = [l for l in content.splitlines() if l.strip().startswith("FROM")] + from_lines = [line for line in content.splitlines() if line.strip().startswith("FROM")] for from_line in from_lines: parts = from_line.split() - if len(parts) >= 2: - image = parts[1] + # 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}")