Fluksio/app
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Device security and robustness (openpanel C0): check code, pnc, lenient report

Browse Source 
- POST /devices/pair answers a 2-character `check`, listed by /devices/pending
  and shown beside each waiting entry; five codes a minute per address, and at
  50 waiting the oldest gives way instead of a 429
- portal-minted device tokens are asked for with the device's nonce (`pnc`) and
  held against it, so unpair and delete revoke them; one without `pnc` stands
  only while the nonce is 0
- a device credential acts as the device (`device_principal`), not as the
  approving superuser, so deleting that account unpairs nothing
- the report is validated port by port: unknown enum values read as `unknown`,
  unreadable ports and sections are left out and logged, `config` may name an
  entry of `config_schemas`; 422 only for no schema 1 or no object
- devices.json is parsed once per change (cached by stat), delete drops the
  report and keeps the nonce, a busy recheck runs in the background
- scope and io only from enabled flows (start/stop tells the devices), a port
  bound twice gives one io row
- the Devices tab's saves cancel and hold back device refetches until settled
This commit is contained in:
Melvin Strobl
2026-10-16 19:26:05 +02:00
parent 13d23f6baf
commit c604cfdf3b
21 changed files with 773 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/api/routes/test_devices.py
+221 -1
View File
@@ -7,6 +7,7 @@ from fastapi.testclient import TestClient
from starlette.websockets import WebSocketDisconnect
from fluksio.api.routes import devices as devices_route
from fluksio.api.routes import oauth
from fluksio.cloud import config as cloud_config
from fluksio.core.config import settings
from fluksio.flow.devices import DeviceState
@@ -18,6 +19,14 @@ PANELS = f"{settings.API_V1_STR}/panels"
DASHBOARDS = f"{settings.API_V1_STR}/dashboards"
@pytest.fixture(autouse=True)
def _pairing_unlimited() -> None:
"""Every test pairs from the test client's one address, which the
per-address limit is about. The test of that limit spends it itself."""
for bucket in [b for b in oauth._hits if b.startswith("pair:")]:
del oauth._hits[bucket]
def _panels(client: TestClient, headers: dict[str, str], config: dict) -> None:
"""Make the stored devices exactly ``config["panels"]``, one call each."""
wanted = {device["id"]: device for device in config["panels"]}
@@ -754,6 +763,16 @@ def test_a_port_wired_twice_or_an_unknown_device_is_an_issue(
("device_port", "gpio.4", False),
("unknown_device", None, False),
}
# The device follows the first of the two, and the tab shows both.
read = client.get(f"{PREFIX}/porch_io", headers=superuser_token_headers).json()
assert [(row["port"], row["message"]) for row in read["io"]] == [
("backlight", "porch.backlight"),
("gpio.4", "porch.doorbell"),
]
assert [b["flow"] for b in read["bindings"] if b["port"] == "gpio.4"] == [
"porch",
"porch_too",
]
# Creating the device is enough, nothing has to be published again.
_panels(
client,
@@ -1192,6 +1211,13 @@ def test_a_pending_code_is_found_by_whichever_worker_is_polled() -> None:
assert polled.get("STALE1") is None
assert polled.count() == 0
# Making room drops the oldest, whichever worker minted it.
for offset, code in enumerate(("OLD111", "MID222", "NEW333")):
later = devices_route._Pending(secret="s", expires=time.time() + 60 + offset)
assert minted.add(code, later)
polled.make_room(3)
assert minted.get("OLD111") is None and polled.count() == 2
def test_removing_the_panel_revokes_its_credential(
client: TestClient, superuser_token_headers: dict[str, str]
@@ -1262,6 +1288,10 @@ def test_a_device_on_this_network_is_adopted_from_the_pending_list(
mine = next(entry for entry in pending if entry["name"] == "knob-3f2a")
assert mine["board"] == "knob" and mine["firmware"] == "1.4"
assert mine["needs_code"] is False
# The check the device shows, to compare before adopting it from here.
assert len(started["check"]) == devices_route.CHECK_LENGTH
assert set(started["check"]) <= set(devices_route.CODE_ALPHABET)
assert mine["check"] == started["check"]
portal = next(entry for entry in pending if entry["remote"])
assert portal["needs_code"] is True
assert len(portal["name"]) == devices_route.HELLO_LENGTH
@@ -1375,7 +1405,9 @@ def test_a_remote_device_is_paired_at_the_portal(
assert calls[0]["url"].endswith("/api/v1/panel-tokens/") # type: ignore[union-attr]
assert calls[0]["headers"]["Authorization"] == "Bearer instance-token" # type: ignore[index]
assert calls[0]["json"] == {"panel": "hallway"} # type: ignore[index]
# With the device's nonce, which the hub puts in the credential.
nonce = app.state.device_store.nonce("hallway")
assert calls[0]["json"] == {"panel": "hallway", "pnc": nonce} # type: ignore[index]
collected = client.get(
f"{PREFIX}/pair/{started['code']}", params={"secret": started["secret"]}
@@ -1646,3 +1678,191 @@ def test_only_the_newest_frame_of_a_batch_is_pushed(tmp_path) -> None:
length = int.from_bytes(frames[0][:4], "big")
assert orjson.loads(frames[0][4 : 4 + length])["digest"] == second["digest"]
assert frames[0][4 + length :] == b"two"
# --------------------------------------------------------------------------
# C0: limits on asking, a lenient report, and what a credential stands on
# --------------------------------------------------------------------------
def test_one_address_asks_for_a_few_codes_a_minute(client: TestClient) -> None:
lan = TestClient(app, client=("192.168.1.77", 50000))
for _ in range(devices_route.PAIR_PER_ADDRESS):
assert lan.post(f"{PREFIX}/pair").status_code == 200
refused = lan.post(f"{PANELS}/pair")
assert refused.status_code == 429
assert refused.headers["retry-after"]
# Another address is not held to this one's.
assert client.post(f"{PREFIX}/pair").status_code == 200
def test_at_the_cap_the_oldest_request_makes_room(
client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
store = devices_route._PendingStore()
store._redis = None
monkeypatch.setattr(devices_route, "_pending", store)
monkeypatch.setattr(devices_route, "PAIR_PER_ADDRESS", 1000)
started = [
client.post(f"{PREFIX}/pair").json()
for _ in range(devices_route.MAX_PENDING + 1)
]
assert store.count() == devices_route.MAX_PENDING
oldest, newest = started[0], started[-1]
assert (
client.get(
f"{PREFIX}/pair/{oldest['code']}", params={"secret": oldest["secret"]}
).status_code
== 404
)
assert (
client.get(
f"{PREFIX}/pair/{newest['code']}", params={"secret": newest["secret"]}
).status_code
== 200
)
def test_a_report_is_read_port_by_port(
client: TestClient,
superuser_token_headers: dict[str, str],
caplog: pytest.LogCaptureFixture,
) -> None:
"""What this version cannot read is left out and logged, not refused.
A shape or direction it does not know reads as unknown, and a named config
schema is put in place.
"""
_panels(client, superuser_token_headers, {"panels": [{"id": "lenient"}]})
device = _pair(client, superuser_token_headers, "lenient")
pin = {"type": "object", "properties": {"invert": {"type": "boolean"}}}
report = {
**REPORT,
"display": {"w": 466, "h": 466, "shape": "hexagon", "touch": False},
"config_schemas": {"pin": pin},
"ports": [
{"port": "gpio.4", "direction": "in", "dtype": "bool", "config": "pin"},
{"port": "gpio.5", "direction": "both", "dtype": "bool", "config": "pin"},
{"port": "Not A Port", "direction": "in", "dtype": "bool"},
{"port": "lidar", "direction": "in", "dtype": "pointcloud"},
],
"features": {"display": True, "camera": "imx415"},
}
with caplog.at_level("WARNING", logger="fluksio.api.routes.devices"):
put = client.put(f"{PREFIX}/lenient/report", headers=device, json=report)
assert put.status_code == 200, put.text
assert "2 ports reported, 3 left out" in put.json()["message"]
logged = " ".join(r.getMessage() for r in caplog.records)
for part in ("'Not A Port'", "'lidar'", "feature 'camera'"):
assert part in logged
listed = client.get(f"{PREFIX}/", headers=superuser_token_headers).json()
stored = next(d for d in listed["data"] if d["id"] == "lenient")["report"]
assert stored["display"]["shape"] == "unknown"
assert stored["ports"] == [
{"port": "gpio.4", "direction": "in", "dtype": "bool", "config": pin},
{"port": "gpio.5", "direction": "unknown", "dtype": "bool", "config": pin},
]
assert stored["features"] == {"display": True}
assert "config_schemas" not in stored
# Only no structure at all is refused.
for body in ([], "report", {"ports": []}):
refused = client.put(f"{PREFIX}/lenient/report", headers=device, json=body)
assert refused.status_code == 422, body
# Deleting the device drops the report and keeps the nonce moving.
store = app.state.device_store
nonce = store.nonce("lenient")
client.delete(f"{PREFIX}/lenient", headers=superuser_token_headers)
assert store.state("lenient").report is None
assert store.nonce("lenient") == nonce + 1
def test_a_busy_engine_rechecks_once_it_is_free(
client: TestClient,
superuser_token_headers: dict[str, str],
monkeypatch: pytest.MonkeyPatch,
) -> None:
from fluksio.flow.controller import RebuildBusy
_panels(client, superuser_token_headers, {"panels": [{"id": "busy"}]})
device = _pair(client, superuser_token_headers, "busy")
controller = app.state.flow_controller
waits: list[float | None] = []
async def recheck(wait: float | None = 5.0) -> None:
waits.append(wait)
if wait is not None:
raise RebuildBusy("busy")
monkeypatch.setattr(controller, "recheck", recheck)
put = client.put(f"{PREFIX}/busy/report", headers=device, json=REPORT)
assert put.status_code == 200, put.text
# The background recheck waits for the lock rather than giving up.
client.get(f"{PREFIX}/busy", headers=device)
assert waits == [5.0, None]
def test_a_device_outlives_the_account_that_paired_it(
client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
"""It acts as itself, not as whoever approved it."""
from tests.utils.user import user_authentication_headers
from tests.utils.utils import random_email, random_lower_string
email, password = random_email(), random_lower_string()
created = client.post(
f"{settings.API_V1_STR}/users/",
headers=superuser_token_headers,
json={"email": email, "password": password, "is_superuser": True},
)
assert created.status_code == 200, created.text
approver = user_authentication_headers(
client=client, email=email, password=password
)
_dashboard(client, superuser_token_headers, "outlived")
_panels(
client,
superuser_token_headers,
{"panels": [{"id": "orphan", "dashboards": ["outlived"]}]},
)
device = _pair(client, approver, "orphan")
gone = client.delete(
f"{settings.API_V1_STR}/users/{created.json()['id']}",
headers=superuser_token_headers,
)
assert gone.status_code == 200, gone.text
assert client.get(f"{PREFIX}/orphan", headers=device).status_code == 200
assert client.get(f"{DASHBOARDS}/outlived", headers=device).status_code == 200
# Still no more than a device: nothing a superuser's route serves.
assert client.get(f"{PREFIX}/", headers=device).status_code == 403
client.delete(f"{PREFIX}/orphan", headers=superuser_token_headers)
def test_a_stopped_flow_wires_no_ports(
client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
_panels(client, superuser_token_headers, {"panels": [{"id": "porch_io"}]})
_publish_flow(client, superuser_token_headers, "porch", [PORCH_NODE])
device = _pair(client, superuser_token_headers, "porch_io")
messages = f"{settings.API_V1_STR}/messages"
def io() -> list[str]:
read = client.get(f"{PREFIX}/porch_io", headers=device).json()
return [row["port"] for row in read["io"]]
assert io() == ["backlight", "gpio.4"]
client.post(f"{FLOWS}/porch/stop", headers=superuser_token_headers)
assert io() == []
assert (
client.post(
f"{messages}/porch.doorbell", headers=device, json={"value": True}
).status_code
== 403
)
client.post(f"{FLOWS}/porch/start", headers=superuser_token_headers)
assert io() == ["backlight", "gpio.4"]
client.delete(f"{FLOWS}/porch", headers=superuser_token_headers)