From 903e0e303d1927efd9bdd5aac5964b06f6bc5389 Mon Sep 17 00:00:00 2001 From: Josh Zhang <39790535+jiashuoz@users.noreply.github.com> Date: Fri, 4 Sep 2026 20:17:59 -0700 Subject: [PATCH 1/4] test(testutil): extract the database helper into a leaf package testutil bundles the contract server, which imports outbound. Once outbound imports sendingpolicy (the provider seam requires the gate's token), any INTERNAL test file in outbound or sendingpolicy that calls testutil.TestDB closes an import cycle. Move db.go into internal/testutil/testdb, a leaf that depends only on identity and migrations, and keep testutil.TestDB / TestDBURL / OpenPreparedTestDB / TruncateAll as forwarding wrappers so every existing caller is unchanged. The contract server and its River test use the two newly exported helpers (Truncate, BaseTestDBURL) instead of package-private ones. The one internal sendingpolicy test imports testdb directly. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01AjfGxvXW6fNKWGFHuo68yX --- .../runtime_attestation_response_loss_test.go | 12 +- internal/testutil/contract_server.go | 3 +- .../testutil/contract_server_river_test.go | 3 +- internal/testutil/db.go | 379 +---------------- internal/testutil/testdb/db.go | 393 ++++++++++++++++++ internal/testutil/{ => testdb}/db_test.go | 16 +- 6 files changed, 426 insertions(+), 380 deletions(-) create mode 100644 internal/testutil/testdb/db.go rename internal/testutil/{ => testdb}/db_test.go (98%) diff --git a/internal/sendingpolicy/runtime_attestation_response_loss_test.go b/internal/sendingpolicy/runtime_attestation_response_loss_test.go index 9759bbbef..56e642d7f 100644 --- a/internal/sendingpolicy/runtime_attestation_response_loss_test.go +++ b/internal/sendingpolicy/runtime_attestation_response_loss_test.go @@ -8,7 +8,7 @@ import ( "time" "github.com/jackc/pgx/v5" - "github.com/tokencanopy/e2a/internal/testutil" + "github.com/tokencanopy/e2a/internal/testutil/testdb" ) var errSyntheticCommitResponseLoss = errors.New("synthetic commit response loss") @@ -19,7 +19,7 @@ func responseLossDigest(hexDigit string) string { func TestRuntimeAttestationCommitResponseLossRereadsExactSuccess(t *testing.T) { ctx := context.Background() - pool := testutil.TestDB(t) + pool := testdb.TestDB(t) m := NewModule(pool, Secrets{}) current, err := m.InspectAttestation(ctx) if err != nil { @@ -58,7 +58,7 @@ func TestRuntimeAttestationCommitResponseLossRereadsExactSuccess(t *testing.T) { func TestRuntimeAttestationCommitResponseLossRereadsAfterCallerCancellation(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - pool := testutil.TestDB(t) + pool := testdb.TestDB(t) m := NewModule(pool, Secrets{}) current, err := m.InspectAttestation(ctx) if err != nil { @@ -97,7 +97,7 @@ func TestRuntimeAttestationCommitResponseLossRereadsAfterCallerCancellation(t *t func TestRuntimeAttestationCommitFailureClassifiesUnchangedState(t *testing.T) { ctx := context.Background() - pool := testutil.TestDB(t) + pool := testdb.TestDB(t) m := NewModule(pool, Secrets{}) current, err := m.InspectAttestation(ctx) if err != nil { @@ -139,7 +139,7 @@ func TestRuntimeAttestationCommitFailureClassifiesUnchangedState(t *testing.T) { func TestRuntimeAttestationCommitResponseLossClassifiesHigherRevisionStale(t *testing.T) { ctx := context.Background() - pool := testutil.TestDB(t) + pool := testdb.TestDB(t) m := NewModule(pool, Secrets{}) other := NewModule(pool, Secrets{}) current, err := m.InspectAttestation(ctx) @@ -199,7 +199,7 @@ func TestRuntimeAttestationConcurrentAbortFenceBothLockOrders(t *testing.T) { runRace := func(t *testing.T, firstIsFence bool) { t.Helper() - pool := testutil.TestDB(t) + pool := testdb.TestDB(t) first := NewModule(pool, Secrets{}) second := NewModule(pool, Secrets{}) prior, err := first.InspectAttestation(ctx) diff --git a/internal/testutil/contract_server.go b/internal/testutil/contract_server.go index eba420de8..86a3e3835 100644 --- a/internal/testutil/contract_server.go +++ b/internal/testutil/contract_server.go @@ -18,6 +18,7 @@ import ( "github.com/tokencanopy/e2a/internal/outbound" "github.com/tokencanopy/e2a/internal/outboundsend" "github.com/tokencanopy/e2a/internal/relay" + "github.com/tokencanopy/e2a/internal/testutil/testdb" "github.com/tokencanopy/e2a/internal/unsubscribe" "github.com/tokencanopy/e2a/internal/usage" "github.com/tokencanopy/e2a/internal/webhook" @@ -281,7 +282,7 @@ func (s *ContractServer) Close(ctx context.Context) error { firstErr = err } s.WSHub.Close() - if err := truncateAll(ctx, s.DBPool); err != nil && firstErr == nil { + if err := testdb.Truncate(ctx, s.DBPool); err != nil && firstErr == nil { firstErr = err } s.DBPool.Close() diff --git a/internal/testutil/contract_server_river_test.go b/internal/testutil/contract_server_river_test.go index cdccb2d7a..dd33982c4 100644 --- a/internal/testutil/contract_server_river_test.go +++ b/internal/testutil/contract_server_river_test.go @@ -6,6 +6,7 @@ import ( "time" "github.com/jackc/pgx/v5/pgxpool" + "github.com/tokencanopy/e2a/internal/testutil/testdb" ) const ( @@ -113,7 +114,7 @@ func requireReachableContractTestDB(t *testing.T) string { contractDBReachabilityTimeout, contractDBPreparationTimeout, func(ctx context.Context) error { - probe, err := pgxpool.New(ctx, baseTestDBURL()) + probe, err := pgxpool.New(ctx, testdb.BaseTestDBURL()) if err != nil { return err } diff --git a/internal/testutil/db.go b/internal/testutil/db.go index cee81312e..c61c49348 100644 --- a/internal/testutil/db.go +++ b/internal/testutil/db.go @@ -2,384 +2,35 @@ package testutil import ( "context" - "crypto/sha256" - "encoding/hex" - "errors" - "fmt" - "net/url" - "os" - "path/filepath" - "strings" "testing" - "github.com/jackc/pgx/v5" - "github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgxpool" - "github.com/tokencanopy/e2a/internal/identity" - "github.com/tokencanopy/e2a/migrations" + "github.com/tokencanopy/e2a/internal/testutil/testdb" ) -const defaultTestDBURL = "postgres://e2a:e2a@localhost:5433/e2a_test?sslmode=disable" +// The database helpers live in the leaf package testdb so that packages this +// package depends on — outbound, and through it sendingpolicy — can use them +// from INTERNAL test files without an import cycle. These wrappers keep every +// existing caller working unchanged; new internal tests in those packages +// should import testdb directly. -type testDBPreparationError struct { - stage string - err error -} - -func (e *testDBPreparationError) Error() string { - return fmt.Sprintf("%s: %v", e.stage, e.err) -} - -func (e *testDBPreparationError) Unwrap() error { - return e.err -} - -// baseTestDBURL is the configured URL before per-package derivation: the -// E2A_TEST_DATABASE_URL override or the local-dev default. Also the admin -// connection target for creating missing package databases. -func baseTestDBURL() string { - if dbURL := os.Getenv("E2A_TEST_DATABASE_URL"); dbURL != "" { - return dbURL - } - return defaultTestDBURL -} - -// TestDBURL returns the database URL tests should use. Inside a `go test` -// binary it derives a PER-PACKAGE database name (_pkg_) so -// packages can run in parallel: the harness truncates tables between tests, -// which made one shared database the documented cross-package flake source -// and forced -p 1 on every DB-backed run. The suffix comes from the test -// binary's name (os.Args[0] = .test — unique per package in this -// repo), so every URL consumer in one test binary — TestDB, hand-built -// pools, the in-process contract server — lands on the same database. -// Non-test binaries (cmd/e2a-contract-server) and E2A_TEST_DB_SHARED=1 get -// the base URL verbatim. Missing databases self-provision on first open -// (see OpenPreparedTestDB). Concurrent sessions, agents, and worktrees are -// isolated by the per-workspace component below, so handing each runner its -// own base URL is no longer required — only useful for pointing a run at an -// entirely separate server. -func TestDBURL() string { - base := baseTestDBURL() - suffix := derivedDBSuffix() - if suffix == "" { - return base - } - u, err := url.Parse(base) - // Only derive on genuine postgres:// URLs. DSN keyword/value form - // ("host=… dbname=…") "parses" into u.Path and would be mangled into - // garbage; pass anything unrecognizable through verbatim. - if err != nil || (u.Scheme != "postgres" && u.Scheme != "postgresql") || - strings.TrimPrefix(u.Path, "/") == "" { - return base - } - // Idempotent: a child .test process handed an already-derived URL (the - // harness's own re-exec tests do this) must not double-suffix it. - if strings.HasSuffix(strings.TrimSuffix(u.Path, "/"), suffix) { - return base - } - u.Path = u.Path + suffix - // Postgres truncates identifiers past maxPostgresIdentifier bytes, and it does - // so SILENTLY. Truncation lands at the END — inside the package component — so - // sibling packages sharing a prefix collapse onto ONE database: internal/identity - // and internal/idempotency both become ..._pkg_ide, then truncate each other's - // tables under -p 4. That is exactly the corruption this derivation exists to - // prevent, so a base too long to derive from has to fail loudly rather than - // quietly reintroduce it. - if name := strings.TrimPrefix(u.Path, "/"); len(name) > maxPostgresIdentifier { - panic(fmt.Sprintf("testutil: derived test database name %q is %d bytes, over Postgres's "+ - "%d-byte identifier limit — Postgres would truncate it silently and collide sibling "+ - "packages onto one database. Shorten the base in E2A_TEST_DATABASE_URL; the derived "+ - "suffix needs %d bytes.", name, len(name), maxPostgresIdentifier, len(suffix))) - } - return u.String() -} - -// maxPostgresIdentifier is Postgres's NAMEDATALEN-1 ceiling for identifiers, -// database names included. Measured in BYTES, which is what len() reports. -const maxPostgresIdentifier = 63 - -// derivedDBSuffix derives the database-name suffix beneath the configured base: -// a per-WORKSPACE component plus a per-PACKAGE component, or "" when the process -// is not a test binary or sharing is forced. -// -// Two dimensions, because per-package alone was not enough. It stops packages in -// ONE run from truncating each other, but every checkout computed the same names, -// so two agents (or two worktrees, or a second terminal) running the same package -// shared a database and corrupted each other. AGENTS.md asked people to hand out -// their own base URL; that is convention, and convention does not scale across -// callers who do not know about each other. Deriving from the module root path -// makes the isolation structural: two checkouts cannot collide even when nobody -// configures anything. -// -// Name length: _ws<8>_pkg_ runs ~40 chars for this repo's longest -// package names, well inside Postgres's 63-byte identifier limit. A much longer -// custom base could push past it, where Postgres truncates silently — keep bases -// short. -func derivedDBSuffix() string { - switch strings.ToLower(os.Getenv("E2A_TEST_DB_SHARED")) { - case "1", "true", "yes": - return "" - } - bin := filepath.Base(os.Args[0]) - if !strings.HasSuffix(bin, ".test") { - return "" - } - name := strings.ToLower(strings.TrimSuffix(bin, ".test")) - sanitized := make([]rune, 0, len(name)) - for _, r := range name { - switch { - case r >= 'a' && r <= 'z', r >= '0' && r <= '9': - sanitized = append(sanitized, r) - default: - sanitized = append(sanitized, '_') - } - } - return workspaceSuffix(moduleRootDir()) + "_pkg_" + string(sanitized) -} - -// workspaceSuffix is the per-checkout component: a short, stable digest of the -// module root's absolute path. Empty when the root cannot be resolved, which -// degrades to the previous per-package-only behavior rather than failing. -// -// Pure and path-taking so it is directly testable: the same path must always -// give the same suffix, and different paths must differ. -func workspaceSuffix(moduleRoot string) string { - if moduleRoot == "" { - return "" - } - sum := sha256.Sum256([]byte(filepath.Clean(moduleRoot))) - return "_ws" + hex.EncodeToString(sum[:])[:8] -} - -// moduleRootDir returns the directory holding go.mod at or above the working -// directory, or "" if there is none. Symlinks are resolved so two paths that -// reach the same checkout derive the same workspace suffix. -func moduleRootDir() string { - dir, err := os.Getwd() - if err != nil { - return "" - } - if resolved, rerr := filepath.EvalSymlinks(dir); rerr == nil { - dir = resolved - } - for { - if _, statErr := os.Stat(filepath.Join(dir, "go.mod")); statErr == nil { - return dir - } - parent := filepath.Dir(dir) - if parent == dir { - return "" - } - dir = parent - } -} +// TestDBURL returns the per-workspace, per-package test database URL. +func TestDBURL() string { return testdb.TestDBURL() } +// OpenPreparedTestDB opens and prepares the database at dbURL. func OpenPreparedTestDB(ctx context.Context, dbURL string) (*pgxpool.Pool, error) { - if dbURL == "" { - dbURL = defaultTestDBURL - } - - pool, err := pgxpool.New(ctx, dbURL) - if err != nil { - return nil, err - } - - if err := pool.Ping(ctx); err != nil { - pool.Close() - // SQLSTATE 3D000 (invalid_catalog_name): the server is up but this - // per-package database doesn't exist yet — self-provision it from - // the base URL's server and retry once. Any other error (server - // down, bad credentials) keeps the caller's skip-vs-fail semantics. - var pgErr *pgconn.PgError - if !errors.As(err, &pgErr) || pgErr.Code != "3D000" { - return nil, err - } - if cerr := createTestDatabase(ctx, dbURL); cerr != nil { - return nil, cerr - } - pool, err = pgxpool.New(ctx, dbURL) - if err != nil { - return nil, err - } - if err := pool.Ping(ctx); err != nil { - pool.Close() - return nil, err - } - } - - if err := runMigrations(ctx, pool); err != nil { - pool.Close() - return nil, &testDBPreparationError{stage: "run migrations", err: err} - } - - if err := truncateAll(ctx, pool); err != nil { - pool.Close() - return nil, &testDBPreparationError{stage: "truncate tables", err: err} - } - - return pool, nil + return testdb.OpenPreparedTestDB(ctx, dbURL) } +// TestDB returns a migrated, truncated pool for this test, skipping when no +// database is reachable. func TestDB(t *testing.T) *pgxpool.Pool { t.Helper() - - ctx := context.Background() - pool, err := OpenPreparedTestDB(ctx, TestDBURL()) - if err != nil { - var preparationErr *testDBPreparationError - if errors.As(err, &preparationErr) { - t.Fatalf("failed to prepare test database: %v", err) - } - t.Skipf("test database not available: %v", err) - } - - t.Cleanup(func() { - TruncateAll(t, pool) - pool.Close() - }) - - return pool + return testdb.TestDB(t) } +// TruncateAll empties every application table. func TruncateAll(t *testing.T, pool *pgxpool.Pool) { t.Helper() - err := truncateAll(context.Background(), pool) - if err != nil { - t.Fatalf("failed to truncate tables: %v", err) - } -} - -// createTestDatabase creates dbURL's database via the base URL's server. -// A concurrent creator racing us is success — the database exists either -// way. Postgres reports that race two ways: 42P04 (duplicate_database, the -// already-committed case) and 23505 (unique violation on -// pg_database_datname_index, the losing side of a true concurrent race — -// empirically what 8 parallel same-name creates produce on PG16). -// -// Error classification is load-bearing for skip-vs-fail: a failure to -// CONNECT to the base URL keeps the caller's "DB unavailable → skip" -// semantics, but a failure to CREATE on a reachable server (e.g. a role -// without CREATEDB) is a preparation error — TestDB must FAIL loudly, not -// silently skip the entire DB tier green. -func createTestDatabase(ctx context.Context, dbURL string) error { - target, err := url.Parse(dbURL) - if err != nil { - return &testDBPreparationError{stage: "parse target db url", err: err} - } - name := strings.TrimPrefix(target.Path, "/") - if name == "" { - return &testDBPreparationError{stage: "derive database name", err: fmt.Errorf("no database name in %s", dbURL)} - } - conn, err := pgx.Connect(ctx, baseTestDBURL()) - if err != nil { - return fmt.Errorf("connect base db to create %s: %w", name, err) - } - defer conn.Close(ctx) - if _, err := conn.Exec(ctx, "CREATE DATABASE "+pgx.Identifier{name}.Sanitize()); err != nil { - var pgErr *pgconn.PgError - if errors.As(err, &pgErr) && (pgErr.Code == "42P04" || pgErr.Code == "23505") { - return nil - } - return &testDBPreparationError{stage: "create database " + name, err: err} - } - return nil -} - -func runMigrations(ctx context.Context, pool *pgxpool.Pool) error { - return identity.RunMigrations(ctx, pool, migrations.FS, identity.ModeAuto) -} - -// truncateAll resets the DB between tests. Most tables are reached implicitly by -// TRUNCATE ... CASCADE via their FK path to users/messages/webhooks, so they need -// no explicit mention. Tables with NO foreign key at all cannot be reached by -// CASCADE, so they need explicit cleanup. Currently that is: -// -// - inbound_intake: written at the SMTP edge BEFORE the agent lookup, so it -// deliberately has no FK. Omitting it left stale dedup rows behind and made -// TestInboundIntake_InsertLoadDedup / _StampProcessAndFail fail on a re-run -// (the "insert must be new" assertions saw the previous run's rows). -// - sender_identity_managed_domains: deliberately survives domain deletion -// until asynchronous provider teardown is confirmed. -// - sending-protection security ledgers: provider operations, budget rows, -// control audit, notice outbox, and feedback provenance deliberately have no -// customer-tree FK so account/message deletion cannot erase them. -// - sending-protection policy state: the event/marker tables have no FK, while -// the runtime-policy and attestation singletons must be restored to their -// migration-owned generation-zero sentinels between tests. -// -// Use DELETE for FK-less tables instead of adding them to TRUNCATE. The test suite -// calls this helper hundreds of times; repeatedly truncating inbound_intake also -// recreates and fsyncs its three indexes and requires an ACCESS EXCLUSIVE lock. -// Any future FK-less table MUST be added to the DELETE section here. -// truncateAllLockTimeout bounds how long cleanup will WAIT ON A LOCK — not how -// long it may take. Cleanup is expected to be lock-free (inbound_intake is -// DELETEd precisely so a concurrent reader's ACCESS SHARE cannot block it), so a -// wait this long means something genuinely holds a conflicting lock. Failing -// fast with SQLSTATE 55P03 (lock_not_available) makes that case -// self-identifying, instead of hanging until the caller's context expires and -// reporting an indistinguishable deadline error. -// -// Deliberately NOT a statement timeout: cleanup is legitimately slow under a -// loaded parallel run (`-p 4` across every package), and slowness must not be -// conflated with a lock conflict — that conflation is what made -// TestTruncateAll_CleansInboundIntakeWithoutExclusiveTableLock flaky in CI. -const truncateAllLockTimeout = "5s" - -func truncateAll(ctx context.Context, pool *pgxpool.Pool) error { - _, err := pool.Exec(ctx, ` - SET LOCAL lock_timeout = '`+truncateAllLockTimeout+`'; - - DELETE FROM inbound_intake; - DELETE FROM sender_identity_managed_domains; - DELETE FROM sending_protection_notice_deliveries; - DELETE FROM sending_protection_notice_events; - DELETE FROM sending_feedback_recipients; - DELETE FROM sending_feedback_events; - DELETE FROM sending_feedback_correlations; - DELETE FROM sending_budget_reservations; - DELETE FROM sending_budget_counters; - DELETE FROM sending_provider_operations; - DELETE FROM account_sending_control_events; - DELETE FROM sending_protection_policy_events; - DELETE FROM sending_protection_runtime_attestation_events; - DELETE FROM sending_ramp_grandfathering; - - -- This registry is append-only in application/migration use; its - -- unconditional trigger intentionally rejects DELETE. The disposable - -- test database bypasses user triggers for this one row-lock-scoped - -- cleanup instead of using TRUNCATE's ACCESS EXCLUSIVE table lock. - SET LOCAL session_replication_role = replica; - DELETE FROM sending_operator_recipient_versions; - SET LOCAL session_replication_role = origin; - - DELETE FROM sending_protection_runtime_policy; - INSERT INTO sending_protection_runtime_policy - (singleton, generation, schema_version, policy, policy_sha256, activated_at, activated_by) - VALUES ( - true, 0, 1, - '{"all_customer_global_daily_recipients":5000,"bounce_min_outcomes":50,"bounce_pause_basis_points":400,"budget_hold_max_days":7,"budget_mode":"disabled","complaint_pause_basis_points":8,"critical_operational_daily_recipients":100,"daily_unlimited_plan_codes":["starter","pro","scale"],"default_account_daily_recipients":100,"detector_interval_seconds":300,"detector_mode":"disabled","detector_window_days":7,"operator_notice_recipient_version":1,"probation_global_daily_recipients":150,"ramp_days":30,"ramp_enabled":false,"ramp_start_daily":150,"ramp_target_daily":2000,"sending_control_audit_retention_days":90,"sending_feedback_post_account_retention_days":30,"shared_domain_account_daily_recipients":50,"shared_reputation_bounce_min_outcomes":1,"tenant_header_canary_account_ids":[],"tenant_header_mode":"disabled","tenant_provisioning_mode":"disabled","tenant_suppression_sync_mode":"disabled","violation_operational_daily_recipients":100}'::jsonb, - '198d8cfb3220b6094a3b8dfe13cb0e2ff97c512ad87ae14609e580ae335c9ce6', - now(), 'migration' - ); - - DELETE FROM sending_protection_runtime_attestation; - INSERT INTO sending_protection_runtime_attestation - (singleton, revision, active_billing_digest, active_billing_contract, - rollback_billing_digest, rollback_billing_contract, updated_by) - VALUES (true, 0, '', 0, '', 0, 'migration'); - - TRUNCATE oauth_pkce_requests, oauth_refresh_tokens, oauth_access_tokens, - oauth_auth_codes, oauth_clients, - usage_summaries, usage_events, webhook_deliveries, - send_attempts, protection_events, messages, - idempotency_keys, api_keys, - agent_identities, domains, - user_sessions, users CASCADE - `) - if err != nil { - return err - } - // Re-seed shared domain (migration seeds it but truncation removes it) - pool.Exec(ctx, `INSERT INTO domains (domain, user_id, verified, verified_at) VALUES ('agents.e2a.dev', NULL, true, now()) ON CONFLICT DO NOTHING`) - return nil + testdb.TruncateAll(t, pool) } diff --git a/internal/testutil/testdb/db.go b/internal/testutil/testdb/db.go new file mode 100644 index 000000000..03440b766 --- /dev/null +++ b/internal/testutil/testdb/db.go @@ -0,0 +1,393 @@ +package testdb + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "net/url" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/tokencanopy/e2a/internal/identity" + "github.com/tokencanopy/e2a/migrations" +) + +const defaultTestDBURL = "postgres://e2a:e2a@localhost:5433/e2a_test?sslmode=disable" + +type testDBPreparationError struct { + stage string + err error +} + +func (e *testDBPreparationError) Error() string { + return fmt.Sprintf("%s: %v", e.stage, e.err) +} + +func (e *testDBPreparationError) Unwrap() error { + return e.err +} + +// baseTestDBURL is the configured URL before per-package derivation: the +// E2A_TEST_DATABASE_URL override or the local-dev default. Also the admin +// connection target for creating missing package databases. +func baseTestDBURL() string { + if dbURL := os.Getenv("E2A_TEST_DATABASE_URL"); dbURL != "" { + return dbURL + } + return defaultTestDBURL +} + +// TestDBURL returns the database URL tests should use. Inside a `go test` +// binary it derives a PER-PACKAGE database name (_pkg_) so +// packages can run in parallel: the harness truncates tables between tests, +// which made one shared database the documented cross-package flake source +// and forced -p 1 on every DB-backed run. The suffix comes from the test +// binary's name (os.Args[0] = .test — unique per package in this +// repo), so every URL consumer in one test binary — TestDB, hand-built +// pools, the in-process contract server — lands on the same database. +// Non-test binaries (cmd/e2a-contract-server) and E2A_TEST_DB_SHARED=1 get +// the base URL verbatim. Missing databases self-provision on first open +// (see OpenPreparedTestDB). Concurrent sessions, agents, and worktrees are +// isolated by the per-workspace component below, so handing each runner its +// own base URL is no longer required — only useful for pointing a run at an +// entirely separate server. +func TestDBURL() string { + base := baseTestDBURL() + suffix := derivedDBSuffix() + if suffix == "" { + return base + } + u, err := url.Parse(base) + // Only derive on genuine postgres:// URLs. DSN keyword/value form + // ("host=… dbname=…") "parses" into u.Path and would be mangled into + // garbage; pass anything unrecognizable through verbatim. + if err != nil || (u.Scheme != "postgres" && u.Scheme != "postgresql") || + strings.TrimPrefix(u.Path, "/") == "" { + return base + } + // Idempotent: a child .test process handed an already-derived URL (the + // harness's own re-exec tests do this) must not double-suffix it. + if strings.HasSuffix(strings.TrimSuffix(u.Path, "/"), suffix) { + return base + } + u.Path = u.Path + suffix + // Postgres truncates identifiers past maxPostgresIdentifier bytes, and it does + // so SILENTLY. Truncation lands at the END — inside the package component — so + // sibling packages sharing a prefix collapse onto ONE database: internal/identity + // and internal/idempotency both become ..._pkg_ide, then truncate each other's + // tables under -p 4. That is exactly the corruption this derivation exists to + // prevent, so a base too long to derive from has to fail loudly rather than + // quietly reintroduce it. + if name := strings.TrimPrefix(u.Path, "/"); len(name) > maxPostgresIdentifier { + panic(fmt.Sprintf("testutil: derived test database name %q is %d bytes, over Postgres's "+ + "%d-byte identifier limit — Postgres would truncate it silently and collide sibling "+ + "packages onto one database. Shorten the base in E2A_TEST_DATABASE_URL; the derived "+ + "suffix needs %d bytes.", name, len(name), maxPostgresIdentifier, len(suffix))) + } + return u.String() +} + +// maxPostgresIdentifier is Postgres's NAMEDATALEN-1 ceiling for identifiers, +// database names included. Measured in BYTES, which is what len() reports. +const maxPostgresIdentifier = 63 + +// derivedDBSuffix derives the database-name suffix beneath the configured base: +// a per-WORKSPACE component plus a per-PACKAGE component, or "" when the process +// is not a test binary or sharing is forced. +// +// Two dimensions, because per-package alone was not enough. It stops packages in +// ONE run from truncating each other, but every checkout computed the same names, +// so two agents (or two worktrees, or a second terminal) running the same package +// shared a database and corrupted each other. AGENTS.md asked people to hand out +// their own base URL; that is convention, and convention does not scale across +// callers who do not know about each other. Deriving from the module root path +// makes the isolation structural: two checkouts cannot collide even when nobody +// configures anything. +// +// Name length: _ws<8>_pkg_ runs ~40 chars for this repo's longest +// package names, well inside Postgres's 63-byte identifier limit. A much longer +// custom base could push past it, where Postgres truncates silently — keep bases +// short. +func derivedDBSuffix() string { + switch strings.ToLower(os.Getenv("E2A_TEST_DB_SHARED")) { + case "1", "true", "yes": + return "" + } + bin := filepath.Base(os.Args[0]) + if !strings.HasSuffix(bin, ".test") { + return "" + } + name := strings.ToLower(strings.TrimSuffix(bin, ".test")) + sanitized := make([]rune, 0, len(name)) + for _, r := range name { + switch { + case r >= 'a' && r <= 'z', r >= '0' && r <= '9': + sanitized = append(sanitized, r) + default: + sanitized = append(sanitized, '_') + } + } + return workspaceSuffix(moduleRootDir()) + "_pkg_" + string(sanitized) +} + +// workspaceSuffix is the per-checkout component: a short, stable digest of the +// module root's absolute path. Empty when the root cannot be resolved, which +// degrades to the previous per-package-only behavior rather than failing. +// +// Pure and path-taking so it is directly testable: the same path must always +// give the same suffix, and different paths must differ. +func workspaceSuffix(moduleRoot string) string { + if moduleRoot == "" { + return "" + } + sum := sha256.Sum256([]byte(filepath.Clean(moduleRoot))) + return "_ws" + hex.EncodeToString(sum[:])[:8] +} + +// moduleRootDir returns the directory holding go.mod at or above the working +// directory, or "" if there is none. Symlinks are resolved so two paths that +// reach the same checkout derive the same workspace suffix. +func moduleRootDir() string { + dir, err := os.Getwd() + if err != nil { + return "" + } + if resolved, rerr := filepath.EvalSymlinks(dir); rerr == nil { + dir = resolved + } + for { + if _, statErr := os.Stat(filepath.Join(dir, "go.mod")); statErr == nil { + return dir + } + parent := filepath.Dir(dir) + if parent == dir { + return "" + } + dir = parent + } +} + +func OpenPreparedTestDB(ctx context.Context, dbURL string) (*pgxpool.Pool, error) { + if dbURL == "" { + dbURL = defaultTestDBURL + } + + pool, err := pgxpool.New(ctx, dbURL) + if err != nil { + return nil, err + } + + if err := pool.Ping(ctx); err != nil { + pool.Close() + // SQLSTATE 3D000 (invalid_catalog_name): the server is up but this + // per-package database doesn't exist yet — self-provision it from + // the base URL's server and retry once. Any other error (server + // down, bad credentials) keeps the caller's skip-vs-fail semantics. + var pgErr *pgconn.PgError + if !errors.As(err, &pgErr) || pgErr.Code != "3D000" { + return nil, err + } + if cerr := createTestDatabase(ctx, dbURL); cerr != nil { + return nil, cerr + } + pool, err = pgxpool.New(ctx, dbURL) + if err != nil { + return nil, err + } + if err := pool.Ping(ctx); err != nil { + pool.Close() + return nil, err + } + } + + if err := runMigrations(ctx, pool); err != nil { + pool.Close() + return nil, &testDBPreparationError{stage: "run migrations", err: err} + } + + if err := truncateAll(ctx, pool); err != nil { + pool.Close() + return nil, &testDBPreparationError{stage: "truncate tables", err: err} + } + + return pool, nil +} + +func TestDB(t *testing.T) *pgxpool.Pool { + t.Helper() + + ctx := context.Background() + pool, err := OpenPreparedTestDB(ctx, TestDBURL()) + if err != nil { + var preparationErr *testDBPreparationError + if errors.As(err, &preparationErr) { + t.Fatalf("failed to prepare test database: %v", err) + } + t.Skipf("test database not available: %v", err) + } + + t.Cleanup(func() { + TruncateAll(t, pool) + pool.Close() + }) + + return pool +} + +// Truncate empties every application table, for callers that own their own +// lifecycle (the contract server's Close) rather than a *testing.T. +func Truncate(ctx context.Context, pool *pgxpool.Pool) error { return truncateAll(ctx, pool) } + +// BaseTestDBURL returns the configured base URL without the per-workspace, +// per-package suffix — the database an administrative probe connects to. +func BaseTestDBURL() string { return baseTestDBURL() } + +func TruncateAll(t *testing.T, pool *pgxpool.Pool) { + t.Helper() + err := truncateAll(context.Background(), pool) + if err != nil { + t.Fatalf("failed to truncate tables: %v", err) + } +} + +// createTestDatabase creates dbURL's database via the base URL's server. +// A concurrent creator racing us is success — the database exists either +// way. Postgres reports that race two ways: 42P04 (duplicate_database, the +// already-committed case) and 23505 (unique violation on +// pg_database_datname_index, the losing side of a true concurrent race — +// empirically what 8 parallel same-name creates produce on PG16). +// +// Error classification is load-bearing for skip-vs-fail: a failure to +// CONNECT to the base URL keeps the caller's "DB unavailable → skip" +// semantics, but a failure to CREATE on a reachable server (e.g. a role +// without CREATEDB) is a preparation error — TestDB must FAIL loudly, not +// silently skip the entire DB tier green. +func createTestDatabase(ctx context.Context, dbURL string) error { + target, err := url.Parse(dbURL) + if err != nil { + return &testDBPreparationError{stage: "parse target db url", err: err} + } + name := strings.TrimPrefix(target.Path, "/") + if name == "" { + return &testDBPreparationError{stage: "derive database name", err: fmt.Errorf("no database name in %s", dbURL)} + } + conn, err := pgx.Connect(ctx, baseTestDBURL()) + if err != nil { + return fmt.Errorf("connect base db to create %s: %w", name, err) + } + defer conn.Close(ctx) + if _, err := conn.Exec(ctx, "CREATE DATABASE "+pgx.Identifier{name}.Sanitize()); err != nil { + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) && (pgErr.Code == "42P04" || pgErr.Code == "23505") { + return nil + } + return &testDBPreparationError{stage: "create database " + name, err: err} + } + return nil +} + +func runMigrations(ctx context.Context, pool *pgxpool.Pool) error { + return identity.RunMigrations(ctx, pool, migrations.FS, identity.ModeAuto) +} + +// truncateAll resets the DB between tests. Most tables are reached implicitly by +// TRUNCATE ... CASCADE via their FK path to users/messages/webhooks, so they need +// no explicit mention. Tables with NO foreign key at all cannot be reached by +// CASCADE, so they need explicit cleanup. Currently that is: +// +// - inbound_intake: written at the SMTP edge BEFORE the agent lookup, so it +// deliberately has no FK. Omitting it left stale dedup rows behind and made +// TestInboundIntake_InsertLoadDedup / _StampProcessAndFail fail on a re-run +// (the "insert must be new" assertions saw the previous run's rows). +// - sender_identity_managed_domains: deliberately survives domain deletion +// until asynchronous provider teardown is confirmed. +// - sending-protection security ledgers: provider operations, budget rows, +// control audit, notice outbox, and feedback provenance deliberately have no +// customer-tree FK so account/message deletion cannot erase them. +// - sending-protection policy state: the event/marker tables have no FK, while +// the runtime-policy and attestation singletons must be restored to their +// migration-owned generation-zero sentinels between tests. +// +// Use DELETE for FK-less tables instead of adding them to TRUNCATE. The test suite +// calls this helper hundreds of times; repeatedly truncating inbound_intake also +// recreates and fsyncs its three indexes and requires an ACCESS EXCLUSIVE lock. +// Any future FK-less table MUST be added to the DELETE section here. +// truncateAllLockTimeout bounds how long cleanup will WAIT ON A LOCK — not how +// long it may take. Cleanup is expected to be lock-free (inbound_intake is +// DELETEd precisely so a concurrent reader's ACCESS SHARE cannot block it), so a +// wait this long means something genuinely holds a conflicting lock. Failing +// fast with SQLSTATE 55P03 (lock_not_available) makes that case +// self-identifying, instead of hanging until the caller's context expires and +// reporting an indistinguishable deadline error. +// +// Deliberately NOT a statement timeout: cleanup is legitimately slow under a +// loaded parallel run (`-p 4` across every package), and slowness must not be +// conflated with a lock conflict — that conflation is what made +// TestTruncateAll_CleansInboundIntakeWithoutExclusiveTableLock flaky in CI. +const truncateAllLockTimeout = "5s" + +func truncateAll(ctx context.Context, pool *pgxpool.Pool) error { + _, err := pool.Exec(ctx, ` + SET LOCAL lock_timeout = '`+truncateAllLockTimeout+`'; + + DELETE FROM inbound_intake; + DELETE FROM sender_identity_managed_domains; + DELETE FROM sending_protection_notice_deliveries; + DELETE FROM sending_protection_notice_events; + DELETE FROM sending_feedback_recipients; + DELETE FROM sending_feedback_events; + DELETE FROM sending_feedback_correlations; + DELETE FROM sending_budget_reservations; + DELETE FROM sending_budget_counters; + DELETE FROM sending_provider_operations; + DELETE FROM account_sending_control_events; + DELETE FROM sending_protection_policy_events; + DELETE FROM sending_protection_runtime_attestation_events; + DELETE FROM sending_ramp_grandfathering; + + -- This registry is append-only in application/migration use; its + -- unconditional trigger intentionally rejects DELETE. The disposable + -- test database bypasses user triggers for this one row-lock-scoped + -- cleanup instead of using TRUNCATE's ACCESS EXCLUSIVE table lock. + SET LOCAL session_replication_role = replica; + DELETE FROM sending_operator_recipient_versions; + SET LOCAL session_replication_role = origin; + + DELETE FROM sending_protection_runtime_policy; + INSERT INTO sending_protection_runtime_policy + (singleton, generation, schema_version, policy, policy_sha256, activated_at, activated_by) + VALUES ( + true, 0, 1, + '{"all_customer_global_daily_recipients":5000,"bounce_min_outcomes":50,"bounce_pause_basis_points":400,"budget_hold_max_days":7,"budget_mode":"disabled","complaint_pause_basis_points":8,"critical_operational_daily_recipients":100,"daily_unlimited_plan_codes":["starter","pro","scale"],"default_account_daily_recipients":100,"detector_interval_seconds":300,"detector_mode":"disabled","detector_window_days":7,"operator_notice_recipient_version":1,"probation_global_daily_recipients":150,"ramp_days":30,"ramp_enabled":false,"ramp_start_daily":150,"ramp_target_daily":2000,"sending_control_audit_retention_days":90,"sending_feedback_post_account_retention_days":30,"shared_domain_account_daily_recipients":50,"shared_reputation_bounce_min_outcomes":1,"tenant_header_canary_account_ids":[],"tenant_header_mode":"disabled","tenant_provisioning_mode":"disabled","tenant_suppression_sync_mode":"disabled","violation_operational_daily_recipients":100}'::jsonb, + '198d8cfb3220b6094a3b8dfe13cb0e2ff97c512ad87ae14609e580ae335c9ce6', + now(), 'migration' + ); + + DELETE FROM sending_protection_runtime_attestation; + INSERT INTO sending_protection_runtime_attestation + (singleton, revision, active_billing_digest, active_billing_contract, + rollback_billing_digest, rollback_billing_contract, updated_by) + VALUES (true, 0, '', 0, '', 0, 'migration'); + + TRUNCATE oauth_pkce_requests, oauth_refresh_tokens, oauth_access_tokens, + oauth_auth_codes, oauth_clients, + usage_summaries, usage_events, webhook_deliveries, + send_attempts, protection_events, messages, + idempotency_keys, api_keys, + agent_identities, domains, + user_sessions, users CASCADE + `) + if err != nil { + return err + } + // Re-seed shared domain (migration seeds it but truncation removes it) + pool.Exec(ctx, `INSERT INTO domains (domain, user_id, verified, verified_at) VALUES ('agents.e2a.dev', NULL, true, now()) ON CONFLICT DO NOTHING`) + return nil +} diff --git a/internal/testutil/db_test.go b/internal/testutil/testdb/db_test.go similarity index 98% rename from internal/testutil/db_test.go rename to internal/testutil/testdb/db_test.go index 8bb29ac93..4e46dafd4 100644 --- a/internal/testutil/db_test.go +++ b/internal/testutil/testdb/db_test.go @@ -1,4 +1,4 @@ -package testutil +package testdb import ( "context" @@ -352,8 +352,8 @@ func TestTestDBURLIsUniquePerWorkspaceAndPackage(t *testing.T) { if !strings.Contains(name, "_ws") { t.Errorf("dbname = %q, want a _ws workspace component", name) } - if !strings.HasSuffix(name, "_pkg_testutil") { - t.Errorf("dbname = %q, want the _pkg_testutil suffix retained", name) + if !strings.HasSuffix(name, "_pkg_testdb") { + t.Errorf("dbname = %q, want the _pkg_testdb suffix retained", name) } if ws := workspaceSuffix(moduleRootDir()); ws == "" || !strings.Contains(name, ws) { t.Errorf("dbname = %q, want it to contain this checkout's suffix %q", name, ws) @@ -396,14 +396,14 @@ func TestTestDBURLDerivesPerPackageDatabase(t *testing.T) { // appends a per-package suffix to the base database name so packages // running in parallel (-p N) cannot truncate each other's rows — the // harness truncates between tests, which made a shared DB the - // documented cross-package flake source. This binary is testutil.test, - // so the derived name is _pkg_testutil. + // documented cross-package flake source. This binary is testdb.test, + // so the derived name is _pkg_testdb. u, err := url.Parse(TestDBURL()) if err != nil { t.Fatalf("parse TestDBURL: %v", err) } - if got := strings.TrimPrefix(u.Path, "/"); !strings.HasSuffix(got, "_pkg_testutil") { - t.Errorf("TestDBURL dbname = %q, want *_pkg_testutil suffix", got) + if got := strings.TrimPrefix(u.Path, "/"); !strings.HasSuffix(got, "_pkg_testdb") { + t.Errorf("TestDBURL dbname = %q, want *_pkg_testdb suffix", got) } // E2A_TEST_DB_SHARED=1 restores the verbatim single-DB behavior (escape @@ -469,7 +469,7 @@ func TestTestDBURLDerivationIsIdempotentAndURLOnly(t *testing.T) { if err != nil { t.Fatalf("parse: %v", err) } - if got := strings.TrimPrefix(u.Path, "/"); strings.Contains(got, "_pkg_testutil_pkg_") { + if got := strings.TrimPrefix(u.Path, "/"); strings.Contains(got, "_pkg_testdb_pkg_") { t.Errorf("double-derived dbname %q", got) } From ec366b35cf64f34f237b724a681db2ad602d2bc1 Mon Sep 17 00:00:00 2001 From: Josh Zhang <39790535+jiashuoz@users.noreply.github.com> Date: Fri, 4 Sep 2026 20:17:59 -0700 Subject: [PATCH 2/4] feat(outbound): require single-use provider authorization MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add ProviderSubmitter, the one path from a composed message to an SMTP transaction with SES. It requires a sendingpolicy.ProviderAuthorization for every call, proves the envelope is the authorized one, derives X-E2A-Provider-Attempt and X-SES-TENANT only from the token (and X-SES-CONFIGURATION-SET from configuration), strips every occurrence of those headers — mixed case and folded forms — from the composed MIME, redeems the token, and only then dials. Every early exit is I/O-free and leaves the token intact; a misconfigured relay is refused before redemption so a retry does not burn an ordinal. A definite permanent rejection is settled as such; an acceptance is settled with the provider's message id, which SettleProvider now binds to the attempt's feedback correlation exactly once (same id replays, a different id is refused). A settlement that fails after acceptance is reported on the result, never as a send error, so the caller retries settlement rather than resubmitting. The legacy tokenless Sender.SubmitOnce path stays for now; Task 7 migrates its callers and makes the relay's socket-opening methods package-private. Tests: zero-network on missing, mismatched, stale, and reused tokens (asserted against a socket counter); exact and single attempt/tenant headers with smuggled spellings removed and the body untouched; a physical retry redeems a distinct ordinal; provider id binding and its conflict rule. Removing the header strip, reordering redemption before envelope validation, or dropping the id binding each fail a test. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01AjfGxvXW6fNKWGFHuo68yX --- internal/outbound/provider_submit.go | 265 +++++++++ .../outbound/provider_submit_internal_test.go | 111 ++++ internal/outbound/provider_submit_test.go | 559 ++++++++++++++++++ internal/sendingpolicy/gate.go | 53 ++ internal/sendingpolicy/provider_token_test.go | 127 ++++ internal/sendingpolicy/types.go | 15 +- 6 files changed, 1129 insertions(+), 1 deletion(-) create mode 100644 internal/outbound/provider_submit.go create mode 100644 internal/outbound/provider_submit_internal_test.go create mode 100644 internal/outbound/provider_submit_test.go create mode 100644 internal/sendingpolicy/provider_token_test.go diff --git a/internal/outbound/provider_submit.go b/internal/outbound/provider_submit.go new file mode 100644 index 000000000..bfd5112e1 --- /dev/null +++ b/internal/outbound/provider_submit.go @@ -0,0 +1,265 @@ +package outbound + +import ( + "bytes" + "context" + "errors" + "fmt" + "strings" + + "github.com/tokencanopy/e2a/internal/delivery" + "github.com/tokencanopy/e2a/internal/sendingpolicy" +) + +// This file is the provider seam: the one place a customer-bound message +// becomes an SMTP transaction with SES. +// +// Everything above it — composition, DKIM, footers, recipient normalization — +// is deliberately token-free, because none of it exposes the shared SES +// reputation. Opening the socket does. So the socket is the thing that +// requires a sendingpolicy.ProviderAuthorization, and the adapter redeems that +// single-use token immediately before dialing, not when the job was picked up +// and not when the message was composed. A decision that went stale in between +// — a pause, a plan change, a policy rotation, a duplicate worker — invalidates +// the token instead of being raced. +// +// The adapter also owns three headers SES reads for provider-side isolation and +// attribution: X-SES-TENANT, X-E2A-Provider-Attempt, and +// X-SES-CONFIGURATION-SET (plus the stable X-E2A-Message-ID correlation marker +// beside them). Their values come only from the token and this deployment's +// configuration. Whatever the composed MIME already carried under those names +// is removed first, every occurrence, so neither a customer nor an upstream +// compose bug can smuggle or duplicate a tenant, attempt, or configuration-set +// selector. + +// ProviderAttemptHeader carries the random attempt correlation id SES echoes +// back in delivery feedback. It is the fallback lookup when the worker died +// between SES accepting the message and the provider id being stored. +const ProviderAttemptHeader = "X-E2A-Provider-Attempt" + +// SESTenantHeader names the SES tenant a submission is attributed to. +const SESTenantHeader = "X-SES-TENANT" + +// SESConfigurationSetHeader selects the SES configuration set (delivery +// feedback destination) for a submission. +const SESConfigurationSetHeader = "X-SES-CONFIGURATION-SET" + +// Sentinel errors for the provider seam. Every one of them is returned before +// any network I/O and before the token is redeemed. +var ( + // ErrAuthorizationRequired means SubmitOnce was called without a token. + // There is no tokenless path to the provider by construction; a caller + // hitting this has bypassed the gate. + ErrAuthorizationRequired = errors.New("outbound: provider submission requires an authorization") + // ErrTenantNameMissing means the token demands a tenant header but carries + // no tenant name. The gate refuses to mint such a token; the adapter checks + // again because the header is the provider-side isolation boundary and + // must never be emitted empty. + ErrTenantNameMissing = errors.New("outbound: authorization requires a tenant header but names no tenant") + // ErrProviderHeaderValue means a provider-owned header value carries a + // line break. The values come from the gate, so this is a defect, not + // input — and a defect here is a header injection, so it fails closed + // rather than being sanitized silently. + ErrProviderHeaderValue = errors.New("outbound: provider header value contains a line break") +) + +// providerOwnedHeaders are removed from the composed MIME before submission, +// matched case-insensitively, folded continuations included. +var providerOwnedHeaders = map[string]struct{}{ + strings.ToLower(SESTenantHeader): {}, + strings.ToLower(ProviderAttemptHeader): {}, + strings.ToLower(SESConfigurationSetHeader): {}, + strings.ToLower(delivery.MessageIDHeader): {}, +} + +// Envelope is what a caller hands the provider seam: the SMTP envelope and the +// composed wire bytes. +type Envelope struct { + // MessageID is the e2a message id stamped as X-E2A-Message-ID, the stable + // correlation marker delivery feedback keys on. Empty for mail that has no + // message row. + MessageID string + // From is the SMTP MAIL FROM address. + From string + // Recipients is the exact final envelope: one entry per distinct mailbox, + // and exactly the set the authorization was minted for. Order is free. + Recipients []string + // Message is the composed MIME. Provider-owned headers in it are stripped. + Message []byte +} + +// ProviderResult reports one accepted provider submission. +type ProviderResult struct { + // ProviderMessageID is the id SES assigned on acceptance. + ProviderMessageID string + // Attempt is the durable attempt that was redeemed for this call. + Attempt sendingpolicy.AttemptRef + // SettlementErr is set when SES accepted the message but the local + // settlement did not commit. The send HAPPENED; the caller must retry + // SettleProvider with the same attempt and provider id, and must never + // resubmit. Delivery feedback carrying the attempt header is the fallback + // if it never does. + SettlementErr error +} + +// ProviderSubmitter is the token-requiring adapter over the SMTP relay. +type ProviderSubmitter struct { + relay *SMTPRelay + gate sendingpolicy.Gate + sesConfigSet string +} + +// NewProviderSubmitter binds the relay to the gate whose tokens it honors. +func NewProviderSubmitter(relay *SMTPRelay, gate sendingpolicy.Gate) *ProviderSubmitter { + return &ProviderSubmitter{relay: relay, gate: gate} +} + +// SetSESConfigurationSet names the configuration set every submission is +// tagged with. Empty means no header (dev/self-host without SES). +func (s *ProviderSubmitter) SetSESConfigurationSet(name string) { s.sesConfigSet = name } + +// SubmitOnce makes exactly one provider call for one authorized attempt. +// +// The sequence is fixed and every early exit is I/O-free: prove the envelope is +// the authorized one, derive the provider headers from the token, rewrite the +// wire bytes, redeem the token, THEN dial. A relay-level retry is not offered +// here on purpose — each physical submission exposes SES once and must be +// charged once, so a retry is a new attempt with a new token, obtained by the +// caller through the gate. +// +// Outcomes: a definite permanent rejection is settled as such and returned; an +// acceptance is settled with the provider id and returned as a result. Anything +// ambiguous (4xx, connection loss, cancellation) is returned unsettled, because +// a message that might have been delivered must not release anything. +func (s *ProviderSubmitter) SubmitOnce(ctx context.Context, auth sendingpolicy.ProviderAuthorization, env Envelope) (ProviderResult, error) { + if s == nil || s.relay == nil || s.gate == nil { + return ProviderResult{}, errors.New("outbound: provider submitter is not wired") + } + if auth.IsZero() { + return ProviderResult{}, ErrAuthorizationRequired + } + if strings.TrimSpace(env.From) == "" { + return ProviderResult{}, errors.New("outbound: envelope sender is empty") + } + headers, err := auth.ValidateEnvelope(env.Recipients) + if err != nil { + return ProviderResult{}, err + } + provider, err := providerHeaderLines(headers, s.sesConfigSet, env.MessageID) + if err != nil { + return ProviderResult{}, err + } + // Refuse a misconfigured relay before the token is spent. Redeeming first + // would invalidate the attempt for a failure that had nothing to do with + // the message, and the caller would burn a fresh ordinal per retry. + if !s.relay.Configured() { + return ProviderResult{}, fmt.Errorf("outbound SMTP relay not configured") + } + wire := append(provider, stripProviderHeaders(env.Message)...) + + if err := s.gate.RedeemProviderCall(ctx, auth); err != nil { + return ProviderResult{}, fmt.Errorf("provider authorization: %w", err) + } + + providerID, sendErr := s.relay.SendOnceContext(ctx, env.From, env.Recipients, wire) + if sendErr != nil { + if IsPermanentSMTPError(sendErr) { + if err := s.gate.SettleProvider(ctx, sendingpolicy.ProviderSettlement{ + Attempt: auth.Attempt(), + Outcome: sendingpolicy.SettlementProviderPermanentlyRejected, + }); err != nil { + return ProviderResult{}, errors.Join(sendErr, fmt.Errorf("settle rejection: %w", err)) + } + } + return ProviderResult{}, sendErr + } + + result := ProviderResult{ProviderMessageID: providerID, Attempt: auth.Attempt()} + if err := s.gate.SettleProvider(ctx, sendingpolicy.ProviderSettlement{ + Attempt: auth.Attempt(), + Outcome: sendingpolicy.SettlementProviderAccepted, + ProviderMessageID: providerID, + }); err != nil { + result.SettlementErr = fmt.Errorf("settle acceptance: %w", err) + } + return result, nil +} + +// providerHeaderLines renders the provider-owned header block from the token's +// view and the deployment configuration — and from nothing else. +// +// Callers cannot pass a tenant name or correlation id; the only way to change +// what SES receives is to change what the gate authorized. +func providerHeaderLines(h sendingpolicy.ProviderHeaders, sesConfigSet, messageID string) ([]byte, error) { + if h.TenantRequired && strings.TrimSpace(h.TenantName) == "" { + return nil, ErrTenantNameMissing + } + for _, v := range []string{h.AttemptCorrelationID, h.TenantName, sesConfigSet, messageID} { + if strings.ContainsAny(v, "\r\n") { + return nil, ErrProviderHeaderValue + } + } + var b bytes.Buffer + if messageID != "" { + b.WriteString(delivery.MessageIDHeader + ": " + messageID + "\r\n") + } + if h.AttemptCorrelationID != "" { + b.WriteString(ProviderAttemptHeader + ": " + h.AttemptCorrelationID + "\r\n") + } + if h.TenantRequired { + b.WriteString(SESTenantHeader + ": " + h.TenantName + "\r\n") + } + if sesConfigSet != "" { + b.WriteString(SESConfigurationSetHeader + ": " + sesConfigSet + "\r\n") + } + return b.Bytes(), nil +} + +// stripProviderHeaders removes every provider-owned header field from the +// header section of a message, leaving every other byte — including the body +// and the original line endings — untouched. +// +// It walks the header section line by line rather than parsing it as a +// message: the bytes were composed by this package or arrived from a customer, +// and a parser that normalized them would change what DKIM signed. A field is +// its name line plus every folded continuation (a line starting with space or +// tab); dropping a field drops its continuations with it. Matching is on the +// lowercased name, so mixed-case spellings are not an evasion. +func stripProviderHeaders(msg []byte) []byte { + out := make([]byte, 0, len(msg)) + rest := msg + dropping := false + for len(rest) > 0 { + nl := bytes.IndexByte(rest, '\n') + var line []byte + if nl < 0 { + line, rest = rest, nil + } else { + line, rest = rest[:nl+1], rest[nl+1:] + } + trimmed := bytes.TrimRight(line, "\r\n") + if len(trimmed) == 0 { + // End of the header section: emit the separator and the body + // verbatim. + out = append(out, line...) + out = append(out, rest...) + return out + } + if trimmed[0] == ' ' || trimmed[0] == '\t' { + if !dropping { + out = append(out, line...) + } + continue + } + dropping = false + if colon := bytes.IndexByte(trimmed, ':'); colon > 0 { + name := strings.ToLower(strings.TrimSpace(string(trimmed[:colon]))) + if _, owned := providerOwnedHeaders[name]; owned { + dropping = true + continue + } + } + out = append(out, line...) + } + return out +} diff --git a/internal/outbound/provider_submit_internal_test.go b/internal/outbound/provider_submit_internal_test.go new file mode 100644 index 000000000..ec506a808 --- /dev/null +++ b/internal/outbound/provider_submit_internal_test.go @@ -0,0 +1,111 @@ +package outbound + +import ( + "errors" + "testing" + + "github.com/tokencanopy/e2a/internal/sendingpolicy" +) + +// Pure unit tests over the adapter's two rewriting steps. The DB-backed +// tests live in the external test package, because the shared test database +// helper transitively imports this package. + +// SmuggledMIME is exported for the external test package. +var SmuggledMIME = smuggledMIME + +// smuggledMIME is customer-shaped wire bytes that try every spelling of the +// provider-owned headers, including a folded one and a body decoy. +func smuggledMIME() []byte { + return []byte("From: agent@agents.e2a.dev\r\n" + + "x-ses-tenant: smuggled-lower\r\n" + + "X-SES-TENANT: smuggled-upper\r\n" + + "X-Ses-Tenant: smuggled-\r\n folded\r\n" + + "X-E2A-Provider-Attempt: cor_forged\r\n" + + "x-e2a-provider-attempt: cor_forged_two\r\n" + + "X-SES-CONFIGURATION-SET: attacker-set\r\n" + + "x-ses-configuration-set:\r\n\tattacker-folded\r\n" + + "X-E2A-Message-ID: msg_forged\r\n" + + "Subject: hello\r\n" + + "\r\n" + + "X-SES-TENANT: body-decoy\r\n" + + "body line\r\n") +} + +func TestProviderHeaderLinesFailClosed(t *testing.T) { + for name, tc := range map[string]struct { + h sendingpolicy.ProviderHeaders + set, id string + wantErr error + want string + }{ + "tenant required but empty": { + h: sendingpolicy.ProviderHeaders{AttemptCorrelationID: "cor_1", TenantRequired: true}, wantErr: ErrTenantNameMissing, + }, + "tenant required but blank": { + h: sendingpolicy.ProviderHeaders{AttemptCorrelationID: "cor_1", TenantRequired: true, TenantName: " \t"}, wantErr: ErrTenantNameMissing, + }, + "line break in tenant": { + h: sendingpolicy.ProviderHeaders{AttemptCorrelationID: "cor_1", TenantRequired: true, TenantName: "t\r\nBcc: x"}, wantErr: ErrProviderHeaderValue, + }, + "line break in config set": { + h: sendingpolicy.ProviderHeaders{AttemptCorrelationID: "cor_1"}, set: "a\nb", wantErr: ErrProviderHeaderValue, + }, + "line break in message id": { + h: sendingpolicy.ProviderHeaders{AttemptCorrelationID: "cor_1"}, id: "m\r", wantErr: ErrProviderHeaderValue, + }, + "no tenant, no set, no id": { + h: sendingpolicy.ProviderHeaders{AttemptCorrelationID: "cor_1"}, want: "X-E2A-Provider-Attempt: cor_1\r\n", + }, + "everything": { + h: sendingpolicy.ProviderHeaders{AttemptCorrelationID: "cor_1", TenantRequired: true, TenantName: "tenant_a"}, set: "cs", id: "msg_1", + want: "X-E2A-Message-ID: msg_1\r\nX-E2A-Provider-Attempt: cor_1\r\nX-SES-TENANT: tenant_a\r\nX-SES-CONFIGURATION-SET: cs\r\n", + }, + } { + got, err := providerHeaderLines(tc.h, tc.set, tc.id) + if !errors.Is(err, tc.wantErr) { + t.Errorf("%s: err = %v, want %v", name, err, tc.wantErr) + } + if string(got) != tc.want { + t.Errorf("%s: headers = %q, want %q", name, got, tc.want) + } + } +} + +func TestStripProviderHeaders(t *testing.T) { + for name, tc := range map[string]struct{ in, want string }{ + "mixed case, duplicates, folded": { + in: string(smuggledMIME()), + want: "From: agent@agents.e2a.dev\r\nSubject: hello\r\n\r\nX-SES-TENANT: body-decoy\r\nbody line\r\n", + }, + "lf-only line endings": { + in: "X-SES-TENANT: a\nSubject: s\nx-ses-configuration-set: b\n c\n\nbody\n", + want: "Subject: s\n\nbody\n", + }, + "headers only, no body separator": { + in: "Subject: s\r\nX-E2A-Provider-Attempt: cor\r\n\tfolded\r\n", + want: "Subject: s\r\n", + }, + "nothing to strip": { + in: "Subject: s\r\nTo: a@example.test\r\n\r\nbody\r\n", + want: "Subject: s\r\nTo: a@example.test\r\n\r\nbody\r\n", + }, + "name prefix is not a match": { + in: "X-SES-TENANT-EXTRA: keep\r\nX-SES-TENANTS: keep\r\n\r\n", + want: "X-SES-TENANT-EXTRA: keep\r\nX-SES-TENANTS: keep\r\n\r\n", + }, + "whitespace before colon still matches": { + in: "X-SES-TENANT : a\r\nSubject: s\r\n\r\n", + want: "Subject: s\r\n\r\n", + }, + "continuation after a kept header is kept": { + in: "Subject: long\r\n subject\r\nX-SES-TENANT: a\r\n b\r\nTo: x@example.test\r\n\r\n", + want: "Subject: long\r\n subject\r\nTo: x@example.test\r\n\r\n", + }, + "empty": {in: "", want: ""}, + } { + if got := string(stripProviderHeaders([]byte(tc.in))); got != tc.want { + t.Errorf("%s:\n got %q\nwant %q", name, got, tc.want) + } + } +} diff --git a/internal/outbound/provider_submit_test.go b/internal/outbound/provider_submit_test.go new file mode 100644 index 000000000..6d6917084 --- /dev/null +++ b/internal/outbound/provider_submit_test.go @@ -0,0 +1,559 @@ +package outbound_test + +import ( + "bufio" + "context" + "errors" + "fmt" + "math/rand" + "net" + "strings" + "sync" + "testing" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/tokencanopy/e2a/internal/config" + "github.com/tokencanopy/e2a/internal/delivery" + "github.com/tokencanopy/e2a/internal/outbound" + "github.com/tokencanopy/e2a/internal/sendingpolicy" + "github.com/tokencanopy/e2a/internal/testutil" +) + +// These tests drive the real gate against real Postgres and a real (fake) +// SMTP listener. Every "no network call" claim is asserted against a socket +// counter, never against the absence of an error, because the failure that +// matters is a connection that happened anyway. Every address is synthetic. + +const ( + psHMAC = `{"active":1,"keys":{"1":"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"}}` + psOperator = `{"commitment_key":"AgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgI","recipients":{"1":"submit-operator@example.test"}}` +) + +type gateFixture struct { + t *testing.T + ctx context.Context + pool *pgxpool.Pool + gate sendingpolicy.Gate + userID string + agent string + tenant string +} + +var psSeq int + +func psID(prefix string) string { + psSeq++ + return fmt.Sprintf("%s_ps_%d_%x", prefix, psSeq, rand.Uint32()) +} + +// newGateFixture builds an enforcing config-source gate with one standard +// account owning one shared-domain agent. mutate adjusts the policy. +func newGateFixture(t *testing.T, mutate func(*sendingpolicy.RuntimePolicy)) *gateFixture { + t.Helper() + ctx := context.Background() + pool := testutil.TestDB(t) + keyring, err := sendingpolicy.LoadKeyring(psHMAC) + if err != nil { + t.Fatalf("load keyring: %v", err) + } + recipients, err := sendingpolicy.LoadOperatorRecipients(psOperator) + if err != nil { + t.Fatalf("load operator map: %v", err) + } + secrets := sendingpolicy.Secrets{Keyring: keyring, Recipients: recipients} + if _, err := sendingpolicy.NewModule(pool, secrets).RegisterOperatorRecipients(ctx, "fixture", "submit test bootstrap"); err != nil { + t.Fatalf("register operator recipients: %v", err) + } + policy := sendingpolicy.DisabledPolicy() + policy.BudgetMode = sendingpolicy.ModeEnforce + if mutate != nil { + mutate(&policy) + } + f := &gateFixture{ + t: t, ctx: ctx, pool: pool, + gate: sendingpolicy.NewGate(pool, secrets, sendingpolicy.PolicySourceConfig, policy), + userID: psID("usr"), + agent: psID("agt"), + tenant: psID("tenant"), + } + if _, err := pool.Exec(ctx, + `INSERT INTO users (id, email, google_subject, account_class) VALUES ($1, $2, $3, 'standard')`, + f.userID, f.userID+"@example.test", "sub_"+f.userID, + ); err != nil { + t.Fatalf("insert user: %v", err) + } + if _, err := pool.Exec(ctx, ` + INSERT INTO account_sending_controls (user_id, ses_tenant_name, ses_tenant_ready, ses_tenant_ready_at) + VALUES ($1, $2, true, now()) + ON CONFLICT (user_id) DO UPDATE + SET ses_tenant_name = EXCLUDED.ses_tenant_name, ses_tenant_ready = true, ses_tenant_ready_at = now()`, + f.userID, f.tenant, + ); err != nil { + t.Fatalf("provision tenant: %v", err) + } + if _, err := pool.Exec(ctx, + `INSERT INTO agent_identities (id, user_id, registered_domain, name) VALUES ($1, $2, 'agents.e2a.dev', $1)`, + f.agent, f.userID, + ); err != nil { + t.Fatalf("insert agent: %v", err) + } + return f +} + +// message inserts an outbound message with `count` distinct recipients. +func (f *gateFixture) message(count int) (string, []string) { + f.t.Helper() + id := psID("msg") + to := make([]string, count) + for i := range to { + to[i] = fmt.Sprintf("rcpt-%s-%d@example.test", id, i) + } + if _, err := f.pool.Exec(f.ctx, + `INSERT INTO messages (id, agent_id, direction, to_recipients, sent_as, status) + VALUES ($1, $2, 'outbound', $3, 'own_address', 'sent')`, id, f.agent, to, + ); err != nil { + f.t.Fatalf("insert message: %v", err) + } + return id, to +} + +// prepare runs the acceptance half the way an API handler does. +func (f *gateFixture) prepare(messageID string) sendingpolicy.OperationRef { + f.t.Helper() + tx, err := f.pool.Begin(f.ctx) + if err != nil { + f.t.Fatalf("begin: %v", err) + } + accept, ref, err := f.gate.PrepareExternalTx(f.ctx, tx, messageID) + if err != nil { + _ = tx.Rollback(f.ctx) + f.t.Fatalf("prepare: %v", err) + } + if accept != sendingpolicy.AcceptanceAccept { + _ = tx.Rollback(f.ctx) + f.t.Fatalf("prepare decision = %v, want accept", accept) + } + if err := tx.Commit(f.ctx); err != nil { + f.t.Fatalf("commit: %v", err) + } + return ref +} + +// authorize runs the worker's sequence — Reserve, then ConsumeAttempt — and +// fails the test on a hold, because every test here is about what happens +// AFTER a token exists. +func (f *gateFixture) authorize(ref sendingpolicy.OperationRef) sendingpolicy.ProviderAuthorization { + f.t.Helper() + early, attempt, err := f.gate.Reserve(f.ctx, ref) + if err != nil { + f.t.Fatalf("reserve: %v", err) + } + if !early.Allow { + f.t.Fatalf("reserve held: %+v", early) + } + decision, auth, err := f.gate.ConsumeAttempt(f.ctx, attempt) + if err != nil { + f.t.Fatalf("consume: %v", err) + } + if !decision.Allow || auth == nil { + f.t.Fatalf("consume held: %+v (token=%v)", decision, auth != nil) + } + return *auth +} + +func (f *gateFixture) callState(operationID string, attempt int) string { + f.t.Helper() + var state string + if err := f.pool.QueryRow(f.ctx, ` + SELECT call_state FROM sending_budget_reservations + WHERE operation_id = $1 AND submission_attempt = $2`, operationID, attempt, + ).Scan(&state); err != nil { + f.t.Fatalf("read call_state: %v", err) + } + return state +} + +func (f *gateFixture) correlation(operationID string, attempt int) (correlationID string, providerMessageID *string) { + f.t.Helper() + err := f.pool.QueryRow(f.ctx, ` + SELECT correlation_id, provider_message_id FROM sending_feedback_correlations + WHERE operation_id = $1 AND submission_attempt = $2`, operationID, attempt, + ).Scan(&correlationID, &providerMessageID) + if err != nil && !errors.Is(err, pgx.ErrNoRows) { + f.t.Fatalf("read correlation: %v", err) + } + return correlationID, providerMessageID +} + +// countingListener accepts and immediately drops connections, counting them. +// A relay pointed at it can never complete a transaction, so any nonzero +// count is a socket that should not have been opened. +func countingListener(t *testing.T) (*outbound.SMTPRelay, func() int) { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = listener.Close() }) + var mu sync.Mutex + count := 0 + go func() { + for { + conn, err := listener.Accept() + if err != nil { + return + } + mu.Lock() + count++ + mu.Unlock() + _ = conn.Close() + } + }() + addr := listener.Addr().(*net.TCPAddr) + relay := outbound.NewSMTPRelay(&config.OutboundSMTPConfig{Host: addr.IP.String(), Port: addr.Port}) + return relay, func() int { mu.Lock(); defer mu.Unlock(); return count } +} + +// acceptingRelay fronts testutil's fake SMTP server. +func acceptingRelay(t *testing.T) (*outbound.SMTPRelay, func() []testutil.SMTPMessage) { + t.Helper() + addr, messages := testutil.FakeSMTPServer(t) + return outbound.NewSMTPRelay(&config.OutboundSMTPConfig{Host: addr.Host, Port: addr.Port}), messages +} + +// rejectingRelay fronts a server that answers every RCPT TO with a 550. +func rejectingRelay(t *testing.T) *outbound.SMTPRelay { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = listener.Close() }) + go func() { + for { + conn, err := listener.Accept() + if err != nil { + return + } + go func(conn net.Conn) { + defer conn.Close() + r := bufio.NewReader(conn) + fmt.Fprint(conn, "220 reject ready\r\n") + for { + line, err := r.ReadString('\n') + if err != nil { + return + } + switch upper := strings.ToUpper(strings.TrimSpace(line)); { + case strings.HasPrefix(upper, "RCPT TO:"): + fmt.Fprint(conn, "550 5.1.1 no such user\r\n") + case upper == "QUIT": + fmt.Fprint(conn, "221 Bye\r\n") + return + default: + fmt.Fprint(conn, "250 OK\r\n") + } + } + }(conn) + } + }() + addr := listener.Addr().(*net.TCPAddr) + return outbound.NewSMTPRelay(&config.OutboundSMTPConfig{Host: addr.IP.String(), Port: addr.Port}) +} + +// headerValues returns every value of `name` in the header section of a +// captured message, case-insensitively, with folded continuations unfolded. +func headerValues(data, name string) []string { + var out []string + current := -1 + for _, line := range strings.Split(data, "\n") { + line = strings.TrimRight(line, "\r") + if line == "" { + break + } + if line[0] == ' ' || line[0] == '\t' { + if current >= 0 { + out[current] += " " + strings.TrimSpace(line) + } + continue + } + current = -1 + if i := strings.IndexByte(line, ':'); i > 0 && strings.EqualFold(strings.TrimSpace(line[:i]), name) { + out = append(out, strings.TrimSpace(line[i+1:])) + current = len(out) - 1 + } + } + return out +} + +func body(data string) string { + if i := strings.Index(data, "\n\n"); i >= 0 { + return data[i+2:] + } + return "" +} + +func TestProviderSubmitterZeroNetworkWithoutAuthorization(t *testing.T) { + f := newGateFixture(t, nil) + relay, sockets := countingListener(t) + s := outbound.NewProviderSubmitter(relay, f.gate) + + _, err := s.SubmitOnce(f.ctx, sendingpolicy.ProviderAuthorization{}, outbound.Envelope{ + From: "agent@agents.e2a.dev", Recipients: []string{"someone@example.test"}, Message: []byte("Subject: x\r\n\r\nbody"), + }) + if !errors.Is(err, outbound.ErrAuthorizationRequired) { + t.Fatalf("err = %v, want outbound.ErrAuthorizationRequired", err) + } + if sockets() != 0 { + t.Fatalf("sockets = %d, want 0", sockets()) + } +} + +// TestProviderSubmitterZeroNetworkOnEnvelopeMismatch proves the envelope check +// runs before redemption: a wrong envelope costs nothing, sends nothing, and +// leaves the token spendable for the right one. +func TestProviderSubmitterZeroNetworkOnEnvelopeMismatch(t *testing.T) { + f := newGateFixture(t, nil) + relay, sockets := countingListener(t) + s := outbound.NewProviderSubmitter(relay, f.gate) + messageID, to := f.message(2) + ref := f.prepare(messageID) + auth := f.authorize(ref) + + for name, envelope := range map[string][]string{ + "extra recipient": append(append([]string(nil), to...), "attacker@example.test"), + "swapped recipient": {to[0], "attacker@example.test"}, + "dropped recipient": {to[0]}, + "duplicated mailbox": {to[0], strings.ToUpper(to[0]), to[1]}, + "empty": nil, + } { + _, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: envelope, Message: []byte("Subject: x\r\n\r\nbody")}) + if err == nil { + t.Fatalf("%s: submitted, want refusal", name) + } + if state := f.callState(ref.ID(), 1); state != "authorized" { + t.Fatalf("%s: call_state = %s, want authorized (token must survive a mismatch)", name, state) + } + } + if sockets() != 0 { + t.Fatalf("sockets = %d, want 0", sockets()) + } +} + +func TestProviderSubmitterZeroNetworkWhenRelayUnconfigured(t *testing.T) { + f := newGateFixture(t, nil) + s := outbound.NewProviderSubmitter(outbound.NewSMTPRelay(&config.OutboundSMTPConfig{}), f.gate) + messageID, to := f.message(1) + ref := f.prepare(messageID) + auth := f.authorize(ref) + + if _, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")}); err == nil { + t.Fatal("submitted through an unconfigured relay") + } + // A misconfigured relay is not the message's fault: the token is intact. + if state := f.callState(ref.ID(), 1); state != "authorized" { + t.Fatalf("call_state = %s, want authorized", state) + } +} + +// TestProviderSubmitterAuthorizedTokenIsSingleUse proves the token is spent by +// the call, not merely checked: a second submission with the same token opens +// no socket. +func TestProviderSubmitterAuthorizedTokenIsSingleUse(t *testing.T) { + f := newGateFixture(t, nil) + relay, captured := acceptingRelay(t) + s := outbound.NewProviderSubmitter(relay, f.gate) + messageID, to := f.message(1) + ref := f.prepare(messageID) + auth := f.authorize(ref) + env := outbound.Envelope{MessageID: messageID, From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")} + + res, err := s.SubmitOnce(f.ctx, auth, env) + if err != nil || res.SettlementErr != nil || res.ProviderMessageID == "" { + t.Fatalf("first submit: res=%+v err=%v", res, err) + } + if state := f.callState(ref.ID(), 1); state != "started" { + t.Fatalf("call_state = %s, want started", state) + } + + _, err = s.SubmitOnce(f.ctx, auth, env) + if !errors.Is(err, sendingpolicy.ErrAuthorizationInvalid) { + t.Fatalf("second submit err = %v, want ErrAuthorizationInvalid", err) + } + if n := len(captured()); n != 1 { + t.Fatalf("provider received %d messages for one token, want 1", n) + } +} + +// TestProviderSubmitterZeroNetworkForStaleAttempt: a token whose ordinal has +// been superseded — the worker died and a later execution re-reserved — opens +// no socket. +func TestProviderSubmitterZeroNetworkForStaleAttempt(t *testing.T) { + f := newGateFixture(t, nil) + relay, sockets := countingListener(t) + s := outbound.NewProviderSubmitter(relay, f.gate) + messageID, to := f.message(1) + ref := f.prepare(messageID) + stale := f.authorize(ref) + if _, next, err := f.gate.Reserve(f.ctx, ref); err != nil || next.Attempt() != 2 { + t.Fatalf("re-reserve: attempt=%v err=%v", next, err) + } + + _, err := s.SubmitOnce(f.ctx, stale, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")}) + if !errors.Is(err, sendingpolicy.ErrAuthorizationInvalid) { + t.Fatalf("err = %v, want ErrAuthorizationInvalid", err) + } + if sockets() != 0 { + t.Fatalf("sockets = %d, want 0", sockets()) + } +} + +// TestProviderSubmitterAttemptHeaderDerivesOnlyFromToken proves the wire +// carries exactly one attempt header, its value is the gate's correlation id, +// every smuggled spelling is gone, and the body is untouched. +func TestProviderSubmitterAttemptHeaderDerivesOnlyFromToken(t *testing.T) { + f := newGateFixture(t, nil) + relay, captured := acceptingRelay(t) + s := outbound.NewProviderSubmitter(relay, f.gate) + s.SetSESConfigurationSet("e2a-delivery-test") + messageID, to := f.message(1) + ref := f.prepare(messageID) + auth := f.authorize(ref) + + res, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{MessageID: messageID, From: "agent@agents.e2a.dev", Recipients: to, Message: outbound.SmuggledMIME()}) + if err != nil || res.SettlementErr != nil { + t.Fatalf("submit: res=%+v err=%v", res, err) + } + msgs := captured() + if len(msgs) != 1 { + t.Fatalf("captured %d messages, want 1", len(msgs)) + } + data := msgs[0].Data + corr, _ := f.correlation(ref.ID(), 1) + if corr == "" { + t.Fatal("no correlation row for the attempt") + } + for name, want := range map[string][]string{ + outbound.ProviderAttemptHeader: {corr}, + outbound.SESConfigurationSetHeader: {"e2a-delivery-test"}, + delivery.MessageIDHeader: {messageID}, + outbound.SESTenantHeader: nil, // policy has the tenant header disabled + } { + if got := headerValues(data, name); strings.Join(got, "|") != strings.Join(want, "|") { + t.Errorf("%s = %q, want %q", name, got, want) + } + } + if got := headerValues(data, "Subject"); len(got) != 1 || got[0] != "hello" { + t.Errorf("customer headers disturbed: Subject = %q", got) + } + if b := body(data); !strings.Contains(b, "X-SES-TENANT: body-decoy") || !strings.Contains(b, "body line") { + t.Errorf("body was rewritten: %q", b) + } + if strings.Contains(data, "smuggled") || strings.Contains(data, "forged") || strings.Contains(data, "attacker") { + t.Errorf("a smuggled header value survived:\n%s", data) + } +} + +// TestProviderSubmitterTenantHeaderIsExactAndSingle: under an enforcing tenant +// policy the wire carries exactly one X-SES-TENANT, whose value is the tenant +// the gate read under lock — not anything the MIME said. +func TestProviderSubmitterTenantHeaderIsExactAndSingle(t *testing.T) { + f := newGateFixture(t, func(p *sendingpolicy.RuntimePolicy) { + p.TenantHeaderMode = sendingpolicy.TenantHeaderEnforce + }) + relay, captured := acceptingRelay(t) + s := outbound.NewProviderSubmitter(relay, f.gate) + messageID, to := f.message(1) + ref := f.prepare(messageID) + auth := f.authorize(ref) + + if _, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: outbound.SmuggledMIME()}); err != nil { + t.Fatalf("submit: %v", err) + } + msgs := captured() + if len(msgs) != 1 { + t.Fatalf("captured %d messages, want 1", len(msgs)) + } + if got := headerValues(msgs[0].Data, outbound.SESTenantHeader); len(got) != 1 || got[0] != f.tenant { + t.Fatalf("%s = %q, want exactly [%q]", outbound.SESTenantHeader, got, f.tenant) + } +} + +// TestProviderSubmitterRetryRedeemsADistinctAttempt: a physical retry is a new +// ordinal with a new token and a new attempt header, never a resubmission +// under the old one. +func TestProviderSubmitterRetryRedeemsADistinctAttempt(t *testing.T) { + f := newGateFixture(t, nil) + relay, captured := acceptingRelay(t) + s := outbound.NewProviderSubmitter(relay, f.gate) + messageID, to := f.message(1) + ref := f.prepare(messageID) + env := outbound.Envelope{MessageID: messageID, From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")} + + first := f.authorize(ref) + if _, err := s.SubmitOnce(f.ctx, first, env); err != nil { + t.Fatalf("first: %v", err) + } + second := f.authorize(ref) + if second.Attempt().Attempt() != 2 { + t.Fatalf("second ordinal = %d, want 2", second.Attempt().Attempt()) + } + if _, err := s.SubmitOnce(f.ctx, second, env); err != nil { + t.Fatalf("second: %v", err) + } + msgs := captured() + if len(msgs) != 2 { + t.Fatalf("captured %d messages, want 2", len(msgs)) + } + c1, _ := f.correlation(ref.ID(), 1) + c2, _ := f.correlation(ref.ID(), 2) + h1 := headerValues(msgs[0].Data, outbound.ProviderAttemptHeader) + h2 := headerValues(msgs[1].Data, outbound.ProviderAttemptHeader) + if len(h1) != 1 || len(h2) != 1 || h1[0] != c1 || h2[0] != c2 || c1 == c2 { + t.Fatalf("attempt headers %v / %v, want distinct correlations %q / %q", h1, h2, c1, c2) + } +} + +func TestProviderSubmitterBindsProviderMessageIDOnAcceptance(t *testing.T) { + f := newGateFixture(t, nil) + relay, _ := acceptingRelay(t) + s := outbound.NewProviderSubmitter(relay, f.gate) + messageID, to := f.message(1) + ref := f.prepare(messageID) + auth := f.authorize(ref) + + res, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")}) + if err != nil || res.SettlementErr != nil { + t.Fatalf("submit: res=%+v err=%v", res, err) + } + if res.Attempt.Attempt() != 1 { + t.Errorf("result attempt = %d, want 1", res.Attempt.Attempt()) + } + _, bound := f.correlation(ref.ID(), 1) + if bound == nil || *bound != res.ProviderMessageID { + t.Fatalf("correlation provider_message_id = %v, want %q", bound, res.ProviderMessageID) + } +} + +// TestProviderSubmitterPermanentRejectionIsSettledNotRetried: a definite 5xx +// consumed the attempt (the socket opened), settles as rejected, binds no +// provider id, and surfaces as a permanent error the worker can classify. +func TestProviderSubmitterPermanentRejectionIsSettledNotRetried(t *testing.T) { + f := newGateFixture(t, nil) + s := outbound.NewProviderSubmitter(rejectingRelay(t), f.gate) + messageID, to := f.message(1) + ref := f.prepare(messageID) + auth := f.authorize(ref) + + _, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")}) + if err == nil || !outbound.IsPermanentSMTPError(err) { + t.Fatalf("err = %v, want a permanent SMTP error", err) + } + if state := f.callState(ref.ID(), 1); state != "started" { + t.Fatalf("call_state = %s, want started (the socket did open)", state) + } + if _, bound := f.correlation(ref.ID(), 1); bound != nil { + t.Fatalf("provider_message_id = %q bound on a rejection", *bound) + } +} diff --git a/internal/sendingpolicy/gate.go b/internal/sendingpolicy/gate.go index 020b5b76c..9301de869 100644 --- a/internal/sendingpolicy/gate.go +++ b/internal/sendingpolicy/gate.go @@ -1624,9 +1624,62 @@ func (m *Module) SettleProvider(ctx context.Context, settlement ProviderSettleme return ErrAttemptStale } + if err := bindProviderMessageID(ctx, tx, settlement); err != nil { + return err + } // The ramp ledger is Task 4's to move. Validating the attempt here — and // taking its locks in the normative order — is what makes that composition // a body change rather than a reshaping of the interface every caller // already uses. return m.commit(ctx, tx, "settle") } + +// bindProviderMessageID records the provider's id on the attempt's feedback +// correlation, exactly once. +// +// It runs after the operation and reservation locks, which is the only place +// the correlation row is ever written after its insert, so no additional key +// joins the normative order. Replaying the same id is a no-op — the +// synchronous success branch and the delayed feedback finalizer both settle +// the same attempt — while a different id is refused outright. A correlation +// that has already aged out of retention is left alone: there is nothing to +// attribute feedback to any more, and failing a late settlement over it would +// only make the caller retry forever. +func bindProviderMessageID(ctx context.Context, tx pgx.Tx, settlement ProviderSettlement) error { + id := strings.TrimSpace(settlement.ProviderMessageID) + if id == "" { + return nil + } + if settlement.Outcome != SettlementProviderAccepted { + return fmt.Errorf("sendingpolicy: a %q settlement cannot carry a provider message id", settlement.Outcome) + } + var bound *string + err := tx.QueryRow(ctx, ` + SELECT provider_message_id + FROM sending_feedback_correlations + WHERE operation_id = $1 AND submission_attempt = $2 + FOR UPDATE`, + settlement.Attempt.operationID, settlement.Attempt.attempt, + ).Scan(&bound) + if errors.Is(err, pgx.ErrNoRows) { + return nil + } + if err != nil { + return fmt.Errorf("sendingpolicy: lock correlation: %w", err) + } + if bound != nil { + if *bound == id { + return nil + } + return ErrProviderMessageIDConflict + } + if _, err := tx.Exec(ctx, ` + UPDATE sending_feedback_correlations + SET provider_message_id = $3 + WHERE operation_id = $1 AND submission_attempt = $2`, + settlement.Attempt.operationID, settlement.Attempt.attempt, id, + ); err != nil { + return fmt.Errorf("sendingpolicy: bind provider message id: %w", err) + } + return nil +} diff --git a/internal/sendingpolicy/provider_token_test.go b/internal/sendingpolicy/provider_token_test.go new file mode 100644 index 000000000..a9b06e20d --- /dev/null +++ b/internal/sendingpolicy/provider_token_test.go @@ -0,0 +1,127 @@ +package sendingpolicy_test + +import ( + "errors" + "testing" + + "github.com/tokencanopy/e2a/internal/sendingpolicy" +) + +// These tests are about what a settlement may bind to the attempt's feedback +// correlation. The provider id is how most delivery feedback finds its +// attempt, so it has to be written exactly once, by an acceptance, and never +// rewritten. + +func (f *fixture) providerMessageID(operationID string, attempt int) *string { + f.t.Helper() + var id *string + if err := f.pool.QueryRow(f.ctx, ` + SELECT provider_message_id FROM sending_feedback_correlations + WHERE operation_id = $1 AND submission_attempt = $2`, operationID, attempt, + ).Scan(&id); err != nil { + f.t.Fatalf("read provider_message_id: %v", err) + } + return id +} + +// redeemed mints a token and redeems it, leaving the attempt in the state a +// settlement expects. +func (f *fixture) redeemed(g sendingpolicy.Gate, agent string) (sendingpolicy.OperationRef, *sendingpolicy.ProviderAuthorization) { + f.t.Helper() + ref, attempt := f.prepareAndReserve(g, agent, 1) + _, auth, err := g.ConsumeAttempt(f.ctx, attempt) + if err != nil || auth == nil { + f.t.Fatalf("authorize: auth=%v err=%v", auth, err) + } + if err := g.RedeemProviderCall(f.ctx, *auth); err != nil { + f.t.Fatalf("redeem: %v", err) + } + return ref, auth +} + +func TestProviderTokenSettlementBindsProviderMessageIDOnce(t *testing.T) { + f := newFixture(t) + g := f.gate(enforcingPolicy(nil)) + ref, auth := f.redeemed(g, f.agent(f.user("standard"))) + + accepted := sendingpolicy.ProviderSettlement{ + Attempt: auth.Attempt(), Outcome: sendingpolicy.SettlementProviderAccepted, ProviderMessageID: "ses-id-one", + } + if err := g.SettleProvider(f.ctx, accepted); err != nil { + t.Fatalf("settle: %v", err) + } + if got := f.providerMessageID(ref.ID(), 1); got == nil || *got != "ses-id-one" { + t.Fatalf("bound = %v, want ses-id-one", got) + } + + // The delayed feedback finalizer settles the same attempt again. + if err := g.SettleProvider(f.ctx, accepted); err != nil { + t.Fatalf("replay: %v", err) + } + + // A different id for the same attempt is two physical sends for one + // charge. Refuse it and keep the first. + conflicting := accepted + conflicting.ProviderMessageID = "ses-id-two" + if err := g.SettleProvider(f.ctx, conflicting); !errors.Is(err, sendingpolicy.ErrProviderMessageIDConflict) { + t.Fatalf("conflict err = %v, want ErrProviderMessageIDConflict", err) + } + if got := f.providerMessageID(ref.ID(), 1); got == nil || *got != "ses-id-one" { + t.Fatalf("bound after conflict = %v, want ses-id-one kept", got) + } +} + +func TestProviderTokenRejectionNeverBindsProviderMessageID(t *testing.T) { + f := newFixture(t) + g := f.gate(enforcingPolicy(nil)) + ref, auth := f.redeemed(g, f.agent(f.user("standard"))) + + // A rejection carrying an id is a caller bug: nothing was accepted, so + // there is nothing to attribute feedback to. + err := g.SettleProvider(f.ctx, sendingpolicy.ProviderSettlement{ + Attempt: auth.Attempt(), Outcome: sendingpolicy.SettlementProviderPermanentlyRejected, ProviderMessageID: "ses-id", + }) + if err == nil { + t.Fatal("a rejection with a provider id was accepted") + } + if got := f.providerMessageID(ref.ID(), 1); got != nil { + t.Fatalf("bound = %q on a refused settlement", *got) + } + + if err := g.SettleProvider(f.ctx, sendingpolicy.ProviderSettlement{ + Attempt: auth.Attempt(), Outcome: sendingpolicy.SettlementProviderPermanentlyRejected, + }); err != nil { + t.Fatalf("plain rejection: %v", err) + } + if got := f.providerMessageID(ref.ID(), 1); got != nil { + t.Fatalf("bound = %q after a rejection", *got) + } +} + +// TestProviderTokenAcceptanceWithoutIDStillSettles keeps the pre-existing +// contract: an acceptance whose id was lost (crash between DATA and the +// result) settles normally and leaves the correlation open for the feedback +// path to bind by attempt header. +func TestProviderTokenAcceptanceWithoutIDStillSettles(t *testing.T) { + f := newFixture(t) + g := f.gate(enforcingPolicy(nil)) + ref, auth := f.redeemed(g, f.agent(f.user("standard"))) + + if err := g.SettleProvider(f.ctx, sendingpolicy.ProviderSettlement{ + Attempt: auth.Attempt(), Outcome: sendingpolicy.SettlementProviderAccepted, + }); err != nil { + t.Fatalf("settle: %v", err) + } + if got := f.providerMessageID(ref.ID(), 1); got != nil { + t.Fatalf("bound = %q, want nothing bound", *got) + } + // And a later settlement that does know the id may still bind it. + if err := g.SettleProvider(f.ctx, sendingpolicy.ProviderSettlement{ + Attempt: auth.Attempt(), Outcome: sendingpolicy.SettlementProviderAccepted, ProviderMessageID: "ses-late", + }); err != nil { + t.Fatalf("late bind: %v", err) + } + if got := f.providerMessageID(ref.ID(), 1); got == nil || *got != "ses-late" { + t.Fatalf("bound = %v, want ses-late", got) + } +} diff --git a/internal/sendingpolicy/types.go b/internal/sendingpolicy/types.go index 4ed728861..0f7320e59 100644 --- a/internal/sendingpolicy/types.go +++ b/internal/sendingpolicy/types.go @@ -403,7 +403,20 @@ func (o SettlementOutcome) valid() bool { type ProviderSettlement struct { Attempt AttemptRef Outcome SettlementOutcome -} + // ProviderMessageID is the id SES assigned when it accepted the message. + // It is bound to the attempt's feedback correlation so delivery feedback + // that arrives by provider id — the common case — resolves to the same + // attempt as feedback that arrives by the random attempt header. Only an + // accepted settlement may carry one; a rejection has nothing to bind. + ProviderMessageID string +} + +// ErrProviderMessageIDConflict means an attempt is being settled with a +// different provider message id than the one already bound to it. One attempt +// is exactly one DATA transaction and SES assigns exactly one id to it, so a +// second, different id is evidence of two physical sends for one charged +// attempt — the invariant this module exists to hold — and is never absorbed. +var ErrProviderMessageIDConflict = errors.New("sendingpolicy: attempt already settled with a different provider message id") // TenantMode is the closed tenant-header state carried by an authorization. // It is resolved under the final account-control lock, so a job that was From 1e9e8a32b7cd2e6c387de9aa2e91ad9220835ebd Mon Sep 17 00:00:00 2001 From: Josh Zhang <39790535+jiashuoz@users.noreply.github.com> Date: Fri, 4 Sep 2026 20:40:20 -0700 Subject: [PATCH 3/4] fix(outbound): harden the provider seam after review Two review passes (correctness + adversarial) over the first cut. Every finding below has a named test, and removing each fix fails it. Gate: - RedeemProviderCall re-proves the abuse pause. ConsumeAttempt linearized it, but that transaction has committed; a pause landing before the dial now invalidates the token. Unlocked read, because the account-control key precedes the operation key this transaction already holds. - SettleProvider requires call_state = 'started', not just a confirmed reservation: a provider id bound to an attempt that never dialed is a ledger claim about a send that did not happen. - The provider message id is normalized to SES's bare form on bind and on compare (NormalizeProviderMessageID). The relay returns it qualified (); SNS feedback carries it bare; the worker and the feedback finalizer must not refuse each other's spelling. - A tenant name that cannot be a header value holds at mint with ses_tenant_unnamed instead of wedging silently in the adapter. Adapter: - X-E2A-Message-ID is derived from the token (a customer operation IS its message id), not trusted from the caller; Envelope.MessageID is gone. - RCPT TO is issued from AuthorizedRecipients(), the canonical envelope the budget priced, never the caller's spelling of it. - A bare CR anywhere in the header section, or a leading continuation line, is refused before redemption (ErrMalformedHeaderSection): a receiver that treats a lone CR as a line break would see a header this walker did not. - Provider headers are emitted in the legacy path's order (configuration set first) so the worker swap is byte-identical for shared headers. - The relay marks a failure after the terminating dot with ErrProviderAcceptanceUnknown; the adapter leaves it unsettled and no classifier calls it permanent. Tests: pause between consume and submit; settle without redeem; bare CR and leading continuation refused with zero sockets; canonical RCPT on the wire; lost 250 unsettled and marked; 4xx unsettled; accepted-but-unsettled reported on the result with a nil error; a positive control proving the socket counter observes a dial; provider id normalization across the three spellings. The socket counter now waits for the accept goroutine. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01AjfGxvXW6fNKWGFHuo68yX --- internal/outbound/provider_submit.go | 97 +++++- .../outbound/provider_submit_internal_test.go | 27 +- internal/outbound/provider_submit_test.go | 313 +++++++++++++++++- internal/outbound/smtp_relay.go | 11 +- internal/sendingpolicy/gate.go | 54 ++- .../provider_attempt_integration_test.go | 15 +- internal/sendingpolicy/provider_token_test.go | 87 +++++ internal/sendingpolicy/types.go | 19 ++ internal/testutil/testdb/db.go | 2 +- 9 files changed, 586 insertions(+), 39 deletions(-) diff --git a/internal/outbound/provider_submit.go b/internal/outbound/provider_submit.go index bfd5112e1..79668a82b 100644 --- a/internal/outbound/provider_submit.go +++ b/internal/outbound/provider_submit.go @@ -61,6 +61,13 @@ var ( // input — and a defect here is a header injection, so it fails closed // rather than being sanitized silently. ErrProviderHeaderValue = errors.New("outbound: provider header value contains a line break") + // ErrMalformedHeaderSection means the composed MIME's header section holds + // a bare carriage return. Receivers disagree on whether a lone CR ends a + // line, so a header hidden behind one might survive stripping here and + // still be honoured by the provider. The composer never emits one — every + // header value is sanitized — so this only fires on a compose defect, and + // it fails the send rather than guess. + ErrMalformedHeaderSection = errors.New("outbound: message header section contains a bare carriage return") ) // providerOwnedHeaders are removed from the composed MIME before submission, @@ -74,11 +81,19 @@ var providerOwnedHeaders = map[string]struct{}{ // Envelope is what a caller hands the provider seam: the SMTP envelope and the // composed wire bytes. +// +// There is deliberately no message id here. The stable X-E2A-Message-ID marker +// that delivery feedback keys on is derived from the token — a customer +// message's operation IS its message id — so a caller cannot stamp one +// message's id on another's send and misroute its bounces. +// +// The sender is the one envelope field the token does not bind: the +// authorization carries recipients and tenant, not MAIL FROM. SES enforces +// identity ownership of the sender domain on its side, and the caller here is +// the trusted worker that composed the message, so the seam only insists the +// sender is present. Binding it would need the gate to learn the composed +// sender at acceptance; that is a gate change, not an adapter one. type Envelope struct { - // MessageID is the e2a message id stamped as X-E2A-Message-ID, the stable - // correlation marker delivery feedback keys on. Empty for mail that has no - // message row. - MessageID string // From is the SMTP MAIL FROM address. From string // Recipients is the exact final envelope: one entry per distinct mailbox, @@ -99,6 +114,12 @@ type ProviderResult struct { // SettleProvider with the same attempt and provider id, and must never // resubmit. Delivery feedback carrying the attempt header is the fallback // if it never does. + // + // One value is not a retry: errors.Is(SettlementErr, + // sendingpolicy.ErrProviderMessageIDConflict) means this attempt was + // already settled with a DIFFERENT provider id — two physical sends for one + // charge. Retrying settlement cannot fix that; it must be surfaced as an + // invariant violation, not absorbed as a transient. SettlementErr error } @@ -145,7 +166,11 @@ func (s *ProviderSubmitter) SubmitOnce(ctx context.Context, auth sendingpolicy.P if err != nil { return ProviderResult{}, err } - provider, err := providerHeaderLines(headers, s.sesConfigSet, env.MessageID) + provider, err := providerHeaderLines(headers, s.sesConfigSet, correlationMessageID(auth)) + if err != nil { + return ProviderResult{}, err + } + stripped, err := stripProviderHeaders(env.Message) if err != nil { return ProviderResult{}, err } @@ -155,14 +180,29 @@ func (s *ProviderSubmitter) SubmitOnce(ctx context.Context, auth sendingpolicy.P if !s.relay.Configured() { return ProviderResult{}, fmt.Errorf("outbound SMTP relay not configured") } - wire := append(provider, stripProviderHeaders(env.Message)...) + wire := append(provider, stripped...) if err := s.gate.RedeemProviderCall(ctx, auth); err != nil { return ProviderResult{}, fmt.Errorf("provider authorization: %w", err) } - providerID, sendErr := s.relay.SendOnceContext(ctx, env.From, env.Recipients, wire) + // RCPT TO is issued from the token's canonical envelope, not the caller's + // spelling of it. ValidateEnvelope has just proved the two name the same + // mailboxes; what goes on the wire is the normalized set the budget priced, + // so a padded or upper-cased entry cannot turn into an SMTP grammar error + // that downstream classifies as the message's own permanent failure. + // + // A failure after the body was fully written (ErrProviderAcceptanceUnknown) + // is neither accepted nor rejected here: it is returned unsettled, because + // the provider may hold the message and only its feedback can say. + providerID, sendErr := s.relay.SendOnceContext(ctx, env.From, auth.AuthorizedRecipients(), wire) if sendErr != nil { + // IsPermanentSMTPError is the worker's retry classifier: any 5xx, + // including one raised before DATA (an AUTH 535, say). Settling such a + // failure as a rejection is conservative in the only direction that + // matters — the provider never took the message, so giving its + // capacity back is correct — and it keeps this seam's verdict identical + // to the one the worker already acts on. if IsPermanentSMTPError(sendErr) { if err := s.gate.SettleProvider(ctx, sendingpolicy.ProviderSettlement{ Attempt: auth.Attempt(), @@ -185,11 +225,24 @@ func (s *ProviderSubmitter) SubmitOnce(ctx context.Context, auth sendingpolicy.P return result, nil } +// correlationMessageID is the value of the stable X-E2A-Message-ID marker for +// a token: the message id for a customer message, nothing for every other +// purpose (operational mail has no message row for feedback to land on). +func correlationMessageID(auth sendingpolicy.ProviderAuthorization) string { + if auth.Purpose() == sendingpolicy.PurposeCustomerMessage { + return auth.Attempt().OperationID() + } + return "" +} + // providerHeaderLines renders the provider-owned header block from the token's // view and the deployment configuration — and from nothing else. // -// Callers cannot pass a tenant name or correlation id; the only way to change -// what SES receives is to change what the gate authorized. +// Callers cannot pass a tenant name, correlation id, or message id; the only +// way to change what SES receives is to change what the gate authorized. The +// order — configuration set, then message id — matches what Sender.SubmitOnce +// emits today, so swapping the worker onto this seam is byte-identical for +// the headers both paths share. func providerHeaderLines(h sendingpolicy.ProviderHeaders, sesConfigSet, messageID string) ([]byte, error) { if h.TenantRequired && strings.TrimSpace(h.TenantName) == "" { return nil, ErrTenantNameMissing @@ -200,6 +253,9 @@ func providerHeaderLines(h sendingpolicy.ProviderHeaders, sesConfigSet, messageI } } var b bytes.Buffer + if sesConfigSet != "" { + b.WriteString(SESConfigurationSetHeader + ": " + sesConfigSet + "\r\n") + } if messageID != "" { b.WriteString(delivery.MessageIDHeader + ": " + messageID + "\r\n") } @@ -209,9 +265,6 @@ func providerHeaderLines(h sendingpolicy.ProviderHeaders, sesConfigSet, messageI if h.TenantRequired { b.WriteString(SESTenantHeader + ": " + h.TenantName + "\r\n") } - if sesConfigSet != "" { - b.WriteString(SESConfigurationSetHeader + ": " + sesConfigSet + "\r\n") - } return b.Bytes(), nil } @@ -225,7 +278,18 @@ func providerHeaderLines(h sendingpolicy.ProviderHeaders, sesConfigSet, messageI // its name line plus every folded continuation (a line starting with space or // tab); dropping a field drops its continuations with it. Matching is on the // lowercased name, so mixed-case spellings are not an evasion. -func stripProviderHeaders(msg []byte) []byte { +// +// Lines end in LF or CRLF. A carriage return anywhere else in the header +// section is refused (ErrMalformedHeaderSection): a receiver that treats a +// bare CR as a line break would see a header this walker did not, and the +// composer never produces one. +func stripProviderHeaders(msg []byte) ([]byte, error) { + // A message whose first line is a folded continuation has nothing to + // continue — except the provider header this adapter is about to prepend, + // whose value it would silently extend. Refuse it. + if len(msg) > 0 && (msg[0] == ' ' || msg[0] == '\t') { + return nil, ErrMalformedHeaderSection + } out := make([]byte, 0, len(msg)) rest := msg dropping := false @@ -237,13 +301,16 @@ func stripProviderHeaders(msg []byte) []byte { } else { line, rest = rest[:nl+1], rest[nl+1:] } + if raw := bytes.TrimSuffix(bytes.TrimSuffix(line, []byte("\n")), []byte("\r")); bytes.IndexByte(raw, '\r') >= 0 { + return nil, ErrMalformedHeaderSection + } trimmed := bytes.TrimRight(line, "\r\n") if len(trimmed) == 0 { // End of the header section: emit the separator and the body // verbatim. out = append(out, line...) out = append(out, rest...) - return out + return out, nil } if trimmed[0] == ' ' || trimmed[0] == '\t' { if !dropping { @@ -261,5 +328,5 @@ func stripProviderHeaders(msg []byte) []byte { } out = append(out, line...) } - return out + return out, nil } diff --git a/internal/outbound/provider_submit_internal_test.go b/internal/outbound/provider_submit_internal_test.go index ec506a808..56b8238e5 100644 --- a/internal/outbound/provider_submit_internal_test.go +++ b/internal/outbound/provider_submit_internal_test.go @@ -57,9 +57,9 @@ func TestProviderHeaderLinesFailClosed(t *testing.T) { "no tenant, no set, no id": { h: sendingpolicy.ProviderHeaders{AttemptCorrelationID: "cor_1"}, want: "X-E2A-Provider-Attempt: cor_1\r\n", }, - "everything": { + "everything, in the legacy path's order": { h: sendingpolicy.ProviderHeaders{AttemptCorrelationID: "cor_1", TenantRequired: true, TenantName: "tenant_a"}, set: "cs", id: "msg_1", - want: "X-E2A-Message-ID: msg_1\r\nX-E2A-Provider-Attempt: cor_1\r\nX-SES-TENANT: tenant_a\r\nX-SES-CONFIGURATION-SET: cs\r\n", + want: "X-SES-CONFIGURATION-SET: cs\r\nX-E2A-Message-ID: msg_1\r\nX-E2A-Provider-Attempt: cor_1\r\nX-SES-TENANT: tenant_a\r\n", }, } { got, err := providerHeaderLines(tc.h, tc.set, tc.id) @@ -73,7 +73,10 @@ func TestProviderHeaderLinesFailClosed(t *testing.T) { } func TestStripProviderHeaders(t *testing.T) { - for name, tc := range map[string]struct{ in, want string }{ + for name, tc := range map[string]struct { + in, want string + wantErr error + }{ "mixed case, duplicates, folded": { in: string(smuggledMIME()), want: "From: agent@agents.e2a.dev\r\nSubject: hello\r\n\r\nX-SES-TENANT: body-decoy\r\nbody line\r\n", @@ -102,9 +105,25 @@ func TestStripProviderHeaders(t *testing.T) { in: "Subject: long\r\n subject\r\nX-SES-TENANT: a\r\n b\r\nTo: x@example.test\r\n\r\n", want: "Subject: long\r\n subject\r\nTo: x@example.test\r\n\r\n", }, + "body may contain bare CR": { + in: "Subject: s\r\n\r\nbinary\rbody\r\n", + want: "Subject: s\r\n\r\nbinary\rbody\r\n", + }, + "bare CR hiding a header is refused": { + in: "Subject: a\rX-SES-TENANT: evil\r\n\r\nbody\r\n", + wantErr: ErrMalformedHeaderSection, + }, + "CR CR LF pseudo-separator is refused": { + in: "Subject: s\r\n\r\r\nX-SES-TENANT: evil\r\n\r\nbody", + wantErr: ErrMalformedHeaderSection, + }, "empty": {in: "", want: ""}, } { - if got := string(stripProviderHeaders([]byte(tc.in))); got != tc.want { + got, err := stripProviderHeaders([]byte(tc.in)) + if !errors.Is(err, tc.wantErr) { + t.Errorf("%s: err = %v, want %v", name, err, tc.wantErr) + } + if string(got) != tc.want { t.Errorf("%s:\n got %q\nwant %q", name, got, tc.want) } } diff --git a/internal/outbound/provider_submit_test.go b/internal/outbound/provider_submit_test.go index 6d6917084..004d4b89d 100644 --- a/internal/outbound/provider_submit_test.go +++ b/internal/outbound/provider_submit_test.go @@ -10,6 +10,7 @@ import ( "strings" "sync" "testing" + "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" @@ -212,7 +213,22 @@ func countingListener(t *testing.T) (*outbound.SMTPRelay, func() int) { }() addr := listener.Addr().(*net.TCPAddr) relay := outbound.NewSMTPRelay(&config.OutboundSMTPConfig{Host: addr.IP.String(), Port: addr.Port}) - return relay, func() int { mu.Lock(); defer mu.Unlock(); return count } + // The accept goroutine runs after the client's dial returns, so a caller + // that reads immediately could miss a connection that did happen. Give a + // connection time to be observed: return as soon as one is, or after a + // grace period that is long compared to a loopback accept. + return relay, func() int { + deadline := time.Now().Add(300 * time.Millisecond) + for { + mu.Lock() + c := count + mu.Unlock() + if c > 0 || time.Now().After(deadline) { + return c + } + time.Sleep(10 * time.Millisecond) + } + } } // acceptingRelay fronts testutil's fake SMTP server. @@ -222,8 +238,8 @@ func acceptingRelay(t *testing.T) (*outbound.SMTPRelay, func() []testutil.SMTPMe return outbound.NewSMTPRelay(&config.OutboundSMTPConfig{Host: addr.Host, Port: addr.Port}), messages } -// rejectingRelay fronts a server that answers every RCPT TO with a 550. -func rejectingRelay(t *testing.T) *outbound.SMTPRelay { +// rejectingRelay fronts a server that answers every RCPT TO with `reply`. +func rejectingRelay(t *testing.T, reply string) *outbound.SMTPRelay { t.Helper() listener, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { @@ -247,7 +263,7 @@ func rejectingRelay(t *testing.T) *outbound.SMTPRelay { } switch upper := strings.ToUpper(strings.TrimSpace(line)); { case strings.HasPrefix(upper, "RCPT TO:"): - fmt.Fprint(conn, "550 5.1.1 no such user\r\n") + fmt.Fprint(conn, reply+"\r\n") case upper == "QUIT": fmt.Fprint(conn, "221 Bye\r\n") return @@ -341,7 +357,10 @@ func TestProviderSubmitterZeroNetworkOnEnvelopeMismatch(t *testing.T) { } } -func TestProviderSubmitterZeroNetworkWhenRelayUnconfigured(t *testing.T) { +// TestProviderSubmitterUnconfiguredRelayLeavesTokenIntact: with no relay host +// there is nothing to dial and nothing to count; what matters is that the +// token survives, because the failure is the deployment's, not the message's. +func TestProviderSubmitterUnconfiguredRelayLeavesTokenIntact(t *testing.T) { f := newGateFixture(t, nil) s := outbound.NewProviderSubmitter(outbound.NewSMTPRelay(&config.OutboundSMTPConfig{}), f.gate) messageID, to := f.message(1) @@ -367,7 +386,7 @@ func TestProviderSubmitterAuthorizedTokenIsSingleUse(t *testing.T) { messageID, to := f.message(1) ref := f.prepare(messageID) auth := f.authorize(ref) - env := outbound.Envelope{MessageID: messageID, From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")} + env := outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")} res, err := s.SubmitOnce(f.ctx, auth, env) if err != nil || res.SettlementErr != nil || res.ProviderMessageID == "" { @@ -421,7 +440,7 @@ func TestProviderSubmitterAttemptHeaderDerivesOnlyFromToken(t *testing.T) { ref := f.prepare(messageID) auth := f.authorize(ref) - res, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{MessageID: messageID, From: "agent@agents.e2a.dev", Recipients: to, Message: outbound.SmuggledMIME()}) + res, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: outbound.SmuggledMIME()}) if err != nil || res.SettlementErr != nil { t.Fatalf("submit: res=%+v err=%v", res, err) } @@ -489,7 +508,7 @@ func TestProviderSubmitterRetryRedeemsADistinctAttempt(t *testing.T) { s := outbound.NewProviderSubmitter(relay, f.gate) messageID, to := f.message(1) ref := f.prepare(messageID) - env := outbound.Envelope{MessageID: messageID, From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")} + env := outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")} first := f.authorize(ref) if _, err := s.SubmitOnce(f.ctx, first, env); err != nil { @@ -531,8 +550,9 @@ func TestProviderSubmitterBindsProviderMessageIDOnAcceptance(t *testing.T) { t.Errorf("result attempt = %d, want 1", res.Attempt.Attempt()) } _, bound := f.correlation(ref.ID(), 1) - if bound == nil || *bound != res.ProviderMessageID { - t.Fatalf("correlation provider_message_id = %v, want %q", bound, res.ProviderMessageID) + want := sendingpolicy.NormalizeProviderMessageID(res.ProviderMessageID) + if bound == nil || *bound != want { + t.Fatalf("correlation provider_message_id = %v, want %q (normalized from %q)", bound, want, res.ProviderMessageID) } } @@ -541,7 +561,8 @@ func TestProviderSubmitterBindsProviderMessageIDOnAcceptance(t *testing.T) { // provider id, and surfaces as a permanent error the worker can classify. func TestProviderSubmitterPermanentRejectionIsSettledNotRetried(t *testing.T) { f := newGateFixture(t, nil) - s := outbound.NewProviderSubmitter(rejectingRelay(t), f.gate) + spy := &spyGate{Gate: f.gate} + s := outbound.NewProviderSubmitter(rejectingRelay(t, "550 5.1.1 no such user"), spy) messageID, to := f.message(1) ref := f.prepare(messageID) auth := f.authorize(ref) @@ -556,4 +577,274 @@ func TestProviderSubmitterPermanentRejectionIsSettledNotRetried(t *testing.T) { if _, bound := f.correlation(ref.ID(), 1); bound != nil { t.Fatalf("provider_message_id = %q bound on a rejection", *bound) } + got := spy.settled() + if len(got) != 1 || got[0].Outcome != sendingpolicy.SettlementProviderPermanentlyRejected || got[0].ProviderMessageID != "" { + t.Fatalf("settlements = %+v, want exactly one permanent rejection without a provider id", got) + } +} + +// spyGate records settlements and can be made to fail them, so a test can see +// the one effect of SubmitOnce that has no observable row yet. +type spyGate struct { + sendingpolicy.Gate + mu sync.Mutex + settleErr error + calls []sendingpolicy.ProviderSettlement +} + +func (g *spyGate) SettleProvider(ctx context.Context, s sendingpolicy.ProviderSettlement) error { + g.mu.Lock() + g.calls = append(g.calls, s) + g.mu.Unlock() + if g.settleErr != nil { + return g.settleErr + } + return g.Gate.SettleProvider(ctx, s) +} + +func (g *spyGate) settled() []sendingpolicy.ProviderSettlement { + g.mu.Lock() + defer g.mu.Unlock() + return append([]sendingpolicy.ProviderSettlement(nil), g.calls...) +} + +// TestProviderSubmitterAmbiguousOutcomeIsNotSettled: a 4xx might still be +// delivered on retry, so nothing is settled and the error is not permanent. +func TestProviderSubmitterAmbiguousOutcomeIsNotSettled(t *testing.T) { + f := newGateFixture(t, nil) + spy := &spyGate{Gate: f.gate} + s := outbound.NewProviderSubmitter(rejectingRelay(t, "451 4.3.0 try again later"), spy) + messageID, to := f.message(1) + ref := f.prepare(messageID) + auth := f.authorize(ref) + + _, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")}) + if err == nil || outbound.IsPermanentSMTPError(err) { + t.Fatalf("err = %v, want a non-permanent SMTP error", err) + } + if got := spy.settled(); len(got) != 0 { + t.Fatalf("settlements = %+v, want none for an ambiguous outcome", got) + } + if state := f.callState(ref.ID(), 1); state != "started" { + t.Fatalf("call_state = %s, want started (the socket did open; a retry needs a new ordinal)", state) + } +} + +// TestProviderSubmitterAcceptedButUnsettledIsReportedNotRetried pins the +// at-least-once contract: SES took the message, so the failure to settle is +// carried on the result with a nil error. A caller that resubmitted here would +// send the message twice. +func TestProviderSubmitterAcceptedButUnsettledIsReportedNotRetried(t *testing.T) { + f := newGateFixture(t, nil) + relay, captured := acceptingRelay(t) + spy := &spyGate{Gate: f.gate, settleErr: errors.New("settle: database unavailable")} + s := outbound.NewProviderSubmitter(relay, spy) + messageID, to := f.message(1) + ref := f.prepare(messageID) + auth := f.authorize(ref) + + res, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")}) + if err != nil { + t.Fatalf("err = %v, want nil — the message was accepted", err) + } + if res.SettlementErr == nil || res.ProviderMessageID == "" || res.Attempt.Attempt() != 1 { + t.Fatalf("result = %+v, want provider id, attempt 1, and a settlement error", res) + } + if n := len(captured()); n != 1 { + t.Fatalf("provider received %d messages, want 1", n) + } + got := spy.settled() + if len(got) != 1 || got[0].Outcome != sendingpolicy.SettlementProviderAccepted || got[0].ProviderMessageID != res.ProviderMessageID { + t.Fatalf("settlement attempted = %+v, want one acceptance carrying %q", got, res.ProviderMessageID) + } + // The caller's recovery is to settle again, idempotently — never to resubmit. + if err := f.gate.SettleProvider(f.ctx, got[0]); err != nil { + t.Fatalf("late settlement: %v", err) + } + if _, bound := f.correlation(ref.ID(), 1); bound == nil || *bound != sendingpolicy.NormalizeProviderMessageID(res.ProviderMessageID) { + t.Fatalf("bound = %v, want the normalized provider id", bound) + } +} + +// TestProviderSubmitterSocketCounterObservesADial is the positive control for +// every zero-network assertion above: a redeemed token that reaches the dial +// is seen by the counter exactly once. +func TestProviderSubmitterSocketCounterObservesADial(t *testing.T) { + f := newGateFixture(t, nil) + relay, sockets := countingListener(t) + s := outbound.NewProviderSubmitter(relay, f.gate) + messageID, to := f.message(1) + ref := f.prepare(messageID) + auth := f.authorize(ref) + + if _, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")}); err == nil { + t.Fatal("a dropped connection was reported as success") + } + if sockets() != 1 { + t.Fatalf("sockets = %d, want exactly 1", sockets()) + } + if state := f.callState(ref.ID(), 1); state != "started" { + t.Fatalf("call_state = %s, want started", state) + } +} + +// vanishingRelay fronts a server that takes the whole DATA body and then +// closes without a 250 — the lost-acceptance shape. +func vanishingRelay(t *testing.T) (*outbound.SMTPRelay, func() int) { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = listener.Close() }) + var mu sync.Mutex + bodies := 0 + go func() { + for { + conn, err := listener.Accept() + if err != nil { + return + } + go func(conn net.Conn) { + defer conn.Close() + r := bufio.NewReader(conn) + fmt.Fprint(conn, "220 vanish ready\r\n") + inData := false + for { + line, err := r.ReadString('\n') + if err != nil { + return + } + if inData { + if strings.TrimRight(line, "\r\n") == "." { + mu.Lock() + bodies++ + mu.Unlock() + return // no 250: the connection just dies + } + continue + } + switch upper := strings.ToUpper(strings.TrimSpace(line)); { + case upper == "DATA": + inData = true + fmt.Fprint(conn, "354 Go ahead\r\n") + case upper == "QUIT": + fmt.Fprint(conn, "221 Bye\r\n") + return + default: + fmt.Fprint(conn, "250 OK\r\n") + } + } + }(conn) + } + }() + addr := listener.Addr().(*net.TCPAddr) + return outbound.NewSMTPRelay(&config.OutboundSMTPConfig{Host: addr.IP.String(), Port: addr.Port}), + func() int { mu.Lock(); defer mu.Unlock(); return bodies } +} + +// TestProviderSubmitterPauseBetweenConsumeAndSubmitOpensNoSocket: the abuse +// pause is re-proved at redemption, so a pause that lands after the token was +// minted still stops the send. +func TestProviderSubmitterPauseBetweenConsumeAndSubmitOpensNoSocket(t *testing.T) { + f := newGateFixture(t, nil) + relay, sockets := countingListener(t) + s := outbound.NewProviderSubmitter(relay, f.gate) + messageID, to := f.message(1) + ref := f.prepare(messageID) + auth := f.authorize(ref) + if _, err := f.pool.Exec(f.ctx, `UPDATE account_sending_controls SET state = 'paused' WHERE user_id = $1`, f.userID); err != nil { + t.Fatal(err) + } + + _, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")}) + if !errors.Is(err, sendingpolicy.ErrAuthorizationInvalid) { + t.Fatalf("err = %v, want ErrAuthorizationInvalid", err) + } + if sockets() != 0 { + t.Fatalf("sockets = %d, want 0", sockets()) + } +} + +// TestProviderSubmitterMalformedHeaderSectionOpensNoSocket: a bare CR in the +// header section, or a leading continuation, is refused before redemption. +func TestProviderSubmitterMalformedHeaderSectionOpensNoSocket(t *testing.T) { + f := newGateFixture(t, nil) + relay, sockets := countingListener(t) + s := outbound.NewProviderSubmitter(relay, f.gate) + messageID, to := f.message(1) + ref := f.prepare(messageID) + auth := f.authorize(ref) + + for name, mime := range map[string]string{ + "bare CR hides a header": "Subject: a\rX-SES-TENANT: evil\r\n\r\nbody", + "CR CR LF pseudo-separator": "Subject: s\r\n\r\r\nX-SES-TENANT: evil\r\n\r\nbody", + "leading continuation": " evil-suffix\r\nSubject: s\r\n\r\nbody", + } { + _, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte(mime)}) + if !errors.Is(err, outbound.ErrMalformedHeaderSection) { + t.Errorf("%s: err = %v, want ErrMalformedHeaderSection", name, err) + } + if state := f.callState(ref.ID(), 1); state != "authorized" { + t.Errorf("%s: call_state = %s, want authorized", name, state) + } + } + if sockets() != 0 { + t.Fatalf("sockets = %d, want 0", sockets()) + } +} + +// TestProviderSubmitterSubmitsTheCanonicalEnvelope: the caller's spelling of a +// recipient is validated but never sent; RCPT TO carries the normalized +// address the token was priced for. +func TestProviderSubmitterSubmitsTheCanonicalEnvelope(t *testing.T) { + f := newGateFixture(t, nil) + relay, captured := acceptingRelay(t) + s := outbound.NewProviderSubmitter(relay, f.gate) + messageID, to := f.message(2) + ref := f.prepare(messageID) + auth := f.authorize(ref) + + padded := []string{" " + strings.ToUpper(to[1]) + " ", to[0]} + if _, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: padded, Message: []byte("Subject: x\r\n\r\nbody")}); err != nil { + t.Fatalf("submit: %v", err) + } + msgs := captured() + if len(msgs) != 1 { + t.Fatalf("captured %d messages, want 1", len(msgs)) + } + want := auth.AuthorizedRecipients() + if strings.Join(msgs[0].Recipients, ",") != strings.Join(want, ",") { + t.Fatalf("RCPT TO = %q, want the canonical %q", msgs[0].Recipients, want) + } +} + +// TestProviderSubmitterLostAcceptanceIsUnsettledAndMarked: the body was +// delivered and the 250 never came. Nothing is settled, and the error carries +// the marker so the worker can tell "maybe sent" from "not sent". +func TestProviderSubmitterLostAcceptanceIsUnsettledAndMarked(t *testing.T) { + f := newGateFixture(t, nil) + relay, bodies := vanishingRelay(t) + spy := &spyGate{Gate: f.gate} + s := outbound.NewProviderSubmitter(relay, spy) + messageID, to := f.message(1) + ref := f.prepare(messageID) + auth := f.authorize(ref) + + _, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")}) + if !errors.Is(err, outbound.ErrProviderAcceptanceUnknown) { + t.Fatalf("err = %v, want ErrProviderAcceptanceUnknown", err) + } + if outbound.IsPermanentSMTPError(err) { + t.Fatalf("err = %v classified permanent; a delivered body must never be", err) + } + if bodies() != 1 { + t.Fatalf("provider took %d bodies, want 1", bodies()) + } + if got := spy.settled(); len(got) != 0 { + t.Fatalf("settlements = %+v, want none while acceptance is unknown", got) + } + if state := f.callState(ref.ID(), 1); state != "started" { + t.Fatalf("call_state = %s, want started", state) + } } diff --git a/internal/outbound/smtp_relay.go b/internal/outbound/smtp_relay.go index d35eda183..6b10a2e70 100644 --- a/internal/outbound/smtp_relay.go +++ b/internal/outbound/smtp_relay.go @@ -19,6 +19,15 @@ import ( var smtpRetryBackoffs = []time.Duration{1 * time.Second, 5 * time.Second, 15 * time.Second} +// ErrProviderAcceptanceUnknown marks a failure that happened AFTER the whole +// message body was handed to the provider: the terminating dot was written and +// the 250 never arrived. The provider may have accepted the message. No +// classifier can call this permanent or transient, and a caller that retries +// it as if nothing was sent will deliver the message twice. The sending +// protection seam leaves such an attempt unsettled; delivery feedback carrying +// the attempt header is the only authoritative answer. +var ErrProviderAcceptanceUnknown = errors.New("outbound smtp: message body delivered, provider acceptance unknown") + type SMTPRelay struct { cfg *config.OutboundSMTPConfig } @@ -289,7 +298,7 @@ func (r *SMTPRelay) sendOnceContext(ctx context.Context, envelopeFrom string, re // 250 response is waiting in the buffer. Read it directly. _, msg, err := text.ReadResponse(250) if err != nil { - return "", fmt.Errorf("data final: %w", err) + return "", fmt.Errorf("data final: %w", errors.Join(ErrProviderAcceptanceUnknown, err)) } // Parse Message-ID from response like "Ok " diff --git a/internal/sendingpolicy/gate.go b/internal/sendingpolicy/gate.go index 9301de869..0fd3b1778 100644 --- a/internal/sendingpolicy/gate.go +++ b/internal/sendingpolicy/gate.go @@ -696,8 +696,12 @@ func (m *Module) readAuthState(ctx context.Context, tx pgx.Tx, ref AttemptRef) ( } // A required header with no name to put in it is not a header. The adapter // would have to either omit it or send an empty value, and both defeat the - // isolation the header exists for, so the send waits for a real tenant. - if st.tenantMode == TenantModeRequired && strings.TrimSpace(st.tenantName) == "" { + // isolation the header exists for, so the send waits for a real tenant. A + // name that cannot be a header VALUE — control characters, whitespace — is + // refused here for the same reason: the adapter would fail closed on it + // anyway, but silently and unretryably, whereas a hold with this reason is + // something an operator can see and repair. + if st.tenantMode == TenantModeRequired && !validTenantName(st.tenantName) { return st, holdDecision(ReasonTenantUnnamed, time.Time{}), nil } @@ -758,6 +762,20 @@ func (m *Module) readAuthState(ctx context.Context, tx pgx.Tx, ref AttemptRef) ( return st, allowDecision(), nil } +// validTenantName reports whether a tenant name can travel as an SMTP header +// value: non-empty, printable ASCII, no whitespace. +func validTenantName(name string) bool { + if strings.TrimSpace(name) == "" { + return false + } + for _, r := range name { + if r <= ' ' || r > '~' { + return false + } + } + return true +} + // tenantModeFor resolves the tenant-header mode for one account. // // Canary is an explicit account list rather than a percentage: a tenant header @@ -1397,6 +1415,27 @@ func (m *Module) RedeemProviderCall(ctx context.Context, auth ProviderAuthorizat return m.invalidate(ctx, tx, auth.attempt) } + // Re-prove the abuse pause. ConsumeAttempt linearized it under the + // account-control lock, but that transaction has committed and a pause + // can land in the gap before the socket opens; a pause is the one control + // whose whole point is "stop now". The read is deliberately UNLOCKED: the + // account-control key precedes the operation key in the normative order + // and this transaction already holds the operation, so locking it here + // would invert the order. A committed pause is visible to a plain read, + // and the check can only refuse, never widen. + if op.Purpose.isCustomer() && op.SourceAccountRef != nil { + var state string + err := tx.QueryRow(ctx, + `SELECT state FROM account_sending_controls WHERE user_id = $1`, *op.SourceAccountRef, + ).Scan(&state) + if err != nil && !errors.Is(err, pgx.ErrNoRows) { + return fmt.Errorf("sendingpolicy: read account control: %w", err) + } + if state == "paused" { + return m.invalidate(ctx, tx, auth.attempt) + } + } + stored, err := lockReservation(ctx, tx, auth.attempt.operationID, auth.attempt.attempt) if err != nil { return err @@ -1620,7 +1659,12 @@ func (m *Module) SettleProvider(ctx context.Context, settlement ProviderSettleme // attempt that was authorized to reach the provider. Accepting a reserved // or released attempt would let a caller advance ramp progress for a send // that never happened once Task 4 hangs the ramp mutation off this call. - if stored.State != "confirmed" { + if stored.State != "confirmed" || stored.CallState != "started" { + // `confirmed` says capacity was charged; `started` says the token was + // redeemed and the socket opened. Only the second proves the provider + // could have seen the message, and a provider id bound to an attempt + // that never dialed is a ledger claim about a send that did not + // happen. return ErrAttemptStale } @@ -1646,7 +1690,7 @@ func (m *Module) SettleProvider(ctx context.Context, settlement ProviderSettleme // attribute feedback to any more, and failing a late settlement over it would // only make the caller retry forever. func bindProviderMessageID(ctx context.Context, tx pgx.Tx, settlement ProviderSettlement) error { - id := strings.TrimSpace(settlement.ProviderMessageID) + id := NormalizeProviderMessageID(settlement.ProviderMessageID) if id == "" { return nil } @@ -1668,7 +1712,7 @@ func bindProviderMessageID(ctx context.Context, tx pgx.Tx, settlement ProviderSe return fmt.Errorf("sendingpolicy: lock correlation: %w", err) } if bound != nil { - if *bound == id { + if NormalizeProviderMessageID(*bound) == id { return nil } return ErrProviderMessageIDConflict diff --git a/internal/sendingpolicy/provider_attempt_integration_test.go b/internal/sendingpolicy/provider_attempt_integration_test.go index f3b3c495f..b480cce7a 100644 --- a/internal/sendingpolicy/provider_attempt_integration_test.go +++ b/internal/sendingpolicy/provider_attempt_integration_test.go @@ -899,9 +899,15 @@ func TestSettleProviderValidatesAndIsIdempotent(t *testing.T) { user := f.user("standard") agent := f.agent(user) _, attempt := f.prepareAndReserve(g, agent, 2) - if _, auth, err := g.ConsumeAttempt(f.ctx, attempt); err != nil || auth == nil { + _, auth, err := g.ConsumeAttempt(f.ctx, attempt) + if err != nil || auth == nil { t.Fatalf("authorize: auth=%v err=%v", auth, err) } + // Settlement reports what the provider did, so the attempt must have + // reached the provider: redeem, as the adapter does before it dials. + if err := g.RedeemProviderCall(f.ctx, *auth); err != nil { + t.Fatalf("redeem: %v", err) + } for i := 0; i < 2; i++ { if err := g.SettleProvider(f.ctx, sendingpolicy.ProviderSettlement{ @@ -1527,9 +1533,14 @@ func TestSettlementRejectsAnAttemptThatWasNeverAuthorized(t *testing.T) { } // Authorized: now it settles. - if _, auth, err := g.ConsumeAttempt(f.ctx, attempt); err != nil || auth == nil { + _, auth, err := g.ConsumeAttempt(f.ctx, attempt) + if err != nil || auth == nil { t.Fatalf("authorize: auth=%v err=%v", auth, err) } + // Authorized is not enough: settlement needs the socket to have opened. + if err := g.RedeemProviderCall(f.ctx, *auth); err != nil { + t.Fatalf("redeem: %v", err) + } if err := g.SettleProvider(f.ctx, sendingpolicy.ProviderSettlement{ Attempt: attempt, Outcome: sendingpolicy.SettlementProviderAccepted, }); err != nil { diff --git a/internal/sendingpolicy/provider_token_test.go b/internal/sendingpolicy/provider_token_test.go index a9b06e20d..5fbabb248 100644 --- a/internal/sendingpolicy/provider_token_test.go +++ b/internal/sendingpolicy/provider_token_test.go @@ -125,3 +125,90 @@ func TestProviderTokenAcceptanceWithoutIDStillSettles(t *testing.T) { t.Fatalf("bound = %v, want ses-late", got) } } + +// TestProviderTokenSettlementNormalizesProviderMessageID: the relay reports +// SES's id qualified () and SES's feedback reports it +// bare. Both spellings must be one binding, or the two writers that settle the +// same attempt will refuse each other. +func TestProviderTokenSettlementNormalizesProviderMessageID(t *testing.T) { + f := newFixture(t) + g := f.gate(enforcingPolicy(nil)) + ref, auth := f.redeemed(g, f.agent(f.user("standard"))) + + qualified := sendingpolicy.ProviderSettlement{ + Attempt: auth.Attempt(), Outcome: sendingpolicy.SettlementProviderAccepted, + ProviderMessageID: "<010f0193abcdef00-000000@us-east-2.amazonses.com>", + } + if err := g.SettleProvider(f.ctx, qualified); err != nil { + t.Fatalf("settle qualified: %v", err) + } + if got := f.providerMessageID(ref.ID(), 1); got == nil || *got != "010f0193abcdef00-000000" { + t.Fatalf("bound = %v, want the bare id", got) + } + bare := qualified + bare.ProviderMessageID = "010f0193abcdef00-000000" + if err := g.SettleProvider(f.ctx, bare); err != nil { + t.Fatalf("settle bare after qualified: %v (want idempotent)", err) + } + bracketed := qualified + bracketed.ProviderMessageID = "<010f0193abcdef00-000000>" + if err := g.SettleProvider(f.ctx, bracketed); err != nil { + t.Fatalf("settle bracketed after qualified: %v (want idempotent)", err) + } + other := qualified + other.ProviderMessageID = "<010f0193abcdef00-000001@us-east-2.amazonses.com>" + if err := g.SettleProvider(f.ctx, other); !errors.Is(err, sendingpolicy.ErrProviderMessageIDConflict) { + t.Fatalf("different id err = %v, want ErrProviderMessageIDConflict", err) + } +} + +// TestProviderTokenSettlementRequiresTheSocketToHaveOpened: an attempt that +// was authorized but never redeemed cannot be settled as accepted — nothing +// reached the provider, so there is no provider outcome to record. +func TestProviderTokenSettlementRequiresTheSocketToHaveOpened(t *testing.T) { + f := newFixture(t) + g := f.gate(enforcingPolicy(nil)) + ref, attempt := f.prepareAndReserve(g, f.agent(f.user("standard")), 1) + if _, auth, err := g.ConsumeAttempt(f.ctx, attempt); err != nil || auth == nil { + t.Fatalf("authorize: auth=%v err=%v", auth, err) + } + + err := g.SettleProvider(f.ctx, sendingpolicy.ProviderSettlement{ + Attempt: attempt, Outcome: sendingpolicy.SettlementProviderAccepted, ProviderMessageID: "ses-id-never-sent", + }) + if !errors.Is(err, sendingpolicy.ErrAttemptStale) { + t.Fatalf("settle without redeem err = %v, want ErrAttemptStale", err) + } + if got := f.providerMessageID(ref.ID(), 1); got != nil { + t.Fatalf("bound = %q for an attempt that never dialed", *got) + } + if _, callState := f.reservationState(ref.ID(), 1); callState != "authorized" { + t.Fatalf("call_state = %s, want authorized (untouched)", callState) + } +} + +// TestProviderTokenHoldsOnATenantNameThatCannotBeAHeader: a tenant name with a +// line break would be refused by the adapter, silently and forever. The gate +// holds the send with a visible reason instead. +func TestProviderTokenHoldsOnATenantNameThatCannotBeAHeader(t *testing.T) { + f := newFixture(t) + g := f.gate(enforcingPolicy(func(p *sendingpolicy.RuntimePolicy) { + p.TenantHeaderMode = sendingpolicy.TenantHeaderEnforce + })) + user := f.user("standard") + agent := f.agent(user) + if _, err := f.pool.Exec(f.ctx, ` + INSERT INTO account_sending_controls (user_id, ses_tenant_name, ses_tenant_ready, ses_tenant_ready_at) + VALUES ($1, $2, true, now()) + ON CONFLICT (user_id) DO UPDATE + SET ses_tenant_name = EXCLUDED.ses_tenant_name, ses_tenant_ready = true, ses_tenant_ready_at = now()`, + user, "good\r\nX-SES-CONFIGURATION-SET: attacker-set", + ); err != nil { + t.Fatal(err) + } + _, ref := f.prepareMessage(g, f.message(agent, "own_address", 1)) + d := f.authorize(g, ref) + if d.Allow || d.Reason != sendingpolicy.ReasonTenantUnnamed { + t.Fatalf("decision = %+v, want a hold with reason %q", d, sendingpolicy.ReasonTenantUnnamed) + } +} diff --git a/internal/sendingpolicy/types.go b/internal/sendingpolicy/types.go index 0f7320e59..7e420791b 100644 --- a/internal/sendingpolicy/types.go +++ b/internal/sendingpolicy/types.go @@ -418,6 +418,25 @@ type ProviderSettlement struct { // attempt — the invariant this module exists to hold — and is never absorbed. var ErrProviderMessageIDConflict = errors.New("sendingpolicy: attempt already settled with a different provider message id") +// NormalizeProviderMessageID reduces a provider message id to the bare form the +// provider itself reports in delivery feedback. +// +// The SMTP relay returns SES's id angle-bracketed and qualified with the +// region domain () because that is the on-wire +// Message-ID replies must anchor on; SES's SNS feedback carries the same id +// BARE. The correlation row exists so feedback can find its attempt, and two +// writers — the synchronous worker and the delayed feedback finalizer — must +// agree on one spelling or the second one is refused as a conflict. Every +// write and comparison goes through this function; readers should too. +func NormalizeProviderMessageID(id string) string { + id = strings.TrimSpace(id) + id = strings.TrimSuffix(strings.TrimPrefix(id, "<"), ">") + if at := strings.IndexByte(id, '@'); at >= 0 { + id = id[:at] + } + return strings.TrimSpace(id) +} + // TenantMode is the closed tenant-header state carried by an authorization. // It is resolved under the final account-control lock, so a job that was // enqueued before a tenant flip still submits with the post-flip header. diff --git a/internal/testutil/testdb/db.go b/internal/testutil/testdb/db.go index 03440b766..2f507690f 100644 --- a/internal/testutil/testdb/db.go +++ b/internal/testutil/testdb/db.go @@ -86,7 +86,7 @@ func TestDBURL() string { // prevent, so a base too long to derive from has to fail loudly rather than // quietly reintroduce it. if name := strings.TrimPrefix(u.Path, "/"); len(name) > maxPostgresIdentifier { - panic(fmt.Sprintf("testutil: derived test database name %q is %d bytes, over Postgres's "+ + panic(fmt.Sprintf("testdb: derived test database name %q is %d bytes, over Postgres's "+ "%d-byte identifier limit — Postgres would truncate it silently and collide sibling "+ "packages onto one database. Shorten the base in E2A_TEST_DATABASE_URL; the derived "+ "suffix needs %d bytes.", name, len(name), maxPostgresIdentifier, len(suffix))) From 615aae616dd6f1f755f8833963aabf30585c02ee Mon Sep 17 00:00:00 2001 From: Josh Zhang <39790535+jiashuoz@users.noreply.github.com> Date: Fri, 4 Sep 2026 21:03:14 -0700 Subject: [PATCH 4/4] fix(outbound): keep the acceptance-unknown marker honest MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Re-review of the previous fix commit found the marker leaking in both directions. The relay's deferred context remap replaced the error wholesale, so a deadline or cancellation after the terminating dot — the likeliest way to lose a 250 — dropped ErrProviderAcceptanceUnknown and classified as a connection outage the worker would re-drive. And the marker was joined to every post-DATA failure, including a coded 554 content rejection, which is the provider's definite answer and must classify permanent. Now the marker survives the remap and is attached only when no reply code came back. Also pins the customer-only guard on the redeem-time pause re-check: the notice telling an account it was paused is sourced from that paused account and must still go out. Reviewer-authored test adopted. And the compare-side normalization of the provider id now has a test that writes a qualified spelling directly and replays bare. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01AjfGxvXW6fNKWGFHuo68yX --- internal/outbound/provider_submit_test.go | 79 ++++++++++++++++++- internal/outbound/smtp_relay.go | 35 +++++--- internal/sendingpolicy/gate.go | 11 ++- internal/sendingpolicy/provider_token_test.go | 57 +++++++++++++ 4 files changed, 169 insertions(+), 13 deletions(-) diff --git a/internal/outbound/provider_submit_test.go b/internal/outbound/provider_submit_test.go index 004d4b89d..c33a3e834 100644 --- a/internal/outbound/provider_submit_test.go +++ b/internal/outbound/provider_submit_test.go @@ -691,12 +691,20 @@ func TestProviderSubmitterSocketCounterObservesADial(t *testing.T) { // vanishingRelay fronts a server that takes the whole DATA body and then // closes without a 250 — the lost-acceptance shape. func vanishingRelay(t *testing.T) (*outbound.SMTPRelay, func() int) { + return afterDotRelay(t, "") +} + +// afterDotRelay fronts a server that takes the whole DATA body and then does +// `then`: "" closes silently, "stall" never answers, anything else is sent as +// the final reply line. +func afterDotRelay(t *testing.T, then string) (*outbound.SMTPRelay, func() int) { t.Helper() listener, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { t.Fatal(err) } - t.Cleanup(func() { _ = listener.Close() }) + stop := make(chan struct{}) + t.Cleanup(func() { close(stop); _ = listener.Close() }) var mu sync.Mutex bodies := 0 go func() { @@ -720,7 +728,17 @@ func vanishingRelay(t *testing.T) (*outbound.SMTPRelay, func() int) { mu.Lock() bodies++ mu.Unlock() - return // no 250: the connection just dies + switch then { + case "": + return // no 250: the connection just dies + case "stall": + <-stop // hold the connection open, silently + return + default: + fmt.Fprint(conn, then+"\r\n") + inData = false + continue + } } continue } @@ -848,3 +866,60 @@ func TestProviderSubmitterLostAcceptanceIsUnsettledAndMarked(t *testing.T) { t.Fatalf("call_state = %s, want started", state) } } + +// TestProviderSubmitterLostAcceptanceSurvivesTheDeadline: the likeliest way +// to lose a 250 is the caller's deadline. The marker must survive the relay's +// context remap, or the worker cannot tell "maybe sent" from "not sent". +func TestProviderSubmitterLostAcceptanceSurvivesTheDeadline(t *testing.T) { + f := newGateFixture(t, nil) + relay, bodies := afterDotRelay(t, "stall") + spy := &spyGate{Gate: f.gate} + s := outbound.NewProviderSubmitter(relay, spy) + messageID, to := f.message(1) + ref := f.prepare(messageID) + auth := f.authorize(ref) + + ctx, cancel := context.WithTimeout(f.ctx, 400*time.Millisecond) + defer cancel() + _, err := s.SubmitOnce(ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")}) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("err = %v, want the deadline", err) + } + if !errors.Is(err, outbound.ErrProviderAcceptanceUnknown) { + t.Fatalf("err = %v, want ErrProviderAcceptanceUnknown preserved through the remap", err) + } + if bodies() != 1 { + t.Fatalf("provider took %d bodies, want 1", bodies()) + } + if got := spy.settled(); len(got) != 0 { + t.Fatalf("settlements = %+v, want none", got) + } +} + +// TestProviderSubmitterPostDataRejectionIsDefinite: SES answers a content +// rejection AFTER the body with an ordinary 554. That is a definite answer — +// settled as rejected, classified permanent, and never marked ambiguous. +func TestProviderSubmitterPostDataRejectionIsDefinite(t *testing.T) { + f := newGateFixture(t, nil) + relay, bodies := afterDotRelay(t, "554 5.6.0 Message rejected") + spy := &spyGate{Gate: f.gate} + s := outbound.NewProviderSubmitter(relay, spy) + messageID, to := f.message(1) + ref := f.prepare(messageID) + auth := f.authorize(ref) + + _, err := s.SubmitOnce(f.ctx, auth, outbound.Envelope{From: "agent@agents.e2a.dev", Recipients: to, Message: []byte("Subject: x\r\n\r\nbody")}) + if err == nil || !outbound.IsPermanentSMTPError(err) { + t.Fatalf("err = %v, want permanent", err) + } + if errors.Is(err, outbound.ErrProviderAcceptanceUnknown) { + t.Fatalf("err = %v carries the ambiguity marker on a definite reply", err) + } + if bodies() != 1 { + t.Fatalf("provider took %d bodies, want 1", bodies()) + } + got := spy.settled() + if len(got) != 1 || got[0].Outcome != sendingpolicy.SettlementProviderPermanentlyRejected { + t.Fatalf("settlements = %+v, want one permanent rejection", got) + } +} diff --git a/internal/outbound/smtp_relay.go b/internal/outbound/smtp_relay.go index 6b10a2e70..b67753423 100644 --- a/internal/outbound/smtp_relay.go +++ b/internal/outbound/smtp_relay.go @@ -187,16 +187,26 @@ func (r *SMTPRelay) sendOnceContext(ctx context.Context, envelopeFrom string, re if err == nil { return } - if ctx.Err() != nil { + // Whether the body had already been handed over must survive the + // remaps below: a cancellation or deadline is the LIKELIEST way to + // lose the 250, and the caller's contract for that shape is "maybe + // sent", not "not sent". + unknown := errors.Is(err, ErrProviderAcceptanceUnknown) + switch { + case ctx.Err() != nil: err = ctx.Err() - return + default: + // The conn deadline is set to the ctx deadline below, so the net + // poller's timer races the context's own timer to the same + // instant. When the poller wins, the I/O error surfaces while + // ctx.Err() is still nil — map it to the deadline error the + // caller contracted for. + if d, ok := ctx.Deadline(); ok && !time.Now().Before(d) && errors.Is(err, os.ErrDeadlineExceeded) { + err = context.DeadlineExceeded + } } - // The conn deadline is set to the ctx deadline below, so the net - // poller's timer races the context's own timer to the same instant. - // When the poller wins, the I/O error surfaces while ctx.Err() is - // still nil — map it to the deadline error the caller contracted for. - if d, ok := ctx.Deadline(); ok && !time.Now().Before(d) && errors.Is(err, os.ErrDeadlineExceeded) { - err = context.DeadlineExceeded + if unknown && !errors.Is(err, ErrProviderAcceptanceUnknown) { + err = errors.Join(ErrProviderAcceptanceUnknown, err) } }() @@ -298,7 +308,14 @@ func (r *SMTPRelay) sendOnceContext(ctx context.Context, envelopeFrom string, re // 250 response is waiting in the buffer. Read it directly. _, msg, err := text.ReadResponse(250) if err != nil { - return "", fmt.Errorf("data final: %w", errors.Join(ErrProviderAcceptanceUnknown, err)) + // A coded reply here is the provider's definite answer to the whole + // message (SES's post-DATA content rejection is an ordinary 554) and + // classifies like any other. Only a reply that never came is + // ambiguous, and only that carries the marker. + if _, coded := smtpCode(err); !coded { + err = errors.Join(ErrProviderAcceptanceUnknown, err) + } + return "", fmt.Errorf("data final: %w", err) } // Parse Message-ID from response like "Ok " diff --git a/internal/sendingpolicy/gate.go b/internal/sendingpolicy/gate.go index 0fd3b1778..0d482b656 100644 --- a/internal/sendingpolicy/gate.go +++ b/internal/sendingpolicy/gate.go @@ -1421,8 +1421,15 @@ func (m *Module) RedeemProviderCall(ctx context.Context, auth ProviderAuthorizat // whose whole point is "stop now". The read is deliberately UNLOCKED: the // account-control key precedes the operation key in the normative order // and this transaction already holds the operation, so locking it here - // would invert the order. A committed pause is visible to a plain read, - // and the check can only refuse, never widen. + // would invert the order. A committed pause is visible to a plain read + // because the transaction runs READ COMMITTED (each statement sees a + // fresh snapshot; under REPEATABLE READ this re-check would silently + // read the snapshot taken at effectivePolicy and prove nothing), and the + // check can only refuse, never widen. The window that remains runs from + // this read to the dial — the reservation update, commit, and return. + // + // Protection notices are exempt: the notice telling an account it was + // paused is SOURCED from that paused account, and it must go out. if op.Purpose.isCustomer() && op.SourceAccountRef != nil { var state string err := tx.QueryRow(ctx, diff --git a/internal/sendingpolicy/provider_token_test.go b/internal/sendingpolicy/provider_token_test.go index 5fbabb248..1e3d2d696 100644 --- a/internal/sendingpolicy/provider_token_test.go +++ b/internal/sendingpolicy/provider_token_test.go @@ -4,6 +4,7 @@ import ( "errors" "testing" + "github.com/jackc/pgx/v5" "github.com/tokencanopy/e2a/internal/sendingpolicy" ) @@ -212,3 +213,59 @@ func TestProviderTokenHoldsOnATenantNameThatCannotBeAHeader(t *testing.T) { t.Fatalf("decision = %+v, want a hold with reason %q", d, sendingpolicy.ReasonTenantUnnamed) } } + +// TestProviderTokenPauseNoticeToAPausedOwnerStillRedeems pins the customer-only +// guard on RedeemProviderCall's pause re-check: the notice telling an account +// it was paused is SOURCED from that paused account, so an unguarded re-read +// would refuse the one email the pause exists to send. +func TestProviderTokenPauseNoticeToAPausedOwnerStillRedeems(t *testing.T) { + f := newFixture(t) + g := f.gate(enforcingPolicy(nil)) + user := f.user("standard") + f.pause(user) + eventID := f.pauseNotice(user) + + var ref sendingpolicy.OperationRef + f.inTx(func(tx pgx.Tx) error { + var err error + ref, err = g.PrepareProtectionNoticeTx(f.ctx, tx, + sendingpolicy.NewProtectionNoticeRef(eventID, sendingpolicy.AudienceOwner)) + return err + }) + _, attempt, err := g.Reserve(f.ctx, ref) + if err != nil { + t.Fatalf("reserve: %v", err) + } + decision, auth, err := g.ConsumeAttempt(f.ctx, attempt) + if err != nil || auth == nil { + t.Fatalf("consume: decision=%+v auth=%v err=%v", decision, auth, err) + } + if err := g.RedeemProviderCall(f.ctx, *auth); err != nil { + t.Fatalf("redeem of a pause notice to a PAUSED owner: %v — the notice must still go out", err) + } +} + +// TestProviderTokenSettlementComparesNormalizedProviderMessageID: a row bound +// by another writer in the qualified spelling must not refuse a bare replay. +func TestProviderTokenSettlementComparesNormalizedProviderMessageID(t *testing.T) { + f := newFixture(t) + g := f.gate(enforcingPolicy(nil)) + ref, auth := f.redeemed(g, f.agent(f.user("standard"))) + if _, err := f.pool.Exec(f.ctx, ` + UPDATE sending_feedback_correlations SET provider_message_id = $3 + WHERE operation_id = $1 AND submission_attempt = $2`, + ref.ID(), 1, "<010f0193abcdef00-000000@us-east-2.amazonses.com>", + ); err != nil { + t.Fatal(err) + } + if err := g.SettleProvider(f.ctx, sendingpolicy.ProviderSettlement{ + Attempt: auth.Attempt(), Outcome: sendingpolicy.SettlementProviderAccepted, ProviderMessageID: "010f0193abcdef00-000000", + }); err != nil { + t.Fatalf("bare replay over a qualified binding: %v", err) + } + if err := g.SettleProvider(f.ctx, sendingpolicy.ProviderSettlement{ + Attempt: auth.Attempt(), Outcome: sendingpolicy.SettlementProviderAccepted, ProviderMessageID: "other-000000", + }); !errors.Is(err, sendingpolicy.ErrProviderMessageIDConflict) { + t.Fatalf("different id err = %v, want ErrProviderMessageIDConflict", err) + } +}