diff --git a/config/assertion-safety-baseline.txt b/config/assertion-safety-baseline.txt index 2715227dfd14..4aa68cc27205 100644 --- a/config/assertion-safety-baseline.txt +++ b/config/assertion-safety-baseline.txt @@ -1461,7 +1461,7 @@ packages/ai/src/utils/credential-redaction.ts 1 packages/ai/src/utils/oauth/openai-chatgpt-jwt.ts 1 packages/ai/src/utils/stream-first-event-timeout.ts 2 packages/ai/src/utils/tls-certificate-errors.ts 1 -packages/gateway-client/src/client.ts 5 +packages/gateway-client/src/client.ts 4 packages/gateway-client/src/gateway-origin-scope.ts 1 packages/gateway-client/src/pending-request.ts 2 packages/gateway-client/src/protocol-client.ts 1 diff --git a/packages/gateway-client/README.md b/packages/gateway-client/README.md index dda77f90ec60..c60df9145ffe 100644 --- a/packages/gateway-client/README.md +++ b/packages/gateway-client/README.md @@ -46,6 +46,37 @@ For device-authenticated Node connections, supply `deviceIdentity` (or storage through `GatewayClientHostDeps`; the package does not load OpenClaw's local identity or credentials automatically. +Token storage callbacks may return their existing synchronous result or a +`Promise`. The client waits for token loading before sending `connect`, and for +issued-token persistence before calling `onHelloOk`. An accepted hello creates a +persistence obligation that survives disconnect or stop; bootstrap credentials +retire after that persistence succeeds. Readiness still belongs to the current +connection. `stop()` requests shutdown synchronously; `stopAndWait()` also waits +for accepted token operations to settle, even when transport closure times out. +A reconnect waits for earlier token operations before reading the token again. + +Accepted asynchronous persistence failures reach `onConnectError` even after the +connection retires. If that callback is absent or throws, `stopAndWait()` rejects +with the first undelivered persistence error after draining accepted work. A later +connection can still load credentials; a reported failure does not poison its +storage queue. Synchronous callback exceptions retain their existing connect-error +behavior. + +Hosts should honor the optional `expectedToken` storage condition. A string +compares the existing row before writing or clearing; `null` on a store means +insert only when no row exists. Omission preserves unconditional storage behavior. +This prevents an older receipt or cleanup from replacing another client's newer +token. Close cleanup waits for pending persistence and, if its result is uncertain, +conditionally clears only the sampled and received tokens. Cleanup before any token +observation retains its existing unconditional behavior. An observed empty cache +does not permit unconditional cleanup. + +When storage callbacks receive `signal` or `assertCurrent`, check them immediately +before admission and before committing a write. These callbacks stay local to the +host; do not send them to a worker. Loads use the current connection lifetime; +cleanup uses the client lifetime. Accepted hello persistence is independent of +transport lifetime and retains the host's normal storage admission checks. + ## Entry points - `@openclaw/gateway-client` exports the Node `GatewayClient`, device-auth diff --git a/packages/gateway-client/src/client-address-utils.ts b/packages/gateway-client/src/client-address-utils.ts index 80e3c9c50ff4..20f14c5f0e8b 100644 --- a/packages/gateway-client/src/client-address-utils.ts +++ b/packages/gateway-client/src/client-address-utils.ts @@ -8,10 +8,25 @@ export function normalizeGatewayErrorText(value: unknown): string { return typeof value === "string" ? value.trim().toLowerCase() : ""; } -export function isSensitiveUrlQueryParamName(key: string): boolean { +function isSensitiveUrlQueryParamName(key: string): boolean { return /(?:token|password|secret|key|auth|credential)/iu.test(key); } +export function isGatewayClientStoppedError(err: unknown): boolean { + const message = err instanceof Error ? err.message : String(err); + return message === "gateway client stopped" || message === "Error: gateway client stopped"; +} + +export function formatGatewayClientErrorForLog(err: unknown): string { + const redactedUrlLikeString = String(err) + .replace(/\/\/([^@/?#\s]+)@/g, "//***:***@") + .replace(/(Authorization:\s*Bearer\s+)[^\s]+/giu, "$1***") + .replace(/([?&])([^=&\s]+)=([^&#\s"'<>)]*)/g, (match, prefix: string, key: string) => + isSensitiveUrlQueryParamName(key) ? `${prefix}${key}=***` : match, + ); + return redactedUrlLikeString; +} + const SHA256_HEX_FINGERPRINT = /^[a-fA-F0-9]{64}$/u; const SHA256_COLON_FINGERPRINT = /^(?:[a-fA-F0-9]{2}:){31}[a-fA-F0-9]{2}$/u; diff --git a/packages/gateway-client/src/client-device-auth.ts b/packages/gateway-client/src/client-device-auth.ts new file mode 100644 index 000000000000..c31ec21779bf --- /dev/null +++ b/packages/gateway-client/src/client-device-auth.ts @@ -0,0 +1,154 @@ +export type DeviceAuthTokenRecord = { + token?: string; + scopes?: string[]; +}; + +export type DeviceAuthTokenObservation = { + // Null records a completed load without a token; undefined means no load completed. + token: string | null | undefined; + receiptToken?: string; + persistence?: Promise; +}; + +export type MaybePromise = T | Promise; + +type DeviceAuthOperation = { + signal?: AbortSignal; + assertCurrent?: () => void; +}; + +export type GatewayClientDeviceAuthStorage = { + loadDeviceAuthToken?: ( + params: DeviceAuthOperation & { + deviceId: string; + role: string; + env?: NodeJS.ProcessEnv; + }, + ) => MaybePromise; + storeDeviceAuthToken?: ( + params: DeviceAuthOperation & { + deviceId: string; + role: string; + token: string; + scopes: string[]; + env?: NodeJS.ProcessEnv; + expectedToken?: string | null; + }, + ) => unknown; + clearDeviceAuthToken?: ( + params: DeviceAuthOperation & { + deviceId: string; + role: string; + env?: NodeJS.ProcessEnv; + expectedToken?: string; + }, + ) => unknown; +}; + +/** Owns pending host storage work for one client across connection generations. */ +export class GatewayClientDeviceAuth { + private readonly operations = new Set>(); + private unreportedPersistenceFailure: Error | undefined; + + constructor(private readonly storage: Required) {} + + load(params: Parameters>[0]) { + const load = () => { + params.signal?.throwIfAborted(); + params.assertCurrent?.(); + return this.track(this.storage.loadDeviceAuthToken(params)); + }; + // A retired connection's accepted store/clear must settle before the next read. + return this.operations.size > 0 ? this.settle().then(load) : load(); + } + + store( + params: Parameters>[0], + onStored: () => void, + onFailed: (error: Error) => boolean, + ) { + const failed = (error: unknown): never => { + const failure = error instanceof Error ? error : new Error(String(error)); + let reported = false; + try { + reported = onFailed(failure); + } catch { + // A broken reporter must not replace or discard the original storage error. + } + if (!reported) { + this.unreportedPersistenceFailure ??= failure; + } + throw failure; + }; + params.signal?.throwIfAborted(); + params.assertCurrent?.(); + const stored = this.storage.storeDeviceAuthToken(params); + return stored instanceof Promise ? this.track(stored.then(onStored).catch(failed)) : onStored(); + } + + clear( + params: Parameters>[0], + observation: DeviceAuthTokenObservation, + canClear: () => boolean, + ): MaybePromise { + const clear = () => { + if (!canClear()) { + return undefined; + } + // A rejected result may follow a committed write. Reconcile only the sampled + // and received token identities; never clear an unrelated replacement. + const knownTokens = [observation.token, observation.receiptToken].filter( + (token): token is string => typeof token === "string", + ); + const expectedTokens: Array = [...new Set(knownTokens)]; + if (observation.token === undefined && expectedTokens.length === 0) { + expectedTokens.push(undefined); + } + const clearToken = (expectedToken: string | undefined) => { + params.signal?.throwIfAborted(); + params.assertCurrent?.(); + const cleared = this.storage.clearDeviceAuthToken({ + ...params, + ...(expectedToken === undefined ? {} : { expectedToken }), + }); + return cleared instanceof Promise ? cleared.then(() => {}) : undefined; + }; + let pending: Promise | undefined; + for (const expectedToken of expectedTokens) { + pending = pending + ? pending.then(() => clearToken(expectedToken)) + : clearToken(expectedToken); + } + return pending; + }; + // Register the entire cleanup now so reconnect and shutdown also join deferred clearing. + return this.track( + observation.persistence ? observation.persistence.then(clear, clear) : clear(), + ); + } + + async settle(): Promise { + await Promise.allSettled(this.operations); + } + + async drain(): Promise { + await this.settle(); + const failure = this.unreportedPersistenceFailure; + this.unreportedPersistenceFailure = undefined; + if (failure) { + throw failure; + } + } + + private track(operation: MaybePromise): MaybePromise { + if (!(operation instanceof Promise)) { + return operation; + } + this.operations.add(operation); + void operation.then( + () => this.operations.delete(operation), + () => this.operations.delete(operation), + ); + return operation; + } +} diff --git a/packages/gateway-client/src/client-upgrade-error.ts b/packages/gateway-client/src/client-upgrade-error.ts new file mode 100644 index 000000000000..ea25cd3a67bc --- /dev/null +++ b/packages/gateway-client/src/client-upgrade-error.ts @@ -0,0 +1,46 @@ +import type { IncomingMessage } from "node:http"; + +const MAX_UPGRADE_ERROR_BODY_BYTES = 2 * 1024; +const UPGRADE_ERROR_BODY_TIMEOUT_MS = 1_000; + +export async function readUpgradeErrorBody(response: IncomingMessage): Promise { + return await new Promise((resolve) => { + const chunks: Buffer[] = []; + let totalBytes = 0; + let settled = false; + const finish = () => { + if (settled) { + return; + } + settled = true; + clearTimeout(timer); + response.off("data", onData); + response.off("end", finish); + response.off("error", finish); + response.off("aborted", finish); + resolve(Buffer.concat(chunks, totalBytes).toString("utf8").replace(/\s+/gu, " ").trim()); + }; + const stop = () => { + finish(); + response.destroy(); + }; + const onData = (chunk: Buffer | string) => { + const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); + const remaining = MAX_UPGRADE_ERROR_BODY_BYTES - totalBytes; + if (remaining > 0) { + const prefix = buffer.subarray(0, remaining); + chunks.push(prefix); + totalBytes += prefix.byteLength; + } + if (buffer.byteLength >= remaining) { + stop(); + } + }; + const timer = setTimeout(stop, UPGRADE_ERROR_BODY_TIMEOUT_MS); + timer.unref?.(); + response.on("data", onData); + response.once("end", finish); + response.once("error", finish); + response.once("aborted", finish); + }); +} diff --git a/packages/gateway-client/src/client.device-auth.test.ts b/packages/gateway-client/src/client.device-auth.test.ts new file mode 100644 index 000000000000..3da0d0f28ece --- /dev/null +++ b/packages/gateway-client/src/client.device-auth.test.ts @@ -0,0 +1,501 @@ +import assert from "node:assert/strict"; +import { EventEmitter } from "node:events"; +import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { createDeferred } from "../../../test/helpers/promise.js"; +import type { + DeviceAuthTokenRecord, + GatewayClient as GatewayClientInstance, + GatewayClientHostDeps, + GatewayClientOptions, +} from "./client.js"; + +class MockWebSocket extends EventEmitter { + static readonly OPEN = 1; + static instances: MockWebSocket[] = []; + readyState = 0; + readonly send = vi.fn<(data: string) => void>(); + + constructor() { + super(); + MockWebSocket.instances.push(this); + } + + open() { + this.readyState = MockWebSocket.OPEN; + this.emit("open"); + this.receive({ + type: "event", + event: "connect.challenge", + payload: { nonce: "synthetic-nonce", ts: 1_800_000_000_000 }, + }); + } + + receive(frame: unknown) { + this.emit("message", JSON.stringify(frame)); + } + + respond(payload: unknown, error?: unknown) { + const sent = this.send.mock.calls[0]; + assert(sent); + const { id } = JSON.parse(sent[0]) as { id: string }; + this.receive({ type: "res", id, ok: !error, payload, error }); + } + + close(code = 1000, reason = "") { + this.readyState = 3; + this.emit("close", code, Buffer.from(reason)); + } + + terminate() { + this.close(); + } +} + +vi.mock("./websocket.js", () => ({ WebSocket: MockWebSocket })); +let GatewayClient: typeof GatewayClientInstance; +const clients: GatewayClientInstance[] = []; + +beforeAll(async () => { + ({ GatewayClient } = await import("./client.js")); +}); +beforeEach(() => { + vi.useFakeTimers(); + MockWebSocket.instances = []; +}); +afterEach(async () => { + for (const client of clients.splice(0)) { + await client.stopAndWait(); + } + vi.useRealTimers(); +}); + +function connect( + hostDeps: GatewayClientHostDeps, + options: Pick< + GatewayClientOptions, + "preferBootstrapToken" | "bootstrapToken" | "token" | "password" + > = {}, + reportError?: ((error: Error) => void) | false, +) { + const onHelloOk = vi.fn(); + const onConnectError = vi.fn((error: Error) => { + if (reportError) { + reportError(error); + } + }); + const onClose = vi.fn(); + const onReconnectPaused = vi.fn(); + const client = new GatewayClient({ + ...options, + url: "ws://127.0.0.1:18789", + deviceIdentity: { + deviceId: "synthetic-device", + privateKeyPem: "synthetic-private-key", + publicKeyPem: "synthetic-public-key", + }, + hostDeps: { + signDevicePayload: () => "synthetic-signature", + publicKeyRawBase64UrlFromPem: () => "synthetic-public-key", + ...hostDeps, + }, + onHelloOk, + onConnectError: reportError === false ? undefined : onConnectError, + onClose, + onReconnectPaused, + }); + clients.push(client); + client.start(); + const socket = MockWebSocket.instances[0]; + assert(socket); + socket.open(); + return { client, socket, onHelloOk, onConnectError, onClose, onReconnectPaused }; +} + +const storedToken = { token: "synthetic-stored-token", scopes: ["operator.read"] }; +const hello = { + type: "hello-ok", + protocol: 4, + auth: { deviceToken: "synthetic-issued-token", role: "operator", scopes: ["operator.read"] }, +}; + +describe("GatewayClient host token storage", () => { + it("awaits a stored token before sending its connect request", async () => { + const loaded = createDeferred(); + const { socket } = connect({ loadDeviceAuthToken: () => loaded.promise }); + expect(socket.send).not.toHaveBeenCalled(); + loaded.resolve(storedToken); + await vi.advanceTimersByTimeAsync(0); + const sent = socket.send.mock.calls[0]; + assert(sent); + expect(JSON.parse(sent[0])).toMatchObject({ + method: "connect", + params: { auth: { deviceToken: storedToken.token }, scopes: storedToken.scopes }, + }); + }); + + it("keeps synchronous storage callbacks and their ignored return values compatible", async () => { + const { socket, onHelloOk } = connect({ + loadDeviceAuthToken: () => storedToken, + storeDeviceAuthToken: () => storedToken, + clearDeviceAuthToken: () => true, + }); + expect(socket.send).toHaveBeenCalledOnce(); + socket.respond(hello); + await vi.advanceTimersByTimeAsync(0); + expect(onHelloOk).toHaveBeenCalledOnce(); + }); + + it.each(["ready", "closing", "stop"] as const)( + "settles token persistence before completing %s", + async (completion) => { + const stored = createDeferred(); + let persistedToken: string | undefined; + const storeDeviceAuthToken = vi.fn< + NonNullable + >(async (params) => { + await stored.promise; + params.signal?.throwIfAborted(); + params.assertCurrent?.(); + persistedToken = params.token; + }); + const { client, socket, onHelloOk, onConnectError } = connect({ + loadDeviceAuthToken: () => storedToken, + storeDeviceAuthToken, + }); + socket.respond(hello); + await vi.advanceTimersByTimeAsync(0); + expect(onHelloOk).not.toHaveBeenCalled(); + expect(storeDeviceAuthToken).toHaveBeenCalledOnce(); + const operation = storeDeviceAuthToken.mock.calls[0]?.[0]; + assert(operation); + expect(operation).toMatchObject({ token: hello.auth.deviceToken, scopes: hello.auth.scopes }); + expect(() => operation.assertCurrent?.()).not.toThrow(); + if (completion === "closing") { + socket.readyState = 2; + } + const stopped = vi.fn(); + const stop = completion === "stop" ? client.stopAndWait().then(stopped) : undefined; + if (completion === "stop") { + expect(operation.signal).toBeUndefined(); + expect(operation.assertCurrent).toBeUndefined(); + await vi.advanceTimersByTimeAsync(0); + expect(stopped).not.toHaveBeenCalled(); + } + stored.resolve(); + await vi.advanceTimersByTimeAsync(0); + await stop; + expect(persistedToken).toBe(hello.auth.deviceToken); + expect(onConnectError).not.toHaveBeenCalled(); + expect(onHelloOk).toHaveBeenCalledTimes(completion === "ready" ? 1 : 0); + }, + ); + + it("reconnects with the durable receipt after bootstrap hello's transport retires", async () => { + const permitStore = createDeferred(); + let cached: DeviceAuthTokenRecord | null = null; + const { socket, onHelloOk } = connect( + { + loadDeviceAuthToken: () => cached, + storeDeviceAuthToken: async (params) => { + await permitStore.promise; + params.signal?.throwIfAborted(); + params.assertCurrent?.(); + cached = { token: params.token, scopes: params.scopes }; + }, + }, + { bootstrapToken: "synthetic-bootstrap", preferBootstrapToken: true }, + ); + try { + socket.respond(hello); + await vi.advanceTimersByTimeAsync(0); + socket.close(1006, "transport retired"); + await vi.advanceTimersByTimeAsync(1_000); + const replacement = MockWebSocket.instances[1]; + assert(replacement); + replacement.open(); + expect(replacement.send).not.toHaveBeenCalled(); + permitStore.resolve(); + await vi.advanceTimersByTimeAsync(0); + const sent = replacement.send.mock.calls[0]; + assert(sent); + expect(JSON.parse(sent[0])).toMatchObject({ + method: "connect", + params: { auth: { deviceToken: hello.auth.deviceToken } }, + }); + expect(JSON.parse(sent[0]).params.auth.bootstrapToken).toBeUndefined(); + expect(onHelloOk).not.toHaveBeenCalled(); + } finally { + permitStore.resolve(); + } + }); + + it.each([false, true])( + "keeps a newer token after observing an empty cache (async load: %s)", + async (asyncLoad) => { + let cached: DeviceAuthTokenRecord | null = null; + const newer = { token: "synthetic-newer-token", scopes: ["operator.read"] }; + const clearDeviceAuthToken = vi.fn< + NonNullable + >((params) => { + if (params.expectedToken === undefined || params.expectedToken === cached?.token) { + cached = null; + } + }); + const { client, socket } = connect({ + loadDeviceAuthToken: () => (asyncLoad ? Promise.resolve(cached) : cached), + clearDeviceAuthToken, + }); + await vi.advanceTimersByTimeAsync(0); + expect(socket.send).toHaveBeenCalledOnce(); + cached = newer; + socket.close(1008, "device token mismatch"); + await client.stopAndWait(); + expect(clearDeviceAuthToken).not.toHaveBeenCalled(); + expect(cached).toEqual(newer); + }, + ); + + it.each([ + { initial: storedToken, result: "committed", replaceWithNewer: false, bootstrap: true }, + { initial: storedToken, result: "committed", replaceWithNewer: true }, + { initial: null, result: "committed", replaceWithNewer: false }, + { initial: null, result: "committed", replaceWithNewer: true }, + { initial: storedToken, result: "uncertain", replaceWithNewer: false }, + { initial: null, result: "uncertain", replaceWithNewer: false }, + { initial: null, result: "failed", replaceWithNewer: true }, + ] as const)("reconciles mismatch cleanup against the accepted receipt: %j", async (entry) => { + const { initial, result, replaceWithNewer } = entry; + const permitStore = createDeferred(); + let cached: DeviceAuthTokenRecord | null = initial; + const newer = { token: "synthetic-newer-token", scopes: ["operator.read"] }; + const clearDeviceAuthToken = vi.fn>( + (params) => { + params.assertCurrent?.(); + if (params.expectedToken === undefined || params.expectedToken === cached?.token) { + cached = null; + } + }, + ); + const { socket, onHelloOk } = connect( + { + loadDeviceAuthToken: () => cached, + storeDeviceAuthToken: async (params) => { + await permitStore.promise; + params.signal?.throwIfAborted(); + params.assertCurrent?.(); + if (result === "failed") { + throw new Error("synthetic persistence rejected"); + } + if ( + params.expectedToken === undefined || + (params.expectedToken === null + ? cached === null + : params.expectedToken === cached?.token) + ) { + cached = { token: params.token, scopes: params.scopes }; + } + if (result === "uncertain") { + throw new Error("synthetic result unavailable"); + } + }, + clearDeviceAuthToken, + }, + "bootstrap" in entry + ? { + preferBootstrapToken: true, + bootstrapToken: "synthetic-bootstrap", + token: "synthetic-shared-token", + password: "synthetic-shared-password", + } + : {}, + ); + try { + socket.respond(hello); + await vi.advanceTimersByTimeAsync(0); + socket.close(1008, "device token mismatch"); + expect(clearDeviceAuthToken).not.toHaveBeenCalled(); + if (replaceWithNewer) { + cached = newer; + } + permitStore.resolve(); + await vi.advanceTimersByTimeAsync(0); + expect(clearDeviceAuthToken).toHaveBeenCalled(); + expect( + clearDeviceAuthToken.mock.calls.every(([params]) => params.expectedToken !== undefined), + ).toBe(true); + expect(cached).toEqual(replaceWithNewer ? newer : null); + expect(onHelloOk).not.toHaveBeenCalled(); + } finally { + permitStore.resolve(); + } + }); + + it.each(["active", "disconnected", "stopped"] as const)( + "reports an accepted persistence rejection exactly once when %s", + async (lifetime) => { + const permitStore = createDeferred(); + const failure = new Error("synthetic async persistence failure"); + const { client, socket, onConnectError } = connect({ + loadDeviceAuthToken: () => storedToken, + storeDeviceAuthToken: async () => { + await permitStore.promise; + throw failure; + }, + }); + try { + socket.respond(hello); + await vi.advanceTimersByTimeAsync(0); + if (lifetime === "disconnected") { + socket.close(1006, "transport retired"); + } + const stopped = lifetime === "stopped" ? client.stopAndWait() : undefined; + permitStore.resolve(); + await vi.advanceTimersByTimeAsync(0); + await stopped; + expect(onConnectError).toHaveBeenCalledExactlyOnceWith(failure); + } finally { + permitStore.resolve(); + } + }, + ); + + it.each(["missing", "throwing"] as const)( + "drains the original undelivered async persistence error with a %s reporter", + async (reporter) => { + const permitStore = createDeferred(); + const failure = new Error("synthetic undelivered persistence failure"); + const { client, socket } = connect( + { + loadDeviceAuthToken: () => storedToken, + storeDeviceAuthToken: async () => { + await permitStore.promise; + throw failure; + }, + }, + {}, + reporter === "missing" + ? false + : () => { + throw new Error("synthetic reporter failure"); + }, + ); + try { + socket.respond(hello); + await vi.advanceTimersByTimeAsync(0); + if (reporter === "throwing") { + const stopped = expect(client.stopAndWait()).rejects.toBe(failure); + permitStore.resolve(); + await vi.advanceTimersByTimeAsync(0); + await stopped; + } else { + socket.close(1006, "transport retired"); + permitStore.resolve(); + await vi.advanceTimersByTimeAsync(1_000); + const replacement = MockWebSocket.instances[1]; + assert(replacement); + replacement.open(); + expect(replacement.send).toHaveBeenCalledOnce(); + await expect(client.stopAndWait()).rejects.toBe(failure); + } + } finally { + permitStore.resolve(); + } + }, + ); + + it("keeps synchronous store exceptions on the existing connect-error path", async () => { + const { client, socket, onConnectError } = connect({ + loadDeviceAuthToken: () => storedToken, + storeDeviceAuthToken: () => { + throw new Error("synthetic synchronous persistence failure"); + }, + }); + socket.respond(hello); + await vi.advanceTimersByTimeAsync(0); + expect(onConnectError).toHaveBeenCalledOnce(); + await expect(client.stopAndWait()).resolves.toBeUndefined(); + }); + + it("retires pending token loading when stopped", async () => { + const loaded = createDeferred(); + const { client, socket, onHelloOk } = connect({ loadDeviceAuthToken: () => loaded.promise }); + const stopped = client.stopAndWait(); + loaded.resolve(storedToken); + await stopped; + await vi.advanceTimersByTimeAsync(0); + expect(socket.send).not.toHaveBeenCalled(); + expect(onHelloOk).not.toHaveBeenCalled(); + }); + + it.each([false, true])( + "preserves rejection and cleanup when the peer closes (already closing: %s)", + async (alreadyClosing) => { + const cleared = createDeferred(); + const clearDeviceAuthToken = vi.fn< + NonNullable + >(() => cleared.promise); + const { socket, onConnectError, onClose, onReconnectPaused } = connect({ + loadDeviceAuthToken: () => storedToken, + clearDeviceAuthToken, + }); + try { + socket.respond(undefined, { + code: "INVALID_REQUEST", + message: "synthetic token rejected", + details: { code: "AUTH_DEVICE_TOKEN_MISMATCH" }, + }); + if (alreadyClosing) { + socket.readyState = 2; + } + await vi.advanceTimersByTimeAsync(0); + expect(clearDeviceAuthToken).toHaveBeenCalledWith( + expect.objectContaining({ expectedToken: storedToken.token }), + ); + expect(onConnectError).toHaveBeenCalledOnce(); + const error = onConnectError.mock.calls[0]?.[0]; + expect(error).toMatchObject({ message: "synthetic token rejected" }); + socket.close(1008, "connect failed"); + const cleanup = clearDeviceAuthToken.mock.calls[0]?.[0]; + assert(cleanup); + expect(() => cleanup.assertCurrent?.()).not.toThrow(); + expect(onClose).toHaveBeenCalledWith( + 1008, + "connect failed", + expect.objectContaining({ connectError: error }), + ); + expect(onReconnectPaused).toHaveBeenCalledWith( + expect.objectContaining({ detailCode: "AUTH_DEVICE_TOKEN_MISMATCH" }), + ); + cleared.resolve(); + await vi.advanceTimersByTimeAsync(1_000); + expect(onConnectError).toHaveBeenCalledOnce(); + expect(MockWebSocket.instances).toHaveLength(1); + } finally { + cleared.resolve(); + } + }, + ); + + it("finishes close cleanup before loading credentials for the replacement connection", async () => { + const cleared = createDeferred(); + const loadDeviceAuthToken = vi.fn(() => storedToken); + const clearDeviceAuthToken = vi.fn>( + () => cleared.promise, + ); + const { socket } = connect({ loadDeviceAuthToken, clearDeviceAuthToken }); + socket.close(1008, "device token mismatch"); + const cleanup = clearDeviceAuthToken.mock.calls[0]?.[0]; + assert(cleanup); + expect(cleanup.expectedToken).toBe(storedToken.token); + expect(() => cleanup.assertCurrent?.()).not.toThrow(); + await vi.advanceTimersByTimeAsync(1_000); + const replacement = MockWebSocket.instances[1]; + assert(replacement); + replacement.open(); + expect(loadDeviceAuthToken).toHaveBeenCalledOnce(); + cleared.resolve(); + await vi.advanceTimersByTimeAsync(0); + expect(loadDeviceAuthToken).toHaveBeenCalledTimes(2); + expect(replacement.send).toHaveBeenCalledOnce(); + }); +}); diff --git a/packages/gateway-client/src/client.ts b/packages/gateway-client/src/client.ts index a2526331b431..7e5e298ab7f7 100644 --- a/packages/gateway-client/src/client.ts +++ b/packages/gateway-client/src/client.ts @@ -21,10 +21,19 @@ import { import { redactSensitiveUrlLikeString } from "@openclaw/net-policy/redact-sensitive-url"; import { isRecord } from "@openclaw/normalization-core/record-coerce"; import { - isSensitiveUrlQueryParamName, + formatGatewayClientErrorForLog, + isGatewayClientStoppedError, normalizeTlsFingerprint, normalizeGatewayErrorText, } from "./client-address-utils.js"; +import { + GatewayClientDeviceAuth, + type DeviceAuthTokenRecord, + type DeviceAuthTokenObservation, + type GatewayClientDeviceAuthStorage, + type MaybePromise, +} from "./client-device-auth.js"; +import { readUpgradeErrorBody } from "./client-upgrade-error.js"; import { buildGatewayConnectAuth, type GatewayConnectAuthSelection, @@ -34,6 +43,7 @@ import { } from "./connect-auth.js"; import { buildDeviceAuthPayloadV3 } from "./device-auth.js"; import { resolveModelCatalogConnect } from "./model-catalog-connect.js"; +import type { GatewayProtocolConnectAuthority } from "./protocol-client-contract.js"; import { GatewayProtocolClient, type GatewayProtocolCloseContext, @@ -46,7 +56,7 @@ import { GatewayProtocolRequestTimeoutError, } from "./protocol-request.js"; import { shouldPauseGatewayReconnect } from "./reconnect-policy.js"; -import { GatewayClientRequestError } from "./request-error.js"; +import { GatewayClientRequestError, markGatewayConnectAssemblyError } from "./request-error.js"; import { DEFAULT_GATEWAY_REQUEST_TIMEOUT_MS, resolveConnectChallengeTimeoutMs, @@ -67,34 +77,14 @@ export type DeviceIdentity = { publicKeyPem: string; }; -export type DeviceAuthTokenRecord = { - token?: string; - scopes?: string[]; -}; +export type { DeviceAuthTokenRecord } from "./client-device-auth.js"; // The package stays reusable by depending on host callbacks for OpenClaw-owned // state: device keys, token storage, proxy routing, logging, and TLS formatting. -export type GatewayClientHostDeps = { +export type GatewayClientHostDeps = GatewayClientDeviceAuthStorage & { loadOrCreateDeviceIdentity?: () => DeviceIdentity | undefined; signDevicePayload?: (privateKeyPem: string, payload: string) => string; publicKeyRawBase64UrlFromPem?: (publicKeyPem: string) => string; - loadDeviceAuthToken?: (params: { - deviceId: string; - role: string; - env?: NodeJS.ProcessEnv; - }) => DeviceAuthTokenRecord | null; - storeDeviceAuthToken?: (params: { - deviceId: string; - role: string; - token: string; - scopes: string[]; - env?: NodeJS.ProcessEnv; - }) => void; - clearDeviceAuthToken?: (params: { - deviceId: string; - role: string; - env?: NodeJS.ProcessEnv; - }) => void; beforeConnect?: () => void; registerGatewayLoopbackBypass?: (url: string) => (() => void) | undefined; logDebug?: (message: string) => void; @@ -141,55 +131,11 @@ type AssembledConnect = { storedScopes: string[] | undefined; storedToken: string | undefined; usingStoredDeviceToken: boolean | undefined; + persistenceFailed?: boolean; }; const DEFAULT_GATEWAY_CLIENT_URL = "ws://127.0.0.1:18789"; const DEFAULT_CLIENT_VERSION = "0.0.0"; -const MAX_UPGRADE_ERROR_BODY_BYTES = 2 * 1024; -const UPGRADE_ERROR_BODY_TIMEOUT_MS = 1_000; - -async function readUpgradeErrorBody(response: IncomingMessage): Promise { - return await new Promise((resolve) => { - const chunks: Buffer[] = []; - let totalBytes = 0; - let settled = false; - const finish = () => { - if (settled) { - return; - } - settled = true; - clearTimeout(timer); - response.off("data", onData); - response.off("end", finish); - response.off("error", finish); - response.off("aborted", finish); - resolve(Buffer.concat(chunks, totalBytes).toString("utf8").replace(/\s+/gu, " ").trim()); - }; - const stop = () => { - finish(); - response.destroy(); - }; - const onData = (chunk: Buffer | string) => { - const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); - const remaining = MAX_UPGRADE_ERROR_BODY_BYTES - totalBytes; - if (remaining > 0) { - const prefix = buffer.subarray(0, remaining); - chunks.push(prefix); - totalBytes += prefix.byteLength; - } - if (buffer.byteLength >= remaining) { - stop(); - } - }; - const timer = setTimeout(stop, UPGRADE_ERROR_BODY_TIMEOUT_MS); - timer.unref?.(); - response.on("data", onData); - response.once("end", finish); - response.once("error", finish); - response.once("aborted", finish); - }); -} - export type GatewayReconnectPausedInfo = { code: number; reason: string; @@ -205,7 +151,7 @@ export type GatewayClientCloseInfo = { connectError?: Error; }; -export { GatewayClientRequestError } from "./request-error.js"; +export { GatewayClientRequestError, isGatewayConnectAssemblyError } from "./request-error.js"; export { isGatewayProtocolResponseError } from "./protocol-request.js"; export class GatewayClientRequestTimeoutError extends GatewayProtocolRequestTimeoutError { @@ -217,27 +163,6 @@ export class GatewayClientRequestTimeoutError extends GatewayProtocolRequestTime class GatewayClientTransportPolicyError extends GatewayWebSocketTransportConfigurationError {} -const GATEWAY_CONNECT_ASSEMBLY_ERROR = Symbol("gateway.connectAssemblyError"); - -type GatewayConnectAssemblyError = Error & { - [GATEWAY_CONNECT_ASSEMBLY_ERROR]?: true; -}; - -function markGatewayConnectAssemblyError(error: Error): Error { - Object.defineProperty(error, GATEWAY_CONNECT_ASSEMBLY_ERROR, { - configurable: true, - value: true, - }); - return error; -} - -export function isGatewayConnectAssemblyError(value: unknown): value is Error { - return ( - value instanceof Error && - (value as GatewayConnectAssemblyError)[GATEWAY_CONNECT_ASSEMBLY_ERROR] === true - ); -} - export type GatewayClientOptions = { url?: string; // ws://127.0.0.1:18789 origin?: string; @@ -302,21 +227,6 @@ export type GatewayClientConnectionMetadata = { preauthHandshakeTimeoutMs?: number; }; -function isGatewayClientStoppedError(err: unknown): boolean { - const message = err instanceof Error ? err.message : String(err); - return message === "gateway client stopped" || message === "Error: gateway client stopped"; -} - -function formatGatewayClientErrorForLog(err: unknown): string { - const redactedUrlLikeString = String(err) - .replace(/\/\/([^@/?#\s]+)@/g, "//***:***@") - .replace(/(Authorization:\s*Bearer\s+)[^\s]+/giu, "$1***") - .replace(/([?&])([^=&\s]+)=([^&#\s"'<>)]*)/g, (match, prefix: string, key: string) => - isSensitiveUrlQueryParamName(key) ? `${prefix}${key}=***` : match, - ); - return redactedUrlLikeString; -} - const FORCE_STOP_TERMINATE_GRACE_MS = 250; const STOP_AND_WAIT_TIMEOUT_MS = 1_000; const MAX_SUPPRESSED_TRANSIENT_PRE_HELLO_CLEAN_CLOSES = 1; @@ -347,7 +257,6 @@ export class GatewayClient { private stopped = false; private useLegacyNodeProtocolEnvelope = false; private nodeProtocolTransitionPending = false; - private suppressNextHelloCallback = false; private pendingDeviceTokenRetry = false; private deviceTokenRetryBudgetUsed = false; private approvalRuntimeTokenCompatibilityDisabled = false; @@ -358,12 +267,15 @@ export class GatewayClient { private tickTimer: NodeJS.Timeout | null = null; private readonly requestTimeoutMs: number; private pendingStop: PendingStop | null = null; + private readonly deviceAuth: GatewayClientDeviceAuth; + private connectionStoredToken?: DeviceAuthTokenObservation & { generation: number }; private transportValidated = false; private suppressedTransientPreHelloCleanCloses = 0; constructor(opts: GatewayClientOptions) { // Defaults keep the package inert until device identity support is used. this.deps = resolveHostDeps(opts.hostDeps); + this.deviceAuth = new GatewayClientDeviceAuth(this.deps); this.opts = { ...opts, deviceIdentity: @@ -389,19 +301,35 @@ export class GatewayClient { createRequestTimeoutError: (method, timeoutMs, requestSent) => new GatewayClientRequestTimeoutError({ method, timeoutMs, requestSent }), createRequestAbortError: createGatewayRequestAbortError, - buildConnectPlan: ({ nonce, challengeTs, serverCapabilities }) => { + buildConnectPlan: ({ nonce, challengeTs, serverCapabilities, generation, ...authority }) => { if (!nonce) { throw new Error("gateway connect challenge missing nonce"); } if (this.opts.deviceIdentity && challengeTs == null) { throw new Error("gateway connect challenge timestamp invalid"); } - return this.assembleConnectParams({ - role: this.opts.role ?? "operator", - nonce, - signedAtMs: challengeTs ?? Date.now(), - serverCapabilities, - }); + const role = this.opts.role ?? "operator"; + const assemble = (storedAuth: DeviceAuthTokenRecord | null) => { + authority.assertCurrent(); + const assembled = this.assembleConnectParams({ + role, + nonce, + signedAtMs: challengeTs ?? Date.now(), + serverCapabilities, + storedAuth, + }); + this.connectionStoredToken = { generation, token: assembled.storedToken ?? null }; + return assembled; + }; + const storedAuth = this.opts.deviceIdentity + ? this.deviceAuth.load({ + deviceId: this.opts.deviceIdentity.deviceId, + role, + env: this.opts.env, + ...authority, + }) + : null; + return storedAuth instanceof Promise ? storedAuth.then(assemble) : assemble(storedAuth); }, buildConnectParams: (assembled) => assembled.params, onConnectPlanError: (error) => { @@ -415,12 +343,8 @@ export class GatewayClient { } return { closeCode: 1008, closeReason: "connect failed", stop: true, error: marked }; }, - onConnectHello: (hello, context) => this.handleConnectHello(hello, context.plan), + onConnectHello: (hello, context) => this.handleConnectHello(hello, context.plan, context), onHello: (hello) => { - if (this.suppressNextHelloCallback) { - this.suppressNextHelloCallback = false; - return; - } this.opts.onHelloOk?.(hello); }, onConnectFailure: (error, context) => this.handleConnectRequestFailure(error, context.plan), @@ -648,12 +572,8 @@ export class GatewayClient { } async stopAndWait(opts?: { timeoutMs?: number }): Promise { - // Some callers need teardown ordering, not just "close requested". Wait for - // the socket to close or the terminate fallback to fire. + // Teardown includes accepted storage work as well as transport closure. const stopPromise = this.beginStop(); - if (!stopPromise) { - return; - } const timeoutMs = opts?.timeoutMs === undefined ? STOP_AND_WAIT_TIMEOUT_MS @@ -673,6 +593,8 @@ export class GatewayClient { if (timeout) { clearTimeout(timeout); } + // The transport deadline must never abandon accepted durable operations. + await this.deviceAuth.drain(); } } @@ -749,11 +671,12 @@ export class GatewayClient { nonce: string; signedAtMs: number; serverCapabilities: readonly string[]; + storedAuth: DeviceAuthTokenRecord | null; }): AssembledConnect { const { role, nonce, signedAtMs } = params; // Auth selection is intentionally centralized: retry decisions depend on // whether a token was explicit, cached, or compatibility-derived. - const selectedAuth = this.selectConnectAuth(role); + const selectedAuth = this.selectConnectAuth(params.storedAuth); const { authDeviceToken, authApprovalRuntimeToken, @@ -953,7 +876,73 @@ export class GatewayClient { }; } - private handleConnectHello(helloOk: HelloOk, assembled: AssembledConnect): void { + private handleConnectHello( + helloOk: HelloOk, + assembled: AssembledConnect, + authority: GatewayProtocolConnectAuthority, + ): MaybePromise { + authority.assertCurrent(); + const role = this.opts.role ?? "operator"; + const authInfo = helloOk.auth; + const observation = this.connectionStoredToken; + const persisted = () => { + if (authInfo?.deviceToken && (authInfo.role ?? role) === role && observation) { + observation.token = authInfo.deviceToken; + } + if (this.opts.preferBootstrapToken) { + // This accepted receipt redeems setup auth even when its transport has retired. + this.opts.token = undefined; + this.opts.bootstrapToken = undefined; + this.opts.password = undefined; + this.opts.preferBootstrapToken = false; + } + }; + let stored: MaybePromise = undefined; + if (authInfo?.deviceToken && this.opts.deviceIdentity) { + const tokenRole = authInfo.role ?? role; + const scopes = + tokenRole === role && authInfo.deviceToken === assembled.storedToken + ? (assembled.storedScopes ?? authInfo.scopes ?? []) + : (authInfo.scopes ?? []); + if (tokenRole === role && observation) { + observation.receiptToken = authInfo.deviceToken; + } + stored = this.deviceAuth.store( + { + deviceId: this.opts.deviceIdentity.deviceId, + role: tokenRole, + token: authInfo.deviceToken, + scopes, + env: this.opts.env, + ...(tokenRole === role ? { expectedToken: assembled.storedToken ?? null } : {}), + }, + persisted, + (error) => { + assembled.persistenceFailed = true; + const reported = this.notifyConnectError(error); + this.logError( + `gateway device token persistence failed: ${formatGatewayClientErrorForLog(error)}`, + ); + return reported; + }, + ); + if (observation && stored instanceof Promise) { + observation.persistence = stored; + } + } else { + persisted(); + } + const complete = () => { + if (authority.signal.aborted || !this.protocol.connected) { + return; + } + authority.assertCurrent(); + this.completeConnectHello(helloOk); + }; + return stored instanceof Promise ? stored.then(complete) : complete(); + } + + private completeConnectHello(helloOk: HelloOk): void { const reconnectWithCurrentNodeProtocol = this.useLegacyNodeProtocolEnvelope && this.shouldNegotiateLegacyNodeProtocol() && @@ -965,47 +954,41 @@ export class GatewayClient { this.pendingDeviceTokenRetry = false; this.deviceTokenRetryBudgetUsed = false; this.suppressedTransientPreHelloCleanCloses = 0; - const role = this.opts.role ?? "operator"; - const authInfo = helloOk.auth; - if (authInfo?.deviceToken && this.opts.deviceIdentity) { - const tokenRole = authInfo.role ?? role; - const scopes = - tokenRole === role && authInfo.deviceToken === assembled.storedToken - ? (assembled.storedScopes ?? authInfo.scopes ?? []) - : (authInfo.scopes ?? []); - this.deps.storeDeviceAuthToken({ - deviceId: this.opts.deviceIdentity.deviceId, - role: tokenRole, - token: authInfo.deviceToken, - scopes, - env: this.opts.env, - }); - } - if (this.opts.preferBootstrapToken) { - // The setup credential is single-use; reconnects must use the stored device token. - this.opts.token = undefined; - this.opts.bootstrapToken = undefined; - this.opts.password = undefined; - this.opts.preferBootstrapToken = false; - } this.tickIntervalMs = typeof helloOk.policy?.tickIntervalMs === "number" ? helloOk.policy.tickIntervalMs : 30_000; if (reconnectWithCurrentNodeProtocol) { // A v4 Gateway accepted the exact-v3 probe as a legacy session. Reconnect // before reporting readiness so node capabilities are not silently filtered. - this.suppressNextHelloCallback = true; this.protocol.resetReconnectBackoff(250); this.protocol.closeSocket(1012, "gateway protocol upgraded"); return; } this.lastTick = Date.now(); this.startTickWatch(); - void assembled; } private handleConnectRequestFailure( error: GatewayProtocolRequestError, assembled: AssembledConnect, + ) { + if (assembled.persistenceFailed) { + return { closeCode: 1008, closeReason: "connect failed" }; + } + const detailCode = + error instanceof GatewayClientRequestError ? readConnectErrorDetailCode(error.details) : null; + const cleared = + this.opts.deviceIdentity && + assembled.usingStoredDeviceToken && + detailCode === ConnectErrorDetailCodes.AUTH_DEVICE_TOKEN_MISMATCH + ? this.clearDeviceToken(this.opts.deviceIdentity.deviceId, { token: assembled.storedToken }) + : undefined; + const decision = this.resolveConnectRequestFailure(error, assembled); + return cleared instanceof Promise ? cleared.then(() => decision) : decision; + } + + private resolveConnectRequestFailure( + error: GatewayProtocolRequestError, + assembled: AssembledConnect, ) { if (this.shouldRetryWithCurrentNodeProtocol(error)) { const resetBackoff = !this.nodeProtocolTransitionPending; @@ -1028,7 +1011,6 @@ export class GatewayClient { return { closeCode: 1008, closeReason: "connect retry" }; } this.nodeProtocolTransitionPending = false; - const role = this.opts.role ?? "operator"; const detailCode = error instanceof GatewayClientRequestError ? readConnectErrorDetailCode(error.details) : null; const shouldRetryWithDeviceToken = shouldRetryGatewayWithDeviceToken({ @@ -1039,21 +1021,6 @@ export class GatewayClient { trustedEndpoint: this.isTrustedDeviceRetryEndpoint(), errorDetails: error instanceof GatewayClientRequestError ? error.details : undefined, }); - if ( - this.opts.deviceIdentity && - assembled.usingStoredDeviceToken && - detailCode === ConnectErrorDetailCodes.AUTH_DEVICE_TOKEN_MISMATCH - ) { - const deviceId = this.opts.deviceIdentity.deviceId; - try { - this.deps.clearDeviceAuthToken({ deviceId, role, env: this.opts.env }); - this.logDebug(`cleared stale device-auth token for device ${deviceId}`); - } catch (clearError) { - this.logDebug( - `failed clearing stale device-auth token for device ${deviceId}: ${String(clearError)}`, - ); - } - } if (shouldRetryWithDeviceToken) { this.pendingDeviceTokenRetry = true; this.deviceTokenRetryBudgetUsed = true; @@ -1147,7 +1114,7 @@ export class GatewayClient { this.notifyConnectError(error); this.logError(`gateway connect failed: ${formatGatewayClientErrorForLog(error)}`); } - this.clearStaleDeviceTokenForClose(context.code, context.reason); + this.clearStaleDeviceTokenForClose(context); if ( shouldPauseGatewayReconnect({ details, @@ -1179,35 +1146,71 @@ export class GatewayClient { }; } - private clearStaleDeviceTokenForClose(code: number, reason: string): void { + private clearStaleDeviceTokenForClose(context: GatewayProtocolCloseContext): void { + const { code, reason, generation } = context; if ( code !== 1008 || !normalizeGatewayErrorText(reason).includes("device token mismatch") || - this.opts.token || - this.opts.password || !this.opts.deviceIdentity ) { return; } - const deviceId = this.opts.deviceIdentity.deviceId; - const role = this.opts.role ?? "operator"; - try { - this.deps.clearDeviceAuthToken({ deviceId, role, env: this.opts.env }); - this.logDebug(`cleared stale device-auth token for device ${deviceId}`); - } catch (error) { + const observation = + this.connectionStoredToken?.generation === generation + ? this.connectionStoredToken + : { token: undefined }; + void this.clearDeviceToken( + this.opts.deviceIdentity.deviceId, + observation, + () => !this.opts.token && !this.opts.password, + ); + } + + private clearDeviceToken( + deviceId: string, + observation: DeviceAuthTokenObservation, + canClear = () => true, + ) { + const failed = (error: unknown) => { this.logDebug( `failed clearing stale device-auth token for device ${deviceId}: ${String(error)}`, ); + }; + const cleared = () => this.logDebug(`cleared stale device-auth token for device ${deviceId}`); + try { + const result = this.deviceAuth.clear( + { + deviceId, + role: this.opts.role ?? "operator", + env: this.opts.env, + // Cleanup follows a received rejection even after its transport retires. + assertCurrent: () => { + if (this.stopped) { + throw new Error("gateway client stopped"); + } + }, + }, + observation, + canClear, + ); + return result instanceof Promise ? result.then(cleared, failed) : cleared(); + } catch (error) { + failed(error); } } - private notifyConnectError(error: Error) { + private notifyConnectError(error: Error): boolean { + if (!this.opts.onConnectError) { + return false; + } try { - this.opts.onConnectError?.(error); + this.opts.onConnectError(error); + return true; } catch (err) { this.logDebug( `gateway client connect error handler error: ${formatGatewayClientErrorForLog(err)}`, ); + return false; } } @@ -1279,14 +1282,7 @@ export class GatewayClient { } } - private selectConnectAuth(role: string): GatewayConnectAuthSelection { - const storedAuth = this.opts.deviceIdentity - ? this.deps.loadDeviceAuthToken({ - deviceId: this.opts.deviceIdentity.deviceId, - role, - env: this.opts.env, - }) - : null; + private selectConnectAuth(storedAuth: DeviceAuthTokenRecord | null): GatewayConnectAuthSelection { return selectGatewayConnectAuth({ token: this.opts.token, bootstrapToken: this.opts.bootstrapToken, diff --git a/packages/gateway-client/src/protocol-client-contract.ts b/packages/gateway-client/src/protocol-client-contract.ts index 34bde97c6124..532168cd2471 100644 --- a/packages/gateway-client/src/protocol-client-contract.ts +++ b/packages/gateway-client/src/protocol-client-contract.ts @@ -14,7 +14,11 @@ export type GatewayProtocolSocketHandlers = { close: (code: number, reason: string) => void; error: (error: Error) => void; }; -type GatewayProtocolConnectContext = { +export type GatewayProtocolConnectAuthority = { + signal: AbortSignal; + assertCurrent: () => void; +}; +type GatewayProtocolConnectContext = GatewayProtocolConnectAuthority & { generation: number; nonce: string | null; challengeTs: number | null | undefined; @@ -66,20 +70,22 @@ export type GatewayProtocolClientOptions = { createRequestError?: (error: Partial) => GatewayProtocolRequestError; createRequestTimeoutError?: (method: string, timeoutMs: number, requestSent: boolean) => Error; createRequestAbortError?: (method: string) => Error; - buildConnectPlan: (params: { - nonce: string | null; - challengeTs: number | null | undefined; - serverCapabilities: readonly string[]; - generation: number; - }) => TPlan | Promise; + buildConnectPlan: ( + params: GatewayProtocolConnectAuthority & { + nonce: string | null; + challengeTs: number | null | undefined; + serverCapabilities: readonly string[]; + generation: number; + }, + ) => TPlan | Promise; buildConnectParams: (plan: TPlan) => unknown; onConnectPlanError?: (error: Error) => GatewayProtocolConnectDecision; - onConnectHello?: (hello: HelloOk, context: GatewayProtocolConnectContext) => void; + onConnectHello?: (hello: HelloOk, context: GatewayProtocolConnectContext) => unknown; onHello?: (hello: HelloOk) => void; onConnectFailure?: ( error: GatewayProtocolRequestError, context: GatewayProtocolConnectContext, - ) => GatewayProtocolConnectDecision; + ) => GatewayProtocolConnectDecision | Promise; resolveClose: (context: GatewayProtocolCloseContext) => GatewayProtocolCloseDecision; onClose?: (context: GatewayProtocolCloseContext, decision: GatewayProtocolCloseDecision) => void; notifyStoppedClose?: boolean; diff --git a/packages/gateway-client/src/protocol-client.handshake.test.ts b/packages/gateway-client/src/protocol-client.handshake.test.ts index f6962ae2754c..2d8aa49db078 100644 --- a/packages/gateway-client/src/protocol-client.handshake.test.ts +++ b/packages/gateway-client/src/protocol-client.handshake.test.ts @@ -15,7 +15,7 @@ function createHandshakeClient( ) { const connections: HandshakeConnection[] = []; const onHello = vi.fn(); - const onConnectHello = vi.fn(); + const onConnectHello = vi.fn(() => ({ ignored: true })); const onClose = vi.fn(); const onTiming = vi.fn(); let nextRequestId = 0; @@ -261,6 +261,8 @@ describe("GatewayProtocolClient connect handshake", () => { challengeTs: 1_700_000_000_123, serverCapabilities: [], generation: 1, + signal: expect.any(AbortSignal), + assertCurrent: expect.any(Function), }); client.stop(); }); @@ -286,6 +288,8 @@ describe("GatewayProtocolClient connect handshake", () => { challengeTs: 1, serverCapabilities: ["model-catalog-snapshot"], generation: 1, + signal: expect.any(AbortSignal), + assertCurrent: expect.any(Function), }); first.handlers.close(1006, "reconnect"); await vi.advanceTimersByTimeAsync(10); @@ -297,6 +301,8 @@ describe("GatewayProtocolClient connect handshake", () => { challengeTs: 1_800_000_000_000, serverCapabilities: [], generation: 2, + signal: expect.any(AbortSignal), + assertCurrent: expect.any(Function), }); } finally { client.stop(); @@ -325,6 +331,8 @@ describe("GatewayProtocolClient connect handshake", () => { challengeTs: null, serverCapabilities: [], generation: 1, + signal: expect.any(AbortSignal), + assertCurrent: expect.any(Function), }); client.stop(); @@ -348,6 +356,8 @@ describe("GatewayProtocolClient connect handshake", () => { challengeTs: null, serverCapabilities: [], generation: 1, + signal: expect.any(AbortSignal), + assertCurrent: expect.any(Function), }); secondClient.client.stop(); }); diff --git a/packages/gateway-client/src/protocol-client.ts b/packages/gateway-client/src/protocol-client.ts index 766816e38b51..62d32fce8940 100644 --- a/packages/gateway-client/src/protocol-client.ts +++ b/packages/gateway-client/src/protocol-client.ts @@ -11,6 +11,7 @@ import type { ConnectTimingState, GatewayProtocolClientOptions, GatewayProtocolCloseContext, + GatewayProtocolConnectAuthority, GatewayProtocolSocket, GatewayProtocolTiming, } from "./protocol-client-contract.js"; @@ -30,6 +31,7 @@ export { export type { GatewayProtocolCloseContext, + GatewayProtocolConnectAuthority, GatewayProtocolSocket, GatewayProtocolSocketHandlers, GatewayProtocolTiming, @@ -45,6 +47,7 @@ export class GatewayProtocolClient { private readonly listeners = new GatewayEventListeners(); private stopped = true; private generation = 0; + private connectionAbort: AbortController | null = null; private lastSeq: number | null = null; private connectNonce: string | null = null; private connectChallengeTs: number | null | undefined; @@ -106,6 +109,7 @@ export class GatewayProtocolClient { stop(): void { this.stopped = true; + this.connectionAbort?.abort(); this.clearHandshakeTimer(); this.reconnectSignal = null; this.reconnectSupervisor.reset(); @@ -142,6 +146,7 @@ export class GatewayProtocolClient { } closeSocket(code?: number, reason?: string): void { + this.connectionAbort?.abort(); this.socket?.close(code, reason); } @@ -226,6 +231,7 @@ export class GatewayProtocolClient { return; } this.generation = generation; + this.connectionAbort = new AbortController(); this.socket = socket; const now = this.nowMs(); this.connectTiming = { @@ -294,6 +300,7 @@ export class GatewayProtocolClient { challengeTs: this.connectChallengeTs, serverCapabilities: this.serverCapabilities, generation, + ...this.connectAuthority(socket, generation), }); } catch (error) { this.handleConnectPlanError(socket, generation, error); @@ -313,7 +320,7 @@ export class GatewayProtocolClient { generation: number, error: unknown, ): void { - if (!this.isActive(socket, generation)) { + if (!this.isConnectCurrent(socket, generation)) { return; } const normalized = error instanceof Error ? error : new Error(String(error)); @@ -329,10 +336,11 @@ export class GatewayProtocolClient { } private sendConnectPlan(socket: GatewayProtocolSocket, generation: number, plan: TPlan): void { - if (!this.isActive(socket, generation) || !socket.isOpen()) { + if (!this.isConnectCurrent(socket, generation)) { return; } const context = { + ...this.connectAuthority(socket, generation), generation, nonce: this.connectNonce, challengeTs: this.connectChallengeTs, @@ -345,7 +353,7 @@ export class GatewayProtocolClient { .then((hello) => { // Closing transports remain current until their close callback runs; // a late response must not publish readiness or reset reconnect backoff. - if (!this.isActive(socket, generation) || !socket.isOpen()) { + if (!this.isConnectCurrent(socket, generation)) { return; } this.helloReceived = true; @@ -353,8 +361,17 @@ export class GatewayProtocolClient { this.connectFailure = undefined; this.reconnectSupervisor.reset(); this.recordTiming("hello", generation, plan); - this.opts.onConnectHello?.(hello, context); - this.invoke("hello", () => this.opts.onHello?.(hello)); + const publishHello = () => { + if (!this.isConnectCurrent(socket, generation)) { + return; + } + this.invoke("hello", () => this.opts.onHello?.(hello)); + }; + const accepted = this.opts.onConnectHello?.(hello, context); + if (accepted instanceof Promise) { + return accepted.then(publishHello); + } + return publishHello(); }) .catch((error: unknown) => { if (!this.isActive(socket, generation)) { @@ -364,18 +381,37 @@ export class GatewayProtocolClient { error instanceof GatewayProtocolRequestError ? error : new GatewayProtocolRequestError({ message: String(error) }); + // Close can arrive while adapter cleanup is pending; retain the received error now. + this.connectFailure = { error: requestError }; const outcome = this.opts.onConnectFailure?.(requestError, context) ?? { closeCode: 1008, closeReason: "connect failed", }; - this.connectFailure = { - error: requestError, - reconnectDelayMs: outcome.reconnectDelayMs, + const applyFailure = (decision: Awaited) => { + if (!this.isActive(socket, generation)) { + return; + } + this.connectFailure = { + error: requestError, + reconnectDelayMs: decision.reconnectDelayMs, + }; + if (decision.stop) { + this.stopped = true; + } + this.connectionAbort?.abort(); + socket.close(decision.closeCode, decision.closeReason); }; - if (outcome.stop) { - this.stopped = true; + if (outcome instanceof Promise) { + return outcome.then(applyFailure); } - socket.close(outcome.closeCode, outcome.closeReason); + return applyFailure(outcome); + }) + .catch((error: unknown) => { + if (!this.isConnectCurrent(socket, generation)) { + return; + } + this.opts.onConnectError?.(error instanceof Error ? error : new Error(String(error))); + this.closeSocket(1008, "connect failed"); }); } @@ -467,6 +503,7 @@ export class GatewayProtocolClient { return; } this.socket = null; + this.connectionAbort?.abort(); this.clearHandshakeTimer(); const context: GatewayProtocolCloseContext = { ...this.closeContext(), @@ -551,6 +588,31 @@ export class GatewayProtocolClient { }; } + private isConnectCurrent(socket: GatewayProtocolSocket, generation: number): boolean { + return ( + this.isActive(socket, generation) && socket.isOpen() && !this.connectionAbort?.signal.aborted + ); + } + + private connectAuthority( + socket: GatewayProtocolSocket, + generation: number, + ): GatewayProtocolConnectAuthority { + const signal = this.connectionAbort?.signal; + if (!signal) { + throw new Error("gateway connection authority is unavailable"); + } + return { + signal, + assertCurrent: () => { + signal.throwIfAborted(); + if (!this.isConnectCurrent(socket, generation)) { + throw new Error("gateway connection retired"); + } + }, + }; + } + private isActive(socket: GatewayProtocolSocket, generation: number): boolean { return !this.stopped && this.socket === socket && this.generation === generation; } diff --git a/packages/gateway-client/src/request-error.ts b/packages/gateway-client/src/request-error.ts index 9bbb5e3816c3..29f050c4009a 100644 --- a/packages/gateway-client/src/request-error.ts +++ b/packages/gateway-client/src/request-error.ts @@ -11,3 +11,25 @@ export class GatewayClientRequestError extends GatewayProtocolRequestError { this.name = "GatewayClientRequestError"; } } + +const GATEWAY_CONNECT_ASSEMBLY_ERROR = Symbol("gateway.connectAssemblyError"); + +type GatewayConnectAssemblyError = Error & { + [GATEWAY_CONNECT_ASSEMBLY_ERROR]?: true; +}; + +export function markGatewayConnectAssemblyError(error: Error): Error { + Object.defineProperty(error, GATEWAY_CONNECT_ASSEMBLY_ERROR, { + configurable: true, + value: true, + }); + return error; +} + +export function isGatewayConnectAssemblyError(value: unknown): value is Error { + return ( + value instanceof Error && + // SAFETY: The Error may carry this module's optional private-symbol marker; absence reads undefined. + (value as GatewayConnectAssemblyError)[GATEWAY_CONNECT_ASSEMBLY_ERROR] === true + ); +} diff --git a/src/gateway/client.device-auth.test.ts b/src/gateway/client.device-auth.test.ts new file mode 100644 index 000000000000..ad6236941f4b --- /dev/null +++ b/src/gateway/client.device-auth.test.ts @@ -0,0 +1,206 @@ +import assert from "node:assert/strict"; +import { beforeEach, expect, it, vi } from "vitest"; +import type { GatewayClientHostDeps } from "../../packages/gateway-client/src/index.js"; +import type { loadDeviceAuthToken } from "../infra/device-auth-store.js"; +import { GatewayClient } from "./client.js"; + +const fixture = vi.hoisted(() => ({ + hosts: [] as GatewayClientHostDeps[], + token: "fixture-existing" as string | undefined, + malformed: false, + load: vi.fn(), + store: vi.fn(), + clear: vi.fn(), + readOnlyLoad: vi.fn(), +})); + +vi.mock("../../packages/gateway-client/src/index.js", async (importOriginal) => ({ + ...(await importOriginal()), + GatewayClient: class { + constructor(private readonly options: { hostDeps: GatewayClientHostDeps }) {} + + start() { + fixture.hosts.push(this.options.hostDeps); + } + }, +})); + +vi.mock("../infra/device-auth-store.js", async (importOriginal) => ({ + ...(await importOriginal()), + loadDeviceAuthToken: fixture.load, + loadOriginDeviceToken: fixture.load, + storeDeviceAuthToken: fixture.store, + storeOriginDeviceToken: fixture.store, + clearDeviceAuthToken: fixture.clear, + clearOriginDeviceToken: fixture.clear, + loadDeviceAuthTokenReadOnly: fixture.readOnlyLoad, + loadOriginDeviceTokenReadOnly: fixture.readOnlyLoad, +})); + +beforeEach(() => { + fixture.hosts.length = 0; + fixture.token = "fixture-existing"; + fixture.malformed = false; + fixture.load + .mockReset() + .mockImplementation((params: Parameters[0]) => { + const entry = + fixture.token === undefined || fixture.malformed + ? null + : { + token: fixture.token, + role: "operator", + scopes: [], + updatedAtMs: 1, + }; + params.onSnapshot?.({ entry, expectedToken: fixture.token ?? null }); + return entry; + }); + fixture.store + .mockReset() + .mockImplementation((params: { token: string; expectedToken?: string | null }) => { + if ( + params.expectedToken === undefined || + (params.expectedToken === null + ? fixture.token === undefined + : params.expectedToken === fixture.token) + ) { + fixture.token = params.token; + return { token: params.token, role: "operator", scopes: [], updatedAtMs: 1 }; + } + return null; + }); + fixture.clear + .mockReset() + .mockImplementation((params: { expectedToken?: string; observedToken?: string }) => { + if ( + params.expectedToken === fixture.token || + (params.observedToken?.trim() === params.expectedToken && + params.observedToken === fixture.token) + ) { + fixture.token = undefined; + return true; + } + return false; + }); + fixture.readOnlyLoad.mockReset(); +}); + +function host(origin: boolean, readOnly = false, explicit = true) { + const client = new GatewayClient({ + url: "wss://gateway.example.test", + ...(origin ? { deviceAuthScope: "wss://gateway.example.test" } : {}), + ...(explicit ? { token: "fixture-shared-auth" } : {}), + ...(readOnly ? { sharedStateMode: "read-only" } : {}), + }); + client.start(); + const deps = fixture.hosts[0]; + assert(deps?.loadDeviceAuthToken && deps.storeDeviceAuthToken && deps.clearDeviceAuthToken); + return { + load: deps.loadDeviceAuthToken, + store: deps.storeDeviceAuthToken, + clear: deps.clearDeviceAuthToken, + }; +} + +const observations = [ + { name: "valid", token: "fixture-existing", malformed: false }, + { name: "malformed scopes", token: "fixture-existing", malformed: true }, + { name: "absent", token: undefined, malformed: false }, + { name: "raw whitespace", token: " fixture-existing ", malformed: false }, +]; + +it.each( + observations.flatMap(({ name, token, malformed }) => + [false, true].flatMap((origin) => + [false, true].map((rotated) => ({ name, token, malformed, origin, rotated })), + ), + ), +)( + "uses exact $name comparison privately (origin: $origin, rotated: $rotated)", + ({ token, malformed, origin, rotated }) => { + fixture.token = token; + fixture.malformed = malformed; + const deps = host(origin); + const scope = { deviceId: "fixture-device", role: "operator" }; + const loaded = deps.load(scope); + assert( + !(loaded instanceof Promise), + "the synchronous facade must return its load result inline", + ); + if (origin || malformed || token === undefined) { + expect(loaded).toBeNull(); + } else { + expect(loaded?.token).toBe(token); + } + if (rotated) { + fixture.token = "fixture-newer"; + } + const stored = deps.store({ + ...scope, + token: "fixture-issued", + scopes: [], + expectedToken: loaded?.token?.trim() ?? null, + }); + expect(stored).toEqual( + rotated + ? null + : { + token: "fixture-issued", + role: "operator", + scopes: [], + updatedAtMs: 1, + }, + ); + expect(fixture.store).toHaveBeenCalledWith({ + ...scope, + ...(origin ? { gatewayScope: "wss://gateway.example.test" } : {}), + token: "fixture-issued", + scopes: [], + expectedToken: token ?? null, + }); + expect(fixture.token).toBe(rotated ? "fixture-newer" : "fixture-issued"); + }, +); + +it.each([false, true])( + "cleans exact raw legacy bytes without redirecting receipt cleanup (origin: %s)", + (origin) => { + fixture.token = " fixture-existing "; + const deps = host(origin, false, false); + const scope = { deviceId: "fixture-device", role: "operator" }; + const loaded = deps.load(scope); + assert( + !(loaded instanceof Promise), + "the synchronous facade must return its load result inline", + ); + expect(deps.clear({ ...scope, expectedToken: "fixture-existing" })).toBe(true); + expect(fixture.token).toBeUndefined(); + expect(fixture.clear).toHaveBeenLastCalledWith({ + ...scope, + ...(origin ? { gatewayScope: "wss://gateway.example.test" } : {}), + expectedToken: "fixture-existing", + observedToken: " fixture-existing ", + }); + fixture.token = "fixture-newer"; + expect(deps.clear({ ...scope, expectedToken: "fixture-issued" })).toBe(false); + expect(fixture.token).toBe("fixture-newer"); + expect(fixture.clear).toHaveBeenLastCalledWith({ + ...scope, + ...(origin ? { gatewayScope: "wss://gateway.example.test" } : {}), + expectedToken: "fixture-issued", + }); + }, +); + +it("keeps explicit read-only origin auth off storage reads and writes", () => { + const deps = host(true, true); + const scope = { deviceId: "fixture-device", role: "operator" }; + expect(deps.load(scope)).toBeNull(); + expect( + deps.store({ ...scope, token: "fixture-issued", scopes: [], expectedToken: null }), + ).toBeUndefined(); + expect(fixture.load).not.toHaveBeenCalled(); + expect(fixture.readOnlyLoad).not.toHaveBeenCalled(); + expect(fixture.store).not.toHaveBeenCalled(); +}); diff --git a/src/gateway/client.test-support.ts b/src/gateway/client.test-support.ts new file mode 100644 index 000000000000..deae5ecfe752 --- /dev/null +++ b/src/gateway/client.test-support.ts @@ -0,0 +1,16 @@ +import { vi } from "vitest"; + +export function waitForFast( + callback: () => T | Promise, + options: { timeout?: number; interval?: number } = {}, +) { + return vi.waitFor(callback, { interval: 1, ...options }); +} + +export function firstMockArg(mock: ReturnType, label: string): unknown { + const [arg] = mock.mock.calls[0] ?? []; + if (arg === undefined) { + throw new Error(`expected ${label}`); + } + return arg; +} diff --git a/src/gateway/client.test.ts b/src/gateway/client.test.ts index b7663cb157da..2bc86d5872d8 100644 --- a/src/gateway/client.test.ts +++ b/src/gateway/client.test.ts @@ -20,13 +20,7 @@ import { } from "../infra/device-identity.js"; import { captureEnv } from "../test-utils/env.js"; import type { GatewayClientOptions } from "./client.js"; - -function waitForFast( - callback: () => T | Promise, - options: { timeout?: number; interval?: number } = {}, -) { - return vi.waitFor(callback, { interval: 1, ...options }); -} +import { firstMockArg, waitForFast } from "./client.test-support.js"; type MockLoggingConfig = { redactPatterns?: string[]; @@ -190,8 +184,7 @@ vi.mock("../infra/device-auth-store.js", async () => { loadDeviceAuthToken: (...args: unknown[]) => loadDeviceAuthTokenMock(...args), loadDeviceAuthTokenReadOnly: (...args: unknown[]) => loadDeviceAuthTokenReadOnlyMock(...args), loadOriginDeviceToken: (...args: unknown[]) => loadOriginDeviceTokenMock(...args), - loadOriginDeviceTokenReadOnly: (...args: unknown[]) => - loadOriginDeviceTokenReadOnlyMock(...args), + loadOriginDeviceTokenReadOnly: loadOriginDeviceTokenReadOnlyMock, storeDeviceAuthToken: (...args: unknown[]) => storeDeviceAuthTokenMock(...args), storeOriginDeviceToken: (...args: unknown[]) => storeOriginDeviceTokenMock(...args), clearDeviceAuthToken: (...args: unknown[]) => clearDeviceAuthTokenMock(...args), @@ -250,13 +243,6 @@ function expectRecordFields( return record; } -function firstMockArg(mock: ReturnType, label: string): unknown { - const [arg] = mock.mock.calls[0] ?? []; - if (arg === undefined) { - throw new Error(`expected ${label}`); - } - return arg; -} function createClientWithIdentity( deviceId: string, onClose: (code: number, reason: string) => void, @@ -738,6 +724,7 @@ describe("GatewayClient close handling", () => { deviceId: "dev-1", role: "operator", env, + assertCurrent: expect.any(Function), }); expect(logDebugMock).toHaveBeenCalledWith("cleared stale device-auth token for device dev-1"); expect(onClose).toHaveBeenCalledWith( @@ -2321,7 +2308,7 @@ describe("GatewayClient connect auth payload", () => { connectId: connect.id, failureDetails: { code: "AUTH_TOKEN_MISMATCH", canRetryWithDeviceToken: true }, }); - expect(loadOriginDeviceTokenMock).not.toHaveBeenCalled(); + expect(loadOriginDeviceTokenMock).toHaveBeenCalledOnce(); expect(onReconnectPaused).toHaveBeenCalledWith({ code: 1008, reason: "connect failed", @@ -2363,6 +2350,7 @@ describe("GatewayClient connect auth payload", () => { token: "stored-origin-token", scopes: ["operator.admin", "operator.read"], env: undefined, + expectedToken: "stored-origin-token", }); }); client.stop(); @@ -2398,6 +2386,7 @@ describe("GatewayClient connect auth payload", () => { token: "issued-origin-token", scopes: ["operator.read"], env: undefined, + expectedToken: null, }); }); expect(storeDeviceAuthTokenMock).not.toHaveBeenCalled(); diff --git a/src/gateway/client.ts b/src/gateway/client.ts index ba3dde4f43c9..4fd734472764 100644 --- a/src/gateway/client.ts +++ b/src/gateway/client.ts @@ -30,7 +30,7 @@ import { import { logDebug, logError } from "../logger.js"; import { redactToolPayloadText } from "../logging/redact.js"; import { registerSecretValueForRedaction } from "../logging/secret-redaction-registry.js"; -import type { DeviceAuthEntry } from "../shared/device-auth.js"; +import { type DeviceAuthEntry, normalizeDeviceAuthRole } from "../shared/device-auth.js"; import { resolveGatewayClientPlatformIdentity } from "../shared/gateway-client-platform.js"; import { VERSION } from "../version.js"; @@ -67,24 +67,64 @@ function createOpenClawGatewayClientHostDeps( const rotationFence = preparedDeviceAuth ? { expectedToken: preparedDeviceAuth.token } : undefined; + let tokenObservation: + | { deviceId: string; role: string; expectedToken: string | null } + | undefined; + const observe = (params: { deviceId: string; role: string }) => { + const deviceId = params.deviceId; + const role = normalizeDeviceAuthRole(params.role); + return (snapshot: { expectedToken: string | null }) => { + tokenObservation = { deviceId, role, expectedToken: snapshot.expectedToken }; + }; + }; + const observedFor = (params: { deviceId: string; role: string }) => + tokenObservation?.deviceId === params.deviceId && + tokenObservation.role === normalizeDeviceAuthRole(params.role) + ? tokenObservation + : undefined; + const writeFence = (params: { deviceId: string; role: string }) => { + if (rotationFence) { + return rotationFence; + } + // Each connection's accepted writes settle before its successor loads another observation. + const observed = observedFor(params); + return observed ? { expectedToken: observed.expectedToken } : undefined; + }; + const clearFence = (params: { deviceId: string; role: string; expectedToken?: string }) => { + const expectedToken = rotationFence?.expectedToken ?? params.expectedToken; + const raw = observedFor(params)?.expectedToken; + return { + ...rotationFence, + ...(typeof raw === "string" && raw !== expectedToken && raw.trim() === expectedToken + ? { observedToken: raw } + : {}), + }; + }; const deviceAuthDeps: Pick< GatewayClientHostDeps, "loadDeviceAuthToken" | "storeDeviceAuthToken" | "clearDeviceAuthToken" > = deviceAuthScope ? { - loadDeviceAuthToken: (params) => - suppressStoredDeviceAuth - ? null - : readOnly - ? loadOriginDeviceTokenReadOnly({ ...params, gatewayScope: deviceAuthScope }) - : loadOriginDeviceToken({ ...params, gatewayScope: deviceAuthScope }), + loadDeviceAuthToken: (params) => { + if (readOnly) { + return suppressStoredDeviceAuth + ? null + : loadOriginDeviceTokenReadOnly({ ...params, gatewayScope: deviceAuthScope }); + } + const load = loadOriginDeviceToken({ + ...params, + gatewayScope: deviceAuthScope, + onSnapshot: observe(params), + }); + return suppressStoredDeviceAuth ? null : load; + }, storeDeviceAuthToken: readOnly ? () => {} : (params) => storeOriginDeviceToken({ ...params, gatewayScope: deviceAuthScope, - ...rotationFence, + ...writeFence(params), }), clearDeviceAuthToken: readOnly ? () => {} @@ -92,7 +132,7 @@ function createOpenClawGatewayClientHostDeps( clearOriginDeviceToken({ ...params, gatewayScope: deviceAuthScope, - ...rotationFence, + ...clearFence(params), }), } : readOnly @@ -102,9 +142,12 @@ function createOpenClawGatewayClientHostDeps( clearDeviceAuthToken: () => {}, } : { - loadDeviceAuthToken, - storeDeviceAuthToken: (params) => storeDeviceAuthToken({ ...params, ...rotationFence }), - clearDeviceAuthToken: (params) => clearDeviceAuthToken({ ...params, ...rotationFence }), + loadDeviceAuthToken: (params) => + loadDeviceAuthToken({ ...params, onSnapshot: observe(params) }), + storeDeviceAuthToken: (params) => + storeDeviceAuthToken({ ...params, ...writeFence(params) }), + clearDeviceAuthToken: (params) => + clearDeviceAuthToken({ ...params, ...clearFence(params) }), }; const preparedDeviceAuthDeps = preparedDeviceAuth ? { ...deviceAuthDeps, loadDeviceAuthToken: () => preparedDeviceAuth } diff --git a/src/tui/gateway-chat.scopes.test.ts b/src/tui/gateway-chat.scopes.test.ts index 2766f2aa1c0f..77dfce9f444e 100644 --- a/src/tui/gateway-chat.scopes.test.ts +++ b/src/tui/gateway-chat.scopes.test.ts @@ -298,6 +298,7 @@ describe("GatewayChatClient operator scopes", () => { token: "approved-tui-device-token", scopes: requestedScopes, env: undefined, + expectedToken: "cached-read-only-device-token", }); const history = approved.client.loadHistory({ sessionKey: "main", limit: 20 });