diff --git a/.github/workflows/deploy-router.yml b/.github/workflows/deploy-router.yml new file mode 100644 index 00000000..aee37cf2 --- /dev/null +++ b/.github/workflows/deploy-router.yml @@ -0,0 +1,41 @@ +name: Deploy Router (Cloudflare) + +on: + push: + branches: + - main + paths: + - router/** + - .github/workflows/deploy-router.yml + workflow_dispatch: + +concurrency: + group: deploy-router-production + cancel-in-progress: false + +permissions: + contents: read + +jobs: + deploy-router: + name: Deploy router + runs-on: ubuntu-latest + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Setup Node.js + uses: actions/setup-node@v4 + with: + node-version: '22' + cache: 'npm' + + - name: Install dependencies + run: npm ci + + - name: Deploy router to Cloudflare + working-directory: router + env: + CLOUDFLARE_API_TOKEN: ${{ secrets.CLOUDFLARE_API_TOKEN }} + CLOUDFLARE_ACCOUNT_ID: ${{ vars.CLOUDFLARE_ACCOUNT_ID }} + run: ../node_modules/.bin/wrangler deploy diff --git a/package-lock.json b/package-lock.json index 9e68a7d3..bb15deaf 100644 --- a/package-lock.json +++ b/package-lock.json @@ -8,7 +8,8 @@ "name": "agentrelay.com", "version": "0.0.0", "workspaces": [ - "web" + "web", + "router" ], "engines": { "node": ">=20.9.0" @@ -1614,6 +1615,13 @@ "node": ">=16" } }, + "node_modules/@cloudflare/workers-types": { + "version": "4.20260623.1", + "resolved": "https://registry.npmjs.org/@cloudflare/workers-types/-/workers-types-4.20260623.1.tgz", + "integrity": "sha512-J/0POl0HeLepbwDE5Yx5c7jQrHFkvCEFu3TS+TQsDDlg/vTs5og7wdGP6eNGXOAntgWUrjcvvKTmVLTP7OrnAg==", + "dev": true, + "license": "MIT OR Apache-2.0" + }, "node_modules/@cspotcode/source-map-support": { "version": "0.8.1", "resolved": "https://registry.npmjs.org/@cspotcode/source-map-support/-/source-map-support-0.8.1.tgz", @@ -9196,6 +9204,34 @@ "node": ">= 0.6" } }, + "node_modules/express/node_modules/path-to-regexp": { + "version": "8.4.2", + "resolved": "https://registry.npmjs.org/path-to-regexp/-/path-to-regexp-8.4.2.tgz", + "integrity": "sha512-qRcuIdP69NPm4qbACK+aDogI5CBDMi1jKe0ry5rSQJz8JVLsC7jV8XpiJjGRLLol3N+R5ihGYcrPLTno6pAdBA==", + "dev": true, + "license": "MIT", + "funding": { + "type": "opencollective", + "url": "https://opencollective.com/express" + } + }, + "node_modules/express/node_modules/router": { + "version": "2.2.0", + "resolved": "https://registry.npmjs.org/router/-/router-2.2.0.tgz", + "integrity": "sha512-nLTrUKm2UyiL7rlhapu/Zl45FwNgkZGaCpZbIHajDYgwlJCOzLSk+cIPAnsEqV955GjILJnKbdQC1nVPz+gAYQ==", + "dev": true, + "license": "MIT", + "dependencies": { + "debug": "^4.4.0", + "depd": "^2.0.0", + "is-promise": "^4.0.0", + "parseurl": "^1.3.3", + "path-to-regexp": "^8.0.0" + }, + "engines": { + "node": ">= 18" + } + }, "node_modules/extend": { "version": "3.0.2", "resolved": "https://registry.npmjs.org/extend/-/extend-3.0.2.tgz", @@ -13934,32 +13970,8 @@ } }, "node_modules/router": { - "version": "2.2.0", - "resolved": "https://registry.npmjs.org/router/-/router-2.2.0.tgz", - "integrity": "sha512-nLTrUKm2UyiL7rlhapu/Zl45FwNgkZGaCpZbIHajDYgwlJCOzLSk+cIPAnsEqV955GjILJnKbdQC1nVPz+gAYQ==", - "dev": true, - "license": "MIT", - "dependencies": { - "debug": "^4.4.0", - "depd": "^2.0.0", - "is-promise": "^4.0.0", - "parseurl": "^1.3.3", - "path-to-regexp": "^8.0.0" - }, - "engines": { - "node": ">= 18" - } - }, - "node_modules/router/node_modules/path-to-regexp": { - "version": "8.4.2", - "resolved": "https://registry.npmjs.org/path-to-regexp/-/path-to-regexp-8.4.2.tgz", - "integrity": "sha512-qRcuIdP69NPm4qbACK+aDogI5CBDMi1jKe0ry5rSQJz8JVLsC7jV8XpiJjGRLLol3N+R5ihGYcrPLTno6pAdBA==", - "dev": true, - "license": "MIT", - "funding": { - "type": "opencollective", - "url": "https://opencollective.com/express" - } + "resolved": "router", + "link": true }, "node_modules/rw": { "version": "1.3.3", @@ -16197,6 +16209,15 @@ "url": "https://github.com/sponsors/wooorm" } }, + "router": { + "version": "0.0.0", + "devDependencies": { + "@cloudflare/workers-types": "^4.20250312.0", + "typescript": "^5.9.3", + "vitest": "^4.1.0", + "wrangler": "^4.91.0" + } + }, "web": { "version": "0.0.1", "dependencies": { diff --git a/package.json b/package.json index 7da5e6bf..75f242b5 100644 --- a/package.json +++ b/package.json @@ -7,13 +7,15 @@ "node": ">=20.9.0" }, "workspaces": [ - "web" + "web", + "router" ], "scripts": { "dev": "npm --workspace web run dev", "build": "npm --workspace web run build", - "test": "npm --workspace web run test", + "test": "npm --workspaces run test", "preview": "npm --workspace web run preview", - "deploy": "npm --workspace web run deploy" + "deploy": "npm --workspace web run deploy", + "deploy:router": "npm --workspace router run deploy" } } diff --git a/router/README.md b/router/README.md new file mode 100644 index 00000000..0c050ae1 --- /dev/null +++ b/router/README.md @@ -0,0 +1,66 @@ +Route precedence: + +- `/cloud*` routes to the Agent Relay Cloud Next.js app. +- `/observer/file*` routes to the RelayFile observer app, expected to be built with `basePath: "/observer/file"`. +- `/observer*` routes to the Relaycast observer app. +- everything else falls back to the legacy proxy target. + +RelayFile observer deployment is wired from the production job in +`.github/workflows/deploy.yml`. The job runs +`scripts/deploy-file-observer-pages.sh`, which packs +`@relayfile/file-observer@latest`, builds it with +`FILE_OBSERVER_BASE_PATH=/observer/file`, and deploys the detected static output +to the Cloudflare Pages project `relayfile-file-observer`. The script creates +that Pages project with the configured production branch when it is missing, so +the first production deploy does not require manual Cloudflare setup. + +``` + ┌────────────────────────────────────┐ + │ @agentworkforce/agentrelay.com │ + │ │ + │ /router (Cloudflare worker) │ + └────────────────────────────────────┘ + + ┌──────────────────────────────────────┐ + │ agentrelay.com │ + └──────────────────────────────────────┘ + │ + │ + │ + All cloud routes │ Observer Routes + ┌───(/cloud/*, /cloud/api)───────────┼────(/observer*, /observer/file*)───────┐ + │ │ │ + │ │ │ + │ │ │ + ▼ │ ▼ +┌────────────────────────────────────┐ │ ┌────────────────────────────────────┐ +│ https://agentrelay.cloud │ │ │ https://relaycast.dev/observer │ +└────────────────────────────────────┘ │ └────────────────────────────────────┘ + │ + │ +┌────────────────────────────────────┐ All Other Routes ┌────────────────────────────────────┐ +│ @agentworkforce/cloud │ (/, /docs, /blog, /openclaw) │ @agentworkforce/relaycast │ +│ │ │ │ │ +│ /web (NextJs App) │ │ │ /packages/observer-dashboard │ +│ │ │ │ /@relayfile/file-observer │ +└────────────────────────────────────┘ │ └────────────────────────────────────┘ + │ + │ + │ + │ + │ + │ + │ + │ + ▼ + ┌────────────────────────────────────┐ + │ https://origin-web.agentrelay.com │ + └────────────────────────────────────┘ + + + ┌────────────────────────────────────┐ + │ @agentworkforce/agentrelay.com │ + │ │ + │ /web (NextJs App) │ + └────────────────────────────────────┘ +``` diff --git a/router/index.ts b/router/index.ts new file mode 100644 index 00000000..e6e85fda --- /dev/null +++ b/router/index.ts @@ -0,0 +1,447 @@ +import { maybeRecord, type RecorderEnv } from "./src/recorder.js"; +import { maybeRateLimit, type RateLimitEnv } from "./src/rate-limit.js"; + +interface Env { + CLOUD_APP_ORIGIN: string; + CLOUD_WEB_WORKER?: { + fetch(request: Request): Promise; + }; + FILE_OBSERVER_ORIGIN?: string; + TRAFFIC_RECORDER?: RecorderEnv["TRAFFIC_RECORDER"]; + ROUTER_CONFIG?: RecorderEnv["ROUTER_CONFIG"]; + RATE_LIMIT_COUNTERS?: RateLimitEnv["RATE_LIMIT_COUNTERS"]; + WEBHOOK_WORKER?: { + fetch(request: Request): Promise; + }; + WEBHOOK_WORKER_ORIGIN?: string; +} + +function hasRecorderEnv(env: Env): env is Env & RecorderEnv { + return Boolean(env.TRAFFIC_RECORDER && env.ROUTER_CONFIG); +} + +// The apex agentrelay.com marketing/docs site. Served by the standalone +// agentrelay.com repo on Cloudflare Workers (OpenNext); the legacy AWS origin +// at origin.agentrelay.net is retired. See AgentWorkforce/agentrelay.com. +const FALLBACK_PROXY_ORIGIN = "https://origin-web.agentrelay.com"; +const OBSERVER_ORIGIN = "https://observer.relaycast.dev"; +const DEFAULT_FILE_OBSERVER_ORIGIN = "https://relayfile-file-observer.pages.dev"; +const PRIMARY_HOST = "agentrelay.com"; +const FILE_OBSERVER_PATH_PREFIX = "/observer/file"; +const OBSERVER_PATH_PREFIX = "/observer"; +const CLOUD_PATH_PREFIX = "/cloud"; +const WEBHOOK_ORIGIN_FLAG_KEY = "WEBHOOK_ORIGIN"; + +// Header set by webhook-worker's queue consumer +// (`packages/webhook-worker/src/queue-consumer.ts`'s `buildForwardHeaders`) on +// outbound forwards to cloud-web's `/api/v1/webhooks/nango`. When the +// `WEBHOOK_ORIGIN` flag is `"worker"`, the router would otherwise redirect +// every `/api/v1/webhooks/nango` POST back to webhook-worker — including +// webhook-worker's own outbound forwards, producing an infinite redelivery +// loop bounded only by `maxRetries`. Honouring this header lets webhook-worker +// reach cloud-web's `routeForwardEvent` → `handleGitHubForward` path which is +// the actual destination of the forward. +const WEBHOOK_WORKER_FORWARDED_HEADER = "x-cloud-webhook-worker-forwarded"; +const WEBHOOK_WORKER_FORWARDED_VALUE = "webhook-worker"; +let loggedPhase5aLambdaEliminated = false; + +// Exact paths the webhook worker handles. Other sub-paths under +// /api/v1/webhooks (notably /api/v1/webhooks/composio/connect/callback, an +// OAuth callback served by Next.js) must continue to route to the Lambda even +// when WEBHOOK_ORIGIN=worker, otherwise they 404 against the Worker. +const WEBHOOK_WORKER_PATHS = new Set([ + "/api/v1/webhooks/composio", + "/api/v1/webhooks/github", + "/api/v1/webhooks/hookdeck", + "/api/v1/webhooks/nango", +]); +const NANGO_WEBHOOK_WORKER_PATH = "/api/v1/webhooks/nango"; + +function isPathWithinPrefix(pathname: string, prefix: string): boolean { + return pathname === prefix || pathname.startsWith(`${prefix}/`); +} + +function isObserverPath(pathname: string): boolean { + return isPathWithinPrefix(pathname, OBSERVER_PATH_PREFIX); +} + +function isFileObserverPath(pathname: string): boolean { + return isPathWithinPrefix(pathname, FILE_OBSERVER_PATH_PREFIX); +} + +function isPrimaryFileObserverPath(hostname: string, pathname: string): boolean { + return hostname === PRIMARY_HOST && isFileObserverPath(pathname); +} + +function isCloudPath(pathname: string): boolean { + return isPathWithinPrefix(pathname, CLOUD_PATH_PREFIX); +} + +// True only for the exact paths the webhook worker knows how to handle. Used +// to gate worker forwarding so unrelated routes under /api/v1/webhooks/* (e.g. +// the Composio OAuth callback) still reach the Lambda. +export function isWebhookWorkerPath(pathname: string): boolean { + return WEBHOOK_WORKER_PATHS.has(stripPathPrefix(pathname, CLOUD_PATH_PREFIX)); +} + +function isNangoWebhookWorkerPath(pathname: string): boolean { + return stripPathPrefix(pathname, CLOUD_PATH_PREFIX) === NANGO_WEBHOOK_WORKER_PATH; +} + +function stripPathPrefix(pathname: string, prefix: string): string { + if (pathname === prefix) { + return "/"; + } + + if (pathname.startsWith(`${prefix}/`)) { + return pathname.slice(prefix.length); + } + + return pathname; +} + +function addPathPrefix(pathname: string, prefix: string): string { + if (!prefix) { + return pathname; + } + + if (pathname === "/") { + return prefix; + } + + if (pathname === prefix || pathname.startsWith(`${prefix}/`) || pathname.startsWith(`${prefix}?`)) { + return pathname; + } + + return `${prefix}${pathname}`; +} + +function hostnameFromHost(host: string, protocol: string): string { + try { + return new URL(`${protocol}//${host}`).hostname; + } catch { + return host.split(":")[0] ?? host; + } +} + +function isPublicFileObserverLocation(hostname: string, pathname: string): boolean { + return isPrimaryFileObserverPath(hostname, pathname); +} + +export function getUpstreamPath(hostname: string, pathname: string): string { + if (isPrimaryFileObserverPath(hostname, pathname)) { + return stripPathPrefix(pathname, FILE_OBSERVER_PATH_PREFIX); + } + + return pathname; +} + +export function getMountPrefix(hostname: string, pathname: string): string { + if (isCloudPath(pathname)) { + return CLOUD_PATH_PREFIX; + } + + if (isPrimaryFileObserverPath(hostname, pathname)) { + return FILE_OBSERVER_PATH_PREFIX; + } + + return ""; +} + +export function rewriteLocation( + location: string, + originUrl: URL, + requestHost: string, + requestProtocol: string, + mountPrefix = "", +): string { + if (!location) { + return location; + } + + try { + const absolute = new URL(location); + if (isPublicFileObserverLocation(absolute.hostname, absolute.pathname)) { + return location; + } + + if (absolute.hostname !== originUrl.hostname) { + return location; + } + + absolute.hostname = requestHost; + absolute.port = ""; + absolute.protocol = requestProtocol; + absolute.pathname = addPathPrefix(absolute.pathname, mountPrefix); + return absolute.toString(); + } catch { + if (location.startsWith("/")) { + try { + const requestHostname = hostnameFromHost(requestHost, requestProtocol); + const locationUrl = new URL(location, `${requestProtocol}//${requestHost}`); + if (isPublicFileObserverLocation(requestHostname, locationUrl.pathname)) { + return location; + } + } catch { + // Fall through to the normal mount-prefix rewrite. + } + + return addPathPrefix(location, mountPrefix); + } + + return location; + } +} + +export function getOrigin(hostname: string, pathname: string, env: Env): string { + // /cloud* defaults to the Next.js cloud app regardless of host. Requests only + // reach this fallback when the cloud-web Worker service binding is absent. + if (isCloudPath(pathname)) { + return env.CLOUD_APP_ORIGIN; + } + + // The production agentrelay.com apex is a split router: + // 1. /observer/file* goes to the RelayFile file observer app + // 2. /observer* stays on the Relaycast observer app + // 3. everything else falls back to the relay web origin + if (hostname === PRIMARY_HOST) { + if (isPrimaryFileObserverPath(hostname, pathname)) { + return env.FILE_OBSERVER_ORIGIN ?? DEFAULT_FILE_OBSERVER_ORIGIN; + } + + if (isObserverPath(pathname)) { + return OBSERVER_ORIGIN; + } + } + + return FALLBACK_PROXY_ORIGIN; +} + +async function shouldUseCloudWebWorker( + pathname: string, + request: Request, + env: Env, +): Promise { + if (!isCloudPath(pathname)) { + return false; + } + + if (await shouldUseWebhookWorker(pathname, request, env)) { + return false; + } + + return true; +} + +function logPhase5aLambdaEliminatedOnce(): void { + if (loggedPhase5aLambdaEliminated) { + return; + } + + loggedPhase5aLambdaEliminated = true; + console.log(JSON.stringify({ router_phase: "5a_lambda_eliminated" })); +} + +export async function readWebhookOriginFlag(env: Env): Promise { + try { + return (await env.ROUTER_CONFIG?.get(WEBHOOK_ORIGIN_FLAG_KEY)) ?? null; + } catch { + return null; + } +} + +export async function shouldUseNangoWebhookWorkerRoute( + pathname: string, + env: Env, +): Promise { + if (!isNangoWebhookWorkerPath(pathname)) { + return false; + } + + const configured = await readWebhookOriginFlag(env); + return configured?.trim().toLowerCase() === "worker"; +} + +async function shouldUseWebhookWorker( + pathname: string, + request: Request, + env: Env, +): Promise { + // Break the redelivery loop: webhook-worker's queue consumer forwards the + // raw envelope back to the same `/api/v1/webhooks/nango` route on + // `origin.agentrelay.cloud` so cloud-web's `handleGitHubForward` can run. + // Without this header check, the router catches that forward and redirects + // it back to webhook-worker, which re-enqueues, ad infinitum (or until + // `maxRetries`). + if (request.headers.get(WEBHOOK_WORKER_FORWARDED_HEADER) === WEBHOOK_WORKER_FORWARDED_VALUE) { + return false; + } + return shouldUseNangoWebhookWorkerRoute(pathname, env); +} + +function buildWebhookWorkerRequest( + request: Request, + requestUrl: URL, + workerOrigin?: string, +): Request { + const targetUrl = new URL(requestUrl.toString()); + targetUrl.pathname = stripPathPrefix(targetUrl.pathname, CLOUD_PATH_PREFIX); + + if (workerOrigin) { + const originUrl = new URL(workerOrigin); + targetUrl.protocol = originUrl.protocol; + targetUrl.hostname = originUrl.hostname; + targetUrl.port = originUrl.port; + } + + const init: RequestInit & { duplex?: "half" } = { + method: request.method, + headers: request.headers, + body: request.body, + redirect: "manual", + duplex: "half", + }; + + return new Request(targetUrl.toString(), init); +} + +export default { + async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise { + const url = new URL(request.url); + + // Per-key rate limiting runs BEFORE any worker routing so a runaway + // workspace gets bounded everywhere — including webhook ingress and + // /cloud* traffic. The bypass list inside maybeRateLimit exempts + // health and observer paths. See packages/router/src/rate-limit.ts + // and docs/security/rate-limiting.md. + const rateLimited = await maybeRateLimit(request, env); + if (rateLimited) { + return rateLimited; + } + + // Clone the request up front so any branch that returns early (cloud-web + // Worker service binding, webhook Worker service binding, etc.) can + // still feed the recorder the original payload. Without this clone, the + // /cloud Worker path bypasses the recorder entirely and the replay + // harness has no corpus to prove equivalence during Phase 4 cutover. + // See Codex P2.6 on bundle PR #647. + const recorderEnv = hasRecorderEnv(env) ? env : null; + const recorderRequestClone = recorderEnv + ? (request.clone() as unknown as Request) + : null; + + if (await shouldUseCloudWebWorker(url.pathname, request, env)) { + logPhase5aLambdaEliminatedOnce(); + if (!env.CLOUD_WEB_WORKER) { + return new Response( + JSON.stringify({ error: "cloud web worker binding unavailable" }), + { + status: 503, + headers: { "content-type": "application/json" }, + }, + ); + } + + const workerResponse = await env.CLOUD_WEB_WORKER.fetch(request); + if (recorderRequestClone && recorderEnv) { + ctx.waitUntil( + maybeRecord(recorderRequestClone, workerResponse.clone(), recorderEnv, ctx), + ); + } + return workerResponse; + } + + if (await shouldUseWebhookWorker(url.pathname, request, env)) { + // Build the forwarded Request inside each branch: request.body is a + // ReadableStream that can back only one Request, so constructing it before + // the env.WEBHOOK_WORKER check would disturb the stream and make the + // WEBHOOK_WORKER_ORIGIN fallback throw a TypeError. + if (env.WEBHOOK_WORKER) { + const workerResponse = await env.WEBHOOK_WORKER.fetch( + buildWebhookWorkerRequest(request, url), + ); + if (recorderRequestClone && recorderEnv) { + ctx.waitUntil( + maybeRecord(recorderRequestClone, workerResponse.clone(), recorderEnv, ctx), + ); + } + return workerResponse; + } + + const workerOrigin = env.WEBHOOK_WORKER_ORIGIN?.trim(); + if (workerOrigin) { + const originResponse = await globalThis.fetch( + buildWebhookWorkerRequest(request, url, workerOrigin), + ); + if (recorderRequestClone && recorderEnv) { + ctx.waitUntil( + maybeRecord(recorderRequestClone, originResponse.clone(), recorderEnv, ctx), + ); + } + return originResponse; + } + } + + const requestHost = request.headers.get("Host") || url.hostname; + const originUrl = new URL(getOrigin(url.hostname, url.pathname, env)); + const mountPrefix = getMountPrefix(url.hostname, url.pathname); + + url.pathname = getUpstreamPath(url.hostname, url.pathname); + url.hostname = originUrl.hostname; + url.port = ""; + url.protocol = "https:"; + + const headers = new Headers(request.headers); + headers.set("X-Forwarded-Host", requestHost); + headers.set("X-Original-Host", requestHost); + headers.set("X-Forwarded-Proto", "https"); + if (mountPrefix) { + headers.set("X-Forwarded-Prefix", mountPrefix); + } + + // Reuse the clone made at the top of fetch() for the recorder. The + // earlier clone covers all branches; making a second one here would + // be wasted bytes on every request and would double-record on the + // Lambda origin path. + const recordingRequest = recorderRequestClone; + const subRequest = new Request(url.toString(), { + method: request.method, + headers, + body: request.body, + redirect: "manual", + }); + + try { + // Use `globalThis.fetch` rather than a bare `fetch` identifier: Cloudflare + // Workers can hoist bare `fetch` off `globalThis` and throw + // `TypeError: Illegal invocation`. See sage `.claude/rules/workers-fetch.md`. + const upstreamResponse = await globalThis.fetch(subRequest); + const responseHeaders = new Headers(upstreamResponse.headers); + + const location = responseHeaders.get("Location"); + if (location) { + responseHeaders.set( + "Location", + rewriteLocation(location, originUrl, requestHost, "https:", mountPrefix), + ); + } + + const response = new Response(upstreamResponse.body, { + status: upstreamResponse.status, + statusText: upstreamResponse.statusText, + headers: responseHeaders, + }); + + if (recordingRequest && hasRecorderEnv(env)) { + ctx.waitUntil(maybeRecord(recordingRequest, response.clone(), env, ctx)); + } + + return response; + } catch (error) { + return new Response(JSON.stringify({ error: (error as Error).message }), { + status: 500, + headers: { "content-type": "application/json" }, + }); + } + }, +}; diff --git a/router/package.json b/router/package.json new file mode 100644 index 00000000..20e08ed7 --- /dev/null +++ b/router/package.json @@ -0,0 +1,17 @@ +{ + "name": "router", + "private": true, + "version": "0.0.0", + "type": "module", + "scripts": { + "test": "vitest run", + "typecheck": "tsc --noEmit", + "deploy": "wrangler deploy" + }, + "devDependencies": { + "@cloudflare/workers-types": "^4.20250312.0", + "typescript": "^5.9.3", + "vitest": "^4.1.0", + "wrangler": "^4.91.0" + } +} diff --git a/router/src/rate-limit.ts b/router/src/rate-limit.ts new file mode 100644 index 00000000..d7c6c734 --- /dev/null +++ b/router/src/rate-limit.ts @@ -0,0 +1,550 @@ +// Per-key request rate limiting for the Cloudflare edge gateways. +// +// Two Workers front the platform: the `AgentRelayRouter` worker +// (agentrelay.com / cloud router) and the `RelayfileApi` worker +// (api.relayfile.dev). A misbehaving client today (e.g. a relayfile +// mount stuck in a polling loop) can hammer either gateway with +// dozens of requests per second; the upstream Lambda has a 10-slot +// concurrency cap, so once a single key saturates the gateway every +// other user starts seeing 429s. The rate limiter in this module +// runs at the edge BEFORE the request reaches the Lambda, so a +// runaway key is contained to its own per-workspace bucket and +// doesn't deplete shared capacity. +// +// Design choices (intentional trade-offs): +// +// * Counter storage is Cloudflare KV (a `RATE_LIMIT_COUNTERS` +// namespace bound to both Workers). KV is eventually consistent +// across edges, so the effective ceiling at high traffic can be +// ~2-3x the configured per-minute limit. For the 60 req/min +// target that means ~180 req/min worst case — still 100x lower +// than the runaway today and good enough for the runaway-client +// class of bug. Sub-second precision and global atomicity would +// require a Durable Object, which we may add later if KV drift +// bites — but the simpler path ships first. +// +// * Buckets are minute-aligned (epoch-minute integer). Each +// `(limitKey, bucket)` pair is one KV key with a 120-second TTL +// so the previous and current minute coexist briefly during the +// turn. Reads are racy w.r.t. writes; that's accepted (see above). +// +// * Limits are read from `ROUTER_CONFIG` KV at request time, so +// operators can dial them up/down or kill-switch the limiter +// entirely via `wrangler kv:key put` without redeploying. +// +// * Defaults: 60 req/min per workspace-or-user-or-ip key, 1000 +// req/min global ceiling (a separate safety-net counter shared +// across all keys). Both defaults are intentionally conservative +// given the actual runaway pattern observed (30+ req/sec from a +// single key). +// +// * Bypass list mirrors the recorder's deny list plus a few +// observability paths — these must never be rate limited because +// they're either health probes, browser asset requests, or +// observer/webhook ingress that already has its own backpressure. + +const ONE_MINUTE_SECONDS = 60; +// KV TTL on counter entries. Must be >= 60 to cover a full bucket +// window even if the request arrives at the very start of the +// minute. We pick 120s so the previous bucket is still readable +// briefly during the turnover, which means a borderline client that +// straddles the boundary still sees its prior count and can't double +// up by ~timing requests across the second. +const COUNTER_TTL_SECONDS = 120; + +// Default config — operators tune via `wrangler kv:key put` on the +// `ROUTER_CONFIG` namespace; we re-read every request. +const DEFAULT_PER_KEY_PER_MIN = 60; +const DEFAULT_GLOBAL_PER_MIN = 1000; + +// KV config keys we read from `ROUTER_CONFIG`. Centralized here so +// the docs and the infra seed step refer to the same names. +export const RATE_LIMIT_CONFIG_KEYS = { + PER_KEY_PER_MIN: "RATE_LIMIT_PER_WORKSPACE_PER_MIN", + GLOBAL_PER_MIN: "RATE_LIMIT_GLOBAL_PER_MIN", + DISABLED: "RATE_LIMIT_DISABLED", + // Comma-separated path prefixes added to the bypass list at + // runtime. Static `BYPASS_PREFIXES` below is the safe core; this + // key lets an operator whitelist a new path (e.g. a freshly-added + // health probe) without redeploying. Values are matched the same + // way as static prefixes — exact match or `${prefix}/...` startswith. + // Example KV value: `/api/k8s-readiness,/internal/uptime` + BYPASS_PREFIXES_EXTRA: "RATE_LIMIT_BYPASS_PREFIXES_EXTRA", +} as const; + +// Paths that must NEVER be rate limited. Health probes, static +// assets, and observability endpoints. The router/relayfile-api +// share this list — both Workers see these paths even though only +// one of them actually serves them, and accidentally rate-limiting +// a health probe would self-DoS the upstream monitoring. +const BYPASS_PREFIXES: readonly string[] = [ + "/api/health", + "/health", + "/observer", + "/_next/static", + "/favicon.ico", + // Webhook ingress has no auth token and no workspace path, so it + // would bucket by IP and trip the 60/min cap with even modest + // burst traffic from a single provider (GitHub, Composio, Nango, + // Hookdeck). The webhook worker already has queue-backed + // backpressure, so rate-limiting at the edge would only cause us + // to drop legitimate events. See Codex P1.4 on bundle PR #647. + "/api/v1/webhooks", +]; + +export interface RateLimitEnv { + // Counter storage — shared across both gateway Workers so a + // workspace's limit isn't accidentally doubled by hitting both. + RATE_LIMIT_COUNTERS?: KVNamespace; + // Configuration knobs — same KV namespace the router already uses + // for other flags (e.g. WEBHOOK_ORIGIN). Optional so unit tests + // can run without it. + ROUTER_CONFIG?: KVNamespace; +} + +export interface RateLimitDecision { + allowed: boolean; + retryAfterSec?: number; + counter: number; + limit: number; + // The key the decision applies to — exposed so callers can put it + // in the 429 body (helps operators correlate gateway logs with the + // offending workspace/user/IP). + key: string; +} + +// Pure path helper so we can unit test it without a real Request. +// Checks ONLY the static hardcoded prefix list — does not consult KV. +// Use `isBypassPathDynamic` when env is available and you want the +// runtime-extensible bypass list to apply. +export function isBypassPath(pathname: string): boolean { + return matchesAnyPrefix(pathname, BYPASS_PREFIXES); +} + +function matchesAnyPrefix( + pathname: string, + prefixes: readonly string[], +): boolean { + for (const prefix of prefixes) { + if (pathname === prefix || pathname.startsWith(`${prefix}/`)) { + return true; + } + } + return false; +} + +// Module-scoped cache for the KV-resolved extra bypass prefixes. +// Refreshed at most once per `BYPASS_EXTRAS_TTL_MS` per Worker +// isolate. Operators tuning the KV value see propagation within +// roughly `BYPASS_EXTRAS_TTL_MS` plus the time CF's eventually- +// consistent KV takes to fan out to edges (typically <60s). This +// is the same staleness window that applies to the existing numeric +// config keys (`RATE_LIMIT_PER_WORKSPACE_PER_MIN`, etc.) — keeping +// the model consistent across all runtime knobs. +const BYPASS_EXTRAS_TTL_MS = 60_000; +let cachedBypassExtras: readonly string[] = []; +let cachedBypassExtrasFetchedAt = 0; + +/** + * Test-only: reset the bypass-extras cache between test cases. + * Production code never calls this. + */ +export function __resetBypassExtrasCacheForTests(): void { + cachedBypassExtras = []; + cachedBypassExtrasFetchedAt = 0; +} + +async function getBypassExtras( + env: RateLimitEnv, + now: number, +): Promise { + if (!env.ROUTER_CONFIG) { + return []; + } + if (now - cachedBypassExtrasFetchedAt < BYPASS_EXTRAS_TTL_MS) { + return cachedBypassExtras; + } + try { + const raw = await env.ROUTER_CONFIG.get( + RATE_LIMIT_CONFIG_KEYS.BYPASS_PREFIXES_EXTRA, + ); + cachedBypassExtras = parseBypassExtras(raw); + cachedBypassExtrasFetchedAt = now; + } catch { + // ROUTER_CONFIG read failures should never break the data plane. + // Keep the last known list (or the empty initial) and try again + // after the TTL. + cachedBypassExtrasFetchedAt = now; + } + return cachedBypassExtras; +} + +/** + * Exported for unit tests. Parses the comma-separated KV value into a + * deduped, trimmed list of prefixes, filtering out anything that + * doesn't look like an absolute path (must start with `/`). Defensive + * parsing — bad operator input shouldn't be able to break the bypass + * check. + */ +export function parseBypassExtras(raw: string | null): readonly string[] { + if (!raw) { + return []; + } + const out: string[] = []; + const seen = new Set(); + for (const entry of raw.split(",")) { + const trimmed = entry.trim(); + if (trimmed.length === 0 || !trimmed.startsWith("/")) { + continue; + } + if (seen.has(trimmed)) { + continue; + } + seen.add(trimmed); + out.push(trimmed); + } + return out; +} + +/** + * Returns true if the path should bypass rate limiting — checks both + * the static `BYPASS_PREFIXES` list and the KV-runtime-tunable + * `RATE_LIMIT_BYPASS_PREFIXES_EXTRA` list. The static check is fast + * (no KV read); the KV list is cached per-isolate for + * `BYPASS_EXTRAS_TTL_MS`. Used by `maybeRateLimit` internally. + * + * @param pathname - the URL pathname to check + * @param env - rate limit env containing the KV bindings + * @param now - wall-clock ms; injectable for tests + */ +export async function isBypassPathDynamic( + pathname: string, + env: RateLimitEnv, + now: number = Date.now(), +): Promise { + if (isBypassPath(pathname)) { + return true; + } + const extras = await getBypassExtras(env, now); + return matchesAnyPrefix(pathname, extras); +} + +// Extract a workspace ID from any of the path shapes both gateways +// see. Matches the literal "workspaces/" segment (Hono routes both +// `/v1/workspaces/:id/...` on the relayfile worker and +// `/cloud/api/v1/workspaces/:id/...` on the cloud router). +function extractWorkspaceId(pathname: string): string | null { + const match = pathname.match(/\/workspaces\/([a-zA-Z0-9_-]+)/); + return match?.[1] ?? null; +} + +// Cheap, non-cryptographic hash for opaque bearer tokens so we can +// shard rate counters by token without ever storing the token. We +// don't need collision resistance against an adversary — we just +// want different tokens to map to different buckets. FNV-1a 32-bit +// fits in a few lines and is fine for this. +function fnv1a32Hex(input: string): string { + let hash = 0x811c9dc5; + for (let i = 0; i < input.length; i++) { + hash ^= input.charCodeAt(i); + // 32-bit multiply, kept unsigned via `>>> 0`. + hash = Math.imul(hash, 0x01000193) >>> 0; + } + return hash.toString(16).padStart(8, "0"); +} + +// The session cookie name used by the web app. Must match +// `SESSION_COOKIE_NAME` in `packages/core/src/session/jwt.ts`. +// Defined inline here to keep the rate-limit module self-contained — +// the router package has no dependency on the web package. +const SESSION_COOKIE_NAME = "agent_relay_session"; + +// Parse a single named cookie out of a Cookie header value. Returns +// null when the cookie is absent or has an empty value. Handles both +// quoted and unquoted values; uses the first match when the name +// appears more than once (which is invalid per RFC 6265 but occurs +// in the wild). +function parseCookieValue(cookieHeader: string, name: string): string | null { + // Regex: look for `; name=` or start-of-string `name=`, then capture + // the value up to the next `;` or end-of-string. + const pattern = new RegExp( + `(?:^|;)\\s*${name}\\s*=\\s*"?([^";]*)"?(?:;|$)`, + ); + const match = cookieHeader.match(pattern); + const value = match?.[1]?.trim() ?? null; + return value && value.length > 0 ? value : null; +} + +// Build a stable limit key from the incoming request. Pure function — +// no network calls, no JWKS verify (too expensive for the hot path). +// Order of precedence reflects which signal is most specific: +// +// 1. workspace ID from the path (covers the actual runaway shape) +// 2. authenticated user via bearer token hash (covers non-workspace +// authenticated traffic — e.g. /api/v1/me) +// 3. session cookie for cookie-authenticated /cloud* traffic. Without +// this, cookie-auth requests have neither a workspace ID nor an +// auth header and collapse to IP buckets — conflating all users +// that share a NAT IP into one rate limit bucket. We hash the +// cookie value with FNV-1a 32 so the rate counter key never +// contains the raw session token. +// 4. CF-connecting-IP (anonymous fallback; X-Forwarded-For untrusted +// because anyone can set it, but `cf-connecting-ip` is set by +// Cloudflare edge and can't be spoofed by the client) +// 5. "anon" — last-resort bucket so a misbehaving anon flood is +// still bounded, just bundled together (acceptable: this is the +// least-trusted traffic anyway) +export function extractLimitKey(request: Request): string { + const url = new URL(request.url); + const workspaceId = extractWorkspaceId(url.pathname); + if (workspaceId) { + return `ws:${workspaceId}`; + } + + const auth = request.headers.get("authorization"); + if (auth) { + // Strip the scheme prefix ("Bearer ", "Basic ", etc.) before + // hashing so the same token doesn't bucket differently when a + // client toggles capitalization or omits the scheme. + const token = auth.replace(/^[A-Za-z]+\s+/, "").trim(); + if (token.length > 0) { + return `user:${fnv1a32Hex(token)}`; + } + } + + const cookieHeader = request.headers.get("cookie"); + if (cookieHeader) { + const sessionToken = parseCookieValue(cookieHeader, SESSION_COOKIE_NAME); + if (sessionToken) { + return `session:${fnv1a32Hex(sessionToken)}`; + } + } + + const ip = + request.headers.get("cf-connecting-ip") ?? + request.headers.get("x-forwarded-for")?.split(",")[0]?.trim() ?? + null; + if (ip) { + return `ip:${ip}`; + } + + return "anon"; +} + +async function readNumberConfig( + kv: KVNamespace | undefined, + configKey: string, + fallback: number, +): Promise { + if (!kv) { + return fallback; + } + try { + const raw = await kv.get(configKey); + if (raw == null) { + return fallback; + } + const parsed = Number(raw); + if (!Number.isFinite(parsed) || parsed <= 0) { + return fallback; + } + return parsed; + } catch { + // ROUTER_CONFIG read failures should never break the data plane — + // we'd rather rate-limit at the default than 500 every request. + return fallback; + } +} + +async function readBooleanConfig( + kv: KVNamespace | undefined, + configKey: string, +): Promise { + if (!kv) { + return false; + } + try { + const raw = await kv.get(configKey); + return raw?.trim().toLowerCase() === "true"; + } catch { + return false; + } +} + +// Increment-and-read with KV. Not atomic — two concurrent requests at +// the same edge can both read N, both write N+1, and we lose a count. +// Accepted per the module-level note. The `now` parameter is injected +// for testability. +async function bumpCounter( + kv: KVNamespace, + bucketKey: string, + now: number, +): Promise { + let prior = 0; + try { + const raw = await kv.get(bucketKey); + if (raw != null) { + const parsed = Number.parseInt(raw, 10); + if (Number.isFinite(parsed) && parsed >= 0) { + prior = parsed; + } + } + } catch { + // Counter-store read failure — treat as zero and continue. + // Failing closed (blocking the request) on a transient KV error + // would amplify a single CF KV hiccup into a customer outage. + prior = 0; + } + + const next = prior + 1; + try { + await kv.put(bucketKey, String(next), { + expirationTtl: COUNTER_TTL_SECONDS, + }); + } catch { + // Same reasoning — log via the surrounding fetch handler if + // observability is needed, but never break the request. + } + void now; // reserved for future jittered TTLs; keeps the signature stable + return next; +} + +function bucketForNow(now: number): number { + return Math.floor(now / 1000 / ONE_MINUTE_SECONDS); +} + +function secondsUntilNextBucket(now: number): number { + const elapsedInMinute = Math.floor(now / 1000) % ONE_MINUTE_SECONDS; + return Math.max(1, ONE_MINUTE_SECONDS - elapsedInMinute); +} + +export interface CheckRateLimitOptions { + // Injected for deterministic tests; defaults to Date.now() in prod. + now?: () => number; +} + +export async function checkRateLimit( + key: string, + env: RateLimitEnv, + options: CheckRateLimitOptions = {}, +): Promise { + const nowFn = options.now ?? (() => Date.now()); + const now = nowFn(); + + const [disabled, perKeyLimit, globalLimit] = await Promise.all([ + readBooleanConfig(env.ROUTER_CONFIG, RATE_LIMIT_CONFIG_KEYS.DISABLED), + readNumberConfig( + env.ROUTER_CONFIG, + RATE_LIMIT_CONFIG_KEYS.PER_KEY_PER_MIN, + DEFAULT_PER_KEY_PER_MIN, + ), + readNumberConfig( + env.ROUTER_CONFIG, + RATE_LIMIT_CONFIG_KEYS.GLOBAL_PER_MIN, + DEFAULT_GLOBAL_PER_MIN, + ), + ]); + + if (disabled || !env.RATE_LIMIT_COUNTERS) { + return { allowed: true, counter: 0, limit: perKeyLimit, key }; + } + + const bucket = bucketForNow(now); + const perKeyBucket = `rl:${key}:${bucket}`; + const globalBucket = `rl:__global__:${bucket}`; + + // Sequential reads/writes — KV doesn't support transactions and + // we want the global counter to include this request even if the + // per-key one rejects. Order is intentional: per-key first so a + // single runaway key doesn't get to skew the global ceiling by + // also incrementing the global bucket *before* its per-key check + // would have rejected it. Net effect: a blocked request still + // contributes 1 to per-key but NOT to global. + const perKeyCount = await bumpCounter( + env.RATE_LIMIT_COUNTERS, + perKeyBucket, + now, + ); + + if (perKeyCount > perKeyLimit) { + return { + allowed: false, + retryAfterSec: secondsUntilNextBucket(now), + counter: perKeyCount, + limit: perKeyLimit, + key, + }; + } + + const globalCount = await bumpCounter( + env.RATE_LIMIT_COUNTERS, + globalBucket, + now, + ); + + if (globalCount > globalLimit) { + return { + allowed: false, + retryAfterSec: secondsUntilNextBucket(now), + counter: globalCount, + limit: globalLimit, + key: "__global__", + }; + } + + return { + allowed: true, + counter: perKeyCount, + limit: perKeyLimit, + key, + }; +} + +// Builds the 429 response. Centralized so both Workers return the +// same shape; clients (relayfile mount, CLI) can rely on it. +export function rateLimitResponse( + retryAfterSec: number, + key: string, +): Response { + const body = JSON.stringify({ + error: "rate_limited", + retryAfter: retryAfterSec, + limitKey: key, + }); + return new Response(body, { + status: 429, + headers: { + "content-type": "application/json", + "retry-after": String(retryAfterSec), + "x-ratelimit-key": key, + }, + }); +} + +// Convenience wrapper that combines bypass check, key extraction, +// the actual limit check, and the 429 response. Both Workers call +// this at the very top of their fetch handler. +// +// Returns: +// * `null` — request is allowed; continue to normal handling +// * `Response` — return this 429 immediately +export async function maybeRateLimit( + request: Request, + env: RateLimitEnv, + options: CheckRateLimitOptions = {}, +): Promise { + const url = new URL(request.url); + // Check both the static `BYPASS_PREFIXES` and the KV-runtime-tunable + // extras (cached 60s per isolate). Static check is fast; KV cache + // means at most one KV read per minute per isolate per Worker. + if (await isBypassPathDynamic(url.pathname, env, options.now?.() ?? Date.now())) { + return null; + } + + const key = extractLimitKey(request); + const decision = await checkRateLimit(key, env, options); + if (decision.allowed) { + return null; + } + return rateLimitResponse(decision.retryAfterSec ?? 60, decision.key); +} diff --git a/router/src/recorder.ts b/router/src/recorder.ts new file mode 100644 index 00000000..be92f1e7 --- /dev/null +++ b/router/src/recorder.ts @@ -0,0 +1,140 @@ +import { redactBody, redactHeaders, shouldRecord } from "./redact.js"; + +const DEFAULT_SAMPLE_RATE = 100; +const MAX_BODY_BYTES = 256 * 1024; +const OVERSIZE_PREFIX = "[OVERSIZE:"; +const textEncoder = new TextEncoder(); + +export interface RecorderEnv { + TRAFFIC_RECORDER: R2Bucket; + ROUTER_CONFIG: KVNamespace; +} + +type RecorderContext = Pick; +type BodyReader = { + body: ReadableStream | null; + bodyUsed: boolean; + headers: Headers; + text(): Promise; +}; + +type TrafficRecord = { + ts: string; + method: string; + path: string; + query: string; + request_headers: Record; + request_body: string | null; + response_status: number; + response_headers: Record; + response_body: string | null; + request_id: string; +}; + +function normalizeSampleRate(value: string | null): number { + const parsed = Number(value); + if (!Number.isFinite(parsed)) { + return DEFAULT_SAMPLE_RATE; + } + + return Math.min(100, Math.max(0, parsed)); +} + +function makeOversizePlaceholder(size: number): string { + return `${OVERSIZE_PREFIX}${size}]`; +} + +async function readBody( + payload: BodyReader | null, + path: string, + contentType: string | null, +): Promise { + if (!payload || payload.body == null || payload.bodyUsed) { + return null; + } + + const body = await payload.text(); + const redactedBody = redactBody(path, body, contentType); + if (redactedBody == null) { + return null; + } + + const bodySize = textEncoder.encode(redactedBody).length; + if (bodySize > MAX_BODY_BYTES) { + return makeOversizePlaceholder(bodySize); + } + + return redactedBody; +} + +function padDatePart(value: number): string { + return String(value).padStart(2, "0"); +} + +function buildRecordKey(timestamp: Date, requestId: string): string { + const year = timestamp.getUTCFullYear(); + const month = padDatePart(timestamp.getUTCMonth() + 1); + const day = padDatePart(timestamp.getUTCDate()); + const hour = padDatePart(timestamp.getUTCHours()); + return `corpus/${year}/${month}/${day}/${hour}/${requestId}.ndjson`; +} + +function getRequestId(request: Request): string { + return ( + request.headers.get("x-request-id") ?? + request.headers.get("cf-ray") ?? + request.headers.get("traceparent") ?? + crypto.randomUUID() + ); +} + +export async function maybeRecord( + request: Request, + response: Response, + env: RecorderEnv, + ctx: RecorderContext, +): Promise { + try { + const url = new URL(request.url); + if (!shouldRecord(url.pathname)) { + return; + } + + const sampleRate = normalizeSampleRate(await env.ROUTER_CONFIG.get("RECORDER_SAMPLE_RATE")); + if (Math.random() * 100 >= sampleRate) { + return; + } + + const requestClone = request.bodyUsed ? null : request.clone(); + const responseClone = response.bodyUsed ? null : response.clone(); + const timestamp = new Date(); + const requestId = getRequestId(request); + + const record: TrafficRecord = { + ts: timestamp.toISOString(), + method: request.method, + path: url.pathname, + query: url.search.startsWith("?") ? url.search.slice(1) : url.search, + request_headers: redactHeaders(request.headers), + request_body: await readBody( + requestClone, + url.pathname, + requestClone?.headers.get("content-type") ?? null, + ), + response_status: response.status, + response_headers: redactHeaders(response.headers), + response_body: await readBody( + responseClone, + url.pathname, + responseClone?.headers.get("content-type") ?? null, + ), + request_id: requestId, + }; + + const line = `${JSON.stringify(record)}\n`; + const key = buildRecordKey(timestamp, requestId); + ctx.waitUntil(env.TRAFFIC_RECORDER.put(key, line)); + } catch (error) { + console.error("traffic recorder failed", error); + } +} diff --git a/router/src/redact.ts b/router/src/redact.ts new file mode 100644 index 00000000..efa3b60d --- /dev/null +++ b/router/src/redact.ts @@ -0,0 +1,136 @@ +const HEADER_DENY_LIST = new Set([ + "authorization", + "cookie", + "set-cookie", + "x-relayauth-token", + "x-api-key", +]); + +const SENSITIVE_HEADER_NAME = /(token|secret|password|signature)/i; + +const FULL_BODY_REDACTION_PATHS = [ + /^\/api\/v1\/workspaces\/[a-zA-Z0-9_-]+\/secrets(?:\/.*)?$/, + /^\/api\/auth(?:\/.*)?$/, + /^\/api\/v1\/auth(?:\/.*)?$/, + /^\/api\/v1\/cli\/auth(?:\/.*)?$/, + /^\/api\/v1\/cli\/login(?:\/.*)?$/, + /^\/api\/v1\/integrations\/nango\/connect-link(?:\/.*)?$/, + /^\/api\/v1\/workspaces\/[a-zA-Z0-9_-]+\/integrations\/connect-session(?:\/.*)?$/, + // Recorder transcript ingest carries full meeting transcript_text/summary_text; + // never persist its body to the traffic corpus. Match with or without the + // /cloud app prefix. + /^\/(?:cloud\/)?api\/v1\/webhooks\/transcripts(?:\/.*)?$/, +]; + +const JSON_SECRET_KEYS = new Set([ + "token", + "accesstoken", + "refreshtoken", + "secret", + "apikey", + "password", + "clientsecret", + "privatekey", +]); + +type HeaderInput = Headers | Iterable<[string, string]> | Record; + +const textEncoder = new TextEncoder(); + +function utf8ByteLength(value: string): number { + return textEncoder.encode(value).length; +} + +function redactValue(value: string): string { + return `[REDACTED:${utf8ByteLength(value)}]`; +} + +function shouldRedactHeader(name: string): boolean { + const normalized = name.toLowerCase(); + return HEADER_DENY_LIST.has(normalized) || SENSITIVE_HEADER_NAME.test(normalized); +} + +function headerEntries(headers: HeaderInput): Iterable<[string, string]> { + if (headers instanceof Headers) { + const entries: Array<[string, string]> = []; + headers.forEach((value, name) => { + entries.push([name, value]); + }); + return entries; + } + + if (Symbol.iterator in Object(headers)) { + return headers as Iterable<[string, string]>; + } + + return Object.entries(headers); +} + +function shouldFullyRedactBody(path: string): boolean { + return FULL_BODY_REDACTION_PATHS.some((pattern) => pattern.test(path)); +} + +function isJsonContentType(contentType: string | null | undefined): boolean { + return typeof contentType === "string" && contentType.toLowerCase().includes("application/json"); +} + +function redactJsonValue(value: unknown): unknown { + if (Array.isArray(value)) { + return value.map((item) => redactJsonValue(item)); + } + + if (!value || typeof value !== "object") { + return value; + } + + const output: Record = {}; + for (const [key, nested] of Object.entries(value)) { + output[key] = JSON_SECRET_KEYS.has(key.toLowerCase()) + ? redactValue(typeof nested === "string" ? nested : JSON.stringify(nested) ?? "") + : redactJsonValue(nested); + } + + return output; +} + +export function redactHeaders(headers: HeaderInput): Record { + const output: Record = {}; + + for (const [name, value] of headerEntries(headers)) { + output[name] = shouldRedactHeader(name) ? redactValue(value) : value; + } + + return output; +} + +export function redactBody( + path: string, + body: string | null | undefined, + contentType?: string | null, +): string | null | undefined { + if (body == null) { + return body; + } + + if (shouldFullyRedactBody(path)) { + return redactValue(body); + } + + if (!isJsonContentType(contentType)) { + return body; + } + + try { + return JSON.stringify(redactJsonValue(JSON.parse(body))); + } catch { + return body; + } +} + +export function shouldRecord(path: string): boolean { + if (path === "/favicon.ico" || path === "/api/health") { + return false; + } + + return path !== "/observer" && !path.startsWith("/observer/"); +} diff --git a/router/test/rate-limit.test.ts b/router/test/rate-limit.test.ts new file mode 100644 index 00000000..369f55b4 --- /dev/null +++ b/router/test/rate-limit.test.ts @@ -0,0 +1,588 @@ +import { beforeEach, describe, expect, it } from "vitest"; + +import { + __resetBypassExtrasCacheForTests, + checkRateLimit, + extractLimitKey, + isBypassPath, + isBypassPathDynamic, + maybeRateLimit, + parseBypassExtras, + rateLimitResponse, + RATE_LIMIT_CONFIG_KEYS, + type RateLimitEnv, +} from "../src/rate-limit.js"; + +// Tiny in-memory KV shim. Counter increments only need get/put; we +// also store an optional `expirationTtl` so a future test could +// assert TTLs, though the current cases don't depend on it. +type StoredEntry = { value: string; expiresAt: number | null }; + +class MemoryKV { + readonly store = new Map(); + private nowFn: () => number; + + constructor(nowFn: () => number = () => Date.now()) { + this.nowFn = nowFn; + } + + async get(key: string): Promise { + const entry = this.store.get(key); + if (!entry) return null; + if (entry.expiresAt !== null && entry.expiresAt <= this.nowFn()) { + this.store.delete(key); + return null; + } + return entry.value; + } + + async put( + key: string, + value: string, + options?: { expirationTtl?: number }, + ): Promise { + const ttl = options?.expirationTtl; + const expiresAt = + typeof ttl === "number" && ttl > 0 ? this.nowFn() + ttl * 1000 : null; + this.store.set(key, { value, expiresAt }); + } + + // Helpers — not part of the KVNamespace surface. + size(): number { + return this.store.size; + } +} + +function asKV(kv: MemoryKV): KVNamespace { + return kv as unknown as KVNamespace; +} + +function buildEnv(opts: { + counters?: MemoryKV; + config?: MemoryKV; +}): RateLimitEnv { + return { + RATE_LIMIT_COUNTERS: opts.counters ? asKV(opts.counters) : undefined, + ROUTER_CONFIG: opts.config ? asKV(opts.config) : undefined, + }; +} + +const FIXED_NOW = 1715683680000; // 2024-05-14T11:28:00Z — middle of an even minute +const fixedNow = () => FIXED_NOW; + +describe("isBypassPath", () => { + it.each([ + ["/api/health", true], + ["/api/health/db", true], + ["/health", true], + ["/observer", true], + ["/observer/session/abc", true], + ["/_next/static/chunks/123.js", true], + ["/favicon.ico", true], + ["/v1/workspaces/ws_123/sync/status", false], + ["/cloud/api/v1/workspaces/ws_123/sandboxes", false], + ["/", false], + ])("path %s bypass=%s", (path, expected) => { + expect(isBypassPath(path)).toBe(expected); + }); +}); + +describe("extractLimitKey", () => { + it("picks workspace id from /v1/workspaces/:id/...", () => { + const req = new Request( + "https://api.relayfile.dev/v1/workspaces/ws_abc123/sync/status", + ); + expect(extractLimitKey(req)).toBe("ws:ws_abc123"); + }); + + it("picks workspace id from /cloud/api/v1/workspaces/:id/...", () => { + const req = new Request( + "https://agentrelay.com/cloud/api/v1/workspaces/ws_xyz/sandboxes", + ); + expect(extractLimitKey(req)).toBe("ws:ws_xyz"); + }); + + it("hashes bearer token when no workspace path is present", () => { + const req = new Request("https://agentrelay.com/cloud/api/v1/me", { + headers: { authorization: "Bearer abcdef0123456789" }, + }); + const key = extractLimitKey(req); + expect(key).toMatch(/^user:[0-9a-f]{8}$/); + }); + + it("buckets different tokens to different user keys", () => { + const a = extractLimitKey( + new Request("https://agentrelay.com/cloud/api/v1/me", { + headers: { authorization: "Bearer aaaaaaaa" }, + }), + ); + const b = extractLimitKey( + new Request("https://agentrelay.com/cloud/api/v1/me", { + headers: { authorization: "Bearer bbbbbbbb" }, + }), + ); + expect(a).not.toBe(b); + expect(a).toMatch(/^user:/); + expect(b).toMatch(/^user:/); + }); + + it("falls back to cf-connecting-ip for anonymous traffic", () => { + const req = new Request("https://agentrelay.com/", { + headers: { "cf-connecting-ip": "203.0.113.5" }, + }); + expect(extractLimitKey(req)).toBe("ip:203.0.113.5"); + }); + + it("prefers cf-connecting-ip over x-forwarded-for", () => { + const req = new Request("https://agentrelay.com/", { + headers: { + "cf-connecting-ip": "203.0.113.5", + "x-forwarded-for": "10.0.0.1, 10.0.0.2", + }, + }); + expect(extractLimitKey(req)).toBe("ip:203.0.113.5"); + }); + + it("uses x-forwarded-for first hop when cf-connecting-ip is missing", () => { + const req = new Request("https://agentrelay.com/", { + headers: { "x-forwarded-for": "10.0.0.1, 10.0.0.2" }, + }); + expect(extractLimitKey(req)).toBe("ip:10.0.0.1"); + }); + + it("keys off session cookie for /cloud* traffic with no auth header", () => { + const req = new Request("https://agentrelay.com/cloud/api/v1/me", { + headers: { cookie: "agent_relay_session=mysecrettoken123" }, + }); + const key = extractLimitKey(req); + expect(key).toMatch(/^session:[0-9a-f]{8}$/); + }); + + it("auth header takes priority over session cookie", () => { + const req = new Request("https://agentrelay.com/cloud/api/v1/me", { + headers: { + authorization: "Bearer bearertoken999", + cookie: "agent_relay_session=mysecrettoken123", + }, + }); + const key = extractLimitKey(req); + // auth header wins — produces a user: key, not session: + expect(key).toMatch(/^user:[0-9a-f]{8}$/); + }); + + it("buckets different session cookie values to different session keys", () => { + const a = extractLimitKey( + new Request("https://agentrelay.com/cloud/api/v1/me", { + headers: { cookie: "agent_relay_session=tokenA" }, + }), + ); + const b = extractLimitKey( + new Request("https://agentrelay.com/cloud/api/v1/me", { + headers: { cookie: "agent_relay_session=tokenB" }, + }), + ); + expect(a).not.toBe(b); + expect(a).toMatch(/^session:/); + expect(b).toMatch(/^session:/); + }); + + it("/cloud* with no cookie and no auth falls back to IP", () => { + const req = new Request("https://agentrelay.com/cloud/api/v1/me", { + headers: { "cf-connecting-ip": "198.51.100.1" }, + }); + expect(extractLimitKey(req)).toBe("ip:198.51.100.1"); + }); + + it("/cloud* with no cookie, no auth, no IP falls back to anon", () => { + const req = new Request("https://agentrelay.com/cloud/api/v1/me"); + expect(extractLimitKey(req)).toBe("anon"); + }); + + it("falls back to anon when no signal is present", () => { + const req = new Request("https://agentrelay.com/"); + expect(extractLimitKey(req)).toBe("anon"); + }); +}); + +describe("checkRateLimit", () => { + let counters: MemoryKV; + let config: MemoryKV; + let env: RateLimitEnv; + + beforeEach(() => { + counters = new MemoryKV(fixedNow); + config = new MemoryKV(fixedNow); + env = buildEnv({ counters, config }); + }); + + it("allows requests under the per-key limit", async () => { + for (let i = 1; i <= 5; i++) { + const result = await checkRateLimit("ws:demo", env, { now: fixedNow }); + expect(result.allowed).toBe(true); + expect(result.counter).toBe(i); + } + }); + + it("returns 429 once the per-key limit is exceeded", async () => { + // Pin the per-workspace limit low for deterministic assertions. + await config.put(RATE_LIMIT_CONFIG_KEYS.PER_KEY_PER_MIN, "3"); + + // Sequential calls — the in-memory KV shim mirrors a single edge + // processing requests in order. KV's eventual consistency across + // edges is acknowledged in the module-level docstring but is not + // what this test asserts. + const results = [ + await checkRateLimit("ws:demo", env, { now: fixedNow }), + await checkRateLimit("ws:demo", env, { now: fixedNow }), + await checkRateLimit("ws:demo", env, { now: fixedNow }), + ]; + expect(results.map((r) => r.allowed)).toEqual([true, true, true]); + + const blocked = await checkRateLimit("ws:demo", env, { now: fixedNow }); + expect(blocked.allowed).toBe(false); + expect(blocked.counter).toBe(4); + expect(blocked.limit).toBe(3); + expect(blocked.retryAfterSec).toBeGreaterThan(0); + expect(blocked.retryAfterSec).toBeLessThanOrEqual(60); + expect(blocked.key).toBe("ws:demo"); + }); + + it("does not pollute the global counter on a per-key reject", async () => { + await config.put(RATE_LIMIT_CONFIG_KEYS.PER_KEY_PER_MIN, "1"); + await checkRateLimit("ws:a", env, { now: fixedNow }); + const blocked = await checkRateLimit("ws:a", env, { now: fixedNow }); + expect(blocked.allowed).toBe(false); + + // Verify the global counter only ever incremented once (for the + // single allowed request) — the blocked one should NOT have + // incremented it. This is intentional so a single misbehaving + // key can't push the global ceiling around. + const bucket = Math.floor(FIXED_NOW / 1000 / 60); + const stored = await counters.get(`rl:__global__:${bucket}`); + expect(stored).toBe("1"); + }); + + it("returns 429 once the global limit is exceeded", async () => { + await config.put(RATE_LIMIT_CONFIG_KEYS.GLOBAL_PER_MIN, "2"); + await config.put(RATE_LIMIT_CONFIG_KEYS.PER_KEY_PER_MIN, "10"); + + expect( + (await checkRateLimit("ws:a", env, { now: fixedNow })).allowed, + ).toBe(true); + expect( + (await checkRateLimit("ws:b", env, { now: fixedNow })).allowed, + ).toBe(true); + + const blocked = await checkRateLimit("ws:c", env, { now: fixedNow }); + expect(blocked.allowed).toBe(false); + // Global rejects surface the synthetic `__global__` key so the + // 429 body tells operators it's the global ceiling, not the + // per-key one. + expect(blocked.key).toBe("__global__"); + }); + + it("respects RATE_LIMIT_DISABLED kill switch", async () => { + await config.put(RATE_LIMIT_CONFIG_KEYS.PER_KEY_PER_MIN, "1"); + await config.put(RATE_LIMIT_CONFIG_KEYS.DISABLED, "true"); + + // Even though the limit is 1, the kill switch should let + // unlimited requests through. + for (let i = 0; i < 5; i++) { + const result = await checkRateLimit("ws:demo", env, { now: fixedNow }); + expect(result.allowed).toBe(true); + } + // No counter writes when disabled — saves KV operations. + expect(counters.size()).toBe(0); + }); + + it("resets per minute (different bucket)", async () => { + await config.put(RATE_LIMIT_CONFIG_KEYS.PER_KEY_PER_MIN, "2"); + + const firstMinute = fixedNow; + const secondMinute = () => FIXED_NOW + 60_000; + + expect( + (await checkRateLimit("ws:demo", env, { now: firstMinute })).allowed, + ).toBe(true); + expect( + (await checkRateLimit("ws:demo", env, { now: firstMinute })).allowed, + ).toBe(true); + expect( + (await checkRateLimit("ws:demo", env, { now: firstMinute })).allowed, + ).toBe(false); + + // Roll forward one full minute — bucket key changes, counter + // starts over. + const reset = await checkRateLimit("ws:demo", env, { now: secondMinute }); + expect(reset.allowed).toBe(true); + expect(reset.counter).toBe(1); + }); + + it("falls back to defaults when ROUTER_CONFIG is unbound", async () => { + const noConfigEnv = buildEnv({ counters }); + const result = await checkRateLimit("ws:demo", noConfigEnv, { + now: fixedNow, + }); + expect(result.allowed).toBe(true); + expect(result.limit).toBe(60); // DEFAULT_PER_KEY_PER_MIN + }); + + it("treats missing RATE_LIMIT_COUNTERS as allow-all (fail open)", async () => { + const noCounterEnv = buildEnv({ config }); + await config.put(RATE_LIMIT_CONFIG_KEYS.PER_KEY_PER_MIN, "1"); + // Without a counter store we cannot enforce, so behavior is + // intentionally permissive (the limit is documented as + // best-effort). This guards against a missing binding silently + // breaking the data plane. + for (let i = 0; i < 10; i++) { + expect( + (await checkRateLimit("ws:demo", noCounterEnv, { now: fixedNow })) + .allowed, + ).toBe(true); + } + }); +}); + +describe("rateLimitResponse", () => { + it("emits a 429 with the documented shape", async () => { + const res = rateLimitResponse(42, "ws:demo"); + expect(res.status).toBe(429); + expect(res.headers.get("retry-after")).toBe("42"); + expect(res.headers.get("x-ratelimit-key")).toBe("ws:demo"); + expect(res.headers.get("content-type")).toBe("application/json"); + const body = (await res.json()) as Record; + expect(body).toEqual({ + error: "rate_limited", + retryAfter: 42, + limitKey: "ws:demo", + }); + }); +}); + +describe("maybeRateLimit", () => { + it("returns null for bypass paths without touching KV", async () => { + const counters = new MemoryKV(fixedNow); + const config = new MemoryKV(fixedNow); + const env = buildEnv({ counters, config }); + + const result = await maybeRateLimit( + new Request("https://api.relayfile.dev/health"), + env, + { now: fixedNow }, + ); + expect(result).toBeNull(); + expect(counters.size()).toBe(0); + }); + + it("returns null when under the limit", async () => { + const counters = new MemoryKV(fixedNow); + const config = new MemoryKV(fixedNow); + const env = buildEnv({ counters, config }); + + const result = await maybeRateLimit( + new Request("https://api.relayfile.dev/v1/workspaces/ws_a/sync/status"), + env, + { now: fixedNow }, + ); + expect(result).toBeNull(); + }); + + it("returns a 429 once limit is exceeded", async () => { + const counters = new MemoryKV(fixedNow); + const config = new MemoryKV(fixedNow); + const env = buildEnv({ counters, config }); + await config.put(RATE_LIMIT_CONFIG_KEYS.PER_KEY_PER_MIN, "1"); + + const url = + "https://api.relayfile.dev/v1/workspaces/ws_a/sync/status"; + const first = await maybeRateLimit(new Request(url), env, { + now: fixedNow, + }); + expect(first).toBeNull(); + + const blocked = await maybeRateLimit(new Request(url), env, { + now: fixedNow, + }); + expect(blocked).not.toBeNull(); + expect(blocked!.status).toBe(429); + expect(blocked!.headers.get("x-ratelimit-key")).toBe("ws:ws_a"); + }); +}); + +describe("parseBypassExtras", () => { + it("returns empty array for null / empty input", () => { + expect(parseBypassExtras(null)).toEqual([]); + expect(parseBypassExtras("")).toEqual([]); + expect(parseBypassExtras(" ")).toEqual([]); + }); + + it("parses comma-separated absolute paths", () => { + expect(parseBypassExtras("/api/k8s-readiness,/internal/uptime")).toEqual([ + "/api/k8s-readiness", + "/internal/uptime", + ]); + }); + + it("trims whitespace around entries", () => { + expect(parseBypassExtras(" /a , /b , /c ")).toEqual(["/a", "/b", "/c"]); + }); + + it("ignores entries that don't start with /", () => { + // Defensive: bad operator input shouldn't poison the list. + expect(parseBypassExtras("/ok,not-a-path,/also-ok")).toEqual([ + "/ok", + "/also-ok", + ]); + }); + + it("dedupes repeated entries", () => { + expect(parseBypassExtras("/x,/x,/y,/x")).toEqual(["/x", "/y"]); + }); +}); + +describe("isBypassPathDynamic", () => { + beforeEach(() => { + __resetBypassExtrasCacheForTests(); + }); + + it("returns true for static bypass paths without touching KV", async () => { + const env = buildEnv({}); // no ROUTER_CONFIG bound + expect(await isBypassPathDynamic("/api/health", env, FIXED_NOW)).toBe(true); + expect(await isBypassPathDynamic("/observer/foo", env, FIXED_NOW)).toBe( + true, + ); + }); + + it("returns false for non-bypass paths when no KV extras set", async () => { + const config = new MemoryKV(fixedNow); + const env = buildEnv({ config }); + expect( + await isBypassPathDynamic("/v1/workspaces/ws_a/sync/status", env, FIXED_NOW), + ).toBe(false); + }); + + it("respects KV-configured extras (exact match)", async () => { + const config = new MemoryKV(fixedNow); + const env = buildEnv({ config }); + await config.put( + RATE_LIMIT_CONFIG_KEYS.BYPASS_PREFIXES_EXTRA, + "/api/k8s-readiness", + ); + expect( + await isBypassPathDynamic("/api/k8s-readiness", env, FIXED_NOW), + ).toBe(true); + }); + + it("respects KV-configured extras (prefix match)", async () => { + const config = new MemoryKV(fixedNow); + const env = buildEnv({ config }); + await config.put( + RATE_LIMIT_CONFIG_KEYS.BYPASS_PREFIXES_EXTRA, + "/internal/uptime", + ); + expect( + await isBypassPathDynamic("/internal/uptime/check", env, FIXED_NOW), + ).toBe(true); + }); + + it("caches KV reads for 60s (does not re-read on every call)", async () => { + const config = new MemoryKV(fixedNow); + const env = buildEnv({ config }); + await config.put( + RATE_LIMIT_CONFIG_KEYS.BYPASS_PREFIXES_EXTRA, + "/api/k8s-readiness", + ); + + // Count KV reads by wrapping the real one. + let reads = 0; + const originalGet = env.ROUTER_CONFIG!.get.bind(env.ROUTER_CONFIG!); + env.ROUTER_CONFIG!.get = ((k: string) => { + reads++; + return originalGet(k); + }) as NonNullable["get"]; + + // First call hits KV. + await isBypassPathDynamic("/v1/x", env, FIXED_NOW); + expect(reads).toBe(1); + // Second call within 60s reuses the cache. + await isBypassPathDynamic("/v1/y", env, FIXED_NOW + 30_000); + expect(reads).toBe(1); + // Past the TTL boundary, fetches again. + await isBypassPathDynamic("/v1/z", env, FIXED_NOW + 61_000); + expect(reads).toBe(2); + }); + + it("falls back to last-known list on KV transient errors", async () => { + const config = new MemoryKV(fixedNow); + const env = buildEnv({ config }); + await config.put( + RATE_LIMIT_CONFIG_KEYS.BYPASS_PREFIXES_EXTRA, + "/cached-ok", + ); + + // Warm the cache once. + await isBypassPathDynamic("/v1/x", env, FIXED_NOW); + expect(await isBypassPathDynamic("/cached-ok", env, FIXED_NOW)).toBe(true); + + // Now break the KV. After TTL expiry the read will throw — we + // should fall back to the cached value rather than crashing. + env.ROUTER_CONFIG!.get = (() => { + throw new Error("KV unavailable"); + }) as NonNullable["get"]; + + expect( + await isBypassPathDynamic("/cached-ok", env, FIXED_NOW + 120_000), + ).toBe(true); + }); + + it("static and dynamic bypass coexist (static still works with KV extras present)", async () => { + const config = new MemoryKV(fixedNow); + const env = buildEnv({ config }); + await config.put( + RATE_LIMIT_CONFIG_KEYS.BYPASS_PREFIXES_EXTRA, + "/internal/uptime", + ); + // Static + dynamic both bypass. + expect(await isBypassPathDynamic("/api/health", env, FIXED_NOW)).toBe(true); + expect( + await isBypassPathDynamic("/internal/uptime", env, FIXED_NOW), + ).toBe(true); + }); +}); + +describe("maybeRateLimit honors KV bypass extras", () => { + beforeEach(() => { + __resetBypassExtrasCacheForTests(); + }); + + it("bypasses a runtime-whitelisted path", async () => { + const counters = new MemoryKV(fixedNow); + const config = new MemoryKV(fixedNow); + const env = buildEnv({ counters, config }); + // Tight limit so anything not bypassed would 429 on call #2. + await config.put(RATE_LIMIT_CONFIG_KEYS.PER_KEY_PER_MIN, "1"); + await config.put( + RATE_LIMIT_CONFIG_KEYS.BYPASS_PREFIXES_EXTRA, + "/api/k8s-readiness", + ); + + // Hit a non-bypassed path twice → second call 429s. + const url = "https://api.relayfile.dev/v1/workspaces/ws_a/sync/status"; + await maybeRateLimit(new Request(url), env, { now: fixedNow }); + const blocked = await maybeRateLimit(new Request(url), env, { + now: fixedNow, + }); + expect(blocked!.status).toBe(429); + + // Hit the runtime-whitelisted path 5 times → never 429s, regardless + // of the very tight per-key limit. + for (let i = 0; i < 5; i++) { + const r = await maybeRateLimit( + new Request("https://api.relayfile.dev/api/k8s-readiness"), + env, + { now: fixedNow }, + ); + expect(r).toBeNull(); + } + }); +}); diff --git a/router/test/recorder.test.ts b/router/test/recorder.test.ts new file mode 100644 index 00000000..bd59dc9f --- /dev/null +++ b/router/test/recorder.test.ts @@ -0,0 +1,192 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { maybeRecord, type RecorderEnv } from "../src/recorder.js"; + +const FIXED_TIME = new Date("2026-05-14T09:08:07.000Z"); +const REQUEST_ID = "123e4567-e89b-12d3-a456-426614174000"; +const KEY_PATTERN = /^corpus\/\d{4}\/\d{2}\/\d{2}\/\d{2}\/[a-f0-9-]+\.ndjson$/; + +function createEnv(sampleRate: string): { + env: RecorderEnv; + put: ReturnType; + get: ReturnType; +} { + const put = vi.fn(async () => undefined); + const get = vi.fn(async () => sampleRate); + + return { + env: { + TRAFFIC_RECORDER: { + put, + } as unknown as R2Bucket, + ROUTER_CONFIG: { + get, + } as unknown as KVNamespace, + }, + put, + get, + }; +} + +function createCtx(): { + ctx: Pick; + waitUntil: ReturnType; + waits: Promise[]; +} { + const waits: Promise[] = []; + const waitUntil = vi.fn((promise: Promise) => { + waits.push(promise); + }); + + return { + ctx: { waitUntil }, + waitUntil, + waits, + }; +} + +describe("maybeRecord", () => { + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(FIXED_TIME); + vi.spyOn(Math, "random").mockReturnValue(0); + }); + + afterEach(() => { + vi.restoreAllMocks(); + vi.useRealTimers(); + }); + + it("writes a record when sampling is 100 percent", async () => { + const { env, put, get } = createEnv("100"); + const { ctx, waitUntil, waits } = createCtx(); + const request = new Request("https://router.test/api/v1/messages?foo=bar", { + method: "POST", + headers: { + "content-type": "application/json", + authorization: "Bearer secret-token", + "x-request-id": REQUEST_ID, + }, + body: JSON.stringify({ + prompt: "hello", + token: "secret-token", + }), + }); + const response = new Response( + JSON.stringify({ + ok: true, + refreshToken: "refresh-me", + }), + { + status: 201, + headers: { + "content-type": "application/json", + "set-cookie": "sid=123", + }, + }, + ); + + await maybeRecord(request, response, env, ctx); + await Promise.all(waits); + + expect(get).toHaveBeenCalledWith("RECORDER_SAMPLE_RATE"); + expect(waitUntil).toHaveBeenCalledOnce(); + expect(put).toHaveBeenCalledOnce(); + + const [key, line] = put.mock.calls[0] as [string, string]; + expect(key).toBe("corpus/2026/05/14/09/123e4567-e89b-12d3-a456-426614174000.ndjson"); + expect(key).toMatch(KEY_PATTERN); + + const record = JSON.parse(line) as Record; + expect(record).toEqual({ + ts: "2026-05-14T09:08:07.000Z", + method: "POST", + path: "/api/v1/messages", + query: "foo=bar", + request_headers: { + authorization: "[REDACTED:19]", + "content-type": "application/json", + "x-request-id": REQUEST_ID, + }, + request_body: JSON.stringify({ + prompt: "hello", + token: "[REDACTED:12]", + }), + response_status: 201, + response_headers: { + "content-type": "application/json", + "set-cookie": "[REDACTED:7]", + }, + response_body: JSON.stringify({ + ok: true, + refreshToken: "[REDACTED:10]", + }), + request_id: REQUEST_ID, + }); + }); + + it("does not write when sampling is 0 percent", async () => { + const { env, put } = createEnv("0"); + const { ctx, waitUntil } = createCtx(); + + await maybeRecord( + new Request("https://router.test/api/v1/messages", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ ok: true }), + }), + new Response("ok"), + env, + ctx, + ); + + expect(waitUntil).not.toHaveBeenCalled(); + expect(put).not.toHaveBeenCalled(); + }); + + it("skips deny-listed paths without touching storage", async () => { + const { env, put, get } = createEnv("100"); + const { ctx, waitUntil } = createCtx(); + + await maybeRecord( + new Request("https://router.test/observer/session/123"), + new Response("ok"), + env, + ctx, + ); + + expect(get).not.toHaveBeenCalled(); + expect(waitUntil).not.toHaveBeenCalled(); + expect(put).not.toHaveBeenCalled(); + }); + + it("replaces oversize bodies with a placeholder before writing", async () => { + const { env, put } = createEnv("100"); + const { ctx, waits } = createCtx(); + const oversized = "x".repeat(256 * 1024 + 1); + + await maybeRecord( + new Request("https://router.test/api/v1/messages", { + method: "POST", + headers: { + "content-type": "text/plain", + "x-request-id": REQUEST_ID, + }, + body: oversized, + }), + new Response("ok", { + headers: { "content-type": "text/plain" }, + }), + env, + ctx, + ); + await Promise.all(waits); + + expect(put).toHaveBeenCalledOnce(); + const [, line] = put.mock.calls[0] as [string, string]; + const record = JSON.parse(line) as { request_body: string; response_body: string }; + + expect(record.request_body).toBe("[OVERSIZE:262145]"); + expect(record.response_body).toBe("ok"); + }); +}); diff --git a/router/test/redact.test.ts b/router/test/redact.test.ts new file mode 100644 index 00000000..13a79c19 --- /dev/null +++ b/router/test/redact.test.ts @@ -0,0 +1,88 @@ +import { describe, expect, it } from "vitest"; + +import { redactBody, redactHeaders, shouldRecord } from "../src/redact.js"; + +describe("router redaction", () => { + it("redacts deny-listed and sensitive headers", () => { + const headers = redactHeaders({ + authorization: "Bearer secret-token", + "x-api-key": "topsecret", + "x-signature": "sig-123", + "content-type": "application/json", + }); + + expect(headers).toEqual({ + authorization: "[REDACTED:19]", + "x-api-key": "[REDACTED:9]", + "x-signature": "[REDACTED:7]", + "content-type": "application/json", + }); + }); + + it("fully redacts bodies for deny-listed paths", () => { + expect( + redactBody( + "/api/v1/workspaces/ws_123/secrets", + JSON.stringify({ apiKey: "secret-value" }), + "application/json", + ), + ).toBe("[REDACTED:25]"); + }); + + it("fully redacts the recorder transcript ingest body (with and without /cloud prefix)", () => { + const body = JSON.stringify({ transcript_text: "private meeting words", summary_text: "x" }); + expect(redactBody("/api/v1/webhooks/transcripts", body, "application/json")).toBe( + `[REDACTED:${body.length}]`, + ); + expect(redactBody("/cloud/api/v1/webhooks/transcripts", body, "application/json")).toBe( + `[REDACTED:${body.length}]`, + ); + }); + + it("scrubs sensitive keys from JSON bodies", () => { + expect( + redactBody( + "/api/v1/messages", + JSON.stringify({ + ok: true, + nested: { + token: "secret-token", + profile: { + password: "swordfish", + }, + }, + items: [ + { + refreshToken: "refresh-me", + }, + ], + }), + "application/json; charset=utf-8", + ), + ).toBe( + JSON.stringify({ + ok: true, + nested: { + token: "[REDACTED:12]", + profile: { + password: "[REDACTED:9]", + }, + }, + items: [ + { + refreshToken: "[REDACTED:10]", + }, + ], + }), + ); + }); + + it("applies shouldRecord route rules", () => { + expect(shouldRecord("/favicon.ico")).toBe(false); + expect(shouldRecord("/api/health")).toBe(false); + expect(shouldRecord("/observer")).toBe(false); + expect(shouldRecord("/observer/session/123")).toBe(false); + expect(shouldRecord("/observer-files")).toBe(true); + expect(shouldRecord("/cloud/dashboard")).toBe(true); + }); +}); diff --git a/router/test/webhook-routing.test.ts b/router/test/webhook-routing.test.ts new file mode 100644 index 00000000..8daa494d --- /dev/null +++ b/router/test/webhook-routing.test.ts @@ -0,0 +1,157 @@ +import { describe, expect, it, vi } from "vitest"; + +import worker from "../index.js"; + +type StoredEntry = { value: string }; + +class MemoryKV { + readonly store = new Map(); + + async get(key: string): Promise { + return this.store.get(key)?.value ?? null; + } + + async put(key: string, value: string): Promise { + this.store.set(key, { value }); + } +} + +function asKV(kv: MemoryKV): KVNamespace { + return kv as unknown as KVNamespace; +} + +type WorkerBinding = { fetch: ReturnType }; + +function makeBinding(): WorkerBinding { + return { + fetch: vi.fn(async () => new Response("ok", { status: 200 })), + }; +} + +function buildEnv(opts: { + webhookOriginFlag?: string; + cloudWebWorker?: WorkerBinding; + webhookWorker?: WorkerBinding; +}) { + const routerConfig = new MemoryKV(); + if (opts.webhookOriginFlag !== undefined) { + routerConfig.store.set("WEBHOOK_ORIGIN", { value: opts.webhookOriginFlag }); + } + return { + CLOUD_APP_ORIGIN: "https://origin.test.invalid", + ROUTER_CONFIG: asKV(routerConfig), + CLOUD_WEB_WORKER: opts.cloudWebWorker, + WEBHOOK_WORKER: opts.webhookWorker, + } as unknown as Parameters[1]; +} + +function buildCtx(): ExecutionContext { + return { + waitUntil: () => undefined, + passThroughOnException: () => undefined, + } as unknown as ExecutionContext; +} + +describe("router webhook routing", () => { + it("routes /api/v1/webhooks/nango to the webhook-worker binding when WEBHOOK_ORIGIN=worker", async () => { + const cloudWebWorker = makeBinding(); + const webhookWorker = makeBinding(); + const env = buildEnv({ + webhookOriginFlag: "worker", + cloudWebWorker, + webhookWorker, + }); + + const request = new Request( + "https://origin.agentrelay.cloud/cloud/api/v1/webhooks/nango", + { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ type: "forward", from: "github-app-oauth" }), + }, + ); + + await worker.fetch(request, env, buildCtx()); + + expect(webhookWorker.fetch).toHaveBeenCalledOnce(); + expect(cloudWebWorker.fetch).not.toHaveBeenCalled(); + }); + + it("bypasses webhook-worker when request carries x-cloud-webhook-worker-forwarded header (loop break)", async () => { + const cloudWebWorker = makeBinding(); + const webhookWorker = makeBinding(); + const env = buildEnv({ + webhookOriginFlag: "worker", + cloudWebWorker, + webhookWorker, + }); + + const request = new Request( + "https://origin.agentrelay.cloud/cloud/api/v1/webhooks/nango", + { + method: "POST", + headers: { + "content-type": "application/json", + "x-cloud-webhook-worker-forwarded": "webhook-worker", + "x-cloud-webhook-worker-request-id": "req-loop-1", + }, + body: JSON.stringify({ type: "forward", from: "github-app-oauth" }), + }, + ); + + await worker.fetch(request, env, buildCtx()); + + expect(cloudWebWorker.fetch).toHaveBeenCalledOnce(); + expect(webhookWorker.fetch).not.toHaveBeenCalled(); + }); + + it("ignores the loop-break header for an unrelated value (don't accidentally widen bypass)", async () => { + const cloudWebWorker = makeBinding(); + const webhookWorker = makeBinding(); + const env = buildEnv({ + webhookOriginFlag: "worker", + cloudWebWorker, + webhookWorker, + }); + + const request = new Request( + "https://origin.agentrelay.cloud/cloud/api/v1/webhooks/nango", + { + method: "POST", + headers: { + "content-type": "application/json", + "x-cloud-webhook-worker-forwarded": "something-else", + }, + body: JSON.stringify({ type: "forward", from: "github-app-oauth" }), + }, + ); + + await worker.fetch(request, env, buildCtx()); + + expect(webhookWorker.fetch).toHaveBeenCalledOnce(); + expect(cloudWebWorker.fetch).not.toHaveBeenCalled(); + }); + + it("routes to cloud-web when WEBHOOK_ORIGIN flag is unset (regression: still works without flag)", async () => { + const cloudWebWorker = makeBinding(); + const webhookWorker = makeBinding(); + const env = buildEnv({ + cloudWebWorker, + webhookWorker, + }); + + const request = new Request( + "https://origin.agentrelay.cloud/cloud/api/v1/webhooks/nango", + { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ type: "forward", from: "github-app-oauth" }), + }, + ); + + await worker.fetch(request, env, buildCtx()); + + expect(cloudWebWorker.fetch).toHaveBeenCalledOnce(); + expect(webhookWorker.fetch).not.toHaveBeenCalled(); + }); +}); diff --git a/router/tsconfig.json b/router/tsconfig.json new file mode 100644 index 00000000..00c428ac --- /dev/null +++ b/router/tsconfig.json @@ -0,0 +1,17 @@ +{ + "compilerOptions": { + "target": "ES2022", + "module": "ES2022", + "moduleResolution": "bundler", + "lib": ["ES2022", "WebWorker"], + "strict": true, + "noEmit": true, + "skipLibCheck": true, + "esModuleInterop": true, + "resolveJsonModule": true, + "isolatedModules": true, + "types": ["@cloudflare/workers-types", "vitest/globals"] + }, + "include": ["index.ts", "src/**/*.ts", "test/**/*.ts"], + "exclude": ["node_modules", "dist"] +} diff --git a/router/wrangler.jsonc b/router/wrangler.jsonc new file mode 100644 index 00000000..8dd58c43 --- /dev/null +++ b/router/wrangler.jsonc @@ -0,0 +1,52 @@ +{ + "$schema": "../web/node_modules/wrangler/config-schema.json", + "name": "agentrelay-router", + "main": "index.ts", + "compatibility_date": "2025-04-01", + "compatibility_flags": ["nodejs_compat"], + + // The router fronts agentrelay.com. It binds resources that remain OWNED BY THE + // CLOUD SST APP in the same Cloudflare account — referenced here by stable + // name/id. These identifiers are cloud-owned; if cloud ever recreates the + // resource the value here must be updated (see the repo's Phase 2 notes). + + // Service bindings reference cloud's Workers by stable prod script name + // (no *.workers.dev URL to pin): /cloud* → the cloud Next.js Worker; + // WEBHOOK_WORKER → the dedicated webhook Worker the router can offload + // /api/v1/webhooks/* to when the ROUTER_CONFIG `WEBHOOK_ORIGIN` flag is "worker" + // (the router code prefers this binding over the WEBHOOK_WORKER_ORIGIN URL). + "services": [ + { "binding": "CLOUD_WEB_WORKER", "service": "cloud-web-worker" }, + { "binding": "WEBHOOK_WORKER", "service": "webhook-worker" } + ], + + // ROUTER_CONFIG: holds the WEBHOOK_ORIGIN ops flag + recorder config. + // RATE_LIMIT_COUNTERS: shared with the relayfile-api worker — MUST be the same + // namespace id so both gateways increment the same per-workspace counters. + // Look up the prod ids: `wrangler kv namespace list` (titles router-config, + // rate-limit-counters). + "kv_namespaces": [ + { "binding": "ROUTER_CONFIG", "id": "e73a3fc0c2514460bdd3afae463e5234" }, + { "binding": "RATE_LIMIT_COUNTERS", "id": "40c2049dd4ce40f4b8f8143d53075731" } + ], + + // Replay/traffic recorder corpus (14-day lifecycle, owned by cloud). + "r2_buckets": [ + { "binding": "TRAFFIC_RECORDER", "bucket_name": "traffic-recorder" } + ], + + "vars": { + "CLOUD_APP_ORIGIN": "https://origin.agentrelay.cloud", + "FILE_OBSERVER_ORIGIN": "https://relayfile-file-observer.pages.dev" + }, + + // Start on a staging hostname for parity verification against the live apex; + // flip "pattern" to "agentrelay.com" at the cutover step. + "routes": [ + { "pattern": "router-next.agentrelay.com", "custom_domain": true } + ], + + "workers_dev": false, + "preview_urls": true, + "observability": { "enabled": true } +}