diff --git a/packages/server/src/controllers/predictions/index.ts b/packages/server/src/controllers/predictions/index.ts index 7865f40e19c..a5de675eea1 100644 --- a/packages/server/src/controllers/predictions/index.ts +++ b/packages/server/src/controllers/predictions/index.ts @@ -104,7 +104,7 @@ const createPrediction = async (req: Request, res: Response, next: NextFunction) const getRateLimiterMiddleware = async (req: Request, res: Response, next: NextFunction) => { try { - return RateLimiterManager.getInstance().getRateLimiter()(req, res, next) + return RateLimiterManager.getInstance().getRateLimiter({ withDefault: true })(req, res, next) } catch (error) { next(error) } diff --git a/packages/server/src/routes/predictions/index.ts b/packages/server/src/routes/predictions/index.ts index f7212de2d67..414eccf1c77 100644 --- a/packages/server/src/routes/predictions/index.ts +++ b/packages/server/src/routes/predictions/index.ts @@ -8,6 +8,7 @@ const router = express.Router() // /prediction is whitelisted, so the global auth never reads the token. Identify Auth0 callers // (browser extension) here; anonymous callers pass through. The data source exists only at request time. +// It runs before the rate limiter so the default limit can count per user, not per office IP. const optionalAAIAuth = (req: Request, res: Response, next: NextFunction) => verifyAAIToken(getRunningExpressApp().AppDataSource, { optional: true })(req, res, next) @@ -17,8 +18,8 @@ const optionalAAIAuth = (req: Request, res: Response, next: NextFunction) => router.post( ['/', '/:id'], getMulterStorage().array('files'), - predictionsController.getRateLimiterMiddleware, optionalAAIAuth, + predictionsController.getRateLimiterMiddleware, predictionsController.createPrediction ) diff --git a/packages/server/src/utils/rateLimit.ts b/packages/server/src/utils/rateLimit.ts index d4dd168a654..e7915e1e8ca 100644 --- a/packages/server/src/utils/rateLimit.ts +++ b/packages/server/src/utils/rateLimit.ts @@ -1,5 +1,5 @@ import { NextFunction, Request, Response } from 'express' -import { rateLimit, RateLimitRequestHandler } from 'express-rate-limit' +import { Options, rateLimit, RateLimitRequestHandler } from 'express-rate-limit' import { IChatFlow, MODE } from '../Interface' import { Mutex } from 'async-mutex' import { RedisStore } from 'rate-limit-redis' @@ -12,9 +12,17 @@ interface CustomListener extends QueueEventsListener { const QUEUE_NAME = 'ratelimit' const QUEUE_EVENT_NAME = 'updateRateLimiter' +const DEFAULT_LIMIT_MESSAGE = 'Too many requests, please try again later.' + +// Default limiter key: per chatflow, per identified user (so people behind one office IP don't share a bucket), else per IP. +export const rateLimitKey = (req: Request): string => { + const id = req.params.id ?? '' + return req.user?.id ? `${id}:user:${req.user.id}` : `${id}:ip:${req.ip}` +} export class RateLimiterManager { private rateLimiters: Record = {} + private defaultLimiter?: RateLimitRequestHandler private rateLimiterMutex: Mutex = new Mutex() private redisClient: Redis private static instance: RateLimiterManager @@ -53,6 +61,36 @@ export class RateLimiterManager { this.queueEventsProducer = new QueueEventsProducer(QUEUE_NAME, { connection: this.getConnection() }) this.queueEvents = new QueueEvents(QUEUE_NAME, { connection: this.getConnection() }) } + + // Opt-in: off unless DEFAULT_RATE_LIMIT_MAX is set, so a release changes nothing for deployments that don't set it. + const max = Number(process.env.DEFAULT_RATE_LIMIT_MAX) + if (Number.isInteger(max) && max > 0) { + const windowSeconds = Number(process.env.DEFAULT_RATE_LIMIT_WINDOW_SECONDS) || 60 + this.defaultLimiter = this.createLimiter('rl:default:', windowSeconds * 1000, max, DEFAULT_LIMIT_MESSAGE, { + keyGenerator: rateLimitKey, + standardHeaders: true, + legacyHeaders: false + }) + } + } + + private createLimiter(prefix: string, windowMs: number, max: number, message: string, extra: Partial = {}) { + if (process.env.MODE === MODE.QUEUE) { + return rateLimit({ + windowMs, + max, + standardHeaders: true, + legacyHeaders: false, + message, + store: new RedisStore({ + prefix, + // @ts-expect-error - Known issue: the `call` function is not present in @types/ioredis + sendCommand: (...args: string[]) => this.redisClient.call(...args) + }), + ...extra + }) + } + return rateLimit({ windowMs, max, message, ...extra }) } getConnection() { @@ -94,26 +132,7 @@ export class RateLimiterManager { public async addRateLimiter(id: string, duration: number, limit: number, message: string): Promise { const release = await this.rateLimiterMutex.acquire() try { - if (process.env.MODE === MODE.QUEUE) { - this.rateLimiters[id] = rateLimit({ - windowMs: duration * 1000, - max: limit, - standardHeaders: true, - legacyHeaders: false, - message, - store: new RedisStore({ - prefix: `rl:${id}`, - // @ts-expect-error - Known issue: the `call` function is not present in @types/ioredis - sendCommand: (...args: string[]) => this.redisClient.call(...args) - }) - }) - } else { - this.rateLimiters[id] = rateLimit({ - windowMs: duration * 1000, - max: limit, - message - }) - } + this.rateLimiters[id] = this.createLimiter(`rl:${id}`, duration * 1000, limit, message) } finally { release() } @@ -125,12 +144,11 @@ export class RateLimiterManager { } } - public getRateLimiter(): (req: Request, res: Response, next: NextFunction) => void { + public getRateLimiter(opts: { withDefault?: boolean } = {}): (req: Request, res: Response, next: NextFunction) => void { return (req: Request, res: Response, next: NextFunction) => { - const id = req.params.id - if (!this.rateLimiters[id]) return next() - const idRateLimiter = this.rateLimiters[id] - return idRateLimiter(req, res, next) + const limiter = this.rateLimiters[req.params.id] ?? (opts.withDefault ? this.defaultLimiter : undefined) + if (!limiter) return next() + return limiter(req, res, next) } } diff --git a/packages/server/test/integration/auth0Fixture.ts b/packages/server/test/integration/auth0Fixture.ts new file mode 100644 index 00000000000..8b7b5f3726a --- /dev/null +++ b/packages/server/test/integration/auth0Fixture.ts @@ -0,0 +1,63 @@ +/** + * Shared setup for local integration tests: a local Auth0-like issuer (RS256 + JWKS) and a Postgres + * DataSource migrated with the app's own migrations. Call setupAuth0AndDb before importing anything + * that reads AUTH0_* at module load (jwtCheck). + */ +import crypto from 'node:crypto' +import http from 'node:http' +import { AddressInfo } from 'node:net' +import { DataSource } from 'typeorm' + +export const AUDIENCE = 'https://api.test.local' +export const ORG = 'org_ias' +const b64 = (o: object) => Buffer.from(JSON.stringify(o)).toString('base64url') + +export const setupAuth0AndDb = async (pgUrl: string) => { + const keys = crypto.generateKeyPairSync('rsa', { modulusLength: 2048 }) + const otherKeys = crypto.generateKeyPairSync('rsa', { modulusLength: 2048 }) + let issuer = '' + const jwk = { ...keys.publicKey.export({ format: 'jwk' }), kid: 'k1', alg: 'RS256', use: 'sig' } + const server = http.createServer((req, res) => { + res.setHeader('Content-Type', 'application/json') + if (req.url === '/.well-known/openid-configuration') return res.end(JSON.stringify({ issuer, jwks_uri: `${issuer}jwks` })) + if (req.url === '/jwks') return res.end(JSON.stringify({ keys: [jwk] })) + res.statusCode = 404 + res.end('{}') + }) + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)) + issuer = `http://127.0.0.1:${(server.address() as AddressInfo).port}/` + + const sign = (claims: Record, key = keys.privateKey) => { + const now = Math.floor(Date.now() / 1000) + const head = b64({ alg: 'RS256', typ: 'JWT', kid: 'k1' }) + const body = b64({ iss: issuer, aud: AUDIENCE, iat: now, exp: now + 600, ...claims }) + const sig = crypto.sign('sha256', Buffer.from(`${head}.${body}`), key).toString('base64url') + return `${head}.${body}.${sig}` + } + + const url = new URL(pgUrl) + Object.assign(process.env, { + AUTH0_ISSUER_BASE_URL: issuer, + AUTH0_AUDIENCE: AUDIENCE, + AUTH0_ORGANIZATION_ID: ORG, + DATABASE_TYPE: 'postgres', + DATABASE_HOST: url.hostname, + DATABASE_PORT: url.port, + DATABASE_USER: decodeURIComponent(url.username), + DATABASE_PASSWORD: decodeURIComponent(url.password), + DATABASE_NAME: url.pathname.slice(1) + }) + + const dsModule = await import('../../src/DataSource') + await dsModule.init() + const ds: DataSource = dsModule.getDataSource() + await ds.initialize() + await ds.query('DROP SCHEMA public CASCADE; CREATE SCHEMA public; CREATE EXTENSION IF NOT EXISTS "uuid-ossp"') + await ds.runMigrations({ transaction: 'each' }) + + const close = async () => { + await ds.destroy() + await new Promise((resolve) => server.close(resolve)) + } + return { ds, sign, otherKey: otherKeys.privateKey, close } +} diff --git a/packages/server/test/integration/prediction/index.test.ts b/packages/server/test/integration/prediction/index.test.ts index 8600c11eb73..6f726c2790e 100644 --- a/packages/server/test/integration/prediction/index.test.ts +++ b/packages/server/test/integration/prediction/index.test.ts @@ -12,11 +12,10 @@ * NODE_OPTIONS is needed because langfuse-core (loaded through flowise-components) runs a dynamic import() at load. */ import crypto from 'node:crypto' -import http from 'node:http' -import { AddressInfo } from 'node:net' import express from 'express' import supertest from 'supertest' import { DataSource } from 'typeorm' +import { ORG, setupAuth0AndDb } from '../auth0Fixture' const mockApp: { AppDataSource?: DataSource } = {} jest.mock('../../../src/utils/getRunningExpressApp', () => ({ @@ -31,30 +30,17 @@ jest.mock('../../../src/services/predictions', () => ({ const PG = process.env.TEST_PG_URL const d = PG ? describe : describe.skip -const AUDIENCE = 'https://api.test.local' -const ORG = 'org_ias' -const b64 = (o: object) => Buffer.from(JSON.stringify(o)).toString('base64url') - d('POST /api/v1/prediction (integration)', () => { jest.setTimeout(180000) - const keys = crypto.generateKeyPairSync('rsa', { modulusLength: 2048 }) - const otherKeys = crypto.generateKeyPairSync('rsa', { modulusLength: 2048 }) - let issuer = '' - let issuerServer: http.Server let ds: DataSource let app: express.Express let entities: { User: any; Organization: any; Workspace: any; WorkspaceUser: any; ChatFlow: any } let userId = '' let chatflowId = '' + let fx: Awaited> + const sign = (claims: Record, key?: crypto.KeyObject) => fx.sign(claims, key) - const sign = (claims: Record, key = keys.privateKey) => { - const now = Math.floor(Date.now() / 1000) - const head = b64({ alg: 'RS256', typ: 'JWT', kid: 'k1' }) - const body = b64({ iss: issuer, aud: AUDIENCE, iat: now, exp: now + 600, ...claims }) - const sig = crypto.sign('sha256', Buffer.from(`${head}.${body}`), key).toString('base64url') - return `${head}.${body}.${sig}` - } const alice = (extra: Record = {}) => sign({ sub: 'auth0|alice', email: 'alice@test.local', name: 'Alice', org_id: ORG, ...extra }) @@ -73,37 +59,9 @@ d('POST /api/v1/prediction (integration)', () => { }) beforeAll(async () => { - const jwk = { ...keys.publicKey.export({ format: 'jwk' }), kid: 'k1', alg: 'RS256', use: 'sig' } - issuerServer = http.createServer((req, res) => { - res.setHeader('Content-Type', 'application/json') - if (req.url === '/.well-known/openid-configuration') return res.end(JSON.stringify({ issuer, jwks_uri: `${issuer}jwks` })) - if (req.url === '/jwks') return res.end(JSON.stringify({ keys: [jwk] })) - res.statusCode = 404 - res.end('{}') - }) - await new Promise((resolve) => issuerServer.listen(0, '127.0.0.1', resolve)) - issuer = `http://127.0.0.1:${(issuerServer.address() as AddressInfo).port}/` - - const url = new URL(PG!) - Object.assign(process.env, { - AUTH0_ISSUER_BASE_URL: issuer, - AUTH0_AUDIENCE: AUDIENCE, - AUTH0_ORGANIZATION_ID: ORG, - DATABASE_TYPE: 'postgres', - DATABASE_HOST: url.hostname, - DATABASE_PORT: url.port, - DATABASE_USER: decodeURIComponent(url.username), - DATABASE_PASSWORD: decodeURIComponent(url.password), - DATABASE_NAME: url.pathname.slice(1) - }) - - // Imported after env is set: jwtCheck reads AUTH0_* at module load. - const dsModule = await import('../../../src/DataSource') - await dsModule.init() - ds = dsModule.getDataSource() - await ds.initialize() - await ds.query('DROP SCHEMA public CASCADE; CREATE SCHEMA public; CREATE EXTENSION IF NOT EXISTS "uuid-ossp"') - await ds.runMigrations({ transaction: 'each' }) + // Before any import that reads AUTH0_* at module load (jwtCheck). + fx = await setupAuth0AndDb(PG!) + ds = fx.ds mockApp.AppDataSource = ds entities = { @@ -150,8 +108,7 @@ d('POST /api/v1/prediction (integration)', () => { }) afterAll(async () => { - await ds?.destroy() - await new Promise((resolve) => issuerServer?.close(resolve)) + await fx?.close() }) beforeEach(() => { @@ -192,7 +149,7 @@ d('POST /api/v1/prediction (integration)', () => { }) it('forged token stays anonymous', async () => { - const forged = sign({ sub: 'auth0|alice', email: 'alice@test.local', org_id: ORG }, otherKeys.privateKey) + const forged = sign({ sub: 'auth0|alice', email: 'alice@test.local', org_id: ORG }, fx.otherKey) await post(chatflowId, `Bearer ${forged}`).then(ok) expect(seenUser).toBeUndefined() }) diff --git a/packages/server/test/integration/rateLimit/index.test.ts b/packages/server/test/integration/rateLimit/index.test.ts new file mode 100644 index 00000000000..2ed537d99d7 --- /dev/null +++ b/packages/server/test/integration/rateLimit/index.test.ts @@ -0,0 +1,99 @@ +/** + * Integration test for the default rate limit on /api/v1/prediction (local only). + * + * Real: the predictions router (optional auth, then the rate limiter), RS256 token verification against a + * local issuer, Postgres with the app's migrations, the RateLimiterManager with DEFAULT_RATE_LIMIT_MAX=2. + * Mocked: getRunningExpressApp, buildChatflow and the streaming check. + * + * Run: same as test/integration/prediction, with `pnpm exec jest test/integration/rateLimit`. + */ +import crypto from 'node:crypto' +import express from 'express' +import supertest from 'supertest' +import { DataSource } from 'typeorm' +import { ORG, setupAuth0AndDb } from '../auth0Fixture' + +const mockApp: { AppDataSource?: DataSource } = {} +jest.mock('../../../src/utils/getRunningExpressApp', () => ({ + getRunningExpressApp: () => mockApp +})) +jest.mock('../../../src/services/predictions', () => ({ + __esModule: true, + default: { buildChatflow: async () => ({ text: 'ok' }) } +})) + +const PG = process.env.TEST_PG_URL +const d = PG ? describe : describe.skip + +d('default rate limit on /api/v1/prediction (integration)', () => { + jest.setTimeout(180000) + + let fx: Awaited> + let app: express.Express + let workspaceId = '' + let ChatFlow: any + let User: any + const user = (name: string) => `Bearer ${fx.sign({ sub: `auth0|${name}`, email: `${name}@test.local`, name, org_id: ORG })}` + + const post = (id: string, authorization?: string) => { + const req = supertest(app).post(`/api/v1/prediction/${id}`).send({ question: 'hi' }) + return authorization ? req.set('Authorization', authorization) : req + } + const statuses = async (id: string, authorization: string | undefined, n: number) => { + const out: number[] = [] + for (let i = 0; i < n; i++) out.push((await post(id, authorization)).status) + return out + } + // Each test gets its own chatflow, so buckets don't leak between tests. + const newChatflow = async () => + (await fx.ds.getRepository(ChatFlow).save({ name: 'cf', flowData: JSON.stringify({ nodes: [], edges: [] }), workspaceId })).id + + beforeAll(async () => { + // The manager is a lazy singleton: set before its first use. + process.env.DEFAULT_RATE_LIMIT_MAX = '2' + fx = await setupAuth0AndDb(PG!) + mockApp.AppDataSource = fx.ds + ChatFlow = (await import('../../../src/database/entities/ChatFlow')).ChatFlow + User = (await import('../../../src/database/entities/User')).User + const WorkspaceUser = (await import('../../../src/enterprise/database/entities/workspace-user.entity')).WorkspaceUser + + const chatflowsService = (await import('../../../src/services/chatflows')).default + jest.spyOn(chatflowsService, 'checkIfChatflowIsValidForStreaming').mockResolvedValue({ isStreaming: false } as any) + const router = (await import('../../../src/routes/predictions')).default + app = express() + app.use(express.json()) + app.use('/api/v1/prediction', router) + app.use((err: any, _req: express.Request, res: express.Response, _next: express.NextFunction) => { + res.status(err.statusCode || err.status || 500).json({ message: err.message }) + }) + + // Provision alice (first signed-in request), on a throwaway chatflow id so no test bucket is touched. + await post(crypto.randomUUID(), user('alice')) + const alice = await fx.ds.getRepository(User).findOneByOrFail({ auth0Id: 'auth0|alice' }) + workspaceId = (await fx.ds.getRepository(WorkspaceUser).findOneByOrFail({ userId: alice.id })).workspaceId + }) + + afterAll(async () => { + delete process.env.DEFAULT_RATE_LIMIT_MAX + await fx?.close() + }) + + it('one user hits the limit; another user on the same IP still gets through', async () => { + const id = await newChatflow() + expect(await statuses(id, user('alice'), 3)).toEqual([200, 200, 429]) + expect(await statuses(id, user('bob'), 1)).toEqual([200]) + }) + + it('anonymous callers on one IP share one bucket', async () => { + expect(await statuses(await newChatflow(), undefined, 3)).toEqual([200, 200, 429]) + }) + + it('a forged token counts as anonymous and creates no user', async () => { + const id = await newChatflow() + const usersBefore = await fx.ds.getRepository(User).count() + const forged = `Bearer ${fx.sign({ sub: 'auth0|mallory', email: 'm@test.local', org_id: ORG }, fx.otherKey)}` + expect(await statuses(id, forged, 2)).toEqual([200, 200]) + expect(await statuses(id, undefined, 1)).toEqual([429]) + expect(await fx.ds.getRepository(User).count()).toBe(usersBefore) + }) +}) diff --git a/packages/server/test/routes/predictions/index.test.ts b/packages/server/test/routes/predictions/index.test.ts index 7a8f86d4053..ca5bc933798 100644 --- a/packages/server/test/routes/predictions/index.test.ts +++ b/packages/server/test/routes/predictions/index.test.ts @@ -29,8 +29,17 @@ jest.mock('../../../src/middlewares/authentication/populateWorkspaceData', () => jest.mock('../../../src/utils', () => ({ getMulterStorage: () => ({ array: () => (_req: any, _res: any, next: any) => next() }) })) +const mockLimiter: { opts?: any; seenUser?: any } = {} jest.mock('../../../src/utils/rateLimit', () => ({ - RateLimiterManager: { getInstance: () => ({ getRateLimiter: () => (_req: any, _res: any, next: any) => next() }) } + RateLimiterManager: { + getInstance: () => ({ + getRateLimiter: (opts: any) => (req: any, _res: any, next: any) => { + mockLimiter.opts = opts + mockLimiter.seenUser = req.user + next() + } + }) + } })) jest.mock('../../../src/utils/getRunningExpressApp', () => ({ getRunningExpressApp: () => ({ @@ -83,6 +92,17 @@ describe('POST /api/v1/prediction/:id', () => { expect((chatflowsService.getChatflowById as jest.Mock).mock.calls[0]).toHaveLength(1) }) + it('runs optional auth before the rate limiter', async () => { + mockJwtMiddleware.mockImplementation((req: any, _res: any, next: any) => { + req.auth = { payload: { sub: 'auth0|u1', email: 'a@b.c', org_id: 'org_valid123' } } + next() + }) + mockLimiter.seenUser = 'not-called' + await supertest(app).post('/api/v1/prediction/cf-1').set('Authorization', 'Bearer a.b.c').send({ question: 'hi' }).expect(200) + expect(mockLimiter.seenUser?.id).toBe('user-1') + expect(mockLimiter.opts).toEqual({ withDefault: true }) + }) + it('no header stays anonymous', async () => { await supertest(app).post('/api/v1/prediction/cf-1').send({ question: 'hi' }).expect(200) expect(seenUser).toBeUndefined() diff --git a/packages/server/test/utils/rateLimit/index.test.ts b/packages/server/test/utils/rateLimit/index.test.ts new file mode 100644 index 00000000000..8395371c900 --- /dev/null +++ b/packages/server/test/utils/rateLimit/index.test.ts @@ -0,0 +1,142 @@ +/** + * RateLimiterManager default limiter — real express-rate-limit (memory store), real manager. + * Queue mode is checked only for how the Redis store is built (ioredis, bullmq and rate-limit-redis mocked). + */ +const mockRedisStore = jest.fn().mockImplementation(() => ({ + init: jest.fn(), + increment: jest.fn().mockResolvedValue({ totalHits: 1, resetTime: undefined }), + decrement: jest.fn(), + resetKey: jest.fn() +})) +jest.mock('rate-limit-redis', () => ({ RedisStore: mockRedisStore })) +jest.mock('ioredis', () => jest.fn().mockImplementation(() => ({ call: jest.fn() }))) +jest.mock('bullmq', () => ({ QueueEvents: jest.fn(), QueueEventsProducer: jest.fn() })) + +import express, { NextFunction, Request, Response } from 'express' +import supertest from 'supertest' +import { RateLimiterManager, rateLimitKey } from '../../../src/utils/rateLimit' + +const ENV = { ...process.env } +afterEach(() => { + process.env = { ...ENV } +}) + +const asUser = (req: Request, _res: Response, next: NextFunction) => { + const id = req.header('x-test-user') + if (id) (req as any).user = { id, email: `${id}@example.com` } + next() +} + +const appFor = (mgr: RateLimiterManager, withDefault = true) => { + const app = express() + const limiter = withDefault ? mgr.getRateLimiter({ withDefault: true }) : mgr.getRateLimiter() + app.post(['/p', '/p/:id'], asUser, limiter, (_req, res) => res.json({ ok: true })) + return app +} + +const statuses = async (app: express.Express, path: string, n: number, user?: string) => { + const out: number[] = [] + for (let i = 0; i < n; i++) { + const r = supertest(app).post(path) + if (user) r.set('x-test-user', user) + out.push((await r).status) + } + return out +} + +describe('RateLimiterManager default limiter', () => { + it('no DEFAULT_RATE_LIMIT_MAX → unlimited chatflow stays unlimited', async () => { + delete process.env.DEFAULT_RATE_LIMIT_MAX + expect(await statuses(appFor(new RateLimiterManager()), '/p/a', 5)).toEqual([200, 200, 200, 200, 200]) + }) + + it.each(['0', '-1', '1.5', 'abc'])('DEFAULT_RATE_LIMIT_MAX=%s is ignored → unlimited', async (value) => { + process.env.DEFAULT_RATE_LIMIT_MAX = value + expect(await statuses(appFor(new RateLimiterManager()), '/p/a', 3)).toEqual([200, 200, 200]) + }) + + it('DEFAULT_RATE_LIMIT_MAX=2 → 3rd call to an unlimited chatflow is 429 with the default message', async () => { + process.env.DEFAULT_RATE_LIMIT_MAX = '2' + const app = appFor(new RateLimiterManager()) + expect(await statuses(app, '/p/a', 2)).toEqual([200, 200]) + const third = await supertest(app).post('/p/a') + expect(third.status).toBe(429) + expect(third.text).toBe('Too many requests, please try again later.') + }) + + it('default counts per chatflow', async () => { + process.env.DEFAULT_RATE_LIMIT_MAX = '2' + const app = appFor(new RateLimiterManager()) + expect(await statuses(app, '/p/a', 2)).toEqual([200, 200]) + expect(await statuses(app, '/p/b', 1)).toEqual([200]) + }) + + it('two users on one IP get separate default buckets', async () => { + process.env.DEFAULT_RATE_LIMIT_MAX = '2' + const app = appFor(new RateLimiterManager()) + expect(await statuses(app, '/p/a', 3, 'u1')).toEqual([200, 200, 429]) + expect(await statuses(app, '/p/a', 1, 'u2')).toEqual([200]) + }) + + it('anonymous callers on one IP share one default bucket per chatflow', async () => { + process.env.DEFAULT_RATE_LIMIT_MAX = '2' + expect(await statuses(appFor(new RateLimiterManager()), '/p/a', 3)).toEqual([200, 200, 429]) + }) + + it('configured chatflow limit wins over the default', async () => { + process.env.DEFAULT_RATE_LIMIT_MAX = '2' + const mgr = new RateLimiterManager() + await mgr.addRateLimiter('a', 60, 4, 'configured') + expect(await statuses(appFor(mgr), '/p/a', 5)).toEqual([200, 200, 200, 200, 429]) + }) + + it('configured chatflow limit still keys by IP', async () => { + const mgr = new RateLimiterManager() + await mgr.addRateLimiter('a', 60, 2, 'configured') + const app = appFor(mgr) + expect(await statuses(app, '/p/a', 2, 'u1')).toEqual([200, 200]) + expect(await statuses(app, '/p/a', 1, 'u2')).toEqual([429]) + }) + + it('getRateLimiter() without withDefault ignores the default', async () => { + process.env.DEFAULT_RATE_LIMIT_MAX = '2' + expect(await statuses(appFor(new RateLimiterManager(), false), '/p/a', 4)).toEqual([200, 200, 200, 200]) + }) + + it('no id → keyed under empty chatflow id, no throw', async () => { + process.env.DEFAULT_RATE_LIMIT_MAX = '2' + expect(await statuses(appFor(new RateLimiterManager()), '/p', 3)).toEqual([200, 200, 429]) + }) + + it('DEFAULT_RATE_LIMIT_WINDOW_SECONDS sets the window', async () => { + process.env.DEFAULT_RATE_LIMIT_MAX = '1' + process.env.DEFAULT_RATE_LIMIT_WINDOW_SECONDS = '120' + const res = await supertest(appFor(new RateLimiterManager())).post('/p/a') + expect(res.status).toBe(200) + expect(Number(res.headers['ratelimit-reset'])).toBeGreaterThan(60) + }) + + it('queue mode → default limiter built with RedisStore prefix rl:default:', () => { + process.env.MODE = 'queue' + process.env.REDIS_URL = 'redis://127.0.0.1:1' + process.env.DEFAULT_RATE_LIMIT_MAX = '2' + mockRedisStore.mockClear() + new RateLimiterManager() + expect(mockRedisStore).toHaveBeenCalledWith(expect.objectContaining({ prefix: 'rl:default:' })) + }) +}) + +describe('rateLimitKey', () => { + it('uses chatflow id and user id when the caller is identified', () => { + expect(rateLimitKey({ params: { id: 'cf' }, ip: '1.1.1.1', user: { id: 'u1' } } as any)).toBe('cf:user:u1') + }) + + it('falls back to chatflow id and IP', () => { + expect(rateLimitKey({ params: { id: 'cf' }, ip: '1.1.1.1' } as any)).toBe('cf:ip:1.1.1.1') + }) + + it('rateLimitKey never contains email', () => { + const key = rateLimitKey({ params: { id: 'cf' }, ip: '1.1.1.1', user: { id: 'u1', email: 'a@b.c' } } as any) + expect(key).not.toContain('a@b.c') + }) +})