Authenticate sockets by header or one-time ticket instead of ?token=

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013GAeg4Jo6xjZzjwy9QDWMQ

backend/fluksio/api/deps.py

@@ -198,6 +198,9 @@ def _panel_may(payload: dict[str, Any], request: Request) -> None:
         allowed = "/" not in name and name in panel.dashboards and published
     elif method == "GET" and path == f"{api}/panels/{panel.id}":
         allowed = True
+    elif method == "POST" and path == f"{api}/flows/ws-ticket":
+        # How its live socket opens; the socket applies the panel's own scope.
+        allowed = True
     elif path.startswith(f"{api}/messages/"):
         # The two a widget speaks: a chart reads a series, a control puts a
         # value in. Percent-decoded, because a message name is a path segment

backend/fluksio/api/routes/flows.py

@@ -23,11 +23,13 @@ from fluksio.api.deps import (
     CurrentUser,
     FlowControllerDep,
     SessionDep,
+    TokenDep,
     decode_token,
     get_code_author,
     get_current_user,
     user_from_token,
 )
+from fluksio.api.socket_auth import TICKET_TTL_S, issue_ticket, socket_token
 from fluksio.core.db import engine
 from fluksio.flow.artifacts import is_reference
 from fluksio.flow.controller import NODE_TYPES, FlowController
@@ -66,8 +68,8 @@ from fluksio.models import Flavor, Message, Run
 router = APIRouter(
     prefix="/flows", tags=["flows"], dependencies=[Depends(get_current_user)]
 )
-# The websocket authenticates from its query string, so it stays off the
-# router that enforces the HTTP bearer scheme.
+# The websocket authenticates from its handshake, a header or a ticket, so it
+# stays off the router that enforces the HTTP bearer scheme.
 ws_router = APIRouter(prefix="/flows", tags=["flows"])
 
 
@@ -1061,13 +1063,31 @@ async def _send(websocket: WebSocket, events: list[dict[str, Any]]) -> None:
     await websocket.send_text(orjson.dumps(payload).decode())
 
 
+class SocketTicket(BaseModel):
+    ticket: str
+    expires_in: float
+
+
+@router.post("/ws-ticket", response_model=SocketTicket)
+def issue_socket_ticket(token: TokenDep) -> Any:
+    """A one-time pass for opening `/flows/ws`, good for thirty seconds.
+
+    A browser cannot set a header on a websocket handshake, and a token in the
+    URL ends up in every access log on the way. The ticket goes there instead
+    and stands for the bearer token of this call, panel credentials included.
+    """
+    return SocketTicket(ticket=issue_ticket(token), expires_in=TICKET_TTL_S)
+
+
 @ws_router.websocket("/ws")
-async def flow_events(websocket: WebSocket, token: str = "") -> None:
+async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -> None:
     """Stream values, node status and execution events as they happen.
 
-    The token goes in the query string because browsers cannot set headers on
-    a websocket handshake.
+    A browser brings a ticket from `POST /flows/ws-ticket`; anything that can
+    set headers sends `Authorization: Bearer`. `?token=` is still read for one
+    release, for clients from before either.
     """
+    token = socket_token(websocket.headers, ticket, token)
 
     # On a thread: authenticating is a database round trip, and the snapshot
     # below reads the whole of state. Neither belongs on the event loop, which

backend/fluksio/api/routes/workers.py

@@ -35,6 +35,7 @@ from fluksio.api.deps import (
     get_current_user,
     worker_claims,
 )
