mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-04 22:35:08 +00:00
* refactor(api-v3): split the 10,469-line blueprint into a package web_interface/blueprints/api_v3.py held 111 routes, 56 helpers and 181 functions in one module -- 9% of the core by line count and three times the next largest file. It becomes a package of nine route modules grouped by path segment, plus __init__.py for the shared imports, constants, Blueprint and helpers. Every route module decorates the SAME api_v3 Blueprint object, so endpoint names stay api_v3.<function>, the URL map is unchanged and app.py is untouched. Verified: 111 routes before, 111 after, byte-identical rules, endpoints and methods, and every endpoint still on the one blueprint. plugins 3,867 config 1,178 starlark 692 system 619 fonts 452 misc 398 wifi 361 display 326 backup 212 __init__ 1,787 (imports, constants, Blueprint, 56 helpers) Two things the URL-map check could not catch, both found by running the suite: 1. PROJECT_ROOT = Path(__file__).parent.parent.parent. Moving the code one directory deeper made that resolve to web_interface/ instead of the project root. Nothing failed at import; it surfaced as ~110 tests failing with 404s and "installation script not found", because every path built from it was one level too shallow. Now parents[3], and test_api_v3_url_map.py asserts PROJECT_ROOT/run.py exists so the next move cannot repeat it. 2. Module-attribute patching. Tests do monkeypatch.setattr(api_v3_module, "_BACKUP_EXPORT_DIR", ...) and a route module that binds such a name by value never sees the patch. The shared code therefore stays in __init__.py rather than moving to a _common submodule -- it has to live on the module the tests patch -- and the eleven names tests patch are read back through the package (_pkg.X) instead of bound by value. Those eleven were found by AST-scanning every setattr in the test tree, not by guessing; "time" is among them, used to drive a fake clock through the second-resolution credential-backup filenames. Test changes are confined to what genuinely moved: patch targets that now name the owning route module, imports of helpers, and six tests that scan the api_v3 source as a file and now read the package directory. Full suite: 4,278 passed, 68 skipped, 0 failed. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014RRtqXDCnvnY6EQwhT5CV9 * fix(api-v3): address CodeRabbit findings from the blueprint-split review Fixes to the api_v3 package split (PR #553), one per finding verified against the actual code: - __init__.py: _redact_credentials only blanked scalar values under a credential-named key; a bare list of secrets under such a key (e.g. tokens: ["a", "b"]) passed through untouched, since the list branch recursed with no memory that its key looked like a credential. Nested dicts still walk normally (a documented, tested behaviour -- a container like secrets: {api_key: ..., note: ...} is a section name, not a value to blank outright), but any value reached under a credential-shaped key is now actually blanked. - __init__.py: the OAuth helper script's raw stderr/stdout went to logger.error unredacted (CWE-532) right next to a comment claiming this was deliberate; the HTTP response already used the existing redact_text helper. Routed the log line through the same helper. - __init__.py / starlark.py: the standalone Starlark manifest fallback (used when the plugin instance isn't loaded) read-modified-wrote manifest.json with no lock, unlike StarlarkAppsPlugin._update_manifest_safe (plugin-repos/starlark-apps/manager.py), which already holds an flock for the same file when the plugin is loaded. Added _starlark_manifest_lock, mirroring that pattern, and wrapped every standalone read-modify-write call site in it. The app-config update route also wrote config.json and the manifest as two separate, non-transactional writes (a second, distinct finding at the same call site); config.json is now rolled back if the manifest write that follows it fails. - backup.py: restore options used bare bool() on values from the request, so {"restore_secrets": "false"} restored secrets anyway (bool("false") is True). Switched to the existing _coerce_to_bool helper already used for this exact purpose elsewhere in the package. - config.py: an automated import-rewrite mangled four user-facing validation strings and their neighbouring comments -- "Invalid start time" had become "Invalid start _pkg.time" (and likewise for "end time") in both the schedule and dim-schedule per-day validation paths. - display.py: `import _pkg.time as time_module` -- _pkg is a local alias for the package, not a real importable module, so this raised ModuleNotFoundError whenever a caller restarted an already-running display service via /display/on-demand/start, after the on-demand request was already written to cache. Fixed to `import time`. Audited the rest of the package for the same `_pkg.<module>` import mistake; every other `_pkg.` reference is a legitimate attribute read-through (`_pkg.time.time()`, `_pkg._get_starlark_plugin()`, ...), not a broken import statement. - fonts.py: validate_file_upload's max_size_mb parameter is silently unused by that helper (it only checks filename/extension) -- the font upload route saved arbitrarily large files as a result. Added the same seek-and-check pattern already used for the sibling .star upload. - wifi.py: two ad hoc, inconsistent bool coercions. POST /wifi/ap/auto-enable used bare bool(), so a JSON string "false" enabled it. POST /wifi/radio's enabled/force parsing recognized real bool and some strings but not int 1/0 (1 is True is False in Python). Factored one small _parse_bool_ish helper local to this file and used it at all three sites. Not changed: the "unknown/misspelled restore option keys default to True" half of the backup.py finding -- the file's own comment documents that a missing key deliberately means "restore everything," matching the already-existing JSON-parse-failure guard a few lines above it; only the bool-coercion defect was a real bug. Added or extended regression tests for every fix, following each area's existing test conventions. Full suite: 4328 passed, 62 skipped, 2 failed on both this branch and origin/main (missing tzdata package breaks two timezone-alias tests in test_onboarding_checklist.py, unrelated to this change) -- no new failures. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01S3bPMESe2TfrGvbs1ef9c5 * fix(api-v3): reject unknown restore option keys CodeRabbit's review of the blueprint split (#553) asked that POST /backup/restore reject option keys outside RestoreOptions' known set. The follow-up commit fixed the bool("false")-is-True bug with _coerce_to_bool but never added the key check: a typo'd or renamed key (e.g. "restoreSecrets") is silently ignored by opts_dict.get(key, True), so the flag stays at its True default and secrets get restored despite the caller's request saying otherwise -- with no indication anything was wrong. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Vmcwf5vMgYqdt8bJTZtiwb * fix(api-v3): address CodeRabbit findings on the blueprint split - _redact_credentials: blank scalar descendants of objects reached through a credential-owned list (e.g. tokens: [{"value": "secret"}]) regardless of field name -- the existing name-based walk only protected direct dict values under a credential key, not list items. - wifi.py: reject enabled/force/auto_enable_ap_mode values _parse_bool_ish can't recognize (400) instead of silently treating them as False, which could disable Wi-Fi or the radio itself. - Starlark manifest locking: lock a stable manifest.json.lock sidecar instead of manifest.json itself, in both the standalone route path (_starlark_manifest_lock) and the plugin path (StarlarkAppsPlugin._save_manifest / _update_manifest_safe). manifest.json is replaced by an atomic rename on every write, which swaps in a fresh inode; a lock held on the old inode does not exclude a second locker that opens the path afresh right after the rename and gets the new inode, so two writers could race despite each holding "a lock". A sidecar that no write ever touches always resolves to the same inode for every locker. Skipped as stale: the "serialize the complete manifest read-modify-write" finding at api_v3/__init__.py -- every standalone handler that calls _write_starlark_manifest is already wrapped in _starlark_manifest_lock() on this branch. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(api-v3): re-check reconciliation findings by the reconciler's own rules Both CodeRabbit findings on the merge commit, verified against the code first. Major, plugins.py: the stale-findings filter derived its own notion of "in config" and "on disk", and both were looser than the reconciliation module's. set(load_config()) also contains system keys, the secrets-file keys load_config() merges in, and non-dict values; and any directory holding a manifest.json counted as installed even when that manifest does not parse. Either looseness clears a finding that is still true -- and a secrets key read as a plugin is the precise bug the filter exists to stop reporting, so reintroducing that asymmetry while re-checking was the wrong way round. The two extractions now live in state_reconciliation.py as config_plugin_ids() and disk_plugin_ids(), with ignored_config_keys() and secrets_top_level_keys() alongside. _get_config_state() and _get_disk_state() use them too, so there is one definition rather than two that can drift. _get_disk_state() re-reads each manifest for version/name after taking membership from the shared extractor; that costs one extra small read per plugin on a path that runs once per boot. Minor, the new test: the fixture assigned api_v3.config_manager and api_v3.plugin_manager directly. Those live on a module-level blueprint singleton, so the mocks leaked into every later test that imports api_v3 -- pointing at a tmp_path already deleted. Both now go through monkeypatch.setattr, which restores them. This is the same pollution class that made an earlier test in this session break seven unrelated ones, so it is worth getting right. Five cases added for the parity itself: a secrets key, a system key and a non-dict value must not clear an "installed but missing from config" finding, and neither an unparseable manifest nor a .standalone-backup- directory may count as installed. All five fail against the looser version. Linux CI on the preceding commit: Core unit tests, plugin harness, CodeQL and CodeRabbit all pass. Codacy reads action_required on every commit of this branch including the first, so it is pre-existing and not from this work. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
414 lines
14 KiB
Python
414 lines
14 KiB
Python
"""Wi-Fi scanning, connection and status routes.
|
|
|
|
Routes decorate the shared `api_v3` Blueprint from ._common, so their
|
|
endpoint names are unchanged by living here.
|
|
"""
|
|
from web_interface.blueprints.api_v3 import (
|
|
api_v3, describe_exception, jsonify, logger, request,
|
|
)
|
|
|
|
|
|
def _parse_bool_ish(value):
|
|
"""Coerce a JSON value that is supposed to be a boolean.
|
|
|
|
A JSON boolean arrives as a real Python bool, but these routes are a
|
|
public HTTP contract and not every caller sends one. `bool(value)` gets
|
|
two common cases wrong: `bool("false")` is True (a non-empty string is
|
|
always truthy), and a plain int does not match an `is True` check
|
|
(`1 is True` is False, since `True` is a distinct singleton from the int
|
|
`1`) -- so a caller sending `{"enabled": 1}` was silently treated as
|
|
False. Recognizes a real bool, "true"/"false"/"1"/"0"/"yes"/"no"
|
|
case-insensitively, and int 1/0.
|
|
|
|
Returns None for anything else, rather than guessing. A supplied-but-
|
|
unrecognized value (e.g. a typo) used to silently become False here,
|
|
which for `enabled` on the radio route could disconnect Wi-Fi the caller
|
|
never asked to turn off -- callers must treat None as a validation
|
|
error, not a real False.
|
|
"""
|
|
if isinstance(value, bool):
|
|
return value
|
|
if isinstance(value, str):
|
|
lowered = value.strip().lower()
|
|
if lowered in ('true', '1', 'yes'):
|
|
return True
|
|
if lowered in ('false', '0', 'no'):
|
|
return False
|
|
return None
|
|
if isinstance(value, int):
|
|
if value == 1:
|
|
return True
|
|
if value == 0:
|
|
return False
|
|
return None
|
|
return None
|
|
|
|
|
|
# WiFi Management Endpoints
|
|
@api_v3.route('/wifi/status', methods=['GET'])
|
|
def get_wifi_status():
|
|
"""Get current WiFi connection status"""
|
|
try:
|
|
from src.wifi_manager import WiFiManager
|
|
|
|
wifi_manager = WiFiManager()
|
|
status = wifi_manager.get_wifi_status()
|
|
|
|
# Get auto-enable setting from config
|
|
auto_enable_ap = wifi_manager.config.get("auto_enable_ap_mode", True) # Default: True (safe due to grace period)
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'data': {
|
|
'connected': status.connected,
|
|
'ssid': status.ssid,
|
|
'ip_address': status.ip_address,
|
|
'signal': status.signal,
|
|
'ap_mode_active': status.ap_mode_active,
|
|
'auto_enable_ap_mode': auto_enable_ap
|
|
}
|
|
})
|
|
except Exception as e:
|
|
logger.error("%s failed", request.path, exc_info=True)
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': 'An error occurred; see logs for details',
|
|
'details': describe_exception(e)
|
|
}), 500
|
|
@api_v3.route('/wifi/scan', methods=['GET'])
|
|
def scan_wifi_networks():
|
|
"""Scan for available WiFi networks
|
|
|
|
If AP mode is active, it will be temporarily disabled during scanning
|
|
and automatically re-enabled afterward. Users connected to the AP will
|
|
be briefly disconnected during this process.
|
|
"""
|
|
try:
|
|
from src.wifi_manager import WiFiManager
|
|
|
|
wifi_manager = WiFiManager()
|
|
|
|
# Check if AP mode is active before scanning (for user notification)
|
|
ap_was_active = wifi_manager._is_ap_mode_active()
|
|
|
|
# Perform the scan (this will handle AP mode disabling/enabling internally)
|
|
networks, _was_cached = wifi_manager.scan_networks()
|
|
|
|
# Convert to dict format
|
|
networks_data = [
|
|
{
|
|
'ssid': net.ssid,
|
|
'signal': net.signal,
|
|
'security': net.security,
|
|
'frequency': net.frequency
|
|
}
|
|
for net in networks
|
|
]
|
|
|
|
response_data = {
|
|
'status': 'success',
|
|
'data': networks_data
|
|
}
|
|
|
|
# Inform user if AP mode was temporarily disabled
|
|
if ap_was_active:
|
|
response_data['message'] = (
|
|
f'Found {len(networks_data)} networks. '
|
|
'Note: AP mode was temporarily disabled during scanning and has been re-enabled. '
|
|
'If you were connected to the setup network, you may need to reconnect.'
|
|
)
|
|
|
|
return jsonify(response_data)
|
|
except Exception as e:
|
|
logger.error("Error scanning WiFi networks", exc_info=True)
|
|
error_message = 'An error occurred while scanning WiFi networks; see logs for details'
|
|
|
|
# Provide more specific error messages for common issues
|
|
error_str = str(e).lower()
|
|
if 'permission' in error_str or 'sudo' in error_str:
|
|
error_message = (
|
|
'Permission error while scanning. '
|
|
'The WiFi scan requires appropriate permissions. '
|
|
'Please ensure the application has necessary privileges.'
|
|
)
|
|
elif 'timeout' in error_str:
|
|
error_message = (
|
|
'WiFi scan timed out. '
|
|
'The scan took too long to complete. '
|
|
'This may happen if the WiFi interface is busy or in use.'
|
|
)
|
|
elif 'no wifi' in error_str or 'not available' in error_str:
|
|
error_message = (
|
|
'WiFi scanning tools are not available. '
|
|
'Please ensure NetworkManager (nmcli) or iwlist is installed.'
|
|
)
|
|
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': error_message
|
|
}), 500
|
|
@api_v3.route('/wifi/connect', methods=['POST'])
|
|
def connect_wifi():
|
|
"""Connect to a WiFi network"""
|
|
try:
|
|
from src.wifi_manager import WiFiManager
|
|
|
|
data = request.get_json(silent=True)
|
|
if not data:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': 'Request body is required'
|
|
}), 400
|
|
|
|
if 'ssid' not in data:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': 'SSID is required'
|
|
}), 400
|
|
|
|
ssid = data['ssid']
|
|
if not ssid or not ssid.strip():
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': 'SSID cannot be empty'
|
|
}), 400
|
|
|
|
ssid = ssid.strip()
|
|
password = data.get('password', '') or ''
|
|
|
|
wifi_manager = WiFiManager()
|
|
success, message = wifi_manager.connect_to_network(ssid, password)
|
|
|
|
if success:
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': message
|
|
})
|
|
else:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': message or 'Failed to connect to network'
|
|
}), 400
|
|
except Exception as e:
|
|
logger.error("Error connecting to WiFi", exc_info=True)
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': 'An error occurred; see logs for details', 'details': describe_exception(e)
|
|
}), 500
|
|
@api_v3.route('/wifi/disconnect', methods=['POST'])
|
|
def disconnect_wifi():
|
|
"""Disconnect from the current WiFi network"""
|
|
try:
|
|
from src.wifi_manager import WiFiManager
|
|
|
|
wifi_manager = WiFiManager()
|
|
success, message = wifi_manager.disconnect_from_network()
|
|
|
|
if success:
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': message
|
|
})
|
|
else:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': message or 'Failed to disconnect from network'
|
|
}), 400
|
|
except Exception as e:
|
|
logger.error("Error disconnecting from WiFi", exc_info=True)
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': 'An error occurred; see logs for details', 'details': describe_exception(e)
|
|
}), 500
|
|
@api_v3.route('/wifi/ap/enable', methods=['POST'])
|
|
def enable_ap_mode():
|
|
"""Enable access point mode"""
|
|
try:
|
|
from src.wifi_manager import WiFiManager
|
|
|
|
wifi_manager = WiFiManager()
|
|
_force_raw = (request.get_json(silent=True) or {}).get('force', False)
|
|
force = _force_raw is True or (isinstance(_force_raw, str) and _force_raw.lower() in ('true', '1'))
|
|
success, message = wifi_manager.enable_ap_mode(force=force)
|
|
|
|
if success:
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': message
|
|
})
|
|
else:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': message
|
|
}), 400
|
|
except Exception as e:
|
|
logger.error("%s failed", request.path, exc_info=True)
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': 'An error occurred; see logs for details',
|
|
'details': describe_exception(e)
|
|
}), 500
|
|
@api_v3.route('/wifi/ap/disable', methods=['POST'])
|
|
def disable_ap_mode():
|
|
"""Disable access point mode"""
|
|
try:
|
|
from src.wifi_manager import WiFiManager
|
|
|
|
wifi_manager = WiFiManager()
|
|
success, message = wifi_manager.disable_ap_mode()
|
|
|
|
if success:
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': message
|
|
})
|
|
else:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': message
|
|
}), 400
|
|
except Exception as e:
|
|
logger.error("%s failed", request.path, exc_info=True)
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': 'An error occurred; see logs for details',
|
|
'details': describe_exception(e)
|
|
}), 500
|
|
@api_v3.route('/wifi/ap/auto-enable', methods=['GET'])
|
|
def get_auto_enable_ap_mode():
|
|
"""Get auto-enable AP mode setting"""
|
|
try:
|
|
from src.wifi_manager import WiFiManager
|
|
|
|
wifi_manager = WiFiManager()
|
|
auto_enable = wifi_manager.config.get("auto_enable_ap_mode", True) # Default: True (safe due to grace period)
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'data': {
|
|
'auto_enable_ap_mode': auto_enable
|
|
}
|
|
})
|
|
except Exception as e:
|
|
logger.error("%s failed", request.path, exc_info=True)
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': 'An error occurred; see logs for details',
|
|
'details': describe_exception(e)
|
|
}), 500
|
|
@api_v3.route('/wifi/ap/auto-enable', methods=['POST'])
|
|
def set_auto_enable_ap_mode():
|
|
"""Set auto-enable AP mode setting"""
|
|
try:
|
|
from src.wifi_manager import WiFiManager
|
|
|
|
data = request.get_json(silent=True)
|
|
if data is None or 'auto_enable_ap_mode' not in data:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': 'auto_enable_ap_mode is required'
|
|
}), 400
|
|
|
|
auto_enable = _parse_bool_ish(data['auto_enable_ap_mode'])
|
|
if auto_enable is None:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': 'auto_enable_ap_mode must be a boolean'
|
|
}), 400
|
|
|
|
wifi_manager = WiFiManager()
|
|
wifi_manager.config["auto_enable_ap_mode"] = auto_enable
|
|
wifi_manager._save_config()
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': f'Auto-enable AP mode set to {auto_enable}',
|
|
'data': {
|
|
'auto_enable_ap_mode': auto_enable
|
|
}
|
|
})
|
|
except Exception as e:
|
|
logger.error("%s failed", request.path, exc_info=True)
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': 'An error occurred; see logs for details',
|
|
'details': describe_exception(e)
|
|
}), 500
|
|
@api_v3.route('/wifi/radio', methods=['GET'])
|
|
def get_wifi_radio():
|
|
"""Get current WiFi radio state (enabled/disabled) and wired-fallback status."""
|
|
try:
|
|
from src.wifi_manager import WiFiManager
|
|
|
|
wifi_manager = WiFiManager()
|
|
state = wifi_manager.get_wifi_radio_state()
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'data': state
|
|
})
|
|
except Exception as e:
|
|
logger.error("Error getting WiFi radio state", exc_info=True)
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': 'An error occurred; see logs for details', 'details': describe_exception(e)
|
|
}), 500
|
|
@api_v3.route('/wifi/radio', methods=['POST'])
|
|
def set_wifi_radio():
|
|
"""Turn the WiFi radio on or off.
|
|
|
|
Body: {"enabled": bool, "force": bool (optional)}. Disabling is refused
|
|
unless Ethernet is connected or force=True, to avoid locking the user out
|
|
of this web interface.
|
|
"""
|
|
try:
|
|
from src.wifi_manager import WiFiManager
|
|
|
|
data = request.get_json(silent=True) or {}
|
|
if 'enabled' not in data:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': 'enabled is required'
|
|
}), 400
|
|
|
|
# Parse defensively: bool("false") is True and a plain int never
|
|
# matches `is True`, so `_parse_bool_ish` handles bool, string and
|
|
# int 1/0 — the endpoint is a public contract, not just the shipped
|
|
# UI (which always sends real JSON booleans). An unrecognized value
|
|
# must be rejected, not silently disable the radio: this is the
|
|
# route that can drop the caller's own connection to this interface.
|
|
enabled = _parse_bool_ish(data['enabled'])
|
|
if enabled is None:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': 'enabled must be a boolean'
|
|
}), 400
|
|
force = _parse_bool_ish(data.get('force', False))
|
|
if force is None:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': 'force must be a boolean'
|
|
}), 400
|
|
|
|
wifi_manager = WiFiManager()
|
|
success, message, reason = wifi_manager.set_wifi_radio(enabled, force=force)
|
|
|
|
if success:
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': message,
|
|
'data': wifi_manager.get_wifi_radio_state()
|
|
})
|
|
else:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': message,
|
|
'reason': reason
|
|
}), 400
|
|
except Exception as e:
|
|
logger.error("Error setting WiFi radio state", exc_info=True)
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': 'An error occurred; see logs for details', 'details': describe_exception(e)
|
|
}), 500
|