Merge remote-tracking branch 'origin/main' into check-412-conflict

# Conflicts:
#	src/image_utils.py

bandit.yaml

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

scripts/audit_plugins.py

@@ -1,344 +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] = []
-        # Local name -> real dotted path, so aliased imports and from-imports
-        # of dangerous APIs (import subprocess as sp; from builtins import
-        # eval as e) are still recognized in visit_Call below.
-        self._aliases: dict[str, str] = {}
-
-    def _add(self, node: ast.AST, severity: str, rule: str, message: str) -> None:
-        self.findings.append(Finding(
-            plugin_id=self.plugin_id,
-            file=str(self.filepath.relative_to(PROJECT_ROOT)),
-            line=getattr(node, "lineno", 0),
-            severity=severity,
-            rule=rule,
-            message=message,
-        ))
-
-    def _resolve(self, local_name: str) -> str:
-        """Resolve a local name through recorded import aliases to its real
-        dotted path (e.g. "sp" -> "subprocess"); unresolved names pass through
-        unchanged."""
-        return self._aliases.get(local_name, local_name)
-
-    def _resolve_call_target(self, func: ast.expr) -> str | None:
-        """Resolve a Call's func node to a fully-qualified dotted target,
-        covering a direct name (bare builtin, aliased import, or
-        from-import: from builtins import eval as e; from subprocess
-        import run; from os import system as s) and module-attribute
-        access (subprocess.run, sp.run, os.system, o.system) uniformly.
-        Returns None for call shapes this doesn't attempt to resolve."""
-        if isinstance(func, ast.Name):
-            return self._resolve(func.id)
-        if isinstance(func, ast.Attribute) and isinstance(func.value, ast.Name):
-            base = self._resolve(func.value.id)
-            return f"{base}.{func.attr}"
-        return None
-
-    def visit_Call(self, node: ast.Call) -> None:
-        target = self._resolve_call_target(node.func)
-        if target is None:
-            self.generic_visit(node)
-            return
-
-        leaf = target.rsplit(".", 1)[-1]
-
-        # eval() / exec() / compile() — arbitrary code execution, whether a
-        # bare call, an aliased import, or a from-import
-        # (from builtins import eval as e; e(...))
-        if leaf == "eval":
-            self._add(node, "CRITICAL", "PLUGIN-001",
-                      "eval() call — arbitrary code execution risk")
-        elif leaf == "exec":
-            self._add(node, "CRITICAL", "PLUGIN-002",
-                      "exec() call — arbitrary code execution risk")
-        elif leaf == "compile":
-            self._add(node, "WARNING", "PLUGIN-003",
-                      "compile() call — dynamic code compilation")
-
-        # subprocess.*(shell=True), whether subprocess.run(...), sp.run(...),
-        # or a from-import (from subprocess import run; run(..., shell=True))
-        if target in {
-            "subprocess.run", "subprocess.call", "subprocess.Popen",
-            "subprocess.check_call", "subprocess.check_output",
-        }:
-            for kw in node.keywords:
-                if (kw.arg == "shell" and
-                        isinstance(kw.value, ast.Constant) and
-                        kw.value.value is True):
-                    self._add(node, "WARNING", "PLUGIN-004",
-                              f"subprocess.{leaf}(shell=True) — "
-                              f"shell injection risk if args include user input")
-
-        # os.system(), whether os.system(...), o.system(...), or a
-        # from-import (from os import system as s; s(...))
-        if target == "os.system":
-            self._add(node, "WARNING", "PLUGIN-005",
-                      "os.system() call — prefer subprocess with list args")
-
-        self.generic_visit(node)
-
-    def visit_Import(self, node: ast.Import) -> None:
-        for alias in node.names:
-            if alias.asname:
-                local, real = alias.asname, alias.name
-            else:
-                # `import os.path` binds the top-level name `os`, not `os.path`
-                local = real = alias.name.split(".")[0]
-            self._aliases[local] = real
-            self._check_import(node, alias.name)
-        self.generic_visit(node)
-
-    def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
-        if node.module:
-            for alias in node.names:
-                local = alias.asname or alias.name
-                self._aliases[local] = f"{node.module}.{alias.name}"
-            self._check_import(node, node.module)
-        self.generic_visit(node)
-
-    def _check_import(self, node: ast.AST, module_name: str) -> None:
-        dangerous = {
-            "ctypes": ("WARNING", "PLUGIN-010", "ctypes import — native code execution"),
-            "cffi": ("WARNING", "PLUGIN-011", "cffi import — native code execution"),
-            "pickle": ("WARNING", "PLUGIN-012",
-                       "pickle import — deserialization can execute arbitrary code"),
-            "marshal": ("WARNING", "PLUGIN-013",
-                        "marshal import — deserialization risk"),
-        }
-        for mod, (severity, rule, msg) in dangerous.items():
-            if module_name == mod or module_name.startswith(mod + "."):
-                self._add(node, severity, rule, msg)
-
-
-# ─────────────────────────────────────────────────────────────────────────────
-# Per-plugin audit
-# ─────────────────────────────────────────────────────────────────────────────
-
-def audit_plugin(plugin_dir: Path) -> list[Finding]:
-    """Audit a single plugin directory. Returns all findings."""
-    findings: list[Finding] = []
-    plugin_id = plugin_dir.name
-
-    # Check for required files
-    for required_file, rule, msg in [
-        ("manifest.json", "PLUGIN-020",
-         "manifest.json missing — plugin may be incomplete"),
-        ("config_schema.json", "PLUGIN-021",
-         "config_schema.json missing — no input validation schema declared"),
-    ]:
-        if not (plugin_dir / required_file).exists():
-            findings.append(Finding(
-                plugin_id=plugin_id,
-                file=str((plugin_dir / required_file).relative_to(PROJECT_ROOT)),
-                line=0,
-                severity="WARNING",
-                rule=rule,
-                message=msg,
-            ))
-
-    # AST analysis of all Python files
-    for py_file in sorted(plugin_dir.rglob("*.py")):
-        try:
-            source = py_file.read_text(encoding="utf-8")
-            tree = ast.parse(source, filename=str(py_file))
-            visitor = _PluginVisitor(py_file, plugin_id)
-            visitor.visit(tree)
-            findings.extend(visitor.findings)
-        except SyntaxError as exc:
-            # A file the visitor can't even parse is a file we can't verify
-            # is safe -- this must block the audit, not just warn.
-            findings.append(Finding(
-                plugin_id=plugin_id,
-                file=str(py_file.relative_to(PROJECT_ROOT)),
-                line=getattr(exc, "lineno", 0) or 0,
-                severity="CRITICAL",
-                rule="PLUGIN-030",
-                message=f"Python syntax error — cannot be parsed: {exc}",
-            ))
-        except OSError as exc:
-            # Same reasoning as SyntaxError: an unreadable file was never
-            # actually scanned, so it must block rather than pass silently.
-            findings.append(Finding(
-                plugin_id=plugin_id,
-                file=str(py_file.relative_to(PROJECT_ROOT)),
-                line=0,
-                severity="CRITICAL",
-                rule="PLUGIN-031",
-                message=f"Could not read file: {exc}",
-            ))
-
-    return findings
-
-
-# ─────────────────────────────────────────────────────────────────────────────
-# Main
-# ─────────────────────────────────────────────────────────────────────────────
-
-def main() -> int:
-    parser = argparse.ArgumentParser(
-        description="LEDMatrix plugin security auditor",
-        formatter_class=argparse.RawDescriptionHelpFormatter,
-    )
-    parser.add_argument("--plugin", "-p", default=None,
-                        help="Audit a specific plugin ID only")
-    parser.add_argument("--output", "-o", default=None,
-                        help="Write JSON results to this file")
-    parser.add_argument("--verbose", "-v", action="store_true",
-                        help="Show all findings, not just summary")
-    args = parser.parse_args()
-
-    print("=" * 60)
-    print("LEDMatrix Plugin Security Audit")
-    print(f"Project root: {PROJECT_ROOT}")
-    print("=" * 60)
-
-    all_findings: list[Finding] = []
-    plugins_scanned = 0
-    plugin_found = args.plugin is None
-
-    for base_dir in PLUGIN_BASE_DIRS:
-        if not base_dir.exists():
-            if args.verbose:
-                print(f"  ⏭️  Skipping {base_dir.name}/ (directory not found)")
-            continue
-
-        base_label = base_dir.relative_to(PROJECT_ROOT)
-        print(f"\n  Scanning {base_label}/")
-
-        for plugin_dir in sorted(base_dir.iterdir()):
-            if not plugin_dir.is_dir():
-                continue
-            if plugin_dir.name.startswith((".", "_")):
-                continue
-            if args.plugin and plugin_dir.name != args.plugin:
-                continue
-            if args.plugin:
-                plugin_found = True
-
-            findings = audit_plugin(plugin_dir)
-            all_findings.extend(findings)
-            plugins_scanned += 1
-
-            critical = [f for f in findings if f.severity == "CRITICAL"]
-            warnings = [f for f in findings if f.severity == "WARNING"]
-
-            if critical:
-                icon, label = "🚨", "CRITICAL"
-            elif warnings:
-                icon, label = "⚠️ ", "WARN   "
-            else:
-                icon, label = "✅", "PASS   "
-
-            print(f"    {icon} [{label}] {plugin_dir.name}"
-                  f" — {len(critical)} critical, {len(warnings)} warnings")
-
-            if args.verbose:
-                for f in findings:
-                    severity_icon = {"CRITICAL": "🚨", "WARNING": "⚠️ ", "INFO": "ℹ️ "}.get(
-                        f.severity, "  "
-                    )
-                    print(f"           {severity_icon} {f.rule} {f.file}:{f.line} — {f.message}")
-
-    if args.plugin and not plugin_found:
-        print(f"\n  🚨 Plugin '{args.plugin}' not found in any of "
-              f"{[str(d.relative_to(PROJECT_ROOT)) for d in PLUGIN_BASE_DIRS]} — "
-              f"nothing was audited")
-        return 1
-
-    # Summary
-    critical_findings = [f for f in all_findings if f.severity == "CRITICAL"]
-    warning_findings = [f for f in all_findings if f.severity == "WARNING"]
-
-    print(f"\n{'=' * 60}")
-    print(f"  Plugins scanned : {plugins_scanned}")
-    print(f"  CRITICAL        : {len(critical_findings)}")
-    print(f"  WARNING         : {len(warning_findings)}")
-
-    if critical_findings:
-        print("\n  🚨 CRITICAL findings:")
-        for f in critical_findings:
-            print(f"    {f.plugin_id} | {Path(f.file).name}:{f.line} | {f.message}")
-
-    # Write JSON output
-    if args.output:
-        output_data = {
-            "timestamp": datetime.now(timezone.utc).isoformat(),
-            "plugins_scanned": plugins_scanned,
-            "summary": {
-                "critical": len(critical_findings),
-                "warnings": len(warning_findings),
-            },
-            "findings": [f.to_dict() for f in all_findings],
-        }
-        Path(args.output).write_text(
-            json.dumps(output_data, indent=2), encoding="utf-8"
-        )
-        print(f"\n  Results written to: {args.output}")
-
-    if critical_findings:
-        print("\n  🚨 Blocking — CRITICAL issues must be resolved")
-        return 1
-
-    print("\n  ✅ No critical issues found")
-    return 0
-
-
-if __name__ == "__main__":
-    sys.exit(main())