+from fluksio.api.socket_auth import socket_token
 from fluksio.core import security
 from fluksio.flow.remote import (
     PROTOCOL,
@@ -240,9 +241,11 @@ def read_runtime() -> str:
 async def attach(websocket: WebSocket, token: str = "") -> None:
     """A worker's connection, for as long as it holds.
 
-    The token goes in the query string for the same reason the dashboard's
-    does: a websocket handshake carries no headers of its own.
+    The token comes as `Authorization: Bearer` on the handshake. `?token=` is
+    still read for one release, so the engine can be upgraded before its
+    workers.
     """
+    token = socket_token(websocket.headers, token=token)
     try:
         claims = await run_in_threadpool(worker_claims, token)
     except InvalidTokenError:

backend/fluksio/api/socket_auth.py

@@ -0,0 +1,60 @@
+"""Where a websocket's credential comes from, and how it stays out of the URL.
+
+A client that can set headers sends `Authorization: Bearer` on the handshake:
+the worker and the SDK. A browser cannot, so it trades its token for a ticket
+over an authenticated REST call and brings the ticket instead. A ticket is good
+once, for thirty seconds, so the URL every proxy logs is worth nothing after.
+
+The engine is one process, so the tickets live in its memory: the tunnel
+connector, which serves the live socket inline, redeems them from the same
+place.
+"""
+
+from __future__ import annotations
+
+import secrets
+import threading
+from collections.abc import Mapping
+from time import monotonic
+
+#: How long a ticket waits to be used: the length of one connect.
+TICKET_TTL_S = 30.0
+
+_tickets: dict[str, tuple[str, float]] = {}
+_lock = threading.Lock()
+
+
+def issue_ticket(token: str) -> str:
+    """A one-time ticket standing in for *token*."""
+    ticket = secrets.token_urlsafe(32)
+    now = monotonic()
+    with _lock:
+        # Swept here rather than on a timer: issuing is what fills the table.
+        for stale in [key for key, (_, due) in _tickets.items() if due <= now]:
+            del _tickets[stale]
+        _tickets[ticket] = (token, now + TICKET_TTL_S)
+    return ticket
+
+
+def redeem_ticket(ticket: str) -> str:
+    """The token a ticket stands for, once; empty if it is unknown or stale."""
+    with _lock:
+        entry = _tickets.pop(ticket, None)
+    if entry is None or entry[1] <= monotonic():
+        return ""
+    return entry[0]
+
+
+def socket_token(headers: Mapping[str, str], ticket: str = "", token: str = "") -> str:
+    """The credential a handshake carries: a header, then a ticket, then a token.
+
+    `token` is the query parameter clients sent before the other two existed.
+    It is read for one more release so an engine can be upgraded before its
+    workers; remove it after that (see NOTEPAD).
+    """
+    header = headers.get("authorization") or ""
+    if header.lower().startswith("bearer "):
+        return header[7:]
+    if ticket:
+        return redeem_ticket(ticket)
+    return token

backend/fluksio/cloud/connector.py

@@ -522,13 +522,18 @@ class CloudConnector:
             panel_scope,
             snapshot_payload,
         )
+        from fluksio.api.socket_auth import redeem_ticket
         from fluksio.core.db import engine
         from fluksio.flow.events import event_bus
 
         stream_id = str(frame["id"])
         path = str(frame.get("path") or "")
         query = str(frame.get("query") or "")
-        token = _query_value(query, "token")
+        # The browser's ticket when it brought one; the hub restates the
+        # session it checked as `token` either way, which is the tunnel's own
+        # channel and stays when the public `?token=` goes.
+        ticket = _query_value(query, "ticket")
+        token = redeem_ticket(ticket) if ticket else _query_value(query, "token")
 
         if not path.startswith(ALLOWED_PREFIX) or not path.endswith("/flows/ws"):
             await socket.send(_dump({"op": "ws_close", "id": stream_id, "code": 1008}))

backend/fluksio/sdk/stream.py

@@ -27,15 +27,15 @@ BACKOFF = (1.0, 5.0, 30.0, 60.0)
 POLL_S = 1.0
 
 
-def socket_url(url: str, token: str) -> str:
+def socket_url(url: str) -> str:
     """The websocket beside an engine's HTTP address.
 
-    The token goes in the query string because a handshake carries no headers
-    of its own — the same reason the browser sends it that way.
+    The token is not in it: it goes as a header on the handshake, since a URL
+    ends up in access logs.
     """
     parts = urlsplit(url)
     scheme = "wss" if parts.scheme == "https" else "ws"
