diff --git a/nodejs/package.json b/nodejs/package.json index 2104a2db843c..7a3998dfeddf 100644 --- a/nodejs/package.json +++ b/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", diff --git a/nodejs/src/capabilities.ts b/nodejs/src/capabilities.ts index d2f2a29e2843..43bc9d288b02 100644 --- a/nodejs/src/capabilities.ts +++ b/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, diff --git a/nodejs/src/cdp/config.ts b/nodejs/src/cdp/config.ts index 0f18e5665623..88e44e7ede63 100644 --- a/nodejs/src/cdp/config.ts +++ b/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' : '', diff --git a/nodejs/src/cdp/smtp-relay/smtp-limits.ts b/nodejs/src/cdp/smtp-relay/smtp-limits.ts new file mode 100644 index 000000000000..768c9f61cdcf --- /dev/null +++ b/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() + + 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) + } + } +} diff --git a/nodejs/src/cdp/smtp-relay/smtp-message.test.ts b/nodejs/src/cdp/smtp-relay/smtp-message.test.ts new file mode 100644 index 000000000000..5ad1b447e8cc --- /dev/null +++ b/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 ', + 'To: a@example.com, B Person , 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" ', '"Smith, Jane" '], + ['an encoded name', 'From: =?UTF-8?Q?Zo=C3=AB?= ', '"Zoë" '], + ['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: ', + 'References: ', + ' ', + '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': '', + References: ' ', + }) + 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', + '', + '', + `--${boundary}`, + 'Content-Type: image/png; name="logo.png"', + 'Content-ID: ', + '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 }) + }) +}) diff --git a/nodejs/src/cdp/smtp-relay/smtp-message.ts b/nodejs/src/cdp/smtp-relay/smtp-message.ts new file mode 100644 index 000000000000..96e53f15d397 --- /dev/null +++ b/nodejs/src/cdp/smtp-relay/smtp-message.ts @@ -0,0 +1,274 @@ +import { createHash } from 'crypto' +import { AddressObject, ParsedMail, simpleParser } from 'mailparser' + +import { REPLIES } from './smtp-replies' + +export const SMTP_TEMPLATE_KEY = 'smtp' +export const MAX_TO_RECIPIENTS = 50 +export const MAX_COPY_RECIPIENTS = 10 +export const MAX_FORWARDED_HEADERS = 10 +const MAX_LISTED_RECIPIENTS = 20 + +const TEMPLATE_KEY_RE = /^[a-zA-Z0-9_-]{1,400}$/ +const FORWARDED_HEADERS = new Set(['in-reply-to', 'references', 'list-unsubscribe', 'list-unsubscribe-post', 'list-id']) +// Used up by the relay, or describing the message the relay takes apart. None of these is dropped from the caller's +// point of view, so none is reported as dropped. +const CONSUMED_HEADERS = new Set([ + 'bcc', + 'cc', + 'content-disposition', + 'content-id', + 'content-transfer-encoding', + 'content-type', + 'date', + 'dkim-signature', + 'from', + 'message-id', + 'mime-version', + 'posthog-idempotency-key', + 'received', + 'reply-to', + 'resend-idempotency-key', + 'return-path', + 'subject', + 'to', + 'user-agent', + 'x-mailer', + 'x-posthog-capture-engagement', + 'x-posthog-distinct-id', + 'x-posthog-template-key', +]) +const RESERVED_HEADER_RE = /^x-(posthog|ses)-/ + +export type SmtpAttachment = { + filename: string + content: string + content_type?: string + content_id?: string +} + +export type SmtpSend = { + idempotencyKey?: string + body: { + to: string + cc?: string[] + bcc?: string[] + template_key: string + subject: string + html?: string + text?: string + from: string + reply_to?: string[] + attachments?: SmtpAttachment[] + headers?: Record + capture_engagement: boolean + track_opens_and_clicks: boolean + distinct_id?: string + } +} + +export type SmtpPlan = { + sends: SmtpSend[] + fromAddress: string + droppedHeaders: string[] +} + +export async function parseSmtpMessage(raw: Buffer): Promise { + try { + // Keeps `cid:` links, which the parser otherwise turns into data URIs, and adds no derived bodies. + return await simpleParser(raw, { + keepCidLinks: true, + skipHtmlToText: true, + skipTextToHtml: true, + skipTextLinks: true, + skipImageLinks: true, + }) + } catch { + throw REPLIES.unreadable() + } +} + +const addressesIn = (field: AddressObject | AddressObject[] | undefined): { name: string; address: string }[] => { + const groups = Array.isArray(field) ? field : field ? [field] : [] + const entries = groups.flatMap((group) => group.value) + return entries.flatMap((entry) => + entry.group + ? entry.group.flatMap((member) => (member.address ? [{ name: member.name, address: member.address }] : [])) + : entry.address + ? [{ name: entry.name, address: entry.address }] + : [] + ) +} + +const normalize = (address: string): string => address.trim().toLowerCase() + +function rawHeaderValue(parsed: ParsedMail, name: string): string | undefined { + const line = parsed.headerLines.find((header) => header.key === name)?.line + if (line === undefined) { + return undefined + } + return line + .slice(line.indexOf(':') + 1) + .replace(/\r?\n[ \t]+/g, ' ') + .trim() +} + +function forwardedHeaders(parsed: ParsedMail): { headers: Record; dropped: string[] } { + const headers: Record = {} + const dropped: string[] = [] + const seen = new Set() + for (const { key, line } of parsed.headerLines) { + if (seen.has(key) || CONSUMED_HEADERS.has(key)) { + continue + } + seen.add(key) + const name = line.slice(0, line.indexOf(':')).trim() + const value = rawHeaderValue(parsed, key) + const forwardable = + !RESERVED_HEADER_RE.test(key) && (FORWARDED_HEADERS.has(key) || key.startsWith('x-')) && !!value + if (forwardable && Object.keys(headers).length < MAX_FORWARDED_HEADERS && value.length <= 998) { + headers[name] = value + } else { + dropped.push(name) + } + } + return { headers, dropped } +} + +function attachmentsOf(parsed: ParsedMail): SmtpAttachment[] { + return parsed.attachments.map((attachment, index) => { + // A file name with a path, or none at all, is refused by the send API. + const baseName = (attachment.filename ?? '').split(/[\\/]/).pop()?.trim() + const contentId = attachment.related ? attachment.cid?.replace(/[<>]/g, '') : undefined + const contentType = attachment.contentType?.split(';')[0].trim() + return { + filename: baseName || `attachment-${index + 1}`, + content: attachment.content.toString('base64'), + ...(contentType && { content_type: contentType }), + ...(contentId && { content_id: contentId }), + } + }) +} + +function displayName(name: string, address: string): string { + return name ? `"${name.replace(/[\\"]/g, '\\$&')}" <${address}>` : address +} + +function senderOf(parsed: ParsedMail, envelopeFrom: string | undefined): { from: string; address: string } { + const senders = addressesIn(parsed.from) + if (senders.length > 1) { + throw REPLIES.severalSenders() + } + const sender = senders[0] ?? (envelopeFrom ? { name: '', address: envelopeFrom } : undefined) + if (!sender) { + throw REPLIES.noSender() + } + return { from: displayName(sender.name, sender.address), address: sender.address } +} + +// The envelope decides who gets the email. The headers only decide who sees whom. +function recipientsOf( + parsed: ParsedMail, + envelopeTo: string[] +): { + to: string[] + cc: string[] + bcc: string[] +} { + const inTo = new Set(addressesIn(parsed.to).map(({ address }) => normalize(address))) + const inCc = new Set(addressesIn(parsed.cc).map(({ address }) => normalize(address))) + const unique = [...new Map(envelopeTo.map((address) => [normalize(address), address.trim()])).values()] + const to = unique.filter((address) => inTo.has(normalize(address))) + const cc = unique.filter((address) => !inTo.has(normalize(address)) && inCc.has(normalize(address))) + const bcc = unique.filter((address) => !inTo.has(normalize(address)) && !inCc.has(normalize(address))) + if (to.length === 0) { + const promoted = cc.shift() ?? bcc.shift() + if (promoted) { + to.push(promoted) + } + } + return { to, cc, bcc } +} + +/** Addresses the message's own headers show to its readers, so never a Bcc. Capped for the recent problems list. */ +export function visibleRecipients(parsed: ParsedMail, envelopeTo: string[]): string[] { + const named = new Set( + [...addressesIn(parsed.to), ...addressesIn(parsed.cc)].map(({ address }) => normalize(address)) + ) + return envelopeTo.filter((address) => named.has(normalize(address))).slice(0, MAX_LISTED_RECIPIENTS) +} + +function idempotencyBase(parsed: ParsedMail): string | undefined { + const explicit = + rawHeaderValue(parsed, 'posthog-idempotency-key') ?? rawHeaderValue(parsed, 'resend-idempotency-key') + if (explicit) { + return `key:${explicit}` + } + return parsed.messageId ? `message-id:${parsed.messageId}` : undefined +} + +const perRecipientKey = (base: string | undefined, to: string): string | undefined => + base + ? `smtp-${createHash('sha256') + .update(`${base}\n${normalize(to)}`) + .digest('hex')}` + : undefined + +function isSigned(parsed: ParsedMail): boolean { + const contentType = parsed.headers.get('content-type') as { value?: string } | undefined + return contentType?.value === 'multipart/signed' || contentType?.value === 'application/pkcs7-mime' +} + +/** Turns one SMTP message into the send API requests that deliver it. Throws an SmtpReply when it can't be sent. */ +export function planSends(parsed: ParsedMail, envelope: { from?: string; to: string[] }): SmtpPlan { + if (isSigned(parsed)) { + throw REPLIES.signed() + } + const { to, cc, bcc } = recipientsOf(parsed, envelope.to) + if (to.length > MAX_TO_RECIPIENTS) { + throw REPLIES.tooManyRecipients(MAX_TO_RECIPIENTS) + } + if (cc.length > MAX_COPY_RECIPIENTS || bcc.length > MAX_COPY_RECIPIENTS) { + throw REPLIES.tooManyRecipients(MAX_COPY_RECIPIENTS) + } + const sender = senderOf(parsed, envelope.from) + const { headers, dropped } = forwardedHeaders(parsed) + const replyTo = addressesIn(parsed.replyTo).map(({ address }) => address) + const attachments = attachmentsOf(parsed) + const templateKey = rawHeaderValue(parsed, 'x-posthog-template-key') + const distinctId = rawHeaderValue(parsed, 'x-posthog-distinct-id') + const base = idempotencyBase(parsed) + const html = typeof parsed.html === 'string' ? parsed.html : '' + // Mail tools can send a subject or a body alone, and the send API needs both. + const text = parsed.text || (html ? '' : ' ') + + // Tracking stays off unless asked for: this mail often carries single-use links, which a rewrite can break. + const tracked = rawHeaderValue(parsed, 'x-posthog-capture-engagement')?.toLowerCase() === 'true' + const shared = { + template_key: templateKey && TEMPLATE_KEY_RE.test(templateKey) ? templateKey : SMTP_TEMPLATE_KEY, + subject: parsed.subject || '(no subject)', + from: sender.from, + capture_engagement: tracked, + track_opens_and_clicks: tracked, + ...(html && { html }), + ...(text && { text }), + ...(replyTo.length > 0 && { reply_to: replyTo }), + ...(attachments.length > 0 && { attachments }), + ...(Object.keys(headers).length > 0 && { headers }), + ...(distinctId && { distinct_id: distinctId }), + } + return { + fromAddress: sender.address, + droppedHeaders: dropped, + sends: to.map((address, index) => ({ + idempotencyKey: perRecipientKey(base, address), + body: { + ...shared, + to: address, + // Copies go out once, with the first send. + ...(index === 0 && cc.length > 0 && { cc }), + ...(index === 0 && bcc.length > 0 && { bcc }), + }, + })), + } +} diff --git a/nodejs/src/cdp/smtp-relay/smtp-relay.service.test.ts b/nodejs/src/cdp/smtp-relay/smtp-relay.service.test.ts new file mode 100644 index 000000000000..88a2df26fa2d --- /dev/null +++ b/nodejs/src/cdp/smtp-relay/smtp-relay.service.test.ts @@ -0,0 +1,374 @@ +import { mkdtempSync, rmSync, writeFileSync } from 'fs' +import net from 'net' +import nodemailer from 'nodemailer' +import { tmpdir } from 'os' +import { join } from 'path' +import { TLSSocket, connect as tlsConnect } from 'tls' + +import { waitForExpect } from '~/tests/helpers/expectations' +import { createTestTlsIdentity } from '~/tests/helpers/tls' + +import { CdpConfig } from '../config' +import { SmtpRelayService } from './smtp-relay.service' +import { REPLIES } from './smtp-replies' +import { SmtpCredential, SmtpSendApi } from './smtp-send-api' + +jest.mock('~/common/utils/posthog', () => ({ captureException: jest.fn() })) + +// The relay under test uses a certificate that this test creates for itself. +// nosemgrep: problem-based-packs.insecure-transport.js-node.bypass-tls-verification.bypass-tls-verification +const TRUST_TEST_CERTIFICATE = { rejectUnauthorized: false } + +describe('SmtpRelayService', () => { + let directory: string + let relay: SmtpRelayService + let port: number + let checkKey: jest.SpyInstance + let send: jest.SpyInstance + let recordActivity: jest.SpyInstance + const credential: SmtpCredential = { apiKey: 'phs_good', teamId: 1 } + + const transport = (auth: { user: string; pass: string } | undefined = { user: 'posthog', pass: 'phs_good' }) => + nodemailer.createTransport({ host: '127.0.0.1', port, tls: TRUST_TEST_CERTIFICATE, auth }) + + const firstLine = (socket: net.Socket): Promise => + new Promise((resolve) => socket.once('data', (data) => resolve(data.toString().split('\r\n')[0]))) + + beforeAll(async () => { + directory = mkdtempSync(join(tmpdir(), 'smtp-relay-test-')) + const identity = await createTestTlsIdentity('localhost') + writeFileSync(join(directory, 'tls.key'), identity.privateKey) + writeFileSync(join(directory, 'tls.crt'), identity.certificate) + await identity.cleanup() + }) + + afterAll(() => { + rmSync(directory, { recursive: true, force: true }) + }) + + const startRelay = async (overrides: Partial = {}): Promise => { + relay = new SmtpRelayService({ + SMTP_RELAY_API_URL: 'http://unused.invalid', + SMTP_RELAY_HOSTNAME: 'localhost', + SMTP_RELAY_MAX_MESSAGE_BYTES: 1024 * 1024, + SMTP_RELAY_REPLY_BUDGET_MS: 4_000, + SMTP_RELAY_LARGE_MESSAGE_BUDGET_MS: 4_000, + SMTP_RELAY_MAX_CLIENTS: 100, + SMTP_RELAY_MAX_CONNECTIONS_PER_IP: 100, + SMTP_RELAY_MAX_CONCURRENT_MESSAGES: 5, + SMTP_RELAY_MAX_CONCURRENT_MESSAGES_PER_TEAM: 5, + SMTP_RELAY_MAX_IN_FLIGHT_BYTES: 10 * 1024 * 1024, + SMTP_RELAY_MAX_IN_FLIGHT_BYTES_PER_TEAM: 10 * 1024 * 1024, + SMTP_RELAY_TLS_DEADLINE_MS: 10_000, + SMTP_RELAY_DATA_DEADLINE_MS: 10_000, + SMTP_RELAY_ACTIVITY_JWT_SECRET: '', + SMTP_RELAY_TLS_WAIT_IMPLICIT_MS: 50, + SMTP_RELAY_PORTS: '0', + SMTP_RELAY_IMPLICIT_TLS_PORTS: '', + SMTP_RELAY_TLS_KEY_PATH: join(directory, 'tls.key'), + SMTP_RELAY_TLS_CERT_PATH: join(directory, 'tls.crt'), + SMTP_RELAY_TLS_WAIT_STARTTLS_MS: 50, + SMTP_RELAY_SHUTDOWN_GRACE_MS: 500, + SMTP_RELAY_MAX_FAILED_LOGINS_PER_IP: 2, + ...overrides, + } as unknown as CdpConfig) + await relay.start() + port = relay.listeningPorts[0] + } + + const restartRelay = async (overrides: Partial): Promise => { + await relay.stop() + await startRelay(overrides) + } + + const closed = (socket: net.Socket): Promise => + new Promise((resolve) => socket.once('close', () => resolve())) + + // Stops at DATA, so a test can decide how the message arrives. + const openData = async (key = 'phs_good'): Promise<{ socket: TLSSocket; reply: () => Promise }> => { + const socket = tlsConnect({ port, host: '127.0.0.1', ...TRUST_TEST_CERTIFICATE }) + socket.on('error', () => undefined) + const replies: string[] = [] + const waiting: ((line: string) => void)[] = [] + let buffered = '' + socket.on('data', (data) => { + buffered += data.toString() + let end = buffered.indexOf('\r\n') + while (end >= 0) { + const line = buffered.slice(0, end) + buffered = buffered.slice(end + 2) + if (line[3] !== '-') { + const next = waiting.shift() + if (next) { + next(line) + } else { + replies.push(line) + } + } + end = buffered.indexOf('\r\n') + } + }) + const reply = (): Promise => + replies.length > 0 ? Promise.resolve(replies.shift()!) : new Promise((resolve) => waiting.push(resolve)) + await reply() + const login = Buffer.from(`\0posthog\0${key}`).toString('base64') + for (const command of [ + 'EHLO test', + `AUTH PLAIN ${login}`, + 'MAIL FROM:', + 'RCPT TO:', + ]) { + socket.write(`${command}\r\n`) + await reply() + } + socket.write('DATA\r\n') + await reply() + return { socket, reply } + } + + beforeEach(async () => { + checkKey = jest.spyOn(SmtpSendApi.prototype, 'checkKey').mockImplementation((key) => { + return key === 'phs_good' ? Promise.resolve(credential) : Promise.reject(REPLIES.invalidCredentials()) + }) + send = jest.spyOn(SmtpSendApi.prototype, 'send').mockResolvedValue('message-1') + recordActivity = jest.spyOn(SmtpSendApi.prototype, 'recordActivity').mockResolvedValue() + await startRelay() + }) + + afterEach(async () => { + await relay.stop() + jest.restoreAllMocks() + }) + + const mail = { from: 'Jane ', to: 'a@example.com', subject: 'Hi', text: 'Hello' } + + it('takes a client that starts with STARTTLS and hands the message to the send API', async () => { + const info = await transport().sendMail(mail) + + expect(info.response).toEqual('250 2.0.0 Ok: queued as message-1') + expect(send.mock.calls[0][1].body).toMatchObject({ to: 'a@example.com', from: '"Jane" ' }) + }) + + it('takes a client that starts with TLS on the same port', async () => { + const info = await nodemailer + .createTransport({ + host: '127.0.0.1', + port, + secure: true, + tls: TRUST_TEST_CERTIFICATE, + auth: { user: 'anything', pass: 'phs_good' }, + }) + .sendMail(mail) + + expect(info.response).toContain('250 2.0.0') + }) + + it('refuses a login before TLS, and offers only logins it can check', async () => { + const socket = net.connect(port, '127.0.0.1') + await firstLine(socket) + + socket.write('EHLO test\r\n') + const capabilities = await new Promise((resolve) => + socket.once('data', (data) => resolve(String(data))) + ) + socket.write('AUTH PLAIN AHBvc3Rob2cAcGhzX2dvb2Q=\r\n') + const reply = await firstLine(socket) + socket.destroy() + + expect(capabilities).toContain('AUTH PLAIN LOGIN\r\n') + expect(capabilities).not.toContain('SMTPUTF8') + expect(capabilities).not.toContain('DSN') + expect(reply).toMatch(/^538 /) + expect(checkKey).not.toHaveBeenCalled() + }) + + it('answers the whole message with the refusal when every send is refused', async () => { + send.mockRejectedValue(REPLIES.noSender()) + + await expect(transport().sendMail(mail)).rejects.toMatchObject({ responseCode: 550 }) + expect(recordActivity).toHaveBeenCalledWith( + credential.teamId, + expect.objectContaining({ kind: 'refused', reply: expect.stringContaining('550') }) + ) + }) + + it.each([ + ['accepts when some recipients are refused', [REPLIES.noSender(), 'id-2'], 250], + ['asks for a retry when a send has no answer, so nobody is lost', [REPLIES.unconfirmed(), 'id-2'], 451], + ])('%s', async (_name, outcomes, code) => { + for (const outcome of outcomes) { + if (typeof outcome === 'string') { + send.mockResolvedValueOnce(outcome) + } else { + send.mockRejectedValueOnce(outcome) + } + } + + const sending = transport().sendMail({ ...mail, to: 'a@example.com, b@example.com' }) + + if (code === 250) { + await expect(sending).resolves.toMatchObject({ response: expect.stringContaining('250 2.0.0') }) + } else { + await expect(sending).rejects.toMatchObject({ responseCode: code }) + } + }) + + it('leaves Bcc recipients out of what it records about a refusal', async () => { + send.mockRejectedValue(REPLIES.noSender()) + + await expect( + transport().sendMail({ ...mail, cc: 'copy@example.com', bcc: 'hidden@example.com' }) + ).rejects.toMatchObject({ responseCode: 550 }) + + expect(recordActivity.mock.calls[0][1].to.sort()).toEqual(['a@example.com', 'copy@example.com']) + }) + + it('locks out a bad key from an address, but not over an outage on our side, and not for other keys', async () => { + checkKey.mockRejectedValueOnce(REPLIES.authUnavailable()) + await expect(transport({ user: 'posthog', pass: 'phs_bad' }).verify()).rejects.toMatchObject({ + responseCode: 454, + }) + for (let attempt = 0; attempt < 2; attempt++) { + await expect(transport({ user: 'posthog', pass: 'phs_bad' }).verify()).rejects.toMatchObject({ + responseCode: 535, + }) + } + + await expect(transport({ user: 'posthog', pass: 'phs_bad' }).verify()).rejects.toMatchObject({ + responseCode: 454, + }) + expect(checkKey).toHaveBeenCalledTimes(3) + await expect(transport({ user: 'posthog', pass: 'phs_good' }).verify()).resolves.toBe(true) + }) + + it('lets many logins that arrive together with one new key through on a single key check', async () => { + const verified = await Promise.all(Array.from({ length: 20 }, () => transport().verify())) + + expect(verified).toEqual(Array(20).fill(true)) + expect(checkKey).toHaveBeenCalledTimes(1) + }) + + it('counts each connection once from accept, and frees its place when it closes', async () => { + await restartRelay({ SMTP_RELAY_MAX_CONNECTIONS_PER_IP: 2 } as Partial) + const held = [net.connect(port, '127.0.0.1'), net.connect(port, '127.0.0.1')] + held[0].write(Buffer.from([0x16])) + await firstLine(held[1]) + + const refused = net.connect(port, '127.0.0.1') + expect(await firstLine(refused)).toMatch(/^421 4\.7\.0 /) + await closed(refused) + + const stillRefused = net.connect(port, '127.0.0.1') + expect(await firstLine(stillRefused)).toMatch(/^421 4\.7\.0 /) + await closed(stillRefused) + + held[0].destroy() + // The relay frees the place when it sees the close, which is after this side sees it. + await waitForExpect(async () => { + const next = net.connect(port, '127.0.0.1') + const line = await firstLine(next) + next.destroy() + expect(line).toMatch(/^220 /) + }) + held[1].destroy() + }) + + it.each([ + ['stops at a TLS hello', (socket: net.Socket) => socket.write(Buffer.from([0x16]))], + ['never starts TLS', (_socket: net.Socket) => undefined], + ])('closes a connection that %s before the TLS deadline passes', async (_name, start) => { + await restartRelay({ SMTP_RELAY_TLS_DEADLINE_MS: 200 } as Partial) + const socket = net.connect(port, '127.0.0.1') + socket.on('error', () => undefined) + socket.on('data', () => undefined) + start(socket) + + await closed(socket) + + await expect(transport().verify()).resolves.toBe(true) + }) + + it('refuses a team over its message slots, while another team still sends', async () => { + await restartRelay({ SMTP_RELAY_MAX_CONCURRENT_MESSAGES_PER_TEAM: 1 } as Partial) + checkKey.mockImplementation((key) => Promise.resolve({ apiKey: key, teamId: key === 'phs_other' ? 2 : 1 })) + const slow = await openData() + + const refused = await openData() + refused.socket.write('Subject: x\r\n\r\nx\r\n.\r\n') + expect(await refused.reply()).toMatch(/^451 4\.3\.2 /) + + const other = await openData('phs_other') + other.socket.write('From: a@example.com\r\nTo: b@example.com\r\n\r\nx\r\n.\r\n') + expect(await other.reply()).toMatch(/^250 /) + for (const { socket } of [slow, refused, other]) { + socket.destroy() + } + }) + + it('answers a message over the byte limit with a retry, and frees its bytes', async () => { + await restartRelay({ SMTP_RELAY_MAX_IN_FLIGHT_BYTES_PER_TEAM: 2048 } as Partial) + const large = await openData() + large.socket.write(`From: a@example.com\r\nTo: b@example.com\r\n\r\n${'x'.repeat(4096)}\r\n.\r\n`) + expect(await large.reply()).toMatch(/^451 4\.3\.2 /) + expect(send).not.toHaveBeenCalled() + + large.socket.write('RSET\r\n') + await large.reply() + await expect(transport().sendMail(mail)).resolves.toMatchObject({ response: expect.stringContaining('250') }) + large.socket.destroy() + }) + + it('ends a message that is still arriving at the deadline, and frees its slot', async () => { + await restartRelay({ + SMTP_RELAY_MAX_CONCURRENT_MESSAGES_PER_TEAM: 1, + SMTP_RELAY_DATA_DEADLINE_MS: 200, + } as Partial) + const slow = await openData() + slow.socket.write('Subject: x\r\n') + + expect(await slow.reply()).toMatch(/^421 4\.4\.2 /) + + await expect(transport().sendMail(mail)).resolves.toMatchObject({ response: expect.stringContaining('250') }) + slow.socket.destroy() + }) + + it('sends a large message to one recipient at a time', async () => { + await restartRelay({ SMTP_RELAY_MAX_MESSAGE_BYTES: 3 * 1024 * 1024 } as Partial) + let inFlight = 0 + let mostInFlight = 0 + send.mockImplementation(async () => { + inFlight += 1 + mostInFlight = Math.max(mostInFlight, inFlight) + await new Promise((resolve) => setImmediate(resolve)) + inFlight -= 1 + return 'id' + }) + + await transport().sendMail({ + ...mail, + to: 'a@example.com, b@example.com, c@example.com', + text: 'x'.repeat(2 * 1024 * 1024), + }) + + expect(send).toHaveBeenCalledTimes(3) + expect(mostInFlight).toEqual(1) + }) + + it('refuses a message over the limit whatever SIZE it declared', async () => { + await expect( + transport().sendMail({ ...mail, envelope: undefined, text: 'x'.repeat(1024 * 1024 + 10) }) + ).rejects.toMatchObject({ responseCode: 552 }) + expect(send).not.toHaveBeenCalled() + }) + + it('tells an open connection to reconnect when it shuts down', async () => { + const socket = tlsConnect({ port, host: '127.0.0.1', ...TRUST_TEST_CERTIFICATE }) + await firstLine(socket) + const goodbye = firstLine(socket) + + await relay.stop() + + expect(await goodbye).toMatch(/^421 4\.3\.2 /) + socket.destroy() + }) +}) diff --git a/nodejs/src/cdp/smtp-relay/smtp-relay.service.ts b/nodejs/src/cdp/smtp-relay/smtp-relay.service.ts new file mode 100644 index 000000000000..0113fcffea4a --- /dev/null +++ b/nodejs/src/cdp/smtp-relay/smtp-relay.service.ts @@ -0,0 +1,627 @@ +import { createHash } from 'crypto' +import { readFileSync, statSync } from 'fs' +import { ParsedMail } from 'mailparser' +import net from 'net' +import { Counter, Histogram } from 'prom-client' +import { SMTPServer, SMTPServerAuthentication, SMTPServerOptions, SMTPServerSession } from 'smtp-server' +import { Readable } from 'stream' + +import { logger } from '~/common/utils/logger' +import { captureException } from '~/common/utils/posthog' +import { HealthCheckResultOk, PluginServerService } from '~/types' + +import { CdpConfig } from '../config' +import { TeamLimit } from './smtp-limits' +import { SmtpPlan, parseSmtpMessage, planSends, visibleRecipients } from './smtp-message' +import { REPLIES, SmtpReply } from './smtp-replies' +import { SmtpActivity, SmtpCredential, SmtpSendApi } from './smtp-send-api' + +const smtpRelayConnectionsTotal = new Counter({ + name: 'cdp_smtp_relay_connections_total', + help: 'Connections to the SMTP relay by how the client started: tls (a TLS hello first) or plain (waited for the banner).', + labelNames: ['mode'], +}) +const smtpRelayRefusedTotal = new Counter({ + name: 'cdp_smtp_relay_refused_total', + help: 'Connections and messages the relay refused to protect itself, by reason.', + labelNames: ['reason'], +}) +const smtpRelayAuthTotal = new Counter({ + name: 'cdp_smtp_relay_auth_total', + help: 'SMTP logins by the reply code the client got, with cached for a login answered from the key cache.', + labelNames: ['result'], +}) +const smtpRelayMessagesTotal = new Counter({ + name: 'cdp_smtp_relay_messages_total', + help: 'Messages through DATA by the reply code the client got.', + labelNames: ['code'], +}) +const smtpRelayReplyMs = new Histogram({ + name: 'cdp_smtp_relay_reply_ms', + help: 'Milliseconds from the end of DATA to the reply, by the reply class. A client that sends with short timeouts gives up after 5 seconds.', + labelNames: ['class'], + buckets: [50, 100, 250, 500, 1_000, 2_000, 3_000, 4_000, 5_000, 10_000, 20_000], +}) + +const KEY_CACHE_TTL_MS = 60_000 +const FAILED_LOGIN_WINDOW_MS = 5 * 60_000 +const FAN_OUT_CONCURRENCY = 5 +// Above this, the client has spent long enough uploading that the budget for short messages no longer applies, and +// each send is one at a time because every request body is another copy of the message in memory. +const LARGE_MESSAGE_BYTES = 1024 * 1024 +const MAX_KEY_CHECKS_PER_IP = 10 +const MAX_TRACKED_FAILED_LOGINS = 50_000 +// Forward secrecy on every TLS 1.2 connection. TLS 1.3 suites are always forward secret and not set by this list. +const TLS_CIPHERS = [ + 'ECDHE-ECDSA-AES128-GCM-SHA256', + 'ECDHE-RSA-AES128-GCM-SHA256', + 'ECDHE-ECDSA-AES256-GCM-SHA384', + 'ECDHE-RSA-AES256-GCM-SHA384', + 'ECDHE-ECDSA-CHACHA20-POLY1305', + 'ECDHE-RSA-CHACHA20-POLY1305', +].join(':') +const HOUSEKEEPING_INTERVAL_MS = 60_000 +const SHUTDOWN_POLL_MS = 100 +const TLS_HELLO_FIRST_BYTE = 0x16 + +type SessionWithKey = SMTPServerSession & { user?: unknown } + +type FailedLogins = { count: number; windowEndsAt: number } + +type OpenConnection = { ip: string; tlsDeadline: NodeJS.Timeout } + +export class SmtpRelayService { + private readonly listeners: net.Server[] = [] + private readonly openConnections = new Map() + private readonly connectionsPerIp = new Map() + private readonly keyChecks = new Map>() + private readonly keyChecksPerIp = new Map() + private lastConnectionId = 0 + private readonly failedLogins = new Map() + private readonly keyCache = new Map() + private plainServer!: SMTPServer + private tlsServer!: SMTPServer + private readonly messageSlots: TeamLimit + private readonly messageBytes: TeamLimit + private certificateMtimeMs = 0 + private housekeepingTimer?: NodeJS.Timeout + private readonly api: SmtpSendApi + + constructor(private readonly config: CdpConfig) { + this.api = new SmtpSendApi(config.SMTP_RELAY_API_URL, config.SMTP_RELAY_ACTIVITY_JWT_SECRET) + this.messageSlots = new TeamLimit( + config.SMTP_RELAY_MAX_CONCURRENT_MESSAGES, + config.SMTP_RELAY_MAX_CONCURRENT_MESSAGES_PER_TEAM + ) + this.messageBytes = new TeamLimit( + config.SMTP_RELAY_MAX_IN_FLIGHT_BYTES, + config.SMTP_RELAY_MAX_IN_FLIGHT_BYTES_PER_TEAM + ) + } + + public get service(): PluginServerService { + return { + id: 'cdp-smtp-relay', + onShutdown: () => this.stop(), + healthcheck: () => new HealthCheckResultOk(), + } + } + + public async start(): Promise { + const certificate = this.readCertificate() + const options: SMTPServerOptions = { + name: this.config.SMTP_RELAY_HOSTNAME, + banner: 'PostHog', + ...certificate, + minVersion: 'TLSv1.2', + ciphers: TLS_CIPHERS, + size: this.config.SMTP_RELAY_MAX_MESSAGE_BYTES, + authMethods: ['PLAIN', 'LOGIN'], + hideSMTPUTF8: true, + hideDSN: true, + hideENHANCEDSTATUSCODES: true, + hideREQUIRETLS: true, + closeTimeout: this.config.SMTP_RELAY_SHUTDOWN_GRACE_MS, + authRequiredMessage: `5.7.0 Authentication required. Use a project secret key as the password.`, + logger: false, + disableReverseLookup: true, + onSecure: (_socket, session, callback) => { + const connection = this.openConnections.get(session.id) + if (connection) { + clearTimeout(connection.tlsDeadline) + } + callback() + }, + onClose: (session) => this.releaseConnection(session.id), + onAuth: (auth, session, callback) => { + this.authenticate(auth, session).then( + (credential) => callback(null, { user: credential as unknown as string }), + (error) => callback(error instanceof Error ? error : REPLIES.authUnavailable()) + ) + }, + onData: (stream, session, callback) => { + this.handleData(stream, session).then( + (reply) => callback(null, reply), + (error) => callback(error instanceof Error ? error : REPLIES.unconfirmed()) + ) + }, + } + this.plainServer = this.buildServer({ ...options, secure: false }) + this.tlsServer = this.buildServer({ ...options, secure: true, needsUpgrade: true }) + + await Promise.all( + this.config.SMTP_RELAY_PORTS.split(',') + .map((port) => parseInt(port.trim())) + .filter((port) => !isNaN(port)) + .map((port) => this.listen(port)) + ) + this.housekeepingTimer = setInterval(() => this.runHousekeeping(), HOUSEKEEPING_INTERVAL_MS) + this.housekeepingTimer.unref() + } + + public get listeningPorts(): number[] { + return this.listeners.map((listener) => (listener.address() as net.AddressInfo).port) + } + + // SMTPServer.close() can't drain these servers, because connections reach them through our own listeners. + public async stop(): Promise { + clearInterval(this.housekeepingTimer) + // A listener's close only finishes once its connections have ended, which happens below. + const closed = Promise.all(this.listeners.map((listener) => new Promise((resolve) => listener.close(resolve)))) + const deadline = Date.now() + this.config.SMTP_RELAY_SHUTDOWN_GRACE_MS + while (this.messageSlots.inUse > 0 && Date.now() < deadline) { + await new Promise((resolve) => setTimeout(resolve, SHUTDOWN_POLL_MS)) + } + for (const server of [this.plainServer, this.tlsServer]) { + for (const connection of server.connections) { + connection.send(421, '4.3.2 Restarting. Reconnect and retry.') + connection.close() + } + } + await closed + } + + private runHousekeeping(): void { + this.reloadCertificate() + const now = Date.now() + for (const [key, failed] of this.failedLogins) { + if (failed.windowEndsAt <= now) { + this.failedLogins.delete(key) + } + } + for (const [key, cached] of this.keyCache) { + if (cached.expiresAt <= now) { + this.keyCache.delete(key) + } + } + } + + private buildServer(options: SMTPServerOptions): SMTPServer { + const server = new SMTPServer(options) + // A client that drops mid-handshake or sends garbage is routine on a public port, and carries no action. + server.on('error', (error) => logger.debug('SMTP relay connection error', { error: error.message })) + return server + } + + private listen(port: number): Promise { + const listener = net.createServer((socket) => this.accept(socket, port)) + this.listeners.push(listener) + return new Promise((resolve, reject) => { + listener.once('error', reject) + listener.listen(port, () => { + listener.off('error', reject) + logger.info('📬', 'SMTP relay listening', { port }) + resolve() + }) + }) + } + + // Any port takes either kind of client, because a TLS and port mismatch is a common way for SMTP setup to fail. A + // client that opens with TLS sends its hello at once, and a plain one waits for the banner. A port people expect + // TLS on waits longer, so a slow TLS client is less likely to get a plain banner. + private accept(socket: net.Socket, port: number): void { + const id = this.admitConnection(socket) + if (!id) { + return + } + const waitMs = this.implicitTlsPorts.has(port) + ? this.config.SMTP_RELAY_TLS_WAIT_IMPLICIT_MS + : this.config.SMTP_RELAY_TLS_WAIT_STARTTLS_MS + let settled = false + const settle = (first: Buffer | null): void => { + if (settled) { + return + } + settled = true + clearTimeout(timer) + socket.pause() + socket.removeListener('data', onData) + socket.removeListener('error', onError) + socket.removeListener('close', onClose) + if (first) { + socket.unshift(first) + } + const isTls = first?.[0] === TLS_HELLO_FIRST_BYTE + smtpRelayConnectionsTotal.inc({ mode: isTls ? 'tls' : 'plain' }) + ;(isTls ? this.tlsServer : this.plainServer).connect(socket, { id }) + } + const onData = (chunk: Buffer): void => settle(chunk) + const onError = (): void => { + socket.destroy() + } + const onClose = (): void => { + settled = true + clearTimeout(timer) + this.releaseConnection(id) + } + const timer = setTimeout(() => settle(null), waitMs) + socket.on('data', onData) + socket.on('error', onError) + socket.on('close', onClose) + } + + private get implicitTlsPorts(): Set { + return new Set( + this.config.SMTP_RELAY_IMPLICIT_TLS_PORTS.split(',') + .map((port) => parseInt(port.trim())) + .filter((port) => !isNaN(port)) + ) + } + + private readCertificate(): { key: Buffer; cert: Buffer } { + const { SMTP_RELAY_TLS_KEY_PATH: keyPath, SMTP_RELAY_TLS_CERT_PATH: certPath } = this.config + if (!keyPath || !certPath) { + throw new Error('The SMTP relay needs SMTP_RELAY_TLS_KEY_PATH and SMTP_RELAY_TLS_CERT_PATH') + } + this.certificateMtimeMs = statSync(certPath).mtimeMs + return { key: readFileSync(keyPath), cert: readFileSync(certPath) } + } + + private reloadCertificate(): void { + try { + if (statSync(this.config.SMTP_RELAY_TLS_CERT_PATH).mtimeMs === this.certificateMtimeMs) { + return + } + const certificate = this.readCertificate() + this.plainServer.updateSecureContext(certificate) + this.tlsServer.updateSecureContext(certificate) + logger.info('📬', 'SMTP relay reloaded its certificate') + } catch (error) { + logger.error('⚠️', 'SMTP relay could not reload its certificate', { error: String(error) }) + captureException(error) + } + } + + // Counted here rather than in the SMTP server's hooks, because those only run after TLS, and a socket holds its + // place from the moment it opens. Returns the connection's ID, or null when it was refused. + private admitConnection(socket: net.Socket): string | null { + socket.on('error', () => socket.destroy()) + const ip = (socket.remoteAddress ?? '').replace(/^::ffff:/, '') + const open = this.connectionsPerIp.get(ip) ?? 0 + const refusal = + this.openConnections.size >= this.config.SMTP_RELAY_MAX_CLIENTS + ? REPLIES.tooManyClients() + : open >= this.config.SMTP_RELAY_MAX_CONNECTIONS_PER_IP + ? REPLIES.tooManyConnections() + : null + if (refusal) { + smtpRelayRefusedTotal.inc({ reason: 'connections' }) + socket.end(`${refusal.line}\r\n`, () => socket.destroy()) + return null + } + this.lastConnectionId += 1 + const id = String(this.lastConnectionId) + this.connectionsPerIp.set(ip, open + 1) + // Nothing can be sent before TLS, so a connection that hasn't finished it by now has no use for its place. + const tlsDeadline = setTimeout(() => { + smtpRelayRefusedTotal.inc({ reason: 'tls_deadline' }) + socket.destroy() + }, this.config.SMTP_RELAY_TLS_DEADLINE_MS) + tlsDeadline.unref() + this.openConnections.set(id, { ip, tlsDeadline }) + return id + } + + private releaseConnection(id: string): void { + const connection = this.openConnections.get(id) + if (!connection) { + return + } + this.openConnections.delete(id) + clearTimeout(connection.tlsDeadline) + const open = this.connectionsPerIp.get(connection.ip) ?? 0 + if (open <= 1) { + this.connectionsPerIp.delete(connection.ip) + } else { + this.connectionsPerIp.set(connection.ip, open - 1) + } + } + + private isLockedOut(key: string): boolean { + const failed = this.failedLogins.get(key) + if (!failed) { + return false + } + if (failed.windowEndsAt <= Date.now()) { + this.failedLogins.delete(key) + return false + } + return failed.count >= this.config.SMTP_RELAY_MAX_FAILED_LOGINS_PER_IP + } + + private recordFailedLogin(key: string): void { + const failed = this.failedLogins.get(key) + if (failed && failed.windowEndsAt > Date.now()) { + failed.count += 1 + return + } + if (this.failedLogins.size >= MAX_TRACKED_FAILED_LOGINS) { + this.failedLogins.delete(this.failedLogins.keys().next().value as string) + } + this.failedLogins.set(key, { count: 1, windowEndsAt: Date.now() + FAILED_LOGIN_WINDOW_MS }) + } + + // Any username is accepted: tools without a username field send the From address or the key itself. + private async authenticate(auth: SMTPServerAuthentication, session: SMTPServerSession): Promise { + const ip = session.remoteAddress + const password = auth.password ?? '' + const cacheKey = createHash('sha256').update(password).digest('hex') + const cached = this.keyCache.get(cacheKey) + if (cached && cached.expiresAt > Date.now()) { + smtpRelayAuthTotal.inc({ result: 'cached' }) + return cached.credential + } + // Per key, because many customers can share one address, and one customer's bad key must not lock out the rest. + const failedLoginKey = `${ip}:${cacheKey}` + if (this.isLockedOut(failedLoginKey)) { + smtpRelayAuthTotal.inc({ result: 'locked_out' }) + throw REPLIES.tooManyFailedLogins() + } + try { + if (!password.startsWith('phs_')) { + throw REPLIES.invalidCredentials() + } + const credential = await this.checkKeyOnce(password, cacheKey, ip) + smtpRelayAuthTotal.inc({ result: '235' }) + return credential + } catch (error) { + if (error instanceof SmtpReply) { + smtpRelayAuthTotal.inc({ result: String(error.responseCode) }) + // An outage on our side must not lock anyone out. + if (!error.isTemporary) { + this.recordFailedLogin(failedLoginKey) + } + } + throw error + } + } + + // Logins that arrive together with the same key share one check, and an address can only start so many checks at + // once, so that made-up keys can't turn into unlimited calls to Django. + private checkKeyOnce(password: string, cacheKey: string, ip: string): Promise { + const running = this.keyChecks.get(cacheKey) + if (running) { + return running + } + const started = this.keyChecksPerIp.get(ip) ?? 0 + if (started >= MAX_KEY_CHECKS_PER_IP) { + return Promise.reject(REPLIES.authUnavailable()) + } + this.keyChecksPerIp.set(ip, started + 1) + const check = this.api + .checkKey(password) + .then((credential) => { + this.keyCache.set(cacheKey, { credential, expiresAt: Date.now() + KEY_CACHE_TTL_MS }) + return credential + }) + .finally(() => { + this.keyChecks.delete(cacheKey) + const remaining = (this.keyChecksPerIp.get(ip) ?? 1) - 1 + if (remaining <= 0) { + this.keyChecksPerIp.delete(ip) + } else { + this.keyChecksPerIp.set(ip, remaining) + } + }) + this.keyChecks.set(cacheKey, check) + return check + } + + private async handleData(stream: Readable & { sizeExceeded?: boolean }, session: SessionWithKey): Promise { + const credential = session.user as unknown as SmtpCredential + const { teamId } = credential + // The socket's idle timeout can't end a client that sends a byte now and then. Destroying the stream ends the + // read below, and the SMTP server sends this reply and closes the connection. + const dataDeadline = setTimeout(() => { + smtpRelayRefusedTotal.inc({ reason: 'data_deadline' }) + stream.destroy(REPLIES.dataTooSlow()) + }, this.config.SMTP_RELAY_DATA_DEADLINE_MS) + dataDeadline.unref() + const held = { bytes: 0 } + const hasSlot = this.messageSlots.tryTake(teamId, 1) + try { + const raw = await this.readOrRefuse(stream, teamId, held, hasSlot) + clearTimeout(dataDeadline) + return await this.deliverAndReply(raw, session, credential) + } finally { + clearTimeout(dataDeadline) + this.messageBytes.give(teamId, held.bytes) + if (hasSlot) { + this.messageSlots.give(teamId, 1) + } + } + } + + private async readOrRefuse( + stream: Readable & { sizeExceeded?: boolean }, + teamId: number, + held: { bytes: number }, + hasSlot: boolean + ): Promise { + try { + if (hasSlot) { + return await this.readMessage(stream, teamId, held) + } + smtpRelayRefusedTotal.inc({ reason: 'slots' }) + for await (const _chunk of stream) { + // Only read so that the client gets its reply. + } + throw REPLIES.busy() + } catch (error) { + if (error instanceof SmtpReply) { + this.observeReply(error, 0) + } + throw error + } + } + + private async deliverAndReply(raw: Buffer, session: SessionWithKey, credential: SmtpCredential): Promise { + const startedAt = performance.now() + const envelope = { + from: session.envelope.mailFrom ? session.envelope.mailFrom.address : undefined, + to: session.envelope.rcptTo.map((recipient) => recipient.address), + } + let visible: string[] = [] + try { + const parsed = await parseSmtpMessage(raw) + visible = visibleRecipients(parsed, envelope.to) + const reply = await this.deliver(raw.length, parsed, envelope, visible, credential) + this.observeReply(null, performance.now() - startedAt) + return reply + } catch (error) { + const reply = error instanceof SmtpReply ? error : REPLIES.unconfirmed() + if (!(error instanceof SmtpReply)) { + captureException(error) + } + this.observeReply(reply, performance.now() - startedAt) + this.recordInBackground(credential, { + kind: 'refused', + reply: reply.line, + from: envelope.from, + to: visible, + }) + throw reply + } + } + + private async readMessage( + stream: Readable & { sizeExceeded?: boolean }, + teamId: number, + held: { bytes: number } + ): Promise { + let chunks: Buffer[] = [] + let received = 0 + let keeping = true + for await (const chunk of stream) { + received += (chunk as Buffer).length + if (!keeping) { + // Past a limit the rest is only read so that the client gets its reply. + continue + } + if ( + received <= this.config.SMTP_RELAY_MAX_MESSAGE_BYTES && + this.messageBytes.tryTake(teamId, (chunk as Buffer).length) + ) { + held.bytes += (chunk as Buffer).length + chunks.push(chunk as Buffer) + } else { + keeping = false + chunks = [] + this.messageBytes.give(teamId, held.bytes) + held.bytes = 0 + } + } + if (stream.sizeExceeded || received > this.config.SMTP_RELAY_MAX_MESSAGE_BYTES) { + throw REPLIES.tooLarge(this.config.SMTP_RELAY_MAX_MESSAGE_BYTES) + } + if (!keeping) { + smtpRelayRefusedTotal.inc({ reason: 'bytes' }) + throw REPLIES.busy() + } + return Buffer.concat(chunks) + } + + private async deliver( + rawBytes: number, + parsed: ParsedMail, + envelope: { from?: string; to: string[] }, + visible: string[], + credential: SmtpCredential + ): Promise { + const plan = planSends(parsed, envelope) + const isLarge = rawBytes > LARGE_MESSAGE_BYTES + const budgetMs = isLarge + ? this.config.SMTP_RELAY_LARGE_MESSAGE_BUDGET_MS + : this.config.SMTP_RELAY_REPLY_BUDGET_MS + const { messageIds, refusals, temporary } = await this.sendAll( + credential, + plan, + Date.now() + budgetMs, + isLarge ? 1 : FAN_OUT_CONCURRENCY + ) + + if (plan.droppedHeaders.length > 0 && messageIds.length > 0) { + this.recordInBackground(credential, { + kind: 'dropped_headers', + from: plan.fromAddress, + to: visible, + message_id: messageIds[0], + dropped_headers: plan.droppedHeaders.slice(0, 20), + }) + } + // Every queued recipient is safe to retry when a key exists, and a whole-message reply is all SMTP clients + // handle well: a partial refusal is treated as success by Django and Nodemailer. + if (refusals.length === plan.sends.length) { + throw refusals[0] + } + if (temporary.length > 0) { + if (messageIds.length > 0) { + throw REPLIES.someUnconfirmed(messageIds.length, plan.sends.length) + } + throw temporary[0] + } + for (const refusal of refusals) { + this.recordInBackground(credential, { kind: 'refused', reply: refusal.line, from: plan.fromAddress }) + } + return `2.0.0 Ok: queued as ${messageIds.join(',')}` + } + + private async sendAll( + credential: SmtpCredential, + plan: SmtpPlan, + deadline: number, + concurrency: number + ): Promise<{ messageIds: string[]; refusals: SmtpReply[]; temporary: SmtpReply[] }> { + const messageIds: string[] = [] + const refusals: SmtpReply[] = [] + const temporary: SmtpReply[] = [] + const queue = [...plan.sends] + const worker = async (): Promise => { + for (let send = queue.shift(); send; send = queue.shift()) { + const remainingMs = deadline - Date.now() + if (remainingMs <= 0) { + temporary.push(REPLIES.unconfirmed()) + continue + } + try { + messageIds.push(await this.api.send(credential, send, remainingMs)) + } catch (error) { + const reply = error instanceof SmtpReply ? error : REPLIES.unconfirmed() + ;(reply.isTemporary ? temporary : refusals).push(reply) + } + } + } + await Promise.all(Array.from({ length: Math.min(concurrency, plan.sends.length) }, () => worker())) + return { messageIds, refusals, temporary } + } + + private observeReply(refusal: SmtpReply | null, durationMs: number): void { + const code = refusal ? refusal.responseCode : 250 + smtpRelayMessagesTotal.inc({ code: String(code) }) + smtpRelayReplyMs.observe({ class: `${Math.floor(code / 100)}xx` }, durationMs) + } + + private recordInBackground(credential: SmtpCredential, activity: SmtpActivity): void { + void this.api.recordActivity(credential.teamId, activity) + } +} diff --git a/nodejs/src/cdp/smtp-relay/smtp-replies.test.ts b/nodejs/src/cdp/smtp-relay/smtp-replies.test.ts new file mode 100644 index 000000000000..213c967b7754 --- /dev/null +++ b/nodejs/src/cdp/smtp-relay/smtp-replies.test.ts @@ -0,0 +1,36 @@ +import { SmtpReply, replyForSendApiError } from './smtp-replies' + +describe('replyForSendApiError', () => { + it.each([ + [{ status: 400, code: 'sender_not_verified', detail: 'No verified sender.' }, 550], + [{ status: 400, code: 'invalid_request', attr: 'to', detail: 'Enter an address.' }, 553], + [{ status: 400, code: 'invalid_request', attr: 'cc__0', detail: 'Enter an address.' }, 553], + [{ status: 400, code: 'invalid_request', attr: 'attachments__0__filename', detail: 'Blocked.' }, 554], + [{ status: 400, code: 'invalid_request', attr: 'attachments', detail: 'Send at most 10 MB.' }, 552], + [{ status: 413 }, 552], + [{ status: 422, code: 'recipient_suppressed' }, 550], + [{ status: 422, code: 'recipient_unreachable' }, 550], + [{ status: 422, code: 'sending_suspended' }, 554], + [{ status: 422, code: 'idempotency_key_reused' }, 554], + [{ status: 401 }, 535], + [{ status: 404 }, 550], + // A client retries a 4xx unchanged, so only what a retry can fix may be temporary. + [{ status: 409 }, 451], + [{ status: 429, code: 'throttled' }, 451], + [{ status: 429, code: 'quota_limited' }, 451], + [{ status: 503 }, 451], + [{ status: 500 }, 451], + ])('answers %j with a %i', (error, code) => { + const reply = replyForSendApiError(error) + + expect(reply.responseCode).toEqual(code) + expect(reply.message).toMatch(/^[245]\.\d\.\d \S/) + }) + + it('keeps a reply to one line of ASCII that fits the limit', () => { + const reply = new SmtpReply(550, `5.6.0 line one\r\nline two é ${'x'.repeat(500)}`) + + expect(reply.message).toMatch(/^5\.6\.0 line one line two \? x+\.\.\.$/) + expect(reply.message.length).toBeLessThanOrEqual(400) + }) +}) diff --git a/nodejs/src/cdp/smtp-relay/smtp-replies.ts b/nodejs/src/cdp/smtp-relay/smtp-replies.ts new file mode 100644 index 000000000000..5c9d2210a756 --- /dev/null +++ b/nodejs/src/cdp/smtp-relay/smtp-replies.ts @@ -0,0 +1,135 @@ +export const SMTP_DOCS_URL = 'https://posthog.com/docs/workflows/smtp' + +export class SmtpReply extends Error { + constructor( + public readonly responseCode: number, + message: string + ) { + super(sanitizeReplyText(message)) + } + + public get isTemporary(): boolean { + return this.responseCode < 500 + } + + /** The whole line as a client sees it. */ + public get line(): string { + return `${this.responseCode} ${this.message}` + } +} + +// A reply is one line of ASCII, and Go's textproto rejects lines under four characters. +const MAX_REPLY_TEXT_LENGTH = 400 + +function sanitizeReplyText(text: string): string { + const oneLine = text + .replace(/[\r\n]+/g, ' ') + .replace(/[^\x20-\x7E]/g, '?') + .trim() + return oneLine.length > MAX_REPLY_TEXT_LENGTH ? `${oneLine.slice(0, MAX_REPLY_TEXT_LENGTH - 3)}...` : oneLine +} + +export const REPLIES = { + tooManyConnections: () => new SmtpReply(421, '4.7.0 Too many connections from this address. Try again shortly.'), + tooManyClients: () => new SmtpReply(421, '4.3.2 PostHog is busy. Try again shortly.'), + tooManyFailedLogins: () => + new SmtpReply( + 454, + '4.7.0 Too many failed logins with this key from this address. Check your key and try again in 5 minutes.' + ), + dataTooSlow: () => new SmtpReply(421, '4.4.2 This message took too long to arrive. Reconnect and retry.'), + invalidCredentials: () => + new SmtpReply(535, `5.7.8 Invalid credentials. Use a project secret key as the password. ${SMTP_DOCS_URL}`), + missingScope: () => + new SmtpReply(535, `5.7.8 This key needs the transactional_email:write scope. ${SMTP_DOCS_URL}`), + notEnabled: () => new SmtpReply(535, '5.7.8 SMTP sending is not turned on for this project.'), + authUnavailable: () => new SmtpReply(454, '4.7.0 Temporary authentication failure. Try again shortly.'), + busy: () => new SmtpReply(451, '4.3.2 PostHog is busy. Retry shortly.'), + tooLarge: (limitBytes: number) => + new SmtpReply( + 552, + `5.3.4 Message too large. The limit is ${Math.floor(limitBytes / (1024 * 1024))} MB. Link to large files instead of attaching them.` + ), + unreadable: () => new SmtpReply(550, '5.6.0 PostHog could not read this message. Check that it is valid MIME.'), + tooManyRecipients: (limit: number) => + new SmtpReply(554, `5.5.3 Too many recipients. The limit is ${limit} per message.`), + signed: () => + new SmtpReply(554, '5.7.1 PostHog rewrites messages before sending, so S/MIME signed mail cannot be sent.'), + noSender: () => new SmtpReply(550, '5.1.7 This message has no From header. Add one on a verified domain.'), + severalSenders: () => new SmtpReply(550, '5.1.7 Use one address in the From header.'), + unconfirmed: () => + new SmtpReply( + 451, + "4.3.0 PostHog couldn't confirm this send, so it may still go out. Retry with the same idempotency key." + ), + someUnconfirmed: (queued: number, total: number) => + new SmtpReply( + 451, + `4.3.0 PostHog queued ${queued} of ${total} recipients and couldn't confirm the rest. Retry to send to the rest.` + ), +} + +export type SendApiError = { + status: number + code?: string + detail?: string + attr?: string +} + +export function replyForSendApiError({ status, code, detail, attr }: SendApiError): SmtpReply { + if (status === 401 || status === 403) { + return new SmtpReply(535, `5.7.8 This key can no longer send email. ${SMTP_DOCS_URL}`) + } + if (status === 404) { + return new SmtpReply(550, '5.7.1 Sending email from code is not turned on for this project.') + } + if (status === 409) { + return new SmtpReply(451, '4.3.0 An earlier send with this key is still processing. Retry shortly.') + } + if (status === 413) { + return new SmtpReply(552, `5.3.4 This email is larger than PostHog can send. ${SMTP_DOCS_URL}`) + } + if (status === 429) { + return code === 'quota_limited' + ? new SmtpReply(451, '4.7.1 This project is over its email quota. Check your billing settings.') + : new SmtpReply(451, '4.7.1 Rate limit reached. Retry shortly.') + } + switch (code) { + case 'recipient_suppressed': + return new SmtpReply( + 550, + '5.7.1 A recipient is suppressed after a hard bounce or complaint. Remove it from the suppression list if it is valid.' + ) + case 'recipient_unreachable': + return new SmtpReply(550, '5.1.2 A recipient cannot receive email. Check for a typo.') + case 'sending_suspended': + case 'sending_paused': + return new SmtpReply(554, '5.7.1 Sending is suspended or paused for this project. See the Reputation tab.') + case 'idempotency_key_reused': + return new SmtpReply(554, '5.7.0 This idempotency key was used for a different email. Use a new key.') + case 'sender_not_verified': + case 'sender_required': + case 'sender_unavailable': + return new SmtpReply(550, `5.7.1 ${detail ?? 'The From address is not a verified sender on this project.'}`) + } + if (status === 400) { + const field = attr?.split('__')[0] + if (field === 'to' || field === 'cc' || field === 'bcc') { + return new SmtpReply( + 553, + "5.1.3 PostHog can't send to this address. Quoted names, IP addresses and non-ASCII domains aren't supported." + ) + } + if (field === 'from') { + return new SmtpReply(550, `5.1.7 The From header is not valid. ${detail ?? ''}`) + } + if (field === 'attachments' && detail?.includes('MB')) { + return new SmtpReply(552, `5.3.4 ${detail} Link to large files instead of attaching them.`) + } + if (field === 'attachments') { + return new SmtpReply(554, `5.7.1 ${detail ?? 'PostHog cannot send one of the attachments.'}`) + } + return new SmtpReply(550, `5.6.0 ${detail ?? 'PostHog could not accept this message.'}`) + } + return REPLIES.unconfirmed() +} diff --git a/nodejs/src/cdp/smtp-relay/smtp-send-api.ts b/nodejs/src/cdp/smtp-relay/smtp-send-api.ts new file mode 100644 index 000000000000..13bf2fa40be3 --- /dev/null +++ b/nodejs/src/cdp/smtp-relay/smtp-send-api.ts @@ -0,0 +1,126 @@ +import { PosthogJwtAudience } from '~/cdp/utils/jwt-utils' +import { ScopedServiceJwt } from '~/cdp/utils/scoped-service-jwt' +import { parseJSON } from '~/common/utils/json-parse' +import { logger } from '~/common/utils/logger' +import { internalFetch } from '~/common/utils/request' + +import type { SmtpSend } from './smtp-message' +import { REPLIES, replyForSendApiError } from './smtp-replies' + +export type SmtpCredential = { apiKey: string; teamId: number } + +export type SmtpActivity = { + kind: 'refused' | 'dropped_headers' + reply?: string + from?: string + to?: string[] + message_id?: string + dropped_headers?: string[] +} + +const AUTH_TIMEOUT_MS = 3_000 +const ACTIVITY_TIMEOUT_MS = 3_000 +const ACTIVITY_TOKEN_TTL_SECONDS = 60 + +const responseJson = (text: string): Record => { + try { + const parsed = parseJSON(text) + return parsed && typeof parsed === 'object' ? parsed : {} + } catch { + return {} + } +} + +/** + * The relay's way into PostHog. Sends use the same API a customer's code calls, with the customer's own key. The list + * of recent problems takes the relay's own token instead, so a customer's key can't add entries to it. + */ +export class SmtpSendApi { + private readonly activityJwt: ScopedServiceJwt + + constructor( + private readonly apiUrl: string, + activitySigningKeys: string + ) { + this.activityJwt = new ScopedServiceJwt(PosthogJwtAudience.SMTP_RELAY_ACTIVITY, activitySigningKeys) + } + + public async checkKey(apiKey: string): Promise { + let response + try { + response = await internalFetch(`${this.apiUrl}/api/projects/@current/messaging/send/smtp_check/`, { + method: 'POST', + headers: { Authorization: `Bearer ${apiKey}`, 'Content-Type': 'application/json' }, + body: '{}', + timeoutMs: AUTH_TIMEOUT_MS, + }) + } catch { + throw REPLIES.authUnavailable() + } + const body = responseJson(await response.text()) + if (response.status === 200 && typeof body.team_id === 'number') { + return { apiKey, teamId: body.team_id } + } + if (response.status === 401) { + throw REPLIES.invalidCredentials() + } + if (response.status === 403) { + throw REPLIES.missingScope() + } + if (response.status === 404) { + throw REPLIES.notEnabled() + } + throw REPLIES.authUnavailable() + } + + /** Resolves to the queued message's ID. Throws the reply the client should get. */ + public async send(credential: SmtpCredential, send: SmtpSend, timeoutMs: number): Promise { + let response + try { + response = await internalFetch(`${this.apiUrl}/api/projects/${credential.teamId}/messaging/send/email/`, { + method: 'POST', + headers: { + Authorization: `Bearer ${credential.apiKey}`, + 'Content-Type': 'application/json', + 'X-PostHog-Send-Source': 'smtp', + ...(send.idempotencyKey && { 'Idempotency-Key': send.idempotencyKey }), + }, + body: JSON.stringify(send.body), + timeoutMs, + }) + } catch { + throw REPLIES.unconfirmed() + } + const body = responseJson(await response.text()) + if (response.status === 202) { + return String(body.message_id) + } + throw replyForSendApiError({ + status: response.status, + code: typeof body.code === 'string' ? body.code : undefined, + detail: typeof body.detail === 'string' ? body.detail : undefined, + attr: typeof body.attr === 'string' ? body.attr : undefined, + }) + } + + /** Best effort, so that a failure here never changes what the client is told. */ + public async recordActivity(teamId: number, activity: SmtpActivity): Promise { + if (!this.activityJwt.enabled) { + return + } + try { + const token = this.activityJwt.mint({ team_id: teamId }, ACTIVITY_TOKEN_TTL_SECONDS) + await internalFetch(`${this.apiUrl}/api/projects/${teamId}/internal/messaging/smtp_activity`, { + method: 'POST', + headers: { Authorization: `Bearer ${token}`, 'Content-Type': 'application/json' }, + body: JSON.stringify(activity), + timeoutMs: ACTIVITY_TIMEOUT_MS, + }) + } catch (error) { + logger.warn('⚠️', 'SMTP relay could not record activity', { + team_id: teamId, + error: error instanceof Error ? error.message : String(error), + }) + } + } +} diff --git a/nodejs/src/cdp/utils/jwt-utils.ts b/nodejs/src/cdp/utils/jwt-utils.ts index f38f0d58272b..e4b75d0ce923 100644 --- a/nodejs/src/cdp/utils/jwt-utils.ts +++ b/nodejs/src/cdp/utils/jwt-utils.ts @@ -23,6 +23,8 @@ export enum PosthogJwtAudience { WORKFLOW_MESSAGE_OPT_OUT = 'posthog:workflows:message_opt_out', // Must match PosthogJwtAudience.CONVERSATIONS_TICKETS in posthog/jwt.py exactly. CONVERSATIONS_TICKETS = 'posthog:conversations:tickets', + // Must match PosthogJwtAudience.SMTP_RELAY_ACTIVITY in posthog/jwt.py exactly. + SMTP_RELAY_ACTIVITY = 'posthog:workflows:smtp_relay_activity', // Must match PosthogJwtAudience.CUSTOMER_ANALYTICS_ACCOUNTS in posthog/jwt.py exactly. CUSTOMER_ANALYTICS_ACCOUNTS = 'posthog:customer_analytics:accounts', } diff --git a/nodejs/src/common/config.ts b/nodejs/src/common/config.ts index 41c2566c2495..2cbbdd1117aa 100644 --- a/nodejs/src/common/config.ts +++ b/nodejs/src/common/config.ts @@ -59,6 +59,7 @@ export enum PluginServerMode { cdp_cyclotron_worker_hogflow = 'cdp-cyclotron-worker-hogflow', cdp_cyclotron_worker_email = 'cdp-cyclotron-worker-email', cdp_api = 'cdp-api', + cdp_smtp_relay = 'cdp-smtp-relay', cdp_legacy_on_event = 'cdp-legacy-on-event', evaluation_scheduler = 'evaluation-scheduler', ingestion_logs = 'ingestion-logs', diff --git a/nodejs/src/server.ts b/nodejs/src/server.ts index 90a4b72cac92..a5cf77bc5587 100644 --- a/nodejs/src/server.ts +++ b/nodejs/src/server.ts @@ -55,6 +55,7 @@ import { hasEmailSigningKey } from './cdp/services/messaging/helpers/tracking-co import { HogInvocationResultsService } from './cdp/services/monitoring/hog-invocation-results.service' import { createSesRateLimiterValkeyPool } from './cdp/services/rate-limiter/rate-limiter-valkey-pool' import { RateLimiterService } from './cdp/services/rate-limiter/rate-limiter.service' +import { SmtpRelayService } from './cdp/smtp-relay/smtp-relay.service' import { EncryptedFields } from './cdp/utils/encryption-utils' import { CleanupResources, NodeServer, ServerLifecycle } from './servers/base-server' import { PluginServerService, PluginsServerConfig, RedisPool } from './types' @@ -95,6 +96,14 @@ export class PluginServer implements NodeServer { const capabilities = getPluginServerCapabilities(this.config) + // The relay's network policy only reaches Django, so waiting for Redis or Postgres would stop it from starting. + if (capabilities.cdpSmtpRelay) { + const relay = new SmtpRelayService(this.config) + await relay.start() + this.lifecycle.services.push(relay.service) + return + } + const needsCdp = !!( capabilities.cdpProcessedEvents || capabilities.cdpDataWarehouseEvents || diff --git a/nodejs/src/types.ts b/nodejs/src/types.ts index 9678003d31e2..eeba1dfc2052 100644 --- a/nodejs/src/types.ts +++ b/nodejs/src/types.ts @@ -152,6 +152,7 @@ export interface PluginServerCapabilities { cdpCyclotronV2Janitor?: boolean cdpRerunWorker?: boolean cdpDlqReplay?: boolean + cdpSmtpRelay?: boolean cdpHogflowScheduler?: boolean cdpHogflowSubscriptionMatcher?: boolean recordingApi?: boolean diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 12fed851602b..9f8db9f3689a 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1792,6 +1792,9 @@ importers: lz4: specifier: ^0.6.5 version: 0.6.5 + mailparser: + specifier: ^3.14.1 + version: 3.14.1 node-fetch: specifier: ^2.6.1 version: 2.6.9(encoding@0.1.13) @@ -1831,6 +1834,9 @@ importers: re2: specifier: ^1.26.1 version: 1.27.0 + smtp-server: + specifier: ^3.21.3 + version: 3.21.3 snappy: specifier: ^7.2.2 version: 7.2.2 @@ -1928,6 +1934,9 @@ importers: '@types/luxon': specifier: ^3.4.2 version: 3.4.2 + '@types/mailparser': + specifier: ^3.9.0 + version: 3.9.0 '@types/node-fetch': specifier: ^2.5.10 version: 2.6.4 @@ -1943,6 +1952,9 @@ importers: '@types/redlock': specifier: ^4.0.1 version: 4.0.4 + '@types/smtp-server': + specifier: ^3.5.13 + version: 3.5.13 '@types/snowflake-sdk': specifier: ^1.5.1 version: 1.6.12 @@ -12346,6 +12358,11 @@ packages: resolution: {integrity: sha512-y2JVHeQsRADEWRLNXV5uf8BAP5PowYU5IBxsSiJY9WCIDS3/RT98g6fFRURefKn+9Dalmp7X7pcHUX5wW2ljYw==} hasBin: true + '@selderee/plugin-htmlparser2@0.12.0': + resolution: {integrity: sha512-oELmoyA6ML9jDRMV3kgcMQFKxUfBU0yFVn6yTctVaLT5ygXnxH52I3TZEgV9EhXJC68/uFvE5Daj1/25c0Xa/A==} + peerDependencies: + selderee: ~0.12.0 + '@shadcn/react@0.3.0': resolution: {integrity: sha512-iKN0NuYe850VDHlxEfvppFrpa7CMV3zArTHusXlaSmeFeQhohGbIqclv6WwPTs/44I8EkXQpcuRt6sXEVKkWqQ==} peerDependencies: @@ -13965,6 +13982,9 @@ packages: '@types/luxon@3.4.2': resolution: {integrity: sha512-TifLZlFudklWlMBfhubvgqTXRzLDI5pCbGa4P8a3wPyUQSW+1xQ5eDsreP9DWHX3tjq1ke96uYG/nwundroWcA==} + '@types/mailparser@3.9.0': + resolution: {integrity: sha512-YnhrRiRVDiEAphFvd72hKa1n8y2b9Q3rgSQiONZQyhQz67h5g4MZHNApET3XWXVEFdInWtHqDHKEwnyjxU3QVw==} + '@types/mapbox__point-geometry@0.1.3': resolution: {integrity: sha512-2W46IOXlu7vC8m3+M5rDqSnuY22GFxxx3xhkoyqyPWrD+eP2iAwNst0A1+umLYjCTJMJTSpiofphn9h9k+Kw+w==} @@ -14128,6 +14148,9 @@ packages: '@types/set-cookie-parser@2.4.10': resolution: {integrity: sha512-GGmQVGpQWUe5qglJozEjZV/5dyxbOOZ0LHe/lqyWssB88Y4svNfst0uqBVscdDeIKl5Jy5+aPSvy7mI9tYRguw==} + '@types/smtp-server@3.5.13': + resolution: {integrity: sha512-S3rGl2KbViH+98/CgHipPIWgtAFnjxLsppxIRbgJHNuZL7Y+py+7kZjT7xS+wOmCxYTn4O7IqLqkvekg99MDVQ==} + '@types/snowflake-sdk@1.6.12': resolution: {integrity: sha512-Ndz6x750TkuvHvdRBH8Cb7T8et2anyyY1j8kFChJ+s8FtURNRKut7DWaq9YdseKXZvqwA3ZYZxKaQRwtSq67eg==} @@ -14693,6 +14716,10 @@ packages: deprecated: v1 is no longer supported. Please update ZenFS (or whatever ZenFS dependent you depend on). hasBin: true + '@zone-eu/mailsplit@5.4.29': + resolution: {integrity: sha512-CCUNycmUjrV5M0wNf23waWAt9eOHczhjLbUQS7k0FNJvR0py9VSqHFt+tSl/D0+XI73VLG0U+PwYtq9R0zsm9Q==} + engines: {node: '>=20.0.0'} + abab@2.0.6: resolution: {integrity: sha512-j2afSsaIENvHZN2B8GOpF566vZ5WVk5opAiMTvWgaQT8DkbOqsTfvNAvHoRGU2zzP8cPoqys+xHTRDWW8L+/BA==} deprecated: Use your platform's native atob() and btoa() methods instead @@ -16267,6 +16294,10 @@ packages: deep-is@0.1.4: resolution: {integrity: sha512-oIPzksmTg4/MriiaYGO+okXDT7ztn/w3Eptv/+gSIdMdKsJo0u4CfYNFJPy+4SKMuCqGw2wxnA+URMg3t8a/bQ==} + deepmerge-ts@8.0.2: + resolution: {integrity: sha512-uqbvqLUMrc6p0MO+WBRtTxY55hmyh94WRwI5a++PZe54X+bfVh59FSN7uWCBCW1CCVjzjnrwzfI8zidE2obMMw==} + engines: {node: '>=16.9.0'} + deepmerge@4.3.1: resolution: {integrity: sha512-3sUqbMEc77XqpdNO7FRyRog+eW3ph+GYCbj+rK+uYyRMuwsVy0rMiVtPn+QJlKFvWP/1PYpapqYn0Me2knFn+A==} engines: {node: '>=0.10.0'} @@ -16552,6 +16583,10 @@ packages: resolution: {integrity: sha512-Q0n9HRi4m6JuGIV1eFlmvJB7ZEVxu93IrMyiMsGC0lrMJMWzRgx6WGquyfQgZVb31vhGgXnfmPNNXmxnOkRBrg==} engines: {node: '>= 0.8'} + encoding-japanese@2.4.0: + resolution: {integrity: sha512-PE0KjlsDy9JSsoU4C2Q8NXb1kC4c+bZRac9ZvHXUE4q1Ht63gScKMdEy9jst9+Fi3M4eDkqzkEhBZvix4blP2A==} + engines: {node: '>=18.0.0'} + encoding-sniffer@0.2.0: resolution: {integrity: sha512-ju7Wq1kg04I3HtiYIOrUrdfdDvkyO9s5XM8QAj/bN61Yo/Vb4vgJxy5vi4Yxk01gWHbrofpPtpxM8bKger9jhg==} @@ -17704,12 +17739,19 @@ packages: html-to-image@1.11.13: resolution: {integrity: sha512-cuOPoI7WApyhBElTTb9oqsawRvZ0rHhaHwghRLlTuffoD1B2aDemlCruLeZrUIIdvG7gs9xeELEPm6PhuASqrg==} + html-to-text@10.0.1: + resolution: {integrity: sha512-GiVhRI1BatGARSCmlXWNCjDT0cWrwBWoeduLoV0WSKAgaV/wa+hUWy5LiQLUs4UwiUrE52ZCMfBGiKD87TDPrg==} + engines: {node: '>=20.19.0'} + html-url-attributes@3.0.1: resolution: {integrity: sha512-ol6UPyBWqsrO6EJySPz2O7ZSr856WDrEzM5zMqp+FJJLGMW35cLYmmZnl0vztAZxRUoNZJFTCohfjuIJ8I4QBQ==} html-void-elements@3.0.0: resolution: {integrity: sha512-bEqo66MRXsUGxWHV5IP0PUiAWwoEjba4VCzg0LjFJBpchPaTfyfCKTG6bc5F8ucKec3q5y6qOdGyYTSBEvhCrg==} + htmlparser2@10.1.0: + resolution: {integrity: sha512-VTZkM9GWRAtEpveh7MSF6SjjrpNVNNVJfFup7xTY3UpFtm67foy9HDVXneLtFVt4pMz5kZtgNcvCniNFb1hlEQ==} + htmlparser2@3.10.1: resolution: {integrity: sha512-IgieNijUMbkDovyoKObU1DUhm1iwNYE/fuifEoEHfd1oZKZDaONBSkal7Y01shxsM49R4XaMdGez3WnF9UfiCQ==} @@ -17796,6 +17838,10 @@ packages: resolution: {integrity: sha512-im9DjEDQ55s9fL4EYzOAv0yMqmMBSZp6G0VvFyTMPKWxiSBHUj9NW/qqLmXUwXrrM7AvqSlTCfvqRb0cM8yYqw==} engines: {node: '>=0.10.0'} + iconv-lite@0.7.3: + resolution: {integrity: sha512-IKXpvIzjnC9XTAUbVBcMfGS0EPaIXtW6v+zr+RRp+hqULEpo0owZax6wyRwPOJbWbzjYspQwusTsfVr0ifh4uQ==} + engines: {node: '>=0.10.0'} + ieee754@1.1.13: resolution: {integrity: sha512-4vf7I2LYV/HaWerSo3XmlMkp5eZ83i+/CDluXi/IGTs/O1sejBNhTtnxzmRZfvOUqj7lZjqHkeTvpgSFDlWZTg==} @@ -17914,6 +17960,9 @@ packages: resolution: {integrity: sha512-Ag3wB2o37wslZS19hZqorUnrnzSkpOVy+IiiDEiTqNubEYpYuHWIf6K4psgN2ZWKExS4xhVCrRVfb/wfW8fWJA==} engines: {node: '>= 10'} + ipv6-normalize@1.0.1: + resolution: {integrity: sha512-Bm6H79i01DjgGTCWjUuCjJ6QDo1HB96PT/xCYuyJUP9WFbVDrLSbG4EZCvOCun2rNswZb0c3e4Jt/ws795esHA==} + is-alphabetical@2.0.1: resolution: {integrity: sha512-FWyyY60MeTNyeSRpkM2Iry0G9hpr7/9kD40mD/cGQEuilcZYS4okz8SN2Q6rLCJ8gbCt6fN+rC+6tMGS99LaxQ==} @@ -18964,6 +19013,9 @@ packages: layout-base@2.0.1: resolution: {integrity: sha512-dp3s92+uNI1hWIpPGH3jK2kxE2lMjdXdr+DH8ynZHpd6PUlH6x6cbuXnoMmiNumznqaNO31xu9e79F0uuZ0JFg==} + leac@0.7.0: + resolution: {integrity: sha512-qMrZeyEekgdRQ9o6a4NAB2EQZrv827GJdn1vnapwSJ90hWRB4TzUSunvacPkxQ2TnNqHNI1/zSt0hlo0crG8Jw==} + less@3.13.1: resolution: {integrity: sha512-SwA1aQXGUvp+P5XdZslUOhhLnClSLIjWvJhmd+Vgib5BFIr9lMNlQwmwUNOjXThF/A0x+MCYYPeWEfeWiLRnTw==} engines: {node: '>=6'} @@ -18986,6 +19038,18 @@ packages: resolution: {integrity: sha512-+bT2uH4E5LGE7h/n3evcS/sQlJXCpIp6ym8OWJ5eV6+67Dsql/LaaT7qJBAt2rzfoa/5QBGBhxDix1dMt2kQKQ==} engines: {node: '>= 0.8.0'} + libbase64@1.4.1: + resolution: {integrity: sha512-Nyh3H90B3gRtrcxIe56Ayy7635J0jmh6AbxbQgQae8wrSPQMH5wFqJChiMTB13yKG512L3YspbA5DzumhFTITA==} + engines: {node: '>=20.0.0'} + + libmime@5.6.3: + resolution: {integrity: sha512-cuTUW/Jybvh5O4/U9p61gyutHOjDtZyGQYuY99+bfnaFwqcDf2Nb+R6hMHY6DDen2H9wLKQSMpofWUgn2TI5JQ==} + engines: {node: '>=20.0.0'} + + libqp@3.0.1: + resolution: {integrity: sha512-7QZxpPX25pga1WiMueduWxmrHMVBFrZPH6ZUaozGYa/H98JH6NVBSET3qVw+4NxzoraIW7rJ81YxSU9IalvCpg==} + engines: {node: '>=20.0.0'} + libsodium-wrappers@0.7.16: resolution: {integrity: sha512-Gtr/WBx4dKjvRL1pvfwZqu7gO6AfrQ0u9vFL+kXihtHf6NfkROR8pjYWn98MFDI3jN19Ii1ZUfPR9afGiPyfHg==} @@ -19150,6 +19214,9 @@ packages: linkify-it@5.0.0: resolution: {integrity: sha512-5aHCbzQRADcdP+ATqnDuhhJ/MRIqDkZX5pyjFHRRysS8vZ5AbqGEoFIb6pYHPZ+L/OC2Lc+xT8uHVVR5CAK/wQ==} + linkify-it@5.0.2: + resolution: {integrity: sha512-ONTm2jCMAVZjgQa/Fy1kScXsuOoF5NPTsoFBdE1KVIZ2vAh/r9+Bqo+0jINCBYnavTPQZz38QzFTme79ENoN3Q==} + linkifyjs@4.3.2: resolution: {integrity: sha512-NT1CJtq3hHIreOianA8aSXn6Cw0JzYOuDQbOrSPe7gqFnCpKP++MQe3ODgO3oh2GJFORkAAdqredOa60z63GbA==} @@ -19347,6 +19414,10 @@ packages: magic-string@0.30.21: resolution: {integrity: sha512-vd2F4YUyEXKGcLHoq+TEyCjxueSeHnFxyyjNp80yg0XV4vUhnDer/lvvlqM/arB5bXQN5K2/3oinyCRyx8T2CQ==} + mailparser@3.14.1: + resolution: {integrity: sha512-Z0LLk7F0yhSCx3xjyL8BMm6dXO3vuWrSAeVhm1F18wVGVBbLAKOsVsCjPyOhJy+KuyDW9PRyDuyiPb7yJCHwVA==} + engines: {node: '>=20.0.0'} + make-dir@2.1.0: resolution: {integrity: sha512-LS9X+dc8KLxXCb8dni79fLIIUA5VyZoyjSMCwTluaXA0o27cCK0bhXkpgw+sTXVpPy/lSO57ilRixqk0vDmtRA==} engines: {node: '>=6'} @@ -20007,6 +20078,10 @@ packages: resolution: {integrity: sha512-OXdegQq03OmXEjt2hZP33W2YPs/E5BcFQks46+G2gAxs4gHOIVD1u7EqlYLYSKsaIpyKCK9Gbk0ta1/gjRSMRQ==} engines: {node: '>=6'} + nodemailer@10.3.1: + resolution: {integrity: sha512-uF0AFsJC07Y1vdW7p580DKAnh7rAO+4UPSSB3mDCPNDufoiaOebV6JmrzpxB14OjEEEiBsNdYoM2Y+fN8M7BBA==} + engines: {node: '>=20.0.0'} + nodemailer@8.0.8: resolution: {integrity: sha512-p+XsnzXGdtIHXUu2ugxdfG+eX2nehsGhMjW9h0CWj1BhE30hrFz0kh0yIM0/VjUgVsRrDj+80ZO+I1nSkGE4tA==} engines: {node: '>=6.0.0'} @@ -20341,6 +20416,9 @@ packages: parse5@7.1.2: resolution: {integrity: sha512-Czj1WaSVpaoj0wbhMzLmWD69anp2WH7FXMB9n1Sy8/ZFF9jolSQVMu1Ij5WIyGmcBmhk7EOndpO4mIpihVqAXw==} + parseley@0.13.1: + resolution: {integrity: sha512-uNBJZzmb60l6p6VWLTmevizNAGnE0xoSf1n0B4q3ntegDNzcS68NRCcBDZTcyXHxt2XhBChsCuqj4M+nChvE/A==} + parseurl@1.3.3: resolution: {integrity: sha512-CiyeOxFT/JZyN5m0z9PfXw4SCBJ6Sygz1Dpl0wqjlhDEGGBP1GnsUVEL0p63hoG1fcj3fHynXi9NYO4nWOL+qQ==} engines: {node: '>= 0.8'} @@ -20421,6 +20499,9 @@ packages: resolution: {integrity: sha512-Q3CG/cYvCO1ye4QKkuH7EXxs3VC/rI1/trd+qX2+PolbaKG0H+bgcZzrTt96mMyRtejk+JMCiLUn3y29W8qmFQ==} engines: {node: '>= 0.10'} + peberminta@0.10.0: + resolution: {integrity: sha512-80B2AsU+I4Qdb0ZAPSfe9UwvGzwkM37IKIFEvdS3D/3Ndgv2bsuJ0bfG1+iEYO+l7Gfd4EUJmuRyq7efLgRMzQ==} + pend@1.2.0: resolution: {integrity: sha512-F3asv42UuXchdzt+xXqfW1OGlVBe+mxa2mqI0pg5yAHZPvFmY3Y6drSf/GQ1A86WgWEN9Kzh/WrgKa6iGcHXLg==} @@ -22166,6 +22247,9 @@ packages: secure-json-parse@2.7.0: resolution: {integrity: sha512-6aU+Rwsezw7VR8/nyvKTx8QpWH9FrcYiXXlqC4z5d5XQBDRqtbfsRjnwGyqbi3gddNtWHuEk9OANUotL26qKUw==} + selderee@0.12.0: + resolution: {integrity: sha512-b1YMh3+DHZp59DLna3qVwQ5iOla/nrI6mLBNW02XxU77M3046Df6VLkoaJyFz20VsGIG5kkp+FK0kg4K4HnUFw==} + semver@5.7.2: resolution: {integrity: sha512-cBznnQ9KjJqU67B52RMC65CMarK2600WFnbkcaiwWq3xy/5haFJlshgnpjovMVJ+Hff49d8GEn0b87C5pDQ10g==} hasBin: true @@ -22325,6 +22409,10 @@ packages: resolution: {integrity: sha512-94hK0Hh8rPqQl2xXc3HsaBoOXKV20MToPkcXvwbISWLEs+64sBq5kFgn2kJDHb1Pry9yrP0dxrCI9RRci7RXKg==} engines: {node: '>= 6.0.0', npm: '>= 3.0.0'} + smtp-server@3.21.3: + resolution: {integrity: sha512-RJRuhJWt+Zs8/c9Pa1Qrzz7RO49iRxrdj+7CXjz+zwRZLVliYyM/vrqF4VxsFH67eLCNW9kM3o5K9tHvlPuiIw==} + engines: {node: '>=20.0.0'} + snappy-wasm@0.3.0: resolution: {integrity: sha512-mT2q7y2K0krf2WdsfR7YZL/vJ+BiTKcvgxIA/gJozPQibWJwV8VOl6TNZLjeUlGH2bqee/570fS/LUkO406Ixg==} @@ -23060,6 +23148,10 @@ packages: resolution: {integrity: sha512-azl+t0z7pw/z958Gy9svOTuzqIk6xq+NSheJzn5MMWtWTFywIacg2wUlzKFGtt3cthx0r2SxMK0yzJOR0IES7Q==} engines: {node: '>=14.0.0'} + tlds@1.261.0: + resolution: {integrity: sha512-QXqwfEl9ddlGBaRFXIvNKK6OhipSiLXuRuLJX5DErz0o0Q0rYxulWLdFryTkV5PkdZct5iMInwYEGe/eR++1AA==} + hasBin: true + tldts-core@6.1.57: resolution: {integrity: sha512-lXnRhuQpx3zU9EONF9F7HfcRLvN1uRYUBIiKL+C/gehC/77XTU+Jye6ui86GA3rU6FjlJ0triD1Tkjt2F/2lEg==} @@ -27967,7 +28059,7 @@ snapshots: '@jest/console@30.0.5': dependencies: '@jest/types': 30.0.5 - '@types/node': 22.18.8 + '@types/node': 26.6.5 chalk: 4.1.2 jest-message-util: 30.0.5 jest-util: 30.0.5 @@ -28109,7 +28201,7 @@ snapshots: dependencies: '@jest/fake-timers': 30.0.5 '@jest/types': 30.0.5 - '@types/node': 22.18.8 + '@types/node': 26.6.5 jest-mock: 30.0.5 '@jest/expect-utils@28.1.3': @@ -31780,6 +31872,12 @@ snapshots: '@segment/fql-ts@1.10.1': {} + '@selderee/plugin-htmlparser2@0.12.0(selderee@0.12.0)': + dependencies: + domelementtype: 2.3.0 + domhandler: 5.0.3 + selderee: 0.12.0 + '@shadcn/react@0.3.0(@types/react@18.3.27)(react@18.3.1)': optionalDependencies: '@types/react': 18.3.27 @@ -33714,6 +33812,11 @@ snapshots: '@types/luxon@3.4.2': {} + '@types/mailparser@3.9.0': + dependencies: + '@types/node': 26.6.5 + iconv-lite: 0.6.3 + '@types/mapbox__point-geometry@0.1.3': {} '@types/mapbox__vector-tile@1.3.3': @@ -33905,6 +34008,11 @@ snapshots: dependencies: '@types/node': 26.6.5 + '@types/smtp-server@3.5.13': + dependencies: + '@types/node': 26.6.5 + '@types/nodemailer': 8.0.0 + '@types/snowflake-sdk@1.6.12': dependencies: '@types/node': 26.6.5 @@ -34608,6 +34716,12 @@ snapshots: readable-stream: 4.7.0 utilium: 1.10.1 + '@zone-eu/mailsplit@5.4.29': + dependencies: + libbase64: 1.4.1 + libmime: 5.6.3 + libqp: 3.0.1 + abab@2.0.6: {} abbrev@5.0.0: {} @@ -36460,6 +36574,8 @@ snapshots: deep-is@0.1.4: {} + deepmerge-ts@8.0.2: {} + deepmerge@4.3.1: {} default-browser-id@5.0.1: {} @@ -36712,6 +36828,8 @@ snapshots: encodeurl@2.0.0: {} + encoding-japanese@2.4.0: {} + encoding-sniffer@0.2.0: dependencies: iconv-lite: 0.6.3 @@ -38300,10 +38418,25 @@ snapshots: html-to-image@1.11.13(patch_hash=f0a5786708aab63eb75c89a2c8de10a8da63fce02a5d9ab588fbe684cab8951e): {} + html-to-text@10.0.1: + dependencies: + '@selderee/plugin-htmlparser2': 0.12.0(selderee@0.12.0) + deepmerge-ts: 8.0.2 + dom-serializer: 2.0.0 + htmlparser2: 10.1.0 + selderee: 0.12.0 + html-url-attributes@3.0.1: {} html-void-elements@3.0.0: {} + htmlparser2@10.1.0: + dependencies: + domelementtype: 2.3.0 + domhandler: 5.0.3 + domutils: 3.2.2 + entities: 7.0.1 + htmlparser2@3.10.1: dependencies: domelementtype: 1.3.1 @@ -38433,6 +38566,10 @@ snapshots: dependencies: safer-buffer: 2.1.2 + iconv-lite@0.7.3: + dependencies: + safer-buffer: 2.1.2 + ieee754@1.1.13: {} ieee754@1.2.1: {} @@ -38545,6 +38682,8 @@ snapshots: ipaddr.js@2.2.0: {} + ipv6-normalize@1.0.1: {} + is-alphabetical@2.0.1: {} is-alphanumerical@2.0.1: @@ -39307,7 +39446,7 @@ snapshots: jest-haste-map@30.0.5: dependencies: '@jest/types': 30.0.5 - '@types/node': 22.18.8 + '@types/node': 26.6.5 anymatch: 3.1.3 fb-watchman: 2.0.2 graceful-fs: 4.2.11 @@ -39486,7 +39625,7 @@ snapshots: jest-mock@30.0.5: dependencies: '@jest/types': 30.0.5 - '@types/node': 22.18.8 + '@types/node': 26.6.5 jest-util: 30.0.5 jest-pnp-resolver@1.2.3(jest-resolve@27.5.1): @@ -39728,7 +39867,7 @@ snapshots: '@jest/test-result': 30.0.5 '@jest/transform': 30.0.5(supports-color@10.2.2) '@jest/types': 30.0.5 - '@types/node': 22.18.8 + '@types/node': 26.6.5 chalk: 4.1.2 cjs-module-lexer: 2.2.0 collect-v8-coverage: 1.0.2 @@ -39934,7 +40073,7 @@ snapshots: dependencies: '@jest/test-result': 30.0.5 '@jest/types': 30.0.5 - '@types/node': 22.18.8 + '@types/node': 26.6.5 ansi-escapes: 4.3.2 chalk: 4.1.2 emittery: 0.13.1 @@ -39956,7 +40095,7 @@ snapshots: jest-worker@30.0.5: dependencies: - '@types/node': 22.18.8 + '@types/node': 26.6.5 '@ungap/structured-clone': 1.3.0 jest-util: 30.0.5 merge-stream: 2.0.0 @@ -40295,6 +40434,8 @@ snapshots: layout-base@2.0.1: {} + leac@0.7.0: {} + less@3.13.1: dependencies: copy-anything: 2.0.6 @@ -40331,6 +40472,17 @@ snapshots: prelude-ls: 1.2.1 type-check: 0.4.0 + libbase64@1.4.1: {} + + libmime@5.6.3: + dependencies: + encoding-japanese: 2.4.0 + iconv-lite: 0.7.3 + libbase64: 1.4.1 + libqp: 3.0.1 + + libqp@3.0.1: {} + libsodium-wrappers@0.7.16: dependencies: libsodium: 0.7.16 @@ -40443,6 +40595,10 @@ snapshots: dependencies: uc.micro: 2.1.0 + linkify-it@5.0.2: + dependencies: + uc.micro: 2.1.0 + linkifyjs@4.3.2: {} lint-staged@15.4.3(supports-color@10.2.2): @@ -40632,6 +40788,17 @@ snapshots: dependencies: '@jridgewell/sourcemap-codec': 1.5.5 + mailparser@3.14.1: + dependencies: + '@zone-eu/mailsplit': 5.4.29 + he: 1.2.0 + html-to-text: 10.0.1 + libmime: 5.6.3 + linkify-it: 5.0.2 + nodemailer: 10.3.1 + punycode.js: 2.3.1 + tlds: 1.261.0 + make-dir@2.1.0: dependencies: pify: 4.0.1 @@ -41613,6 +41780,8 @@ snapshots: long-timeout: 0.1.1 sorted-array-functions: 1.3.0 + nodemailer@10.3.1: {} + nodemailer@8.0.8: {} nopt@10.0.1: @@ -42108,6 +42277,11 @@ snapshots: dependencies: entities: 4.5.0 + parseley@0.13.1: + dependencies: + leac: 0.7.0 + peberminta: 0.10.0 + parseurl@1.3.3: {} partial-json@0.1.7: {} @@ -42168,6 +42342,8 @@ snapshots: sha.js: 2.4.12 to-buffer: 1.2.2 + peberminta@0.10.0: {} + pend@1.2.0: {} pg-connection-string@2.5.0: {} @@ -44451,6 +44627,10 @@ snapshots: secure-json-parse@2.7.0: {} + selderee@0.12.0: + dependencies: + parseley: 0.13.1 + semver@5.7.2: {} semver@6.3.1: {} @@ -44680,6 +44860,12 @@ snapshots: smart-buffer@4.2.0: {} + smtp-server@3.21.3: + dependencies: + ipv6-normalize: 1.0.1 + nodemailer: 10.3.1 + punycode.js: 2.3.1 + snappy-wasm@0.3.0: {} snappy@7.2.2: @@ -45576,6 +45762,8 @@ snapshots: tinyspy@4.0.4: {} + tlds@1.261.0: {} + tldts-core@6.1.57: {} tldts-core@7.0.28: {} diff --git a/posthog/jwt.py b/posthog/jwt.py index 6e41684c9058..82d0c171eb68 100644 --- a/posthog/jwt.py +++ b/posthog/jwt.py @@ -41,6 +41,7 @@ class PosthogJwtAudience(Enum): WORKFLOW_VISION_REQUEST = "posthog:workflows:vision_request" WORKFLOW_ADD_TO_STATIC_COHORT = "posthog:workflows:add_to_static_cohort" WORKFLOW_MESSAGE_OPT_OUT = "posthog:workflows:message_opt_out" + SMTP_RELAY_ACTIVITY = "posthog:workflows:smtp_relay_activity" CONVERSATIONS_TICKETS = "posthog:conversations:tickets" CUSTOMER_ANALYTICS_ACCOUNTS = "posthog:customer_analytics:accounts" SECURITY_HUB_RULES = "posthog:security_hub:rules" diff --git a/posthog/settings/data_stores.py b/posthog/settings/data_stores.py index f80dc3fa2cd0..440529eadd26 100644 --- a/posthog/settings/data_stores.py +++ b/posthog/settings/data_stores.py @@ -661,6 +661,15 @@ def _apply_product_db_ssl_options(db: str, options: dict) -> None: ) ) +# Verifies the tokens the SMTP relay sends when it records a refused message. The dev/test value must match the plugin +# server's minting default. Empty outside dev/test, so the route refuses every request until the secret is provisioned. +SMTP_RELAY_ACTIVITY_JWT_SECRETS = get_list( + get_from_env( + "SMTP_RELAY_ACTIVITY_JWT_SECRET", + "local-dev-smtp-relay-activity-jwt" if DEBUG or TEST else "", + ) +) + # Verifies the scoped JWTs the CDP worker's conversations ticket actions send to the internal # ticket route (the worker mints, Django verifies; products/conversations/backend/api/internal.py). # Comma-separated, newest first. Empty outside dev/test, so the internal route rejects every diff --git a/posthog/urls.py b/posthog/urls.py index f5469962a975..f58c13990522 100644 --- a/posthog/urls.py +++ b/posthog/urls.py @@ -52,6 +52,7 @@ from products.experiments.backend.routes import web_experiments from products.messaging.backend.presentation.views.customerio_webhook import CustomerIOWebhookView from products.messaging.backend.presentation.views.recipient_preferences import preferences_page, update_preferences +from products.messaging.backend.presentation.views.smtp_activity import InternalSmtpActivityView from products.notebooks.backend.facade.sql_v2 import ( notebook_sql_v2_callback, notebook_sql_v2_data_plane, @@ -289,6 +290,11 @@ "api/projects//internal/customer_analytics/account/custom_property_values", csrf_exempt(CustomerAnalyticsInternalAccountCustomPropertiesView.as_view()), ), + # Where the SMTP relay records refused messages (auth: scoped service JWT) + path( + "api/projects//internal/messaging/smtp_activity", + csrf_exempt(InternalSmtpActivityView.as_view()), + ), # Test setup endpoint (only available in TEST mode) path("api/setup_test//", csrf_exempt(playwright_setup.setup_test)), opt_slash_path( diff --git a/products/messaging/backend/facade/contracts.py b/products/messaging/backend/facade/contracts.py index 7ee542e2b905..4f5d6d61143c 100644 --- a/products/messaging/backend/facade/contracts.py +++ b/products/messaging/backend/facade/contracts.py @@ -343,3 +343,8 @@ class TransactionalEmailResults: revenue: float # Hours from the send to the first conversion, for the median converted recipient. None when nobody converted. median_hours_to_convert: float | None + + +class SmtpActivityKind(LabeledStrEnum): + REFUSED = "refused", "Refused" + DROPPED_HEADERS = "dropped_headers", "Dropped headers" diff --git a/products/messaging/backend/facade/smtp.py b/products/messaging/backend/facade/smtp.py new file mode 100644 index 000000000000..0364455217f5 --- /dev/null +++ b/products/messaging/backend/facade/smtp.py @@ -0,0 +1,4 @@ +from products.messaging.backend.facade.contracts import SmtpActivityKind +from products.messaging.backend.services.smtp_activity import recent_smtp_activity, record_smtp_activity + +__all__ = ["SmtpActivityKind", "recent_smtp_activity", "record_smtp_activity"] diff --git a/products/messaging/backend/presentation/views/message_templates.py b/products/messaging/backend/presentation/views/message_templates.py index ae1d229b11d0..fdfbe1f827ca 100644 --- a/products/messaging/backend/presentation/views/message_templates.py +++ b/products/messaging/backend/presentation/views/message_templates.py @@ -424,6 +424,7 @@ class DesignPatchSerializer(serializers.Serializer): TRANSACTIONAL_EMAIL_API_FLAG = "workflows-transactional-email-api" +SMTP_RELAY_FLAG = "workflows-smtp-relay" FIRST_CODE_SEND_EVENT = "email first code send" @@ -958,6 +959,7 @@ def _report_transactional_email_send(self, request: Any, properties: dict[str, A if is_authenticated_via_project_secret_api_key(request) else "user", "authorized_by": transactional_email_authorized_by(request, TRANSACTIONAL_EMAIL_SEND_SCOPE), + "send_source": "smtp" if request.headers.get("X-PostHog-Send-Source") == "smtp" else "api", **properties, }, groups=groups(self.organization, self.team), @@ -1267,16 +1269,27 @@ def make_transactional(self, request: Any, *args: Any, **kwargs: Any) -> Respons return Response(data) +class SmtpRelayCallThrottle(PersonalOrProjectSecretApiKeyRateThrottle): + scope = "transactional_email_smtp_relay" + rate = "600/minute" + + +class SmtpCheckResponseSerializer(serializers.Serializer): + team_id = serializers.IntegerField(help_text="ID of the project the key sends for.") + + class MessagingSendViewSet(TransactionalEmailSendMixin, TeamAndOrgViewSetMixin, viewsets.ViewSet): scope_object = "hog_flow" permission_classes = [IsAuthenticated] authentication_classes = [ProjectSecretAPIKeyAuthentication] - psak_allowed_actions = ["email"] + psak_allowed_actions = ["email", "smtp_check"] transactional_email_send_actions = ("email",) # Access rules on hog_flow name workflow ids, never template ids, so no rule names one template. records_have_object_access_rules = False def get_throttles(self): + if self.action == "smtp_check": + return [SmtpRelayCallThrottle()] return transactional_email_throttles() def read_object(self) -> MessageTemplateRow | None: @@ -1318,6 +1331,19 @@ def email(self, request: Any, *args: Any, **kwargs: Any) -> Response: template = self.get_object() return self._queue_transactional_email(request, template, route="send_email") + @extend_schema( + request=None, + responses={200: SmtpCheckResponseSerializer}, + summary="Check a key for the SMTP relay", + description="Says which project a key sends for, so the SMTP relay can answer a login. Answers 404 when SMTP " + "sending is not turned on for the project. Needs the `transactional_email:write` scope.", + ) + @action(detail=False, methods=["POST"], required_scopes=[TRANSACTIONAL_EMAIL_SEND_SCOPE]) + def smtp_check(self, request: Any, *args: Any, **kwargs: Any) -> Response: + _require_flag(self.team, SMTP_RELAY_FLAG) + require_email_from_code(self.team) + return Response(SmtpCheckResponseSerializer({"team_id": self.team_id}).data) + def _send_code_email(self, request: Any) -> Response: with transactional_email_stage("feature_flag"): require_email_from_code(self.team) diff --git a/products/messaging/backend/presentation/views/smtp_activity.py b/products/messaging/backend/presentation/views/smtp_activity.py new file mode 100644 index 000000000000..727f71b794b5 --- /dev/null +++ b/products/messaging/backend/presentation/views/smtp_activity.py @@ -0,0 +1,71 @@ +from typing import Any + +from drf_spectacular.utils import extend_schema +from rest_framework import serializers, status +from rest_framework.permissions import IsAuthenticated +from rest_framework.request import Request +from rest_framework.response import Response +from rest_framework.views import APIView + +from posthog.auth import ScopedServiceJWTAuthentication +from posthog.jwt import PosthogJwtAudience +from posthog.scoped_service_jwt import ScopedServiceJwtPurpose + +from products.messaging.backend.facade.smtp import SmtpActivityKind, record_smtp_activity + +SMTP_RELAY_ACTIVITY_PURPOSE = ScopedServiceJwtPurpose( + audience=PosthogJwtAudience.SMTP_RELAY_ACTIVITY, + settings_name="SMTP_RELAY_ACTIVITY_JWT_SECRETS", +) + + +class SmtpRelayActivityJWTAuthentication(ScopedServiceJWTAuthentication): + purpose = SMTP_RELAY_ACTIVITY_PURPOSE + + +class SmtpActivitySerializer(serializers.Serializer): + kind = serializers.ChoiceField( + choices=SmtpActivityKind.choices, help_text="`refused` or `dropped_headers`, what happened to the message." + ) + reply = serializers.CharField( + required=False, allow_blank=True, max_length=512, help_text="The reply line the SMTP client got." + ) + # A malformed address is exactly what a refusal needs to show, so these aren't validated as addresses. + sender = serializers.CharField( + required=False, allow_blank=True, max_length=320, help_text="The envelope sender of the message." + ) + to = serializers.ListField( + child=serializers.CharField(max_length=320), + required=False, + max_length=20, + help_text="Up to 20 recipients named in the To and Cc headers. Bcc recipients are never listed.", + ) + message_id = serializers.CharField( + required=False, allow_blank=True, max_length=128, help_text="ID of the queued send, when one was queued." + ) + dropped_headers = serializers.ListField( + child=serializers.CharField(max_length=128), + required=False, + max_length=20, + help_text="Names of the headers PostHog could not forward, up to 20.", + ) + + def get_fields(self) -> dict[str, serializers.Field]: + fields = super().get_fields() + fields["from"] = fields.pop("sender") + return fields + + +class InternalSmtpActivityView(APIView): + """Keeps a short per-project list of refused messages and dropped headers, so the SMTP setup page can show why mail + didn't arrive. Only the SMTP relay holds the key for these tokens, and the route is not exposed publicly.""" + + authentication_classes = [SmtpRelayActivityJWTAuthentication] + permission_classes = [IsAuthenticated] + + @extend_schema(exclude=True) + def post(self, request: Request, team_id: str, *args: Any, **kwargs: Any) -> Response: + serializer = SmtpActivitySerializer(data=request.data) + serializer.is_valid(raise_exception=True) + record_smtp_activity(int(team_id), serializer.validated_data) + return Response(status=status.HTTP_204_NO_CONTENT) diff --git a/products/messaging/backend/services/smtp_activity.py b/products/messaging/backend/services/smtp_activity.py new file mode 100644 index 000000000000..86b935cd6429 --- /dev/null +++ b/products/messaging/backend/services/smtp_activity.py @@ -0,0 +1,28 @@ +"""A short list per project of what the SMTP relay refused or dropped, for the setup page. It holds addresses and the +reply a client got, never message content, and expires on its own.""" + +import json +import time +from typing import Any + +from posthog.redis import get_client + +SMTP_ACTIVITY_LIMIT = 50 +SMTP_ACTIVITY_TTL_SECONDS = 7 * 24 * 60 * 60 + + +def _key(team_id: int) -> str: + return f"@posthog/messaging/smtp-activity/{team_id}" + + +def record_smtp_activity(team_id: int, activity: dict[str, Any]) -> None: + key = _key(team_id) + pipeline = get_client().pipeline() + pipeline.lpush(key, json.dumps({**activity, "at": time.time()})) + pipeline.ltrim(key, 0, SMTP_ACTIVITY_LIMIT - 1) + pipeline.expire(key, SMTP_ACTIVITY_TTL_SECONDS) + pipeline.execute() + + +def recent_smtp_activity(team_id: int) -> list[dict[str, Any]]: + return [json.loads(entry) for entry in get_client().lrange(_key(team_id), 0, SMTP_ACTIVITY_LIMIT - 1)] diff --git a/products/messaging/backend/tests/api/test_smtp_relay.py b/products/messaging/backend/tests/api/test_smtp_relay.py new file mode 100644 index 000000000000..2f6b2f56b4dd --- /dev/null +++ b/products/messaging/backend/tests/api/test_smtp_relay.py @@ -0,0 +1,104 @@ +from typing import Any + +from posthog.test.base import APIBaseTest +from unittest.mock import patch + +from django.test import override_settings + +from parameterized import parameterized +from rest_framework import status +from rest_framework.test import APIClient + +from posthog.models import ProjectSecretAPIKey, Team +from posthog.models.personal_api_key import hash_key_value +from posthog.models.utils import generate_random_token_secret +from posthog.test.authentication_checks import covers_authentication + +from products.messaging.backend.presentation.views.smtp_activity import ( + SMTP_RELAY_ACTIVITY_PURPOSE, + SmtpRelayActivityJWTAuthentication, +) +from products.messaging.backend.services.smtp_activity import SMTP_ACTIVITY_LIMIT, recent_smtp_activity + +SMTP_FLAGS = ["workflows-smtp-relay", "workflows-email-from-code"] + + +def _flags_on(*keys: str) -> Any: + return patch("posthoganalytics.feature_enabled", side_effect=lambda flag, *_args, **_kwargs: flag in keys) + + +class TestSmtpRelayApi(APIBaseTest): + def _key_client(self, scopes: list[str], team: Team | None = None) -> APIClient: + token = generate_random_token_secret() + ProjectSecretAPIKey.objects.create( + team=team or self.team, label=f"smtp {token[:8]}", secure_value=hash_key_value(token), scopes=scopes + ) + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Bearer {token}") + return client + + @parameterized.expand( + [ + ("can_send", ["transactional_email:write"], SMTP_FLAGS, status.HTTP_200_OK), + ("legacy_scope", ["hog_flow:write"], SMTP_FLAGS, status.HTTP_200_OK), + ("read_only_key", ["transactional_email:read"], SMTP_FLAGS, status.HTTP_403_FORBIDDEN), + ("relay_flag_off", ["transactional_email:write"], ["workflows-email-from-code"], status.HTTP_404_NOT_FOUND), + ("code_sends_flag_off", ["transactional_email:write"], ["workflows-smtp-relay"], status.HTTP_404_NOT_FOUND), + ] + ) + def test_check_says_which_project_a_key_sends_for( + self, _name: str, scopes: list[str], flags: list[str], expected: int + ) -> None: + with _flags_on(*flags): + response = self._key_client(scopes).post("/api/projects/@current/messaging/send/smtp_check/", {}) + + assert response.status_code == expected, response.content + if expected == status.HTTP_200_OK: + assert response.json() == {"team_id": self.team.id} + + def test_check_refuses_an_unknown_key(self) -> None: + client = APIClient() + client.credentials(HTTP_AUTHORIZATION="Bearer phs_notarealkey") + + with _flags_on(*SMTP_FLAGS): + response = client.post("/api/projects/@current/messaging/send/smtp_check/", {}) + + assert response.status_code == status.HTTP_401_UNAUTHORIZED + + @covers_authentication(SmtpRelayActivityJWTAuthentication) + @override_settings(SMTP_RELAY_ACTIVITY_JWT_SECRETS=["activity-key"]) + def test_activity_is_newest_first_capped_and_kept_per_project(self) -> None: + other_team = Team.objects.create(organization=self.organization, name="Other") + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Bearer {SMTP_RELAY_ACTIVITY_PURPOSE.mint({'team_id': self.team.id})}") + + for index in range(SMTP_ACTIVITY_LIMIT + 2): + response = client.post( + f"/api/projects/{self.team.id}/internal/messaging/smtp_activity", + {"kind": "refused", "reply": f"550 number {index}", "from": "a@example.com", "to": ["b@example.com"]}, + format="json", + ) + assert response.status_code == status.HTTP_204_NO_CONTENT, response.content + + activity = recent_smtp_activity(self.team.id) + other = recent_smtp_activity(other_team.id) + assert len(activity) == SMTP_ACTIVITY_LIMIT + assert activity[0]["reply"] == f"550 number {SMTP_ACTIVITY_LIMIT + 1}" + assert activity[0]["from"] == "a@example.com" + assert other == [] + + @override_settings(SMTP_RELAY_ACTIVITY_JWT_SECRETS=["activity-key"]) + def test_activity_only_takes_the_relays_token_for_the_same_project(self) -> None: + other_team = Team.objects.create(organization=self.organization, name="Other") + url = f"/api/projects/{self.team.id}/internal/messaging/smtp_activity" + body = {"kind": "refused", "reply": "550 made up"} + other_team_token = APIClient() + other_team_token.credentials( + HTTP_AUTHORIZATION=f"Bearer {SMTP_RELAY_ACTIVITY_PURPOSE.mint({'team_id': other_team.id})}" + ) + + for client in (self._key_client(["transactional_email:write"]), other_team_token): + response = client.post(url, body, format="json") + + assert response.status_code in (status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN), response.content + assert recent_smtp_activity(self.team.id) == [] diff --git a/products/messaging/frontend/generated/api.schemas.ts b/products/messaging/frontend/generated/api.schemas.ts index 483fcbe24ebd..bc8af09fc3e9 100644 --- a/products/messaging/frontend/generated/api.schemas.ts +++ b/products/messaging/frontend/generated/api.schemas.ts @@ -177,6 +177,11 @@ export interface SendTransactionalEmailResponseApi { replayed: boolean } +export interface SmtpCheckResponseApi { + /** ID of the project the key sends for. */ + team_id: number +} + /** * * `marketing` - Marketing * * `transactional` - Transactional diff --git a/products/messaging/frontend/generated/api.ts b/products/messaging/frontend/generated/api.ts index 25f56be7d99e..c476c5bc9956 100644 --- a/products/messaging/frontend/generated/api.ts +++ b/products/messaging/frontend/generated/api.ts @@ -44,6 +44,7 @@ import type { SendTransactionalEmailApi, SendTransactionalEmailResponseApi, SetPreferencesApi, + SmtpCheckResponseApi, StartProviderImportApi, SupabaseAuthEmailSettingsApi, TopicPreferencesApi, @@ -137,6 +138,24 @@ export const messagingSendEmailCreate = async ( }) } +export const getMessagingSendSmtpCheckCreateUrl = (projectId: string) => { + return `/api/projects/${projectId}/messaging/send/smtp_check/` +} + +/** + * Says which project a key sends for, so the SMTP relay can answer a login. Answers 404 when SMTP sending is not turned on for the project. Needs the `transactional_email:write` scope. + * @summary Check a key for the SMTP relay + */ +export const messagingSendSmtpCheckCreate = async ( + projectId: string, + options?: RequestInit +): Promise => { + return apiMutator(getMessagingSendSmtpCheckCreateUrl(projectId), { + ...options, + method: 'POST', + }) +} + export const getMessagingCategoriesListUrl = (projectId: string, params?: MessagingCategoriesListParams) => { const normalizedParams = new URLSearchParams() diff --git a/services/mcp/src/api/generated.ts b/services/mcp/src/api/generated.ts index 7bcbad851e68..19f60273fdd9 100644 --- a/services/mcp/src/api/generated.ts +++ b/services/mcp/src/api/generated.ts @@ -108750,6 +108750,11 @@ export namespace Schemas { has_more?: boolean; } + export interface SmtpCheckResponse { + /** ID of the project the key sends for. */ + team_id: number; + } + /** * * `Snowflake` - Snowflake */