scripts/generate_report.py

@@ -1,356 +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 exactly equal to one of these (not a substring match -- a
-# real secret that merely contains one of these words as part of its actual
-# value must still be reported) are known template placeholders.
-_GITLEAKS_SUPPRESS_EXACT_VALUES = {
-    "YOUR_YOUTUBE_API_KEY",
-    "YOUR_YOUTUBE_CHANNEL_ID",
-    "YOUR_GITHUB_PERSONAL_ACCESS_TOKEN",
-}
-
-# Findings in these files are suppressed regardless of value -- they are
-# template/example files that are expected to only ever contain placeholders.
-_GITLEAKS_SUPPRESS_PATHS = [
-    "config_secrets.template.json",
-    "config.template.json",
-]
-
-
-# ─────────────────────────────────────────────────────────────────────────────
-# Helpers
-# ─────────────────────────────────────────────────────────────────────────────
-
-def _load(path: Path) -> tuple[dict | list | None, str | None]:
-    """Load a JSON artifact file.
-
-    Returns (data, error): error is None on success (data is whatever was
-    parsed, which may legitimately be an empty list/dict for a clean scan);
-    otherwise error is a human-readable reason the artifact is unavailable,
-    distinguishing "missing/malformed artifact" from "valid empty result" so
-    callers don't silently treat a broken CI job as a clean pass.
-    """
-    if not path.exists():
-        return None, f"artifact not found: {path}"
-    try:
-        return json.loads(path.read_text(encoding="utf-8")), None
-    except (json.JSONDecodeError, OSError) as exc:
-        return None, f"could not read/parse {path}: {exc}"
-
-
-def _md_sanitize_cell(value: object) -> str:
-    """Escape/normalize a value so scanner-controlled content (a matched
-    secret, a bandit issue_text, a file path) can't alter the Markdown
-    table's structure: pipes would add bogus columns, newlines would break
-    out of the row (or forge a fake header/separator line)."""
-    text = str(value)
-    text = text.replace("\\", "\\\\").replace("|", "\\|")
-    text = text.replace("\r\n", " ").replace("\n", " ").replace("\r", " ")
-    return text
-
-
-def _md_table_row(*cells: str) -> str:
-    return "| " + " | ".join(_md_sanitize_cell(c) for c in cells) + " |"
-
-
-# ─────────────────────────────────────────────────────────────────────────────
-# Per-tool summarizers
-# Returns: (markdown_lines: list[str], critical_count: int, available: bool)
-# `available=False` means the artifact was missing or malformed -- distinct
-# from a valid scan that simply found nothing -- so the caller can report
-# INCOMPLETE instead of silently counting it as a clean pass.
-# ─────────────────────────────────────────────────────────────────────────────
-
-def _summarize_bandit(artifact_dir: Path) -> tuple[list[str], int, bool]:
-    data, error = _load(artifact_dir / "sast-results" / "bandit-results.json")
-    if error:
-        return [f"_bandit results unavailable: {error}_"], 0, False
-
-    results = data.get("results", [])
-    high = [r for r in results if r.get("issue_severity") == "HIGH"]
-    medium = [r for r in results if r.get("issue_severity") == "MEDIUM"]
-    low = [r for r in results if r.get("issue_severity") == "LOW"]
-
-    lines = [
-        f"**Bandit**: {len(high)} HIGH · {len(medium)} MEDIUM · {len(low)} LOW"
-    ]
-
-    if high:
-        lines += [
-            "",
-            "| Severity | File | Line | Issue |",
-            "| --- | --- | --- | --- |",
-        ]
-        for r in high[:10]:
-            fname = Path(r.get("filename", "")).name
-            lines.append(_md_table_row(
-                "HIGH", f"`{fname}`",
-                str(r.get("line_number", "?")),
-                r.get("issue_text", "")
-            ))
-        if len(high) > 10:
-            lines.append(f"_… and {len(high) - 10} more HIGH findings_")
-
-    return lines, len(high), True
-
-
-def _summarize_pip_audit(artifact_dir: Path) -> tuple[list[str], int, bool]:
-    data, error = _load(artifact_dir / "dependency-audit-results" / "pip-audit-results.json")
-    if error:
-        return [f"_pip-audit results unavailable: {error}_"], 0, False
-
-    # pip-audit JSON format: {"dependencies": [{"name": ..., "vulns": [...]}]}
-    vulns: list[dict] = []
-    for dep in data.get("dependencies", []):
-        for v in dep.get("vulns", []):
-            vulns.append({"package": dep.get("name", "?"), **v})
-
-    lines = [f"**pip-audit**: {len(vulns)} vulnerabilities found"]
-
-    if vulns:
-        lines += ["", "| Package | ID | Fix |", "| --- | --- | --- |"]
-        for v in vulns[:10]:
-            fix = v.get("fix_versions", ["none"])
-            fix_str = ", ".join(fix) if fix else "none"
-            lines.append(_md_table_row(
-                v.get("package", "?"),
-                v.get("id", "?"),
-                fix_str,
-            ))
-
-    # Treat known vulnerabilities as warnings, not critical (they may be unavoidable)
-    return lines, 0, True
-
-
-def _summarize_gitleaks(artifact_dir: Path) -> tuple[list[str], int, bool]:
-    data, error = _load(artifact_dir / "secrets-scan-results" / "gitleaks-results.json")
-    if error:
-        return [f"_gitleaks results unavailable: {error}_"], 0, False
-
-    if not isinstance(data, list):
-        data = []
-
-    real_findings = []
-    suppressed = 0
-    for finding in data:
-        secret_val = str(finding.get("Secret", "") or finding.get("Match", ""))
-        file_name = Path(finding.get("File", "")).name
-        if (secret_val in _GITLEAKS_SUPPRESS_EXACT_VALUES
-                or file_name in _GITLEAKS_SUPPRESS_PATHS):
-            suppressed += 1
-        else:
-            real_findings.append(finding)
-
-    lines = [
-        f"**Gitleaks**: {len(real_findings)} finding(s) "
-        f"({suppressed} suppressed as template placeholders)"
-    ]
-
-    if real_findings:
-        lines += ["", "| Rule | File | Line | Description |", "| --- | --- | --- | --- |"]
-        for f in real_findings[:10]:
-            fname = Path(f.get("File", "")).name
-            lines.append(_md_table_row(
-                f.get("RuleID", "?"),
-                f"`{fname}`",
-                str(f.get("StartLine", "?")),
-                f.get("Description", ""),
-            ))
-
-    critical = len(real_findings)  # any real secret is critical
-    return lines, critical, True
-
-
-def _summarize_security_proofs(artifact_dir: Path) -> tuple[list[str], int, bool]:
-    data, error = _load(artifact_dir / "security-proofs-results" / "security-proofs-results.json")
-    if error:
-        return [f"_security proofs results unavailable: {error}_"], 0, False
-
-    if not isinstance(data, list):
-        data = []
-
-    critical = [r for r in data if r.get("severity") == "CRITICAL"]
-    warnings = [r for r in data if r.get("severity") == "WARNING"]
-    passed = [r for r in data if r.get("severity") == "PASS"]
-    skipped = [r for r in data if r.get("severity") == "SKIP"]
-
-    lines = [
-        f"**Security Proofs**: "
-        f"{len(passed)} PASS · {len(warnings)} WARN · "
-        f"{len(critical)} CRITICAL · {len(skipped)} SKIP",
-        "",
-    ]
-
-    _icon = {"PASS": "✅", "INFO": "ℹ️", "WARNING": "⚠️",  # nosec B105 - severity labels, not credentials
-             "CRITICAL": "🚨", "SKIP": "⏭️"}
-    for r in data:
-        icon = _icon.get(r.get("severity", ""), "❓")
-        lines.append(
-            f"- {icon} **{r.get('test_id', '?')}**: {r.get('message', '')}"
-        )
-        if r.get("details") and r.get("severity") in ("CRITICAL", "WARNING"):
-            lines.append(f"  - _{r['details']}_")
-
-    return lines, len(critical), True
-
-
-def _summarize_plugin_audit(artifact_dir: Path) -> tuple[list[str], int, bool]:
-    data, error = _load(artifact_dir / "plugin-audit-results" / "plugin-audit-results.json")
-    if error:
-        return [f"_plugin audit results unavailable: {error}_"], 0, False
-
-    summary = data.get("summary", {})
-    findings = data.get("findings", [])
-    critical_findings = [f for f in findings if f.get("severity") == "CRITICAL"]
-    warning_findings = [f for f in findings if f.get("severity") == "WARNING"]
-
-    lines = [
-        f"**Plugin Audit**: {data.get('plugins_scanned', '?')} plugins scanned — "
-        f"{summary.get('critical', 0)} CRITICAL · {summary.get('warnings', 0)} WARNINGS"
-    ]
-
-    if critical_findings:
-        lines += ["", "| Plugin | File | Line | Rule | Message |",
-                  "| --- | --- | --- | --- | --- |"]
-        for f in critical_findings[:10]:
-            fname = Path(f.get("file", "")).name
-            lines.append(_md_table_row(
-                f.get("plugin_id", "?"),
-                f"`{fname}`",
-                str(f.get("line", "?")),
-                f.get("rule", "?"),
-                f.get("message", ""),
-            ))
-
-    if warning_findings and not critical_findings:
-        lines.append(f"\n_{len(warning_findings)} warning(s) found — see artifact for details_")
-
-    return lines, summary.get("critical", 0), True
-
-
-# ─────────────────────────────────────────────────────────────────────────────
-# Main
-# ─────────────────────────────────────────────────────────────────────────────
-
-def main() -> int:
-    parser = argparse.ArgumentParser(
-        description="Generate consolidated security audit report",
-        formatter_class=argparse.RawDescriptionHelpFormatter,
-    )
-    parser.add_argument("--artifact-dir", required=True,
-                        help="Directory containing downloaded CI artifacts")
-    parser.add_argument("--output", "-o", required=True,
-                        help="Output Markdown file path")
-    parser.add_argument("--verbose", "-v", action="store_true")
-    args = parser.parse_args()
-
-    artifact_dir = Path(args.artifact_dir)
-    timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
-
-    bandit_lines, bandit_crit, bandit_ok = _summarize_bandit(artifact_dir)
-    pip_audit_lines, pip_audit_crit, pip_audit_ok = _summarize_pip_audit(artifact_dir)
-    gitleaks_lines, gitleaks_crit, gitleaks_ok = _summarize_gitleaks(artifact_dir)
-    proofs_lines, proofs_crit, proofs_ok = _summarize_security_proofs(artifact_dir)
-    plugins_lines, plugins_crit, plugins_ok = _summarize_plugin_audit(artifact_dir)
-
-    unavailable_tools = [
-        name for name, ok in [
-            ("bandit", bandit_ok), ("pip-audit", pip_audit_ok),
-            ("gitleaks", gitleaks_ok), ("security-proofs", proofs_ok),
-            ("plugin-audit", plugins_ok),
-        ] if not ok
-    ]
-
-    total_critical = bandit_crit + pip_audit_crit + gitleaks_crit + proofs_crit + plugins_crit
-    if unavailable_tools:
-        # A missing/malformed artifact means that tool's checks never
-        # actually ran -- this must not be reported as a clean PASS just
-        # because the *artifacts that did load* found nothing.
-        overall = "INCOMPLETE ⚠️"
-    elif total_critical > 0:
-        overall = "ACTION REQUIRED 🚨"
-    else:
-        overall = "PASSED ✅"
-
-    def section(title: str, lines: list[str]) -> str:
-        return f"### {title}\n\n" + "\n".join(lines) + "\n"
-
-    incomplete_note = (
-        f"\n_⚠️ Incomplete: results unavailable for {', '.join(unavailable_tools)} "
-        f"— see the corresponding section(s) below for details_\n"
-        if unavailable_tools else ""
-    )
-
-    report = f"""## 🔒 Security Audit — {overall}
-
-_Generated: {timestamp}_
-{incomplete_note}
-| Critical | High/Warn | Overall |
-| :---: | :---: | :---: |
-| {'🚨 ' + str(total_critical) if total_critical else '✅ 0'} | ⚠️ see below | {overall} |
-
----
-
-{section('SAST — Bandit', bandit_lines)}
-{section('Dependencies — pip-audit', pip_audit_lines)}
-{section('Secrets — Gitleaks', gitleaks_lines)}
-{section('LEDMatrix Security Proofs', proofs_lines)}
-{section('Plugin Security Audit', plugins_lines)}
----
-
-_Total critical findings: **{total_critical}**_
-"""
-
-    output_path = Path(args.output)
-    output_path.write_text(report, encoding="utf-8")
-
-    if args.verbose:
-        print(f"  Report written to: {output_path}")
-        print(f"  Status: {overall}")
-        print(f"  Critical findings: {total_critical}")
-        print(f"    bandit={bandit_crit}  pip-audit={pip_audit_crit}  "
-              f"gitleaks={gitleaks_crit}  proofs={proofs_crit}  plugins={plugins_crit}")
-        if unavailable_tools:
-            print(f"  Unavailable: {', '.join(unavailable_tools)}")
-
-    if unavailable_tools:
-        return 1
-
-    return 0
-
-
-if __name__ == "__main__":
-    sys.exit(main())

scripts/prove_security.py

@@ -1,593 +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 hashlib
-import json
-import re
-import sys
-from dataclasses import dataclass, asdict
-from pathlib import Path
-
-PROJECT_ROOT = Path(__file__).resolve().parent.parent
-
-
-# ─────────────────────────────────────────────────────────────────────────────
-# Result dataclass
-# ─────────────────────────────────────────────────────────────────────────────
-
-@dataclass
-class TestResult:
-    test_id: str
-    severity: str   # PASS | INFO | WARNING | CRITICAL | SKIP
-    message: str
-    details: str = ""
-
-    def to_dict(self) -> dict:
-        return asdict(self)
-
-    @property
-    def icon(self) -> str:
-        return {
-            "PASS": "✅",  # nosec B105 - severity label, not a credential
-            "INFO": "ℹ️ ",
-            "WARNING": "⚠️ ",
-            "CRITICAL": "🚨",
-            "SKIP": "⏭️ ",
-        }.get(self.severity, "❓")
-
-
-# ─────────────────────────────────────────────────────────────────────────────
-# T1: Plugin Loading / Zip Slip
-# ─────────────────────────────────────────────────────────────────────────────
-
-def test_t1a_zip_slip_protection() -> TestResult:
-    """
-    Verify that zip-slip protection actually guards zip extraction in
-    store_manager.py.
-
-    A whole-file substring check for "is_relative_to"/"Zip-slip detected"
-    would pass even if the guard existed somewhere unrelated, or covered
-    only one of several extract()/extractall() call sites. Instead, this
-    walks the AST: for every extract()/extractall() call, it confirms an
-    is_relative_to() check (and the "Zip-slip detected" log) appears
-    earlier in that same enclosing function -- validate-then-bulk-extract
-    (validate every member, then call extractall() only after all passed)
-    counts as protecting the call, since it covers the same member list.
-    """
-    store_manager = PROJECT_ROOT / "src" / "plugin_system" / "store_manager.py"
-    if not store_manager.exists():
-        return TestResult("T1a", "CRITICAL",
-                          "store_manager.py not found",
-                          f"Expected at {store_manager}")
-
-    content = store_manager.read_text(encoding="utf-8")
-    try:
-        tree = ast.parse(content, filename=str(store_manager))
-    except SyntaxError as exc:
-        return TestResult("T1a", "CRITICAL",
-                          "store_manager.py could not be parsed",
-                          str(exc))
-
-    extraction_sites = 0
-    unprotected: list[str] = []
-
-    for func in ast.walk(tree):
-        if not isinstance(func, (ast.FunctionDef, ast.AsyncFunctionDef)):
-            continue
-
-        extract_calls = [
-            node for node in ast.walk(func)
-            if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute)
-            and node.func.attr in ("extract", "extractall")
-        ]
-        if not extract_calls:
-            continue
-        extraction_sites += len(extract_calls)
-
-        guard_lines = [
-            n.lineno for n in ast.walk(func)
-            if isinstance(n, ast.Attribute) and n.attr == "is_relative_to"
-        ]
-        has_zip_slip_log = any(
-            isinstance(n, ast.Constant) and isinstance(n.value, str)
-            and "Zip-slip detected" in n.value
-            for n in ast.walk(func)
-        )
-
-        for call in extract_calls:
-            guarded = has_zip_slip_log and any(g < call.lineno for g in guard_lines)
-            if not guarded:
-                unprotected.append(
-                    f"{func.name}() line {call.lineno}: {call.func.attr}() call not "
-                    f"clearly preceded by an is_relative_to() guard + Zip-slip log "
-                    f"in the same function"
-                )
-
-    if extraction_sites == 0:
-        return TestResult("T1a", "WARNING",
-                          "No zipfile extract()/extractall() calls found in store_manager.py",
-                          "Verify plugin installation no longer extracts zip archives, "
-                          "or that this check still targets the right file")
-
-    if unprotected:
-        return TestResult("T1a", "CRITICAL",
-                          f"{len(unprotected)} of {extraction_sites} zip extraction "
-                          f"call(s) not clearly guarded",
-                          "; ".join(unprotected))
-
-    return TestResult("T1a", "PASS",
-                      "Zip-slip protection verified",
-                      f"All {extraction_sites} extract()/extractall() call(s) in "
-                      f"store_manager.py are preceded by an is_relative_to() guard "
-                      f"with a Zip-slip log in the same function")
-
-
-def test_t1b_dangerous_plugin_calls() -> list[TestResult]:
-    """
-    Scan plugin directories for dangerous function calls (eval, exec).
-    These represent arbitrary code execution risks in plugin code.
-    """
-    results = []
-    plugin_dirs = [
-        PROJECT_ROOT / "plugins",
-        PROJECT_ROOT / "plugin-repos",
-    ]
-
-    violations: list[str] = []
-    files_scanned = 0
-
-    scan_errors: list[str] = []
-
-    for base in plugin_dirs:
-        if not base.exists():
-            continue
-        for plugin_dir in sorted(base.iterdir()):
-            if not plugin_dir.is_dir() or plugin_dir.name.startswith(('.', '_')):
-                continue
-            for py_file in plugin_dir.rglob("*.py"):
-                files_scanned += 1
-                try:
-                    source = py_file.read_text(encoding="utf-8")
-                    tree = ast.parse(source, filename=str(py_file))
-                    for node in ast.walk(tree):
-                        if isinstance(node, ast.Call) and isinstance(node.func, ast.Name):
-                            if node.func.id in ("eval", "exec"):
-                                rel = py_file.relative_to(PROJECT_ROOT)
-                                violations.append(
-                                    f"{rel}:{node.lineno} — {node.func.id}() call")
-                except (SyntaxError, OSError) as exc:
-                    # A file we couldn't parse/read was never actually
-                    # scanned for eval()/exec() -- that must block this
-                    # test, not silently pass as if it were clean.
-                    rel = py_file.relative_to(PROJECT_ROOT)
-                    scan_errors.append(f"{rel} — {type(exc).__name__}: {exc}")
-
-    if scan_errors:
-        results.append(TestResult(
-            "T1b", "CRITICAL",
-            f"{len(scan_errors)} plugin file(s) could not be scanned for eval()/exec()",
-            "; ".join(scan_errors[:10])
-        ))
-
-    if violations:
-        results.append(TestResult(
-            "T1b", "CRITICAL",
-            f"Dangerous function calls found in plugins ({len(violations)} instance(s))",
-            "; ".join(violations[:10])
-        ))
-    elif not scan_errors:
-        results.append(TestResult(
-            "T1b", "PASS",
-            "No eval()/exec() calls found in plugins",
-            f"{files_scanned} plugin Python files scanned"
-        ))
-
-    return results
-
-
-# ─────────────────────────────────────────────────────────────────────────────
-# T2: API Surface Inventory
-# ─────────────────────────────────────────────────────────────────────────────
-
-def test_t2a_api_surface_inventory() -> TestResult:
-    """
-    Document the API surface area.
-
-    This app intentionally has no authentication (local-only Raspberry Pi
-    design, documented in web_interface/app.py). This test produces an
-    inventory for audit purposes and warns only if the design-intent comment
-    is removed from app.py (which would indicate someone deleted the rationale
-    without adding auth, rather than a deliberate undocumented change).
-    """
-    api_file = PROJECT_ROOT / "web_interface" / "blueprints" / "api_v3.py"
-    app_file = PROJECT_ROOT / "web_interface" / "app.py"
-
-    if not api_file.exists():
-        return TestResult("T2a", "WARNING", "api_v3.py not found", str(api_file))
-
-    api_content = api_file.read_text(encoding="utf-8")
-    routes = re.findall(r"@api_v3\.route\('([^']+)'", api_content)
-
-    csrf_documented = False
-    if app_file.exists():
-        app_content = app_file.read_text(encoding="utf-8")
-        csrf_documented = "CSRF protection disabled for local-only" in app_content
-
-    summary = (
-        f"{len(routes)} API routes in api_v3.py. "
-        f"No auth decorators (intentional local-only design). "
-        f"CSRF disabled: {'YES — design intent documented in app.py' if csrf_documented else 'YES — but design intent comment NOT found in app.py'}. "
-        f"Rate limiting: 1000/min."
-    )
-
-    if not csrf_documented:
-        return TestResult(
-            "T2a", "WARNING",
-            "CSRF is disabled but the design-intent comment is missing from app.py",
-            "Add the rationale comment back, or add proper CSRF protection if "
-            "the app is now internet-facing"
-        )
-
-    # There is currently no config mechanism that actually enforces the
-    # local-only boundary the design-intent comment describes -- app.py
-    # hardcodes host='0.0.0.0' unconditionally, so nothing here can confirm
-    # this deployment is in fact LAN-only. Reporting this as mere INFO
-    # understates that: an unauthenticated, CSRF-disabled API surface is a
-    # real risk the moment this ever runs somewhere other than a home LAN,
-    # documented rationale or not.
-    return TestResult(
-        "T2a", "WARNING",
-        "API surface has no auth and CSRF disabled; enforcement of the "
-        "documented local-only boundary cannot be confirmed",
-        summary
-    )
-
-
-# ─────────────────────────────────────────────────────────────────────────────
-# T3: Secrets & Credential Handling
-# ─────────────────────────────────────────────────────────────────────────────
-
-# Patterns that suggest real credentials (must be >8 chars, not placeholders)
-_SECRET_PATTERNS = [
-    (r'(?i)password\s*=\s*["\'](?!none|empty|placeholder|example|test|default|""|'')[^"\']{8,}["\']', "WARNING", "password"),
-    (r'(?i)api[_-]?key\s*=\s*["\'](?!none|empty|placeholder|YOUR_|example|test)[^"\']{16,}["\']', "WARNING", "api_key"),
-    (r'(?i)secret\s*=\s*["\'](?!none|empty|placeholder|YOUR_|example|test)[^"\']{16,}["\']', "WARNING", "secret"),
-    # Real GitHub token pattern
-    (r'ghp_[a-zA-Z0-9]{36}', "CRITICAL", "github_token"),
-    # Generic long bearer tokens
-    (r'Bearer\s+[a-zA-Z0-9\-_\.]{32,}', "WARNING", "bearer_token"),
-]
-
-_TEMPLATE_SKIP_STRINGS = [
-    "YOUR_", "PLACEHOLDER", "_HERE", "example.com", "config_secrets.template",
-    "prove_security",  # this file itself
-]
-
-_SCAN_DIRS = ["src", "web_interface", "scripts"]
-
-
-def test_t3a_hardcoded_secrets() -> TestResult:
-    """Scan source code for hardcoded credentials."""
-    violations: list[str] = []
-
-    for dir_name in _SCAN_DIRS:
-        scan_dir = PROJECT_ROOT / dir_name
-        if not scan_dir.exists():
-            continue
-        for py_file in scan_dir.rglob("*.py"):
-            # Skip test files and this script
-            if "test" in str(py_file).lower() or "prove_security" in str(py_file):
-                continue
-            try:
-                content = py_file.read_text(encoding="utf-8")
-            except OSError:
-                continue
-
-            for pattern, severity, pattern_type in _SECRET_PATTERNS:
-                for match in re.finditer(pattern, content):
-                    line_content = match.group(0)
-                    # Skip lines containing template placeholder strings.
-                    # line_content is only used for this in-memory check --
-                    # it must never be stored or included in output below.
-                    if any(skip in line_content for skip in _TEMPLATE_SKIP_STRINGS):
-                        continue
-                    rel = py_file.relative_to(PROJECT_ROOT)
-                    line_no = content[: match.start()].count("\n") + 1
-                    # Redacted fingerprint lets the same finding be recognized
-                    # across scans without ever reporting the matched
-                    # credential itself (which would otherwise get published
-                    # into CI logs, JSON artifacts, and PR comments -- wider
-                    # exposure than the original leak).
-                    fingerprint = hashlib.sha256(line_content.encode()).hexdigest()[:12]
-                    violations.append(
-                        f"[{severity}] {rel}:{line_no} — {pattern_type} "
-                        f"(fingerprint {fingerprint})"
-                    )
-
-    critical_violations = [v for v in violations if "[CRITICAL]" in v]
-    if critical_violations:
-        return TestResult(
-            "T3a", "CRITICAL",
-            f"Hardcoded secrets found ({len(critical_violations)} critical)",
-            "; ".join(critical_violations[:5])
-        )
-    if violations:
-        return TestResult(
-            "T3a", "WARNING",
-            f"Potential hardcoded secrets found ({len(violations)} instance(s))",
-            "; ".join(violations[:5])
-        )
-
-    return TestResult("T3a", "PASS", "No hardcoded secrets detected",
-                      f"Scanned {', '.join(_SCAN_DIRS)}")
-
-
-def test_t3b_plaintext_password_storage() -> TestResult:
-    """
-    Check for user account password storage without hashing.
-
-    The LEDMatrix app has no user account system, so this should produce INFO.
-    It would only CRITICAL if someone added user auth and stored passwords without hashing.
-
-    We require all three of: a password *variable assignment or DB operation*,
-    a clear storage call (INSERT / db commit / ORM save), and no hashing lib present
-    — to avoid false positives from files that contain 'password' for WiFi handling
-    and '.save()' for image/file saving in unrelated functions.
-    """
-    hashing_libs = ["bcrypt", "argon2", "pbkdf2", "scrypt",
-                    "generate_password_hash", "hashpw", "make_password"]
-    # Patterns that indicate password being stored in a database / ORM context.
-    # Must be specific enough to avoid matching set.add(), file.save(), etc.
-    db_storage_patterns = ["INSERT INTO", "db.session", "session.add(", "session.commit(", "orm.save"]
-
-    password_storage_found = False
-
-    for dir_name in _SCAN_DIRS:
-        scan_dir = PROJECT_ROOT / dir_name
-        if not scan_dir.exists():
-            continue
-        for py_file in scan_dir.rglob("*.py"):
-            try:
-                content = py_file.read_text(encoding="utf-8")
-            except OSError:
-                continue
-            # Require DB/ORM context specifically — not just any .save() call
-            if ("password" in content.lower() and
-                    any(store in content for store in db_storage_patterns) and
-                    not any(h in content for h in hashing_libs)):
-                password_storage_found = True
-
-    if password_storage_found:
-        return TestResult(
-            "T3b", "CRITICAL",
-            "Potential plaintext password storage in database/ORM detected",
-            "Found password + database storage operations without a recognized hashing library"
-        )
-
-    return TestResult("T3b", "INFO",
-                      "No plaintext password storage detected",
-                      "App has no user account system — expected result")
-
-
-# ─────────────────────────────────────────────────────────────────────────────
-# T4: Path Traversal
-# ─────────────────────────────────────────────────────────────────────────────
-
-def test_t4a_path_traversal() -> TestResult:
-    """
-    Verify static file serving uses send_from_directory (safe) rather than
-    open() with user-supplied paths. Also checks for extractall() calls that
-    lack the is_relative_to() guard.
-    """
-    issues: list[str] = []
-
-    app_file = PROJECT_ROOT / "web_interface" / "app.py"
-    if app_file.exists():
-        content = app_file.read_text(encoding="utf-8")
-        # The file-serve route should use send_from_directory or commonpath
-        if "send_from_directory" not in content and "commonpath" not in content:
-            issues.append("app.py: file-serve routes may not use send_from_directory/commonpath")
-
-    # Check all extractall() calls have a preceding is_relative_to guard
-    for py_file in (PROJECT_ROOT / "src").rglob("*.py"):
-        try:
-            content = py_file.read_text(encoding="utf-8")
-        except OSError:
-            continue
-        if "extractall(" in content and "is_relative_to" not in content:
-            rel = py_file.relative_to(PROJECT_ROOT)
-            issues.append(f"{rel}: extractall() without is_relative_to() guard")
-
-    if issues:
-        return TestResult(
-            "T4a", "WARNING",
-            f"Potential path traversal patterns found ({len(issues)})",
-            "; ".join(issues)
-        )
-
-    return TestResult("T4a", "PASS",
-                      "Path traversal mitigations verified",
-                      "send_from_directory/commonpath used for file serving; "
-                      "extractall() calls have is_relative_to() guards")
-
-
-# ─────────────────────────────────────────────────────────────────────────────
-# T5: Auth Bypass Patterns
-# ─────────────────────────────────────────────────────────────────────────────
-
-def test_t5a_auth_bypass_patterns() -> TestResult:
-    """
-    Look for broken auth bypass patterns — not the intentional no-auth design
-    (T2a covers that), but patterns that suggest auth was INTENDED to exist
-    but has an exploitable bypass: broad substring matching, debug-mode skips,
-    or if-True conditions.
-    """
-    bypass_signals = [
-        (r'if\s+True\s*:', "if True: bypass"),
-        (r'if\s+debug\s*:', "debug-mode auth skip"),
-        (r'request\.path\s+in\s+', "substring path matching in auth (Huntarr pattern)"),
-        (r'EXEMPT_ROUTES\s*=', "exempt routes list"),
-    ]
-
-    findings: list[str] = []
-
-    for dir_name in ["src", "web_interface"]:
-        scan_dir = PROJECT_ROOT / dir_name
-        if not scan_dir.exists():
-            continue
-        for py_file in scan_dir.rglob("*.py"):
-            try:
-                content = py_file.read_text(encoding="utf-8")
-            except OSError:
-                continue
-            for pattern, label in bypass_signals:
-                if re.search(pattern, content):
-                    # Only flag if the file also contains auth-related terms
-                    if any(auth in content.lower() for auth in
-                           ["auth", "login", "authenticate", "token", "permission"]):
-                        rel = py_file.relative_to(PROJECT_ROOT)
-                        findings.append(f"{rel}: {label}")
-
-    if findings:
-        return TestResult(
-            "T5a", "WARNING",
-            f"Potential auth bypass patterns found ({len(findings)})",
-            "; ".join(findings[:5])
-        )
-
-    return TestResult("T5a", "PASS",
-                      "No auth bypass patterns detected",
-                      "Checked src/ and web_interface/ for bypass signals")
-
-
-# ─────────────────────────────────────────────────────────────────────────────
-# T6: Docker / Container Hardening
-# ─────────────────────────────────────────────────────────────────────────────
-
-def test_t6_docker_hardening() -> TestResult:
-    """Container security — skipped if no Dockerfile exists."""
-    dockerfile = PROJECT_ROOT / "Dockerfile"
-    if not dockerfile.exists():
-        return TestResult("T6", "SKIP",
-                          "No Dockerfile found — container security scan not applicable",
-                          "If Docker support is added in future, enable hadolint/trivy scanning "
-                          "in .github/workflows/security-audit.yml")
-
-    content = dockerfile.read_text(encoding="utf-8")
-    issues: list[str] = []
-
-    # Check for non-root USER directive
-    user_lines = [l for l in content.splitlines() if l.strip().startswith("USER")]
-    if not user_lines or user_lines[-1].strip() == "USER root":
-        issues.append("Container runs as root — use USER directive to drop privileges")
-
-    # Check for pinned base image tags. A tag (even a specific version, not
-    # just :latest) is mutable -- the same tag can point to a different
-    # image later. Only a @sha256 digest is truly immutable/reproducible.
-    from_lines = [line for line in content.splitlines() if line.strip().startswith("FROM")]
-    for from_line in from_lines:
-        parts = from_line.split()
-        # FROM [--platform=<platform>] <image> [AS <name>] -- skip an
-        # optional --platform= flag so it's never mistaken for the image
-        # token itself (which would falsely report it as unpinned).
-        image_parts = [p for p in parts[1:] if not p.startswith("--platform=")]
-        if image_parts:
-            image = image_parts[0]
-            if "@sha256:" not in image:
-                issues.append(f"Base image not pinned to a digest: {image}")
-
-    if issues:
-        return TestResult("T6", "WARNING",
-                          f"Dockerfile hardening issues ({len(issues)})",
-                          "; ".join(issues))
-
-    return TestResult("T6", "PASS", "Dockerfile hardening checks passed", "")
-
-
-# ─────────────────────────────────────────────────────────────────────────────
-# Runner
-# ─────────────────────────────────────────────────────────────────────────────
-
-def main() -> int:
-    parser = argparse.ArgumentParser(
-        description="LEDMatrix security proof tests",
-        formatter_class=argparse.RawDescriptionHelpFormatter,
-    )
-    parser.add_argument("--output", "-o", default=None,
-                        help="Write JSON results to this file")
-    parser.add_argument("--verbose", "-v", action="store_true",
-                        help="Show details for each check")
-    args = parser.parse_args()
-
-    print("=" * 60)
-    print("LEDMatrix Security Proof Tests")
-    print(f"Project root: {PROJECT_ROOT}")
-    print("=" * 60)
-
-    all_results: list[TestResult] = []
-
-    # Run all test groups
-    all_results.append(test_t1a_zip_slip_protection())
-    all_results.extend(test_t1b_dangerous_plugin_calls())
-    all_results.append(test_t2a_api_surface_inventory())
-    all_results.append(test_t3a_hardcoded_secrets())
-    all_results.append(test_t3b_plaintext_password_storage())
-    all_results.append(test_t4a_path_traversal())
-    all_results.append(test_t5a_auth_bypass_patterns())
-    all_results.append(test_t6_docker_hardening())
-
-    # Print results
-    print()
-    for r in all_results:
-        line = f"  {r.icon} [{r.severity:<8}] {r.test_id}: {r.message}"
-        print(line)
-        if args.verbose and r.details:
-            print(f"             {r.details}")
-
-    # Tally
-    critical = [r for r in all_results if r.severity == "CRITICAL"]
-    warnings = [r for r in all_results if r.severity == "WARNING"]
-    passed = [r for r in all_results if r.severity == "PASS"]
-    skipped = [r for r in all_results if r.severity == "SKIP"]
-
-    print()
-    print(f"  Results: {len(passed)} PASS  {len(warnings)} WARN  "
-          f"{len(critical)} CRITICAL  {len(skipped)} SKIP")
-
-    # Write JSON output
-    if args.output:
-        output_data = [r.to_dict() for r in all_results]
-        Path(args.output).write_text(
-            json.dumps(output_data, indent=2), encoding="utf-8"
-        )
-        print(f"  Results written to: {args.output}")
-
-    if critical:
-        print(f"\n  🚨 {len(critical)} CRITICAL issue(s) found — blocking")
-        return 1
-
-    if warnings:
-        print(f"\n  ⚠️  {len(warnings)} warning(s) found — non-blocking")
-
-    print("\n  ✅ All checks passed (warnings are non-blocking)")
-    return 0
-
-
-if __name__ == "__main__":
-    sys.exit(main())

