|
1 | | -import type { RequestHandler } from "express"; |
2 | 1 | import { Ratelimit } from "@upstash/ratelimit"; |
| 2 | +import { LRUCache } from "lru-cache"; |
3 | 3 | import { env } from "~/env.server"; |
| 4 | +import type { RedisWithClusterOptions } from "~/redis.server"; |
| 5 | +import { singleton } from "~/utils/singleton"; |
4 | 6 | import { logger } from "./logger.server"; |
5 | 7 | import { RateLimiter, type Duration } from "./rateLimiter.server"; |
6 | 8 |
|
7 | | -const ipLimiter = new RateLimiter({ |
8 | | - keyPrefix: "webhook-ingress-ip", |
9 | | - limiter: Ratelimit.fixedWindow( |
10 | | - env.WEBHOOK_INGRESS_IP_RATE_LIMIT_TOKENS, |
11 | | - env.WEBHOOK_INGRESS_IP_RATE_LIMIT_WINDOW as Duration |
12 | | - ), |
13 | | -}); |
| 9 | +export type WebhookIngressIpLimitResult = |
| 10 | + | { success: true } |
| 11 | + | { success: false; retryAfterSeconds: number }; |
14 | 12 |
|
15 | | -// Coarse per-IP gate mounted in server.ts ahead of the Remix handler. The |
16 | | -// per-opaqueId limiter (webhookIngressRateLimit.server) is the real protection. |
17 | | -export const webhookIngressIpRateLimiter: RequestHandler = async (req, res, next) => { |
18 | | - if (!req.path.startsWith("/webhooks/v1/ingest/")) return next(); |
19 | | - const ip = |
20 | | - (req.headers["x-forwarded-for"] as string)?.split(",")[0]?.trim() || req.ip || "unknown"; |
21 | | - try { |
22 | | - const { success } = await ipLimiter.limit(ip); |
23 | | - if (!success) { |
24 | | - res.status(429).json({ error: "Too many requests" }); |
25 | | - return; |
26 | | - } |
27 | | - } catch (error) { |
28 | | - logger.warn("webhookIngressIpRateLimiter: limiter error, allowing request", { error }); |
29 | | - } |
30 | | - next(); |
31 | | -}; |
| 13 | +/** |
| 14 | + * Per-IP budget for ingress requests to endpoint ids that don't exist, so one source can't flood the |
| 15 | + * ingress with made-up ids. A request to a real endpoint never spends it: that endpoint's own limit |
| 16 | + * bounds it, so a provider IP shared by many customers is never throttled by another customer's |
| 17 | + * traffic. Callers pass the last X-Forwarded-For hop, which the load balancer appends; a request |
| 18 | + * without one isn't limited, and a limiter failure lets the request through. Each process remembers |
| 19 | + * an IP that used up its budget until the window resets, so `blockedFor` can refuse it before any |
| 20 | + * other work. A blocked IP can still reach a real endpoint the process hasn't cached yet: each |
| 21 | + * blocked IP gets its own small lookup budget (`takeBlockedLookup`), so made-up ids cost at most |
| 22 | + * that many database reads per second per source, and one source can't spend another's. |
| 23 | + */ |
| 24 | +export function createWebhookIngressIpRateLimiter(options: { |
| 25 | + tokens: number; |
| 26 | + window: Duration; |
| 27 | + redis?: RedisWithClusterOptions; |
| 28 | + blockedLookupsPerSecond?: number; |
| 29 | +}) { |
| 30 | + const limiter = new RateLimiter({ |
| 31 | + keyPrefix: "webhook-ingress-ip", |
| 32 | + limiter: Ratelimit.fixedWindow(options.tokens, options.window), |
| 33 | + logFailure: false, |
| 34 | + redis: options.redis, |
| 35 | + }); |
| 36 | + const blockedUntil = new LRUCache<string, number>({ max: 10_000 }); |
| 37 | + const lookupsPerSecond = options.blockedLookupsPerSecond ?? 5; |
| 38 | + const lookupBuckets = new LRUCache<string, { tokens: number; refilledAt: number }>({ |
| 39 | + max: 10_000, |
| 40 | + }); |
| 41 | + |
| 42 | + return { |
| 43 | + /** Whether this process may look up an uncached endpoint id for blocked `ip` right now. */ |
| 44 | + takeBlockedLookup(ip: string): boolean { |
| 45 | + const now = Date.now(); |
| 46 | + const bucket = lookupBuckets.get(ip) ?? { tokens: lookupsPerSecond, refilledAt: now }; |
| 47 | + const tokens = Math.min( |
| 48 | + lookupsPerSecond, |
| 49 | + bucket.tokens + ((now - bucket.refilledAt) / 1000) * lookupsPerSecond |
| 50 | + ); |
| 51 | + const taken = tokens >= 1; |
| 52 | + lookupBuckets.set(ip, { tokens: taken ? tokens - 1 : tokens, refilledAt: now }); |
| 53 | + return taken; |
| 54 | + }, |
| 55 | + /** Seconds until `ip` may send requests for unknown endpoints again, or undefined if it may now. */ |
| 56 | + blockedFor(ip: string | null | undefined): number | undefined { |
| 57 | + if (!ip) return undefined; |
| 58 | + const until = blockedUntil.get(ip); |
| 59 | + if (until === undefined) return undefined; |
| 60 | + const now = Date.now(); |
| 61 | + if (until <= now) { |
| 62 | + blockedUntil.delete(ip); |
| 63 | + return undefined; |
| 64 | + } |
| 65 | + return Math.max(1, Math.ceil((until - now) / 1000)); |
| 66 | + }, |
| 67 | + async limit(ip: string | null | undefined): Promise<WebhookIngressIpLimitResult> { |
| 68 | + if (!ip) return { success: true }; |
| 69 | + try { |
| 70 | + const result = await limiter.limit(ip); |
| 71 | + if (result.success) return { success: true }; |
| 72 | + blockedUntil.set(ip, result.reset); |
| 73 | + return { |
| 74 | + success: false, |
| 75 | + retryAfterSeconds: Math.max(1, Math.ceil((result.reset - Date.now()) / 1000)), |
| 76 | + }; |
| 77 | + } catch (error) { |
| 78 | + logger.warn("webhookIngressIpRateLimiter: limiter error, allowing request", { error }); |
| 79 | + return { success: true }; |
| 80 | + } |
| 81 | + }, |
| 82 | + }; |
| 83 | +} |
| 84 | + |
| 85 | +export const webhookIngressIpRateLimiter = singleton("webhookIngressIpRateLimiter", () => |
| 86 | + createWebhookIngressIpRateLimiter({ |
| 87 | + tokens: env.WEBHOOK_INGRESS_IP_RATE_LIMIT_TOKENS, |
| 88 | + window: env.WEBHOOK_INGRESS_IP_RATE_LIMIT_WINDOW as Duration, |
| 89 | + }) |
| 90 | +); |
0 commit comments