From 8d5f7e27b2ffd233e262fbe4e00f09bc201ce068 Mon Sep 17 00:00:00 2001 From: xingzhou Date: Sat, 11 Jul 2026 00:58:53 +0800 Subject: [PATCH] fix(sms): replayed Twilio webhooks process again after high inbound traffic (#101107) * fix(sms): replayed Twilio webhooks process again after high inbound traffic * fix(sms): harden Twilio replay saturation * test(sms): expose response header spy --------- Co-authored-by: Peter Steinberger --- docs/channels/sms.md | 5 +- extensions/sms/src/twilio.test.ts | 12 +-- extensions/sms/src/twilio.ts | 20 ++--- extensions/sms/src/webhook.test.ts | 139 ++++++++++++++++++++++++----- extensions/sms/src/webhook.ts | 118 +++++++++++++++++------- 5 files changed, 224 insertions(+), 70 deletions(-) diff --git a/docs/channels/sms.md b/docs/channels/sms.md index cec10b8c015a..6efff7991a2b 100644 --- a/docs/channels/sms.md +++ b/docs/channels/sms.md @@ -343,7 +343,7 @@ The first message should create a pairing request. The second message should rec ## Webhook security -By default, OpenClaw validates `X-Twilio-Signature` using `publicWebhookUrl` and `authToken`. Keep `publicWebhookUrl` byte-for-byte aligned with the URL configured in Twilio, including scheme, host, path, and query string. +By default, OpenClaw validates `X-Twilio-Signature` using `publicWebhookUrl` and `authToken`. Keep the endpoint portion of `publicWebhookUrl` byte-for-byte aligned with the URL configured in Twilio, including scheme, host, path, and query string. OpenClaw excludes Twilio [connection-override](https://www.twilio.com/docs/usage/webhooks/webhooks-connection-overrides) fragments (`#...`) from signature computation, as Twilio requires. The webhook route also enforces, independent of signature validation: @@ -351,8 +351,11 @@ The webhook route also enforces, independent of signature validation: - Rate limit of 30 requests per minute per source IP (HTTP 429 above that). - The payload `AccountSid` must match the configured `accountSid` (HTTP 403 otherwise). - Replayed `MessageSid` values are deduplicated for 10 minutes. +- Each SMS account's replay cache retains up to 10,000 live message SIDs. When every slot is live, new webhooks for that account fail closed with HTTP 429 and a `Retry-After` header until the oldest slot expires. - Request bodies over 32 KB are rejected. +Twilio does not retry HTTP 429 by default or document support for `Retry-After`. The `#rp=4xx` and `#rp=all` connection overrides opt into 4xx retries, but Twilio caps the complete retry transaction at 15 seconds, so retries can still finish before a replay-cache slot expires. Configure a fallback URL when another handler must receive failed deliveries; treat a 429 as a fail-closed rejection, not reliable backpressure. + For local tunnel testing only, you can set: ```json5 diff --git a/extensions/sms/src/twilio.test.ts b/extensions/sms/src/twilio.test.ts index b6b2b0632019..d3517b13fe57 100644 --- a/extensions/sms/src/twilio.test.ts +++ b/extensions/sms/src/twilio.test.ts @@ -639,29 +639,29 @@ describe("Twilio SMS helpers", () => { ).rejects.toThrow("Twilio SMS send response did not include a Message SID."); }); - it("preserves the configured public webhook path when adding a request query", () => { + it("excludes a connection override fragment when adding a request query", () => { expect( resolveTwilioWebhookSignatureUrl({ req: { url: "/webhooks/sms?foo=bar" } as never, - publicWebhookUrl: "https://gateway.example.com/base", + publicWebhookUrl: "https://gateway.example.com/base#rp=4xx", }), ).toBe("https://gateway.example.com/base?foo=bar"); }); - it("keeps an explicit configured public webhook query", () => { + it("keeps an explicit configured query but excludes its connection override fragment", () => { expect( resolveTwilioWebhookSignatureUrl({ req: { url: "/webhooks/sms?foo=request" } as never, - publicWebhookUrl: "https://gateway.example.com/base?foo=configured", + publicWebhookUrl: "https://gateway.example.com/base?foo=configured#rp=all", }), ).toBe("https://gateway.example.com/base?foo=configured"); }); - it("does not reserialize the configured public webhook URL", () => { + it("strips a connection override fragment without reserializing the configured URL", () => { expect( resolveTwilioWebhookSignatureUrl({ req: { url: "/webhooks/sms" } as never, - publicWebhookUrl: "https://gateway.example.com:443/webhooks/sms", + publicWebhookUrl: "https://gateway.example.com:443/webhooks/sms#rp=4xx", }), ).toBe("https://gateway.example.com:443/webhooks/sms"); }); diff --git a/extensions/sms/src/twilio.ts b/extensions/sms/src/twilio.ts index 4fe2b7bf3871..7658f7978443 100644 --- a/extensions/sms/src/twilio.ts +++ b/extensions/sms/src/twilio.ts @@ -133,28 +133,26 @@ function requestSearch(req: IncomingMessage): string { } } -function configuredUrlHasQuery(url: string): boolean { +function stripUrlFragment(url: string): string { const hashIndex = url.indexOf("#"); - const beforeHash = hashIndex === -1 ? url : url.slice(0, hashIndex); - return beforeHash.includes("?"); + return hashIndex === -1 ? url : url.slice(0, hashIndex); } export function resolveTwilioWebhookSignatureUrl(params: { req: IncomingMessage; publicWebhookUrl: string; }): string { - if (configuredUrlHasQuery(params.publicWebhookUrl)) { - return params.publicWebhookUrl; + // Twilio connection overrides live in the fragment but are excluded from its + // signature input. Strip without URL reserialization so exact port/path bytes survive. + const signatureBaseUrl = stripUrlFragment(params.publicWebhookUrl); + if (signatureBaseUrl.includes("?")) { + return signatureBaseUrl; } const search = requestSearch(params.req); if (!search) { - return params.publicWebhookUrl; + return signatureBaseUrl; } - const hashIndex = params.publicWebhookUrl.indexOf("#"); - if (hashIndex === -1) { - return `${params.publicWebhookUrl}${search}`; - } - return `${params.publicWebhookUrl.slice(0, hashIndex)}${search}${params.publicWebhookUrl.slice(hashIndex)}`; + return `${signatureBaseUrl}${search}`; } export class TwilioSmsApiError extends Error { diff --git a/extensions/sms/src/webhook.test.ts b/extensions/sms/src/webhook.test.ts index d95c3a0bccc2..25928d6e8db6 100644 --- a/extensions/sms/src/webhook.test.ts +++ b/extensions/sms/src/webhook.test.ts @@ -5,7 +5,11 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import type { SmsChannelRuntime } from "./inbound.js"; import { computeTwilioSignature, parseTwilioFormBody } from "./twilio.js"; import type { ResolvedSmsAccount } from "./types.js"; -import { createSmsWebhookHandler, resetSmsWebhookReplayCacheForTest } from "./webhook.js"; +import { + createSmsWebhookHandler, + createSmsWebhookReplayGuard, + resetSmsWebhookReplayGuardsForTest, +} from "./webhook.js"; const dispatchSmsInboundEvent = vi.hoisted(() => vi.fn(async () => undefined)); @@ -13,7 +17,7 @@ vi.mock("./inbound.js", () => ({ dispatchSmsInboundEvent, })); -function createAccount(): ResolvedSmsAccount { +function createAccount(overrides: Partial = {}): ResolvedSmsAccount { return { accountId: "default", enabled: true, @@ -28,63 +32,158 @@ function createAccount(): ResolvedSmsAccount { dmPolicy: "pairing", allowFrom: [], textChunkLimit: 1500, + ...overrides, }; } -function createRequest(body: string, signature: string): IncomingMessage { +function createRequest( + body: string, + signature: string, + remoteAddress = "127.0.0.1", +): IncomingMessage { const req = Readable.from([body]) as IncomingMessage; req.method = "POST"; req.headers = { "x-twilio-signature": signature }; Object.defineProperty(req, "socket", { - value: { remoteAddress: "127.0.0.1" }, + value: { remoteAddress }, }); return req; } -function createResponse(): ServerResponse & { body?: string } { +type TestResponse = ServerResponse & { + body?: string; + setHeaderMock: ReturnType; +}; + +function createResponse(): TestResponse { + const setHeaderMock = vi.fn(); return { statusCode: 200, - setHeader: vi.fn(), + setHeader: setHeaderMock, + setHeaderMock, end: vi.fn(function (this: ServerResponse & { body?: string }, body?: string) { this.body = body; return this; }), - } as unknown as ServerResponse & { body?: string }; + } as unknown as TestResponse; +} + +function createSignedSmsPayload(messageSid: string): { body: string; signature: string } { + const body = `AccountSid=AC123&From=%2B15551234567&To=%2B15557654321&Body=hello&MessageSid=${messageSid}`; + return { + body, + signature: computeTwilioSignature({ + url: "https://gateway.example.com/webhooks/sms", + authToken: "secret", + form: parseTwilioFormBody(body), + }), + }; +} + +function createMessageSid(index: number): string { + return `SM${index.toString(16).padStart(32, "0")}`; } describe("createSmsWebhookHandler", () => { beforeEach(() => { dispatchSmsInboundEvent.mockClear(); - resetSmsWebhookReplayCacheForTest(); + resetSmsWebhookReplayGuardsForTest(); }); - it("dedupes replayed signed Twilio webhooks by message SID", async () => { - const body = - "AccountSid=AC123&From=%2B15551234567&To=%2B15557654321&Body=hello&MessageSid=SM123"; - const signature = computeTwilioSignature({ - url: "https://gateway.example.com/webhooks/sms", - authToken: "secret", - form: parseTwilioFormBody(body), - }); + it("validates a fragmentless signature and preserves dedupe across handler reloads", async () => { + const { body, signature } = createSignedSmsPayload(createMessageSid(1)); const handler = createSmsWebhookHandler({ cfg: {}, - account: createAccount(), + account: createAccount({ + publicWebhookUrl: "https://gateway.example.com/webhooks/sms#rp=4xx", + }), channelRuntime: {} as SmsChannelRuntime, }); const firstRes = createResponse(); await handler(createRequest(body, signature), firstRes); const replayRes = createResponse(); - await handler(createRequest(body, signature), replayRes); + const reloadedHandler = createSmsWebhookHandler({ + cfg: {}, + account: createAccount({ + publicWebhookUrl: "https://gateway.example.com/webhooks/sms#rp=4xx", + }), + channelRuntime: {} as SmsChannelRuntime, + }); + await reloadedHandler(createRequest(body, signature), replayRes); expect(firstRes.statusCode).toBe(200); expect(replayRes.statusCode).toBe(200); expect(dispatchSmsInboundEvent).toHaveBeenCalledTimes(1); }); + it("prunes only the expired insertion prefix without refreshing replays", () => { + let nowMs = 0; + const replayGuard = createSmsWebhookReplayGuard({ + ttlMs: 10, + maxKeys: 2, + now: () => nowMs, + }); + const first = createMessageSid(2); + const second = createMessageSid(3); + const overflow = createMessageSid(4); + + 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", async () => { + let nowMs = 1_000; + const webhookReplayGuard = createSmsWebhookReplayGuard({ + ttlMs: 10_000, + maxKeys: 2, + now: () => nowMs, + }); + const handler = createSmsWebhookHandler( + { + cfg: {}, + account: createAccount(), + channelRuntime: {} as SmsChannelRuntime, + }, + webhookReplayGuard, + ); + const first = createSignedSmsPayload(createMessageSid(5)); + const second = createSignedSmsPayload(createMessageSid(6)); + const overflow = createSignedSmsPayload(createMessageSid(7)); + + await handler(createRequest(first.body, first.signature), createResponse()); + await handler(createRequest(second.body, second.signature), createResponse()); + const overflowRes = createResponse(); + await handler(createRequest(overflow.body, overflow.signature), overflowRes); + const repeatedOverflowRes = createResponse(); + await handler(createRequest(overflow.body, overflow.signature), repeatedOverflowRes); + const firstReplayRes = createResponse(); + await handler(createRequest(first.body, first.signature), firstReplayRes); + + expect(overflowRes.statusCode).toBe(429); + expect(repeatedOverflowRes.statusCode).toBe(429); + expect(overflowRes.setHeaderMock).toHaveBeenCalledWith("Retry-After", "10"); + expect(firstReplayRes.statusCode).toBe(200); + expect(dispatchSmsInboundEvent).toHaveBeenCalledTimes(2); + + nowMs += 10_000; + const afterExpiryRes = createResponse(); + await handler(createRequest(overflow.body, overflow.signature), afterExpiryRes); + + expect(afterExpiryRes.statusCode).toBe(200); + expect(dispatchSmsInboundEvent).toHaveBeenCalledTimes(3); + }); + it("rejects signed webhooks for a different Twilio account", async () => { - const body = - "AccountSid=AC-other&From=%2B15551234567&To=%2B15557654321&Body=hello&SmsMessageSid=SM123"; + const body = `AccountSid=AC-other&From=%2B15551234567&To=%2B15557654321&Body=hello&SmsMessageSid=${createMessageSid(8)}`; const signature = computeTwilioSignature({ url: "https://gateway.example.com/webhooks/sms", authToken: "secret", diff --git a/extensions/sms/src/webhook.ts b/extensions/sms/src/webhook.ts index fd4b73cba8d5..a84f8a61e7b1 100644 --- a/extensions/sms/src/webhook.ts +++ b/extensions/sms/src/webhook.ts @@ -1,5 +1,6 @@ // Sms plugin module implements webhook behavior. import type { IncomingMessage, ServerResponse } from "node:http"; +import { performance } from "node:perf_hooks"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; import { createFixedWindowRateLimiter } from "openclaw/plugin-sdk/webhook-ingress"; import { dispatchSmsInboundEvent, type SmsChannelRuntime } from "./inbound.js"; @@ -19,7 +20,77 @@ const rateLimiter = createFixedWindowRateLimiter({ }); const REPLAY_CACHE_TTL_MS = 10 * 60_000; const REPLAY_CACHE_MAX_KEYS = 10_000; -const replayCache = new Map(); + +type ReplayCacheDecision = + | { kind: "accepted" } + | { kind: "replayed" } + | { kind: "saturated"; retryAfterMs: number }; + +type SmsWebhookReplayGuard = { + remember: (messageSid: string) => ReplayCacheDecision; +}; + +const replayGuardsByAccount = new Map(); + +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" }; + }, + }; +} + +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; +} + +export function resetSmsWebhookReplayGuardsForTest(): void { + replayGuardsByAccount.clear(); +} type SmsWebhookLog = { info?: (message: string) => void; @@ -45,31 +116,11 @@ function rateLimitKey(req: IncomingMessage): string { return req.socket?.remoteAddress ?? "unknown"; } -function rememberWebhookMessage(params: { - accountId: string; - messageSid: string; - now?: number; -}): boolean { - const now = params.now ?? Date.now(); - for (const [key, expiresAt] of replayCache) { - if (expiresAt > now && replayCache.size <= REPLAY_CACHE_MAX_KEYS) { - break; - } - replayCache.delete(key); - } - const key = `${params.accountId}:${params.messageSid}`; - if ((replayCache.get(key) ?? 0) > now) { - return false; - } - replayCache.set(key, now + REPLAY_CACHE_TTL_MS); - return true; -} - -export function resetSmsWebhookReplayCacheForTest(): void { - replayCache.clear(); -} - -export function createSmsWebhookHandler(params: SmsWebhookHandlerParams) { +// Each account route owns its guard so one saturated account cannot block sibling accounts. +export function createSmsWebhookHandler( + params: SmsWebhookHandlerParams, + webhookReplayGuard: SmsWebhookReplayGuard = resolveSmsWebhookReplayGuard(params.account), +) { return async (req: IncomingMessage, res: ServerResponse) => { if (req.method !== "POST") { respondTwiml(res, 405, "Method not allowed"); @@ -118,16 +169,19 @@ export function createSmsWebhookHandler(params: SmsWebhookHandlerParams) { respondTwiml(res, 403, "Invalid account"); return true; } - if ( - !rememberWebhookMessage({ - accountId: params.account.accountId, - messageSid: msg.messageSid, - }) - ) { + const replayDecision = webhookReplayGuard.remember(msg.messageSid); + if (replayDecision.kind === "replayed") { params.log?.warn?.(`SMS webhook ignored replayed message ${msg.messageSid}`); respondTwiml(res, 200); 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; + } void dispatchSmsInboundEvent({ cfg: params.cfg,