feat(workflows): add the smtp relay

nodejs/package.json

@@ -132,6 +132,7 @@
         "lru-cache": "^11.0.0",
         "luxon": "^3.4.4",
         "lz4": "^0.6.5",
+        "mailparser": "^3.14.1",
         "node-fetch": "^2.6.1",
         "node-rdkafka": "^3.6.1",
         "node-schedule": "^2.1.0",
@@ -145,6 +146,7 @@
         "puppeteer": "^24.3.0",
         "puppeteer-capture": "^1.43.0",
         "re2": "^1.26.1",
+        "smtp-server": "^3.21.3",
         "snappy": "^7.2.2",
         "structured-headers": "^2.0.3",
         "tail": "^2.2.6",
@@ -163,6 +165,7 @@
         "@babel/types": "^7.20.2",
         "@bufbuild/protoc-gen-es": "^2.11.0",
         "@eslint-community/eslint-plugin-eslint-comments": "^4.4.1",
+        "@jest/test-sequencer": "^30.0.5",
         "@segment/actions-core": "^3.151.0",
         "@trivago/prettier-plugin-sort-imports": "^5.2.2",
         "@types/adm-zip": "^0.4.34",
@@ -178,12 +181,13 @@
         "@types/lodash": "^4.17.16",
         "@types/long": "4.x.x",
         "@types/luxon": "^3.4.2",
+        "@types/mailparser": "^3.9.0",
         "@types/node-fetch": "^2.5.10",
         "@types/node-schedule": "^2.1.0",
         "@types/nodemailer": "^8.0.0",
         "@types/pg": "~8.6.6",
-        "@jest/test-sequencer": "^30.0.5",
         "@types/redlock": "^4.0.1",
+        "@types/smtp-server": "^3.5.13",
         "@types/snowflake-sdk": "^1.5.1",
         "@types/supertest": "^6.0.2",
         "@typescript-eslint/eslint-plugin": "^8.58.2",

nodejs/src/capabilities.ts

@@ -188,6 +188,10 @@ export function getPluginServerCapabilities(
             }
         // Deliberately absent from CAPABILITIES_CDP: a replay is a bounded, operator-triggered run,
         // not something a local or combined stack should start on its own.
