Bound a panel credential where the route check cannot reach

Three things the security pass on the portal pairing turned up. The first two
were already true of a screen on the local network; what changed is that a
panel credential is now presentable from the internet, which is what makes
them worth closing rather than recording.

The artifact endpoint authenticates for itself, because a worker's credential
has to open it and that token is no use anywhere else. It resolved the caller
without handing over the request, so the one credential that is scoped by
route was judged by no route at all — a panel could read and write the store
as whoever approved it. It passes the request it already holds now.

The websocket has no route to judge either, and there the bound has to be on
what is sent: a panel is given the values its own dashboards draw and nothing
else — no node status, no logs, no shape of the graph. The keys stay in the
message, emptied, because a screen on a wall runs the bundle it was paired
with. `messages_for` reads that set off the published dashboards, and is the
walk the `/messages/` allowlist has wanted for a while.

And locality is no longer a header anyone can type. The marker the connector
stamps is a value minted per process, so reaching this API directly cannot buy
a device the credential meant for one that cannot reach it at all.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017F9RnYCJgASuBTcAjxmnsp

backend/app/api/deps.py

@@ -142,13 +142,18 @@ def _gate_panel(payload: dict[str, Any], request: Request | None) -> dict[str, A
     return payload
 
 
-def user_from_token(session: Session, token: str) -> User | None:
+def user_from_token(
+    session: Session, token: str, request: Request | None = None
+) -> User | None:
     """Resolve a bearer token to its user, or None if it does not hold up.
 
-    Shared with the websocket, which cannot use the HTTP security scheme.
+    Shared with the websocket, which cannot use the HTTP security scheme, and
+    with the artifact endpoint, which accepts a worker's credential as well as
+    a person's. Pass the request wherever there is one: a panel's credential is
+    scoped by route, and without it the scope check cannot run.
     """
     try:
-        token_data = TokenPayload(**decode_token(token))
+        token_data = TokenPayload(**decode_token(token, request))
     except (InvalidTokenError, ValidationError):
         return None
     user = session.get(User, token_data.sub)

backend/app/api/routes/artifacts.py

@@ -38,7 +38,9 @@ def artifact_caller(request: Request) -> str:
     else:
         return f"worker:{claims.get('sub')}"
     with Session(engine) as session:
-        user = user_from_token(session, token)
+        # With the request, so a credential that is scoped by route — a wall
+        # panel's — is judged against this one rather than waved through.
+        user = user_from_token(session, token, request)
     if user is None:
         raise HTTPException(status_code=401, detail="Not authenticated")
     return user.email

backend/app/api/routes/flows.py

@@ -13,12 +13,14 @@ from fastapi import (
     WebSocketDisconnect,
 )
 from fastapi.concurrency import run_in_threadpool
+from jwt.exceptions import InvalidTokenError
 from pydantic import BaseModel
 from sqlmodel import Session
 
 from app.api.deps import (
     CurrentUser,
     FlowControllerDep,
+    decode_token,
     get_current_user,
     user_from_token,
 )
@@ -27,6 +29,7 @@ from app.flow.controller import FlowController
 from app.flow.dashboards import DashboardStore
 from app.flow.events import event_bus
 from app.flow.messages import qualify
+from app.flow.panels import messages_for
 from app.flow.pipeline import ValidationIssue
 from app.flow.runs import RunRejected
 from app.flow.schemas import (
@@ -730,14 +733,34 @@ def read_message_history(
 # -----------------------------------------------------------------------------
 
 
-def snapshot_payload(controller: FlowController) -> dict[str, Any]:
+def snapshot_payload(
+    controller: FlowController, only: set[str] | None = None
+) -> dict[str, Any]:
     """Everything a client needs to catch up, sent the moment it connects.
 
     Also sent by the tunnel connector, which serves this websocket inline: the
     two have to agree, so they build the message here rather than each their
     own. ``emits`` is what the bus counted while nobody was listening — a
     client that reconnects between two pages would otherwise start from zero.
+
+    ``only`` bounds it to a set of message names, which is what a wall panel
+    gets: the values its own dashboards draw, and none of the rest — no node
+    status, no logs, no shape of the graph. A panel renders none of that, and
+    a screen may be hanging somewhere nobody here can see.
     """
+    if only is not None:
+        # The same keys, emptied rather than dropped: a screen already hanging
+        # runs whatever bundle it was paired with, and the shape of this
+        # message is what that bundle reads.
+        return {
+            "type": "snapshot",
+            "values": {k: v for k, v in controller.values().items() if k in only},
+            "nodes": [],
+            "issues": [],
+            "paused": [],
+            "logs": [],
+            "emits": {},
+        }
     return {
         "type": "snapshot",
         "values": controller.values(),
@@ -749,6 +772,32 @@ def snapshot_payload(controller: FlowController) -> dict[str, Any]:
     }
 
 
+def panel_scope(token: str, app: Any) -> set[str] | None:
+    """The messages this credential is bounded to, or None if it is a person's.
+
+    The socket is the one authenticated surface the route check cannot reach —
+    a handshake has no route to judge — so a panel is bounded by what it is
+    sent instead of by what it asks for.
+    """
+    try:
+        panel = str(decode_token(token).get("panel") or "")
+    except InvalidTokenError:
+        return None
+    if not panel:
+        return None
+    store: DashboardStore | None = getattr(app.state, "dashboard_store", None)
+    return messages_for(panel, store) if store is not None else set()
+
+
+def event_for_panel(event: dict[str, Any], only: set[str]) -> bool:
+    """Whether a panel's socket should carry this event.
+
+    The same bound as the snapshot above, applied to the stream that follows
+    it: a value the panel draws, and nothing else on the bus.
+    """
+    return event.get("type") == "message_value" and str(event.get("name") or "") in only
+
+
 @ws_router.websocket("/ws")
 async def flow_events(websocket: WebSocket, token: str = "") -> None:
     """Stream values, node status and execution events as they happen.
@@ -761,6 +810,7 @@ async def flow_events(websocket: WebSocket, token: str = "") -> None:
     if user is None:
         await websocket.close(code=1008)
         return
+    only = panel_scope(token, websocket.app)
 
     await websocket.accept()
 
@@ -768,7 +818,7 @@ async def flow_events(websocket: WebSocket, token: str = "") -> None:
         websocket.app.state, "flow_controller", None
     )
     if controller is not None:
-        await websocket.send_json(snapshot_payload(controller))
+        await websocket.send_json(snapshot_payload(controller, only))
 
     async with event_bus.subscribe() as queue:
         receiver = asyncio.create_task(websocket.receive_text())
@@ -782,7 +832,10 @@ async def flow_events(websocket: WebSocket, token: str = "") -> None:
                     # The client went away.
                     sender.cancel()
                     break
-                await websocket.send_json(sender.result())
+                event = sender.result()
+                if only is not None and not event_for_panel(event, only):
+                    continue
+                await websocket.send_json(event)
         except WebSocketDisconnect:
             pass
         finally:

backend/app/api/routes/panels.py

@@ -236,7 +236,9 @@ def start_pairing(request: Request) -> Any:
     while code in _pending:
         code = "".join(secrets.choice(CODE_ALPHABET) for _ in range(CODE_LENGTH))
 
-    remote = request.headers.get("x-fluksio-via") == "portal"
+    remote = secrets.compare_digest(
+        request.headers.get(cloud_config.VIA_HEADER, ""), cloud_config.VIA_PORTAL
+    )
     entry = _Pending(secrets.token_urlsafe(16), _describe(request), remote)
     _pending[code] = entry
     return PairStarted(code=code, secret=entry.secret)

backend/app/cloud/config.py

@@ -17,6 +17,7 @@ from __future__ import annotations
 import json
 import logging
 import os
+import secrets
 from dataclasses import dataclass
 from typing import Any
 
@@ -27,6 +28,15 @@ from app.core.config import settings
 
 logger = logging.getLogger(__name__)
 
+#: Marks a request the connector replayed off the tunnel. The value is minted
+#: per process and never leaves it, because the header itself is not evidence:
+#: anything that can reach this API directly can set one, and the difference
+#: decides whether a pairing device is handed a credential the whole internet
+#: can present. The connector overwrites it on every frame, so a browser
+#: sending its own gets nowhere from either direction.
+VIA_HEADER = "x-fluksio-via"
+VIA_PORTAL = secrets.token_urlsafe(16)
+
 
 @dataclass(frozen=True)
 class CloudConfig:

backend/app/cloud/connector.py

@@ -222,8 +222,9 @@ class CloudConnector:
             # Set here rather than trusted from the frame: a browser can send
             # any header it likes through the proxy, and this one decides where
             # a pairing device's credential is minted. Arriving on this socket
-            # is the only thing that makes it true.
-            headers["x-fluksio-via"] = "portal"
+            # is the only thing that makes it true, and the value is this
+            # process's own so nothing off the network can imitate it.
+            headers[cloud_config.VIA_HEADER] = cloud_config.VIA_PORTAL
 
             async with self._client() as client:
                 request = client.build_request(
@@ -298,7 +299,11 @@ class CloudConnector:
         from sqlmodel import Session
 
         from app.api.deps import user_from_token
-        from app.api.routes.flows import snapshot_payload
+        from app.api.routes.flows import (
+            event_for_panel,
+            panel_scope,
+            snapshot_payload,
+        )
         from app.core.db import engine
         from app.flow.events import event_bus
 
@@ -316,6 +321,7 @@ class CloudConnector:
                     _dump({"op": "ws_close", "id": stream_id, "code": 1008})
                 )
                 return
+        only = panel_scope(token, self._app)
 
         await socket.send(_dump({"op": "ws_open_ok", "id": stream_id}))
 
@@ -326,7 +332,7 @@ class CloudConnector:
                     {
                         "op": "ws_msg",
                         "id": stream_id,
-                        "text": _dump(snapshot_payload(controller)),
+                        "text": _dump(snapshot_payload(controller, only)),
                     }
                 )
             )
@@ -335,6 +341,8 @@ class CloudConnector:
             async with event_bus.subscribe() as queue:
                 while True:
                     event = await queue.get()
+                    if 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)})
                     )

backend/app/flow/panels.py

@@ -17,6 +17,7 @@ from pathlib import Path
 from pydantic import BaseModel, Field, field_validator
 
 from app.core.config import settings
+from app.flow.dashboards import DashboardNotFound, DashboardStore
 from app.flow.schemas import _validate_name
 
 
@@ -75,3 +76,26 @@ def find(panel_id: str) -> PanelDef | None:
         if panel.id == panel_id:
             return panel
     return None
+
+
+def messages_for(panel_id: str, store: DashboardStore) -> set[str]:
+    """Every message the widgets of this panel's dashboards read or write.
+
+    What a screen is entitled to see, as its own dashboards define it. Read
+    from the published documents, since that is what a panel draws, and empty
+    for a panel that is gone — which is the same answer as "nothing".
+    """
+    panel = find(panel_id)
+    if panel is None:
+        return set()
+    names: set[str] = set()
+    for dashboard in panel.dashboards:
+        try:
+            defn = store.read(dashboard)
+        except DashboardNotFound:
+            continue
+        for widget in defn.widgets:
+            names.update(widget.messages)
+            if widget.target:
+                names.add(widget.target)
+    return names