diff --git a/backend/fluksio/api/deps.py b/backend/fluksio/api/deps.py index 7b06c92..7a266d6 100644 --- a/backend/fluksio/api/deps.py +++ b/backend/fluksio/api/deps.py @@ -177,10 +177,12 @@ def _device_digests(device: DeviceDef, request: Request) -> set[str]: controller: FlowController | None = getattr( request.app.state, "flow_controller", None ) - if controller is None: - return set() - names = _device_messages(device, request) - if not names: + return digests_of(controller, _device_messages(device, request)) + + +def digests_of(controller: FlowController | None, names: set[str]) -> set[str]: + """The artifacts these messages point at now, and the last few before.""" + if controller is None or not names: return set() found: set[str] = set() for value in controller.state.get_present(sorted(names)).values(): @@ -191,6 +193,19 @@ def _device_digests(device: DeviceDef, request: Request) -> set[str]: return found +def hidden_from(user: User, app: Any, *, device: bool = False) -> frozenset[str]: + """The camera messages ``user`` may not see (``DeviceDef.viewers``). + + None for a superuser, and none for a device credential, which its + dashboards bound instead: a superuser assigned them, camera tiles + included. ``device`` says the caller is one. + """ + if device or user.is_superuser: + return frozenset() + store: DeviceStore | None = getattr(app.state, "device_store", None) + return store.restricted() if store is not None else frozenset() + + def _device_may(payload: dict[str, Any], request: Request) -> None: """Refuse anything a wall panel has no business asking for. @@ -520,6 +535,18 @@ def get_current_user(request: Request, session: SessionDep, token: TokenDep) -> CurrentUser = Annotated[User, Depends(get_current_user)] +def get_hidden(request: Request, user: CurrentUser) -> frozenset[str]: + """The camera messages the caller of this request may not see.""" + device = getattr(request.state, "device", None) is not None + return hidden_from(user, request.app, device=device) + + +HiddenDep = Annotated[frozenset[str], Depends(get_hidden)] + +#: Why a camera's values, bytes or live view are refused. +CAMERA_IS_SUPERUSERS = "Only a superuser may see this camera" + + def get_flow_controller(request: Request) -> FlowController: controller: FlowController | None = getattr( request.app.state, "flow_controller", None diff --git a/backend/fluksio/api/routes/artifacts.py b/backend/fluksio/api/routes/artifacts.py index 48a1595..2667259 100644 --- a/backend/fluksio/api/routes/artifacts.py +++ b/backend/fluksio/api/routes/artifacts.py @@ -18,8 +18,11 @@ from sqlmodel import Session from starlette.concurrency import run_in_threadpool from fluksio.api.deps import ( + CAMERA_IS_SUPERUSERS, FlowControllerDep, + digests_of, get_current_active_superuser, + hidden_from, user_from_token, worker_claims, ) @@ -63,6 +66,8 @@ def artifact_caller(request: Request) -> str: user = user_from_token(session, token, request) if user is None: raise HTTPException(status_code=401, detail="Not authenticated") + # For the camera check a fetch is held to, which a worker is not. + request.state.person = user return user.email @@ -103,26 +108,30 @@ async def put_artifact( watching now, which the oldest of falls out of memory rather than being kept. A worker on another host or a device publishing a camera sends this. It is read into memory, where the ring holds it, and never touches the - disk. + disk, so it is held to ``MAX_FRAME_BYTES`` rather than the store's cap. """ store = _store(request) - cap = settings.MAX_ARTIFACT_BYTES + ring = store.volatile if volatile else None + cap = settings.MAX_FRAME_BYTES if ring is not None else settings.MAX_ARTIFACT_BYTES declared = request.headers.get("content-length") if cap and declared and declared.isdigit() and int(declared) > cap: raise HTTPException( status_code=413, detail=f"An artifact may be at most {cap} bytes" ) - ring = store.volatile if volatile else None if ring is not None: - body = bytearray() + # The chunks as they came, joined once: the frame is copied into + # memory one time, never past the cap. + chunks: list[bytes] = [] + size = 0 async for chunk in request.stream(): - body += chunk - if cap and len(body) > cap: + size += len(chunk) + if cap and size > cap: raise HTTPException( status_code=413, detail=f"An artifact may be at most {cap} bytes", ) - return await run_in_threadpool(ring.put, [bytes(body)], name, media_type) + chunks.append(chunk) + return await run_in_threadpool(ring.put, chunks, name, media_type) handle = tempfile.NamedTemporaryFile(dir=store.root, delete=False) written = 0 try: @@ -158,6 +167,15 @@ def get_artifact( scrubbing through a file asks for them. """ store = _store(request) + user = getattr(request.state, "person", None) + if user is not None: + # What a camera only superusers may see points at, now or in its last + # few frames. A worker's node reads anything, as node code can. + device = getattr(request.state, "device", None) is not None + hidden = hidden_from(user, request.app, device=device) + controller = getattr(request.app.state, "flow_controller", None) + if hidden and digest in digests_of(controller, set(hidden)): + raise HTTPException(status_code=403, detail=CAMERA_IS_SUPERUSERS) path = store.path(digest) # A live frame is held in memory, not on disk. frame = store.volatile.get(digest) if path is None and store.volatile else None diff --git a/backend/fluksio/api/routes/devices.py b/backend/fluksio/api/routes/devices.py index be0548b..9a854c6 100644 --- a/backend/fluksio/api/routes/devices.py +++ b/backend/fluksio/api/routes/devices.py @@ -43,12 +43,14 @@ from datetime import datetime, timedelta from typing import Any, cast import httpx +import orjson import redis from fastapi import APIRouter, Depends, HTTPException, Request from fastapi.concurrency import run_in_threadpool from pydantic import BaseModel, Field, ValidationError, field_validator from fluksio.api.deps import ( + CAMERA_IS_SUPERUSERS, CurrentUser, DeviceStoreDep, get_current_active_superuser, @@ -120,6 +122,10 @@ HELLO_LENGTH = 64 #: The most a capability report may weigh. A panel with every header pin #: reported is a few kB. REPORT_MAX = 16 * 1024 +#: The most it may weigh as kept, with each named ``config_schemas`` entry put +#: in place: thirty pins naming one schema send it once and keep it thirty +#: times. The P4 panel's is about 7 kB. +REPORT_STORED_MAX = 64 * 1024 class _Pending(BaseModel): @@ -690,9 +696,13 @@ async def read_ice(device_id: str, store: DeviceStoreDep, user: CurrentUser) -> STUN and TURN servers, the TURN ones with credentials of the caller's own that expire. Empty when none are set, which is enough on one network. + Only for whoever may see the device's camera (``DeviceDef.viewers``). """ - if await run_in_threadpool(store.find, device_id) is None: + device = await run_in_threadpool(store.find, device_id) + if device is None: raise _not_found(device_id) + if not user.is_superuser and device.viewers != "members": + raise HTTPException(status_code=403, detail=CAMERA_IS_SUPERUSERS) servers = [IceServer(**server) for server in ice_servers(str(user.id))] return IceServers(ice_servers=servers, ttl=ICE_TTL_S) @@ -745,6 +755,12 @@ async def put_report(device_id: str, request: Request, store: DeviceStoreDep) -> include_url=False, include_input=False, include_context=False ), ) + if len(orjson.dumps(report.stored())) > REPORT_STORED_MAX: + raise HTTPException( + status_code=413, + detail=f"A report is at most {REPORT_STORED_MAX // 1024} KB once " + "its config_schemas are put in place", + ) for part in left_out: logger.warning("Device '%s' reported %s, left out", device_id, part) first = await run_in_threadpool(store.report, device_id, report) diff --git a/backend/fluksio/api/routes/flows.py b/backend/fluksio/api/routes/flows.py index 7cf7571..775a6ff 100644 --- a/backend/fluksio/api/routes/flows.py +++ b/backend/fluksio/api/routes/flows.py @@ -23,14 +23,17 @@ from pydantic import BaseModel from sqlmodel import Session, col, select from fluksio.api.deps import ( + CAMERA_IS_SUPERUSERS, CODE_IS_SUPERUSERS, CurrentUser, FlowControllerDep, + HiddenDep, SessionDep, TokenDep, decode_token, get_code_author, get_current_user, + hidden_from, user_from_token, ) from fluksio.api.routes.messages import port_source @@ -279,10 +282,15 @@ def _require_enabled(controller: FlowController, name: str) -> None: ) -def _flow_state(controller: FlowController, name: str) -> FlowStatePublic: +def _flow_state( + controller: FlowController, name: str, hidden: frozenset[str] +) -> FlowStatePublic: + """The flow's values, but those of a camera the caller may not see.""" return FlowStatePublic( values={ - key: MessageValue(**value) for key, value in controller.values(name).items() + key: MessageValue(**value) + for key, value in controller.values(name).items() + if key not in hidden }, nodes=controller.node_statuses(name), ) @@ -892,6 +900,7 @@ async def run_flow( request: Request, controller: FlowControllerDep, user: CurrentUser, + hidden: HiddenDep, ) -> Any: """Run every node of a flow once. @@ -919,14 +928,14 @@ async def run_flow( ) except RunRejected as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc - return _flow_state(controller, name) + return _flow_state(controller, name, hidden) inputs = {qualify(name, key): value for key, value in body.inputs.items()} if controller.store.has_draft(name): await run_in_threadpool(controller.run_preview, name, inputs) else: await run_in_threadpool(controller.run_flow, name, inputs) - return _flow_state(controller, name) + return _flow_state(controller, name, hidden) @router.post("/{name}/nodes/{node_id}/trigger", response_model=FlowStatePublic) @@ -935,6 +944,7 @@ async def trigger_node( node_id: str, body: TriggerRequest, controller: FlowControllerDep, + hidden: HiddenDep, ) -> Any: """Feed values into a single node.""" _require_enabled(controller, name) @@ -950,7 +960,7 @@ async def trigger_node( # The canvas already has the failure from the bus; the person who # clicked gets to read it too, instead of a stack trace in the log. raise HTTPException(status_code=400, detail=error) - return _flow_state(controller, name) + return _flow_state(controller, name, hidden) @router.post("/{name}/nodes/{node_id}/cancel", response_model=Message) @@ -995,10 +1005,10 @@ def acknowledge_node_error( @router.get("/{name}/state", response_model=FlowStatePublic) -def read_flow_state(name: str, controller: FlowControllerDep) -> Any: +def read_flow_state(name: str, controller: FlowControllerDep, hidden: HiddenDep) -> Any: """The last value seen on every message of this flow.""" _read_flow(controller, name) - return _flow_state(controller, name) + return _flow_state(controller, name, hidden) @router.get("/{name}/history/{message}", response_model=MessageHistory) @@ -1006,6 +1016,7 @@ def read_message_history( name: str, message: str, controller: FlowControllerDep, + hidden: HiddenDep, ) -> Any: """The recent values of one message, for plotting. @@ -1014,6 +1025,8 @@ def read_message_history( """ _read_flow(controller, name) key = qualify(name, message) + if key in hidden: + raise HTTPException(status_code=403, detail=CAMERA_IS_SUPERUSERS) return MessageHistory( message=key, numeric=as_number(controller.state.get(key)) is not None, @@ -1034,6 +1047,7 @@ def snapshot_payload( only: set[str] | None = None, device: str = "", uplink: str = "", + hidden: frozenset[str] = frozenset(), ) -> dict[str, Any]: """Everything a client needs to catch up, sent the moment it connects. @@ -1052,6 +1066,8 @@ def snapshot_payload( A device's socket also says which of its ports have ``demand`` now, and which ``uplink`` its frames take: ``binary`` on its own socket, ``http`` through the portal, whose tunnel carries text only. + + ``hidden`` are the camera messages a person may not see, left out. """ if only is not None: # The same keys, emptied rather than dropped: a screen already hanging @@ -1071,9 +1087,12 @@ def snapshot_payload( payload["demand"] = demand.ports(device) payload["uplink"] = uplink return payload + values = controller.values() return { "type": "snapshot", - "values": controller.values(), + "values": ( + {k: v for k, v in values.items() if k not in hidden} if hidden else values + ), "nodes": [s.model_dump() for s in controller.node_statuses()], "issues": [i.model_dump() for i in controller.issues], "paused": controller.paused_flows(), @@ -1096,6 +1115,41 @@ def device_of(token: str, app: Any) -> str: return "" +class Credential: + """Whether a device socket's credential still stands, asked per frame. + + Checked in full only once what it is checked against has moved: the + device files, which a device's save, deletion or a revert commits to the + flow store, and the nonces in ``devices.json``, which an unpair bumps. + Otherwise a frame costs a stat of that file. + """ + + def __init__(self, token: str, app: Any, device_id: str) -> None: + self.token = token + self.app = app + self.device = device_id + self._seen: tuple[int, int] | None = None + + def _key(self) -> tuple[int, int] | None: + devices: DeviceStore | None = getattr(self.app.state, "device_store", None) + if devices is None: + return None + try: + return devices.flows.revision, devices.nonce(self.device) + except (OSError, ValueError): + return None + + async def revoked(self) -> bool: + key = self._key() + if key is not None and key == self._seen: + return False + if await run_in_threadpool(revoked, self.token, self.app): + return True + # The key read before the check, so a change meanwhile is checked. + self._seen = key + return False + + def revoked(token: str, app: Any) -> bool: """Whether a device's credential has stopped standing since its handshake. @@ -1155,6 +1209,21 @@ def event_for_panel(event: dict[str, Any], only: set[str]) -> bool: return event.get("type") == "message_value" and str(event.get("name") or "") in only +def carried( + event: dict[str, Any], only: set[str] | None, hidden: frozenset[str] +) -> bool: + """Whether a socket carries this event on the bus. + + A panel's what it draws (``event_for_panel``), a person's everything but + the values of the cameras ``hidden`` from them. + """ + if only is not None: + return event_for_panel(event, only) + return not ( + hidden and event.get("type") == "message_value" and event.get("name") in hidden + ) + + # How many events one frame may carry. A ceiling rather than a target: the # batch is whatever the bus happens to hold, and a client that has been away # should not be handed the whole queue in one message. @@ -1166,14 +1235,19 @@ MAX_FRAME_EVENTS = 64 MAX_MEDIA_NAMES = 8 -def wanted_names(frame: str, only: set[str] | None) -> set[str] | None: +def wanted_names( + frame: str, only: set[str] | None, hidden: frozenset[str] = frozenset() +) -> set[str] | None: """What this client is asking to be sent bytes for, or None if it said something else. A client asks by name — the tiles it is drawing — and is answered only for - the names a panel credential would have been allowed anyway. Bytes are the - expensive thing on this socket, so nothing is pushed until something says - it is looking at it. + the names a panel credential would have been allowed anyway, and none of + the ``hidden`` cameras. Bytes are the expensive thing on this socket, so + nothing is pushed until something says it is looking at it. + + Past the cap the names it gave last are kept: a client lists the tile that + came on screen most recently last, so one scrolled back to keeps working. """ try: payload = orjson.loads(frame) @@ -1186,15 +1260,16 @@ def wanted_names(frame: str, only: set[str] | None) -> set[str] | None: return set() asked = {str(name) for name in names} if len(asked) > MAX_MEDIA_NAMES: - kept = list(dict.fromkeys(str(name) for name in names))[:MAX_MEDIA_NAMES] + kept = list(dict.fromkeys(str(name) for name in names))[-MAX_MEDIA_NAMES:] logger.warning( "A socket asked for live media on %d messages; only %d are served, " - "the first it named: %s", + "the last it named: %s", len(asked), MAX_MEDIA_NAMES, ", ".join(kept), ) asked = set(kept) + asked -= hidden return asked if only is None else asked & only @@ -1271,28 +1346,47 @@ def parse_frame(data: bytes) -> tuple[dict[str, Any], bytes] | None: return header, data[4 + length :] +#: Why a device's frame was dropped, by the key it is logged under once per +#: socket. Fixed, so nothing the device chose reaches the log. +FRAME_REFUSALS = { + "not_a_frame": "not a frame", + "not_a_device": "only a device sends frames", + "no_engine": "the engine is not running", + "not_wired": "it names no message one of the device's ports publishes", + "not_image": "the port it names is not wired as an image", + "media_type": "its media type is not an image type", + "too_large": "it is larger than MAX_FRAME_BYTES", + "undeclared": "no flow declares its message", + "refused": "its message refused the value", +} + + def ingest_frame(app: Any, device: DeviceDef, data: bytes) -> str: """Store a frame a device sent on its socket and publish it. - Returns why it was refused, or "" once it is published. + Returns the key in ``FRAME_REFUSALS`` it was refused for, or "" once it + is published. The same frame over HTTP is two requests, each authenticated: the upload, then the publish. Here the socket was authenticated once, at its handshake, and the frame is held and published in one step. Only a message one of the device's own ``in`` ports is wired to, as an ``image``, and the value is credited to the device node holding that port, as an - HTTP publish is. + HTTP publish is. The frame is named after that message. """ + cap = settings.MAX_FRAME_BYTES + if cap and len(data) > cap: + return "too_large" parsed = parse_frame(data) if parsed is None or parsed[0].get("type") != "frame": - return "not a frame" + return "not_a_frame" header, payload = parsed name = str(header.get("name") or "") devices: DeviceStore | None = getattr(app.state, "device_store", None) store: Any = getattr(app.state, "artifact_store", None) controller: FlowController | None = getattr(app.state, "flow_controller", None) if devices is None or store is None or controller is None: - return "the engine is not running" + return "no_engine" binding = next( ( b @@ -1302,24 +1396,19 @@ def ingest_frame(app: Any, device: DeviceDef, data: bytes) -> str: None, ) if binding is None: - return f"{name!r} is not a message one of its ports publishes" + return "not_wired" if binding.dtype != "image": - return f"its port {binding.port!r} is wired as {binding.dtype}, not image" + return "not_image" media_type = str(header.get("media_type") or "") if not _IMAGE_TYPE.match(media_type): - return f"{media_type!r} is not an image type" - cap = settings.MAX_ARTIFACT_BYTES - if cap and len(payload) > cap: - return f"a frame may be at most {cap} bytes" - reference = store.put( - [payload], name=binding.port, media_type=media_type, volatile=True - ) + return "media_type" + reference = store.put([payload], name=name, media_type=media_type, volatile=True) try: controller.publish_message(name, reference, port_source(devices, device, name)) except KeyError: - return f"no flow declares {name!r}" - except TypeError as exc: - return str(exc) + return "undeclared" + except TypeError: + return "refused" return "" @@ -1407,6 +1496,11 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") - viewer = object() # Each reason a frame was refused for, logged once per socket. refused: set[str] = set() + # The cameras a person may not see. Looked at again with each batch, so + # a device switched to members, or back, takes effect on open sockets. + hidden = hidden_from(user, websocket.app, device=only is not None) + # A device's credential, checked again per frame once it may have moved. + credential = Credential(token, websocket.app, device) if device else None await websocket.accept() @@ -1420,11 +1514,36 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") - if controller is None: return [] payload = await run_in_threadpool( - snapshot_payload, controller, only, device, "binary" + snapshot_payload, controller, only, device, "binary", hidden ) await websocket.send_text(orjson.dumps(payload).decode()) return list(payload.get("demand", [])) + async def catch_up() -> bool: + """Rescope a panel and send the snapshot again, demand included. + + False once a device whose credential stopped standing is closed with + 1008, which sends it to check: otherwise its frames would still be + taken here. + """ + nonlocal only, device_def, wanted + if only is not None: + if device and await run_in_threadpool(revoked, token, websocket.app): + await websocket.close(code=1008) + return False + # The scope was resolved once, at the handshake. A panel pointed + # at another dashboard would otherwise fetch the new document and + # then draw tiles nothing ever updates. ``or set()`` because a + # panel that was deleted resolves to None, the same as a person's + # token — and that would widen this socket to everything on the + # bus. + only = panel_scope(token, websocket.app) or set() + device_def = _device_def(websocket.app, device) + wanted &= only + demand.want(viewer, wanted) + await send_snapshot() + return True + told = await send_snapshot() async with event_bus.subscribe() as queue: @@ -1477,23 +1596,27 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") - if isinstance(frame, bytes): # A camera's frame. Refused ones are dropped and the # socket stays open: one bad frame is not a reason to - # take the device's live values away. + # take the device's live values away. A device that + # was unpaired meanwhile is closed instead. + if credential is not None and await credential.revoked(): + await websocket.close(code=1008) + return reason = ( await run_in_threadpool( ingest_frame, websocket.app, device_def, frame ) if device_def is not None - else "only a device sends frames" + else "not_a_device" ) if reason and reason not in refused: refused.add(reason) logger.warning( "Dropped a frame from %s: %s", device or "a person's socket", - reason, + FRAME_REFUSALS[reason], ) else: - asked = wanted_names(frame, only) + asked = wanted_names(frame, only, hidden) if asked is not None: wanted = asked demand.want(viewer, wanted) @@ -1506,6 +1629,7 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") - devices=getattr( websocket.app.state, "device_store", None ), + superuser=user.is_superuser, ) receiver = asyncio.create_task(_receive(websocket)) if sender not in done: @@ -1523,48 +1647,46 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") - except asyncio.QueueEmpty: break + # A panel this far behind lost events the bus dropped for it: + # its demand, a dashboard published or its own unpairing. It + # catches up once this batch is out, as after a dashboard. + resync = only is not None and event_bus.dropped(queue) + if only is None: + fresh = hidden_from(user, websocket.app) + if fresh != hidden: + # A camera was opened or closed to members: the + # snapshot gives back what may now be seen. + hidden = fresh + wanted -= hidden + demand.want(viewer, wanted) + resync = True + out: list[dict[str, Any]] = [] for event in batch: - if event.get("type") == "device_demand": + kind = event.get("type") + if kind == "device_demand": # For the device it names, and for nobody else. if device and event.get("device") == device: await websocket.send_text( demand_frame(list(event.get("ports") or [])) ) continue - if only is not None and event.get("type") == "dashboard_changed": - # An unpaired or deleted device is closed with 1008, - # which sends it to check its credential: otherwise - # its frames would still be taken here. - if device and await run_in_threadpool( - revoked, token, websocket.app - ): - await websocket.close(code=1008) - return - # The scope was resolved once, at the handshake. A - # panel pointed at another dashboard would otherwise - # fetch the new document and then draw tiles nothing - # ever updates. ``or set()`` because a panel that was - # deleted resolves to None, the same as a person's - # token — and that would widen this socket to - # everything on the bus. - only = panel_scope(token, websocket.app) or set() - device_def = _device_def(websocket.app, device) - wanted &= only - demand.want(viewer, wanted) + if only is not None and kind == "dashboard_changed": if out: await _send(websocket, out) out = [] - await send_snapshot() - if only is not None and not event_for_panel(event, only): - continue - out.append(event) + if not await catch_up(): + return + if carried(event, only, hidden): + out.append(event) if out: # The bytes first: a tile that has the frame when the # value lands draws it in one pass rather than two. for frame in media_frames(out, wanted, store): await websocket.send_bytes(frame) await _send(websocket, out) + if resync and not await catch_up(): + return except (WebSocketDisconnect, RuntimeError): # A peer that goes away mid-send takes the RuntimeError route # ("websocket.send after websocket.close") rather than the clean diff --git a/backend/fluksio/api/routes/messages.py b/backend/fluksio/api/routes/messages.py index 93ed718..4a2e4e5 100644 --- a/backend/fluksio/api/routes/messages.py +++ b/backend/fluksio/api/routes/messages.py @@ -12,7 +12,12 @@ from fastapi import APIRouter, Depends, HTTPException, Request from fastapi.concurrency import run_in_threadpool from pydantic import BaseModel -from fluksio.api.deps import FlowControllerDep, get_current_user +from fluksio.api.deps import ( + CAMERA_IS_SUPERUSERS, + FlowControllerDep, + HiddenDep, + get_current_user, +) from fluksio.flow.devices import DeviceDef, DeviceStore from fluksio.flow.messages import flow_of from fluksio.flow.pipeline import ValueSource @@ -66,9 +71,20 @@ class MessagePoints(BaseModel): @router.get("/", response_model=MessagesPublic) -async def read_messages(controller: FlowControllerDep) -> Any: - """Every message any published flow declares, with its last value.""" +async def read_messages(controller: FlowControllerDep, hidden: HiddenDep) -> Any: + """Every message any published flow declares, with its last value. + + A camera the caller may not see is listed without its value, so a tile + can still be bound to it. + """ infos = await run_in_threadpool(controller.message_catalog) + if hidden: + infos = [ + info.model_copy(update={"value": None, "ts": None}) + if info.name in hidden + else info + for info in infos + ] return MessagesPublic(data=infos, count=len(infos)) @@ -131,8 +147,12 @@ async def publish_message( @router.get("/{name}/history", response_model=MessagePoints) -def read_message_history(name: str, controller: FlowControllerDep) -> Any: +def read_message_history( + name: str, controller: FlowControllerDep, hidden: HiddenDep +) -> Any: """The series behind a chart. Numbers and flags — nothing else plots.""" + if name in hidden: + raise HTTPException(status_code=403, detail=CAMERA_IS_SUPERUSERS) if controller.pipeline is None: return MessagePoints(message=name, numeric=False, points=[]) series = controller.state.history(name) diff --git a/backend/fluksio/cli.py b/backend/fluksio/cli.py index ba0206a..aaef877 100644 --- a/backend/fluksio/cli.py +++ b/backend/fluksio/cli.py @@ -581,6 +581,9 @@ def cmd_serve(args: argparse.Namespace) -> int: # compressing each one again per open socket was a quarter of the # engine's time with four cameras. ws_per_message_deflate=False, + # A device's socket that never finishes closing would otherwise + # hold a stop, and a container's restart, for good. + timeout_graceful_shutdown=5, ) ) try: diff --git a/backend/fluksio/cloud/connector.py b/backend/fluksio/cloud/connector.py index 42ead1e..cfa3cf3 100644 --- a/backend/fluksio/cloud/connector.py +++ b/backend/fluksio/cloud/connector.py @@ -669,11 +669,11 @@ class CloudConnector: """ from sqlmodel import Session - from fluksio.api.deps import user_from_token + from fluksio.api.deps import hidden_from, user_from_token from fluksio.api.routes.flows import ( + carried, demand_frame, device_of, - event_for_panel, panel_scope, revoked, snapshot_payload, @@ -719,12 +719,14 @@ class CloudConnector: await socket.send(_dump({"op": "ws_open_ok", "id": stream_id})) + # The cameras a person may not see, looked at again per event. + hidden = hidden_from(user, self._app, device=only is not None) controller = getattr(self._app.state, "flow_controller", None) told: list[str] = [] if controller is not None: # A device here sends its frames over HTTP: the tunnel carries # text only. - snapshot = snapshot_payload(controller, only, device, "http") + snapshot = snapshot_payload(controller, only, device, "http", hidden) told = list(snapshot.get("demand", [])) await socket.send( _dump({"op": "ws_msg", "id": stream_id, "text": _dump(snapshot)}) @@ -743,7 +745,7 @@ class CloudConnector: def heard(text: str) -> None: nonlocal wanted - asked = wanted_names(text, only) + asked = wanted_names(text, only, hidden) if asked is not None: wanted = asked demand.want(viewer, asked) @@ -754,6 +756,7 @@ class CloudConnector: person=str(user.id) if only is None else "", device=device, devices=getattr(self._app.state, "device_store", None), + superuser=user.is_superuser, ) self._listeners[stream_id] = heard @@ -793,15 +796,37 @@ class CloudConnector: if sender not in done: continue event = sender.result() - if event.get("type") == "device_demand": + # Behind far enough to lose events, a device catches up + # as after a dashboard: the lost one may have been its + # demand or its unpairing. A person whose cameras opened + # or closed is sent what may be seen now. + resync = only is not None and event_bus.dropped(queue) + if only is None: + fresh = hidden_from(user, self._app) + if fresh != hidden: + hidden = fresh + wanted -= hidden + demand.want(viewer, wanted) + resync = True + kind = event.get("type") + if kind == "device_demand": # For the device it names, and for nobody else. if device and event.get("device") == device: text = demand_frame(list(event.get("ports") or [])) await socket.send( _dump({"op": "ws_msg", "id": stream_id, "text": text}) ) + elif only is not None and kind == "dashboard_changed": + resync = True + elif carried(event, only, hidden): + await socket.send( + _dump( + {"op": "ws_msg", "id": stream_id, "text": _dump(event)} + ) + ) + if not resync: continue - if only is not None and event.get("type") == "dashboard_changed": + if only is not None: if device and revoked(token, self._app): # Unpaired or deleted: 1008 sends it to check. await socket.send( @@ -817,24 +842,26 @@ class CloudConnector: only = panel_scope(token, self._app) or set() wanted &= only demand.want(viewer, wanted) - if controller is not None: - catch_up = snapshot_payload( - controller, only, device, "http" + if controller is not None: + catch_up = snapshot_payload( + controller, only, device, "http", hidden + ) + await socket.send( + _dump( + { + "op": "ws_msg", + "id": stream_id, + "text": _dump(catch_up), + } ) - await socket.send( - _dump( - { - "op": "ws_msg", - "id": stream_id, - "text": _dump(catch_up), - } - ) + ) + if kind == "dashboard_changed" and only is not None: + # After the snapshot, as the device socket sends it. + await socket.send( + _dump( + {"op": "ws_msg", "id": stream_id, "text": _dump(event)} ) - elif only is not None and not event_for_panel(event, only): - continue - await socket.send( - _dump({"op": "ws_msg", "id": stream_id, "text": _dump(event)}) - ) + ) except asyncio.CancelledError: raise except Exception as exc: diff --git a/backend/fluksio/core/config.py b/backend/fluksio/core/config.py index 205145e..6164cc9 100644 --- a/backend/fluksio/core/config.py +++ b/backend/fluksio/core/config.py @@ -221,6 +221,11 @@ class Settings(BaseSettings): #: Docker's default 64 MB `/dev/shm`: raise `shm_size` with it when #: workers publish frames. ARTIFACT_VOLATILE_BYTES: int = 48 * 1024 * 1024 + #: The most one live frame may weigh: a volatile `PUT /artifacts` is + #: refused with 413 above it, and a device's frame on its socket is + #: dropped. Held in memory whole, which is why it is far below the + #: durable store's cap. Keep it under uvicorn's 16 MiB websocket limit. + MAX_FRAME_BYTES: int = 4 * 1024 * 1024 # Without a Redis host the engine keeps its state in memory. REDIS_HOST: str | None = None REDIS_PORT: int = 6379 diff --git a/backend/fluksio/flow/artifacts.py b/backend/fluksio/flow/artifacts.py index be76a47..c6bc0a2 100644 --- a/backend/fluksio/flow/artifacts.py +++ b/backend/fluksio/flow/artifacts.py @@ -306,6 +306,32 @@ class VolatileStore: return True return valid_digest(digest) and self._file(digest).is_file() + def spill(self, digest: str) -> None: + """Write a frame held in memory to the directory, for a worker here. + + A node in this container opens a frame with ``load_artifact``, which + looks in the directory: the worker is another process and cannot see + this one's memory. Called before a call naming the frame. The + directory's timer trims it like a worker's own frames. + """ + with self._lock: + data = self._frames.get(digest) + if data is None: + return + target = self._file(digest) + if target.exists(): + return + target.parent.mkdir(parents=True, exist_ok=True) + handle = tempfile.NamedTemporaryFile(dir=target.parent, delete=False) + try: + with handle: + handle.write(data) + os.replace(handle.name, target) + except BaseException: + with contextlib.suppress(OSError): + os.unlink(handle.name) + raise + @property def size(self) -> int: """Bytes held in memory.""" diff --git a/backend/fluksio/flow/devices.py b/backend/fluksio/flow/devices.py index 6698b45..cafef27 100644 --- a/backend/fluksio/flow/devices.py +++ b/backend/fluksio/flow/devices.py @@ -68,6 +68,9 @@ PORT_PATTERN = re.compile(r"^[a-z0-9_.]+$") #: The node type that stands for a device in a flow. DEVICE_TYPE = "device" +#: The dtypes a device's ``viewers`` setting covers: what a camera sends. +CAMERA_DTYPES = frozenset({"image", "video"}) + @dataclass(frozen=True) class Binding: @@ -104,6 +107,7 @@ class DeviceBinding(BaseModel): port: str message: str direction: Literal["in", "out"] + dtype: str = "" def bindings_of(flow: FlowDef) -> list[Binding]: @@ -149,6 +153,11 @@ class DeviceDef(BaseModel): #: touched rather than pointed at. A fact about the screen, not about any #: dashboard on it. touch: bool = False + #: Who may see what the device's camera sends, its ``image`` and ``video`` + #: ``in`` messages, their bytes and its live view: superusers only, or + #: every member too. A device credential is bounded by its dashboards + #: instead, which a superuser assigned. + viewers: Literal["superusers", "members"] = "superusers" #: Bumped on every save. A save based on an older one is refused. version: int = 0 @@ -401,6 +410,9 @@ class DeviceStore: self._states_seen: ( tuple[tuple[int, int, int], dict[str, DeviceState]] | None ) = None + #: The camera messages only superusers may see, at the flow store + #: revision they were worked out at. A device save commits there too. + self._restricted: tuple[int, frozenset[str]] | None = None def _file(self, device_id: str) -> Path: return self.root / f"{checked(device_id)}.json" @@ -459,6 +471,30 @@ class DeviceStore: found.append(binding) return found + def restricted(self) -> frozenset[str]: + """The messages only a superuser may see: what the cameras of devices + whose ``viewers`` are superusers publish, their ``image`` and ``video`` + ``in`` ports. + + Asked on every value a person's socket carries, so worked out again + only once the flow store commits, which every device save does. + """ + revision = self.flows.revision + held = self._restricted + if held is not None and held[0] == revision: + return held[1] + wiring = self.wiring() + found = frozenset( + binding.message + for device in self.list() + if device.viewers == "superusers" + for binding in wiring.get(device.id, ()) + if binding.direction == "in" and binding.dtype in CAMERA_DTYPES + ) + # Keyed by the revision read before, so a commit meanwhile is seen. + self._restricted = (revision, found) + return found + def io(self, device_id: str) -> list[DeviceIo]: """The ports as the device reads them.""" return [ @@ -481,6 +517,7 @@ class DeviceStore: port=b.port, message=b.message, direction=b.direction, + dtype=b.dtype, ) for b in self.wiring().get(device_id, []) ), diff --git a/backend/fluksio/flow/events.py b/backend/fluksio/flow/events.py index 98c56bc..8f00ba3 100644 --- a/backend/fluksio/flow/events.py +++ b/backend/fluksio/flow/events.py @@ -2,7 +2,8 @@ Nodes execute in a thread pool; websocket clients live on the event loop. Publishers are therefore thread-safe and never block: a subscriber that cannot -keep up loses its oldest queued events rather than stalling the engine. +keep up loses its oldest queued events rather than stalling the engine, and +can ask whether it did (``dropped``) to catch up another way. """ from __future__ import annotations @@ -49,6 +50,9 @@ class EventBus: #: may already have moved past, and a device's credential reaches #: these as well as the current one. self.digests: dict[str, deque[str]] = {} + #: Subscribers that lost an event since they last asked. Touched on + #: the event loop only, by the flush and the subscribers. + self._dropped: set[asyncio.Queue[dict[str, Any]]] = set() def bind(self, loop: asyncio.AbstractEventLoop) -> None: """Attach the bus to the running event loop (called once at startup).""" @@ -108,8 +112,20 @@ class EventBus: queue.get_nowait() except asyncio.QueueEmpty: pass + self._dropped.add(queue) queue.put_nowait(event) + def dropped(self, queue: asyncio.Queue[dict[str, Any]]) -> bool: + """Whether this subscriber lost events since it last asked. + + A device's socket that did catches up with a snapshot: what it lost + may have been its demand, or the event that tells it it was unpaired. + """ + if queue in self._dropped: + self._dropped.discard(queue) + return True + return False + @asynccontextmanager async def subscribe(self) -> AsyncIterator[asyncio.Queue[dict[str, Any]]]: """Yield a queue receiving every event published while subscribed.""" @@ -119,6 +135,7 @@ class EventBus: yield queue finally: self._subscribers.discard(queue) + self._dropped.discard(queue) event_bus = EventBus() diff --git a/backend/fluksio/flow/live.py b/backend/fluksio/flow/live.py index 396356f..02bcc66 100644 --- a/backend/fluksio/flow/live.py +++ b/backend/fluksio/flow/live.py @@ -172,19 +172,21 @@ class LiveSessions: person: str, device: str, devices: Any, + superuser: bool = False, ) -> None: """One signalling message from a socket. ``person`` is the id of the account holding the socket and ``device`` - the device whose credential does, one of them "". Any person may watch - any device, camera privacy being deferred. A device speaks only for - the sessions it was asked to serve on this socket. + the device whose credential does, one of them "". A person watches a + device whose ``viewers`` are members, or any as a ``superuser``. A + device speaks only for the sessions it was asked to serve on this + socket. """ session_id = str(message.get("session") or "") if not _SESSION.match(session_id): return if message.get("type") == "live" and person: - self._from_viewer(message, session_id, endpoint, person, devices) + self._from_viewer(message, session_id, endpoint, person, devices, superuser) elif message.get("type") == "signal" and device: self._from_device(message, session_id, endpoint, device) @@ -197,6 +199,7 @@ class LiveSessions: endpoint: Endpoint, who: str, devices: Any, + superuser: bool, ) -> None: device_id = str(message.get("device") or "") action = message.get("action") @@ -205,7 +208,7 @@ class LiveSessions: if session is not None: # Said twice, or somebody else's id: neither gets a second one. return - reason = self._refusal(device_id, devices) + reason = self._refusal(device_id, devices, superuser) if reason: endpoint.deliver( _viewer_message(device_id, session_id, "bye", reason=reason) @@ -289,8 +292,12 @@ class LiveSessions: if not session.viewer.deliver(sent): self._end(session, "overflow", tell_viewer=False) - def _refusal(self, device_id: str, devices: Any) -> str: - """Why this device cannot serve a live view now, or "".""" + def _refusal(self, device_id: str, devices: Any, superuser: bool) -> str: + """Why this device cannot serve this viewer a live view now, or "".""" + if not superuser: + found = devices.find(device_id) if devices is not None else None + if found is None or found.viewers != "members": + return "forbidden" if not self._devices.get(device_id): return "offline" reported = devices.reported_ports(device_id) if devices is not None else None diff --git a/backend/fluksio/flow/nodes/device.py b/backend/fluksio/flow/nodes/device.py index a1cab8a..937d9d2 100644 --- a/backend/fluksio/flow/nodes/device.py +++ b/backend/fluksio/flow/nodes/device.py @@ -13,6 +13,8 @@ reads back as its ``io`` (see ``fluksio.flow.devices``). from __future__ import annotations +import logging +import threading from collections.abc import Iterable from typing import TYPE_CHECKING, Any @@ -21,10 +23,18 @@ from pydantic import BaseModel, Field from fluksio.flow.devices import PORT_PATTERN, presence from fluksio.flow.messages import MessageSpec from fluksio.flow.nodes.base import Node +from fluksio.flow.pipeline import ValueSource if TYPE_CHECKING: from fastapi import FastAPI +logger = logging.getLogger(__name__) + +#: How long a device is away before its flags read false: the floor before +#: its node is shown degraded (``controller.HEALTH_FLOOR``), so a reconnect +#: or a restart of this instance leaves them as they were. +OFFLINE_GRACE_S = 60.0 + def _never(**_: Any) -> None: """What the node would compute, were it ever run: nothing.""" @@ -45,7 +55,7 @@ class DeviceNode(Node): ), ) - __slots__ = ("cfg",) + __slots__ = ("_away", "cfg") def __init__( self, @@ -55,6 +65,8 @@ class DeviceNode(Node): name: str | None = None, ): self.cfg = self.Params.model_validate(params or {}) + #: Counting down from the device going away to its flags reading false. + self._away: threading.Timer | None = None super().__init__( f=_never, requires=requires, provides=provides, params=params, name=name ) @@ -79,11 +91,49 @@ class DeviceNode(Node): async def stop(self, app: FastAPI | None = None) -> None: if self.cfg.device: presence.unwatch(self.cfg.device, self._presence) + self._stay() def _presence(self, online: bool) -> None: # Degraded rather than down: a panel switched off or asleep is a thing # to notice, not a fault of the flow wiring it. if online: + self._stay() self.report_health("ok") else: self.report_health("degraded", "Offline") + self._stay() + self._away = threading.Timer(OFFLINE_GRACE_S, self._gone) + self._away.daemon = True + self._away.start() + + def _stay(self) -> None: + if self._away is not None: + self._away.cancel() + self._away = None + + def _gone(self) -> None: + """The device has been away for the grace: its flags read false. + + A ``bool`` it publishes says something is so right now, such as + ``person``, and nothing tells a device that went away to say it is + no longer so. Credited to this node, as the device's own values are. + """ + pipeline = self._pipeline + if presence.online(self.cfg.device) or pipeline is None: + return + flags = { + spec.name: False + for spec in self.output_ports + if spec.name + and spec.dtype.value == "bool" + and pipeline.state.get(spec.name) is True + } + if not flags: + return + source = ValueSource( + kind="node", id=self.id, label=self.cfg.device, detail="offline" + ) + try: + pipeline.publish(flags, source) + except Exception: + logger.exception("Could not reset the flags of '%s'", self.cfg.device) diff --git a/backend/fluksio/flow/workers.py b/backend/fluksio/flow/workers.py index e678ae0..19dffc0 100644 --- a/backend/fluksio/flow/workers.py +++ b/backend/fluksio/flow/workers.py @@ -24,13 +24,14 @@ import subprocess import sys import threading import time -from collections.abc import Callable +from collections.abc import Callable, Iterator from pathlib import Path from typing import Any import orjson from fluksio_worker import worker_main as _worker_main +from fluksio.flow.artifacts import is_reference from fluksio.flow.events import EventBus from fluksio.flow.nodes.base import NodeOutputError @@ -129,6 +130,19 @@ def _remote_class(name: str) -> type[RemoteError]: return cls +def _digests(value: Any) -> Iterator[str]: + """Every artifact digest a node's inputs name, however deep.""" + if isinstance(value, dict): + if is_reference(value): + yield str(value["digest"]) + return + for item in value.values(): + yield from _digests(item) + elif isinstance(value, list): + for item in value: + yield from _digests(item) + + class _Worker: """One subprocess, and the framing of one request/response over its pipes.""" @@ -237,10 +251,16 @@ class PythonWorkerPool: events: EventBus | None = None, env: dict[str, str] | None = None, max_rss_mb: int = 0, + spill: Callable[[str], None] | None = None, ) -> None: self.python = python self.size = size self.events = events + # Called with each artifact a call's inputs name, before the call: a + # frame the engine holds in memory is written where a worker in this + # container finds it (``VolatileStore.spill``), since it is another + # process. + self.spill = spill # A worker holding more than this when its call returns is retired # rather than kept warm; 0 keeps them all. A library like JAX keeps what # it allocated for as long as the process lives, and warm is the point @@ -307,6 +327,7 @@ class PythonWorkerPool: events=self.events, env={**self.env, **extra}, max_rss_mb=self.max_rss_mb, + spill=self.spill, ) child._generation = self._generation child.start() @@ -593,6 +614,9 @@ class PythonWorkerPool: run_id: str = "", on_event: Callable[[dict[str, Any]], None] | None = None, ) -> Any: + if self.spill is not None: + for digest in _digests(kwargs): + self.spill(digest) response = self._request( { "op": "run", diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py index ff42b6f..5a969ce 100644 --- a/backend/fluksio/main.py +++ b/backend/fluksio/main.py @@ -329,6 +329,9 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: # what it needs overrides this; one that says nothing gets a share. **fair_share_env(accountant.cpus, settings.FLOW_MAX_WORKERS), }, + # The engine's own frames are in its memory, which a worker cannot + # read: each one a call names is written beside the workers'. + spill=volatile.spill if volatile else None, ) pool.start() # On a thread: stopping a worker waits up to five seconds on each child, diff --git a/backend/tests/api/routes/test_camera_access.py b/backend/tests/api/routes/test_camera_access.py new file mode 100644 index 0000000..c9d5089 --- /dev/null +++ b/backend/tests/api/routes/test_camera_access.py @@ -0,0 +1,228 @@ +"""Who may see a device's camera: superusers only, or every member too. + +openpanel ROADMAP C4. ``viewers`` on a device covers its ``image`` and +``video`` messages, their bytes, frames on the socket and its live view. A +device credential is bounded by its dashboards instead. +""" + +from typing import Any + +import pytest +from fastapi.testclient import TestClient + +from fluksio.core.config import settings +from fluksio.flow.demand import demand +from tests.api.routes.test_cameras import ( + CAMERA_NODE, + REPORT, + _ticket, + frame, + wait_for, +) +from tests.api.routes.test_devices import ( + FLOWS, + _dashboard_with, + _pair, + _panels, + _publish_flow, +) + +API = settings.API_V1_STR +PREFIX = f"{API}/devices" +SOCKET = f"{API}/flows/ws" + + +@pytest.fixture +def camera(client: TestClient, superuser_token_headers: dict[str, str]): + """``cam1`` with a camera wired in ``cams``, and a frame on it.""" + _panels(client, superuser_token_headers, {"panels": [{"id": "cam1"}]}) + device = _pair(client, superuser_token_headers, "cam1") + reported = client.put(f"{PREFIX}/cam1/report", headers=device, json=REPORT) + assert reported.status_code == 200, reported.text + _publish_flow(client, superuser_token_headers, "cams", [CAMERA_NODE]) + controller = client.app.state.flow_controller + with client.websocket_connect(SOCKET, headers=device) as ws: + ws.receive_json() + for index in range(3): + ws.send_bytes(frame("cams.camera", b"\xff\xd8 frame %d" % index)) + wait_for( + lambda: (controller.values().get("cams.camera") or {}) + .get("value", {}) + .get("size") + == len(b"\xff\xd8 frame 2") + ) + controller.publish_message("cams.person", True) + yield device + client.delete(f"{FLOWS}/cams", headers=superuser_token_headers) + client.delete(f"{PREFIX}/cam1", headers=superuser_token_headers) + + +def _viewers(client: TestClient, headers: dict[str, str], viewers: str) -> None: + current = next( + one + for one in client.get(f"{PREFIX}/", headers=headers).json()["data"] + if one["id"] == "cam1" + ) + body = { + key: current[key] for key in ("id", "title", "dashboards", "touch", "version") + } + saved = client.put( + f"{PREFIX}/cam1", headers=headers, json={**body, "viewers": viewers} + ) + assert saved.status_code == 200, saved.text + + +def _surfaces(client: TestClient, headers: dict[str, str]) -> dict[str, Any]: + """What one account gets of the camera on every surface but the socket.""" + controller = client.app.state.flow_controller + digest = controller.values()["cams.camera"]["value"]["digest"] + catalog = { + info["name"]: info + for info in client.get(f"{API}/messages/", headers=headers).json()["data"] + } + return { + "catalog": catalog["cams.camera"]["value"] is not None, + "person": catalog["cams.person"]["value"], + "history": client.get( + f"{API}/messages/cams.camera/history", headers=headers + ).status_code, + "flow_history": client.get( + f"{FLOWS}/cams/history/camera", headers=headers + ).status_code, + "state": "cams.camera" + in client.get(f"{FLOWS}/cams/state", headers=headers).json()["values"], + "artifact": client.get( + f"{API}/artifacts/{digest}", headers=headers + ).status_code, + "ice": client.get(f"{PREFIX}/cam1/ice", headers=headers).status_code, + } + + +def _live_start(client: TestClient, headers: dict[str, str]) -> str: + """The reason a viewer's live view of cam1 ends with right away.""" + with client.websocket_connect(_ticket(client, headers)) as ws: + ws.receive_json() + ws.send_json( + {"type": "live", "device": "cam1", "session": "s-access", "action": "start"} + ) + while True: + message = ws.receive_json() + if message.get("type") == "live": + return str(message.get("reason")) + + +@pytest.mark.usefixtures("camera") +def test_a_camera_is_for_superusers_until_opened_to_members( + client: TestClient, + superuser_token_headers: dict[str, str], + normal_user_token_headers: dict[str, str], +) -> None: + member, admin = normal_user_token_headers, superuser_token_headers + open_to_all = { + "catalog": True, + "person": True, + "history": 200, + "flow_history": 200, + "state": True, + "artifact": 200, + "ice": 200, + } + assert _surfaces(client, admin) == open_to_all + assert _surfaces(client, member) == { + "catalog": False, + # Not a camera: its bool is anyone's. + "person": True, + "history": 403, + "flow_history": 403, + "state": False, + "artifact": 403, + "ice": 403, + } + # The frames before the current one too. + store = client.app.state.artifact_store + older = store.put([b"\xff\xd8 older"], media_type="image/jpeg", volatile=True) + client.app.state.flow_controller.publish_message("cams.camera", older) + newer = store.put([b"\xff\xd8 newer"], media_type="image/jpeg", volatile=True) + client.app.state.flow_controller.publish_message("cams.camera", newer) + fetched = client.get(f"{API}/artifacts/{older['digest']}", headers=member) + assert fetched.status_code == 403 + assert _live_start(client, member) == "forbidden" + assert _live_start(client, admin) == "offline" + + _viewers(client, admin, "members") + assert _surfaces(client, member) == open_to_all + assert _live_start(client, member) == "offline" + + +@pytest.mark.usefixtures("camera") +def test_a_members_socket_carries_no_frames_of_a_closed_camera( + client: TestClient, + superuser_token_headers: dict[str, str], + normal_user_token_headers: dict[str, str], + camera: dict[str, str], +) -> None: + """Neither its values, nor the bytes, nor demand on the device; and the + socket is given them once the camera is opened.""" + with client.websocket_connect(_ticket(client, normal_user_token_headers)) as ws: + snapshot = ws.receive_json() + assert "cams.camera" not in snapshot["values"] + assert "cams.person" in snapshot["values"] + ws.send_json({"type": "media", "names": ["cams.camera", "cams.person"]}) + wait_for(lambda: demand.viewed() == {"cams.person"}) + assert demand.ports("cam1") == [] + + with client.websocket_connect(SOCKET, headers=camera) as device: + device.receive_json() + device.send_bytes(frame("cams.camera", b"\xff\xd8 not for members")) + client.app.state.flow_controller.publish_message("cams.person", False) + # The person's flag arrives, the frame before it did not. + while True: + message = ws.receive_json() + names = [ + event.get("name") + for event in message.get("events", [message]) + if event.get("type") == "message_value" + ] + assert "cams.camera" not in names + if "cams.person" in names: + break + + _viewers(client, superuser_token_headers, "members") + while True: + message = ws.receive_json() + if message.get("type") == "snapshot": + break + assert "cams.camera" in message["values"] + + +@pytest.mark.usefixtures("camera") +def test_a_wall_panel_shows_the_camera_its_dashboards_carry( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + """Assigned by a superuser, so the panel's own scope is what bounds it.""" + _dashboard_with( + client, + superuser_token_headers, + "panel_cams", + [ + { + "id": "w_cam", + "type": "media", + "config": {"message": "cams.camera", "dtype": "image"}, + } + ], + ) + _panels( + client, + superuser_token_headers, + {"panels": [{"id": "cam1"}, {"id": "hall", "dashboards": ["panel_cams"]}]}, + ) + panel = _pair(client, superuser_token_headers, "hall") + try: + value = client.app.state.flow_controller.values()["cams.camera"]["value"] + fetched = client.get(f"{API}/artifacts/{value['digest']}", headers=panel) + assert fetched.status_code == 200 + with client.websocket_connect(SOCKET, headers=panel) as ws: + assert "cams.camera" in ws.receive_json()["values"] + finally: + client.delete(f"{PREFIX}/hall", headers=superuser_token_headers) diff --git a/backend/tests/api/routes/test_cameras.py b/backend/tests/api/routes/test_cameras.py index 8e8c543..f027187 100644 --- a/backend/tests/api/routes/test_cameras.py +++ b/backend/tests/api/routes/test_cameras.py @@ -244,10 +244,14 @@ def test_a_frame_the_device_may_not_send_is_dropped( current = controller.values()["cams.camera"]["value"] data = client.app.state.artifact_store.volatile.get(current["digest"]) assert data == b"\xff\xd8 after" + from fluksio.api.routes.flows import FRAME_REFUSALS + + # By fixed keys, never in the words the device chose. said = [r.getMessage() for r in caplog.records if "Dropped a frame" in r.message] - assert any("'person' is wired as bool" in line for line in said) - assert any("not a message one of its ports publishes" in line for line in said) - assert any("only a device sends frames" in line for line in said) + for key in ("not_image", "not_wired", "media_type", "not_a_frame"): + assert f"Dropped a frame from cam1: {FRAME_REFUSALS[key]}" in said + assert any(FRAME_REFUSALS["not_a_device"] in line for line in said) + assert not any("elsewhere" in line or "text/html" in line for line in said) def test_an_unpaired_device_is_closed_and_its_frames_stop( @@ -264,6 +268,106 @@ def test_an_unpaired_device_is_closed_and_its_frames_stop( assert closed.value.code == 1008 +def test_a_frame_past_the_cap_is_dropped( + client: TestClient, camera, monkeypatch: pytest.MonkeyPatch +) -> None: + controller = client.app.state.flow_controller + monkeypatch.setattr(settings, "MAX_FRAME_BYTES", 1000) + with client.websocket_connect(SOCKET, headers=camera) as device: + device.receive_json() + device.send_bytes(frame("cams.camera", b"\xff" * 2000)) + device.send_bytes(frame("cams.camera", b"\xff\xd8 small")) + wait_for(lambda: (controller.values().get("cams.camera") or {}).get("value")) + digest = controller.values()["cams.camera"]["value"]["digest"] + store = client.app.state.artifact_store + assert store.volatile.get(digest) == b"\xff\xd8 small" + + +def test_a_frame_is_named_after_its_message(client: TestClient, camera) -> None: + controller = client.app.state.flow_controller + with client.websocket_connect(SOCKET, headers=camera) as device: + device.receive_json() + device.send_bytes(frame("cams.camera", b"\xff\xd8 named")) + value = wait_for( + lambda: (controller.values().get("cams.camera") or {}).get("value") + ) + assert value["name"] == "cams.camera" + + +def test_an_unpaired_device_is_closed_at_its_next_frame( + client: TestClient, camera +) -> None: + """Without the event an unpair sends, which a full queue may have lost.""" + from starlette.websockets import WebSocketDisconnect + + with client.websocket_connect(SOCKET, headers=camera) as device: + device.receive_json() + device.send_bytes(frame("cams.camera", b"\xff\xd8 one")) + client.app.state.device_store.bump("cam1") + device.send_bytes(frame("cams.camera", b"\xff\xd8 two")) + with pytest.raises(WebSocketDisconnect) as closed: + while True: + device.receive_json() + assert closed.value.code == 1008 + store = client.app.state.artifact_store + current = client.app.state.flow_controller.values()["cams.camera"]["value"] + assert store.volatile.get(current["digest"]) == b"\xff\xd8 one" + + +def test_a_device_that_lost_events_catches_up( + client: TestClient, camera, monkeypatch: pytest.MonkeyPatch +) -> None: + """Its demand and its unpairing come again with a snapshot, after the + events it did get.""" + from starlette.websockets import WebSocketDisconnect + + controller = client.app.state.flow_controller + with client.websocket_connect(SOCKET, headers=camera) as device: + assert device.receive_json()["demand"] == [] + # Demand changes while the bus drops what the socket has queued. + monkeypatch.setattr(event_bus, "dropped", lambda queue: True) + demand.want("someone", {"cams.camera"}) + try: + seen = [] + while not seen or seen[-1].get("type") != "snapshot": + seen.append(device.receive_json()) + assert seen[-1]["demand"] == ["camera"] + finally: + demand.drop("someone") + + # And an unpair it never heard of closes it at the next batch. + client.app.state.device_store.bump("cam1") + controller.publish_message("cams.person", True) + with pytest.raises(WebSocketDisconnect) as closed: + while True: + device.receive_json() + assert closed.value.code == 1008 + + +def test_a_report_is_bounded_once_its_schemas_are_in_place( + client: TestClient, camera +) -> None: + """Thirty pins naming one big schema are thirty copies of it, kept.""" + from fluksio.api.routes.devices import REPORT_MAX, REPORT_STORED_MAX + + schema = { + "type": "object", + "properties": {f"p{i}": {"type": "integer"} for i in range(250)}, + } + pins = [ + {"port": f"gpio.{i}", "direction": "in", "dtype": "bool", "config": "pin"} + for i in range(60) + ] + report = {**REPORT, "config_schemas": {"pin": schema}, "ports": pins} + raw = orjson.dumps(report) + assert len(raw) < REPORT_MAX + answer = client.put(f"{PREFIX}/cam1/report", headers=camera, content=raw) + assert answer.status_code == 413, answer.text + kept = client.app.state.device_store.state("cam1").report + assert len(orjson.dumps(kept)) < REPORT_STORED_MAX + assert kept["board"] == "nano" + + def test_a_screen_may_fetch_the_last_few_frames(client: TestClient, camera) -> None: """A remote screen fetching frame by frame is often one behind.""" controller = client.app.state.flow_controller @@ -290,7 +394,8 @@ def test_a_viewer_asks_for_a_few_cameras_at_most() -> None: names = [f"cams.c{i}" for i in range(MAX_MEDIA_NAMES + 3)] asked = wanted_names(json.dumps({"type": "media", "names": names}), None) - assert asked == set(names[:MAX_MEDIA_NAMES]) + # The newest, which a page lists last: a tile scrolled back to keeps working. + assert asked == set(names[-MAX_MEDIA_NAMES:]) def test_viewers_through_the_portal_count_too( diff --git a/backend/tests/api/routes/test_flows.py b/backend/tests/api/routes/test_flows.py index 9250335..82cd461 100644 --- a/backend/tests/api/routes/test_flows.py +++ b/backend/tests/api/routes/test_flows.py @@ -1,5 +1,6 @@ import asyncio from datetime import UTC, datetime +from types import SimpleNamespace import pytest from fastapi.testclient import TestClient @@ -519,7 +520,9 @@ async def test_the_live_socket_ends_when_its_client_goes( because only SIGKILL ended the process. This returning is what ends that. """ monkeypatch.setattr( - flows_route, "user_from_token", lambda session, token, **_: object() + flows_route, + "user_from_token", + lambda session, token, **_: SimpleNamespace(id="someone", is_superuser=True), ) class Peer: diff --git a/backend/tests/api/routes/test_live.py b/backend/tests/api/routes/test_live.py index 874bf03..d2360f5 100644 --- a/backend/tests/api/routes/test_live.py +++ b/backend/tests/api/routes/test_live.py @@ -337,6 +337,7 @@ def test_a_full_outbox_ends_the_session_instead_of_dropping() -> None: person="someone", device="", devices=Devices(), + superuser=True, ) assert registry.count("cam9") == 1 # The viewer never reads, and the device trickles on. diff --git a/backend/tests/flow/test_device_node.py b/backend/tests/flow/test_device_node.py index b44a20d..3598e1d 100644 --- a/backend/tests/flow/test_device_node.py +++ b/backend/tests/flow/test_device_node.py @@ -3,6 +3,8 @@ import asyncio from pathlib import Path +import pytest + from fluksio.flow.alerts import Alert, AlertManager, AlertsConfig, Channel, Rule from fluksio.flow.controller import HEALTH_FLOOR, FlowController from fluksio.flow.devices import presence @@ -165,3 +167,41 @@ def test_a_device_away_after_a_restart_alerts_nobody(tmp_path: Path) -> None: ] assert sent == [] presence.forget("wall") + + +def test_a_device_gone_for_the_grace_reads_false( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A flag such as `person` stops saying someone is there once the device + has been away for the grace, and a device back in time keeps it.""" + from fluksio.flow.nodes import device as device_node + + monkeypatch.setattr(device_node, "OFFLINE_GRACE_S", 0.1) + controller = _controller(tmp_path) + + def bell() -> object: + return controller.values("hall")["hall.bell"]["value"] + + async def scenario() -> list[object]: + await controller.reload() + presence.opened("wall") + controller.publish_message("hall.bell", True) + seen = [bell()] + # Back within the grace: nothing changes. + presence.closed("wall") + presence.opened("wall") + await asyncio.sleep(0.25) + seen.append(bell()) + # Away for good. + presence.closed("wall") + await asyncio.sleep(0.05) + seen.append(bell()) + await asyncio.sleep(0.25) + seen.append(bell()) + # What reads it hears so too. + seen.append(controller.values("hall")["hall.rung"]["value"]) + await controller._teardown() + presence.forget("wall") + return seen + + assert asyncio.run(scenario()) == [True, True, True, False, False] diff --git a/backend/tests/flow/test_event_bus.py b/backend/tests/flow/test_event_bus.py index 7db2d17..9838b87 100644 --- a/backend/tests/flow/test_event_bus.py +++ b/backend/tests/flow/test_event_bus.py @@ -69,6 +69,25 @@ def test_a_slow_subscriber_still_loses_its_oldest() -> None: assert oldest == "n50" +def test_a_subscriber_that_lost_events_is_told_once() -> None: + """What a device's socket catches up by: it may have lost its demand.""" + + async def run() -> tuple[bool, bool, bool]: + bus = EventBus() + bus.bind(asyncio.get_running_loop()) + async with bus.subscribe() as kept, bus.subscribe() as behind: + for i in range(QUEUE_SIZE): + bus.publish(_event(i)) + await asyncio.sleep(0.01) + while not kept.empty(): + kept.get_nowait() + bus.publish(_event(QUEUE_SIZE)) + await asyncio.sleep(0.01) + return bus.dropped(kept), bus.dropped(behind), bus.dropped(behind) + + assert asyncio.run(run()) == (False, True, False) + + def test_nothing_is_buffered_with_nobody_listening() -> None: """An engine with no browser open holds no events.""" diff --git a/backend/tests/flow/test_workers.py b/backend/tests/flow/test_workers.py index b24843a..54a1775 100644 --- a/backend/tests/flow/test_workers.py +++ b/backend/tests/flow/test_workers.py @@ -6,9 +6,9 @@ import time from collections.abc import Iterator import pytest -from fluksio_worker.worker_main import ARTIFACT_DIR_ENV +from fluksio_worker.worker_main import ARTIFACT_DIR_ENV, ARTIFACT_VOLATILE_DIR_ENV -from fluksio.flow.artifacts import ArtifactStore +from fluksio.flow.artifacts import ArtifactStore, VolatileStore from fluksio.flow.controller import EmitSink, FlowController from fluksio.flow.messages import DType, MessageSpec from fluksio.flow.nodes import Node @@ -473,6 +473,43 @@ def test_a_node_saves_and_loads_an_artifact(tmp_path): pool.stop() +def test_a_node_opens_a_camera_frame_held_in_memory(tmp_path): + """A frame from a device's socket is in the engine's memory only, which a + worker cannot read: the pool writes it out for the call naming it.""" + store = ArtifactStore(tmp_path / "artifacts") + store.volatile = VolatileStore(tmp_path / "ring", 1024 * 1024) + pool = PythonWorkerPool( + python=sys.executable, + size=1, + env={ + ARTIFACT_DIR_ENV: str(store.root), + ARTIFACT_VOLATILE_DIR_ENV: str(store.volatile.root), + }, + spill=store.volatile.spill, + ) + pool.start() + try: + frame = store.put([b"\xff\xd8 a frame"], media_type="image/jpeg", volatile=True) + assert not store.volatile._file(frame["digest"]).exists() + loaded = pool.run( + "demo", + "look", + "import fluksio\n" + "def process(snapshot, earlier):\n" + " with open(fluksio.load_artifact(snapshot), 'rb') as f:\n" + " return {'bytes': f.read().decode('latin-1')}\n", + # Nested, as a node handing on a list of frames would get them. + {"snapshot": frame, "earlier": [frame]}, + "demo.look", + timeout=10, + ) + assert loaded == {"bytes": "\xff\xd8 a frame"} + # On the ring's directory, which the engine trims, never the volume. + assert store.path(frame["digest"]) is None + finally: + pool.stop() + + def test_a_node_passes_audio_to_the_next_one(tmp_path): # The media path end to end: one node writes a clip and declares what kind # of bytes it is, the next opens it. Only the reference travels. diff --git a/backend/tests/test_artifacts.py b/backend/tests/test_artifacts.py index 857a80f..649a2f3 100644 --- a/backend/tests/test_artifacts.py +++ b/backend/tests/test_artifacts.py @@ -240,6 +240,55 @@ def test_an_upload_may_ask_for_the_ring( store.volatile = previous +def test_a_ring_never_holds_more_than_its_bound_and_one_frame(tmp_path): + """Frames of every size, the bound never passed by more than the newest.""" + ring = VolatileStore(tmp_path / "ring", limit_bytes=10_000) + sizes = [700, 9_000, 30, 12_000, 4_000, 4_000, 2_500, 9_999, 1, 6_000] * 3 + for index, size in enumerate(sizes): + ring.put([bytes([index % 256]) * size]) + assert ring.size <= ring.limit_bytes or len(ring._frames) == 1 + assert ring.size <= ring.limit_bytes + size + + +def test_a_volatile_upload_is_held_to_the_frame_cap( + client: TestClient, + superuser_token_headers: dict[str, str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A frame is held in memory whole, so it is refused past a few MB before + any of it is read, and a chunked one as soon as it passes the cap.""" + store = client.app.state.artifact_store + previous = store.volatile + store.volatile = VolatileStore( + store.root.parent / "test-ring", limit_bytes=1_000_000 + ) + monkeypatch.setattr(settings, "MAX_FRAME_BYTES", 1000) + url = f"{settings.API_V1_STR}/artifacts?volatile=1&media_type=image%2Fjpeg" + try: + declared = client.put(url, headers=superuser_token_headers, content=b"x" * 1001) + assert declared.status_code == 413 + + def chunks(): + for _ in range(5): + yield b"x" * 400 + + chunked = client.put(url, headers=superuser_token_headers, content=chunks()) + assert chunked.status_code == 413 + assert store.volatile.size == 0 + + fits = client.put(url, headers=superuser_token_headers, content=b"x" * 1000) + assert fits.status_code == 200, fits.text + # The durable store keeps its own, larger cap. + kept = client.put( + f"{settings.API_V1_STR}/artifacts", + headers=superuser_token_headers, + content=b"y" * 5000, + ) + assert kept.status_code == 200, kept.text + finally: + store.volatile = previous + + def test_a_type_that_can_run_script_is_served_sandboxed( client: TestClient, superuser_token_headers: dict[str, str] ) -> None: diff --git a/docker/compose.dev.yml b/docker/compose.dev.yml index 55aecc0..eee2d45 100644 --- a/docker/compose.dev.yml +++ b/docker/compose.dev.yml @@ -2,12 +2,21 @@ # published host ports, hot reload and the mail/browser test helpers. # Never used in production — the Makefile only layers it for `make dev*`. +# Every container's log rotates: json-file, at most 3 files of 10 MB each. +# Docker's default keeps one file that grows until the host's disk is full. +x-logging: &logging + driver: json-file + options: + max-size: "10m" + max-file: "3" + services: # Local Traefik. Services stay reachable both on their published ports and # on http://api.localhost / http://app.localhost via the routers in # compose.yml. Dashboard at http://localhost:8090. proxy: + logging: *logging # A patch tag, so a host holding an older cached 3.6 pulls this one. 3.6.25 # knows --accesslog.fields.queryparameters, which older 3.6 refuses. image: traefik:3.6.25 @@ -61,6 +70,7 @@ services: # them, and 1883 in particular tends to be taken by the real broker on a # machine that runs this alongside anything else. mosquitto: + logging: *logging image: eclipse-mosquitto:2 restart: "no" # The config the image ships for exactly this: listener on 1883, anonymous @@ -68,6 +78,7 @@ services: command: mosquitto -c /mosquitto-no-auth.conf influxdb: + logging: *logging image: influxdb:2.7 restart: "no" # Dev credentials, and no volume: the container comes up initialised and a @@ -151,6 +162,7 @@ services: API_DOCS_ENABLED: "true" mailcatcher: + logging: *logging image: schickling/mailcatcher ports: - "1080:1080" @@ -162,6 +174,7 @@ services: restart: "no" playwright: + logging: *logging build: context: .. dockerfile: frontend/Dockerfile.playwright diff --git a/docker/compose.traefik.yml b/docker/compose.traefik.yml index 541c80b..5a0fa9f 100644 --- a/docker/compose.traefik.yml +++ b/docker/compose.traefik.yml @@ -1,5 +1,14 @@ +# Every container's log rotates: json-file, at most 3 files of 10 MB each. +# Docker's default keeps one file that grows until the host's disk is full. +x-logging: &logging + driver: json-file + options: + max-size: "10m" + max-file: "3" + services: traefik: + logging: *logging # A patch tag, so a host holding an older cached 3.6 pulls this one. 3.6.25 # knows --accesslog.fields.queryparameters, which older 3.6 refuses. image: traefik:3.6.25 diff --git a/docker/compose.yml b/docker/compose.yml index 4269833..6ef08ec 100644 --- a/docker/compose.yml +++ b/docker/compose.yml @@ -6,11 +6,20 @@ # Paths are relative to this directory, which compose uses as the project # directory (build contexts therefore point at `..`, the repo root). +# Every container's log rotates: json-file, at most 3 files of 10 MB each. +# Docker's default keeps one file that grows until the host's disk is full. +x-logging: &logging + driver: json-file + options: + max-size: "10m" + max-file: "3" + services: # Umami's database, and nothing else's. The engine keeps its own state in # SQLite on the flow-data volume, which is why this is behind a profile now. db: + logging: *logging image: postgres:18 container_name: fluksio-db restart: always @@ -36,6 +45,7 @@ services: - POSTGRES_DB=${POSTGRES_DB:-app} redis: + logging: *logging image: redis:8.2-alpine container_name: fluksio-redis restart: always @@ -57,6 +67,7 @@ services: - app-redis-data:/data adminer: + logging: *logging image: adminer container_name: fluksio-adminer restart: always @@ -73,6 +84,7 @@ services: - ADMINER_DESIGN=pepa-linha-dark backend: + logging: *logging image: '${DOCKER_IMAGE_BACKEND?Variable not set}:${TAG-latest}' container_name: fluksio-api restart: always @@ -177,6 +189,7 @@ services: # Static documentation site (zensical), served at docs.${DOMAIN}. Content and # config live in ../docs/ and ../zensical.toml; `make docs-serve` previews it. docs: + logging: *logging container_name: fluksio-docs restart: always security_opt: @@ -211,6 +224,7 @@ services: # authority: on a machine that runs anything besides this stack, that is a # deliberate operator decision. `make up` opts in; the dev stacks do not. autoheal: + logging: *logging image: willfarrell/autoheal:latest container_name: fluksio-autoheal profiles: ["autoheal"] @@ -236,6 +250,7 @@ services: # goes in through a config file rather than the command line, which `ps` # shows. See DEPLOY.md for ports, DNS and certificates. coturn: + logging: *logging image: coturn/coturn:4.6.3 container_name: fluksio-coturn profiles: ["turn"] @@ -282,6 +297,7 @@ services: # `analytics` profile, so a plain `make up` never starts it; bring it up with # `make umami`, which also provisions its database. umami: + logging: *logging container_name: fluksio-umami # Pinned by digest so a bump is a deliberate step: this is an internet-facing # third-party dashboard, not a library. diff --git a/docs/code/api.md b/docs/code/api.md index f1674d7..38e530e 100644 --- a/docs/code/api.md +++ b/docs/code/api.md @@ -174,7 +174,10 @@ curl "$FLUKSIO/artifacts/" \ The upload returns a reference with the content digest. Identical bytes are stored once. Node code uses `save_artifact` and `load_artifact` instead. `?volatile=true` holds the bytes in the engine's memory for a live frame, where -newer frames push them out, instead of storing them. +newer frames push them out, instead of storing them. Such a frame is at most +`MAX_FRAME_BYTES` (4 MB), and a larger one is refused with 413. A camera only +superusers may see answers 403 to anyone else for the frames its messages +hold, the last 8 included (see [Who may watch a camera](../interface/dashboards.md#who-may-watch-a-camera)). A superuser can remove stored bytes with `DELETE /artifacts/`. While a run, a current message or a flow input's initial value still references the @@ -200,8 +203,8 @@ on its own, so this is for reclaiming space sooner. | `GET` | `/devices/pair/{code}/device` | what is waiting on a code, before it is approved | | `POST` | `/devices/{id}/pair` | adopt by `{code}`, or by `{request}` from the pending list for a request from your network | | `POST` | `/devices/{id}/unpair` | revoke the device credential, one the portal issued included | -| `PUT` | `/devices/{id}/report` | what the device has, from its own credential only; at most 16 KB, replaces the last report, read leniently (see [What a device reports](../interface/dashboards.md#what-a-device-reports)) | -| `GET` | `/devices/{id}/ice` | STUN and TURN servers for a live view of the device, as `{ice_servers: [{urls, username?, credential?}], ttl}`, with TURN credentials of the caller's own; not for a device credential | +| `PUT` | `/devices/{id}/report` | what the device has, from its own credential only; at most 16 KB, and 64 KB once its named `config_schemas` are put in place (413 above either), replaces the last report, read leniently (see [What a device reports](../interface/dashboards.md#what-a-device-reports)) | +| `GET` | `/devices/{id}/ice` | STUN and TURN servers for a live view of the device, as `{ice_servers: [{urls, username?, credential?}], ttl}`, with TURN credentials of the caller's own; 403 for whoever may not watch its camera, and for a device credential | `/panels/pair`, `/panels/pair/{code}` and `GET /panels/{id}` answer as their `/devices/` counterparts, for firmware that has not updated yet. @@ -254,7 +257,7 @@ The socket sends a snapshot followed by node, value, health, and run events. A client asks for the bytes of the media it shows with `{"type": "media", "names": [...]}`, which replaces what it asked for before, -for at most 8 messages. Each frame held in memory then arrives as a binary +for at most 8 messages, the last 8 it names when it names more. Each frame held in memory then arrives as a binary message before the value naming it: four bytes of header length (big endian), a JSON header `{"type": "media", "name", "digest", "media_type", "ts"}`, then the bytes. Messages are sent without per-message compression. @@ -264,17 +267,26 @@ A device's socket is bounded to its own messages. Its snapshot carries `"uplink"`, `"binary"` on its own socket and `"http"` through the portal. When that set changes it is sent `{"type": "demand", "ports": [...]}`. A port has demand while a socket asks for the media of the message it is wired to, or a -node of a running flow reads that message. +node of a running flow reads that message. A socket that falls so far behind +that events are dropped for it is sent a fresh snapshot, demand included, and +is closed with 1008 if it was unpaired meanwhile. + +A person who may not watch a camera (see +[Who may watch a camera](../interface/dashboards.md#who-may-watch-a-camera)) +gets none of its values in the snapshot or after it, and no frames when asking +for them. Switching the camera's setting sends that person's open sockets a +fresh snapshot. With a `binary` uplink the device sends a camera frame as a binary message in the same layout, with the header `{"type": "frame", "name": "", "media_type": "image/jpeg", "ts": }`. The message has to be one of -its `in` ports wired as `image`. The frame is held in memory and published -in one step, credited to its device node. A frame that does not qualify is -logged and dropped and the socket stays open. With `http` it uploads with -`PUT /artifacts?volatile=true&name=` and publishes with -`POST /messages/`. A device that is unpaired or deleted while -connected is closed with 1008. +its `in` ports wired as `image`, and the whole message at most +`MAX_FRAME_BYTES`. The frame is held in memory and published in one step, +named after the message and credited to its device node. A frame that does not +qualify is dropped, the reason logged once per socket, and the socket stays +open. With `http` it uploads with `PUT /artifacts?volatile=true&name=` +and publishes with `POST /messages/`. A device that is unpaired or +deleted while connected is closed with 1008, at the latest with its next frame. ### Live view @@ -294,10 +306,11 @@ the viewer. In order: | 4 | sends and hears `candidate`, the bare `candidate:...` line | hears and sends `candidate` | | 5 | sends `bye`, or hears `bye` or `busy` with a `reason` | hears `bye` with a `reason`, or sends `bye` or `busy` | -Any signed-in person may watch any device. A device speaks only for the -sessions it was handed on that socket. The server answers `busy` itself once a -device has 2 sessions, and `bye` with `offline` or `unsupported` when the -device is not connected or reports no `live` port. A session the viewer has not +A superuser may watch any device, anyone else a device whose camera is open +to members, and `start` is answered `bye` with `forbidden` otherwise. A device +speaks only for the sessions it was handed on that socket. The server answers +`busy` itself once a device has 2 sessions, and `bye` with `offline` or +`unsupported` when the device is not connected or reports no `live` port. A session the viewer has not answered within 10 seconds ends with `timeout`, and either socket closing ends its sessions with `closed` or `offline`. A socket that falls 64 signalling messages behind ends the session with `overflow` instead of losing one. diff --git a/docs/interface/dashboards.md b/docs/interface/dashboards.md index 06f6356..dac6eb1 100644 --- a/docs/interface/dashboards.md +++ b/docs/interface/dashboards.md @@ -552,12 +552,20 @@ lights up on the canvas. `out` messages are read only. The node is shown degraded, *Offline*, in amber while the device is not connected: a warning on Home, not a fault, since a panel switched off or asleep breaks nothing. +A `bool` the device publishes, such as a camera's `person`, says something is +so right now, and a device that went away cannot say it no longer is. Once the +device has been away for 60 seconds, the same wait as before its node shows +*Offline*, each of its `bool` messages that reads true is set to false by its +device node. A device back within that wait keeps its values, so a restart of +either side changes nothing. Its other values keep the last one it sent. + ### What a device reports A paired device says what it has with `PUT /devices/{id}/report`, from its own credential only. openpanel sends it after pairing and once per boot. The report -replaces the one before and is at most 16 KB. It is read leniently, so a newer -firmware still reports to an older instance: +replaces the one before and is at most 16 KB, and 64 KB as kept, once named +`config_schemas` are put in place. It is read leniently, so a newer firmware +still reports to an older instance: - keys this version does not know are dropped - a value it does not know for a display `shape` or a port `direction` reads as @@ -635,6 +643,29 @@ A device may fetch the last 8 artifacts each of its messages pointed at, not only the current one, so a screen fetching a camera frame by frame through the portal is not refused a frame the camera has just replaced. +#### Who may watch a camera + +**Who may watch** in the Camera section of the device's page decides who sees +what its camera sends, its `image` and `video` messages: **Superusers**, the +default, or **All members**. A member of the portal is a member here. For +anyone it leaves out: + +- the messages' values are left out of the live socket, `GET /messages/` and a + flow's state, and their history answers 403 +- the frames those messages hold, the last 8 included, answer 403, and a + socket asking for their bytes gets none +- a live view is refused with `forbidden`, and so are TURN credentials +- a dashboard tile or the device's preview says *No access to this camera* + +A wall panel shows the camera when one of its dashboards carries it, whoever +watches the wall: a superuser assigned it the dashboards. + +The setting covers the camera's own messages, not what a flow makes of them. +Anyone who edits flows can wire a node that passes a picture on under another +name, and node code reads whatever it is handed. How long a kept `snapshot` +stays and a light on the device that shows the camera is on are not part of +it yet. + #### Live view A media tile showing a device's `camera`, and the preview on the device's diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index 777cc25..e997d33 100644 --- a/docs/reference/configuration.md +++ b/docs/reference/configuration.md @@ -169,14 +169,18 @@ MIG entries are passed on as written. | `ARTIFACT_GC_GRACE_S` | `3600` | minimum age before cleanup | | `ARTIFACT_VOLATILE_DIR` | auto-detected | where local workers write live frames, under `/dev/shm` or the temporary directory | | `ARTIFACT_VOLATILE_BYTES` | `50331648` (48 MiB) | live-media ring size, in the engine's memory and again in that directory; `0` stores volatile saves durably | +| `MAX_FRAME_BYTES` | `4194304` (4 MiB) | the largest live frame: a volatile upload past it gets 413, a device's frame on its socket is dropped; keep it under 16 MiB, the websocket's limit | Cleanup removes durable artifacts no longer referenced by recorded runs or current messages, and pauses while a run is active. Volatile artifacts live in a bounded ring in the engine's memory, and the oldest disappear as it fills. Recorded run outputs are copied into durable storage. A worker in the engine's container writes its frames to `ARTIFACT_VOLATILE_DIR` instead, which the -engine trims to the same size every few seconds. If workers publish frames -under Docker, raise `shm_size` with the ring size. +engine trims to the same size every few seconds. A frame the engine holds is +written there too before a node in this container is handed it, so +`load_artifact` finds it. If workers publish or read frames under Docker, +raise `shm_size` with the ring size. The ring never holds more than its size +plus one frame. ## Agents diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index 9b2907c..e6a96e0 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -933,6 +933,11 @@ export const DeviceBindingSchema = { type: 'string', enum: ['in', 'out'], title: 'Direction' + }, + dtype: { + type: 'string', + title: 'Dtype', + default: '' } }, type: 'object', @@ -976,6 +981,12 @@ export const DeviceDefSchema = { title: 'Touch', default: false }, + viewers: { + type: 'string', + enum: ['superusers', 'members'], + title: 'Viewers', + default: 'superusers' + }, version: { type: 'integer', title: 'Version', @@ -1011,6 +1022,12 @@ export const DeviceDocumentSchema = { title: 'Touch', default: false }, + viewers: { + type: 'string', + enum: ['superusers', 'members'], + title: 'Viewers', + default: 'superusers' + }, version: { type: 'integer', title: 'Version', @@ -1090,6 +1107,12 @@ export const DevicePublicSchema = { title: 'Touch', default: false }, + viewers: { + type: 'string', + enum: ['superusers', 'members'], + title: 'Viewers', + default: 'superusers' + }, version: { type: 'integer', title: 'Version', diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index 5f6261a..23b8061 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -271,7 +271,7 @@ export class ArtifactsService { * watching now, which the oldest of falls out of memory rather than being * kept. A worker on another host or a device publishing a camera sends this. * It is read into memory, where the ring holds it, and never touches the - * disk. + * disk, so it is held to ``MAX_FRAME_BYTES`` rather than the store's cap. * @param data The data for the request. * @param data.name * @param data.mediaType @@ -1015,6 +1015,7 @@ export class DevicesService { * * STUN and TURN servers, the TURN ones with credentials of the caller's own * that expire. Empty when none are set, which is enough on one network. + * Only for whoever may see the device's camera (``DeviceDef.viewers``). * @param data The data for the request. * @param data.deviceId * @returns IceServers Successful Response @@ -1958,6 +1959,9 @@ export class MessagesService { /** * Read Messages * Every message any published flow declares, with its last value. + * + * A camera the caller may not see is listed without its value, so a tile + * can still be bound to it. * @returns MessagesPublic Successful Response * @throws ApiError */ diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 48e5a11..fa2fc59 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -238,6 +238,7 @@ export type DeviceBinding = { port: string; message: string; direction: 'in' | 'out'; + dtype?: string; }; export type direction = 'in' | 'out'; @@ -254,9 +255,12 @@ export type DeviceDef = { title?: string; dashboards?: Array<(string)>; touch?: boolean; + viewers?: 'superusers' | 'members'; version?: number; }; +export type viewers = 'superusers' | 'members'; + /** * A device as it reads itself: what it shows, and what its ports carry. * @@ -268,6 +272,7 @@ export type DeviceDocument = { title?: string; dashboards?: Array<(string)>; touch?: boolean; + viewers?: 'superusers' | 'members'; version?: number; io?: Array; bindings?: Array; @@ -293,6 +298,7 @@ export type DevicePublic = { title?: string; dashboards?: Array<(string)>; touch?: boolean; + viewers?: 'superusers' | 'members'; version?: number; io?: Array; bindings?: Array; diff --git a/frontend/src/components/Dashboard/MediaWidget.tsx b/frontend/src/components/Dashboard/MediaWidget.tsx index 0f47e4a..7fdcfc6 100644 --- a/frontend/src/components/Dashboard/MediaWidget.tsx +++ b/frontend/src/components/Dashboard/MediaWidget.tsx @@ -6,7 +6,7 @@ import { isRef, useArtifactUrl } from "@/lib/media" import { cn } from "@/lib/utils" import { useBoundValue } from "./dataContext" import { config, text } from "./ui/core/config" -import { useLiveDevice, useLiveView } from "./useLiveView" +import { useCameraHidden, useLiveDevice, useLiveView } from "./useLiveView" import type { WidgetProps } from "./widgets" /** @@ -66,10 +66,15 @@ export function MediaView({ const ref = isRef(value) ? value : null const kind = (ref?.media_type ?? dtype).split("/")[0] const playing = kind === "audio" || kind === "video" - const camera = useLiveDevice(stream ? "" : message) + // Only superusers may see this camera: nothing is asked of it. + const hidden = useCameraHidden(message) + const camera = useLiveDevice(stream || hidden ? "" : message) const live = useLiveView(camera) const onAir = live.phase === "playing" - const active = (shown || playing) && !onAir + // A stream URL draws the picture itself, all but sound: the message's + // bytes would only be demand on a camera nothing shows. + const draws = (!stream || kind === "audio") && !hidden + const active = (shown || playing) && !onAir && draws const url = useArtifactUrl( active ? ref : null, active ? message || undefined : undefined, @@ -116,6 +121,13 @@ export function MediaView({ if (!message && !stream) { return

