From 13d23f6bafd32793d8f5ef3dfd4b5468c44a768a Mon Sep 17 00:00:00 2001 From: Melvin Strobl Date: Fri, 16 Oct 2026 18:18:42 +0200 Subject: [PATCH] Restart safety: transient credential checks are 1013/503, not a refusal - A device credential that cannot be checked now (device store not running, nonces unreadable) raises DeviceUnavailable: 503 over HTTP, and the /flows/ws handshake and the tunnel stream accept and close with 1013. 1008 (a 403 before accepting) stays for a revoked or invalid one - Device files are written through a temporary file, so a read mid-save no longer looks like a deleted device - The first report turns Touch friendly on only for a device paired since pairings are noted (DeviceState.paired_at), so devices paired before keep the switch they were given - NodeDef reads `device`/`device_policy` from older clients as the worker label, for a release - Tests: transient vs revoked socket, the touch default, the old worker keys, and a restart with a device node and a node_health rule sending no push --- backend/fluksio/api/deps.py | 15 +++--- backend/fluksio/api/routes/devices.py | 12 ++++- backend/fluksio/api/routes/flows.py | 23 +++++++-- backend/fluksio/cloud/connector.py | 24 ++++++---- backend/fluksio/flow/devices.py | 25 +++++++++- backend/fluksio/flow/schemas.py | 21 +++++++++ backend/fluksio/main.py | 8 +++- backend/tests/api/routes/test_devices.py | 60 ++++++++++++++++++++++++ backend/tests/flow/test_device_node.py | 52 +++++++++++++++++++- backend/tests/flow/test_store.py | 11 +++++ docs/interface/dashboards.md | 3 +- 11 files changed, 228 insertions(+), 26 deletions(-) diff --git a/backend/fluksio/api/deps.py b/backend/fluksio/api/deps.py index 29c690d..0c061d8 100644 --- a/backend/fluksio/api/deps.py +++ b/backend/fluksio/api/deps.py @@ -19,7 +19,7 @@ from fluksio.flow import devices from fluksio.flow.artifacts import is_reference from fluksio.flow.controller import FlowController from fluksio.flow.dashboards import DashboardDef, DashboardStore -from fluksio.flow.devices import DeviceDef, DeviceStore, presence +from fluksio.flow.devices import DeviceDef, DeviceStore, DeviceUnavailable, presence from fluksio.flow.workers import PythonWorkerPool from fluksio.models import OAuthClient, TokenPayload, User, WorkerToken @@ -59,12 +59,15 @@ def _device_for(payload: dict[str, Any], app: Any) -> DeviceDef: The claim is still called ``panel``: tokens in circulation say it, and nothing pairs again for a rename. + + What cannot be checked right now raises ``DeviceUnavailable`` rather than + refusing: a device told its credential is refused forgets it. """ - store: DeviceStore | None = ( - getattr(app.state, "device_store", None) if app else None - ) + if app is None: + raise InvalidTokenError("Nothing to check a device credential against") + store: DeviceStore | None = getattr(app.state, "device_store", None) if store is None: - raise InvalidTokenError("The device store is not running") + raise DeviceUnavailable("The device store is not running") device = store.find(str(payload.get("panel", ""))) if device is None: raise InvalidTokenError("This device no longer exists") @@ -72,7 +75,7 @@ def _device_for(payload: dict[str, Any], app: Any) -> DeviceDef: try: nonce = store.nonce(device.id) except (OSError, ValueError) as exc: - raise InvalidTokenError(f"The device nonces are unreadable: {exc}") + raise DeviceUnavailable(f"The device nonces are unreadable: {exc}") if int(payload.get("pnc") or 0) != nonce: raise InvalidTokenError("This device was paired again") presence.seen(device.id) diff --git a/backend/fluksio/api/routes/devices.py b/backend/fluksio/api/routes/devices.py index b233a1d..9193c74 100644 --- a/backend/fluksio/api/routes/devices.py +++ b/backend/fluksio/api/routes/devices.py @@ -625,7 +625,8 @@ async def put_report(device_id: str, request: Request, store: DeviceStoreDep) -> The first report of a touch screen makes the device touch friendly, as the switch would have been set by hand. Later ones leave the setting to - whoever changed it since. + whoever changed it since, and so does the first report of a device paired + before pairings were noted: its switch was set by hand already. """ device: DeviceDef | None = getattr(request.state, "device", None) if device is None or device.id != device_id: @@ -650,7 +651,13 @@ async def put_report(device_id: str, request: Request, store: DeviceStoreDep) -> ), ) first = await run_in_threadpool(store.report, device_id, report) - if first and report.display and report.display.touch and not device.touch: + if ( + first + and report.display + and report.display.touch + and not device.touch + and (await run_in_threadpool(store.state, device_id)).paired_at + ): with suppress(StaleVersion, DeviceNotFound): await run_in_threadpool( store.write, @@ -724,6 +731,7 @@ def _adopt( ) entry.device_id = device_id _pending.save(code, entry) + store.paired(device_id) return f"Paired {entry.name or entry.device} with {device_id}" diff --git a/backend/fluksio/api/routes/flows.py b/backend/fluksio/api/routes/flows.py index 0785f8f..8eb6f96 100644 --- a/backend/fluksio/api/routes/flows.py +++ b/backend/fluksio/api/routes/flows.py @@ -1,6 +1,7 @@ """The flow API. Everything the editor can do is available here first.""" import asyncio +import logging import os import time from pathlib import Path @@ -68,6 +69,8 @@ from fluksio.flow.store import ( ) from fluksio.models import Flavor, Message, Run +logger = logging.getLogger(__name__) + router = APIRouter( prefix="/flows", tags=["flows"], dependencies=[Depends(get_current_user)] ) @@ -1221,15 +1224,25 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") - # every other socket and every request is sharing. def _authenticate() -> Any: with Session(engine) as session: - return user_from_token(session, token, app=websocket.app) + user = user_from_token(session, token, app=websocket.app) + only = panel_scope(token, websocket.app) if user is not None else None + # A device is online for as long as this socket is open. + device = device_of(token, websocket.app) if only is not None else "" + return user, only, device - user = await run_in_threadpool(_authenticate) + try: + user, only, device = await run_in_threadpool(_authenticate) + except Exception as exc: + # Not a refusal: the database or the device store could not say. A + # device takes 1008, or a 403 before accepting, as its credential + # being gone and checks it, so this one is told to try again later. + logger.warning("Could not check a socket's credential: %s", exc) + await websocket.accept() + await websocket.close(code=1013) + return if user is None: await websocket.close(code=1008) return - only = panel_scope(token, websocket.app) - # A device is online for as long as this socket is open. - device = device_of(token, websocket.app) if only is not None else "" # Nothing until a client says it is drawing something: bytes are what this # socket cannot afford to send speculatively. wanted: set[str] = set() diff --git a/backend/fluksio/cloud/connector.py b/backend/fluksio/cloud/connector.py index 4e78e82..a9e96e8 100644 --- a/backend/fluksio/cloud/connector.py +++ b/backend/fluksio/cloud/connector.py @@ -663,15 +663,21 @@ class CloudConnector: if not path.startswith("/api/v1/") or not path.endswith("/flows/ws"): await socket.send(_dump({"op": "ws_close", "id": stream_id, "code": 1008})) return - with Session(engine) as session: - if user_from_token(session, token, app=self._app) is None: - await socket.send( - _dump({"op": "ws_close", "id": stream_id, "code": 1008}) - ) - return - only = panel_scope(token, self._app) - # A remote device is online for as long as this stream is open. - device = device_of(token, self._app) if only is not None else "" + try: + with Session(engine) as session: + user = user_from_token(session, token, app=self._app) + only = panel_scope(token, self._app) if user is not None else None + # A remote device is online for as long as this stream is open. + device = device_of(token, self._app) if only is not None else "" + except Exception as exc: + # Could not tell, which is not a refusal: 1008 is what makes a + # device check whether it is still paired. + logger.warning("Could not check a stream's credential: %s", exc) + await socket.send(_dump({"op": "ws_close", "id": stream_id, "code": 1013})) + return + if user is None: + await socket.send(_dump({"op": "ws_close", "id": stream_id, "code": 1008})) + return await socket.send(_dump({"op": "ws_open_ok", "id": stream_id})) diff --git a/backend/fluksio/flow/devices.py b/backend/fluksio/flow/devices.py index b39cdf4..f61911e 100644 --- a/backend/fluksio/flow/devices.py +++ b/backend/fluksio/flow/devices.py @@ -159,6 +159,9 @@ class DeviceState(BaseModel): #: What the device says it has, written by the device itself. report: dict[str, Any] | None = None reported_at: datetime | None = None + #: When a pairing here last minted the device a credential. None for one + #: paired before this was kept, whose settings were all chosen by hand. + paired_at: datetime | None = None _STATES = TypeAdapter(dict[str, DeviceState]) @@ -231,6 +234,15 @@ class DeviceExists(ValueError): self.id = device_id +class DeviceUnavailable(RuntimeError): + """Whether a device credential stands cannot be told right now. + + Not a refusal: a device told its credential is refused pairs again, so + a store that is not running or a nonce file that does not read answers + "try again later" instead. + """ + + class DeviceStore: """Device definitions in the flow repository, their nonces beside it. @@ -260,7 +272,11 @@ class DeviceStore: def _write(self, device: DeviceDef) -> None: path = self._file(device.id) path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(device.model_dump_json(indent=2) + "\n") + # Through a temporary file: a credential checked against a half + # written one would read as a deleted device and be refused. + tmp = path.with_name(path.name + ".tmp") + tmp.write_text(device.model_dump_json(indent=2) + "\n") + os.replace(tmp, path) # -- what the published flows wire ------------------------------------ @@ -433,6 +449,13 @@ class DeviceStore: self._write_states(states) return state.nonce + def paired(self, device_id: str) -> None: + """Note that a pairing here just minted this device a credential.""" + with self._state_lock: + states = self._read_states() + states.setdefault(device_id, DeviceState()).paired_at = datetime.now(UTC) + self._write_states(states) + def report(self, device_id: str, report: DeviceReport) -> bool: """Keep what the device says it has, in place of what it said before. diff --git a/backend/fluksio/flow/schemas.py b/backend/fluksio/flow/schemas.py index 2962f12..c63f4f3 100644 --- a/backend/fluksio/flow/schemas.py +++ b/backend/fluksio/flow/schemas.py @@ -226,6 +226,27 @@ class NodeDef(BaseModel): ), ) + @model_validator(mode="before") + @classmethod + def _old_worker_keys(cls, data: Any) -> Any: + """Read ``device`` and ``device_policy`` as the worker label they were. + + What a ``fluksio sync`` or SDK from before the rename to ``worker`` + sends, which would otherwise be dropped without a word. A new key + sent beside an old one wins. + + ponytail: drop after a release, once clients have caught up. + """ + if not isinstance(data, dict) or not ( + "device" in data or "device_policy" in data + ): + return data + data = dict(data) + for old, new in (("device", "worker"), ("device_policy", "worker_policy")): + if old in data: + data.setdefault(new, data.pop(old)) + return data + @field_validator("id") @classmethod def _check_id(cls, value: str) -> str: diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py index 1cd7a00..0b1e44c 100644 --- a/backend/fluksio/main.py +++ b/backend/fluksio/main.py @@ -34,7 +34,7 @@ from fluksio.flow.alerts import AlertManager from fluksio.flow.artifacts import ArtifactStore, VolatileStore from fluksio.flow.controller import FlowController, RebuildBusy from fluksio.flow.dashboards import DashboardStore -from fluksio.flow.devices import DeviceStore +from fluksio.flow.devices import DeviceStore, DeviceUnavailable from fluksio.flow.events import event_bus from fluksio.flow.executor import ExecutionService from fluksio.flow.metrics import MetricsCollector @@ -538,6 +538,12 @@ async def invalid_name(request: Request, exc: InvalidName) -> JSONResponse: # n return JSONResponse(status_code=400, content={"detail": str(exc)}) +@app.exception_handler(DeviceUnavailable) +async def device_unavailable(request: Request, exc: DeviceUnavailable) -> JSONResponse: # noqa: ARG001 + """A device credential that cannot be checked now is not a refused one.""" + return JSONResponse(status_code=503, content={"detail": str(exc)}) + + @app.exception_handler(RebuildBusy) async def rebuild_busy(request: Request, exc: RebuildBusy) -> JSONResponse: # noqa: ARG001 """Every route that deploys something answers a wedged rebuild the same way. diff --git a/backend/tests/api/routes/test_devices.py b/backend/tests/api/routes/test_devices.py index 4c97d72..e99c937 100644 --- a/backend/tests/api/routes/test_devices.py +++ b/backend/tests/api/routes/test_devices.py @@ -2,7 +2,9 @@ import threading +import pytest from fastapi.testclient import TestClient +from starlette.websockets import WebSocketDisconnect from fluksio.api.routes import devices as devices_route from fluksio.cloud import config as cloud_config @@ -1053,6 +1055,64 @@ def test_unpairing_a_screen_leaves_its_panel_standing( assert client.post(f"{PREFIX}/kitchen/unpair").status_code == 401 +def test_a_device_is_refused_only_when_its_credential_is_gone( + client: TestClient, + superuser_token_headers: dict[str, str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + """ "Cannot say right now" is no refusal: a refused device pairs again.""" + _panels(client, superuser_token_headers, {"panels": [{"id": "hallway"}]}) + device = _pair(client, superuser_token_headers, "hallway") + socket = f"{settings.API_V1_STR}/flows/ws" + store = app.state.device_store + + def unreadable(_: str) -> int: + raise ValueError("torn") + + # The nonces do not read: the socket is accepted and told to try again + # later, and the device's own read answers 503. + monkeypatch.setattr(store, "nonce", unreadable) + with ( + client.websocket_connect(socket, headers=device) as ws, + pytest.raises(WebSocketDisconnect) as later, + ): + ws.receive_json() + assert later.value.code == 1013 + assert client.get(f"{PREFIX}/hallway", headers=device).status_code == 503 + monkeypatch.undo() + with client.websocket_connect(socket, headers=device) as ws: + assert "values" in ws.receive_json() + + # Unpaired, it is refused before it is accepted, a 403 on the wire. + client.post(f"{PREFIX}/hallway/unpair", headers=superuser_token_headers) + with ( + pytest.raises(WebSocketDisconnect) as refused, + client.websocket_connect(socket, headers=device), + ): + pass + assert refused.value.code == 1008 + assert client.get(f"{PREFIX}/hallway", headers=device).status_code == 401 + + +def test_only_a_device_paired_here_turns_touch_on_with_its_report( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + """One paired before reports for the first time with an update, its + switch set by hand: a knob left off on purpose stays off.""" + _panels(client, superuser_token_headers, {"panels": [{"id": "knob_old"}]}) + device = _pair(client, superuser_token_headers, "knob_old") + store = app.state.device_store + with store._state_lock: + states = store._read_states() + states["knob_old"].paired_at = None + store._write_states(states) + + put = client.put(f"{PREFIX}/knob_old/report", headers=device, json=REPORT) + assert put.status_code == 200, put.text + read = client.get(f"{PREFIX}/knob_old", headers=device).json() + assert read["touch"] is False + + class _FakeRedis: """Just enough Redis for the pending-code store: strings and one zset. diff --git a/backend/tests/flow/test_device_node.py b/backend/tests/flow/test_device_node.py index 5d8ca19..b44a20d 100644 --- a/backend/tests/flow/test_device_node.py +++ b/backend/tests/flow/test_device_node.py @@ -3,8 +3,10 @@ import asyncio from pathlib import Path -from fluksio.flow.controller import FlowController +from fluksio.flow.alerts import Alert, AlertManager, AlertsConfig, Channel, Rule +from fluksio.flow.controller import HEALTH_FLOOR, FlowController from fluksio.flow.devices import presence +from fluksio.flow.events import EventBus from fluksio.flow.messages import DType, MessageSpec from fluksio.flow.schemas import FlowDef, FlowInput, NodeDef from fluksio.flow.store import FlowStore @@ -115,3 +117,51 @@ def test_an_offline_device_is_a_warning_not_a_fault(tmp_path: Path) -> None: ["hall.panel"], [("node_degraded", True)], ] + + +def test_a_device_away_after_a_restart_alerts_nobody(tmp_path: Path) -> None: + """No push, and degraded on screen only once the floor has passed. + + A restart starts every device node without its device, which comes back + within seconds, and a rule on every node's health is what sends phones a + push. + """ + controller = _controller(tmp_path) + events: list[dict] = [] + original = controller._publish + controller._publish = lambda event: (events.append(event), original(event)) # type: ignore[method-assign] + sent: list[Alert] = [] + alerts = AlertManager( + EventBus(), + AlertsConfig( + channels=[Channel(name="phone", kind="ntfy", config={"topic": "t"})], + rules=[Rule(events=["node_health"], channels=["phone"])], + ), + ) + + async def capture(channel: Channel, alert: Alert) -> None: + sent.append(alert) + + alerts.send = capture # type: ignore[method-assign] + + async def scenario() -> list[list[str]]: + await controller.reload() # the restart: the device is not back yet + surfaced = [[e.id for e in controller.unhealthy_nodes(health="degraded")]] + presence.opened("wall") + presence.closed("wall") # and it drops again + surfaced.append([e.id for e in controller.unhealthy_nodes(health="degraded")]) + controller.loaded["hall.panel"].health_since -= HEALTH_FLOOR + surfaced.append([e.id for e in controller.unhealthy_nodes(health="degraded")]) + for event in events: + await alerts.handle(event) + await controller._teardown() + return surfaced + + assert asyncio.run(scenario()) == [[], [], ["hall.panel"]] + assert [e["health"] for e in events if e["type"] == "node_health"] == [ + "degraded", + "ok", + "degraded", + ] + assert sent == [] + presence.forget("wall") diff --git a/backend/tests/flow/test_store.py b/backend/tests/flow/test_store.py index 6fee24d..f57f162 100644 --- a/backend/tests/flow/test_store.py +++ b/backend/tests/flow/test_store.py @@ -193,3 +193,14 @@ def test_opening_the_store_renames_a_nodes_device_to_worker(store: FlowStore): "prefer", ) assert published.version == 2 + + +def test_a_client_from_before_the_rename_still_labels_its_workers(): + """An older `fluksio sync` or SDK sends `device`, read as `worker` for a release.""" + node = NodeDef.model_validate( + {"id": "fit", "device": "gpu", "device_policy": "prefer"} + ) + assert (node.worker, node.worker_policy) == ("gpu", "prefer") + # Beside the new key, the new key wins. + both = NodeDef.model_validate({"id": "fit", "device": "cpu", "worker": "gpu"}) + assert both.worker == "gpu" diff --git a/docs/interface/dashboards.md b/docs/interface/dashboards.md index bb16a8b..fc23451 100644 --- a/docs/interface/dashboards.md +++ b/docs/interface/dashboards.md @@ -565,7 +565,8 @@ sends only while somebody wants it. The device's page shows the report under **Capabilities**, with each port's binding or **Use in flow** for it. A reported screen becomes a canvas preset in the dashboard settings, named after the device, and the first report of a -touch screen turns **Touch friendly** on. +touch screen turns **Touch friendly** on. A device paired before this version +keeps the switch as it was set. ### Before device nodes