-    return urlunsplit((scheme, parts.netloc, "/api/v1/flows/ws", f"token={token}", ""))
+    return urlunsplit((scheme, parts.netloc, "/api/v1/flows/ws", "", ""))
 
 
 def subscribe(
@@ -61,7 +61,11 @@ def subscribe(
     attempt = 0
     while not stop.is_set():
         try:
-            with connect(socket_url(url, token), open_timeout=POLL_S * 5) as socket:
+            with connect(
+                socket_url(url),
+                additional_headers={"Authorization": f"Bearer {token}"},
+                open_timeout=POLL_S * 5,
+            ) as socket:
                 attempt = 0
                 while not stop.is_set():
                     try:

backend/tests/api/routes/test_flows.py

@@ -521,6 +521,7 @@ async def test_the_live_socket_ends_when_its_client_goes(
 
         def __init__(self) -> None:
             self.app = client.app
+            self.headers: dict[str, str] = {}
             self.frames = ['{"type": "media", "names": []}']
             self.sent: list[str] = []
 

backend/tests/api/routes/test_panels.py

@@ -892,6 +892,18 @@ def test_a_panels_socket_carries_only_what_it_draws(
     # A person's credential is not bounded at all.
     assert panel_scope(superuser_token_headers["Authorization"][7:], client.app) is None
 
+    # It opens the socket with a ticket like any browser, and the ticket
+    # carries the panel's bound rather than widening it.
+    ticket = client.post(
+        f"{settings.API_V1_STR}/flows/ws-ticket",
+        headers={"Authorization": f"Bearer {token}"},
+    )
+    assert ticket.status_code == 200, ticket.text
+    with client.websocket_connect(
+        f"{settings.API_V1_STR}/flows/ws?ticket={ticket.json()['ticket']}"
+    ) as ws:
+        assert ws.receive_json()["nodes"] == []
+
 
 def test_a_panel_fetches_only_the_media_its_tiles_are_showing(
     client: TestClient, superuser_token_headers: dict[str, str]

backend/tests/api/routes/test_socket_auth.py

@@ -0,0 +1,72 @@
+"""A socket's credential comes as a header or a one-time ticket, not in its URL."""
+
+from time import monotonic
+
+import pytest
+from fastapi.testclient import TestClient
+from starlette.websockets import WebSocketDisconnect
+
+from fluksio.core.config import settings
+from fluksio.flow.remote import PROTOCOL
+
+FLOWS = f"{settings.API_V1_STR}/flows"
+ATTACH = f"{settings.API_V1_STR}/workers/attach"
+
+
+def ticket(client: TestClient, headers: dict[str, str]) -> str:
+    answer = client.post(f"{FLOWS}/ws-ticket", headers=headers)
+    assert answer.status_code == 200, answer.text
+    return str(answer.json()["ticket"])
+
+
+def opens(client: TestClient, url: str, headers: dict[str, str] | None = None) -> bool:
+    try:
+        with client.websocket_connect(url, headers=headers or {}) as ws:
+            assert ws.receive_json()["type"] == "snapshot"
+    except WebSocketDisconnect:
+        return False
+    return True
+
+
+def test_the_live_socket_takes_a_header_and_still_a_token(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    assert opens(client, f"{FLOWS}/ws", superuser_token_headers)
+    token = superuser_token_headers["Authorization"][7:]
+    assert opens(client, f"{FLOWS}/ws?token={token}")
+    assert not opens(client, f"{FLOWS}/ws")
+
+
+def test_a_ticket_opens_the_socket_once(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    once = ticket(client, superuser_token_headers)
+    assert opens(client, f"{FLOWS}/ws?ticket={once}")
+    assert not opens(client, f"{FLOWS}/ws?ticket={once}")
+
+
+def test_a_ticket_lapses_after_thirty_seconds(
+    client: TestClient,
+    superuser_token_headers: dict[str, str],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    stale = ticket(client, superuser_token_headers)
+    later = monotonic() + 31
+    monkeypatch.setattr("fluksio.api.socket_auth.monotonic", lambda: later)
+    assert not opens(client, f"{FLOWS}/ws?ticket={stale}")
+
+
+def test_a_worker_attaches_with_a_header_and_still_a_token(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    token = client.post(
+        f"{settings.API_V1_STR}/workers/tokens",
+        headers=superuser_token_headers,
+        json={"name": "header-box"},
+    ).json()["token"]
+    hello = {"op": "hello", "protocol": PROTOCOL}
+    bearer = {"Authorization": f"Bearer {token}"}
+    for url, headers in ((ATTACH, bearer), (f"{ATTACH}?token={token}", {})):
+        with client.websocket_connect(url, headers=headers) as ws:
+            ws.send_json(hello)
+            assert ws.receive_json()["op"] == "welcome"

backend/tests/test_cli.py

@@ -1059,11 +1059,9 @@ def test_a_duration_reads_the_way_an_age_does() -> None:
 
 
 def test_the_dashboard_subscribes_where_the_browser_does() -> None:
-    """The same socket, and the token where a handshake can carry it."""
+    """The same socket, with the token kept out of its URL."""
     from fluksio.sdk.stream import socket_url
 
-    assert socket_url("http://127.0.0.1:8000", "abc") == (
-        "ws://127.0.0.1:8000/api/v1/flows/ws?token=abc"
-    )
+    assert socket_url("http://127.0.0.1:8000") == "ws://127.0.0.1:8000/api/v1/flows/ws"
     # TLS on the one side is TLS on the other.
-    assert socket_url("https://engine.example.com", "t").startswith("wss://")
+    assert socket_url("https://engine.example.com").startswith("wss://")

backend/tests/test_cloud.py

@@ -579,3 +579,35 @@ def test_a_member_the_portal_vouches_for_is_adopted(
         "fluksio.cloud.enroll.httpx.get", side_effect=httpx.ConnectError("no route")
     ):
         assert user_from_token(db, stranger) is None
+
+
+def test_the_tunnel_opens_the_live_socket_with_a_ticket(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    """A browser under the portal brings a ticket, which the engine redeems."""
+    import json
+
+    from fluksio.cloud.connector import CloudConnector
+
+    ticket = client.post(
+        f"{settings.API_V1_STR}/flows/ws-ticket", headers=superuser_token_headers
+    ).json()["ticket"]
+    connector = CloudConnector(client.app)
+    frame = {"id": "s1", "path": "/api/v1/flows/ws", "query": f"ticket={ticket}"}
+
+    async def opened() -> list[str]:
+        sent: list[str] = []
+
+        class Socket:
+            async def send(self, text: str) -> None:
+                sent.append(text)
+
+        task = asyncio.create_task(connector._stream(Socket(), frame))
+        while len(sent) < 1:
+            await asyncio.sleep(0.01)
+        task.cancel()
+        return [str(json.loads(text)["op"]) for text in sent]
+
+    assert asyncio.run(opened())[0] == "ws_open_ok"
+    # Used once: the same ticket again is refused.
+    assert asyncio.run(opened()) == ["ws_close"]

docs/code/api.md

@@ -233,20 +233,24 @@ the process is dead.
 | `GET` | `/workers/tokens` | the worker credentials minted, without the tokens (superuser) |
 | `POST` | `/workers/tokens/{jti}/revoke` | refuse one from now on and drop a worker attached with it (superuser) |
 | `GET` | `/workers/runtime` | the node runner's source, for a host without pip |
-| `WS` | `/workers/attach` | where a worker dials in |
+| `WS` | `/workers/attach` | where a worker dials in, with `Authorization: Bearer <worker token>` |
 
 See [Remote workers](workers.md).
 
 ## Live events
 
 ```text
-ws://127.0.0.1:8000/api/v1/flows/ws?token=<access token>
+ws://127.0.0.1:8000/api/v1/flows/ws
 ```
 
-The websocket authenticates from its query string. It sends a snapshot on
-connect and then every engine event: node executions, values published, health
-changes, run started and finished. This is what the canvas and the dashboards
-draw from.
+A client that can set headers sends `Authorization: Bearer <access token>` on
+the handshake. A browser cannot, so it first calls `POST /flows/ws-ticket` and
+opens `/flows/ws?ticket=<ticket>`: a ticket is good for one connect, within 30
+seconds, so the URL that access logs keep is worth nothing afterwards.
+`?token=` is still accepted for one more release. The socket sends a snapshot
+on connect and then every engine event: node executions, values published,
+health changes, run started and finished. This is what the canvas and the
+dashboards draw from.
 
 ## Errors
 

docs/code/workers.md

@@ -199,6 +199,12 @@ A token minted before tokens carried their labels and a `jti` is refused. Mint
 each existing worker host a new one with `POST /workers/tokens`, naming the
 labels it should carry, and swap it in for `FLUKSIO_WORKER_TOKEN`.
 
+**Upgrade the engine before its workers.** A worker sends its token as an
+`Authorization` header on the handshake, where it stays out of access logs, and
+an engine from before that reads it only from the URL: it refuses the newer
+worker at the handshake. The engine still takes the old `?token=` for one
+release, so older workers keep attaching to a newer engine meanwhile.
+
 ## Machines from a batch scheduler
 
 A cluster is not a machine that attaches and stays; it is a queue somebody else

frontend/src/client/schemas.gen.ts

@@ -3692,6 +3692,22 @@ export const ShareRequestSchema = {
     title: 'ShareRequest'
 } as const;
 
+export const SocketTicketSchema = {
+    properties: {
+        ticket: {
+            type: 'string',
+            title: 'Ticket'
+        },
+        expires_in: {
+            type: 'number',
+            title: 'Expires In'
+        }
+    },
+    type: 'object',
+    required: ['ticket', 'expires_in'],
+    title: 'SocketTicket'
+} as const;
+
 export const StoredKeySchema = {
     properties: {
         key: {

frontend/src/client/types.gen.ts

@@ -1241,6 +1241,11 @@ export type ShareRequest = {
     lib_name: string;
 };
 
+export type SocketTicket = {
+    ticket: string;
+    expires_in: number;
+};
+
 export type StoredKey = {
     key: string;
     size: number;
@@ -1835,6 +1840,8 @@ export type FlowsReadMessageHistoryData = {
 
 export type FlowsReadMessageHistoryResponse = (MessageHistory);
 
+export type FlowsIssueSocketTicketResponse = (SocketTicket);
+
 export type LoginLoginAccessTokenData = {
     formData: Body_login_login_access_token;
 };

frontend/src/components/Flow/useFlowSocket.ts

@@ -2,13 +2,12 @@ import type { QueryClient } from "@tanstack/react-query"
 import { useQueryClient } from "@tanstack/react-query"
 import { useEffect } from "react"
 
-import { OpenAPI } from "@/client"
+import { ApiError, FlowsService, OpenAPI } from "@/client"
 import { dashboardKeys, panelKeys } from "@/components/Dashboard/queries"
 import { healthKeys } from "@/components/Health/queries"
 import { runKeys } from "@/components/Runs/queries"
 import { connectionStore } from "@/lib/connectionStore"
 import { parseMediaFrame } from "@/lib/media"
-import { apiToken } from "@/lib/portal"
 import { type LogLine, liveStore, type ValueSource } from "./liveStore"
 import { flowKeys } from "./queries"
 
@@ -126,17 +125,19 @@ type FlowEvent =
       ts?: number
     }
 
-function socketUrl(): string {
+async function socketUrl(): Promise<string> {
   const base = String(OpenAPI.BASE || window.location.origin)
   // Concatenated rather than resolved: an absolute path as the second argument
   // to `new URL` discards the base's own path, which under a portal
   // (`https://host/i/{id}`) would aim the socket at the wrong place entirely.
   const url = new URL(`${base.replace(/\/$/, "")}/api/v1/flows/ws`)
   url.protocol = url.protocol === "https:" ? "wss:" : "ws:"
-  // Browsers cannot set headers on a websocket handshake, so the token rides
-  // in the query string. Read per attempt, so a reconnect after the old one
-  // expired carries whatever is current rather than what we started with.
-  url.searchParams.set("token", apiToken())
+  // Browsers cannot set headers on a websocket handshake, and a token in the
+  // URL ends up in every access log on the way. So the token is traded for a
+  // one-time ticket, fetched per attempt: it is good for one connect, and the
+  // call carries whatever token is current.
+  const { ticket } = await FlowsService.issueSocketTicket()
+  url.searchParams.set("ticket", ticket)
   return url.toString()
 }
 
@@ -159,6 +160,8 @@ let retry = RECONNECT_MIN
 let watchers = 0
 /** Consecutive 1008s, cleared by a handshake the server accepts. */
 let rejected = 0
+/** A ticket is being fetched, so a socket is on its way. */
+let connecting = false
 let client: QueryClient | null = null
 const authHandlers = new Set<() => void>()
 /** Floor re-reads still owed, per node, so a recovery can cancel its own. */
@@ -206,9 +209,36 @@ function sendWanted() {
 
 liveStore.onWanted(sendWanted)
 
+/** The token was refused, by the socket or by the ticket call before it. */
+function refused(): boolean {
+  rejected += 1
+  if (rejected < AUTH_ATTEMPTS) return false
+  for (const handler of authHandlers) handler()
+  return true
+}
+
 function connect() {
+  if (watchers === 0 || socket || timer || connecting) return
+  connecting = true
+  socketUrl().then(
+    (url) => {
+      connecting = false
+      open(url)
+    },
+    (error: unknown) => {
+      connecting = false
+      if (watchers === 0) return
+      // A 401 is the 1008 of the ticket call; anything else is the engine
+      // not answering yet, which the backoff is for.
+      if (error instanceof ApiError && error.status === 401 && refused()) return
+      schedule()
+    },
+  )
+}
+
+function open(url: string) {
   if (watchers === 0 || socket || timer) return
-  const ws = new WebSocket(socketUrl())
+  const ws = new WebSocket(url)
   ws.binaryType = "arraybuffer"
   socket = ws
 
@@ -422,13 +452,7 @@ function connect() {
     liveStore.setConnected(false)
     connectionStore.setSocketOpen(false)
     if (watchers === 0) return
-    if (event.code === 1008) {
-      rejected += 1
-      if (rejected >= AUTH_ATTEMPTS) {
-        for (const handler of authHandlers) handler()
-        return
-      }
-    }
+    if (event.code === 1008 && refused()) return
     // 1013 is the portal saying the instance is not attached — the one
     // close code that means "offline" rather than "the socket dropped".
     if (event.code === 1013) {

uv.lock

@@ -1021,7 +1021,7 @@ dependencies = [
 ]
 
 [package.metadata]
-requires-dist = [{ name = "websockets", specifier = ">=12" }]
+requires-dist = [{ name = "websockets", specifier = ">=14" }]
 
 [[package]]
 name = "frozenlist"

worker/README.md

@@ -19,6 +19,9 @@ its own wheels without the engine ever installing them. Mint a token from the
 engine with `POST /api/v1/workers/tokens`; it carries this worker's name and
 labels, and `--labels` may repeat those but not change them.
 
+It sends the token as an `Authorization` header, which an engine from before
+that does not read: upgrade the engine first, then its workers.
+
 A plain `ws://` address is refused unless the engine is on this machine, since
 the token and node secrets would cross the network in clear; `--insecure`
 allows it on a network you trust.

worker/fluksio_worker/agent.py

@@ -156,8 +156,13 @@ class Agent:
                 backoff = min(MAX_BACKOFF_S, backoff * 2)
 
     async def _session(self) -> None:
-        url = f"{self.args.url}?token={self.args.token}"
-        async with websockets.connect(url, max_size=None, ping_interval=20) as socket:
+        # A header rather than the query string, which access logs keep.
+        async with websockets.connect(
+            self.args.url,
+            additional_headers={"Authorization": f"Bearer {self.args.token}"},
+            max_size=None,
+            ping_interval=20,
+        ) as socket:
             await socket.send(
                 json.dumps(
                     {

worker/pyproject.toml

@@ -22,7 +22,7 @@ classifiers = [
     "Topic :: System :: Distributed Computing",
 ]
 dependencies = [
-    "websockets>=12",
+    "websockets>=14",
 ]
 
 [project.urls]