From f5d18ae38ea137984c4cbb15e5448964d210f910 Mon Sep 17 00:00:00 2001 From: ocakciomursel-pixel Date: Thu, 15 Oct 2026 09:52:26 +0300 Subject: [PATCH] fix(webhooks): constrain subscription URLs against SSRF and add webhook unit tests (#280, #170) --- .github/workflows/build-check.yml | 3 +- .../services/webhook-delivery.service.spec.ts | 452 ++++++++++++++++++ .../services/webhook-delivery.service.ts | 133 +++++- .../services/webhook-event.service.spec.ts | 172 +++++++ .../services/webhook-hmac.service.spec.ts | 94 ++++ .../webhook-processor.service.spec.ts | 67 +++ .../webhook-subscription.service.spec.ts | 229 +++++++++ .../services/webhook-subscription.service.ts | 20 +- .../webhook-url-validator.service.spec.ts | 228 +++++++++ .../services/webhook-url-validator.service.ts | 448 +++++++++++++++++ .../webhooks/webhook.controller.ts | 17 +- src/infrastructure/webhooks/webhook.module.ts | 3 + 12 files changed, 1844 insertions(+), 22 deletions(-) create mode 100644 src/infrastructure/webhooks/services/webhook-delivery.service.spec.ts create mode 100644 src/infrastructure/webhooks/services/webhook-event.service.spec.ts create mode 100644 src/infrastructure/webhooks/services/webhook-hmac.service.spec.ts create mode 100644 src/infrastructure/webhooks/services/webhook-processor.service.spec.ts create mode 100644 src/infrastructure/webhooks/services/webhook-subscription.service.spec.ts create mode 100644 src/infrastructure/webhooks/services/webhook-url-validator.service.spec.ts create mode 100644 src/infrastructure/webhooks/services/webhook-url-validator.service.ts diff --git a/.github/workflows/build-check.yml b/.github/workflows/build-check.yml index 4dd7ee0..028278a 100644 --- a/.github/workflows/build-check.yml +++ b/.github/workflows/build-check.yml @@ -28,12 +28,13 @@ jobs: - name: Run TypeScript check run: npx tsc --noEmit - - name: Test GraphQL gateway and cursor pagination + - name: Test GraphQL gateway, cursor pagination, and webhooks run: >- npm test -- --runInBand src/graphql src/discovery/reviews/agent-reviews.service.spec.ts src/common/pagination/cursor-pagination.service.spec.ts + src/infrastructure/webhooks - name: Build project run: npm run build diff --git a/src/infrastructure/webhooks/services/webhook-delivery.service.spec.ts b/src/infrastructure/webhooks/services/webhook-delivery.service.spec.ts new file mode 100644 index 0000000..76b6d7f --- /dev/null +++ b/src/infrastructure/webhooks/services/webhook-delivery.service.spec.ts @@ -0,0 +1,452 @@ +import { HttpService } from "@nestjs/axios"; +import { of, throwError } from "rxjs"; +import { Repository } from "typeorm"; +import { WebhookDeadLetter } from "src/infrastructure/webhooks/entities/webhook-dead-letter.entity"; +import { + WebhookDelivery, + WebhookDeliveryStatus, +} from "src/infrastructure/webhooks/entities/webhook-delivery.entity"; +import { + WebhookEvent, + WebhookEventStatus, +} from "src/infrastructure/webhooks/entities/webhook-event.entity"; +import { + WebhookSubscription, + WebhookSubscriptionStatus, +} from "src/infrastructure/webhooks/entities/webhook-subscription.entity"; +import { WebhookDeliveryService } from "./webhook-delivery.service"; +import { WebhookHmacService } from "./webhook-hmac.service"; +import { + DnsLookupFn, + WebhookUrlValidatorService, +} from "./webhook-url-validator.service"; + +describe("WebhookDeliveryService", () => { + let service: WebhookDeliveryService; + let deliveryRepo: jest.Mocked< + Pick< + Repository, + "create" | "save" | "findOne" | "find" | "createQueryBuilder" + > + >; + let deadLetterRepo: jest.Mocked< + Pick, "create" | "save" | "findOne" | "find"> + >; + let eventRepo: jest.Mocked< + Pick, "findOne" | "createQueryBuilder"> + >; + let subscriptionRepo: jest.Mocked< + Pick, "save" | "findOne" | "find"> + >; + let httpService: jest.Mocked>; + let hmacService: WebhookHmacService; + let dnsResolver: jest.MockedFunction; + let urlValidator: WebhookUrlValidatorService; + let eventQueryBuilder: { + update: jest.Mock; + set: jest.Mock; + where: jest.Mock; + execute: jest.Mock; + }; + + const makeSubscription = ( + overrides: Partial = {}, + ): WebhookSubscription => + ({ + id: "sub-1", + userId: "user-1", + url: "https://hooks.example.com/receive", + signingKey: "whsec_test_key_123456", + events: ["agent.created"], + status: WebhookSubscriptionStatus.ACTIVE, + maxRetries: 3, + retryDelayMs: 1000, + backoffMultiplier: 2, + timeoutMs: 5000, + rateLimitPerMinute: 10, + ...overrides, + }) as WebhookSubscription; + + const makeEvent = (overrides: Partial = {}): WebhookEvent => + ({ + id: "evt-1", + eventType: "agent.created", + aggregateId: "ag-1", + payload: { agentId: "ag-1" }, + status: WebhookEventStatus.DELIVERING, + deliveryCount: 1, + successCount: 0, + failureCount: 0, + createdAt: new Date("2026-01-01T00:00:00.000Z"), + ...overrides, + }) as WebhookEvent; + + beforeEach(() => { + eventQueryBuilder = { + update: jest.fn().mockReturnThis(), + set: jest.fn().mockReturnThis(), + where: jest.fn().mockReturnThis(), + execute: jest.fn().mockResolvedValue({}), + }; + + deliveryRepo = { + create: jest.fn((data) => ({ id: "deliv-1", attempts: 0, ...data }) as any), + save: jest.fn(async (entity) => entity as any), + findOne: jest.fn(), + find: jest.fn().mockResolvedValue([]), + createQueryBuilder: jest.fn(), + }; + + deadLetterRepo = { + create: jest.fn((data) => ({ id: "dlq-1", ...data }) as any), + save: jest.fn(async (entity) => entity as any), + findOne: jest.fn(), + find: jest.fn().mockResolvedValue([]), + }; + + eventRepo = { + findOne: jest.fn(), + createQueryBuilder: jest.fn().mockReturnValue(eventQueryBuilder as any), + }; + + subscriptionRepo = { + save: jest.fn(async (sub) => sub as WebhookSubscription), + findOne: jest.fn(), + find: jest.fn().mockResolvedValue([]), + }; + + httpService = { + post: jest.fn(), + }; + + hmacService = new WebhookHmacService(); + dnsResolver = jest + .fn, Parameters>() + .mockResolvedValue([{ address: "93.184.216.34", family: 4 }]); + urlValidator = new WebhookUrlValidatorService(); + urlValidator.setLookupFn(dnsResolver); + + service = new WebhookDeliveryService( + deliveryRepo as unknown as Repository, + deadLetterRepo as unknown as Repository, + eventRepo as unknown as Repository, + subscriptionRepo as unknown as Repository, + httpService as unknown as HttpService, + hmacService, + urlValidator, + ); + }); + + describe("createDeliveries", () => { + it("creates PENDING delivery records with maxAttempts from subscription.maxRetries", async () => { + const event = makeEvent(); + const sub = makeSubscription({ maxRetries: 5 }); + + const created = await service.createDeliveries(event, [sub]); + + expect(created).toHaveLength(1); + expect(deliveryRepo.create).toHaveBeenCalledWith({ + subscriptionId: "sub-1", + eventId: "evt-1", + status: WebhookDeliveryStatus.PENDING, + maxAttempts: 5, + }); + expect(deliveryRepo.save).toHaveBeenCalledTimes(1); + }); + }); + + describe("processDelivery — 2xx success", () => { + it("marks delivery SUCCESS, signs headers with HMAC-SHA256, disables redirects, and increments event successCount", async () => { + const subscription = makeSubscription(); + const event = makeEvent(); + const delivery: WebhookDelivery = { + id: "deliv-1", + eventId: event.id, + subscriptionId: subscription.id, + status: WebhookDeliveryStatus.PENDING, + attempts: 0, + maxAttempts: 3, + createdAt: new Date(), + } as WebhookDelivery; + + deliveryRepo.findOne.mockResolvedValue(delivery); + subscriptionRepo.findOne.mockResolvedValue(subscription); + eventRepo.findOne.mockResolvedValue(event); + + httpService.post.mockReturnValue( + of({ + status: 200, + data: { ok: true }, + headers: { "content-type": "application/json" }, + } as any), + ); + + const result = await service.processDelivery("deliv-1"); + + expect(result.status).toBe(WebhookDeliveryStatus.SUCCESS); + expect(result.attempts).toBe(1); + expect(result.statusCode).toBe(200); + expect(result.deliveredAt).toBeInstanceOf(Date); + expect(httpService.post).toHaveBeenCalledWith( + "https://hooks.example.com/receive", + expect.any(String), + expect.objectContaining({ + maxRedirects: 0, + headers: expect.objectContaining({ + "X-Webhook-Event-Id": "evt-1", + "X-Webhook-Event-Type": "agent.created", + "X-Webhook-Signature": expect.stringMatching(/^sha256=[a-f0-9]{64}$/), + }), + }), + ); + expect(eventQueryBuilder.update).toHaveBeenCalledWith(WebhookEvent); + expect(eventQueryBuilder.where).toHaveBeenCalledWith("id = :id", { + id: "evt-1", + }); + }); + }); + + describe("processDelivery — 5xx retry and dead-letter escalation", () => { + it("marks delivery FAILED and schedules exponential backoff nextRetryAt when attempts < maxAttempts", async () => { + const subscription = makeSubscription({ + maxRetries: 3, + retryDelayMs: 1000, + backoffMultiplier: 2, + }); + const event = makeEvent(); + const delivery: WebhookDelivery = { + id: "deliv-retry", + eventId: event.id, + subscriptionId: subscription.id, + status: WebhookDeliveryStatus.PENDING, + attempts: 0, + maxAttempts: 3, + createdAt: new Date(), + } as WebhookDelivery; + + deliveryRepo.findOne.mockResolvedValue(delivery); + subscriptionRepo.findOne.mockResolvedValue(subscription); + eventRepo.findOne.mockResolvedValue(event); + + httpService.post.mockReturnValue( + of({ + status: 503, + data: "Service Unavailable", + headers: {}, + } as any), + ); + + const before = Date.now(); + const result = await service.processDelivery("deliv-retry"); + + expect(result.status).toBe(WebhookDeliveryStatus.FAILED); + expect(result.attempts).toBe(1); + expect(result.statusCode).toBe(503); + expect(result.nextRetryAt).toBeInstanceOf(Date); + expect(result.nextRetryAt.getTime()).toBeGreaterThanOrEqual(before + 900); + expect((result.metadata as any).attempts).toHaveLength(1); + expect(deadLetterRepo.save).not.toHaveBeenCalled(); + }); + + it("moves delivery to DEAD_LETTERED and records a WebhookDeadLetter when attempts reach maxAttempts", async () => { + const subscription = makeSubscription({ maxRetries: 3 }); + const event = makeEvent(); + const delivery: WebhookDelivery = { + id: "deliv-dlq", + eventId: event.id, + subscriptionId: subscription.id, + status: WebhookDeliveryStatus.FAILED, + attempts: 2, + maxAttempts: 3, + createdAt: new Date(), + } as WebhookDelivery; + + deliveryRepo.findOne.mockResolvedValue(delivery); + subscriptionRepo.findOne.mockResolvedValue(subscription); + eventRepo.findOne.mockResolvedValue(event); + + httpService.post.mockReturnValue( + throwError(() => new Error("connect ECONNREFUSED")), + ); + + const result = await service.processDelivery("deliv-dlq"); + + expect(result.status).toBe(WebhookDeliveryStatus.DEAD_LETTERED); + expect(result.attempts).toBe(3); + expect(deadLetterRepo.create).toHaveBeenCalledWith( + expect.objectContaining({ + deliveryId: delivery.id, + subscriptionId: subscription.id, + eventId: event.id, + url: subscription.url, + totalAttempts: 3, + userId: subscription.userId, + }), + ); + expect(deadLetterRepo.save).toHaveBeenCalledTimes(1); + expect(eventQueryBuilder.update).toHaveBeenCalledWith(WebhookEvent); + }); + }); + + describe("SSRF, redirect, rate-limit, and response truncation safeguards", () => { + it("blocks delivery-time DNS rebinding to a private/metadata IP without making an HTTP request", async () => { + const subscription = makeSubscription({ + url: "https://rebind.example.com/hook", + }); + const event = makeEvent(); + const delivery: WebhookDelivery = { + id: "deliv-rebind", + eventId: event.id, + subscriptionId: subscription.id, + status: WebhookDeliveryStatus.PENDING, + attempts: 0, + maxAttempts: 2, + createdAt: new Date(), + } as WebhookDelivery; + + deliveryRepo.findOne.mockResolvedValue(delivery); + subscriptionRepo.findOne.mockResolvedValue(subscription); + eventRepo.findOne.mockResolvedValue(event); + + dnsResolver.mockResolvedValueOnce([ + { address: "169.254.169.254", family: 4 }, + ]); + + const result = await service.processDelivery("deliv-rebind"); + + expect(httpService.post).not.toHaveBeenCalled(); + expect(result.status).toBe(WebhookDeliveryStatus.FAILED); + expect(result.errorMessage).toContain("SSRF validation failed"); + }); + + it("rejects 3xx redirect responses instead of following them to internal services", async () => { + const subscription = makeSubscription(); + const event = makeEvent(); + const delivery: WebhookDelivery = { + id: "deliv-redirect", + eventId: event.id, + subscriptionId: subscription.id, + status: WebhookDeliveryStatus.PENDING, + attempts: 0, + maxAttempts: 2, + createdAt: new Date(), + } as WebhookDelivery; + + deliveryRepo.findOne.mockResolvedValue(delivery); + subscriptionRepo.findOne.mockResolvedValue(subscription); + eventRepo.findOne.mockResolvedValue(event); + + httpService.post.mockReturnValue( + of({ + status: 302, + data: "Found", + headers: { location: "http://169.254.169.254/latest/meta-data/" }, + } as any), + ); + + const result = await service.processDelivery("deliv-redirect"); + + expect(result.status).toBe(WebhookDeliveryStatus.FAILED); + expect(result.statusCode).toBe(302); + expect(result.errorMessage).toContain( + "Redirect responses (HTTP 302) are not allowed", + ); + }); + + it("enforces per-subscription outbound rateLimitPerMinute", async () => { + const subscription = makeSubscription({ rateLimitPerMinute: 1 }); + const event = makeEvent(); + const makeDeliv = (id: string): WebhookDelivery => + ({ + id, + eventId: event.id, + subscriptionId: subscription.id, + status: WebhookDeliveryStatus.PENDING, + attempts: 0, + maxAttempts: 2, + createdAt: new Date(), + }) as WebhookDelivery; + + subscriptionRepo.findOne.mockResolvedValue(subscription); + eventRepo.findOne.mockResolvedValue(event); + httpService.post.mockReturnValue( + of({ status: 200, data: "ok", headers: {} } as any), + ); + + const first = makeDeliv("deliv-rate-1"); + deliveryRepo.findOne.mockResolvedValueOnce(first); + const res1 = await service.processDelivery("deliv-rate-1"); + expect(res1.status).toBe(WebhookDeliveryStatus.SUCCESS); + + const second = makeDeliv("deliv-rate-2"); + deliveryRepo.findOne.mockResolvedValueOnce(second); + const res2 = await service.processDelivery("deliv-rate-2"); + + expect(res2.status).toBe(WebhookDeliveryStatus.FAILED); + expect(res2.errorMessage).toContain("Outbound rate limit exceeded"); + expect(httpService.post).toHaveBeenCalledTimes(1); + }); + + it("truncates oversized response bodies before storing on delivery records", async () => { + const subscription = makeSubscription(); + const event = makeEvent(); + const delivery: WebhookDelivery = { + id: "deliv-large-body", + eventId: event.id, + subscriptionId: subscription.id, + status: WebhookDeliveryStatus.PENDING, + attempts: 0, + maxAttempts: 2, + createdAt: new Date(), + } as WebhookDelivery; + + deliveryRepo.findOne.mockResolvedValue(delivery); + subscriptionRepo.findOne.mockResolvedValue(subscription); + eventRepo.findOne.mockResolvedValue(event); + + const hugeBody = "X".repeat(5000); + httpService.post.mockReturnValue( + of({ status: 200, data: hugeBody, headers: {} } as any), + ); + + const result = await service.processDelivery("deliv-large-body"); + + expect(result.responseBody.length).toBeLessThan(550); + expect(result.responseBody).toContain("...[truncated]"); + }); + }); + + describe("retryDeadLetter", () => { + it("creates a fresh PENDING delivery and marks the dead letter retried for its owner", async () => { + const subscription = makeSubscription({ userId: "user-1", maxRetries: 4 }); + const deadLetter: WebhookDeadLetter = { + id: "dlq-1", + subscriptionId: subscription.id, + eventId: "evt-1", + deliveryId: "deliv-old", + userId: "user-1", + retried: false, + } as WebhookDeadLetter; + + deadLetterRepo.findOne.mockResolvedValue(deadLetter); + subscriptionRepo.findOne.mockResolvedValue(subscription); + + const requeued = await service.retryDeadLetter("dlq-1", "user-1"); + + expect(requeued).not.toBeNull(); + expect(requeued.status).toBe(WebhookDeliveryStatus.PENDING); + expect(requeued.maxAttempts).toBe(4); + expect(deadLetter.retried).toBe(true); + expect(deadLetterRepo.save).toHaveBeenCalledWith(deadLetter); + }); + + it("returns null when a user attempts to retry another user's dead letter", async () => { + deadLetterRepo.findOne.mockResolvedValue(null); + + const res = await service.retryDeadLetter("dlq-other", "user-attacker"); + + expect(res).toBeNull(); + expect(deliveryRepo.save).not.toHaveBeenCalled(); + }); + }); +}); diff --git a/src/infrastructure/webhooks/services/webhook-delivery.service.ts b/src/infrastructure/webhooks/services/webhook-delivery.service.ts index 43b030d..e4f9f6c 100644 --- a/src/infrastructure/webhooks/services/webhook-delivery.service.ts +++ b/src/infrastructure/webhooks/services/webhook-delivery.service.ts @@ -1,8 +1,10 @@ -import { Injectable, Logger } from "@nestjs/common"; +import { Injectable, Logger, Optional } from "@nestjs/common"; import { InjectRepository } from "@nestjs/typeorm"; import { Repository } from "typeorm"; import { HttpService } from "@nestjs/axios"; import { firstValueFrom } from "rxjs"; +import * as http from "http"; +import * as https from "https"; import { WebhookDelivery, WebhookDeliveryStatus, @@ -14,6 +16,10 @@ import { } from "../entities/webhook-subscription.entity"; import { WebhookEvent } from "../entities/webhook-event.entity"; import { WebhookHmacService } from "./webhook-hmac.service"; +import { WebhookUrlValidatorService } from "./webhook-url-validator.service"; + +const MAX_STORED_RESPONSE_BODY_CHARS = 512; +const OUTBOUND_WINDOW_MS = 60_000; export interface DeliveryResult { statusCode?: number; @@ -39,6 +45,8 @@ export interface WebhookMetrics { @Injectable() export class WebhookDeliveryService { private readonly logger = new Logger(WebhookDeliveryService.name); + private readonly urlValidator: WebhookUrlValidatorService; + private readonly deliveryWindows = new Map(); constructor( @InjectRepository(WebhookDelivery) @@ -51,7 +59,41 @@ export class WebhookDeliveryService { private readonly subscriptionRepo: Repository, private readonly httpService: HttpService, private readonly hmacService: WebhookHmacService, - ) {} + @Optional() urlValidator?: WebhookUrlValidatorService, + ) { + this.urlValidator = urlValidator ?? new WebhookUrlValidatorService(); + } + + /** + * Enforce per-subscription outbound delivery rate limit (sliding 60s window). + */ + private consumeOutboundRateLimit(subscription: WebhookSubscription): boolean { + const limit = subscription.rateLimitPerMinute ?? 10; + if (limit <= 0) return false; + const now = Date.now(); + const cutoff = now - OUTBOUND_WINDOW_MS; + const recent = (this.deliveryWindows.get(subscription.id) || []).filter( + (t) => t > cutoff, + ); + if (recent.length >= limit) { + this.deliveryWindows.set(subscription.id, recent); + return false; + } + recent.push(now); + this.deliveryWindows.set(subscription.id, recent); + return true; + } + + private sanitizeResponseBody(data: unknown): string | undefined { + if (data === undefined || data === null) return undefined; + const raw = + typeof data === "string" ? data : JSON.stringify(data) ?? String(data); + const truncated = + raw.length > MAX_STORED_RESPONSE_BODY_CHARS + ? `${raw.slice(0, MAX_STORED_RESPONSE_BODY_CHARS)}...[truncated]` + : raw; + return truncated; + } /** * Create delivery records for an event against all matching subscriptions. @@ -75,7 +117,8 @@ export class WebhookDeliveryService { } /** - * Execute a single delivery attempt with HMAC signing. + * Execute a single delivery attempt with HMAC signing, delivery-time SSRF + * re-resolution, redirect blocking, and outbound rate limiting. */ async executeDelivery(delivery: WebhookDelivery): Promise { const subscription = await this.subscriptionRepo.findOne({ @@ -91,6 +134,27 @@ export class WebhookDeliveryService { }; } + // Re-validate and re-resolve URL at delivery time to prevent DNS rebinding + let validatedTarget; + try { + validatedTarget = await this.urlValidator.validateWebhookUrl( + subscription.url, + ); + } catch (err: any) { + return { + durationMs: 0, + error: `SSRF validation failed: ${err.message || "Blocked webhook destination"}`, + }; + } + + // Outbound rate limit check per subscription + if (!this.consumeOutboundRateLimit(subscription)) { + return { + durationMs: 0, + error: `Outbound rate limit exceeded (${subscription.rateLimitPerMinute ?? 10}/min)`, + }; + } + const payload = JSON.stringify({ id: event.id, type: event.eventType, @@ -114,22 +178,46 @@ export class WebhookDeliveryService { delivery.lastAttemptAt = new Date(); await this.deliveryRepo.save(delivery); + // Pin the socket lookup to the validated DNS record so a second lookup + // inside Node's HTTP agent cannot rebind to a private IP. + const pinned = validatedTarget.resolvedAddresses[0]; + const lookup = pinned + ? ( + _hostname: string, + options: any, + cb: (err: NodeJS.ErrnoException | null, address: any, family?: number) => void, + ) => { + if (typeof options === "function") { + options(null, pinned.address, pinned.family); + return; + } + if (options && options.all) { + cb(null, [{ address: pinned.address, family: pinned.family }]); + return; + } + cb(null, pinned.address, pinned.family); + } + : undefined; + + const httpAgent = lookup ? new http.Agent({ lookup }) : undefined; + const httpsAgent = lookup ? new https.Agent({ lookup }) : undefined; + const startTime = Date.now(); try { const response = await firstValueFrom( this.httpService.post(subscription.url, payload, { headers, timeout: subscription.timeoutMs || 30000, - validateStatus: () => true, // Don't throw for 4xx/5xx + maxRedirects: 0, // Never follow redirects to prevent 302-to-internal SSRF + httpAgent, + httpsAgent, + validateStatus: () => true, // Don't throw for 3xx/4xx/5xx }), ); const durationMs = Date.now() - startTime; const statusCode = response.status; - const responseBody = - typeof response.data === "string" - ? response.data.substring(0, 10000) - : JSON.stringify(response.data).substring(0, 10000); + const responseBody = this.sanitizeResponseBody(response.data); const respHeaders: Record = {}; if (response.headers) { @@ -138,6 +226,16 @@ export class WebhookDeliveryService { } } + if (statusCode >= 300 && statusCode < 400) { + return { + statusCode, + responseBody, + durationMs, + responseHeaders: respHeaders, + error: `Redirect responses (HTTP ${statusCode}) are not allowed for webhook deliveries`, + }; + } + const isSuccess = statusCode >= 200 && statusCode < 300; return { statusCode, @@ -309,16 +407,23 @@ export class WebhookDeliveryService { } /** - * Reattempt a dead-lettered delivery. + * Reattempt a dead-lettered delivery, scoped to the owning user when provided. */ - async retryDeadLetter(deadLetterId: string): Promise { - const dl = await this.deadLetterRepo.findOne({ - where: { id: deadLetterId }, - }); + async retryDeadLetter( + deadLetterId: string, + userId?: string, + ): Promise { + const where: any = { id: deadLetterId }; + if (userId) { + where.userId = userId; + } + const dl = await this.deadLetterRepo.findOne({ where }); if (!dl || dl.retried) return null; const subscription = await this.subscriptionRepo.findOne({ - where: { id: dl.subscriptionId }, + where: userId + ? { id: dl.subscriptionId, userId } + : { id: dl.subscriptionId }, }); if ( !subscription || diff --git a/src/infrastructure/webhooks/services/webhook-event.service.spec.ts b/src/infrastructure/webhooks/services/webhook-event.service.spec.ts new file mode 100644 index 0000000..cca7339 --- /dev/null +++ b/src/infrastructure/webhooks/services/webhook-event.service.spec.ts @@ -0,0 +1,172 @@ +import { Queue } from "bull"; +import { Repository } from "typeorm"; +import { WebhookDelivery } from "src/infrastructure/webhooks/entities/webhook-delivery.entity"; +import { + WebhookEvent, + WebhookEventStatus, +} from "src/infrastructure/webhooks/entities/webhook-event.entity"; +import { WebhookSubscription } from "src/infrastructure/webhooks/entities/webhook-subscription.entity"; +import { WebhookDeliveryService } from "./webhook-delivery.service"; +import { WebhookEventService } from "./webhook-event.service"; +import { WebhookSubscriptionService } from "./webhook-subscription.service"; + +describe("WebhookEventService", () => { + let service: WebhookEventService; + let eventRepo: jest.Mocked< + Pick, "create" | "save" | "findOne" | "find"> + >; + let subscriptionService: jest.Mocked< + Pick + >; + let deliveryService: jest.Mocked< + Pick + >; + let deliveryQueue: jest.Mocked>; + + beforeEach(() => { + eventRepo = { + create: jest.fn((data) => ({ + id: "evt-1", + createdAt: new Date("2026-01-01T00:00:00.000Z"), + deliveryCount: 0, + ...data, + })) as any, + save: jest.fn(async (event) => event as WebhookEvent), + findOne: jest.fn(), + find: jest.fn(), + }; + + subscriptionService = { + findActiveForEvent: jest.fn(), + }; + + deliveryService = { + createDeliveries: jest.fn(), + }; + + deliveryQueue = { + add: jest.fn().mockResolvedValue({ id: "job-1" } as any), + }; + + service = new WebhookEventService( + eventRepo as unknown as Repository, + subscriptionService as unknown as WebhookSubscriptionService, + deliveryService as unknown as WebhookDeliveryService, + deliveryQueue as unknown as Queue, + ); + }); + + describe("publishEvent", () => { + it("marks the event DELIVERED with deliveryCount=0 when no active subscribers match", async () => { + subscriptionService.findActiveForEvent.mockResolvedValue([]); + + const event = await service.publishEvent({ + eventType: "agent.created", + payload: { agentId: "ag-1" }, + }); + + expect(event.status).toBe(WebhookEventStatus.DELIVERED); + expect(event.deliveryCount).toBe(0); + expect(deliveryService.createDeliveries).not.toHaveBeenCalled(); + expect(deliveryQueue.add).not.toHaveBeenCalled(); + }); + + it("creates deliveries and enqueues a Bull job per matching subscriber", async () => { + const subscribers = [ + { id: "sub-1" } as WebhookSubscription, + { id: "sub-2" } as WebhookSubscription, + ]; + const deliveries = [ + { id: "deliv-1", subscriptionId: "sub-1" } as WebhookDelivery, + { id: "deliv-2", subscriptionId: "sub-2" } as WebhookDelivery, + ]; + + subscriptionService.findActiveForEvent.mockResolvedValue(subscribers); + deliveryService.createDeliveries.mockResolvedValue(deliveries); + + const event = await service.publishEvent({ + eventType: "agent.updated", + payload: { agentId: "ag-9" }, + aggregateId: "ag-9", + metadata: { source: "api" }, + }); + + expect(deliveryService.createDeliveries).toHaveBeenCalledWith( + expect.objectContaining({ id: "evt-1", eventType: "agent.updated" }), + subscribers, + ); + expect(deliveryQueue.add).toHaveBeenCalledTimes(2); + expect(deliveryQueue.add).toHaveBeenNthCalledWith( + 1, + "deliver", + { deliveryId: "deliv-1", eventId: "evt-1" }, + { attempts: 1, removeOnComplete: 100, removeOnFail: 200 }, + ); + expect(deliveryQueue.add).toHaveBeenNthCalledWith( + 2, + "deliver", + { deliveryId: "deliv-2", eventId: "evt-1" }, + { attempts: 1, removeOnComplete: 100, removeOnFail: 200 }, + ); + expect(event.status).toBe(WebhookEventStatus.DELIVERING); + expect(event.deliveryCount).toBe(2); + }); + }); + + describe("getEvent & listEvents", () => { + it("retrieves an event by ID", async () => { + const stored = { id: "evt-1" } as WebhookEvent; + eventRepo.findOne.mockResolvedValue(stored); + + await expect(service.getEvent("evt-1")).resolves.toBe(stored); + expect(eventRepo.findOne).toHaveBeenCalledWith({ + where: { id: "evt-1" }, + }); + }); + + it("lists events with optional eventType filter, limit, and offset", async () => { + eventRepo.find.mockResolvedValue([]); + + await service.listEvents("agent.created", 25, 10); + + expect(eventRepo.find).toHaveBeenCalledWith({ + where: { eventType: "agent.created" }, + order: { createdAt: "DESC" }, + take: 25, + skip: 10, + }); + }); + }); + + describe("replayEvent", () => { + it("returns null if the original event does not exist", async () => { + eventRepo.findOne.mockResolvedValue(null); + + await expect(service.replayEvent("missing-id")).resolves.toBeNull(); + }); + + it("re-publishes the original event payload with replayedFrom metadata", async () => { + const original = { + id: "evt-orig", + eventType: "agent.created", + payload: { agentId: "ag-1" }, + aggregateId: "ag-1", + metadata: { env: "staging" }, + } as unknown as WebhookEvent; + eventRepo.findOne.mockResolvedValue(original); + subscriptionService.findActiveForEvent.mockResolvedValue([]); + + const replayed = await service.replayEvent("evt-orig"); + + expect(replayed).not.toBeNull(); + expect(eventRepo.create).toHaveBeenCalledWith( + expect.objectContaining({ + eventType: "agent.created", + payload: { agentId: "ag-1" }, + aggregateId: "ag-1", + metadata: { env: "staging", replayedFrom: "evt-orig" }, + }), + ); + }); + }); +}); diff --git a/src/infrastructure/webhooks/services/webhook-hmac.service.spec.ts b/src/infrastructure/webhooks/services/webhook-hmac.service.spec.ts new file mode 100644 index 0000000..d3980c1 --- /dev/null +++ b/src/infrastructure/webhooks/services/webhook-hmac.service.spec.ts @@ -0,0 +1,94 @@ +import { WebhookHmacService } from "./webhook-hmac.service"; + +describe("WebhookHmacService", () => { + let service: WebhookHmacService; + + beforeEach(() => { + service = new WebhookHmacService(); + }); + + describe("sign", () => { + it("produces a deterministic sha256= prefixed hex digest", () => { + const payload = JSON.stringify({ id: "evt_1", eventType: "agent.created" }); + const secret = "whsec_test_secret_key_123456"; + + const sig1 = service.sign(payload, secret); + const sig2 = service.sign(payload, secret); + + expect(sig1).toBe(sig2); + expect(sig1).toMatch(/^sha256=[a-f0-9]{64}$/); + }); + + it("produces different signatures for different secrets or payloads", () => { + const payload = '{"event":"order.paid"}'; + const sigA = service.sign(payload, "secret-a"); + const sigB = service.sign(payload, "secret-b"); + const sigDifferentPayload = service.sign('{"event":"order.failed"}', "secret-a"); + + expect(sigA).not.toBe(sigB); + expect(sigA).not.toBe(sigDifferentPayload); + }); + }); + + describe("buildSignedHeaders", () => { + it("includes standard webhook metadata headers and HMAC signature", () => { + const payload = '{"id":"evt_42"}'; + const secret = "whsec_signing_key"; + const headers = service.buildSignedHeaders( + secret, + payload, + "evt_42", + "agent.updated", + ); + + expect(headers["Content-Type"]).toBe("application/json"); + expect(headers["X-Webhook-Event-Id"]).toBe("evt_42"); + expect(headers["X-Webhook-Event-Type"]).toBe("agent.updated"); + expect(headers["X-Webhook-Signature"]).toBe(service.sign(payload, secret)); + expect(headers["User-Agent"]).toBe("AlianStructure-Webhook/1.0"); + expect(Number(headers["X-Webhook-Timestamp"])).not.toBeNaN(); + }); + + it("merges custom extra headers", () => { + const headers = service.buildSignedHeaders( + "whsec_key", + "{}", + "evt_1", + "test.event", + { "X-Custom-Header": "custom-value" }, + ); + + expect(headers["X-Custom-Header"]).toBe("custom-value"); + expect(headers["X-Webhook-Event-Id"]).toBe("evt_1"); + }); + }); + + describe("verify", () => { + const payload = '{"id":"evt_99","data":{"amount":100}}'; + const secret = "whsec_verify_secret"; + + it("returns true for a valid signature produced by sign()", () => { + const signature = service.sign(payload, secret); + expect(service.verify(payload, secret, signature)).toBe(true); + }); + + it("returns false when verified with the wrong secret", () => { + const signature = service.sign(payload, secret); + expect(service.verify(payload, "wrong_secret", signature)).toBe(false); + }); + + it("returns false when the payload has been tampered with", () => { + const signature = service.sign(payload, secret); + const tampered = '{"id":"evt_99","data":{"amount":999999}}'; + expect(service.verify(tampered, secret, signature)).toBe(false); + }); + + it("returns false cleanly without throwing on wrong-length or malformed signatures", () => { + expect(service.verify(payload, secret, "")).toBe(false); + expect(service.verify(payload, secret, "sha256=deadbeef")).toBe(false); + expect(service.verify(payload, secret, "not-a-valid-signature")).toBe( + false, + ); + }); + }); +}); diff --git a/src/infrastructure/webhooks/services/webhook-processor.service.spec.ts b/src/infrastructure/webhooks/services/webhook-processor.service.spec.ts new file mode 100644 index 0000000..516df0c --- /dev/null +++ b/src/infrastructure/webhooks/services/webhook-processor.service.spec.ts @@ -0,0 +1,67 @@ +import { Job } from "bull"; +import { + WebhookDelivery, + WebhookDeliveryStatus, +} from "src/infrastructure/webhooks/entities/webhook-delivery.entity"; +import { WebhookDeliveryService } from "./webhook-delivery.service"; +import { + WebhookDeliveryJobData, + WebhookProcessor, +} from "./webhook-processor.service"; + +describe("WebhookProcessor", () => { + let processor: WebhookProcessor; + let deliveryService: jest.Mocked>; + + beforeEach(() => { + deliveryService = { + processDelivery: jest.fn(), + }; + processor = new WebhookProcessor( + deliveryService as unknown as WebhookDeliveryService, + ); + }); + + it("delegates delivery processing to WebhookDeliveryService and returns summary", async () => { + const mockDelivery: Partial = { + id: "deliv-1", + status: WebhookDeliveryStatus.SUCCESS, + attempts: 1, + statusCode: 200, + durationMs: 42, + }; + deliveryService.processDelivery.mockResolvedValue( + mockDelivery as WebhookDelivery, + ); + + const job = { + id: "job-101", + data: { deliveryId: "deliv-1", eventId: "evt-1" }, + } as unknown as Job; + + const result = await processor.handleDelivery(job); + + expect(deliveryService.processDelivery).toHaveBeenCalledWith("deliv-1"); + expect(result).toEqual({ + status: WebhookDeliveryStatus.SUCCESS, + deliveryId: "deliv-1", + attempts: 1, + statusCode: 200, + durationMs: 42, + }); + }); + + it("rethrows errors from WebhookDeliveryService so Bull can record job failure", async () => { + const failure = new Error("Database connection lost"); + deliveryService.processDelivery.mockRejectedValue(failure); + + const job = { + id: "job-102", + data: { deliveryId: "deliv-2", eventId: "evt-2" }, + } as unknown as Job; + + await expect(processor.handleDelivery(job)).rejects.toThrow( + "Database connection lost", + ); + }); +}); diff --git a/src/infrastructure/webhooks/services/webhook-subscription.service.spec.ts b/src/infrastructure/webhooks/services/webhook-subscription.service.spec.ts new file mode 100644 index 0000000..cfa95e4 --- /dev/null +++ b/src/infrastructure/webhooks/services/webhook-subscription.service.spec.ts @@ -0,0 +1,229 @@ +import { BadRequestException, NotFoundException } from "@nestjs/common"; +import { Repository } from "typeorm"; +import { + WebhookSubscription, + WebhookSubscriptionStatus, +} from "src/infrastructure/webhooks/entities/webhook-subscription.entity"; +import { WebhookSubscriptionService } from "./webhook-subscription.service"; +import { WebhookUrlValidatorService } from "./webhook-url-validator.service"; + +describe("WebhookSubscriptionService", () => { + let service: WebhookSubscriptionService; + let subscriptionRepo: jest.Mocked< + Pick< + Repository, + "create" | "save" | "find" | "findOne" | "createQueryBuilder" + > + >; + let urlValidator: WebhookUrlValidatorService; + + beforeEach(() => { + subscriptionRepo = { + create: jest.fn((entity) => ({ + id: "sub-1", + createdAt: new Date("2026-01-01T00:00:00.000Z"), + updatedAt: new Date("2026-01-01T00:00:00.000Z"), + ...entity, + })) as any, + save: jest.fn(async (entity) => entity as WebhookSubscription), + find: jest.fn(), + findOne: jest.fn(), + createQueryBuilder: jest.fn(), + }; + + urlValidator = new WebhookUrlValidatorService(); + urlValidator.setLookupFn(async () => [ + { address: "93.184.216.34", family: 4 }, + ]); + + service = new WebhookSubscriptionService( + subscriptionRepo as unknown as Repository, + urlValidator, + ); + }); + + describe("create", () => { + it("validates the URL, generates a whsec_ signing key, and applies default retry/rate settings", async () => { + const result = await service.create("user-1", { + url: "https://hooks.example.com/events", + events: ["agent.created", "agent.updated"], + description: "Production webhook", + }); + + expect(result.userId).toBe("user-1"); + expect(result.url).toBe("https://hooks.example.com/events"); + expect(result.signingKey).toMatch(/^whsec_[a-f0-9]{64}$/); + expect(result.maxRetries).toBe(5); + expect(result.retryDelayMs).toBe(1000); + expect(result.backoffMultiplier).toBe(2); + expect(result.timeoutMs).toBe(30000); + expect(result.rateLimitPerMinute).toBe(10); + expect(result.status).toBe(WebhookSubscriptionStatus.ACTIVE); + expect(subscriptionRepo.save).toHaveBeenCalledTimes(1); + }); + + it("rejects SSRF / loopback / metadata URLs before persisting", async () => { + await expect( + service.create("user-1", { + url: "http://169.254.169.254/latest/meta-data/", + events: ["*"], + }), + ).rejects.toThrow(BadRequestException); + + expect(subscriptionRepo.save).not.toHaveBeenCalled(); + }); + }); + + describe("findAll & findOne", () => { + it("scopes findAll to the requesting userId ordered by createdAt DESC", async () => { + subscriptionRepo.find.mockResolvedValue([]); + + await service.findAll("user-42"); + + expect(subscriptionRepo.find).toHaveBeenCalledWith({ + where: { userId: "user-42" }, + order: { createdAt: "DESC" }, + }); + }); + + it("returns the subscription when id and userId match", async () => { + const existing = { + id: "sub-1", + userId: "user-1", + } as WebhookSubscription; + subscriptionRepo.findOne.mockResolvedValue(existing); + + const found = await service.findOne("sub-1", "user-1"); + expect(found).toBe(existing); + expect(subscriptionRepo.findOne).toHaveBeenCalledWith({ + where: { id: "sub-1", userId: "user-1" }, + }); + }); + + it("throws NotFoundException when subscription does not exist or belongs to another user", async () => { + subscriptionRepo.findOne.mockResolvedValue(null); + + await expect(service.findOne("sub-1", "other-user")).rejects.toThrow( + NotFoundException, + ); + }); + }); + + describe("findActiveForEvent", () => { + it("matches active subscriptions subscribed to the exact event type or wildcard '*'", async () => { + const wildcardSub = { + id: "sub-wildcard", + status: WebhookSubscriptionStatus.ACTIVE, + events: ["*"], + } as WebhookSubscription; + const exactSub = { + id: "sub-exact", + status: WebhookSubscriptionStatus.ACTIVE, + events: ["agent.created", "agent.deleted"], + } as WebhookSubscription; + const unrelatedSub = { + id: "sub-unrelated", + status: WebhookSubscriptionStatus.ACTIVE, + events: ["payment.completed"], + } as WebhookSubscription; + + subscriptionRepo.find.mockResolvedValue([ + wildcardSub, + exactSub, + unrelatedSub, + ]); + + const matches = await service.findActiveForEvent("agent.created"); + + expect(subscriptionRepo.find).toHaveBeenCalledWith({ + where: { status: WebhookSubscriptionStatus.ACTIVE }, + }); + expect(matches).toEqual([wildcardSub, exactSub]); + }); + }); + + describe("update", () => { + it("re-validates a newly supplied URL and saves updates", async () => { + const existing = { + id: "sub-1", + userId: "user-1", + url: "https://hooks.example.com/old", + description: "Old", + } as WebhookSubscription; + subscriptionRepo.findOne.mockResolvedValue(existing); + + const updated = await service.update("sub-1", "user-1", { + url: "https://hooks.example.com/new", + description: "New", + }); + + expect(updated.url).toBe("https://hooks.example.com/new"); + expect(updated.description).toBe("New"); + }); + + it("rejects updating a subscription URL to a private/loopback target", async () => { + const existing = { + id: "sub-1", + userId: "user-1", + url: "https://hooks.example.com/old", + } as WebhookSubscription; + subscriptionRepo.findOne.mockResolvedValue(existing); + + await expect( + service.update("sub-1", "user-1", { + url: "https://127.0.0.1/internal", + }), + ).rejects.toThrow(BadRequestException); + + expect(subscriptionRepo.save).not.toHaveBeenCalled(); + }); + }); + + describe("remove, rotateKey, and toggleStatus", () => { + it("soft-disables a subscription on remove()", async () => { + const existing = { + id: "sub-1", + userId: "user-1", + status: WebhookSubscriptionStatus.ACTIVE, + } as WebhookSubscription; + subscriptionRepo.findOne.mockResolvedValue(existing); + + await service.remove("sub-1", "user-1"); + + expect(existing.status).toBe(WebhookSubscriptionStatus.DISABLED); + expect(subscriptionRepo.save).toHaveBeenCalledWith(existing); + }); + + it("generates a distinct signing key on rotateKey()", async () => { + const existing = { + id: "sub-1", + userId: "user-1", + signingKey: "whsec_old_key", + } as WebhookSubscription; + subscriptionRepo.findOne.mockResolvedValue(existing); + + const rotated = await service.rotateKey("sub-1", "user-1"); + + expect(rotated.newSigningKey).toMatch(/^whsec_[a-f0-9]{64}$/); + expect(rotated.newSigningKey).not.toBe("whsec_old_key"); + expect(existing.signingKey).toBe(rotated.newSigningKey); + }); + + it("updates subscription status on toggleStatus()", async () => { + const existing = { + id: "sub-1", + userId: "user-1", + status: WebhookSubscriptionStatus.ACTIVE, + } as WebhookSubscription; + subscriptionRepo.findOne.mockResolvedValue(existing); + + const paused = await service.toggleStatus( + "sub-1", + "user-1", + WebhookSubscriptionStatus.PAUSED, + ); + + expect(paused.status).toBe(WebhookSubscriptionStatus.PAUSED); + }); + }); +}); diff --git a/src/infrastructure/webhooks/services/webhook-subscription.service.ts b/src/infrastructure/webhooks/services/webhook-subscription.service.ts index 8197cca..284f0d8 100644 --- a/src/infrastructure/webhooks/services/webhook-subscription.service.ts +++ b/src/infrastructure/webhooks/services/webhook-subscription.service.ts @@ -1,4 +1,9 @@ -import { Injectable, Logger, NotFoundException } from "@nestjs/common"; +import { + Injectable, + Logger, + NotFoundException, + Optional, +} from "@nestjs/common"; import { InjectRepository } from "@nestjs/typeorm"; import { Repository } from "typeorm"; import { randomBytes } from "crypto"; @@ -10,20 +15,26 @@ import { CreateWebhookSubscriptionDto, UpdateWebhookSubscriptionDto, } from "../dto/webhook.dto"; +import { WebhookUrlValidatorService } from "./webhook-url-validator.service"; @Injectable() export class WebhookSubscriptionService { private readonly logger = new Logger(WebhookSubscriptionService.name); + private readonly urlValidator: WebhookUrlValidatorService; constructor( @InjectRepository(WebhookSubscription) private readonly subscriptionRepo: Repository, - ) {} + @Optional() urlValidator?: WebhookUrlValidatorService, + ) { + this.urlValidator = urlValidator ?? new WebhookUrlValidatorService(); + } async create( userId: string, dto: CreateWebhookSubscriptionDto, ): Promise { + await this.urlValidator.validateWebhookUrl(dto.url); const signingKey = this.generateSigningKey(); const subscription = this.subscriptionRepo.create({ userId, @@ -42,7 +53,7 @@ export class WebhookSubscriptionService { }); const saved = await this.subscriptionRepo.save(subscription); this.logger.log( - `Webhook subscription created: ${saved.id} for user ${userId}, url=${dto.url}`, + `Webhook subscription created: ${saved.id} for user ${userId}`, ); return { ...saved, signingKey }; } @@ -93,6 +104,9 @@ export class WebhookSubscriptionService { dto: UpdateWebhookSubscriptionDto, ): Promise { const sub = await this.findOne(id, userId); + if (dto.url !== undefined) { + await this.urlValidator.validateWebhookUrl(dto.url); + } Object.assign(sub, dto); const saved = await this.subscriptionRepo.save(sub); this.logger.log(`Webhook subscription updated: ${id}`); diff --git a/src/infrastructure/webhooks/services/webhook-url-validator.service.spec.ts b/src/infrastructure/webhooks/services/webhook-url-validator.service.spec.ts new file mode 100644 index 0000000..62692af --- /dev/null +++ b/src/infrastructure/webhooks/services/webhook-url-validator.service.spec.ts @@ -0,0 +1,228 @@ +import { BadRequestException } from "@nestjs/common"; +import { + DnsLookupFn, + WebhookUrlValidatorService, +} from "./webhook-url-validator.service"; + +describe("WebhookUrlValidatorService", () => { + const originalEnv = { ...process.env }; + let publicDnsResolver: jest.MockedFunction; + let validator: WebhookUrlValidatorService; + + beforeEach(() => { + process.env = { ...originalEnv }; + delete process.env.WEBHOOK_ALLOW_INSECURE_HTTP; + delete process.env.WEBHOOK_ALLOW_PRIVATE_NETWORKS; + delete process.env.WEBHOOK_ALLOWED_PORTS; + process.env.NODE_ENV = "test"; + + publicDnsResolver = jest + .fn, Parameters>() + .mockResolvedValue([{ address: "93.184.216.34", family: 4 }]); + + validator = new WebhookUrlValidatorService(); + validator.setLookupFn(publicDnsResolver); + }); + + afterAll(() => { + process.env = originalEnv; + }); + + describe("valid public endpoints", () => { + it("accepts a valid https URL resolving to a public IPv4 address", async () => { + const result = await validator.validateWebhookUrl( + "https://hooks.example.com/webhooks/receive", + ); + + expect(result.url.toString()).toBe( + "https://hooks.example.com/webhooks/receive", + ); + expect(result.hostname).toBe("hooks.example.com"); + expect(result.port).toBe(443); + expect(result.resolvedAddresses).toEqual([ + { address: "93.184.216.34", family: 4 }, + ]); + expect(publicDnsResolver).toHaveBeenCalledWith("hooks.example.com"); + }); + + it("accepts a public IPv4 literal in the 172.32.0.0/16 range", async () => { + const result = await validator.validateWebhookUrl( + "https://172.32.0.10/webhook", + ); + + expect(result.resolvedAddresses).toEqual([ + { address: "172.32.0.10", family: 4 }, + ]); + expect(publicDnsResolver).not.toHaveBeenCalled(); + }); + + it("accepts a public IPv6 literal", async () => { + const result = await validator.validateWebhookUrl( + "https://[2606:2800:220:1:248:1893:25c8:1946]/hook", + ); + + expect(result.resolvedAddresses).toEqual([ + { address: "2606:2800:220:1:248:1893:25c8:1946", family: 6 }, + ]); + }); + }); + + describe("scheme and syntax restrictions", () => { + it.each([ + "", + "not-a-url", + "file:///etc/passwd", + "gopher://127.0.0.1:6379/_INFO", + "ftp://hooks.example.com/resource", + "data:text/plain;base64,SGVsbG8=", + ])("rejects disallowed or malformed URL: %s", async (input) => { + await expect(validator.validateWebhookUrl(input)).rejects.toThrow( + BadRequestException, + ); + }); + + it("rejects http:// URLs by default outside development", async () => { + await expect( + validator.validateWebhookUrl("http://hooks.example.com/webhook"), + ).rejects.toThrow(BadRequestException); + }); + + it("allows http:// URLs when WEBHOOK_ALLOW_INSECURE_HTTP=true outside production", async () => { + process.env.WEBHOOK_ALLOW_INSECURE_HTTP = "true"; + + const result = await validator.validateWebhookUrl( + "http://hooks.example.com/webhook", + ); + + expect(result.url.toString()).toBe("http://hooks.example.com/webhook"); + expect(result.port).toBe(80); + }); + + it("rejects URLs with embedded credentials", async () => { + await expect( + validator.validateWebhookUrl( + "https://admin:secret@hooks.example.com/webhook", + ), + ).rejects.toThrow(BadRequestException); + }); + }); + + describe("port allowlist", () => { + it.each([ + "https://hooks.example.com:6379/hook", + "https://hooks.example.com:5432/hook", + "https://hooks.example.com:8080/hook", + "https://hooks.example.com:22/hook", + ])("rejects non-allowlisted port in %s", async (url) => { + await expect(validator.validateWebhookUrl(url)).rejects.toThrow( + BadRequestException, + ); + }); + + it("allows custom ports when WEBHOOK_ALLOWED_PORTS is configured", async () => { + process.env.WEBHOOK_ALLOWED_PORTS = "8443,9443"; + + const result = await validator.validateWebhookUrl( + "https://hooks.example.com:8443/hook", + ); + + expect(result.url.toString()).toBe("https://hooks.example.com:8443/hook"); + expect(result.port).toBe(8443); + }); + }); + + describe("SSRF private, loopback, link-local, and metadata protection", () => { + it.each([ + ["loopback 127.0.0.1", "https://127.0.0.1/hook"], + ["loopback 127.255.255.254", "https://127.255.255.254/hook"], + ["localhost hostname", "https://localhost/hook"], + ["subdomain of localhost", "https://api.localhost/hook"], + ["internal TLD", "https://redis.internal/hook"], + ["local mDNS TLD", "https://printer.local/hook"], + [ + "GCP metadata hostname", + "https://metadata.google.internal/computeMetadata/v1/", + ], + [ + "AWS/GCP/Azure metadata IP", + "https://169.254.169.254/latest/meta-data/", + ], + ["RFC 1918 10.0.0.0/8", "https://10.0.0.1/hook"], + ["RFC 1918 172.16.0.0/12 start", "https://172.16.0.1/hook"], + ["RFC 1918 172.16.0.0/12 end", "https://172.31.255.254/hook"], + ["RFC 1918 192.168.0.0/16", "https://192.168.1.1/hook"], + ["CGNAT 100.64.0.0/10", "https://100.64.0.1/hook"], + ["unspecified 0.0.0.0", "https://0.0.0.0/hook"], + ["IPv6 loopback ::1", "https://[::1]/hook"], + ["IPv6 unspecified ::", "https://[::]/hook"], + ["IPv4-mapped IPv6 loopback", "https://[::ffff:127.0.0.1]/hook"], + ["IPv4-mapped IPv6 metadata", "https://[::ffff:169.254.169.254]/hook"], + ["IPv6 unique-local fc00::/7", "https://[fd00::1]/hook"], + ["IPv6 link-local fe80::/10", "https://[fe80::1]/hook"], + ["short-form IPv4 127.1", "https://127.1/hook"], + ["decimal integer IPv4 2130706433", "https://2130706433/hook"], + ["hex IPv4 0x7f000001", "https://0x7f000001/hook"], + ["octal IPv4 0177.0.0.1", "https://0177.0.0.1/hook"], + ])("rejects %s (%s)", async (_label, url) => { + await expect(validator.validateWebhookUrl(url)).rejects.toThrow( + BadRequestException, + ); + }); + + it("rejects a public hostname whose DNS resolves to a private IPv4 address", async () => { + publicDnsResolver.mockResolvedValueOnce([ + { address: "169.254.169.254", family: 4 }, + ]); + + await expect( + validator.validateWebhookUrl("https://evil.example.com/hook"), + ).rejects.toThrow(BadRequestException); + }); + + it("rejects a hostname when any resolved DNS record is private (mixed records)", async () => { + publicDnsResolver.mockResolvedValueOnce([ + { address: "93.184.216.34", family: 4 }, + { address: "10.0.0.42", family: 4 }, + ]); + + await expect( + validator.validateWebhookUrl("https://dual.example.com/hook"), + ).rejects.toThrow(BadRequestException); + }); + + it("rejects a hostname when DNS lookup fails or returns no records", async () => { + publicDnsResolver.mockRejectedValueOnce(new Error("ENOTFOUND")); + + await expect( + validator.validateWebhookUrl("https://nonexistent.example.com/hook"), + ).rejects.toThrow(BadRequestException); + }); + + it("allows private targets only when WEBHOOK_ALLOW_PRIVATE_NETWORKS=true outside production", async () => { + process.env.WEBHOOK_ALLOW_INSECURE_HTTP = "true"; + process.env.WEBHOOK_ALLOW_PRIVATE_NETWORKS = "true"; + process.env.WEBHOOK_ALLOWED_PORTS = "8080"; + + const result = await validator.validateWebhookUrl( + "http://127.0.0.1:8080/local-hook", + ); + + expect(result.hostname).toBe("127.0.0.1"); + expect(result.port).toBe(8080); + }); + + it("never allows private networks or insecure http in NODE_ENV=production", async () => { + process.env.NODE_ENV = "production"; + process.env.WEBHOOK_ALLOW_PRIVATE_NETWORKS = "true"; + process.env.WEBHOOK_ALLOW_INSECURE_HTTP = "true"; + + await expect( + validator.validateWebhookUrl("https://127.0.0.1/hook"), + ).rejects.toThrow(BadRequestException); + + await expect( + validator.validateWebhookUrl("http://hooks.example.com/hook"), + ).rejects.toThrow(BadRequestException); + }); + }); +}); diff --git a/src/infrastructure/webhooks/services/webhook-url-validator.service.ts b/src/infrastructure/webhooks/services/webhook-url-validator.service.ts new file mode 100644 index 0000000..a0e9c5b --- /dev/null +++ b/src/infrastructure/webhooks/services/webhook-url-validator.service.ts @@ -0,0 +1,448 @@ +import { BadRequestException, Injectable } from "@nestjs/common"; +import * as dns from "dns"; +import * as net from "net"; + +export interface ValidatedWebhookTarget { + url: URL; + hostname: string; + port: number; + resolvedAddresses: Array<{ address: string; family: 4 | 6 }>; +} + +export type DnsLookupFn = ( + hostname: string, +) => Promise>; + +const DEFAULT_LOOKUP: DnsLookupFn = (hostname: string) => + dns.promises.lookup(hostname, { all: true, verbatim: true }); + +const BLOCKED_HOSTNAMES = new Set([ + "localhost", + "localhost.localdomain", + "metadata.google.internal", + "metadata", + "instance-data", +]); + +const BLOCKED_HOSTNAME_SUFFIXES = [ + ".localhost", + ".local", + ".internal", + ".home.arpa", + ".localdomain", +]; + +/** + * Parse a dotted-quad IPv4 string into an unsigned 32-bit integer. + * Returns null if the string is not a canonical 4-octet IPv4 address. + */ +export function parseIpv4ToUint32(ip: string): number | null { + const parts = ip.split("."); + if (parts.length !== 4) return null; + let result = 0; + for (const part of parts) { + if (!/^(0|[1-9]\d*)$/.test(part)) return null; + const octet = Number(part); + if (!Number.isInteger(octet) || octet < 0 || octet > 255) return null; + result = ((result << 8) | octet) >>> 0; + } + return result >>> 0; +} + +function inCidrv4(ipUint32: number, baseIp: string, prefixLen: number): boolean { + const base = parseIpv4ToUint32(baseIp); + if (base === null) return false; + const mask = prefixLen === 0 ? 0 : (~0 << (32 - prefixLen)) >>> 0; + return (ipUint32 & mask) >>> 0 === (base & mask) >>> 0; +} + +/** + * Expand an IPv6 address into 8 16-bit words (0..0xffff), or null if invalid. + */ +export function parseIpv6Words(ip: string): number[] | null { + // Strip zone index if present (e.g. fe80::1%eth0) + const zoneIdx = ip.indexOf("%"); + const clean = (zoneIdx !== -1 ? ip.slice(0, zoneIdx) : ip).toLowerCase(); + + if (!clean || clean.includes(":::")) return null; + + let working = clean; + // Handle embedded IPv4 tail (e.g. ::ffff:127.0.0.1 or ::ffff:169.254.169.254) + const lastColon = working.lastIndexOf(":"); + if (lastColon !== -1 && working.slice(lastColon + 1).includes(".")) { + const ipv4Part = working.slice(lastColon + 1); + const v4 = parseIpv4ToUint32(ipv4Part); + if (v4 === null) return null; + const hi = ((v4 >>> 16) & 0xffff).toString(16); + const lo = (v4 & 0xffff).toString(16); + working = `${working.slice(0, lastColon)}:${hi}:${lo}`; + } + + const doubleColonParts = working.split("::"); + if (doubleColonParts.length > 2) return null; + + const parseSide = (side: string): number[] | null => { + if (!side) return []; + const tokens = side.split(":"); + const nums: number[] = []; + for (const t of tokens) { + if (!/^[0-9a-f]{1,4}$/.test(t)) return null; + nums.push(parseInt(t, 16)); + } + return nums; + }; + + if (doubleColonParts.length === 1) { + const words = parseSide(doubleColonParts[0]); + return words && words.length === 8 ? words : null; + } + + const left = parseSide(doubleColonParts[0]); + const right = parseSide(doubleColonParts[1]); + if (!left || !right) return null; + const missing = 8 - (left.length + right.length); + if (missing < 1) return null; + return [...left, ...new Array(missing).fill(0), ...right]; +} + +/** + * Returns true if the given IPv4 or IPv6 literal belongs to a loopback, + * private, link-local, cloud-metadata, CGNAT, multicast,unspecified, or + * reserved address range. + */ +export function isBlockedIpAddress(ip: string): boolean { + const trimmed = ip.trim().replace(/^\[|\]$/g, ""); + const v4 = parseIpv4ToUint32(trimmed); + if (v4 !== null) { + return ( + inCidrv4(v4, "0.0.0.0", 8) || // "This" network (RFC 1122) + inCidrv4(v4, "10.0.0.0", 8) || // Private-use (RFC 1918) + inCidrv4(v4, "100.64.0.0", 10) || // Shared / CGNAT (RFC 6598) + inCidrv4(v4, "127.0.0.0", 8) || // Loopback (RFC 1122) + inCidrv4(v4, "169.254.0.0", 16) || // Link-local + cloud metadata 169.254.169.254 (RFC 3927) + inCidrv4(v4, "172.16.0.0", 12) || // Private-use (RFC 1918) + inCidrv4(v4, "192.0.0.0", 24) || // IETF Protocol Assignments (RFC 6890) + inCidrv4(v4, "192.0.2.0", 24) || // TEST-NET-1 (RFC 5737) + inCidrv4(v4, "192.88.99.0", 24) || // 6to4 relay anycast (RFC 7526) + inCidrv4(v4, "192.168.0.0", 16) || // Private-use (RFC 1918) + inCidrv4(v4, "198.18.0.0", 15) || // Benchmarking (RFC 2544) + inCidrv4(v4, "198.51.100.0", 24) || // TEST-NET-2 (RFC 5737) + inCidrv4(v4, "203.0.113.0", 24) || // TEST-NET-3 (RFC 5737) + inCidrv4(v4, "224.0.0.0", 4) || // Multicast (RFC 5771) + inCidrv4(v4, "240.0.0.0", 4) // Reserved + broadcast 255.255.255.255 + ); + } + + const words = parseIpv6Words(trimmed); + if (!words) { + // Fail closed on unparseable IP strings + return true; + } + + const isAllZeroPrefix = (count: number) => + words.slice(0, count).every((w) => w === 0); + + // Unspecified ::/128 + if (isAllZeroPrefix(8)) return true; + + // Loopback ::1/128 + if (isAllZeroPrefix(7) && words[7] === 1) return true; + + // IPv4-mapped ::ffff:0:0/96 or IPv4-compatible ::/96 + if (isAllZeroPrefix(5) && (words[5] === 0xffff || words[5] === 0)) { + const embeddedV4 = ((words[6] << 16) | words[7]) >>> 0; + const dotted = [ + (embeddedV4 >>> 24) & 0xff, + (embeddedV4 >>> 16) & 0xff, + (embeddedV4 >>> 8) & 0xff, + embeddedV4 & 0xff, + ].join("."); + return isBlockedIpAddress(dotted); + } + + // IPv4-translated ::ffff:0:0:0/96 (RFC 2765) + if ( + words[0] === 0 && + words[1] === 0 && + words[2] === 0 && + words[3] === 0 && + words[4] === 0xffff && + words[5] === 0 + ) { + const embeddedV4 = ((words[6] << 16) | words[7]) >>> 0; + const dotted = [ + (embeddedV4 >>> 24) & 0xff, + (embeddedV4 >>> 16) & 0xff, + (embeddedV4 >>> 8) & 0xff, + embeddedV4 & 0xff, + ].join("."); + return isBlockedIpAddress(dotted); + } + + // NAT64 well-known prefix 64:ff9b::/96 and 64:ff9b:1::/48 + if (words[0] === 0x0064 && words[1] === 0xff9b) { + const embeddedV4 = ((words[6] << 16) | words[7]) >>> 0; + const dotted = [ + (embeddedV4 >>> 24) & 0xff, + (embeddedV4 >>> 16) & 0xff, + (embeddedV4 >>> 8) & 0xff, + embeddedV4 & 0xff, + ].join("."); + return isBlockedIpAddress(dotted); + } + + // 6to4 2002::/16 embeds an IPv4 address in words[1..2] + if (words[0] === 0x2002) { + const embeddedV4 = ((words[1] << 16) | words[2]) >>> 0; + const dotted = [ + (embeddedV4 >>> 24) & 0xff, + (embeddedV4 >>> 16) & 0xff, + (embeddedV4 >>> 8) & 0xff, + embeddedV4 & 0xff, + ].join("."); + return isBlockedIpAddress(dotted); + } + + // Teredo 2001:0000::/32 + if (words[0] === 0x2001 && words[1] === 0x0000) return true; + + // Documentation 2001:db8::/32 + if (words[0] === 0x2001 && words[1] === 0x0db8) return true; + + // Discard-only 100::/64 + if (words[0] === 0x0100 && words[1] === 0 && words[2] === 0 && words[3] === 0) + return true; + + // Unique local fc00::/7 (fc00..fdff) + if ((words[0] & 0xfe00) === 0xfc00) return true; + + // Link-local fe80::/10 (fe80..febf) + if ((words[0] & 0xffc0) === 0xfe80) return true; + + // Site-local deprecated fec0::/10 + if ((words[0] & 0xffc0) === 0xfec0) return true; + + // Multicast ff00::/8 + if ((words[0] & 0xff00) === 0xff00) return true; + + return false; +} + +/** + * Detect obfuscated numeric/octal/hex host encodings such as `127.1`, + * `0x7f000001`, `2130706433`, or `0177.0.0.1` that URL parsers may normalize. + */ +function hasObfuscatedIpHostname(rawUrl: string, parsedHost: string): boolean { + // Check raw host authority before URL normalization + const match = rawUrl.match(/^[a-zA-Z][a-zA-Z0-9+.-]*:\/\/([^/?#]+)/); + if (match) { + const authority = match[1]; + const hostPort = authority.includes("@") + ? authority.slice(authority.lastIndexOf("@") + 1) + : authority; + const rawHost = hostPort.startsWith("[") + ? hostPort.slice(0, hostPort.indexOf("]") + 1) + : hostPort.split(":")[0]; + + // Pure integer host (e.g. 2130706433) or hex host (0x7f000001) + if (/^\d+$/.test(rawHost) || /^0x[0-9a-f]+$/i.test(rawHost)) { + return true; + } + + // Dotted quad with fewer than 4 parts of all-digits/hex/octal (e.g. 127.1) + const labels = rawHost.split("."); + if ( + labels.length >= 2 && + labels.every( + (l) => /^\d+$/.test(l) || /^0x[0-9a-f]+$/i.test(l) || /^0[0-7]+$/.test(l), + ) + ) { + if (labels.length !== 4) return true; + // Reject leading-zero octal octets like 0177.0.0.1 + if (labels.some((l) => l.length > 1 && l.startsWith("0"))) return true; + } + } + + if (/^\d+$/.test(parsedHost) || /^0x[0-9a-f]+$/i.test(parsedHost)) { + return true; + } + + return false; +} + +@Injectable() +export class WebhookUrlValidatorService { + private lookupFn: DnsLookupFn = DEFAULT_LOOKUP; + + /** + * Override the DNS lookup implementation (used in unit tests to remain offline). + */ + setLookupFn(fn: DnsLookupFn): void { + this.lookupFn = fn; + } + + private isInsecureHttpAllowed(): boolean { + const env = process.env.NODE_ENV || "production"; + if (process.env.WEBHOOK_ALLOW_INSECURE_HTTP === "true" && env !== "production") { + return true; + } + return env === "development"; + } + + private isPrivateNetworkOptOutEnabled(): boolean { + const env = process.env.NODE_ENV || "production"; + return ( + env !== "production" && + process.env.WEBHOOK_ALLOW_PRIVATE_NETWORKS === "true" + ); + } + + private getAllowedPorts(protocol: string): Set { + const ports = new Set(); + if (protocol === "https:") { + ports.add(443); + } else if (protocol === "http:") { + ports.add(80); + ports.add(443); + } + + const extra = process.env.WEBHOOK_ALLOWED_PORTS; + if (extra) { + for (const token of extra.split(",")) { + const parsed = Number(token.trim()); + if (Number.isInteger(parsed) && parsed >= 1 && parsed <= 65535) { + ports.add(parsed); + } + } + } + return ports; + } + + /** + * Validate a webhook URL at subscription registration/update time or right + * before outbound delivery (defeating DNS rebinding). + */ + async validateWebhookUrl(rawUrl: string): Promise { + if (!rawUrl || typeof rawUrl !== "string") { + throw new BadRequestException("Webhook URL must be a non-empty string"); + } + + const trimmed = rawUrl.trim(); + let parsed: URL; + try { + parsed = new URL(trimmed); + } catch { + throw new BadRequestException("Invalid webhook URL format"); + } + + // 1. Scheme allowlist + const protocol = parsed.protocol.toLowerCase(); + const allowHttp = this.isInsecureHttpAllowed(); + if (protocol !== "https:" && !(allowHttp && protocol === "http:")) { + throw new BadRequestException( + `Webhook URL scheme "${protocol.replace(":", "")}" is not allowed; use https`, + ); + } + + // 2. Reject embedded credentials (user:pass@host) + if (parsed.username || parsed.password) { + throw new BadRequestException( + "Webhook URL must not contain embedded credentials", + ); + } + + const hostname = parsed.hostname.toLowerCase().replace(/^\[|\]$/g, ""); + if (!hostname) { + throw new BadRequestException("Webhook URL must include a valid hostname"); + } + + // 3. Port validation + const effectivePort = parsed.port + ? Number(parsed.port) + : protocol === "https:" + ? 443 + : 80; + const allowedPorts = this.getAllowedPorts(protocol); + if (!allowedPorts.has(effectivePort)) { + throw new BadRequestException( + `Webhook URL port ${effectivePort} is not permitted`, + ); + } + + if (this.isPrivateNetworkOptOutEnabled()) { + return { + url: parsed, + hostname, + port: effectivePort, + resolvedAddresses: [], + }; + } + + // 4. Reject obfuscated numeric/octal/hex IP encodings and reserved hostnames + if (hasObfuscatedIpHostname(trimmed, hostname)) { + throw new BadRequestException( + "Webhook URL hostname uses a disallowed numeric or obfuscated IP encoding", + ); + } + + if ( + BLOCKED_HOSTNAMES.has(hostname) || + BLOCKED_HOSTNAME_SUFFIXES.some((s) => hostname.endsWith(s)) + ) { + throw new BadRequestException( + `Webhook URL hostname "${hostname}" is not permitted`, + ); + } + + // 5. Direct IP literal check + const ipVersion = net.isIP(hostname); + if (ipVersion === 4 || ipVersion === 6) { + if (isBlockedIpAddress(hostname)) { + throw new BadRequestException( + `Webhook URL target IP "${hostname}" is in a forbidden private, loopback, or reserved range`, + ); + } + return { + url: parsed, + hostname, + port: effectivePort, + resolvedAddresses: [{ address: hostname, family: ipVersion }], + }; + } + + // 6. Resolve DNS and verify every A/AAAA record is globally routable + let records: Array<{ address: string; family: number }>; + try { + records = await this.lookupFn(hostname); + } catch { + throw new BadRequestException( + `Webhook URL hostname "${hostname}" could not be resolved`, + ); + } + + if (!Array.isArray(records) || records.length === 0) { + throw new BadRequestException( + `Webhook URL hostname "${hostname}" resolved to no addresses`, + ); + } + + const validated: Array<{ address: string; family: 4 | 6 }> = []; + for (const rec of records) { + const addr = rec?.address; + const family = rec?.family === 6 ? 6 : 4; + if (!addr || isBlockedIpAddress(addr)) { + throw new BadRequestException( + `Webhook URL hostname "${hostname}" resolves to a forbidden internal or reserved IP address`, + ); + } + validated.push({ address: addr, family }); + } + + return { + url: parsed, + hostname, + port: effectivePort, + resolvedAddresses: validated, + }; + } +} diff --git a/src/infrastructure/webhooks/webhook.controller.ts b/src/infrastructure/webhooks/webhook.controller.ts index a6ab69f..d73d111 100644 --- a/src/infrastructure/webhooks/webhook.controller.ts +++ b/src/infrastructure/webhooks/webhook.controller.ts @@ -236,8 +236,14 @@ export class WebhookController { @HttpCode(HttpStatus.OK) @ApiOperation({ summary: "Retry a dead-lettered delivery" }) @ApiParam({ name: "id" }) - async retryDeadLetter(@Param("id") id: string) { - const delivery = await this.deliveryService.retryDeadLetter(id); + async retryDeadLetter( + @Param("id") id: string, + @CurrentUser() user: AuthUser, + ) { + if (!user?.id) { + throw new BadRequestException("User identity not found in request"); + } + const delivery = await this.deliveryService.retryDeadLetter(id, user.id); if (!delivery) { return { success: false, message: "Dead letter not found or already retried" }; } @@ -248,8 +254,11 @@ export class WebhookController { @Get("metrics") @ApiOperation({ summary: "Get webhook delivery metrics and reliability stats" }) - async getMetrics() { - const metrics = await this.deliveryService.getMetrics(); + async getMetrics(@CurrentUser() user: AuthUser) { + if (!user?.id) { + throw new BadRequestException("User identity not found in request"); + } + const metrics = await this.deliveryService.getMetrics(user.id); return { success: true, metrics }; } } diff --git a/src/infrastructure/webhooks/webhook.module.ts b/src/infrastructure/webhooks/webhook.module.ts index 553bf11..885462c 100644 --- a/src/infrastructure/webhooks/webhook.module.ts +++ b/src/infrastructure/webhooks/webhook.module.ts @@ -15,6 +15,7 @@ import { WebhookEventService } from "./services/webhook-event.service"; import { WebhookDeliveryService } from "./services/webhook-delivery.service"; import { WebhookHmacService } from "./services/webhook-hmac.service"; import { WebhookProcessor } from "./services/webhook-processor.service"; +import { WebhookUrlValidatorService } from "./services/webhook-url-validator.service"; @Module({ imports: [ @@ -45,6 +46,7 @@ import { WebhookProcessor } from "./services/webhook-processor.service"; ], controllers: [WebhookController], providers: [ + WebhookUrlValidatorService, WebhookSubscriptionService, WebhookEventService, WebhookDeliveryService, @@ -52,6 +54,7 @@ import { WebhookProcessor } from "./services/webhook-processor.service"; WebhookProcessor, ], exports: [ + WebhookUrlValidatorService, WebhookEventService, WebhookDeliveryService, WebhookSubscriptionService,