Pick a message.

} + if (hidden) { + return ( +

+ No access to this camera +

+ ) + } if (!ref) { return

Nothing published yet.

} diff --git a/frontend/src/components/Dashboard/useLiveView.ts b/frontend/src/components/Dashboard/useLiveView.ts index c45e97a..4225b78 100644 --- a/frontend/src/components/Dashboard/useLiveView.ts +++ b/frontend/src/components/Dashboard/useLiveView.ts @@ -1,13 +1,18 @@ import { useQuery } from "@tanstack/react-query" import { useCallback, useEffect, useRef, useState } from "react" -import { LIVE_PORT, reportedPorts } from "@/components/Devices/ports" +import { + cameraMessages, + LIVE_PORT, + reportedPorts, +} from "@/components/Devices/ports" import { devicesQueryOptions } from "@/components/Devices/queries" import { type LiveSignal, listenLive, sendLive, } from "@/components/Flow/useFlowSocket" +import useAuth from "@/hooks/useAuth" import { appRoute } from "@/lib/portal" /** How long a live view gets from the tap to its first frame. */ @@ -45,6 +50,29 @@ export function useLiveDevice(message: string): string { return device?.id ?? "" } +/** + * Whether this viewer may not see the camera publishing `message`: a + * device's image or video while its `viewers` are superusers, for anyone + * else. The server holds the frames back too, this only says why the tile + * stays empty. + * + * A wall panel shows what its dashboards carry, which a superuser assigned, + * and cannot read the device list anyway. + */ +export function useCameraHidden(message: string): boolean { + const { user } = useAuth() + const enabled = + Boolean(message) && + !appRoute().startsWith("/panel") && + user?.is_superuser === false + const { data } = useQuery({ ...devicesQueryOptions(), enabled }) + if (!enabled) return false + return (data?.data ?? []).some( + (device) => + device.viewers !== "members" && cameraMessages(device).includes(message), + ) +} + /** * A live view of one device's camera: the browser's own `RTCPeerConnection` * answering the device's offer, signalled over the page's socket. diff --git a/frontend/src/components/Devices/DeviceDetail.tsx b/frontend/src/components/Devices/DeviceDetail.tsx index 01085c5..8c66e6b 100644 --- a/frontend/src/components/Devices/DeviceDetail.tsx +++ b/frontend/src/components/Devices/DeviceDetail.tsx @@ -44,6 +44,7 @@ import { type DeviceReport, deviceNodeOf, freeNodeId, + hasCamera, LIVE_PORT, type ReportedPort, reportedPorts, @@ -141,7 +142,7 @@ export function DeviceDetail({ id }: { id: string }) {
- +
@@ -219,33 +220,81 @@ function Basics({ ) } +/** Who may see a device's camera, as the setting reads. */ +const VIEWERS = { + superusers: "Superusers", + members: "All members", +} as const + /** - * What the device's camera sees, once a flow wires its `camera` port. + * Who may watch the device's camera, and what it sees once a flow wires its + * `camera` port. * * Watching is what turns the camera on: it sends frames only while this * preview, a tile or a flow reading them wants them. */ -function Camera({ device }: { device: DevicePublic }) { +function Camera({ + device, + canEdit, + onChange, +}: { + device: DevicePublic + canEdit: boolean + onChange: (changes: Partial) => void +}) { + if (!hasCamera(device)) return null const bound = (device.bindings ?? []).find( (binding) => binding.port === "camera" && binding.direction === "in", ) - if (!bound) return null return (
-
- +
+ Who may watch + + Its pictures, their history and its live view: superusers only, or + every member of this instance. A wall panel shows the camera when + one of its dashboards carries it. Whoever edits flows can still pass + the pictures on through a node. + +
+ {bound ? ( +
+ +
+ ) : null}
) diff --git a/frontend/src/components/Devices/ports.ts b/frontend/src/components/Devices/ports.ts index cc58499..f1fbf91 100644 --- a/frontend/src/components/Devices/ports.ts +++ b/frontend/src/components/Devices/ports.ts @@ -63,6 +63,30 @@ export function screensOf(devices: DevicePublic[] = []) { }) } +/** What a device's `viewers` setting covers: what a camera sends. */ +const CAMERA_DTYPES: readonly string[] = ["image", "video"] + +/** The messages a device's camera publishes, its image and video `in` ports. */ +export const cameraMessages = (device: DevicePublic) => + (device.bindings ?? []) + .filter( + (binding) => + binding.direction === "in" && + CAMERA_DTYPES.includes(binding.dtype ?? ""), + ) + .map((binding) => binding.message) + +/** + * Whether a device has a camera to say who may watch: one it reports, or an + * image or video port wired in a flow. + */ +export const hasCamera = (device: DevicePublic) => + cameraMessages(device).length > 0 || + (reportedPorts(device) ?? []).some( + (port) => + port.direction === "in" && CAMERA_DTYPES.includes(port.dtype ?? ""), + ) + /** Whether a reported port may be wired in `direction`. `live` never is. */ export const wiresAs = (port: ReportedPort, direction: "in" | "out") => port.port !== LIVE_PORT &&