Harden smtpwebhook delivery and service runtime

2026-09-02 12:20:04 +02:00
parent f8c0c4f70c
commit d451da5109
3 changed files with 637 additions and 102 deletions
@@ -1,3 +1,6 @@
+const crypto = require('crypto');
+const fs = require('fs');
+const path = require('path');
 const SMTPServer = require('smtp-server').SMTPServer;
 const config = require('./config');
 const { parseEmail } = require('./services/emailParser');
@@ -14,9 +17,10 @@ const logger = winston.createLogger({
    winston.format.json()
  ),
  transports: [
-    new winston.transports.Console(),
+    // Supervisor captures stdout/stderr. A regular file avoids EPIPE loops
+    // when Supervisor rotates its child logs.
    new winston.transports.DailyRotateFile({
-      filename: 'logs/application-%DATE%.log',
+      filename: path.join(config.LOG_DIR, 'application-%DATE%.log'),
      datePattern: 'YYYY-MM-DD',
      zippedArchive: true,
      maxSize: '20m',
@@ -26,102 +30,343 @@ const logger = winston.createLogger({
 });

 function validateConfig() {
-  const requiredKeys = ['PORT', 'SMTP_SECURE', 'WEBHOOK_URL', 'WEBHOOK_CONCURRENCY'];
-  for (const key of requiredKeys) {
-    if (!(key in config)) {
-      throw new Error(`Missing required configuration: ${key}`);
-    }
+  if (!config.WEBHOOK_URL) {
+    throw new Error('Missing required configuration: WEBHOOK_URL');
+  }
+  if (!/^https?:\/\//i.test(config.WEBHOOK_URL)) {
+    throw new Error('WEBHOOK_URL must use http:// or https://');
+  }
+  if (!Number.isInteger(config.PORT) || config.PORT < 1 || config.PORT > 65535) {
+    throw new Error('PORT must be an integer between 1 and 65535');
+  }
+  if (!Number.isInteger(config.MAX_FILE_SIZE) || config.MAX_FILE_SIZE < 0) {
+    throw new Error('MAX_FILE_SIZE must be a non-negative integer');
+  }
+  if (!Number.isInteger(config.MAX_MESSAGE_SIZE) || config.MAX_MESSAGE_SIZE < 0) {
+    throw new Error('MAX_MESSAGE_SIZE must be a non-negative integer');
+  }
+  if (!Number.isInteger(config.MAX_SMTP_CLIENTS) || config.MAX_SMTP_CLIENTS < 1) {
+    throw new Error('MAX_SMTP_CLIENTS must be a positive integer');
+  }
+  if (!Number.isInteger(config.WEBHOOK_CONCURRENCY) || config.WEBHOOK_CONCURRENCY < 1) {
+    throw new Error('WEBHOOK_CONCURRENCY must be a positive integer');
+  }
+  if (!Number.isInteger(config.WEBHOOK_QUEUE_RETRIES) || config.WEBHOOK_QUEUE_RETRIES < 0) {
+    throw new Error('WEBHOOK_QUEUE_RETRIES must be a non-negative integer');
+  }
+  if (!Number.isInteger(config.WEBHOOK_QUEUE_RETRY_DELAY) || config.WEBHOOK_QUEUE_RETRY_DELAY < 1000) {
+    throw new Error('WEBHOOK_QUEUE_RETRY_DELAY must be at least 1000 milliseconds');
+  }
+  if (!Number.isInteger(config.WEBHOOK_QUEUE_TIMEOUT) || config.WEBHOOK_QUEUE_TIMEOUT < 1000) {
+    throw new Error('WEBHOOK_QUEUE_TIMEOUT must be at least 1000 milliseconds');
+  }
+  if (config.SMTP_SECURE && (!config.TLS_KEY_PATH || !config.TLS_CERT_PATH)) {
+    throw new Error('SMTP_SECURE=true requires TLS_KEY_PATH and TLS_CERT_PATH');
  }
 }

-const webhookQueue = new Queue(async function (parsed, cb) {
-  const maxRetries = 3;
-  let retries = 0;
-
-  const attemptWebhook = async () => {
-    try {
-      await sendToWebhook(parsed);
-      logger.info('Successfully sent to webhook');
-      cb(null);
-    } catch (error) {
-      logger.error('Webhook error:', { message: error.message, stack: error.stack });
-      if (error.response) {
-        logger.error('Webhook response error:', { 
-          status: error.response.status, 
-          data: error.response.data 
-        });
-      }
-      if (retries < maxRetries) {
-        retries++;
-        logger.info(`Retrying webhook (attempt ${retries}/${maxRetries})`);
-        setTimeout(attemptWebhook, 1000 * retries);
-      } else {
-        cb(error);
-      }
-    }
-  };
-
-  attemptWebhook();
-}, { concurrent: config.WEBHOOK_CONCURRENCY || 5 });
-
-const server = new SMTPServer({
-  onData(stream, session, callback) {
-    parseEmail(stream)
-      .then(parsed => {
-        webhookQueue.push(parsed);
-        logger.info('Email added to queue', { queueSize: webhookQueue.getStats().total });
-        callback();
-      })
-      .catch(error => {
-        logger.error('Parsing error:', { message: error.message, stack: error.stack });
-        callback(new Error('Failed to parse email'));
-      });
-  },
-  onError(error) {
-    logger.error('SMTP server error:', { message: error.message, stack: error.stack });
-  },
-  disabledCommands: ['AUTH'],
-  secure: config.SMTP_SECURE
-});
-
-server.listen(config.PORT, '0.0.0.0', err => {
-  if (err) {
-    logger.error('Failed to start SMTP server:', { message: err.message, stack: err.stack });
-    process.exit(1);
-  }
-  logger.info(`SMTP server listening on port ${config.PORT} on all interfaces`);
-});
-
-function gracefulShutdown(reason) {
-  logger.info(`Shutting down: ${reason}`);
-  server.close(() => {
-    logger.info('Server closed. Exiting process.');
-    process.exit(0);
-  });
-}
-
-process.on('uncaughtException', (err) => {
-  logger.error('Uncaught exception:', { message: err.message, stack: err.stack });
-  gracefulShutdown('Uncaught exception');
-});
-
-process.on('unhandledRejection', (reason, promise) => {
-  logger.error('Unhandled Rejection:', { reason: reason, promise: promise });
-  gracefulShutdown('Unhandled rejection');
-});
-
-process.on('SIGTERM', () => {
-  gracefulShutdown('SIGTERM signal received');
-});
-
-process.on('SIGINT', () => {
-  gracefulShutdown('SIGINT signal received');
-});
-
-// Add configuration validation at startup
 try {
  validateConfig();
 } catch (error) {
  logger.error('Configuration error:', { message: error.message });
  process.exit(1);
-}
+}
+
+const pendingDir = path.join(config.SPOOL_DIR, 'pending');
+const failedDir = path.join(config.SPOOL_DIR, 'failed');
+let shuttingDown = false;
+let spoolScanTimer = null;
+let activeJobs = 0;
+const activeJobWaiters = new Set();
+const enqueuedJobIds = new Set();
+
+try {
+  fs.mkdirSync(pendingDir, { recursive: true, mode: 0o700 });
+  fs.mkdirSync(failedDir, { recursive: true, mode: 0o700 });
+} catch (error) {
+  logger.error('Failed to initialize spool directories:', {
+    message: error.message,
+    stack: error.stack
+  });
+  process.exit(1);
+}
+
+function notifyActiveJobWaiters() {
+  if (activeJobs !== 0) return;
+  for (const waiter of activeJobWaiters) waiter();
+  activeJobWaiters.clear();
+}
+
+function waitForActiveJobs(timeout) {
+  if (activeJobs === 0) return Promise.resolve();
+
+  return new Promise(resolve => {
+    let timer;
+    const onDone = () => {
+      clearTimeout(timer);
+      activeJobWaiters.delete(onDone);
+      resolve();
+    };
+
+    timer = setTimeout(onDone, timeout);
+    activeJobWaiters.add(onDone);
+  });
+}
+
+function delay(milliseconds) {
+  return new Promise(resolve => setTimeout(resolve, milliseconds));
+}
+
+function logWebhookError(error) {
+  const message = error instanceof Error ? error.message : String(error);
+  const stack = error instanceof Error ? error.stack : undefined;
+  logger.error('Webhook error:', { message, stack });
+
+  if (error && error.response) {
+    let data = error.response.data;
+    if (typeof data === 'string') data = data.slice(0, 500);
+    logger.error('Webhook response error:', {
+      status: error.response.status,
+      data
+    });
+  }
+}
+
+async function sendWithRetry(parsed) {
+  const maxRetries = 3;
+  let retries = 0;
+
+  while (true) {
+    try {
+      await sendToWebhook(parsed);
+      return;
+    } catch (error) {
+      logWebhookError(error);
+      if (retries >= maxRetries) throw error;
+
+      retries++;
+      logger.info(`Retrying webhook (attempt ${retries}/${maxRetries})`);
+      await delay(1000 * retries);
+    }
+  }
+}
+
+async function persistEmail(parsed) {
+  const id = `${Date.now()}-${process.pid}-${crypto.randomUUID()}`;
+  const filePath = path.join(pendingDir, `${id}.json`);
+  const temporaryPath = `${filePath}.tmp`;
+
+  let fileHandle;
+  try {
+    fileHandle = await fs.promises.open(temporaryPath, 'w', 0o600);
+    await fileHandle.writeFile(JSON.stringify(parsed), 'utf8');
+    await fileHandle.sync();
+    await fileHandle.close();
+    fileHandle = undefined;
+    await fs.promises.rename(temporaryPath, filePath);
+    return { id, filePath };
+  } catch (error) {
+    if (fileHandle) {
+      await fileHandle.close().catch(() => {});
+    }
+    await fs.promises.unlink(temporaryPath).catch(() => {});
+    throw error;
+  }
+}
+
+async function quarantineFile(filePath) {
+  const target = path.join(
+    failedDir,
+    `${path.basename(filePath)}.${Date.now()}.failed`
+  );
+  await fs.promises.rename(filePath, target);
+  return target;
+}
+
+async function processWebhookJob(job, cb) {
+  activeJobs++;
+
+  try {
+    let parsed;
+    try {
+      parsed = JSON.parse(await fs.promises.readFile(job.filePath, 'utf8'));
+    } catch (error) {
+      if (error.code === 'ENOENT') {
+        logger.warn('Spool job disappeared before processing', { jobId: job.id });
+        cb(null);
+        return;
+      }
+
+      try {
+        const failedPath = await quarantineFile(job.filePath);
+        logger.error('Invalid spool job moved to failed spool', {
+          jobId: job.id,
+          failedPath,
+          message: error.message
+        });
+        cb(null);
+      } catch (quarantineError) {
+        cb(quarantineError);
+      }
+      return;
+    }
+
+    await sendWithRetry(parsed);
+    await fs.promises.unlink(job.filePath);
+    logger.info('Successfully sent to webhook', { jobId: job.id });
+    cb(null);
+  } catch (error) {
+    logWebhookError(error);
+    cb(error);
+  } finally {
+    activeJobs--;
+    notifyActiveJobWaiters();
+  }
+}
+
+const webhookQueue = new Queue(processWebhookJob, {
+  concurrent: config.WEBHOOK_CONCURRENCY,
+  maxRetries: config.WEBHOOK_QUEUE_RETRIES,
+  retryDelay: config.WEBHOOK_QUEUE_RETRY_DELAY,
+  maxTimeout: config.WEBHOOK_QUEUE_TIMEOUT
+});
+
+webhookQueue.on('error', error => {
+  logger.error('Webhook queue error:', { message: error.message, stack: error.stack });
+});
+
+webhookQueue.on('task_failed', (jobId, error) => {
+  logger.error('Webhook queue task exhausted retries:', {
+    jobId,
+    message: error && error.message ? error.message : String(error)
+  });
+});
+
+function enqueueSpoolFile(filePath) {
+  const id = path.basename(filePath, '.json');
+  if (enqueuedJobIds.has(id)) return;
+
+  enqueuedJobIds.add(id);
+  const ticket = webhookQueue.push({ id, filePath });
+  ticket.once('finish', () => enqueuedJobIds.delete(id));
+  ticket.once('failed', () => enqueuedJobIds.delete(id));
+}
+
+async function scanSpool() {
+  if (shuttingDown) return;
+
+  try {
+    const entries = await fs.promises.readdir(pendingDir, { withFileTypes: true });
+    for (const entry of entries) {
+      if (entry.isFile() && entry.name.endsWith('.json')) {
+        enqueueSpoolFile(path.join(pendingDir, entry.name));
+      }
+    }
+  } catch (error) {
+    logger.error('Failed to scan spool:', { message: error.message, stack: error.stack });
+  }
+}
+
+const smtpOptions = {
+  maxClients: config.MAX_SMTP_CLIENTS,
+  socketTimeout: config.SMTP_SOCKET_TIMEOUT,
+  closeTimeout: config.SMTP_CLOSE_TIMEOUT,
+  size: config.MAX_MESSAGE_SIZE || undefined,
+  onConnect(session, callback) {
+    if (
+      config.SMTP_ALLOWED_IPS.length > 0 &&
+      !config.SMTP_ALLOWED_IPS.includes(session.remoteAddress)
+    ) {
+      const error = new Error('Connection not allowed');
+      error.responseCode = 421;
+      callback(error);
+      return;
+    }
+    callback();
+  },
+  onData(stream, session, callback) {
+    parseEmail(stream)
+      .then(async parsed => {
+        const job = await persistEmail(parsed);
+        enqueueSpoolFile(job.filePath);
+        logger.info('Email persisted and added to queue', {
+          jobId: job.id,
+          queued: webhookQueue.length
+        });
+        callback();
+      })
+      .catch(error => {
+        logger.error('Parsing or persistence error:', {
+          message: error.message,
+          stack: error.stack
+        });
+        callback(new Error('Failed to persist email'));
+      });
+  },
+  disabledCommands: ['AUTH'],
+  secure: config.SMTP_SECURE
+};
+
+if (config.SMTP_SECURE) {
+  smtpOptions.key = fs.readFileSync(config.TLS_KEY_PATH);
+  smtpOptions.cert = fs.readFileSync(config.TLS_CERT_PATH);
+}
+
+const server = new SMTPServer(smtpOptions);
+
+let serverReady = false;
+server.on('error', error => {
+  logger.error('SMTP server error:', { message: error.message, stack: error.stack });
+  if (!serverReady) {
+    // Let the file transport flush the startup error before exiting. Since the
+    // listener never came up, there are no active handles keeping the process
+    // alive; the non-zero exit code still lets Supervisor restart it.
+    process.exitCode = 1;
+  }
+});
+
+function gracefulShutdown(reason) {
+  if (shuttingDown) return;
+  shuttingDown = true;
+  logger.info(`Shutting down: ${reason}`);
+  clearInterval(spoolScanTimer);
+  webhookQueue.pause();
+
+  const cleanShutdown = reason === 'SIGTERM signal received' || reason === 'SIGINT signal received';
+  const exitCode = cleanShutdown ? 0 : 1;
+  const shutdownDeadline = setTimeout(() => {
+    logger.error('Shutdown timeout reached; exiting with pending spool jobs preserved');
+    process.exit(exitCode);
+  }, 15000);
+
+  server.close(async () => {
+    await waitForActiveJobs(10000);
+    clearTimeout(shutdownDeadline);
+    logger.info('Server closed. Exiting process.');
+    process.exit(exitCode);
+  });
+}
+
+process.on('uncaughtException', error => {
+  logger.error('Uncaught exception:', { message: error.message, stack: error.stack });
+  gracefulShutdown('Uncaught exception');
+});
+
+process.on('unhandledRejection', reason => {
+  const error = reason instanceof Error
+    ? { message: reason.message, stack: reason.stack }
+    : { message: String(reason) };
+  logger.error('Unhandled Rejection:', error);
+  gracefulShutdown('Unhandled rejection');
+});
+
+process.on('SIGTERM', () => gracefulShutdown('SIGTERM signal received'));
+process.on('SIGINT', () => gracefulShutdown('SIGINT signal received'));
+
+server.listen(config.PORT, '0.0.0.0', () => {
+  serverReady = true;
+  logger.info(`SMTP server listening on port ${config.PORT} on all interfaces`);
+  void scanSpool();
+  spoolScanTimer = setInterval(() => void scanSpool(), 60 * 1000);
+  spoolScanTimer.unref();
+});