From 752d91cf5cc37e6661a2f51b043e4d112e9242d3 Mon Sep 17 00:00:00 2001 From: tegwick Date: Mon, 10 Aug 2026 12:29:55 +0200 Subject: [PATCH] Add single-use registration verification --- src/email_connect/transactional.py | 67 ++++++++++++++++++- tests/test_transactional.py | 15 +++++ ...004-transactional-mail-delivery-service.md | 6 +- 3 files changed, 86 insertions(+), 2 deletions(-) diff --git a/src/email_connect/transactional.py b/src/email_connect/transactional.py index 6804d55..804e391 100644 --- a/src/email_connect/transactional.py +++ b/src/email_connect/transactional.py @@ -3,13 +3,17 @@ from __future__ import annotations import hmac +import hashlib import json import os import smtplib import sqlite3 +import secrets +from datetime import datetime, timedelta, timezone from email.message import EmailMessage from email.utils import parseaddr from http import HTTPStatus +from threading import RLock from wsgiref.simple_server import make_server ALLOWED_EVENTS = {"family_member.invited", "family_invitation.resent"} @@ -18,7 +22,14 @@ ALLOWED_EVENTS = {"family_member.invited", "family_invitation.resent"} class SQLiteDeliveryStore: def __init__(self, path: str) -> None: 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)") + 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 + )""") self.db.commit() def reference(self, event_id: str) -> str | None: @@ -29,6 +40,28 @@ class SQLiteDeliveryStore: with self.db: self.db.execute("INSERT INTO deliveries VALUES (?, ?)", (event_id, reference)) + 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("INSERT INTO verifications VALUES (?,?,?,?,?,?,?,?,?,NULL)", ( + request_id, handle_hash, payload["registration_id"], payload["normalized_email"], + payload["preferred_username"], payload["client_id"], payload["tenant"], + payload.get("display_name"), expires_at, + )) + return request_id + + def consume_verification(self, handle_hash: str) -> dict: + with self.lock, self.db: + row = self.db.execute("SELECT * FROM verifications WHERE handle_hash=?", (handle_hash,)).fetchone() + if row is None or row[9] is not None: + raise ValueError("verification_invalid") + if datetime.fromisoformat(row[8]) <= datetime.now(timezone.utc): + raise ValueError("verification_expired") + self.db.execute("UPDATE verifications SET consumed_at=? WHERE request_id=?", + (datetime.now(timezone.utc).isoformat(), row[0])) + return {"request_id":row[0],"registration_id":row[2],"email":row[3], + "preferred_username":row[4],"client_id":row[5],"tenant":row[6],"display_name":row[7]} + class SMTPProvider: def __init__(self, host: str, port: int, username: str, password: str, sender: str) -> None: @@ -53,7 +86,8 @@ class TransactionalApplication: def __call__(self, environ, start_response): if environ.get("PATH_INFO") in ("/healthz", "/readyz"): return self._json(start_response, HTTPStatus.OK, {"status":"ok"}) - if environ.get("PATH_INFO") != "/v1/send" or environ.get("REQUEST_METHOD") != "POST": + path = environ.get("PATH_INFO") + if path not in ("/v1/send", "/v1/registration-verifications", "/v1/registration-verifications/consume") 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"}) @@ -61,6 +95,10 @@ class TransactionalApplication: 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/registration-verifications": + return self._request_verification(start_response, payload) + if path == "/v1/registration-verifications/consume": + return self._consume_verification(start_response, 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: raise ValueError("template_not_allowed") @@ -80,6 +118,33 @@ class TransactionalApplication: return self._json(start_response, HTTPStatus.SERVICE_UNAVAILABLE, {"error":"provider_unavailable","retryable":True}) return self._json(start_response, HTTPStatus.ACCEPTED, {"status":"accepted","reference":reference}) + def _request_verification(self, start_response, payload): + required = ("registration_id", "normalized_email", "preferred_username", "client_id", "tenant", "correlation_id") + if any(not payload.get(key) for key in required) or not _valid_address(str(payload["normalized_email"])): + raise ValueError("invalid_verification_request") + handle = secrets.token_urlsafe(32) + expires_at = datetime.now(timezone.utc) + timedelta(minutes=30) + request_id = self.store.create_verification( + payload, hashlib.sha256(handle.encode()).hexdigest(), expires_at.isoformat() + ) + self.provider.send( + str(payload["normalized_email"]), "Verify your NetKingdom registration", + f"Continue securely at {self.portal_url}/registration/verify?handle={handle}\nThis link expires in 30 minutes.\n", + ) + return self._json(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()) + return self._json(start_response, HTTPStatus.OK, { + "purpose":"public-registration", "verification_id":f"fvr_{secrets.token_hex(12)}", + **evidence, "source_system":"email-connect", + "assurance":{"mailbox_control":True}, + }) + @staticmethod def _json(start_response, status, payload): body=json.dumps(payload).encode(); start_response(f"{status.value} {status.phrase}", [("Content-Type","application/json"),("Content-Length",str(len(body)))]) diff --git a/tests/test_transactional.py b/tests/test_transactional.py index 6da206c..3f3c2e5 100644 --- a/tests/test_transactional.py +++ b/tests/test_transactional.py @@ -21,3 +21,18 @@ def test_rejects_auth_recipient_template_and_key(tmp_path): bad=event(); bad["data"]["primary_email"]="bad\n@example.test"; assert invoke(app,bad)[1]["error"]=="invalid_recipient" bad=event(); bad["type"]="arbitrary.send"; assert invoke(app,bad)[1]["error"]=="template_not_allowed" assert invoke(app,event(),key="other")[1]["error"]=="idempotency_key_mismatch" + +def test_registration_verification_is_digest_only_and_single_use(tmp_path): + provider=Provider(); store=SQLiteDeliveryStore(str(tmp_path/"mail.db")); app=TransactionalApplication(store,provider,"opaque","https://users.example") + request={"registration_id":"reg-1","normalized_email":"person@example.test","preferred_username":"person","client_id":"coulomb-social","tenant":"tenant:coulomb","correlation_id":"corr-1"} + raw=json.dumps(request).encode(); result={}; response=json.loads(b"".join(app({"PATH_INFO":"/v1/registration-verifications","REQUEST_METHOD":"POST","CONTENT_LENGTH":str(len(raw)),"wsgi.input":io.BytesIO(raw),"HTTP_AUTHORIZATION":"Bearer opaque"},lambda s,h:result.update(status=s)))) + assert result["status"].startswith("202") and response["accepted"] + text=provider.calls[0][2]; handle=text.split("handle=",1)[1].splitlines()[0] + stored=store.db.execute("SELECT handle_hash FROM verifications").fetchone()[0] + assert handle not in stored and stored + consume=json.dumps({"handle":handle}).encode(); env={"PATH_INFO":"/v1/registration-verifications/consume","REQUEST_METHOD":"POST","CONTENT_LENGTH":str(len(consume)),"wsgi.input":io.BytesIO(consume),"HTTP_AUTHORIZATION":"Bearer opaque"} + evidence=json.loads(b"".join(app(env,lambda s,h:result.update(status=s)))) + assert evidence["purpose"]=="public-registration" and evidence["email"]=="person@example.test" + env["wsgi.input"]=io.BytesIO(consume) + replay=json.loads(b"".join(app(env,lambda s,h:result.update(status=s)))) + assert replay["error"]=="verification_invalid" diff --git a/workplans/EMAIL-WP-0004-transactional-mail-delivery-service.md b/workplans/EMAIL-WP-0004-transactional-mail-delivery-service.md index 5a827f9..016d569 100644 --- a/workplans/EMAIL-WP-0004-transactional-mail-delivery-service.md +++ b/workplans/EMAIL-WP-0004-transactional-mail-delivery-service.md @@ -62,7 +62,11 @@ stable retryable/permanent classifications without leaking SMTP details. Done 2026-08-09: SQLite idempotency, provider-neutral injection, bounded STARTTLS SMTP calls, duplicate suppression, and redacted provider-unavailable -responses are implemented. All 22 repository tests pass. +responses are implemented. The same narrow service now issues public- +registration mailbox challenges: plaintext handles exist only in the message, +SQLite retains a SHA-256 digest, evidence is purpose/binding scoped, expiry is +enforced, and consumption is atomic and single-use. All 23 repository tests +pass. ## T03 - Establish custody and deploy the service