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>
This commit is contained in:
Claude
2026-09-11 10:35:27 +00:00
parent 6e06b60615
commit 7c3e4b8a38
7 changed files with 192 additions and 25 deletions
+24 -1
View File
@@ -195,7 +195,7 @@ class TestRadio:
@pytest.mark.parametrize("raw,expected", [
(True, True), ("true", True), ("1", True), ("yes", True),
(False, False), ("false", False), ("off", False), (0, False),
(False, False), ("false", False), ("no", False), (0, False),
])
def test_enabled_coercion_is_string_aware(
self, api_v3_client, wifi_manager, raw, expected):
@@ -228,6 +228,22 @@ class TestRadio:
wifi_manager.set_wifi_radio.side_effect = RuntimeError("boom")
assert api_v3_client.post(self.URL, json={"enabled": True}).status_code == 500
def test_unrecognized_enabled_value_is_rejected_not_treated_as_false(
self, api_v3_client, wifi_manager):
# An invalid `enabled` used to silently fall back to False, which
# can disconnect Wi-Fi (or, with force=true, drop the caller's own
# connection to this interface) even though nothing asked for that.
response = api_v3_client.post(self.URL, json={"enabled": "typo"})
assert response.status_code == 400
wifi_manager.set_wifi_radio.assert_not_called()
def test_unrecognized_force_value_is_rejected_not_treated_as_false(
self, api_v3_client, wifi_manager):
response = api_v3_client.post(
self.URL, json={"enabled": False, "force": "typo"})
assert response.status_code == 400
wifi_manager.set_wifi_radio.assert_not_called()
class TestAutoEnableApMode:
URL = "/api/v3/wifi/ap/auto-enable"
@@ -259,6 +275,13 @@ class TestAutoEnableApMode:
api_v3_client.post(self.URL, json={"auto_enable_ap_mode": "false"})
assert wifi_manager.config["auto_enable_ap_mode"] is False
def test_unrecognized_value_is_rejected_not_treated_as_false(
self, api_v3_client, wifi_manager):
wifi_manager.config = {}
response = api_v3_client.post(self.URL, json={"auto_enable_ap_mode": "typo"})
assert response.status_code == 400
assert "auto_enable_ap_mode" not in wifi_manager.config
class TestRadioEnabledAndForceAcceptIntegers:
"""`{"enabled": 1}` / `{"enabled": 0}` used to be mishandled: the old
+11
View File
@@ -125,6 +125,17 @@ def test_a_list_nested_inside_a_credential_named_container_is_also_blanked():
assert out["auth"]["note"] == "keep"
def test_objects_inside_a_credential_named_list_are_blanked_regardless_of_field_names():
"""A list item has no field name of its own to test against
`_CREDENTIAL_NAME_PARTS`, so an object reached through a credential-owned
list must be blanked outright rather than walked by field name -- unlike
the dict-directly-under-a-credential-key case in the test above.
"""
config = {"tokens": [{"value": "secret", "kind": "bearer"}]}
out = _redact_credentials(config)
assert out["tokens"] == [{"value": "", "kind": ""}]
def test_credential_shaped_container_with_named_fields_still_only_blanks_those():
"""Unchanged behaviour for the case the container test above already
covers: a dict whose sub-keys are semantically named fields is walked
+45
View File
@@ -6,6 +6,7 @@ frame for the whole display_duration instead of rotating on -- the same class
of defect as a display() that returns None.
"""
import json
import sys
import types
from pathlib import Path
@@ -197,6 +198,50 @@ class TestInstalledAppsTakeTurns:
assert p.display(force_clear=True) is False
class TestManifestLockSidecarIsStable:
"""_save_manifest and _update_manifest_safe must lock a sidecar file that
a manifest write never replaces. manifest.json itself is swapped for a
fresh inode by every atomic write (temp file + rename); a lock taken on
manifest.json directly would not exclude a second locker whose fresh
os.open() lands on that new inode right after a write, so two writers
could still race each other despite each believing it held the lock.
web_interface._starlark_manifest_lock (web_interface/blueprints/api_v3)
must lock this same sidecar name for that guarantee to hold across both
the plugin-loaded and standalone paths.
"""
def _plugin_with_manifest_dir(self, manager_module, tmp_path):
p = _plugin(manager_module)
p.manifest_file = tmp_path / "manifest.json"
p.manifest_lock_file = tmp_path / "manifest.json.lock"
return p
def test_the_lock_sidecar_is_not_the_manifest_file(self, manager_module, tmp_path):
p = self._plugin_with_manifest_dir(manager_module, tmp_path)
assert p.manifest_lock_file != p.manifest_file
assert p.manifest_lock_file.name == "manifest.json.lock"
def test_the_locked_files_inode_survives_repeated_writes(self, manager_module, tmp_path):
p = self._plugin_with_manifest_dir(manager_module, tmp_path)
assert p._save_manifest({"apps": {}})
lock_ino_before = p.manifest_lock_file.stat().st_ino
for app_id in ("one", "two", "three"):
def _update(manifest, app_id=app_id):
manifest.setdefault("apps", {})[app_id] = {"enabled": True}
assert p._update_manifest_safe(_update)
lock_ino_after = p.manifest_lock_file.stat().st_ino
assert lock_ino_after == lock_ino_before, (
"the locked file's inode changed across writes -- a locker that "
"opened it before this write and one that opens it after would "
"no longer contend for the same lock")
manifest = json.loads(p.manifest_file.read_text())
assert set(manifest["apps"]) == {"one", "two", "three"}
class TestAnimationsRunAtFrameRate:
def test_the_plugin_asks_for_the_high_fps_loop(self, manager_module):
"""The controller reads this attribute; a multi-frame app is otherwise
@@ -413,9 +413,41 @@ class TestManifestLockPreventsLostUpdates:
apps_dir.mkdir()
monkeypatch.setattr(module, '_STARLARK_APPS_DIR', apps_dir)
monkeypatch.setattr(module, '_STARLARK_MANIFEST_FILE', apps_dir / 'manifest.json')
monkeypatch.setattr(module, '_STARLARK_MANIFEST_LOCK_FILE', apps_dir / 'manifest.json.lock')
module._write_starlark_manifest({'apps': {}})
return apps_dir
def test_the_locked_file_survives_a_manifest_write(self, starlark_dir):
"""_write_starlark_manifest replaces manifest.json with a fresh inode
on every write (temp file + rename). If the lock were taken on that
same file, a second locker's fresh os.open() right after the rename
would land on the new inode -- unguarded, because only the old,
now-orphaned inode was ever locked -- and two writers could race
despite each believing it "held the lock" (see the docstring on
_starlark_manifest_lock). Locking a sidecar path that no write ever
touches or renames over closes that: the inode identity of what gets
locked must not change across writes.
"""
import os
from web_interface.blueprints import api_v3 as module
with module._starlark_manifest_lock():
manifest = module._read_starlark_manifest()
lock_ino_before = os.stat(module._STARLARK_MANIFEST_LOCK_FILE).st_ino
for app_id in ('one', 'two', 'three'):
with module._starlark_manifest_lock():
manifest = module._read_starlark_manifest()
manifest.setdefault('apps', {})[app_id] = {'enabled': True}
assert module._write_starlark_manifest(manifest)
lock_ino_after = os.stat(module._STARLARK_MANIFEST_LOCK_FILE).st_ino
assert lock_ino_after == lock_ino_before, (
"the locked file's inode changed across writes -- a locker that "
"opened it before this write and one that opens it after would "
"no longer contend for the same lock")
def test_two_concurrent_updates_are_both_kept(self, starlark_dir):
import threading
import time as _time