diff --git a/packages/reflex-base/news/+session-cookie-bootstrap.feature.md b/packages/reflex-base/news/+session-cookie-bootstrap.feature.md new file mode 100644 index 00000000000..dae61bdf544 --- /dev/null +++ b/packages/reflex-base/news/+session-cookie-bootstrap.feature.md @@ -0,0 +1 @@ +Establish browser sessions over the WebSocket while hydration begins, then persist them through HTTP-only cookies with automatic refresh. Browser storage is isolated for each backend application. Idle tabs renew their session cookies automatically; transient cookie-exchange failures retry without blocking WebSocket events, and queued uploads remain cancellable. diff --git a/packages/reflex-base/src/reflex_base/.templates/web/utils/helpers/upload.js b/packages/reflex-base/src/reflex_base/.templates/web/utils/helpers/upload.js index 2e489c23059..f8ee1187f94 100644 --- a/packages/reflex-base/src/reflex_base/.templates/web/utils/helpers/upload.js +++ b/packages/reflex-base/src/reflex_base/.templates/web/utils/helpers/upload.js @@ -14,6 +14,7 @@ import { parseJson } from "$/utils/helpers/json"; * @param refs The refs object to store the abort controller in. * @param getBackendURL Function to get the backend URL. * @param getToken Function to get the Reflex token. + * @param waitForSession Wait for the session cookie before starting HTTP traffic. * * @returns The response from posting to the UPLOADURL endpoint. */ @@ -28,6 +29,7 @@ export const uploadFiles = async ( refs, getBackendURL, getToken, + waitForSession, ) => { files = files ?? []; @@ -38,6 +40,29 @@ export const uploadFiles = async ( return false; } + const controller = new AbortController(); + refs[upload_ref_name] = controller; + if (waitForSession) { + let abortWaiting; + const cancelled = new Promise((resolve) => { + abortWaiting = resolve; + controller.signal.addEventListener("abort", abortWaiting, { once: true }); + }); + try { + await Promise.race([waitForSession(), cancelled]); + } catch (error) { + console.log("Upload session error:", error.message); + delete refs[upload_ref_name]; + return false; + } finally { + controller.signal.removeEventListener("abort", abortWaiting); + } + } + if (controller.signal.aborted || !socket.connected) { + delete refs[upload_ref_name]; + return false; + } + // Track how many partial updates have been processed for this upload. let resp_idx = 0; const eventHandler = (progressEvent) => { @@ -73,7 +98,6 @@ export const uploadFiles = async ( }); }; - const controller = new AbortController(); const formdata = new FormData(); // Bound handler args are sent as a JSON form field that must precede the @@ -89,7 +113,6 @@ export const uploadFiles = async ( }); // Send the file to the server. - refs[upload_ref_name] = controller; return new Promise((resolve, reject) => { const xhr = new XMLHttpRequest(); @@ -154,6 +177,7 @@ export const uploadFiles = async ( // Configure and send request xhr.open("POST", getBackendURL(env.UPLOAD)); + xhr.withCredentials = true; xhr.setRequestHeader("Reflex-Client-Token", getToken()); xhr.setRequestHeader("Reflex-Event-Handler", handler); // Instrumentation hook (installed by reflex-otel): may add trace headers. diff --git a/packages/reflex-base/src/reflex_base/.templates/web/utils/state.js b/packages/reflex-base/src/reflex_base/.templates/web/utils/state.js index 2336a33340a..4244f23ae50 100644 --- a/packages/reflex-base/src/reflex_base/.templates/web/utils/state.js +++ b/packages/reflex-base/src/reflex_base/.templates/web/utils/state.js @@ -42,12 +42,8 @@ const SAME_DOMAIN_HOSTNAMES = ["localhost", "0.0.0.0", "::", "0:0:0:0:0:0:0:0"]; // Global variable to hold the token. let token; -// A token generated for the transport warmed up before the app mounted. It is -// saved to the session storage by getToken, once the mounted app connects. -let unsavedToken; - -// Key for the token in the session storage. -const TOKEN_KEY = "token"; +// Share cookie exchange and refresh requests in this tab. +let initialSessionExchange; // create cookie instance const cookies = new Cookies(); @@ -87,7 +83,7 @@ const locationRef = { }; /** - * Generate a UUID (Used for session tokens). + * Generate a UUID for component identifiers. * Taken from: https://stackoverflow.com/questions/105034/how-do-i-create-a-guid-uuid * @returns A UUID. */ @@ -116,25 +112,17 @@ export const getToken = () => { return token; } if (typeof window !== "undefined") { - if (!window.sessionStorage.getItem(TOKEN_KEY)) { - window.sessionStorage.setItem(TOKEN_KEY, unsavedToken ?? generateUUID()); - } - token = window.sessionStorage.getItem(TOKEN_KEY); + token = window.sessionStorage.getItem(clientTokenStorageKey()) || ""; } - return token; + return token || ""; }; -/** - * Get the token for the current session without saving a new one. - * - * A new token only reaches the session storage when the mounted app - * connects, so anything waiting for it there sees the rendered page. - * @returns The saved token, or a new one that getToken saves later. - */ +/** Get the saved client token to warm the transport before the app mounts. */ const peekToken = () => token || - window.sessionStorage.getItem(TOKEN_KEY) || - (unsavedToken ??= generateUUID()); + window.sessionStorage.getItem(clientTokenStorageKey()) || + window.sessionStorage.getItem(LEGACY_TOKEN_STORAGE_KEY) || + ""; /** * Get the URL for the backend server @@ -166,6 +154,65 @@ export const getBackendURL = (url_str) => { return endpoint; }; +/** Namespace browser state by the backend URL, including its port and path. */ +const sessionStorageKey = () => `reflex:${getBackendURL(EVENTURL).href}`; +const LEGACY_TOKEN_STORAGE_KEY = "token"; +const clientTokenStorageKey = () => `${sessionStorageKey()}:token`; +const sessionRefreshStorageKey = () => `${sessionStorageKey()}:expires_at`; + +/** Store only the non-secret tab routing token in JavaScript-accessible storage. */ +const setToken = (value) => { + token = value; + window.sessionStorage.setItem(clientTokenStorageKey(), value); +}; + +/** Wait until credentialed HTTP calls can use the initial session cookie. */ +export const waitForSession = async () => { + await initialSessionExchange; +}; + +/** Exchange an in-band credential or refresh the HTTP-only session cookie. */ +const ensureSession = async (sessionCredential, controller) => { + const endpoint = getBackendURL(EVENTURL); + endpoint.protocol = endpoint.protocol === "wss:" ? "https:" : "http:"; + endpoint.pathname = endpoint.pathname.replace( + /\/_event\/?$/, + "/_reflex/session", + ); + endpoint.search = ""; + const headers = { "Content-Type": "application/json" }; + const previousToken = getToken(); + if (previousToken) headers["Reflex-Client-Token"] = previousToken; + // Abort a stalled cookie exchange after ten seconds. + const timeout = window.setTimeout(() => controller.abort(), 10000); + try { + const response = await window.fetch(endpoint.href, { + method: "POST", + credentials: "include", + headers, + body: JSON.stringify( + sessionCredential ? { session_token: sessionCredential } : {}, + ), + signal: controller.signal, + }); + if (!response.ok) { + throw new Error(`Session request failed: ${response.status}`); + } + const session = await response.json(); + if ( + typeof session.client_token !== "string" || + !session.client_token || + !Number.isFinite(session.expires_in) || + session.expires_in <= 0 + ) { + throw new Error("Invalid session response"); + } + return session; + } finally { + window.clearTimeout(timeout); + } +}; + /** * Check if the backend is disabled. * @@ -192,6 +239,7 @@ const createSocket = (endpoint, transports, token) => protocols: [reflexEnvironment.version], autoUnref: false, autoConnect: false, + withCredentials: true, query: { token }, reconnection: false, }); @@ -222,6 +270,15 @@ if (typeof window !== "undefined") { return; } try { + // A cold transport would capture cookies before the shared session lock. + if ( + !( + Number(window.localStorage.getItem(sessionRefreshStorageKey())) > + Date.now() + ) + ) { + return; + } warmSocket = createSocket( getBackendURL(EVENTURL), [env.TRANSPORT], @@ -716,7 +773,8 @@ export const applyRestEvent = async (event, socket, navigate, params) => { for (const name of event.payload.__reflex_event_arg_names ?? []) { extra_args[name] = event.payload[name]; } - // Start upload, but do not wait for it, which would block other events. + // The helper reserves its cancellation controller before waiting for cookies. + // Do not await it here: ordinary WebSocket events must keep draining. uploadFiles( event.name, event.payload.files, @@ -728,6 +786,7 @@ export const applyRestEvent = async (event, socket, navigate, params) => { refs, getBackendURL, getToken, + waitForSession, ); } }; @@ -891,7 +950,7 @@ export const connect = async ( // Socket already allocated, just reconnect it if needed. if (socket.current) { if (!socket.current.connected) { - socket.current.reconnect(); + await socket.current.reconnect(); } return; } @@ -931,8 +990,48 @@ export const connect = async ( warmSocket ?? createSocket(endpoint, transports, session_token); warmSocket = null; cancelWarmup(); - socket.current.auth = bootAuth(true); - socket.current.wait_connect = !socket.current.connected; + socket.current.wait_connect = false; + let firstConnect = true; + let finishBootstrap; + let exchangingSession = false; + let reconnectingSession = false; + let sessionTimer; + let sessionController; + let resolveSession; + let rejectSession; + let sessionFailures = 0; + const currentSocket = socket.current; + const requestSessionRefresh = () => { + window.clearTimeout(sessionTimer); + if (socket.current === currentSocket && currentSocket.connected) { + currentSocket.emit("session_refresh"); + } + }; + const finishSessionExchange = () => { + resolveSession?.(); + resolveSession = rejectSession = undefined; + initialSessionExchange = undefined; + }; + const stopSessionExchange = (preserveReadiness = false) => { + window.clearTimeout(sessionTimer); + sessionController?.abort(); + sessionController = undefined; + exchangingSession = false; + if (!preserveReadiness) { + reconnectingSession = false; + rejectSession?.(new Error("Session connection closed")); + resolveSession = rejectSession = undefined; + initialSessionExchange = undefined; + } + }; + const releaseBootstrap = () => { + finishBootstrap?.(); + finishBootstrap = null; + }; + socket.current.releaseSessionBootstrap = () => { + stopSessionExchange(); + releaseBootstrap(); + }; // Ensure undefined fields in events are sent as null instead of removed socket.current.io.encoder.replacer = (k, v) => (v === undefined ? null : v); socket.current.io.decoder.tryParse = (str) => { @@ -943,17 +1042,53 @@ export const connect = async ( } }; // Set up a reconnect helper function - socket.current.reconnect = () => { + socket.current.reconnect = async () => { if ( socket.current && !socket.current.connected && !socket.current.wait_connect ) { socket.current.wait_connect = true; - socket.current.io.opts.query = { token: getToken() }; // Update token for reconnect. - // A reconnect rehydrates in full: the reducers no longer hold the defaults. - socket.current.auth = bootAuth(false); - socket.current.connect(); + const current = socket.current; + const start = () => { + if (socket.current !== current) { + releaseBootstrap(); + return; + } + current.io.opts.query = { + token: + getToken() || + window.sessionStorage.getItem(LEGACY_TOKEN_STORAGE_KEY) || + "", + }; + current.auth = bootAuth(firstConnect); + current.connect(); + }; + const key = sessionStorageKey(); + if ( + window.navigator?.locks && + !( + Number(window.localStorage.getItem(sessionRefreshStorageKey())) > + Date.now() + ) + ) { + // Only cold tabs wait. The first socket hydrates while its in-band + // credential becomes a cookie; the next tab then shares that cookie. + window.navigator.locks + .request( + `${key}:session`, + () => + new Promise((resolve) => { + finishBootstrap = resolve; + start(); + }), + ) + .catch((error) => { + if (socket.current === current) onConnectError(error); + }); + } else { + start(); + } } }; @@ -973,7 +1108,7 @@ export const connect = async ( console.log("Socket is disconnected, attempting to reconnect "); socket.current.reconnect(); } else { - console.log("Socket is reconnected "); + requestSessionRefresh(); } } } @@ -1002,6 +1137,15 @@ export const connect = async ( // Once the socket is open, hydrate the page. socket.current.on("connect", async () => { + if (reconnectingSession) { + reconnectingSession = false; + if (!exchangingSession) finishSessionExchange(); + } + if (!exchangingSession) { + releaseBootstrap(); + requestSessionRefresh(); + } + firstConnect = false; socket.current.wait_connect = false; reported_missing_substates.clear(); setConnectErrors([]); @@ -1014,7 +1158,8 @@ export const connect = async ( } }); - socket.current.on("connect_error", (error) => { + const onConnectError = (error) => { + releaseBootstrap(); socket.current.wait_connect = false; let n_connect_errors = 0; setConnectErrors((connectErrors) => { @@ -1024,13 +1169,20 @@ export const connect = async ( }); window.setTimeout(() => { if (socket.current && !socket.current.connected) { - socket.current.reconnect(); + return socket.current.reconnect(); } }, 200 * n_connect_errors); // Incremental backoff - }); + }; + socket.current.on("connect_error", onConnectError); socket.current.on("disconnect", (reason, details) => { + stopSessionExchange( + reconnectingSession && reason === "io client disconnect", + ); + releaseBootstrap(); + firstConnect = false; socket.current.wait_connect = false; + socket.current.rehydrate = true; window.__reflex_otel?.onSocketDisconnect?.(reason); const try_reconnect = reason !== "io server disconnect" && reason !== "io client disconnect"; @@ -1117,13 +1269,66 @@ export const connect = async ( } }); socket.current.on("new_token", async (new_token) => { - token = new_token; - window.sessionStorage.setItem(TOKEN_KEY, new_token); + setToken(new_token); }); + socket.current.on("session_refresh", requestSessionRefresh); + socket.current.on("session_token", async (sessionCredential) => { + if (sessionController) return; + const controller = new AbortController(); + sessionController = controller; + exchangingSession = true; + window.clearTimeout(sessionTimer); + if (!resolveSession) { + initialSessionExchange = new Promise((resolve, reject) => { + resolveSession = resolve; + rejectSession = reject; + }); + // Readiness may have no HTTP consumer when the connection closes. + initialSessionExchange.catch(() => {}); + } + try { + const previousToken = getToken(); + const session = await ensureSession(sessionCredential, controller); + if (sessionController !== controller) return; + setToken(session.client_token); + const refreshDelay = + (session.refresh_in ?? session.expires_in / 2) * 1000; + window.localStorage.setItem( + sessionRefreshStorageKey(), + Date.now() + refreshDelay, + ); + sessionFailures = 0; + setConnectErrors([]); + sessionTimer = window.setTimeout(requestSessionRefresh, refreshDelay); + if (currentSocket.connected && getToken() !== previousToken) { + // Keep waiting HTTP calls queued until this cookie's socket is accepted. + reconnectingSession = true; + currentSocket.disconnect(); + await currentSocket.reconnect(); + } else { + finishSessionExchange(); + } + } catch (error) { + if (sessionController !== controller) return; + window.localStorage.removeItem(sessionRefreshStorageKey()); + setConnectErrors((errors) => [...errors.slice(-9), error]); + // Retry on the live socket after one second, backing off to thirty seconds. + // Each attempt requests a fresh credential, including after cookie expiry. + const delay = Math.min(30000, 1000 * 2 ** Math.min(sessionFailures++, 5)); + sessionTimer = window.setTimeout(requestSessionRefresh, delay); + } finally { + if (sessionController === controller) { + sessionController = undefined; + exchangingSession = false; + releaseBootstrap(); + } + } + }); + socket.current.on(STATE_MISMATCH_EVENT, handleStateMismatch); document.addEventListener("visibilitychange", checkVisibility); - socket.current.connect(); + await socket.current.reconnect(); }; /** @@ -1498,6 +1703,7 @@ export const useEventLoop = ( return () => { mounted.current = false; if (socket.current) { + socket.current.releaseSessionBootstrap?.(); socket.current.disconnect(); socket.current.off(); socket.current = null; diff --git a/tests/units/compiler/state_js.test.mjs b/tests/units/compiler/state_js.test.mjs index df8472228fb..f895e0dc765 100644 --- a/tests/units/compiler/state_js.test.mjs +++ b/tests/units/compiler/state_js.test.mjs @@ -10,6 +10,7 @@ const source = await readFile( ), "utf8", ); +const tokenKey = "reflex:ws://localhost:8000/_event:token"; /** Evaluate the actual frontend module with controlled transport and browser APIs. */ async function setup({ @@ -27,11 +28,14 @@ async function setup({ const updates = []; const window = new EventTarget(); window.location = new URL("http://localhost:3000/page?query=value"); - const storage = new Map(storedToken ? [["token", storedToken]] : []); + const storage = new Map(storedToken ? [[tokenKey, storedToken]] : []); window.sessionStorage = { getItem: (key) => storage.get(key), setItem: (key, value) => storage.set(key, value), }; + window.localStorage = { + getItem: () => (storedToken ? String(Date.now() + 3600000) : undefined), + }; const document = new EventTarget(); document.cookie = disabled ? "backend-enabled=false" : ""; document.visibilityState = hidden ? "hidden" : "visible"; @@ -50,6 +54,7 @@ async function setup({ clearTimeout: (fn) => timers.delete(fn), }); window.setTimeout = context.setTimeout; + window.clearTimeout = context.clearTimeout; function io(url, opts) { const handlers = new Map(); @@ -70,6 +75,7 @@ async function setup({ on(name, handler) { handlers.set(name, handler); }, + emit() {}, connect() { // A ready transport can deliver these synchronously on namespace connect. for (const name of [ @@ -196,18 +202,15 @@ test("warm the transport without hydrating, then reuse it with every handler att assert.deepEqual(app.updates, [7]); assert.deepEqual(app.firstHydrates, [true]); assert.equal(app.timers.size, 0); - assert.equal(app.window.sessionStorage.getItem("token"), "assigned-token"); + assert.equal(app.window.sessionStorage.getItem(tokenKey), "assigned-token"); assert.equal(app.socket.current.auth.event.router_data.pathname, "/page"); }); -test("a new session's token is saved only once the mounted app connects", async () => { +test("a cold session stores only the token assigned by the server", async () => { const app = await setup({ storedToken: null }); app.flush(); - const warm = app.sockets[0]; - const token = warm.io.opts.query.token; - assert.ok(token); - // Waiting for the token in session storage must mean the app has mounted. - assert.equal(app.window.sessionStorage.getItem("token"), undefined); + assert.equal(app.sockets.length, 0); + assert.equal(app.window.sessionStorage.getItem(tokenKey), undefined); const saved = []; const setItem = app.window.sessionStorage.setItem; app.window.sessionStorage.setItem = (key, value) => { @@ -215,15 +218,15 @@ test("a new session's token is saved only once the mounted app connects", async setItem(key, value); }; await app.connect(); - assert.equal(app.socket.current, warm); - assert.deepEqual(saved, [token, "assigned-token"]); + assert.equal(app.socket.current.io.opts.query.token, ""); + assert.deepEqual(saved, ["assigned-token"]); }); test("a warm transport carrying another session's token is replaced", async () => { const app = await setup(); app.flush(); const warm = app.sockets[0]; - app.window.sessionStorage.setItem("token", "replaced-token"); + app.window.sessionStorage.setItem(tokenKey, "replaced-token"); await app.connect(); assert.equal(warm.disconnects, 1); assert.equal(app.sockets.length, 2); diff --git a/tests/units/reflex_base/client_event_queue_runtime.mjs b/tests/units/reflex_base/client_event_queue_runtime.mjs index 2befa24459f..9e04ee66357 100644 --- a/tests/units/reflex_base/client_event_queue_runtime.mjs +++ b/tests/units/reflex_base/client_event_queue_runtime.mjs @@ -3,14 +3,27 @@ import { SourceTextModule, SyntheticModule } from "node:vm"; /** Evaluate the complete frontend module with isolated dependency stubs. */ export async function createQueueRuntime( source, - { uploadFiles, throttle, dependencies: dependencyOverrides } = {}, + { + uploadFiles, + throttle, + dependencies: dependencyOverrides, + io, + env, + browser, + initialState = {}, + } = {}, ) { const unused = () => { throw new Error("Unexpected frontend dependency in queue test"); }; const app = { - initialEvents: () => [], - initialState: {}, + initialEvents: (first) => [ + { + name: "reflex___state.test.hydrate_and_load", + payload: first ? { hashes: { test_state: "compiled-defaults" } } : {}, + }, + ], + initialState, onLoadInternalEvent: unused, state_name: "test_state", exception_state_name: "test_exception_state", @@ -23,10 +36,11 @@ export async function createQueueRuntime( document: {}, localStorage: { clear() {}, removeItem() {} }, sessionStorage: { clear() {}, removeItem() {} }, + ...browser, }, - "socket.io-client": { default: unused }, + "socket.io-client": { default: io ?? unused }, mergician: { mergician: unused }, - "$/env.json": { default: {} }, + "$/env.json": { default: env ?? {} }, "$/reflex.json": { default: {} }, "universal-cookie": { default: class { diff --git a/tests/units/reflex_base/client_session.mjs b/tests/units/reflex_base/client_session.mjs new file mode 100644 index 00000000000..1c9e981afa7 --- /dev/null +++ b/tests/units/reflex_base/client_session.mjs @@ -0,0 +1,574 @@ +import assert from "node:assert/strict"; +import fs from "node:fs"; +import { test } from "node:test"; +import { createQueueRuntime } from "./client_event_queue_runtime.mjs"; + +const source = fs.readFileSync(process.argv[2], "utf8"); +const tick = () => new Promise((resolve) => setImmediate(resolve)); + +/** Create browser storage with Web Storage's missing-value behavior. */ +function storage() { + const values = new Map(); + return { + getItem: (key) => values.get(key) ?? null, + setItem: (key, value) => values.set(key, String(value)), + removeItem: (key) => values.delete(key), + clear: () => values.clear(), + values, + }; +} + +/** Serialize callbacks as the browser's exclusive Web Locks do. */ +function locks() { + let tail = Promise.resolve(); + return { + request: (_name, callback) => { + const request = tail.then(callback); + tail = request.catch(() => {}); + return request; + }, + }; +} + +/** Evaluate the frontend with a browser and manually driven Socket.IO peer. */ +async function browserSession({ + fetch, + localStorage = storage(), + sessionStorage = storage(), + webLocks, + warm = false, + eventURL = "ws://localhost:8000/prefix/_event", +} = {}) { + const callbacks = new Map(); + const timeouts = new Map(); + const delays = new Map(); + const errors = []; + const connections = []; + const warmups = []; + const requests = []; + const emitted = []; + const uploads = []; + const socket = { current: null }; + const window = { + location: { hostname: "localhost", protocol: "http:" }, + navigator: { locks: webLocks }, + localStorage, + sessionStorage, + addEventListener() {}, + removeEventListener() {}, + setTimeout(callback, delay) { + const id = Symbol(); + timeouts.set(id, callback); + delays.set(id, delay); + return id; + }, + clearTimeout: (id) => timeouts.delete(id), + async fetch(url, options) { + requests.push([String(url), options]); + return fetch + ? await fetch(url, options) + : { + ok: true, + json: async () => ({ + client_token: options.headers["Reflex-Client-Token"], + expires_in: 3600, + }), + }; + }, + }; + const runtime = await createQueueRuntime(source, { + uploadFiles: async (...args) => { + try { + await args[10]?.(); + uploads.push(args); + } catch {} + }, + browser: { + window, + document: { + cookie: "", + visibilityState: "visible", + addEventListener() {}, + removeEventListener() {}, + }, + localStorage, + sessionStorage, + }, + env: { EVENT: eventURL, TRANSPORT: "websocket" }, + initialState: warm + ? { test_state: {}, test_exception_state: {} } + : undefined, + io: (_url, options) => { + const current = { + connected: false, + io: { + opts: options, + encoder: {}, + decoder: {}, + open(callback) { + warmups.push(options.query.token); + callback(); + }, + }, + on: (event, callback) => callbacks.set(event, callback), + emit: (event, payload) => emitted.push([event, payload]), + connect() { + connections.push(this.io.opts.query.token); + return this; + }, + disconnect() { + this.connected = false; + callbacks.get("disconnect")?.("io client disconnect"); + return this; + }, + }; + if (options.autoConnect !== false) current.connect(); + return current; + }, + }); + return { + runtime, + socket, + callbacks, + timeouts, + delays, + connections, + warmups, + requests, + emitted, + uploads, + errors, + localStorage, + sessionStorage, + connect: async () => { + await runtime.connect( + socket, + {}, + ["websocket"], + (update) => { + errors.splice( + 0, + errors.length, + ...(typeof update === "function" ? update(errors) : update), + ); + }, + {}, + () => {}, + { current: {} }, + ); + await tick(); + }, + async accept(token = "bound-tab", credential) { + await callbacks.get("new_token")(token); + const exchange = credential + ? callbacks.get("session_token")(credential) + : undefined; + socket.current.connected = true; + await callbacks.get("connect")(); + return { exchange }; + }, + }; +} + +test("cold websocket hydrates before cookie exchange completes", async () => { + let finish; + const tab = await browserSession({ + fetch: () => + new Promise((resolve) => { + finish = resolve; + }), + }); + assert.equal(tab.runtime.getToken(), ""); + await tab.connect(); + assert.deepEqual(tab.connections, [""]); + assert.equal( + tab.socket.current.auth.event.name, + "reflex___state.test.hydrate_and_load", + ); + assert.deepEqual(tab.socket.current.auth.event.payload.hashes, { + test_state: "compiled-defaults", + }); + assert.equal(tab.requests.length, 0); + const { exchange } = await tab.accept("bound-tab", "signed-credential"); + assert.equal(tab.runtime.event_queue.length, 0); + tab.runtime.event_queue.push({ name: "reflex___state.test.after_hydrate" }); + await tab.runtime.processEvent(tab.socket.current, () => {}, { current: {} }); + assert.equal(tab.emitted.length, 1); + const [url, options] = tab.requests[0]; + assert.equal(url, "http://localhost:8000/prefix/_reflex/session"); + assert.equal(options.credentials, "include"); + assert.equal(options.headers["Content-Type"], "application/json"); + assert.deepEqual(JSON.parse(options.body), { + session_token: "signed-credential", + }); + assert.equal(tab.socket.current.io.opts.withCredentials, true); + finish({ + ok: true, + json: async () => ({ client_token: "bound-tab", expires_in: 3600 }), + }); + await exchange; + assert.equal(tab.runtime.getToken(), "bound-tab"); + assert.equal(tab.sessionStorage.getItem("token"), null); + for (const value of [ + ...tab.sessionStorage.values.values(), + ...tab.localStorage.values.values(), + ]) { + assert.notEqual(value, "signed-credential"); + } +}); + +test("legacy tokens are presented for validation without being adopted", async () => { + const sessionStorage = storage(); + sessionStorage.setItem("token", "legacy-unbound"); + const tab = await browserSession({ sessionStorage }); + assert.equal(tab.runtime.getToken(), ""); + await tab.connect(); + assert.deepEqual(tab.connections, ["legacy-unbound"]); + await tab.accept("bound-tab"); + assert.equal(tab.runtime.getToken(), "bound-tab"); + assert.equal(sessionStorage.getItem("token"), "legacy-unbound"); +}); + +test("cold tabs wait for the first cookie while its socket hydrates", async () => { + const webLocks = locks(); + const localStorage = storage(); + let finish; + const first = await browserSession({ + webLocks, + localStorage, + fetch: () => + new Promise((resolve) => { + finish = resolve; + }), + }); + const second = await browserSession({ webLocks, localStorage }); + await Promise.all([first.connect(), first.connect(), second.connect()]); + assert.deepEqual(first.connections, [""]); + assert.deepEqual(second.connections, []); + const { exchange } = await first.accept("first-tab", "credential"); + assert.equal(first.socket.current.connected, true); + assert.deepEqual(second.connections, []); + finish({ + ok: true, + json: async () => ({ client_token: "first-tab", expires_in: 3600 }), + }); + await exchange; + await tick(); + assert.deepEqual(second.connections, [""]); + await second.accept("second-tab"); + assert.equal(second.requests.length, 0); +}); + +test("fresh hint reconnects directly without HTTP or a lock", async () => { + const first = await browserSession(); + await first.connect(); + await (await first.accept("existing-tab", "credential")).exchange; + const tab = await browserSession({ + warm: true, + localStorage: first.localStorage, + sessionStorage: first.sessionStorage, + webLocks: { + request() { + throw new Error("fresh session must not wait"); + }, + }, + }); + await tick(); + assert.deepEqual(tab.warmups, ["existing-tab"]); + assert.deepEqual(tab.connections, []); + await tab.connect(); + assert.deepEqual(tab.connections, ["existing-tab"]); + assert.equal(tab.requests.length, 0); + await tab.accept("existing-tab"); + tab.socket.current.disconnect(); + await tab.socket.current.reconnect(); + assert.deepEqual(tab.connections, ["existing-tab", "existing-tab"]); + assert.equal(tab.socket.current.auth.event.payload.hashes, undefined); + assert.equal(tab.requests.length, 0); +}); + +test("cold transport cannot capture cookies before the session lock", async () => { + const tab = await browserSession({ warm: true, webLocks: locks() }); + await tick(); + assert.deepEqual(tab.warmups, []); + await tab.connect(); + assert.deepEqual(tab.connections, [""]); + await (await tab.accept("bound-tab", "credential")).exchange; +}); + +test("cookie-wins exchange adopts its token and reconnects the provisional socket", async () => { + const tab = await browserSession({ + fetch: async () => ({ + ok: true, + json: async () => ({ client_token: "cookie-tab", expires_in: 3600 }), + }), + }); + await tab.connect(); + await (await tab.accept("provisional-tab", "credential")).exchange; + assert.deepEqual(tab.connections, ["", "cookie-tab"]); + assert.equal(tab.runtime.getToken(), "cookie-tab"); +}); + +for (const teardown of [false, true]) { + test(`cookie-wins readiness waits for reconnect and respects teardown=${teardown}`, async () => { + let finish; + const tab = await browserSession({ + fetch: () => + new Promise((resolve) => { + finish = resolve; + }), + }); + await tab.connect(); + const { exchange } = await tab.accept("provisional-tab", "credential"); + let ready = false; + const pending = tab.runtime.waitForSession().then(() => { + ready = true; + }); + pending.catch(() => {}); + await tab.runtime.applyRestEvent( + { name: "upload", handler: "uploadFiles", payload: { files: [] } }, + tab.socket.current, + ); + finish({ + ok: true, + json: async () => ({ client_token: "cookie-tab", expires_in: 3600 }), + }); + await exchange; + await tick(); + assert.equal(tab.socket.current.connected, false); + assert.equal(ready, false); + assert.equal(tab.uploads.length, 0); + if (teardown) { + tab.socket.current.releaseSessionBootstrap(); + await assert.rejects(pending, /Session connection closed/); + await tick(); + assert.equal(tab.uploads.length, 0); + } else { + await tab.accept("cookie-tab"); + await pending; + await tick(); + assert.equal(ready, true); + assert.equal(tab.uploads.length, 1); + assert.equal(tab.runtime.getToken(), "cookie-tab"); + } + }); +} + +test("refresh and duplicate-tab updates preserve a socket with matching identity", async () => { + const tab = await browserSession(); + await tab.connect(); + await tab.accept("first-tab"); + await tab.callbacks.get("new_token")("duplicated-tab"); + await tab.callbacks.get("session_refresh")(); + assert.equal(tab.emitted.at(-1)[0], "session_refresh"); + await tab.callbacks.get("session_token")("renewed-credential"); + assert.equal( + tab.requests[0][1].headers["Reflex-Client-Token"], + "duplicated-tab", + ); + assert.deepEqual(JSON.parse(tab.requests[0][1].body), { + session_token: "renewed-credential", + }); + assert.deepEqual(tab.connections, [""]); +}); + +test("failed exchange releases the cold-tab lock and surfaces failure", async () => { + const webLocks = locks(); + const first = await browserSession({ + webLocks, + fetch: async () => ({ ok: false, status: 403 }), + }); + const second = await browserSession({ webLocks }); + await first.connect(); + await second.connect(); + await (await first.accept("first-tab", "credential")).exchange; + await tick(); + assert.deepEqual(second.connections, [""]); + assert.equal(first.errors.length, 1); + assert.equal(first.runtime.getToken(), "first-tab"); +}); + +test("stalled exchange times out without holding up hydration", async () => { + const tab = await browserSession({ + fetch: (_url, { signal }) => + new Promise((_resolve, reject) => { + signal.addEventListener("abort", () => reject(new Error("timed out"))); + }), + }); + await tab.connect(); + const { exchange } = await tab.accept("first-tab", "credential"); + assert.equal(tab.socket.current.connected, true); + tab.timeouts.values().next().value(); + await exchange; + assert.equal(tab.errors[0].message, "timed out"); +}); + +test("disconnecting before acceptance releases the cold-tab lock", async () => { + const webLocks = locks(); + const first = await browserSession({ webLocks }); + const second = await browserSession({ webLocks }); + await first.connect(); + await second.connect(); + first.socket.current.disconnect(); + await tick(); + assert.deepEqual(second.connections, [""]); +}); + +test("backend apps have independent token and freshness storage", async () => { + const first = await browserSession(); + await first.connect(); + await (await first.accept("first-tab", "credential")).exchange; + const second = await browserSession({ + localStorage: first.localStorage, + sessionStorage: first.sessionStorage, + eventURL: "ws://localhost:9000/other/_event", + }); + assert.equal(second.runtime.getToken(), ""); + await second.connect(); + assert.deepEqual(second.connections, [""]); +}); + +test("uploads wait for the cookie while following WebSocket events drain", async () => { + let finish; + const tab = await browserSession({ + fetch: () => + new Promise((resolve) => { + finish = resolve; + }), + }); + await tab.connect(); + const { exchange } = await tab.accept("first-tab", "credential"); + await tab.runtime.queueEvents( + [ + { + name: "reflex___state.test.upload", + handler: "uploadFiles", + payload: { files: [] }, + }, + { name: "reflex___state.test.ordinary" }, + ], + tab.socket, + false, + () => {}, + { current: {} }, + ); + assert.equal(tab.uploads.length, 0); + assert.equal(tab.emitted.length, 1); + finish({ + ok: true, + json: async () => ({ client_token: "first-tab", expires_in: 3600 }), + }); + await exchange; + await tick(); + assert.equal(tab.uploads.length, 1); +}); + +test("failed initial exchange holds HTTP readiness until recovery or disconnect", async () => { + const tab = await browserSession({ + fetch: async () => ({ ok: false, status: 403 }), + }); + await tab.connect(); + await (await tab.accept("first-tab", "credential")).exchange; + let ready = false; + const pending = tab.runtime.waitForSession().then(() => { + ready = true; + }); + pending.catch(() => {}); + await tick(); + assert.equal(ready, false); + await tab.runtime.applyRestEvent( + { name: "upload", handler: "uploadFiles", payload: { files: [] } }, + tab.socket.current, + ); + await tick(); + assert.equal(tab.uploads.length, 0); + tab.socket.current.disconnect(); + await assert.rejects(pending, /Session connection closed/); + await tick(); + assert.equal(tab.uploads.length, 0); +}); + +test("idle sessions schedule renewal without an application event", async () => { + const tab = await browserSession({ + fetch: async () => ({ + ok: true, + json: async () => ({ + client_token: "bound-tab", + expires_in: 100, + refresh_in: 20, + }), + }), + }); + await tab.connect(); + await (await tab.accept("bound-tab", "credential")).exchange; + const timer = [...tab.timeouts.keys()].find( + (id) => tab.delays.get(id) === 20000, + ); + assert.ok(timer, "renewal uses the server's configured refresh interval"); + tab.timeouts.get(timer)(); + await tick(); + assert.equal(tab.emitted.at(-1)[0], "session_refresh"); + await tab.callbacks.get("session_token")("renewed-credential"); + assert.equal( + JSON.parse(tab.requests.at(-1)[1].body).session_token, + "renewed-credential", + ); + tab.socket.current.disconnect(); + assert.equal(tab.timeouts.size, 0, "disconnect clears renewal timers"); +}); + +test("transient exchange failure retries and recovers HTTP readiness on the live socket", async () => { + let fail = true; + const tab = await browserSession({ + fetch: async () => + fail + ? { ok: false, status: 503 } + : { + ok: true, + json: async () => ({ client_token: "bound-tab", expires_in: 100 }), + }, + }); + await tab.connect(); + await (await tab.accept("bound-tab", "credential")).exchange; + let ready = false; + const readiness = tab.runtime.waitForSession().then(() => { + ready = true; + }); + // Observe rejection so a failing regression does not leak an unhandled promise. + readiness.catch(() => {}); + await tick(); + assert.equal(ready, false); + fail = false; + for (const callback of [...tab.timeouts.values()]) callback(); + await tick(); + assert.equal(tab.emitted.at(-1)?.[0], "session_refresh"); + await tab.callbacks.get("session_token")("fresh-credential"); + await readiness; + assert.equal(ready, true); + await tab.runtime.waitForSession(); + assert.deepEqual(tab.connections, [""]); +}); + +test("an exchange completing after disconnect cannot replace a newer connection's token", async () => { + let finish; + const tab = await browserSession({ + fetch: () => + new Promise((resolve) => { + finish = resolve; + }), + }); + await tab.connect(); + const { exchange } = await tab.accept("first-tab", "credential"); + const ready = tab.runtime.waitForSession(); + ready.catch(() => {}); + tab.socket.current.disconnect(); + await assert.rejects(ready, /Session connection closed/); + await tab.callbacks.get("new_token")("next-tab"); + finish({ + ok: true, + json: async () => ({ client_token: "stale-tab", expires_in: 100 }), + }); + await exchange; + assert.equal(tab.runtime.getToken(), "next-tab"); + assert.equal(tab.timeouts.size, 0); +}); diff --git a/tests/units/reflex_base/client_state_updates.mjs b/tests/units/reflex_base/client_state_updates.mjs index 9302dae924a..a7efbcdff05 100644 --- a/tests/units/reflex_base/client_state_updates.mjs +++ b/tests/units/reflex_base/client_state_updates.mjs @@ -45,7 +45,7 @@ async function createConnection( }; const body = element("body"); const socket = { - connected: true, + connected: false, io: { opts: {}, encoder: {}, decoder: {} }, on: (name, handler) => handlers.set(name, handler), emit: (name, value) => sent.push({ name, value }), @@ -72,6 +72,7 @@ async function createConnection( reload: () => reloads.push(true), }, sessionStorage, + clearTimeout() {}, addEventListener() {}, removeEventListener() {}, }, diff --git a/tests/units/reflex_base/client_upload.mjs b/tests/units/reflex_base/client_upload.mjs new file mode 100644 index 00000000000..54d0f5e1f9c --- /dev/null +++ b/tests/units/reflex_base/client_upload.mjs @@ -0,0 +1,116 @@ +import assert from "node:assert/strict"; +import fs from "node:fs"; +import path from "node:path"; +import { test } from "node:test"; +import { createContext, SourceTextModule, SyntheticModule } from "node:vm"; + +const source = fs.readFileSync( + path.join(path.dirname(process.argv[2]), "helpers/upload.js"), + "utf8", +); + +/** Run the actual upload helper with a held cookie exchange and fake XHR. */ +async function pendingUpload() { + const requests = []; + const context = createContext({ + AbortController, + FormData, + console, + window: {}, + XMLHttpRequest: class { + upload = {}; + open() {} + setRequestHeader() {} + send() { + requests.push(this); + } + abort() { + this.onabort(); + } + }, + }); + const module = new SourceTextModule(source, { context }); + await module.link((specifier) => { + const exports = + specifier === "$/env.json" + ? { default: { UPLOAD: "/upload" } } + : { parseJson: JSON.parse }; + return new SyntheticModule( + Object.keys(exports), + function () { + for (const [name, value] of Object.entries(exports)) + this.setExport(name, value); + }, + { context }, + ); + }); + await module.evaluate(); + let finish; + const readiness = new Promise((resolve) => { + finish = resolve; + }); + const refs = {}; + const socket = { connected: true }; + const start = () => + module.namespace.uploadFiles( + "handler", + [], + "test", + null, + {}, + {}, + socket, + refs, + (value) => value, + () => "token", + () => readiness, + ); + return { start, finish, refs, requests, socket }; +} + +test("queued uploads are cancellable before the cookie exchange finishes", async () => { + const upload = await pendingUpload(); + const pending = upload.start(); + assert.ok(upload.refs.__upload_controllers_test); + assert.equal(upload.requests.length, 0); + upload.refs.__upload_controllers_test.abort(); + assert.equal(await pending, false); + assert.equal(upload.requests.length, 0); + assert.equal(upload.refs.__upload_controllers_test, undefined); +}); + +test("teardown prevents a queued upload from starting later", async () => { + const upload = await pendingUpload(); + const pending = upload.start(); + assert.equal(upload.requests.length, 0); + upload.socket.connected = false; + upload.finish(); + assert.equal(await pending, false); + assert.equal(upload.requests.length, 0); + assert.equal(upload.refs.__upload_controllers_test, undefined); +}); + +test("duplicate queued uploads reserve their controller before readiness", async () => { + const upload = await pendingUpload(); + const pending = upload.start(); + assert.equal(await upload.start(), false); + assert.equal(upload.requests.length, 0); + upload.refs.__upload_controllers_test.abort(); + upload.finish(); + await pending; +}); + +test("a ready upload starts once and cleans up its controller", async () => { + const upload = await pendingUpload(); + const pending = upload.start(); + assert.equal(upload.requests.length, 0); + upload.finish(); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(upload.requests.length, 1); + const request = upload.requests[0]; + assert.equal(request.withCredentials, true); + request.status = 200; + request.onload(); + assert.equal((await pending).status, 200); + assert.equal(upload.refs.__upload_controllers_test, undefined); +}); diff --git a/tests/units/reflex_base/test_client_event_queue.py b/tests/units/reflex_base/test_client_event_queue.py index ccbbbc05a82..2c21774d083 100644 --- a/tests/units/reflex_base/test_client_event_queue.py +++ b/tests/units/reflex_base/test_client_event_queue.py @@ -12,6 +12,8 @@ "script", [ "client_event_queue.mjs", + "client_session.mjs", + "client_upload.mjs", "client_state_updates.mjs", "state_string_helpers.mjs", ],