mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-05 23:05:10 +00:00
Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8879886e50 |
@@ -19,6 +19,16 @@ accepts both, but the store flags the old spelling as deprecated
|
|||||||
|
|
||||||
## Unreleased
|
## Unreleased
|
||||||
|
|
||||||
|
- 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
|
## 3.8.2
|
||||||
|
|
||||||
The display hands freed memory back to the OS (#774), and sports consolidation
|
The display hands freed memory back to the OS (#774), and sports consolidation
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ Centralized error handling for web interface.
|
|||||||
Provides helpers for consistent error responses across API endpoints.
|
Provides helpers for consistent error responses across API endpoints.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
|
import errno
|
||||||
from typing import Any, Optional
|
from typing import Any, Optional
|
||||||
from flask import jsonify
|
from flask import jsonify
|
||||||
|
|
||||||
@@ -20,10 +21,9 @@ logger = get_logger(__name__)
|
|||||||
_MAX_DETAIL_LENGTH = 400
|
_MAX_DETAIL_LENGTH = 400
|
||||||
|
|
||||||
|
|
||||||
def describe_exception(exc: BaseException,
|
def describe_exception(exc: BaseException) -> str:
|
||||||
max_length: int = _MAX_DETAIL_LENGTH) -> 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
|
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
|
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
|
*including* the log viewer, because journalctl could not be executed. The
|
||||||
underlying `[Errno 5] Input/output error` named the fault immediately.
|
underlying `[Errno 5] Input/output error` named the fault immediately.
|
||||||
|
|
||||||
Returns "TypeName: message", credentials redacted and length capped. The
|
So the type and errno still go back -- "OSError:EIO", "PermissionError:
|
||||||
type alone is worth carrying -- a bare PermissionError says more than any
|
EACCES", "TimeoutExpired" -- but never the exception's message, which can
|
||||||
generic sentence.
|
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:
|
Args:
|
||||||
exc: The exception to describe
|
exc: The exception to describe
|
||||||
max_length: Truncate beyond this many characters
|
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
A single-line description, never empty
|
"TypeName" or "TypeName:ERRNO", never empty
|
||||||
"""
|
"""
|
||||||
message = str(exc).strip()
|
code = type(exc).__name__
|
||||||
text = f"{type(exc).__name__}: {message}" if message else type(exc).__name__
|
exc_errno = getattr(exc, 'errno', None)
|
||||||
return redact_text(text, max_length)
|
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:
|
def redact_text(text: str, max_length: int = _MAX_DETAIL_LENGTH) -> str:
|
||||||
|
|||||||
+9
-9
@@ -1369,7 +1369,7 @@ class WiFiManager:
|
|||||||
self.enable_ap_mode(force=True)
|
self.enable_ap_mode(force=True)
|
||||||
except Exception as ap_error: # nosec B110 - last-resort; do not re-raise, but log for debugging
|
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)
|
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]:
|
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,
|
"""Force the setup AP up after a connect that left no working network,
|
||||||
@@ -1585,7 +1585,7 @@ class WiFiManager:
|
|||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(f"Error connecting with nmcli: {e}")
|
logger.error(f"Error connecting with nmcli: {e}")
|
||||||
self._show_led_message("Connection error", duration=5)
|
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
|
# 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
|
# 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"
|
return False, "nmcli is required to disconnect from WiFi"
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(f"Error disconnecting from WiFi: {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:
|
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)"
|
return False, "No WiFi tools available (nmcli, hostapd, or dnsmasq required)"
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(f"Error in enable_ap_mode: {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:
|
def _mark_forced(self) -> None:
|
||||||
"""Record that AP mode was forced on, so the periodic check leaves it
|
"""Record that AP mode was forced on, so the periodic check leaves it
|
||||||
@@ -2099,10 +2099,10 @@ class WiFiManager:
|
|||||||
return True, "AP mode enabled"
|
return True, "AP mode enabled"
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(f"Error starting AP services: {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:
|
except Exception as e:
|
||||||
logger.error(f"Error enabling AP mode: {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]:
|
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}")
|
logger.error(f"Error starting AP mode with nmcli: {e}")
|
||||||
self._remove_nm_dnsmasq_captive_conf()
|
self._remove_nm_dnsmasq_captive_conf()
|
||||||
self._show_led_message("Setup mode error", duration=5)
|
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:
|
def _get_ap_status_nmcli(self) -> Dict:
|
||||||
"""
|
"""
|
||||||
@@ -2409,10 +2409,10 @@ class WiFiManager:
|
|||||||
return True, "AP mode disabled"
|
return True, "AP mode disabled"
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(f"Error stopping AP services: {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:
|
except Exception as e:
|
||||||
logger.error(f"Error disabling AP mode: {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):
|
def _create_hostapd_config(self):
|
||||||
"""Create hostapd configuration file"""
|
"""Create hostapd configuration file"""
|
||||||
|
|||||||
@@ -147,7 +147,8 @@ class TestOneBadConfigSectionDoesNotBlankTheList:
|
|||||||
side_effect=RuntimeError("disk is gone"))
|
side_effect=RuntimeError("disk is gone"))
|
||||||
resp = api_v3_client.get('/api/v3/display/modes')
|
resp = api_v3_client.get('/api/v3/display/modes')
|
||||||
assert resp.status_code == 500
|
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):
|
def test_credentials_in_the_exception_are_redacted(self, api_v3_module, api_v3_client):
|
||||||
"""describe_exception is what makes returning detail safe."""
|
"""describe_exception is what makes returning detail safe."""
|
||||||
|
|||||||
@@ -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"]
|
||||||
@@ -95,9 +95,10 @@ class TestRefreshPluginStore:
|
|||||||
RuntimeError("failed at /home/user/LEDMatrix/src/secret.py line 42"))
|
RuntimeError("failed at /home/user/LEDMatrix/src/secret.py line 42"))
|
||||||
body = api_v3_client.post(self.URL, json={}).get_json()
|
body = api_v3_client.post(self.URL, json={}).get_json()
|
||||||
assert "Traceback" not in str(body)
|
assert "Traceback" not in str(body)
|
||||||
# `details` is describe_exception output: one line, type-named,
|
# `details` is describe_exception output: the type, never the
|
||||||
# credential-redacted. It may quote the message, but never a stack.
|
# message or a stack.
|
||||||
assert body["details"].startswith("RuntimeError:")
|
assert body["details"] == "RuntimeError"
|
||||||
|
assert "secret.py" not in str(body)
|
||||||
assert "\n" not in body["details"]
|
assert "\n" not in body["details"]
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -11,26 +11,32 @@ than even logging it.
|
|||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from src.web_interface.error_handler import describe_exception
|
from src.web_interface.error_handler import describe_exception, redact_text
|
||||||
|
|
||||||
|
|
||||||
class TestDescribeException:
|
class TestDescribeException:
|
||||||
def test_names_the_type_and_message(self):
|
"""describe_exception is a reason code: type and errno, never the message.
|
||||||
detail = describe_exception(OSError(5, "Input/output error", "systemctl"))
|
|
||||||
assert detail == "OSError: [Errno 5] Input/output error: 'systemctl'"
|
|
||||||
|
|
||||||
def test_the_reported_failure_is_legible(self):
|
The message can quote paths, URLs or credentials (CodeQL
|
||||||
# The whole point: this string is the diagnosis.
|
py/stack-trace-exposure), so it goes to the log; the code still names the
|
||||||
assert "Input/output error" in describe_exception(
|
fault, as "[Errno 5]" did.
|
||||||
OSError(5, "Input/output error", "systemctl"))
|
"""
|
||||||
|
|
||||||
|
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):
|
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(PermissionError()) == "PermissionError"
|
||||||
assert describe_exception(Exception()) == "Exception"
|
assert describe_exception(Exception()) == "Exception"
|
||||||
|
|
||||||
def test_message_is_kept_when_present(self):
|
def test_the_message_never_reaches_the_code(self):
|
||||||
assert describe_exception(ValueError("bad port")) == "ValueError: bad port"
|
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:
|
class TestCredentialRedaction:
|
||||||
@@ -56,47 +62,44 @@ class TestCredentialRedaction:
|
|||||||
("authorization: barecredential", "barecredential"),
|
("authorization: barecredential", "barecredential"),
|
||||||
])
|
])
|
||||||
def test_credentials_never_reach_the_response(self, secret_text, leaked):
|
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 leaked not in detail
|
||||||
assert "<redacted>" in detail
|
assert "<redacted>" in detail
|
||||||
|
|
||||||
def test_the_parameter_name_survives_redaction(self):
|
def test_the_parameter_name_survives_redaction(self):
|
||||||
# Knowing *which* credential was involved is part of the diagnosis.
|
# 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
|
assert "api_key" in detail
|
||||||
|
|
||||||
def test_unknown_schemes_keep_their_name(self):
|
def test_unknown_schemes_keep_their_name(self):
|
||||||
for scheme in ("ApiKey", "Negotiate", "NTLM", "AWS4-HMAC-SHA256"):
|
for scheme in ("ApiKey", "Negotiate", "NTLM", "AWS4-HMAC-SHA256"):
|
||||||
detail = describe_exception(
|
detail = redact_text("Authorization: %s SECRETVALUE" % scheme)
|
||||||
RuntimeError("Authorization: %s SECRETVALUE" % scheme))
|
|
||||||
assert scheme in detail, detail
|
assert scheme in detail, detail
|
||||||
assert "SECRETVALUE" not in detail, detail
|
assert "SECRETVALUE" not in detail, detail
|
||||||
|
|
||||||
def test_auth_scheme_and_username_survive(self):
|
def test_auth_scheme_and_username_survive(self):
|
||||||
# Which kind of credential, and whose, without the credential itself.
|
# Which kind of credential, and whose, without the credential itself.
|
||||||
assert "Bearer" in describe_exception(
|
assert "Bearer" in redact_text("Authorization: Bearer eyJ.SECRET.sig")
|
||||||
RuntimeError("Authorization: Bearer eyJ.SECRET.sig"))
|
assert "user" in redact_text("https://user:hunter2@example.com")
|
||||||
assert "user" in describe_exception(
|
|
||||||
RuntimeError("https://user:hunter2@example.com"))
|
|
||||||
|
|
||||||
def test_non_secret_context_is_preserved(self):
|
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 "city=Tampa" in detail
|
||||||
assert "<redacted>" not in detail
|
assert "<redacted>" not in detail
|
||||||
|
|
||||||
|
|
||||||
class TestBounds:
|
class TestBounds:
|
||||||
def test_long_messages_are_truncated(self):
|
def test_long_messages_are_truncated(self):
|
||||||
detail = describe_exception(ValueError("x" * 5000))
|
detail = redact_text("x" * 5000)
|
||||||
assert len(detail) <= 400
|
assert len(detail) <= 400
|
||||||
|
|
||||||
def test_newlines_are_collapsed_to_one_line(self):
|
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 "\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):
|
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:
|
class TestHandlersCarryDetail:
|
||||||
@@ -319,10 +322,10 @@ class TestHandlersCarryDetail:
|
|||||||
assert resp.status_code == 405, "a wrong method must stay a 405"
|
assert resp.status_code == 405, "a wrong method must stay a 405"
|
||||||
assert resp.get_json()["error_code"] == "METHOD_NOT_ALLOWED"
|
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")
|
resp = client.get("/boom")
|
||||||
assert resp.status_code == 500
|
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):
|
def test_global_handler_reports_the_underlying_error(self):
|
||||||
from flask import Flask, jsonify
|
from flask import Flask, jsonify
|
||||||
@@ -345,4 +348,5 @@ class TestHandlersCarryDetail:
|
|||||||
client = app.test_client()
|
client = app.test_client()
|
||||||
body = client.get("/boom").get_json()
|
body = client.get("/boom").get_json()
|
||||||
assert body["error_code"] == "UNKNOWN_ERROR"
|
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
|
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()
|
body = client.get("/api/v3/plugins/installed").get_json()
|
||||||
for secret in ("SECRET123", "pw1", "K1"):
|
for secret in ("SECRET123", "pw1", "K1", "forced failure"):
|
||||||
assert secret not in body["details"]
|
assert secret not in str(body)
|
||||||
assert "<redacted>" in body["details"]
|
assert body["details"] == "RuntimeError"
|
||||||
assert body["details"].startswith("RuntimeError: forced failure")
|
|
||||||
|
|
||||||
|
|
||||||
def _raise_415():
|
def _raise_415():
|
||||||
@@ -218,7 +217,7 @@ class TestPluginActionStep1:
|
|||||||
encoding="utf-8")
|
encoding="utf-8")
|
||||||
return d
|
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
|
from unittest.mock import MagicMock
|
||||||
manager = MagicMock()
|
manager = MagicMock()
|
||||||
manager.get_plugin_directory.return_value = str(plugin_dir)
|
manager.get_plugin_directory.return_value = str(plugin_dir)
|
||||||
@@ -232,5 +231,6 @@ class TestPluginActionStep1:
|
|||||||
|
|
||||||
assert resp.status_code == 500
|
assert resp.status_code == 500
|
||||||
body = resp.get_json()
|
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'
|
assert body["message"] == 'An error occurred; see logs for details'
|
||||||
|
|||||||
@@ -946,21 +946,27 @@ class TestTheStoreReportsWhyItIsEmpty:
|
|||||||
|
|
||||||
|
|
||||||
class TestACrashCarriesItsDetail:
|
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',
|
with patch('web_interface.blueprints.api_v3._get_tronbyte_repository_class',
|
||||||
side_effect=ImportError("No module named 'yaml'")):
|
side_effect=ImportError("No module named 'yaml'")):
|
||||||
body = client.get('/api/v3/starlark/repository/browse').get_json()
|
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',
|
with patch('web_interface.blueprints.api_v3._get_starlark_plugin',
|
||||||
side_effect=RuntimeError("plugin manager is not attached")):
|
side_effect=RuntimeError("plugin manager is not attached")):
|
||||||
body = client.get('/api/v3/starlark/status').get_json()
|
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:
|
class TestTheListingIsNotCappedAtOneThousand:
|
||||||
|
|||||||
@@ -222,7 +222,7 @@ def _save_config_atomic(config_manager, config_data, create_backup=True):
|
|||||||
config_manager.save_config(config_data)
|
config_manager.save_config(config_data)
|
||||||
return True, None
|
return True, None
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
return False, str(e)
|
return False, f"Failed to save configuration ({describe_exception(e)})"
|
||||||
def _coerce_to_bool(value):
|
def _coerce_to_bool(value):
|
||||||
"""
|
"""
|
||||||
Coerce a form value to a proper Python boolean.
|
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 value.lower() in ('true', 'on', '1', 'yes')
|
||||||
return False
|
return False
|
||||||
def _get_display_service_status():
|
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:
|
try:
|
||||||
result = subprocess.run(
|
result = subprocess.run(
|
||||||
['systemctl', 'is-active', 'ledmatrix'],
|
['systemctl', 'is-active', 'ledmatrix'],
|
||||||
@@ -254,26 +258,18 @@ def _get_display_service_status():
|
|||||||
text=True,
|
text=True,
|
||||||
timeout=3
|
timeout=3
|
||||||
)
|
)
|
||||||
|
if result.stderr.strip():
|
||||||
|
logger.debug('systemctl is-active ledmatrix: %s', result.stderr.strip())
|
||||||
return {
|
return {
|
||||||
'active': result.stdout.strip() == 'active',
|
'active': result.stdout.strip() == 'active',
|
||||||
'returncode': result.returncode,
|
'returncode': result.returncode,
|
||||||
'stdout': result.stdout.strip(),
|
|
||||||
'stderr': result.stderr.strip()
|
|
||||||
}
|
}
|
||||||
except subprocess.TimeoutExpired:
|
except subprocess.TimeoutExpired:
|
||||||
return {
|
logger.warning('systemctl is-active ledmatrix timed out')
|
||||||
'active': False,
|
return {'active': False, 'returncode': -1}
|
||||||
'returncode': -1,
|
except Exception:
|
||||||
'stdout': '',
|
logger.warning('Could not query ledmatrix.service status', exc_info=True)
|
||||||
'stderr': 'timeout'
|
return {'active': False, 'returncode': -1}
|
||||||
}
|
|
||||||
except Exception as err:
|
|
||||||
return {
|
|
||||||
'active': False,
|
|
||||||
'returncode': -1,
|
|
||||||
'stdout': '',
|
|
||||||
'stderr': str(err)
|
|
||||||
}
|
|
||||||
def _run_systemctl_command(args):
|
def _run_systemctl_command(args):
|
||||||
"""Run a systemctl command safely."""
|
"""Run a systemctl command safely."""
|
||||||
try:
|
try:
|
||||||
@@ -295,18 +291,26 @@ def _run_systemctl_command(args):
|
|||||||
'stderr': 'timeout'
|
'stderr': 'timeout'
|
||||||
}
|
}
|
||||||
except Exception as err:
|
except Exception as err:
|
||||||
|
logger.warning('%s failed', ' '.join(args), exc_info=True)
|
||||||
return {
|
return {
|
||||||
'returncode': -1,
|
'returncode': -1,
|
||||||
'stdout': '',
|
'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():
|
def _ensure_display_service_running():
|
||||||
"""Ensure the ledmatrix display service is running."""
|
"""Ensure the ledmatrix display service is running."""
|
||||||
status = _get_display_service_status()
|
status = _get_display_service_status()
|
||||||
if status.get('active'):
|
if status.get('active'):
|
||||||
status['started'] = False
|
status['started'] = False
|
||||||
return status
|
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()
|
service_status = _get_display_service_status()
|
||||||
result['started'] = result.get('returncode') == 0
|
result['started'] = result.get('returncode') == 0
|
||||||
result['active'] = service_status.get('active')
|
result['active'] = service_status.get('active')
|
||||||
@@ -314,7 +318,8 @@ def _ensure_display_service_running():
|
|||||||
return result
|
return result
|
||||||
def _stop_display_service():
|
def _stop_display_service():
|
||||||
"""Stop the ledmatrix 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()
|
status = _get_display_service_status()
|
||||||
result['active'] = status.get('active')
|
result['active'] = status.get('active')
|
||||||
result['status'] = status
|
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)
|
success = api_v3.plugin_store_manager.uninstall_plugin(plugin_id)
|
||||||
except Exception as remove_err:
|
except Exception as remove_err:
|
||||||
_rollback()
|
_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:
|
if not success:
|
||||||
_rollback()
|
_rollback()
|
||||||
|
|||||||
@@ -983,7 +983,6 @@ def stop_pixlet_editor():
|
|||||||
'status': 'error',
|
'status': 'error',
|
||||||
'message': 'Editor force-stopped, but the display could not be '
|
'message': 'Editor force-stopped, but the display could not be '
|
||||||
'restarted automatically - start it manually.',
|
'restarted automatically - start it manually.',
|
||||||
'details': (result.get('stderr') or '').strip(),
|
|
||||||
'data': {'running': False}}), 500
|
'data': {'running': False}}), 500
|
||||||
return jsonify({'status': 'success',
|
return jsonify({'status': 'success',
|
||||||
'message': 'Editor force-stopped; the display has been '
|
'message': 'Editor force-stopped; the display has been '
|
||||||
|
|||||||
@@ -689,7 +689,7 @@ def execute_system_action():
|
|||||||
logger.warning("install_base_requirements timed out for %s", label)
|
logger.warning("install_base_requirements timed out for %s", label)
|
||||||
except OSError as install_err:
|
except OSError as install_err:
|
||||||
all_ok = False
|
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)
|
logger.warning("install_base_requirements errored for %s: %s", label, install_err)
|
||||||
return jsonify({
|
return jsonify({
|
||||||
'status': 'success' if all_ok else 'error',
|
'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'})
|
return jsonify({'status': 'error', 'message': 'Command timed out', 'returncode': -1, 'stderr': 'timeout'})
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error("execute_system_action failed: %s", e, exc_info=True)
|
logger.error("execute_system_action failed: %s", e, exc_info=True)
|
||||||
detail = describe_exception(e)
|
|
||||||
resp = {
|
resp = {
|
||||||
'status': 'error',
|
'status': 'error',
|
||||||
'message': _sudo_hint_for(detail) or 'Action failed; see logs for details',
|
'message': _sudo_hint_for(str(e)) or 'Action failed; see logs for details',
|
||||||
'details': detail,
|
'details': describe_exception(e),
|
||||||
}
|
}
|
||||||
return jsonify(resp), 500
|
return jsonify(resp), 500
|
||||||
@api_v3.route('/system/git-info', methods=['GET'])
|
@api_v3.route('/system/git-info', methods=['GET'])
|
||||||
|
|||||||
@@ -67,7 +67,7 @@ def _run_background_connect(ssid, password):
|
|||||||
payload = _connect_result_payload(ssid, success, message)
|
payload = _connect_result_payload(ssid, success, message)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error("Background WiFi connect failed", exc_info=True)
|
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)
|
_record_connect_result(ssid, payload)
|
||||||
|
|
||||||
|
|
||||||
@@ -276,7 +276,7 @@ def connect_wifi():
|
|||||||
try:
|
try:
|
||||||
success, message = wifi_manager.connect_to_network(ssid, password)
|
success, message = wifi_manager.connect_to_network(ssid, password)
|
||||||
except Exception as e:
|
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
|
raise
|
||||||
payload = _connect_result_payload(ssid, success, message)
|
payload = _connect_result_payload(ssid, success, message)
|
||||||
_record_connect_result(ssid, payload)
|
_record_connect_result(ssid, payload)
|
||||||
|
|||||||
@@ -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
|
except Exception as e: # a broken template must not fail the update itself
|
||||||
if type(e).__name__ != 'UnitsUnreadable':
|
if type(e).__name__ != 'UnitsUnreadable':
|
||||||
logger.warning("Could not compare the installed systemd units with the new templates: %s", e)
|
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
|
# 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.
|
# it set 0644): only root can compare them, so let the helper decide.
|
||||||
stale = []
|
stale = []
|
||||||
@@ -110,7 +110,7 @@ def refresh_after_update(run=None, systemd_dir=None, helper_source=None, helper_
|
|||||||
timeout=TIMEOUT_SECONDS)
|
timeout=TIMEOUT_SECONDS)
|
||||||
except (subprocess.SubprocessError, OSError) as e:
|
except (subprocess.SubprocessError, OSError) as e:
|
||||||
logger.warning("Refreshing the systemd units failed: %s", 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:
|
if result.returncode == 0:
|
||||||
# The helper says what it did: "units refreshed: a b" or "units: up to date".
|
# 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()
|
done = next((line.split(':', 1)[1].split() for line in (result.stdout or '').splitlines()
|
||||||
|
|||||||
Reference in New Issue
Block a user