From 0fcfddc342302b98d23888f8ff2e506b01a7846e Mon Sep 17 00:00:00 2001 From: tegwick Date: Sun, 13 Sep 2026 22:11:49 +0200 Subject: [PATCH] Implement P05 checked services and safe selected delivery recovery Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a092fe-13b1-7f12-ac74-7d258af4d79c --- .forgejo/workflows/acceptance.yaml | 21 +++ .forgejo/workflows/image.yaml | 56 ++++++ Containerfile | 2 +- src/email_connect/transactional.py | 263 +++++++++++++++++++++++------ tests/test_transactional.py | 184 ++++++++++++++++++-- 5 files changed, 459 insertions(+), 67 deletions(-) create mode 100644 .forgejo/workflows/acceptance.yaml create mode 100644 .forgejo/workflows/image.yaml diff --git a/.forgejo/workflows/acceptance.yaml b/.forgejo/workflows/acceptance.yaml new file mode 100644 index 0000000..58d7275 --- /dev/null +++ b/.forgejo/workflows/acceptance.yaml @@ -0,0 +1,21 @@ +name: Transactional mail acceptance +on: + push: + branches: [main] + paths: ["src/**", "tests/**", "pyproject.toml", ".forgejo/workflows/acceptance.yaml"] + workflow_dispatch: +jobs: + mail: + runs-on: ubuntu-latest + container: + image: python:3.12-slim@sha256:d764629ce0ddd8c71fd371e9901efb324a95789d2315a47db7e4d27e78f1b0e9 + steps: + - name: Test exact source commit + run: | + set -eu + mkdir source + python -c 'import os, urllib.request; urllib.request.urlretrieve("https://forgejo.coulomb.social/" + os.environ["GITHUB_REPOSITORY"] + "/archive/" + os.environ["GITHUB_SHA"] + ".tar.gz", "source.tar.gz")' + tar xzf source.tar.gz -C source --strip-components=1 + cd source + pip install . pytest + PYTHONDONTWRITEBYTECODE=1 PYTHONPATH=src python -m pytest -p no:cacheprovider -q diff --git a/.forgejo/workflows/image.yaml b/.forgejo/workflows/image.yaml new file mode 100644 index 0000000..55a3cb7 --- /dev/null +++ b/.forgejo/workflows/image.yaml @@ -0,0 +1,56 @@ +name: Build and Publish Container Image + +# Modelled on tenant-engine/.forgejo/workflows/image.yaml. Images are built +# by CI from a tarball of the pushed commit, never from a workstation tree. + +on: + push: + branches: + - main + paths: + - ".forgejo/workflows/image.yaml" + - "Containerfile" + - "src/**" + - "pyproject.toml" + - "README.md" + - "LICENSE" + workflow_dispatch: + +env: + REGISTRY: forgejo.coulomb.social + IMAGE_NAME: coulomb/email-connect + DOCKER_HOST: tcp://127.0.0.1:2375 + +jobs: + build-and-push: + runs-on: container-build + steps: + - name: Build and push image + env: + REGISTRY_USER: ${{ secrets.REGISTRY_USER }} + REGISTRY_TOKEN: ${{ secrets.REGISTRY_TOKEN }} + run: | + set -eu + REF="${GITHUB_SHA:-main}" + SHORT="${REF:0:7}" + mkdir -p buildctx "${HOME}/bin" + wget -qO /tmp/repo.tar.gz \ + "https://forgejo.coulomb.social/${GITHUB_REPOSITORY}/archive/${SHORT}.tar.gz" + tar xzf /tmp/repo.tar.gz -C buildctx --strip-components=1 + wget -qO- https://download.docker.com/linux/static/stable/x86_64/docker-27.3.1.tgz \ + | tar xz --strip-components=1 -C "${HOME}/bin" docker/docker + export PATH="${HOME}/bin:${PATH}" + echo "${REGISTRY_TOKEN}" | docker login "${REGISTRY}" -u "${REGISTRY_USER}" --password-stdin + IMAGE="${REGISTRY}/${IMAGE_NAME}" + docker build -f buildctx/Containerfile -t "${IMAGE}:latest" -t "${IMAGE}:main-${SHORT}" buildctx + docker push "${IMAGE}:latest" + docker push "${IMAGE}:main-${SHORT}" + echo "pushed ${IMAGE}:latest and ${IMAGE}:main-${SHORT}" + + - name: Report immutable digest + run: | + set -eu + export PATH="${HOME}/bin:${PATH}" + IMAGE="${REGISTRY}/${IMAGE_NAME}" + SHORT="${GITHUB_SHA:0:7}" + docker inspect --format='{{index .RepoDigests 0}}' "${IMAGE}:main-${SHORT}" diff --git a/Containerfile b/Containerfile index ffae8d6..520006d 100644 --- a/Containerfile +++ b/Containerfile @@ -1,4 +1,4 @@ -FROM python:3.12-slim +FROM python:3.12-slim@sha256:d764629ce0ddd8c71fd371e9901efb324a95789d2315a47db7e4d27e78f1b0e9 RUN useradd --system --uid 10001 --create-home email-connect WORKDIR /app COPY pyproject.toml README.md LICENSE ./ diff --git a/src/email_connect/transactional.py b/src/email_connect/transactional.py index e883e6b..68c659f 100644 --- a/src/email_connect/transactional.py +++ b/src/email_connect/transactional.py @@ -8,6 +8,8 @@ import ipaddress import json import os import smtplib +import ssl +import time import sqlite3 import secrets from datetime import datetime, timedelta, timezone @@ -36,26 +38,72 @@ class SQLiteDeliveryStore: self.db.execute( "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 ( + 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 - )""" - ) - self.db.execute( - """CREATE TABLE IF NOT EXISTS suppressions ( + )""") + 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") + self.db.execute( + "CREATE TABLE IF NOT EXISTS send_attempts (event_id TEXT PRIMARY KEY, fingerprint TEXT NOT NULL, state TEXT NOT NULL)" + ) self.db.commit() + def reserve_send(self, event_id, fingerprint): + with self.lock: + self.db.execute("BEGIN IMMEDIATE") + try: + row = self.db.execute( + "SELECT fingerprint,state FROM send_attempts WHERE event_id=?", + (event_id,), + ).fetchone() + if row and row[0] != fingerprint: + raise ValueError("idempotency_payload_mismatch") + if self.reference(event_id): + return False + if row and row[1] == "in_flight": + raise ProviderError("delivery_outcome_unknown", retryable=False) + self.db.execute( + "INSERT OR REPLACE INTO send_attempts VALUES (?,?,?)", + (event_id, fingerprint, "in_flight"), + ) + return True + finally: + self.db.commit() + + def send_failed(self, event_id, definitive): + if definitive: + with self.db: + self.db.execute( + "UPDATE send_attempts SET state='failed' WHERE event_id=?", + (event_id,), + ) + + def delivery_status(self, event_id): + reference = self.reference(event_id) + if reference: + return {"state": "provider_accepted", "reference": reference} + row = self.db.execute( + "SELECT state FROM send_attempts WHERE event_id=?", (event_id,) + ).fetchone() + return { + "state": ( + "unknown" + if row and row[0] == "in_flight" + else "failed" if row else "not_found" + ) + } + def reference(self, event_id: str) -> str | None: row = self.db.execute( "SELECT provider_ref FROM deliveries WHERE event_id=?", (event_id,) @@ -64,7 +112,9 @@ 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( @@ -79,7 +129,9 @@ class SQLiteDeliveryStore: (email.lower(), reason, datetime.now(timezone.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: request_id = f"vrq_{secrets.token_hex(12)}" with self.db: self.db.execute( @@ -177,63 +229,111 @@ class SMTPProvider: ) self.security = security + def check(self): + """Authenticate the configured SMTP lane without submitting a message.""" + with smtplib.SMTP(self.host, self.port, timeout=5) as smtp: + if self.security == "starttls": + smtp.starttls(context=ssl.create_default_context()) + smtp.login(self.username, self.password) + smtp.noop() + def send(self, recipient: str, subject: str, text: str) -> str: message = EmailMessage() - message["From"], message["To"], message["Subject"] = self.sender, recipient, subject + message["From"], message["To"], message["Subject"] = ( + self.sender, + recipient, + subject, + ) # RFC 5322 wants a Message-ID on outgoing mail, and it is the only handle # 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 + + def failure(code, *, retryable): + error = ProviderError(code, retryable=retryable) + error.definitive = not submitted or ( + not accepted and code in {"temporary_deferral", "permanent_rejection"} + ) + return error + try: with smtplib.SMTP(self.host, self.port, timeout=10) as smtp: if self.security == "starttls": - smtp.starttls() + smtp.starttls(context=ssl.create_default_context()) smtp.login(self.username, self.password) + submitted = True smtp.send_message(message) + accepted = True except TimeoutError as exc: - raise ProviderError("provider_timeout", retryable=True) from exc + raise failure("provider_timeout", retryable=True) from exc except smtplib.SMTPRecipientsRefused as exc: # 5xx recipient refusals are permanent; never surface SMTP text. - raise ProviderError("permanent_rejection", retryable=False) from exc + raise failure("permanent_rejection", retryable=False) from exc except smtplib.SMTPResponseException as exc: code = int(getattr(exc, "smtp_code", 0) or 0) if 400 <= code < 500: - raise ProviderError("temporary_deferral", retryable=True) from exc + raise failure("temporary_deferral", retryable=True) from exc if 500 <= code < 600: - raise ProviderError("permanent_rejection", retryable=False) from exc - raise ProviderError("provider_unavailable", retryable=True) from exc + raise failure("permanent_rejection", retryable=False) from exc + raise failure("provider_unavailable", retryable=True) from exc except (OSError, smtplib.SMTPException) as exc: - raise ProviderError("provider_unavailable", retryable=True) from exc + raise failure("provider_unavailable", retryable=True) from exc return message["Message-ID"] or f"smtp:{abs(hash((recipient, subject)))}" class TransactionalApplication: - def __init__(self, store: SQLiteDeliveryStore, provider, bearer_token: str, portal_url: str) -> None: + def __init__( + self, store: SQLiteDeliveryStore, provider, bearer_token: str, portal_url: str + ) -> None: self.store, self.provider, self.token = store, provider, bearer_token self.portal_url = portal_url.rstrip("/") + self._status_cache = None + self._status_time = 0.0 def __call__(self, environ, start_response): if environ.get("PATH_INFO") in ("/healthz", "/readyz"): return self._json(start_response, HTTPStatus.OK, {"status": "ok"}) path = environ.get("PATH_INFO") - if path not in ( - "/v1/send", - "/v1/registration-verifications", - "/v1/registration-verifications/consume", - "/v1/registration-verifications/cancel", - ) or environ.get("REQUEST_METHOD") != "POST": - return self._json(start_response, HTTPStatus.NOT_FOUND, {"error": "not_found"}) + if ( + path + not in ( + "/v1/send", + "/v1/status", + "/v1/delivery-status", + "/v1/registration-verifications", + "/v1/registration-verifications/consume", + "/v1/registration-verifications/cancel", + ) + or environ.get("REQUEST_METHOD") != "POST" + ): + 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: raise ValueError("invalid_size") payload = json.loads(environ["wsgi.input"].read(length)) + if path == "/v1/status": + return self._status(start_response) + if path == "/v1/delivery-status": + return self._json( + start_response, + HTTPStatus.OK, + self.store.delivery_status(str(payload["event_id"])), + ) if path == "/v1/registration-verifications": return self._request_verification(start_response, payload) if path == "/v1/registration-verifications/consume": @@ -243,7 +343,9 @@ 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, @@ -251,7 +353,9 @@ 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( @@ -260,31 +364,77 @@ class TransactionalApplication: {"error": "provider_unavailable", "retryable": True}, ) + def _status(self, start_response): + if self._status_cache is None or time.monotonic() - self._status_time >= 30: + checks = {"database": "failed", "smtp_authentication": "failed"} + try: + with self.store.lock: + self.store.db.execute("SELECT 1").fetchone() + checks["database"] = "ok" + except Exception: + pass + try: + self.provider.check() + checks["smtp_authentication"] = "ok" + except Exception: + pass + self._status_cache = { + "checks": checks, + "checked_at": datetime.now(timezone.utc).isoformat(), + "message_sent": False, + "inbox_receipt_verified": False, + } + self._status_time = time.monotonic() + return self._json(start_response, HTTPStatus.OK, self._status_cache) + def _send_invitation(self, start_response, environ, payload): 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): raise ValueError("invalid_recipient") if self.store.is_suppressed(recipient): raise ValueError("recipient_suppressed") - existing = self.store.reference(event_id) - if existing: - return self._json( - start_response, HTTPStatus.OK, {"status": "duplicate", "reference": existing} - ) invitation_id = str(payload.get("data", {}).get("invitation_id", "")) if not invitation_id: raise ValueError("invitation_id_required") - reference = self.provider.send( - recipient, - "Your NetKingdom invitation", - f"You have been invited. Continue securely at {self.portal_url}/invitations/{invitation_id}\n", - ) - self.store.record(event_id, reference) + fingerprint = hashlib.sha256( + json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() + ).hexdigest() + if not self.store.reserve_send(event_id, fingerprint): + return self._json( + start_response, + HTTPStatus.OK, + {"status": "duplicate", "reference": self.store.reference(event_id)}, + ) + try: + reference = self.provider.send( + recipient, + "Your NetKingdom invitation", + f"You have been invited. Continue securely at {self.portal_url}/invitations/{invitation_id}\n", + ) + self.store.record(event_id, reference) + except ProviderError as exc: + definitive = getattr( + exc, + "definitive", + exc.code in {"temporary_deferral", "permanent_rejection"}, + ) + self.store.send_failed(event_id, definitive) + if not definitive: + raise ProviderError( + "delivery_outcome_unknown", retryable=False + ) from None + raise + except Exception: + # Preserve the durable reservation: SMTP may have accepted the message. + raise ProviderError("delivery_outcome_unknown", retryable=False) from None # Provider acceptance is transport evidence only — not identity or auth. return self._json( start_response, @@ -326,14 +476,18 @@ class TransactionalApplication: "These links expire in 30 minutes.\n", ) return self._json( - start_response, HTTPStatus.ACCEPTED, {"request_id": request_id, "accepted": True} + start_response, + HTTPStatus.ACCEPTED, + {"request_id": request_id, "accepted": True}, ) def _consume_verification(self, start_response, payload): handle = str(payload.get("handle") or "") if len(handle) < 32: raise ValueError("verification_invalid") - evidence = self.store.consume_verification(hashlib.sha256(handle.encode()).hexdigest()) + evidence = self.store.consume_verification( + hashlib.sha256(handle.encode()).hexdigest() + ) # mailbox_control is channel evidence only; never an authorization decision. return self._json( start_response, @@ -353,7 +507,9 @@ class TransactionalApplication: handle = str(payload.get("handle") or "") if len(handle) < 32: raise ValueError("verification_invalid") - evidence = self.store.cancel_verification(hashlib.sha256(handle.encode()).hexdigest()) + evidence = self.store.cancel_verification( + hashlib.sha256(handle.encode()).hexdigest() + ) return self._json( start_response, HTTPStatus.OK, @@ -392,7 +548,12 @@ 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: @@ -405,10 +566,14 @@ def main() -> None: security=os.environ.get("EMAIL_CONNECT_SMTP_SECURITY", "starttls"), ) app = TransactionalApplication( - SQLiteDeliveryStore(os.environ.get("EMAIL_CONNECT_DATABASE_PATH", "/data/email-connect.db")), + 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"], ) - with make_server("0.0.0.0", int(os.environ.get("EMAIL_CONNECT_HTTP_PORT", "8080")), app) as server: + with make_server( + "0.0.0.0", int(os.environ.get("EMAIL_CONNECT_HTTP_PORT", "8080")), app + ) as server: server.serve_forever() diff --git a/tests/test_transactional.py b/tests/test_transactional.py index 5cf388b..c04043f 100644 --- a/tests/test_transactional.py +++ b/tests/test_transactional.py @@ -51,7 +51,9 @@ def invoke(app, payload, token="opaque", key=None, path="/v1/send"): "HTTP_AUTHORIZATION": f"Bearer {token}", } if path == "/v1/send": - env["HTTP_IDEMPOTENCY_KEY"] = key if key is not None else payload.get("id", "evt-1") + env["HTTP_IDEMPOTENCY_KEY"] = ( + key if key is not None else payload.get("id", "evt-1") + ) body = b"".join(app(env, lambda s, h: result.update(status=s))) return result["status"], json.loads(body) @@ -59,7 +61,10 @@ def invoke(app, payload, token="opaque", key=None, path="/v1/send"): def test_sends_fixed_template_once(tmp_path): provider = Provider() app = TransactionalApplication( - SQLiteDeliveryStore(str(tmp_path / "mail.db")), provider, "opaque", "https://users.example" + SQLiteDeliveryStore(str(tmp_path / "mail.db")), + provider, + "opaque", + "https://users.example", ) status, body = invoke(app, event()) assert status.startswith("202") @@ -71,7 +76,10 @@ def test_sends_fixed_template_once(tmp_path): def test_rejects_auth_recipient_template_and_key(tmp_path): app = TransactionalApplication( - SQLiteDeliveryStore(str(tmp_path / "mail.db")), Provider(), "opaque", "https://users.example" + SQLiteDeliveryStore(str(tmp_path / "mail.db")), + Provider(), + "opaque", + "https://users.example", ) assert invoke(app, event(), token="bad")[0].startswith("401") bad = event() @@ -86,7 +94,10 @@ def test_rejects_auth_recipient_template_and_key(tmp_path): def test_resend_is_separate_event_and_sends_again(tmp_path): provider = Provider() app = TransactionalApplication( - SQLiteDeliveryStore(str(tmp_path / "mail.db")), provider, "opaque", "https://users.example" + SQLiteDeliveryStore(str(tmp_path / "mail.db")), + provider, + "opaque", + "https://users.example", ) assert invoke(app, event("evt-1", "family_member.invited"))[0].startswith("202") status, body = invoke(app, event("evt-2", "family_invitation.resent")) @@ -141,7 +152,7 @@ def test_smtp_provider_maps_recipient_refused_to_permanent(): def __exit__(self, *a): return False - def starttls(self): + def starttls(self, *, context=None): return None def login(self, *a): @@ -251,7 +262,9 @@ def test_registration_cancellation_is_single_use_and_prevents_verification(tmp_p "wsgi.input": io.BytesIO(payload), "HTTP_AUTHORIZATION": "Bearer opaque", } - canceled = json.loads(b"".join(app(cancel_env, lambda s, h: result.update(status=s)))) + canceled = json.loads( + b"".join(app(cancel_env, lambda s, h: result.update(status=s))) + ) assert canceled["purpose"] == "public-registration-cancel" assert canceled["authorization"] is False consume_env = { @@ -261,7 +274,9 @@ def test_registration_cancellation_is_single_use_and_prevents_verification(tmp_p "wsgi.input": io.BytesIO(payload), "HTTP_AUTHORIZATION": "Bearer opaque", } - rejected = json.loads(b"".join(app(consume_env, lambda s, h: result.update(status=s)))) + rejected = json.loads( + b"".join(app(consume_env, lambda s, h: result.update(status=s))) + ) assert rejected["error"] == "verification_invalid" cancel_env["wsgi.input"] = io.BytesIO(payload) replay = json.loads(b"".join(app(cancel_env, lambda s, h: result.update(status=s)))) @@ -272,10 +287,15 @@ def test_mailbox_evidence_is_not_authorization(tmp_path): """Provider acceptance and mailbox challenge never grant user-engine authority.""" provider = Provider() app = TransactionalApplication( - SQLiteDeliveryStore(str(tmp_path / "mail.db")), provider, "opaque", "https://users.example" + SQLiteDeliveryStore(str(tmp_path / "mail.db")), + provider, + "opaque", + "https://users.example", ) _, accepted = invoke(app, event()) - assert accepted.get("authorization") is None or accepted.get("authorization") is False + assert ( + accepted.get("authorization") is None or accepted.get("authorization") is False + ) assert accepted["evidence_ceiling"] == "provider_accepted" # Ceiling is explicit; user-engine must not elevate this to authz. assert "authorized" not in accepted @@ -299,7 +319,8 @@ class RecordingSMTP: def __exit__(self, *exc): return False - def starttls(self): + def starttls(self, *, context=None): + assert context.check_hostname self.starttls_calls += 1 def login(self, username, password): @@ -312,7 +333,9 @@ class RecordingSMTP: def test_smtp_provider_defaults_to_starttls(monkeypatch): monkeypatch.setattr(smtplib, "SMTP", RecordingSMTP) RecordingSMTP.instances = [] - provider = SMTPProvider("smtp.example.com", 587, "user", "pass", "noreply@example.com") + provider = SMTPProvider( + "smtp.example.com", 587, "user", "pass", "noreply@example.com" + ) assert provider.security == "starttls" provider.send("person@example.test", "Subject", "body") @@ -322,7 +345,9 @@ def test_smtp_provider_defaults_to_starttls(monkeypatch): def test_plaintext_security_skips_starttls_on_loopback(monkeypatch): monkeypatch.setattr(smtplib, "SMTP", RecordingSMTP) RecordingSMTP.instances = [] - provider = SMTPProvider("127.0.0.1", 3025, "user", "pass", "noreply@example.com", security="plaintext") + provider = SMTPProvider( + "127.0.0.1", 3025, "user", "pass", "noreply@example.com", security="plaintext" + ) provider.send("person@example.test", "Subject", "body") smtp = RecordingSMTP.instances[0] @@ -331,17 +356,142 @@ def test_plaintext_security_skips_starttls_on_loopback(monkeypatch): assert len(smtp.sent) == 1 -@pytest.mark.parametrize("host", ["127.0.0.1", "127.0.1.1", "::1", "[::1]", "localhost", "LOCALHOST"]) +@pytest.mark.parametrize( + "host", ["127.0.0.1", "127.0.1.1", "::1", "[::1]", "localhost", "LOCALHOST"] +) def test_plaintext_security_is_allowed_for_loopback_hosts(host): - assert SMTPProvider(host, 3025, "user", "pass", "noreply@example.com", security="plaintext") + assert SMTPProvider( + host, 3025, "user", "pass", "noreply@example.com", security="plaintext" + ) -@pytest.mark.parametrize("host", ["smtp.example.com", "10.0.0.5", "::ffff:10.0.0.5", "", "localhost.example.com"]) +@pytest.mark.parametrize( + "host", + ["smtp.example.com", "10.0.0.5", "::ffff:10.0.0.5", "", "localhost.example.com"], +) def test_plaintext_security_is_refused_for_non_loopback_hosts(host): with pytest.raises(ValueError, match="loopback"): - SMTPProvider(host, 3025, "user", "pass", "noreply@example.com", security="plaintext") + SMTPProvider( + host, 3025, "user", "pass", "noreply@example.com", security="plaintext" + ) def test_unsupported_security_mode_is_refused(): with pytest.raises(ValueError, match="security mode"): - SMTPProvider("127.0.0.1", 3025, "user", "pass", "noreply@example.com", security="none") + SMTPProvider( + "127.0.0.1", 3025, "user", "pass", "noreply@example.com", security="none" + ) + + +def test_status_authenticates_without_sending_and_redacts_errors(tmp_path): + class Diagnostic(Provider): + def check(self): + pass + + provider = Diagnostic() + app = TransactionalApplication( + SQLiteDeliveryStore(str(tmp_path / "status.db")), + provider, + "opaque", + "https://users.example", + ) + assert invoke(app, {}, path="/v1/status", token="bad")[0].startswith("401") + status, body = invoke(app, {}, path="/v1/status") + assert body["checks"] == {"database": "ok", "smtp_authentication": "ok"} + assert body["message_sent"] is False and provider.calls == [] + + def unavailable(): + raise RuntimeError("private-smtp-password") + + provider.check = unavailable + app._status_cache = None + body = invoke(app, {}, path="/v1/status")[1] + assert body["checks"][ + "smtp_authentication" + ] == "failed" and "private-smtp" not in json.dumps(body) + + +def test_lost_post_send_record_never_resends_after_restart(tmp_path): + path = str(tmp_path / "lost.db") + store = SQLiteDeliveryStore(path) + provider = Provider() + app = TransactionalApplication(store, provider, "opaque", "https://users.example") + + def failed_write(*args): + raise RuntimeError("private-db-error") + + store.record = failed_write + assert invoke(app, event())[1]["error"] == "delivery_outcome_unknown" + app = TransactionalApplication( + SQLiteDeliveryStore(path), provider, "opaque", "https://users.example" + ) + assert invoke(app, event())[1]["error"] == "delivery_outcome_unknown" + assert len(provider.calls) == 1 + assert ( + invoke(app, {"event_id": "evt-1"}, path="/v1/delivery-status")[1]["state"] + == "unknown" + ) + + +def test_definitive_failure_retries_but_changed_payload_is_rejected(tmp_path): + provider = FailingProvider(ProviderError("temporary_deferral", retryable=True)) + app = TransactionalApplication( + SQLiteDeliveryStore(str(tmp_path / "retry.db")), + provider, + "opaque", + "https://users.example", + ) + assert invoke(app, event())[0].startswith("503") + app.provider = Provider() + assert invoke(app, event())[0].startswith("202") + assert ( + invoke(app, event(email="different@example.test"))[1]["error"] + == "idempotency_payload_mismatch" + ) + assert len(app.provider.calls) == 1 + + +def test_disconnect_after_smtp_acceptance_never_releases_reservation( + tmp_path, monkeypatch +): + import smtplib + + class SMTP: + sends = 0 + + def __init__(self, *args, **kwargs): + pass + + def __enter__(self): + return self + + def __exit__(self, *args): + raise smtplib.SMTPResponseException(550, b"private quit failure") + + def login(self, *args): + pass + + def send_message(self, message): + self.__class__.sends += 1 + + monkeypatch.setattr("email_connect.transactional.smtplib.SMTP", SMTP) + from email_connect.transactional import SMTPProvider + + app = TransactionalApplication( + SQLiteDeliveryStore(str(tmp_path / "quit.db")), + SMTPProvider( + "127.0.0.1", + 3025, + "fixture", + "fixture", + "from@example.test", + security="plaintext", + ), + "opaque", + "https://users.example", + ) + for _ in range(2): + status, body = invoke(app, event()) + assert body["error"] == "delivery_outcome_unknown" + assert "private" not in str(body) + assert SMTP.sends == 1