Compare commits

..
Author SHA1 Message Date
ChuckBuilds 22f0a96cbb fix(security): stop interpolating req_file/pip-output into log calls
The previous commit's redaction (mutating result.stderr/stdout right after
each subprocess.run()) didn't clear CodeQL's clear-text-logging alerts --
same lesson as the path-injection fix earlier in this PR: a static
analyzer can't tell "this value was already sanitised two lines up" from
"this is still the raw tainted value" just by looking at a single log
call in isolation, so it conservatively keeps flagging it regardless of
what the redaction function actually does.

Removed all dynamic interpolation (req_file, result.stderr) from the 3
flagged logger.warning() calls entirely, replacing them with fixed
messages plus (for the one that had it) result.returncode, which is a
plain int with no possible taint. The full redacted detail is still
available where it actually matters -- in the returned
CompletedProcess.stderr/stdout and the "note" text -- just not duplicated
into a log line a scanner has to reason about in isolation.

Re-verified: all 6 test_permission_utils.py tests still pass (they assert
on the returned result, not log call arguments), plus the full
test_plugin_loader.py/test_store_manager_caches.py/test_plugin_system.py
suite (71 passed, 1 pre-existing deselect, 4 subtests).
2026-07-10 16:28:29 -04:00
ChuckBuilds 2a99cf6e64 fix(security): redact URL credentials from pip subprocess output before logging
CodeQL flagged 3 clear-text-logging-of-secrets alerts in
install_requirements_file() (src/common/permission_utils.py:353,360,371).
Pre-existing on main, unrelated to this PR's own diff, but now visible
since the path-injection alerts that previously took priority in the
annotation list are fixed.

The underlying risk is real: pip can echo a private index URL's embedded
basic-auth credentials (from a requirements.txt --index-url line or
PIP_INDEX_URL) back verbatim in its own stderr/stdout on failure, and this
function both logs that output directly and returns it to callers --
store_manager.py's _install_dependencies() logs result.stderr from this
same function too.

Added _redact_url_credentials(), applied immediately after each of the two
subprocess.run() calls (mutating result.stderr/stdout in place) rather
than patching each log call site individually. This closes the leak at
the source: every downstream use -- the three flagged log lines, the
"note" string embedded in the returned stdout, and store_manager.py's own
logging of the returned result -- gets the redacted text for free.

Verified the fixed-phrase "denied" check (`"a password is required" in
result.stderr`) is unaffected, since URL syntax and those phrases don't
overlap -- covered explicitly by
test_does_not_touch_denied_check_phrases. Added
test/test_permission_utils.py (6 tests) covering the redaction helper
directly and both subprocess.run() call sites (the sudo-wrapper branch,
which this repo's scripts/fix_perms/safe_pip_install.sh makes live, and
the no-wrapper fallback branch). All pass.
2026-07-10 16:18:51 -04:00
ChuckBuilds b89653c836 fix(security): replace basename-only sanitiser with a trusted-enumeration check
The previous commit's os.path.basename() + os.path.join() pattern (which a
pre-existing code comment claimed CodeQL recognises as a sanitiser) did not
actually clear the alert -- the next CodeQL run still flagged the same 2
sink lines, plus a new one at the os.path.join() call itself. Taking a
substring of tainted data apparently isn't treated as a barrier by this
query, whatever the comment assumed.

Replaced it with find_trusted_subdir(): enumerate the trusted plugins_dir
via os.scandir() and only use a name that scandir itself produced, matched
by equality against the caller's requested name. The path is then built
from that enumerated entry, not from the caller's string -- a value
sourced from iterating a trusted, non-tainted directory carries no taint
regardless of what it happens to equal, which is a stronger and more
conventional allowlist-style barrier than string-stripping. Applied
identically in both PluginLoader.install_dependencies() and
StoreManager._install_dependencies(), sharing one implementation.

Re-verified: all 65 tests across test_plugin_loader.py (20, including the
2 new security regression tests), test_store_manager_caches.py (35),
test_plugin_system.py (10) pass, plus the full CI plugin-safety suite
(test_harness.py/test_visual_rendering.py/test_plugin_matrix.py: 52
passed, 2 pre-existing skips).
2026-07-10 15:17:25 -04:00
ChuckBuilds d59a66133b fix(security): close path-injection gap in dependency-satisfaction checks
CodeQL flagged 2 new high-severity "uncontrolled data used in path
expression" alerts at the open() calls inside this PR's new
requirements_has_real_deps()/requirements_are_satisfied() -- both are
reachable from paths that were never run through the basename+trusted-base
sanitiser this codebase already uses elsewhere:

