feat(event-process): SignatureValidator registry + 7 provider validators (EVO-1210)

.env.example

@@ -229,3 +229,17 @@ EVOAI_CRM_CIRCUIT_RECOVERY_MS=60000
 EVO_AUTH_API_TOKEN=
 EVO_AUTH_TIMEOUT_MS=10000
 EVO_AUTH_CACHE_TTL_MS=30000
+
+# Webhook signature validation — RUN_MODE=event-process (story 3.4 / EVO-1210)
+# Per-provider secrets for src/runners/event-process/validators/. When a secret
+# is unset that provider fails-closed (drops every event) and logs an
+# "unconfigured-secrets" warning at boot. SendGrid is opt-in: with no key it
+# passes through unverified. SES needs no secret (AWS SNS certificate-based).
+EVOLUTION_API_WEBHOOK_TOKEN=
+SPARKPOST_WEBHOOK_USER=
+SPARKPOST_WEBHOOK_PASSWORD=
+SENDGRID_WEBHOOK_VERIFICATION_KEY=
+MAILERSEND_WEBHOOK_SECRET=
+RESEND_WEBHOOK_SECRET=
+MANDRILL_WEBHOOK_SECRET=
+MANDRILL_WEBHOOK_URL=

package.json

@@ -88,6 +88,8 @@
     "reflect-metadata": "^0.2.2",
     "rxjs": "^7.8.1",
     "snappy": "^7.3.2",
+    "sns-validator": "^0.3.5",
+    "svix": "^1.95.2",
     "swagger-ui-express": "^5.0.1",
     "typeorm": "^0.3.21",
     "typeorm-extension": "^3.7.0",

src/runners/event-process/event-process.module.ts

@@ -1,17 +1,25 @@
 import { Module } from '@nestjs/common';
 import { EventProcessService } from './services/event-process.service';
 import { EventsReceivedConsumer } from './services/events-received.consumer';
+import { SignatureValidatorRegistry } from './services/signature-validator.registry';
+import { EventProcessMetrics } from './metrics/event-process-metrics';
 
 /**
  * Runner module for RUN_MODE=event-process (story 3.3 / EVO-1208).
  *
  * Boots the `events.received.<platform>` consumer end of the webhook pipeline.
  * IMESSAGE_BROKER (BrokerModule, @Global) and CorrelationContext
- * (CorrelationModule, @Global) come from their own modules, so this module only
- * declares the consumer + stub handler. Imported conditionally from
- * AppModule.forRoot() when AppFactory.shouldStartEventProcess() is true.
+ * (CorrelationModule, @Global) come from their own modules, and ConfigService
+ * is global (ConfigModule.forRoot isGlobal), so this module only declares the
+ * consumer, handler, signature-validator registry and metrics. Imported
+ * conditionally from AppModule.forRoot() when shouldStartEventProcess() is true.
  */
 @Module({
-  providers: [EventProcessService, EventsReceivedConsumer],
+  providers: [
+    EventProcessService,
+    EventsReceivedConsumer,
+    SignatureValidatorRegistry,
+    EventProcessMetrics,
+  ],
 })
 export class EventProcessModule {}

src/runners/event-process/metrics/event-process-metrics.ts

@@ -0,0 +1,28 @@
+import { Injectable } from '@nestjs/common';
+import { Counter, register } from 'prom-client';
+
+const SIGNATURE_INVALID_METRIC = 'evo_webhook_signature_invalid_total';
+
+/**
+ * Prometheus counters for the event-process webhook pipeline (story 3.4).
+ *
+ * The counter is fetched from the global registry if it already exists so that
+ * re-instantiating this provider (e.g. across test modules) does not throw the
+ * "metric already registered" error prom-client raises on duplicate names.
+ */
+@Injectable()
+export class EventProcessMetrics {
+  readonly signatureInvalid: Counter<string>;
+
+  constructor() {
+    this.signatureInvalid =
+      (register.getSingleMetric(SIGNATURE_INVALID_METRIC) as
+        | Counter<string>
+        | undefined) ??
+      new Counter({
+        name: SIGNATURE_INVALID_METRIC,
+        help: 'Webhook envelopes dropped because the signature was missing, invalid or unverifiable',
+        labelNames: ['platform', 'reason'],
+      });
+  }
+}

