diff --git a/scripts/e2e-edge.sh b/scripts/e2e-edge.sh index d8e960b..4cebebc 100755 --- a/scripts/e2e-edge.sh +++ b/scripts/e2e-edge.sh @@ -335,6 +335,50 @@ if [ -n "$NO_CAMERA" ]; then ok "a printer with no camera is refused with 404, blaming the right half" fi +# --- 5c. the operator console's live feed ---------------------------------- + +# The user-facing socket, driven for real. The unit tests can only mirror its +# composition, because the route imports ../../db; this is the only thing that +# proves it is wired into the app at all, that the Supabase token verifies on +# the upgrade, and that a gateway's telemetry actually reaches a console. +echo "==> /ws/telemetry: connecting as the operator and waiting for a push" +WS_OUT="$WORK/telemetry.json" +if bun scripts/watch-telemetry.ts "ws://localhost:$PORT" "$USER_TOKEN" 12 > "$WS_OUT" 2>"$WORK/ws.err"; then + python3 - "$WS_OUT" <<'PYCHK' || fail "the telemetry socket did not behave: $(cat "$WS_OUT" "$WORK/ws.err")" +import json, sys +d = json.load(open(sys.argv[1])) +# "bearer" and not the token: the runtime echoes the first offered subprotocol +# back in the response header, so the credential must not be in that position. +assert d["negotiated"] == "bearer", f'echoed subprotocol was {d["negotiated"]!r}, expected "bearer"' +kinds = [f.get("type") for f in d["frames"]] +assert "ready" in kinds, f"no ready frame; got {kinds}" +ready = next(f for f in d["frames"] if f.get("type") == "ready") +assert ready["visiblePrinters"] >= 1, f"ready says {ready['visiblePrinters']} visible printers" +pushes = [f for f in d["frames"] if f.get("type") == "printer.telemetry"] +assert pushes, f"no printer.telemetry push arrived; got {kinds}" +# A push must be the printer resource, not a telemetry-shaped subset, and must +# carry real readings rather than a touched row of nulls. +p = pushes[-1]["printer"] +for key in ("id", "farmId", "name", "vendor"): + assert key in p, f"push is missing {key}: {p}" +assert p.get("vendorState"), f"push carries no vendor state: {p}" +print(f' ok: {len(pushes)} push(es); last was {p["name"]} ' + f'state={p["vendorState"]} nozzle={p.get("nozzleTempC")} camera={p.get("cameraStream")}') +PYCHK +else + fail "could not connect to /ws/telemetry: $(cat "$WORK/ws.err")" +fi +ok "the console socket authenticated and received live telemetry" + +# A socket with no valid credential must not open at all - the whole point of +# checking at the upgrade rather than in a first message. +echo "==> an unauthenticated console socket must be refused" +if bun scripts/watch-telemetry.ts "ws://localhost:$PORT" "not-a-real-token" 5 \ + > "$WORK/ws-bad.json" 2>&1; then + fail "a socket opened with a bogus token: $(cat "$WORK/ws-bad.json")" +fi +ok "a bogus token is refused before any socket exists" + # --- 6. the negative case -------------------------------------------------- echo "==> an unclaimed printer must be refused, not guessed at" diff --git a/scripts/watch-telemetry.ts b/scripts/watch-telemetry.ts new file mode 100644 index 0000000..146f3a0 --- /dev/null +++ b/scripts/watch-telemetry.ts @@ -0,0 +1,38 @@ +/** + * Connects to /ws/telemetry as a user and prints what arrives, then exits. + * + * Exists for scripts/e2e-edge.sh, which can then assert on real frames from + * the real route rather than on a mirrored composition in a unit test. Also + * useful by hand: it is the quickest way to see whether a farm's telemetry is + * actually reaching a console. + * + * Usage: bun scripts/watch-telemetry.ts [seconds] + */ +const [base, token, secondsArg] = process.argv.slice(2); +if (!base || !token) { + console.error("usage: watch-telemetry.ts [seconds]"); + process.exit(2); +} +const seconds = Number(secondsArg ?? 15); + +// The token goes in the subprotocol, not the URL - a browser cannot set +// headers, and a query parameter would put a bearer credential in every +// access log. "bearer" first: the server echoes the first offered value back, +// and the token must not be what comes back. +const ws = new WebSocket(`${base}/ws/telemetry`, ["bearer", token]); + +const frames: unknown[] = []; +ws.onmessage = (event) => { + frames.push(typeof event.data === "string" ? JSON.parse(event.data) : event.data); +}; +ws.onerror = () => { + console.error("REFUSED: the socket would not open"); + process.exit(1); +}; + +setTimeout(() => { + // Printed as one JSON document so the caller can parse it rather than + // scraping log lines. + console.log(JSON.stringify({ negotiated: ws.protocol, frames }, null, 2)); + process.exit(0); +}, seconds * 1000); diff --git a/src/auth/ws_subprotocol.ts b/src/auth/ws_subprotocol.ts new file mode 100644 index 0000000..ce18244 --- /dev/null +++ b/src/auth/ws_subprotocol.ts @@ -0,0 +1,50 @@ +import { UnauthorizedError } from "@continuum/ts-core"; + +/** + * Pulls a bearer token out of a WebSocket's `Sec-WebSocket-Protocol` header. + * + * A leaf module with no imports but the error type, like + * `auth/edge_upgrade.ts` next to it, and for the same reason: this is an + * authentication boundary and should not need a database to test. + * + * ## Why the subprotocol and not a query parameter + * + * A browser cannot set headers on a WebSocket - `new WebSocket(url, protos)` + * is the whole of what JavaScript controls about the handshake. So the usual + * `Authorization: Bearer` is unavailable, and the common workaround is + * `?token=...`, which puts a bearer credential in a URL and therefore into + * access logs, proxy logs and error breadcrumbs. This codebase already goes + * out of its way to avoid that: continuum-proxy hand-writes a `Debug` so a + * presigned URL prints as ``. + * + * The subprotocol field is a header, is not logged by default, and - unlike + * a message-based handshake - is available *before* the upgrade, so an + * unauthenticated peer never gets a socket at all. The gateway socket gets + * the same property through a real `Authorization` header, which it can use + * because it is not a browser. + * + * The client sends `["bearer", ""]`. "bearer" goes first because the + * server must echo exactly one offered value back or the browser fails the + * connection, and the runtime echoes the first - so the token itself must + * not be in that position, or it would come back in the response header. + */ +const SCHEME = "bearer"; + +export function tokenFromSubprotocol(header: string | null | undefined): string { + const offered = (header ?? "") + .split(",") + .map((part) => part.trim()) + .filter((part) => part.length > 0); + + // Exactly two: the scheme and the token. Being strict here rather than + // scanning for "the entry that looks like a JWT" means a client that gets + // the order or the count wrong is told so, instead of working by accident + // until the day something else is in the list. + if (offered.length !== 2 || offered[0] !== SCHEME) { + throw new UnauthorizedError( + `WebSocket subprotocol must be ["${SCHEME}", ""]`, + ); + } + + return offered[1]!; +} diff --git a/src/index.ts b/src/index.ts index d8284c9..17f509f 100644 --- a/src/index.ts +++ b/src/index.ts @@ -11,6 +11,7 @@ import { farmRoutes } from "./routes/farms"; import { printerRoutes } from "./routes/printers"; import { jobRoutes } from "./routes/jobs"; import { edgeWsRoute } from "./routes/ws/edge"; +import { telemetryWsRoute } from "./routes/ws/telemetry"; import { startDispatcher } from "./dispatch"; const app = new Elysia() @@ -62,6 +63,7 @@ const app = new Elysia() .use(printerRoutes) .use(jobRoutes) .use(edgeWsRoute) + .use(telemetryWsRoute) .listen(Number(process.env.PORT ?? 3000)); console.log(`continuum-backend listening on :${app.server?.port}`); diff --git a/src/routes/ws/edge.ts b/src/routes/ws/edge.ts index 2b4b391..14c0aad 100644 --- a/src/routes/ws/edge.ts +++ b/src/routes/ws/edge.ts @@ -13,6 +13,8 @@ import { advanceJobsFrom } from "../../dispatch"; import { enqueueFrameIngestionBatch } from "../../queue"; import { edgeJwtSecret } from "../../auth/edge_secret"; import { farmFromEdgeUpgrade } from "../../auth/edge_upgrade"; +import { publish } from "../../telemetry/broadcast"; +import { printerProjection } from "../../db/projections"; /** * Edge-gateway uplink. A continuum-proxy gateway connects here with a @@ -259,7 +261,7 @@ async function persistTelemetry(conn: EdgeConn, batch: PrinterTelemetry[]) { const status = mapStatus(snapshot.state, snapshot.print_error); - await db + const [updated] = await db .update(printers) .set({ // `?? null` rather than leaving the key out: a field the gateway @@ -285,7 +287,27 @@ async function persistTelemetry(conn: EdgeConn, batch: PrinterTelemetry[]) { lastSeenAt: new Date(), updatedAt: new Date(), }) - .where(eq(printers.id, printerId)); + .where(eq(printers.id, printerId)) + // Returned through the same projection the REST routes use, so what a + // watching console is pushed is byte-for-byte what it would have got + // from `GET /printers/:id`. Deriving it from `snapshot` instead would + // be a second translation of the same data, free to drift - and it + // would be missing the columns the gateway does not report at all, + // like `name` and `vendor`. + .returning(printerProjection.returning); + + // Push to anyone watching. After the write, so a console cannot be told + // about a state that failed to persist - and `publish` swallows a + // subscriber's error, so one broken browser tab cannot abort a whole + // farm's telemetry batch here. + if (updated) { + publish({ + ...updated, + // Serialised, because this goes to JSON and a Date would otherwise + // be stringified inconsistently with the REST response. + lastSeenAt: updated.lastSeenAt?.toISOString() ?? null, + }); + } // The printer's row is the latest snapshot; the job's row is the history // that outlives it. This is the only thing that moves a job out of @@ -370,11 +392,20 @@ export const edgeWsRoute = new Elysia() * * Throwing here fails the upgrade with an HTTP status, so a peer without a * valid token never reaches `open` - which is the whole point of - * authenticating at the handshake rather than in the first frame. Modelled - * on `requireUser`'s scoped resolve, for the same reason: one verification - * per connection, in one place, ahead of every handler. + * authenticating at the handshake rather than in the first frame. + * + * The default `local` scope, NOT `scoped` like `requireUser`, and the + * difference is load-bearing. `requireUser` is `.use()`d inside each route + * file and has to propagate `user` to that file's routes. This resolve + * belongs to the one socket defined just below it, and `scoped` here + * propagates it to the parent app, where it then gates every route + * registered *after* this plugin. Measured: with `scoped`, a second ws + * plugin's own route is refused, because this check runs first and finds + * no edge token. Nothing shipped broken only because /ws/telemetry was the + * first route mounted after this one - and it was refused outright until + * this was changed. */ - .resolve({ as: "scoped" }, async ({ headers, edgeJwt }): Promise<{ edgeFarmId: string }> => ({ + .resolve(async ({ headers, edgeJwt }): Promise<{ edgeFarmId: string }> => ({ edgeFarmId: await farmFromEdgeUpgrade(headers.authorization, (token) => edgeJwt.verify(token)), })) .ws("/ws/edge/v1", { diff --git a/src/routes/ws/telemetry.ts b/src/routes/ws/telemetry.ts new file mode 100644 index 0000000..944fd5d --- /dev/null +++ b/src/routes/ws/telemetry.ts @@ -0,0 +1,149 @@ +import { Elysia } from "elysia"; +import { JwtVerificationError, UnauthorizedError } from "@continuum/ts-core"; +import type { AppPrinterState, AppTelemetryMessage } from "@continuum/ts-types"; +import { db } from "../../db"; +import { farms, printers } from "../../db/schema"; +import { visibleFarms, visiblePrinters } from "../../auth"; +import { jwtVerifier } from "../../auth/verifier"; +import { tokenFromSubprotocol } from "../../auth/ws_subprotocol"; +import { maySee, subscribe } from "../../telemetry/broadcast"; + +/** + * The operator console's live fleet feed. + * + * Read-only and push-only: the browser sends nothing after the handshake, + * and a frame arriving from it is ignored. Everything that changes a printer + * goes over HTTP, where the role checks and the audit trail already live - + * `POST /printers/:id/commands`. A socket that can also issue commands is a + * second authorization surface to get right for no gain. + * + * ## Not /ws/edge/v1 + * + * That socket authenticates as a *farm*, with an edge JWT that any gateway + * for that farm holds, and it accepts telemetry. This one authenticates as a + * *user*, with a Supabase access token, and accepts nothing. Pointing a + * browser at the gateway socket would mean handing the browser a credential + * that can drive printers. + * + * ## The wire format + * + * `continuum/v1/app/telemetry_message.schema.json`, mirrored in + * `@continuum/ts-types` as `AppTelemetryMessageSchema`. camelCase, matching + * this backend's HTTP routes rather than the gateway protocol - see that + * file. The payload is the printer resource itself, so a client applies a + * push onto the row it already fetched from `GET /printers`. + */ + +/** + * The farms a user can see, resolved once when their socket opens. + * + * Once, not per message: a farm's telemetry arrives every few seconds per + * printer, and re-deriving visibility each time would put a database round + * trip on every push to every viewer. The cost is staleness in one + * direction - someone added to an organisation mid-connection does not see + * its printers until they reconnect, which a page reload does. + * + * Staleness in the *other* direction is the one that would matter, and it is + * bounded by the same thing: someone removed from an organisation keeps + * receiving its telemetry until their socket drops. That is a real gap worth + * knowing about. Closing it properly means invalidating open sockets on a + * membership change, which needs a signal this backend does not have yet; a + * periodic re-resolve would narrow the window but not close it, so it is + * written down rather than half-done. Telemetry is also the least sensitive + * thing in the system - nozzle temperatures, not credentials - and a removed + * member can be cut off immediately by anything that matters, because every + * HTTP route re-checks on each request. + */ +async function visibleFarmIds(userId: string): Promise> { + const rows = await db.select({ id: farms.id }).from(farms).where(visibleFarms(userId)); + return new Set(rows.map((row) => row.id)); +} + +/** + * How many printers this user can see, for the `ready` message. + * + * Counting farms would be the cheaper query and the wrong answer: a farm + * with no printers claimed yet is exactly the first-run state this message + * exists to make legible. + */ +async function countVisiblePrinters(userId: string): Promise { + const rows = await db.select({ id: printers.id }).from(printers).where(visiblePrinters(userId)); + return rows.length; +} + +/** One unsubscribe per open socket, removed when it closes. */ +const unsubscribers = new Map void>(); + +export const telemetryWsRoute = new Elysia() + /** + * Verifies the user's access token on the upgrade, before a socket exists. + * + * Throwing fails the upgrade with an HTTP status, so an unauthenticated + * browser never reaches `open`. Same shape as the edge socket's resolve + * next door; the difference is only where the token comes from and which + * verifier checks it. + * + * The default `local` scope deliberately - see the note on the edge + * route's resolve. `scoped` would push this check onto every route + * mounted after this plugin, which would then demand a WebSocket + * subprotocol from plain HTTP requests. + */ + .resolve(async ({ headers }): Promise<{ viewer: { id: string } }> => { + const token = tokenFromSubprotocol(headers["sec-websocket-protocol"]); + + try { + const claims = await jwtVerifier.verify(token); + if (!claims.sub) throw new UnauthorizedError("Access token has no subject claim"); + return { viewer: { id: claims.sub } }; + } catch (err) { + // Never pass the reason through - "expired" versus "bad signature" + // tells a forger which part to fix next. Logged instead, exactly as + // `requireUser` does it. + if (err instanceof JwtVerificationError) { + console.warn( + `[telemetry] token rejected: ${err.cause instanceof Error ? err.cause.message : err.message}`, + ); + throw new UnauthorizedError("Invalid or expired access token"); + } + throw err; + } + }) + .ws("/ws/telemetry", { + async open(ws) { + const { viewer } = ws.data as { viewer: { id: string } }; + const allowed = await visibleFarmIds(viewer.id); + + const send = (message: AppTelemetryMessage) => ws.send(message); + + const unsubscribe = subscribe((printer: AppPrinterState) => { + // The line that keeps one organisation's telemetry out of another's + // console. `maySee` is in `telemetry/broadcast.ts` so it can be + // tested without standing up a socket. + if (!maySee(allowed, printer)) return; + send({ type: "printer.telemetry", printer }); + }); + + // Cleanup is attached to the socket rather than kept in a module-level + // map: there is exactly one unsubscribe per socket and the close + // handler is the only thing that needs it. A leaked subscriber would + // push into a dead socket forever. + unsubscribers.set(ws.id, unsubscribe); + + // Sent before any telemetry, so a client knows the difference between + // "nothing to show you" and "not connected yet". Counting farms the + // user can see is not the same as counting printers, so this asks for + // the printers. + send({ type: "ready", visiblePrinters: await countVisiblePrinters(viewer.id) }); + }, + + message(_ws, _message) { + // Deliberately nothing. This socket is push-only; see the module + // comment. Ignoring rather than closing, because a well-meaning client + // sending a keepalive should not be punished for it. + }, + + close(ws) { + unsubscribers.get(ws.id)?.(); + unsubscribers.delete(ws.id); + }, + }); diff --git a/src/telemetry/broadcast.ts b/src/telemetry/broadcast.ts new file mode 100644 index 0000000..ef7c984 --- /dev/null +++ b/src/telemetry/broadcast.ts @@ -0,0 +1,81 @@ +import type { AppPrinterState } from "@continuum/ts-types"; + +/** + * Fan-out from the gateway socket to whoever is watching, with no imports but + * a type - the same reason as `dispatch/job_status.ts`, `stream.ts` and + * `auth/edge_upgrade.ts`: `../db` opens a connection at import time, and the + * part worth testing here is which subscriber is allowed to see what. + * + * ## In-process only, and that is a real limit + * + * A `Set` of callbacks in one Node/Bun process. Two backend instances behind + * a load balancer would each only push the telemetry their own gateways + * happen to be connected to, so an operator would see half a fleet depending + * on which instance their socket landed on. + * + * Not fixed here because the fix is Valkey pub/sub - already a dependency, + * already used by the AI frame queue - and swapping this `Set` for a channel + * subscription is a contained change once there is a second instance to + * justify it. Doing it now would mean testing a message broker to deliver + * events between two objects in the same process. Written down rather than + * discovered: see PLAN.md. + */ + +/** What a subscriber is handed. `farmId` is what authorization turns on. */ +export type Listener = (printer: AppPrinterState) => void; + +const listeners = new Set(); + +/** + * Registers a subscriber and returns its own removal. + * + * Returning the unsubscribe rather than exposing a `remove(listener)` means + * a caller cannot get the identity wrong, and a socket's close handler has + * exactly one thing to call. Leaking a subscriber here would keep pushing + * into a dead socket forever. + */ +export function subscribe(listener: Listener): () => void { + listeners.add(listener); + return () => listeners.delete(listener); +} + +/** + * Publishes one printer's new state to every subscriber. + * + * Each listener is called inside its own try/catch. One viewer's socket + * throwing on send - which a socket closing mid-iteration does - must not + * stop the other viewers from getting the same update, and must not + * propagate into `persistTelemetry`, where it would abort a whole farm's + * telemetry batch over one broken browser tab. + */ +export function publish(printer: AppPrinterState): void { + for (const listener of listeners) { + try { + listener(printer); + } catch (err) { + console.warn("[telemetry] a subscriber threw while being notified", err); + } + } +} + +/** How many subscribers are attached. For tests and `/health`. */ +export function subscriberCount(): number { + return listeners.size; +} + +/** + * Whether a user watching `visibleFarms` should be sent this printer. + * + * Separate from the socket so it can be tested without one, because this is + * the function that decides whether one organisation's operator sees + * another's machines. The set is resolved once when a socket opens - see + * `routes/ws/telemetry.ts` for why, and for what that costs. + */ +export function maySee(visibleFarms: ReadonlySet, printer: AppPrinterState): boolean { + // A printer with no farm cannot be matched against anything, so it is not + // sent. There is no such row - `farm_id` is not nullable - but this is the + // one place where being wrong means leaking another org's machine, so it + // does not rely on that staying true. + if (!printer.farmId) return false; + return visibleFarms.has(printer.farmId); +} diff --git a/tests/telemetry.test.ts b/tests/telemetry.test.ts new file mode 100644 index 0000000..79f8c13 --- /dev/null +++ b/tests/telemetry.test.ts @@ -0,0 +1,283 @@ +import { describe, expect, it } from "bun:test"; +import { Elysia } from "elysia"; +import { FormatRegistry } from "@sinclair/typebox"; +import { Value } from "@sinclair/typebox/value"; +import { AppTelemetryMessageSchema, type AppPrinterState } from "@continuum/ts-types"; +import { UnauthorizedError } from "@continuum/ts-core"; +import { maySee, publish, subscribe, subscriberCount } from "../src/telemetry/broadcast"; +import { tokenFromSubprotocol } from "../src/auth/ws_subprotocol"; + +/** + * TypeBox knows no string formats until something registers them. Elysia + * does that for itself at runtime, which is why `format: "uuid"` works on + * the HTTP routes - but a bare `Value.Check` here does not, and an + * unregistered format fails every value rather than being ignored. So the + * schema keeps its format and the test learns the format, instead of the + * schema being weakened to suit the test. + */ +FormatRegistry.Set("uuid", (value) => + /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(value), +); + +const printer = (over: Partial = {}): AppPrinterState => ({ + id: "3f2b1c5e-0000-4000-8000-000000000001", + farmId: "3f2b1c5e-0000-4000-8000-0000000000aa", + name: "H2C", + vendor: "bambu", + status: "idle", + vendorState: "FINISH", + vendorErrorCode: 0, + nozzleTempC: 30, + nozzleTargetTempC: 0, + bedTempC: 23, + bedTargetTempC: 0, + progressPercent: 100, + currentLayer: 134, + totalLayers: 134, + remainingMinutes: 0, + currentJobName: "a-real-job", + activeAlerts: 6, + cameraStream: "h2c", + lastSeenAt: "2026-10-18T13:40:00.000Z", + ...over, +}); + +/** + * The function that decides whether one organisation's operator sees + * another's machines. It lives in its own module precisely so this test does + * not need a socket or a database. + */ +describe("maySee", () => { + const mine = "3f2b1c5e-0000-4000-8000-0000000000aa"; + const theirs = "3f2b1c5e-0000-4000-8000-0000000000bb"; + + it("sends a printer in a farm the viewer can see", () => { + expect(maySee(new Set([mine]), printer({ farmId: mine }))).toBe(true); + }); + + it("does not send a printer from another organisation's farm", () => { + expect(maySee(new Set([mine]), printer({ farmId: theirs }))).toBe(false); + }); + + it("sends nothing to a viewer who can see no farms", () => { + expect(maySee(new Set(), printer())).toBe(false); + }); + + /** + * `farm_id` is not nullable, so there is no such row today. Asserted + * anyway: this is the one function where being wrong leaks another + * organisation's machines, so it should not depend on a schema constraint + * staying as it is. + */ + it("refuses a printer with no farm rather than matching it against anything", () => { + expect(maySee(new Set([mine]), printer({ farmId: undefined as never }))).toBe(false); + expect(maySee(new Set([""]), printer({ farmId: "" as never }))).toBe(false); + }); +}); + +describe("the broadcast fan-out", () => { + it("delivers to every subscriber and stops on unsubscribe", () => { + const seen: string[] = []; + const off1 = subscribe((p) => seen.push(`a:${p.name}`)); + const off2 = subscribe((p) => seen.push(`b:${p.name}`)); + + publish(printer({ name: "first" })); + expect(seen).toEqual(["a:first", "b:first"]); + + off1(); + publish(printer({ name: "second" })); + expect(seen).toEqual(["a:first", "b:first", "b:second"]); + + off2(); + publish(printer({ name: "third" })); + expect(seen).toEqual(["a:first", "b:first", "b:second"]); + expect(subscriberCount()).toBe(0); + }); + + /** + * A socket closing mid-iteration makes its `send` throw. If that + * propagated, it would come out of `persistTelemetry` and abort a whole + * farm's telemetry batch - one dead browser tab taking out the fleet. + */ + it("keeps delivering after a subscriber throws, and does not propagate", () => { + const seen: string[] = []; + const off1 = subscribe(() => { + throw new Error("socket already closed"); + }); + const off2 = subscribe((p) => seen.push(p.name)); + + expect(() => publish(printer({ name: "survivor" }))).not.toThrow(); + expect(seen).toEqual(["survivor"]); + + off1(); + off2(); + }); +}); + +/** + * A browser cannot set headers on a WebSocket, so the token rides in the + * subprotocol field rather than a query parameter - which would put a bearer + * credential into access and proxy logs. + */ +describe("tokenFromSubprotocol", () => { + it("takes the token from a well-formed offer", () => { + expect(tokenFromSubprotocol("bearer, tok-abc")).toBe("tok-abc"); + // Whatever whitespace the runtime joins the list with. + expect(tokenFromSubprotocol("bearer,tok-abc")).toBe("tok-abc"); + }); + + it("refuses anything else", () => { + for (const header of [ + undefined, + null, + "", + "bearer", + "tok-abc", + // Reversed: the scheme must be first, because the runtime echoes the + // first offered value back and the token must not be what comes back. + "tok-abc, bearer", + "basic, tok-abc", + "bearer, tok-abc, extra", + ]) { + expect(() => tokenFromSubprotocol(header)).toThrow(UnauthorizedError); + } + }); +}); + +/** + * That what the backend actually pushes matches the published contract. + * + * The payload is built from `printerProjection`, and the schema lives in + * continuum-schemas - two places that can drift. This is the test that + * notices. It is also the reason the frontend can be written against the + * schema rather than against a guess, which is how the last version of that + * store ended up consuming fields no backend ever sent. + */ +describe("the pushed messages match continuum/v1/app", () => { + function check(value: unknown) { + const ok = Value.Check(AppTelemetryMessageSchema, value); + if (!ok) { + const errors = [...Value.Errors(AppTelemetryMessageSchema, value)].map( + (e) => `${e.path || "/"}: ${e.message} (got ${JSON.stringify(e.value)})`, + ); + throw new Error(`validation failed:\n ${errors.join("\n ")}`); + } + } + + it("accepts a full telemetry push", () => { + check({ type: "printer.telemetry", printer: printer() }); + }); + + /** + * The state the schema's nullability exists for: a claimed printer whose + * gateway has never reported. Every reading is null, and a client that + * renders `?? 0` would draw this as a cold printer rather than an unknown + * one. + */ + it("accepts a printer that has never reported anything", () => { + check({ + type: "printer.telemetry", + printer: { + id: "3f2b1c5e-0000-4000-8000-000000000002", + farmId: "3f2b1c5e-0000-4000-8000-0000000000aa", + name: "never seen", + vendor: "bambu", + status: null, + vendorState: null, + nozzleTempC: null, + bedTempC: null, + progressPercent: null, + cameraStream: null, + lastSeenAt: null, + }, + }); + }); + + it("accepts the ready message", () => { + check({ type: "ready", visiblePrinters: 0 }); + check({ type: "ready", visiblePrinters: 12 }); + }); + + it("rejects the shape the old frontend invented", () => { + // Every field of this was made up against nothing: a `data` envelope, + // `printerId`, `nozzleTempC` at the top level, `progressPct`. Asserted + // so nobody reintroduces it by writing the client first. + expect( + Value.Check(AppTelemetryMessageSchema, { + type: "printer.telemetry", + data: { printerId: "p1p", nozzleTempC: 30, progressPct: 50 }, + }), + ).toBe(false); + }); + + it("rejects a push with no printer and an unknown message type", () => { + expect(Value.Check(AppTelemetryMessageSchema, { type: "printer.telemetry" })).toBe(false); + expect(Value.Check(AppTelemetryMessageSchema, { type: "printer.update", printer: printer() })).toBe(false); + }); +}); + +/** + * That the subprotocol check actually gates the upgrade, so an + * unauthenticated browser never gets a socket. + * + * Mirrors the real route's composition rather than importing it, because + * `routes/ws/telemetry.ts` pulls in `../../db` and so a live Postgres - the + * same constraint as `tests/edge_upgrade.test.ts`. What this pins is the + * framework behaviour the design rests on: that a throwing `resolve` fails + * the HTTP upgrade, and that the runtime echoes back the *first* offered + * subprotocol, which is why "bearer" is first and the token second. If it + * echoed the token instead, the credential would come back in a response + * header. `scripts/e2e-edge.sh` drives the real route. + */ +describe("the subprotocol gates the websocket upgrade", () => { + const app = new Elysia() + .resolve({ as: "scoped" }, ({ headers }): { viewer: string } => { + const token = tokenFromSubprotocol(headers["sec-websocket-protocol"]); + if (token !== "good-token") throw new UnauthorizedError("Invalid or expired access token"); + return { viewer: "user-1" }; + }) + .ws("/ws/telemetry", { + open(ws) { + ws.send(JSON.stringify({ viewer: (ws.data as { viewer: string }).viewer })); + }, + }) + .listen(0); + + const base = `localhost:${app.server!.port}`; + + function attempt(protocols?: string[]): Promise<{ result: string; negotiated?: string }> { + return new Promise((resolve) => { + const ws = protocols + ? new WebSocket(`ws://${base}/ws/telemetry`, protocols) + : new WebSocket(`ws://${base}/ws/telemetry`); + const done = (result: string) => { + const negotiated = ws.protocol; + try { + ws.close(); + } catch { + // Already closed by the server. + } + resolve({ result, negotiated }); + }; + ws.onmessage = (event) => done(`opened:${event.data}`); + ws.onerror = () => done("refused"); + ws.onclose = () => done("refused"); + setTimeout(() => done("timed out"), 5_000); + }); + } + + it("opens for a valid token and echoes back the scheme, not the token", async () => { + const { result, negotiated } = await attempt(["bearer", "good-token"]); + expect(result).toBe('opened:{"viewer":"user-1"}'); + // The credential must not be what comes back in the response header. + expect(negotiated).toBe("bearer"); + }); + + it("never opens a socket without a valid token", async () => { + expect((await attempt()).result).toBe("refused"); + expect((await attempt(["bearer", "forged"])).result).toBe("refused"); + expect((await attempt(["bearer"])).result).toBe("refused"); + // Scheme second: rejected, because the first value is what gets echoed. + expect((await attempt(["good-token", "bearer"])).result).toBe("refused"); + }); +});