- PluginLoader.install_dependencies() only applied that sanitiser when its
  optional plugins_dir argument was actually passed; the "no plugins_dir"
  branch trusted plugin_dir_real directly. Made plugins_dir required (not
  Optional) so that branch can't exist, and added an explicit guard in
  load_plugin() so install_deps=True without a plugins_dir fails loudly
  instead of silently. Production's only real caller (PluginManager) always
  passes plugins_dir already; the harness/dev-server/render-plugin callers
  all use install_deps=False and are unaffected.

- StoreManager._install_dependencies() never sanitised plugin_path at all,
  and its call sites ultimately derive that path from a plugin's own
  manifest.json "id" field (install_plugin_from_url) -- a malicious plugin
  could otherwise point requirements_file outside plugins_dir. Applied the
  same os.path.basename()-based containment pattern PluginLoader already
  uses (and that CodeQL recognises as a real sanitiser).

Added test_install_dependencies_requires_plugins_dir and
test_install_dependencies_rejects_path_outside_plugins_dir to lock in the
actual security property, not just quiet the scanner. Verified: all 20
tests in test_plugin_loader.py pass, plus the PR's existing test plan
(test_plugin_system.py, test_store_manager_caches.py: 53 passed) and the
full CI plugin-safety suite (test_harness.py, test_visual_rendering.py,
test_plugin_matrix.py: 52 passed, 2 pre-existing skips) all still pass.
2026-07-10 15:10:49 -04:00
ChuckBuildsandClaude Sonnet 5 d86dc5914b fix(plugins): replace dependency marker files with a real satisfaction check
The .dependencies_installed hash-marker system only tracked "was this exact
requirements.txt hashed before" — not whether the packages it names are
actually present. That made it fragile (a wiped venv, a manually removed
package, or a lost/corrupted marker forces a needless full pip reinstall or,
worse, a false skip) and produced dead weight for the ~10 plugins whose
requirements.txt is comment-only (they still paid a pip subprocess on first
boot before a marker existed).

