feat(workflows): restrict sandbox email to verified org members

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.

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)

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)
+    })
+})

nodejs/src/cdp/services/managers/organization-members.service.ts

@@ -0,0 +1,130 @@
+import { PostgresRouter, PostgresUse } from '~/common/utils/db/postgres'
+import { LazyLoader } from '~/common/utils/lazy-loader'
+
+type OrganizationMemberSnapshot = {
+    verifiedEmails: ReadonlySet<string>
+    unverifiedEmails: ReadonlySet<string>
+    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<OrganizationMemberSnapshot>
+    private organizations: LazyLoader<TeamOrganizationSnapshot>
+
+    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
+                     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'
+                )
+                const members: Record<string, { verified: Set<string>; unverified: Set<string> }> = Object.fromEntries(
+                    organizationIds.map((id) => [id, { verified: new Set<string>(), unverified: new Set<string>() }])
+                )
+                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<TeamOrganizationSnapshot> {
+        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<BlockedRecipient[]> {
+        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()),
+            }))
+    }
+}