From ee2b4e5acc590c6e0a67dd547ea88f7a2ac8c13c Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 17 Jul 2026 11:11:13 +0100 Subject: [PATCH] fix(sms): prevent silent loss of acknowledged Twilio messages (#109866) * fix(sms): adopt durable ingress drain with ack gated on MessageSid enqueue Twilio inbound webhooks acked 200 then dispatched detached behind a 10-minute in-memory replay guard; Twilio does not reliably retry inbound messages, so a crash between ack and dispatch silently lost the SMS. The raw Twilio form now enqueues durably (event_id = MessageSid, lane per sender) before the TwiML 200; dispatch, retry, dead-letter, and tombstones run through the core drain with a startup + interval pump for restart recovery. Payload/account validation moved to parse-at-dispatch and dead-letters as non-retryable. The replay guard and its saturation 429 path are deleted; 24h/20k tombstones strictly cover the old 10min/10k window. Autoreview P1 (ambiguous non-adopted settlement) rejected with evidence: bindIngressLifecycleToReplyOptions threads onAdopted/onDeferred/onAbandoned into reply admission, so deferral and abandonment settle via core lifecycle callbacks; normal non-adopted return = terminal local handling, matching the Telegram reference mapping (skipped -> completed). Part of #109657 wave 1. * fix(sms): keep ingress payload type internal * fix(sms): preserve Twilio message SID aliases * fix(sms): preserve ingress ordering and timestamps * test(sms): satisfy type and lint gates --- extensions/sms/src/gateway.test.ts | 15 + extensions/sms/src/gateway.ts | 59 ++-- extensions/sms/src/inbound.test.ts | 7 + extensions/sms/src/inbound.ts | 9 +- extensions/sms/src/ingress-spool.test.ts | 272 ++++++++++++++++++ extensions/sms/src/ingress-spool.ts | 129 +++++++++ extensions/sms/src/twilio.ts | 19 +- .../sms/src/webhook-replay-guard.test.ts | 51 ---- extensions/sms/src/webhook-replay-guard.ts | 56 ---- extensions/sms/src/webhook.test.ts | 155 +++++++--- extensions/sms/src/webhook.ts | 77 ++--- 11 files changed, 611 insertions(+), 238 deletions(-) create mode 100644 extensions/sms/src/ingress-spool.test.ts create mode 100644 extensions/sms/src/ingress-spool.ts delete mode 100644 extensions/sms/src/webhook-replay-guard.test.ts delete mode 100644 extensions/sms/src/webhook-replay-guard.ts diff --git a/extensions/sms/src/gateway.test.ts b/extensions/sms/src/gateway.test.ts index ae74652e04f4..0f4940ccac4f 100644 --- a/extensions/sms/src/gateway.test.ts +++ b/extensions/sms/src/gateway.test.ts @@ -4,6 +4,16 @@ import { startSmsGatewayAccount } from "./gateway.js"; import type { SmsChannelRuntime } from "./inbound.js"; import type { ResolvedSmsAccount } from "./types.js"; +const drainSmsIngress = vi.hoisted(() => vi.fn(async () => undefined)); +const disposeSmsIngress = vi.hoisted(() => vi.fn()); +const createSmsIngressSpool = vi.hoisted(() => + vi.fn(() => ({ + enqueue: vi.fn(), + drainOnce: drainSmsIngress, + dispose: disposeSmsIngress, + })), +); + const { registeredRoutes, registerPluginHttpRoute, waitUntilAbort } = vi.hoisted(() => { const routeCleanups: Array<() => void> = []; return { @@ -19,6 +29,8 @@ const { registeredRoutes, registerPluginHttpRoute, waitUntilAbort } = vi.hoisted vi.mock("openclaw/plugin-sdk/channel-outbound", () => ({ waitUntilAbort })); +vi.mock("./ingress-spool.js", () => ({ createSmsIngressSpool })); + vi.mock("openclaw/plugin-sdk/webhook-ingress", () => ({ createFixedWindowRateLimiter: () => ({ clear: vi.fn(), @@ -51,6 +63,9 @@ describe("startSmsGatewayAccount", () => { beforeEach(() => { registerPluginHttpRoute.mockClear(); waitUntilAbort.mockClear(); + createSmsIngressSpool.mockClear(); + drainSmsIngress.mockClear(); + disposeSmsIngress.mockClear(); }); afterEach(() => { diff --git a/extensions/sms/src/gateway.ts b/extensions/sms/src/gateway.ts index 53bee5c57457..de9828ddd2f3 100644 --- a/extensions/sms/src/gateway.ts +++ b/extensions/sms/src/gateway.ts @@ -1,10 +1,12 @@ // Sms plugin module implements gateway behavior. import { waitUntilAbort } from "openclaw/plugin-sdk/channel-outbound"; import { registerPluginHttpRoute } from "openclaw/plugin-sdk/webhook-ingress"; +import { createSmsIngressSpool } from "./ingress-spool.js"; import type { ResolvedSmsAccount } from "./types.js"; import { createSmsWebhookHandler, type SmsWebhookHandlerParams } from "./webhook.js"; const CHANNEL_ID = "sms"; +const SMS_INGRESS_DRAIN_INTERVAL_MS = 500; const activeRoutes = new Map void>(); const activeRoutePaths = new Map(); @@ -52,7 +54,8 @@ export function collectSmsStartupWarnings(account: ResolvedSmsAccount): string[] function registerSmsWebhookRoute(params: { cfg: SmsWebhookHandlerParams["cfg"]; account: ResolvedSmsAccount; - channelRuntime: SmsWebhookHandlerParams["channelRuntime"]; + channelRuntime: Parameters[0]["channelRuntime"]; + abortSignal: AbortSignal; log?: SmsGatewayLog; }): () => void { const key = routeKey(params.account); @@ -65,29 +68,49 @@ function registerSmsWebhookRoute(params: { } activeRoutes.get(key)?.(); activeRoutePaths.delete(webhookPath); - const unregister = registerPluginHttpRoute({ - path: webhookPath, - auth: "plugin", - pluginId: CHANNEL_ID, - accountId: params.account.accountId, - log: (msg) => params.log?.info?.(msg), - handler: createSmsWebhookHandler(params), - }); - activeRoutes.set(key, unregister); - activeRoutePaths.set(webhookPath, params.account.accountId); - return () => { - unregister(); - activeRoutes.delete(key); - if (activeRoutePaths.get(webhookPath) === params.account.accountId) { - activeRoutePaths.delete(webhookPath); - } + const ingress = createSmsIngressSpool(params); + const requestDrain = () => { + void ingress.drainOnce().catch((error: unknown) => { + params.log?.error?.( + `SMS ingress drain failed: ${error instanceof Error ? error.message : String(error)}`, + ); + }); }; + const drainTimer = setInterval(requestDrain, SMS_INGRESS_DRAIN_INTERVAL_MS); + drainTimer.unref?.(); + try { + const unregisterRoute = registerPluginHttpRoute({ + path: webhookPath, + auth: "plugin", + pluginId: CHANNEL_ID, + accountId: params.account.accountId, + log: (msg) => params.log?.info?.(msg), + handler: createSmsWebhookHandler({ ...params, ingress }), + }); + const unregister = () => { + clearInterval(drainTimer); + ingress.dispose(); + unregisterRoute(); + activeRoutes.delete(key); + if (activeRoutePaths.get(webhookPath) === params.account.accountId) { + activeRoutePaths.delete(webhookPath); + } + }; + activeRoutes.set(key, unregister); + activeRoutePaths.set(webhookPath, params.account.accountId); + requestDrain(); + return unregister; + } catch (error) { + clearInterval(drainTimer); + ingress.dispose(); + throw error; + } } export async function startSmsGatewayAccount(params: { cfg: SmsWebhookHandlerParams["cfg"]; account: ResolvedSmsAccount; - channelRuntime: SmsWebhookHandlerParams["channelRuntime"]; + channelRuntime: Parameters[0]["channelRuntime"]; abortSignal: AbortSignal; log?: SmsGatewayLog; }) { diff --git a/extensions/sms/src/inbound.test.ts b/extensions/sms/src/inbound.test.ts index 4952248855c2..3c7030622461 100644 --- a/extensions/sms/src/inbound.test.ts +++ b/extensions/sms/src/inbound.test.ts @@ -38,6 +38,7 @@ function createRuntime() { const resolveAgentRoute = vi.fn(); const run = vi.fn< (params: { + turnAdoptionLifecycle?: { onAdopted: () => void | Promise }; adapter: { ingest: (msg: { from: string; @@ -93,6 +94,7 @@ describe("dispatchSmsInboundEvent", () => { cfg: {}, account: createAccount(), channelRuntime: runtime, + receivedAt: 1_700_000_000_000, msg: { from: "+15551234567", to: "+15557654321", @@ -130,6 +132,7 @@ describe("dispatchSmsInboundEvent", () => { }); buildContext.mockReturnValue({ SessionKey: "agent:main:sms:direct:+15551234567" }); resolveStorePath.mockReturnValue("/tmp/openclaw-sessions"); + const turnAdoptionLifecycle = { onAdopted: vi.fn(async () => undefined) }; await dispatchSmsInboundEvent({ cfg: {}, @@ -138,6 +141,8 @@ describe("dispatchSmsInboundEvent", () => { allowFrom: ["+15551234567"], }), channelRuntime: runtime, + receivedAt: 1_700_000_000_123, + turnAdoptionLifecycle, msg: { from: "+15551234567", to: "+15557654321", @@ -148,6 +153,7 @@ describe("dispatchSmsInboundEvent", () => { }); const runParams = expectDefined(run.mock.calls[0]?.[0], "SMS inbound run parameters"); + expect(runParams.turnAdoptionLifecycle).toBe(turnAdoptionLifecycle); const ingested = runParams.adapter.ingest({ from: "+15551234567", to: "+15557654321", @@ -164,6 +170,7 @@ describe("dispatchSmsInboundEvent", () => { ); expect(buildContext).toHaveBeenCalledWith( expect.objectContaining({ + timestamp: 1_700_000_000_123, from: "sms:+15551234567", sender: expect.objectContaining({ id: "+15551234567" }), conversation: expect.objectContaining({ id: "+15551234567" }), diff --git a/extensions/sms/src/inbound.ts b/extensions/sms/src/inbound.ts index 9033b7cb44f9..f824ac33ae40 100644 --- a/extensions/sms/src/inbound.ts +++ b/extensions/sms/src/inbound.ts @@ -89,6 +89,10 @@ export async function dispatchSmsInboundEvent(params: { account: ResolvedSmsAccount; msg: SmsInboundMessage; channelRuntime: SmsChannelRuntime; + receivedAt: number; + turnAdoptionLifecycle?: NonNullable< + Parameters[0]["turnAdoptionLifecycle"] + >; log?: SmsLog; }): Promise { const from = normalizeSmsPhoneNumber(params.msg.from); @@ -127,10 +131,13 @@ export async function dispatchSmsInboundEvent(params: { channel: CHANNEL_ID, accountId: params.account.accountId, raw: params.msg, + ...(params.turnAdoptionLifecycle + ? { turnAdoptionLifecycle: params.turnAdoptionLifecycle } + : {}), adapter: { ingest: (msg) => ({ id: msg.messageSid, - timestamp: Date.now(), + timestamp: params.receivedAt, rawText: msg.body, textForAgent: msg.body, textForCommands: msg.body, diff --git a/extensions/sms/src/ingress-spool.test.ts b/extensions/sms/src/ingress-spool.test.ts new file mode 100644 index 000000000000..fb7763a45ae6 --- /dev/null +++ b/extensions/sms/src/ingress-spool.test.ts @@ -0,0 +1,272 @@ +// Sms tests cover durable Twilio webhook admission and replay. +import { mkdtemp, realpath, rm } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { createChannelIngressQueueForTests } from "openclaw/plugin-sdk/plugin-state-test-runtime"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { SmsChannelRuntime } from "./inbound.js"; +import { createSmsIngressSpool } from "./ingress-spool.js"; +import type { ResolvedSmsAccount } from "./types.js"; + +type SmsIngressPayload = { + version: 1; + form: Record; +}; + +const account: ResolvedSmsAccount = { + accountId: "default", + enabled: true, + accountSid: "AC123", + authToken: "secret", + fromNumber: "+15557654321", + messagingServiceSid: "", + defaultTo: "", + webhookPath: "/webhooks/sms", + publicWebhookUrl: "https://gateway.example.com/webhooks/sms", + dangerouslyDisableSignatureValidation: false, + dmPolicy: "pairing", + allowFrom: [], + textChunkLimit: 1500, +}; + +const stateDirs: string[] = []; +const disposers: Array<() => void> = []; +type SmsIngressDeliver = NonNullable[0]["deliver"]>; +type SmsIngressSpool = ReturnType; + +async function createStateDir(): Promise { + const created = await mkdtemp(path.join(os.tmpdir(), "openclaw-sms-ingress-")); + const resolved = await realpath(created); + stateDirs.push(resolved); + return resolved; +} + +function createQueue(stateDir: string) { + return createChannelIngressQueueForTests({ + channelId: "sms", + accountId: account.accountId, + stateDir, + }); +} + +function form(messageSid: string): Record { + return { + AccountSid: account.accountSid, + From: "+15551234567", + To: "+15557654321", + Body: "hello", + MessageSid: messageSid, + }; +} + +async function drainSpool(spool: SmsIngressSpool): Promise { + await spool.drainOnce(); + await spool.waitForIdle(); +} + +afterEach(async () => { + for (const dispose of disposers.splice(0).toReversed()) { + dispose(); + } + for (const stateDir of stateDirs.splice(0).toReversed()) { + await rm(stateDir, { recursive: true, force: true }); + } +}); + +describe("createSmsIngressSpool", () => { + it("recovers an uncompleted message with a fresh drain instance", async () => { + const stateDir = await createStateDir(); + const first = createSmsIngressSpool({ + cfg: {}, + account, + channelRuntime: {} as SmsChannelRuntime, + queue: createQueue(stateDir), + deliver: vi.fn(async () => undefined), + }); + disposers.push(first.dispose); + await first.enqueue(form("SM-restart")); + first.dispose(); + + const deliver = vi.fn(async (_message, lifecycle) => { + await lifecycle.onAdopted(); + }); + const recovered = createSmsIngressSpool({ + cfg: {}, + account, + channelRuntime: {} as SmsChannelRuntime, + queue: createQueue(stateDir), + deliver, + }); + disposers.push(recovered.dispose); + await drainSpool(recovered); + + expect(deliver).toHaveBeenCalledOnce(); + }); + + it("keeps a completed MessageSid tombstone from dispatching twice", async () => { + const stateDir = await createStateDir(); + const deliver = vi.fn(async (_message, lifecycle) => { + await lifecycle.onAdopted(); + }); + const spool = createSmsIngressSpool({ + cfg: {}, + account, + channelRuntime: {} as SmsChannelRuntime, + queue: createQueue(stateDir), + deliver, + }); + disposers.push(spool.dispose); + + expect(await spool.enqueue(form("SM-completed"))).toMatchObject({ + kind: "accepted", + duplicate: false, + }); + await drainSpool(spool); + expect(await spool.enqueue(form("SM-completed"))).toMatchObject({ + kind: "completed", + duplicate: true, + }); + await drainSpool(spool); + + expect(deliver).toHaveBeenCalledOnce(); + }); + + it.each(["SmsSid", "SmsMessageSid"])("accepts the legacy %s event id alias", async (key) => { + const stateDir = await createStateDir(); + const deliver = vi.fn(async (_message, lifecycle) => { + await lifecycle.onAdopted(); + }); + const spool = createSmsIngressSpool({ + cfg: {}, + account, + channelRuntime: {} as SmsChannelRuntime, + queue: createQueue(stateDir), + deliver, + }); + disposers.push(spool.dispose); + const rawForm = form("SM-alias"); + delete rawForm.MessageSid; + rawForm[key] = "SM-alias"; + + expect(await spool.enqueue(rawForm)).toMatchObject({ kind: "accepted", duplicate: false }); + await drainSpool(spool); + + expect(deliver).toHaveBeenCalledWith( + expect.objectContaining({ messageSid: "SM-alias" }), + expect.any(Object), + expect.any(Number), + ); + }); + + it("uses the canonical sender as the durable lane", async () => { + const stateDir = await createStateDir(); + const queue = createQueue(stateDir); + const spool = createSmsIngressSpool({ + cfg: {}, + account, + channelRuntime: {} as SmsChannelRuntime, + queue, + deliver: vi.fn(async () => undefined), + }); + disposers.push(spool.dispose); + + await spool.enqueue({ ...form("SM-canonical-lane"), From: "RcS:+1 (555) 123-4567" }); + + expect(await queue.listPending()).toEqual([ + expect.objectContaining({ laneKey: "sender:+15551234567" }), + ]); + }); + + it("replays with the original webhook receipt timestamp", async () => { + const stateDir = await createStateDir(); + const receivedAt = 1_700_000_000_456; + const now = vi + .spyOn(Date, "now") + .mockReturnValueOnce(receivedAt) + .mockReturnValue(receivedAt + 60_000); + const first = createSmsIngressSpool({ + cfg: {}, + account, + channelRuntime: {} as SmsChannelRuntime, + queue: createQueue(stateDir), + deliver: vi.fn(async () => undefined), + }); + disposers.push(first.dispose); + await first.enqueue(form("SM-received-at")); + first.dispose(); + now.mockRestore(); + + const deliver = vi.fn(async (_message, lifecycle) => { + await lifecycle.onAdopted(); + }); + const recovered = createSmsIngressSpool({ + cfg: {}, + account, + channelRuntime: {} as SmsChannelRuntime, + queue: createQueue(stateDir), + deliver, + }); + disposers.push(recovered.dispose); + await drainSpool(recovered); + + expect(deliver).toHaveBeenCalledWith(expect.any(Object), expect.any(Object), receivedAt); + }); + + it("preserves the old handler-reload replay guard scenario with a tombstone", async () => { + const stateDir = await createStateDir(); + const firstDeliver = vi.fn(async (_message, lifecycle) => { + await lifecycle.onAdopted(); + }); + const first = createSmsIngressSpool({ + cfg: {}, + account, + channelRuntime: {} as SmsChannelRuntime, + queue: createQueue(stateDir), + deliver: firstDeliver, + }); + disposers.push(first.dispose); + await first.enqueue(form("SM-handler-reload")); + await drainSpool(first); + first.dispose(); + + const reloadedDeliver = vi.fn(async () => undefined); + const reloaded = createSmsIngressSpool({ + cfg: {}, + account, + channelRuntime: {} as SmsChannelRuntime, + queue: createQueue(stateDir), + deliver: reloadedDeliver, + }); + disposers.push(reloaded.dispose); + expect(await reloaded.enqueue(form("SM-handler-reload"))).toMatchObject({ + kind: "completed", + duplicate: true, + }); + await drainSpool(reloaded); + + expect(firstDeliver).toHaveBeenCalledOnce(); + expect(reloadedDeliver).not.toHaveBeenCalled(); + }); + + it.each([ + ["invalid payload", { MessageSid: "SM-invalid", From: "+15551234567" }], + ["account mismatch", { ...form("SM-account"), AccountSid: "AC-other" }], + ])("dead-letters a permanent %s failure", async (_label, rawForm) => { + const stateDir = await createStateDir(); + const deliver = vi.fn(async () => undefined); + const spool = createSmsIngressSpool({ + cfg: {}, + account, + channelRuntime: {} as SmsChannelRuntime, + queue: createQueue(stateDir), + deliver, + }); + disposers.push(spool.dispose); + + await spool.enqueue(rawForm); + await drainSpool(spool); + + expect(await spool.enqueue(rawForm)).toMatchObject({ kind: "failed", duplicate: true }); + expect(deliver).not.toHaveBeenCalled(); + }); +}); diff --git a/extensions/sms/src/ingress-spool.ts b/extensions/sms/src/ingress-spool.ts new file mode 100644 index 000000000000..d52148cc570b --- /dev/null +++ b/extensions/sms/src/ingress-spool.ts @@ -0,0 +1,129 @@ +// Sms plugin module owns durable Twilio webhook admission and replay. +import { + bindIngressLifecycleToReplyOptions, + createChannelIngressDrain, + type ChannelIngressQueue, +} from "openclaw/plugin-sdk/channel-outbound"; +import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; +import { dispatchSmsInboundEvent, type SmsChannelRuntime } from "./inbound.js"; +import { getSmsRuntime } from "./runtime.js"; +import { + buildTwilioInboundMessage, + resolveTwilioInboundSender, + resolveTwilioMessageSid, +} from "./twilio.js"; +import type { ResolvedSmsAccount, SmsInboundMessage } from "./types.js"; + +const SMS_INGRESS_PAYLOAD_VERSION = 1; +// Tombstones dominate the retired 10-minute / 10,000-key replay cache. +const SMS_COMPLETED_TTL_MS = 24 * 60 * 60 * 1000; +const SMS_COMPLETED_MAX_ENTRIES = 20_000; +const SMS_FAILED_TTL_MS = 30 * 24 * 60 * 60 * 1000; +const SMS_FAILED_MAX_ENTRIES = 1_000; + +type SmsIngressPayload = { + version: typeof SMS_INGRESS_PAYLOAD_VERSION; + form: Record; +}; + +type SmsIngressLifecycle = ReturnType< + typeof bindIngressLifecycleToReplyOptions +>["turnAdoptionLifecycle"]; + +class SmsIngressPermanentError extends Error {} + +function parseSmsIngressPayload( + payload: SmsIngressPayload, + account: ResolvedSmsAccount, +): SmsInboundMessage { + if (payload.version !== SMS_INGRESS_PAYLOAD_VERSION) { + throw new SmsIngressPermanentError("SMS ingress payload version is invalid."); + } + const message = buildTwilioInboundMessage(payload.form); + if (!message) { + throw new SmsIngressPermanentError("SMS ingress payload is invalid."); + } + if (message.accountSid && message.accountSid !== account.accountSid) { + throw new SmsIngressPermanentError("SMS ingress payload has an invalid Twilio account."); + } + return message; +} + +export function createSmsIngressSpool(params: { + cfg: OpenClawConfig; + account: ResolvedSmsAccount; + channelRuntime: SmsChannelRuntime; + queue?: ChannelIngressQueue; + abortSignal?: AbortSignal; + log?: { info?: (message: string) => void; warn?: (message: string) => void }; + deliver?: ( + message: SmsInboundMessage, + lifecycle: SmsIngressLifecycle, + receivedAt: number, + ) => Promise; +}) { + const queue = + params.queue ?? + getSmsRuntime().state.openChannelIngressQueue({ + accountId: params.account.accountId, + }); + const deliver = + params.deliver ?? + (async (message: SmsInboundMessage, lifecycle: SmsIngressLifecycle, receivedAt: number) => { + await dispatchSmsInboundEvent({ + cfg: params.cfg, + account: params.account, + channelRuntime: params.channelRuntime, + msg: message, + receivedAt, + turnAdoptionLifecycle: lifecycle, + log: params.log, + }); + }); + const drain = createChannelIngressDrain({ + queue, + ...(params.abortSignal ? { abortSignal: params.abortSignal } : {}), + ...(params.log?.warn ? { onLog: (message: string) => params.log?.warn?.(message) } : {}), + resolveNonRetryableFailure: (error) => + error instanceof SmsIngressPermanentError + ? { reason: "invalid-payload", message: error.message } + : null, + dispatchClaimedEvent: async (event, lifecycle) => { + await deliver( + parseSmsIngressPayload(event.payload, params.account), + bindIngressLifecycleToReplyOptions(lifecycle).turnAdoptionLifecycle, + event.receivedAt, + ); + }, + }); + return { + enqueue: async (form: Record) => { + const receivedAt = Date.now(); + const eventId = resolveTwilioMessageSid(form); + if (!eventId) { + throw new Error("SMS webhook is missing MessageSid."); + } + const sender = resolveTwilioInboundSender(form); + await queue.prune({ + completedTtlMs: SMS_COMPLETED_TTL_MS, + completedMaxEntries: SMS_COMPLETED_MAX_ENTRIES, + failedTtlMs: SMS_FAILED_TTL_MS, + failedMaxEntries: SMS_FAILED_MAX_ENTRIES, + }); + const result = await queue.enqueue( + eventId, + { version: SMS_INGRESS_PAYLOAD_VERSION, form }, + { + receivedAt, + laneKey: sender ? `sender:${sender}` : `event:${eventId}`, + }, + ); + return { kind: result.kind, duplicate: result.duplicate }; + }, + drainOnce: async () => { + await drain.drainOnce(); + }, + waitForIdle: drain.waitForIdle, + dispose: () => drain.dispose(), + }; +} diff --git a/extensions/sms/src/twilio.ts b/extensions/sms/src/twilio.ts index d7ed89277907..276eb0b339ea 100644 --- a/extensions/sms/src/twilio.ts +++ b/extensions/sms/src/twilio.ts @@ -234,23 +234,32 @@ function parseTwilioInboundFrom(raw: string): string | null { return phoneNumber; } +export function resolveTwilioInboundSender(form: Record): string { + return parseTwilioInboundFrom(firstTrimmedString(form.From)) ?? ""; +} + export function buildTwilioInboundMessage(form: Record): SmsInboundMessage | null { // Signature verification owns the untouched form. Canonicalize only after // that boundary so Twilio channel prefixes never change its signed input. - const from = parseTwilioInboundFrom(firstTrimmedString(form.From)); + const from = resolveTwilioInboundSender(form); const to = firstTrimmedString(form.To); const body = firstString(form.Body); const accountSid = firstTrimmedString(form.AccountSid); - const messageSid = - firstTrimmedString(form.MessageSid) || - firstTrimmedString(form.SmsSid) || - firstTrimmedString(form.SmsMessageSid); + const messageSid = resolveTwilioMessageSid(form); if (!from || !to || !body || !messageSid) { return null; } return { accountSid, from, to, body, messageSid }; } +export function resolveTwilioMessageSid(form: Record): string { + return ( + firstTrimmedString(form.MessageSid) || + firstTrimmedString(form.SmsSid) || + firstTrimmedString(form.SmsMessageSid) + ); +} + export async function readTwilioWebhookForm(req: IncomingMessage): Promise> { const body = await readRequestBodyWithLimit(req, { maxBytes: WEBHOOK_BODY_LIMIT_BYTES, diff --git a/extensions/sms/src/webhook-replay-guard.test.ts b/extensions/sms/src/webhook-replay-guard.test.ts deleted file mode 100644 index 674c5d51650d..000000000000 --- a/extensions/sms/src/webhook-replay-guard.test.ts +++ /dev/null @@ -1,51 +0,0 @@ -import { describe, expect, it } from "vitest"; -import { createSmsWebhookReplayGuard } from "./webhook-replay-guard.js"; - -describe("createSmsWebhookReplayGuard", () => { - it("prunes only the expired insertion prefix without refreshing replays", () => { - let nowMs = 0; - const replayGuard = createSmsWebhookReplayGuard({ - ttlMs: 10, - maxKeys: 2, - now: () => nowMs, - }); - - expect(replayGuard.remember("first")).toEqual({ kind: "accepted" }); - nowMs = 2; - expect(replayGuard.remember("second")).toEqual({ kind: "accepted" }); - nowMs = 5; - expect(replayGuard.remember("first")).toEqual({ kind: "replayed" }); - expect(replayGuard.remember("overflow")).toEqual({ - kind: "saturated", - retryAfterMs: 5, - }); - - nowMs = 10; - expect(replayGuard.remember("overflow")).toEqual({ kind: "accepted" }); - expect(replayGuard.remember("second")).toEqual({ kind: "replayed" }); - }); - - it("keeps live replay keys and fails closed until capacity expires", () => { - let nowMs = 1_000; - const replayGuard = createSmsWebhookReplayGuard({ - ttlMs: 10_000, - maxKeys: 2, - now: () => nowMs, - }); - - expect(replayGuard.remember("first")).toEqual({ kind: "accepted" }); - expect(replayGuard.remember("second")).toEqual({ kind: "accepted" }); - expect(replayGuard.remember("overflow")).toEqual({ - kind: "saturated", - retryAfterMs: 10_000, - }); - expect(replayGuard.remember("overflow")).toEqual({ - kind: "saturated", - retryAfterMs: 10_000, - }); - expect(replayGuard.remember("first")).toEqual({ kind: "replayed" }); - - nowMs += 10_000; - expect(replayGuard.remember("overflow")).toEqual({ kind: "accepted" }); - }); -}); diff --git a/extensions/sms/src/webhook-replay-guard.ts b/extensions/sms/src/webhook-replay-guard.ts deleted file mode 100644 index b68b7b736581..000000000000 --- a/extensions/sms/src/webhook-replay-guard.ts +++ /dev/null @@ -1,56 +0,0 @@ -import { performance } from "node:perf_hooks"; - -const REPLAY_CACHE_TTL_MS = 10 * 60_000; -const REPLAY_CACHE_MAX_KEYS = 10_000; - -type ReplayCacheDecision = - | { kind: "accepted" } - | { kind: "replayed" } - | { kind: "saturated"; retryAfterMs: number }; - -export type SmsWebhookReplayGuard = { - remember: (messageSid: string) => ReplayCacheDecision; -}; - -export function createSmsWebhookReplayGuard( - options: { - ttlMs?: number; - maxKeys?: number; - now?: () => number; - } = {}, -): SmsWebhookReplayGuard { - const ttlMs = options.ttlMs ?? REPLAY_CACHE_TTL_MS; - const maxKeys = options.maxKeys ?? REPLAY_CACHE_MAX_KEYS; - const now = options.now ?? (() => performance.now()); - const entries = new Map(); - - const pruneExpired = (nowMs: number) => { - // Fixed TTLs on a monotonic clock expire in insertion order, so only inspect - // the expired prefix. Full live caches stay O(1) instead of rescanning 10k keys. - for (const [key, expiresAt] of entries) { - if (expiresAt > nowMs) { - break; - } - entries.delete(key); - } - }; - - return { - remember: (messageSid) => { - const nowMs = now(); - pruneExpired(nowMs); - if (entries.has(messageSid)) { - return { kind: "replayed" }; - } - if (entries.size >= maxKeys) { - const oldestExpiresAt = entries.values().next().value ?? nowMs; - return { - kind: "saturated", - retryAfterMs: Math.max(0, oldestExpiresAt - nowMs), - }; - } - entries.set(messageSid, nowMs + ttlMs); - return { kind: "accepted" }; - }, - }; -} diff --git a/extensions/sms/src/webhook.test.ts b/extensions/sms/src/webhook.test.ts index 3cc9e72c4555..7ebc2b79da91 100644 --- a/extensions/sms/src/webhook.test.ts +++ b/extensions/sms/src/webhook.test.ts @@ -3,11 +3,13 @@ import { createHmac } from "node:crypto"; import type { IncomingMessage, ServerResponse } from "node:http"; import { Readable } from "node:stream"; import { beforeEach, describe, expect, it, vi } from "vitest"; -import type { SmsChannelRuntime } from "./inbound.js"; import type { ResolvedSmsAccount } from "./types.js"; import { createSmsWebhookHandler } from "./webhook.js"; -const dispatchSmsInboundEvent = vi.hoisted(() => vi.fn(async () => undefined)); +const enqueueSmsIngress = vi.hoisted(() => + vi.fn(async () => ({ kind: "accepted" as const, duplicate: false })), +); +const drainSmsIngress = vi.hoisted(() => vi.fn(async () => undefined)); const runDetachedWebhookWork = vi.hoisted(() => vi.fn((run: () => Promise) => run())); vi.mock("openclaw/plugin-sdk/webhook-request-guards", async (importOriginal) => { @@ -16,13 +18,16 @@ vi.mock("openclaw/plugin-sdk/webhook-request-guards", async (importOriginal) => return { ...actual, runDetachedWebhookWork }; }); -vi.mock("./inbound.js", () => ({ - dispatchSmsInboundEvent, -})); - let testAccountSequence = 0; let activeAccountId = "test-0"; +function createIngress() { + return { + enqueue: enqueueSmsIngress, + drainOnce: drainSmsIngress, + }; +} + function parseTestTwilioForm(body: string): Record { return Object.fromEntries(new URLSearchParams(body)); } @@ -100,18 +105,21 @@ function createRequest( type TestResponse = ServerResponse & { body?: string; setHeaderMock: ReturnType; + endMock: ReturnType; }; function createResponse(): TestResponse { const setHeaderMock = vi.fn(); + const endMock = vi.fn(function (this: ServerResponse & { body?: string }, body?: string) { + this.body = body; + return this; + }); return { statusCode: 200, setHeader: setHeaderMock, setHeaderMock, - end: vi.fn(function (this: ServerResponse & { body?: string }, body?: string) { - this.body = body; - return this; - }), + end: endMock, + endMock, } as unknown as TestResponse; } @@ -142,39 +150,92 @@ function createMessageSid(index: number): string { describe("createSmsWebhookHandler", () => { beforeEach(() => { - dispatchSmsInboundEvent.mockClear(); + enqueueSmsIngress.mockReset(); + enqueueSmsIngress.mockResolvedValue({ kind: "accepted", duplicate: false }); + drainSmsIngress.mockClear(); runDetachedWebhookWork.mockClear(); activeAccountId = `test-${++testAccountSequence}`; }); - it("validates a fragmentless signature and preserves dedupe across handler reloads", async () => { + it("validates a fragmentless signature before enqueuing the raw Twilio form", async () => { const { body, signature } = createSignedSmsPayload(createMessageSid(1)); const handler = createSmsWebhookHandler({ cfg: {}, account: createAccount({ publicWebhookUrl: "https://gateway.example.com/webhooks/sms#rp=4xx", }), - channelRuntime: {} as SmsChannelRuntime, + ingress: createIngress(), }); - const firstRes = createResponse(); - await handler(createRequest(body, signature), firstRes); - const replayRes = createResponse(); - const reloadedHandler = createSmsWebhookHandler({ - cfg: {}, - account: createAccount({ - publicWebhookUrl: "https://gateway.example.com/webhooks/sms#rp=4xx", - }), - channelRuntime: {} as SmsChannelRuntime, - }); - await reloadedHandler(createRequest(body, signature), replayRes); + const res = createResponse(); + await handler(createRequest(body, signature), res); - expect(firstRes.statusCode).toBe(200); - expect(replayRes.statusCode).toBe(200); - expect(dispatchSmsInboundEvent).toHaveBeenCalledTimes(1); + expect(res.statusCode).toBe(200); + expect(enqueueSmsIngress).toHaveBeenCalledWith(parseTestTwilioForm(body)); expect(runDetachedWebhookWork).toHaveBeenCalledTimes(1); }); + it("does not acknowledge when the durable enqueue fails", async () => { + const { body, signature } = createSignedSmsPayload(createMessageSid(2)); + enqueueSmsIngress.mockRejectedValueOnce(new Error("sqlite unavailable")); + const handler = createSmsWebhookHandler({ + cfg: {}, + account: createAccount(), + ingress: createIngress(), + }); + const res = createResponse(); + + await expect(handler(createRequest(body, signature), res)).rejects.toThrow( + "sqlite unavailable", + ); + + expect(res.endMock).not.toHaveBeenCalled(); + expect(runDetachedWebhookWork).not.toHaveBeenCalled(); + }); + + it("rejects a signed webhook without a stable MessageSid", async () => { + const body = "AccountSid=AC123&From=%2B15551234567&To=%2B15557654321&Body=hello"; + const signature = computeTestTwilioSignature({ + url: "https://gateway.example.com/webhooks/sms", + authToken: "secret", + form: parseTestTwilioForm(body), + }); + const handler = createSmsWebhookHandler({ + cfg: {}, + account: createAccount(), + ingress: createIngress(), + }); + const res = createResponse(); + + await handler(createRequest(body, signature), res); + + expect(res.statusCode).toBe(400); + expect(enqueueSmsIngress).not.toHaveBeenCalled(); + }); + + it("accepts the legacy SmsMessageSid event id alias", async () => { + const body = + "AccountSid=AC123&From=%2B15551234567&To=%2B15557654321&Body=hello&SmsMessageSid=SM-alias"; + const signature = computeTestTwilioSignature({ + url: "https://gateway.example.com/webhooks/sms", + authToken: "secret", + form: parseTestTwilioForm(body), + }); + const handler = createSmsWebhookHandler({ + cfg: {}, + account: createAccount(), + ingress: createIngress(), + }); + const res = createResponse(); + + await handler(createRequest(body, signature), res); + + expect(res.statusCode).toBe(200); + expect(enqueueSmsIngress).toHaveBeenCalledWith( + expect.objectContaining({ SmsMessageSid: "SM-alias" }), + ); + }); + it("validates the raw RCS form before canonicalizing its sender", async () => { const messageSid = createMessageSid(9); const { body, signature } = createSignedSmsPayload(messageSid, { @@ -184,7 +245,7 @@ describe("createSmsWebhookHandler", () => { const handler = createSmsWebhookHandler({ cfg: {}, account: createAccount(), - channelRuntime: {} as SmsChannelRuntime, + ingress: createIngress(), }); expect(parseTestTwilioForm(body).From).toBe("RcS:+1 (555) 123-4567"); @@ -193,21 +254,19 @@ describe("createSmsWebhookHandler", () => { await handler(createRequest(body, signature), res); expect(res.statusCode).toBe(200); - expect(dispatchSmsInboundEvent).toHaveBeenCalledWith( + expect(enqueueSmsIngress).toHaveBeenCalledWith( expect.objectContaining({ - msg: { - accountSid: "AC123", - from: "+15551234567", - to: "rcs:example-agent", - body: "hello", - messageSid, - }, + AccountSid: "AC123", + From: "RcS:+1 (555) 123-4567", + To: "rcs:example-agent", + Body: "hello", + MessageSid: messageSid, }), ); }); - it("rejects signed webhooks for a different Twilio account", async () => { - const body = `AccountSid=AC-other&From=%2B15551234567&To=%2B15557654321&Body=hello&SmsMessageSid=${createMessageSid(8)}`; + it("durably accepts a signed account mismatch for non-retryable drain classification", async () => { + const body = `AccountSid=AC-other&From=%2B15551234567&To=%2B15557654321&Body=hello&MessageSid=${createMessageSid(8)}`; const signature = computeTestTwilioSignature({ url: "https://gateway.example.com/webhooks/sms", authToken: "secret", @@ -216,14 +275,16 @@ describe("createSmsWebhookHandler", () => { const handler = createSmsWebhookHandler({ cfg: {}, account: createAccount(), - channelRuntime: {} as SmsChannelRuntime, + ingress: createIngress(), }); const res = createResponse(); await handler(createRequest(body, signature), res); - expect(res.statusCode).toBe(403); - expect(dispatchSmsInboundEvent).not.toHaveBeenCalled(); + expect(res.statusCode).toBe(200); + expect(enqueueSmsIngress).toHaveBeenCalledWith( + expect.objectContaining({ AccountSid: "AC-other" }), + ); }); it("does not let unsigned proxy traffic consume the same client's signed webhook rate limit", async () => { @@ -231,7 +292,7 @@ describe("createSmsWebhookHandler", () => { const handler = createSmsWebhookHandler({ cfg: { gateway: { trustedProxies: ["127.0.0.1"] } }, account, - channelRuntime: {} as SmsChannelRuntime, + ingress: createIngress(), }); const unsignedBody = "AccountSid=AC123&From=%2B15550000000&To=%2B15557654321&Body=bad&MessageSid=SM-bad"; @@ -264,7 +325,7 @@ describe("createSmsWebhookHandler", () => { ); expect(accepted.statusCode).toBe(200); - expect(dispatchSmsInboundEvent).toHaveBeenCalledTimes(1); + expect(enqueueSmsIngress).toHaveBeenCalledTimes(1); }); it("scopes signed webhook rate limits to one SMS account and route", async () => { @@ -278,12 +339,12 @@ describe("createSmsWebhookHandler", () => { const supportHandler = createSmsWebhookHandler({ cfg: {}, account: supportAccount, - channelRuntime: {} as SmsChannelRuntime, + ingress: createIngress(), }); const defaultHandler = createSmsWebhookHandler({ cfg: {}, account: defaultAccount, - channelRuntime: {} as SmsChannelRuntime, + ingress: createIngress(), }); for (let i = 0; i < 30; i += 1) { @@ -318,7 +379,7 @@ describe("createSmsWebhookHandler", () => { const handler = createSmsWebhookHandler({ cfg: { gateway: { trustedProxies: ["127.0.0.1"] } }, account, - channelRuntime: {} as SmsChannelRuntime, + ingress: createIngress(), }); for (let i = 0; i < 30; i += 1) { @@ -349,6 +410,6 @@ describe("createSmsWebhookHandler", () => { ); expect(overBudgetRes.statusCode).toBe(429); - expect(dispatchSmsInboundEvent).toHaveBeenCalledTimes(30); + expect(enqueueSmsIngress).toHaveBeenCalledTimes(30); }); }); diff --git a/extensions/sms/src/webhook.ts b/extensions/sms/src/webhook.ts index 951ef475612b..75c5d4408775 100644 --- a/extensions/sms/src/webhook.ts +++ b/extensions/sms/src/webhook.ts @@ -6,16 +6,14 @@ import { resolveRequestClientIp, } from "openclaw/plugin-sdk/webhook-ingress"; import { runDetachedWebhookWork } from "openclaw/plugin-sdk/webhook-request-guards"; -import { dispatchSmsInboundEvent, type SmsChannelRuntime } from "./inbound.js"; import { - buildTwilioInboundMessage, readTwilioWebhookForm, respondTwiml, + resolveTwilioMessageSid, resolveTwilioWebhookSignatureUrl, verifyTwilioSignature, } from "./twilio.js"; import type { ResolvedSmsAccount } from "./types.js"; -import { createSmsWebhookReplayGuard, type SmsWebhookReplayGuard } from "./webhook-replay-guard.js"; const INVALID_REQUEST_MAX_REQUESTS = 300; const CALLBACK_DISPATCH_MAX_REQUESTS = 30; @@ -33,20 +31,6 @@ const callbackDispatchRateLimiter = createFixedWindowRateLimiter({ windowMs: 60_000, maxTrackedKeys: 5_000, }); -const replayGuardsByAccount = new Map(); - -function resolveSmsWebhookReplayGuard(account: ResolvedSmsAccount): SmsWebhookReplayGuard { - // Config reloads replace route handlers. Keep the guard with the Twilio account - // identity so retries cannot cross that lifecycle boundary or block sibling accounts. - const key = `${account.accountId}\0${account.accountSid}`; - const existing = replayGuardsByAccount.get(key); - if (existing) { - return existing; - } - const created = createSmsWebhookReplayGuard(); - replayGuardsByAccount.set(key, created); - return created; -} type SmsWebhookLog = { info?: (message: string) => void; @@ -57,7 +41,10 @@ type SmsWebhookLog = { export type SmsWebhookHandlerParams = { cfg: OpenClawConfig; account: ResolvedSmsAccount; - channelRuntime: SmsChannelRuntime; + ingress: { + enqueue: (form: Record) => Promise<{ duplicate: boolean }>; + drainOnce: () => Promise; + }; log?: SmsWebhookLog; }; @@ -94,9 +81,8 @@ function rejectInvalidRequestRateLimit(params: { return true; } -// Each account route owns its guard so one saturated account cannot block sibling accounts. +// Each account route owns one durable ingress adapter. export function createSmsWebhookHandler(params: SmsWebhookHandlerParams) { - const webhookReplayGuard = resolveSmsWebhookReplayGuard(params.account); return async (req: IncomingMessage, res: ServerResponse) => { if (req.method !== "POST") { respondTwiml(res, 405, "Method not allowed"); @@ -138,22 +124,6 @@ export function createSmsWebhookHandler(params: SmsWebhookHandlerParams) { } } - const msg = buildTwilioInboundMessage(form); - if (!msg) { - if (invalidRequestRateLimited) { - return rejectInvalidRequestRateLimit({ key, log: params.log, res }); - } - respondTwiml(res, 400, "Missing SMS payload"); - return true; - } - if (msg.accountSid && msg.accountSid !== params.account.accountSid) { - if (invalidRequestRateLimited) { - return rejectInvalidRequestRateLimit({ key, log: params.log, res }); - } - params.log?.warn?.("SMS webhook rejected mismatched Twilio AccountSid"); - respondTwiml(res, 403, "Invalid account"); - return true; - } if (invalidRequestRateLimited && params.account.dangerouslyDisableSignatureValidation) { return rejectInvalidRequestRateLimit({ key, log: params.log, res }); } @@ -162,36 +132,23 @@ export function createSmsWebhookHandler(params: SmsWebhookHandlerParams) { respondTwiml(res, 429, "Rate limit exceeded"); return true; } - const replayDecision = webhookReplayGuard.remember(msg.messageSid); - if (replayDecision.kind === "replayed") { - params.log?.warn?.(`SMS webhook ignored replayed message ${msg.messageSid}`); - respondTwiml(res, 200); + const messageSid = resolveTwilioMessageSid(form); + if (!messageSid) { + respondTwiml(res, 400, "Missing MessageSid"); return true; } - if (replayDecision.kind === "saturated") { - const retryAfterSeconds = Math.max(1, Math.ceil(replayDecision.retryAfterMs / 1000)); - params.log?.warn?.("SMS webhook replay cache is full of unexpired message SIDs"); - res.setHeader("Retry-After", String(retryAfterSeconds)); - respondTwiml(res, 429, "Replay cache saturated"); - return true; + // Signature validation owns the parsed-but-otherwise-raw Twilio form. + // A 200 is impossible until SQLite commits this exact transport envelope. + const verdict = await params.ingress.enqueue(form); + if (verdict.duplicate) { + params.log?.warn?.(`SMS webhook ignored replayed message ${messageSid}`); } - - // Reserve the detached task before the HTTP admission is released; - // otherwise later queue work inherits a released admission root. - void runDetachedWebhookWork(() => - dispatchSmsInboundEvent({ - cfg: params.cfg, - account: params.account, - msg, - channelRuntime: params.channelRuntime, - log: params.log, - }), - ).catch((err: unknown) => { + // Reserve detached work under HTTP admission; it only pumps the durable drain. + void runDetachedWebhookWork(() => params.ingress.drainOnce()).catch((err: unknown) => { params.log?.error?.( - `SMS webhook dispatch failed: ${err instanceof Error ? err.message : String(err)}`, + `SMS ingress drain failed: ${err instanceof Error ? err.message : String(err)}`, ); }); - respondTwiml(res, 200); return true; };