Live view over WebRTC (openpanel L1, server and browser)

A device reporting a `live` port streams its camera to the browser's own
RTCPeerConnection. The flows socket carries the signalling as text, so it
crosses the portal: `live` messages on a person's socket, `signal` on a
device's (start, sdp, candidate, bye, busy). The server routes by session,
holds a device to two viewers, ends a session when either socket closes or
no answer comes within 10 s, and gives each socket an outbox of its own
that ends a session rather than drop a message. TURN credentials are
minted per session in coturn's REST scheme (STUN_URLS, TURN_URLS,
TURN_SECRET), the viewer's with the device's offer; GET /devices/{id}/ice
hands them out too. The tunnel keeps a stream's early frames instead of
dropping them past eight and forgets them when the stream or link goes.

A media tile on a device's camera, and the Devices tab's preview, play
live when tapped, pause frames meanwhile, and fall back to frames with a
short note on busy, failure or no picture within 10 s. coturn is an
optional compose service behind the `turn` profile.

backend/fluksio/api/routes/devices.py

@@ -71,6 +71,7 @@ from fluksio.flow.devices import (
     read_report,
 )
 from fluksio.flow.events import event_bus
+from fluksio.flow.live import ICE_TTL_S, ice_servers
 from fluksio.flow.state import CONNECT_TIMEOUT_S, SOCKET_TIMEOUT_S
 from fluksio.flow.store import StaleVersion
 from fluksio.models import Message, User
@@ -387,6 +388,20 @@ class DevicesPublic(BaseModel):
     frontend_host: str = ""
 
 
+class IceServer(BaseModel):
+    """One STUN or TURN server, as `RTCPeerConnection` takes it."""
+
+    urls: list[str]
+    username: str | None = None
+    credential: str | None = None
+
+
+class IceServers(BaseModel):
+    ice_servers: list[IceServer] = Field(default_factory=list)
+    #: Seconds the TURN credentials stay good.
+    ttl: int = 0
+
+
 def _describe(request: Request) -> str:
     """A line naming the device behind a pairing request.
 
@@ -665,6 +680,23 @@ async def read_device(device_id: str, store: DeviceStoreDep) -> Any:
     return await run_in_threadpool(_document, device, store)
 
 
+@router.get(
+    "/{device_id}/ice",
+    response_model=IceServers,
+    response_model_exclude_none=True,
+)
+async def read_ice(device_id: str, store: DeviceStoreDep, user: CurrentUser) -> Any:
+    """Where a browser looks for a way to this device's live view.
+
+    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.
+    """
+    if await run_in_threadpool(store.find, device_id) is None:
+        raise _not_found(device_id)
+    servers = [IceServer(**server) for server in ice_servers(str(user.id))]
+    return IceServers(ice_servers=servers, ttl=ICE_TTL_S)
+
+
 panels_router.add_api_route(
     "/{device_id}",
     read_device,

backend/fluksio/api/routes/flows.py

@@ -37,6 +37,7 @@ from fluksio.api.routes.messages import port_source
 from fluksio.api.socket_auth import TICKET_TTL_S, issue_ticket, socket_token
 from fluksio.core.config import settings
 from fluksio.core.db import engine
+from fluksio.flow import live as live_view
 from fluksio.flow.artifacts import is_reference
 from fluksio.flow.controller import NODE_TYPES, FlowController, follow_header
 from fluksio.flow.dashboards import DashboardStore
@@ -1369,6 +1370,9 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -
     [...]}``) and is pushed those frames. A device also sends its camera's
     frames here, as binary frames in the layout it is pushed them in, and is
     told when its on-demand ports have demand.
+
+    A live view's signalling rides here too (``fluksio.flow.live``): ``live``
+    messages on a person's socket, ``signal`` messages on a device's.
     """
     token = socket_token(websocket.headers, ticket, token)
 
@@ -1425,6 +1429,9 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -
 
     async with event_bus.subscribe() as queue:
         receiver = asyncio.create_task(_receive(websocket))
+        # Where a live view this socket is one end of reaches it. Its own
+        # outbox rather than the bus's queue, which drops when full.
+        endpoint = live_view.Endpoint()
         if device:
             presence.opened(device)
             # Demand that moved between the snapshot and the subscription
@@ -1432,12 +1439,23 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -
             ports = demand.ports(device)
             if ports != told:
                 await websocket.send_text(demand_frame(ports))
+            live_view.sessions.attach(device, endpoint)
+        outbox = asyncio.create_task(endpoint.next())
         try:
             while True:
                 sender = asyncio.create_task(queue.get())
                 done, _ = await asyncio.wait(
-                    {sender, receiver}, return_when=asyncio.FIRST_COMPLETED
+                    {sender, receiver, outbox}, return_when=asyncio.FIRST_COMPLETED
                 )
+                if outbox in done:
+                    # Signalling, for this socket alone.
+                    if sender not in done:
+                        sender.cancel()
+                    text = outbox.result()
+                    outbox = asyncio.create_task(endpoint.next())
+                    await websocket.send_text(text)
+                    if receiver not in done and sender not in done:
+                        continue
                 if receiver in done:
                     if sender not in done:
                         # Nobody will ever read this one: the next turn makes
