diff --git a/apps/agent/agent/lib/tasks.ts b/apps/agent/agent/lib/tasks.ts index 9d8a912c7..c1fa3ec49 100644 --- a/apps/agent/agent/lib/tasks.ts +++ b/apps/agent/agent/lib/tasks.ts @@ -42,11 +42,7 @@ export async function claimDue( const onlyMode = "only" in kinds; const claimed = await db.$queryRaw` - UPDATE "agentTask" AS t - SET "leasedUntil" = ${until}, - "startedAt" = COALESCE(t."startedAt", ${now}), - "attempts" = t."attempts" + 1 - FROM ( + WITH due AS MATERIALIZED ( SELECT t2.id FROM "agentTask" AS t2 WHERE t2."finishedAt" IS NULL AND t2."dueAt" <= ${now} @@ -59,7 +55,12 @@ export async function claimDue( ORDER BY t2."priority" DESC, t2."dueAt" ASC LIMIT ${limit} FOR UPDATE SKIP LOCKED - ) AS due + ) + UPDATE "agentTask" AS t + SET "leasedUntil" = ${until}, + "startedAt" = COALESCE(t."startedAt", ${now}), + "attempts" = t."attempts" + 1 + FROM due WHERE t.id = due.id RETURNING t.id, t."contactId", t."companyId", t."dealId", t.kind, t.reason, t.payload, t.budget, t.attempts, t.priority, t."dueAt"; @@ -72,23 +73,30 @@ export async function claimDue( export async function retireExhausted( limit: number = DISPATCH.reconcile.retire, + kinds?: { only: readonly string[] }, ): Promise { + const only = kinds?.only ?? []; + if (kinds && only.length === 0) return []; + const now = new Date(); return db.$queryRaw` - UPDATE "agentTask" AS t - SET "finishedAt" = ${now}, - "outcome" = ${RETIRED_OUTCOME} - WHERE t.id IN ( + WITH exhausted AS MATERIALIZED ( SELECT c.id FROM "agentTask" AS c WHERE c."finishedAt" IS NULL AND c."attempts" >= ${MAX_ATTEMPTS} AND (c."leasedUntil" IS NULL OR c."leasedUntil" < ${now}) + AND (${only}::text[] = '{}' OR c.kind = ANY(${only}::text[])) ORDER BY c."dueAt" ASC LIMIT ${limit} FOR UPDATE SKIP LOCKED ) + UPDATE "agentTask" AS t + SET "finishedAt" = ${now}, + "outcome" = ${RETIRED_OUTCOME} + FROM exhausted + WHERE t.id = exhausted.id RETURNING t.id, t."contactId", t."companyId", t."dealId", t.kind; `; } diff --git a/apps/agent/test/tasks.integration.spec.ts b/apps/agent/test/tasks.integration.spec.ts index 3100f0c1f..7ebf8fdfb 100644 --- a/apps/agent/test/tasks.integration.spec.ts +++ b/apps/agent/test/tasks.integration.spec.ts @@ -1,6 +1,5 @@ import { afterEach, beforeEach, describe, expect, it } from "bun:test"; import { db } from "@crm/db"; -import { DIRECT_KINDS } from "@crm/db/agent-tasks"; import { claimDue, completeTask, @@ -9,9 +8,14 @@ import { scheduleTask, } from "../agent/lib/tasks"; -const kind = "test-lease"; +const suffix = process.env.TEST_RUN_ID ?? "tasks-integration-spec"; +const kind = `test-lease-${suffix}`; -const RESEARCH = { except: DIRECT_KINDS } as const; +const RESEARCH = { only: [kind] } as const; + +function retire(limit?: number) { + return retireExhausted(limit, RESEARCH); +} async function clear() { await db.agentTask.deleteMany({ where: { kind } }); @@ -44,6 +48,13 @@ async function expire(taskId: string) { }); } +async function exhaust() { + for (let attempt = 0; attempt < MAX_ATTEMPTS; attempt++) { + const claimed = await claimDue(10, RESEARCH); + for (const task of claimed) await expire(task.id); + } +} + async function someone() { return db.contact.create({ data: { @@ -141,6 +152,27 @@ describe("claimDue", () => { expect(await claimDue(10, RESEARCH)).toHaveLength(0); }); + + it("gives each due row to one caller and no caller more than its limit", async () => { + const mine: string[] = []; + for (let row = 0; row < 8; row++) mine.push((await queue()).id); + + const results = await Promise.all( + Array.from({ length: 4 }, () => claimDue(2, RESEARCH)), + ); + for (const claimed of results) { + expect(claimed.length).toBeLessThanOrEqual(2); + } + + const ids = results.flatMap((claimed) => claimed.map((task) => task.id)); + expect(new Set(ids).size).toBe(ids.length); + for (const id of ids) expect(mine).toContain(id); + + const leased = await db.agentTask.count({ + where: { id: { in: mine }, attempts: 1, leasedUntil: { not: null } }, + }); + expect(leased).toBe(ids.length); + }); }); describe("retireExhausted", () => { @@ -153,7 +185,7 @@ describe("retireExhausted", () => { await expire(task.id); } - const retired = await retireExhausted(); + const retired = await retire(); expect(retired.map((t) => t.id)).toContain(task.id); expect(retired.find((t) => t.id === task.id)?.contactId).toBe(contact.id); @@ -170,14 +202,14 @@ describe("retireExhausted", () => { if (attempt < MAX_ATTEMPTS - 1) await expire(task.id); } - expect(await retireExhausted()).toHaveLength(0); + expect(await retire()).toHaveLength(0); }); it("leaves work that still has attempts left", async () => { await queue(); await claimDue(10, RESEARCH); - expect(await retireExhausted()).toHaveLength(0); + expect(await retire()).toHaveLength(0); }); it("retires no more rows than the limit allows", async () => { @@ -190,7 +222,7 @@ describe("retireExhausted", () => { } for (let pass = 0; pass < 3; pass++) { - expect((await retireExhausted(2)).length).toBeLessThanOrEqual(2); + expect((await retire(2)).length).toBeLessThanOrEqual(2); } const open = await db.agentTask.count({ @@ -198,6 +230,29 @@ describe("retireExhausted", () => { }); expect(open).toBe(0); }); + + it("gives each exhausted row to one caller and no caller more than its limit", async () => { + const mine: string[] = []; + for (let row = 0; row < 8; row++) mine.push((await queue()).id); + + await exhaust(); + + const results = await Promise.all( + Array.from({ length: 4 }, () => retire(2)), + ); + for (const retired of results) { + expect(retired.length).toBeLessThanOrEqual(2); + } + + const ids = results.flatMap((retired) => retired.map((task) => task.id)); + expect(new Set(ids).size).toBe(ids.length); + for (const id of ids) expect(mine).toContain(id); + + const open = await db.agentTask.count({ + where: { id: { in: mine }, finishedAt: null }, + }); + expect(open).toBe(mine.length - ids.length); + }); }); describe("completeTask", () => { diff --git a/apps/api/package.json b/apps/api/package.json index 151aec0f8..7ae3c11d0 100644 --- a/apps/api/package.json +++ b/apps/api/package.json @@ -51,6 +51,7 @@ "zod": "^4.4.3" }, "devDependencies": { + "@better-auth/api-key": "1.6.25", "@crm/typescript-config": "workspace:*", "@nestjs/testing": "^11.0.1", "@types/bun": "^1.3.14", diff --git a/apps/api/src/archive/archive-retention.controller.ts b/apps/api/src/archive/archive-retention.controller.ts index 5050b43df..d1b8d3d85 100644 --- a/apps/api/src/archive/archive-retention.controller.ts +++ b/apps/api/src/archive/archive-retention.controller.ts @@ -20,6 +20,7 @@ import { ApiTags, } from "@nestjs/swagger"; import { AllowAnonymous } from "@thallesp/nestjs-better-auth"; +import { timingSafeEquals } from "../common/timing-safe-equal"; import { CompaniesService } from "../companies/companies.service"; import type { EnvironmentVariables } from "../config/env.validation"; import { ContactsService } from "../contacts/contacts.service"; @@ -103,14 +104,3 @@ export class ArchiveRetentionController { return { retentionDays, companies, contacts, deals }; } } - -function timingSafeEquals(a: string, b: string): boolean { - if (a.length !== b.length) return false; - - let mismatch = 0; - for (let index = 0; index < a.length; index += 1) { - mismatch |= a.charCodeAt(index) ^ b.charCodeAt(index); - } - - return mismatch === 0; -} diff --git a/apps/api/src/common/timing-safe-equal.ts b/apps/api/src/common/timing-safe-equal.ts new file mode 100644 index 000000000..31373ce2b --- /dev/null +++ b/apps/api/src/common/timing-safe-equal.ts @@ -0,0 +1,10 @@ +import { timingSafeEqual } from "node:crypto"; + +export function timingSafeEquals(a: string, b: string): boolean { + const bufferA = Buffer.from(a); + const bufferB = Buffer.from(b); + + if (bufferA.length !== bufferB.length) return false; + + return timingSafeEqual(bufferA, bufferB); +} diff --git a/apps/api/src/config/env.validation.ts b/apps/api/src/config/env.validation.ts index 08cb676c2..984fd3dad 100644 --- a/apps/api/src/config/env.validation.ts +++ b/apps/api/src/config/env.validation.ts @@ -8,6 +8,7 @@ import { Max, Min, MinLength, + ValidateIf, validateSync, } from "class-validator"; @@ -120,7 +121,12 @@ export class EnvironmentVariables { AGENT_URL?: string; @IsOptional() + @ValidateIf((_environment, value) => value !== "") @IsString() + @MinLength(32, { + message: + "AGENT_BRIDGE_SECRET must be at least 32 characters. Generate one with: openssl rand -base64 32", + }) AGENT_BRIDGE_SECRET?: string; @IsOptional() diff --git a/apps/api/src/create-app.ts b/apps/api/src/create-app.ts index 4a436e71c..267c3202e 100644 --- a/apps/api/src/create-app.ts +++ b/apps/api/src/create-app.ts @@ -109,7 +109,7 @@ export async function createApp(): Promise { restBridge = createOpenApiExpressMiddleware({ router: appRouter, - createContext: ({ req }) => createBaseTrpcContext(req), + createContext: ({ req, res }) => createBaseTrpcContext(req, res), }); return app; diff --git a/apps/api/src/currency/rates.controller.ts b/apps/api/src/currency/rates.controller.ts index 1f99af8f5..eb54d07f6 100644 --- a/apps/api/src/currency/rates.controller.ts +++ b/apps/api/src/currency/rates.controller.ts @@ -18,6 +18,7 @@ import { ApiTags, } from "@nestjs/swagger"; import { AllowAnonymous } from "@thallesp/nestjs-better-auth"; +import { timingSafeEquals } from "../common/timing-safe-equal"; import type { EnvironmentVariables } from "../config/env.validation"; import { ConversionService } from "./conversion.service"; import { RatesService } from "./rates.service"; @@ -81,14 +82,3 @@ export class RatesController { return { ...refresh, converted: filled.converted, missing: filled.missing }; } } - -function timingSafeEquals(a: string, b: string): boolean { - if (a.length !== b.length) return false; - - let mismatch = 0; - for (let index = 0; index < a.length; index += 1) { - mismatch |= a.charCodeAt(index) ^ b.charCodeAt(index); - } - - return mismatch === 0; -} diff --git a/apps/api/src/logging/all-exceptions.filter.ts b/apps/api/src/logging/all-exceptions.filter.ts index fb8f46038..17b7b8a63 100644 --- a/apps/api/src/logging/all-exceptions.filter.ts +++ b/apps/api/src/logging/all-exceptions.filter.ts @@ -42,7 +42,7 @@ export class AllExceptionsFilter implements ExceptionFilter { const payload = { message: describe(exception), method: request.method, - path: request.originalUrl, + path: request.path, statusCode: status, exception: exception?.constructor?.name, }; diff --git a/apps/api/src/logging/request-logger.middleware.ts b/apps/api/src/logging/request-logger.middleware.ts index 699549fec..ca876bf29 100644 --- a/apps/api/src/logging/request-logger.middleware.ts +++ b/apps/api/src/logging/request-logger.middleware.ts @@ -1,4 +1,5 @@ import { randomUUID } from "node:crypto"; +import { API_KEY_HEADER } from "@crm/auth"; import { Injectable, Logger, type NestMiddleware } from "@nestjs/common"; import type { NextFunction, Request, Response } from "express"; import { type RequestContext, runInRequestContext } from "./request-context"; @@ -18,7 +19,7 @@ export class RequestLoggerMiddleware implements NestMiddleware { const context: RequestContext = { requestId, method: request.method, - path: request.originalUrl, + path: request.path, }; const startedAt = process.hrtime.bigint(); @@ -56,6 +57,7 @@ export class RequestLoggerMiddleware implements NestMiddleware { durationMs: Number(durationMs.toFixed(1)), ip: request.ip, userAgent: request.get("user-agent"), + apiKeyPresented: Boolean(request.get(API_KEY_HEADER)), }; if (statusCode >= 500) { diff --git a/apps/api/src/sync/sync.controller.ts b/apps/api/src/sync/sync.controller.ts index 7464f607c..7ab614eac 100644 --- a/apps/api/src/sync/sync.controller.ts +++ b/apps/api/src/sync/sync.controller.ts @@ -18,6 +18,7 @@ import { ApiTags, } from "@nestjs/swagger"; import { AllowAnonymous } from "@thallesp/nestjs-better-auth"; +import { timingSafeEquals } from "../common/timing-safe-equal"; import type { EnvironmentVariables } from "../config/env.validation"; import { MailboxSyncService } from "./mailbox-sync.service"; @@ -88,14 +89,3 @@ export class SyncController { return this.sync.runDue(); } } - -function timingSafeEquals(a: string, b: string): boolean { - if (a.length !== b.length) return false; - - let mismatch = 0; - for (let index = 0; index < a.length; index += 1) { - mismatch |= a.charCodeAt(index) ^ b.charCodeAt(index); - } - - return mismatch === 0; -} diff --git a/apps/api/src/telemetry/rollup.service.ts b/apps/api/src/telemetry/rollup.service.ts index c53b77a43..ad4787024 100644 --- a/apps/api/src/telemetry/rollup.service.ts +++ b/apps/api/src/telemetry/rollup.service.ts @@ -128,7 +128,13 @@ export class RollupService { ]); return { - properties: { ...shape, ...agent, ...ledger, ...crm }, + properties: { + ...shape, + ...agent, + ...ledger, + ...crm, + api_key_rate_limited: counters.api_key_rate_limited ?? 0, + }, counters, }; } diff --git a/apps/api/src/telemetry/telemetry.controller.ts b/apps/api/src/telemetry/telemetry.controller.ts index 90ba0aaa7..df6fe8484 100644 --- a/apps/api/src/telemetry/telemetry.controller.ts +++ b/apps/api/src/telemetry/telemetry.controller.ts @@ -18,6 +18,7 @@ import { ApiTags, } from "@nestjs/swagger"; import { AllowAnonymous } from "@thallesp/nestjs-better-auth"; +import { timingSafeEquals } from "../common/timing-safe-equal"; import type { EnvironmentVariables } from "../config/env.validation"; import { RollupService } from "./rollup.service"; @@ -71,14 +72,3 @@ export class TelemetryController { return this.rollup.run(); } } - -function timingSafeEquals(a: string, b: string): boolean { - if (a.length !== b.length) return false; - - let mismatch = 0; - for (let index = 0; index < a.length; index += 1) { - mismatch |= a.charCodeAt(index) ^ b.charCodeAt(index); - } - - return mismatch === 0; -} diff --git a/apps/api/src/tracking/tracking.controller.ts b/apps/api/src/tracking/tracking.controller.ts index 2a3603e79..a6c1b0de1 100644 --- a/apps/api/src/tracking/tracking.controller.ts +++ b/apps/api/src/tracking/tracking.controller.ts @@ -33,6 +33,7 @@ import { import { AllowAnonymous } from "@thallesp/nestjs-better-auth"; import type { Response } from "express"; import { z } from "zod"; +import { timingSafeEquals } from "../common/timing-safe-equal"; import type { EnvironmentVariables } from "../config/env.validation"; import { InjectDatabase } from "../database/database.constants"; import { TrackingConfigService } from "./tracking-config.service"; @@ -297,14 +298,3 @@ function startOfDay(at: Date): Date { return day; } - -function timingSafeEquals(a: string, b: string): boolean { - if (a.length !== b.length) return false; - - let mismatch = 0; - for (let index = 0; index < a.length; index += 1) { - mismatch |= a.charCodeAt(index) ^ b.charCodeAt(index); - } - - return mismatch === 0; -} diff --git a/apps/api/src/trpc/trpc.context.ts b/apps/api/src/trpc/trpc.context.ts index 873d5a801..2cb573154 100644 --- a/apps/api/src/trpc/trpc.context.ts +++ b/apps/api/src/trpc/trpc.context.ts @@ -1,25 +1,88 @@ import { auth } from "@crm/auth"; +import { bumpCounter, COUNTERS } from "@crm/telemetry"; import { Injectable } from "@nestjs/common"; +import { TRPCError } from "@trpc/server"; import { fromNodeHeaders } from "better-auth/node"; -import type { Request } from "express"; +import type { Request, Response } from "express"; import type { ContextOptions, TRPCContext } from "nestjs-trpc"; +import { z } from "zod"; import type { BaseTrpcContext } from "./context.types"; +const betterAuthApiErrorShape = z + .object({ + name: z.literal("APIError"), + status: z.string(), + message: z.string().optional(), + body: z + .object({ + details: z + .object({ tryAgainIn: z.number().optional() }) + .nullable() + .optional(), + }) + .nullable() + .optional(), + }) + .nullable() + .catch(null); + +const AUTHENTICATION_FAILURE_STATUSES = new Set(["UNAUTHORIZED", "FORBIDDEN"]); + +const RETRY_AFTER_HEADER = "Retry-After"; + +const MS_PER_SECOND = 1000; + +export function retryAfterSeconds( + tryAgainInMs: number | undefined, +): number | null { + if (tryAgainInMs === undefined || !Number.isFinite(tryAgainInMs)) return null; + + return Math.max(0, Math.ceil(tryAgainInMs / MS_PER_SECOND)); +} + export async function createBaseTrpcContext( req: Request | undefined, + res?: Response, ): Promise { - const session = req - ? await auth.api - .getSession({ headers: fromNodeHeaders(req.headers) }) - .catch(() => null) - : null; - return { req, session }; + if (!req) { + return { req, session: null }; + } + + try { + const session = await auth.api.getSession({ + headers: fromNodeHeaders(req.headers), + }); + return { req, session }; + } catch (error) { + const apiError = betterAuthApiErrorShape.parse(error); + + if (apiError?.status === "TOO_MANY_REQUESTS") { + void bumpCounter(COUNTERS.apiKeyRateLimited); + + const seconds = retryAfterSeconds(apiError.body?.details?.tryAgainIn); + if (seconds !== null) { + res?.setHeader(RETRY_AFTER_HEADER, String(seconds)); + } + + throw new TRPCError({ + code: "TOO_MANY_REQUESTS", + message: apiError.message ?? "Too many requests.", + }); + } + + if (apiError && AUTHENTICATION_FAILURE_STATUSES.has(apiError.status)) { + return { req, session: null }; + } + + throw error; + } } @Injectable() export class TrpcContext implements TRPCContext { async create(opts: ContextOptions): Promise { const req = "req" in opts ? opts.req : undefined; - return createBaseTrpcContext(req); + const res = "res" in opts ? opts.res : undefined; + return createBaseTrpcContext(req, res); } } diff --git a/apps/api/test/api-key-rate-limit.integration.spec.ts b/apps/api/test/api-key-rate-limit.integration.spec.ts new file mode 100644 index 000000000..9de51a3e1 --- /dev/null +++ b/apps/api/test/api-key-rate-limit.integration.spec.ts @@ -0,0 +1,204 @@ +import { afterAll, beforeAll, describe, expect, it } from "bun:test"; +import { db } from "@crm/db"; +import { backfillApiKeyRateLimit } from "@crm/db/api-key-rate-limit"; +import { + type RateLimitedAuth, + rateLimitedAuth, +} from "./support/rate-limited-auth"; + +const suffix = process.env.TEST_RUN_ID ?? "api-key-rate-limit-spec"; +const userId = `user-${suffix}`; + +async function issueKey( + auth: RateLimitedAuth, + name: string, +): Promise<{ id: string; key: string }> { + const created = await auth.api.createApiKey({ + body: { userId, name }, + }); + + return { id: created.id, key: created.key }; +} + +function headersWithKey(key: string): Headers { + return new Headers({ "x-api-key": key }); +} + +async function deniedStatus( + auth: RateLimitedAuth, + key: string, +): Promise { + try { + await auth.api.getSession({ headers: headersWithKey(key) }); + return undefined; + } catch (error) { + return (error as { status?: string }).status; + } +} + +beforeAll(async () => { + await db.apikey.deleteMany({ where: { referenceId: userId } }); + await db.user.deleteMany({ where: { id: userId } }); + await db.user.create({ + data: { + id: userId, + name: "Rate Limit Test", + email: `${userId}@example.test`, + }, + }); +}); + +afterAll(async () => { + await db.apikey.deleteMany({ where: { referenceId: userId } }); + await db.user.deleteMany({ where: { id: userId } }); +}); + +describe("apiKey plugin rate limiting (isolated test instance, crm_test only)", () => { + it("allows requests at or under the limit", async () => { + const auth = rateLimitedAuth({ maxRequests: 3, timeWindowMs: 60_000 }); + const { key } = await issueKey(auth, "under-limit"); + + for (let attempt = 0; attempt < 3; attempt += 1) { + const session = await auth.api.getSession({ + headers: headersWithKey(key), + }); + expect(session?.user.id).toBe(userId); + } + }); + + it("denies the request that exceeds the limit with TOO_MANY_REQUESTS", async () => { + const auth = rateLimitedAuth({ maxRequests: 2, timeWindowMs: 60_000 }); + const { key } = await issueKey(auth, "over-limit"); + + await auth.api.getSession({ headers: headersWithKey(key) }); + await auth.api.getSession({ headers: headersWithKey(key) }); + + const status = await deniedStatus(auth, key); + expect(status).toBe("TOO_MANY_REQUESTS"); + }); + + it("carries a positive tryAgainIn on the rejection", async () => { + const auth = rateLimitedAuth({ maxRequests: 1, timeWindowMs: 60_000 }); + const { key } = await issueKey(auth, "retry-after"); + + await auth.api.getSession({ headers: headersWithKey(key) }); + + const error = await auth.api + .getSession({ headers: headersWithKey(key) }) + .then( + () => null, + (error) => error as { body?: { details?: { tryAgainIn?: number } } }, + ); + expect(error).not.toBeNull(); + expect(error?.body?.details?.tryAgainIn).toBeGreaterThan(0); + }); + + it("never lets the stored counter exceed the limit under concurrency", async () => { + const maxRequests = 5; + const auth = rateLimitedAuth({ maxRequests, timeWindowMs: 60_000 }); + const { id, key } = await issueKey(auth, "concurrency"); + + const attempts = 10; + const results = await Promise.allSettled( + Array.from({ length: attempts }, () => + auth.api.getSession({ headers: headersWithKey(key) }), + ), + ); + + const succeeded = results.filter((r) => r.status === "fulfilled").length; + const denied = results.filter( + (r) => + r.status === "rejected" && + (r.reason as { status?: string })?.status === "TOO_MANY_REQUESTS", + ).length; + + expect(succeeded).toBeLessThanOrEqual(maxRequests); + expect(succeeded + denied).toBe(attempts); + + const row = await db.apikey.findUniqueOrThrow({ + where: { id }, + select: { requestCount: true }, + }); + expect(row.requestCount ?? 0).toBeLessThanOrEqual(maxRequests); + }, 15_000); +}); + +describe("backfillApiKeyRateLimit (crm_test only, never run against crm)", () => { + it("applies the policy only to keys with no policy yet", async () => { + const auth = rateLimitedAuth({ maxRequests: 10, timeWindowMs: 60_000 }); + const legacy = await issueKey(auth, "legacy-no-policy"); + + await db.apikey.update({ + where: { id: legacy.id }, + data: { + rateLimitEnabled: false, + rateLimitMax: null, + rateLimitTimeWindow: null, + }, + }); + + const before = await db.apikey.findUniqueOrThrow({ + where: { id: legacy.id }, + select: { rateLimitEnabled: true, rateLimitMax: true }, + }); + expect(before.rateLimitEnabled).toBe(false); + expect(before.rateLimitMax).toBeNull(); + + const first = await backfillApiKeyRateLimit(db, { + maxRequests: 7, + timeWindowMs: 3_600_000, + }); + expect(first.updated).toBeGreaterThanOrEqual(1); + + const after = await db.apikey.findUniqueOrThrow({ + where: { id: legacy.id }, + select: { + rateLimitEnabled: true, + rateLimitMax: true, + rateLimitTimeWindow: true, + }, + }); + expect(after.rateLimitEnabled).toBe(true); + expect(after.rateLimitMax).toBe(7); + expect(after.rateLimitTimeWindow).toBe(3_600_000); + + const second = await backfillApiKeyRateLimit(db, { + maxRequests: 999, + timeWindowMs: 1, + }); + expect(second.alreadyPolicied).toBeGreaterThan(0); + + const unchanged = await db.apikey.findUniqueOrThrow({ + where: { id: legacy.id }, + select: { rateLimitMax: true }, + }); + expect(unchanged.rateLimitMax).toBe(7); + }); + + it("rejects invalid policies before writing any keys", async () => { + await expect( + backfillApiKeyRateLimit(db, { maxRequests: 0, timeWindowMs: 1 }), + ).rejects.toThrow(RangeError); + await expect( + backfillApiKeyRateLimit(db, { maxRequests: 1, timeWindowMs: -1 }), + ).rejects.toThrow(RangeError); + }); + + it("leaves an already-policied key's limit untouched", async () => { + const auth = rateLimitedAuth({ maxRequests: 4, timeWindowMs: 5_000 }); + const configured = await issueKey(auth, "already-policied"); + + const result = await backfillApiKeyRateLimit(db, { + maxRequests: 999, + timeWindowMs: 1, + }); + expect(result.updated).toBe(0); + + const row = await db.apikey.findUniqueOrThrow({ + where: { id: configured.id }, + select: { rateLimitMax: true, rateLimitTimeWindow: true }, + }); + expect(row.rateLimitMax).toBe(4); + expect(row.rateLimitTimeWindow).toBe(5_000); + }); +}); diff --git a/apps/api/test/auth.e2e.spec.ts b/apps/api/test/auth.e2e.spec.ts index d5d79131d..14828a9ae 100644 --- a/apps/api/test/auth.e2e.spec.ts +++ b/apps/api/test/auth.e2e.spec.ts @@ -1,8 +1,31 @@ -import { afterAll, beforeAll, describe, expect, it } from "bun:test"; +import { + afterAll, + afterEach, + beforeAll, + describe, + expect, + it, + spyOn, +} from "bun:test"; import type { INestApplication } from "@nestjs/common"; import { Test, type TestingModule } from "@nestjs/testing"; import request from "supertest"; +function betterAuthError( + status: string, + message?: string, + tryAgainInMs?: number, +): Error { + const error = new Error(message ?? status); + error.name = "APIError"; + return Object.assign(error, { + status, + ...(tryAgainInMs !== undefined && { + body: { details: { tryAgainIn: tryAgainInMs } }, + }), + }); +} + const fallback = (key: string, value: string) => { if (!process.env[key]) { process.env[key] = value; @@ -21,9 +44,12 @@ fallback("GOOGLE_CLIENT_SECRET", "test-google-client-secret"); describe("Auth (e2e)", () => { let app: INestApplication; + let auth: typeof import("@crm/auth").auth; + let getSessionSpy: ReturnType | undefined; beforeAll(async () => { const { AppModule } = await import("../src/app.module"); + ({ auth } = await import("@crm/auth")); const moduleFixture: TestingModule = await Test.createTestingModule({ imports: [AppModule], @@ -37,6 +63,11 @@ describe("Auth (e2e)", () => { await app.close(); }); + afterEach(() => { + getSessionSpy?.mockRestore(); + getSessionSpy = undefined; + }); + it("rejects an unauthenticated request to a guarded route", async () => { await request(app.getHttpServer()).get("/auth/me").expect(401); }); @@ -78,4 +109,29 @@ describe("Auth (e2e)", () => { expect(response.status).toBe(401); }); + + it("answers 429, not 401 or 500, when the session lookup is rate limited", async () => { + getSessionSpy = spyOn(auth.api, "getSession").mockRejectedValue( + betterAuthError("TOO_MANY_REQUESTS", "Rate limit exceeded."), + ); + + const response = await request(app.getHttpServer()).get( + "/api/trpc/sso.settings", + ); + + expect(response.status).toBe(429); + expect(response.headers["retry-after"]).toBeUndefined(); + }); + + it("carries Retry-After, in seconds, on a rate-limited response", async () => { + getSessionSpy = spyOn(auth.api, "getSession").mockRejectedValue( + betterAuthError("TOO_MANY_REQUESTS", "Rate limit exceeded.", 4_500), + ); + + const response = await request(app.getHttpServer()) + .get("/api/trpc/sso.settings") + .expect(429); + + expect(response.headers["retry-after"]).toBe("5"); + }); }); diff --git a/apps/api/test/env.validation.spec.ts b/apps/api/test/env.validation.spec.ts new file mode 100644 index 000000000..7667478bd --- /dev/null +++ b/apps/api/test/env.validation.spec.ts @@ -0,0 +1,27 @@ +import { describe, expect, it } from "bun:test"; +import "reflect-metadata"; +import { validateEnv } from "../src/config/env.validation"; + +const required = { + DATABASE_URL: "postgresql://postgres:postgres@localhost:5432/crm", + BETTER_AUTH_SECRET: "test-secret-at-least-32-characters-long", + ALLOWED_SIGN_IN: "example.com", +}; + +describe("environment validation", () => { + it("allows the bridge secret to be absent", () => { + expect(validateEnv(required).AGENT_BRIDGE_SECRET).toBeUndefined(); + }); + + it("treats an empty bridge secret as an unset optional value", () => { + expect( + validateEnv({ ...required, AGENT_BRIDGE_SECRET: "" }).AGENT_BRIDGE_SECRET, + ).toBe(""); + }); + + it("requires configured bridge secrets to be at least 32 characters", () => { + expect(() => + validateEnv({ ...required, AGENT_BRIDGE_SECRET: "too-short" }), + ).toThrow("AGENT_BRIDGE_SECRET must be at least 32 characters"); + }); +}); diff --git a/apps/api/test/logging.spec.ts b/apps/api/test/logging.spec.ts index f5ca0308c..e093ca162 100644 --- a/apps/api/test/logging.spec.ts +++ b/apps/api/test/logging.spec.ts @@ -1,4 +1,5 @@ import { describe, expect, it, spyOn } from "bun:test"; +import { Logger } from "@nestjs/common"; import type { NextFunction, Request, Response } from "express"; import { ContextLogger } from "../src/logging/context-logger"; import { @@ -122,38 +123,83 @@ describe("request context", () => { }); }); +type LoggedRequest = { + message: string; + path: string; + statusCode: number; + apiKeyPresented: boolean; +}; + +const LOG_LEVELS = ["log", "warn", "error", "verbose"] as const; + +type LogLevel = (typeof LOG_LEVELS)[number]; + +type LoggedCall = { level: LogLevel; payload: LoggedRequest }; + type MiddlewareRun = { requestId: string; seen: string | undefined; + events: string[]; + finish: () => LoggedCall; }; +type RequestTarget = { originalUrl: string; path: string }; + describe("RequestLoggerMiddleware", () => { - function run(headers: Record): MiddlewareRun { + function run( + headers: Record, + target: RequestTarget = { originalUrl: "/things", path: "/things" }, + statusCode = 200, + ): MiddlewareRun { const middleware = new RequestLoggerMiddleware(); let requestId = ""; let seen: string | undefined; + const listeners = new Map void>(); const request = { method: "GET", - originalUrl: "/things", - path: "/things", + ...target, ip: "127.0.0.1", get: (name: string) => headers[name.toLowerCase()], } as unknown as Request; const response = { - statusCode: 200, + statusCode, setHeader: (_name: string, value: string) => { requestId = value; }, - on: () => response, + on: (event: string, listener: () => void) => { + listeners.set(event, listener); + return response; + }, } as unknown as Response; middleware.use(request, response, (() => { seen = getRequestContext()?.requestId; }) as NextFunction); - return { requestId, seen }; + const finish = (): LoggedCall => { + const calls: LoggedCall[] = []; + const spies = LOG_LEVELS.map((level) => + spyOn(Logger.prototype, level).mockImplementation( + (payload: LoggedRequest) => { + calls.push({ level, payload }); + }, + ), + ); + + try { + listeners.get("finish")?.(); + expect(calls).toHaveLength(1); + return calls[0] as LoggedCall; + } finally { + for (const spy of spies) { + spy.mockRestore(); + } + } + }; + + return { requestId, seen, events: [...listeners.keys()], finish }; } it("stamps a request id and exposes it to the handler", () => { @@ -175,4 +221,44 @@ describe("RequestLoggerMiddleware", () => { expect(requestId).not.toContain("\n"); expect(requestId).toMatch(/^[0-9a-f-]{36}$/); }); + + it("logs the path without its query string", () => { + const logged = run( + {}, + { originalUrl: "/api/test?token=secret&foo=bar", path: "/api/test" }, + ).finish().payload; + + expect(logged.path).toBe("/api/test"); + expect(logged.message).not.toContain("?"); + expect(JSON.stringify(logged)).not.toContain("secret"); + }); + + it("records that an api key was presented without logging it", () => { + const logged = run({ "x-api-key": "crm_live_secret" }).finish().payload; + + expect(logged.apiKeyPresented).toBe(true); + expect(JSON.stringify(logged)).not.toContain("crm_live_secret"); + }); + + it("records that no api key was presented", () => { + expect(run({}).finish().payload.apiKeyPresented).toBe(false); + }); + + it("logs once the response emits finish", () => { + expect(run({}).events).toEqual(["finish"]); + }); + + it.each([ + [200, "/things", "log"], + [404, "/things", "warn"], + [500, "/things", "error"], + [200, "/health", "verbose"], + [503, "/health", "error"], + ] as const)("logs a %i on %s at %s", (statusCode, path, level) => { + const logged = run({}, { originalUrl: path, path }, statusCode).finish(); + + expect(logged.level).toBe(level); + expect(logged.payload.statusCode).toBe(statusCode); + expect(logged.payload.path).toBe(path); + }); }); diff --git a/apps/api/test/support/rate-limited-auth.ts b/apps/api/test/support/rate-limited-auth.ts new file mode 100644 index 000000000..a0dc9ad82 --- /dev/null +++ b/apps/api/test/support/rate-limited-auth.ts @@ -0,0 +1,32 @@ +import { apiKey } from "@better-auth/api-key"; +import { db } from "@crm/db"; +import { betterAuth } from "better-auth"; +import { prismaAdapter } from "better-auth/adapters/prisma"; + +export type RateLimitedAuthOptions = { + maxRequests: number; + timeWindowMs: number; +}; + +export function rateLimitedAuth(options: RateLimitedAuthOptions) { + return betterAuth({ + secret: "test-secret-at-least-32-characters-long", + baseURL: "http://localhost:3001", + database: prismaAdapter(db, { provider: "postgresql" }), + emailAndPassword: { enabled: false }, + plugins: [ + apiKey({ + enableSessionForAPIKeys: true, + defaultKeyLength: 32, + deferUpdates: true, + rateLimit: { + enabled: true, + maxRequests: options.maxRequests, + timeWindow: options.timeWindowMs, + }, + }), + ], + }); +} + +export type RateLimitedAuth = ReturnType; diff --git a/apps/api/test/trpc-context.spec.ts b/apps/api/test/trpc-context.spec.ts new file mode 100644 index 000000000..9b1107c6f --- /dev/null +++ b/apps/api/test/trpc-context.spec.ts @@ -0,0 +1,184 @@ +import { afterEach, describe, expect, it, spyOn } from "bun:test"; +import { auth } from "@crm/auth"; +import { TRPCError } from "@trpc/server"; +import type { Request, Response } from "express"; +import { + createBaseTrpcContext, + retryAfterSeconds, +} from "../src/trpc/trpc.context"; + +function betterAuthError( + status: string, + message?: string, + tryAgainInMs?: number, +): Error { + const error = new Error(message ?? status); + error.name = "APIError"; + return Object.assign(error, { + status, + ...(tryAgainInMs !== undefined && { + body: { details: { tryAgainIn: tryAgainInMs } }, + }), + }); +} + +function fakeRequest(): Request { + return { headers: {} } as unknown as Request; +} + +function fakeResponse() { + const headers = new Map(); + const res = { + setHeader: (name: string, value: string) => { + headers.set(name, value); + }, + } as unknown as Response; + return { res, headers }; +} + +let getSessionSpy: ReturnType | undefined; + +afterEach(() => { + getSessionSpy?.mockRestore(); + getSessionSpy = undefined; +}); + +describe("createBaseTrpcContext", () => { + it("returns no session when there is no request", async () => { + const context = await createBaseTrpcContext(undefined); + + expect(context.session).toBeNull(); + }); + + it("returns the resolved session when auth.api.getSession succeeds", async () => { + const fakeSession = { user: { id: "u1" } }; + getSessionSpy = spyOn(auth.api, "getSession").mockResolvedValue( + fakeSession as never, + ); + + const context = await createBaseTrpcContext(fakeRequest()); + + expect(context.session).toBe(fakeSession as never); + }); + + it("treats an expired or invalid session as no session, not an error", async () => { + getSessionSpy = spyOn(auth.api, "getSession").mockRejectedValue( + betterAuthError("UNAUTHORIZED"), + ); + + const context = await createBaseTrpcContext(fakeRequest()); + + expect(context.session).toBeNull(); + }); + + it("treats a malformed API key as no session, not an error", async () => { + getSessionSpy = spyOn(auth.api, "getSession").mockRejectedValue( + betterAuthError("FORBIDDEN"), + ); + + const context = await createBaseTrpcContext(fakeRequest()); + + expect(context.session).toBeNull(); + }); + + it("does not turn a rate-limited API key into an absent session", async () => { + getSessionSpy = spyOn(auth.api, "getSession").mockRejectedValue( + betterAuthError("TOO_MANY_REQUESTS", "Rate limit exceeded."), + ); + + await expect(createBaseTrpcContext(fakeRequest())).rejects.toBeInstanceOf( + TRPCError, + ); + }); + + it("propagates a rate limit as a TOO_MANY_REQUESTS TRPCError, not UNAUTHORIZED", async () => { + getSessionSpy = spyOn(auth.api, "getSession").mockRejectedValue( + betterAuthError("TOO_MANY_REQUESTS", "Rate limit exceeded."), + ); + + try { + await createBaseTrpcContext(fakeRequest()); + throw new Error("expected createBaseTrpcContext to throw"); + } catch (error) { + expect(error).toBeInstanceOf(TRPCError); + expect((error as TRPCError).code).toBe("TOO_MANY_REQUESTS"); + expect((error as TRPCError).message).toBe("Rate limit exceeded."); + } + }); + + it("does not silently swallow an unrecognised auth failure", async () => { + getSessionSpy = spyOn(auth.api, "getSession").mockRejectedValue( + new Error("boom"), + ); + + await expect(createBaseTrpcContext(fakeRequest())).rejects.toThrow("boom"); + }); + + it("does not silently swallow an APIError with an unexpected status", async () => { + getSessionSpy = spyOn(auth.api, "getSession").mockRejectedValue( + betterAuthError("INTERNAL_SERVER_ERROR", "Database unavailable."), + ); + + await expect(createBaseTrpcContext(fakeRequest())).rejects.toThrow( + "Database unavailable.", + ); + }); + + it("sets Retry-After, in seconds, when the rate limiter names a wait", async () => { + getSessionSpy = spyOn(auth.api, "getSession").mockRejectedValue( + betterAuthError("TOO_MANY_REQUESTS", "Rate limit exceeded.", 4_500), + ); + const { res, headers } = fakeResponse(); + + await expect( + createBaseTrpcContext(fakeRequest(), res), + ).rejects.toBeInstanceOf(TRPCError); + + expect(headers.get("Retry-After")).toBe("5"); + }); + + it("does not set Retry-After when the rate limiter names no wait", async () => { + getSessionSpy = spyOn(auth.api, "getSession").mockRejectedValue( + betterAuthError("TOO_MANY_REQUESTS", "Rate limit exceeded."), + ); + const { res, headers } = fakeResponse(); + + await expect( + createBaseTrpcContext(fakeRequest(), res), + ).rejects.toBeInstanceOf(TRPCError); + + expect(headers.has("Retry-After")).toBe(false); + }); + + it("never sets Retry-After for a non-rate-limit failure", async () => { + getSessionSpy = spyOn(auth.api, "getSession").mockRejectedValue( + betterAuthError("UNAUTHORIZED"), + ); + const { res, headers } = fakeResponse(); + + await createBaseTrpcContext(fakeRequest(), res); + + expect(headers.size).toBe(0); + }); +}); + +describe("retryAfterSeconds", () => { + it("returns null when there is nothing to wait for", () => { + expect(retryAfterSeconds(undefined)).toBeNull(); + }); + + it("rounds up to the next whole second", () => { + expect(retryAfterSeconds(1)).toBe(1); + expect(retryAfterSeconds(1_000)).toBe(1); + expect(retryAfterSeconds(1_001)).toBe(2); + expect(retryAfterSeconds(4_500)).toBe(5); + }); + + it("never returns a negative value", () => { + expect(retryAfterSeconds(-5_000)).toBe(0); + }); + + it("treats zero as zero, not as absent", () => { + expect(retryAfterSeconds(0)).toBe(0); + }); +}); diff --git a/bun.lock b/bun.lock index 1c6df3e63..6e00abb74 100644 --- a/bun.lock +++ b/bun.lock @@ -65,6 +65,7 @@ "zod": "^4.4.3", }, "devDependencies": { + "@better-auth/api-key": "1.6.25", "@crm/typescript-config": "workspace:*", "@nestjs/testing": "^11.0.1", "@types/bun": "^1.3.14", diff --git a/docs/telemetry.md b/docs/telemetry.md index 6e1ed36aa..282c615b1 100644 --- a/docs/telemetry.md +++ b/docs/telemetry.md @@ -174,6 +174,12 @@ sent. | `tracking_capped` | Submissions the hourly contact cap refused. Matched against `CONTACT_CAP_REASON` (`@crm/db/tracking`) exactly, so the wording is a constant two files share rather than prose one of them may reword | | `tracking_paused` | Whether collection is paused | +#### API keys + +| Property | What it is | +| --- | --- | +| `api_key_rate_limited` | `429` responses from an API key's own rate limit, in the window. A count, never which key | + ### The setup funnel Eight events, each sent once per install, ever. Each carries the timestamp of the thing it diff --git a/packages/db/package.json b/packages/db/package.json index 4935913b1..1cf14f438 100644 --- a/packages/db/package.json +++ b/packages/db/package.json @@ -6,6 +6,7 @@ "exports": { ".": "./src/index.ts", "./agent-tasks": "./src/agent-tasks.ts", + "./api-key-rate-limit": "./src/api-key-rate-limit.ts", "./attribution": "./src/attribution.ts", "./blob": "./src/blob.ts", "./client": "./src/client.ts", diff --git a/packages/db/src/api-key-rate-limit.ts b/packages/db/src/api-key-rate-limit.ts new file mode 100644 index 000000000..4a52cbb4d --- /dev/null +++ b/packages/db/src/api-key-rate-limit.ts @@ -0,0 +1,49 @@ +import type { Db } from "./client"; +import type { Prisma } from "./generated/prisma/client"; + +export type ApiKeyRateLimitPolicy = { + maxRequests: number; + timeWindowMs: number; +}; + +export type ApiKeyRateLimitBackfillResult = { + updated: number; + alreadyPolicied: number; +}; + +const WITHOUT_POLICY: Prisma.ApikeyWhereInput = { + OR: [{ rateLimitMax: null }, { rateLimitTimeWindow: null }], +}; + +function isPositiveInteger(value: number): boolean { + return Number.isSafeInteger(value) && value > 0; +} + +export async function backfillApiKeyRateLimit( + db: Db, + policy: ApiKeyRateLimitPolicy, +): Promise { + if ( + !isPositiveInteger(policy.maxRequests) || + !isPositiveInteger(policy.timeWindowMs) + ) { + throw new RangeError( + "API key rate-limit policy values must be positive integers.", + ); + } + + const alreadyPolicied = await db.apikey.count({ + where: { NOT: WITHOUT_POLICY }, + }); + + const { count: updated } = await db.apikey.updateMany({ + where: WITHOUT_POLICY, + data: { + rateLimitEnabled: true, + rateLimitMax: policy.maxRequests, + rateLimitTimeWindow: policy.timeWindowMs, + }, + }); + + return { updated, alreadyPolicied }; +} diff --git a/packages/telemetry/src/allowlist.ts b/packages/telemetry/src/allowlist.ts index 5880dc2bd..83d096e1e 100644 --- a/packages/telemetry/src/allowlist.ts +++ b/packages/telemetry/src/allowlist.ts @@ -43,6 +43,8 @@ export const ALLOWED_PROPERTIES = [ "sandbox_used", "agent_conversations", + "api_key_rate_limited", + "facts_by_status", "facts_by_band", "facts_by_method", diff --git a/packages/telemetry/src/install.ts b/packages/telemetry/src/install.ts index de2a5a593..d87e6aea7 100644 --- a/packages/telemetry/src/install.ts +++ b/packages/telemetry/src/install.ts @@ -119,6 +119,7 @@ export async function reachedMilestones(): Promise { export const COUNTERS = { budgetExhausted: "budget_exhausted", + apiKeyRateLimited: "api_key_rate_limited", } as const; export async function bumpCounter(name: string, by = 1): Promise {