fix(webhooks): harden dispatcher SSRF pin and response limits

AI-Co-Authored-By: Grok

src/db.js

@@ -1224,12 +1224,16 @@ export function completeWebhookDeliverySuccess(id, { responseStatus = null, body
   return publicWebhookDelivery(requireDb().prepare('SELECT * FROM webhook_deliveries WHERE id = ?').get(id));
 }
 
-export function completeWebhookDeliveryFailure(id, { responseStatus = null, bodyPreview = '', error = '' } = {}) {
+export function completeWebhookDeliveryFailure(
+  id,
+  { responseStatus = null, bodyPreview = '', error = '', permanent = false } = {}
+) {
   const row = requireDb().prepare('SELECT * FROM webhook_deliveries WHERE id = ?').get(id);
   if (!row) return null;
   const attemptCount = Number(row.attempt_count || 0) + 1;
   const completedAt = now();
-  const exhausted = attemptCount >= MAX_WEBHOOK_ATTEMPTS;
+  // Permanent failures (SSRF blocked, invalid URL, missing secret) skip backoff and die immediately.
+  const exhausted = permanent || attemptCount >= MAX_WEBHOOK_ATTEMPTS;
   const nextStatus = exhausted ? 'dead' : 'pending';
   const nextAttemptAt = exhausted
     ? completedAt

src/webhook-dispatcher.js

@@ -1,4 +1,6 @@
 import dns from 'node:dns';
+import http from 'node:http';
+import https from 'node:https';
 import net from 'node:net';
 import {
   claimWebhookDeliveries,
@@ -11,6 +13,7 @@ import { eventTypeForStatus, signWebhookBody } from './webhook-model.js';
 const DEFAULT_INTERVAL_MS = 10_000;
 const DEFAULT_BATCH_SIZE = 3;
 const FETCH_TIMEOUT_MS = 10_000;
+const MAX_RESPONSE_BODY_BYTES = 4096;
 const USER_AGENT = 'MailHub-Webhook/1.0';
 
 const blockedAddresses = new net.BlockList();
@@ -85,9 +88,12 @@ export function isLoopbackIpAddress(address) {
 
 /**
  * Validate webhook URL scheme and resolved addresses (fail closed).
- * @returns {Promise<URL>}
+ * Resolves DNS once and returns every allowed address so callers can pin the TCP connection
+ * (avoids TOCTOU / DNS rebinding between validation and fetch).
+ *
+ * @returns {Promise<{ url: URL, addresses: string[], pinnedAddress: string }>}
  */
-export async function assertSafeWebhookUrl(
+export async function resolveSafeWebhookTarget(
   rawUrl,
   {
     allowHttpLocal = String(process.env.WEBHOOK_ALLOW_HTTP_LOCAL || '') === '1',
@@ -128,7 +134,7 @@ export async function assertSafeWebhookUrl(
     if (isBlockedIpAddress(hostname) && !allowLoopback) {
       throw new Error(`Webhook URL resolves to a blocked address (${hostname})`);
     }
-    return parsed;
+    return { url: parsed, addresses: [hostname], pinnedAddress: hostname };
   }
 
   let records;
@@ -143,19 +149,156 @@ export async function assertSafeWebhookUrl(
     throw new Error('Webhook DNS lookup returned no addresses');
   }
 
+  const addresses = [];
   for (const record of list) {
     const address = typeof record === 'string' ? record : record.address;
+    if (!address) continue;
     const allowLoopback = allowHttpLocal && loopbackHost && isLoopbackIpAddress(address);
     if (isBlockedIpAddress(address) && !allowLoopback) {
       throw new Error(`Webhook URL resolves to a blocked address (${address})`);
     }
+    addresses.push(address);
+  }
+
+  if (addresses.length === 0) {
+    throw new Error('Webhook DNS lookup returned no addresses');
   }
 
-  return parsed;
+  // All addresses are public (or allowed loopback); pin the first to avoid a second DNS lookup.
+  return { url: parsed, addresses, pinnedAddress: addresses[0] };
+}
+
+/**
+ * Validate webhook URL scheme and resolved addresses (fail closed).
+ * @returns {Promise<URL>}
+ */
+export async function assertSafeWebhookUrl(
+  rawUrl,
+  {
+    allowHttpLocal = String(process.env.WEBHOOK_ALLOW_HTTP_LOCAL || '') === '1',
+    dnsLookup = defaultDnsLookup
+  } = {}
+) {
+  const target = await resolveSafeWebhookTarget(rawUrl, { allowHttpLocal, dnsLookup });
+  return target.url;
+}
+
+/**
+ * Build a URL that connects to a pinned IP while preserving path/query/port/protocol.
+ * Callers must set Host + TLS servername to the original hostname.
+ */
+export function buildPinnedWebhookUrl(parsedUrl, pinnedAddress) {
+  const pinned = new URL(String(parsedUrl));
+  pinned.hostname = pinnedAddress;
+  return pinned;
+}
+
+/**
+ * Default transport: connect to the pinned IP with original Host / TLS SNI.
+ * Does not re-resolve DNS (prevents rebinding between assert and connect).
+ */
+export function pinnedWebhookFetch(requestUrl, options = {}) {
+  const parsed = new URL(String(requestUrl));
+  const isHttps = parsed.protocol === 'https:';
+  const transport = isHttps ? https : http;
+  const connectHost = parsed.hostname.replace(/^\[|\]$/g, '');
+  const port = parsed.port ? Number(parsed.port) : isHttps ? 443 : 80;
+  const path = `${parsed.pathname || '/'}${parsed.search || ''}`;
+  const headers = { ...(options.headers || {}) };
+  const servername = options.servername || headers.Host || headers.host || connectHost;
+  // Ensure Host header reflects the original hostname when provided via servername/options.
+  if (!headers.Host && !headers.host) {
+    headers.Host = servername;
+  }
+
+  const signal = options.signal;
+  const body = options.body;
+
+  return new Promise((resolve, reject) => {
+    let settled = false;
+    const fail = (error) => {
+      if (settled) return;
+      settled = true;
+      cleanup();
+      reject(error);
+    };
+    const succeed = (value) => {
+      if (settled) return;
+      settled = true;
+      cleanup();
+      resolve(value);
+    };
+
+    const req = transport.request(
+      {
+        protocol: parsed.protocol,
+        hostname: connectHost,
+        port,
+        path,
+        method: options.method || 'GET',
+        headers,
+        servername: isHttps ? String(servername).replace(/:\d+$/, '').replace(/^\[|\]$/g, '') : undefined,
+        timeout: FETCH_TIMEOUT_MS
+      },
+      (res) => {
+        succeed({
+          status: res.statusCode || 0,
+          ok: (res.statusCode || 0) >= 200 && (res.statusCode || 0) < 300,
+          headers: res.headers,
+          body: res,
+          async text() {
+            return readLimitedStream(res, MAX_RESPONSE_BODY_BYTES);
+          }
+        });
+      }
+    );
+
+    const onAbort = () => {
+      const error = new Error('Webhook request timed out');
+      error.name = signal?.reason?.name === 'TimeoutError' ? 'TimeoutError' : 'AbortError';
+      req.destroy(error);
+      fail(error);
+    };
+
+    function cleanup() {
+      if (signal) {
+        signal.removeEventListener?.('abort', onAbort);
+      }
+      req.removeAllListeners('timeout');
+      req.removeAllListeners('error');
+    }
+
+    if (signal) {
+      if (signal.aborted) {
+        onAbort();
+        return;
+      }
+      signal.addEventListener('abort', onAbort, { once: true });
+    }
+
+    req.on('timeout', () => {
+      const error = new Error('Webhook request timed out');
+      error.name = 'TimeoutError';
+      req.destroy(error);
+      fail(error);
+    });
+    req.on('error', (error) => {
+      if (error?.name === 'TimeoutError' || error?.name === 'AbortError') {
+        fail(error);
+        return;
+      }
+      fail(error);
+    });
+
+    if (body != null && body !== '') {
+      req.write(body);
+    }
+    req.end();
+  });
 }
 
 export async function deliverOne(item, {
-  fetchImpl = globalThis.fetch.bind(globalThis),
+  fetchImpl = pinnedWebhookFetch,
   completeSuccess = completeWebhookDeliverySuccess,
   completeFailure = completeWebhookDeliveryFailure,
   dnsLookup = defaultDnsLookup,
@@ -176,15 +319,18 @@ export async function deliverOne(item, {
   const secret = webhook?.secret;
   if (!url || !secret) {
     return completeFailure(deliveryId, {
-      error: 'Webhook target missing url or secret'
+      error: 'Webhook target missing url or secret',
+      permanent: true
     });
   }
 
+  let target;
   try {
-    await assertSafeWebhookUrl(url, { allowHttpLocal, dnsLookup });
+    target = await resolveSafeWebhookTarget(url, { allowHttpLocal, dnsLookup });
   } catch (error) {
     return completeFailure(deliveryId, {
-      error: error.message || 'Webhook URL blocked'
+      error: error.message || 'Webhook URL blocked',
+      permanent: true
     });
   }
 
@@ -198,21 +344,28 @@ export async function deliverOne(item, {
   }
 
   const signature = signWebhookBody(rawBody, secret, nowSeconds());
+  const originalHost = target.url.host;
+  const originalHostname = target.url.hostname.replace(/^\[|\]$/g, '');
+  const pinnedUrl = buildPinnedWebhookUrl(target.url, target.pinnedAddress);
   const headers = {
     'Content-Type': 'application/json',
     'User-Agent': USER_AGENT,
+    Host: originalHost,
     'X-MailHub-Signature': signature,
     'X-MailHub-Event': eventHeader,
     'X-MailHub-Delivery': `whd_${deliveryId}`
   };
 
   try {
-    const response = await fetchImpl(url, {
+    const response = await fetchImpl(pinnedUrl.href, {
       method: 'POST',
       headers,
       body: rawBody,
       redirect: 'manual',
-      signal: AbortSignal.timeout(FETCH_TIMEOUT_MS)
+      signal: AbortSignal.timeout(FETCH_TIMEOUT_MS),
+      // Used by pinnedWebhookFetch; ignored by plain fetch mocks.
+      servername: originalHostname,
+      pinnedAddress: target.pinnedAddress
     });
     const status = Number(response?.status) || 0;
     const bodyPreview = await readBodyPreview(response);
@@ -237,7 +390,7 @@ export async function deliverOne(item, {
 }
 
 export async function processWebhookBatch({
-  fetchImpl = globalThis.fetch.bind(globalThis),
+  fetchImpl = pinnedWebhookFetch,
   batchSize = DEFAULT_BATCH_SIZE,
   claim = claimWebhookDeliveries,
   completeSuccess = completeWebhookDeliverySuccess,
@@ -296,7 +449,7 @@ export function startWebhookWorker({
   enabled = true,
   intervalMs = DEFAULT_INTERVAL_MS,
   batchSize = DEFAULT_BATCH_SIZE,
-  fetchImpl = globalThis.fetch.bind(globalThis),
+  fetchImpl = pinnedWebhookFetch,
   logger = console
 } = {}) {
   if (!enabled) return null;
@@ -350,14 +503,117 @@ async function defaultDnsLookup(hostname, options) {
   return dns.promises.lookup(hostname, options);
 }
 
+/**
+ * Read at most MAX_RESPONSE_BODY_BYTES from the response, then abort the rest.
+ * Prefer body streams so large payloads never buffer fully into memory.
+ */
 async function readBodyPreview(response) {
-  if (!response || typeof response.text !== 'function') return '';
+  if (!response) return '';
   try {
-    const text = await response.text();
-    return String(text || '').slice(0, 2048);
+    if (response.body && typeof response.body.getReader === 'function') {
+      return await readLimitedWebStream(response.body, MAX_RESPONSE_BODY_BYTES);
+    }
+    if (response.body && typeof response.body.on === 'function') {
+      return await readLimitedStream(response.body, MAX_RESPONSE_BODY_BYTES);
+    }
+    if (typeof response.text === 'function') {
+      const text = await response.text();
+      return String(text || '').slice(0, MAX_RESPONSE_BODY_BYTES);
+    }
   } catch {
     return '';
   }
+  return '';
+}
+
+async function readLimitedWebStream(stream, maxBytes) {
+  const reader = stream.getReader();
+  const chunks = [];
+  let total = 0;
+  try {
+    while (total < maxBytes) {
+      const { done, value } = await reader.read();
+      if (done) break;
+      if (!value) continue;
+      const chunk = Buffer.from(value);
+      chunks.push(chunk);
+      total += chunk.byteLength;
+      if (total >= maxBytes) break;
+    }
+  } finally {
+    try {
+      await reader.cancel();
+    } catch {
+      // ignore cancel errors
+    }
+  }
+  if (chunks.length === 0) return '';
+  return Buffer.concat(chunks).subarray(0, maxBytes).toString('utf8');
+}
+
+function readLimitedStream(stream, maxBytes) {
+  return new Promise((resolve) => {
+    if (!stream || typeof stream.on !== 'function') {
+      resolve('');
+      return;
+    }
+
+    const chunks = [];
+    let total = 0;
+    let settled = false;
+
+    const finish = () => {
+      if (settled) return;
+      settled = true;
+      stream.removeListener?.('data', onData);
+      stream.removeListener?.('end', onEnd);
+      stream.removeListener?.('error', onEnd);
+      stream.removeListener?.('close', onEnd);
+      if (typeof stream.destroy === 'function' && !stream.destroyed) {
+        try {
+          stream.destroy();
+        } catch {
+          // ignore
+        }
+      }
+      if (chunks.length === 0) {
+        resolve('');
+        return;
+      }
+      resolve(Buffer.concat(chunks).subarray(0, maxBytes).toString('utf8'));
+    };
+
+    const onData = (chunk) => {
+      if (settled) return;
+      const buf = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
+      const remaining = maxBytes - total;
+      if (remaining <= 0) {
+        finish();
+        return;
+      }
+      if (buf.byteLength > remaining) {
+        chunks.push(buf.subarray(0, remaining));
+        total += remaining;
+        finish();
+        return;
+      }
+      chunks.push(buf);
+      total += buf.byteLength;
+      if (total >= maxBytes) finish();
+    };
+    const onEnd = () => finish();
+
+    // Already flowing / ended
+    if (stream.readableEnded || stream.destroyed) {
+      finish();
+      return;
+    }
+
+    stream.on('data', onData);
+    stream.on('end', onEnd);
+    stream.on('error', onEnd);
+    stream.on('close', onEnd);
+  });
 }
 
 function safePositiveInt(value, fallback) {

test/webhook-dispatcher.test.js

@@ -2,6 +2,7 @@ import assert from 'node:assert/strict';
 import { mkdtempSync } from 'node:fs';
 import { tmpdir } from 'node:os';
 import path from 'node:path';
+import { Readable } from 'node:stream';
 import { test } from 'node:test';
 
 import {
@@ -14,8 +15,10 @@ import {
 } from '../src/db.js';
 import {
   assertSafeWebhookUrl,
+  buildPinnedWebhookUrl,
   isBlockedIpAddress,
   processWebhookBatch,
+  resolveSafeWebhookTarget,
   startWebhookWorker,
   stopWebhookWorker
 } from '../src/webhook-dispatcher.js';
@@ -62,7 +65,34 @@ test('assertSafeWebhookUrl requires https and blocks private DNS results', async
   assert.equal(local.hostname, '127.0.0.1');
 });
 
-test('posts signed body and marks success on 2xx', async () => {
+test('resolveSafeWebhookTarget returns pinned public address and rejects mixed private results', async () => {
+  const target = await resolveSafeWebhookTarget('https://hooks.example.com/mail?x=1', {
+    dnsLookup: async () => [
+      { address: '1.1.1.1', family: 4 },
+      { address: '8.8.8.8', family: 4 }
+    ]
+  });
+  assert.equal(target.url.hostname, 'hooks.example.com');
+  assert.deepEqual(target.addresses, ['1.1.1.1', '8.8.8.8']);
+  assert.equal(target.pinnedAddress, '1.1.1.1');
+  assert.equal(
+    buildPinnedWebhookUrl(target.url, target.pinnedAddress).href,
+    'https://1.1.1.1/mail?x=1'
+  );
+
+  await assert.rejects(
+    () =>
+      resolveSafeWebhookTarget('https://hooks.example.com/hook', {
+        dnsLookup: async () => [
+          { address: '1.1.1.1', family: 4 },
+          { address: '10.0.0.1', family: 4 }
+        ]
+      }),
+    /blocked/i
+  );
+});
+
+test('posts signed body to pinned IP with Host/SNI and marks success on 2xx', async () => {
   initDatabase(tempDataDir(), 'test-secret');
   const alice = createUser({ username: 'alice', email: 'alice@example.com', password: 'password123' });
   const webhook = createWebhook(alice.id, {
@@ -102,11 +132,15 @@ test('posts signed body and marks success on 2xx', async () => {
 
   assert.equal(result.claimed, 1);
   assert.equal(fetchCalls.length, 1);
-  assert.equal(fetchCalls[0].url, 'https://hooks.example.com/mail');
+  // Connect by pinned IP (no second DNS); Host/SNI keep original hostname.
+  assert.equal(fetchCalls[0].url, 'https://1.1.1.1/mail');
   assert.equal(fetchCalls[0].options.method, 'POST');
   assert.equal(fetchCalls[0].options.redirect, 'manual');
   assert.equal(fetchCalls[0].options.headers['Content-Type'], 'application/json');
   assert.equal(fetchCalls[0].options.headers['User-Agent'], 'MailHub-Webhook/1.0');
+  assert.equal(fetchCalls[0].options.headers.Host, 'hooks.example.com');
+  assert.equal(fetchCalls[0].options.servername, 'hooks.example.com');
+  assert.equal(fetchCalls[0].options.pinnedAddress, '1.1.1.1');
   assert.equal(fetchCalls[0].options.headers['X-MailHub-Event'], 'email.sent');
   assert.equal(fetchCalls[0].options.headers['X-MailHub-Delivery'], `whd_${before.id}`);
   assert.equal(
@@ -207,7 +241,7 @@ test('marks dead after max attempts', async () => {
   assert.equal(dead.attemptCount, MAX_WEBHOOK_ATTEMPTS);
 });
 
-test('rejects private IP targets without calling fetch', async () => {
+test('rejects private IP targets as permanent dead without calling fetch', async () => {
   initDatabase(tempDataDir(), 'test-secret');
   const alice = createUser({ username: 'alice', email: 'alice@example.com', password: 'password123' });
   createWebhook(alice.id, {
@@ -238,12 +272,98 @@ test('rejects private IP targets without calling fetch', async () => {
 
   assert.equal(fetchCalled, false);
   const [delivery] = listWebhookDeliveries(alice.id);
-  assert.equal(delivery.status, 'pending');
+  assert.equal(delivery.status, 'dead');
   assert.equal(delivery.attemptCount, 1);
   assert.match(delivery.error, /blocked/i);
   assert.equal(claimWebhookDeliveries(5).length, 0);
 });
 
+test('missing webhook secret marks delivery permanently dead', async () => {
+  initDatabase(tempDataDir(), 'test-secret');
+  const alice = createUser({ username: 'alice', email: 'alice@example.com', password: 'password123' });
+  const webhook = createWebhook(alice.id, {
+    name: 'NoSecret',
+    url: 'https://hooks.example.com/no-secret',
+    events: ['sent']
+  });
+  enqueueWebhookDeliveries({
+    id: 15,
+    userId: alice.id,
+    domainId: null,
+    status: 'sent',
+    sender: 'noreply@example.com',
+    recipients: ['user@example.com'],
+    subject: 'Secret',
+    detail: '',
+    queueId: 'Q15'
+  });
+
+  const delivery = listWebhookDeliveries(alice.id)[0];
+  await processWebhookBatch({
+    claim: () => [
+      {
+        delivery: { ...delivery, status: 'processing' },
+        webhook: {
+          id: webhook.id,
+          url: webhook.url,
+          secret: ''
+        }
+      }
+    ],
+    reap: () => 0,
+    fetchImpl: async () => {
+      throw new Error('should not fetch');
+    }
+  });
+
+  const [after] = listWebhookDeliveries(alice.id);
+  assert.equal(after.status, 'dead');
+  assert.equal(after.attemptCount, 1);
+  assert.match(after.error, /missing url or secret/i);
+});
+
+test('bounds response body preview without consuming unbounded text()', async () => {
+  initDatabase(tempDataDir(), 'test-secret');
+  const alice = createUser({ username: 'alice', email: 'alice@example.com', password: 'password123' });
+  createWebhook(alice.id, {
+    name: 'Body',
+    url: 'https://hooks.example.com/body',
+    events: ['sent']
+  });
+  enqueueWebhookDeliveries({
+    id: 16,
+    userId: alice.id,
+    domainId: null,
+    status: 'sent',
+    sender: 'noreply@example.com',
+    recipients: ['user@example.com'],
+    subject: 'Body',
+    detail: '',
+    queueId: 'Q16'
+  });
+
+  const huge = 'x'.repeat(20_000);
+  let textCalls = 0;
+  await processWebhookBatch({
+    dnsLookup: async () => [{ address: '1.1.1.1', family: 4 }],
+    fetchImpl: async () => ({
+      status: 200,
+      body: Readable.from([Buffer.from(huge)]),
+      text: async () => {
+        textCalls += 1;
+        return huge;
+      }
+    })
+  });
+
+  assert.equal(textCalls, 0);
+  const [delivery] = listWebhookDeliveries(alice.id);
+  assert.equal(delivery.status, 'success');
+  assert.ok(delivery.responseBodyPreview.length <= 2048);
+  assert.ok(delivery.responseBodyPreview.length > 0);
+  assert.match(delivery.responseBodyPreview, /^x+$/);
+});
+
 test('startWebhookWorker can be skipped and stopped', async () => {
   assert.equal(startWebhookWorker({ enabled: false }), null);
   const handle = startWebhookWorker({