From 7197eef3ebeb5ac294da51ca073fff33277ed429 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sun, 12 Jul 2026 09:33:14 +0100 Subject: [PATCH] fix(gateway): queue agent runtime identity verification (#105143) * fix(gateway): queue agent runtime identity verification * chore: keep release changelog unchanged * fix(gateway): recheck runtime context expiry after lock wait --- src/agents/tools/gateway.test.ts | 6 +- .../agent-runtime-identity-token.test.ts | 67 +++++++++++++++++-- src/gateway/agent-runtime-identity-token.ts | 20 +++--- .../server/ws-connection/message-handler.ts | 5 +- src/infra/exec-approvals.ts | 12 ++++ 5 files changed, 91 insertions(+), 19 deletions(-) diff --git a/src/agents/tools/gateway.test.ts b/src/agents/tools/gateway.test.ts index 95de2ba07815..8207462aa34c 100644 --- a/src/agents/tools/gateway.test.ts +++ b/src/agents/tools/gateway.test.ts @@ -497,7 +497,7 @@ describe("gateway tool defaults", () => { sessionId: "session-1", }); expect(token).toEqual(expect.any(String)); - expect(verifyAgentRuntimeIdentityToken(token)).toMatchObject({ + await expect(verifyAgentRuntimeIdentityToken(token)).resolves.toMatchObject({ messageActionContext: { sessionId: "session-1", requesterAccountId: "default", @@ -704,7 +704,9 @@ describe("gateway tool defaults", () => { turnSourceAccountId: "work", turnSourceThreadId: 42, }); - expect(verifyAgentRuntimeIdentityToken(call.agentRuntimeIdentityToken ?? "")).toMatchObject({ + await expect( + verifyAgentRuntimeIdentityToken(call.agentRuntimeIdentityToken ?? ""), + ).resolves.toMatchObject({ agentId: "ops", sessionKey: "agent:ops:telegram:direct:alice", }); diff --git a/src/gateway/agent-runtime-identity-token.test.ts b/src/gateway/agent-runtime-identity-token.test.ts index 4fa09d7c84aa..b598acae325f 100644 --- a/src/gateway/agent-runtime-identity-token.test.ts +++ b/src/gateway/agent-runtime-identity-token.test.ts @@ -59,7 +59,7 @@ describe("agent runtime identity token", () => { expect(persistedToken).not.toHaveLength(0); const secondProcess = await importRuntimeTokenModule(); - expect(secondProcess.verifyAgentRuntimeIdentityToken(token)).toEqual({ + await expect(secondProcess.verifyAgentRuntimeIdentityToken(token)).resolves.toEqual({ kind: "agentRuntime", agentId: "main", sessionKey: "session-1", @@ -70,7 +70,9 @@ describe("agent runtime identity token", () => { const home = useTempHome(); const runtimeToken = await importRuntimeTokenModule(); - expect(runtimeToken.verifyAgentRuntimeIdentityToken("not-a-valid-token")).toBeUndefined(); + await expect( + runtimeToken.verifyAgentRuntimeIdentityToken("not-a-valid-token"), + ).resolves.toBeUndefined(); expect(fs.existsSync(execApprovalsPath(home))).toBe(false); }); @@ -91,7 +93,7 @@ describe("agent runtime identity token", () => { }); expect(secondToken).not.toBe(token); - expect(secondProcess.verifyAgentRuntimeIdentityToken(token)).toBeUndefined(); + await expect(secondProcess.verifyAgentRuntimeIdentityToken(token)).resolves.toBeUndefined(); }); it("round-trips signed message action context and rejects it after expiry", async () => { @@ -113,7 +115,7 @@ describe("agent runtime identity token", () => { }, }); - expect(runtimeToken.verifyAgentRuntimeIdentityToken(token, 4000)).toMatchObject({ + await expect(runtimeToken.verifyAgentRuntimeIdentityToken(token, 4000)).resolves.toMatchObject({ kind: "agentRuntime", agentId: "main", sessionKey: "session-1", @@ -129,6 +131,61 @@ describe("agent runtime identity token", () => { }, }, }); - expect(runtimeToken.verifyAgentRuntimeIdentityToken(token, 5000)).toBeUndefined(); + await expect( + runtimeToken.verifyAgentRuntimeIdentityToken(token, 5000), + ).resolves.toBeUndefined(); + }); + + it("queues parallel verifications behind a same-process approvals update", async () => { + useTempHome(); + const runtimeToken = await importRuntimeTokenModule(); + const { updateExecApprovals } = await import("../infra/exec-approvals.js"); + const token = await runtimeToken.mintAgentRuntimeIdentityToken({ + agentId: "main", + sessionKey: "session-1", + }); + let verifications: Array> = []; + + await updateExecApprovals({ + update: () => { + // Verification can begin while another parallel agent call still owns + // the process-local approvals lock. It must queue behind that owner. + verifications = Array.from({ length: 8 }, () => + runtimeToken.verifyAgentRuntimeIdentityToken(token), + ); + return null; + }, + }); + + await expect(Promise.all(verifications)).resolves.toEqual( + Array.from({ length: 8 }, () => ({ + kind: "agentRuntime", + agentId: "main", + sessionKey: "session-1", + })), + ); + }); + + it("rechecks message action expiry after waiting for an approvals update", async () => { + useTempHome(); + const runtimeToken = await importRuntimeTokenModule(); + const { updateExecApprovals } = await import("../infra/exec-approvals.js"); + const token = await runtimeToken.mintAgentRuntimeIdentityToken({ + agentId: "main", + sessionKey: "session-1", + messageActionContext: { expiresAtMs: 5000 }, + }); + const nowSpy = vi.spyOn(Date, "now").mockReturnValue(4000); + let verification!: ReturnType; + + await updateExecApprovals({ + update: () => { + verification = runtimeToken.verifyAgentRuntimeIdentityToken(token); + nowSpy.mockReturnValue(5000); + return null; + }, + }); + + await expect(verification).resolves.toBeUndefined(); }); }); diff --git a/src/gateway/agent-runtime-identity-token.ts b/src/gateway/agent-runtime-identity-token.ts index 265f017eae19..db8914e51e1c 100644 --- a/src/gateway/agent-runtime-identity-token.ts +++ b/src/gateway/agent-runtime-identity-token.ts @@ -4,7 +4,7 @@ import { isRecord } from "@openclaw/normalization-core/record-coerce"; import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; import { normalizeChatType } from "../channels/chat-type.js"; import type { ChannelId, ChannelThreadingToolContext } from "../channels/plugins/types.public.js"; -import { ensureExecApprovalsSnapshot, loadExecApprovals } from "../infra/exec-approvals.js"; +import { ensureExecApprovalsSnapshot, loadExecApprovalsAsync } from "../infra/exec-approvals.js"; import { normalizeAgentId } from "../routing/session-key.js"; import type { AgentRuntimeMessageActionContext } from "./message-action-turn-capability.js"; @@ -25,8 +25,8 @@ type AgentRuntimeIdentityTokenPayload = { messageActionContext?: AgentRuntimeMessageActionContext; }; -function readSharedAgentRuntimeIdentitySecret(): string | null { - return loadExecApprovals().socket?.token?.trim() || null; +async function readSharedAgentRuntimeIdentitySecret(): Promise { + return (await loadExecApprovalsAsync()).socket?.token?.trim() || null; } async function requireSharedAgentRuntimeIdentitySecret(): Promise { @@ -192,10 +192,10 @@ export async function mintAgentRuntimeIdentityToken(params: { } /** Validate a presented agent runtime token and return the internal caller identity. */ -export function verifyAgentRuntimeIdentityToken( +export async function verifyAgentRuntimeIdentityToken( value: string | null | undefined, - nowMs: number = Date.now(), -): AgentRuntimeIdentity | undefined { + nowMs?: number, +): Promise { const token = value?.trim(); if (!token) { return undefined; @@ -204,12 +204,12 @@ export function verifyAgentRuntimeIdentityToken( if (!payloadPart || !signature || extra.length > 0) { return undefined; } - const payload = decodePayload(payloadPart, nowMs); - if (!payload) { + const sharedSecret = await readSharedAgentRuntimeIdentitySecret(); + if (!sharedSecret || !signatureMatches(signature, signPayload(sharedSecret, payloadPart))) { return undefined; } - const sharedSecret = readSharedAgentRuntimeIdentitySecret(); - if (!sharedSecret || !signatureMatches(signature, signPayload(sharedSecret, payloadPart))) { + const payload = decodePayload(payloadPart, nowMs ?? Date.now()); + if (!payload) { return undefined; } return { diff --git a/src/gateway/server/ws-connection/message-handler.ts b/src/gateway/server/ws-connection/message-handler.ts index eb768c7a434e..159909c97369 100644 --- a/src/gateway/server/ws-connection/message-handler.ts +++ b/src/gateway/server/ws-connection/message-handler.ts @@ -2234,7 +2234,7 @@ export function attachGatewayWsMessageHandler(params: GatewayWsMessageHandlerPar connectParams.client.id === GATEWAY_CLIENT_IDS.GATEWAY_CLIENT && connectParams.client.mode === GATEWAY_CLIENT_MODES.BACKEND; let trustedAgentRuntimeIdentity: - | ReturnType + | Awaited> | undefined; if (typeof agentRuntimeIdentityToken === "string") { if (!canAcceptAgentRuntimeIdentity) { @@ -2249,7 +2249,8 @@ export function attachGatewayWsMessageHandler(params: GatewayWsMessageHandlerPar close(1008, truncateCloseReason(message)); return; } - trustedAgentRuntimeIdentity = verifyAgentRuntimeIdentityToken(agentRuntimeIdentityToken); + trustedAgentRuntimeIdentity = + await verifyAgentRuntimeIdentityToken(agentRuntimeIdentityToken); if (!trustedAgentRuntimeIdentity) { const message = "invalid agent runtime identity token"; markHandshakeFailure("agent-runtime-identity-invalid", { diff --git a/src/infra/exec-approvals.ts b/src/infra/exec-approvals.ts index 7c190b8c2fca..96feeaab3f31 100644 --- a/src/infra/exec-approvals.ts +++ b/src/infra/exec-approvals.ts @@ -1110,6 +1110,18 @@ export function loadExecApprovals(): ExecApprovalsFile { } } +export async function loadExecApprovalsAsync(): Promise { + try { + return await withExecApprovalsReadLock(resolveExecApprovalsPath(), async () => + loadExecApprovalsUnlocked(), + ); + } catch { + // Match the synchronous reader's fail-closed contract while allowing + // same-process async writers to finish instead of rejecting valid state. + return createFailClosedExecApprovalsFallback(); + } +} + type ExecApprovalsSyncLock = { descriptor: number; lockPath: string;