+        case PluginServerMode.cdp_smtp_relay:
+            return {
+                cdpSmtpRelay: true,
+            }
         case PluginServerMode.cdp_dlq_replay:
             return {
                 cdpDlqReplay: true,

nodejs/src/cdp/config.ts

@@ -179,6 +179,35 @@ export type CdpConfig = ClickhouseConfig & {
     CDP_EMAIL_ATTACHMENTS_S3_ENDPOINT: string
     CDP_EMAIL_ATTACHMENTS_S3_ACCESS_KEY_ID: string
     CDP_EMAIL_ATTACHMENTS_S3_SECRET_ACCESS_KEY: string
+    // The URL of Django that the SMTP relay calls, with the customer's own key.
+    SMTP_RELAY_API_URL: string
+    SMTP_RELAY_HOSTNAME: string
+    SMTP_RELAY_PORTS: string
+    // Ports where people expect TLS from the start. Every port also takes the other kind of client.
+    SMTP_RELAY_IMPLICIT_TLS_PORTS: string
+    SMTP_RELAY_TLS_KEY_PATH: string
+    SMTP_RELAY_TLS_CERT_PATH: string
+    SMTP_RELAY_TLS_WAIT_IMPLICIT_MS: number
+    SMTP_RELAY_TLS_WAIT_STARTTLS_MS: number
+    SMTP_RELAY_MAX_MESSAGE_BYTES: number
+    SMTP_RELAY_REPLY_BUDGET_MS: number
+    SMTP_RELAY_LARGE_MESSAGE_BUDGET_MS: number
+    SMTP_RELAY_MAX_CLIENTS: number
+    SMTP_RELAY_MAX_CONNECTIONS_PER_IP: number
+    SMTP_RELAY_MAX_FAILED_LOGINS_PER_IP: number
+    SMTP_RELAY_MAX_CONCURRENT_MESSAGES: number
+    SMTP_RELAY_MAX_CONCURRENT_MESSAGES_PER_TEAM: number
+    // Limits on the message bytes a relay pod holds. A message takes several times its size in memory while it is
+    // parsed and sent, so these stay well under the pod's memory limit.
+    SMTP_RELAY_MAX_IN_FLIGHT_BYTES: number
+    SMTP_RELAY_MAX_IN_FLIGHT_BYTES_PER_TEAM: number
+    // From the moment a connection opens until TLS finishes, whatever the client sends in that time.
+    SMTP_RELAY_TLS_DEADLINE_MS: number
+    // From DATA until the end of the message, whatever the client sends in that time.
+    SMTP_RELAY_DATA_DEADLINE_MS: number
+    // Signs the relay's calls that record refused messages. Empty turns off that recording.
+    SMTP_RELAY_ACTIVITY_JWT_SECRET: string
+    SMTP_RELAY_SHUTDOWN_GRACE_MS: number
     SES_ENDPOINT: string
     SES_ACCESS_KEY_ID: string
     SES_SECRET_ACCESS_KEY: string
@@ -442,6 +471,29 @@ export function getDefaultCdpConfig(): CdpConfig {
         CDP_EMAIL_ATTACHMENTS_S3_ENDPOINT: isDevEnv() ? 'http://localhost:19000' : '',
         CDP_EMAIL_ATTACHMENTS_S3_ACCESS_KEY_ID: isDevEnv() ? 'object_storage_root_user' : '',
         CDP_EMAIL_ATTACHMENTS_S3_SECRET_ACCESS_KEY: isDevEnv() ? 'object_storage_root_password' : '',
+        SMTP_RELAY_API_URL: isDevEnv() ? 'http://localhost:8000' : '',
+        SMTP_RELAY_HOSTNAME: 'localhost',
+        SMTP_RELAY_PORTS: '465,587,2465,2587',
+        SMTP_RELAY_IMPLICIT_TLS_PORTS: '465,2465',
+        SMTP_RELAY_TLS_KEY_PATH: '',
+        SMTP_RELAY_TLS_CERT_PATH: '',
+        SMTP_RELAY_TLS_WAIT_IMPLICIT_MS: 1_500,
+        SMTP_RELAY_TLS_WAIT_STARTTLS_MS: 400,
+        // Fits 10 MiB of attachments after base64, plus the message bodies.
+        SMTP_RELAY_MAX_MESSAGE_BYTES: 15 * 1024 * 1024,
+        SMTP_RELAY_REPLY_BUDGET_MS: 4_000,
+        SMTP_RELAY_LARGE_MESSAGE_BUDGET_MS: 20_000,
+        SMTP_RELAY_MAX_CLIENTS: 2_000,
+        SMTP_RELAY_MAX_CONNECTIONS_PER_IP: 50,
+        SMTP_RELAY_MAX_FAILED_LOGINS_PER_IP: 10,
+        SMTP_RELAY_MAX_CONCURRENT_MESSAGES: 50,
+        SMTP_RELAY_MAX_CONCURRENT_MESSAGES_PER_TEAM: 10,
+        SMTP_RELAY_MAX_IN_FLIGHT_BYTES: 120 * 1024 * 1024,
+        SMTP_RELAY_MAX_IN_FLIGHT_BYTES_PER_TEAM: 45 * 1024 * 1024,
+        SMTP_RELAY_TLS_DEADLINE_MS: 10_000,
+        SMTP_RELAY_DATA_DEADLINE_MS: 180_000,
+        SMTP_RELAY_ACTIVITY_JWT_SECRET: isTestEnv() || isDevEnv() ? 'local-dev-smtp-relay-activity-jwt' : '',
+        SMTP_RELAY_SHUTDOWN_GRACE_MS: 25_000,
         SES_ENDPOINT: isTestEnv() || isDevEnv() ? 'http://localhost:4566' : '',
         SES_ACCESS_KEY_ID: isTestEnv() || isDevEnv() ? 'test' : '',
         SES_SECRET_ACCESS_KEY: isTestEnv() || isDevEnv() ? 'test' : '',

nodejs/src/cdp/smtp-relay/smtp-limits.ts

@@ -0,0 +1,34 @@
+/** Counts what is in use, in total and per team. A caller over either limit gets a refusal instead of a place in line. */
+export class TeamLimit {
+    private used = 0
+    private readonly usedByTeam = new Map<number, number>()
+
+    constructor(
+        private readonly total: number,
+        private readonly perTeam: number
+    ) {}
+
+    public get inUse(): number {
+        return this.used
+    }
+
+    public tryTake(teamId: number, amount: number): boolean {
+        const teamUsed = this.usedByTeam.get(teamId) ?? 0
+        if (this.used + amount > this.total || teamUsed + amount > this.perTeam) {
+            return false
+        }
+        this.used += amount
+        this.usedByTeam.set(teamId, teamUsed + amount)
+        return true
+    }
+
+    public give(teamId: number, amount: number): void {
+        this.used -= amount
+        const teamUsed = (this.usedByTeam.get(teamId) ?? 0) - amount
+        if (teamUsed <= 0) {
+            this.usedByTeam.delete(teamId)
+        } else {
+            this.usedByTeam.set(teamId, teamUsed)
+        }
+    }
+}

nodejs/src/cdp/smtp-relay/smtp-message.test.ts

@@ -0,0 +1,200 @@
+import { parseSmtpMessage, planSends } from './smtp-message'
+import { SmtpReply } from './smtp-replies'
+
+const raw = (headers: string[], body = 'Hello'): Buffer =>
+    Buffer.from([...headers, 'Content-Type: text/plain; charset=utf-8', '', body, ''].join('\r\n'))
+
+const plan = async (headers: string[], envelopeTo: string[], from = 'envelope@example.com', body?: string) =>
+    planSends(await parseSmtpMessage(raw(headers, body)), { from, to: envelopeTo })
+
+describe('planSends', () => {
+    it('sends one email per To, with copies on the first only, and takes who gets it from the envelope', async () => {
+        const { sends } = await plan(
+            [
+                'From: Jane <jane@example.com>',
+                'To: a@example.com, B Person <b@example.com>, not-in-envelope@example.com',
+                'Cc: c@example.com',
+                'Subject: Hi',
+            ],
+            ['a@example.com', 'B@EXAMPLE.COM', 'c@example.com', 'hidden@example.com']
+        )
+
+        expect(sends.map(({ body }) => [body.to, body.cc, body.bcc])).toEqual([
+            ['a@example.com', ['c@example.com'], ['hidden@example.com']],
+            ['B@EXAMPLE.COM', undefined, undefined],
+        ])
+    })
+
+    it('still sends when every recipient is hidden', async () => {
+        const { sends } = await plan(
+            ['From: jane@example.com', 'To: undisclosed-recipients:;', 'Subject: Hi'],
+            ['a@example.com', 'b@example.com']
+        )
+
+        expect(sends.map(({ body }) => [body.to, body.bcc])).toEqual([['a@example.com', ['b@example.com']]])
+    })
+
+    it.each([
+        ['a name with a comma', 'From: "Smith, Jane" <jane@example.com>', '"Smith, Jane" <jane@example.com>'],
+        ['an encoded name', 'From: =?UTF-8?Q?Zo=C3=AB?= <jane@example.com>', '"Zoë" <jane@example.com>'],
+        ['no name', 'From: jane@example.com', 'jane@example.com'],
+    ])('keeps the sender with %s', async (_name, header, from) => {
+        const { sends } = await plan([header, 'To: a@example.com', 'Subject: Hi'], ['a@example.com'])
+
+        expect(sends[0].body.from).toEqual(from)
+    })
+
+    it('falls back to the envelope sender and refuses two senders', async () => {
+        const { sends } = await plan(['To: a@example.com', 'Subject: Hi'], ['a@example.com'], 'envelope@example.com')
+        expect(sends[0].body.from).toEqual('envelope@example.com')
+
+        await expect(
+            plan(['From: a@example.com, b@example.com', 'To: a@example.com'], ['a@example.com'])
+        ).rejects.toBeInstanceOf(SmtpReply)
+    })
+
+    it('fills what the send API requires but mail tools may leave out', async () => {
+        const { sends } = await plan(['From: jane@example.com', 'To: a@example.com'], ['a@example.com'], undefined, '')
+
+        expect(sends[0].body).toMatchObject({ subject: '(no subject)', template_key: 'smtp' })
+        expect(sends[0].body.text).toBeTruthy()
+    })
+
+    it('forwards the headers the send API takes and reports the others', async () => {
+        const { sends, droppedHeaders } = await plan(
+            [
+                'From: jane@example.com',
+                'To: a@example.com',
+                'Subject: Hi',
+                'X-Entity-Ref-ID: order-1',
+                'In-Reply-To: <a@example.com>',
+                'References: <a@example.com>',
+                '  <b@example.com>',
+                'X-PostHog-Tracking-Code: forged',
+                'X-SES-TENANT: team-2',
+                'Sender: other@example.com',
+                'X-Mailer: PHPMailer',
+            ],
+            ['a@example.com']
+        )
+
+        expect(sends[0].body.headers).toEqual({
+            'X-Entity-Ref-ID': 'order-1',
+            'In-Reply-To': '<a@example.com>',
+            References: '<a@example.com> <b@example.com>',
+        })
+        expect(droppedHeaders).toEqual(['X-PostHog-Tracking-Code', 'X-SES-TENANT', 'Sender'])
+    })
+
+    it.each([
+        ['nothing', [], false],
+        ['the header', ['X-PostHog-Capture-Engagement: true'], true],
+        ['a header set to something else', ['X-PostHog-Capture-Engagement: yes'], false],
+    ])('tracks opens and clicks with %s', async (_name, extra, tracked) => {
+        const { sends } = await plan(
+            ['From: jane@example.com', 'To: a@example.com', 'Subject: Hi', ...extra],
+            ['a@example.com']
+        )
+
+        expect(sends[0].body).toMatchObject({ capture_engagement: tracked, track_opens_and_clicks: tracked })
+    })
+
+    it('gives each recipient its own stable idempotency key', async () => {
+        const headers = ['From: j@example.com', 'To: a@example.com, b@example.com', 'Subject: Hi', 'Message-ID: <1@x>']
+        const [first, retry] = [
+            await plan(headers, ['a@example.com', 'b@example.com']),
+            await plan(headers, ['a@example.com', 'b@example.com']),
+        ]
+
+        expect(first.sends[0].idempotencyKey).toEqual(retry.sends[0].idempotencyKey)
+        expect(first.sends[0].idempotencyKey).not.toEqual(first.sends[1].idempotencyKey)
+        expect(first.sends[0].idempotencyKey).toMatch(/^[!-~]{1,255}$/)
+    })
+
+    it.each([['PostHog-Idempotency-Key: 1'], ['Resend-Idempotency-Key: 1']])(
+        'prefers the explicit %s over the Message-ID',
+        async (header) => {
+            const headers = (messageId: string) => [
+                'From: j@example.com',
+                'To: a@example.com',
+                'Subject: Hi',
+                `Message-ID: ${messageId}`,
+                header,
+            ]
+            const [first, second] = [
+                await plan(headers('<1@x>'), ['a@example.com']),
+                await plan(headers('<2@x>'), ['a@example.com']),
+            ]
+
+            expect(first.sends[0].idempotencyKey).toEqual(second.sends[0].idempotencyKey)
+        }
+    )
+
+    it('leaves a message with no key or Message-ID without an idempotency key', async () => {
+        const { sends } = await plan(['From: j@example.com', 'To: a@example.com', 'Subject: Hi'], ['a@example.com'])
+
+        expect(sends[0].idempotencyKey).toBeUndefined()
+    })
+
+    it('keeps cid links, marks inline files and cleans file names', async () => {
+        const boundary = 'b1'
+        const message = Buffer.from(
+            [
+                'From: j@example.com',
+                'To: a@example.com',
+                'Subject: Hi',
+                `Content-Type: multipart/related; boundary=${boundary}`,
+                '',
+                `--${boundary}`,
+                'Content-Type: text/html',
+                '',
+                '<img src="cid:logo">',
+                `--${boundary}`,
+                'Content-Type: image/png; name="logo.png"',
+                'Content-ID: <logo>',
+                'Content-Disposition: inline; filename="logo.png"',
+                'Content-Transfer-Encoding: base64',
+                '',
+                'aGk=',
+                `--${boundary}`,
+                'Content-Type: application/pdf; name="../../a.pdf"',
+                'Content-Disposition: attachment; filename="../../a.pdf"',
+                'Content-Transfer-Encoding: base64',
+                '',
+                'aGk=',
+                `--${boundary}--`,
+                '',
+            ].join('\r\n')
+        )
+
+        const { sends } = planSends(await parseSmtpMessage(message), { to: ['a@example.com'] })
+
+        expect(sends[0].body.html).toContain('cid:logo')
+        expect(sends[0].body.attachments?.map(({ filename, content_id }) => [filename, content_id])).toEqual([
+            ['logo.png', 'logo'],
+            ['a.pdf', undefined],
+        ])
+    })
+
+    it.each([
+        ['more To addresses than a message can take', 51, 0],
+        ['more Bcc addresses than a message can take', 1, 11],
+    ])('refuses the whole message with %s', async (_name, toCount, bccCount) => {
+        const to = Array.from({ length: toCount }, (_, i) => `t${i}@example.com`)
+        const bcc = Array.from({ length: bccCount }, (_, i) => `h${i}@example.com`)
+
+        await expect(
+            plan(['From: j@example.com', `To: ${to.join(', ')}`, 'Subject: Hi'], [...to, ...bcc])
+        ).rejects.toMatchObject({ responseCode: 554 })
+    })
+
+    it('refuses signed mail, which rebuilding would break', async () => {
+        const message = Buffer.from(
+            'From: j@example.com\r\nTo: a@example.com\r\nContent-Type: multipart/signed; boundary=b\r\n\r\n--b\r\n\r\nhi\r\n--b--\r\n'
+        )
+
+        await expect(
+            parseSmtpMessage(message).then((parsed) => planSends(parsed, { to: ['a@example.com'] }))
+        ).rejects.toMatchObject({ responseCode: 554 })
+    })
+})