src/runners/event-process/services/event-process.service.spec.ts

@@ -1,20 +1,39 @@
 import { EventProcessService } from './event-process.service';
+import { SignatureValidatorRegistry } from './signature-validator.registry';
+import { EventProcessMetrics } from '../metrics/event-process-metrics';
 
 describe('EventProcessService', () => {
-  const service = new EventProcessService();
+  let service: EventProcessService;
+  let validate: jest.Mock;
+  let forPlatform: jest.Mock;
+  let inc: jest.Mock;
 
   const validEnvelope = {
     platform: 'evolution-api',
-    rawPayload: { hello: 'world' },
-    headers: { 'x-test': '1' },
+    rawPayload: 'raw-body-bytes',
+    headers: { apikey: 'tok' },
     receivedAt: '2026-06-08T12:00:00.000Z',
     sourceIp: '203.0.113.10',
     ingestionId: '00000000-0000-4000-8000-000000000000',
     correlationId: '11111111-1111-4111-8111-111111111111',
   };
 
-  it('resolves for a valid events.received envelope', async () => {
+  beforeEach(() => {
+    validate = jest.fn().mockReturnValue(true);
+    forPlatform = jest
+      .fn()
+      .mockReturnValue({ platform: 'evolution-api', validate });
+    inc = jest.fn();
+    service = new EventProcessService(
+      { for: forPlatform } as unknown as SignatureValidatorRegistry,
+      { signatureInvalid: { inc } } as unknown as EventProcessMetrics,
+    );
+  });
+
+  it('processes a valid envelope whose signature verifies (no drop metric)', async () => {
     await expect(service.handle(validEnvelope)).resolves.toBeUndefined();
+    expect(validate).toHaveBeenCalledWith('raw-body-bytes', { apikey: 'tok' });
+    expect(inc).not.toHaveBeenCalled();
   });
 
   it('throws for a payload that is not a valid envelope', async () => {
@@ -28,4 +47,40 @@ describe('EventProcessService', () => {
       service.handle({ ...validEnvelope, platform: 'not-a-platform' }),
     ).rejects.toThrow();
   });
+
+  it('drops (acks, no throw) and counts the metric when the signature is invalid (AC2)', async () => {
+    validate.mockReturnValue(false);
+
+    await expect(service.handle(validEnvelope)).resolves.toBeUndefined();
+
+    expect(inc).toHaveBeenCalledWith({
+      platform: 'evolution-api',
+      reason: 'invalid_signature',
+    });
+  });
+
+  it('drops with a warning when no validator is registered for the platform (AC3)', async () => {
+    forPlatform.mockReturnValue(null);
+
+    await expect(
+      service.handle({ ...validEnvelope, platform: 'unknown' }),
+    ).resolves.toBeUndefined();
+
+    expect(validate).not.toHaveBeenCalled();
+    expect(inc).toHaveBeenCalledWith({
+      platform: 'unknown',
+      reason: 'no_validator',
+    });
+  });
+
+  it('awaits an async validator (SES/SNS path)', async () => {
+    validate.mockResolvedValue(false);
+
+    await expect(service.handle(validEnvelope)).resolves.toBeUndefined();
+
+    expect(inc).toHaveBeenCalledWith({
+      platform: 'evolution-api',
+      reason: 'invalid_signature',
+    });
+  });
 });

src/runners/event-process/services/event-process.service.ts

@@ -5,6 +5,8 @@ import {
   isEventsReceivedContract,
 } from 'src/shared/broker/contracts/events-received.contract';
 import { TerminalError } from 'src/shared/errors/terminal-error';
+import { SignatureValidatorRegistry } from './signature-validator.registry';
+import { EventProcessMetrics } from '../metrics/event-process-metrics';
 
 /**
  * Thrown when a consumed message is not a valid `events.received` envelope.
@@ -26,6 +28,11 @@ export class InvalidEnvelopeError extends TerminalError {}
 export class EventProcessService {
   private readonly logger = new CustomLoggerService(EventProcessService.name);
 
+  constructor(
+    private readonly validators: SignatureValidatorRegistry,
+    private readonly metrics: EventProcessMetrics,
+  ) {}
+
   async handle(envelope: unknown): Promise<void> {
     if (!isEventsReceivedContract(envelope)) {
       throw new InvalidEnvelopeError(
@@ -34,6 +41,8 @@ export class EventProcessService {
     }
     const valid: EventsReceivedContract = envelope;
 
+    if (!(await this.hasValidSignature(valid))) return;
+
     this.logger.log('event-process.handle', {
       action: 'event-process.handle',
       platform: valid.platform,
@@ -46,4 +55,61 @@ export class EventProcessService {
 
     return Promise.resolve();
   }
+
+  /**
+   * Resolves the provider's signature validator and runs it. Returns false —
+   * meaning "drop, but ack" — when no validator is registered for the platform
+   * (e.g. `unknown`) or the signature does not verify. Dropping is a plain
+   * `return` upstream, never a throw, so the broker acks and never redelivers a
+   * payload we have already rejected.
+   */
+  private async hasValidSignature(
+    valid: EventsReceivedContract,
+  ): Promise<boolean> {
+    const validator = this.validators.for(valid.platform);
+    if (!validator) {
+      this.logger.warn('event-process.signature.no-validator', {
+        action: 'event-process.signature.no-validator',
+        platform: valid.platform,
+        correlationId: valid.correlationId,
+        ingestionId: valid.ingestionId,
+      });
+      this.metrics.signatureInvalid.inc({
+        platform: valid.platform,
+        reason: 'no_validator',
+      });
+      return false;
+    }
+
+    // HMAC validators need the exact bytes the provider signed. The receiver
+    // (story 3.1) preserves rawPayload as the raw UTF-8 string; a non-string
+    // here means an upstream change broke that invariant and HMAC checks would
+    // silently fail, so surface it loudly.
+    if (typeof valid.rawPayload !== 'string') {
+      this.logger.warn('event-process.signature.non-string-payload', {
+        action: 'event-process.signature.non-string-payload',
+        platform: valid.platform,
+        correlationId: valid.correlationId,
+        ingestionId: valid.ingestionId,
+      });
+    }
+    const rawPayload =
+      typeof valid.rawPayload === 'string'
+        ? valid.rawPayload
+        : JSON.stringify(valid.rawPayload ?? '');
+
+    if (await validator.validate(rawPayload, valid.headers)) return true;
+
+    this.logger.warn('event-process.signature.invalid', {
+      action: 'event-process.signature.invalid',
+      platform: valid.platform,
+      correlationId: valid.correlationId,
+      ingestionId: valid.ingestionId,
+    });
+    this.metrics.signatureInvalid.inc({
+      platform: valid.platform,
+      reason: 'invalid_signature',
+    });
+    return false;
+  }
 }

src/runners/event-process/services/signature-validator.registry.spec.ts

@@ -0,0 +1,31 @@
+import { ConfigService } from '@nestjs/config';
+import { SignatureValidatorRegistry } from './signature-validator.registry';
+
+describe('SignatureValidatorRegistry', () => {
+  const config = {
+    get: jest.fn().mockReturnValue('configured-secret'),
+  } as unknown as ConfigService;
+  const registry = new SignatureValidatorRegistry(config);
+
+  it.each([
+    'evolution-api',
+    'sparkpost',
+    'sendgrid',
+    'mailersend',
+    'resend',
+    'ses',
+    'mandrill',
+  ])('resolves a validator for %s', (platform) => {
+    const validator = registry.for(platform);
+    expect(validator).not.toBeNull();
+    expect(validator?.platform).toBe(platform);
+  });
+
+  it('returns null for the unknown platform', () => {
+    expect(registry.for('unknown')).toBeNull();
+  });
+
+  it('returns null for an unregistered platform', () => {
+    expect(registry.for('not-a-platform')).toBeNull();
+  });
+});