${paragraph}
`) + .join('')}`, + }, + ) + } catch (error) { + console.warn('email-outbound-pause-notification-failed', error) + } +} diff --git a/packages/worker/src/email/outbound-abuse.workers.test.ts b/packages/worker/src/email/outbound-abuse.workers.test.ts new file mode 100644 index 0000000000..b24f863625 --- /dev/null +++ b/packages/worker/src/email/outbound-abuse.workers.test.ts @@ -0,0 +1,225 @@ +import { env } from 'cloudflare:workers' +import { expect, test } from 'vitest' +import { createStableUserIdFromEmail } from '#worker/user-id.ts' +import { consoleWarn } from '#worker/test-support/console-spies.ts' +import { processCloudflareEmailDeliveryEvent } from './delivery-events.ts' +import { + applyOutboundEmailAbusePause, + emailOutboundPausedMessage, + outboundEmailBouncePauseThresholdPerDay, +} from './outbound-abuse.ts' +import { sendOutboundEmail } from './outbound.ts' +import { insertEmailMessage } from './repo.ts' +import { ensureEmailTestSchema } from './test-schema.ts' + +const platformBaseUrl = 'https://kody.example.com' + +async function seedVerifiedAccount(input: { email: string }) { + const username = `sender-${crypto.randomUUID().slice(0, 8)}` + const stableUserId = await createStableUserIdFromEmail(input.email) + await env.APP_DB.prepare( + `INSERT INTO users (username, email, password_hash, email_verified_at, plan, stable_user_id) + VALUES (?, ?, ?, ?, 'max', ?)`, + ) + .bind( + username, + input.email, + 'test-password-hash', + new Date().toISOString(), + stableUserId, + ) + .run() + return { username, stableUserId } +} + +async function recordProviderEvent(input: { + providerMessageId: string + eventId: string + status: 'delivered' | 'bounced' | 'complained' + eventTimestamp: string +}) { + const result = await processCloudflareEmailDeliveryEvent({ + db: env.APP_DB, + body: { + type: `cf.email.sending.message.${input.status}`, + source: { + type: 'email.sending', + zoneId: 'zone-1', + domain: 'inbox.kody.example.com', + }, + payload: { + eventId: input.eventId, + messageId: input.providerMessageId, + sender: 'user@inbox.kody.example.com', + recipient: 'recipient@example.net', + terminal: true, + delivery: { status: input.status }, + }, + metadata: { + accountId: 'account-1', + eventSubscriptionId: 'subscription-1', + eventSchemaVersion: 1, + eventTimestamp: input.eventTimestamp, + }, + }, + }) + expect(result.outcome).not.toBe('invalid') + return result +} + +async function readPauseTimestamp(stableUserId: string) { + const row = await env.APP_DB.prepare( + `SELECT email_outbound_paused_at FROM users WHERE stable_user_id = ?`, + ) + .bind(stableUserId) + .first<{ email_outbound_paused_at: string | null }>() + return row?.email_outbound_paused_at ?? null +} + +test('a spam complaint pauses outbound email once and blocks further sends', async () => { + consoleWarn.mockImplementation(() => {}) + await ensureEmailTestSchema(env.APP_DB) + const accountEmail = `complainer-${crypto.randomUUID()}@example.com` + const { stableUserId } = await seedVerifiedAccount({ email: accountEmail }) + const providerMessageId = `provider-${crypto.randomUUID()}` + await insertEmailMessage({ + db: env.APP_DB, + message: { + direction: 'outbound', + userId: stableUserId, + fromAddress: 'user@inbox.kody.example.com', + toAddresses: ['recipient@example.net'], + subject: 'Complaint pause', + processingStatus: 'sent', + providerMessageId, + sentAt: '2026-07-17T20:00:00.000Z', + }, + }) + await recordProviderEvent({ + providerMessageId, + eventId: 'event-complained', + status: 'complained', + eventTimestamp: new Date().toISOString(), + }) + + const first = await applyOutboundEmailAbusePause({ + env, + userId: stableUserId, + deliveryStatus: 'complained', + eventRecorded: true, + }) + expect(first.paused).toBe(true) + const pausedAt = await readPauseTimestamp(stableUserId) + expect(pausedAt).not.toBeNull() + expect( + consoleWarn.mock.calls.some((call) => call[0] === 'email-outbound-paused'), + ).toBe(true) + + // Replayed queue messages must not re-pause or move the timestamp. + const second = await applyOutboundEmailAbusePause({ + env, + userId: stableUserId, + deliveryStatus: 'complained', + eventRecorded: false, + }) + expect(second.paused).toBe(false) + expect(await readPauseTimestamp(stableUserId)).toBe(pausedAt) + + await expect( + sendOutboundEmail({ + env: { + ...env, + APP_BASE_URL: platformBaseUrl, + EMAIL: { + async send() { + throw new Error('Paused accounts must not reach the provider.') + }, + }, + }, + userId: stableUserId, + accountEmail, + recipientPolicy: 'self', + subject: 'Blocked while paused', + text: 'Body', + }), + ).rejects.toThrow(emailOutboundPausedMessage) +}) + +test('bounces below the daily threshold do not pause; reaching it does', async () => { + consoleWarn.mockImplementation(() => {}) + await ensureEmailTestSchema(env.APP_DB) + const accountEmail = `bouncer-${crypto.randomUUID()}@example.com` + const { stableUserId } = await seedVerifiedAccount({ email: accountEmail }) + const now = new Date() + + for ( + let attempt = 1; + attempt <= outboundEmailBouncePauseThresholdPerDay; + attempt += 1 + ) { + const providerMessageId = `provider-${crypto.randomUUID()}` + await insertEmailMessage({ + db: env.APP_DB, + message: { + direction: 'outbound', + userId: stableUserId, + fromAddress: 'user@inbox.kody.example.com', + toAddresses: ['recipient@example.net'], + subject: `Bounce ${attempt}`, + processingStatus: 'sent', + providerMessageId, + sentAt: now.toISOString(), + }, + }) + await recordProviderEvent({ + providerMessageId, + eventId: `event-bounced-${attempt}`, + status: 'bounced', + eventTimestamp: now.toISOString(), + }) + const result = await applyOutboundEmailAbusePause({ + env, + userId: stableUserId, + deliveryStatus: 'bounced', + eventRecorded: true, + now, + }) + if (attempt < outboundEmailBouncePauseThresholdPerDay) { + expect(result.paused).toBe(false) + expect(await readPauseTimestamp(stableUserId)).toBeNull() + } else { + expect(result.paused).toBe(true) + expect(await readPauseTimestamp(stableUserId)).not.toBeNull() + } + } +}) + +test('delivered events never pause outbound email', async () => { + await ensureEmailTestSchema(env.APP_DB) + const accountEmail = `deliverer-${crypto.randomUUID()}@example.com` + const { stableUserId } = await seedVerifiedAccount({ email: accountEmail }) + const result = await applyOutboundEmailAbusePause({ + env, + userId: stableUserId, + deliveryStatus: 'delivered', + eventRecorded: true, + }) + expect(result.paused).toBe(false) + expect(await readPauseTimestamp(stableUserId)).toBeNull() +}) + +test('a complaint signal that was never persisted does not pause on its own', async () => { + await ensureEmailTestSchema(env.APP_DB) + const accountEmail = `phantom-${crypto.randomUUID()}@example.com` + const { stableUserId } = await seedVerifiedAccount({ email: accountEmail }) + // Conflicting-duplicate queue outcome: the complained event's insert + // was deduped by provider_event_id, so no persisted complaint backs it. + const result = await applyOutboundEmailAbusePause({ + env, + userId: stableUserId, + deliveryStatus: 'complained', + eventRecorded: false, + }) + expect(result.paused).toBe(false) + expect(await readPauseTimestamp(stableUserId)).toBeNull() +}) diff --git a/packages/worker/src/email/outbound.ts b/packages/worker/src/email/outbound.ts index 7b7e6d3ef8..03372d4357 100644 --- a/packages/worker/src/email/outbound.ts +++ b/packages/worker/src/email/outbound.ts @@ -1,5 +1,9 @@ import { base64ToBytes } from '@kody-internal/shared/base64.ts' import { getErrorMessage } from '@kody-internal/shared/error-message.ts' +import { + accountSuspendedMessage, + getAccountRestrictionsByStableUserId, +} from '#app/account-suspension.ts' import { sendCloudflareEmail } from '#app/email/cloudflare-email.ts' import { isAccountEmailVerified } from '#app/email-verification.ts' import { normalizeEmail } from '#app/normalize-email.ts' @@ -12,6 +16,7 @@ import { } from '#worker/entitlements/service.ts' import { recordUsage } from '#worker/usage/record-usage.ts' import { normalizeEmailAddress } from './address.ts' +import { emailOutboundPausedMessage } from './outbound-abuse.ts' import { resolveUserPlatformSender } from './platform-address.ts' import { createEmailThread, @@ -433,6 +438,21 @@ export async function sendOutboundEmail( if (!accountEmailVerified) { throw new Error('Account email must be verified before sending email.') } + // Abuse gates: a suspended account cannot send at all, and an + // account paused by the delivery-event abuse monitor (spam + // complaints / repeated bounces) stays blocked until an admin + // clears the pause. Both protect the shared platform sending + // domain's reputation. + const restrictions = await getAccountRestrictionsByStableUserId({ + db: input.env.APP_DB, + stableUserId: input.userId, + }) + if (restrictions?.suspendedAt) { + throw new Error(accountSuspendedMessage) + } + if (restrictions?.emailOutboundPausedAt) { + throw new Error(emailOutboundPausedMessage) + } // Sends reference the platform-provisioned sender identity. It is // normally created alongside the default inbox at signup; ensuring it // here also covers accounts provisioned before the identity existed. diff --git a/packages/worker/src/email/outbound.workers.test.ts b/packages/worker/src/email/outbound.workers.test.ts index ee7001ae40..8ad4c18a7d 100644 --- a/packages/worker/src/email/outbound.workers.test.ts +++ b/packages/worker/src/email/outbound.workers.test.ts @@ -188,6 +188,30 @@ test('sendOutboundEmail sends from the platform-assigned username address to the expect(listed.map((message) => message.id)).toContain(result.message.id) }) +test('sendOutboundEmail rejects suspended accounts', async () => { + silenceIncidentalRuntimeWarnings() + await ensureEmailTestSchema(env.APP_DB) + const accountEmail = `suspended-${crypto.randomUUID()}@example.com` + const userId = await createStableUserIdFromEmail(accountEmail) + await seedVerifiedAccount({ email: accountEmail }) + await env.APP_DB.prepare( + `UPDATE users SET suspended_at = ? WHERE stable_user_id = ?`, + ) + .bind(new Date().toISOString(), userId) + .run() + + await expect( + sendOutboundEmail({ + env: createBindingSendEnv(), + userId, + accountEmail, + recipientPolicy: 'self', + subject: 'Blocked while suspended', + text: 'Body', + }), + ).rejects.toThrow('This account is suspended') +}) + test('sendOutboundEmail rejects non-self recipients under the self policy', async () => { // Usage recording degrades with a warn when the usage_rollups table is // not part of this test's schema; that is incidental to the policy. diff --git a/packages/worker/src/email/service.ts b/packages/worker/src/email/service.ts index 3403603359..9bd36bd25b 100644 --- a/packages/worker/src/email/service.ts +++ b/packages/worker/src/email/service.ts @@ -564,7 +564,12 @@ export async function recordBoundedEmailRejectionEvent(input: { inboxId: string recipient: string reason: string - phase: 'entitlement' | 'size' | 'account-verification' | 'system-limit' + phase: + | 'entitlement' + | 'size' + | 'account-verification' + | 'account-suspension' + | 'system-limit' now?: Date }) { const now = input.now ?? new Date() diff --git a/packages/worker/src/entitlements/plans.ts b/packages/worker/src/entitlements/plans.ts index 0e115e60e6..43789f2c44 100644 --- a/packages/worker/src/entitlements/plans.ts +++ b/packages/worker/src/entitlements/plans.ts @@ -153,6 +153,15 @@ export type PlanLimits = { maxStorageBytes: number /** Maximum concurrently active workflow runs. */ maxConcurrentWorkflows: number + /** Maximum MCP execute-tool runs per UTC day. */ + maxExecuteCallsPerDay: number + /** + * Maximum sandbox outbound fetches (through the fetch gateway) per UTC + * day. Bounds cost abuse and third-party hammering from user code; the + * shared Worker egress identity means one user's fetch flood can burn + * reputation for the whole deployment. + */ + maxOutboundFetchesPerDay: number } export const entitlementResources = [ @@ -168,6 +177,8 @@ export const entitlementResources = [ 'secrets', 'storage_bytes', 'concurrent_workflows', + 'execute_calls_per_day', + 'outbound_fetches_per_day', ] as const export type EntitlementResource = (typeof entitlementResources)[number] @@ -186,6 +197,8 @@ export const entitlementResourceLabels: Record