From 30fb28b0ec9edda26e831b55c5537a693a5091ad Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Mon, 19 Oct 2026 09:28:41 +0000 Subject: [PATCH 1/2] feat(workflows): restrict sandbox email to verified org members Co-Authored-By: Claude Opus 5.5 --- .../internal/workflows-email-sending-tiers.md | 1 + nodejs/src/cdp/cdp-services.ts | 7 +- .../organization-members.service.test.ts | 74 ++ .../managers/organization-members.service.ts | 127 +++ .../services/messaging/email.service.test.ts | 758 +++++++++++++++++- .../cdp/services/messaging/email.service.ts | 280 ++++++- .../messaging/sandbox-email-sender.ts | 26 +- nodejs/src/cdp/utils.ts | 1 + 8 files changed, 1220 insertions(+), 54 deletions(-) create mode 100644 nodejs/src/cdp/services/managers/organization-members.service.test.ts create mode 100644 nodejs/src/cdp/services/managers/organization-members.service.ts diff --git a/docs/internal/workflows-email-sending-tiers.md b/docs/internal/workflows-email-sending-tiers.md index fa14233c1387..d1378b2726b2 100644 --- a/docs/internal/workflows-email-sending-tiers.md +++ b/docs/internal/workflows-email-sending-tiers.md @@ -85,5 +85,6 @@ Decay, suspension drops, admin recomputes, and the backfill stay silent. Sandbox email uses a fixed SES tenant, configuration set, and From address. Set `SES_SANDBOX_TENANT_NAME`, `SES_SANDBOX_CONFIGURATION_SET`, and `SES_SANDBOX_FROM_ADDRESS` on the worker. Django reads the same `SES_SANDBOX_FROM_ADDRESS` for the address it stores and shows. The address must belong to the shared SES identity. Sandbox sends disable click tracking and unsubscribe headers because the complete stack restricts them to verified organization members for internal testing. Marketing delivery to external audiences requires a verified sender. +Every To, CC, and BCC address must belong to an active organization member with a verified email address. Membership snapshots expire within one minute, and the worker checks recipients again immediately before delivery. A failed or stale membership check skips the send and explains the blocked addresses in the run log. HTML is parsed and normalized before the identification footer is appended. Root clipping and visibility styles are overridden while template typography is preserved; malformed markup may render differently from an own-domain send. Preview text is parsed separately so escaped template values remain readable. The visible footer is best effort across email clients and arbitrary template CSS. Signed message tracking headers retain sender attribution independently of the template. diff --git a/nodejs/src/cdp/cdp-services.ts b/nodejs/src/cdp/cdp-services.ts index ae8961673f6a..3b60a0e40a33 100644 --- a/nodejs/src/cdp/cdp-services.ts +++ b/nodejs/src/cdp/cdp-services.ts @@ -31,6 +31,7 @@ import { InvocationResultsService } from './services/invocation-results.service' import { HogFunctionManagerService } from './services/managers/hog-function-manager.service' import { HogFunctionTemplateManagerService } from './services/managers/hog-function-template-manager.service' import { IntegrationManagerService } from './services/managers/integration-manager.service' +import { OrganizationMembersService } from './services/managers/organization-members.service' import { PersonsManagerService } from './services/managers/persons-manager.service' import { RecipientsManagerService } from './services/managers/recipients-manager.service' import { TeamWorkflowsConfigService } from './services/managers/team-workflows-config.service' @@ -486,8 +487,10 @@ export function createCdpCoreServices( configurationSetName: config.SES_SANDBOX_CONFIGURATION_SET, fromAddress: config.SES_SANDBOX_FROM_ADDRESS, }, - deps.teamManager - ) + deps.teamManager, + redis + ), + new OrganizationMembersService(deps.postgres) ) const recipientTokensService = new RecipientTokensService(config.ENCRYPTION_SALT_KEYS, config.SITE_URL) const hogInputsService = new HogInputsService(deps.integrationManager, recipientTokensService, deps.encryptedFields) diff --git a/nodejs/src/cdp/services/managers/organization-members.service.test.ts b/nodejs/src/cdp/services/managers/organization-members.service.test.ts new file mode 100644 index 000000000000..e52ebfceb53e --- /dev/null +++ b/nodejs/src/cdp/services/managers/organization-members.service.test.ts @@ -0,0 +1,74 @@ +import { PostgresRouter } from '~/common/utils/db/postgres' + +import { OrganizationMembersService } from './organization-members.service' + +describe('sandbox organization membership decisions', () => { + it('fails closed when a warmed organization mapping expires during a member refresh', async () => { + let now = -10_000 + const organizationId = '00000000-0000-0000-0000-000000000001' + const query = jest.fn((_use, _sql, params, operation) => { + if (operation === 'fetch-sandbox-team-organization') { + return Promise.resolve({ + rows: params[0].map((id: number) => ({ id, organization_id: organizationId })), + }) + } + if (now > 0) { + now = 63_000 + } + return Promise.resolve({ + rows: [{ organization_id: organizationId, email: 'member@example.com', is_email_verified: true }], + }) + }) + const service = new OrganizationMembersService({ query } as unknown as PostgresRouter, () => now) + expect(await service.getBlockedRecipients(2, ['member@example.com'])).toEqual([]) + now = 0 + await service.getOrganizationSnapshot(1) + now = 58_000 + + await expect(service.getBlockedRecipients(1, ['member@example.com'])).rejects.toThrow('expired') + }) + + it('rejects a member response that outlasts its one-minute decision', async () => { + let now = 0 + const organizationId = '00000000-0000-0000-0000-000000000001' + const query = jest + .fn() + .mockResolvedValueOnce({ rows: [{ id: 1, organization_id: organizationId }] }) + .mockImplementationOnce(() => { + now = 60_000 + return Promise.resolve({ + rows: [{ organization_id: organizationId, email: 'member@example.com', is_email_verified: true }], + }) + }) + const service = new OrganizationMembersService({ query } as unknown as PostgresRouter, () => now) + + await expect(service.getBlockedRecipients(1, ['member@example.com'])).rejects.toThrow('expired') + }) + + it('shares a bounded organization lookup across repeated sends and refreshes it after a minute', async () => { + let now = 0 + let organizationId = '00000000-0000-0000-0000-000000000001' + const query = jest.fn((_use, _sql, _params, operation) => + Promise.resolve({ + rows: + operation === 'fetch-sandbox-team-organization' + ? [{ id: 1, organization_id: organizationId }] + : [{ organization_id: organizationId, email: 'member@example.com', is_email_verified: true }], + }) + ) + const service = new OrganizationMembersService({ query } as unknown as PostgresRouter, () => now) + + expect(await service.getBlockedRecipients(1, ['member@example.com'])).toEqual([]) + expect(await service.getBlockedRecipients(1, ['member@example.com'])).toEqual([]) + expect( + query.mock.calls.filter(([, , , operation]) => operation === 'fetch-sandbox-team-organization') + ).toHaveLength(1) + + now = 60_001 + organizationId = '00000000-0000-0000-0000-000000000002' + expect(await service.getBlockedRecipients(1, ['member@example.com'])).toEqual([]) + expect( + query.mock.calls.filter(([, , , operation]) => operation === 'fetch-sandbox-team-organization') + ).toHaveLength(2) + }) +}) diff --git a/nodejs/src/cdp/services/managers/organization-members.service.ts b/nodejs/src/cdp/services/managers/organization-members.service.ts new file mode 100644 index 000000000000..14dd567233cf --- /dev/null +++ b/nodejs/src/cdp/services/managers/organization-members.service.ts @@ -0,0 +1,127 @@ +import { PostgresRouter, PostgresUse } from '~/common/utils/db/postgres' +import { LazyLoader } from '~/common/utils/lazy-loader' + +type OrganizationMemberSnapshot = { + verifiedEmails: ReadonlySet + unverifiedEmails: ReadonlySet + expiresAt: number +} + +export type BlockedRecipient = { readonly address: string; readonly unverifiedMember: boolean } + +export type TeamOrganizationSnapshot = { readonly organizationId: string; readonly expiresAt: number } + +const MAX_SNAPSHOT_AGE_MS = 60_000 +const REFRESH_AGE_MS = MAX_SNAPSHOT_AGE_MS - 1_000 + +export class OrganizationMembersService { + private snapshots: LazyLoader + private organizations: LazyLoader + + constructor( + postgres: PostgresRouter, + private now: () => number = Date.now + ) { + this.organizations = new LazyLoader({ + name: 'sandbox_team_organizations', + refreshAgeMs: REFRESH_AGE_MS, + refreshJitterMs: 0, + bufferMs: 0, + loader: async (teamIds) => { + const expiresAt = this.now() + MAX_SNAPSHOT_AGE_MS + const { rows } = await postgres.query<{ id: number; organization_id: string }>( + PostgresUse.COMMON_WRITE, + 'SELECT id, organization_id FROM posthog_team WHERE id = ANY($1::bigint[])', + [teamIds.map(Number)], + 'fetch-sandbox-team-organization' + ) + return Object.fromEntries( + rows.map((team) => [String(team.id), { organizationId: team.organization_id, expiresAt }]) + ) + }, + }) + this.snapshots = new LazyLoader({ + name: 'organization_email_members', + refreshAgeMs: REFRESH_AGE_MS, + refreshJitterMs: 0, + bufferMs: 0, + loader: async (organizationIds) => { + const expiresAt = this.now() + MAX_SNAPSHOT_AGE_MS + const { rows } = await postgres.query<{ + organization_id: string + email: string + is_email_verified: boolean | null + }>( + PostgresUse.COMMON_WRITE, + `SELECT membership.organization_id, users.email, users.is_email_verified + FROM posthog_organizationmembership AS membership + JOIN posthog_user AS users ON users.id = membership.user_id + WHERE membership.organization_id = ANY($1::uuid[]) AND users.is_active = true`, + [organizationIds], + 'fetch-organization-email-members' + ) + const members: Record; unverified: Set }> = Object.fromEntries( + organizationIds.map((id) => [id, { verified: new Set(), unverified: new Set() }]) + ) + for (const row of rows) { + const { verified, unverified } = members[row.organization_id] + ;(row.is_email_verified === true ? verified : unverified).add(row.email.trim().toLowerCase()) + } + return Object.fromEntries( + Object.entries(members).map(([id, { verified, unverified }]) => [ + id, + { verifiedEmails: verified, unverifiedEmails: unverified, expiresAt }, + ]) + ) + }, + }) + } + + public async getOrganizationSnapshot(teamId: number): Promise { + const key = String(teamId) + let snapshot = await this.organizations.get(key) + if (snapshot && this.now() >= snapshot.expiresAt) { + this.organizations.markForRefresh(key) + snapshot = await this.organizations.get(key) + } + if (!snapshot) { + throw new Error('Could not identify the organization for this team') + } + if (this.now() >= snapshot.expiresAt) { + this.organizations.markForRefresh(key) + throw new Error('The organization lookup expired') + } + return snapshot + } + + public async getBlockedRecipients( + teamId: number, + recipients: string[], + organizationSnapshot?: TeamOrganizationSnapshot + ): Promise { + const expiresAt = this.now() + MAX_SNAPSHOT_AGE_MS + const organization = organizationSnapshot ?? (await this.getOrganizationSnapshot(teamId)) + if (this.now() >= organization.expiresAt) { + throw new Error('The organization lookup expired') + } + const { organizationId } = organization + let snapshot = await this.snapshots.get(organizationId) + if (snapshot && this.now() >= snapshot.expiresAt && this.now() < Math.min(expiresAt, organization.expiresAt)) { + this.snapshots.markForRefresh(organizationId) + snapshot = await this.snapshots.get(organizationId) + } + if (!snapshot) { + throw new Error('Could not check organization members') + } + if (this.now() >= Math.min(expiresAt, snapshot.expiresAt, organization.expiresAt)) { + this.snapshots.markForRefresh(organizationId) + throw new Error('The organization member check expired') + } + return recipients + .filter((email) => !snapshot.verifiedEmails.has(email.trim().toLowerCase())) + .map((address) => ({ + address, + unverifiedMember: snapshot.unverifiedEmails.has(address.trim().toLowerCase()), + })) + } +} diff --git a/nodejs/src/cdp/services/messaging/email.service.test.ts b/nodejs/src/cdp/services/messaging/email.service.test.ts index 657fffaa9d98..ea7408b4e358 100644 --- a/nodejs/src/cdp/services/messaging/email.service.test.ts +++ b/nodejs/src/cdp/services/messaging/email.service.test.ts @@ -20,12 +20,18 @@ import { defineLuaTokenBucketV2 } from '~/common/redis/redis-token-bucket-v2.lua import { defineLuaTokenBucketV3 } from '~/common/redis/redis-token-bucket-v3.lua' import { RedisClient, RedisClientPipeline, RedisV2, createRedisV2PoolFromConfig } from '~/common/redis/redis-v2' import { closeHub, createHub } from '~/common/utils/db/hub' -import { PostgresUse } from '~/common/utils/db/postgres' +import { PostgresRouter, PostgresUse } from '~/common/utils/db/postgres' import * as posthog from '~/common/utils/posthog' import { waitForExpect } from '~/tests/helpers/expectations' -import { createTestTeamFixture } from '~/tests/helpers/sql' +import { + createOrganization, + createOrganizationMembership, + createTestTeamFixture, + createUser, +} from '~/tests/helpers/sql' import { Hub, Team } from '../../../types' +import { OrganizationMembersService } from '../managers/organization-members.service' import { RecipientsManagerService } from '../managers/recipients-manager.service' import { TeamWorkflowsConfigService } from '../managers/team-workflows-config.service' import { RateLimiterService } from '../rate-limiter/rate-limiter.service' @@ -230,6 +236,31 @@ describe('EmailService', () => { }) describe('sandbox sender', () => { let capture: jest.SpyInstance + let memberId: number + let memberIds: number[] + let extraOrganizationIds: string[] + const memberEmail = (name = 'test'): string => `${name}-${team.id}@example.com` + const createSandboxParams = ( + params: Partial = {} + ): CyclotronInvocationQueueParametersEmailType => + createEmailParams({ to: { email: memberEmail(), name: 'Example member' }, ...params }) + const createMember = async (email: string, organizationId = team.organization_id): Promise => { + const id = await createUser(hub.postgres, email) + memberIds.push(id) + await createOrganizationMembership(hub.postgres, organizationId, id) + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'UPDATE posthog_user SET email = $1, is_active = true, is_email_verified = true WHERE id = $2', + [email, id], + 'test:verify-sandbox-member' + ) + return id + } + const createOtherOrganization = async (): Promise => { + const id = await createOrganization(hub.postgres) + extraOrganizationIds.push(id) + return id + } let sandboxRedisClient: Redis.Redis | undefined const createSandboxLimiter = async (name: string): Promise => { sandboxRedisClient = await hub.redisPool.acquire() @@ -250,6 +281,7 @@ describe('EmailService', () => { enabled: boolean, tierLimiter: RateLimiterService | null = null, messageAssetsService?: MessageAssetsService, + membersPostgres: PostgresRouter = hub.postgres, routeSettings: Partial = {} ): EmailService => { const sandboxService = new EmailService( @@ -282,8 +314,19 @@ describe('EmailService', () => { fromAddress: 'fixed-sandbox@example.com', ...routeSettings, }, - hub.teamManager - ) + hub.teamManager, + { + useClient: async (_options, callback) => { + const client = await hub.redisPool.acquire() + try { + return await callback(client as unknown as RedisClient) + } finally { + await hub.redisPool.release(client) + } + }, + } + ), + new OrganizationMembersService(membersPostgres) ) sendEmailSpy = jest.spyOn(sandboxService.sesV2Client!, 'send') as jest.SpyInstance sendEmailSpy.mockResolvedValue({ MessageId: 'sandbox-message-id' }) @@ -292,6 +335,11 @@ describe('EmailService', () => { beforeEach(async () => { capture = jest.spyOn(posthog, 'captureTeamEvent').mockImplementation(() => {}) + memberIds = [] + extraOrganizationIds = [] + memberId = await createMember(memberEmail()) + await createMember(memberEmail('cc')) + await createMember(memberEmail('bcc')) await insertIntegration(hub.postgres, team.id, { id: getIntegrationId(4), kind: 'email', @@ -303,11 +351,46 @@ describe('EmailService', () => { verified: true, }, }) - invocation.queueParameters = createEmailParams({ from: { integrationId: 4 } }) + invocation.queueParameters = createSandboxParams({ + from: { integrationId: 4 }, + }) Object.assign(invocation, { hogFlow: { id: invocation.functionId, team_id: team.id } }) }) afterEach(async () => { + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'DELETE FROM posthog_messagesuppression WHERE team_id = $1', + [team.id], + 'test:delete-sandbox-suppressions' + ) + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'DELETE FROM posthog_organizationmembership WHERE user_id = ANY($1::integer[])', + [memberIds], + 'test:delete-sandbox-memberships' + ) + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'DELETE FROM posthog_user WHERE id = ANY($1::integer[])', + [memberIds], + 'test:delete-sandbox-members' + ) + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'DELETE FROM posthog_organization WHERE id = ANY($1::uuid[])', + [extraOrganizationIds], + 'test:delete-sandbox-organizations' + ) + const client = await hub.redisPool.acquire() + try { + const keys = await client.keys(`@posthog/workflows-sandbox-blocked/${team.id}/*`) + if (keys.length) { + await client.del(...keys) + } + } finally { + await hub.redisPool.release(client) + } capture.mockRestore() if (sandboxRedisClient) { await hub.redisPool.release(sandboxRedisClient) @@ -315,16 +398,28 @@ describe('EmailService', () => { } }) + it('skips a sandbox send to a person without an email address as a missing recipient', async () => { + service = createSandboxService(true) + invocation.queueParameters = createSandboxParams({ + from: { integrationId: 4 }, + to: { email: '', name: 'Example member' }, + }) + + const result = await service.executeSendEmail(invocation) + + expect(sendEmailSpy).not.toHaveBeenCalled() + expect(result.metrics.map((m) => m.metric_name)).toEqual(['email_missing_recipient']) + expect(capture).not.toHaveBeenCalled() + }) + it.each([false, true])( 'permits sandbox destination invocations only as explicit tests (%s)', async (isTest) => { service = createSandboxService(true) const destination = createExampleInvocation({ team_id: team.id, id: 'email-destination' }) destination.state.vmState = { stack: [] } as any - destination.queueParameters = createEmailParams({ from: { integrationId: 4 } }) - + destination.queueParameters = createSandboxParams({ from: { integrationId: 4 } }) const result = await service.executeSendEmail(destination, isTest) - expect(result.error).toBeUndefined() expect(result.skipped === true).toBe(!isTest) expect(sendEmailSpy).toHaveBeenCalledTimes(isTest ? 1 : 0) @@ -340,6 +435,627 @@ describe('EmailService', () => { } ) + it('deduplicates blocked analytics per reason and send kind across sender instances while preserving every skip log', async () => { + for (let attempt = 0; attempt < 2; attempt++) { + service = createSandboxService(true) + invocation.state.vmState = { stack: [] } as any + invocation.queueParameters = createSandboxParams({ + from: { integrationId: 4 }, + to: { email: 'outside@example.com' }, + }) + const result = await service.executeSendEmail(invocation) + expect(result.skipped).toBe(true) + expect(result.logs).toContainEqual( + expect.objectContaining({ message: expect.stringContaining('outside@example.com') }) + ) + } + expect(capture).toHaveBeenCalledTimes(1) + + service = createSandboxService(true) + invocation.queueParameters = createSandboxParams({ + from: { integrationId: 4 }, + to: { email: 'outside@example.com' }, + }) + await service.executeSendEmail(invocation, true) + expect(capture).toHaveBeenCalledTimes(2) + expect(capture).toHaveBeenLastCalledWith( + expect.anything(), + 'workflows sandbox email blocked', + expect.objectContaining({ reason: 'recipient_not_member', is_test: true }) + ) + + service = createSandboxService(false) + invocation.queueParameters = createSandboxParams({ from: { integrationId: 4 } }) + await service.executeSendEmail(invocation) + expect(capture).toHaveBeenCalledTimes(3) + expect(capture).toHaveBeenLastCalledWith( + expect.anything(), + 'workflows sandbox email blocked', + expect.objectContaining({ reason: 'switch_off' }) + ) + }) + + it.each([false, true])('blocks a non-member To address (isTest=%s)', async (isTest) => { + service = createSandboxService(true) + invocation.queueParameters = createSandboxParams({ + from: { integrationId: 4 }, + to: { email: 'outside@example.com' }, + }) + + const result = await service.executeSendEmail(invocation, isTest) + + expect(sendEmailSpy).not.toHaveBeenCalled() + expect(result).toMatchObject({ finished: true, skipped: true, metrics: [] }) + expect(result.error).toBeUndefined() + expect(result.invocation.state.vmState?.stack).toEqual([{ success: false }]) + expect(result.logs).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + level: 'info', + message: + 'Skipping send: the sandbox sender only sends to active organization members with verified email addresses. Blocked addresses: outside@example.com. Verify your own domain to send to anyone.', + }), + ]) + ) + expect(capture).toHaveBeenCalledWith( + expect.objectContaining({ id: team.id }), + 'workflows sandbox email blocked', + { reason: 'recipient_not_member', is_test: isTest, blocked_recipient_count: 1 } + ) + }) + + it.each([false].flatMap((isTest) => [false, true].map((checkFailed) => [isTest, checkFailed] as const)))( + 'packs all blocked addresses and guidance into few long log rows (isTest=%s, checkFailed=%s)', + async (isTest, checkFailed) => { + const outside = Array.from( + { length: 200 }, + (_, index) => + `${'a'.repeat(60)}${index}@${'b'.repeat(48)}.${'c'.repeat(48)}.${'d'.repeat(48)}.example.com` + ) + const membersPostgres = checkFailed ? new PostgresRouter(hub) : hub.postgres + if (checkFailed) { + await membersPostgres.end() + } + service = createSandboxService(true, null, undefined, membersPostgres) + invocation.queueParameters = createSandboxParams({ + from: { integrationId: 4 }, + cc: outside.slice(0, 100).join(', '), + bcc: outside.slice(100).join(', '), + }) + + const result = await service.executeSendEmail(invocation, isTest) + + expect(sendEmailSpy).not.toHaveBeenCalled() + expect(result).toMatchObject({ finished: true, skipped: true, metrics: [], messageAssets: [] }) + expect(result.error).toBeUndefined() + expect(result.invocation.state.vmState?.stack).toEqual([{ success: false }]) + const messages = result.logs.map(({ message }) => message).join('\n') + for (const email of checkFailed ? [memberEmail(), ...outside] : outside) { + expect(messages).toContain(email) + } + expect( + result.logs.filter(({ message }) => outside.some((email) => message.includes(email))) + ).toHaveLength(5) + expect(messages.toLowerCase()).toContain('verify your own domain to send to anyone.') + expect(messages).not.toContain('(truncated)') + expect(capture).toHaveBeenCalledTimes(1) + expect(capture).toHaveBeenCalledWith( + expect.objectContaining({ id: team.id }), + 'workflows sandbox email blocked', + { + reason: checkFailed ? 'check_failed' : 'recipient_not_member', + is_test: isTest, + blocked_recipient_count: checkFailed ? 201 : 200, + } + ) + } + ) + + it.each( + [false].flatMap((isTest) => + [ + 'cc', + 'bcc', + 'cc list', + 'bcc list', + 'multiple', + 'unverified', + 'legacy', + 'inactive', + 'other organization', + ].map((recipient) => [isTest, recipient] as const) + ) + )('blocks all delivery for %s / %s', async (isTest, recipient) => { + let blocked = ['outside@example.com'] + let params = createSandboxParams({ from: { integrationId: 4 } }) + if (recipient === 'cc' || recipient === 'bcc') { + params = { ...params, [recipient]: 'Outside member ' } + blocked = ['OUTSIDE@example.com'] + } else if (recipient === 'cc list' || recipient === 'bcc list') { + const field = recipient === 'cc list' ? 'cc' : 'bcc' + params = { ...params, [field]: `"Example, colleague" <${memberEmail(field)}>, outside@example.com` } + } else if (recipient === 'multiple') { + params = { ...params, cc: 'outside@example.com', bcc: 'another@example.com' } + blocked = ['outside@example.com', 'another@example.com'] + } else if (recipient === 'other organization') { + blocked = [memberEmail('outside')] + await createMember(blocked[0], await createOtherOrganization()) + params = { ...params, to: { email: blocked[0] } } + } else { + blocked = [memberEmail()] + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'UPDATE posthog_user SET is_active = $1, is_email_verified = $2 WHERE id = $3', + [recipient !== 'inactive', recipient === 'legacy' ? null : recipient === 'inactive', memberId], + 'test:change-sandbox-member-status' + ) + } + service = createSandboxService(true) + invocation.queueParameters = params + + const result = await service.executeSendEmail(invocation, isTest) + + expect(sendEmailSpy).not.toHaveBeenCalled() + expect(result).toMatchObject({ finished: true, skipped: true, metrics: [], messageAssets: [] }) + expect(result.error).toBeUndefined() + expect(result.invocation.queueScheduledAt).toBeUndefined() + expect(result.invocation.state.vmState?.stack).toEqual([{ success: false }]) + const unverifiedMember = recipient === 'unverified' || recipient === 'legacy' + expect(result.logs).toContainEqual( + expect.objectContaining({ + level: 'info', + message: unverifiedMember + ? `Skipping send: the sandbox sender only sends to organization members with verified email addresses. Unverified addresses: ${blocked.join(', ')}. Verify your email address in PostHog, or ask these members to verify theirs.` + : `Skipping send: the sandbox sender only sends to active organization members with verified email addresses. Blocked addresses: ${blocked.join(', ')}. Verify your own domain to send to anyone.`, + }) + ) + expect(capture).toHaveBeenCalledTimes(1) + expect(capture).toHaveBeenCalledWith( + expect.objectContaining({ id: team.id }), + 'workflows sandbox email blocked', + { + reason: unverifiedMember ? 'recipient_unverified' : 'recipient_not_member', + is_test: isTest, + blocked_recipient_count: blocked.length, + } + ) + expect(result.capturedPostHogEvents).toEqual([]) + }) + + it.each( + [false].flatMap((isTest) => + [ + 'Example colleague', + '"member@example.com"', + '"Example, colleague"', + '"Example" Colleague', + ].flatMap((name) => + [false, true].map((storedMixedCase) => [isTest, name, storedMixedCase] as const) + ) + ) + )( + 'matches bare addresses without case sensitivity (isTest=%s, name=%s, storedMixedCase=%s)', + async (isTest, name, storedMixedCase) => { + if (storedMixedCase) { + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'UPDATE posthog_user SET email = initcap(email) WHERE id = ANY($1::integer[])', + [memberIds], + 'test:store-mixed-case-sandbox-members' + ) + } + service = createSandboxService(true) + invocation.queueParameters = createSandboxParams({ + from: { integrationId: 4 }, + to: { email: memberEmail().toUpperCase() }, + cc: `${name} <${memberEmail('cc').toUpperCase()}>`, + bcc: ` ${name} <${memberEmail('bcc').toUpperCase()}> `, + }) + + const result = await service.executeSendEmail(invocation, isTest) + + expect(result).toMatchObject({ finished: true }) + expect(result.skipped).not.toBe(true) + expect(result.error).toBeUndefined() + expect(result.invocation.state.vmState?.stack).toEqual([{ success: true }]) + expect(sendEmailSpy).toHaveBeenCalledTimes(1) + expect((sendEmailSpy.mock.calls[0][0] as SendEmailCommand).input.Destination).toEqual({ + ToAddresses: [memberEmail().toUpperCase()], + CcAddresses: [memberEmail('cc').toUpperCase()], + BccAddresses: [memberEmail('bcc').toUpperCase()], + }) + expect(capture).toHaveBeenCalledWith( + expect.objectContaining({ id: team.id }), + 'workflows sandbox email sent', + { recipient_count: 3, source: isTest ? 'test' : 'workflow', is_test: isTest } + ) + } + ) + + it.each( + [false].flatMap((isTest) => + ['cc', 'bcc'].flatMap((field) => [false, true].map((list) => [isTest, field, list] as const)) + ) + )('accepts escaped quoted display names (isTest=%s, field=%s, list=%s)', async (isTest, field, list) => { + service = createSandboxService(true) + invocation.queueParameters = createSandboxParams({ + from: { integrationId: 4 }, + [field]: `"Example \\"colleague, teammate\\"" <${memberEmail(field)}>${list ? `, ${memberEmail(field === 'cc' ? 'bcc' : 'cc')}` : ''}`, + }) + const result = await service.executeSendEmail(invocation, isTest) + expect(result.error).toBeUndefined() + expect(result.skipped).not.toBe(true) + expect(result.invocation.state.vmState?.stack).toEqual([{ success: true }]) + expect(sendEmailSpy).toHaveBeenCalledTimes(1) + const destination = (sendEmailSpy.mock.calls[0][0] as SendEmailCommand).input.Destination + expect(field === 'cc' ? destination?.CcAddresses : destination?.BccAddresses).toEqual([ + memberEmail(field), + ...(list ? [memberEmail(field === 'cc' ? 'bcc' : 'cc')] : []), + ]) + expect(capture).toHaveBeenCalledWith( + expect.objectContaining({ id: team.id }), + 'workflows sandbox email sent', + { recipient_count: list ? 3 : 2, source: isTest ? 'test' : 'workflow', is_test: isTest } + ) + }) + + it.each( + [false].flatMap((isTest) => + ['cc', 'bcc'].flatMap((field) => + [false, true].flatMap((suppressed) => + [false, true].map((wrapped) => [isTest, field, suppressed, wrapped] as const) + ) + ) + ) + )( + 'keeps a quoted mailbox intact (isTest=%s, field=%s, suppressed=%s, wrapped=%s)', + async (isTest, field, suppressed, wrapped) => { + const email = `"example,<${team.id}>"@example.com` + await createMember(email) + if (suppressed) { + await new EmailSuppressionService( + hub.postgres, + emailSuppressionConfigFromEnv() + ).recordHardBounces(team.id, [email]) + } + service = createSandboxService(true) + invocation.queueParameters = createSandboxParams({ + from: { integrationId: 4 }, + [field]: wrapped ? `Example colleague <${email}>` : email, + }) + const result = await service.executeSendEmail(invocation, isTest) + expect(result.error).toBeUndefined() + expect(result.invocation.state.vmState?.stack).toEqual([{ success: !suppressed }]) + if (suppressed) { + expect(sendEmailSpy).not.toHaveBeenCalled() + expect(result.logs).toContainEqual( + expect.objectContaining({ message: expect.stringContaining(email) }) + ) + expect(capture).not.toHaveBeenCalled() + } else { + expect(result.skipped).not.toBe(true) + expect(sendEmailSpy).toHaveBeenCalledTimes(1) + const destination = (sendEmailSpy.mock.calls[0][0] as SendEmailCommand).input.Destination + expect(field === 'cc' ? destination?.CcAddresses : destination?.BccAddresses).toEqual([email]) + expect(capture).toHaveBeenCalledWith( + expect.objectContaining({ id: team.id }), + 'workflows sandbox email sent', + { recipient_count: 2, source: isTest ? 'test' : 'workflow', is_test: isTest } + ) + } + } + ) + + it.each([false])('rechecks membership after slow send preparation (isTest=%s)', async (isTest) => { + const now = Date.now() + const clock = jest.spyOn(Date, 'now').mockReturnValue(now) + const preparation = jest.spyOn(EmailSuppressionService.prototype, 'isSuppressed') + const suppress = preparation.getMockImplementation()! + try { + service = createSandboxService(true) + const first = await service.executeSendEmail(invocation, isTest) + expect(first.error).toBeUndefined() + expect(sendEmailSpy).toHaveBeenCalledTimes(1) + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'DELETE FROM posthog_organizationmembership WHERE user_id = $1', + [memberId], + 'test:revoke-sandbox-member-before-preparation' + ) + clock.mockReturnValue(now + 58_000) + preparation.mockImplementation(async function (this: EmailSuppressionService, ...args) { + const suppressed = await suppress.apply(this, args) + clock.mockReturnValue(now + 63_000) + return suppressed + }) + invocation.queueParameters = createSandboxParams({ + from: { integrationId: 4 }, + cc: memberEmail('cc'), + }) + const result = await service.executeSendEmail(invocation, isTest) + expect(result).toMatchObject({ finished: true, skipped: true, metrics: [], messageAssets: [] }) + expect(result.error).toBeUndefined() + expect(result.invocation.state.vmState?.stack.at(-1)).toEqual({ success: false }) + expect(sendEmailSpy).toHaveBeenCalledTimes(1) + expect(capture).toHaveBeenLastCalledWith( + expect.objectContaining({ id: team.id }), + 'workflows sandbox email blocked', + { reason: 'recipient_not_member', is_test: isTest, blocked_recipient_count: 1 } + ) + } finally { + clock.mockRestore() + preparation.mockRestore() + } + }) + + it.each([false].flatMap((isTest) => [false, true].map((warm) => [isTest, warm] as const)))( + 'fails closed when member SQL fails (isTest=%s, warm=%s)', + async (isTest, warm) => { + const membersPostgres = new PostgresRouter({ + DATABASE_URL: hub.DATABASE_URL, + POSTGRES_CONNECTION_POOL_SIZE: 1, + }) + const now = Date.now() + const clock = jest.spyOn(Date, 'now').mockReturnValue(now) + try { + await membersPostgres.query( + PostgresUse.COMMON_WRITE, + 'CREATE TEMP TABLE posthog_team AS SELECT id, organization_id FROM public.posthog_team WHERE id = $1', + [team.id], + 'test:own-sandbox-team-lookup' + ) + service = createSandboxService(true, null, undefined, membersPostgres) + if (warm) { + const first = await service.executeSendEmail(invocation, isTest) + expect(first.error).toBeUndefined() + expect(sendEmailSpy).toHaveBeenCalledTimes(1) + clock.mockReturnValue(now + 61_000) + } + await membersPostgres.query( + PostgresUse.COMMON_WRITE, + 'SET search_path = pg_temp', + [], + 'test:fail-sandbox-member-lookup' + ) + const { rows } = await membersPostgres.query<{ organization_id: string }>( + PostgresUse.COMMON_WRITE, + 'SELECT organization_id FROM posthog_team WHERE id = $1', + [team.id], + 'test:confirm-sandbox-team-lookup' + ) + expect(rows).toEqual([{ organization_id: team.organization_id }]) + invocation.queueParameters = createSandboxParams({ + from: { integrationId: 4 }, + cc: memberEmail('cc'), + bcc: memberEmail('bcc'), + }) + const result = await service.executeSendEmail(invocation, isTest) + expect(result).toMatchObject({ finished: true, skipped: true, metrics: [], messageAssets: [] }) + expect(result.error).toBeUndefined() + expect(result.invocation.state.vmState?.stack.at(-1)).toEqual({ success: false }) + expect(sendEmailSpy).toHaveBeenCalledTimes(warm ? 1 : 0) + expect(result.logs).toContainEqual( + expect.objectContaining({ + message: `Skipping send: could not check organization members for these addresses: ${memberEmail()}, ${memberEmail('cc')}, ${memberEmail('bcc')}. Try again, or verify your own domain to send to anyone.`, + }) + ) + expect(capture).toHaveBeenLastCalledWith( + expect.objectContaining({ id: team.id }), + 'workflows sandbox email blocked', + { reason: 'check_failed', is_test: isTest, blocked_recipient_count: 3 } + ) + } finally { + clock.mockRestore() + await membersPostgres.end() + } + } + ) + + it.each([false])( + 'blocks previous organization members after a project transfer (isTest=%s)', + async (isTest) => { + const otherOrganization = await createOtherOrganization() + const now = Date.now() + const clock = jest.spyOn(Date, 'now').mockReturnValue(now) + try { + service = createSandboxService(true) + const first = await service.executeSendEmail(invocation, isTest) + expect(first.error).toBeUndefined() + expect(sendEmailSpy).toHaveBeenCalledTimes(1) + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'UPDATE posthog_team SET organization_id = $1 WHERE id = $2', + [otherOrganization, team.id], + 'test:transfer-sandbox-team' + ) + clock.mockReturnValue(now + 61_000) + invocation.queueParameters = createSandboxParams({ from: { integrationId: 4 } }) + + const transferred = await service.executeSendEmail(invocation, isTest) + + expect(transferred).toMatchObject({ finished: true, skipped: true, metrics: [] }) + expect(transferred.error).toBeUndefined() + expect(transferred.invocation.state.vmState?.stack.at(-1)).toEqual({ success: false }) + expect(sendEmailSpy).toHaveBeenCalledTimes(1) + expect(transferred.logs).toContainEqual( + expect.objectContaining({ + message: `Skipping send: the sandbox sender only sends to active organization members with verified email addresses. Blocked addresses: ${memberEmail()}. Verify your own domain to send to anyone.`, + }) + ) + expect(capture).toHaveBeenLastCalledWith( + expect.objectContaining({ id: team.id }), + 'workflows sandbox email blocked', + { reason: 'recipient_not_member', is_test: isTest, blocked_recipient_count: 1 } + ) + } finally { + clock.mockRestore() + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'UPDATE posthog_team SET organization_id = $1 WHERE id = $2', + [team.organization_id, team.id], + 'test:restore-sandbox-team' + ) + } + } + ) + + it.each( + [{ failure: 'a closed member connection', closeConnection: true }].flatMap((failure) => + [false, true].map((isTest) => ({ ...failure, isTest })) + ) + )( + 'fails closed on $failure and bypasses it for own senders (isTest=$isTest)', + async ({ closeConnection, isTest }) => { + const membersPostgres = new PostgresRouter(hub) + if (closeConnection) { + await membersPostgres.end() + } + const memberQueries = jest.spyOn(membersPostgres, 'query') + try { + service = createSandboxService(true, null, undefined, membersPostgres) + + const result = await service.executeSendEmail(invocation, isTest) + + expect(sendEmailSpy).not.toHaveBeenCalled() + expect(result).toMatchObject({ finished: true, skipped: true, metrics: [], messageAssets: [] }) + expect(result.error).toBeUndefined() + expect(result.invocation.state.vmState?.stack).toEqual([{ success: false }]) + expect(result.logs).toContainEqual( + expect.objectContaining({ + level: 'info', + message: `Skipping send: could not check organization members for these addresses: ${memberEmail()}. Try again, or verify your own domain to send to anyone.`, + }) + ) + expect(capture).toHaveBeenCalledWith( + expect.objectContaining({ id: team.id }), + 'workflows sandbox email blocked', + { reason: 'check_failed', is_test: isTest, blocked_recipient_count: 1 } + ) + + invocation.queueParameters = createSandboxParams({ from: { integrationId: 1 } }) + memberQueries.mockClear() + const ownSender = await service.executeSendEmail(invocation, isTest) + expect(ownSender.error).toBeUndefined() + expect(ownSender.skipped).not.toBe(true) + expect(sendEmailSpy).toHaveBeenCalledTimes(1) + expect(memberQueries).not.toHaveBeenCalled() + } finally { + memberQueries.mockRestore() + if (!closeConnection) { + await membersPostgres.end() + } + } + } + ) + + it.each(['membership', 'verification', 'active status', 'lookup failure'])( + 'refreshes cached members within a minute after %s changes', + async (change) => { + const membersPostgres = new PostgresRouter(hub) + const now = Date.now() + const clock = jest.spyOn(Date, 'now').mockReturnValue(now) + let databaseClosed = false + try { + service = createSandboxService(true, null, undefined, membersPostgres) + const params = createSandboxParams({ from: { integrationId: 4 } }) + const first = await service.executeSendEmail(invocation) + expect(first.error).toBeUndefined() + expect(first.skipped).not.toBe(true) + expect(sendEmailSpy).toHaveBeenCalledTimes(1) + if (change === 'membership') { + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'DELETE FROM posthog_organizationmembership WHERE user_id = $1 AND organization_id = $2', + [memberId, team.organization_id], + 'test:remove-sandbox-member' + ) + } else if (change !== 'lookup failure') { + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'UPDATE posthog_user SET is_active = $1, is_email_verified = $2 WHERE id = $3', + [change !== 'active status', change === 'verification' ? null : true, memberId], + 'test:revoke-sandbox-member' + ) + } + clock.mockReturnValue(now + 30_000) + invocation.queueParameters = params + const cached = await service.executeSendEmail(invocation) + expect(cached.error).toBeUndefined() + expect(cached.skipped).not.toBe(true) + expect(sendEmailSpy).toHaveBeenCalledTimes(2) + + if (change === 'lookup failure') { + await membersPostgres.end() + databaseClosed = true + } + clock.mockReturnValue(now + 60_000) + invocation.queueParameters = params + const expired = await service.executeSendEmail(invocation) + expect(expired).toMatchObject({ finished: true, skipped: true, metrics: [] }) + expect(expired.error).toBeUndefined() + expect(expired.invocation.state.vmState?.stack.at(-1)).toEqual({ success: false }) + expect(sendEmailSpy).toHaveBeenCalledTimes(2) + expect(capture).toHaveBeenLastCalledWith( + expect.objectContaining({ id: team.id }), + 'workflows sandbox email blocked', + { + reason: + change === 'lookup failure' + ? 'check_failed' + : change === 'verification' + ? 'recipient_unverified' + : 'recipient_not_member', + is_test: false, + blocked_recipient_count: 1, + } + ) + } finally { + clock.mockRestore() + if (!databaseClosed) { + await membersPostgres.end() + } + } + } + ) + + it.each(['cc', 'bcc'] as const)( + 'blocks ambiguous %s mailboxes rather than sending a partially checked list', + async (field) => { + service = createSandboxService(true) + invocation.queueParameters = createSandboxParams({ + from: { integrationId: 4 }, + [field]: `Example <${memberEmail('cc')}> `, + }) + const result = await service.executeSendEmail(invocation) + expect(sendEmailSpy).not.toHaveBeenCalled() + expect(result).toMatchObject({ finished: true, skipped: true }) + expect(result.invocation.state.vmState?.stack).toEqual([{ success: false }]) + expect(result.logs).toContainEqual( + expect.objectContaining({ + message: expect.stringContaining('outside@example.com'), + }) + ) + } + ) + + it.each([false, true])( + 'sends only the checked address when the To display name contains mailbox syntax (utm=%s)', + async (utmEnabled) => { + service = createSandboxService(true) + invocation.hogFunction.metadata = { utm_tags_enabled: utmEnabled } + invocation.queueParameters = createSandboxParams({ + from: { integrationId: 4 }, + to: { email: memberEmail(), name: 'Example" , "Example' }, + }) + const result = await service.executeSendEmail(invocation) + expect(result.error).toBeUndefined() + expect(sendEmailSpy).toHaveBeenCalledTimes(1) + expect((sendEmailSpy.mock.calls[0][0] as SendEmailCommand).input.Destination?.ToAddresses).toEqual([ + `"Example outside@example.com, Example" <${memberEmail()}>`, + ]) + } + ) + it.each([ ['the switch is off', false, false, {}], ['the switch is off', true, false, {}], @@ -349,7 +1065,7 @@ describe('EmailService', () => { ] as const)( 'skips when the sandbox route is unavailable because %s (isTest=%s)', async (_reason, isTest, enabled: boolean, routeSettings: Partial) => { - service = createSandboxService(enabled, null, undefined, routeSettings) + service = createSandboxService(enabled, null, undefined, hub.postgres, routeSettings) const result = await service.executeSendEmail(invocation, isTest) expect(result).toMatchObject({ finished: true, skipped: true, metrics: [] }) @@ -389,11 +1105,11 @@ describe('EmailService', () => { } as unknown as IngestionOutputs<'message_assets'> service = createSandboxService(true, null, new MessageAssetsService(outputs)) invocation.state.actionId = 'send-email' - invocation.queueParameters = createEmailParams({ + invocation.queueParameters = createSandboxParams({ from: { integrationId: 4, email: 'override@example.com', name: 'Custom sender' }, replyTo: 'reply@example.com', - cc: 'cc@example.com', - bcc: 'bcc@example.com', + cc: memberEmail('cc'), + bcc: memberEmail('bcc'), text: htmlOnly ? undefined : 'Hello there.', html, }) @@ -418,9 +1134,9 @@ describe('EmailService', () => { FromEmailAddress: '"Example organization via PostHog" ', FeedbackForwardingEmailAddress: 'fixed-sandbox@example.com', Destination: { - ToAddresses: ['"Test User" '], - CcAddresses: ['cc@example.com'], - BccAddresses: ['bcc@example.com'], + ToAddresses: [`"Example member" <${memberEmail()}>`], + CcAddresses: [memberEmail('cc')], + BccAddresses: [memberEmail('bcc')], }, Content: { Simple: { @@ -520,7 +1236,7 @@ describe('EmailService', () => { const limiter = await createSandboxLimiter('sandbox-rejected-html-budget-test') service = createSandboxService(true, limiter, new MessageAssetsService(outputs)) invocation.state.actionId = 'send-email' - const params = createEmailParams({ from: { integrationId: 4 }, text, html }) + const params = createSandboxParams({ from: { integrationId: 4 }, text, html }) invocation.queueParameters = params const result = await service.executeSendEmail(invocation, isTest) @@ -543,9 +1259,9 @@ describe('EmailService', () => { expect(result.capturedPostHogEvents.some((event) => event.event === '$workflows_email_sent')).toBe( false ) - expect(params).toEqual(createEmailParams({ from: { integrationId: 4 }, text, html })) + expect(params).toEqual(createSandboxParams({ from: { integrationId: 4 }, text, html })) - invocation.queueParameters = createEmailParams({ from: { integrationId: 1 } }) + invocation.queueParameters = createSandboxParams({ from: { integrationId: 1 } }) const ownSender = await service.executeSendEmail(invocation) expect(ownSender.error).toBeUndefined() expect(ownSender.finished).toBe(true) @@ -555,7 +1271,7 @@ describe('EmailService', () => { it('sends untracked with the fixed identity and organization footer for text-only content', async () => { service = createSandboxService(true) - invocation.queueParameters = createEmailParams({ + invocation.queueParameters = createSandboxParams({ from: { integrationId: 4 }, html: '', text: 'Hello there.', @@ -627,7 +1343,7 @@ describe('EmailService', () => { expect(sandbox.error).toBeUndefined() expect(sendEmailSpy).toHaveBeenCalledTimes(1) - invocation.queueParameters = createEmailParams({ from: { integrationId: 1 } }) + invocation.queueParameters = createSandboxParams({ from: { integrationId: 1 } }) const ownSender = await service.executeSendEmail(invocation) expect(ownSender).toMatchObject({ finished: true }) expect(ownSender.error).toBeUndefined() @@ -637,7 +1353,7 @@ describe('EmailService', () => { expect(exhausted.finished).toBe(false) expect(sendEmailSpy).toHaveBeenCalledTimes(2) - invocation.queueParameters = createEmailParams({ from: { integrationId: 4 } }) + invocation.queueParameters = createSandboxParams({ from: { integrationId: 4 } }) const afterExhaustion = await service.executeSendEmail(invocation) expect(afterExhaustion).toMatchObject({ finished: true }) expect(afterExhaustion.error).toBeUndefined() diff --git a/nodejs/src/cdp/services/messaging/email.service.ts b/nodejs/src/cdp/services/messaging/email.service.ts index aafcefff8fb3..2ff9f7e51116 100644 --- a/nodejs/src/cdp/services/messaging/email.service.ts +++ b/nodejs/src/cdp/services/messaging/email.service.ts @@ -12,11 +12,12 @@ import { IntegrationType, MessageAssetRow, } from '~/cdp/types' -import { createAddLogFunction, logEntry } from '~/cdp/utils' +import { MAX_UNTRUNCATED_LOG_LENGTH, createAddLogFunction, logEntry, sanitizeLogMessage } from '~/cdp/utils' import { createInvocationResult } from '~/cdp/utils/invocation-utils' import { logger } from '~/common/utils/logger' import { IntegrationManagerService } from '../managers/integration-manager.service' +import { OrganizationMembersService, TeamOrganizationSnapshot } from '../managers/organization-members.service' import { RecipientManagerRecipient, RecipientsManagerService } from '../managers/recipients-manager.service' import { TeamWorkflowsConfigService } from '../managers/team-workflows-config.service' import { RateLimiterService } from '../rate-limiter/rate-limiter.service' @@ -345,6 +346,115 @@ export function parseAddressList(value?: string): string[] | undefined { return result.length > 0 ? result : undefined } +type QuoteScan = { unquotedIndexes: number[]; balanced: boolean } + +function scanQuotes(text: string): QuoteScan { + const unquotedIndexes: number[] = [] + let quoted = false + let escaped = false + for (let index = 0; index < text.length; index++) { + const character = text[index] + if (escaped) { + escaped = false + } else if (quoted && character === '\\') { + escaped = true + } else if (character === '"') { + quoted = !quoted + } else if (!quoted) { + unquotedIndexes.push(index) + } + } + return { unquotedIndexes, balanced: !quoted } +} + +function splitOutsideQuotes(text: string): string[] { + const commas = scanQuotes(text).unquotedIndexes.filter((index) => text[index] === ',') + return [-1, ...commas].map((comma, position) => text.slice(comma + 1, commas[position] ?? text.length)) +} + +function angleAddress(entry: string): string | undefined { + const { unquotedIndexes, balanced } = scanQuotes(entry) + const opening = unquotedIndexes.find((index) => entry[index] === '<') + const closing = entry.length - 1 + if (!balanced || opening === undefined || entry[closing] !== '>' || /[\r\n]/.test(entry)) { + return undefined + } + const unquotedBetween = (start: number, end: number): string => + unquotedIndexes + .filter((index) => index >= start && index < end) + .map((index) => entry[index]) + .join('') + const mailbox = entry.slice(opening + 1, closing).trim() + const isFramed = !/[>@;]/.test(unquotedBetween(0, opening)) && !/[<>]/.test(unquotedBetween(opening + 1, closing)) + return isFramed && mailbox ? mailbox : undefined +} + +function sandboxAddressList(value?: string): string[] { + return splitOutsideQuotes(value ?? '') + .map((entry) => entry.trim()) + .filter(Boolean) + .map((entry) => angleAddress(entry) ?? entry) +} + +type SandboxRecipientBlockReason = 'recipient_not_member' | 'recipient_unverified' | 'check_failed' + +const SANDBOX_RECIPIENT_BLOCK_COPY: Record< + SandboxRecipientBlockReason, + { explanation: string; label: string; guidance: string } +> = { + recipient_not_member: { + explanation: + 'Skipping send: the sandbox sender only sends to active organization members with verified email addresses', + label: '. Blocked addresses: ', + guidance: 'Verify your own domain to send to anyone.', + }, + recipient_unverified: { + explanation: + 'Skipping send: the sandbox sender only sends to organization members with verified email addresses', + label: '. Unverified addresses: ', + guidance: 'Verify your email address in PostHog, or ask these members to verify theirs.', + }, + check_failed: { + explanation: 'Skipping send: could not check organization members', + label: ' for these addresses: ', + guidance: 'Try again, or verify your own domain to send to anyone.', + }, +} + +const BLOCKED_ADDRESSES_LABEL = 'Blocked addresses: ' + +function splitByLength(text: string, maxLength: number): string[] { + const chunks = [''] + for (const character of text) { + if (chunks[chunks.length - 1].length + character.length > maxLength) { + chunks.push('') + } + chunks[chunks.length - 1] += character + } + return chunks +} + +function packAddressRows(addresses: string[], maxLength: number): string[] { + const rows: string[] = [] + let row = '' + for (const piece of addresses.flatMap((address) => splitByLength(address, maxLength))) { + const joined = row ? `${row}, ${piece}` : piece + if (row && joined.length > maxLength) { + rows.push(row) + row = piece + } else { + row = joined + } + } + return [...rows, row] +} + +function blockedAddressLogRows(addresses: string[]): string[] { + return packAddressRows(addresses, MAX_UNTRUNCATED_LOG_LENGTH - BLOCKED_ADDRESSES_LABEL.length).map( + (addressList) => `${BLOCKED_ADDRESSES_LABEL}${addressList}` + ) +} + export class EmailService { sesV2Client: SESv2Client | null @@ -363,7 +473,8 @@ export class EmailService { private messageAssetsService?: MessageAssetsService, private workflowEmailRateLimiter: RateLimiterService | null = null, private teamEmailRateLimiter: RateLimiterService | null = null, - private sandboxSender: SandboxEmailSender | null = null + private sandboxSender: SandboxEmailSender | null = null, + private organizationMembers: OrganizationMembersService | null = null ) { this.sesV2Client = this.sesConfig.sesRegion ? new SESv2Client({ @@ -407,6 +518,7 @@ export class EmailService { let assetRow: MessageAssetRow | null = null let trackingEnabled = true let deliveryParams = params + let sandboxRecipients: string[] | undefined try { // Team-level kill switches: staff suspend all workflow email for a team whose sender @@ -512,13 +624,6 @@ export class EmailService { } const from = this.resolveFromSender(integration, params.from, addLog) - if (isSandbox && (params.from.email || params.from.name || params.replyTo)) { - addLog( - 'info', - 'Ignoring custom sender and Reply-To settings: the sandbox sender uses a fixed identity.' - ) - } - // An empty `to` comes from the customer's data (a person without an email property, or no // person at all), not from a fault in the send path. SES rejects it as an invalid address, // so record a skip with its own metric instead of an `email_failed`. Otherwise one project @@ -543,11 +648,39 @@ export class EmailService { return result } + if (isSandbox) { + const cc = sandboxAddressList(params.cc) + const bcc = sandboxAddressList(params.bcc) + sandboxRecipients = [params.to.email, ...cc, ...bcc] + if (!(await this.checkSandboxRecipients(result, sandboxRecipients, isTest))) { + return result + } + deliveryParams = { + ...params, + to: { + email: params.to.email.trim(), + name: params.to.name ? sanitizeFromName(params.to.name) : undefined, + }, + cc: cc.join(', ') || undefined, + bcc: bcc.join(', ') || undefined, + } + } + if (isSandbox && (params.from.email || params.from.name || params.replyTo)) { + addLog( + 'info', + 'Ignoring custom sender and Reply-To settings: the sandbox sender uses a fixed identity.' + ) + } + // Single choke point for the suppression check — every send path lands here regardless // of whether the invocation came from a workflow action or an email destination hog // function. Checking here means callers can't bypass it by taking a different upstream // route. Covers `to`, `cc`, and `bcc`; a suppressed address anywhere blocks the send. - const skipReason = await this.buildSuppressionSkipReason(invocation.teamId, params) + const skipReason = await this.buildSuppressionSkipReason( + invocation.teamId, + deliveryParams, + sandboxRecipients + ) if (skipReason) { addLog('info', skipReason) if (!isTest) { @@ -616,7 +749,10 @@ export class EmailService { // Charged per recipient, not per send: SES counts every to/cc/bcc address against its // own quota, so a send with many copies must spend that many tokens. const capRecipients = - 1 + extractEmailsFromAddressList(params.cc).length + extractEmailsFromAddressList(params.bcc).length + sandboxRecipients?.length ?? + 1 + + extractEmailsFromAddressList(deliveryParams.cc).length + + extractEmailsFromAddressList(deliveryParams.bcc).length const capDelay = isSandbox ? null : await this.claimTeamSendingBudget(invocation, isTest, capRecipients) if (capDelay) { result.finished = false @@ -637,18 +773,18 @@ export class EmailService { // A reply-to set on the email step wins over the sender's default. const paramsWithReplyTo = - parseAddressList(params.replyTo) || !integration.config.reply_to - ? params - : { ...params, replyTo: integration.config.reply_to } + parseAddressList(deliveryParams.replyTo) || !integration.config.reply_to + ? deliveryParams + : { ...deliveryParams, replyTo: integration.config.reply_to } // Tagged before click tracking wraps the links, so the tags land on the destination URL. const metadata = invocation.hogFunction.metadata const sendParams = - metadata?.utm_tags_enabled === true && params.html + metadata?.utm_tags_enabled === true && paramsWithReplyTo.html ? { ...paramsWithReplyTo, html: addUtmTagsToEmail( - params.html, + paramsWithReplyTo.html, resolveUtmTags( { utm_source: 'posthog', @@ -682,7 +818,18 @@ export class EmailService { from.name, invocation.teamId ) - await this.sendEmailWithSES(result, deliveryParams, from, trackingEnabled, integration, isTest) + if ( + !(await this.sendEmailWithSES( + result, + deliveryParams, + from, + trackingEnabled, + integration, + isTest + )) + ) { + return result + } await this.sandboxSender!.capture(invocation.teamId, isTest, { type: 'sent', recipientCount: capRecipients, @@ -912,6 +1059,69 @@ export class EmailService { return null } + private async checkSandboxRecipients( + result: CyclotronJobInvocationResult, + recipients: string[], + isTest: boolean, + organization?: TeamOrganizationSnapshot + ): Promise { + const invocation = result.invocation + let blockedRecipients: string[] + let reason: SandboxRecipientBlockReason + try { + if (!this.organizationMembers) { + throw new Error('Organization member checks are not configured') + } + const blocked = await this.organizationMembers.getBlockedRecipients( + invocation.teamId, + recipients, + organization + ) + blockedRecipients = blocked.map(({ address }) => address) + reason = blocked.every(({ unverifiedMember }) => unverifiedMember) + ? 'recipient_unverified' + : 'recipient_not_member' + } catch (error) { + blockedRecipients = recipients + reason = 'check_failed' + logger.warn('Could not check sandbox email recipients', { + teamId: invocation.teamId, + error: error instanceof Error ? error.message : 'Unknown error', + }) + } + if (!blockedRecipients.length) { + return true + } + this.logSandboxRecipientBlock(result, blockedRecipients, reason) + result.skipped = true + result.invocation.state.vmState?.stack.push({ success: false }) + await this.sandboxSender!.capture(invocation.teamId, isTest, { + type: 'blocked', + reason, + blockedRecipientCount: blockedRecipients.length, + }) + return false + } + + private logSandboxRecipientBlock( + result: CyclotronJobInvocationResult, + blockedRecipients: string[], + reason: SandboxRecipientBlockReason + ): void { + const addLog = createAddLogFunction(result.logs) + const addresses = blockedRecipients.map((address) => address.trim()) + const { explanation, label, guidance } = SANDBOX_RECIPIENT_BLOCK_COPY[reason] + const message = `${explanation}${label}${addresses.join(', ')}. ${guidance}` + if (sanitizeLogMessage([message]) === message) { + addLog('info', message) + return + } + addLog('info', `${explanation}. ${guidance}`) + for (const row of blockedAddressLogRows(addresses)) { + addLog('info', row) + } + } + // Returns a human-readable log string when any destination address is suppressed for the team, // or null when the send should proceed. Scans to + cc + bcc — SES delivers to every list, so a // suppressed address anywhere blocks the whole send. `cc` and `bcc` can be comma-separated @@ -919,14 +1129,14 @@ export class EmailService { // matching against the normalized suppression identifier. private async buildSuppressionSkipReason( teamId: number, - params: CyclotronInvocationQueueParametersEmailType + params: CyclotronInvocationQueueParametersEmailType, + recipientEmails?: string[] ): Promise { - const recipients: string[] = [] - if (params.to?.email && params.to.email.trim()) { - recipients.push(params.to.email.trim()) - } - recipients.push(...extractEmailsFromAddressList(params.cc)) - recipients.push(...extractEmailsFromAddressList(params.bcc)) + const recipients = recipientEmails ?? [ + ...(params.to?.email?.trim() ? [params.to.email.trim()] : []), + ...extractEmailsFromAddressList(params.cc), + ...extractEmailsFromAddressList(params.bcc), + ] if (recipients.length === 0) { return null } @@ -1163,7 +1373,7 @@ export class EmailService { trackingEnabled: boolean, integration: IntegrationType, isTest = false - ): Promise { + ): Promise { if (!this.sesV2Client) { throw new Error('SES is not configured - set SES_REGION and AWS credentials') } @@ -1259,19 +1469,30 @@ export class EmailService { } const replyToAddresses = parseAddressList(params.replyTo) - const ccAddresses = parseAddressList(params.cc) - const bccAddresses = parseAddressList(params.bcc) + const parseRecipients = integration.config.provider === 'sandbox' ? sandboxAddressList : parseAddressList + const ccAddresses = parseRecipients(params.cc) + const bccAddresses = parseRecipients(params.bcc) if (replyToAddresses) { sendEmailParams.ReplyToAddresses = replyToAddresses } - if (ccAddresses) { + if (ccAddresses?.length) { sendEmailParams.Destination!.CcAddresses = ccAddresses } - if (bccAddresses) { + if (bccAddresses?.length) { sendEmailParams.Destination!.BccAddresses = bccAddresses } + if ( + integration.config.provider === 'sandbox' && + !(await this.checkSandboxRecipients( + result, + [params.to.email, ...(ccAddresses ?? []), ...(bccAddresses ?? [])], + isTest + )) + ) { + return false + } try { const response = await this.sesV2Client.send(new SendEmailCommand(sendEmailParams)) if (!response.MessageId) { @@ -1285,6 +1506,7 @@ export class EmailService { const message = error instanceof Error ? error.message : String(error) throw new Error(`Failed to send email via SES: ${message}`) } + return true } private generateUnsubscribeHeaders( diff --git a/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts b/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts index 8a271c97538d..0186bcff2c23 100644 --- a/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts +++ b/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts @@ -1,6 +1,7 @@ import { type DefaultTreeAdapterMap, defaultTreeAdapter, parse, parseFragment, serialize } from 'parse5' import { CyclotronInvocationQueueParametersEmailType } from '~/cdp/schema/cyclotron' +import { RedisV2 } from '~/common/redis/redis-v2' import { logger } from '~/common/utils/logger' import { captureTeamEvent } from '~/common/utils/posthog' import { TeamManager } from '~/common/utils/team-manager' @@ -28,7 +29,11 @@ function emptyRouteSettings(config: SandboxEmailSenderConfig): string[] { type SandboxEmailOutcome = | { type: 'sent'; recipientCount: number } - | { type: 'blocked'; reason: 'switch_off'; blockedRecipientCount: number } + | { + type: 'blocked' + reason: 'switch_off' | 'recipient_not_member' | 'recipient_unverified' | 'check_failed' + blockedRecipientCount: number + } function htmlBody(document: DefaultTreeAdapterMap['document']): DefaultTreeAdapterMap['element'] | undefined { const root = document.childNodes.find(defaultTreeAdapter.isElementNode) @@ -116,7 +121,8 @@ export class SandboxEmailSender { constructor( public readonly config: SandboxEmailSenderConfig, - private teamManager: TeamManager + private teamManager: TeamManager, + private blockedEventRedis: Pick | null = null ) { this.config = { ...config, @@ -160,6 +166,22 @@ export class SandboxEmailSender { public async capture(teamId: number, isTest: boolean, outcome: SandboxEmailOutcome): Promise { try { + if (outcome.type === 'blocked') { + const firstBlock = await this.blockedEventRedis?.useClient( + { name: 'sandbox-blocked-event', timeout: 1000 }, + (client) => + client.set( + `@posthog/workflows-sandbox-blocked/${teamId}/${outcome.reason}/${isTest ? 'test' : 'live'}`, + '1', + 'EX', + 3600, + 'NX' + ) + ) + if (firstBlock !== 'OK') { + return + } + } const team = await this.teamManager.getTeam(teamId) if (team) { const properties = diff --git a/nodejs/src/cdp/utils.ts b/nodejs/src/cdp/utils.ts index 78401817b4b8..a28cd207380e 100644 --- a/nodejs/src/cdp/utils.ts +++ b/nodejs/src/cdp/utils.ts @@ -15,6 +15,7 @@ import { HogFunctionInvocationGlobals, HogFunctionType, LogEntry, LogEntrySerial export const CDP_TEST_ID = '[CDP-TEST-HIDDEN]' export const MAX_LOG_LENGTH = 10000 const TRUNCATION_SUFFIX = '... (truncated)' +export const MAX_UNTRUNCATED_LOG_LENGTH = MAX_LOG_LENGTH - TRUNCATION_SUFFIX.length const REDACTED = '***REDACTED***' // Sync with person.py and constants.tsx From 222ac4e5617c4aa7d1cfa0f612b6b894b386ae45 Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Mon, 26 Oct 2026 04:56:14 +0000 Subject: [PATCH 2/2] feat(workflows): stop sandbox email for deactivated organizations A deactivated organization now has no verified members, so the sandbox sender blocks every send within the one-minute member snapshot window. Only an explicit false deactivates, matching the web middleware. Co-Authored-By: Claude Opus 5.5 --- .../managers/organization-members.service.ts | 5 +++- .../services/messaging/email.service.test.ts | 26 +++++++++++++++++++ 2 files changed, 30 insertions(+), 1 deletion(-) diff --git a/nodejs/src/cdp/services/managers/organization-members.service.ts b/nodejs/src/cdp/services/managers/organization-members.service.ts index 14dd567233cf..9c39ab4194ee 100644 --- a/nodejs/src/cdp/services/managers/organization-members.service.ts +++ b/nodejs/src/cdp/services/managers/organization-members.service.ts @@ -56,7 +56,10 @@ export class OrganizationMembersService { `SELECT membership.organization_id, users.email, users.is_email_verified FROM posthog_organizationmembership AS membership JOIN posthog_user AS users ON users.id = membership.user_id - WHERE membership.organization_id = ANY($1::uuid[]) AND users.is_active = true`, + JOIN posthog_organization AS organization ON organization.id = membership.organization_id + WHERE membership.organization_id = ANY($1::uuid[]) + AND users.is_active = true + AND organization.is_active IS NOT FALSE`, [organizationIds], 'fetch-organization-email-members' ) diff --git a/nodejs/src/cdp/services/messaging/email.service.test.ts b/nodejs/src/cdp/services/messaging/email.service.test.ts index ea7408b4e358..3ca0fce8c2f5 100644 --- a/nodejs/src/cdp/services/messaging/email.service.test.ts +++ b/nodejs/src/cdp/services/messaging/email.service.test.ts @@ -256,6 +256,14 @@ describe('EmailService', () => { ) return id } + const setOrganizationActive = async (isActive: boolean | null): Promise => { + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'UPDATE posthog_organization SET is_active = $1 WHERE id = $2', + [isActive, team.organization_id], + 'test:set-sandbox-organization-active' + ) + } const createOtherOrganization = async (): Promise => { const id = await createOrganization(hub.postgres) extraOrganizationIds.push(id) @@ -563,6 +571,7 @@ describe('EmailService', () => { 'legacy', 'inactive', 'other organization', + 'deactivated organization', ].map((recipient) => [isTest, recipient] as const) ) )('blocks all delivery for %s / %s', async (isTest, recipient) => { @@ -581,6 +590,9 @@ describe('EmailService', () => { blocked = [memberEmail('outside')] await createMember(blocked[0], await createOtherOrganization()) params = { ...params, to: { email: blocked[0] } } + } else if (recipient === 'deactivated organization') { + blocked = [memberEmail()] + await setOrganizationActive(false) } else { blocked = [memberEmail()] await hub.postgres.query( @@ -622,6 +634,20 @@ describe('EmailService', () => { expect(result.capturedPostHogEvents).toEqual([]) }) + it.each([null, true])( + 'sends to a verified member of an organization that was never deactivated (is_active=%s)', + async (isActive) => { + await setOrganizationActive(isActive) + service = createSandboxService(true) + + const result = await service.executeSendEmail(invocation) + + expect(result.error).toBeUndefined() + expect(result.skipped).not.toBe(true) + expect(sendEmailSpy).toHaveBeenCalledTimes(1) + } + ) + it.each( [false].flatMap((isTest) => [