From 7c6aff64765d51b3cb40d1ac996671b5c4d2f199 Mon Sep 17 00:00:00 2001 From: Jackson Miller Date: Fri, 6 Nov 2026 18:45:03 -0500 Subject: [PATCH] fix(server): a revoked or expired session's preview stream closes The preview stream socket authenticated its session only when it opened, so a revoked or expired device kept receiving frames and, with the preview-operate scope, kept sending input until it disconnected. Race the viewer against SessionStore.awaitInvalidation, as the RPC socket does. Co-Authored-By: Claude Opus 5.5 --- .../src/preview/ServerBrowserStream.test.ts | 95 ++++++++++++++++++- .../server/src/preview/ServerBrowserStream.ts | 8 ++ 2 files changed, 102 insertions(+), 1 deletion(-) diff --git a/apps/server/src/preview/ServerBrowserStream.test.ts b/apps/server/src/preview/ServerBrowserStream.test.ts index c4a709ed8acc4..3429b37f3b18e 100644 --- a/apps/server/src/preview/ServerBrowserStream.test.ts +++ b/apps/server/src/preview/ServerBrowserStream.test.ts @@ -12,6 +12,7 @@ import { type AuthEnvironmentScope, } from "@t3tools/contracts"; import * as Context from "effect/Context"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; @@ -21,6 +22,7 @@ import * as Schema from "effect/Schema"; import { HttpRouter, HttpServer } from "effect/http"; import * as EnvironmentAuth from "../auth/EnvironmentAuth.ts"; +import * as SessionStore from "../auth/SessionStore.ts"; import * as PreviewBrowserHost from "./PreviewBrowserHost.ts"; import * as ServerBrowser from "./ServerBrowser.ts"; import { routeLayer } from "./ServerBrowserStream.ts"; @@ -30,6 +32,7 @@ const platformLayer = NodeHttpPlatform.layer.pipe(Layer.provideMerge(NodeService const makeAuth = ( scopes: ReadonlyArray, error?: EnvironmentAuth.ServerAuthCredentialError, + invalidation: Effect.Effect = Effect.never, ) => { const requests: string[] = []; const layer = Layer.mock(EnvironmentAuth.EnvironmentAuth, { @@ -45,7 +48,17 @@ const makeAuth = ( }); }, }); - return { layer, requests }; + return { + layer: Layer.merge( + layer, + Layer.mock(SessionStore.SessionStore, { + cookieName: "test-session", + legacyCookieName: undefined, + awaitInvalidation: () => invalidation, + }), + ), + requests, + }; }; const mutations = [ @@ -152,6 +165,86 @@ it.effect.each([ }).pipe(Effect.scoped), ); +it.effect.each([false, true])( + "closes an established viewer when its session is invalidated (operate=%s)", + (operate) => + Effect.gen(function* () { + const invalidated = yield* Deferred.make(); + const detached = yield* Deferred.make(); + const inputStarted = yield* Deferred.make(); + const inputInterrupted = yield* Deferred.make(); + const output = yield* Queue.make(); + const auth = makeAuth( + operate + ? [AuthOrchestrationReadScope, AuthPreviewOperateScope] + : [AuthOrchestrationReadScope], + undefined, + Deferred.await(invalidated), + ); + const browser = ServerBrowser.ServerBrowser.of({ + clearProfile: () => Effect.void, + profileReporter: Effect.succeed(() => Effect.void), + openDownload: () => Effect.succeedNone, + answerFileChooser: () => Effect.succeed(false), + attachViewer: () => + Effect.gen(function* () { + yield* Effect.addFinalizer(() => Deferred.succeed(detached, undefined)); + return { + output, + input: () => + Deferred.succeed(inputStarted, undefined).pipe( + Effect.andThen(Effect.never), + Effect.onInterrupt(() => Deferred.succeed(inputInterrupted, undefined)), + ), + }; + }), + }); + const services = yield* Layer.build( + HttpRouter.serve( + routeLayer.pipe( + Layer.provide(Layer.succeed(ServerBrowser.ServerBrowser, browser)), + Layer.provide(platformLayer), + ), + { disableListenLog: true }, + ).pipe(Layer.provideMerge(NodeHttpServer.layerTest), Layer.provide(auth.layer)), + ); + const server = Context.get(services, HttpServer.HttpServer); + const origin = HttpServer.formatAddress(server.address).replace(/^http/, "ws"); + const received = Promise.withResolvers(); + const closed = Promise.withResolvers(); + const frames: unknown[] = []; + const socket = yield* Effect.acquireRelease( + Effect.sync(() => { + const socket = new WebSocket( + `${origin}/api/preview-stream/ws?threadId=thread&tabId=tab&wsTicket=one-use-ticket`, + ); + socket.addEventListener("message", (event) => { + frames.push(event.data); + received.resolve(); + }); + socket.addEventListener("close", () => closed.resolve()); + socket.addEventListener("error", () => received.reject(new Error("stream failed"))); + return socket; + }), + (socket) => Effect.sync(() => socket.close()), + ); + yield* Queue.offer(output, { _tag: "frame", data: new Uint8Array([1]), ack: Effect.void }); + yield* Effect.promise(() => received.promise); + if (operate) { + socket.send(JSON.stringify({ type: "takeControl" })); + yield* Deferred.await(inputStarted); + } + yield* Deferred.succeed(invalidated, undefined); + // The viewer detaches, any running input is interrupted, and the socket closes. + yield* Deferred.await(detached); + yield* Effect.promise(() => closed.promise); + if (operate) yield* Deferred.await(inputInterrupted); + yield* Queue.offer(output, { _tag: "frame", data: new Uint8Array([2]), ack: Effect.void }); + expect(frames).toHaveLength(1); + expect(socket.readyState).toBe(WebSocket.CLOSED); + }).pipe(Effect.scoped), +); + it.effect("sends replies ahead of stale frames and only the newest frame after an ack", () => Effect.gen(function* () { const output = yield* Queue.make(); diff --git a/apps/server/src/preview/ServerBrowserStream.ts b/apps/server/src/preview/ServerBrowserStream.ts index e46a0c76adb53..137692f8e23a4 100644 --- a/apps/server/src/preview/ServerBrowserStream.ts +++ b/apps/server/src/preview/ServerBrowserStream.ts @@ -20,6 +20,7 @@ import { } from "effect/http"; import * as Socket from "effect/socket/Socket"; +import * as SessionStore from "../auth/SessionStore.ts"; import { authenticateMediaRequest } from "../auth/http.ts"; import { assetResponseHeaders } from "../http.ts"; import * as PreviewBrowserHost from "./PreviewBrowserHost.ts"; @@ -71,6 +72,7 @@ const makeHandler = Effect.fnUntraced(function* (browser: ServerBrowser.ServerBr return HttpServerResponse.text("Not Found", { status: 404 }); } const session = yield* authenticateMediaRequest(AuthOrchestrationReadScope); + const sessions = yield* SessionStore.SessionStore; const params = url.value.searchParams; const canOperate = session.scopes.includes(AuthPreviewOperateScope) && params.get("interactive") !== "false"; @@ -201,6 +203,12 @@ const makeHandler = Effect.fnUntraced(function* (browser: ServerBrowser.ServerBr return yield* Effect.raceFirst(Effect.forever(sendOutput), Effect.forever(receiveInput)); }), ).pipe( + // Like the RPC socket: a revoked or expired session ends the stream and its input. + Effect.raceFirst( + sessions + .awaitInvalidation(session.sessionId, session.proofKeyThumbprint) + .pipe(Effect.as(HttpServerResponse.empty())), + ), Effect.catchTags({ ServerBrowserLaunchError: (error) => Effect.logWarning("server preview browser failed to start", { cause: error.cause }).pipe(