@@ -1479,11 +1497,21 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -
                         if asked is not None:
                             wanted = asked
                             demand.want(viewer, wanted)
+                        elif (signal := live_view.parse(frame)) is not None:
+                            live_view.sessions.heard(
+                                signal,
+                                endpoint,
+                                person=str(user.id) if only is None else "",
+                                device=device,
+                                devices=getattr(
+                                    websocket.app.state, "device_store", None
+                                ),
+                            )
                     receiver = asyncio.create_task(_receive(websocket))
                     if sender not in done:
                         continue
-                # `sender` is done from here: `FIRST_COMPLETED` returns with at
-                # least one of the two, so `receiver` missing means this one is.
+                # `sender` is done from here: the turns where it is not have
+                # gone round again above.
 
                 # Everything the bus has right now, not just the one event
                 # that woke this: a cascade puts a dozen in at once, and one
@@ -1544,7 +1572,9 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -
             pass
         finally:
             receiver.cancel()
+            outbox.cancel()
             demand.drop(viewer)
+            live_view.sessions.detach(endpoint)
             if device:
                 presence.closed(device)
 

backend/fluksio/cloud/connector.py

@@ -73,7 +73,9 @@ ALLOWED_PATHS = frozenset(
 #: reference to a task that was still running.
 MAX_IN_FLIGHT = 256
 #: How many of a client's frames are kept for a stream not listening yet.
-MAX_EARLY = 8
+#: One past it closes the stream rather than losing a frame: a live view's
+#: candidates are worth nothing with one missing.
+MAX_EARLY = 64
 #: How often to look for a config that appeared while the engine was up.
 #: Enrolment is something a person does and then waits on, so this is the
 #: delay they sit through — short enough not to be worth a restart.
@@ -299,7 +301,20 @@ class CloudConnector:
                             # A browser speaks as soon as the hub accepts it,
                             # before this end has checked its credential.
                             early = self._early.setdefault(stream, [])
-                            early[:] = [*early[-(MAX_EARLY - 1) :], text]
+                            if len(early) < MAX_EARLY:
+                                early.append(text)
+                            else:
+                                logger.warning(
+                                    "Closing stream %s: %d frames before it listened",
+                                    stream,
+                                    MAX_EARLY,
+                                )
+                                self._cancel_stream(stream)
+                                await socket.send(
+                                    _dump(
+                                        {"op": "ws_close", "id": stream, "code": 1011}
+                                    )
+                                )
                     elif op == "ws_close":
                         self._cancel_stream(str(frame.get("id") or ""))
                     elif op == "member_removed":
@@ -321,6 +336,7 @@ class CloudConnector:
                     task.cancel()
                 self._calls.clear()
                 self._streams.clear()
+                self._early.clear()
         return attached
 
     @staticmethod
@@ -631,12 +647,16 @@ class CloudConnector:
                 "Refusing stream %s: %d already open", stream_id, len(self._streams)
             )
             return
-        existing = self._streams.pop(stream_id, None)
-        if existing is not None:
-            existing.cancel()
+        self._cancel_stream(stream_id)
         task = asyncio.create_task(self._stream(socket, frame))
         self._streams[stream_id] = task
