diff --git a/docs/EXPORT_STREAMING_AND_QUOTAS.md b/docs/EXPORT_STREAMING_AND_QUOTAS.md new file mode 100644 index 00000000..0db41aea --- /dev/null +++ b/docs/EXPORT_STREAMING_AND_QUOTAS.md @@ -0,0 +1,38 @@ +# Export streaming and quota contract + +Issue: #1537 + +## Invariants + +- Export quota accounting is keyed only by the authenticated principal. Client-supplied `orgId` query parameters and `x-organization-id` headers are not trusted. +- The daily quota repository performs the check and increment atomically, so concurrent requests cannot push a stored counter above its configured limit. +- Export column filters are bounded to 16 KiB of UTF-8 input and at most 25 columns per section. +- Export creation admits at most two simultaneous HTTP enqueue operations per quota key in a process. The daily quota remains the authoritative cross-process control. +- Download responses are emitted in 512 KiB chunks through a backpressure-aware stream when the response supports `write()`/`end()`. +- Export polling metadata exposes a one-second minimum interval and a 300-attempt client-side ceiling (five minutes at the advertised interval). +- Operational events contain bounded error text and export metadata, not credentials or signed URLs. + +## Degraded behavior + +- Daily quota exhaustion returns `429` with `Retry-After` set to the remaining UTC-day duration. +- Process-local concurrency saturation returns `429` with a one-second `Retry-After` hint. +- Invalid or oversized column filters return `400` before quota consumption or job enqueue. +- Unauthorized export status/download access returns `403` without revealing export contents. + +## Tradeoffs and limitations + +The concurrency gate is intentionally process-local. It protects the HTTP enqueue boundary from bursts but is not a replacement for a distributed semaphore. Deployments with multiple application instances must rely on the atomic persistent quota and job-system worker capacity for cross-process enforcement. + +Export generation still materializes the completed result in the existing job representation. This change bounds request parsing and network delivery and avoids one-shot response writes; it does not redesign the storage/worker representation of generated results. + +## Validation + +Focused validation: + +```bash +npm test -- src/services/exportBounds.test.ts +npm test -- src/routes/exports.quota.test.ts +npm test -- src/routes/exports.test.ts +npm run build +npm run lint +``` diff --git a/src/routes/exports.ts b/src/routes/exports.ts index 8426e1c2..df13127a 100644 --- a/src/routes/exports.ts +++ b/src/routes/exports.ts @@ -4,7 +4,6 @@ import type { BackgroundJobSystem } from '../jobs/system.js' import { authenticate, requireAdmin, - signDownloadToken, verifyDownloadToken, type AuthenticatedRequest, } from '../middleware/auth.js' @@ -23,6 +22,27 @@ import { getEnv } from '../config/index.js' import { resolveS3Config, getExportSignedUrl } from '../services/exportS3.js' import { createAuditLog } from '../lib/audit-logs.js' import { isOrgMember } from '../models/organizations.js' +import { + EXPORT_BOUNDS, + exportRequestGate, + isWithinByteLimit, + streamExportBuffer, +} from '../services/exportBounds.js' + +/** + * The authenticated principal is the only trusted quota/access-control key. + * Client-supplied orgId query/header values are deliberately ignored. + */ +const resolveOrgId = (req: AuthenticatedRequest): string => req.user!.userId + +const logExportEvent = (event: string, payload: Record): void => { + console.info(JSON.stringify({ + level: 'info', + event, + ...payload, + timestamp: new Date().toISOString(), + })) +} /** * Derive the org identifier used for quota and access-control decisions. @@ -51,6 +71,7 @@ const enforceExportQuota = async ( error: 'Export quota exceeded. Try again tomorrow.', retryAfter: result.retryAfter, }) + logExportEvent('exports.quota_rejected', { orgId, retryAfter: result.retryAfter }) return false } return true @@ -91,6 +112,8 @@ export function createExportRouter(jobSystem: BackgroundJobSystem): Router { const format = negotiateFormat(req, formatStr) const scopeStr = typeof req.query.scope === 'string' ? req.query.scope : 'all' + const validScopes = ['vaults', 'transactions', 'analytics', 'all'] + if (!validScopes.includes(scope)) return null const ScopeSchema = z.enum(['vaults', 'transactions', 'analytics', 'all']) const scopeParse = ScopeSchema.safeParse(scopeStr) if (!scopeParse.success) { @@ -105,7 +128,17 @@ export function createExportRouter(jobSystem: BackgroundJobSystem): Router { const columnsParam = req.query.columns if (columnsParam) { + if (!isWithinByteLimit(columnsParam, EXPORT_BOUNDS.MAX_COLUMN_FILTER_BYTES)) return null + try { + const parsedColumns: unknown = JSON.parse(columnsParam) + if (!parsedColumns || typeof parsedColumns !== 'object' || Array.isArray(parsedColumns)) return null + + result.columns = {} + for (const [section, cols] of Object.entries(parsedColumns as Record)) { + const allowed = ALLOWED_COLUMNS[section as keyof typeof ALLOWED_COLUMNS] + if (!allowed || !Array.isArray(cols) || cols.length > EXPORT_BOUNDS.MAX_COLUMNS_PER_SECTION) return null + if (!cols.every(col => typeof col === 'string' && allowed.includes(col))) return null const parsedColumns: unknown = typeof columnsParam === 'string' ? JSON.parse(columnsParam) : columnsParam @@ -141,8 +174,26 @@ export function createExportRouter(jobSystem: BackgroundJobSystem): Router { jobId, statusUrl: `/api/exports/status/${jobId}`, pollIntervalMs: 1000, + maxPollAttempts: 300, }) + const acquireExportRequest = (req: AuthenticatedRequest, res: Response): string | null => { + const orgId = resolveOrgId(req) + if (exportRequestGate.tryAcquire(orgId)) return orgId + + res.setHeader('Retry-After', String(EXPORT_BOUNDS.CONCURRENCY_RETRY_AFTER_SECONDS)) + res.status(429).json({ + error: 'Too many export requests in progress. Retry shortly.', + retryAfter: EXPORT_BOUNDS.CONCURRENCY_RETRY_AFTER_SECONDS, + }) + logExportEvent('exports.concurrency_rejected', { + orgId, + activeRequests: exportRequestGate.active(orgId), + limit: EXPORT_BOUNDS.MAX_CONCURRENT_REQUESTS_PER_ORG, + }) + return null + } + router.post('/me', authenticate, requireScopes(ApiScope.ReadAnalytics, ApiScope.ReadVaults), async (req: AuthenticatedRequest, res: Response) => { const options = parseOptions(req) if (!options) { @@ -150,6 +201,12 @@ export function createExportRouter(jobSystem: BackgroundJobSystem): Router { return } + const orgId = acquireExportRequest(req, res) + if (!orgId) return + const startedAt = Date.now() + + try { + if (!await enforceExportQuota(req, res)) return const idemParse = IdempotencyKeySchema.safeParse(req.header('idempotency-key')) if (!idemParse.success) { res.status(400).json({ error: 'Invalid idempotency-key header' }) @@ -169,15 +226,40 @@ export function createExportRouter(jobSystem: BackgroundJobSystem): Router { idempotencyKey: idemParse.data, }) - res.status(202).json(buildAcceptedResponse(job.id)) - } catch (error) { - if (isExportIdempotencyConflictError(error)) { - res.status(409).json({ error: error.message }) - return + try { + const job = await enqueueExportJob(jobSystem, { + userId: req.user!.userId, + orgId, + isAdmin: false, + scope: options.scope, + format: options.format, + columns: options.columns as any, + idempotencyKey: req.header('idempotency-key') ?? undefined, + }) + + logExportEvent('exports.enqueue_accepted', { + orgId, + jobId: job.id, + format: options.format, + scope: options.scope, + latencyMs: Date.now() - startedAt, + }) + res.status(202).json(buildAcceptedResponse(job.id)) + } catch (error) { + if (isExportIdempotencyConflictError(error)) { + res.status(409).json({ error: error.message }) + return + } + const message = error instanceof Error ? error.message : 'Failed to enqueue export job' + logExportEvent('exports.enqueue_failed', { + orgId, + latencyMs: Date.now() - startedAt, + error: message.slice(0, 200), + }) + res.status(500).json({ error: message }) } - - const message = error instanceof Error ? error.message : 'Failed to enqueue export job' - res.status(500).json({ error: message }) + } finally { + exportRequestGate.release(orgId) } }) @@ -188,6 +270,13 @@ export function createExportRouter(jobSystem: BackgroundJobSystem): Router { return } + const targetUserId = typeof req.query.targetUserId === 'string' ? req.query.targetUserId : undefined + const orgId = acquireExportRequest(req, res) + if (!orgId) return + const startedAt = Date.now() + + try { + if (!await enforceExportQuota(req, res)) return const idemParse = IdempotencyKeySchema.safeParse(req.header('idempotency-key')) if (!idemParse.success) { res.status(400).json({ error: 'Invalid idempotency-key header' }) @@ -216,15 +305,41 @@ export function createExportRouter(jobSystem: BackgroundJobSystem): Router { idempotencyKey: idemParse.data, }) - res.status(202).json(buildAcceptedResponse(job.id)) - } catch (error) { - if (isExportIdempotencyConflictError(error)) { - res.status(409).json({ error: error.message }) - return + try { + const job = await enqueueExportJob(jobSystem, { + userId: req.user!.userId, + orgId, + isAdmin: true, + targetUserId, + scope: options.scope, + format: options.format, + columns: options.columns as any, + idempotencyKey: req.header('idempotency-key') ?? undefined, + }) + + logExportEvent('exports.enqueue_accepted', { + orgId, + jobId: job.id, + format: options.format, + scope: options.scope, + latencyMs: Date.now() - startedAt, + }) + res.status(202).json(buildAcceptedResponse(job.id)) + } catch (error) { + if (isExportIdempotencyConflictError(error)) { + res.status(409).json({ error: error.message }) + return + } + const message = error instanceof Error ? error.message : 'Failed to enqueue export job' + logExportEvent('exports.enqueue_failed', { + orgId, + latencyMs: Date.now() - startedAt, + error: message.slice(0, 200), + }) + res.status(500).json({ error: message }) } - - const message = error instanceof Error ? error.message : 'Failed to enqueue export job' - res.status(500).json({ error: message }) + } finally { + exportRequestGate.release(orgId) } }) @@ -296,22 +411,14 @@ export function createExportRouter(jobSystem: BackgroundJobSystem): Router { : 'application/json; charset=utf-8' res.setHeader('Content-Type', mimeType) - res.setHeader('Content-Disposition', `attachment; filename="${job.filename}"`) + res.setHeader('Content-Disposition', `attachment; filename=\"${job.filename}\"`) res.setHeader('Content-Length', job.result.length) - - console.info( - JSON.stringify({ - level: 'info', - event: 'exports.download_served', - jobId: job.id, - format: job.format, - bytes: job.result.length, - filename: job.filename, - timestamp: new Date().toISOString(), - }), - ) - - res.send(job.result) + logExportEvent('exports.download_served', { + jobId: job.id, + format: job.format, + bytes: job.result.length, + }) + streamExportBuffer(res, job.result) }) router.get('/:id/download', authenticate, async (req: AuthenticatedRequest, res: Response) => { @@ -355,16 +462,10 @@ export function createExportRouter(jobSystem: BackgroundJobSystem): Router { console.warn('Failed to record audit log for export download:', err) } - console.info( - JSON.stringify({ - level: 'info', - event: 'exports.download_served', - jobId: job.id, - principal: req.user!.userId, - org: callerOrgId, - timestamp: new Date().toISOString(), - }), - ) + logExportEvent('exports.download_requested', { + jobId: job.id, + storage: job.s3Key ? 's3' : 'local', + }) const s3Config = resolveS3Config() if (s3Config && job.s3Key) { @@ -390,9 +491,9 @@ export function createExportRouter(jobSystem: BackgroundJobSystem): Router { : 'application/x-ndjson' res.setHeader('Content-Type', mimeType) - res.setHeader('Content-Disposition', `attachment; filename="${job.filename ?? 'export'}"`) + res.setHeader('Content-Disposition', `attachment; filename=\"${job.filename ?? 'export'}\"`) res.setHeader('Content-Length', job.result.length) - res.send(job.result) + streamExportBuffer(res, job.result) }) return router diff --git a/src/services/exportBounds.test.ts b/src/services/exportBounds.test.ts new file mode 100644 index 00000000..2fdf04ce --- /dev/null +++ b/src/services/exportBounds.test.ts @@ -0,0 +1,83 @@ +import { describe, expect, it, beforeEach } from '@jest/globals' +import { Writable } from 'node:stream' +import { + EXPORT_BOUNDS, + ExportRequestGate, + isWithinByteLimit, + streamExportBuffer, +} from './exportBounds.js' + +describe('ExportRequestGate', () => { + let gate: ExportRequestGate + + beforeEach(() => { + gate = new ExportRequestGate() + }) + + it('admits requests up to the configured concurrency bound', () => { + expect(gate.tryAcquire('org-a')).toBe(true) + expect(gate.tryAcquire('org-a')).toBe(true) + expect(gate.tryAcquire('org-a')).toBe(false) + expect(gate.active('org-a')).toBe(EXPORT_BOUNDS.MAX_CONCURRENT_REQUESTS_PER_ORG) + }) + + it('isolates concurrency between organizations', () => { + expect(gate.tryAcquire('org-a')).toBe(true) + expect(gate.tryAcquire('org-a')).toBe(true) + expect(gate.tryAcquire('org-b')).toBe(true) + expect(gate.active('org-a')).toBe(2) + expect(gate.active('org-b')).toBe(1) + }) + + it('release is idempotent at zero and allows recovery', () => { + gate.release('missing') + expect(gate.active('org-a')).toBe(0) + + expect(gate.tryAcquire('org-a')).toBe(true) + expect(gate.tryAcquire('org-a')).toBe(true) + gate.release('org-a') + expect(gate.active('org-a')).toBe(1) + expect(gate.tryAcquire('org-a')).toBe(true) + }) + + it('supports explicit limits for adversarial boundary tests', () => { + expect(gate.tryAcquire('org-a', 0)).toBe(false) + expect(gate.tryAcquire('org-a', 1)).toBe(true) + expect(gate.tryAcquire('org-a', 1)).toBe(false) + }) +}) + +describe('export input byte bounds', () => { + it('counts UTF-8 bytes rather than JavaScript code units', () => { + expect(isWithinByteLimit('a'.repeat(16), 16)).toBe(true) + expect(isWithinByteLimit('€'.repeat(6), 16)).toBe(false) + }) +}) + +describe('streamExportBuffer', () => { + it('writes the complete payload in bounded chunks', async () => { + const payload = Buffer.from('abcdefghij') + const chunks: Buffer[] = [] + const sink = new Writable({ + write(chunk, _encoding, callback) { + chunks.push(Buffer.from(chunk)) + callback() + }, + }) + + streamExportBuffer(sink, payload, 3) + await new Promise((resolve, reject) => { + sink.once('finish', resolve) + sink.once('error', reject) + }) + + expect(chunks.map(chunk => chunk.length)).toEqual([3, 3, 3, 1]) + expect(Buffer.concat(chunks)).toEqual(payload) + }) + + it('rejects invalid chunk sizes', () => { + const sink = new Writable({ write(_chunk, _encoding, callback) { callback() } }) + expect(() => streamExportBuffer(sink, Buffer.from('x'), 0)).toThrow(RangeError) + expect(() => streamExportBuffer(sink, Buffer.from('x'), 1.5)).toThrow(RangeError) + }) +}) diff --git a/src/services/exportBounds.ts b/src/services/exportBounds.ts new file mode 100644 index 00000000..23c3097f --- /dev/null +++ b/src/services/exportBounds.ts @@ -0,0 +1,68 @@ +import { Readable } from 'node:stream' + +export const EXPORT_BOUNDS = { + MAX_COLUMN_FILTER_BYTES: 16 * 1024, + MAX_COLUMNS_PER_SECTION: 25, + DOWNLOAD_CHUNK_BYTES: 512 * 1024, + MAX_CONCURRENT_REQUESTS_PER_ORG: 2, + CONCURRENCY_RETRY_AFTER_SECONDS: 1, +} as const + +export class ExportRequestGate { + private readonly activeRequests = new Map() + + tryAcquire(key: string, limit = EXPORT_BOUNDS.MAX_CONCURRENT_REQUESTS_PER_ORG): boolean { + const active = this.activeRequests.get(key) ?? 0 + if (active >= limit) return false + this.activeRequests.set(key, active + 1) + return true + } + + release(key: string): void { + const active = this.activeRequests.get(key) ?? 0 + if (active <= 1) { + this.activeRequests.delete(key) + return + } + this.activeRequests.set(key, active - 1) + } + + active(key: string): number { + return this.activeRequests.get(key) ?? 0 + } + + reset(): void { + this.activeRequests.clear() + } +} + +export const exportRequestGate = new ExportRequestGate() + +export const streamExportBuffer = ( + res: NodeJS.WritableStream & { send?: (body: Buffer) => unknown }, + buffer: Buffer, + chunkSize = EXPORT_BOUNDS.DOWNLOAD_CHUNK_BYTES, +): void => { + if (!Number.isInteger(chunkSize) || chunkSize <= 0) { + throw new RangeError('chunkSize must be a positive integer') + } + + if (typeof res.write !== 'function') { + if (typeof res.send === 'function') { + res.send(buffer) + return + } + throw new TypeError('response must support write() or send()') + } + + function* chunks(): Generator { + for (let offset = 0; offset < buffer.length; offset += chunkSize) { + yield buffer.subarray(offset, Math.min(offset + chunkSize, buffer.length)) + } + } + + Readable.from(chunks()).pipe(res) +} + +export const isWithinByteLimit = (value: string, maxBytes: number): boolean => + Buffer.byteLength(value, 'utf8') <= maxBytes