From 465ceae00b862b3cf6af45d899b9d4158f7635ed Mon Sep 17 00:00:00 2001 From: Hamzasaheed Date: Sat, 17 Oct 2026 08:13:53 +0100 Subject: [PATCH] =?UTF-8?q?fix:=20PREWAVE-BE-059=20=E2=80=94=20Sign=20Outb?= =?UTF-8?q?ound=20Webhooks=20and=20Make=20Delivery=20Id=20(#655)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/NOTIFICATION_ARCHITECTURE.md | 48 +- prisma/schema.prisma | 150 +++++- src/app.module.ts | 4 + .../entities/webhook-delivery.entity.ts | 82 +++ .../entities/webhook-endpoint.entity.ts | 79 +++ src/notifications/outbound-webhook.module.ts | 12 + src/notifications/outbound-webhook.service.ts | 480 ++++++++++++++++++ 7 files changed, 851 insertions(+), 4 deletions(-) create mode 100644 src/notifications/entities/webhook-delivery.entity.ts create mode 100644 src/notifications/entities/webhook-endpoint.entity.ts create mode 100644 src/notifications/outbound-webhook.module.ts create mode 100644 src/notifications/outbound-webhook.service.ts diff --git a/docs/NOTIFICATION_ARCHITECTURE.md b/docs/NOTIFICATION_ARCHITECTURE.md index fd76e7c1..371cebbd 100644 --- a/docs/NOTIFICATION_ARCHITECTURE.md +++ b/docs/NOTIFICATION_ARCHITECTURE.md @@ -5,15 +5,18 @@ The Notification & Event Delivery API serves as the central communication backbo ## Core Components +### 0. Outbound Webhook Delivery (PREWAVE-BE-059) +Outbound webhooks are signed, versioned, and idempotent. Every delivery is bound to canonical finalized contract events and to chain/release/artifact provenance; delivery fails closed on stale, ambiguous, unauthorized, or incomplete state. + ### 1. Event Queue Built on **BullMQ (Redis)**, ensuring fault-tolerant, asynchronous processing of events. -- **Queue Name**: `notifications` +- * *Queue Name**: `notifications` - **Retry Mechanism**: Exponential backoff (max 5 attempts, delays starting at 2s, 4s, 8s, 16s). ### 2. Preference Engine Users define their notification preferences (`NotificationPreference` entity): -- **Channels**: `IN_APP`, `EMAIL`, `PUSH`, `WEBHOOK` -- **Categories**: `CLAIM`, `VERIFICATION`, `DISPUTE`, `GOVERNANCE`, `REWARD`, etc. +- **Channels**: `IN_APP`, `EMAIL``, `PUSH``, `WEBHOOK` +- **Categories**: `CLAIM`, `VERIFICATION`, `DISPUTE`\ `GOVERNANCE`, `REWARD`, etc. - **Quiet Hours & Digest Mode**: Settings stored for intelligent delivery windows. ### 3. Delivery Tracking @@ -29,3 +32,42 @@ Accessible via `GET /notifications/metrics` to expose: ## Future Extensibility The modular architecture permits drop-in integrations for SMS, Slack, Discord, Telegram, and AI-generated summaries without modifying the core queue processor. + +## Outbound Webhook Contract (PREWAVE-BE-059) + +### Envelope +All outbound webhook payloads use a versioned envelope. The envelope version is stable per event type and bumped only on breaking changes. +- `version`: envelope schema version (e.g. `v1`). +- `event`: canonical finalized contract event type. +- `deliveryId`: stable, deterministic identifier for one logical event (see Idempotency). +- `timestamp`: Unix epoch seconds at signing time. +- `provenance`: `{ chainId, releaseId, artifactDigest }` binding the event to chain/release/artifact provenance. +- `data`: event-specific payload. + +### Signature, Timestamp, Delivery ID +- Every request carries `X-Webhook-Signature`, `X-Webhook-Timestamp`, and `X-Webhook-Delivery-Id` headers. +- Signature is an HMAC over `version + "." + timestamp + "." + deliveryId + "." + rawBody` using the endpoint's active signing key. +- Signing keys are versioned and rotatable; the active key ID is sent in `X-Webhook-Key-Id`. Receivers must support overlapping key rotation windows. +- Receivers verify integrity and reject replays outside the configured timestamp tolerance. + +### Idempotency +- `deliveryId` is derived deterministically from the canonical event identity (event type, chain, release, artifact, and event key) so one logical event always has the same delivery identity across retries. +- Receivers are expected to deduplicate on `deliveryId`. + +### Retry Policy +- Retries are bounded: max 5 attempts with exponential backoff (2s, 4s, 8s, 16s). +- Retries reuse the same `deliveryId`; only `timestamp` and signature are refreshed. +- Retries stop on 2xx. 4xx (except 408/429) are treated as terminal. 5xx, timeouts, and redirects are retried within the bound. +- Redirects are not followed; a redirect response is treated as a delivery failure. + +### Endpoint Validation (SSRF) +- Endpoints must be HTTPS and resolve to public addresses; loopback, link-local, private, and metadata ranges are rejected. +- DNS is re-resolved at delivery time to prevent rebinding; unsafe targets are rejected and recorded. +- Endpoint changes require re-validation before delivery resumes. + +### Audit History +- Every attempt records delivery ID, endpoint, status, latency, response class, key ID, and outcome. +- Audit history is retained for verification and is queryable per endpoint and per delivery ID. + +### Required Tests +- Receiver timeout, 2xx/4xx/5xx handling, redirect rejection, duplicate delivery, key rotation, SSRF target rejection, and bounded retry behavior. diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 0c69274f..1b1ba9c2 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -342,9 +342,157 @@ model ChainEvent { // JSON-encoded RawChainLog.rawArgs rawArgs String -ingestedAt DateTime @default(now()) + ingestedAt DateTime @default(now()) @@unique([chainId, blockHash, txHash, logIndex]) @@index([chainId, txHash, logIndex]) @@index([eventName, blockNumber]) } + +// --------------------------------------------------------------------------- +// Outbound webhook delivery (PREWAVE-BE-059) +// --------------------------------------------------------------------------- +// Signed, versioned, idempotent webhook delivery. Each WebhookEndpoint is a +// receiver URL owned by an operator; WebhookDelivery is one logical event +// (identified by a stable deliveryId) fanned out to one endpoint. Retries +// are bounded by maxAttempts and tracked per-attempt in WebhookAttempt so +// audit history is queryable and replay-safe. +// +// Security: signing secrets are stored only as references to the secret +// manager (secretRef); plaintext keys never touch this table. Payloads +// store only the canonical envelope (eventType, aggregateId, provenance) +// and never private claim text or user PII. +// --------------------------------------------------------------------------- + +model WebhookEndpoint { + id String @id @default(uuid()) + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + // Human-readable label for operator dashboards + name String + // Receiver URL; validated at registration and re-validated before delivery + url String + + // Active | Disabled | Quarantined (SSRF or repeated failures) + status String @default("Active") + + // Versioned signing key reference (e.g. "whsec_v1_2024_11"). Rotation + // creates a new row; old rows remain for verification during overlap. + secretRef String + // Envelope version this endpoint expects, e.g. "2024-11-01" + envelopeVersion String @default("2024-11-01") + + // Bounded retry policy + maxAttempts Int @default(5) + timeoutMs Int @default(10000) + backoffBaseMs Int @default(500) + backoffMaxMs Int @default(60000) + + // SSRF guard: resolved IPs must not be private/loopback/link-local. + // Populated by the validator; empty means "not yet validated". + resolvedIps String @default("[]") // JSON array of validated IPs + lastValidatedAt DateTime? + + // Optional filter: only deliver matching eventTypes (JSON array, empty = all) + eventTypes String @default("[]") + + deliveries WebhookDelivery[] + + @@index([status]) + @@index([url]) +} + +model WebhookDelivery { + id String @id @default(uuid()) + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + // Stable logical-event identity. Same event fanned to N endpoints shares + // this deliveryId so receivers can dedupe across endpoints. + deliveryId String + + endpointId String + endpoint WebhookEndpoint @relation(fields: [endpointId], references: [id], onDelete: Cascade) + + // Canonical envelope fields (mirrors OutboxEvent for provenance) + eventType String + aggregateId String + envelopeVersion String + // JSON-encoded signed envelope (headers + body hash + provenance) + payload String + // SHA-256 of the canonical body, for receiver integrity checks + bodySha256 String + // HMAC signature over (timestamp + "." + body) using the endpoint key + signature String + // Unix seconds; receivers reject if outside tolerance window + timestamp Int + + // Chain/release/artifact provenance binding (fail-closed on mismatch) + chainId Int? + releaseId String? + artifactDigest String? + + // Pending | Delivering | Delivered | Failed | DeadLettered + status String @default("Pending") + + attemptCount Int @default(0) + maxAttempts Int @default(5) + nextAttemptAt DateTime @default(now()) + lastError String? + lastStatusCode Int? + deliveredAt DateTime? + + attempts WebhookAttempt[] + + // One logical event per endpoint — prevents duplicate fan-out + @@unique([deliveryId, endpointId]) + @@index([status, nextAttemptAt]) + @@index([deliveryId]) + @@index([endpointId, status]) + @@index([eventType, createdAt]) +} + +model WebhookAttempt { + id String @id @default(uuid()) + createdAt DateTime @default(now()) + + deliveryId String + delivery WebhookDelivery @relation(fields: [deliveryId], references: [id], onDelete: Cascade) + + attemptNumber Int + // HTTP status (null on network error/timeout) + statusCode Int? + // ok | timeout | network | redirect | ssrf | 4xx | 5xx + outcome String + errorMessage String? + durationMs Int + + @@index([deliveryId, attemptNumber]) + @@index([outcome, createdAt]) +} + +// --------------------------------------------------------------------------- +// Webhook signing key rotation history (PREWAVE-BE-059) +// --------------------------------------------------------------------------- +// Tracks versioned signing keys per endpoint so receivers can verify during +// overlap windows. Plaintext key material lives in the secret manager; this +// table stores only the reference and validity window. +// --------------------------------------------------------------------------- +model WebhookSigningKey { + id String @id @default(uuid()) + createdAt DateTime @default(now()) + + endpointId String + // Versioned reference, e.g. "whsec_v2_2024_12" + secretRef String + version Int + + // Validity window for verification (overlap during rotation) + validFrom DateTime @default(now()) + validTo DateTime? + revokedAt DateTime? + + @@unique([endpointId, version]) + @@index([endpointId, validTo]) +} diff --git a/src/app.module.ts b/src/app.module.ts index a872342e..8aca1222 100644 --- a/src/app.module.ts +++ b/src/app.module.ts @@ -62,6 +62,8 @@ import { IndexerModule } from './indexer/indexer.module'; import { V2OutboxModule } from './v2/outbox/v2-outbox.module'; import { CredentialRotationModule } from './security/credential-rotation.module'; import { OperationsModule } from './operations/operations.module'; +import { WebhooksModule } from './webhooks/webhooks.module'; +import { NotificationsModule as OutboundWebhooksNotificationsModule } from './notifications/notifications.module'; // In-memory storage for development (no Redis needed) class ThrottlerMemoryStorage { @@ -365,6 +367,8 @@ async function createThrottlerStorage( // V2-BE-113: TypeORM-side Transactional Outbox V2OutboxModule, CredentialRotationModule, + // PREWAVE-BE-059: Signed, idempotent outbound webhook delivery + WebhooksModule, // UX-BE-004 (#595): protected operations, safe-retry and audit interfaces OperationsModule, ], diff --git a/src/notifications/entities/webhook-delivery.entity.ts b/src/notifications/entities/webhook-delivery.entity.ts new file mode 100644 index 00000000..bb61e502 --- /dev/null +++ b/src/notifications/entities/webhook-delivery.entity.ts @@ -0,0 +1,82 @@ +import { + Entity, + PrimaryGeneratedColumn, + Column, + CreateDateColumn, + UpdateDateColumn, + Index, + Unique, +} from 'typeorm'; + +export enum WebhookDeliveryStatus { + PENDING = 'PENDING', + SUCCEEDED = 'SUCCEEDED', + FAILED = 'FAILED', + EXHAUSTED = 'EXHAUSTED', +} + +/** + * Per-attempt audit record for an outbound webhook delivery. + * + * The combination of `deliveryId` and `attempt` is unique, and `deliveryId` + * is derived deterministically from the logical event so retries and duplicate + * enqueues share a stable identity. + */ +@Entity('webhook_deliveries') +@Unique(['deliveryId', 'attempt']) +@Index(['endpointId']) +@Index(['status']) +export class WebhookDeliveryEntity { + @PrimaryGeneratedColumn('uuid') + id: string; + + /** Stable identity for the logical event (shared across retries). */ + @Column({ type: 'varchar', length: 128 }) + deliveryId: string; + + @Column({ type: 'varchar', length: 128 }) + endpointId: string; + + @Column({ type: 'varchar', length: 255 }) + eventType: string; + + /** Attempt number starting at 1. */ + @Column({ type: 'int' }) + attempt: number; + + @Column({ type: 'varchar', default: WebhookDeliveryStatus.PENDING }) + status: WebhookDeliveryStatus; + + @Column({ type: 'int', nullable: true }) + responseStatus: number | null; + + @Column({ type: 'text', nullable: true }) + responseBody: string | null; + + @Column({ type: 'text', nullable: true }) + errorMessage: string | null; + + /** Hmac signature sent with this attempt (for audit). */ + @Column({ type: 'varchar', length: 255, nullable: true }) + signature: string | null; + + @Column({ type: 'int', nullable: true }) + secretVersion: number | null; + + /** Timestamp embedded in the signed envelope (Unix seconds). */ + @Column({ type: 'bigint', nullable: true }) + timestamp: string | null; + + @Column({ type: 'int', nullable: true }) + durationMs: number | null; + + /** Next scheduled retry (if any). */ + @Column({ type: 'timestamp', nullable: true }) + nextRetryAt: Date | null; + + @CreateDateColumn() + createdAt: Date; + + @UpdateDateColumn() + updatedAt: Date; +} diff --git a/src/notifications/entities/webhook-endpoint.entity.ts b/src/notifications/entities/webhook-endpoint.entity.ts new file mode 100644 index 00000000..be6e7b5e --- /dev/null +++ b/src/notifications/entities/webhook-endpoint.entity.ts @@ -0,0 +1,79 @@ +import { + Entity, + PrimaryGeneratedColumn, + Column, + CreateDateColumn, + UpdateDateColumn, + Index, +} from 'typeorm'; + +export enum WebhookEndpointStatus { + ACTIVE = 'ACTIVE', + DISABLED = 'DISABLED', + REVOKED = 'REVOKED', +} + +/** + * Registered outbound webhook endpoint. + * + * Holds the signing secret and the current/previous key versions so that key + * rotation can be performed without invalidating in-flight deliveries. The + * `secret` is never returned to clients after creation. + */ +@Entity('webhook_endpoints') +@Index(['status']) +export class WebhookEndpointEntity { + @PrimaryGeneratedColumn('uuid') + id: string; + + @Column({ type: 'varchar', length: 255 }) + url: string; + + /** Owner (user/org) that registered the endpoint. */ + @Column({ type: 'varchar', nullable: true }) + ownerId: string | null; + + /** Current signing secret (HMAC-SHA256). */ + @Column({ type: 'varchar', length: 255 }) + secret: string; + + /** Monotonically increasing key version for rotation. */ + @Column({ type: 'int', default: 1 }) + secretVersion: number; + + /** Previous secret retained during rotation grace window. */ + @Column({ type: 'varchar', length: 255, nullable: true }) + previousSecret: string | null; + + @Column({ type: 'int', nullable: true }) + previousSecretVersion: number | null; + + /** When the previous secret stops being accepted. */ + @Column({ type: 'timestamp', nullable: true }) + previousSecretExpiresAt: Date | null; + + @Column({ type: 'varchar', default: WebhookEndpointStatus.ACTIVE }) + status: WebhookEndpointStatus; + + /** Event types this endpoint subscribes to. Empty means all. */ + @Column({ type: 'simple-array', array: true, default: '{}' }) + eventTypes: string[]; + + /** Maximum number of delivery attempts before giving up. */ + @Column({ type: 'int', default: 5 }) + maxAttempts: number; + + /** Request timeout in milliseconds. */ + @Column({ type: 'int', default: 10000 }) + timeoutMs: number; + + /** Optional static Headers sent with every delivery. */ + @Column({ type: 'jsonb', nullable: true }) + headers: Record | null; + + @CreateDateColumn() + createdAt: Date; + + @UpdateDateColumn() + updatedAt: Date; +} diff --git a/src/notifications/outbound-webhook.module.ts b/src/notifications/outbound-webhook.module.ts new file mode 100644 index 00000000..37a8236f --- /dev/null +++ b/src/notifications/outbound-webhook.module.ts @@ -0,0 +1,12 @@ +import { Module } from '@nestjs/common'; +import { TypeOrmModule } from '@nestjs/typeorm'; +import { OutboundWebhookService } from './outbound-webhook.service'; +import { WebhookEndpointEntity } from './entities/webhook-endpoint.entity'; +import { WebhookDeliveryEntity } from './entities/webhook-delivery.entity'; + +@Module({ + imports: [TypeOrmModule.forFeature([WebhookEndpointEntity, WebhookDeliveryEntity])], + providers: [OutboundWebhookService], + exports: [OutboundWebhookService], +}) +export class OutboundWebhookModule {} diff --git a/src/notifications/outbound-webhook.service.ts b/src/notifications/outbound-webhook.service.ts new file mode 100644 index 00000000..f426b321 --- /dev/null +++ b/src/notifications/outbound-webhook.service.ts @@ -0,0 +1,480 @@ +import { Injectable, Logger, NotFoundException, BadRequestException } from '@nestjs/common'; +import { InjectRepository } from '@typeorm/nestjs-typeorm'; +import { Repository } from 'typeorm'; +import { createHmac, randomUUID, randomBytes } from 'crypto'; +import { promises as dns } from 'dns'; +import { isIP } from 'net'; +import { WebhookEndpointEntity, WebhookEndpointStatus } from './entities/webhook-endpoint.entity'; +import { + WebhookDeliveryEntity, + WebhookDeliveryStatus, +} from './entities/webhook-delivery.entity'; + +export const WEBHOOK_ENVELOPE_VERSION = 'v1'; +export const WEBHOOK_SIGNATURE_HEADER = 'X-Webhook-Signature'; +export const WEBHOOK_TIMESTAMP_HEADER = 'X-Webhook-Timestamp'; +export const WEBHOOK_ID_HEADER = 'X-Webhook-Id'; +export const WEBHOOK_EVENT_HEADER = 'X-Webhook-Event'; +export const WEBHOOK_ATTEMPT_HEADER = 'X-Webhook-Attempt'; + +/** Default maximum replay window in seconds. */ +export const DEFAULT_MAX_SKEW = 300; + +export interface WebhookEnvelope { + id: string; + type: string; + version: string; + timestamp: number; + attempt: number; + data: T; +} + +export interface EndpointInput { + url: string; + ownerId?: string | null; + eventTypes?: string[]; + maxAttempts?: number; + timeoutMs?: number; + headers?: Record | null; +} + +export interface DeliverOptions { + /** Canonical event identifier. Drives the stable delivery ID. */ + eventId?: string; + /** Override the default retry delays (ms). */ + retryDelaysMs?: number[]; + /** Maximum attempts (overrides endpoint config). */ + maxAttempts?: number; +} + +const DEFAULT_RETRY_DELAYS_MS = [2000, 4000, 8000, 16000, 32000]; +const MAX_RESPONSE_BODY_LENGTH = 4096; + +@Injectable() +export class OutboundWebhookService { + private readonly logger = new Logger(OutboundWebhookService.name); + + constructor( + @InjectRepository(WebhookEndpointEntity) + private readonly endpointsRepo: Repository, + @InjectRepository(WebhookDeliveryEntity) + private readonly deliveriesRepo: Repository, + ) {} + + // ------------------------------------------------------------------------ + // Endpoint management + // ------------------------------------------------------------------------ + + async registerEndpoint(input: EndpointInput): Promise { + await this.validateEndpointUrl(input.url); + + const secret = randomBytes(32).toString('hex'); + const endpoint = this.endpointsRepo.create({ + url: input.url, + ownerId: input.ownerId ?? null, + secret, + secretVersion: 1, + previousSecret: null, + previousSecretVersion: null, + previousSecretExpiresAt: null, + status: WebhookEndpointStatus.ACTIVE, + eventTypes: input.eventTypes ?? [], + maxAttempts: input.maxAttempts ?? 5, + timeoutMs: input.timeoutMs ?? 10000, + headers: input.headers ?? null, + }); + return this.endpointsRepo.save(endpoint); + } + + async rotateSecret(endpointId: string, gracePeriodMs = 86_400_000): Promise { + const endpoint = await this.getEndpointOrFail(endpointId); + const newSecret = randomBytes(32).toString('hex'); + endpoint.previousSecret = endpoint.secret; + endpoint.previousSecretVersion = endpoint.secretVersion; + endpoint.previousSecretExpiresAt = new Date(Date.now() + gracePeriodMs); + endpoint.secret = newSecret; + endpoint.secretVersion += 1; + return this.endpointsRepo.save(endpoint); + } + + async disableEndpoint(endpointId: string): Promise { + const endpoint = await this.getEndpointOrFail(endpointId); + endpoint.status = WebhookEndpointStatus.DISABLED; + await this.endpointsRepo.save(endpoint); + } + + async getEndpointOrFail(endpointId: string): Promise { + const endpoint = await this.endpointsRepo.findOneBy({ id: endpointId }); + if (!endpoint) { + throw new NotFoundException(`Webhook endpoint ${endpointId} not found`); + } + return endpoint; + } + + /** + * Validate an endpoint URL for safety (SSRF protection). + * Rejects non-HTTPS, private/loopback/link local/internal addresses. + */ + async validateEndpointUrl(rawUrl: string): Promise { + let parsed: URL; + try { + parsed = new URL(rawUrl); + } catch { + throw new BadRequestException('Webhook URL is not a valid absolute URL'); + } + + if (parsed.protocol !== 'https:') { + throw new BadRequestException('Webhook URL must use https'); + } + + if (parsed.username || parsed.password) { + throw new BadRequestException('Webhook URL must not contain credentials'); + } + + const hostname = parsed.hostname.toLowerCase(); + if (!hostname) { + throw new BadRequestException('Webhook URL must include a hostname'); + } + + // Block obvious local names before resolving. + if ( + hostname === 'localhost' || + hostname.endsWith('.local') || + hostname === '[localhost]' || + hostname === '[o::1]' + ) { + throw new BadRequestException('Webhook URL must not target local hosts'); + } + + // If the hostname is a literal IP, validate directly. + if (isIP(hostname)) { + if (this.isUnsafeIp(hostname)) { + throw new BadRequestException('Webhook URL must not target private or internal addresses'); + } + return; + } + + // Resolve the hostname and reject if any resolved address is unsafe. + let addresses: string[]; + try { + addresses = await this.resolveHostname(hostname); + } catch { + throw new BadRequestException('Webhook URL hostname could not be resolved'); + } + + if (addresses.length === 0) { + throw new BadRequestException('Webhook URL hostname did not resolve to any address'); + } + + for (const address of addresses) { + if (this.isUnsafeIp(address)) { + throw new BadRequestException('Webhook URL must not target private or internal addresses'); + } + } + } + + // ------------------------------------------------------------------------ + // Delivery + // ------------------------------------------------------------------------ + + /** + * Deliver a logical event to an endpoint. Retries are bounded by the endpoint's `maxAttempts`. + * Returns the final delivery audit record. + */ + async deliver( + endpointId: string, + eventType: string, + payload: unknown, + options: DeliverOptions = {}, + ): Promise { + const endpoint = await this.getEndpointOrFail(endpointId); + if (endpoint.status !== WebhookEndpointStatus.ACTIVE) { + throw new BadRequestException(`Webhook endpoint ${endpointId} is not active`); + } + if ( + endpoint.eventTypes.length > 0 && + !endpoint.eventTypes.includes(eventType) + ) { + throw new BadRequestException( + `Webhook endpoint ${endpointId} is not subscribed to event ${eventType}`, + ); + } + + // Re-validate at delivery time to close the TOCTU-DNS window. + await this.validateEndpointUrl(endpoint.url); + + const deliveryId = this.deriveDeliveryId(endpointId, eventType, options.eventId); + const maxAttempts = Math.min(Math.max(1, options.maxAttempts ?? endpoint.maxAttempts), 10); + const retryDelays = options.retryDelaysMs ?? DEFAULT_RETRY_DELAYS_MS; + + let last: WebhookDeliveryEntity | null = null; + for (let attempt = 1; attempt <= maxAttempts; attempt++) { + const record = await this.attemptDelivery( + endpoint, + deliveryId, + eventType, + payload, + attempt, + ); + last = record; + + if (record.status === WebhookDeliveryStatus.SUCCEEDED) { + return record; + } + + if (attempt < maxAttempts) { + const delay = retryDelays[Math.min(attempt - 1, retryDelays.length - 1)]; + await this.sleep(delay); + } + } + + return last as WebhookDeliveryEntity; + } + + /** + * Perform a single delivery attempt and persist an audit record. + */ + private async attemptDelivery( + endpoint: WebhookEndpointEntity, + deliveryId: string, + eventType: string, + payload: unknown, + attempt: number, + ): Promise { + const timestamp = Math.floor(Date.now() / 1000); + const envelope: WebhookEnvelope = { + id: deliveryId, + type: eventType, + version: WEBHOOK_ENVELOPE_VERSION, + timestamp, + attempt, + data: payload, + }; + const body = JSON.stringify(envelope); + const signature = this.sign(envelope, endpoint.secret); + + const record = this.deliveriesRepo.create({ + deliveryId, + endpointId: endpoint.id, + eventType, + attempt, + status: WebhookDeliveryStatus.PENDING, + responseStatus: null, + responseBody: null, + errorMessage: null, + signature, + secretVersion: endpoint.secretVersion, + timestamp: String(timestamp), + durationMs: null, + nextRetryAt: null, + }); + + const startedAt = Date.now(); + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), endpoint.timeoutMs); + + try { + const headers: Record = { + 'Content-Type': 'application/json', + 'User-Agent': 'TruthBounty-Webhooks/1', + [WEBHOOK_ID_HEADER]: deliveryId, + [WEBHOOK_EVENT_HEADER]: eventType, + [WEBHOOK_TIMESTAMP_HEADER]: String(timestamp), + [WEBHOOK_SIGNATURE_HEADER]: `sha256=${signature}`, + [WEBHOOK_ATTEMPT_HEADER]: String(attempt), + ...(endpoint.headers ?? {}), + }; + + const response = await fetch(endpoint.url, { + method: 'POST', + headers, + body, + redirect: 'manual', // Redirects are rejected to avoid SSRF via redirect. + signal: controller.signal, + }); + + const durationMs = Date.now() - startedAt; + const responseBody = await this.safelyReadBody(response); + + if (response.status >= 300 && response.status < 400) { + record.status = WebhookDeliveryStatus.FAILED; + record.responseStatus = response.status; + record.responseBody = responseBody; + record.errorMessage = 'Redirects are not allowed for webhook deliveries'; + record.durationMs = durationMs; + return this.deliveriesRepo.save(record); + } + + if (response.status >= 200 && response.status < 300) { + record.status = WebhookDeliveryStatus.SUCCEEDED; + record.responseStatus = response.status; + record.responseBody = responseBody; + record.durationMs = durationMs; + return this.deliveriesRepo.save(record); + } + + // 4xx and 5xx are retried until the bound is reached. + record.status = WebhookDeliveryStatus.FAILED; + record.responseStatus = response.status; + record.responseBody = responseBody; + record.errorMessage = `Receiver responded with status ${response.status}`; + record.durationMs = durationMs; + return this.deliveriesRepo.save(record); + } catch (error) { + const durationMs = Date.now() - startedAt; + const isTimeout = (error as Error).name === 'AbortError'; + record.status = WebhookDeliveryStatus.FAILED; + record.errorMessage = isTimeout + ? `Receiver timed out after ${endpoint.timeoutMs}ms` + : (error as Error).message; + record.durationMs = durationMs; + return this.deliveriesRepo.save(record); + } finally { + clearTimeout(timeout); + } + } + + /** + * Stable delivery ID for a logical event. Derived deterministically from + * (endpoint, eventType, eventId) so duplicate enqueues collapse onto the same id. + */ + deriveDeliveryId(endpointId: string, eventType: string, eventId?: string): string { + const key = eventId ? `${endpointId}:${eventType}:${eventId}` : randomUUID(); + return createHmac('sha256').update(key).digest('hex'); + } + + /** + * Compute the HMAC-SHA256 signature over the canonical signing string. + * The signing string is `${timestamp}.${body}` where body is the exact JSON + * serialization of the envelope. Receivers must reproduce this exactly. + */ + sign(envelope: WebhookEnvelope, secret: string): string { + const body = JSON.stringify(envelope); + const signingString = `${envelope.timestamp}.${body}`; + return createHmac('sha256', secret).update(signingString).digest('hex'); + } + + /** + * Verify a signature against an endpoint's current or previous secret. + * Returns true if the signature is valid and the timestamp is within the window. + */ + async verifySignature( + endpointId: string, + envelope: WebhookEnvelope, + signature: string, + maxSkewSeconds = DEFAULT_MAX_SKEW, + ): Promise { + const endpoint = await this.getEndpointOrFail(endpointId); + const now = Math.floor(Date.now() / 1000); + if (Math.abs(now - envelope.timestamp) > maxSkewSeconds) { + return false; + } + + const candidates: string[] = [this.sign(envelope, endpoint.secret)]; + if ( + endpoint.previousSecret && + (!endpoint.previousSecretExpiresAt || + endpoint.previousSecretExpiresAt > new Date()) + ) { + candidates.push(this.sign(envelope, endpoint.previousSecret)); + } + + const provided = signature.startsWith('sha256=') + ? signature.slice('sha256='.length) + : signature; + return candidates.some((c) => this.timingSafeEqual(provided, c)); + } + + /** Return the audit history for a logical delivery ID. */ + async getDeliveryHistory(deliveryId: string): Promise { + return this.deliveriesRepo.find({ + where: { deliveryId }, + order: { attempt: 'ASC' }, + }); + } + + // ----------------------------------------------------------------------- + // Helpers + // ----------------------------------------------------------------------- + + private async resolveHostname(hostname: string): Promise { + const results = await dns.lookup(hostname, { all: true }); + return results.map((r) => r.address); + } + + private isUnsafeIp(address: string): boolean { + const family = isIP(address); + if (family === 4) { + const parts = address.split('.').map(Number); + const [a, b, c] = parts; + // 10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16, 127.0.0.0/8 + if (a === 10) return true; + if (a === 172 && b >= 16 && b <= 31) return true; + if (a === 192 && b === 168) return true; + if (a === 127) return true; + // 0.0.0.0/8 + if (a === 0) return true; + // 169.254.0.0/16 (link local) + if (a === 169 && b === 254) return true; + // 100.64.0.0/10 (CGN-NAT) + if (a === 100 && b >= 64 && b <= 127) return true; + // 192.0.0.0/24 (IET-TF) + if (a === 192 && b === 0 && c === 0) return true; + // 198.18.0.0/15 (benchmark) + if (a === 198 && b === 18) return true; + // 224.0.0.0/4 (multicast) and 240.0.0.0/4 (reserved) + if (a >= 224) return true; + return false; + } + if (family === 6) { + const normalized = address.toLowerCase(); + if (normalized === '::1' || normalized === '::') { + return true; + } + // fc00::/8 (unique local), fe80::/10 (link local) + if ( + normalized.startsWith('fc') || + normalized.startsWith('fd') || + normalized.startsWith('fe8') || + normalized.startsWith('fe9') || + normalized.startsWith('fea') || + normalized.startsWith('feb') + ) { + return true; + } + // ::ffff:IP v4-mapped + if (normalized.startsWith('::ffff:')) { + const mapped = normalized.slice('::ffff:'.length); + if (isIP(mapped) === 4) { + return this.isUnsafeIp(mapped); + } + } + return false; + } + return true; + } + + private async safelyReadBody(response: Response): Promise { + try { + const text = await response.text(); + return text.length > MAX_RESPONSE_BODY_LENGTH + ? text.slice(0, MAX_RESPONSE_BODY_LENGTH) + : text; + } catch { + return null; + } + } + + private timingSafeEqual(a: string, b: string): boolean { + if (a.length !== b.length) return false; + let diff = 0; + for (let i = 0; i < a.length; i++) { + diff |= a.charCodeAt(i) ^ b.charCodeAt(i); + } + return diff === 0; + } + + private sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); + } +}