fix(sms): isolate validated webhook quotas per sender (#104862)

* fix(sms): isolate validated webhook quotas per sender * fix(sms): isolate validated webhook quotas per sender --------- Co-authored-by: clawSean <260045960+clawSean@users.noreply.github.com> Co-authored-by: Vincent Koc <vincentkoc@ieee.org>
2026-08-12 21:53:00 -06:00 · 2026-07-29 10:23:24 -07:00
parent b44890bf2d
commit 864259486b
3 changed files with 194 additions and 24 deletions
@@ -409,6 +409,146 @@ describe("createSmsWebhookHandler", () => {
    expect(defaultRes.statusCode).toBe(200);
  });

+  it("meters the validated dispatch quota per sender, not per shared egress address", async () => {
+    const warn = vi.fn();
+    const handler = createSmsWebhookHandler({
+      cfg: {},
+      account: createAccount(),
+      ingress: createIngress(),
+      log: { warn },
+    });
+
+    for (let i = 0; i < 30; i += 1) {
+      const { body, signature } = createSignedSmsPayload(createMessageSid(500 + i));
+      const res = createResponse();
+      await handler(createRequest(body, signature, { remoteAddress: "203.0.113.30" }), res);
+      expect(res.statusCode).toBe(200);
+    }
+    // Equivalent Twilio RCS address syntax canonicalizes into the same sender bucket.
+    const overQuota = createSignedSmsPayload(createMessageSid(530), {
+      from: "RCS:+1 (555) 123-4567",
+    });
+    const overQuotaRes = createResponse();
+    await handler(
+      createRequest(overQuota.body, overQuota.signature, { remoteAddress: "203.0.113.30" }),
+      overQuotaRes,
+    );
+    expect(overQuotaRes.statusCode).toBe(429);
+    expect(enqueueSmsIngress).toHaveBeenCalledTimes(30);
+    expect(warn).toHaveBeenCalledWith("SMS webhook callback rate limit exceeded");
+    expect(warn).not.toHaveBeenCalledWith(expect.stringContaining("+15551234567"));
+
+    // Same Twilio egress address, different validated sender: must still dispatch.
+    const otherSender = createSignedSmsPayload(createMessageSid(531), { from: "+15559998888" });
+    const otherSenderRes = createResponse();
+    await handler(
+      createRequest(otherSender.body, otherSender.signature, { remoteAddress: "203.0.113.30" }),
+      otherSenderRes,
+    );
+    expect(otherSenderRes.statusCode).toBe(200);
+    expect(enqueueSmsIngress).toHaveBeenCalledTimes(31);
+
+    // Changing Twilio egress addresses cannot widen the sender-scoped budget.
+    const stillLimited = createSignedSmsPayload(createMessageSid(532));
+    const stillLimitedRes = createResponse();
+    await handler(
+      createRequest(stillLimited.body, stillLimited.signature, { remoteAddress: "203.0.113.31" }),
+      stillLimitedRes,
+    );
+    expect(stillLimitedRes.statusCode).toBe(429);
+    expect(enqueueSmsIngress).toHaveBeenCalledTimes(31);
+  });
+
+  it("bounds aggregate validated callback fan-out across distinct senders", async () => {
+    const handler = createSmsWebhookHandler({
+      cfg: {},
+      account: createAccount(),
+      ingress: createIngress(),
+    });
+
+    for (let i = 0; i < 300; i += 1) {
+      const distinctSender = `+1555${i.toString().padStart(7, "0")}`;
+      const { body, signature } = createSignedSmsPayload(createMessageSid(900 + i), {
+        from: distinctSender,
+      });
+      const res = createResponse();
+      await handler(createRequest(body, signature), res);
+      expect(res.statusCode).toBe(200);
+    }
+
+    const overAggregate = createSignedSmsPayload(createMessageSid(1_200), {
+      from: "+15559999999",
+    });
+    const overAggregateRes = createResponse();
+    await handler(createRequest(overAggregate.body, overAggregate.signature), overAggregateRes);
+
+    expect(overAggregateRes.statusCode).toBe(429);
+    expect(enqueueSmsIngress).toHaveBeenCalledTimes(300);
+  });
+
+  it("restores a rate limited sender after the fixed dispatch window expires", async () => {
+    vi.useFakeTimers({ toFake: ["Date"] });
+    try {
+      const handler = createSmsWebhookHandler({
+        cfg: {},
+        account: createAccount(),
+        ingress: createIngress(),
+      });
+
+      for (let i = 0; i < 30; i += 1) {
+        const { body, signature } = createSignedSmsPayload(createMessageSid(600 + i));
+        await handler(createRequest(body, signature), createResponse());
+      }
+      const throttled = createSignedSmsPayload(createMessageSid(630));
+      const throttledRes = createResponse();
+      await handler(createRequest(throttled.body, throttled.signature), throttledRes);
+      expect(throttledRes.statusCode).toBe(429);
+
+      vi.setSystemTime(Date.now() + 60_001);
+      const recovered = createSignedSmsPayload(createMessageSid(631));
+      const recoveredRes = createResponse();
+      await handler(createRequest(recovered.body, recovered.signature), recoveredRes);
+
+      expect(recoveredRes.statusCode).toBe(200);
+      expect(enqueueSmsIngress).toHaveBeenCalledTimes(31);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it("shares one quota for invalid signed senders without throttling a valid sender", async () => {
+    const handler = createSmsWebhookHandler({
+      cfg: {},
+      account: createAccount(),
+      ingress: createIngress(),
+    });
+
+    for (let i = 0; i < 30; i += 1) {
+      const invalidSender = createSignedSmsPayload(createMessageSid(800 + i), {
+        from: "not-a-phone",
+      });
+      const res = createResponse();
+      await handler(createRequest(invalidSender.body, invalidSender.signature), res);
+      expect(res.statusCode).toBe(200);
+    }
+
+    const invalidOverQuota = createSignedSmsPayload(createMessageSid(830), {
+      from: "still-not-a-phone",
+    });
+    const invalidOverQuotaRes = createResponse();
+    await handler(
+      createRequest(invalidOverQuota.body, invalidOverQuota.signature),
+      invalidOverQuotaRes,
+    );
+    expect(invalidOverQuotaRes.statusCode).toBe(429);
+
+    const validSender = createSignedSmsPayload(createMessageSid(831));
+    const validSenderRes = createResponse();
+    await handler(createRequest(validSender.body, validSender.signature), validSenderRes);
+    expect(validSenderRes.statusCode).toBe(200);
+    expect(enqueueSmsIngress).toHaveBeenCalledTimes(31);
+  });
+
  it("keeps validation-disabled webhook dispatches on the stricter callback budget", async () => {
    const account = createAccount({ dangerouslyDisableSignatureValidation: true });
    const handler = createSmsWebhookHandler({
@@ -418,13 +558,14 @@ describe("createSmsWebhookHandler", () => {
    });

    for (let i = 0; i < 30; i += 1) {
-      const valid = createSignedBody({
-        account,
-        messageSid: `SM-disabled-${i}`,
+      // Rotate From: without signature validation it is unauthenticated input and
+      // must not widen the address-keyed budget.
+      const { body } = createSignedSmsPayload(createMessageSid(700 + i), {
+        from: `+1555000${1000 + i}`,
      });
      const res = createResponse();
      await handler(
-        createRequest(valid.body, "unused-signature", {
+        createRequest(body, "unused-signature", {
          headers: { "x-forwarded-for": "203.0.113.20" },
        }),
        res,
@@ -432,10 +573,7 @@ describe("createSmsWebhookHandler", () => {
      expect(res.statusCode).toBe(200);
    }

-    const overBudget = createSignedBody({
-      account,
-      messageSid: "SM-disabled-over-budget",
-    });
+    const overBudget = createSignedSmsPayload(createMessageSid(760), { from: "+15550009999" });
    const overBudgetRes = createResponse();
    await handler(
      createRequest(overBudget.body, "unused-signature", {
@@ -8,6 +8,7 @@ import {
 import {
  readTwilioWebhookForm,
  respondTwiml,
+  resolveTwilioInboundSender,
  resolveTwilioMessageSid,
  resolveTwilioWebhookSignatureUrl,
  verifyTwilioSignature,
@@ -32,6 +33,12 @@ const callbackDispatchRateLimiter = createFixedWindowRateLimiter({
  windowMs: 60_000,
  maxTrackedKeys: 5_000,
 });
+const VALIDATED_AGGREGATE_MAX_REQUESTS = 300;
+const validatedAggregateRateLimiter = createFixedWindowRateLimiter({
+  maxRequests: VALIDATED_AGGREGATE_MAX_REQUESTS,
+  windowMs: 60_000,
+  maxTrackedKeys: 1_000,
+});

 type SmsWebhookLog = {
  info?: (message: string) => void;
@@ -67,8 +74,12 @@ function resolvedClientAddress(params: { cfg: OpenClawConfig; req: IncomingMessa
  );
 }

-function rateLimitKey(params: { account: ResolvedSmsAccount; clientAddress: string }): string {
-  return `${params.account.accountId}:${params.account.webhookPath}:${params.clientAddress}`;
+function rateLimitKey(params: { account: ResolvedSmsAccount; subject: string }): string {
+  return `${params.account.accountId}:${params.account.webhookPath}:${params.subject}`;
+}
+
+function accountRouteRateLimitKey(account: ResolvedSmsAccount): string {
+  return `${account.accountId}:${account.webhookPath}`;
 }

 function rejectInvalidRequestRateLimit(params: {
@@ -90,15 +101,15 @@ export function createSmsWebhookHandler(params: SmsWebhookHandlerParams) {
    }

    const clientAddress = resolvedClientAddress({ cfg: params.cfg, req });
-    const key = rateLimitKey({ account: params.account, clientAddress });
-    const invalidRequestRateLimited = invalidRequestRateLimiter.isRateLimited(key);
+    const clientAddressKey = rateLimitKey({ account: params.account, subject: clientAddress });
+    const invalidRequestRateLimited = invalidRequestRateLimiter.isRateLimited(clientAddressKey);

    let form: Record<string, string>;
    try {
      form = await readTwilioWebhookForm(req);
    } catch {
      if (invalidRequestRateLimited) {
-        return rejectInvalidRequestRateLimit({ key, log: params.log, res });
+        return rejectInvalidRequestRateLimit({ key: clientAddressKey, log: params.log, res });
      }
      respondTwiml(res, 400, "Invalid request body");
      return true;
@@ -116,7 +127,7 @@ export function createSmsWebhookHandler(params: SmsWebhookHandlerParams) {
      });
      if (!ok) {
        if (invalidRequestRateLimited) {
-          return rejectInvalidRequestRateLimit({ key, log: params.log, res });
+          return rejectInvalidRequestRateLimit({ key: clientAddressKey, log: params.log, res });
        }
        params.log?.warn?.("SMS webhook rejected invalid Twilio signature");
        respondTwiml(res, 403, "Invalid signature");
@@ -125,13 +136,34 @@ export function createSmsWebhookHandler(params: SmsWebhookHandlerParams) {
    }

    if (invalidRequestRateLimited && params.account.dangerouslyDisableSignatureValidation) {
-      return rejectInvalidRequestRateLimit({ key, log: params.log, res });
+      return rejectInvalidRequestRateLimit({ key: clientAddressKey, log: params.log, res });
    }
-    if (callbackDispatchRateLimiter.isRateLimited(key)) {
-      params.log?.warn?.(`SMS webhook rate limit exceeded for ${key}`);
+    // Twilio egress IPs are shared across unrelated senders: an address-keyed quota
+    // would let one flooding sender 429 every sender behind that IP, so validated
+    // callbacks meter on the canonical signature-covered From value (invalid or absent
+    // From values share one bucket).
+    // With validation disabled nothing authenticates From and rotating it would bypass
+    // the cap, so unauthenticated traffic stays on the fail-closed client address key.
+    const dispatchKey = params.account.dangerouslyDisableSignatureValidation
+      ? clientAddressKey
+      : rateLimitKey({ account: params.account, subject: resolveTwilioInboundSender(form) });
+    if (callbackDispatchRateLimiter.isRateLimited(dispatchKey)) {
+      params.log?.warn?.("SMS webhook callback rate limit exceeded");
      respondTwiml(res, 429, "Rate limit exceeded");
      return true;
    }
+    // Sender fairness must not remove bounded admission for a signed fan-out.
+    // Keep the aggregate route ceiling separate from the sender limiter so one
+    // sender cannot monopolize the route, while many distinct valid senders also
+    // cannot create unbounded durable-ingress pressure.
+    if (!params.account.dangerouslyDisableSignatureValidation) {
+      const aggregateKey = accountRouteRateLimitKey(params.account);
+      if (validatedAggregateRateLimiter.isRateLimited(aggregateKey)) {
+        params.log?.warn?.(`SMS webhook aggregate rate limit exceeded for ${aggregateKey}`);
+        respondTwiml(res, 429, "Rate limit exceeded");
+        return true;
+      }
+    }
    const messageSid = resolveTwilioMessageSid(form);
    if (!messageSid) {
      respondTwiml(res, 400, "Missing MessageSid");