From 5259b8db0a157bab99c4fd879d63c7b547e3dc07 Mon Sep 17 00:00:00 2001 From: stroblme Date: Wed, 7 Oct 2026 08:58:22 +0200 Subject: [PATCH] Authenticate sockets by header or one-time ticket instead of ?token= Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_013GAeg4Jo6xjZzjwy9QDWMQ --- backend/fluksio/api/deps.py | 3 + backend/fluksio/api/routes/flows.py | 30 ++++++-- backend/fluksio/api/routes/workers.py | 7 +- backend/fluksio/api/socket_auth.py | 60 ++++++++++++++++ backend/fluksio/cloud/connector.py | 7 +- backend/fluksio/sdk/stream.py | 14 ++-- backend/tests/api/routes/test_flows.py | 1 + backend/tests/api/routes/test_panels.py | 12 ++++ backend/tests/api/routes/test_socket_auth.py | 72 +++++++++++++++++++ backend/tests/test_cli.py | 8 +-- backend/tests/test_cloud.py | 32 +++++++++ docs/code/api.md | 16 +++-- docs/code/workers.md | 6 ++ frontend/src/client/schemas.gen.ts | 16 +++++ frontend/src/client/sdk.gen.ts | 19 ++++- frontend/src/client/types.gen.ts | 7 ++ frontend/src/components/Flow/useFlowSocket.ts | 54 ++++++++++---- uv.lock | 2 +- worker/README.md | 3 + worker/fluksio_worker/agent.py | 9 ++- worker/pyproject.toml | 2 +- 21 files changed, 336 insertions(+), 44 deletions(-) create mode 100644 backend/fluksio/api/socket_auth.py create mode 100644 backend/tests/api/routes/test_socket_auth.py diff --git a/backend/fluksio/api/deps.py b/backend/fluksio/api/deps.py index 1770d65..03497b3 100644 --- a/backend/fluksio/api/deps.py +++ b/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 diff --git a/backend/fluksio/api/routes/flows.py b/backend/fluksio/api/routes/flows.py index ed992ca..d4f2a6f 100644 --- a/backend/fluksio/api/routes/flows.py +++ b/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 diff --git a/backend/fluksio/api/routes/workers.py b/backend/fluksio/api/routes/workers.py index 8fe2163..ca6ef74 100644 --- a/backend/fluksio/api/routes/workers.py +++ b/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: diff --git a/backend/fluksio/api/socket_auth.py b/backend/fluksio/api/socket_auth.py new file mode 100644 index 0000000..8a5d583 --- /dev/null +++ b/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 diff --git a/backend/fluksio/cloud/connector.py b/backend/fluksio/cloud/connector.py index 100a074..0135c05 100644 --- a/backend/fluksio/cloud/connector.py +++ b/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})) diff --git a/backend/fluksio/sdk/stream.py b/backend/fluksio/sdk/stream.py index ee4e250..6a8d8a6 100644 --- a/backend/fluksio/sdk/stream.py +++ b/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: diff --git a/backend/tests/api/routes/test_flows.py b/backend/tests/api/routes/test_flows.py index 5ef9cbc..1e8f708 100644 --- a/backend/tests/api/routes/test_flows.py +++ b/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] = [] diff --git a/backend/tests/api/routes/test_panels.py b/backend/tests/api/routes/test_panels.py index c4d2785..12c67c0 100644 --- a/backend/tests/api/routes/test_panels.py +++ b/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] diff --git a/backend/tests/api/routes/test_socket_auth.py b/backend/tests/api/routes/test_socket_auth.py new file mode 100644 index 0000000..6ec7e98 --- /dev/null +++ b/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" diff --git a/backend/tests/test_cli.py b/backend/tests/test_cli.py index f4853fb..61cb816 100644 --- a/backend/tests/test_cli.py +++ b/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://") diff --git a/backend/tests/test_cloud.py b/backend/tests/test_cloud.py index 5712ead..83ced0a 100644 --- a/backend/tests/test_cloud.py +++ b/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"] diff --git a/docs/code/api.md b/docs/code/api.md index cb9135b..ea08cec 100644 --- a/docs/code/api.md +++ b/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 ` | See [Remote workers](workers.md). ## Live events ```text -ws://127.0.0.1:8000/api/v1/flows/ws?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 ` on +the handshake. A browser cannot, so it first calls `POST /flows/ws-ticket` and +opens `/flows/ws?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 diff --git a/docs/code/workers.md b/docs/code/workers.md index fbdc9ba..a62ae7b 100644 --- a/docs/code/workers.md +++ b/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 diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index e5b5e0c..fb7dbfa 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/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: { diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index 05607b4..8cab379 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -3,7 +3,7 @@ import type { CancelablePromise } from './core/CancelablePromise'; import { OpenAPI } from './core/OpenAPI'; import { request as __request } from './core/request'; -import type { AiReadAiConfigResponse, AiUpdateAiConfigData, AiUpdateAiConfigResponse, AiDeleteAiConfigResponse, AiListAiModelsData, AiListAiModelsResponse, AiReadAiStatusResponse, AiReadControlNodesResponse, AiReadChatResponse, AiChatData, AiChatResponse, AiClearChatResponse, AlertsReadAlertsConfigResponse, AlertsSaveAlertsConfigData, AlertsSaveAlertsConfigResponse, AlertsReadWebpushKeyResponse, AlertsAddWebpushSubscriptionData, AlertsAddWebpushSubscriptionResponse, AlertsRemoveWebpushSubscriptionData, AlertsRemoveWebpushSubscriptionResponse, AlertsTestChannelData, AlertsTestChannelResponse, ArtifactsPutArtifactData, ArtifactsPutArtifactResponse, ArtifactsGetArtifactData, ArtifactsGetArtifactResponse, CloudReadStatusResponse, CloudEnrollData, CloudEnrollResponse, CloudAddRemoteUserData, CloudAddRemoteUserResponse, CloudDisconnectResponse, ConnectorsReadConnectorStateData, ConnectorsReadConnectorStateResponse, ConnectorsClearConnectorStateData, ConnectorsClearConnectorStateResponse, ConnectorsDeleteConnectorStateData, ConnectorsDeleteConnectorStateResponse, DashboardsReadDashboardsResponse, DashboardsReadDashboardData, DashboardsReadDashboardResponse, DashboardsCreateDashboardData, DashboardsCreateDashboardResponse, DashboardsSaveDashboardData, DashboardsSaveDashboardResponse, DashboardsDeleteDashboardData, DashboardsDeleteDashboardResponse, DashboardsGenerateResultsDashboardData, DashboardsGenerateResultsDashboardResponse, DashboardsPublishDashboardData, DashboardsPublishDashboardResponse, DashboardsDiscardDashboardDraftData, DashboardsDiscardDashboardDraftResponse, DashboardsRenameDashboardData, DashboardsRenameDashboardResponse, FlavorsReadFlavorsResponse, FlavorsCreateFlavorData, FlavorsCreateFlavorResponse, FlavorsUpdateFlavorData, FlavorsUpdateFlavorResponse, FlavorsDeleteFlavorData, FlavorsDeleteFlavorResponse, FlowsReadFlowsResponse, FlowsReadNodeTypesResponse, FlowsReadGraphResponse, FlowsReadLibraryResponse, FlowsDeleteSharedNodeData, FlowsDeleteSharedNodeResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsRenameFlowData, FlowsRenameFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsShareNodeData, FlowsShareNodeResponse, FlowsUnshareNodeData, FlowsUnshareNodeResponse, FlowsStartFlowData, FlowsStartFlowResponse, FlowsStopFlowData, FlowsStopFlowResponse, FlowsPauseFlowData, FlowsPauseFlowResponse, FlowsResumeFlowData, FlowsResumeFlowResponse, FlowsStepFlowData, FlowsStepFlowResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsCancelNodeData, FlowsCancelNodeResponse, FlowsAcknowledgeNodeErrorData, FlowsAcknowledgeNodeErrorResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, MessagesReadMessagesResponse, MessagesPublishMessageData, MessagesPublishMessageResponse, MessagesReadMessageHistoryData, MessagesReadMessageHistoryResponse, ModulesReadModulesResponse, ModulesApplyModulesData, ModulesApplyModulesResponse, ModulesRefreshModulesResponse, OauthRegisterClientData, OauthRegisterClientResponse, OauthAuthorizeValidateData, OauthAuthorizeValidateResponse, OauthAuthorizeData, OauthAuthorizeResponse, OauthTokenData, OauthTokenResponse, OauthReadClientsResponse, OauthRevokeClientData, OauthRevokeClientResponse, ObservabilityReadSummaryResponse, ObservabilityReadTimeseriesData, ObservabilityReadTimeseriesResponse, ObservabilityReadFlowRollupsData, ObservabilityReadFlowRollupsResponse, ObservabilityReadRunsData, ObservabilityReadRunsResponse, ObservabilityReadEventsData, ObservabilityReadEventsResponse, ObservabilityReadDeadLettersData, ObservabilityReadDeadLettersResponse, PanelsReadPanelsResponse, PanelsSavePanelsData, PanelsSavePanelsResponse, PanelsStartPairingResponse, PanelsPollPairingData, PanelsPollPairingResponse, PanelsPendingDeviceData, PanelsPendingDeviceResponse, PanelsApprovePairingData, PanelsApprovePairingResponse, PanelsUnpairPanelData, PanelsUnpairPanelResponse, PanelsReadPanelData, PanelsReadPanelResponse, PrivateCreateUserData, PrivateCreateUserResponse, RunsCreateRunData, RunsCreateRunResponse, RunsCreateSweepData, RunsCreateSweepResponse, RunsImportRunsData, RunsImportRunsResponse, RunsReadRunsData, RunsReadRunsResponse, RunsDeleteRunsData, RunsDeleteRunsResponse, RunsReadOverviewResponse, RunsExportMetricsData, RunsExportMetricsResponse, RunsExportRunsData, RunsExportRunsResponse, RunsReadMetricNamesData, RunsReadMetricNamesResponse, RunsReadRunData, RunsReadRunResponse, RunsDeleteRunData, RunsDeleteRunResponse, RunsCancelRunData, RunsCancelRunResponse, RunsRetryRunData, RunsRetryRunResponse, RunsReadMetricsData, RunsReadMetricsResponse, RunsCompareMetricData, RunsCompareMetricResponse, SearchReadSearchIndexResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsReadConfigResponse, UtilsHealthCheckResponse, UtilsHealthResponse, WorkersReadWorkersResponse, WorkersReadResourcesResponse, WorkersReadTokensResponse, WorkersIssueTokenData, WorkersIssueTokenResponse, WorkersRevokeTokenData, WorkersRevokeTokenResponse, WorkersReadRuntimeResponse } from './types.gen'; +import type { AiReadAiConfigResponse, AiUpdateAiConfigData, AiUpdateAiConfigResponse, AiDeleteAiConfigResponse, AiListAiModelsData, AiListAiModelsResponse, AiReadAiStatusResponse, AiReadControlNodesResponse, AiReadChatResponse, AiChatData, AiChatResponse, AiClearChatResponse, AlertsReadAlertsConfigResponse, AlertsSaveAlertsConfigData, AlertsSaveAlertsConfigResponse, AlertsReadWebpushKeyResponse, AlertsAddWebpushSubscriptionData, AlertsAddWebpushSubscriptionResponse, AlertsRemoveWebpushSubscriptionData, AlertsRemoveWebpushSubscriptionResponse, AlertsTestChannelData, AlertsTestChannelResponse, ArtifactsPutArtifactData, ArtifactsPutArtifactResponse, ArtifactsGetArtifactData, ArtifactsGetArtifactResponse, CloudReadStatusResponse, CloudEnrollData, CloudEnrollResponse, CloudAddRemoteUserData, CloudAddRemoteUserResponse, CloudDisconnectResponse, ConnectorsReadConnectorStateData, ConnectorsReadConnectorStateResponse, ConnectorsClearConnectorStateData, ConnectorsClearConnectorStateResponse, ConnectorsDeleteConnectorStateData, ConnectorsDeleteConnectorStateResponse, DashboardsReadDashboardsResponse, DashboardsReadDashboardData, DashboardsReadDashboardResponse, DashboardsCreateDashboardData, DashboardsCreateDashboardResponse, DashboardsSaveDashboardData, DashboardsSaveDashboardResponse, DashboardsDeleteDashboardData, DashboardsDeleteDashboardResponse, DashboardsGenerateResultsDashboardData, DashboardsGenerateResultsDashboardResponse, DashboardsPublishDashboardData, DashboardsPublishDashboardResponse, DashboardsDiscardDashboardDraftData, DashboardsDiscardDashboardDraftResponse, DashboardsRenameDashboardData, DashboardsRenameDashboardResponse, FlavorsReadFlavorsResponse, FlavorsCreateFlavorData, FlavorsCreateFlavorResponse, FlavorsUpdateFlavorData, FlavorsUpdateFlavorResponse, FlavorsDeleteFlavorData, FlavorsDeleteFlavorResponse, FlowsReadFlowsResponse, FlowsReadNodeTypesResponse, FlowsReadGraphResponse, FlowsReadLibraryResponse, FlowsDeleteSharedNodeData, FlowsDeleteSharedNodeResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsRenameFlowData, FlowsRenameFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsShareNodeData, FlowsShareNodeResponse, FlowsUnshareNodeData, FlowsUnshareNodeResponse, FlowsStartFlowData, FlowsStartFlowResponse, FlowsStopFlowData, FlowsStopFlowResponse, FlowsPauseFlowData, FlowsPauseFlowResponse, FlowsResumeFlowData, FlowsResumeFlowResponse, FlowsStepFlowData, FlowsStepFlowResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsCancelNodeData, FlowsCancelNodeResponse, FlowsAcknowledgeNodeErrorData, FlowsAcknowledgeNodeErrorResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, FlowsIssueSocketTicketResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, MessagesReadMessagesResponse, MessagesPublishMessageData, MessagesPublishMessageResponse, MessagesReadMessageHistoryData, MessagesReadMessageHistoryResponse, ModulesReadModulesResponse, ModulesApplyModulesData, ModulesApplyModulesResponse, ModulesRefreshModulesResponse, OauthRegisterClientData, OauthRegisterClientResponse, OauthAuthorizeValidateData, OauthAuthorizeValidateResponse, OauthAuthorizeData, OauthAuthorizeResponse, OauthTokenData, OauthTokenResponse, OauthReadClientsResponse, OauthRevokeClientData, OauthRevokeClientResponse, ObservabilityReadSummaryResponse, ObservabilityReadTimeseriesData, ObservabilityReadTimeseriesResponse, ObservabilityReadFlowRollupsData, ObservabilityReadFlowRollupsResponse, ObservabilityReadRunsData, ObservabilityReadRunsResponse, ObservabilityReadEventsData, ObservabilityReadEventsResponse, ObservabilityReadDeadLettersData, ObservabilityReadDeadLettersResponse, PanelsReadPanelsResponse, PanelsSavePanelsData, PanelsSavePanelsResponse, PanelsStartPairingResponse, PanelsPollPairingData, PanelsPollPairingResponse, PanelsPendingDeviceData, PanelsPendingDeviceResponse, PanelsApprovePairingData, PanelsApprovePairingResponse, PanelsUnpairPanelData, PanelsUnpairPanelResponse, PanelsReadPanelData, PanelsReadPanelResponse, PrivateCreateUserData, PrivateCreateUserResponse, RunsCreateRunData, RunsCreateRunResponse, RunsCreateSweepData, RunsCreateSweepResponse, RunsImportRunsData, RunsImportRunsResponse, RunsReadRunsData, RunsReadRunsResponse, RunsDeleteRunsData, RunsDeleteRunsResponse, RunsReadOverviewResponse, RunsExportMetricsData, RunsExportMetricsResponse, RunsExportRunsData, RunsExportRunsResponse, RunsReadMetricNamesData, RunsReadMetricNamesResponse, RunsReadRunData, RunsReadRunResponse, RunsDeleteRunData, RunsDeleteRunResponse, RunsCancelRunData, RunsCancelRunResponse, RunsRetryRunData, RunsRetryRunResponse, RunsReadMetricsData, RunsReadMetricsResponse, RunsCompareMetricData, RunsCompareMetricResponse, SearchReadSearchIndexResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsReadConfigResponse, UtilsHealthCheckResponse, UtilsHealthResponse, WorkersReadWorkersResponse, WorkersReadResourcesResponse, WorkersReadTokensResponse, WorkersIssueTokenData, WorkersIssueTokenResponse, WorkersRevokeTokenData, WorkersRevokeTokenResponse, WorkersReadRuntimeResponse } from './types.gen'; export class AiService { /** @@ -1365,6 +1365,23 @@ export class FlowsService { } }); } + + /** + * Issue Socket Ticket + * 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. + * @returns SocketTicket Successful Response + * @throws ApiError + */ + public static issueSocketTicket(): CancelablePromise { + return __request(OpenAPI, { + method: 'POST', + url: '/api/v1/flows/ws-ticket' + }); + } } export class LoginService { diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 476aad8..8d2d3c9 100644 --- a/frontend/src/client/types.gen.ts +++ b/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; }; diff --git a/frontend/src/components/Flow/useFlowSocket.ts b/frontend/src/components/Flow/useFlowSocket.ts index c0cde04..43b1a0f 100644 --- a/frontend/src/components/Flow/useFlowSocket.ts +++ b/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 { 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) { diff --git a/uv.lock b/uv.lock index 6bcab51..eb11543 100644 --- a/uv.lock +++ b/uv.lock @@ -1021,7 +1021,7 @@ dependencies = [ ] [package.metadata] -requires-dist = [{ name = "websockets", specifier = ">=12" }] +requires-dist = [{ name = "websockets", specifier = ">=14" }] [[package]] name = "frozenlist" diff --git a/worker/README.md b/worker/README.md index 7109a65..aac4dec 100644 --- a/worker/README.md +++ b/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. diff --git a/worker/fluksio_worker/agent.py b/worker/fluksio_worker/agent.py index e9a12fc..4f80ffd 100644 --- a/worker/fluksio_worker/agent.py +++ b/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( { diff --git a/worker/pyproject.toml b/worker/pyproject.toml index 45c735a..58850cf 100644 --- a/worker/pyproject.toml +++ b/worker/pyproject.toml @@ -22,7 +22,7 @@ classifiers = [ "Topic :: System :: Distributed Computing", ] dependencies = [ - "websockets>=12", + "websockets>=14", ] [project.urls]