src/common/permission_utils.py

@@ -146,60 +146,6 @@ def ensure_file_permissions(path: Path, mode: int = 0o644) -> None:
         raise
 
 
-_shared_group_gid_cache: Optional[int] = None
-
-
-def get_shared_group_gid() -> Optional[int]:
-    """
-    Return the gid that should own config/secrets files shared between the
-    root-run ``ledmatrix.service`` (main display) and the non-root user that
-    ``ledmatrix-web.service`` runs as (see install_web_service.sh, which sets
-    ``User=$SUDO_USER``).
-
-    Resolved once from the project root directory's current group (normally
-    the login user's group from the initial ``git clone``), since that user
-    is stable across reinstalls unlike any single file's ownership.
-
-    Returns:
-        The gid, or None if it cannot be determined.
-    """
-    global _shared_group_gid_cache
-    if _shared_group_gid_cache is not None:
-        return _shared_group_gid_cache
-    try:
-        project_root = Path(__file__).resolve().parent.parent.parent
-        _shared_group_gid_cache = project_root.stat().st_gid
-        return _shared_group_gid_cache
-    except OSError:
-        return None
-
-
-def ensure_shared_group_ownership(path: Path) -> None:
-    """
-    Best-effort chgrp of ``path`` to the shared group (see
-    :func:`get_shared_group_gid`) when running as root.
-
-    Only root can change a file's group to one the calling process isn't a
-    member of, which is exactly the case that causes the web interface
-    (running as a non-root user) to get ``PermissionError`` reading files
-    the root-run display service just wrote with a 0o640/2775 mode: the mode
-    is group-readable, but without this the group is root's, not the web
-    user's. Silently does nothing if not running as root or on any error —
-    this is a hardening step, not a required one.
-    """
-    if os.geteuid() != 0:
-        return
-    gid = get_shared_group_gid()
-    if gid is None:
-        return
-    try:
-        if path.exists() and path.stat().st_gid != gid:
-            os.chown(path, -1, gid)
-            logger.debug(f"Set shared group ownership (gid {gid}) on {path}")
-    except OSError as e:
-        logger.debug(f"Could not set shared group ownership on {path}: {e}")
-
-
 def get_config_file_mode(file_path: Path) -> int:
     """
     Return appropriate permission mode for config files.

src/config_manager.py

@@ -38,7 +38,6 @@ from src.config_manager_atomic import (
 from src.common.permission_utils import (
     ensure_directory_permissions,
     ensure_file_permissions,
-    ensure_shared_group_ownership,
     get_config_file_mode,
     get_config_dir_mode
 )
@@ -235,11 +234,6 @@ class ConfigManager:
 
             # Load and merge secrets if they exist (be permissive on errors)
             if os.path.exists(self.secrets_path):
-                # Self-heal stale group ownership (e.g. the root-run display
-                # service wrote this file before the web user was granted
-                # group access) before every load attempt; no-op unless
-                # running as root and the group is already wrong.
-                ensure_shared_group_ownership(Path(self.secrets_path))
                 try:
                     with open(self.secrets_path, 'r') as f:
                         secrets = json.load(f)
@@ -369,7 +363,6 @@ class ConfigManager:
         # Set proper file permissions after creation
         config_path_obj = Path(self.config_path)
         ensure_file_permissions(config_path_obj, get_config_file_mode(config_path_obj))
-        ensure_shared_group_ownership(config_path_obj)
         
         self.logger.info(f"Created config.json from template at {os.path.abspath(self.config_path)}")
 
@@ -482,11 +475,6 @@ class ConfigManager:
             self.logger.error(error_msg)
             raise ConfigError(error_msg, config_path=path_to_load)
 
-        if file_type == "secrets":
-            # Best-effort self-heal: no-op unless running as root and the
-            # group is stale (see load_config for why this can happen).
-            ensure_shared_group_ownership(Path(path_to_load))
-
         try:
             with open(path_to_load, 'r') as f:
                 return json.load(f)
@@ -494,18 +482,7 @@ class ConfigManager:
             error_msg = f"Error parsing {file_type} configuration file: {path_to_load}"
             self.logger.error(error_msg, exc_info=True)
             raise ConfigError(error_msg, config_path=path_to_load) from e
-        except PermissionError as e:
-            if file_type == "secrets":
-                # Match load_config()'s tolerance: a secrets file the web
-                # process can't read (e.g. written 0640 by the root-run
-                # display service before the group was fixed up) shouldn't
-                # 500 the settings page — degrade to "no secrets" instead.
-                self.logger.warning(f"Secrets file not readable ({path_to_load}): {e}. Returning empty secrets.")
-                return {}
-            error_msg = f"Error loading {file_type} configuration file {path_to_load}: {str(e)}"
-            self.logger.error(error_msg, exc_info=True)
-            raise ConfigError(error_msg, config_path=path_to_load) from e
-        except (IOError, OSError) as e:
+        except (IOError, OSError, PermissionError) as e:
             error_msg = f"Error loading {file_type} configuration file {path_to_load}: {str(e)}"
             self.logger.error(error_msg, exc_info=True)
             raise ConfigError(error_msg, config_path=path_to_load) from e
@@ -562,7 +539,6 @@ class ConfigManager:
                 # Ensure final file has correct permissions
                 try:
                     ensure_file_permissions(path_obj, file_mode)
-                    ensure_shared_group_ownership(path_obj)
                 except OSError as perm_error:
                     # If we can't set permissions but file was written, log warning but don't fail
                     self.logger.warning(

src/config_manager_atomic.py

@@ -17,7 +17,6 @@ from enum import Enum
 
 from src.exceptions import ConfigError
 from src.logging_config import get_logger
-from src.common.permission_utils import ensure_shared_group_ownership
 
 
 class SaveResultStatus(Enum):
@@ -411,13 +410,6 @@ class AtomicConfigManager:
             # This is important because temp files may have different permissions
             # and we need root service to be able to read config.json
             os.chmod(destination, target_mode)
-
-            # Also fix group ownership when this save is running as root
-            # (the display service): 0o640 alone only helps the non-root web
-            # user read a root-written secrets file if its group already
-            # matches the web user's group, which isn't guaranteed. See
-            # permission_utils.ensure_shared_group_ownership for why.
-            ensure_shared_group_ownership(destination)
             
         except Exception as e:
             raise ConfigError(f"Error during atomic move: {e}") from e

src/display_controller.py

@@ -1133,29 +1133,6 @@ class DisplayController:
         remaining = self.on_demand_expires_at - time.time()
         return max(0.0, remaining)
 
-    def _publish_current_mode_state(self) -> None:
-        """Publish the currently active display mode/plugin to cache for the web UI."""
-        try:
-            state = {
-                'mode': self.current_display_mode,
-                'plugin_id': self.mode_to_plugin_id.get(self.current_display_mode),
-                'mode_index': self.current_mode_index,
-                'total_modes': len(self.available_modes),
-                'on_demand_active': self.on_demand_active,
-                'is_display_active': self.is_display_active,
-                'last_updated': time.time(),
-            }
-            self.cache_manager.set('display_current_state', state)
-            self._last_published_mode = self.current_display_mode
-        except (OSError, RuntimeError, ValueError, TypeError) as err:
-            logger.error("Failed to publish current display state: %s", err, exc_info=True)
-
-    def _publish_current_mode_state_if_changed(self) -> None:
-        """Publish current mode state only when it actually changed, to avoid
-        writing to the shared cache on every render tick."""
-        if self.current_display_mode != getattr(self, '_last_published_mode', None):
-            self._publish_current_mode_state()
-
     def _publish_on_demand_state(self) -> None:
         """Publish current on-demand state to cache for external consumers."""
         try:
@@ -1675,7 +1652,6 @@ class DisplayController:
             logger.info("Starting display with cached data (fast startup mode)")
             self.current_display_mode = self.available_modes[self.current_mode_index] if self.available_modes else 'none'
             logger.info(f"Initial mode set to: {self.current_display_mode} (index: {self.current_mode_index}, total modes: {len(self.available_modes)})")
-            self._publish_current_mode_state()
             
             while True:
                 # Apply plugin enable/disable edits saved via the web UI. The
@@ -1736,11 +1712,9 @@ class DisplayController:
                         logger.debug(f"Error clearing display when inactive: {e}")
                     
                     logger.info(f"Display not active (is_display_active={self.is_display_active}), sleeping...")
-                    self._publish_current_mode_state()
                     self._sleep_with_plugin_updates(60)
                     continue
                 
-                self._publish_current_mode_state_if_changed()
                 logger.debug("Display active, processing mode: %s", self.current_display_mode)
                 
                 # Plugins update on their own schedules - no forced sync updates needed

src/logging_config.py

@@ -139,41 +139,23 @@ def setup_logging(
             sys.stderr.write(f"Warning: Could not set up file logging to {log_file}: {e}\n")
 
 
-class PluginLoggerAdapter(logging.LoggerAdapter):
-    """LoggerAdapter that stamps every record with its plugin_id.
-
-    A plain `logging.Logger` attribute (the old approach) is never copied
-    onto individual `LogRecord`s, so `ContextualFormatter`/`StructuredFormatter`
-    only ever saw `plugin_id` on calls that explicitly passed
-    `extra={'plugin_id': ...}` (i.e. `log_with_context`). This adapter injects
-    it into `extra` on every call, so `self.logger.info(...)` in plugin code
-    is tagged automatically.
-    """
-
-    def process(self, msg, kwargs):
-        extra = dict(kwargs.get('extra') or {})
-        extra.setdefault('plugin_id', self.extra.get('plugin_id'))
-        kwargs['extra'] = extra
-        return msg, kwargs
-
-
-def get_logger(name: str, plugin_id: Optional[str] = None):
+def get_logger(name: str, plugin_id: Optional[str] = None) -> logging.Logger:
     """
     Get a logger with consistent configuration.
