Compare commits

..
Author SHA1 Message Date
ChuckandClaude Opus 5.5 8879886e50 fix(web): keep exception messages out of API responses (py/stack-trace-exposure)
CodeQL had ~40 open py/stack-trace-exposure alerts on main. Almost all
flowed through describe_exception(), which returned "TypeName: message"
(redacted, capped); the rest through _run_systemctl_command's str(err),
WiFiManager's `return False, str(e)`, unit_refresh's f-strings and two
str(e)/f"{err}" messages in api_v3/__init__.py.

describe_exception() now returns a reason code -- the type, plus the
errno symbol for an OSError ("OSError:EIO", "PermissionError:EACCES") --
and logs the redacted message itself. That keeps what #538 wanted (a
failing disk still says EIO in the response) without quoting paths,
URLs or library internals, and fixes every call site at once; the
test_no_api_v3_handler_discards_its_exception policy still holds.

Service results: _get_display_service_status returns active/returncode
only, and the on-demand start/stop `service` result keeps
returncode/active/started/status but drops systemctl stdout/stderr
(logged on failure). Nothing in web_interface/static, the templates or
the MQTT bridge reads those fields. The Starlark SIGKILL-restart error
no longer returns systemctl stderr as `details`.

WiFi, unit-refresh, config-save and plugin-removal failures now say
what failed with the reason code and point at the log. display.py is
untouched (draft #773 edits it).

Tests: test_api_v3_no_exception_text.py drives one route per affected
file with a marker in the exception message and asserts it never
reaches the body; all 13 fail on origin/main, and targeted mutations
(drop the service filter, put stderr back, str(e) in WiFiManager,
{e} in unit_refresh, {install_err} in system.py, message back in
describe_exception) each fail at least one. Tests that asserted the old
message-in-details contract now assert the reason code.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-05 16:41:07 -04:00
15 changed files with 280 additions and 144 deletions
+9 -9
View File
@@ -19,15 +19,15 @@ accepts both, but the store flags the old spelling as deprecated
## Unreleased
### Tooling
- `test/test_sports_helpers.py`'s parity tests pass again with
`LEDMATRIX_PLUGINS` set. The scoreboards deleted their copies of the
`sports_helpers` bodies and constants when they adopted `SportsHelpersMixin`
(ledmatrix-plugins #563/#564), and the 19 tests still expected them. A copy
that is gone now counts as adopted when the plugin imports
`src.common.sports_helpers`, as the stage 3/4 and game-over parity tests
already do; a copy that remains must still match.
- Web API error responses no longer carry an exception's message (CodeQL
`py/stack-trace-exposure`). `describe_exception()` now returns a reason
code -- the exception type, plus the errno for an `OSError`
(`OSError:EIO`, `PermissionError:EACCES`) -- and logs the message
instead, so `details` still names the fault without quoting paths, URLs
or library internals. The display service status and the on-demand
start/stop `service` results keep `active`, `returncode` and `started`
but drop systemctl's `stdout`/`stderr`; WiFi, unit-refresh and
config-save failures say what failed and point at the log.
## 3.8.2
+15 -11
View File
@@ -4,6 +4,7 @@ Centralized error handling for web interface.
Provides helpers for consistent error responses across API endpoints.
"""
import errno
from typing import Any, Optional
from flask import jsonify
@@ -20,10 +21,9 @@ logger = get_logger(__name__)
_MAX_DETAIL_LENGTH = 400
def describe_exception(exc: BaseException,
max_length: int = _MAX_DETAIL_LENGTH) -> str:
def describe_exception(exc: BaseException) -> str:
"""
One-line, safe-to-return description of an exception.
Machine-readable reason code for an exception, safe to return over HTTP.
The generic "an error occurred; see logs for details" tells a user nothing
and, when the failure is bad enough, the logs are unreachable too: a device
@@ -31,20 +31,24 @@ def describe_exception(exc: BaseException,
*including* the log viewer, because journalctl could not be executed. The
underlying `[Errno 5] Input/output error` named the fault immediately.
Returns "TypeName: message", credentials redacted and length capped. The
type alone is worth carrying -- a bare PermissionError says more than any
generic sentence.
So the type and errno still go back -- "OSError:EIO", "PermissionError:
EACCES", "TimeoutExpired" -- but never the exception's message, which can
quote paths, URLs, credentials or a library's internals (CodeQL
py/stack-trace-exposure). The message is logged here instead, so every
reason code a client sees has its full text in the log.
Args:
exc: The exception to describe
max_length: Truncate beyond this many characters
Returns:
A single-line description, never empty
"TypeName" or "TypeName:ERRNO", never empty
"""
message = str(exc).strip()
text = f"{type(exc).__name__}: {message}" if message else type(exc).__name__
return redact_text(text, max_length)
code = type(exc).__name__
exc_errno = getattr(exc, 'errno', None)
if isinstance(exc_errno, int) and exc_errno in errno.errorcode:
code = f"{code}:{errno.errorcode[exc_errno]}"
logger.warning("Error reported to the client as %s: %s", code, redact_text(str(exc)))
return code
def redact_text(text: str, max_length: int = _MAX_DETAIL_LENGTH) -> str:
+9 -9
View File
@@ -1369,7 +1369,7 @@ class WiFiManager:
self.enable_ap_mode(force=True)
except Exception as ap_error: # nosec B110 - last-resort; do not re-raise, but log for debugging
logger.error("Last-resort AP mode enable failed in recovery path: %s", ap_error, exc_info=True)
return False, str(e)
return False, f"Connection failed ({type(e).__name__}); see logs for details"
def _failsafe_ap(self, enabled_msg: str, failed_msg: str) -> Tuple[bool, str]:
"""Force the setup AP up after a connect that left no working network,
@@ -1585,7 +1585,7 @@ class WiFiManager:
except Exception as e:
logger.error(f"Error connecting with nmcli: {e}")
self._show_led_message("Connection error", duration=5)
return False, str(e)
return False, f"Connection failed ({type(e).__name__}); see logs for details"
# 802.11 caps an SSID at 32 octets. Control characters cannot appear in a
# real one, and a leading "-" would be read by nmcli as an option rather
@@ -1725,7 +1725,7 @@ class WiFiManager:
return False, "nmcli is required to disconnect from WiFi"
except Exception as e:
logger.error(f"Error disconnecting from WiFi: {e}")
return False, str(e)
return False, f"Disconnect failed ({type(e).__name__}); see logs for details"
def _ensure_wifi_radio_enabled(self, max_retries: int = 3) -> bool:
"""
@@ -2004,7 +2004,7 @@ class WiFiManager:
return False, "No WiFi tools available (nmcli, hostapd, or dnsmasq required)"
except Exception as e:
logger.error(f"Error in enable_ap_mode: {e}")
return False, str(e)
return False, f"Could not enable AP mode ({type(e).__name__}); see logs for details"
def _mark_forced(self) -> None:
"""Record that AP mode was forced on, so the periodic check leaves it
@@ -2099,10 +2099,10 @@ class WiFiManager:
return True, "AP mode enabled"
except Exception as e:
logger.error(f"Error starting AP services: {e}")
return False, str(e)
return False, f"Could not enable AP mode ({type(e).__name__}); see logs for details"
except Exception as e:
logger.error(f"Error enabling AP mode: {e}")
return False, str(e)
return False, f"Could not enable AP mode ({type(e).__name__}); see logs for details"
def _enable_ap_mode_nmcli_hotspot(self) -> Tuple[bool, str]:
"""
@@ -2227,7 +2227,7 @@ class WiFiManager:
logger.error(f"Error starting AP mode with nmcli: {e}")
self._remove_nm_dnsmasq_captive_conf()
self._show_led_message("Setup mode error", duration=5)
return False, str(e)
return False, f"Could not enable AP mode ({type(e).__name__}); see logs for details"
def _get_ap_status_nmcli(self) -> Dict:
"""
@@ -2409,10 +2409,10 @@ class WiFiManager:
return True, "AP mode disabled"
except Exception as e:
logger.error(f"Error stopping AP services: {e}")
return False, str(e)
return False, f"Could not disable AP mode ({type(e).__name__}); see logs for details"
except Exception as e:
logger.error(f"Error disabling AP mode: {e}")
return False, str(e)
return False, f"Could not disable AP mode ({type(e).__name__}); see logs for details"
def _create_hostapd_config(self):
"""Create hostapd configuration file"""
+2 -1
View File
@@ -147,7 +147,8 @@ class TestOneBadConfigSectionDoesNotBlankTheList:
side_effect=RuntimeError("disk is gone"))
resp = api_v3_client.get('/api/v3/display/modes')
assert resp.status_code == 500
assert 'disk is gone' in resp.get_json()['details']
assert resp.get_json()['details'] == 'RuntimeError'
assert 'disk is gone' not in json.dumps(resp.get_json())
def test_credentials_in_the_exception_are_redacted(self, api_v3_module, api_v3_client):
"""describe_exception is what makes returning detail safe."""
+147
View File
@@ -0,0 +1,147 @@
"""No API response carries an exception's message (CodeQL py/stack-trace-exposure).
One representative route per file that had open alerts. Each forces a failure
whose message holds a marker and asserts the marker is nowhere in the body:
the message goes to the log, the client gets a fixed message plus a reason
code (describe_exception: the type, and the errno for an OSError).
"""
import json
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent))
from test._api_v3_test_helpers import api_v3_client, api_v3_module # noqa: F401,E402
LEAK = "LEAKED-/home/pi/secret token=abc123"
API = "web_interface.blueprints.api_v3"
def _assert_no_leak(response):
body = response.get_data(as_text=True)
assert "LEAKED" not in body, body
assert "abc123" not in body, body
return json.loads(body)
def test_display_service_status_drops_systemctl_output(api_v3_module, api_v3_client,
monkeypatch):
"""display.py: the on-demand routes return the service status verbatim."""
api_v3_module.api_v3.cache_manager.get.return_value = None
monkeypatch.setattr(f"{API}.display.display_state.read_state", lambda: None)
with patch(f"{API}.subprocess.run", side_effect=OSError(13, LEAK)):
body = _assert_no_leak(api_v3_client.get("/api/v3/display/on-demand/status"))
assert body["data"]["service"] == {"active": False, "returncode": -1}
@pytest.mark.parametrize("helper", ["_ensure_display_service_running",
"_stop_display_service"])
def test_service_results_keep_returncode_but_not_output(api_v3_module, helper):
"""display.py start/stop: returncode/active/started stay, stdout/stderr go."""
failed = MagicMock(returncode=1, stdout=LEAK, stderr=LEAK)
with patch(f"{API}.subprocess.run", return_value=failed):
result = getattr(api_v3_module, helper)()
assert "LEAKED" not in json.dumps(result)
assert result["returncode"] == 1 and result["active"] is False
assert "stdout" not in result and "stderr" not in result
def test_wifi_connect_failure(api_v3_client):
"""wifi.py: a raising connect, and the attempt /wifi/status reports after."""
with patch("src.wifi_manager.WiFiManager") as cls:
cls.return_value._is_ap_mode_active.return_value = False
cls.return_value.connect_to_network.side_effect = RuntimeError(LEAK)
body = _assert_no_leak(api_v3_client.post(
"/api/v3/wifi/connect", json={"ssid": "HomeNet", "password": "pw"}))
assert body["details"] == "RuntimeError"
cls.return_value.get_wifi_status.return_value = MagicMock(
connected=False, ssid=None, ip_address=None, signal=0, ap_mode_active=False)
cls.return_value.config = {}
status = _assert_no_leak(api_v3_client.get("/api/v3/wifi/status"))
assert status["data"]["last_connect_attempt"]["message"] == (
"Failed to connect to network (RuntimeError)")
def test_wifi_manager_messages_carry_no_exception_text():
"""src/wifi_manager.py: its (success, message) is what the wifi routes return."""
from src.wifi_manager import WiFiManager
manager = WiFiManager.__new__(WiFiManager) # no __init__: no host access
manager.get_wifi_status = MagicMock(side_effect=OSError(5, LEAK))
success, message = manager.disconnect_from_network()
assert success is False
assert "LEAKED" not in message and "OSError" in message
def test_system_action_exception(api_v3_client):
"""system.py: execute_system_action's catch-all."""
with patch("subprocess.run", side_effect=OSError(5, LEAK)):
body = _assert_no_leak(api_v3_client.post(
"/api/v3/system/action", json={"action": "stop_display"}))
assert body["details"] == "OSError:EIO"
def test_calendar_registration_failure(api_v3_client, tmp_path, monkeypatch):
"""plugin_calendar.py: the auth script could not be run."""
plugin_dir = tmp_path / "calendar"
plugin_dir.mkdir()
(plugin_dir / "credentials.json").write_text("{}", encoding="utf-8")
(plugin_dir / "calendar_registration.py").write_text("", encoding="utf-8")
monkeypatch.setattr(f"{API}._calendar_plugin_dir", lambda: plugin_dir)
with patch(f"{API}.subprocess.run", side_effect=OSError(13, LEAK)):
body = _assert_no_leak(api_v3_client.post(
"/api/v3/plugins/calendar/authenticate", json={"code": "x"}))
assert "EACCES" in body["message"]
def test_health_failure(api_v3_client, monkeypatch):
"""misc.py: get_health's catch-all."""
def boom():
raise RuntimeError(LEAK)
monkeypatch.setattr(f"{API}.misc._get_display_service_status", boom)
body = _assert_no_leak(api_v3_client.get("/api/v3/health"))
assert body["details"] == "RuntimeError"
def test_config_route_failure(api_v3_module, api_v3_client):
"""error_handler.py: create_error_response, as config.py's routes use it."""
api_v3_module.api_v3.config_manager.load_config.side_effect = RuntimeError(LEAK)
body = _assert_no_leak(api_v3_client.get("/api/v3/config/schedule"))
assert body["details"] == "RuntimeError"
def test_plugin_route_failure(api_v3_module, api_v3_client):
"""plugins.py: an unhandled error in a plugin route."""
api_v3_module.api_v3.plugin_catalog.get_all_plugin_info.side_effect = RuntimeError(LEAK)
body = _assert_no_leak(api_v3_client.get("/api/v3/plugins/installed"))
assert body["details"] == "RuntimeError"
def test_starlark_route_failure(api_v3_client):
"""starlark.py: one of its catch-alls."""
with patch(f"{API}._get_starlark_plugin", side_effect=RuntimeError(LEAK)):
body = _assert_no_leak(api_v3_client.get("/api/v3/starlark/status"))
assert body["details"] == "RuntimeError"
def test_unit_refresh_failure(monkeypatch):
"""system.py git_pull: perform_core_update appends unit_refresh's message."""
from web_interface import unit_refresh
def boom(*_a, **_k):
raise RuntimeError(LEAK)
monkeypatch.setattr(unit_refresh, "stale_units", boom)
result = unit_refresh.refresh_after_update()
assert result["status"] == unit_refresh.FAILED
assert "LEAKED" not in result["message"]
def test_install_base_requirements_failure(api_v3_client):
"""system.py: a pip install that could not start, in the action's output."""
with patch(f"{API}.system._pip_install_requirements", side_effect=OSError(5, LEAK)):
body = _assert_no_leak(api_v3_client.post(
"/api/v3/system/action", json={"action": "install_base_requirements"}))
assert "Failed: OSError:EIO" in body["output"]
+4 -3
View File
@@ -95,9 +95,10 @@ class TestRefreshPluginStore:
RuntimeError("failed at /home/user/LEDMatrix/src/secret.py line 42"))
body = api_v3_client.post(self.URL, json={}).get_json()
assert "Traceback" not in str(body)
# `details` is describe_exception output: one line, type-named,
# credential-redacted. It may quote the message, but never a stack.
assert body["details"].startswith("RuntimeError:")
# `details` is describe_exception output: the type, never the
# message or a stack.
assert body["details"] == "RuntimeError"
assert "secret.py" not in str(body)
assert "\n" not in body["details"]
+11 -41
View File
@@ -8,9 +8,7 @@ loses those tests with it.
The parity class is what keeps "byte-identical" true after this lands. Point
LEDMATRIX_PLUGINS at a ledmatrix-plugins checkout and every promoted body is
compared, as a docstring-stripped AST, against every plugin copy that carries
it. A copy that is gone counts as adopted when the plugin imports
src.common.sports_helpers (plugins#563/#564 did that for every scoreboard).
Without the variable it skips rather than fails, since core CI has no
it. Without the variable it skips rather than fails, since core CI has no
plugins checkout; ledmatrix-plugins CI runs the same comparison against core
(scripts/check_sports_helpers_parity.py, ledmatrix-plugins#495).
"""
@@ -574,24 +572,6 @@ def _core_definitions():
return out
def _sports_source(root, sport):
return (root / f"{sport}-scoreboard" / "sports.py").read_text(encoding="utf-8")
def _adopted(source):
"""Gone is fine once the plugin uses the module; otherwise the finder is
not seeing its copy."""
name = sports_helpers.__name__
for node in ast.walk(ast.parse(source)):
if isinstance(node, ast.ImportFrom):
if node.module == name or any(
f"{node.module}.{a.name}" == name for a in node.names):
return True
elif isinstance(node, ast.Import) and any(a.name == name for a in node.names):
return True
return False
class TestParityWithPlugins:
@pytest.mark.parametrize("name", sorted(PROMOTED))
def test_body_matches_every_plugin_copy(self, name):
@@ -600,10 +580,10 @@ class TestParityWithPlugins:
ours = _dump(_core_definitions()[name])
drifted, missing = [], []
for sport in carriers:
source = _sports_source(root, sport)
theirs = _definitions(ast.parse(source))[where].get(plugin_name)
defs = _definitions(ast.parse(
(root / f"{sport}-scoreboard" / "sports.py").read_text(encoding="utf-8")))
theirs = defs[where].get(plugin_name)
if theirs is None:
if not _adopted(source):
missing.append(sport)
elif _dump(theirs) != ours:
drifted.append(sport)
@@ -614,20 +594,10 @@ class TestParityWithPlugins:
@pytest.mark.parametrize("sport", SCOREBOARDS)
def test_constants_match(self, sport):
source = _sports_source(_plugins_root(), sport)
defs = _definitions(ast.parse(source))
expected = {
("module", "_MIN_WINDOW_DAYS"): MIN_WINDOW_DAYS,
("module", "_MAX_WINDOW_DAYS"): MAX_WINDOW_DAYS,
("SportsCore", "_DWELL_REENTRY_GAP_SECONDS"):
SportsHelpersMixin._DWELL_REENTRY_GAP_SECONDS,
}
missing = []
for (where, name), value in expected.items():
node = defs[where].get(name)
if node is None:
if not _adopted(source):
missing.append(name)
else:
assert math.isclose(ast.literal_eval(node.value), value), name
assert missing == [], f"not found in {sport}: {missing}"
root = _plugins_root()
defs = _definitions(ast.parse(
(root / f"{sport}-scoreboard" / "sports.py").read_text(encoding="utf-8")))
assert ast.literal_eval(defs["module"]["_MIN_WINDOW_DAYS"].value) == MIN_WINDOW_DAYS
assert ast.literal_eval(defs["module"]["_MAX_WINDOW_DAYS"].value) == MAX_WINDOW_DAYS
gap = defs["SportsCore"]["_DWELL_REENTRY_GAP_SECONDS"].value
assert math.isclose(ast.literal_eval(gap), SportsHelpersMixin._DWELL_REENTRY_GAP_SECONDS)
+31 -27
View File
@@ -11,26 +11,32 @@ than even logging it.
import pytest
from src.web_interface.error_handler import describe_exception
from src.web_interface.error_handler import describe_exception, redact_text
class TestDescribeException:
def test_names_the_type_and_message(self):
detail = describe_exception(OSError(5, "Input/output error", "systemctl"))
assert detail == "OSError: [Errno 5] Input/output error: 'systemctl'"
"""describe_exception is a reason code: type and errno, never the message.
def test_the_reported_failure_is_legible(self):
# The whole point: this string is the diagnosis.
assert "Input/output error" in describe_exception(
OSError(5, "Input/output error", "systemctl"))
The message can quote paths, URLs or credentials (CodeQL
py/stack-trace-exposure), so it goes to the log; the code still names the
fault, as "[Errno 5]" did.
"""
def test_an_oserror_names_its_errno(self):
assert describe_exception(
OSError(5, "Input/output error", "systemctl")) == "OSError:EIO"
def test_a_bare_exception_still_names_its_type(self):
# A PermissionError with no message still says more than "unknown".
assert describe_exception(PermissionError()) == "PermissionError"
assert describe_exception(Exception()) == "Exception"
def test_message_is_kept_when_present(self):
assert describe_exception(ValueError("bad port")) == "ValueError: bad port"
def test_the_message_never_reaches_the_code(self):
assert describe_exception(ValueError("bad port /etc/secret")) == "ValueError"
def test_the_message_is_logged_instead(self, caplog):
describe_exception(RuntimeError("disk on fire token=abc123"))
assert "disk on fire" in caplog.text
assert "abc123" not in caplog.text
class TestCredentialRedaction:
@@ -56,47 +62,44 @@ class TestCredentialRedaction:
("authorization: barecredential", "barecredential"),
])
def test_credentials_never_reach_the_response(self, secret_text, leaked):
detail = describe_exception(RuntimeError(secret_text))
detail = redact_text(secret_text)
assert leaked not in detail
assert "<redacted>" in detail
def test_the_parameter_name_survives_redaction(self):
# Knowing *which* credential was involved is part of the diagnosis.
detail = describe_exception(RuntimeError("https://x/y?api_key=SEC123"))
detail = redact_text("https://x/y?api_key=SEC123")
assert "api_key" in detail
def test_unknown_schemes_keep_their_name(self):
for scheme in ("ApiKey", "Negotiate", "NTLM", "AWS4-HMAC-SHA256"):
detail = describe_exception(
RuntimeError("Authorization: %s SECRETVALUE" % scheme))
detail = redact_text("Authorization: %s SECRETVALUE" % scheme)
assert scheme in detail, detail
assert "SECRETVALUE" not in detail, detail
def test_auth_scheme_and_username_survive(self):
# Which kind of credential, and whose, without the credential itself.
assert "Bearer" in describe_exception(
RuntimeError("Authorization: Bearer eyJ.SECRET.sig"))
assert "user" in describe_exception(
RuntimeError("https://user:hunter2@example.com"))
assert "Bearer" in redact_text("Authorization: Bearer eyJ.SECRET.sig")
assert "user" in redact_text("https://user:hunter2@example.com")
def test_non_secret_context_is_preserved(self):
detail = describe_exception(RuntimeError("https://api.x.com/v1?city=Tampa"))
detail = redact_text("https://api.x.com/v1?city=Tampa")
assert "city=Tampa" in detail
assert "<redacted>" not in detail
class TestBounds:
def test_long_messages_are_truncated(self):
detail = describe_exception(ValueError("x" * 5000))
detail = redact_text("x" * 5000)
assert len(detail) <= 400
def test_newlines_are_collapsed_to_one_line(self):
detail = describe_exception(ValueError("line one\nline two\tthree"))
detail = redact_text("line one\nline two\tthree")
assert "\n" not in detail and "\t" not in detail
assert detail == "ValueError: line one line two three"
assert detail == "line one line two three"
def test_custom_length_is_honoured(self):
assert len(describe_exception(ValueError("y" * 500), max_length=50)) <= 50
assert len(redact_text("y" * 500, max_length=50)) <= 50
class TestHandlersCarryDetail:
@@ -319,10 +322,10 @@ class TestHandlersCarryDetail:
assert resp.status_code == 405, "a wrong method must stay a 405"
assert resp.get_json()["error_code"] == "METHOD_NOT_ALLOWED"
# A genuine server fault still reports as one, with its detail.
# A genuine server fault still reports as one, with its reason code.
resp = client.get("/boom")
assert resp.status_code == 500
assert "Input/output error" in resp.get_json()["details"]
assert resp.get_json()["details"] == "OSError:EIO"
def test_global_handler_reports_the_underlying_error(self):
from flask import Flask, jsonify
@@ -345,4 +348,5 @@ class TestHandlersCarryDetail:
client = app.test_client()
body = client.get("/boom").get_json()
assert body["error_code"] == "UNKNOWN_ERROR"
assert "Input/output error" in body["details"]
assert body["details"] == "OSError:EIO"
assert "Input/output error" not in str(body)
@@ -114,12 +114,11 @@ def test_the_answer_is_what_the_catch_all_returned(client, caplog, method, url,
assert records[-1].exc_info[1] is FORCED
def test_credentials_are_redacted_from_the_detail(client):
def test_the_exception_message_never_reaches_the_detail(client):
body = client.get("/api/v3/plugins/installed").get_json()
for secret in ("SECRET123", "pw1", "K1"):
assert secret not in body["details"]
assert "<redacted>" in body["details"]
assert body["details"].startswith("RuntimeError: forced failure")
for secret in ("SECRET123", "pw1", "K1", "forced failure"):
assert secret not in str(body)
assert body["details"] == "RuntimeError"
def _raise_415():
@@ -218,7 +217,7 @@ class TestPluginActionStep1:
encoding="utf-8")
return d
def test_the_script_error_reaches_the_response(self, plugin_dir, monkeypatch):
def test_the_script_error_is_reported_by_type(self, plugin_dir, monkeypatch):
from unittest.mock import MagicMock
manager = MagicMock()
manager.get_plugin_directory.return_value = str(plugin_dir)
@@ -232,5 +231,6 @@ class TestPluginActionStep1:
assert resp.status_code == 500
body = resp.get_json()
assert body["details"] == "RuntimeError: the auth script failed"
assert body["details"] == "RuntimeError"
assert "the auth script failed" not in str(body)
assert body["message"] == 'An error occurred; see logs for details'
@@ -946,21 +946,27 @@ class TestTheStoreReportsWhyItIsEmpty:
class TestACrashCarriesItsDetail:
"""Seventeen Starlark handlers answered 5xx with no detail at all."""
"""Seventeen Starlark handlers answered 5xx with no detail at all.
def test_browse_returns_the_exception_detail(self, client):
The detail is a reason code (the exception type), not the exception's
message, which stays in the log (CodeQL py/stack-trace-exposure).
"""
def test_browse_returns_the_reason_code(self, client):
with patch('web_interface.blueprints.api_v3._get_tronbyte_repository_class',
side_effect=ImportError("No module named 'yaml'")):
body = client.get('/api/v3/starlark/repository/browse').get_json()
assert 'yaml' in body.get('details', ''), body
assert body.get('details') == 'ImportError', body
assert 'yaml' not in str(body), body
def test_status_returns_the_exception_detail(self, client):
def test_status_returns_the_reason_code(self, client):
with patch('web_interface.blueprints.api_v3._get_starlark_plugin',
side_effect=RuntimeError("plugin manager is not attached")):
body = client.get('/api/v3/starlark/status').get_json()
assert 'plugin manager is not attached' in body.get('details', ''), body
assert body.get('details') == 'RuntimeError', body
assert 'plugin manager is not attached' not in str(body), body
class TestTheListingIsNotCappedAtOneThousand:
+26 -21
View File
@@ -222,7 +222,7 @@ def _save_config_atomic(config_manager, config_data, create_backup=True):
config_manager.save_config(config_data)
return True, None
except Exception as e:
return False, str(e)
return False, f"Failed to save configuration ({describe_exception(e)})"
def _coerce_to_bool(value):
"""
Coerce a form value to a proper Python boolean.
@@ -246,7 +246,11 @@ def _coerce_to_bool(value):
return value.lower() in ('true', 'on', '1', 'yes')
return False
def _get_display_service_status():
"""Return status information about the ledmatrix service."""
"""Return status information about the ledmatrix service.
active/returncode only: this goes back in API responses, and systemctl's
output (or an exception's text) is logged rather than returned.
"""
try:
result = subprocess.run(
['systemctl', 'is-active', 'ledmatrix'],
@@ -254,26 +258,18 @@ def _get_display_service_status():
text=True,
timeout=3
)
if result.stderr.strip():
logger.debug('systemctl is-active ledmatrix: %s', result.stderr.strip())
return {
'active': result.stdout.strip() == 'active',
'returncode': result.returncode,
'stdout': result.stdout.strip(),
'stderr': result.stderr.strip()
}
except subprocess.TimeoutExpired:
return {
'active': False,
'returncode': -1,
'stdout': '',
'stderr': 'timeout'
}
except Exception as err:
return {
'active': False,
'returncode': -1,
'stdout': '',
'stderr': str(err)
}
logger.warning('systemctl is-active ledmatrix timed out')
return {'active': False, 'returncode': -1}
except Exception:
logger.warning('Could not query ledmatrix.service status', exc_info=True)
return {'active': False, 'returncode': -1}
def _run_systemctl_command(args):
"""Run a systemctl command safely."""
try:
@@ -295,18 +291,26 @@ def _run_systemctl_command(args):
'stderr': 'timeout'
}
except Exception as err:
logger.warning('%s failed', ' '.join(args), exc_info=True)
return {
'returncode': -1,
'stdout': '',
'stderr': str(err)
'stderr': describe_exception(err)
}
def _public_service_result(result):
"""A _run_systemctl_command result fit for a response: no stdout/stderr."""
if result.get('returncode') != 0:
logger.error('systemctl exited %s: %s', result.get('returncode'),
(result.get('stderr') or '').strip())
return {k: v for k, v in result.items() if k not in ('stdout', 'stderr')}
def _ensure_display_service_running():
"""Ensure the ledmatrix display service is running."""
status = _get_display_service_status()
if status.get('active'):
status['started'] = False
return status
result = _run_systemctl_command(['sudo', 'systemctl', 'start', 'ledmatrix.service'])
result = _public_service_result(
_run_systemctl_command(['sudo', 'systemctl', 'start', 'ledmatrix.service']))
service_status = _get_display_service_status()
result['started'] = result.get('returncode') == 0
result['active'] = service_status.get('active')
@@ -314,7 +318,8 @@ def _ensure_display_service_running():
return result
def _stop_display_service():
"""Stop the ledmatrix display service."""
result = _run_systemctl_command(['sudo', 'systemctl', 'stop', 'ledmatrix.service'])
result = _public_service_result(
_run_systemctl_command(['sudo', 'systemctl', 'stop', 'ledmatrix.service']))
status = _get_display_service_status()
result['active'] = status.get('active')
result['status'] = status
@@ -712,7 +717,7 @@ def _do_transactional_uninstall(plugin_id, preserve_config):
success = api_v3.plugin_store_manager.uninstall_plugin(plugin_id)
except Exception as remove_err:
_rollback()
return False, f"Failed to remove plugin {plugin_id}: {remove_err}"
return False, f"Failed to remove plugin {plugin_id} ({describe_exception(remove_err)})"
if not success:
_rollback()
@@ -983,7 +983,6 @@ def stop_pixlet_editor():
'status': 'error',
'message': 'Editor force-stopped, but the display could not be '
'restarted automatically - start it manually.',
'details': (result.get('stderr') or '').strip(),
'data': {'running': False}}), 500
return jsonify({'status': 'success',
'message': 'Editor force-stopped; the display has been '
+3 -4
View File
@@ -689,7 +689,7 @@ def execute_system_action():
logger.warning("install_base_requirements timed out for %s", label)
except OSError as install_err:
all_ok = False
outputs.append(f"== {label} ==\nFailed: {install_err}")
outputs.append(f"== {label} ==\nFailed: {describe_exception(install_err)}")
logger.warning("install_base_requirements errored for %s: %s", label, install_err)
return jsonify({
'status': 'success' if all_ok else 'error',
@@ -784,11 +784,10 @@ def execute_system_action():
return jsonify({'status': 'error', 'message': 'Command timed out', 'returncode': -1, 'stderr': 'timeout'})
except Exception as e:
logger.error("execute_system_action failed: %s", e, exc_info=True)
detail = describe_exception(e)
resp = {
'status': 'error',
'message': _sudo_hint_for(detail) or 'Action failed; see logs for details',
'details': detail,
'message': _sudo_hint_for(str(e)) or 'Action failed; see logs for details',
'details': describe_exception(e),
}
return jsonify(resp), 500
@api_v3.route('/system/git-info', methods=['GET'])
+2 -2
View File
@@ -67,7 +67,7 @@ def _run_background_connect(ssid, password):
payload = _connect_result_payload(ssid, success, message)
except Exception as e:
logger.error("Background WiFi connect failed", exc_info=True)
payload = {'status': 'error', 'message': describe_exception(e)}
payload = {'status': 'error', 'message': f'Failed to connect to network ({describe_exception(e)})'}
_record_connect_result(ssid, payload)
@@ -276,7 +276,7 @@ def connect_wifi():
try:
success, message = wifi_manager.connect_to_network(ssid, password)
except Exception as e:
_record_connect_result(ssid, {'status': 'error', 'message': describe_exception(e)})
_record_connect_result(ssid, {'status': 'error', 'message': f'Failed to connect to network ({describe_exception(e)})'})
raise
payload = _connect_result_payload(ssid, success, message)
_record_connect_result(ssid, payload)
+2 -2
View File
@@ -86,7 +86,7 @@ def refresh_after_update(run=None, systemd_dir=None, helper_source=None, helper_
except Exception as e: # a broken template must not fail the update itself
if type(e).__name__ != 'UnitsUnreadable':
logger.warning("Could not compare the installed systemd units with the new templates: %s", e)
return _result(FAILED, f'The service settings could not be checked: {e}.')
return _result(FAILED, 'The service settings could not be checked; see logs for details.')
# Units installed mode 0600 (install_service.sh run on its own, before
# it set 0644): only root can compare them, so let the helper decide.
stale = []
@@ -110,7 +110,7 @@ def refresh_after_update(run=None, systemd_dir=None, helper_source=None, helper_
timeout=TIMEOUT_SECONDS)
except (subprocess.SubprocessError, OSError) as e:
logger.warning("Refreshing the systemd units failed: %s", e)
return _result(FAILED, f'Updating the service settings ({names}) failed: {e}.', stale)
return _result(FAILED, f'Updating the service settings ({names}) failed; see logs for details.', stale)
if result.returncode == 0:
# The helper says what it did: "units refreshed: a b" or "units: up to date".
done = next((line.split(':', 1)[1].split() for line in (result.stdout or '').splitlines()