Replace it with requirements_are_satisfied() in plugin_loader.py, which
checks each real requirement line against importlib.metadata directly, so
install_dependencies() only shells out to pip when something is actually
missing or version-mismatched. Drops the marker file entirely: removed all
marker read/write sites in plugin_loader.py and store_manager.py, the
now-pointless marker-cleanup step in the git-update path, the unused legacy
marker implementation in plugin_manager.py, and the already-stale
clear_dependency_markers.sh script.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KEZK1P1Q1fu5pcuVrkrCFZ
2026-07-10 12:08:22 -04:00
7 changed files with 16 additions and 142 deletions
+8 -35
View File
@@ -33,8 +33,7 @@ else:
from contextlib import contextmanager from contextlib import contextmanager
from PIL import Image, ImageDraw, ImageFont from PIL import Image, ImageDraw, ImageFont
import time import time
from collections import OrderedDict from typing import Dict, Any, List, Optional
from typing import Dict, Any, List, Optional, Tuple
import logging import logging
import math import math
import freetype import freetype
@@ -181,25 +180,14 @@ class DisplayManager:
# the logical image is blitted to the matrix unchanged. # the logical image is blitted to the matrix unchanged.
self._double_sided = None # dict {copies, axis, logical_width, logical_height} or None self._double_sided = None # dict {copies, axis, logical_width, logical_height} or None
self._physical_image = None # full-chain buffer reused each frame when tiling self._physical_image = None # full-chain buffer reused each frame when tiling
# Text-width measurement cache: (text, id(font)) -> (width, font_ref) # Text-width measurement cache: (text, id(font)) -> pixel_width
# Avoids re-measuring the same string+font on every display() call. # Avoids re-measuring the same string+font on every display() call.
# LRU-bounded: keys embed the TEXT, so changing strings (a clock, a
# live score) would otherwise grow it forever on a 24/7 service.
# Entries hold a strong reference to the font so its id() can't be
# recycled by a different font object — an id-keyed cache without
# the reference can return the WRONG width after garbage collection.
# Cleared on _load_fonts() so stale entries don't survive a font reload. # Cleared on _load_fonts() so stale entries don't survive a font reload.
self._text_width_cache: "OrderedDict[tuple, Tuple[int, Any]]" = OrderedDict() self._text_width_cache: Dict[tuple, int] = {}
self._TEXT_WIDTH_CACHE_MAX = 1024
# Snapshot settings for web preview integration (service writes, web reads) # Snapshot settings for web preview integration (service writes, web reads)
self._snapshot_path = "/tmp/led_matrix_preview.png" # nosec B108 - fixed path intentional; web UI reads same path self._snapshot_path = "/tmp/led_matrix_preview.png" # nosec B108 - fixed path intentional; web UI reads same path
self._snapshot_min_interval_sec = 0.2 # max ~5 fps self._snapshot_min_interval_sec = 0.2 # max ~5 fps
self._last_snapshot_ts = 0.0 self._last_snapshot_ts = 0.0
# Snapshot failures are logged as warnings, rate-limited so a
# persistent failure (e.g. an unwritable file) can't spam the log —
# but is never silent: the snapshot's mtime doubles as the web UI's
# hardware-liveness signal, so a quiet failure makes health checks lie.
self._snapshot_fail_log_ts = 0.0
# Scrolling state tracking for graceful updates # Scrolling state tracking for graceful updates
self._scrolling_state = { self._scrolling_state = {
@@ -711,15 +699,12 @@ class DisplayManager:
Results are cached by (text, font identity) so plugins that measure Results are cached by (text, font identity) so plugins that measure
the same string every frame (e.g. to centre a score) pay only one the same string every frame (e.g. to centre a score) pay only one
measurement per unique (text, font) pair. The entry keeps the font measurement per unique (text, font) pair.
alive so its id() can't be recycled, and the cache is LRU-bounded so
ever-changing text (clocks, tickers) can't grow it without limit.
""" """
cache_key = (text, id(font)) cache_key = (text, id(font))
cached = self._text_width_cache.get(cache_key) cached = self._text_width_cache.get(cache_key)
if cached is not None: if cached is not None:
self._text_width_cache.move_to_end(cache_key) return cached
return cached[0]
try: try:
if isinstance(font, freetype.Face): if isinstance(font, freetype.Face):
@@ -734,9 +719,7 @@ class DisplayManager:
logger.error("Error getting text width: %s", e) logger.error("Error getting text width: %s", e)
return 0 return 0
self._text_width_cache[cache_key] = (width, font) self._text_width_cache[cache_key] = width
while len(self._text_width_cache) > self._TEXT_WIDTH_CACHE_MAX:
self._text_width_cache.popitem(last=False)
return width return width
def get_font_height(self, font): def get_font_height(self, font):
@@ -1181,15 +1164,5 @@ class DisplayManager:
pass pass
self._last_snapshot_ts = now self._last_snapshot_ts = now
except Exception as e: except Exception as e:
# Snapshot failures must never break displaybut they must not # Snapshot failures should never break display; log at debug to avoid noise
# be silent either: the snapshot's mtime is the web UI's display logger.debug(f"Snapshot write skipped: {e}")
# mirror AND its hardware-liveness proxy, so a quietly failing
# write freezes the mirror and makes health checks lie (seen in
# the field: a stale root-owned /tmp file froze it for a day).
# Warn at most once per 5 minutes to avoid log spam.
if (now - self._snapshot_fail_log_ts) > 300:
self._snapshot_fail_log_ts = now
logger.warning("Snapshot write failing (web preview/health "
"mirror is stale): %s", e)
else:
logger.debug(f"Snapshot write skipped: {e}")
+5 -18
View File
@@ -35,7 +35,6 @@ import urllib.request
import zipfile import zipfile
import tempfile import tempfile
import time import time
from collections import OrderedDict
from pathlib import Path from pathlib import Path
from PIL import ImageFont from PIL import ImageFont
from typing import Dict, Tuple, Optional, Union, Any, List from typing import Dict, Tuple, Optional, Union, Any, List
@@ -59,13 +58,7 @@ class FontManager:
# Font discovery and catalog # Font discovery and catalog
self.font_catalog: Dict[str, str] = {} # family_name -> file_path self.font_catalog: Dict[str, str] = {} # family_name -> file_path
self.font_cache: Dict[str, Union[ImageFont.FreeTypeFont, freetype.Face]] = {} # (family, size) -> font self.font_cache: Dict[str, Union[ImageFont.FreeTypeFont, freetype.Face]] = {} # (family, size) -> font
# (text, id(font)) -> ((width, height, baseline), font_ref). self.metrics_cache: Dict[str, Tuple[int, int, int]] = {} # (text, font_id) -> (width, height, baseline)
# LRU-bounded — keys embed the measured TEXT, so changing strings
# (clocks, live scores) would otherwise grow it forever. Entries
# keep the font alive so its id() can't be recycled by a different
# font object (which would silently return wrong metrics).
self.metrics_cache: "OrderedDict[Any, Tuple[Tuple[int, int, int], Any]]" = OrderedDict()
self._METRICS_CACHE_MAX = 1024
# Plugin font management # Plugin font management
self.plugin_fonts: Dict[str, Dict[str, Any]] = {} # plugin_id -> font_manifest self.plugin_fonts: Dict[str, Dict[str, Any]] = {} # plugin_id -> font_manifest
@@ -514,14 +507,10 @@ class FontManager:
Returns: Returns:
Tuple of (width, height, baseline_offset) Tuple of (width, height, baseline_offset)
""" """
# Key on the text itself (hash(text) could collide) + font identity; cache_key = f"{hash(text)}_{id(font)}"
# the entry below keeps the font referenced so the id stays valid.
cache_key = (text, id(font))
cached = self.metrics_cache.get(cache_key) if cache_key in self.metrics_cache:
if cached is not None: return self.metrics_cache[cache_key]
self.metrics_cache.move_to_end(cache_key)
return cached[0]
try: try:
if isinstance(font, freetype.Face): if isinstance(font, freetype.Face):
@@ -558,9 +547,7 @@ class FontManager:
baseline = 10 baseline = 10
result = (width, height, baseline) result = (width, height, baseline)
self.metrics_cache[cache_key] = (result, font) self.metrics_cache[cache_key] = result
while len(self.metrics_cache) > self._METRICS_CACHE_MAX:
self.metrics_cache.popitem(last=False)
return result return result
def get_font_height(self, font: Union[ImageFont.FreeTypeFont, freetype.Face]) -> int: def get_font_height(self, font: Union[ImageFont.FreeTypeFont, freetype.Face]) -> int:
+1 -12
View File
@@ -11,7 +11,6 @@ import json
import sys import sys
import time import time
import threading import threading
import types
from pathlib import Path from pathlib import Path
from typing import Dict, List, Optional, Any from typing import Dict, List, Optional, Any
import logging import logging
@@ -744,18 +743,8 @@ class PluginManager:
# If resource monitor exists, wrap the call # If resource monitor exists, wrap the call
def monitored_update(): def monitored_update():
self.resource_monitor.monitor_call(plugin_id, plugin_instance.update) self.resource_monitor.monitor_call(plugin_id, plugin_instance.update)
# SimpleNamespace stores `update` as an *instance*
# attribute, so attribute lookup returns the plain
# function object as-is. A dynamically-built class
# (`type(..., {'update': monitored_update})`) instead
# stores it as a *class* attribute, which the
# descriptor protocol turns into a bound method on
# access -- silently prepending the instance as an
# implicit first argument to a function that takes
# none, raising "monitored_update() takes 0
# positional arguments but 1 was given" on every call.
success = self.plugin_executor.execute_update( success = self.plugin_executor.execute_update(
types.SimpleNamespace(update=monitored_update), type('obj', (object,), {'update': monitored_update})(),
plugin_id plugin_id
) )
else: else:
@@ -454,18 +454,6 @@ class VisualTestDisplayManager:
"""Check if display is currently scrolling.""" """Check if display is currently scrolling."""
return self._scrolling_state['is_scrolling'] return self._scrolling_state['is_scrolling']
def process_deferred_updates(self):
"""Process any deferred updates (no-op for testing).
Several ticker-style plugins (news, odds-ticker, leaderboard,
stock-news, stocks) call this unconditionally between
set_scrolling_state() and their scroll-position update, mirroring the
real display_manager's deferred-update queue. This double has no such
queue, so there is nothing to process — the no-op just lets those
plugins render under the harness instead of raising AttributeError.
"""
pass
# ------------------------------------------------------------------ # ------------------------------------------------------------------
# Utility methods # Utility methods
# ------------------------------------------------------------------ # ------------------------------------------------------------------
+1 -5
View File
@@ -38,11 +38,7 @@ def mock_cache_manager():
mock._memory_cache_timestamps = {} mock._memory_cache_timestamps = {}
mock.cache_dir = "/tmp/test_cache" mock.cache_dir = "/tmp/test_cache"
def mock_get(key: str, max_age: Optional[int] = 300, def mock_get(key: str, max_age: int = 300) -> Optional[Dict]:
memory_ttl: Optional[int] = None) -> Optional[Dict]:
# Signature mirrors CacheManager.get — keep in sync or callers
# passing keyword args (health tracker, resource monitor) break
# only in tests, hiding real-API compatibility.
return mock._memory_cache.get(key) return mock._memory_cache.get(key)
def mock_set(key: str, data: Dict, ttl: Optional[int] = None) -> None: def mock_set(key: str, data: Dict, ttl: Optional[int] = None) -> None:
-10
View File
@@ -172,16 +172,6 @@ class TestVisualDisplayManager:
vdm.set_scrolling_state(False) vdm.set_scrolling_state(False)
assert vdm.is_currently_scrolling() is False assert vdm.is_currently_scrolling() is False
def test_process_deferred_updates_is_noop(self):
# Ticker-style plugins (news, odds-ticker, leaderboard, stock-news,
# stocks) call this unconditionally alongside set_scrolling_state();
# it must exist and be harmless so those plugins render under the
# harness instead of raising AttributeError.
vdm = VisualTestDisplayManager(width=128, height=32)
vdm.set_scrolling_state(True)
vdm.process_deferred_updates() # should not raise
assert vdm.is_currently_scrolling() is True
def test_format_date_with_ordinal(self): def test_format_date_with_ordinal(self):
from datetime import datetime from datetime import datetime
vdm = VisualTestDisplayManager(width=128, height=32) vdm = VisualTestDisplayManager(width=128, height=32)
+1 -50
View File
@@ -3,7 +3,6 @@ from unittest.mock import MagicMock, patch
from pathlib import Path from pathlib import Path
from src.plugin_system.plugin_manager import PluginManager from src.plugin_system.plugin_manager import PluginManager
from src.plugin_system.plugin_state import PluginState from src.plugin_system.plugin_state import PluginState
from src.plugin_system.resource_monitor import PluginResourceMonitor
class TestPluginManager: class TestPluginManager:
"""Test PluginManager functionality.""" """Test PluginManager functionality."""
@@ -75,58 +74,10 @@ class TestPluginManager:
# No manifest in pm.plugin_manifests # No manifest in pm.plugin_manifests
result = pm.load_plugin("non_existent_plugin") result = pm.load_plugin("non_existent_plugin")
assert result is False assert result is False
assert pm.state_manager.get_state("non_existent_plugin") == PluginState.ERROR assert pm.state_manager.get_state("non_existent_plugin") == PluginState.ERROR
def test_run_scheduled_updates_calls_update_with_resource_monitor(
self, mock_config_manager, mock_display_manager, mock_cache_manager
):
"""Regression test: run_scheduled_updates() must actually call a
plugin's update() when self.resource_monitor is set (as it is in
every real deployment -- display_controller.py and web_interface/
app.py both assign a real PluginResourceMonitor after construction).
Previously, the resource_monitor branch wrapped the call in a
function stored as a *class* attribute on a dynamically-built type
(`type('obj', (object,), {'update': monitored_update})()`), which
the descriptor protocol turns into a bound method on access --
silently passing the synthetic instance as an implicit first
argument to monitored_update(), which takes none. Every plugin's
scheduled update failed with "monitored_update() takes 0 positional
arguments but 1 was given" and was silently swallowed into a
circuit-breaker retry loop that never succeeded, so plugin data
(scores, odds, etc.) never refreshed.
"""
with patch('src.plugin_system.plugin_manager.ensure_directory_permissions'):
pm = PluginManager(
plugins_dir="plugins",
config_manager=mock_config_manager,
display_manager=mock_display_manager,
cache_manager=mock_cache_manager
)
plugin_instance = MagicMock()
plugin_instance.enabled = True
plugin_instance.update = MagicMock()
pm.plugins["test_plugin"] = plugin_instance
pm.plugin_manifests["test_plugin"] = {"update_interval": 10}
pm.state_manager.set_state("test_plugin", PluginState.ENABLED)
# Plain MagicMock, not the mock_cache_manager fixture: this test
# is about run_scheduled_updates() actually invoking update()
# through the resource-monitor wrapper, not about
# PluginResourceMonitor's own cache-backed metrics persistence
# (which calls cache_manager.get(..., memory_ttl=...) --
# a kwarg the fixture's mock_get() doesn't accept).
pm.resource_monitor = PluginResourceMonitor(MagicMock())
pm.run_scheduled_updates(current_time=time.time())
plugin_instance.update.assert_called_once()
assert "test_plugin" in pm.plugin_last_update
assert pm.state_manager.get_state("test_plugin") == PluginState.ENABLED
class TestPluginLoader: class TestPluginLoader:
"""Test PluginLoader functionality.""" """Test PluginLoader functionality."""