coulomb/email-connect
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Implement P05 checked services and safe selected delivery recovery

Browse source 
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a092fe-13b1-7f12-ac74-7d258af4d79c
This commit is contained in:
tegwick 2026-09-13 22:11:49 +02:00
parent 21eebffc56
commit 0fcfddc342
5 changed files with 459 additions and 67 deletions
Download patch file Download diff file Expand all files Collapse all files

263
src/email_connect/transactional.py
View file

@ -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()