diff --git a/docs/caller-admission-2026-10-08.md b/docs/caller-admission-2026-10-08.md new file mode 100644 index 0000000..aff7899 --- /dev/null +++ b/docs/caller-admission-2026-10-08.md @@ -0,0 +1,75 @@ +# Scoped transactional callers — source review, 2026-10-08 + +The receiver can isolate explicitly admitted callers while retaining the existing +single-caller startup path until deployment admission. No live credentials, +mailboxes, sender, chart, image, ledger or timer changed in this source increment. + +Set EMAIL_CONNECT_CALLERS_FILE to an owner-controlled read-only JSON policy file. +If present, invalid policy fails startup before ledger migration; there is no +fallback to EMAIL_CONNECT_INGEST_TOKEN. Schema version1 has a callers list (1–32). +Each entry needs caller_id, source, absolute token_file, bindings (exact +[tenant, client] pairs), operations, templates, rate_per_minute, burst and +max_payload_bytes. Credentials are independently delivered files; inline tokens, +shared tokens and duplicate caller IDs are rejected. Tokens never enter policy +receipts or repr output. Limits are positive integers: rate <=6000/minute, +burst <=600 requests, payload <=128KiB. These are implementation ceilings, +not approved runtime quotas. + +Operations: send, delivery-status, status, verify, consume, cancel, suppress, +unsuppress. Send templates are only family_member.invited and +family_invitation.resent. Payload source must match the authenticated policy; +verification and invitation tenant/client pairs must be explicitly admitted. +Each caller's invitation IDs, verification reservations and handles are isolated. +An event's immutable payload cannot change on retry. Status exposes a caller-local +opaque reference and transport evidence, never raw provider diagnostics. +Confirmed temporary non-acceptance can retry; permanent rejection is terminal; +uncertain acceptance remains terminal for automatic retry across restart. +Historical failed invitations lacking retry evidence also remain terminal. + +A persisted atomic SQLite token bucket counts authorized requests, including +status, validation failures and duplicates. Concurrent connections and restarts +share its allowance. Changing caller identity resets its budget; admission must +treat that as a quota/security change. Backward wall-clock movement grants no +extra tokens. Oversized requests are rejected before reading their bodies. +Public health only proves receiver liveness; authenticated status is separately +admitted and consumes allowance. + +Suppression is global recipient protection. suppress/unsuppress are privileged +platform operations, to admit only for a named operations owner, never generic +application callers. They do not inspect mailboxes or automatically classify +bounces. Sender identity/TLS remain governed by the existing SMTP provider; +caller policy does not grant arbitrary subjects/bodies/From or a bulk-mail API. + +## Existing original caller and rollback + +Only caller_id/source user-engine may explicitly adopt legacy verification rows +(adopt_legacy_verifications=true). Its granted binding pairs still constrain +consume/cancel. legacy_invitation_binding may explicitly admit one binding for +old invitation payloads lacking tenant/client, and preserves old event keys and +accepted receipts. Reserved ec2_ input IDs cannot collide with scoped storage IDs. +There is no inferred legacy adoption. Removing or narrowing a policy immediately +denies its operations without deleting transport/idempotency evidence. + +Startup adds caller_id to verification rows, error/retryable to send_attempts, +and caller_buckets. Capture fingerprints include all six tables; old five-table +captures remain readable. Rehearsal source preserves scoped accepted/uncertain +verification reservations and budgets without provider calls. Helpers remain +pinned to the current admitted image: repin capture and rehearsal together to +the candidate's immutable image before new-schema capture or scoped activation. +Do not downgrade after new-schema rows exist without an admitted compatible +recovery plan; old receiver code can lose retry restrictions or fail inserts. + +## Validation and outstanding admission + +Synthetic caller/legacy tests:66 passed, zero skips, network-denied app gate. +Full owner suite including loopback GreenMail and scoped ledger rehearsal: +148 passed, zero skips. Receiver/caller module and caller-test Ruff checks pass. +No live recipient, identity, token, handle or message body is a fixture. + +EMAIL-WP-0008-T03 retains admission: choose actual caller/binding/template and +operation sets, quotas, separate token custody/delivery, original-user-engine +transition, platform sender/reply/bounce responsibilities, exact image/schema +rollout and rollback, and independent ledger offsite recovery. Existing live +EMAIL-WP-0007-T03/T04 and RPF-WP-0053-T05/T06 retain archive admission and atomic +credential-pair custody. CSOC-WP-0010-T04/T05/T06 retain productive flow and +owner-spanning browser acceptance. Source checks alone close none of these gates. diff --git a/scripts/ledger_archive.py b/scripts/ledger_archive.py index 432b486..e1a43b8 100644 --- a/scripts/ledger_archive.py +++ b/scripts/ledger_archive.py @@ -72,7 +72,7 @@ def fingerprint(path): "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'" ) } - if tables != TABLES: + if tables not in (TABLES, TABLES | {"caller_buckets"}): raise Refused("unexpected_ledger_schema") schema = list( database.execute("SELECT type,name,tbl_name,sql FROM sqlite_master ORDER BY type,name") @@ -82,7 +82,7 @@ def fingerprint(path): "user_version": database.execute("PRAGMA user_version").fetchone()[0], "tables": {}, } - for name in sorted(TABLES): + for name in sorted(tables): rows = [ json.dumps( row, separators=(",", ":"), default=lambda value: {"bytes_hex": value.hex()} diff --git a/scripts/rehearse-ledger.py b/scripts/rehearse-ledger.py index 34d6542..5a13769 100644 --- a/scripts/rehearse-ledger.py +++ b/scripts/rehearse-ledger.py @@ -41,10 +41,13 @@ class ForbiddenProvider: app=TransactionalApplication(store,ForbiddenProvider(),'synthetic-recovery-token','https://issuer.example.test') checks={'accepted_verifications':0,'unknown_verifications':0,'invalidated_handles':0, 'suppressions':0,'accepted_deliveries':0,'unknown_sends':0} -query=("SELECT v.registration_id,v.email,v.username,v.client_id,v.tenant,v.display_name,a.state " +caller_column="v.caller_id" if "caller_id" in {r[1] for r in store.db.execute("PRAGMA table_info(verifications)")} else "''" +query=("SELECT v.registration_id,v.email,v.username,v.client_id,v.tenant,v.display_name,a.state,"+caller_column+" " "FROM verifications v JOIN verification_attempts a ON a.request_id=v.request_id " "WHERE a.state IN ('accepted','in_flight')") for row in store.db.execute(query).fetchall(): + app=TransactionalApplication(store,ForbiddenProvider(),'synthetic-recovery-token', + 'https://issuer.example.test',**({'verification_namespace':row[7]} if row[7] else {})) payload=dict(zip(['registration_id','normalized_email','preferred_username','client_id','tenant','display_name'],row[:6])) payload['correlation_id']='synthetic-recovery-probe' raw=json.dumps(payload).encode();response={} diff --git a/src/email_connect/callers.py b/src/email_connect/callers.py new file mode 100644 index 0000000..e49b100 --- /dev/null +++ b/src/email_connect/callers.py @@ -0,0 +1,368 @@ +"""Explicitly admitted caller policies for fixed transactional templates. + +No implicit shared-bearer fallback exists in this mode. The old receiver remains +the default until the owning deployment selects an admitted policy file. +""" + +from __future__ import annotations + +import hashlib +import hmac +import io +import json +import math +import re +import secrets +import time +from dataclasses import dataclass, field +from http import HTTPStatus +from pathlib import Path + +IDENTIFIER = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}\Z") +OPERATIONS = { + "/v1/send": "send", + "/v1/delivery-status": "delivery-status", + "/v1/status": "status", + "/v1/registration-verifications": "verify", + "/v1/registration-verifications/consume": "consume", + "/v1/registration-verifications/cancel": "cancel", + "/v1/suppressions": "suppress", + "/v1/suppressions/remove": "unsuppress", +} +TEMPLATES = frozenset({"family_member.invited", "family_invitation.resent"}) +ERRORS = frozenset( + { + "template_not_allowed", + "invalid_recipient", + "recipient_suppressed", + "invitation_id_required", + "idempotency_key_mismatch", + "idempotency_payload_mismatch", + "invalid_verification_request", + "verification_invalid", + "verification_expired", + "delivery_outcome_unknown", + "provider_timeout", + "provider_unavailable", + "temporary_deferral", + "permanent_rejection", + "invalid_size", + } +) + + +@dataclass(frozen=True) +class CallerPolicy: + caller_id: str + token: str = field(repr=False) + source: str + bindings: frozenset[tuple[str, str]] + operations: frozenset[str] + templates: frozenset[str] + rate_per_minute: int + burst: int + max_payload_bytes: int + adopt_legacy_verifications: bool = False + legacy_invitation_binding: tuple[str, str] | None = None + + def __post_init__(self): + if not isinstance(self.caller_id, str) or not IDENTIFIER.fullmatch(self.caller_id): + raise ValueError("invalid_caller_policy") + if not isinstance(self.source, str) or not IDENTIFIER.fullmatch(self.source): + raise ValueError("invalid_caller_policy") + if not isinstance(self.token, str) or not re.fullmatch(r"[!-~]{16,512}", self.token): + raise ValueError("invalid_caller_credential") + if ( + not self.operations + or not self.operations <= set(OPERATIONS.values()) + or not self.templates <= TEMPLATES + or not self.bindings + ): + raise ValueError("invalid_caller_policy") + for binding in self.bindings: + if len(binding) != 2 or any( + not isinstance(v, str) or not IDENTIFIER.fullmatch(v) for v in binding + ): + raise ValueError("invalid_caller_binding") + for value, maximum in ( + (self.rate_per_minute, 6000), + (self.burst, 600), + (self.max_payload_bytes, 128 * 1024), + ): + if type(value) is not int or not 0 < value <= maximum: + raise ValueError("invalid_caller_limits") + if type(self.adopt_legacy_verifications) is not bool: + raise ValueError("invalid_legacy_admission") + if self.adopt_legacy_verifications or self.legacy_invitation_binding is not None: + if self.caller_id != "user-engine" or self.source != "user-engine": + raise ValueError("legacy_admission_requires_original_caller") + if ( + self.legacy_invitation_binding is not None + and self.legacy_invitation_binding not in self.bindings + ): + raise ValueError("invalid_legacy_binding") + + +def load_callers(path): + """Owner-controlled non-secret policy plus independently delivered token files.""" + try: + data = json.loads(Path(path).read_text(encoding="utf-8")) + if data["schema_version"] != 1 or not isinstance(data["callers"], list): + raise ValueError + policies = [] + fields = set(CallerPolicy.__dataclass_fields__) - {"token"} + for entry in data["callers"]: + if set(entry) - (fields | {"token_file"}) or "token" in entry: + raise ValueError + values = dict(entry) + token_path = Path(values.pop("token_file")) + if not token_path.is_absolute(): + raise ValueError + # Kubernetes projections legitimately use symlinks. The projection + # and policy mounts are owner-controlled, read-only deployment inputs. + with token_path.open("rb") as stream: + raw_token = stream.read(515) + if len(raw_token) > 514: + raise ValueError + token = raw_token.decode("ascii").strip() + values["token"] = token + values["bindings"] = frozenset(tuple(v) for v in values["bindings"]) + for key in ("operations", "templates"): + values[key] = frozenset(values[key]) + if values.get("legacy_invitation_binding") is not None: + values["legacy_invitation_binding"] = tuple(values["legacy_invitation_binding"]) + policies.append(CallerPolicy(**values)) + validate_callers(policies) + return policies + except (OSError, UnicodeError, ValueError, TypeError, KeyError): + raise ValueError("caller_configuration_invalid") from None + + +def validate_callers(policies): + if not policies or len(policies) > 32: + raise ValueError("invalid_caller_count") + if len({p.caller_id for p in policies}) != len(policies): + raise ValueError("duplicate_caller_identity") + if len({p.token for p in policies}) != len(policies): + raise ValueError("shared_caller_credential") + + +class ScopedTransactionalApplication: + def __init__(self, store, provider, portal_url, policies, *, clock=time.time): + from .transactional import TransactionalApplication + + validate_callers(policies) + self.store, self.policies, self.clock = store, tuple(policies), clock + self.internal_token = secrets.token_urlsafe(32) + self.apps = { + p.caller_id: TransactionalApplication( + store, + provider, + self.internal_token, + portal_url, + verification_namespace="" if p.adopt_legacy_verifications else p.caller_id, + verification_bindings=p.bindings, + ) + for p in policies + } + + @staticmethod + def _json(start_response, status, payload, *, retry_after=None): + body = json.dumps(payload).encode() + headers = [ + ("Content-Type", "application/json"), + ("Content-Length", str(len(body))), + ("Cache-Control", "private, no-store"), + ] + if retry_after is not None: + headers.append(("Retry-After", str(retry_after))) + start_response(f"{status.value} {status.phrase}", headers) + return [body] + + @staticmethod + def _event_key(policy, event_id): + if ( + not isinstance(event_id, str) + or not IDENTIFIER.fullmatch(event_id) + or event_id.startswith("ec2_") + ): + raise ValueError("invalid_event_id") + if policy.legacy_invitation_binding is not None: + return event_id + return ( + "ec2_" + hashlib.sha256(json.dumps([policy.caller_id, event_id]).encode()).hexdigest() + ) + + @staticmethod + def _reference(policy, event_id): + return ( + "dvr_" + + hashlib.sha256(json.dumps([policy.caller_id, event_id]).encode()).hexdigest()[:32] + ) + + def __call__(self, environ, start_response): + path = environ.get("PATH_INFO") + if path in {"/healthz", "/readyz"} and environ.get("REQUEST_METHOD") in {"GET", "HEAD"}: + return self._json(start_response, HTTPStatus.OK, {"status": "ok"}) + if path not in OPERATIONS or environ.get("REQUEST_METHOD") != "POST": + return self._json(start_response, HTTPStatus.NOT_FOUND, {"error": "not_found"}) + authorization = str(environ.get("HTTP_AUTHORIZATION", "")) + if len(authorization) > 1024: + return self._json(start_response, HTTPStatus.UNAUTHORIZED, {"error": "unauthorized"}) + try: + matches = [ + p + for p in self.policies + if hmac.compare_digest( + authorization.encode("utf-8"), + f"Bearer {p.token}".encode(), + ) + ] + except UnicodeError: + matches = [] + if len(matches) != 1: + return self._json(start_response, HTTPStatus.UNAUTHORIZED, {"error": "unauthorized"}) + policy = matches[0] + if OPERATIONS[path] not in policy.operations: + return self._json( + start_response, HTTPStatus.FORBIDDEN, {"error": "operation_not_allowed"} + ) + try: + now = self.clock() + if not math.isfinite(now): + raise RuntimeError + allowed, wait = self.store.request_budget( + policy.caller_id, + rate=policy.rate_per_minute, + burst=policy.burst, + now=now, + ) + if not allowed: + return self._json( + start_response, + HTTPStatus.TOO_MANY_REQUESTS, + {"error": "rate_limited", "retryable": True}, + retry_after=wait, + ) + length = int(environ.get("CONTENT_LENGTH") or 0) + if not 0 < length <= policy.max_payload_bytes: + return self._json( + start_response, HTTPStatus.REQUEST_ENTITY_TOO_LARGE, {"error": "invalid_size"} + ) + raw = environ["wsgi.input"].read(length) + if len(raw) != length: + raise ValueError("invalid_request") + payload = json.loads(raw) + if not isinstance(payload, dict): + raise ValueError("invalid_request") + return self._dispatch(policy, path, payload, environ, start_response) + except ValueError as exc: + code = str(exc) + permitted = ERRORS | { + "invalid_request", + "invalid_event_id", + "caller_scope_denied", + "invalid_suppression_request", + } + return self._json( + start_response, + HTTPStatus.BAD_REQUEST, + {"error": code if code in permitted else "invalid_request"}, + ) + except Exception: + return self._json( + start_response, + HTTPStatus.SERVICE_UNAVAILABLE, + {"error": "operation_unconfirmed", "retryable": False}, + ) + + def _dispatch(self, policy, path, payload, environ, start_response): + from .transactional import _valid_address + + external_id = None + if path == "/v1/delivery-status": + external_id = payload.get("event_id") + status = self.store.delivery_status(self._event_key(policy, external_id)) + result = {"event_id": external_id, "state": status["state"]} + if status["state"] == "provider_accepted": + result["reference"] = self._reference(policy, external_id) + result["evidence_ceiling"] = "provider_accepted" + return self._json(start_response, HTTPStatus.OK, result) + if path in {"/v1/suppressions", "/v1/suppressions/remove"}: + recipient = payload.get("recipient") + if ( + not isinstance(recipient, str) + or not _valid_address(recipient) + or len(recipient) > 254 + ): + raise ValueError("invalid_suppression_request") + if path.endswith("/remove"): + self.store.unsuppress(recipient) + else: + reason = payload.get("reason", "manual") + if reason not in {"manual", "hard_bounce", "complaint", "unsubscribe"}: + raise ValueError("invalid_suppression_request") + with self.store.lock: + self.store.suppress(recipient, reason) + return self._json(start_response, HTTPStatus.OK, {"status": "updated"}) + if path == "/v1/send": + external_id = payload.get("id") + event_key = self._event_key(policy, external_id) + if environ.get("HTTP_IDEMPOTENCY_KEY") != external_id: + raise ValueError("idempotency_key_mismatch") + if ( + payload.get("source") != policy.source + or not isinstance(payload.get("type"), str) + or payload["type"] not in policy.templates + ): + raise ValueError("template_not_allowed") + binding = (payload.get("tenant"), payload.get("client_id")) + if binding == (None, None) and policy.legacy_invitation_binding is not None: + binding = policy.legacy_invitation_binding + if not all(isinstance(v, str) for v in binding) or binding not in policy.bindings: + raise ValueError("caller_scope_denied") + data = payload.get("data") + if ( + not isinstance(data, dict) + or not isinstance(data.get("invitation_id"), str) + or not IDENTIFIER.fullmatch(data["invitation_id"]) + ): + raise ValueError("invalid_request") + recipient = data.get("primary_email") + if not isinstance(recipient, str) or len(recipient) > 254: + raise ValueError("invalid_recipient") + payload = dict(payload, id=event_key, source="user-engine") + if path == "/v1/registration-verifications": + binding = (payload.get("tenant"), payload.get("client_id")) + if not all(isinstance(v, str) for v in binding) or binding not in policy.bindings: + raise ValueError("caller_scope_denied") + raw = json.dumps(payload).encode() + internal = dict(environ, CONTENT_LENGTH=str(len(raw))) + internal["wsgi.input"] = io.BytesIO(raw) + internal["HTTP_AUTHORIZATION"] = f"Bearer {self.internal_token}" + if external_id is not None: + internal["HTTP_IDEMPOTENCY_KEY"] = self._event_key(policy, external_id) + response = {} + body = json.loads( + b"".join( + self.apps[policy.caller_id]( + internal, + lambda status, headers: response.update(status=int(status.split()[0])), + ) + ) + ) + status = HTTPStatus(response["status"]) + if "error" in body: + error = body["error"] if body["error"] in ERRORS else "invalid_request" + body = { + "error": error, + **({"retryable": body["retryable"]} if "retryable" in body else {}), + } + if external_id is not None and status in {HTTPStatus.OK, HTTPStatus.ACCEPTED}: + body = { + "event_id": external_id, + "status": body["status"], + "reference": self._reference(policy, external_id), + "evidence_ceiling": "provider_accepted", + } + return self._json(start_response, status, body) diff --git a/src/email_connect/transactional.py b/src/email_connect/transactional.py index e3103c6..d4600f8 100644 --- a/src/email_connect/transactional.py +++ b/src/email_connect/transactional.py @@ -2,17 +2,18 @@ from __future__ import annotations -import hmac import hashlib +import hmac import ipaddress import json +import math import os +import secrets import smtplib +import sqlite3 import ssl import time -import sqlite3 -import secrets -from datetime import datetime, timedelta, timezone +from datetime import UTC, datetime, timedelta from email.message import EmailMessage from email.utils import make_msgid, parseaddr from http import HTTPStatus @@ -36,42 +37,85 @@ class SQLiteDeliveryStore: self.db = sqlite3.connect(path, check_same_thread=False) self.lock = RLock() self.db.execute( - "CREATE TABLE IF NOT EXISTS deliveries (event_id TEXT PRIMARY KEY, provider_ref TEXT NOT NULL)" + "CREATE TABLE IF NOT EXISTS deliveries (event_id TEXT PRIMARY KEY, " + "provider_ref TEXT NOT NULL)" ) self.db.execute("""CREATE TABLE IF NOT EXISTS verifications ( request_id TEXT PRIMARY KEY, handle_hash TEXT UNIQUE NOT NULL, registration_id TEXT NOT NULL, email TEXT NOT NULL, username TEXT NOT NULL, client_id TEXT NOT NULL, tenant TEXT NOT NULL, display_name TEXT, - expires_at TEXT NOT NULL, consumed_at TEXT, canceled_at TEXT + expires_at TEXT NOT NULL, consumed_at TEXT, canceled_at TEXT, + caller_id TEXT NOT NULL DEFAULT '' )""") self.db.execute("""CREATE TABLE IF NOT EXISTS suppressions ( email TEXT PRIMARY KEY NOT NULL, reason TEXT NOT NULL, created_at TEXT NOT NULL )""") - columns = { - row[1] for row in self.db.execute("PRAGMA table_info(verifications)") - } + columns = {row[1] for row in self.db.execute("PRAGMA table_info(verifications)")} if "canceled_at" not in columns: self.db.execute("ALTER TABLE verifications ADD COLUMN canceled_at TEXT") + if "caller_id" not in columns: + self.db.execute( + "ALTER TABLE verifications ADD COLUMN caller_id TEXT NOT NULL DEFAULT ''" + ) self.db.execute( - "CREATE TABLE IF NOT EXISTS send_attempts (event_id TEXT PRIMARY KEY, fingerprint TEXT NOT NULL, state TEXT NOT NULL)" + "CREATE TABLE IF NOT EXISTS send_attempts (event_id TEXT PRIMARY KEY, " + "fingerprint TEXT NOT NULL, state TEXT NOT NULL)" ) + send_columns = {row[1] for row in self.db.execute("PRAGMA table_info(send_attempts)")} + for name, kind in (("error", "TEXT"), ("retryable", "INTEGER")): + if name not in send_columns: + self.db.execute(f"ALTER TABLE send_attempts ADD COLUMN {name} {kind}") self.db.execute("""CREATE TABLE IF NOT EXISTS verification_attempts ( operation_key TEXT PRIMARY KEY, fingerprint TEXT NOT NULL, request_id TEXT NOT NULL, state TEXT NOT NULL, error TEXT, retryable INTEGER, provider_ref TEXT )""") + self.db.execute("""CREATE TABLE IF NOT EXISTS caller_buckets ( + caller_id TEXT PRIMARY KEY, tokens REAL NOT NULL, updated_at REAL NOT NULL + )""") self.db.commit() - def reserve_verification(self, payload, handle_hash, expires_at): + def request_budget(self, caller_id, *, rate, burst, now): + """One durable token bucket shared by connections, restarts and operations.""" + with self.lock, self.db: + self.db.execute("BEGIN IMMEDIATE") + row = self.db.execute( + "SELECT tokens,updated_at FROM caller_buckets WHERE caller_id=?", + (caller_id,), + ).fetchone() + tokens, previous = row if row else (float(burst), now) + tokens = min(float(burst), tokens + max(0, now - previous) * rate / 60) + allowed = tokens >= 1 + self.db.execute( + "INSERT OR REPLACE INTO caller_buckets VALUES (?,?,?)", + (caller_id, tokens - int(allowed), max(now, previous)), + ) + wait = max(1, math.ceil((1 - tokens) * 60 / rate)) if not allowed else 0 + return allowed, wait + + def unsuppress(self, email): + with self.lock, self.db: + self.db.execute("DELETE FROM suppressions WHERE email=?", (email.lower(),)) + + def reserve_verification(self, payload, handle_hash, expires_at, *, caller_id=""): # Registration identity, not a tracing ID, defines the logical operation. identity = [payload[key] for key in ("tenant", "client_id", "registration_id")] + if caller_id: + identity.insert(0, caller_id) operation_key = hashlib.sha256(json.dumps(identity).encode()).hexdigest() - binding = {key: payload.get(key) for key in ( - "registration_id", "normalized_email", "preferred_username", - "client_id", "tenant", "display_name", - )} + binding = { + key: payload.get(key) + for key in ( + "registration_id", + "normalized_email", + "preferred_username", + "client_id", + "tenant", + "display_name", + ) + } fingerprint = hashlib.sha256( json.dumps(binding, sort_keys=True, separators=(",", ":")).encode() ).hexdigest() @@ -79,12 +123,19 @@ class SQLiteDeliveryStore: self.db.execute("BEGIN IMMEDIATE") row = self.db.execute( "SELECT fingerprint,request_id,state,error,retryable " - "FROM verification_attempts WHERE operation_key=?", (operation_key,), + "FROM verification_attempts WHERE operation_key=?", + (operation_key,), ).fetchone() if not row: legacy = self.db.execute( - "SELECT 1 FROM verifications WHERE registration_id=? AND client_id=? AND tenant=? LIMIT 1", - (payload["registration_id"], payload["client_id"], payload["tenant"]), + "SELECT 1 FROM verifications WHERE registration_id=? AND client_id=? AND " + "tenant=? AND caller_id=? LIMIT 1", + ( + payload["registration_id"], + payload["client_id"], + payload["tenant"], + caller_id, + ), ).fetchone() if legacy: # Pre-upgrade challenges have no trustworthy send outcome. @@ -99,7 +150,9 @@ class SQLiteDeliveryStore: raise ProviderError("delivery_outcome_unknown", retryable=False) if not row[4]: raise ProviderError(row[3], retryable=False) - request_id = self._insert_verification(payload, handle_hash, expires_at) + request_id = self._insert_verification( + payload, handle_hash, expires_at, caller_id=caller_id + ) self.db.execute( "INSERT OR REPLACE INTO verification_attempts VALUES (?,?,?,?,NULL,NULL,NULL)", (operation_key, fingerprint, request_id, "in_flight"), @@ -110,7 +163,8 @@ class SQLiteDeliveryStore: with self.lock, self.db: self.db.execute( "UPDATE verification_attempts SET state='accepted',provider_ref=? " - "WHERE request_id=? AND state='in_flight'", (reference, request_id), + "WHERE request_id=? AND state='in_flight'", + (reference, request_id), ) def verification_failed(self, request_id, error): @@ -120,7 +174,7 @@ class SQLiteDeliveryStore: self.db.execute("BEGIN IMMEDIATE") self.db.execute( "UPDATE verifications SET canceled_at=? WHERE request_id=?", - (datetime.now(timezone.utc).isoformat(), request_id), + (datetime.now(UTC).isoformat(), request_id), ) self.db.execute( "UPDATE verification_attempts SET state='failed',error=?,retryable=? " @@ -133,7 +187,7 @@ class SQLiteDeliveryStore: self.db.execute("BEGIN IMMEDIATE") try: row = self.db.execute( - "SELECT fingerprint,state FROM send_attempts WHERE event_id=?", + "SELECT fingerprint,state,error,retryable FROM send_attempts WHERE event_id=?", (event_id,), ).fetchone() if row and row[0] != fingerprint: @@ -142,20 +196,23 @@ class SQLiteDeliveryStore: return False if row and row[1] == "in_flight": raise ProviderError("delivery_outcome_unknown", retryable=False) + if row and row[1] == "failed" and not row[3]: + raise ProviderError(row[2] or "delivery_outcome_unknown", retryable=False) self.db.execute( - "INSERT OR REPLACE INTO send_attempts VALUES (?,?,?)", + "INSERT OR REPLACE INTO send_attempts (event_id,fingerprint,state) VALUES " + "(?,?,?)", (event_id, fingerprint, "in_flight"), ) return True finally: self.db.commit() - def send_failed(self, event_id, definitive): + def send_failed(self, event_id, definitive, error): if definitive: with self.db: self.db.execute( - "UPDATE send_attempts SET state='failed' WHERE event_id=?", - (event_id,), + "UPDATE send_attempts SET state='failed',error=?,retryable=? WHERE event_id=?", + (error.code, int(error.retryable), event_id), ) def delivery_status(self, event_id): @@ -167,9 +224,7 @@ class SQLiteDeliveryStore: ).fetchone() return { "state": ( - "unknown" - if row and row[0] == "in_flight" - else "failed" if row else "not_found" + "unknown" if row and row[0] == "in_flight" else "failed" if row else "not_found" ) } @@ -181,9 +236,7 @@ class SQLiteDeliveryStore: def record(self, event_id: str, reference: str) -> None: with self.db: - self.db.execute( - "INSERT INTO deliveries VALUES (?, ?)", (event_id, reference) - ) + self.db.execute("INSERT INTO deliveries VALUES (?, ?)", (event_id, reference)) def is_suppressed(self, email: str) -> bool: row = self.db.execute( @@ -195,22 +248,20 @@ class SQLiteDeliveryStore: with self.db: self.db.execute( "INSERT OR REPLACE INTO suppressions VALUES (?,?,?)", - (email.lower(), reason, datetime.now(timezone.utc).isoformat()), + (email.lower(), reason, datetime.now(UTC).isoformat()), ) - def create_verification( - self, payload: dict, handle_hash: str, expires_at: str - ) -> str: + def create_verification(self, payload: dict, handle_hash: str, expires_at: str) -> str: with self.lock, self.db: return self._insert_verification(payload, handle_hash, expires_at) - def _insert_verification(self, payload, handle_hash, expires_at): + def _insert_verification(self, payload, handle_hash, expires_at, *, caller_id=""): request_id = f"vrq_{secrets.token_hex(12)}" self.db.execute( """INSERT INTO verifications (request_id, handle_hash, registration_id, email, username, - client_id, tenant, display_name, expires_at, consumed_at, canceled_at) - VALUES (?,?,?,?,?,?,?,?,?,NULL,NULL)""", + client_id, tenant, display_name, expires_at, consumed_at, canceled_at, caller_id) + VALUES (?,?,?,?,?,?,?,?,?,NULL,NULL,?)""", ( request_id, handle_hash, @@ -221,11 +272,12 @@ class SQLiteDeliveryStore: payload["tenant"], payload.get("display_name"), expires_at, + caller_id, ), ) return request_id - def consume_verification(self, handle_hash: str) -> dict: + def consume_verification(self, handle_hash: str, *, caller_id="", bindings=None) -> dict: with self.lock, self.db: self.db.execute("BEGIN IMMEDIATE") row = self.db.execute( @@ -233,11 +285,13 @@ class SQLiteDeliveryStore: ).fetchone() if row is None or row[9] is not None or row[10] is not None: raise ValueError("verification_invalid") - if datetime.fromisoformat(row[8]) <= datetime.now(timezone.utc): + if row[11] != caller_id or (bindings is not None and (row[6], row[5]) not in bindings): + raise ValueError("verification_invalid") + if datetime.fromisoformat(row[8]) <= datetime.now(UTC): raise ValueError("verification_expired") self.db.execute( "UPDATE verifications SET consumed_at=? WHERE request_id=?", - (datetime.now(timezone.utc).isoformat(), row[0]), + (datetime.now(UTC).isoformat(), row[0]), ) return { "request_id": row[0], @@ -249,7 +303,7 @@ class SQLiteDeliveryStore: "display_name": row[7], } - def cancel_verification(self, handle_hash: str) -> dict: + def cancel_verification(self, handle_hash: str, *, caller_id="", bindings=None) -> dict: with self.lock, self.db: self.db.execute("BEGIN IMMEDIATE") row = self.db.execute( @@ -257,11 +311,13 @@ class SQLiteDeliveryStore: ).fetchone() if row is None or row[9] is not None or row[10] is not None: raise ValueError("verification_invalid") - if datetime.fromisoformat(row[8]) <= datetime.now(timezone.utc): + if row[11] != caller_id or (bindings is not None and (row[6], row[5]) not in bindings): + raise ValueError("verification_invalid") + if datetime.fromisoformat(row[8]) <= datetime.now(UTC): raise ValueError("verification_expired") self.db.execute( "UPDATE verifications SET canceled_at=? WHERE request_id=?", - (datetime.now(timezone.utc).isoformat(), row[0]), + (datetime.now(UTC).isoformat(), row[0]), ) return { "request_id": row[0], @@ -322,9 +378,7 @@ class SMTPProvider: # that ties an accepted send to a message later observed in a mailbox. # Without it the reference below fell back to hash(), which Python # randomizes per process and which collides for equal recipient/subject. - message["Message-ID"] = make_msgid( - domain=self.sender.rpartition("@")[2] or None - ) + message["Message-ID"] = make_msgid(domain=self.sender.rpartition("@")[2] or None) message.set_content(text) submitted = False accepted = False @@ -363,10 +417,19 @@ class SMTPProvider: class TransactionalApplication: def __init__( - self, store: SQLiteDeliveryStore, provider, bearer_token: str, portal_url: str + self, + store: SQLiteDeliveryStore, + provider, + bearer_token: str, + portal_url: str, + *, + verification_namespace="", + verification_bindings=None, ) -> None: self.store, self.provider, self.token = store, provider, bearer_token self.portal_url = portal_url.rstrip("/") + self.verification_namespace = verification_namespace + self.verification_bindings = verification_bindings self._status_cache = None self._status_time = 0.0 @@ -386,15 +449,11 @@ class TransactionalApplication: ) or environ.get("REQUEST_METHOD") != "POST" ): - return self._json( - start_response, HTTPStatus.NOT_FOUND, {"error": "not_found"} - ) + return self._json(start_response, HTTPStatus.NOT_FOUND, {"error": "not_found"}) if not hmac.compare_digest( str(environ.get("HTTP_AUTHORIZATION", "")), f"Bearer {self.token}" ): - return self._json( - start_response, HTTPStatus.UNAUTHORIZED, {"error": "unauthorized"} - ) + return self._json(start_response, HTTPStatus.UNAUTHORIZED, {"error": "unauthorized"}) try: length = int(environ.get("CONTENT_LENGTH") or 0) if length <= 0 or length > 128 * 1024: @@ -417,9 +476,7 @@ class TransactionalApplication: return self._send_invitation(start_response, environ, payload) except ProviderError as exc: status = ( - HTTPStatus.SERVICE_UNAVAILABLE - if exc.retryable - else HTTPStatus.UNPROCESSABLE_ENTITY + HTTPStatus.SERVICE_UNAVAILABLE if exc.retryable else HTTPStatus.UNPROCESSABLE_ENTITY ) return self._json( start_response, @@ -427,9 +484,7 @@ class TransactionalApplication: {"error": exc.code, "retryable": exc.retryable}, ) except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc: - return self._json( - start_response, HTTPStatus.BAD_REQUEST, {"error": str(exc)} - ) + return self._json(start_response, HTTPStatus.BAD_REQUEST, {"error": str(exc)}) except (TimeoutError, OSError, smtplib.SMTPException): # Safety net: never leak provider exception text. return self._json( @@ -454,7 +509,7 @@ class TransactionalApplication: pass self._status_cache = { "checks": checks, - "checked_at": datetime.now(timezone.utc).isoformat(), + "checked_at": datetime.now(UTC).isoformat(), "message_sent": False, "inbox_receipt_verified": False, } @@ -465,10 +520,7 @@ class TransactionalApplication: event_id = str(payload["id"]) if environ.get("HTTP_IDEMPOTENCY_KEY") != event_id: raise ValueError("idempotency_key_mismatch") - if ( - payload.get("source") != "user-engine" - or payload.get("type") not in ALLOWED_EVENTS - ): + if payload.get("source") != "user-engine" or payload.get("type") not in ALLOWED_EVENTS: raise ValueError("template_not_allowed") recipient = str(payload.get("data", {}).get("primary_email", "")) if not _valid_address(recipient): @@ -491,7 +543,8 @@ class TransactionalApplication: reference = self.provider.send( recipient, "Your NetKingdom invitation", - f"You have been invited. Continue securely at {self.portal_url}/invitations/{invitation_id}\n", + f"You have been invited. Continue securely at " + f"{self.portal_url}/invitations/{invitation_id}\n", ) self.store.record(event_id, reference) except ProviderError as exc: @@ -500,11 +553,9 @@ class TransactionalApplication: "definitive", exc.code in {"temporary_deferral", "permanent_rejection"}, ) - self.store.send_failed(event_id, definitive) + self.store.send_failed(event_id, definitive, exc) if not definitive: - raise ProviderError( - "delivery_outcome_unknown", retryable=False - ) from None + raise ProviderError("delivery_outcome_unknown", retryable=False) from None raise except Exception: # Preserve the durable reservation: SMTP may have accepted the message. @@ -530,17 +581,20 @@ class TransactionalApplication: "tenant", "correlation_id", ) - if any(not isinstance(payload.get(key), str) or not payload[key] for key in required) or not _valid_address( - str(payload["normalized_email"]) - ): + if any( + not isinstance(payload.get(key), str) or not payload[key] for key in required + ) or not _valid_address(str(payload["normalized_email"])): raise ValueError("invalid_verification_request") email = str(payload["normalized_email"]) if self.store.is_suppressed(email): raise ValueError("recipient_suppressed") handle = secrets.token_urlsafe(32) - expires_at = datetime.now(timezone.utc) + timedelta(minutes=30) + expires_at = datetime.now(UTC) + timedelta(minutes=30) request_id, send = self.store.reserve_verification( - payload, hashlib.sha256(handle.encode()).hexdigest(), expires_at.isoformat() + payload, + hashlib.sha256(handle.encode()).hexdigest(), + expires_at.isoformat(), + caller_id=self.verification_namespace, ) if send: try: @@ -548,13 +602,15 @@ class TransactionalApplication: email, "Verify your NetKingdom registration", f"Continue securely at {self.portal_url}/registration/verify?handle={handle}\n" - f"Cancel this request at {self.portal_url}/registration/cancel?handle={handle}\n" + f"Cancel this request at {self.portal_url}/registration/cancel" + f"?handle={handle}\n" "These links expire in 30 minutes.\n", ) self.store.verification_accepted(request_id, reference) except ProviderError as exc: definitive = getattr( - exc, "definitive", + exc, + "definitive", exc.code in {"temporary_deferral", "permanent_rejection"}, ) if not definitive: @@ -577,7 +633,9 @@ class TransactionalApplication: if len(handle) < 32: raise ValueError("verification_invalid") evidence = self.store.consume_verification( - hashlib.sha256(handle.encode()).hexdigest() + hashlib.sha256(handle.encode()).hexdigest(), + caller_id=self.verification_namespace, + bindings=self.verification_bindings, ) # mailbox_control is channel evidence only; never an authorization decision. return self._json( @@ -599,7 +657,9 @@ class TransactionalApplication: if len(handle) < 32: raise ValueError("verification_invalid") evidence = self.store.cancel_verification( - hashlib.sha256(handle.encode()).hexdigest() + hashlib.sha256(handle.encode()).hexdigest(), + caller_id=self.verification_namespace, + bindings=self.verification_bindings, ) return self._json( start_response, @@ -639,15 +699,15 @@ def _is_loopback_host(host: str) -> bool: def _valid_address(value: str) -> bool: _, address = parseaddr(value) - return ( - address == value - and "@" in address - and "\n" not in address - and "\r" not in address - ) + return address == value and "@" in address and "\n" not in address and "\r" not in address def main() -> None: + policies = None + if "EMAIL_CONNECT_CALLERS_FILE" in os.environ: + from .callers import ScopedTransactionalApplication, load_callers + + policies = load_callers(os.environ["EMAIL_CONNECT_CALLERS_FILE"]) provider = SMTPProvider( os.environ["EMAIL_CONNECT_SMTP_HOST"], int(os.environ.get("EMAIL_CONNECT_SMTP_PORT", "587")), @@ -656,14 +716,23 @@ def main() -> None: os.environ["EMAIL_CONNECT_SENDER"], security=os.environ.get("EMAIL_CONNECT_SMTP_SECURITY", "starttls"), ) - app = TransactionalApplication( - SQLiteDeliveryStore( - os.environ.get("EMAIL_CONNECT_DATABASE_PATH", "/data/email-connect.db") - ), - provider, - os.environ["EMAIL_CONNECT_INGEST_TOKEN"].strip(), - os.environ["EMAIL_CONNECT_PORTAL_URL"], + store = SQLiteDeliveryStore( + os.environ.get("EMAIL_CONNECT_DATABASE_PATH", "/data/email-connect.db") ) + if policies is not None: + app = ScopedTransactionalApplication( + store, + provider, + os.environ["EMAIL_CONNECT_PORTAL_URL"], + policies, + ) + else: + app = TransactionalApplication( + store, + provider, + os.environ["EMAIL_CONNECT_INGEST_TOKEN"].strip(), + os.environ["EMAIL_CONNECT_PORTAL_URL"], + ) with make_server( "0.0.0.0", int(os.environ.get("EMAIL_CONNECT_HTTP_PORT", "8080")), app ) as server: diff --git a/tests/test_callers.py b/tests/test_callers.py new file mode 100644 index 0000000..03a1282 --- /dev/null +++ b/tests/test_callers.py @@ -0,0 +1,414 @@ +"""Admitted caller contract: only synthetic tokens, providers, recipients and ledgers.""" + +import io +import json +from concurrent.futures import ThreadPoolExecutor + +import pytest +from email_connect.callers import CallerPolicy, ScopedTransactionalApplication, load_callers +from email_connect.transactional import ProviderError, SQLiteDeliveryStore, TransactionalApplication + + +class Provider: + def __init__(self): + self.calls = [] + + def send(self, *args): + self.calls.append(args) + return "synthetic-reference-recipient=member@example.test-token=must-not-leak" + + def check(self): + return None + + +def policy(name="caller-a", **changes): + values = dict( + caller_id=name, + token="synthetic-token-" + name, + source=name, + bindings=frozenset({("tenant:a", "client:a")}), + operations=frozenset({"send", "delivery-status", "verify", "consume", "cancel"}), + templates=frozenset({"family_member.invited"}), + rate_per_minute=60, + burst=100, + max_payload_bytes=32768, + ) + values.update(changes) + return CallerPolicy(**values) + + +def event(caller="caller-a", **changes): + payload = { + "id": "evt-1", + "type": "family_member.invited", + "source": caller, + "tenant": "tenant:a", + "client_id": "client:a", + "data": {"primary_email": "member@example.test", "invitation_id": "invite-1"}, + } + payload.update(changes) + return payload + + +def verification(**changes): + result = { + "registration_id": "reg-1", + "normalized_email": "member@example.test", + "preferred_username": "member", + "tenant": "tenant:a", + "client_id": "client:a", + "correlation_id": "trace-1", + } + result.update(changes) + return result + + +def invoke(app, payload, caller="caller-a", path="/v1/send", token=None, key=None): + raw = json.dumps(payload).encode() + response = {} + env = { + "PATH_INFO": path, + "REQUEST_METHOD": "POST", + "CONTENT_LENGTH": str(len(raw)), + "wsgi.input": io.BytesIO(raw), + "HTTP_AUTHORIZATION": "Bearer " + (token or "synthetic-token-" + caller), + "HTTP_IDEMPOTENCY_KEY": key if key is not None else payload.get("id", ""), + } + body = b"".join( + app( + env, + lambda status, headers: response.update( + status=int(status.split()[0]), + headers=dict(headers), + ), + ) + ) + return response["status"], json.loads(body), response["headers"] + + +def application(tmp_path, policies=None, clock=None): + store = SQLiteDeliveryStore(str(tmp_path / "ledger.db")) + provider = Provider() + app = ScopedTransactionalApplication( + store, + provider, + "https://issuer.example.test", + policies or [policy(), policy("caller-b")], + **({"clock": clock} if clock is not None else {}), + ) + return app, store, provider + + +def test_credentials_determine_caller_and_status_is_scoped_and_redacted(tmp_path): + app, _, provider = application(tmp_path) + assert invoke(app, event(), token="wrong-credential")[0] == 401 + assert invoke(app, event("caller-b"))[0] == 400 + accepted, body, headers = invoke(app, event()) + assert accepted == 202 and body["event_id"] == "evt-1" and body["reference"].startswith("dvr_") + assert headers["Cache-Control"] == "private, no-store" + status, record, _ = invoke(app, {"event_id": "evt-1"}, path="/v1/delivery-status") + assert status == 200 and record["state"] == "provider_accepted" + assert record["reference"] == body["reference"] + _, other, _ = invoke(app, {"event_id": "evt-1"}, caller="caller-b", path="/v1/delivery-status") + assert other["state"] == "not_found" + assert len(provider.calls) == 1 + serialized = json.dumps([body, record, other]) + assert "member@example.test" not in serialized and "must-not-leak" not in serialized + + +@pytest.mark.parametrize( + "change", + [ + {"tenant": "tenant:b"}, + {"client_id": "client:b"}, + {"tenant": ["tenant:a"]}, + {"type": "unapproved-template"}, + {"type": []}, + {"source": "user-engine"}, + ], +) +def test_scope_and_template_denials_never_submit(tmp_path, change): + app, _, provider = application(tmp_path) + assert invoke(app, event(**change))[0] == 400 + assert provider.calls == [] + + +def test_retry_and_payload_identity_are_independent_across_callers_and_restart(tmp_path): + app, store, provider = application(tmp_path) + first = invoke(app, event())[1] + assert invoke(app, event())[1]["status"] == "duplicate" + assert ( + invoke( + app, event(data={"primary_email": "other@example.test", "invitation_id": "invite-1"}) + )[0] + == 400 + ) + assert invoke(app, event("caller-b"), caller="caller-b")[0] == 202 + assert len(provider.calls) == 2 + store.db.close() + restarted, _, empty_provider = application(tmp_path) + assert invoke(restarted, event())[1]["reference"] == first["reference"] + assert empty_provider.calls == [] + + +def test_unknown_submission_survives_restart_without_resend(tmp_path): + app, store, provider = application(tmp_path) + + def unknown(*args): + provider.calls.append(args) + raise OSError("synthetic SMTP recipient/token must not leak") + + provider.send = unknown + status, body, _ = invoke(app, event()) + assert status == 422 and body == {"error": "delivery_outcome_unknown", "retryable": False} + store.db.close() + restarted, _, next_provider = application(tmp_path) + assert invoke(restarted, event())[1] == body + assert len(provider.calls) == 1 and next_provider.calls == [] + + +def test_rate_budget_survives_restart_and_backward_clock_and_is_per_caller(tmp_path): + now = [100.0] + policies = [policy(burst=1, rate_per_minute=2), policy("caller-b", burst=1, rate_per_minute=2)] + app, store, provider = application(tmp_path, policies, clock=lambda: now[0]) + assert invoke(app, event())[0] == 202 + status, _, headers = invoke(app, event(id="evt-2")) + assert status == 429 and headers["Retry-After"] == "30" and len(provider.calls) == 1 + assert invoke(app, event("caller-b"), caller="caller-b")[0] == 202 + store.db.close() + restarted, _, next_provider = application(tmp_path, policies, clock=lambda: now[0]) + assert invoke(restarted, event(id="evt-2"))[0] == 429 + now[0] = 90 + assert invoke(restarted, event(id="evt-2"))[0] == 429 + now[0] = 130 + assert invoke(restarted, event(id="evt-2"))[0] == 202 + assert len(next_provider.calls) == 1 + + +def test_concurrent_connections_cannot_exceed_request_burst(tmp_path): + first, store, provider = application(tmp_path, [policy(burst=1)]) + second = ScopedTransactionalApplication( + SQLiteDeliveryStore(str(tmp_path / "ledger.db")), + provider, + "https://issuer.example.test", + [policy(burst=1)], + ) + with ThreadPoolExecutor(max_workers=2) as workers: + responses = list(workers.map(lambda app: invoke(app, event()), [first, second])) + assert sorted(r[0] for r in responses) == [202, 429] + assert len(provider.calls) == 1 + + +def test_verification_owner_and_binding_checked_before_consumption_or_cancellation(tmp_path): + app, store, provider = application(tmp_path) + request = verification() + first = invoke(app, request, path="/v1/registration-verifications")[1] + assert invoke(app, request, path="/v1/registration-verifications")[1] == first + handle = provider.calls[0][2].split("handle=", 1)[1].splitlines()[0] + for path in ("/consume", "/cancel"): + assert ( + invoke( + app, + {"handle": handle}, + caller="caller-b", + path="/v1/registration-verifications" + path, + )[0] + == 400 + ) + assert invoke(app, request, caller="caller-b", path="/v1/registration-verifications")[0] == 202 + assert len(provider.calls) == 2 + legacy = TransactionalApplication(store, provider, "legacy", "https://issuer.example.test") + raw = json.dumps({"handle": handle}).encode() + response = {} + legacy( + { + "PATH_INFO": "/v1/registration-verifications/consume", + "REQUEST_METHOD": "POST", + "CONTENT_LENGTH": str(len(raw)), + "wsgi.input": io.BytesIO(raw), + "HTTP_AUTHORIZATION": "Bearer legacy", + }, + lambda status, headers: response.update(status=int(status.split()[0])), + ) + assert response["status"] == 400 + limited = ScopedTransactionalApplication( + store, + provider, + "https://issuer.example.test", + [policy(bindings=frozenset({("tenant:b", "client:b")}))], + ) + assert ( + invoke(limited, {"handle": handle}, path="/v1/registration-verifications/consume")[0] == 400 + ) + assert invoke(app, {"handle": handle}, path="/v1/registration-verifications/consume")[0] == 200 + assert invoke(app, {"handle": handle}, path="/v1/registration-verifications/cancel")[0] == 400 + + +def test_suppression_and_unsuppression_require_explicit_owner_operations(tmp_path): + ops = policy("mail-operator", operations=frozenset({"suppress", "unsuppress"})) + app, _, provider = application(tmp_path, [policy(), ops]) + payload = {"recipient": "member@example.test", "reason": "complaint"} + assert invoke(app, payload, path="/v1/suppressions")[0] == 403 + assert invoke(app, payload, caller="mail-operator", path="/v1/suppressions")[0] == 200 + assert invoke(app, event())[1]["error"] == "recipient_suppressed" + assert not provider.calls + assert invoke(app, payload, caller="mail-operator", path="/v1/suppressions/remove")[0] == 200 + assert invoke(app, event())[0] == 202 + + +def test_original_caller_legacy_reservations_and_links_need_explicit_adoption(tmp_path): + app, store, provider = application(tmp_path) + payload = verification() + legacy = TransactionalApplication(store, provider, "legacy", "https://issuer.example.test") + raw = json.dumps(payload).encode() + result = {} + legacy( + { + "PATH_INFO": "/v1/registration-verifications", + "REQUEST_METHOD": "POST", + "CONTENT_LENGTH": str(len(raw)), + "wsgi.input": io.BytesIO(raw), + "HTTP_AUTHORIZATION": "Bearer legacy", + }, + lambda status, headers: result.update(status=status), + ) + handle = provider.calls[0][2].split("handle=", 1)[1].splitlines()[0] + adopter = policy("user-engine", adopt_legacy_verifications=True) + adopted = ScopedTransactionalApplication( + store, provider, "https://issuer.example.test", [adopter] + ) + assert ( + invoke(adopted, payload, caller="user-engine", path="/v1/registration-verifications")[0] + == 202 + ) + assert len(provider.calls) == 1 + assert ( + invoke( + adopted, + {"handle": handle}, + caller="user-engine", + path="/v1/registration-verifications/consume", + )[0] + == 200 + ) + with pytest.raises(ValueError, match="original_caller"): + policy(adopt_legacy_verifications=True) + + +def test_payload_limit_unknown_operation_and_provider_status_have_no_implicit_grant(tmp_path): + app, _, provider = application(tmp_path, [policy(max_payload_bytes=100)]) + assert invoke(app, event(data={"primary_email": "x" * 1000}))[0] == 413 + assert invoke(app, {}, path="/v1/status")[0] == 403 + assert invoke(app, {}, path="/unknown")[0] == 404 + assert not provider.calls + + +@pytest.mark.parametrize( + "change", + [ + {"burst": 0}, + {"rate_per_minute": True}, + {"max_payload_bytes": 200000}, + {"token": ""}, + {"bindings": frozenset()}, + {"templates": frozenset({"arbitrary-body"})}, + ], +) +def test_policy_refuses_unbounded_or_unauthorized_configuration(change): + with pytest.raises(ValueError): + policy(**change) + + +def test_policy_loader_uses_independent_files_and_never_echoes_credentials(tmp_path): + token_file = tmp_path / "token" + token_file.write_text("synthetic-token-caller-a\n") + entry = dict( + caller_id="caller-a", + source="caller-a", + bindings=[["tenant:a", "client:a"]], + operations=["send"], + templates=["family_member.invited"], + rate_per_minute=10, + burst=3, + max_payload_bytes=8192, + token_file=str(token_file), + ) + config = tmp_path / "policy.json" + config.write_text(json.dumps({"schema_version": 1, "callers": [entry]})) + loaded = load_callers(config) + assert loaded[0].token == "synthetic-token-caller-a" + assert "synthetic-token" not in repr(loaded) + config.write_text( + json.dumps({"schema_version": 1, "callers": [entry, dict(entry, caller_id="caller-b")]}) + ) + with pytest.raises(ValueError, match="caller_configuration_invalid") as error: + load_callers(config) + assert "synthetic-token" not in str(error.value) + entry["token"] = "synthetic-embedded-secret" + config.write_text(json.dumps({"schema_version": 1, "callers": [entry]})) + with pytest.raises(ValueError, match="caller_configuration_invalid"): + load_callers(config) + + +@pytest.mark.parametrize("retryable", [False, True]) +def test_definitive_failure_retains_retry_policy_after_restart(tmp_path, retryable): + app, store, provider = application(tmp_path) + + def reject(*args): + provider.calls.append(args) + raise ProviderError( + "temporary_deferral" if retryable else "permanent_rejection", retryable=retryable + ) + + provider.send = reject + first = invoke(app, event()) + assert first[0] == (503 if retryable else 422) + store.db.close() + restarted, _, next_provider = application(tmp_path) + result = invoke(restarted, event()) + assert result[0] == (202 if retryable else 422) + assert len(next_provider.calls) == int(retryable) + + +def test_legacy_invitation_adoption_preserves_accepted_receipt(tmp_path): + store = SQLiteDeliveryStore(str(tmp_path / "ledger.db")) + provider = Provider() + legacy = TransactionalApplication( + store, provider, "legacy-token", "https://issuer.example.test" + ) + old = event("user-engine") + old.pop("tenant") + old.pop("client_id") + assert invoke(legacy, old, token="legacy-token")[0] == 202 + admitted = policy("user-engine", legacy_invitation_binding=("tenant:a", "client:a")) + scoped = ScopedTransactionalApplication( + store, provider, "https://issuer.example.test", [admitted] + ) + assert invoke(scoped, old, caller="user-engine")[1]["status"] == "duplicate" + assert len(provider.calls) == 1 + assert invoke(scoped, dict(old, id="ec2_reserved"), caller="user-engine")[0] == 400 + + +def test_historical_failed_reservation_is_conservatively_terminal(tmp_path): + import sqlite3 + + path = tmp_path / "old.db" + with sqlite3.connect(path) as db: + db.execute( + "CREATE TABLE send_attempts (event_id TEXT PRIMARY KEY, " + "fingerprint TEXT NOT NULL, state TEXT NOT NULL)" + ) + db.execute("INSERT INTO send_attempts VALUES ('old-event','old-fingerprint','failed')") + store = SQLiteDeliveryStore(str(path)) + with pytest.raises(ProviderError, match="delivery_outcome_unknown"): + store.reserve_send("old-event", "old-fingerprint") + + +def test_invalid_policy_refuses_startup_before_database_mutation(tmp_path, monkeypatch): + from email_connect.transactional import main + + monkeypatch.setenv("EMAIL_CONNECT_CALLERS_FILE", str(tmp_path / "missing.json")) + monkeypatch.setenv("EMAIL_CONNECT_DATABASE_PATH", str(tmp_path / "not-created.db")) + with pytest.raises(ValueError, match="caller_configuration_invalid"): + main() + assert not (tmp_path / "not-created.db").exists() diff --git a/tests/test_ledger_backup.py b/tests/test_ledger_backup.py index 82c7126..b9081f9 100644 --- a/tests/test_ledger_backup.py +++ b/tests/test_ledger_backup.py @@ -433,3 +433,45 @@ def test_helper_has_no_secret_or_token_mount_and_requires_no_egress(modules): }, labels, ) + + +def test_scoped_ledger_rehearsal_preserves_ownership_and_budgets(modules, tmp_path, monkeypatch, capsys): + archive, _, _ = modules + path = tmp_path / "source.sqlite3" + synthetic_store(path) + store = SQLiteDeliveryStore(str(path)) + payload = { + "registration_id": "scoped", "normalized_email": "scoped@example.test", + "preferred_username": "synthetic", "client_id": "synthetic-client", + "tenant": "synthetic-tenant", "correlation_id": "trace-scoped", + } + expires = (datetime.now(UTC) + timedelta(days=1)).isoformat() + request, _ = store.reserve_verification( + payload, hashlib.sha256(b"scoped-handle").hexdigest(), expires, caller_id="caller-a" + ) + store.verification_accepted(request, "synthetic-scoped-reference") + store.reserve_verification( + dict(payload, registration_id="scoped-unknown"), + hashlib.sha256(b"scoped-unknown-handle").hexdigest(), expires, caller_id="caller-b" + ) + assert store.request_budget("caller-a", rate=2, burst=1, now=100)[0] + store.db.close() + before = archive.fingerprint(path) + assert before["tables"]["caller_buckets"]["rows"] == 1 + script = Path(__file__).resolve().parents[1] / "scripts/rehearse-ledger.py" + spec = importlib.util.spec_from_file_location("rehearsal", script) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + recovered = tmp_path / "recovered.sqlite3" + code = module.CODE.replace("Path('/tmp/recovered.sqlite3')", "Path(" + repr(str(recovered)) + ")") + monkeypatch.setattr(sys, "stdin", io.TextIOWrapper(io.BytesIO(path.read_bytes()))) + monkeypatch.setattr(__import__("os"), "getuid", lambda: 10001) + exec(compile(code, str(script), "exec"), {}) + receipt = json.loads(capsys.readouterr().out) + assert receipt["checks"]["accepted_verifications"] == 4 + assert receipt["checks"]["unknown_verifications"] == 2 + assert receipt["mail_sent"] is False + assert archive.fingerprint(recovered) == before + restarted = SQLiteDeliveryStore(str(recovered)) + assert not restarted.request_budget("caller-a", rate=2, burst=1, now=100)[0] + restarted.db.close() diff --git a/workplans/EMAIL-WP-0008-scoped-transactional-callers.md b/workplans/EMAIL-WP-0008-scoped-transactional-callers.md new file mode 100644 index 0000000..050eb30 --- /dev/null +++ b/workplans/EMAIL-WP-0008-scoped-transactional-callers.md @@ -0,0 +1,70 @@ +--- +id: EMAIL-WP-0008 +type: workplan +title: "Isolate explicitly admitted transactional mail callers" +domain: infotech +repo: email-connect +status: active +owner: codex +topic_slug: netkingdom +created: "2026-10-08" +updated: "2026-10-08" +origin: residual +origin_ref: CSOC-WP-0010 +--- + +CSOC-WP-0010-T04 needs independent caller credentials, bounded template/tenant +permissions, durable limits and scoped diagnostics. This record owns receiver +source and its admission; current live single-caller SMTP remains unchanged. + +## T01 — Implement explicit caller policies and durable isolation + +```task +id: EMAIL-WP-0008-T01 +status: done +priority: high +``` + +Opt-in read-only policy/token-file loader, fixed operation/template/binding +admission, exact credential/source identity, per-caller invitations/verifications, +atomic persistent request budgets, bounded input, redacted scoped status and +privileged suppression controls. Invalid selected policy refuses startup without +legacy fallback. Permanent invitation failures now remain terminal across +restart; confirmed temporary failures can retry. Unknown outcomes never resend. +Explicit original-caller legacy adoption preserves accepted evidence. + +## T02 — Prove isolation, retry policy and recovered-ledger continuity + +```task +id: EMAIL-WP-0008-T02 +status: done +priority: high +``` + +Synthetic tests cover cross-caller handle denial without mutation, scope changes, +independent retries, permanent/temporary/unknown outcome, restart, concurrent +budgets, backward clock, size/config failure and safe status. App owner gate +binds exact tests to modules with no network or inherited secrets. Ledger +fingerprint includes rate budgets; rehearsal must preserve scoped accepted and +unknown evidence and quota rows without sending. Final owner gate:148passed, zero skips, including the loopback-only GreenMail +harness. Network-denied app owner gate:66passed, zero skips. Ruff passes for +receiver and caller module/tests. T03 remains waiting for runtime admission. + +## T03 — Admit original caller transition, runtime custody and recovery + +```task +id: EMAIL-WP-0008-T03 +status: wait +priority: high +needs_human: true +intervention_note: "Actual callers/scopes/quotas and separate token custody, platform sender/reply/bounce ownership, schema rollout/rollback and independent ledger recovery admission remain unapproved; source tests do not activate them." +``` + +Retain this live admission record before any completion. Choose real binding and +operation sets/quotas, independent credential delivery and explicit legacy +adoption. Review exact immutable image and additive schema migration/rollback. +Repin guarded capture/rehearsal helpers together. EMAIL-WP-0007-T03/T04 and +RPF-WP-0053-T05/T06 retain offsite archive/atomic credential pair custody. +CSOC-WP-0010-T04/T05/T06 retain platform operations and productive/browser review. +Review: docs/caller-admission-2026-10-08.md. No sender, caller, token, image, +mailbox, production ledger, network admission, upload or timer is activated.