From d7b3be8c347810c022adeeb86631bb0c2d1fb915 Mon Sep 17 00:00:00 2001 From: omoh5 Date: Mon, 19 Oct 2026 22:59:26 +0000 Subject: [PATCH 1/2] feat(webhooks): add delivery retries and HMAC signature verification --- .env.example | 8 + API_DOCUMENTATION.md | 35 ++ docs/configuration.md | 23 +- package-lock.json | 1 + src/app.module.ts | 7 +- .../guards/webhook-signature.guard.spec.ts | 352 +++++------------- src/common/guards/webhook-signature.guard.ts | 144 ++----- src/config/env.validation.spec.ts | 24 ++ src/config/env.validation.ts | 10 + src/config/queue.config.ts | 10 + src/main.ts | 2 +- .../services/webhook-audit.service.spec.ts | 2 +- .../services/webhook-audit.service.ts | 10 +- .../services/webhook-delivery.service.spec.ts | 9 + .../services/webhook-delivery.service.ts | 3 +- .../webhook-delivery.integration.spec.ts | 191 ++++++++++ .../webhooks/webhook-failure-audit.spec.ts | 36 +- src/modules/webhooks/webhook.module.ts | 11 +- src/modules/webhooks/webhook.service.spec.ts | 8 +- .../webhooks/webhooks.processor.spec.ts | 37 +- src/modules/webhooks/webhooks.processor.ts | 106 ++++-- src/queues/dlq.processor.ts | 13 +- src/queues/queue-failure-listener.spec.ts | 32 +- src/queues/queue-failure-listener.ts | 23 +- src/queues/webhook.queue.spec.ts | 30 +- src/queues/webhook.queue.ts | 51 +-- src/utils/backoff.util.spec.ts | 22 +- src/utils/backoff.util.ts | 8 +- 28 files changed, 721 insertions(+), 487 deletions(-) create mode 100644 src/modules/webhooks/webhook-delivery.integration.spec.ts diff --git a/.env.example b/.env.example index b05eb3d2..1bded0b6 100644 --- a/.env.example +++ b/.env.example @@ -61,6 +61,14 @@ STORAGE_SECRET_KEY=astroid-secret # Queue (BullMQ) — shares Redis by default QUEUE_PREFIX=astroid QUEUE_CONCURRENCY=5 +WEBHOOK_DELIVERY_MAX_ATTEMPTS=5 +WEBHOOK_RETRY_DELAY_MS=2000 +WEBHOOK_HTTP_TIMEOUT_MS=5000 +WEBHOOK_FAILED_JOB_RETENTION_SECONDS=86400 +WEBHOOK_DEAD_LETTER_RETENTION_SECONDS=604800 +# Optional; required to accept signed POST /api/v1/webhooks/receive requests. +# Generate and store a unique secret (at least 32 characters) in a secret manager. +# WEBHOOK_INGRESS_SECRET= # Rate limiting (@nestjs/throttler - per-tier steady-state limits per TTL window) # Counters are stored in Redis so every API replica enforces the same budget. diff --git a/API_DOCUMENTATION.md b/API_DOCUMENTATION.md index 18facfc5..de71fee4 100644 --- a/API_DOCUMENTATION.md +++ b/API_DOCUMENTATION.md @@ -421,6 +421,41 @@ export function verifyAstroidWebhook({ secret, headers, rawBody }) { } ``` +### Delivery retries and failures + +Outgoing deliveries are asynchronous and provide **at-least-once**, not +exactly-once, semantics. The event ID remains stable across retries; consumers +should use it as an idempotency key. By default, BullMQ makes five total +attempts with exponential backoff starting at 2 seconds and 20% jitter. Operators +can configure the attempt limit, base delay, HTTP timeout, and failed-job +retention with `WEBHOOK_DELIVERY_MAX_ATTEMPTS`, `WEBHOOK_RETRY_DELAY_MS`, +`WEBHOOK_HTTP_TIMEOUT_MS`, and `WEBHOOK_FAILED_JOB_RETENTION_SECONDS`. + +Network errors, timeouts, HTTP 408, HTTP 429, and 5xx responses are retried. +Other 3xx/4xx responses are treated as permanent and stop immediately. Redirects +are not followed. Exhausted jobs are routed to the shared dead-letter queue and +recorded in the webhook delivery audit trail. Failed BullMQ jobs are retained +for 24 hours by default; the dead-letter queue retains failed records for seven +days by default. `WEBHOOK_DEAD_LETTER_RETENTION_SECONDS` configures that +retention. Operators can inspect and re-drive dead-letter jobs through the +existing queue administration tools. + +### Inbound signature verification + +`POST /{API_PREFIX}/webhooks/receive` requires both +`x-astroid-timestamp` and `x-astroid-signature`. The inbound signature is the +lowercase 64-character hex HMAC-SHA256 digest of UTF-8 +`.` concatenated with the exact raw request body bytes. The request +timestamp must be within five minutes of server time. Verification uses a +constant-time comparison; missing, malformed, stale, or invalid signatures are +rejected before the ingress handler runs. + +The route uses the dedicated `WEBHOOK_INGRESS_SECRET`, which must contain at +least 32 characters. It is separate from each outbound subscription's +`whsec_...` secret; when unset, inbound requests are rejected. Deployments +should provide it through a secret manager. Nest captures raw request bytes +before JSON parsing so whitespace and key order are preserved for verification. + ## Health Probes (`/health`) The liveness and readiness probes are served **outside** the API prefix, so diff --git a/docs/configuration.md b/docs/configuration.md index 5d19e2b5..73582ab5 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -136,10 +136,25 @@ are rejected: ### Queues (BullMQ) -| Variable | Default | Description | -| ------------------- | --------- | ----------------------------- | -| `QUEUE_PREFIX` | `astroid` | Key prefix for BullMQ queues. | -| `QUEUE_CONCURRENCY` | `5` | Default worker concurrency. | +| Variable | Default | Description | +| ----------------------------------------- | ------- | ----------- | +| `QUEUE_PREFIX` | `astroid` | Key prefix for BullMQ queues. | +| `QUEUE_CONCURRENCY` | `5` | Default worker concurrency. | +| `WEBHOOK_DELIVERY_MAX_ATTEMPTS` | `5` | Maximum webhook delivery attempts, including the initial delivery; bounded to 25. | +| `WEBHOOK_RETRY_DELAY_MS` | `2000` | Initial exponential retry delay in milliseconds; bounded to 300000. | +| `WEBHOOK_HTTP_TIMEOUT_MS` | `5000` | Outbound request timeout in milliseconds; bounded to 120000. | +| `WEBHOOK_FAILED_JOB_RETENTION_SECONDS` | `86400` | How long BullMQ retains failed webhook jobs before cleanup; bounded to 30 days. | +| `WEBHOOK_DEAD_LETTER_RETENTION_SECONDS` | `604800` | How long the dead-letter queue retains webhook failures; bounded to 30 days. | + +Webhook retries use BullMQ exponential backoff with 20% jitter. Permanent +redirect and 4xx responses stop retrying; HTTP 408, 429, 5xx, and network or +timeout failures are retried up to the configured attempt limit. + +### Webhook ingress + +| Variable | Default | Description | +| -------- | ------- | ----------- | +| `WEBHOOK_INGRESS_SECRET` | _(unset)_ | Dedicated secret for inbound callbacks. Must be at least 32 characters when set. Requests are rejected while unset. Never reuse an outbound subscriber secret. | ### Rate limiting diff --git a/package-lock.json b/package-lock.json index a5389c49..b72149a4 100644 --- a/package-lock.json +++ b/package-lock.json @@ -5911,6 +5911,7 @@ "version": "2.3.3", "resolved": "https://registry.npmjs.org/fsevents/-/fsevents-2.3.3.tgz", "integrity": "sha512-5xoDfX+fL7faATnagmWPpbFtwh/R77WmMMqqHGS65C3vvB0YHrgF+B1YmZ3441tMj5n63k0212XNoJwzlhffQw==", + "dev": true, "hasInstallScript": true, "license": "MIT", "optional": true, diff --git a/src/app.module.ts b/src/app.module.ts index f0252c1c..aca3f662 100644 --- a/src/app.module.ts +++ b/src/app.module.ts @@ -1,4 +1,4 @@ -import { MiddlewareConsumer, Module, NestModule } from '@nestjs/common'; +import { MiddlewareConsumer, Module, NestModule, RequestMethod } from '@nestjs/common'; import { APP_FILTER, APP_GUARD, APP_INTERCEPTOR } from '@nestjs/core'; import { ConfigService } from '@nestjs/config'; import { ThrottlerModule } from '@nestjs/throttler'; @@ -16,6 +16,7 @@ import { REDIS_CLIENT } from './common/locks/locks.constants'; import { EncryptionModule } from './common/encryption/encryption.module'; import { RequestIdMiddleware } from './middleware/request-id.middleware'; import { StructuredRequestLoggingMiddleware } from './middleware/structured-request-logging.middleware'; +import { RawBodyMiddleware } from './common/middleware/raw-body.middleware'; import { REQUEST_ID_HEADER } from './common/constants/headers'; import { JwtAuthGuard } from './common/guards/jwt-auth.guard'; @@ -156,5 +157,9 @@ export class AppModule implements NestModule { configure(consumer: MiddlewareConsumer): void { consumer.apply(RequestIdMiddleware, StructuredRequestLoggingMiddleware).forRoutes('*'); consumer.apply(RequestMetricsMiddleware).forRoutes('*'); + consumer.apply(RawBodyMiddleware).forRoutes({ + path: 'webhooks/receive', + method: RequestMethod.POST, + }); } } diff --git a/src/common/guards/webhook-signature.guard.spec.ts b/src/common/guards/webhook-signature.guard.spec.ts index 9bf6772d..100b3213 100644 --- a/src/common/guards/webhook-signature.guard.spec.ts +++ b/src/common/guards/webhook-signature.guard.spec.ts @@ -1,307 +1,131 @@ -import { describe, it, expect, vi } from 'vitest'; +import { describe, expect, it, vi } from 'vitest'; import { ExecutionContext, UnauthorizedException } from '@nestjs/common'; import { ConfigService } from '@nestjs/config'; -import * as crypto from 'crypto'; +import { createHmac } from 'crypto'; import { WebhookSignatureGuard } from './webhook-signature.guard'; import type { WebhookSecretResolver } from './webhook-signature.guard'; -import { - WEBHOOK_SIGNATURE_HEADER, - WEBHOOK_TIMESTAMP_HEADER, -} from '../constants/headers'; +import { WEBHOOK_SIGNATURE_HEADER, WEBHOOK_TIMESTAMP_HEADER } from '../constants/headers'; describe('WebhookSignatureGuard', () => { const secret = 'test-webhook-secret'; - const mockConfigService = { - get: (key: string) => { - if (key === 'WEBHOOK_SECRET') return secret; - return null; - }, + const config = { + get: (key: string) => (key === 'WEBHOOK_INGRESS_SECRET' ? secret : undefined), } as unknown as ConfigService; - - const guard = new WebhookSignatureGuard(mockConfigService); - - function createMockContext(headers: Record, body: unknown): ExecutionContext { + const guard = new WebhookSignatureGuard(config); + + function context( + timestamp: string, + signature: string, + rawBody: Buffer | undefined, + integrationId?: string, + ): ExecutionContext { return { switchToHttp: () => ({ getRequest: () => ({ - headers, - body, - rawBody: typeof body === 'string' ? body : JSON.stringify(body), + headers: { + [WEBHOOK_SIGNATURE_HEADER]: signature, + [WEBHOOK_TIMESTAMP_HEADER]: timestamp, + ...(integrationId ? { 'x-integration-id': integrationId } : {}), + }, + body: rawBody?.toString('utf8'), + rawBody, }), }), } as unknown as ExecutionContext; } - it('should accept valid signature and fresh timestamp', () => { - const timestamp = Math.floor(Date.now() / 1000); - const body = { event: 'soroban.contract.invoked', ledger: 12345 }; - const rawBody = JSON.stringify(body); - - const hmac = crypto.createHmac('sha256', secret); - hmac.update(`${timestamp}.${rawBody}`); - const signature = hmac.digest('hex'); + function signatureFor(key: string, timestamp: string, body: Buffer): string { + return createHmac('sha256', key) + .update(Buffer.concat([Buffer.from(`${timestamp}.`, 'utf8'), body])) + .digest('hex'); + } - const context = createMockContext( - { - [WEBHOOK_SIGNATURE_HEADER]: signature, - [WEBHOOK_TIMESTAMP_HEADER]: timestamp.toString(), - }, - body, - ); + it('accepts a fresh signature over the exact raw request bytes', async () => { + const timestamp = Math.floor(Date.now() / 1000).toString(); + const rawBody = Buffer.from('{ "event": "test", "data": {"a":1} }'); + const requestContext = context(timestamp, signatureFor(secret, timestamp, rawBody), rawBody); - expect(guard.canActivate(context)).toBe(true); + await expect(guard.canActivate(requestContext)).resolves.toBe(true); }); - it('should reject requests with expired timestamp (> 5 minutes)', () => { - const timestamp = Math.floor(Date.now() / 1000) - 400; // ~6.6 minutes ago - const body = { event: 'soroban.contract.invoked' }; - const rawBody = JSON.stringify(body); - - const hmac = crypto.createHmac('sha256', secret); - hmac.update(`${timestamp}.${rawBody}`); - const signature = hmac.digest('hex'); + it('accepts the versioned v1= signature format used for outbound deliveries', async () => { + const timestamp = Math.floor(Date.now() / 1000).toString(); + const rawBody = Buffer.from('{"event":"test"}'); + const requestContext = context(timestamp, `v1=${signatureFor(secret, timestamp, rawBody)}`, rawBody); - const context = createMockContext( - { - [WEBHOOK_SIGNATURE_HEADER]: signature, - [WEBHOOK_TIMESTAMP_HEADER]: timestamp.toString(), - }, - body, - ); - - expect(() => guard.canActivate(context)).toThrow(UnauthorizedException); + await expect(guard.canActivate(requestContext)).resolves.toBe(true); }); - it('should reject requests with modified body / invalid signature', () => { - const timestamp = Math.floor(Date.now() / 1000); - const body = { event: 'soroban.contract.invoked', amount: 100 }; - const tamperedBody = { event: 'soroban.contract.invoked', amount: 999 }; - - const hmac = crypto.createHmac('sha256', secret); - hmac.update(`${timestamp}.${JSON.stringify(body)}`); - const signature = hmac.digest('hex'); + it('rejects missing or malformed signatures and timestamps', async () => { + const timestamp = Math.floor(Date.now() / 1000).toString(); + const rawBody = Buffer.from('{}'); + const valid = signatureFor(secret, timestamp, rawBody); - const context = createMockContext( - { - [WEBHOOK_SIGNATURE_HEADER]: signature, - [WEBHOOK_TIMESTAMP_HEADER]: timestamp.toString(), - }, - tamperedBody, + await expect(guard.canActivate(context(timestamp, '', rawBody))).rejects.toBeInstanceOf( + UnauthorizedException, ); - - expect(() => guard.canActivate(context)).toThrow(UnauthorizedException); - }); - - it('should reject requests with missing signature header', () => { - const context = createMockContext( - { [WEBHOOK_TIMESTAMP_HEADER]: Math.floor(Date.now() / 1000).toString() }, - { foo: 'bar' }, + await expect(guard.canActivate(context(timestamp, 'not-a-signature', rawBody))).rejects.toBeInstanceOf( + UnauthorizedException, ); - expect(() => guard.canActivate(context)).toThrow(UnauthorizedException); - }); - - it('should reject requests with missing timestamp header', () => { - const context = createMockContext( - { [WEBHOOK_SIGNATURE_HEADER]: 'abc123' }, - { foo: 'bar' }, + await expect(guard.canActivate(context('not-a-timestamp', valid, rawBody))).rejects.toBeInstanceOf( + UnauthorizedException, ); - expect(() => guard.canActivate(context)).toThrow(UnauthorizedException); - }); - - it('should reject requests with non-numeric timestamp', () => { - const context = createMockContext( - { - [WEBHOOK_SIGNATURE_HEADER]: 'abc123', - [WEBHOOK_TIMESTAMP_HEADER]: 'not-a-number', - }, - { foo: 'bar' }, + await expect(guard.canActivate(context(timestamp, valid, undefined))).rejects.toBeInstanceOf( + UnauthorizedException, ); - expect(() => guard.canActivate(context)).toThrow(UnauthorizedException); }); - it('should reject when no secret is configured', () => { - const noSecretConfig = { - get: () => null, - } as unknown as ConfigService; - const guardNoSecret = new WebhookSignatureGuard(noSecretConfig); + it('rejects a modified body and an expired timestamp', async () => { + const timestamp = Math.floor(Date.now() / 1000).toString(); + const signedBody = Buffer.from('{"value":1}'); + const tamperedBody = Buffer.from('{"value":2}'); + const signature = signatureFor(secret, timestamp, signedBody); - const timestamp = Math.floor(Date.now() / 1000); - const context = createMockContext( - { - [WEBHOOK_SIGNATURE_HEADER]: 'abc123', - [WEBHOOK_TIMESTAMP_HEADER]: timestamp.toString(), - }, - { foo: 'bar' }, + await expect(guard.canActivate(context(timestamp, signature, tamperedBody))).rejects.toBeInstanceOf( + UnauthorizedException, ); - expect(() => guardNoSecret.canActivate(context)).toThrow(UnauthorizedException); + const expired = (Number(timestamp) - 400).toString(); + await expect( + guard.canActivate(context(expired, signatureFor(secret, expired, signedBody), signedBody)), + ).rejects.toBeInstanceOf(UnauthorizedException); }); - describe('per-integration secret resolver', () => { - it('should use custom secret resolver when provided', () => { - const customSecret = 'integration-specific-secret'; - const secretResolver: WebhookSecretResolver = vi.fn().mockReturnValue(customSecret); - - const guardWithResolver = new WebhookSignatureGuard(mockConfigService, { - secretResolver, - }); - - const timestamp = Math.floor(Date.now() / 1000); - const body = { event: 'test' }; - const rawBody = JSON.stringify(body); - - const hmac = crypto.createHmac('sha256', customSecret); - hmac.update(`${timestamp}.${rawBody}`); - const signature = hmac.digest('hex'); - - const context = createMockContext( - { - [WEBHOOK_SIGNATURE_HEADER]: signature, - [WEBHOOK_TIMESTAMP_HEADER]: timestamp.toString(), - 'x-integration-id': 'integration-123', - }, - body, - ); - - expect(guardWithResolver.canActivate(context)).toBe(true); - expect(secretResolver).toHaveBeenCalledWith( - expect.objectContaining({ headers: expect.any(Object) }), - 'integration-123', - ); - }); - - it('should fall back to global secret when resolver returns undefined', () => { - const secretResolver: WebhookSecretResolver = vi.fn().mockReturnValue(undefined); - - const guardWithResolver = new WebhookSignatureGuard(mockConfigService, { - secretResolver, - }); + it('rejects requests when no inbound secret is configured', async () => { + const guardWithoutSecret = new WebhookSignatureGuard({ get: () => undefined } as unknown as ConfigService); + const timestamp = Math.floor(Date.now() / 1000).toString(); + const rawBody = Buffer.from('{}'); - const timestamp = Math.floor(Date.now() / 1000); - const body = { event: 'test' }; - const rawBody = JSON.stringify(body); - - const hmac = crypto.createHmac('sha256', secret); - hmac.update(`${timestamp}.${rawBody}`); - const signature = hmac.digest('hex'); - - const context = createMockContext( - { - [WEBHOOK_SIGNATURE_HEADER]: signature, - [WEBHOOK_TIMESTAMP_HEADER]: timestamp.toString(), - }, - body, - ); - - expect(guardWithResolver.canActivate(context)).toBe(true); - }); - - it('should reject with custom secret when signature does not match', () => { - const customSecret = 'integration-specific-secret'; - const secretResolver: WebhookSecretResolver = vi.fn().mockReturnValue(customSecret); - - const guardWithResolver = new WebhookSignatureGuard(mockConfigService, { - secretResolver, - }); - - const timestamp = Math.floor(Date.now() / 1000); - const body = { event: 'test' }; - const rawBody = JSON.stringify(body); - - // Sign with the WRONG secret - const hmac = crypto.createHmac('sha256', 'wrong-secret'); - hmac.update(`${timestamp}.${rawBody}`); - const signature = hmac.digest('hex'); - - const context = createMockContext( - { - [WEBHOOK_SIGNATURE_HEADER]: signature, - [WEBHOOK_TIMESTAMP_HEADER]: timestamp.toString(), - }, - body, - ); - - expect(() => guardWithResolver.canActivate(context)).toThrow(UnauthorizedException); - }); + await expect( + guardWithoutSecret.canActivate( + context(timestamp, signatureFor(secret, timestamp, rawBody), rawBody), + ), + ).rejects.toBeInstanceOf(UnauthorizedException); }); - describe('custom tolerance', () => { - it('should respect custom tolerance window', () => { - const guardShortTolerance = new WebhookSignatureGuard(mockConfigService, { - toleranceSeconds: 60, // 1 minute - }); - - const timestamp = Math.floor(Date.now() / 1000) - 90; // 90 seconds ago - const body = { event: 'test' }; - const rawBody = JSON.stringify(body); - - const hmac = crypto.createHmac('sha256', secret); - hmac.update(`${timestamp}.${rawBody}`); - const signature = hmac.digest('hex'); - - const context = createMockContext( - { - [WEBHOOK_SIGNATURE_HEADER]: signature, - [WEBHOOK_TIMESTAMP_HEADER]: timestamp.toString(), - }, - body, - ); - - expect(() => guardShortTolerance.canActivate(context)).toThrow(UnauthorizedException); - }); + it('awaits an integration-specific secret resolver', async () => { + const integrationSecret = 'integration-specific-secret'; + const resolver: WebhookSecretResolver = vi.fn().mockResolvedValue(integrationSecret); + const guarded = new WebhookSignatureGuard(config, { secretResolver: resolver }); + const timestamp = Math.floor(Date.now() / 1000).toString(); + const rawBody = Buffer.from('{"event":"partner.callback"}'); + + await expect( + guarded.canActivate( + context(timestamp, signatureFor(integrationSecret, timestamp, rawBody), rawBody, 'partner-1'), + ), + ).resolves.toBe(true); + expect(resolver).toHaveBeenCalledWith(expect.any(Object), 'partner-1'); }); - describe('raw body extraction', () => { - it('should prefer rawBody over parsed body for HMAC computation', () => { - const timestamp = Math.floor(Date.now() / 1000); - const rawBody = '{"event":"test","data":{"a":1}}'; - - const hmac = crypto.createHmac('sha256', secret); - hmac.update(`${timestamp}.${rawBody}`); - const signature = hmac.digest('hex'); - - const request = { - headers: { - [WEBHOOK_SIGNATURE_HEADER]: signature, - [WEBHOOK_TIMESTAMP_HEADER]: timestamp.toString(), - }, - body: { event: 'test', data: { a: 1 } }, // parsed body (may have different key order) - rawBody: Buffer.from(rawBody), - }; - - const context = { - switchToHttp: () => ({ - getRequest: () => request, - }), - } as unknown as ExecutionContext; - - expect(guard.canActivate(context)).toBe(true); - }); - - it('should fall back to JSON.stringify when rawBody is not set', () => { - const timestamp = Math.floor(Date.now() / 1000); - const body = { event: 'test' }; - const rawBody = JSON.stringify(body); - - const hmac = crypto.createHmac('sha256', secret); - hmac.update(`${timestamp}.${rawBody}`); - const signature = hmac.digest('hex'); - - const request = { - headers: { - [WEBHOOK_SIGNATURE_HEADER]: signature, - [WEBHOOK_TIMESTAMP_HEADER]: timestamp.toString(), - }, - body, - }; - - const context = { - switchToHttp: () => ({ - getRequest: () => request, - }), - } as unknown as ExecutionContext; + it('does not fall back to a global secret when an integration resolver cannot select one', async () => { + const resolver: WebhookSecretResolver = vi.fn().mockResolvedValue(undefined); + const guarded = new WebhookSignatureGuard(config, { secretResolver: resolver }); + const timestamp = Math.floor(Date.now() / 1000).toString(); + const rawBody = Buffer.from('{}'); - expect(guard.canActivate(context)).toBe(true); - }); + await expect( + guarded.canActivate(context(timestamp, signatureFor(secret, timestamp, rawBody), rawBody)), + ).rejects.toBeInstanceOf(UnauthorizedException); }); }); diff --git a/src/common/guards/webhook-signature.guard.ts b/src/common/guards/webhook-signature.guard.ts index 54317966..9441515d 100644 --- a/src/common/guards/webhook-signature.guard.ts +++ b/src/common/guards/webhook-signature.guard.ts @@ -7,61 +7,28 @@ import { UnauthorizedException, } from '@nestjs/common'; import { ConfigService } from '@nestjs/config'; -import * as crypto from 'crypto'; import { Request } from 'express'; +import { createHmac, timingSafeEqual } from 'crypto'; import { WEBHOOK_SIGNATURE_HEADER, WEBHOOK_TIMESTAMP_HEADER, } from '../constants/headers'; -/** - * Resolves the HMAC secret for a given webhook integration. - * Returns the secret string, or undefined if no secret is configured. - */ export type WebhookSecretResolver = ( request: Request, integrationId?: string, ) => string | undefined | Promise; -/** - * Configuration options for `WebhookSignatureGuard`. - */ export interface WebhookSignatureGuardOptions { - /** - * Custom secret resolver for per-integration webhook secrets. - * When provided, this function is called first to resolve the secret. - * If it returns `undefined`, the guard falls back to global env-var secrets. - */ secretResolver?: WebhookSecretResolver; - /** - * Timestamp tolerance in seconds (default: 300 = 5 minutes). - * Requests with timestamps older than this are rejected to prevent replay attacks. - */ toleranceSeconds?: number; } /** - * Validates HMAC-SHA256 signature and timestamp freshness on incoming - * webhook events from external partner services and oracle providers. - * - * Signature calculation: HMAC-SHA256(timestamp + '.' + rawBody, secret) - * - * Security features: - * - Constant-time comparison via `crypto.timingSafeEqual` prevents timing attacks - * - Timestamp tolerance (default 5 min) prevents replay attacks - * - Configurable per-integration secret resolution for multi-tenant setups - * - Raw body buffering ensures accurate HMAC computation - * - * @example - * ```ts - * // Apply with default global secret: - * @UseGuards(WebhookSignatureGuard) - * - * // Apply with per-integration secret resolver: - * @UseGuards(new WebhookSignatureGuard(configService, { - * secretResolver: (req) => getIntegrationSecret(req.headers['x-integration-id']), - * })) - * ``` + * Verifies inbound signatures in the format + * HMAC-SHA256(secret, timestamp + "." + exact raw request bytes), encoded as + * 64 hexadecimal characters. The ingress secret is deliberately distinct from + * per-subscriber outbound signing secrets. */ @Injectable() export class WebhookSignatureGuard implements CanActivate { @@ -77,103 +44,58 @@ export class WebhookSignatureGuard implements CanActivate { this.secretResolver = options?.secretResolver; } - canActivate(context: ExecutionContext): boolean { - const request = context.switchToHttp().getRequest(); + async canActivate(context: ExecutionContext): Promise { + const request = context.switchToHttp().getRequest(); + const signature = request.headers[WEBHOOK_SIGNATURE_HEADER]; + const timestamp = request.headers[WEBHOOK_TIMESTAMP_HEADER]; - const signature = request.headers[WEBHOOK_SIGNATURE_HEADER] as string; - const timestampStr = request.headers[WEBHOOK_TIMESTAMP_HEADER] as string; - - if (!signature) { - throw new UnauthorizedException('Missing webhook signature header'); + if (typeof signature !== 'string' || typeof timestamp !== 'string') { + throw new UnauthorizedException('Missing webhook signature headers'); } - - if (!timestampStr) { - throw new UnauthorizedException('Missing webhook timestamp header'); + if (!/^\d{1,12}$/.test(timestamp)) { + throw new UnauthorizedException('Malformed webhook signature'); } - const timestamp = parseInt(timestampStr, 10); - if (isNaN(timestamp)) { - throw new UnauthorizedException('Invalid webhook timestamp'); + const signatureMatch = /^(?:v1=)?([0-9a-fA-F]{64})$/.exec(signature.trim()); + if (!signatureMatch) { + throw new UnauthorizedException('Malformed webhook signature'); } + const normalizedSignature = signatureMatch[1].toLowerCase(); + const timestampSeconds = Number(timestamp); const now = Math.floor(Date.now() / 1000); - if (Math.abs(now - timestamp) > this.toleranceSeconds) { + if (Math.abs(now - timestampSeconds) > this.toleranceSeconds) { throw new UnauthorizedException('Webhook timestamp expired or out of tolerance'); } - // Resolve the secret: try per-integration resolver first, then fall back to global env vars. - const secret = this.resolveSecret(request); + const secret = await this.resolveSecret(request); if (!secret) { - this.logger.warn('No webhook secret configured; rejecting request'); + this.logger.warn('No webhook ingress secret configured; rejecting request'); throw new UnauthorizedException('No webhook signing secret configured'); } - const payload = this.extractPayload(request); - - const expectedPayloadToSign = `${timestamp}.${payload}`; - const hmac = crypto.createHmac('sha256', secret); - hmac.update(expectedPayloadToSign); - const expectedSignature = hmac.digest('hex'); + if (!Buffer.isBuffer(request.rawBody)) { + throw new UnauthorizedException('Raw webhook body is unavailable'); + } - const signatureBuffer = Buffer.from(signature, 'utf8'); - const expectedBuffer = Buffer.from(expectedSignature, 'utf8'); + const expected = createHmac('sha256', secret) + .update(Buffer.concat([Buffer.from(`${timestamp}.`, 'utf8'), request.rawBody])) + .digest(); + const received = Buffer.from(normalizedSignature, 'hex'); - if ( - signatureBuffer.length !== expectedBuffer.length || - !crypto.timingSafeEqual(signatureBuffer, expectedBuffer) - ) { + if (received.length !== expected.length || !timingSafeEqual(received, expected)) { throw new UnauthorizedException('Invalid webhook cryptographic signature'); } return true; } - /** - * Resolves the HMAC signing secret for the incoming request. - * Priority: per-integration resolver → WEBHOOK_SECRET → STELLAR_WEBHOOK_SECRET. - */ - private resolveSecret(request: Request): string | undefined { - // Per-integration secret resolver (synchronous or async) - const integrationId = request.headers['x-integration-id'] as string | undefined; + private async resolveSecret(request: Request): Promise { if (this.secretResolver) { - const resolved = this.secretResolver(request, integrationId); - // Handle both sync and async resolvers - if (resolved && typeof (resolved as Promise).then === 'function') { - // Async resolver — store the promise for later use - // Note: canActivate doesn't support async in sync mode, so we - // log a warning. For async secret resolution, use the async guard variant. - this.logger.warn( - 'Async secret resolver returned a Promise; use the async canActivate variant for async secret resolution', - ); - } else if (resolved) { - return resolved as string; - } - } - - // Global env-var fallback - return ( - this.configService.get('WEBHOOK_SECRET') || - this.configService.get('STELLAR_WEBHOOK_SECRET') || - this.configService.get('WEBHOOK_SIGNING_SECRET') || - undefined - ); - } - - /** - * Extracts the request payload for HMAC computation. - * Prefers the raw body captured by `RawBodyMiddleware`, falling back to - * the parsed body if raw body is not available. - */ - private extractPayload(request: Request): string { - const rawBody = (request as Request & { rawBody?: Buffer | string }).rawBody; - if (rawBody !== undefined) { - return Buffer.isBuffer(rawBody) ? rawBody.toString('utf8') : rawBody; - } - - if (typeof request.body === 'string') { - return request.body; + const integrationId = request.headers['x-integration-id']; + return this.secretResolver(request, typeof integrationId === 'string' ? integrationId : undefined); } - return JSON.stringify(request.body ?? {}); + return this.configService.get('WEBHOOK_INGRESS_SECRET') || undefined; } } diff --git a/src/config/env.validation.spec.ts b/src/config/env.validation.spec.ts index 9d93cc15..4d37f213 100644 --- a/src/config/env.validation.spec.ts +++ b/src/config/env.validation.spec.ts @@ -50,6 +50,30 @@ describe('assertValidEnvironment', () => { expect(env.REDIS_HOST).toBe('localhost'); expect(env.STELLAR_USE_MOCK).toBe(true); expect(env.DATABASE_URL).toBe(REQUIRED_ENV.DATABASE_URL); + expect(env.WEBHOOK_DELIVERY_MAX_ATTEMPTS).toBe(5); + expect(env.WEBHOOK_RETRY_DELAY_MS).toBe(2_000); + expect(env.WEBHOOK_HTTP_TIMEOUT_MS).toBe(5_000); + expect(env.WEBHOOK_FAILED_JOB_RETENTION_SECONDS).toBe(86_400); + expect(env.WEBHOOK_DEAD_LETTER_RETENTION_SECONDS).toBe(604_800); + }); + + it('bounds webhook retries and retention to prevent unbounded work or storage', () => { + const error = validationError({ + ...REQUIRED_ENV, + WEBHOOK_DELIVERY_MAX_ATTEMPTS: '26', + WEBHOOK_RETRY_DELAY_MS: '0', + WEBHOOK_HTTP_TIMEOUT_MS: '120001', + WEBHOOK_FAILED_JOB_RETENTION_SECONDS: '2592001', + WEBHOOK_DEAD_LETTER_RETENTION_SECONDS: '2592001', + }); + + expect(error.issues.map((issue) => issue.key).sort()).toEqual([ + 'WEBHOOK_DEAD_LETTER_RETENTION_SECONDS', + 'WEBHOOK_DELIVERY_MAX_ATTEMPTS', + 'WEBHOOK_FAILED_JOB_RETENTION_SECONDS', + 'WEBHOOK_HTTP_TIMEOUT_MS', + 'WEBHOOK_RETRY_DELAY_MS', + ]); }); it('accepts the documented .env.example as-is', () => { diff --git a/src/config/env.validation.ts b/src/config/env.validation.ts index e86ec988..dd2e885d 100644 --- a/src/config/env.validation.ts +++ b/src/config/env.validation.ts @@ -89,6 +89,15 @@ export const storageEnvSchema = z.object({ export const queueEnvSchema = z.object({ QUEUE_PREFIX: z.string().default('astroid'), QUEUE_CONCURRENCY: z.coerce.number().int().positive().default(5), + WEBHOOK_DELIVERY_MAX_ATTEMPTS: z.coerce.number().int().positive().max(25).default(5), + WEBHOOK_RETRY_DELAY_MS: z.coerce.number().int().positive().max(300_000).default(2_000), + WEBHOOK_HTTP_TIMEOUT_MS: z.coerce.number().int().positive().max(120_000).default(5_000), + WEBHOOK_FAILED_JOB_RETENTION_SECONDS: z.coerce.number().int().positive().max(2_592_000).default(86_400), + WEBHOOK_DEAD_LETTER_RETENTION_SECONDS: z.coerce.number().int().positive().max(2_592_000).default(604_800), +}); + +export const webhookEnvSchema = z.object({ + WEBHOOK_INGRESS_SECRET: z.string().min(32).optional(), }); export const throttleEnvSchema = z.object({ @@ -200,6 +209,7 @@ export const environmentSchema = z ...stellarEnvSchema.shape, ...storageEnvSchema.shape, ...queueEnvSchema.shape, + ...webhookEnvSchema.shape, ...throttleEnvSchema.shape, ...rateLimitEnvSchema.shape, ...metricsEnvSchema.shape, diff --git a/src/config/queue.config.ts b/src/config/queue.config.ts index f6c37250..40841f85 100644 --- a/src/config/queue.config.ts +++ b/src/config/queue.config.ts @@ -4,6 +4,11 @@ import { queueEnvSchema, validateEnv } from './env.validation'; export type QueueConfig = { prefix: string; concurrency: number; + webhookDeliveryMaxAttempts: number; + webhookRetryDelayMs: number; + webhookHttpTimeoutMs: number; + webhookFailedJobRetentionSeconds: number; + webhookDeadLetterRetentionSeconds: number; }; export const queueConfig = registerAs('queue', (): QueueConfig => { @@ -11,5 +16,10 @@ export const queueConfig = registerAs('queue', (): QueueConfig => { return { prefix: queueEnv.QUEUE_PREFIX, concurrency: queueEnv.QUEUE_CONCURRENCY, + webhookDeliveryMaxAttempts: queueEnv.WEBHOOK_DELIVERY_MAX_ATTEMPTS, + webhookRetryDelayMs: queueEnv.WEBHOOK_RETRY_DELAY_MS, + webhookHttpTimeoutMs: queueEnv.WEBHOOK_HTTP_TIMEOUT_MS, + webhookFailedJobRetentionSeconds: queueEnv.WEBHOOK_FAILED_JOB_RETENTION_SECONDS, + webhookDeadLetterRetentionSeconds: queueEnv.WEBHOOK_DEAD_LETTER_RETENTION_SECONDS, }; }); diff --git a/src/main.ts b/src/main.ts index 0c15d5f5..769f572c 100644 --- a/src/main.ts +++ b/src/main.ts @@ -24,7 +24,7 @@ async function bootstrap() { // that every migration shipped with this build has been applied — a // pending or failed migration throws there, so `create` rejects and the // process exits before accepting any traffic. - const app = await NestFactory.create(AppModule, { bufferLogs: true }); + const app = await NestFactory.create(AppModule, { bufferLogs: true, rawBody: true }); const config = app.get(ConfigService); const appConfig = config.getOrThrow('app'); const prisma = app.get(PrismaService); diff --git a/src/modules/webhooks/services/webhook-audit.service.spec.ts b/src/modules/webhooks/services/webhook-audit.service.spec.ts index ab85bbb5..525063cb 100644 --- a/src/modules/webhooks/services/webhook-audit.service.spec.ts +++ b/src/modules/webhooks/services/webhook-audit.service.spec.ts @@ -39,7 +39,7 @@ describe('WebhookAuditService', () => { entity: 'Webhook', entityId: 'wh-1', newValue: expect.objectContaining({ - url: 'https://example.com/hook', + endpointOrigin: 'https://example.com', eventName: 'transaction.completed', eventId: 'event-1', attemptsMade: 5, diff --git a/src/modules/webhooks/services/webhook-audit.service.ts b/src/modules/webhooks/services/webhook-audit.service.ts index 46fafe7f..5701b0f0 100644 --- a/src/modules/webhooks/services/webhook-audit.service.ts +++ b/src/modules/webhooks/services/webhook-audit.service.ts @@ -18,6 +18,14 @@ export interface WebhookDeliveryFailure { /** Audit action recorded for a webhook that exhausted every retry attempt. */ export const WEBHOOK_DELIVERY_FAILED_ACTION = 'WEBHOOK_DELIVERY_FAILED'; +function safeEndpointOrigin(endpoint: string): string { + try { + return new URL(endpoint).origin; + } catch { + return '[invalid endpoint]'; + } +} + /** * Writes permanently failed webhook deliveries into the compliance audit trail. * @@ -47,7 +55,7 @@ export class WebhookAuditService { entity: 'Webhook', entityId: failure.webhookId, newValue: { - url: failure.url, + endpointOrigin: safeEndpointOrigin(failure.url), eventName: failure.eventName ?? null, eventId: failure.eventId ?? null, attemptsMade: failure.attemptsMade, diff --git a/src/modules/webhooks/services/webhook-delivery.service.spec.ts b/src/modules/webhooks/services/webhook-delivery.service.spec.ts index e3238bce..f34324cd 100644 --- a/src/modules/webhooks/services/webhook-delivery.service.spec.ts +++ b/src/modules/webhooks/services/webhook-delivery.service.spec.ts @@ -127,6 +127,15 @@ describe('WebhookDeliveryService', () => { expect(callArgs?.[2]?.removeOnFail).toEqual({ age: 24 * 3600 }); }); + it('does not include the subscriber URL in queue logs', async () => { + const logger = { debug: vi.fn(), error: vi.fn() }; + Object.assign(service, { logger }); + + await service.queueDelivery(createJobData({ url: 'https://example.com/hook?token=secret-value' })); + + expect(logger.debug.mock.calls.flat().join(' ')).not.toContain('secret-value'); + }); + it('handles queueing multiple webhooks sequentially', async () => { const jobData1 = createJobData({ eventId: 'event-1' }); const jobData2 = createJobData({ eventId: 'event-2' }); diff --git a/src/modules/webhooks/services/webhook-delivery.service.ts b/src/modules/webhooks/services/webhook-delivery.service.ts index c17d51a7..52f25e63 100644 --- a/src/modules/webhooks/services/webhook-delivery.service.ts +++ b/src/modules/webhooks/services/webhook-delivery.service.ts @@ -47,11 +47,10 @@ export class WebhookDeliveryService { ...(hasMetadata ? { metadata } : {}), }; await this.webhookQueue.add(WEBHOOK_JOB_NAME, jobData, webhookJobOptions); - this.logger.debug(`Queued webhook delivery for ${data.eventName} to ${data.url}`); + this.logger.debug(`Queued webhook delivery for ${data.eventName} (${data.webhookId})`); } catch (error) { this.logger.error('Failed to queue webhook delivery'); throw error; } } } - diff --git a/src/modules/webhooks/webhook-delivery.integration.spec.ts b/src/modules/webhooks/webhook-delivery.integration.spec.ts new file mode 100644 index 00000000..4175c81d --- /dev/null +++ b/src/modules/webhooks/webhook-delivery.integration.spec.ts @@ -0,0 +1,191 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import type { Job } from 'bullmq'; +import { createHmac } from 'crypto'; +import { WebhookDispatcher } from './webhook.dispatcher'; +import { WebhookDeliveryService } from './services/webhook-delivery.service'; +import { WebhooksProcessor } from './webhooks.processor'; +import type { WebhookJobData } from './types/webhook-job.types'; +import { WEBHOOK_JOB_NAME, webhookJobOptions } from '../../queues/webhook.queue'; +import { verifyWebhookSignature } from './utils/signing'; +import { WebhookSignatureGuard } from '../../common/guards/webhook-signature.guard'; +import { WebhookIngressController } from './webhook-ingress.controller'; + +describe('webhook delivery integration', () => { + afterEach(() => { + vi.restoreAllMocks(); + vi.unstubAllGlobals(); + }); + + it('dispatches an event through BullMQ options and delivers the same signed payload', async () => { + const secret = 'whsec_integration_test_secret'; + const endpoint = 'https://subscriber.example.test/webhook'; + const queueAdd = vi.fn().mockResolvedValue({ id: 'job-1' }); + const deliveryService = new WebhookDeliveryService({ add: queueAdd } as never); + const dispatcher = new WebhookDispatcher( + { + findEnabledForEvent: vi.fn().mockResolvedValue([ + { id: 'wh-1', organizationId: 'org-1', url: endpoint, secret }, + ]), + } as never, + deliveryService, + ); + const prisma = { + webhook: { + findFirst: vi.fn().mockResolvedValue({ secret, url: endpoint, enabled: true }), + }, + webhookDelivery: { upsert: vi.fn().mockResolvedValue({}) }, + }; + const processor = new WebhooksProcessor(prisma as never); + const response = { ok: true, status: 202 }; + const fetch = vi.fn().mockResolvedValue(response); + vi.stubGlobal('fetch', fetch); + + await dispatcher.dispatch({ + eventId: 'evt-stable-1', + name: 'transaction.completed', + organizationId: 'org-1', + aggregateType: 'Transaction', + aggregateId: 'txn-1', + requestId: 'req-1', + correlationId: 'corr-1', + payload: { transactionId: 'txn-1', amount: '25' }, + occurredAt: new Date('2026-10-19T20:00:00.000Z'), + }); + + const [name, data, options] = queueAdd.mock.calls[0] as [ + string, + WebhookJobData, + typeof webhookJobOptions, + ]; + expect(name).toBe(WEBHOOK_JOB_NAME); + expect(options).toEqual(webhookJobOptions); + expect(options.attempts).toBeGreaterThan(1); + expect(options.backoff).toEqual(expect.objectContaining({ type: 'exponential' })); + expect(data).not.toHaveProperty('secret'); + + await processor.process({ + id: 'job-1', + name, + data, + attemptsMade: 0, + opts: options, + } as Job); + + expect(fetch).toHaveBeenCalledTimes(1); + const [url, request] = fetch.mock.calls[0]; + expect(url).toBe(endpoint); + expect(request.method).toBe('POST'); + expect(Buffer.isBuffer(request.body)).toBe(true); + expect(verifyWebhookSignature( + secret, + request.headers['x-astroid-timestamp'], + data.eventId, + request.body, + request.headers['x-astroid-signature'], + )).toBe(true); + expect(JSON.parse(request.body.toString('utf8'))).toEqual({ + event: 'transaction.completed', + occurredAt: '2026-10-19T20:00:00.000Z', + aggregateType: 'Transaction', + aggregateId: 'txn-1', + data: { transactionId: 'txn-1', amount: '25' }, + }); + }); + + it('retries a transient endpoint failure and keeps the event identity stable', async () => { + const secret = 'whsec_retry_test_secret'; + const endpoint = 'https://subscriber.example.test/webhook'; + const queueAdd = vi.fn().mockResolvedValue({ id: 'job-retry' }); + const dispatcher = new WebhookDispatcher( + { findEnabledForEvent: vi.fn().mockResolvedValue([{ id: 'wh-2', organizationId: 'org-1', url: endpoint }]) } as never, + new WebhookDeliveryService({ add: queueAdd } as never), + ); + const processor = new WebhooksProcessor({ + webhook: { + findFirst: vi.fn().mockResolvedValue({ secret, url: endpoint, enabled: true }), + }, + } as never); + const fetch = vi + .fn() + .mockResolvedValueOnce({ ok: false, status: 503 }) + .mockResolvedValueOnce({ ok: true, status: 200 }); + vi.stubGlobal('fetch', fetch); + + await dispatcher.dispatch({ + eventId: 'evt-retry-stable', + name: 'transaction.completed', + organizationId: 'org-1', + aggregateType: 'Transaction', + aggregateId: 'txn-2', + payload: { transactionId: 'txn-2' }, + occurredAt: new Date('2026-10-19T20:00:00.000Z'), + }); + const [name, data, options] = queueAdd.mock.calls[0] as [ + string, + WebhookJobData, + typeof webhookJobOptions, + ]; + const job = (attemptsMade: number) => + ({ id: 'job-retry', name, data, attemptsMade, opts: options }) as Job; + + await expect(processor.process(job(0))).rejects.toThrow('HTTP 503'); + await expect(processor.process(job(1))).resolves.toMatchObject({ success: true, statusCode: 200 }); + + expect(fetch).toHaveBeenCalledTimes(2); + expect(data.eventId).toBe('evt-retry-stable'); + expect(fetch.mock.calls[0][1].body).toEqual(fetch.mock.calls[1][1].body); + expect(fetch.mock.calls[0][1].headers['x-astroid-event-id']).toBe('evt-retry-stable'); + expect(fetch.mock.calls[1][1].headers['x-astroid-event-id']).toBe('evt-retry-stable'); + }); + + it('builds the receiver signature from the exact serialized delivery bytes', () => { + const secret = 'whsec_integration_test_secret'; + const timestamp = '1792430400'; + const eventId = 'evt-byte-exact'; + const rawBytes = Buffer.from('{ "amount": 25, "unit": "XLM" }'); + const signature = createHmac('sha256', secret) + .update(Buffer.concat([Buffer.from(`v1.${timestamp}.${eventId}.`), rawBytes])) + .digest('hex'); + + expect(verifyWebhookSignature(secret, timestamp, eventId, rawBytes, `v1=${signature}`)).toBe(true); + expect( + verifyWebhookSignature(secret, timestamp, eventId, Buffer.from('{"amount":25,"unit":"XLM"}'), `v1=${signature}`), + ).toBe(false); + }); + + it('allows only a correctly signed inbound callback to reach its handler', async () => { + const secret = 'ingress_secret_for_integration_tests'; + const timestamp = Math.floor(Date.now() / 1000).toString(); + const rawBody = Buffer.from( + JSON.stringify({ eventId: 'evt-inbound', eventType: 'partner.updated', data: { id: 'p-1' } }), + ); + const signature = createHmac('sha256', secret) + .update(Buffer.concat([Buffer.from(`${timestamp}.`), rawBody])) + .digest('hex'); + const ingressService = { handle: vi.fn().mockResolvedValue(undefined) }; + const controller = new WebhookIngressController(ingressService as never); + const guard = new WebhookSignatureGuard( + { get: () => secret } as never, + ); + const makeContext = (receivedSignature: string) => + ({ + switchToHttp: () => ({ + getRequest: () => ({ + headers: { + 'x-astroid-signature': receivedSignature, + 'x-astroid-timestamp': timestamp, + }, + rawBody, + body: JSON.parse(rawBody.toString('utf8')), + }), + }), + }) as never; + + await guard.canActivate(makeContext(signature)); + await controller.receive(JSON.parse(rawBody.toString('utf8'))); + expect(ingressService.handle).toHaveBeenCalledTimes(1); + + await expect(guard.canActivate(makeContext(`${signature.slice(0, -1)}0`))).rejects.toThrow(); + expect(ingressService.handle).toHaveBeenCalledTimes(1); + }); +}); diff --git a/src/modules/webhooks/webhook-failure-audit.spec.ts b/src/modules/webhooks/webhook-failure-audit.spec.ts index d94e07fe..4f6e1bc8 100644 --- a/src/modules/webhooks/webhook-failure-audit.spec.ts +++ b/src/modules/webhooks/webhook-failure-audit.spec.ts @@ -31,7 +31,15 @@ describe('WebhooksProcessor terminal failures', () => { beforeEach(() => { recordTerminalFailure = vi.fn().mockResolvedValue(undefined); processor = new WebhooksProcessor( - { webhook: { findFirst: vi.fn().mockResolvedValue({ secret: 'whsec_test' }) } } as never, + { + webhook: { + findFirst: vi.fn().mockResolvedValue({ + secret: 'whsec_test', + url: jobData.url, + enabled: true, + }), + }, + } as never, undefined, { recordTerminalFailure } as unknown as WebhookAuditService, ); @@ -75,7 +83,7 @@ describe('WebhooksProcessor terminal failures', () => { eventName: 'transaction.completed', eventId: 'event-1', attemptsMade: 5, - failedReason: expect.stringContaining('HTTP 503'), + failedReason: 'HTTP 503', responseStatus: 503, }); }); @@ -87,7 +95,6 @@ describe('WebhooksProcessor terminal failures', () => { statusText: 'Not Found', text: () => Promise.resolve('Not Found'), }); - await expect(processor.process(makeJob(0))).rejects.toBeInstanceOf(UnrecoverableError); expect(recordTerminalFailure).toHaveBeenCalledTimes(1); @@ -96,6 +103,29 @@ describe('WebhooksProcessor terminal failures', () => { ); }); + it('uses the current subscription URL instead of a stale URL in queued job data', async () => { + const currentUrl = 'https://current.example.com/hook'; + processor = new WebhooksProcessor( + { + webhook: { + findFirst: vi.fn().mockResolvedValue({ + secret: 'whsec_test', + url: currentUrl, + enabled: true, + }), + }, + } as never, + ); + fetchSpy.mockResolvedValue({ ok: true, status: 200 }); + + await processor.process({ + ...makeJob(0), + data: { ...jobData, url: 'https://stale.example.com/hook' }, + } as Job); + + expect(fetchSpy).toHaveBeenCalledWith(currentUrl, expect.any(Object)); + }); + it('never lets an audit failure mask the original delivery error', async () => { const logger = { warn: vi.fn(), error: vi.fn(), debug: vi.fn(), log: vi.fn() }; Object.assign(processor, { logger }); diff --git a/src/modules/webhooks/webhook.module.ts b/src/modules/webhooks/webhook.module.ts index 884316c9..5affa224 100644 --- a/src/modules/webhooks/webhook.module.ts +++ b/src/modules/webhooks/webhook.module.ts @@ -1,4 +1,4 @@ -import { MiddlewareConsumer, Module, NestModule, RequestMethod } from '@nestjs/common'; +import { Module } from '@nestjs/common'; import { BullModule } from '@nestjs/bullmq'; import { WebhookController } from './webhook.controller'; import { WebhookIngressController } from './webhook-ingress.controller'; @@ -12,7 +12,6 @@ import { WebhooksProcessor } from './webhooks.processor'; import { createWebhookQueueOptions } from '../../queues/webhook.queue'; import { redisConfig } from '../../config/redis.config'; import { MetricsModule } from '../metrics/metrics.module'; -import { RawBodyMiddleware } from '../../common/middleware/raw-body.middleware'; import { WebhookSignatureGuard } from '../../common/guards/webhook-signature.guard'; import { SlidingWindowThrottlerGuard } from '../../common/guards/sliding-window-throttler.guard'; @@ -52,10 +51,4 @@ import { SlidingWindowThrottlerGuard } from '../../common/guards/sliding-window- ], exports: [WebhookService], }) -export class WebhookModule implements NestModule { - configure(consumer: MiddlewareConsumer): void { - consumer - .apply(RawBodyMiddleware) - .forRoutes({ path: 'webhooks/receive', method: RequestMethod.POST }); - } -} +export class WebhookModule {} diff --git a/src/modules/webhooks/webhook.service.spec.ts b/src/modules/webhooks/webhook.service.spec.ts index 6f839941..87f7bef3 100644 --- a/src/modules/webhooks/webhook.service.spec.ts +++ b/src/modules/webhooks/webhook.service.spec.ts @@ -62,7 +62,9 @@ describe('Webhook signing & delivery', () => { beforeEach(() => { processor = new WebhooksProcessor({ - webhook: { findFirst: vi.fn().mockResolvedValue({ secret: SECRET }) }, + webhook: { + findFirst: vi.fn().mockResolvedValue({ secret: SECRET, url: URL, enabled: true }), + }, } as never); fetchSpy = vi.fn(); vi.stubGlobal('fetch', fetchSpy); @@ -142,7 +144,7 @@ describe('Webhook signing & delivery', () => { it('loads the secret by webhook and organization rather than from the job payload', async () => { const fallbackSecret = 'fallback-secret-123'; - const findFirst = vi.fn().mockResolvedValue({ secret: fallbackSecret }); + const findFirst = vi.fn().mockResolvedValue({ secret: fallbackSecret, url: URL, enabled: true }); const processorWithDatabaseSecret = new WebhooksProcessor({ webhook: { findFirst } } as never); fetchSpy.mockResolvedValue({ ok: true, status: 200, text: () => Promise.resolve('OK') }); @@ -163,7 +165,7 @@ describe('Webhook signing & delivery', () => { const [, opts] = fetchSpy.mock.calls[0]; expect(findFirst).toHaveBeenCalledWith({ where: { id: 'wh-1', organizationId: 'org-1' }, - select: { secret: true }, + select: { secret: true, url: true, enabled: true }, }); expect(job.data).not.toHaveProperty('secret'); expect(opts.headers['x-astroid-signature']).toMatch(/^v1=[0-9a-f]{64}$/); diff --git a/src/modules/webhooks/webhooks.processor.spec.ts b/src/modules/webhooks/webhooks.processor.spec.ts index 362e21e5..efc32356 100644 --- a/src/modules/webhooks/webhooks.processor.spec.ts +++ b/src/modules/webhooks/webhooks.processor.spec.ts @@ -35,7 +35,9 @@ describe('WebhooksProcessor', () => { beforeEach(() => { mockPrisma = { - webhook: { findFirst: vi.fn().mockResolvedValue({ secret: WEBHOOK_SECRET }) }, + webhook: { + findFirst: vi.fn().mockResolvedValue({ secret: WEBHOOK_SECRET, url: WEBHOOK_URL, enabled: true }), + }, }; // Access private property via type assertion processor = new WebhooksProcessor(mockPrisma as never); @@ -163,6 +165,16 @@ describe('WebhooksProcessor', () => { const job = createMockJob(); await expect(processor.process(job)).rejects.toThrow('HTTP 503'); }); + + it('retries HTTP 429 responses rather than treating them as permanent', async () => { + fetchSpy.mockResolvedValue({ + ok: false, + status: 429, + text: () => Promise.resolve('Too Many Requests'), + }); + + await expect(processor.process(createMockJob())).rejects.toThrow('HTTP 429'); + }); }); describe('transient error handling (retriable)', () => { @@ -263,5 +275,28 @@ describe('WebhooksProcessor', () => { expect(warn.mock.calls.flat().join(' ')).not.toContain(secretEcho); expect(error.mock.calls.flat().join(' ')).not.toContain(secretEcho); }); + + it('uses the configured outbound HTTP timeout', async () => { + const timeout = vi.spyOn(AbortSignal, 'timeout').mockReturnValue(new AbortController().signal); + const configuredProcessor = new WebhooksProcessor( + mockPrisma as never, + undefined, + undefined, + { get: () => ({ + prefix: 'test', + concurrency: 1, + webhookDeliveryMaxAttempts: 4, + webhookRetryDelayMs: 1_000, + webhookHttpTimeoutMs: 9_000, + webhookFailedJobRetentionSeconds: 3_600, + webhookDeadLetterRetentionSeconds: 7_200, + }) } as never, + ); + fetchSpy.mockResolvedValue({ ok: true, status: 200 }); + + await configuredProcessor.process(createMockJob()); + + expect(timeout).toHaveBeenCalledWith(9_000); + }); }); }); diff --git a/src/modules/webhooks/webhooks.processor.ts b/src/modules/webhooks/webhooks.processor.ts index ef37c5d2..0a0ffa2f 100644 --- a/src/modules/webhooks/webhooks.processor.ts +++ b/src/modules/webhooks/webhooks.processor.ts @@ -1,12 +1,14 @@ import { Processor, WorkerHost } from '@nestjs/bullmq'; import { Inject, Logger, Optional } from '@nestjs/common'; import { Job, UnrecoverableError } from 'bullmq'; +import { ConfigService } from '@nestjs/config'; import { Queues } from '../../queues/queues.constants'; import { WebhookJobData, WebhookJobResult } from './types/webhook-job.types'; import { signWebhookPayload, WEBHOOK_SIGNATURE_VERSION } from './utils/signing'; import { PrismaService } from '../../database/prisma.service'; import { WorkerMetricsService } from '../../modules/metrics/worker-metrics.service'; import { WebhookAuditService } from './services/webhook-audit.service'; +import { queueConfig, QueueConfig } from '../../config/queue.config'; import { WEBHOOK_DELIVERY_HEADER, WEBHOOK_EVENT_HEADER, @@ -18,11 +20,11 @@ import { /** * BullMQ job processor for webhook event delivery with exponential backoff + jitter. - * Implements the retry strategy required by issue #125: - * - 5 max attempts - * - Exponential backoff with 2000ms base and 20% randomized jitter + * Implements the configured retry strategy: + * - A bounded maximum number of attempts + * - Exponential backoff with randomized jitter * (prevents thundering herd against subscriber endpoints) - * - Non-transient error detection (400,401,403,404,422) prevents infinite retries + * - Permanent 3xx/4xx errors stop immediately; 408, 429, 5xx and network errors retry * - Persistent delivery status tracking (PENDING → RETRYING → FAILED/DELIVERED) * - Fail-safe: retry failures never crash the master process * @@ -38,6 +40,7 @@ import { OnModuleDestroy } from '@nestjs/common'; @Processor(Queues.Webhooks) export class WebhooksProcessor extends WorkerHost implements OnModuleDestroy { private readonly logger = new Logger(WebhooksProcessor.name); + private readonly config: QueueConfig; async onModuleDestroy(): Promise { if (this.worker) { @@ -58,14 +61,14 @@ export class WebhooksProcessor extends WorkerHost implements OnModuleDestroy { }); } } - private static readonly NON_TRANSIENT_STATUSES = new Set([400, 401, 403, 404, 422]); - constructor( @Optional() @Inject(PrismaService) private readonly prisma?: PrismaService, @Optional() private readonly workerMetrics?: WorkerMetricsService, @Optional() private readonly webhookAudit?: WebhookAuditService, + @Optional() configService?: ConfigService, ) { super(); + this.config = configService?.get('queue') ?? queueConfig(); } /** @@ -77,13 +80,14 @@ export class WebhooksProcessor extends WorkerHost implements OnModuleDestroy { job: Job, failedReason: string, responseStatus?: number, + endpoint = job.data.url, ): Promise { if (!this.webhookAudit) return; try { await this.webhookAudit.recordTerminalFailure({ webhookId: job.data.webhookId, organizationId: job.data.organizationId, - url: job.data.url, + url: endpoint, eventName: job.data.eventName, eventId: job.data.eventId, attemptsMade: job.attemptsMade + 1, @@ -98,37 +102,46 @@ export class WebhooksProcessor extends WorkerHost implements OnModuleDestroy { } } - private async resolveSecret(webhookId: string, organizationId: string): Promise { + private async resolveSubscription( + webhookId: string, + organizationId: string, + ): Promise<{ secret: string; url: string }> { const client = this.prisma?.workerClient ?? this.prisma; - if (!client) throw new UnrecoverableError('Webhook signing secret is unavailable'); + if (!client) throw new UnrecoverableError('Webhook subscription is unavailable'); const webhook = await client.webhook.findFirst({ where: { id: webhookId, organizationId }, - select: { secret: true }, + select: { secret: true, url: true, enabled: true }, }); - if (!webhook?.secret) throw new UnrecoverableError('Webhook signing secret is unavailable'); - return webhook.secret; + if (!webhook?.secret || !webhook.enabled || !webhook.url) { + throw new UnrecoverableError('Webhook subscription is unavailable'); + } + return { secret: webhook.secret, url: webhook.url }; } async process(job: Job): Promise { const jobName = job.name ?? 'webhook-delivery'; const execute = async (): Promise => { - const { webhookId, organizationId, url, eventName, payload, eventId, metadata } = job.data; + const { webhookId, organizationId, eventName, payload, eventId, metadata } = job.data; const requestTrace = metadata?.requestId ? ` requestId=${metadata.requestId}` : ''; - this.logger.debug(`Processing webhook ${webhookId} event ${eventName} attempt ${job.attemptsMade + 1}/5${requestTrace}`); + const maxAttempts = job.opts?.attempts ?? this.config.webhookDeliveryMaxAttempts; + this.logger.debug( + `Processing webhook ${webhookId} event ${eventName} attempt ${job.attemptsMade + 1}/${maxAttempts}${requestTrace}`, + ); let responseStatus: number | undefined; let errorMessage: string | undefined; - let isNonTransient = false; - + let endpoint = job.data.url; try { const body = Buffer.from(JSON.stringify(payload), 'utf8'); const timestamp = Math.floor(Date.now() / 1000).toString(); - const effectiveSecret = await this.resolveSecret(webhookId, organizationId); - const signature = signWebhookPayload(effectiveSecret, timestamp, eventId, body); + const subscription = await this.resolveSubscription(webhookId, organizationId); + endpoint = subscription.url; + const signature = signWebhookPayload(subscription.secret, timestamp, eventId, body); - const response = await fetch(url, { + const response = await fetch(subscription.url, { method: 'POST', + redirect: 'manual', headers: { 'content-type': 'application/json', [WEBHOOK_SIGNATURE_HEADER]: signature, @@ -140,39 +153,51 @@ export class WebhooksProcessor extends WorkerHost implements OnModuleDestroy { 'user-agent': 'Astroid-Webhook-Bot/1.0', }, body, - signal: AbortSignal.timeout(5000), + signal: AbortSignal.timeout(this.config.webhookHttpTimeoutMs), }); responseStatus = response.status; if (!response.ok) { errorMessage = `HTTP ${response.status}`; - isNonTransient = WebhooksProcessor.NON_TRANSIENT_STATUSES.has(response.status); + const isNonTransient = + response.status >= 300 && + response.status < 500 && + response.status !== 408 && + response.status !== 429; this.logger.warn(`Webhook ${webhookId} responded ${response.status}${requestTrace}`); if (isNonTransient) { - await this.persistState({ - webhookId, - organizationId, - eventName, - eventId, - payload, - status: 'FAILED', - attempts: job.attemptsMade + 1, - lastError: errorMessage, - responseStatus, - }); - // Non-transient (4xx): record the abandoned delivery before BullMQ - // moves it straight to the failed set. - await this.auditTerminalFailure(job, errorMessage ?? 'HTTP error', responseStatus); throw new UnrecoverableError(errorMessage); } throw new Error(errorMessage); } this.logger.debug(`Webhook ${webhookId} delivered successfully${requestTrace}`); } catch (error) { - if (error instanceof UnrecoverableError) throw error; - errorMessage = error instanceof Error ? error.message : 'Delivery attempt failed'; - const isLastAttempt = job.attemptsMade >= 4; - this.logger.error(`Webhook ${webhookId} failed attempt ${job.attemptsMade + 1}/5${requestTrace}`); + if (error instanceof UnrecoverableError) { + errorMessage = error.message; + await this.persistState({ + webhookId, + organizationId, + eventName, + eventId, + payload, + status: 'FAILED', + attempts: job.attemptsMade + 1, + lastError: errorMessage, + responseStatus, + }); + await this.auditTerminalFailure(job, errorMessage, responseStatus, endpoint); + throw error; + } + errorMessage = + responseStatus !== undefined + ? `HTTP ${responseStatus}` + : error instanceof Error && ['AbortError', 'TimeoutError'].includes(error.name) + ? 'Webhook request timed out' + : 'Webhook request failed'; + const isLastAttempt = job.attemptsMade >= maxAttempts - 1; + this.logger.error( + `Webhook ${webhookId} failed attempt ${job.attemptsMade + 1}/${maxAttempts}${requestTrace}`, + ); await this.persistState({ webhookId, organizationId, @@ -188,7 +213,7 @@ export class WebhooksProcessor extends WorkerHost implements OnModuleDestroy { this.logger.error(`Webhook ${webhookId} exhausted all retry attempts${requestTrace}`); // Retries are exhausted: the delivery is dead-lettered by the queue // failure listener, so record it permanently in the audit trail. - await this.auditTerminalFailure(job, errorMessage ?? 'unknown error', responseStatus); + await this.auditTerminalFailure(job, errorMessage ?? 'unknown error', responseStatus, endpoint); } throw error; } @@ -259,4 +284,3 @@ export class WebhooksProcessor extends WorkerHost implements OnModuleDestroy { } } } - diff --git a/src/queues/dlq.processor.ts b/src/queues/dlq.processor.ts index f8cb47bf..c15e98a7 100644 --- a/src/queues/dlq.processor.ts +++ b/src/queues/dlq.processor.ts @@ -41,9 +41,16 @@ export class DlqProcessor extends WorkerHost { `[DLQ-JOB-RECEIVED] Job ${originalJobId ?? job.id} from queue '${originalQueue}' (${originalJobName ?? 'unknown'}) permanently failed after ${attemptsMade} attempts. Reason: ${failedReason ?? 'Unknown'}`, ); - this.logger.debug( - `[DLQ-JOB-DETAILS] Payload: ${JSON.stringify(payload)} | Stacktrace: ${JSON.stringify(stacktrace ?? [])} | FailedAt: ${failedAt}`, - ); + if (originalQueue === Queues.Webhooks) { + const delivery = payload && typeof payload === 'object' ? (payload as Record) : {}; + this.logger.debug( + `[DLQ-JOB-DETAILS] Webhook ${String(delivery.webhookId ?? 'unknown')} event ${String(delivery.eventId ?? 'unknown')} | FailedAt: ${failedAt}`, + ); + } else { + this.logger.debug( + `[DLQ-JOB-DETAILS] Payload: ${JSON.stringify(payload)} | Stacktrace: ${JSON.stringify(stacktrace ?? [])} | FailedAt: ${failedAt}`, + ); + } await this.recordDeadLetterAudit(job.data); diff --git a/src/queues/queue-failure-listener.spec.ts b/src/queues/queue-failure-listener.spec.ts index cbc91a82..ed190bbf 100644 --- a/src/queues/queue-failure-listener.spec.ts +++ b/src/queues/queue-failure-listener.spec.ts @@ -94,6 +94,36 @@ describe('QueueFailureListener', () => { }); describe('handleFailed', () => { + it('logs safe webhook metadata without the sensitive delivery payload', async () => { + getJob.mockResolvedValue( + exhaustedJob({ + data: { + webhookId: 'wh-1', + organizationId: 'org-1', + eventName: 'transaction.completed', + eventId: 'evt-1', + payload: { customerEmail: 'private@example.com' }, + }, + }), + ); + + const record = await listener.handleFailed(Queues.Webhooks, { + jobId: 'job-123', + failedReason: 'HTTP 500', + }); + + expect(errorSpy).toHaveBeenCalledTimes(1); + const logged = loggedRecord(errorSpy); + expect(logged.payload).toEqual({ + webhookId: 'wh-1', + organizationId: 'org-1', + eventName: 'transaction.completed', + eventId: 'evt-1', + }); + expect(JSON.stringify(logged)).not.toContain('private@example.com'); + expect(record).not.toBeNull(); + }); + it('logs a structured error payload with the full failure context', async () => { getJob.mockResolvedValue(exhaustedJob()); @@ -228,7 +258,7 @@ describe('QueueFailureListener', () => { const line = String(errorSpy.mock.calls[0][0]); expect(line).not.toContain('whsec_live'); expect(line).not.toContain('abc.def.ghi'); - expect(loggedRecord(errorSpy).payload).toEqual({ webhookId: 'wh-1', secret: '[REDACTED]' }); + expect(loggedRecord(errorSpy).payload).toEqual({ webhookId: 'wh-1' }); const dlqAdd = add.mock.calls.find((call: unknown[]) => String(call[0]).startsWith('dlq:')); expect((dlqAdd?.[1] as { payload: unknown }).payload).toEqual({ diff --git a/src/queues/queue-failure-listener.ts b/src/queues/queue-failure-listener.ts index 65edbab7..c3f3fb6f 100644 --- a/src/queues/queue-failure-listener.ts +++ b/src/queues/queue-failure-listener.ts @@ -5,6 +5,7 @@ import { redisConfig } from '../config/redis.config'; import { isTerminalJobFailure } from '../workers/dlq.processor'; import { RequestContext } from '../common/context/request-context'; import { scrubForLog, scrubString } from '../utils/log-scrubber.util'; +import { queueConfig } from '../config/queue.config'; /** Correlation identifiers recovered from the job payload, when present. */ export interface JobTraceContext { @@ -187,7 +188,8 @@ export class QueueFailureListener implements OnModuleInit, OnModuleDestroy { ...record, failedReason: record.failedReason && scrubString(record.failedReason), stacktrace: record.stacktrace?.map(scrubString), - payload: scrubForLog(record.payload), + payload: + record.queue === Queues.Webhooks ? this.webhookLogMetadata(record.payload) : scrubForLog(record.payload), }); if (record.event === 'stalled') { this.logger.warn( @@ -196,6 +198,7 @@ export class QueueFailureListener implements OnModuleInit, OnModuleDestroy { ); return; } + this.logger.error( line, record.terminal @@ -204,6 +207,17 @@ export class QueueFailureListener implements OnModuleInit, OnModuleDestroy { ); } + private webhookLogMetadata(payload: unknown): unknown { + if (!payload || typeof payload !== 'object') return '[webhook payload omitted]'; + const data = payload as Record; + return { + webhookId: data.webhookId, + organizationId: data.organizationId, + eventName: data.eventName, + eventId: data.eventId, + }; + } + /** * Enqueues a copy of the terminal failure on the dead-letter queue for * administrative triage. Failures are logged, never propagated. @@ -231,9 +245,12 @@ export class QueueFailureListener implements OnModuleInit, OnModuleDestroy { }, }; + const webhookRetentionSeconds = + record.queue === Queues.Webhooks ? queueConfig().webhookDeadLetterRetentionSeconds : undefined; + const retentionSeconds = webhookRetentionSeconds ?? 7 * 24 * 3_600; await queue.add(`dlq:${record.queue}:${record.jobId}`, data, { - removeOnComplete: { count: 5_000 }, - removeOnFail: { age: 7 * 24 * 3_600 }, + removeOnComplete: { count: 5_000, age: retentionSeconds }, + removeOnFail: { age: retentionSeconds }, }); } catch (error) { this.logger.error( diff --git a/src/queues/webhook.queue.spec.ts b/src/queues/webhook.queue.spec.ts index 7964e112..a7d4f688 100644 --- a/src/queues/webhook.queue.spec.ts +++ b/src/queues/webhook.queue.spec.ts @@ -6,6 +6,7 @@ import { WEBHOOK_DEAD_LETTER_QUEUE, WEBHOOK_JOB_NAME, WEBHOOK_MAX_ATTEMPTS, + createWebhookJobOptions, createWebhookQueueOptions, webhookJobOptions, } from './webhook.queue'; @@ -30,18 +31,43 @@ describe('webhook queue configuration', () => { expect(options.defaultJobOptions).toEqual(webhookJobOptions); }); + it('uses configured attempt, delay, and failure retention values consistently', () => { + const config = { + prefix: 'test', + concurrency: 1, + webhookDeliveryMaxAttempts: 3, + webhookRetryDelayMs: 1_500, + webhookHttpTimeoutMs: 8_000, + webhookFailedJobRetentionSeconds: 3_600, + webhookDeadLetterRetentionSeconds: 7_200, + }; + const options = createWebhookQueueOptions(config); + const settings = options.settings as unknown as { + backoffStrategy: (attemptsMade: number) => number; + }; + + expect(createWebhookJobOptions(config)).toMatchObject({ + attempts: 3, + backoff: { type: 'exponential', delay: 1_500 }, + removeOnFail: { age: 3_600 }, + }); + expect(options.defaultJobOptions).toEqual(createWebhookJobOptions(config)); + expect(settings.backoffStrategy(1)).toBeGreaterThanOrEqual(1_500); + expect(settings.backoffStrategy(1)).toBeLessThan(1_800); + }); + it('attaches a jittered backoff strategy so retries never fire in lockstep', () => { const settings = createWebhookQueueOptions().settings as unknown as { backoffStrategy: (attemptsMade: number) => number; }; - const firstRetry = settings.backoffStrategy(0); + const firstRetry = settings.backoffStrategy(1); expect(firstRetry).toBeGreaterThanOrEqual(2_000); expect(firstRetry).toBeLessThan(2_400); // The second retry doubles the base delay while still staying inside the // 20% jitter envelope. - const secondRetry = settings.backoffStrategy(1); + const secondRetry = settings.backoffStrategy(2); expect(secondRetry).toBeGreaterThanOrEqual(4_000); expect(secondRetry).toBeLessThan(4_800); }); diff --git a/src/queues/webhook.queue.ts b/src/queues/webhook.queue.ts index 9fc3c8ca..c9b789e0 100644 --- a/src/queues/webhook.queue.ts +++ b/src/queues/webhook.queue.ts @@ -1,17 +1,22 @@ import type { JobsOptions } from 'bullmq'; import type { RegisterQueueOptions } from '@nestjs/bullmq'; -import { webhookBackoffStrategy } from '../utils/backoff.util'; +import { createWebhookBackoffStrategy } from '../utils/backoff.util'; import { Queues } from './queues.constants'; +import { queueConfig, QueueConfig } from '../config/queue.config'; /** BullMQ job name used for every outbound webhook delivery. */ export const WEBHOOK_JOB_NAME = 'webhook-delivery'; -/** Total delivery attempts (1 initial + 4 retries) before a webhook is dead-lettered. */ +/** Default total delivery attempts (1 initial + 4 retries). */ export const WEBHOOK_MAX_ATTEMPTS = 5; -/** Base delay in milliseconds for the exponential backoff between attempts. */ +/** Default base delay in milliseconds for exponential backoff. */ export const WEBHOOK_BACKOFF_BASE_DELAY_MS = 2_000; +/** Default HTTP request timeout in milliseconds. */ +export const WEBHOOK_HTTP_TIMEOUT_MS = 5_000; +/** Default retention for exhausted webhook jobs, in seconds. */ +export const WEBHOOK_FAILED_JOB_RETENTION_SECONDS = 24 * 3_600; /** Queue that permanently failed webhook deliveries are routed to for inspection. */ export const WEBHOOK_DEAD_LETTER_QUEUE = Queues.DeadLetter; @@ -19,36 +24,34 @@ export const WEBHOOK_DEAD_LETTER_QUEUE = Queues.DeadLetter; /** * Retry policy applied to every queued webhook delivery. * - * - `attempts: 5` bounds the work spent on a dead endpoint. - * - `backoff: exponential @ 2000ms` spaces retries out (2s, 4s, 8s, 16s). - * - `removeOnFail.age: 24h` keeps exhausted jobs inspectable (and re-drivable) + * - A configured attempt limit bounds the work spent on a dead endpoint. + * - Exponential backoff spaces retries out using a configured base delay. + * - Configured failed-job retention keeps exhausted jobs inspectable * long enough for an operator to act, without growing Redis forever. */ -export const webhookJobOptions: JobsOptions = { - attempts: WEBHOOK_MAX_ATTEMPTS, - backoff: { - type: 'exponential', - delay: WEBHOOK_BACKOFF_BASE_DELAY_MS, - }, - removeOnComplete: { count: 1_000 }, - removeOnFail: { age: 24 * 3_600 }, -}; +export function createWebhookJobOptions(config: QueueConfig = queueConfig()): JobsOptions { + return { + attempts: config.webhookDeliveryMaxAttempts, + backoff: { + type: 'exponential', + delay: config.webhookRetryDelayMs, + }, + removeOnComplete: { count: 1_000 }, + removeOnFail: { age: config.webhookFailedJobRetentionSeconds }, + }; +} + +export const webhookJobOptions: JobsOptions = createWebhookJobOptions(); /** * Builds the BullMQ registration for the webhook queue. - * - * The custom `backoffStrategy` adds ±20% randomized jitter on top of the - * exponential delay so a fleet of failing subscribers is not retried in - * lockstep (thundering herd). BullMQ reads the strategy from - * `queue.opts.settings.backoffStrategy` at retry time; `@nestjs/bullmq` does not - * surface that field on `RegisterQueueOptions`, hence the narrow cast. */ -export function createWebhookQueueOptions(): RegisterQueueOptions { +export function createWebhookQueueOptions(config: QueueConfig = queueConfig()): RegisterQueueOptions { return { name: Queues.Webhooks, - defaultJobOptions: webhookJobOptions, + defaultJobOptions: createWebhookJobOptions(config), settings: { - backoffStrategy: webhookBackoffStrategy, + backoffStrategy: createWebhookBackoffStrategy(config.webhookRetryDelayMs), } as RegisterQueueOptions['settings'], }; } diff --git a/src/utils/backoff.util.spec.ts b/src/utils/backoff.util.spec.ts index 28f9ec13..47f55634 100644 --- a/src/utils/backoff.util.spec.ts +++ b/src/utils/backoff.util.spec.ts @@ -71,17 +71,17 @@ describe('backoff.util', () => { it('returns a delay consistent with the backoff algorithm', () => { vi.mocked(Math.random).mockReturnValue(0); - // First retry (attemptsMade = 0 → attempt 1) - const delay1 = webhookBackoffStrategy(0, 'exponential', new Error('test')) as number; + // BullMQ passes 1 for the first failure that schedules a retry. + const delay1 = webhookBackoffStrategy(1, 'exponential', new Error('test')) as number; expect(delay1).toBe(2000); - // Second retry (attemptsMade = 1 → attempt 2) - const delay2 = webhookBackoffStrategy(1, 'exponential', new Error('test')) as number; + // BullMQ passes 2 for the second retry. + const delay2 = webhookBackoffStrategy(2, 'exponential', new Error('test')) as number; expect(delay2).toBe(4000); - // Third retry (attemptsMade = 2 → attempt 3) - const delay3 = webhookBackoffStrategy(2, 'exponential', new Error('test')) as number; - expect(delay3).toBe(8000); + // Third retry. + const thirdRetryDelay = webhookBackoffStrategy(3, 'exponential', new Error('test')) as number; + expect(thirdRetryDelay).toBe(8000); }); it('adds jitter to prevent thundering herd', () => { @@ -96,10 +96,10 @@ describe('backoff.util', () => { const uniqueDelays = new Set(delays); expect(uniqueDelays.size).toBeGreaterThan(1); - // All delays should be in the range [4000, 4800) + // BullMQ's attemptsMade = 1 is the first retry: [2000, 2400). for (const delay of delays) { - expect(delay).toBeGreaterThanOrEqual(4000); - expect(delay).toBeLessThanOrEqual(4800); + expect(delay).toBeGreaterThanOrEqual(2000); + expect(delay).toBeLessThanOrEqual(2400); } }); @@ -124,7 +124,7 @@ describe('backoff.util', () => { it('works when err and job are not provided', () => { vi.mocked(Math.random).mockReturnValue(0); - const delay = webhookBackoffStrategy(0) as number; + const delay = webhookBackoffStrategy(1) as number; expect(delay).toBe(2000); }); }); diff --git a/src/utils/backoff.util.ts b/src/utils/backoff.util.ts index 92cd8cd1..ffdb28f6 100644 --- a/src/utils/backoff.util.ts +++ b/src/utils/backoff.util.ts @@ -50,5 +50,11 @@ export function exponentialBackoffWithJitter( * @see https://docs.bullmq.io/guide/retrying-failing-jobs#custom-backoff-strategy */ export const webhookBackoffStrategy: BackoffStrategy = (attemptsMade) => { - return exponentialBackoffWithJitter(attemptsMade + 1, DEFAULT_BASE_DELAY_MS, DEFAULT_JITTER_FACTOR); + return exponentialBackoffWithJitter(Math.max(attemptsMade, 1), DEFAULT_BASE_DELAY_MS, DEFAULT_JITTER_FACTOR); }; + +/** Creates a webhook retry strategy using the configured base delay. */ +export function createWebhookBackoffStrategy(baseDelayMs: number): BackoffStrategy { + return (attemptsMade) => + exponentialBackoffWithJitter(Math.max(attemptsMade, 1), baseDelayMs, DEFAULT_JITTER_FACTOR); +} From f4fdc867e4eaeef52b3e2532058bb9a2dc48c37f Mon Sep 17 00:00:00 2001 From: omoh5 Date: Tue, 20 Oct 2026 09:55:52 +0000 Subject: [PATCH 2/2] feat(webhooks): add delivery retries and HMAC signature verification --- .../webhook-delivery.integration.spec.ts | 35 +++++++++++++++---- .../webhooks/webhook-failure-audit.spec.ts | 18 ++++++++++ .../webhooks/webhooks.processor.spec.ts | 6 ++++ src/modules/webhooks/webhooks.processor.ts | 24 ++++++++----- src/queues/queue-failure-listener.ts | 6 +++- src/queues/webhook.queue.ts | 21 +++++++++-- 6 files changed, 92 insertions(+), 18 deletions(-) diff --git a/src/modules/webhooks/webhook-delivery.integration.spec.ts b/src/modules/webhooks/webhook-delivery.integration.spec.ts index 4175c81d..8f395960 100644 --- a/src/modules/webhooks/webhook-delivery.integration.spec.ts +++ b/src/modules/webhooks/webhook-delivery.integration.spec.ts @@ -10,6 +10,21 @@ import { verifyWebhookSignature } from './utils/signing'; import { WebhookSignatureGuard } from '../../common/guards/webhook-signature.guard'; import { WebhookIngressController } from './webhook-ingress.controller'; +const createDestinationPolicy = () => ({ + post: async ( + url: string, + options: { headers: Record; body: Buffer; timeoutMs: number }, + ) => { + const response = await fetch(url, { + method: 'POST', + headers: options.headers, + body: options.body, + signal: AbortSignal.timeout(options.timeoutMs), + }); + return { status: response.status }; + }, +}); + describe('webhook delivery integration', () => { afterEach(() => { vi.restoreAllMocks(); @@ -35,7 +50,7 @@ describe('webhook delivery integration', () => { }, webhookDelivery: { upsert: vi.fn().mockResolvedValue({}) }, }; - const processor = new WebhooksProcessor(prisma as never); + const processor = new WebhooksProcessor(prisma as never, undefined, undefined, createDestinationPolicy()); const response = { ok: true, status: 202 }; const fetch = vi.fn().mockResolvedValue(response); vi.stubGlobal('fetch', fetch); @@ -100,11 +115,16 @@ describe('webhook delivery integration', () => { { findEnabledForEvent: vi.fn().mockResolvedValue([{ id: 'wh-2', organizationId: 'org-1', url: endpoint }]) } as never, new WebhookDeliveryService({ add: queueAdd } as never), ); - const processor = new WebhooksProcessor({ - webhook: { - findFirst: vi.fn().mockResolvedValue({ secret, url: endpoint, enabled: true }), - }, - } as never); + const processor = new WebhooksProcessor( + { + webhook: { + findFirst: vi.fn().mockResolvedValue({ secret, url: endpoint, enabled: true }), + }, + } as never, + undefined, + undefined, + createDestinationPolicy(), + ); const fetch = vi .fn() .mockResolvedValueOnce({ ok: false, status: 503 }) @@ -185,7 +205,8 @@ describe('webhook delivery integration', () => { await controller.receive(JSON.parse(rawBody.toString('utf8'))); expect(ingressService.handle).toHaveBeenCalledTimes(1); - await expect(guard.canActivate(makeContext(`${signature.slice(0, -1)}0`))).rejects.toThrow(); + const invalidSignature = `${signature.slice(0, -1)}${signature.endsWith('0') ? '1' : '0'}`; + await expect(guard.canActivate(makeContext(invalidSignature))).rejects.toThrow(); expect(ingressService.handle).toHaveBeenCalledTimes(1); }); }); diff --git a/src/modules/webhooks/webhook-failure-audit.spec.ts b/src/modules/webhooks/webhook-failure-audit.spec.ts index ea36360a..a538527c 100644 --- a/src/modules/webhooks/webhook-failure-audit.spec.ts +++ b/src/modules/webhooks/webhook-failure-audit.spec.ts @@ -5,6 +5,21 @@ import { WebhooksProcessor } from './webhooks.processor'; import { WebhookAuditService } from './services/webhook-audit.service'; import { WebhookJobData } from './types/webhook-job.types'; +const createDestinationPolicy = () => ({ + post: async ( + url: string, + options: { headers: Record; body: Buffer; timeoutMs: number }, + ) => { + const response = await fetch(url, { + method: 'POST', + headers: options.headers, + body: options.body, + signal: AbortSignal.timeout(options.timeoutMs), + }); + return { status: response.status }; + }, +}); + /** * Terminal-failure behaviour of the webhook processor: retries are scheduled * while attempts remain, and a delivery that exhausts them (or hits a @@ -127,6 +142,9 @@ describe('WebhooksProcessor terminal failures', () => { }), }, } as never, + undefined, + undefined, + createDestinationPolicy(), ); fetchSpy.mockResolvedValue({ ok: true, status: 200 }); diff --git a/src/modules/webhooks/webhooks.processor.spec.ts b/src/modules/webhooks/webhooks.processor.spec.ts index a0e3f281..c6e60cd1 100644 --- a/src/modules/webhooks/webhooks.processor.spec.ts +++ b/src/modules/webhooks/webhooks.processor.spec.ts @@ -292,6 +292,12 @@ describe('WebhooksProcessor', () => { mockPrisma as never, undefined, undefined, + { + post: async (_url, options) => { + AbortSignal.timeout(options.timeoutMs); + return { status: 200 }; + }, + }, { get: () => ({ prefix: 'test', concurrency: 1, diff --git a/src/modules/webhooks/webhooks.processor.ts b/src/modules/webhooks/webhooks.processor.ts index e36c7e8e..f5439208 100644 --- a/src/modules/webhooks/webhooks.processor.ts +++ b/src/modules/webhooks/webhooks.processor.ts @@ -1,5 +1,6 @@ import { Processor, WorkerHost } from '@nestjs/bullmq'; import { Inject, Logger, Optional } from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; import { Job, UnrecoverableError } from 'bullmq'; import { PermanentJobError } from '../../queues/queue-policy'; import { Queues } from '../../queues/queues.constants'; @@ -73,9 +74,10 @@ export class WebhooksProcessor extends WorkerHost implements OnModuleDestroy { @Optional() private readonly webhookAudit?: WebhookAuditService, @Optional() @Inject(WebhookDestinationPolicy) private readonly destinationPolicy: WebhookDestinationClient = new WebhookDestinationPolicy(), + @Optional() @Inject(ConfigService) private readonly configService?: ConfigService, ) { super(); - this.config = configService?.get('queue') ?? queueConfig(); + this.config = this.configService?.get('queue') ?? queueConfig(); } /** @@ -120,7 +122,7 @@ export class WebhooksProcessor extends WorkerHost implements OnModuleDestroy { select: { secret: true, url: true, enabled: true }, }); if (!webhook?.secret) throw new PermanentJobError('Webhook signing secret is unavailable'); - return webhook.secret; + return { secret: webhook.secret, url: webhook.url }; } async process(job: Job): Promise { @@ -144,7 +146,7 @@ export class WebhooksProcessor extends WorkerHost implements OnModuleDestroy { endpoint = subscription.url; const signature = signWebhookPayload(subscription.secret, timestamp, eventId, body); - const response = await this.destinationPolicy.post(url, { + const response = await this.destinationPolicy.post(endpoint, { headers: { 'content-type': 'application/json', [WEBHOOK_SIGNATURE_HEADER]: signature, @@ -156,14 +158,18 @@ export class WebhooksProcessor extends WorkerHost implements OnModuleDestroy { 'user-agent': 'Astroid-Webhook-Bot/1.0', }, body, - timeoutMs: 5000, + timeoutMs: this.config.webhookHttpTimeoutMs, }); responseStatus = response.status; if (response.status < 200 || response.status >= 300) { const isRedirect = response.status >= 300 && response.status < 400; errorMessage = isRedirect ? 'Webhook endpoint redirects are not supported' : `HTTP ${response.status}`; - isNonTransient = isRedirect || WebhooksProcessor.NON_TRANSIENT_STATUSES.has(response.status); + const isNonTransient = + isRedirect || + (response.status >= 400 && + response.status < 500 && + ![408, 425, 429].includes(response.status)); this.logger.warn(`Webhook ${webhookId} responded ${response.status}${requestTrace}`); if (isNonTransient) { await this.persistState({ @@ -180,7 +186,7 @@ export class WebhooksProcessor extends WorkerHost implements OnModuleDestroy { // Non-transient (4xx): record the abandoned delivery before BullMQ // moves it straight to the failed set. await this.auditTerminalFailure(job, errorMessage ?? 'HTTP error', responseStatus); - throw new PermanentJobError(errorMessage); + throw new PermanentJobError(errorMessage); } throw new Error(errorMessage); } @@ -206,8 +212,10 @@ export class WebhooksProcessor extends WorkerHost implements OnModuleDestroy { : error instanceof Error && /^HTTP \d{3}$/.test(error.message) ? error.message : 'Webhook delivery failed'; - const isLastAttempt = job.attemptsMade >= 4; - this.logger.error(`Webhook ${webhookId} failed attempt ${job.attemptsMade + 1}/5${requestTrace}`); + const isLastAttempt = job.attemptsMade >= maxAttempts - 1; + this.logger.error( + `Webhook ${webhookId} failed attempt ${job.attemptsMade + 1}/${maxAttempts}${requestTrace}`, + ); await this.persistState({ webhookId, organizationId, diff --git a/src/queues/queue-failure-listener.ts b/src/queues/queue-failure-listener.ts index 515ae9df..2e83cfa7 100644 --- a/src/queues/queue-failure-listener.ts +++ b/src/queues/queue-failure-listener.ts @@ -218,6 +218,10 @@ export class QueueFailureListener implements OnModuleInit, OnModuleDestroy { const safeReason = record.failedReason ?? 'unknown reason'; const line = JSON.stringify({ ...record, + payload: + record.queue === Queues.Webhooks + ? this.webhookLogMetadata(record.payload) + : record.payload, }); if (record.event === 'stalled') { this.logger.warn( @@ -278,7 +282,7 @@ export class QueueFailureListener implements OnModuleInit, OnModuleDestroy { const retentionSeconds = webhookRetentionSeconds ?? 7 * 24 * 3_600; await queue.add(`dlq:${record.queue}:${record.jobId}`, data, { removeOnComplete: { count: 5_000 }, - removeOnFail: { age: 7 * 24 * 3_600, count: 5_000 }, + removeOnFail: { age: retentionSeconds, count: 5_000 }, }); } catch (error) { this.logger.error( diff --git a/src/queues/webhook.queue.ts b/src/queues/webhook.queue.ts index e2a4f66e..905b38d7 100644 --- a/src/queues/webhook.queue.ts +++ b/src/queues/webhook.queue.ts @@ -1,8 +1,10 @@ import type { JobsOptions } from 'bullmq'; import type { RegisterQueueOptions } from '@nestjs/bullmq'; +import { queueConfig, type QueueConfig } from '../config/queue.config'; import { Queues } from './queues.constants'; import { buildJobOptions, createQueueRegistrationOptions } from './queue-policy'; +import { createWebhookBackoffStrategy } from '../utils/backoff.util'; /** BullMQ job name used for every outbound webhook delivery. */ export const WEBHOOK_JOB_NAME = 'webhook-delivery'; @@ -41,6 +43,21 @@ export const webhookJobOptions: JobsOptions = buildJobOptions({ /** * Builds the BullMQ registration for the webhook queue. */ -export function createWebhookQueueOptions(): RegisterQueueOptions { - return createQueueRegistrationOptions(Queues.Webhooks, webhookJobOptions); +export function createWebhookJobOptions(config: QueueConfig = queueConfig()): JobsOptions { + return buildJobOptions({ + attempts: config.webhookDeliveryMaxAttempts, + backoff: { type: 'exponential', delay: config.webhookRetryDelayMs }, + removeOnFail: { age: config.webhookFailedJobRetentionSeconds }, + }); +} + +export function createWebhookQueueOptions(config: QueueConfig = queueConfig()): RegisterQueueOptions { + const options = createQueueRegistrationOptions(Queues.Webhooks, createWebhookJobOptions(config)); + return { + ...options, + settings: { + ...options.settings, + backoffStrategy: createWebhookBackoffStrategy(config.webhookRetryDelayMs), + } as RegisterQueueOptions['settings'], + }; }