-
+    
     Args:
         name: Logger name (typically __name__)
         plugin_id: Optional plugin ID for automatic context
-
+        
     Returns:
-        Configured logger instance (or a PluginLoggerAdapter when plugin_id
-        is given, which supports the same .debug/.info/.warning/.error API)
+        Configured logger instance
     """
     logger = logging.getLogger(name)
-
+    
+    # Add plugin_id as attribute for formatters
     if plugin_id:
-        return PluginLoggerAdapter(logger, {'plugin_id': plugin_id})
-
+        logger.plugin_id = plugin_id
+    
     return logger
 
 

src/plugin_system/base_plugin.py

@@ -86,9 +86,7 @@ class BasePlugin(ABC):
         self.display_manager: Any = display_manager
         self.cache_manager: Any = cache_manager
         self.plugin_manager: Any = plugin_manager
-        # get_logger returns a PluginLoggerAdapter here (plugin_id given), which
-        # stamps every record with plugin_id so it survives into formatted output.
-        self.logger = get_logger(f"plugin.{plugin_id}", plugin_id=plugin_id)
+        self.logger: logging.Logger = get_logger(f"plugin.{plugin_id}", plugin_id=plugin_id)
         self.enabled: bool = config.get("enabled", True)
 
         self.logger.info("Initialized plugin: %s", plugin_id)

web_interface/blueprints/api_v3.py

@@ -6891,29 +6891,6 @@ def list_plugin_assets():
         logger.error('Unhandled exception', exc_info=True)
         return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details'}), 500
 
-@api_v3.route('/display/current-status', methods=['GET'])
-def get_current_display_status():
-    """Return the display mode/plugin currently intended to be shown.
-
-    Published by the display process (display_controller._publish_current_mode_state)
-    to the shared cache whenever the active mode changes, so the web UI (e.g. the
-    System Logs page) can show what's on screen without querying the display
-    process directly.
-    """
-    try:
-        cache = _ensure_cache_manager()
-        state = cache.get('display_current_state', max_age=120)
-        if state is None:
-            state = {
-                'mode': None,
-                'plugin_id': None,
-                'last_updated': None,
-            }
-        return jsonify({'status': 'success', 'data': state})
-    except Exception:
-        logger.error('Error in get_current_display_status', exc_info=True)
-        return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details'}), 500
-
 @api_v3.route('/logs', methods=['GET'])
 def get_logs():
     """Get system logs from journalctl"""

web_interface/templates/v3/partials/logs.html

@@ -4,17 +4,6 @@
         <p class="mt-1 text-sm text-gray-600">View real-time logs from the LED matrix service for troubleshooting.</p>
     </div>
 
-    <!-- Currently displayed plugin -->
-    <div id="current-plugin-banner" class="hidden mb-4 flex items-center gap-2 text-sm bg-blue-50 border border-blue-200 text-blue-800 rounded-lg px-3 py-2">
-        <i class="fas fa-tv"></i>
-        <span>Now showing: <strong id="current-plugin-mode">-</strong>
-            <span id="current-plugin-id-wrap" class="hidden">(plugin: <code id="current-plugin-id" class="bg-blue-100 px-1 rounded"></code>)</span>
-        </span>
-        <button id="current-plugin-filter-btn" class="hidden ml-auto btn bg-blue-600 hover:bg-blue-700 text-white px-2 py-0.5 rounded text-xs">
-            Filter to this plugin
-        </button>
-    </div>
-
     <!-- Controls -->
     <div class="flex flex-wrap items-center justify-between gap-4 mb-6">
         <div class="flex items-center space-x-4">
@@ -38,11 +27,6 @@
                 <option value="INFO">Info & Above</option>
             </select>
 
-            <!-- Plugin Filter -->
-            <select id="log-plugin-filter" class="form-control text-sm">
-                <option value="">All Plugins</option>
-            </select>
-
             <!-- Search -->
             <div class="relative">
                 <input type="text" id="log-search" placeholder="Search logs..." class="form-control text-sm pl-8 pr-4 py-1 w-48">
@@ -155,7 +139,6 @@ window._filteredLogs = [];
     const realtimeToggle = document.getElementById('log-realtime-toggle');
     const refreshBtn = document.getElementById('refresh-logs-btn');
     const levelFilter = document.getElementById('log-level-filter');
-    const pluginFilter = document.getElementById('log-plugin-filter');
     const searchInput = document.getElementById('log-search');
     const autoscrollToggle = document.getElementById('log-autoscroll');
     const clearBtn = document.getElementById('clear-logs-btn');
@@ -177,11 +160,6 @@ window._filteredLogs = [];
         levelFilter.parentNode.replaceChild(newFilter, levelFilter);
         newFilter.addEventListener('change', filterLogs);
     }
-    if (pluginFilter) {
-        const newFilter = pluginFilter.cloneNode(true);
-        pluginFilter.parentNode.replaceChild(newFilter, pluginFilter);
-        newFilter.addEventListener('change', filterLogs);
-    }
     if (searchInput) {
         const newInput = searchInput.cloneNode(true);
         searchInput.parentNode.replaceChild(newInput, searchInput);
@@ -202,24 +180,6 @@ window._filteredLogs = [];
         downloadBtn.parentNode.replaceChild(newBtn, downloadBtn);
         newBtn.addEventListener('click', downloadLogs);
     }
-    const currentPluginFilterBtn = document.getElementById('current-plugin-filter-btn');
-    if (currentPluginFilterBtn) {
-        const newBtn = currentPluginFilterBtn.cloneNode(true);
-        currentPluginFilterBtn.parentNode.replaceChild(newBtn, currentPluginFilterBtn);
-        newBtn.addEventListener('click', function() {
-            const pluginFilterEl = document.getElementById('log-plugin-filter');
-            if (pluginFilterEl && window._currentPluginId) {
-                pluginFilterEl.value = window._currentPluginId;
-                filterLogs();
-            }
-        });
-    }
-
-    refreshCurrentPluginStatus();
-    if (window._currentPluginPollTimer) {
-        clearInterval(window._currentPluginPollTimer);
-    }
-    window._currentPluginPollTimer = setInterval(refreshCurrentPluginStatus, 5000);
 
     // Handle window resize for responsive height
     window.addEventListener('resize', function() {
@@ -324,25 +284,37 @@ function processLogs(logsText, append = false) {
         // Skip empty lines
         if (!line.trim()) return;
 
-        // journalctl (--output=short-iso) emits: "YYYY-MM-DDTHH:MM:SS+ZZZZ hostname service[pid]: message"
-        // Example: "2024-01-15T10:23:45+0000 raspberrypi ledmatrix[1234]: INFO - plugin.nhl_scoreboard - [Plugin: nhl_scoreboard] Updated scores"
-        // Also accept the older syslog-style "MMM DD HH:MM:SS" timestamp for compatibility.
-
+        // Try to parse journalctl format: "MMM DD HH:MM:SS hostname service[pid]: message"
+        // Example: "Oct 13 14:23:45 raspberrypi ledmatrix[1234]: INFO: Starting display"
+        
         let timestamp = '';
         let level = 'INFO';
         let message = line;
-        let plugin = '';
-        let rest = null;
-
-        const isoMatch = line.match(/^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:[+-]\d{2}:?\d{2}|Z))\s+(.*)$/);
-        const syslogMatch = !isoMatch && line.match(/^([A-Z][a-z]{2}\s+\d{1,2}\s+\d{2}:\d{2}:\d{2})\s+(.*)$/);
-
-        if (isoMatch) {
-            timestamp = isoMatch[1];
-            rest = isoMatch[2];
-        } else if (syslogMatch) {
-            timestamp = syslogMatch[1];
-            rest = syslogMatch[2];
+        
+        // Extract timestamp (first part before hostname)
+        const timestampMatch = line.match(/^([A-Z][a-z]{2}\s+\d{1,2}\s+\d{2}:\d{2}:\d{2})/);
+        if (timestampMatch) {
+            timestamp = timestampMatch[1];
+            
+            // Find the message part (after service name and pid)
+            const messageMatch = line.match(/:\s*(.+)$/);
+            if (messageMatch) {
+                message = messageMatch[1];
+                
+                // Detect log level from message
+                if (message.match(/\b(ERROR|CRITICAL|FATAL)\b/i)) {
+                    level = 'ERROR';
+                } else if (message.match(/\b(WARNING|WARN)\b/i)) {
+                    level = 'WARNING';
+                } else if (message.match(/\bDEBUG\b/i)) {
+                    level = 'DEBUG';
+                } else if (message.match(/\bINFO\b/i)) {
+                    level = 'INFO';
+                }
+                
+                // Clean up level prefix from message if it exists
+                message = message.replace(/^(ERROR|WARNING|WARN|INFO|DEBUG):\s*/i, '');
+            }
         } else {
             // If no timestamp, use current time
             timestamp = new Date().toLocaleString('en-US', {
@@ -355,58 +327,14 @@ function processLogs(logsText, append = false) {
             });
         }
 
-        if (rest !== null) {
-            // Find the message part (after hostname + service name/pid)
-            const messageMatch = rest.match(/:\s*(.+)$/);
-            if (messageMatch) {
-                message = messageMatch[1];
-
-                // Detect log level from message
-                if (message.match(/\b(ERROR|CRITICAL|FATAL)\b/i)) {
-                    level = 'ERROR';
-                } else if (message.match(/\b(WARNING|WARN)\b/i)) {
-                    level = 'WARNING';
-                } else if (message.match(/\bDEBUG\b/i)) {
-                    level = 'DEBUG';
-                } else if (message.match(/\bINFO\b/i)) {
-                    level = 'INFO';
-                }
-
-                // Clean up level prefix from message if it exists
-                message = message.replace(/^(ERROR|WARNING|WARN|INFO|DEBUG)\s*[-:]\s*/i, '');
-
-                // journalctl already carries its own timestamp; strip the app's
-                // internal "YYYY-MM-DD HH:MM:SS.mmm - LEVEL - logger.name - "
-                // prefix (see ContextualFormatter in src/logging_config.py) so
-                // it isn't duplicated in the displayed message.
-                message = message.replace(
-                    /^\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2}(?:\.\d+)?\s*-\s*(ERROR|WARNING|WARN|INFO|DEBUG|CRITICAL)\s*-\s*[\w.]+\s*-\s*/i,
-                    ''
-                );
-
-                // Extract the "[Plugin: <id>]" context tag emitted by
-                // ContextualFormatter for every plugin logger (see
-                // src/logging_config.py PluginLoggerAdapter), and pull it out
-                // of the displayed message into its own field.
-                const pluginMatch = message.match(/^\[Plugin:\s*([^\]]+)\]\s*/);
-                if (pluginMatch) {
-                    plugin = pluginMatch[1].trim();
-                    message = message.replace(pluginMatch[0], '');
-                }
-            } else {
-                message = rest;
-            }
-        }
-
         const logEntry = {
             timestamp: timestamp,
             level: level,
-            plugin: plugin,
             message: message,
             raw: line,
             id: Date.now() + Math.random()
         };
-
+        
         // Don't add duplicate entries when appending
         if (!append || !window._allLogs.find(log => log.raw === line)) {
             window._allLogs.push(logEntry);
@@ -418,26 +346,9 @@ function processLogs(logsText, append = false) {
         window._allLogs = window._allLogs.slice(-window._MAX_LOGS);
     }
 
-    updatePluginFilterOptions();
     filterLogs();
 }
 
-function updatePluginFilterOptions() {
-    const pluginFilterEl = document.getElementById('log-plugin-filter');
-    if (!pluginFilterEl) return;
-
-    const previousValue = pluginFilterEl.value;
-    const plugins = Array.from(new Set(window._allLogs.map(log => log.plugin).filter(Boolean))).sort();
-
-    pluginFilterEl.innerHTML = '<option value="">All Plugins</option>' +
-        plugins.map(p => `<option value="${escapeHtml(p)}">${escapeHtml(p)}</option>`).join('');
-
-    // Restore previous selection if it's still a valid option
-    if (previousValue && plugins.includes(previousValue)) {
-        pluginFilterEl.value = previousValue;
-    }
-}
-
 function renderLogs() {
     if (window._filteredLogs.length === 0) {
         showEmptyState();
@@ -453,14 +364,10 @@ function renderLogs() {
     window._filteredLogs.forEach(log => {
         const logElement = document.createElement('div');
         logElement.className = `log-entry py-1 px-2 hover:bg-gray-800 rounded transition-colors duration-150 ${getLogLevelClass(log.level)}`;
-        const pluginBadge = log.plugin
-            ? `<span class="log-plugin flex-shrink-0 px-2 py-0.5 rounded text-xs font-semibold bg-purple-700 text-white" title="Plugin: ${escapeHtml(log.plugin)}">${escapeHtml(log.plugin)}</span>`
-            : '';
         logElement.innerHTML = `
             <div class="flex items-start gap-3 text-xs font-mono">
                 <span class="log-timestamp text-gray-400 flex-shrink-0 w-32">${escapeHtml(log.timestamp)}</span>
                 <span class="log-level flex-shrink-0 px-2 py-0.5 rounded text-xs font-semibold ${getLogLevelBadgeClass(log.level)}">${log.level}</span>
-                ${pluginBadge}
                 <span class="log-message flex-1 ${getLogLevelTextClass(log.level)} break-words">${escapeHtml(log.message)}</span>
             </div>
         `;
@@ -503,12 +410,10 @@ function getLogLevelTextClass(level) {
 
 function filterLogs() {
     const levelFilterEl = document.getElementById('log-level-filter');
-    const pluginFilterEl = document.getElementById('log-plugin-filter');
     const searchEl = document.getElementById('log-search');
     if (!levelFilterEl || !searchEl) return;
-
+    
     const levelFilter = levelFilterEl.value;
-    const pluginFilter = pluginFilterEl ? pluginFilterEl.value : '';
     const searchTerm = searchEl.value.toLowerCase();
 
     window._filteredLogs = window._allLogs.filter(log => {
@@ -525,14 +430,8 @@ function filterLogs() {
             }
         }
 
-        // Plugin filter
-        if (pluginFilter && log.plugin !== pluginFilter) {
-            return false;
-        }
-
         // Search filter
-        if (searchTerm && !log.message.toLowerCase().includes(searchTerm) &&
-            !(log.plugin && log.plugin.toLowerCase().includes(searchTerm))) {
+        if (searchTerm && !log.message.toLowerCase().includes(searchTerm)) {
             return false;
         }
 
@@ -718,45 +617,10 @@ function escapeHtml(text) {
     return div.innerHTML;
 }
 
-function refreshCurrentPluginStatus() {
-    fetch('/api/v3/display/current-status')
-        .then(response => response.json())
-        .then(data => {
-            if (data.status !== 'success' || !data.data) return;
-            const state = data.data;
-            const banner = document.getElementById('current-plugin-banner');
-            const modeEl = document.getElementById('current-plugin-mode');
-            const idWrap = document.getElementById('current-plugin-id-wrap');
-            const idEl = document.getElementById('current-plugin-id');
-            const filterBtn = document.getElementById('current-plugin-filter-btn');
-            if (!banner || !modeEl) return;
-
-            window._currentPluginId = state.plugin_id || null;
-
-            modeEl.textContent = state.mode || 'unknown';
-            if (state.plugin_id) {
-                idEl.textContent = state.plugin_id;
-                idWrap.classList.remove('hidden');
-                if (filterBtn) filterBtn.classList.remove('hidden');
-            } else {
-                idWrap.classList.add('hidden');
-                if (filterBtn) filterBtn.classList.add('hidden');
-            }
-            banner.classList.remove('hidden');
-        })
-        .catch(() => {
-            // Silently ignore - banner just stays hidden/stale
-        });
-}
-
 // Cleanup on page unload
 window.addEventListener('beforeunload', function() {
     if (window._logsEventSource) {
         window._logsEventSource.close();
     }
-    if (window._currentPluginPollTimer) {
-        clearInterval(window._currentPluginPollTimer);
-        window._currentPluginPollTimer = null;
-    }
 });
 </script>