smtp.ts · Source

smtp.tsapps/worker/src/lib/smtp.tsraw ← lib10,474 B← Back to the app
1/**
2 * apps/worker — minimal SMTP client (node:net only, no deps).
3 *
4 * Implements the bare RFC 5321 dialogue needed for outbound mail:
5 *
6 *   connect → 220 → EHLO factory.local → 250 (+capabilities)
7 *   → [STARTTLS → 220 → EHLO again] when STARTTLS is offered and enabled
8 *   → MAIL FROM:<from> → 250
9 *   → RCPT TO:<to>      → 250
10 *   → DATA               → 354
11 *   → <headers>\r\n<body (dot-stuffed)>\r\n.\r\n → 250
12 *   → QUIT               → 221
13 *
14 * - dot-stuffing: a body line beginning with "." is prefixed with an extra
15 *   "." (RFC 5321 §4.5.2) and the message terminates with a lone ".\r\n",
16 * - responses are matched by 3-digit code; multi-line replies join on the
17 *   space-before-the-code boundary,
18 * - timeouts on every command (default 30s) — a hung SMTP server must not
19 *   wedge the email queue forever (the job still retries via BullMQ).
20 *
21 * The client is exported separately so the command sequence is unit-testable
22 * against a fake socket (see __tests__/smtp.test.ts).
23 */
24import { connect, type Socket } from "node:net";
25import { connect as tlsConnect, type TLSSocket } from "node:tls";
26 
27export interface SmtpSendOptions {
28  host: string;
29  port: number;
30  user: string;
31  pass: string;
32  from: string;
33  to: string[];
34  subject: string;
35  html?: string;
36  text?: string;
37  /** Force STARTTLS when the server advertises it (default true). */
38  startTls?: boolean;
39  localName?: string;
40  commandTimeoutMs?: number;
41}
42 
43export class SmtpError extends Error {
44  constructor(message: string, public code?: number) {
45    super(message);
46    this.name = "SmtpError";
47  }
48}
49 
50interface Pending {
51  resolve: (line: string) => void;
52  reject: (err: Error) => void;
53  timer: NodeJS.Timeout;
54}
55 
56/**
57 * Stateful line reader: buffers socket data, splits CRLF lines, resolves the
58 * next pending command promise per line. Multi-line 250/4xx replies are
59 * joined by the caller (each physical line arrives separately; we resolve on
60 * the LAST line of a reply — detected by the space-vs-hyphen convention:
61 * continuation lines have a "-" in column 4).
62 */
63export class SmtpSession {
64  private buf = Buffer.alloc(0);
65  private pending: Pending | null = null;
66  private lastReply = "";
67  private closed = false;
68 
69  constructor(private socket: Socket | TLSSocket) {
70    socket.on("data", (chunk) => this.onData(chunk));
71    socket.on("error", (err) => this.fail(err));
72    socket.on("close", () => this.fail(new SmtpError("smtp connection closed")));
73  }
74 
75  private onData(chunk: Buffer): void {
76    this.buf = Buffer.concat([this.buf, chunk]);
77    let idx: number;
78    while ((idx = this.buf.indexOf(0x0a)) >= 0) {
79      const line = this.buf.subarray(0, idx).toString("utf8").replace(/\r$/, "");
80      this.buf = this.buf.subarray(idx + 1);
81      this.handleLine(line);
82    }
83  }
84 
85  private handleLine(line: string): void {
86    // RFC 5321: continuation lines are "NNN-text", final lines "NNN text".
87    const isContinuation = line.length > 3 && line[3] === "-";
88    if (isContinuation) {
89      this.lastReply += line + "\n";
90      return;
91    }
92    this.lastReply += line + "\n";
93    if (this.pending) {
94      const p = this.pending;
95      this.pending = null;
96      clearTimeout(p.timer);
97      p.resolve(this.lastReply.trimEnd());
98      this.lastReply = "";
99    }
100  }
101 
102  /** Send a raw command line (CRLF added). */
103  write(line: string): void {
104    this.socket.write(line + "\r\n");
105  }
106 
107  /** Write arbitrary bytes (used for the DATA payload). */
108  writeRaw(data: string): void {
109    this.socket.write(data);
110  }
111 
112  private fail(err: Error): void {
113    if (this.closed) return;
114    this.closed = true;
115    if (this.pending) {
116      const p = this.pending;
117      this.pending = null;
118      clearTimeout(p.timer);
119      p.reject(err);
120    }
121  }
122 
123  /** Wait for the next complete reply. */
124  expect(commandTimeoutMs: number): Promise<string> {
125    if (this.pending) {
126      return Promise.reject(new SmtpError("command already in flight"));
127    }
128    if (this.lastReply) {
129      const r = this.lastReply.trimEnd();
130      this.lastReply = "";
131      return Promise.resolve(r);
132    }
133    return new Promise((resolve, reject) => {
134      const timer = setTimeout(() => {
135        this.pending = null;
136        reject(new SmtpError("smtp command timeout"));
137      }, commandTimeoutMs);
138      this.pending = { resolve, reject, timer };
139    });
140  }
141 
142  codeOf(reply: string): number {
143    return Number(reply.slice(0, 3));
144  }
145 
146  end(): void {
147    this.closed = true;
148    this.socket.end();
149  }
150 
151  destroy(): void {
152    this.closed = true;
153    this.socket.destroy();
154  }
155}
156 
157function requireCode(session: SmtpSession, reply: string, expected: number[], what: string): void {
158  const code = session.codeOf(reply);
159  if (!expected.includes(code)) {
160    throw new SmtpError(`smtp ${what} rejected: ${reply}`, code);
161  }
162}
163 
164/**
165 * Send one email. Resolves on 250-after-DATA; throws SmtpError on any
166 * protocol/transport failure (BullMQ retries the job).
167 */
168export async function smtpSend(opts: SmtpSendOptions): Promise<void> {
169  const timeout = opts.commandTimeoutMs ?? 30_000;
170  const localName = opts.localName ?? "factory.local";
171 
172  await new Promise<void>((resolve, reject) => {
173    const socket: Socket = connect({ host: opts.host, port: opts.port });
174    let session: SmtpSession | null = null;
175    let done = false;
176    let upgraded = false;
177 
178    const fail = (err: Error): void => {
179      if (done) return;
180      done = true;
181      socket.destroy();
182      reject(err);
183    };
184 
185    socket.setTimeout(timeout, () => fail(new SmtpError("smtp connect timeout")));
186    socket.once("error", (err) => fail(err));
187 
188    socket.once("connect", async () => {
189      try {
190        session = new SmtpSession(socket);
191        const banner = await session.expect(timeout);
192        requireCode(session, banner, [220], "banner");
193 
194        // EHLO
195        let ehloReply = await ehlo(session, localName, timeout);
196        if ((opts.startTls ?? true) && /STARTTLS/i.test(ehloReply) && !upgraded) {
197          upgraded = true;
198          // 1) ask, 2) get "220 Ready to start TLS", 3) upgrade the socket.
199          await cmd(session, "STARTTLS", timeout, [220], "STARTTLS");
200          const secure = await new Promise<TLSSocket>((res, rej) => {
201            const t = tlsConnect({ socket, servername: opts.host }, () => res(t));
202            t.once("error", rej);
203          });
204          session = new SmtpSession(secure);
205          ehloReply = await ehlo(session, localName, timeout);
206        }
207 
208        // AUTH LOGIN (only when the server advertises it and we have creds)
209        if (opts.user && /AUTH\s+LOGIN/i.test(ehloReply)) {
210          const r1 = await cmd(session, `AUTH LOGIN`, timeout, [334], "AUTH LOGIN");
211          void r1;
212          const r2 = await cmd(session, Buffer.from(opts.user, "utf8").toString("base64"), timeout, [334], "auth user");
213          void r2;
214          await cmd(session, Buffer.from(opts.pass, "utf8").toString("base64"), timeout, [235], "auth pass");
215        }
216 
217        // MAIL FROM / RCPT TO
218        await cmd(session, `MAIL FROM:<${opts.from}>`, timeout, [250], "MAIL FROM");
219        for (const to of opts.to) {
220          await cmd(session, `RCPT TO:<${to}>`, timeout, [250, 251], "RCPT TO");
221        }
222 
223        // DATA
224        await cmd(session, "DATA", timeout, [354], "DATA");
225        const body = buildMessage(opts);
226        const stuffed = dotStuff(body);
227        session.writeRaw(stuffed + "\r\n.\r\n");
228        const dataOk = await session.expect(timeout);
229        requireCode(session, dataOk, [250], "DATA accept");
230 
231        await cmd(session, "QUIT", timeout, [221], "QUIT");
232        done = true;
233        socket.end();
234        resolve();
235      } catch (err) {
236        fail(err instanceof Error ? err : new SmtpError(String(err)));
237      }
238    });
239  });
240}
241 
242async function ehlo(
243  session: SmtpSession,
244  localName: string,
245  timeout: number
246): Promise<string> {
247  const reply = await cmd(session, `EHLO ${localName}`, timeout, [250], "EHLO");
248  return reply;
249}
250 
251async function cmd(
252  session: SmtpSession,
253  line: string,
254  timeout: number,
255  expected: number[],
256  what: string
257): Promise<string> {
258  session.write(line);
259  const reply = await session.expect(timeout);
260  requireCode(session, reply, expected, what);
261  return reply;
262}
263 
264/** Build the full RFC 5322 message (headers + multipart body). */
265export function buildMessage(opts: {
266  from: string;
267  to: string[];
268  subject: string;
269  html?: string;
270  text?: string;
271  messageId?: string;
272}): string {
273  const boundary = "----factory-" + Date.now().toString(36) + Math.random().toString(36).slice(2, 8);
274  const lines: string[] = [
275    `From: ${opts.from}`,
276    `To: ${opts.to.join(", ")}`,
277    `Subject: ${encodeHeader(opts.subject)}`,
278    "MIME-Version: 1.0",
279    "Content-Type: multipart/alternative; boundary=" + boundary,
280    `Message-ID: ${opts.messageId ?? "<" + Date.now() + "@factory.local>"}`,
281    "Date: " + new Date().toUTCString(),
282    ""
283  ];
284  const parts: string[] = [];
285  if (opts.text) {
286    parts.push(
287      `--${boundary}\r\nContent-Type: text/plain; charset=utf-8\r\nContent-Transfer-Encoding: base64\r\n\r\n${toBase64Wrapped(opts.text)}`
288    );
289  }
290  if (opts.html ?? opts.text) {
291    const html = opts.html ?? `<pre>${escapeHtml(opts.text ?? "")}</pre>`;
292    parts.push(
293      `--${boundary}\r\nContent-Type: text/html; charset=utf-8\r\nContent-Transfer-Encoding: base64\r\n\r\n${toBase64Wrapped(html)}`
294    );
295  }
296  lines.push(...parts, `--${boundary}--`, "");
297  return lines.join("\r\n");
298}
299 
300/** RFC 5321 §4.5.2 dot-stuffing. */
301export function dotStuff(message: string): string {
302  return message
303    .split("\r\n")
304    .map((line) => (line.startsWith(".") ? "." + line : line))
305    .join("\r\n");
306}
307 
308function toBase64Wrapped(s: string): string {
309  const b64 = Buffer.from(s, "utf8").toString("base64");
310  // 76-char lines per RFC 2045
311  return b64.match(/.{1,76}/g)?.join("\r\n") ?? "";
312}
313 
314function escapeHtml(s: string): string {
315  return s.replace(/&/g, "&amp;").replace(/</g, "&lt;").replace(/>/g, "&gt;");
316}
317 
318/** Encode a UTF-8 header value (Subject) with RFC 2047 words. */
319export function encodeHeader(value: string): string {
320  if (/^[\x20-\x7e]*$/.test(value)) return value;
321  const b64 = Buffer.from(value, "utf8").toString("base64");
322  return b64
323    .match(/.{1,75}/g)
324    ?.map((chunk, i, arr) => (i === arr.length - 1 ? `=?utf-8?B?${chunk}?=` : `=?utf-8?B?${chunk}?=`))
325    ?.join("") ?? value;
326}
327