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

docs/internal/workflows-email-sending-tiers.md

@@ -74,3 +74,15 @@ Decay, suspension drops, admin recomputes, and the backfill stay silent.
   A send denied because the limiter itself failed keeps the shorter token bucket cadence, at most 5 to 10 minutes.
 - Gmail provides no per-message feedback loop, so complaint rates (ours and AWS's) cannot see Gmail complaints at all.
 - The `HOGFLOW_BATCH_TRIGGER_ELEVATED_TEAM_IDS` allowlist elevates a team to the top tier in Django only: the UI, the admin, and the batch audience cap show top-tier numbers, but the worker's send-time buckets read the stored tier. An allowlisted team below the top tier is therefore throttled at its stored tier while being told otherwise, and its shadow-mode delay counts overstate real impact. Before enforcing, pin each allowlisted team at the top tier from the Django admin and empty the allowlist, rather than plumbing the list into the worker.
+
+### Sandbox delivery
+
+The worker sends through a `sandbox` email integration only when `WORKFLOWS_SANDBOX_SENDER_ENABLED` is on and `SES_SANDBOX_TENANT_NAME`, `SES_SANDBOX_CONFIGURATION_SET`, and `SES_SANDBOX_FROM_ADDRESS` are set.
+Otherwise every send through it skips with a reason in the run log and a `workflows sandbox email blocked` event with reason `switch_off`.
+Every To, CC, and BCC address must belong to an active member of the team's active organization with a verified email address.
+Membership snapshots expire within one minute, and the worker checks recipients again right before delivery.
+A failed or stale membership check skips the send and lists the blocked addresses in the run log.
+The blocked event fires at most once an hour per team, reason, and send kind (workflow or test).
+Sandbox sends never spend the team's tier budget, because they reach only its own members.
+The sandbox route is not built yet, so a send that passes every check still skips as unavailable.
+Keep the switch off in production until the sandbox caps are deployed.

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'
@@ -43,6 +44,7 @@ import { MessageAssetsService } from './services/messaging/message-assets.servic
 import { PushNotificationService } from './services/messaging/push-notification.service'
 import { RecipientPreferencesService } from './services/messaging/recipient-preferences.service'
 import { RecipientTokensService } from './services/messaging/recipient-tokens.service'
+import { SandboxEmailSender } from './services/messaging/sandbox-email-sender'
 import { HogFunctionMonitoringService } from './services/monitoring/hog-function-monitoring.service'
 import { HogInvocationResultsService } from './services/monitoring/hog-invocation-results.service'
 import { HogWatcherService } from './services/monitoring/hog-watcher.service'
@@ -171,6 +173,10 @@ export type CdpCoreServicesConfig = Pick<
         | 'SES_ENDPOINT'
         | 'SES_TRACKED_CONFIGURATION_SET'
         | 'SES_UNTRACKED_CONFIGURATION_SET'
+        | 'WORKFLOWS_SANDBOX_SENDER_ENABLED'
+        | 'SES_SANDBOX_TENANT_NAME'
+        | 'SES_SANDBOX_CONFIGURATION_SET'
+        | 'SES_SANDBOX_FROM_ADDRESS'
         | 'EMAIL_SUPPRESSION_TRANSIENT_BOUNCE_THRESHOLD'
         | 'EMAIL_TEAM_SENDING_CAP_MODE'
         | 'EMAIL_TEAM_SENDING_CAP_HOURLY_BY_TIER'
@@ -482,7 +488,18 @@ export function createCdpCoreServices(
         emailStatusService,
         messageAssetsService,
         workflowEmailRateLimiter,
-        teamEmailRateLimiter
+        teamEmailRateLimiter,
+        new SandboxEmailSender(
+            {
+                enabled: config.WORKFLOWS_SANDBOX_SENDER_ENABLED,
+                tenantName: config.SES_SANDBOX_TENANT_NAME,
+                configurationSetName: config.SES_SANDBOX_CONFIGURATION_SET,
+                fromAddress: config.SES_SANDBOX_FROM_ADDRESS,
+            },
+            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/config.ts

@@ -171,6 +171,11 @@ export type CdpConfig = ClickhouseConfig & {
     // Configuration set without open/click tracking (same delivery/bounce/complaint event destination).
     // Empty means not provisioned: tracking-off sends fall back to the tracked set with a warning.
     SES_UNTRACKED_CONFIGURATION_SET: string
+    // The sandbox sender delivers only with the switch on and all three route settings set.
+    WORKFLOWS_SANDBOX_SENDER_ENABLED: boolean
+    SES_SANDBOX_TENANT_NAME: string
+    SES_SANDBOX_CONFIGURATION_SET: string
+    SES_SANDBOX_FROM_ADDRESS: string
     // Comma-separated allowlist of SNS Topic ARNs the SES webhook accepts events from. Empty string
     // means no restriction (dev/test); production should set this to the workflow SES topic ARN(s).
     SES_ALLOWED_SNS_TOPIC_ARNS: string
@@ -417,6 +422,10 @@ export function getDefaultCdpConfig(): CdpConfig {
         SES_REGION: isTestEnv() || isDevEnv() ? 'us-east-1' : '',
         SES_TRACKED_CONFIGURATION_SET: 'posthog-messaging',
         SES_UNTRACKED_CONFIGURATION_SET: '',
+        WORKFLOWS_SANDBOX_SENDER_ENABLED: false,
+        SES_SANDBOX_TENANT_NAME: '',
+        SES_SANDBOX_CONFIGURATION_SET: '',
+        SES_SANDBOX_FROM_ADDRESS: '',
         SES_ALLOWED_SNS_TOPIC_ARNS: '',
         EMAIL_SUPPRESSION_TRANSIENT_BOUNCE_THRESHOLD: 5,
 

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

@@ -0,0 +1,41 @@
+import { PostgresRouter } from '~/common/utils/db/postgres'
+
+import { OrganizationMembersService } from './organization-members.service'
+
+describe('OrganizationMembersService', () => {
+    const memberRow = (email: string): { team_id: number; email: string; is_email_verified: boolean } => ({
+        team_id: 1,
+        email,
+        is_email_verified: true,
+    })
+
+    it('fails closed when every member lookup outlasts its one-minute decision', async () => {
+        let now = 0
+        const query = jest.fn(() => {
+            now += 60_000
+            return Promise.resolve({ rows: [memberRow('member@example.com')] })
+        })
+        const service = new OrganizationMembersService({ query } as unknown as PostgresRouter, () => now)
+
+        await expect(service.getBlockedRecipients(1, ['member@example.com'])).rejects.toThrow('expired')
+        expect(query).toHaveBeenCalledTimes(2)
+    })
+
+    it('shares one member lookup across sends and reloads it after a minute', async () => {
+        let now = 0
+        let member = 'member@example.com'
+        const query = jest.fn(() => Promise.resolve({ rows: [memberRow(member)] }))
+        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).toHaveBeenCalledTimes(1)
+
+        now = 60_001
+        member = 'new-member@example.com'
+        expect(await service.getBlockedRecipients(1, ['member@example.com'])).toEqual([
+            { address: 'member@example.com', unverifiedMember: false },
+        ])
+        expect(query).toHaveBeenCalledTimes(2)
+    })
+})

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

@@ -0,0 +1,86 @@
+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 }
+
+const MAX_SNAPSHOT_AGE_MS = 60_000
+const REFRESH_AGE_MS = MAX_SNAPSHOT_AGE_MS - 1_000
+
+const normalizeEmail = (email: string): string => email.trim().toLowerCase()
+
+export class OrganizationMembersService {
+    private snapshots: LazyLoader<OrganizationMemberSnapshot>
+
+    constructor(
+        postgres: PostgresRouter,
+        private now: () => number = Date.now
+    ) {
+        this.snapshots = new LazyLoader({
+            name: 'sandbox_team_organization_members',
+            refreshAgeMs: REFRESH_AGE_MS,
+            refreshJitterMs: 0,
+            bufferMs: 0,
+            loader: async (teamIds) => {
+                const expiresAt = this.now() + MAX_SNAPSHOT_AGE_MS
+                // Only an explicit false deactivates an organization, matching the web middleware.
+                const { rows } = await postgres.query<{ team_id: number; email: string; is_email_verified: boolean }>(
+                    PostgresUse.COMMON_WRITE,
+                    `SELECT team.id AS team_id, users.email, users.is_email_verified
+                     FROM posthog_team AS team
+                     JOIN posthog_organization AS organization ON organization.id = team.organization_id
+                     JOIN posthog_organizationmembership AS membership ON membership.organization_id = organization.id
+                     JOIN posthog_user AS users ON users.id = membership.user_id
+                     WHERE team.id = ANY($1::bigint[])
+                       AND organization.is_active IS NOT FALSE
+                       AND users.is_active = true`,
+                    [teamIds.map(Number)],
+                    'fetch-sandbox-team-organization-members'
+                )
+                const members = new Map(
+                    teamIds.map((id) => [id, { verified: new Set<string>(), unverified: new Set<string>() }])
+                )
+                for (const row of rows) {
+                    const { verified, unverified } = members.get(String(row.team_id))!
+                    ;(row.is_email_verified === true ? verified : unverified).add(normalizeEmail(row.email))
+                }
+                return Object.fromEntries(
+                    [...members].map(([id, { verified, unverified }]) => [
+                        id,
+                        { verifiedEmails: verified, unverifiedEmails: unverified, expiresAt },
+                    ])
+                )
+            },
+        })
+    }
+
+    public async getBlockedRecipients(teamId: number, recipients: string[]): Promise<BlockedRecipient[]> {
+        const snapshot = await this.getFreshSnapshot(String(teamId))
+        return recipients
+            .filter((email) => !snapshot.verifiedEmails.has(normalizeEmail(email)))
+            .map((address) => ({ address, unverifiedMember: snapshot.unverifiedEmails.has(normalizeEmail(address)) }))
+    }
+
+    // A snapshot older than a minute never decides a send: it is reloaded once, and a reload that
+    // itself outlasts the minute fails the check rather than trusting stale membership.
+    private async getFreshSnapshot(teamId: string): Promise<OrganizationMemberSnapshot> {
+        let snapshot = await this.snapshots.get(teamId)
+        if (snapshot && this.now() >= snapshot.expiresAt) {
+            this.snapshots.markForRefresh(teamId)
+            snapshot = await this.snapshots.get(teamId)
+        }
+        if (!snapshot) {
+            throw new Error('Could not check organization members')
+        }
+        if (this.now() >= snapshot.expiresAt) {
+            this.snapshots.markForRefresh(teamId)
+            throw new Error('The organization member check expired')
+        }
+        return snapshot
+    }
+}