fix(devices): address pre-merge review findings
Closes the issues surfaced by the pre-merge code review of the expand-device-support branch. CRITICAL #2 -- update_device double-encrypts secrets in memory. storage/device_store.py round-tripped through device.to_dict() which encrypts hue_username / hue_client_key / ble_govee_key / nanoleaf_token via _enc(), but Device.__init__ does not decrypt. The cached self._items[device_id] thus held ciphertext where plaintext belonged, breaking runtime auth for paired devices on any update -- even an innocuous rename. Sourcing kwargs from vars(device) directly avoids the round-trip. Regression tests cover Nanoleaf and Hue. HIGH #3 -- secrets leaked in GET /api/v1/devices response. DeviceResponse previously returned nanoleaf_token / hue_username / hue_client_key in plaintext (decrypted server-side from storage), defeating the encryption-at-rest. Replaced with nanoleaf_paired and hue_paired booleans. ble_govee_key intentionally stays -- it's a user-managed value pasted from a third-party tool, must remain visible for edit. Frontend types.ts + the one nanoleaf_token reader updated to the boolean. HIGH #4 -- SSRF surface. validate_lan_host() added to net_classify.py; called from each new driver's validate_device (DDP / Yeelight / WiZ / LIFX / Govee / OPC / Nanoleaf) and from pair_device. Rejects literal public IPs with a descriptive ValueError; non-IP hostnames pass through (mDNS labels, bare hostnames). RFC6890 ranges (documentation, former class E) are accepted as LAN-like since Python's ipaddress.is_private treats them so -- correct policy for LedGrab. HIGH #5 -- decrypt failure deletes the device row. _dec() now catches the exception, logs an error, and returns "" instead of propagating. Without the fix, a regenerated data/.secret_key would silently make every Hue / Nanoleaf / BLE-Govee device disappear from the device list on next startup. Regression test asserts a corrupt envelope leaves the device hydratable. HIGH #6 -- update_device route does not rstrip("/") for non-WLED. Moved the trim before the WLED-specific scheme inference so every device type gets consistent URL normalization between create and update. MEDIUM #7 -- Govee discovery port 4002 collision. Added a lazily- initialized module-level asyncio.Lock that serializes concurrent discover_govee_devices() calls; the previous behavior had the second parallel scan silently return [] when the first still held port 4002. Error message also clarified to mention another Govee tool. MEDIUM #8 -- Nanoleaf discover() leaked browser tasks on cancellation. Moved the browser cancel loop into the finally block so an interrupted mDNS scan still tears them down. MEDIUM #9 -- pair endpoint logged user-supplied URL with exc_info=True. Added _sanitize_url_for_log() that strips userinfo + fragment, and demoted the log from exc_info to type(exc).__name__ + str(exc) so a hostile receiver's response body can't end up in the log file. LOW -- Nanoleaf was the only client without a .port property. Added one (returns NANOLEAF_PORT, fixed) for cross-driver symmetry. LOW -- no end-to-end pair-then-create coverage. Added TestPairThenCreateFlow.test_pair_then_create_persists_encrypted_token which exercises the full path: POST /api/v1/devices/pair returned fields, store.create_device, then asserts (a) in-memory plaintext, (b) to_config() plaintext, (c) persisted ciphertext, (d) API response strip + paired-boolean. Tests: 1379 pass (was 1358 -- 21 new regression tests added). ruff clean. TypeScript clean.
This commit is contained in:
@@ -340,3 +340,80 @@ class TestPairDevice:
|
||||
def test_missing_required_fields_returns_422(self, client):
|
||||
resp = client.post("/api/v1/devices/pair", json={"device_type": "nanoleaf"})
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
class TestPairThenCreateFlow:
|
||||
"""End-to-end coverage: pair, then persist; assert the token is
|
||||
encrypted at rest and decrypted in to_config(), and that the API
|
||||
response strips the secret.
|
||||
|
||||
Closes the LOW gap in the pre-merge review: pair-route tests stopped
|
||||
at the 200 response, never carrying the returned fields through to
|
||||
storage to verify the round-trip and the response-strip.
|
||||
"""
|
||||
|
||||
def test_pair_then_create_persists_encrypted_token(self, client, device_store):
|
||||
from ledgrab.api.routes.devices import _device_to_response
|
||||
from ledgrab.core.devices import led_client as _led_client
|
||||
from ledgrab.core.devices.led_client import DeviceHealth
|
||||
|
||||
class _NanoleafLikeStub:
|
||||
@property
|
||||
def device_type(self):
|
||||
return "nanoleaf_like_stub"
|
||||
|
||||
@property
|
||||
def capabilities(self):
|
||||
return {"manual_led_count", "requires_pairing"}
|
||||
|
||||
async def pair_device(self, url):
|
||||
return {"nanoleaf_token": "secret-paired-token"}
|
||||
|
||||
async def validate_device(self, url):
|
||||
return {"led_count": 9}
|
||||
|
||||
def create_client(self, config, *, deps):
|
||||
raise AssertionError("not used")
|
||||
|
||||
async def check_health(self, url, http_client, prev_health=None):
|
||||
return DeviceHealth(online=True)
|
||||
|
||||
_led_client.register_provider(_NanoleafLikeStub())
|
||||
try:
|
||||
# Step 1: pair via the route
|
||||
pair_resp = client.post(
|
||||
"/api/v1/devices/pair",
|
||||
json={"device_type": "nanoleaf_like_stub", "url": "stub://1.2.3.4"},
|
||||
)
|
||||
assert pair_resp.status_code == 200, pair_resp.text
|
||||
fields = pair_resp.json()["fields"]
|
||||
assert fields == {"nanoleaf_token": "secret-paired-token"}
|
||||
|
||||
# Step 2: persist via the store (skip the route's create path
|
||||
# which would require a real validate_device handshake)
|
||||
device = device_store.create_device(
|
||||
name="E2E Paired",
|
||||
url="stub://1.2.3.4",
|
||||
led_count=9,
|
||||
device_type="nanoleaf",
|
||||
**fields,
|
||||
)
|
||||
|
||||
# In-memory device holds plaintext
|
||||
assert device.nanoleaf_token == "secret-paired-token"
|
||||
|
||||
# to_config surfaces plaintext to the provider
|
||||
config = device.to_config()
|
||||
assert config.nanoleaf_token == "secret-paired-token"
|
||||
|
||||
# Persisted row holds ciphertext (envelope prefix)
|
||||
persisted = device.to_dict()
|
||||
assert persisted["nanoleaf_token"].startswith("ENC:v1:")
|
||||
assert persisted["nanoleaf_token"] != "secret-paired-token"
|
||||
|
||||
# API response strips the token; only a boolean flag remains
|
||||
resp = _device_to_response(device).model_dump()
|
||||
assert "nanoleaf_token" not in resp
|
||||
assert resp.get("nanoleaf_paired") is True
|
||||
finally:
|
||||
_led_client._provider_registry.pop("nanoleaf_like_stub", None)
|
||||
|
||||
@@ -270,3 +270,81 @@ def test_update_device_ignores_unknown_fields(device_store):
|
||||
# Should not raise even with an unknown kwarg
|
||||
updated = device_store.update_device(device.id, nonexistent_field="foo")
|
||||
assert updated.name == "Test WLED"
|
||||
|
||||
|
||||
def test_update_device_preserves_plaintext_secrets(device_store):
|
||||
"""update_device must not corrupt encrypted-at-rest secrets in memory.
|
||||
|
||||
Regression test for the to_dict() round-trip bug: to_dict() encrypts
|
||||
hue_username / hue_client_key / ble_govee_key / nanoleaf_token via
|
||||
_enc(), but Device.__init__ does not decrypt -- so the old
|
||||
update_device path would leave the cached Device holding ciphertext
|
||||
where plaintext belongs, breaking runtime auth for paired devices
|
||||
until server restart. See pre-merge review CRITICAL #2.
|
||||
"""
|
||||
device = device_store.create_device(
|
||||
name="Office panels",
|
||||
url="nanoleaf://192.168.1.99",
|
||||
led_count=9,
|
||||
device_type="nanoleaf",
|
||||
nanoleaf_token="plaintext-secret-token",
|
||||
)
|
||||
assert device.nanoleaf_token == "plaintext-secret-token"
|
||||
|
||||
# An innocuous rename must not touch the secret.
|
||||
updated = device_store.update_device(device.id, name="Renamed panels")
|
||||
assert (
|
||||
updated.nanoleaf_token == "plaintext-secret-token"
|
||||
), "rename round-tripped the secret through to_dict() and corrupted it"
|
||||
|
||||
# The cached _items entry must also hold plaintext (it is what
|
||||
# provider.create_client() reads).
|
||||
cached = device_store.get_device(device.id)
|
||||
assert cached.nanoleaf_token == "plaintext-secret-token"
|
||||
|
||||
# And to_config() must surface the plaintext token to the provider.
|
||||
config = cached.to_config()
|
||||
assert config.nanoleaf_token == "plaintext-secret-token"
|
||||
|
||||
|
||||
def test_update_device_preserves_hue_secrets(device_store):
|
||||
"""Same bug applies to hue_username / hue_client_key / ble_govee_key."""
|
||||
device = device_store.create_device(
|
||||
name="Hue bridge",
|
||||
url="http://192.168.1.50",
|
||||
led_count=10,
|
||||
device_type="hue",
|
||||
hue_username="plain-hue-user",
|
||||
hue_client_key="plain-client-key",
|
||||
)
|
||||
|
||||
updated = device_store.update_device(device.id, name="Renamed bridge")
|
||||
|
||||
assert updated.hue_username == "plain-hue-user"
|
||||
assert updated.hue_client_key == "plain-client-key"
|
||||
|
||||
|
||||
def test_from_dict_clears_corrupt_secret_envelope(device_store):
|
||||
"""A corrupt nanoleaf_token must not delete the device row.
|
||||
|
||||
Regression test for HIGH #5: previously _dec() would raise on a bad
|
||||
envelope, propagate up through BaseSqliteStore._load, and the entire
|
||||
device row would silently disappear. The fix clears the offending
|
||||
field and logs an error instead.
|
||||
"""
|
||||
payload = {
|
||||
"id": "device_abc12345",
|
||||
"name": "Will survive",
|
||||
"url": "nanoleaf://192.168.1.50",
|
||||
"led_count": 9,
|
||||
"device_type": "nanoleaf",
|
||||
# Looks like an envelope (ENC:v1: prefix) but the base64 body is
|
||||
# truncated / corrupt -- the AES-GCM auth tag will fail to verify.
|
||||
"nanoleaf_token": "ENC:v1:not-real-base64-cipher-text",
|
||||
}
|
||||
|
||||
restored = Device.from_dict(payload)
|
||||
|
||||
# The device hydrates; the broken secret is cleared, not crashed.
|
||||
assert restored.id == "device_abc12345"
|
||||
assert restored.nanoleaf_token == ""
|
||||
|
||||
@@ -322,24 +322,34 @@ def test_provider_device_type_and_capabilities():
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_provider_pair_device_returns_nanoleaf_token(respx_mock):
|
||||
respx_mock.post(f"http://1.2.3.4:{NANOLEAF_PORT}/api/v1/new").mock(
|
||||
# Use a LAN address; validate_lan_host now rejects public IPs at the
|
||||
# provider boundary (review HIGH #4).
|
||||
respx_mock.post(f"http://192.168.1.50:{NANOLEAF_PORT}/api/v1/new").mock(
|
||||
return_value=httpx.Response(200, json={"auth_token": "ttoken"}),
|
||||
)
|
||||
provider = NanoleafDeviceProvider()
|
||||
|
||||
fields = await provider.pair_device("nanoleaf://1.2.3.4")
|
||||
fields = await provider.pair_device("nanoleaf://192.168.1.50")
|
||||
|
||||
assert fields == {"nanoleaf_token": "ttoken"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_provider_pair_device_propagates_pairing_not_ready(respx_mock):
|
||||
respx_mock.post(f"http://1.2.3.4:{NANOLEAF_PORT}/api/v1/new").mock(
|
||||
respx_mock.post(f"http://192.168.1.50:{NANOLEAF_PORT}/api/v1/new").mock(
|
||||
return_value=httpx.Response(403),
|
||||
)
|
||||
provider = NanoleafDeviceProvider()
|
||||
with pytest.raises(PairingNotReady):
|
||||
await provider.pair_device("nanoleaf://1.2.3.4")
|
||||
await provider.pair_device("nanoleaf://192.168.1.50")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_provider_pair_device_rejects_public_ip():
|
||||
"""validate_lan_host fires before pair_nanoleaf even runs."""
|
||||
provider = NanoleafDeviceProvider()
|
||||
with pytest.raises(ValueError, match="LedGrab is LAN-only"):
|
||||
await provider.pair_device("nanoleaf://1.1.1.1")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
||||
@@ -135,3 +135,55 @@ def test_is_loopback_recognises_loopback(host: str) -> None:
|
||||
)
|
||||
def test_is_loopback_rejects_other(host) -> None:
|
||||
assert is_loopback(host) is False
|
||||
|
||||
|
||||
# ─── validate_lan_host ────────────────────────────────────────────
|
||||
|
||||
|
||||
from ledgrab.utils.net_classify import validate_lan_host # noqa: E402
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"host",
|
||||
[
|
||||
"127.0.0.1",
|
||||
"10.0.0.5",
|
||||
"192.168.1.50",
|
||||
"172.16.0.10",
|
||||
"fe80::1",
|
||||
"fc00::1",
|
||||
"0.0.0.0",
|
||||
# IPv6 with brackets and zone id
|
||||
"[fe80::1%eth0]",
|
||||
# Plain hostnames pass through (mDNS / bare-label)
|
||||
"bulb.local",
|
||||
"office-strip",
|
||||
"controller.lan",
|
||||
"", # empty also passes (caller handles validation upstream)
|
||||
],
|
||||
)
|
||||
def test_validate_lan_host_accepts_local(host) -> None:
|
||||
"""Loopback / private / link-local / hostnames must not raise."""
|
||||
validate_lan_host(host)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"host",
|
||||
[
|
||||
"1.1.1.1", # Cloudflare DNS (routable public)
|
||||
"8.8.8.8", # Google DNS
|
||||
"224.0.0.1", # multicast — not a LAN device address
|
||||
"2606:4700:4700::1111", # Cloudflare DNS IPv6
|
||||
],
|
||||
)
|
||||
def test_validate_lan_host_rejects_public_or_multicast(host) -> None:
|
||||
"""Reject genuinely-public addresses and multicast targets.
|
||||
|
||||
Note: Python's :func:`ipaddress.ip_address(...).is_private` treats
|
||||
RFC6890 ranges (documentation 192.0.2/24, former class E 240/4, etc.)
|
||||
as private, so those are accepted as "local-ish" by our classifier --
|
||||
which is the correct policy for LedGrab: those ranges aren't routable
|
||||
public addresses, and refusing them would be unhelpful theatre.
|
||||
"""
|
||||
with pytest.raises(ValueError, match="LedGrab is LAN-only"):
|
||||
validate_lan_host(host)
|
||||
|
||||
Reference in New Issue
Block a user