fix: remove content accidentally bundled into the dead-code-removal commit

The previous commit's git add/commit swept in a lot of unrelated,
unreviewed work alongside the intended dead-code deletions: a new Plugin
Composer web UI, new security-audit/CI tooling, a new march-madness
plugin, and 23 local-development-only symlinks under plugin-repos/ (per
scripts/setup_plugin_repos.py's own docstring, these are meant to be
generated locally, never committed -- .gitignore has no entry for them,
which is how they slipped in).

Removed here, split into their own PRs instead (except plugin-repos/*
symlinks and march-madness/ncaa_logos, which are dropped rather than
carried forward -- see PR discussion):
- web_interface/blueprints/composer.py + composer-app.js +
  composer-canvas.js + composer.html + manager.py.j2
- scripts/prove_security.py, audit_plugins.py, generate_report.py
- .github/workflows/security-audit.yml, .github/workflows/tests.yml,
  bandit.yaml
- All plugin-repos/* symlinks (local dev artifacts, not meant to be
  committed at all)
- plugin-repos/march-madness/* and the 4 new assets/sports/ncaa_logos/*
  PNGs it needed (left out of every split PR pending a decision on
  whether march-madness belongs in the core repo or the plugin monorepo)

This PR now contains only what its title describes: the dead-code
removal from the previous commit, untouched.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KEZK1P1Q1fu5pcuVrkrCFZ
This commit is contained in:
ChuckBuilds
2026-07-14 16:40:34 -04:00
co-authored by Claude Sonnet 5
parent 73a5304194
commit b0a404ef26
53 changed files with 0 additions and 7651 deletions
-296
View File
@@ -1,296 +0,0 @@
#!/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())
-297
View File
@@ -1,297 +0,0 @@
#!/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):
<artifact-dir>/
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())
-504
View File
@@ -1,504 +0,0 @@
#!/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",
f"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())