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
6 changed files with 11 additions and 120 deletions
+5 -20
View File
@@ -199,10 +199,6 @@ class DisplayController:
self.wifi_status_file = WIFI_STATUS_FILE self.wifi_status_file = WIFI_STATUS_FILE
self.wifi_status_active = False self.wifi_status_active = False
self.wifi_status_expires_at: Optional[float] = None self.wifi_status_expires_at: Optional[float] = None
# _check_wifi_status_message throttle state (checked at frame rate,
# stat'd at most once per second)
self._wifi_status_check_ts = 0.0
self._wifi_status_last_result: Optional[Dict[str, Any]] = None
# Plugin display() signature cache — must be initialised before the plugin # Plugin display() signature cache — must be initialised before the plugin
# loading loop below so the .pop() invalidation at load time is always safe. # loading loop below so the .pop() invalidation at load time is always safe.
@@ -1639,7 +1635,7 @@ class DisplayController:
self._sleep_with_plugin_updates(60) self._sleep_with_plugin_updates(60)
continue continue
logger.debug("Display active, processing mode: %s", self.current_display_mode) logger.info(f"Display active, processing mode: {self.current_display_mode}")
# Plugins update on their own schedules - no forced sync updates needed # Plugins update on their own schedules - no forced sync updates needed
# Each plugin has its own update_interval and background services # Each plugin has its own update_interval and background services
@@ -1807,7 +1803,7 @@ class DisplayController:
if self.plugin_manager and hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker: if self.plugin_manager and hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
should_skip = self.plugin_manager.health_tracker.should_skip_plugin(plugin_id) should_skip = self.plugin_manager.health_tracker.should_skip_plugin(plugin_id)
if should_skip: if should_skip:
logger.info("Skipping plugin %s due to circuit breaker (mode: %s)", plugin_id, active_mode) logger.info(f"Skipping plugin {plugin_id} due to circuit breaker (mode: {active_mode})")
display_result = False display_result = False
# Skip to next mode - let existing logic handle it # Skip to next mode - let existing logic handle it
manager_to_display = None manager_to_display = None
@@ -1865,7 +1861,7 @@ class DisplayController:
if isinstance(result, bool): if isinstance(result, bool):
display_result = result display_result = result
if not display_result: if not display_result:
logger.info("Plugin %s display() returned False for mode %s", plugin_id, active_mode) logger.info(f"Plugin {plugin_id} display() returned False for mode {active_mode}")
# Record success if display completed without exception # Record success if display completed without exception
if self.plugin_manager and hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker: if self.plugin_manager and hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
@@ -2358,16 +2354,6 @@ class DisplayController:
Returns None on any error or if message is expired/invalid. Returns None on any error or if message is expired/invalid.
""" """
try: try:
# Throttle the existence stat to ~1 Hz: this runs on every render
# iteration (60+ fps), and the file usually doesn't exist — the
# status message's lifetime is measured in seconds anyway.
# Both attributes are initialised in __init__.
now = time.time()
if (now - self._wifi_status_check_ts) < 1.0:
return self._wifi_status_last_result
self._wifi_status_check_ts = now
self._wifi_status_last_result = None
# Check if file exists # Check if file exists
if not self.wifi_status_file or not self.wifi_status_file.exists(): if not self.wifi_status_file or not self.wifi_status_file.exists():
return None return None
@@ -2418,14 +2404,13 @@ class DisplayController:
pass pass
return None return None
# Message is valid and not expired — cache for the throttle window # Message is valid and not expired
self._wifi_status_last_result = { return {
'message': message, 'message': message,
'timestamp': timestamp, 'timestamp': timestamp,
'duration': duration, 'duration': duration,
'expires_at': expires_at 'expires_at': expires_at
} }
return self._wifi_status_last_result
except Exception as e: except Exception as e:
# Catch-all for any unexpected errors - log but don't break the display # Catch-all for any unexpected errors - log but don't break the display
+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
# ------------------------------------------------------------------ # ------------------------------------------------------------------
+4 -16
View File
@@ -66,10 +66,6 @@ class RenderPipeline:
else display_manager.height else display_manager.height
) )
# Reusable blank frame for cycle-end pushes (allocated lazily,
# re-blacked before each reuse)
self._blank_frame = None
# ScrollHelper for optimized scrolling # ScrollHelper for optimized scrolling
self.scroll_helper = ScrollHelper( self.scroll_helper = ScrollHelper(
self.display_width, self.display_width,
@@ -238,19 +234,11 @@ class RenderPipeline:
) )
# Push blank immediately so the hardware never shows any # Push blank immediately so the hardware never shows any
# post-wrap content while the coordinator recomposes the # post-wrap content while the coordinator recomposes the
# next cycle (~100 ms). The blank is allocated once and # next cycle (~100 ms).
# reused across cycle wraps (fresh paste each time in case
# a consumer drew on the previous one).
try: try:
if self._blank_frame is None or self._blank_frame.size != ( from PIL import Image as _Image
self.display_width, self.display_height): blank = _Image.new('RGB', (self.display_width, self.display_height))
self._blank_frame = Image.new( self.display_manager.image = blank
'RGB', (self.display_width, self.display_height))
else:
self._blank_frame.paste(
(0, 0, 0),
(0, 0, self.display_width, self.display_height))
self.display_manager.image = self._blank_frame
self.display_manager.update_display() self.display_manager.update_display()
except Exception: except Exception:
logger.exception("Failed to write blank frame to display at cycle end") logger.exception("Failed to write blank frame to display at cycle end")
-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)
-49
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."""
@@ -79,54 +78,6 @@ class TestPluginManager:
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."""