-        task.add_done_callback(lambda _t: self._streams.pop(stream_id, None))
+        task.add_done_callback(lambda done: self._stream_done(stream_id, done))
+
+    def _stream_done(self, stream_id: str, task: asyncio.Task[None]) -> None:
+        # Its own entry only: an id the hub reused belongs to the newer task.
+        if self._streams.get(stream_id) is task:
+            del self._streams[stream_id]
+            self._early.pop(stream_id, None)
 
     async def _stream(self, socket: Any, frame: dict[str, Any]) -> None:
         """Feed the engine's live events down the tunnel.
@@ -661,6 +681,7 @@ class CloudConnector:
         )
         from fluksio.api.socket_auth import redeem_ticket
         from fluksio.core.db import engine
+        from fluksio.flow import live as live_view
         from fluksio.flow.demand import demand
         from fluksio.flow.devices import presence
         from fluksio.flow.events import event_bus
@@ -714,6 +735,11 @@ class CloudConnector:
         # are fetched one by one through the portal.
         viewer = ("tunnel", stream_id)
         wanted: set[str] = set()
+        # Where a live view this stream is one end of reaches it. Its
+        # signalling is text, so it crosses, on an outbox of its own.
+        endpoint = live_view.Endpoint()
+        outbox: asyncio.Task[str] | None = None
+        sender: asyncio.Task[dict[str, Any]] | None = None
 
         def heard(text: str) -> None:
             nonlocal wanted
@@ -721,6 +747,14 @@ class CloudConnector:
             if asked is not None:
                 wanted = asked
                 demand.want(viewer, asked)
+            elif (signal := live_view.parse(text)) is not None:
+                live_view.sessions.heard(
+                    signal,
+                    endpoint,
+                    person=str(user.id) if only is None else "",
+                    device=device,
+                    devices=getattr(self._app.state, "device_store", None),
+                )
 
         self._listeners[stream_id] = heard
         for text in self._early.pop(stream_id, []):
@@ -728,8 +762,10 @@ class CloudConnector:
 
         if device:
             presence.opened(device)
+            live_view.sessions.attach(device, endpoint)
         try:
             async with event_bus.subscribe() as queue:
+                outbox = asyncio.create_task(endpoint.next())
                 if device and (ports := demand.ports(device)) != told:
                     await socket.send(
                         _dump(
@@ -741,7 +777,22 @@ class CloudConnector:
                         )
                     )
                 while True:
-                    event = await queue.get()
+                    sender = asyncio.create_task(queue.get())
+                    done, _ = await asyncio.wait(
+                        {sender, outbox}, return_when=asyncio.FIRST_COMPLETED
+                    )
+                    if outbox in done:
+                        # Signalling, for this stream alone.
+                        if sender not in done:
+                            sender.cancel()
+                        text = outbox.result()
+                        outbox = asyncio.create_task(endpoint.next())
+                        await socket.send(
+                            _dump({"op": "ws_msg", "id": stream_id, "text": text})
+                        )
+                        if sender not in done:
+                            continue
+                    event = sender.result()
                     if event.get("type") == "device_demand":
                         # For the device it names, and for nobody else.
                         if device and event.get("device") == device:
@@ -793,9 +844,13 @@ class CloudConnector:
                     _dump({"op": "ws_close", "id": stream_id, "code": 1011})
                 )
         finally:
+            for task in (outbox, sender):
+                if task is not None:
+                    task.cancel()
             self._listeners.pop(stream_id, None)
             self._early.pop(stream_id, None)
             demand.drop(viewer)
+            live_view.sessions.detach(endpoint)
             if device:
                 presence.closed(device)
 
@@ -867,6 +922,9 @@ class CloudConnector:
                 )
 
     def _cancel_stream(self, stream_id: str) -> None:
+        # The frames it was holding go with it, including those of a stream
+        # cancelled before it ever ran its own cleanup.
+        self._early.pop(stream_id, None)
         task = self._streams.pop(stream_id, None)
         if task is not None:
             task.cancel()

backend/fluksio/core/config.py

@@ -229,6 +229,18 @@ class Settings(BaseSettings):
         list[AnyUrl] | str, BeforeValidator(parse_cors)
     ] = []
 
+    #: Where a live view's two ends look for a way to each other, comma
+    #: separated. None by default: on one network the peers find each other
+    #: without, and a third-party server would learn who watches what. coturn
+    #: serves STUN as well, so its `stun:` address goes here.
+    STUN_URLS: Annotated[list[str] | str, BeforeValidator(parse_cors)] = []
+    #: The relay a live view falls back to when the two ends cannot reach each
+    #: other, as `turn:` and `turns:` URLs. Used only with `TURN_SECRET`, the
+    #: secret the TURN server shares (coturn's `static-auth-secret`), from
+    #: which each viewer and device gets credentials of its own.
+    TURN_URLS: Annotated[list[str] | str, BeforeValidator(parse_cors)] = []
+    TURN_SECRET: str = ""
+
     @model_validator(mode="before")
     @classmethod
     def _derive_data_paths(cls, data: Any) -> Any:

backend/fluksio/flow/live.py

@@ -0,0 +1,365 @@
+"""Live view: WebRTC signalling between a viewer's socket and a device's.
+
+A dashboard tile asks a device's camera for H.264 over WebRTC. The video goes
+from the device to the browser directly, or through a TURN relay, never
+through here: this only carries the few messages that set a connection up.
+Both ends already hold a ``/flows/ws`` socket, which also crosses the portal
+tunnel because it carries text, so signalling rides on it.
+
+A person's socket speaks ``{"type": "live", "device", "session", "action"}``
+and a device's ``{"type": "signal", "session", "action"}``. ``start`` from
+the viewer is handed to the device with ICE servers of its own, the device
+offers (``sdp``), which reaches the viewer with ICE servers of the viewer's
+own, the viewer answers (``sdp``), both trickle ``candidate``, and ``bye``
+from either side ends it. A device past its cap answers ``busy``, and so does
+this end once a device has ``MAX_SESSIONS``.
+
+Signalling never rides the event bus, which drops a slow socket's oldest
+events: one lost candidate is a connection that never comes up. Each socket
+is an :class:`Endpoint` with an outbox of its own that its loop drains, so
+only that socket writes to its connection, and a full outbox ends the session
+instead of losing a message. One process holds every socket, so one registry
+is the whole picture.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import hashlib
+import hmac
+import logging
+import re
+import time
+from dataclasses import dataclass, field
+from typing import Any
+
+import orjson
+
+from fluksio.core.config import settings
+
+logger = logging.getLogger(__name__)
+
+#: The port a device reports when it can stream live. Signalled, never wired.
+LIVE_PORT = "live"
+#: Live viewers per device. The NANO encodes once for all of them, and S3
+#: measured two at 720p; past that a tile stays on frames.
+MAX_SESSIONS = 2
+#: How long a session has to get from ``start`` to the viewer's answer.
+ANSWER_TIMEOUT_S = 10.0
+#: How long TURN credentials stay good. Minted per session, for whoever is
+#: one end of it. coturn checks them on every refresh, so a view relayed
+#: through it ends this long after it started.
+ICE_TTL_S = 3600
+#: Signalling messages one socket may have waiting. A session trickles a
+#: dozen candidates at most, so a socket this far behind is stuck.
+OUTBOX = 64
+
+_SESSION = re.compile(r"^[A-Za-z0-9_-]{8,64}$")
+#: An offer is one to three kilobytes, a candidate a line.
+MAX_SDP = 16 * 1024
+MAX_CANDIDATE = 1024
+#: Which actions are passed on once a session stands.
+_RELAYED = ("sdp", "candidate")
+
+
+def ice_servers(who: str) -> list[dict[str, Any]]:
+    """The STUN and TURN servers for one end, with TURN credentials of its own.
+
+    coturn's REST scheme (``use-auth-secret``): the username is
+    ``<expiry>:<who>`` and the password the base64 HMAC-SHA1 of it under the
+    shared secret, so coturn checks it without asking anyone.
+    """
+    servers: list[dict[str, Any]] = []
+    if settings.STUN_URLS:
+        servers.append({"urls": list(settings.STUN_URLS)})
+    if settings.TURN_URLS and settings.TURN_SECRET:
+        username = f"{int(time.time()) + ICE_TTL_S}:{who}"
+        digest = hmac.new(
+            settings.TURN_SECRET.encode(), username.encode(), hashlib.sha1
+        ).digest()
+        servers.append(
+            {
+                "urls": list(settings.TURN_URLS),
+                "username": username,
+                "credential": base64.b64encode(digest).decode(),
+            }
+        )
+    return servers
+
+
+def parse(frame: str) -> dict[str, Any] | None:
+    """A signalling message a socket sent, or None for anything else."""
+    try:
+        payload = orjson.loads(frame)
+    except orjson.JSONDecodeError:
+        return None
+    if isinstance(payload, dict) and payload.get("type") in ("live", "signal"):
+        return payload
+    return None
+
+
+class Endpoint:
+    """One socket signalling can reach, and what is waiting to go out on it."""
+
+    def __init__(self) -> None:
+        self._outbox: asyncio.Queue[str] = asyncio.Queue(maxsize=OUTBOX)
+
+    def deliver(self, message: dict[str, Any]) -> bool:
+        """Put a message in the outbox. False when it is full."""
+        try:
+            self._outbox.put_nowait(orjson.dumps(message).decode())
+        except asyncio.QueueFull:
+            return False
+        return True
+
+    async def next(self) -> str:
+        """The next message for this socket to send, as text."""
+        return await self._outbox.get()
+
+
+@dataclass(eq=False)
+class Session:
+    id: str
+    device: str
+    viewer: Endpoint
+    camera: Endpoint
+    #: Who is watching, whose TURN credentials the viewer is handed.
+    who: str
+    offered: bool = False
+    answered: bool = False
+    started: float = field(default_factory=time.monotonic)
+    timer: asyncio.TimerHandle | None = None
+
+
+class LiveSessions:
+    """Every live view being set up or running, and the devices that serve them.
+
+    Called only from the event loop, by the sockets themselves.
+    """
+
+    def __init__(self) -> None:
+        self._sessions: dict[str, Session] = {}
+        #: Each device's open sockets, newest last. A device reconnecting can
+        #: hold two for a moment; a new session goes to the newer one.
+        self._devices: dict[str, list[Endpoint]] = {}
+
+    def attach(self, device_id: str, endpoint: Endpoint) -> None:
+        """A device's socket opened."""
+        self._devices.setdefault(device_id, []).append(endpoint)
+
+    def detach(self, endpoint: Endpoint) -> None:
+        """A socket closed: every session it was one end of is over."""
+        for device_id, endpoints in list(self._devices.items()):
+            if endpoint in endpoints:
+                endpoints.remove(endpoint)
+                if not endpoints:
+                    del self._devices[device_id]
+        for session in list(self._sessions.values()):
+            if session.viewer is endpoint:
+                self._end(session, "closed", tell_viewer=False)
+            elif session.camera is endpoint:
+                self._end(session, "offline", tell_camera=False)
+
+    def count(self, device_id: str) -> int:
+        return sum(1 for s in self._sessions.values() if s.device == device_id)
+
+    def heard(
+        self,
+        message: dict[str, Any],
+        endpoint: Endpoint,
+        *,
+        person: str,
+        device: str,
+        devices: Any,
+    ) -> 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.
+        """
+        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)
+        elif message.get("type") == "signal" and device:
+            self._from_device(message, session_id, endpoint, device)
+
+    # -- the two directions ---------------------------------------------------
+
+    def _from_viewer(
+        self,
+        message: dict[str, Any],
+        session_id: str,
+        endpoint: Endpoint,
+        who: str,
+        devices: Any,
+    ) -> None:
+        device_id = str(message.get("device") or "")
+        action = message.get("action")
+        session = self._sessions.get(session_id)
+        if action == "start":
+            if session is not None:
+                # Said twice, or somebody else's id: neither gets a second one.
+                return
+            reason = self._refusal(device_id, devices)
+            if reason:
+                endpoint.deliver(
+                    _viewer_message(device_id, session_id, "bye", reason=reason)
+                )
+                return
+            if self.count(device_id) >= MAX_SESSIONS:
+                endpoint.deliver(_viewer_message(device_id, session_id, "busy"))
+                return
+            session = Session(
+                session_id, device_id, endpoint, self._devices[device_id][-1], who
+            )
+            session.timer = asyncio.get_running_loop().call_later(
+                ANSWER_TIMEOUT_S, self._expire, session
+            )
+            self._sessions[session_id] = session
+            logger.info("Live view %s on %s asked for", session_id, device_id)
+            self._to_camera(
+                session,
+                {
+                    "action": "start",
+                    "ice_servers": ice_servers(f"device-{device_id}"),
+                },
+            )
+            return
+        if (
+            session is None
+            or session.viewer is not endpoint
+            or session.device != device_id
+        ):
+            return
+        if action == "bye":
+            self._end(session, "closed", tell_viewer=False)
+            return
+        relayed = _relayed(message, action)
+        if relayed is None:
+            return
+        if action == "sdp" and session.offered and not session.answered:
+            session.answered = True
+            if session.timer is not None:
+                session.timer.cancel()
+        self._to_camera(session, relayed)
+
+    def _from_device(
+        self,
+        message: dict[str, Any],
+        session_id: str,
+        endpoint: Endpoint,
+        device_id: str,
+    ) -> None:
+        session = self._sessions.get(session_id)
+        if (
+            session is None
+            or session.device != device_id
+            or session.camera is not endpoint
+        ):
+            return
+        action = message.get("action")
+        if action in ("bye", "busy"):
+            reason = str(message.get("reason") or "")[:64] or "closed"
+            self._end(session, reason, tell_camera=False, action=str(action))
+            return
+        relayed = _relayed(message, action)
+        if relayed is None:
+            return
+        if action == "sdp" and not session.offered:
+            # The viewer builds its connection on the offer, so its ICE
+            # servers come with it: minted only for a session that got a slot.
+            session.offered = True
+            relayed["ice_servers"] = ice_servers(session.who)
+        self._to_viewer(session, relayed)
+
+    # -- delivering and ending ------------------------------------------------
+
+    def _to_camera(self, session: Session, message: dict[str, Any]) -> None:
+        sent = {"type": "signal", "session": session.id, **message}
+        if not session.camera.deliver(sent):
+            self._end(session, "overflow", tell_camera=False)
+
+    def _to_viewer(self, session: Session, message: dict[str, Any]) -> None:
+        sent = _viewer_message(session.device, session.id, **message)
+        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 ""."""
+        if not self._devices.get(device_id):
+            return "offline"
+        reported = devices.reported_ports(device_id) if devices is not None else None
+        if LIVE_PORT not in (reported or {}):
+            return "unsupported"
+        return ""
+
+    def _expire(self, session: Session) -> None:
+        if self._sessions.get(session.id) is session and not session.answered:
+            self._end(session, "timeout")
+
+    def _end(
+        self,
+        session: Session,
+        reason: str,
+        *,
+        tell_viewer: bool = True,
+        tell_camera: bool = True,
+        action: str = "bye",
+    ) -> None:
+        if self._sessions.pop(session.id, None) is not session:
+            return
+        if session.timer is not None:
+            session.timer.cancel()
+        # Told once, without a retry: a side whose outbox is full is stuck,
+        # and its own timeout or a closing socket ends it there.
+        if tell_viewer:
+            session.viewer.deliver(
+                _viewer_message(session.device, session.id, action, reason=reason)
+            )
+        if tell_camera:
+            session.camera.deliver(
+                {
+                    "type": "signal",
+                    "session": session.id,
+                    "action": "bye",
+                    "reason": reason,
+                }
+            )
+        logger.info(
+            "Live view %s on %s ended after %.1f s: %s",
+            session.id,
+            session.device,
+            time.monotonic() - session.started,
+            reason,
+        )
+
+
+def _relayed(message: dict[str, Any], action: Any) -> dict[str, Any] | None:
+    """What of an ``sdp`` or ``candidate`` message is passed on, or None."""
+    if action not in _RELAYED:
+        return None
+    value = message.get(action)
+    limit = MAX_SDP if action == "sdp" else MAX_CANDIDATE
+    if not isinstance(value, str) or not value or len(value) > limit:
+        return None
+    return {"action": action, action: value}
+
+
+def _viewer_message(
+    device: str, session: str, action: str, **rest: Any
+) -> dict[str, Any]:
+    return {
+        "type": "live",
+        "device": device,
+        "session": session,
+        "action": action,
+        **rest,
+    }
+
+
+sessions = LiveSessions()

