fix(email): keep the webhook payload out of the logs, and bound the response read (#1225) (#1233)

Round 4 of external review, on the merged commit. Both findings are
consequences of the round-3 streaming change, which is exactly why the round
was worth running.

An AxiosError carries the request it failed on: `config.data` is the ENTIRE
serialised message, base64 attachments included, and `config.headers` holds
the signature. emailProcessor logs the error object and winston serialises it,
so a DNS blip or a refused connection wrote password-reset links, guest
recovery codes and multi-megabyte invoices into combined.log — verified
against axios rather than assumed. Every rejection is now caught and replaced
with a message-and-code-only error, so nothing downstream can serialise the
request back out of it.

readBounded had no deadline. axios' `timeout` covers the response HEADERS, and
with responseType 'stream' it has already resolved by the time the body is
read — so a receiver that answered 2xx and never closed its body left the
await hanging, the queue row stayed pending, and the next processor pass sent
the same message again. An unclosed stream was duplicate email. There is now a
10s wall clock that destroys the stream, with the timer unref'd so a hung read
cannot hold the process open at exit.

23 transport tests (2 new, both failing without these fixes), 63 across the
email suites, eslint clean.

Co-authored-by: Paul Nothaft <[email protected]>

backend/__tests__/services/emailWebhookTransport.test.js

@@ -48,6 +48,7 @@ beforeEach(() => {
   process.env.EMAIL_WEBHOOK_SECRET = SECRET;
   transport.__testing.setAllowPrivateUrls(false);
   transport.__testing.resetSecretWarning();
+  transport.__testing.setReadTimeout(10000);
   validateExternalUrlAsync.mockResolvedValue({ valid: true, reason: 'ok' });
   axios.post.mockResolvedValue({ status: 200, data: streamOf('') });
 });
@@ -257,6 +258,45 @@ describe('delivery result', () => {
     expect(result.messageId).toEqual(expect.any(String));
   });
 
+  it('never lets the request body escape inside a network error', async () => {
+    // An AxiosError carries config.data — the whole serialised message, base64
+    // attachments included — and config.headers holds the signature. Callers
+    // log the error object and winston serialises it, so propagating the raw
+    // error writes password-reset links and invoices into combined.log.
+    const axiosError = Object.assign(new Error('connect ECONNREFUSED'), {
+      code: 'ECONNREFUSED',
+      config: {
+        data: JSON.stringify({ html: 'RESET-LINK-SECRET' }),
+        headers: { 'X-PicPeak-Signature': 'the-signature' },
+      },
+    });
+    axios.post.mockRejectedValue(axiosError);
+
+    const caught = await transport.send(MAIL).catch((e) => e);
+    expect(caught).toBeInstanceOf(Error);
+    expect(caught.config).toBeUndefined();
+    const serialised = JSON.stringify({ msg: caught.message, ...caught });
+    expect(serialised).not.toContain('RESET-LINK-SECRET');
+    expect(serialised).not.toContain('the-signature');
+    // The useful part still survives for diagnosis.
+    expect(caught.message).toContain('ECONNREFUSED');
+  });
+
+  it('gives up on a response stream that never closes, instead of hanging', async () => {
+    // axios' timeout covers the headers; with responseType 'stream' it has
+    // already resolved. Without a deadline this await hangs, the queue row
+    // stays pending, and the next pass sends the same email again.
+    const { PassThrough } = require('stream');
+    const neverEnds = new PassThrough(); // written to by nobody, never ended
+    axios.post.mockResolvedValue({ status: 200, data: neverEnds });
+    transport.__testing.setReadTimeout(25);
+
+    const result = await transport.send(MAIL);
+    expect(result.messageId).toEqual(expect.any(String));
+    // The stream is torn down rather than left dangling.
+    expect(neverEnds.destroyed).toBe(true);
+  });
+
   it('prefers a messageId the receiver reports', async () => {
     axios.post.mockResolvedValue({ status: 200, data: streamOf(JSON.stringify({ messageId: 'from-n8n-123' })) });
     expect((await transport.send(MAIL)).messageId).toBe('from-n8n-123');

backend/src/services/emailWebhookTransport.js

@@ -29,6 +29,10 @@ const { validateExternalUrlAsync } = require('../utils/networkValidation');
 
 const SIGNATURE_HEADER = 'X-PicPeak-Signature';
 const HTTP_TIMEOUT_MS = 15000;
+// Wall clock for reading the response body. Separate from HTTP_TIMEOUT_MS,
+// which axios applies to the headers only. Mutable so a test can shrink it
+// rather than actually waiting ten seconds.
+let READ_TIMEOUT_MS = 10000;
 
 // Attachments are base64 in the JSON body, which inflates them by a third.
 // Invoices and quotes are the real users of this and run to a few hundred KB;
@@ -137,14 +141,6 @@ async function encodeAttachments(attachments) {
   return encoded;
 }
 
-/**
- * Recipients as a flat list of single addresses.
- *
- * Splits inside array elements too, not just bare strings: sendRawEmail wraps a
- * string cc in an array before it reaches here, so "a@x, b@y" arrives as ONE
- * element. Passing that through would put a combined address in the payload,
- * which a relay treating each element as one mailbox rejects or misaddresses.
- */
 /**
  * Read at most MAX_RESPONSE_BYTES from a response stream, then stop.
  *
@@ -160,7 +156,27 @@ function readBounded(stream) {
   return new Promise((resolve) => {
     const chunks = [];
     let size = 0;
-    const finish = () => resolve(Buffer.concat(chunks).toString('utf8'));
+    let done = false;
+    // A deadline, not just a size cap. axios' `timeout` covers the response
+    // HEADERS; with responseType 'stream' it has already resolved by the time
+    // we get here, so a receiver that answers 2xx and then never closes its
+    // body — or trickles bytes forever — would leave this await hanging. The
+    // queue row would stay pending and the next processor pass would POST the
+    // same message again, so an unclosed stream becomes duplicate email.
+    const timer = setTimeout(() => {
+      stream.destroy();
+      finish();
+    }, READ_TIMEOUT_MS);
+    // Node's unref keeps a hung read from holding the process open at exit.
+    if (typeof timer.unref === 'function') timer.unref();
+
+    function finish() {
+      if (done) return;
+      done = true;
+      clearTimeout(timer);
+      resolve(Buffer.concat(chunks).toString('utf8'));
+    }
+
     stream.on('data', (chunk) => {
       size += chunk.length;
       if (size <= MAX_RESPONSE_BYTES) {
@@ -171,11 +187,19 @@ function readBounded(stream) {
       }
     });
     stream.on('end', finish);
-    stream.on('error', () => resolve(''));
+    stream.on('error', finish);
     stream.on('close', finish);
   });
 }
 
+/**
+ * Recipients as a flat list of single addresses.
+ *
+ * Splits inside array elements too, not just bare strings: sendRawEmail wraps a
+ * string cc in an array before it reaches here, so "a@x, b@y" arrives as ONE
+ * element. Passing that through would put a combined address in the payload,
+ * which a relay treating each element as one mailbox rejects or misaddresses.
+ */
 function normalizeRecipients(value) {
   if (!value) return [];
   const parts = Array.isArray(value) ? value : [value];
@@ -238,31 +262,47 @@ async function send(mail) {
   const rawBody = JSON.stringify(payload);
   const signature = signPayload(secret, rawBody);
 
-  const response = await axios.post(url, rawBody, {
+  // Every axios rejection is caught and replaced. An AxiosError carries the
-    headers: {
+  // request it failed on — `config.data` is the ENTIRE serialised message,
-      'Content-Type': 'application/json',
+  // base64 attachments included, and `config.headers` holds the signature.
-      [SIGNATURE_HEADER]: signature,
+  // Callers log the error object (emailProcessor's `logger.error('Error
-    },
+  // sending template email:', error)`), and winston serialises it, so a DNS
-    timeout: HTTP_TIMEOUT_MS,
+  // blip or a connection refusal would write password-reset links, recovery
-    // Resolve on any status so a 4xx/5xx becomes our error message rather than
+  // codes and multi-megabyte invoices into combined.log — the log file being
-    // axios's, which does not say which webhook failed.
+  // exactly where none of that belongs. Only the message survives.
-    validateStatus: () => true,
+  let response;
-    maxRedirects: 0,
+  try {
-    // Streamed, NOT buffered with maxContentLength. axios enforces that limit
+    response = await axios.post(url, rawBody, {
-    // while reading, so a receiver that delivered the mail and then echoed a
+      headers: {
-    // large body would make this throw AFTER a successful delivery — the queue
+        'Content-Type': 'application/json',
-    // would retry and the recipient would get the same email again. Reading it
+        [SIGNATURE_HEADER]: signature,
-    // ourselves means an oversized response costs us the messageId, never a
+      },
-    // duplicate send.
+      timeout: HTTP_TIMEOUT_MS,
-    responseType: 'stream',
+      // Resolve on any status so a 4xx/5xx becomes our error message rather than
-    // Byte length, not String#length. axios enforces this against the UTF-8
+      // axios's, which does not say which webhook failed.
-    // buffer it sends, while rawBody.length counts UTF-16 code units — every
+      validateStatus: () => true,
-    // umlaut is 2 bytes and every CJK character 3, so a German or Japanese
+      maxRedirects: 0,
-    // message would blow past a code-unit budget and axios would reject it
+      // Streamed, NOT buffered with maxContentLength. axios enforces that limit
-    // before posting. With base64 attachments in the body the gap is easily
+      // while reading, so a receiver that delivered the mail and then echoed a
-    // more than the slack.
+      // large body would make this throw AFTER a successful delivery — the queue
-    maxBodyLength: Buffer.byteLength(rawBody, 'utf8') + 1024,
+      // would retry and the recipient would get the same email again. Reading it
-  });
+      // ourselves means an oversized response costs us the messageId, never a
+      // duplicate send.
+      responseType: 'stream',
+      // Byte length, not String#length. axios enforces this against the UTF-8
+      // buffer it sends, while rawBody.length counts UTF-16 code units — every
+      // umlaut is 2 bytes and every CJK character 3, so a German or Japanese
+      // message would blow past a code-unit budget and axios would reject it
+      // before posting. With base64 attachments in the body the gap is easily
+      // more than the slack.
+      maxBodyLength: Buffer.byteLength(rawBody, 'utf8') + 1024,
+    });
+  } catch (err) {
+    // Message and code only — deliberately NOT the error object, so nothing
+    // downstream can serialise the request back out of it.
+    const detail = err && err.code ? `${err.code}: ${err.message}` : (err && err.message) || 'request failed';
+    throw new Error(`email webhook request failed (${detail})`);
+  }
 
   // Status first: it is the delivery verdict, and it is known before a single
   // byte of the body is read.
@@ -290,6 +330,7 @@ module.exports = {
     MAX_ATTACHMENT_BYTES,
     encodeAttachments,
     setAllowPrivateUrls(value) { allowPrivateUrls = !!value; },
+    setReadTimeout(ms) { READ_TIMEOUT_MS = ms; },
     resetSecretWarning() { warnedAboutMissingSecret = false; },
   },
 };