backend/tests/api/routes/test_live.py

@@ -0,0 +1,441 @@
+"""Live view: WebRTC signalling relayed between a viewer's socket and a device's.
+
+openpanel ROADMAP L1. The server routes by session, holds a device to two
+viewers, ends a session when either socket goes or no answer comes, and hands
+out TURN credentials in coturn's REST scheme.
+"""
+
+import asyncio
+import base64
+import hashlib
+import hmac
+import json
+import time
+from typing import Any
+
+import pytest
+from fastapi.testclient import TestClient
+
+from fluksio.api.routes import oauth
+from fluksio.cloud.connector import CloudConnector
+from fluksio.core.config import settings
+from fluksio.flow import live as live_view
+from tests.api.routes.test_cameras import REPORT, wait_for
+from tests.api.routes.test_devices import _pair, _panels
+
+PREFIX = f"{settings.API_V1_STR}/devices"
+SOCKET = f"{settings.API_V1_STR}/flows/ws"
+LIVE_REPORT = {
+    **REPORT,
+    "ports": [
+        *REPORT["ports"],
+        {"port": "live", "direction": "in", "dtype": "video", "on_demand": True},
+    ],
+}
+
+
+@pytest.fixture(scope="module")
+def cams(client: TestClient, superuser_token_headers: dict[str, str]):
+    """``cam1`` reports a live port, ``cam2`` does not. Yields their headers."""
+    for bucket in [b for b in oauth._hits if b.startswith("pair:")]:
+        del oauth._hits[bucket]
+    _panels(
+        client, superuser_token_headers, {"panels": [{"id": "cam1"}, {"id": "cam2"}]}
+    )
+    cam1 = _pair(client, superuser_token_headers, "cam1")
+    cam2 = _pair(client, superuser_token_headers, "cam2")
+    for headers, report in ((cam1, LIVE_REPORT), (cam2, REPORT)):
+        device = "cam1" if headers is cam1 else "cam2"
+        put = client.put(f"{PREFIX}/{device}/report", headers=headers, json=report)
+        assert put.status_code == 200, put.text
+    yield cam1, cam2
+    for device in ("cam1", "cam2"):
+        client.delete(f"{PREFIX}/{device}", headers=superuser_token_headers)
+
+
+def _viewer(client: TestClient, headers: dict[str, str]) -> Any:
+    ticket = client.post(f"{settings.API_V1_STR}/flows/ws-ticket", headers=headers)
+    return client.websocket_connect(f"{SOCKET}?ticket={ticket.json()['ticket']}")
+
+
+def _next(ws: Any, kind: str) -> dict[str, Any]:
+    """The next message of this type on a socket, skipping the rest."""
+    while True:
+        message = ws.receive_json()
+        if message.get("type") == kind:
+            return message
+
+
+def _start(ws: Any, session: str, device: str = "cam1") -> None:
+    ws.send_json(
+        {"type": "live", "device": device, "session": session, "action": "start"}
+    )
+
+
+def test_signalling_is_relayed_both_ways(
+    client: TestClient, superuser_token_headers: dict[str, str], cams
+) -> None:
+    """start, the device's offer, the answer, candidates both ways, bye."""
+    cam1, _ = cams
+    with (
+        client.websocket_connect(SOCKET, headers=cam1) as device,
+        _viewer(client, superuser_token_headers) as viewer,
+    ):
+        device.receive_json()
+        _start(viewer, "session-one")
+        start = _next(device, "signal")
+        assert start["action"] == "start"
+        assert start["session"] == "session-one"
+        assert start["ice_servers"] == []
+
+        device.send_json(
+            {
+                "type": "signal",
+                "session": "session-one",
+                "action": "sdp",
+                "sdp": "v=0 offer",
+            }
+        )
+        assert _next(viewer, "live") == {
+            "type": "live",
+            "device": "cam1",
+            "session": "session-one",
+            "action": "sdp",
+            "sdp": "v=0 offer",
+            "ice_servers": [],
+        }
+        viewer.send_json(
+            {
+                "type": "live",
+                "device": "cam1",
+                "session": "session-one",
+                "action": "sdp",
+                "sdp": "v=0 answer",
+            }
+        )
+        assert _next(device, "signal")["sdp"] == "v=0 answer"
+        viewer.send_json(
+            {
+                "type": "live",
+                "device": "cam1",
+                "session": "session-one",
+                "action": "candidate",
+                "candidate": "candidate:1 1 udp 1 10.0.0.2 5000 typ host",
+            }
+        )
+        assert _next(device, "signal")["candidate"].startswith("candidate:1 ")
+        device.send_json(
+            {
+                "type": "signal",
+                "session": "session-one",
+                "action": "candidate",
+                "candidate": "candidate:2 1 udp 1 10.0.0.3 5000 typ host",
+            }
+        )
+        assert _next(viewer, "live")["candidate"].startswith("candidate:2 ")
+
+        viewer.send_json(
+            {
+                "type": "live",
+                "device": "cam1",
+                "session": "session-one",
+                "action": "bye",
+            }
+        )
+        bye = _next(device, "signal")
+        assert (bye["action"], bye["session"]) == ("bye", "session-one")
+        assert live_view.sessions.count("cam1") == 0
+
+
+def test_a_third_viewer_is_busy_and_closing_ends_the_sessions(
+    client: TestClient, superuser_token_headers: dict[str, str], cams
+) -> None:
+    cam1, _ = cams
+    with client.websocket_connect(SOCKET, headers=cam1) as device:
+        device.receive_json()
+        with _viewer(client, superuser_token_headers) as viewer:
+            _start(viewer, "session-a")
+            _start(viewer, "session-b")
+            _start(viewer, "session-c")
+            busy = _next(viewer, "live")
+            assert (busy["session"], busy["action"]) == ("session-c", "busy")
+            asked = {_next(device, "signal")["session"] for _ in range(2)}
+            assert asked == {"session-a", "session-b"}
+        # The viewer's socket closed: the device hears bye for both.
+        ended = {_next(device, "signal")["session"] for _ in range(2)}
+        assert ended == {"session-a", "session-b"}
+        wait_for(lambda: live_view.sessions.count("cam1") == 0)
+
+    with _viewer(client, superuser_token_headers) as viewer:
+        with client.websocket_connect(SOCKET, headers=cam1) as device:
+            device.receive_json()
+            _start(viewer, "session-d")
+            _next(device, "signal")
+            # The device answers busy itself, past its own cap.
+            device.send_json(
+                {"type": "signal", "session": "session-d", "action": "busy"}
+            )
+            assert _next(viewer, "live")["action"] == "busy"
+
+            _start(viewer, "session-e")
+            _next(device, "signal")
+        # The device's socket closed.
+        bye = _next(viewer, "live")
+        assert (bye["session"], bye["action"], bye["reason"]) == (
+            "session-e",
+            "bye",
+            "offline",
+        )
+
+
+def test_a_session_without_an_answer_times_out(
+    client: TestClient,
+    superuser_token_headers: dict[str, str],
+    cams,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(live_view, "ANSWER_TIMEOUT_S", 0.2)
+    cam1, _ = cams
+    with (
+        client.websocket_connect(SOCKET, headers=cam1) as device,
+        _viewer(client, superuser_token_headers) as viewer,
+    ):
+        device.receive_json()
+        _start(viewer, "session-slow")
+        _next(device, "signal")
+        bye = _next(viewer, "live")
+        assert (bye["action"], bye["reason"]) == ("bye", "timeout")
+        told = _next(device, "signal")
+        assert (told["action"], told["reason"]) == ("bye", "timeout")
+
+
+def test_only_a_person_starts_and_only_the_device_speaks_for_its_session(
+    client: TestClient, superuser_token_headers: dict[str, str], cams
+) -> None:
+    cam1, cam2 = cams
+    with _viewer(client, superuser_token_headers) as viewer:
+        # Not connected, then connected without a live port.
+        _start(viewer, "session-off")
+        assert _next(viewer, "live")["reason"] == "offline"
+        with client.websocket_connect(SOCKET, headers=cam2) as other:
+            other.receive_json()
+            _start(viewer, "session-old", device="cam2")
+            assert _next(viewer, "live")["reason"] == "unsupported"
+
+            with client.websocket_connect(SOCKET, headers=cam1) as device:
+                device.receive_json()
+                # A device's credential cannot start one.
+                _start(other, "session-dev")
+                _start(viewer, "session-ok")
+                assert _next(device, "signal")["session"] == "session-ok"
+                assert live_view.sessions.count("cam1") == 1
+
+                # Another device cannot speak for it.
+                other.send_json(
+                    {
+                        "type": "signal",
+                        "session": "session-ok",
+                        "action": "sdp",
+                        "sdp": "x",
+                    }
+                )
+                time.sleep(0.1)
+                device.send_json(
+                    {
+                        "type": "signal",
+                        "session": "session-ok",
+                        "action": "sdp",
+                        "sdp": "y",
+                    }
+                )
+                assert _next(viewer, "live")["sdp"] == "y"
+
+                # Nor can another viewer, for a session it does not hold.
+                with _viewer(client, superuser_token_headers) as stranger:
+                    stranger.send_json(
+                        {
+                            "type": "live",
+                            "device": "cam1",
+                            "session": "session-ok",
+                            "action": "bye",
+                        }
+                    )
+                    time.sleep(0.1)
+                assert live_view.sessions.count("cam1") == 1
+
+
+def test_ice_servers_carry_coturn_credentials(
+    client: TestClient,
+    superuser_token_headers: dict[str, str],
+    cams,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """``<expiry>:<who>``, and the base64 HMAC-SHA1 of it under the secret."""
+    monkeypatch.setattr(settings, "STUN_URLS", ["stun:turn.example.com:3478"])
+    monkeypatch.setattr(
+        settings,
+        "TURN_URLS",
+        ["turn:turn.example.com:3478", "turns:turn.example.com:5349?transport=tcp"],
+    )
+    monkeypatch.setattr(settings, "TURN_SECRET", "s3cret")
+    cam1, _ = cams
+
+    def check(server: dict[str, Any], who: str) -> None:
+        expiry, _, name = server["username"].partition(":")
+        assert name == who
+        assert 0 < int(expiry) - time.time() <= live_view.ICE_TTL_S
+        mac = hmac.new(b"s3cret", server["username"].encode(), hashlib.sha1)
+        assert server["credential"] == base64.b64encode(mac.digest()).decode()
+
+    answer = client.get(f"{PREFIX}/cam1/ice", headers=superuser_token_headers)
+    assert answer.status_code == 200, answer.text
+    stun, turn = answer.json()["ice_servers"]
+    assert stun == {"urls": ["stun:turn.example.com:3478"]}
+    me = client.get(f"{settings.API_V1_STR}/users/me", headers=superuser_token_headers)
+    check(turn, me.json()["id"])
+
+    assert (
+        client.get(f"{PREFIX}/nope/ice", headers=superuser_token_headers).status_code
+        == 404
+    )
+    # A device gets its servers in `start`, not here.
+    assert client.get(f"{PREFIX}/cam1/ice", headers=cam1).status_code == 403
+
+    with (
+        client.websocket_connect(SOCKET, headers=cam1) as device,
+        _viewer(client, superuser_token_headers) as viewer,
+    ):
+        device.receive_json()
+        _start(viewer, "session-ice")
+        check(_next(device, "signal")["ice_servers"][1], "device-cam1")
+        # The viewer's come with the offer, minted for this session's user.
+        device.send_json(
+            {"type": "signal", "session": "session-ice", "action": "sdp", "sdp": "v=0"}
+        )
+        check(_next(viewer, "live")["ice_servers"][1], me.json()["id"])
+
+
+def test_a_full_outbox_ends_the_session_instead_of_dropping() -> None:
+    """A socket that stops reading loses its session, never a message."""
+
+    class Devices:
+        def reported_ports(self, _device: str) -> dict:
+            return {"live": {}}
+
+    async def run() -> None:
+        registry = live_view.LiveSessions()
+        camera, viewer = live_view.Endpoint(), live_view.Endpoint()
+        registry.attach("cam9", camera)
+        registry.heard(
+            {
+                "type": "live",
+                "device": "cam9",
+                "session": "s-overflow",
+                "action": "start",
+            },
+            viewer,
+            person="someone",
+            device="",
+            devices=Devices(),
+        )
+        assert registry.count("cam9") == 1
+        # The viewer never reads, and the device trickles on.
+        for index in range(live_view.OUTBOX + 1):
+            registry.heard(
+                {
+                    "type": "signal",
+                    "session": "s-overflow",
+                    "action": "candidate",
+                    "candidate": f"candidate:{index} 1 udp 1 10.0.0.9 5000 typ host",
+                },
+                camera,
+                person="",
+                device="cam9",
+                devices=None,
+            )
+        assert registry.count("cam9") == 0
+        start, bye = [json.loads(await camera.next()) for _ in range(2)]
+        assert start["action"] == "start"
+        assert (bye["action"], bye["reason"]) == ("bye", "overflow")
+
+    asyncio.run(run())
+
+
+def test_the_tunnel_carries_signalling(
+    client: TestClient, superuser_token_headers: dict[str, str], cams
+) -> None:
+    """A viewer through the portal, a device on this network, and back."""
+    cam1, _ = cams
+    ticket = client.post(
+        f"{settings.API_V1_STR}/flows/ws-ticket", headers=superuser_token_headers
+    ).json()["ticket"]
+    connector = CloudConnector(client.app)
+    sent: list[dict[str, Any]] = []
+
+    class Socket:
+        async def send(self, text: str) -> None:
+            sent.append(json.loads(text))
+
+    def said() -> list[dict[str, Any]]:
+        return [
+            json.loads(frame["text"])
+            for frame in sent
+            if frame["op"] == "ws_msg" and '"live"' in frame["text"]
+        ]
+
+    frame = {"id": "t1", "path": "/api/v1/flows/ws", "query": f"ticket={ticket}"}
+    assert client.portal is not None
+    with client.websocket_connect(SOCKET, headers=cam1) as device:
+        device.receive_json()
+        stream = client.portal.start_task_soon(connector._stream, Socket(), frame)
+        try:
+            wait_for(lambda: "t1" in connector._listeners)
+            hear = connector._listeners["t1"]
+            # The stream subscribes to the bus once it has said its snapshot.
+            wait_for(lambda: any(f["op"] == "ws_msg" for f in sent))
+            time.sleep(0.1)
+            client.portal.call(
+                hear,
+                json.dumps(
+                    {
+                        "type": "live",
+                        "device": "cam1",
+                        "session": "tunnel-1",
+                        "action": "start",
+                    }
+                ),
+            )
+            assert _next(device, "signal")["session"] == "tunnel-1"
+            device.send_json(
+                {"type": "signal", "session": "tunnel-1", "action": "sdp", "sdp": "v=0"}
+            )
+            offer = wait_for(lambda: said())[0]
+            assert (offer["session"], offer["sdp"]) == ("tunnel-1", "v=0")
+        finally:
+            stream.cancel()
+            wait_for(stream.done)
+        # The stream went: the device hears bye.
+        assert _next(device, "signal")["action"] == "bye"
+
+
+def test_frames_held_for_a_stream_go_with_it(client: TestClient) -> None:
+    """A stream refused before it listened leaves nothing behind."""
+    connector = CloudConnector(client.app)
+    sent: list[str] = []
+
+    class Socket:
+        async def send(self, text: str) -> None:
+            sent.append(text)
+
+    frame = {"id": "t2", "path": "/api/v1/flows/ws", "query": "ticket=spent"}
+    assert client.portal is not None
+
+    def open_and_speak() -> None:
+        connector._start_stream(Socket(), frame)
+        # What the link does with a frame for a stream not listening yet,
+        # here before the stream has even started to run.
+        connector._early["t2"] = ['{"type": "media", "names": []}']
+
+    client.portal.call(open_and_speak)
+    wait_for(lambda: "t2" not in connector._streams)
+    assert "t2" not in connector._early