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

Complete EMAIL-WP-0004 transactional invitation mail delivery.
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
Details
CI Smoke / container-smoke (push) Successful in 1s
Details

Browse source 
Ship the railiance01 deploy package (OpenBao/ESO custody, NetworkPolicy,
probes), provider failure classification and suppression, T04 unit and live
proof, and non-secret NK-WP-0024 hand-back evidence. Workplan finished.
This commit is contained in:
tegwick 2026-08-12 13:32:11 +02:00
parent 5d1c5e1c5c
commit 04897a149f
18 changed files with 1269 additions and 117 deletions
Download patch file Download diff file Expand all files Collapse all files

307
src/email_connect/transactional.py
View file

@ -19,54 +19,109 @@ from wsgiref.simple_server import make_server
ALLOWED_EVENTS = {"family_member.invited", "family_invitation.resent"}
class ProviderError(Exception):
"""Provider outcome with stable redacted code and retry classification."""
def __init__(self, code: str, *, retryable: bool) -> None:
super().__init__(code)
self.code = code
self.retryable = retryable
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 (
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, canceled_at TEXT
)""")
)"""
)
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)")}
if "canceled_at" not in columns:
self.db.execute("ALTER TABLE verifications ADD COLUMN canceled_at TEXT")
self.db.commit()
def reference(self, event_id: str) -> str | None:
row = self.db.execute("SELECT provider_ref FROM deliveries WHERE event_id=?", (event_id,)).fetchone()
row = self.db.execute(
"SELECT provider_ref FROM deliveries WHERE event_id=?", (event_id,)
).fetchone()
return row[0] if row else None
def record(self, event_id: str, reference: str) -> None:
with self.db:
self.db.execute("INSERT INTO deliveries VALUES (?, ?)", (event_id, reference))
def is_suppressed(self, email: str) -> bool:
row = self.db.execute(
"SELECT 1 FROM suppressions WHERE email=?", (email.lower(),)
).fetchone()
return row is not None
def suppress(self, email: str, reason: str = "manual") -> None:
with self.db:
self.db.execute(
"INSERT OR REPLACE INTO suppressions VALUES (?,?,?)",
(email.lower(), reason, datetime.now(timezone.utc).isoformat()),
)
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
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)""", (
request_id, handle_hash, payload["registration_id"], payload["normalized_email"],
payload["preferred_username"], payload["client_id"], payload["tenant"],
payload.get("display_name"), expires_at,
))
VALUES (?,?,?,?,?,?,?,?,?,NULL,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()
row = self.db.execute(
"SELECT * FROM verifications WHERE handle_hash=?", (handle_hash,)
).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):
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]}
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],
}
def cancel_verification(self, handle_hash: str) -> dict:
with self.lock, self.db:
@ -81,22 +136,50 @@ class SQLiteDeliveryStore:
"UPDATE verifications SET canceled_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]}
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:
self.host, self.port, self.username, self.password, self.sender = host, port, username, password, sender
self.host, self.port, self.username, self.password, self.sender = (
host,
port,
username,
password,
sender,
)
def send(self, recipient: str, subject: str, text: str) -> str:
message = EmailMessage()
message["From"], message["To"], message["Subject"] = self.sender, recipient, subject
message.set_content(text)
with smtplib.SMTP(self.host, self.port, timeout=10) as smtp:
smtp.starttls()
smtp.login(self.username, self.password)
smtp.send_message(message)
try:
with smtplib.SMTP(self.host, self.port, timeout=10) as smtp:
smtp.starttls()
smtp.login(self.username, self.password)
smtp.send_message(message)
except TimeoutError as exc:
raise ProviderError("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
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
if 500 <= code < 600:
raise ProviderError("permanent_rejection", retryable=False) from exc
raise ProviderError("provider_unavailable", retryable=True) from exc
except (OSError, smtplib.SMTPException) as exc:
raise ProviderError("provider_unavailable", retryable=True) from exc
return message["Message-ID"] or f"smtp:{abs(hash((recipient, subject)))}"
@ -107,15 +190,23 @@ 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"})
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 not hmac.compare_digest(str(environ.get("HTTP_AUTHORIZATION", "")), f"Bearer {self.token}"):
return self._json(start_response, HTTPStatus.UNAUTHORIZED, {"error":"unauthorized"})
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 not hmac.compare_digest(
str(environ.get("HTTP_AUTHORIZATION", "")), f"Bearer {self.token}"
):
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")
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)
@ -123,71 +214,141 @@ class TransactionalApplication:
return self._consume_verification(start_response, payload)
if path == "/v1/registration-verifications/cancel":
return self._cancel_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")
recipient = str(payload.get("data", {}).get("primary_email", ""))
if not _valid_address(recipient): raise ValueError("invalid_recipient")
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)
return self._send_invitation(start_response, environ, payload)
except ProviderError as exc:
status = (
HTTPStatus.SERVICE_UNAVAILABLE if exc.retryable else HTTPStatus.UNPROCESSABLE_ENTITY
)
return self._json(
start_response,
status,
{"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):
return self._json(start_response, HTTPStatus.SERVICE_UNAVAILABLE, {"error":"provider_unavailable","retryable":True})
return self._json(start_response, HTTPStatus.ACCEPTED, {"status":"accepted","reference":reference})
# Safety net: never leak provider exception text.
return self._json(
start_response,
HTTPStatus.SERVICE_UNAVAILABLE,
{"error": "provider_unavailable", "retryable": True},
)
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:
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)
# Provider acceptance is transport evidence only — not identity or auth.
return self._json(
start_response,
HTTPStatus.ACCEPTED,
{
"status": "accepted",
"reference": reference,
"event_id": event_id,
"evidence_ceiling": "provider_accepted",
},
)
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"])):
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")
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)
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",
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"
"These links expire in 30 minutes.\n",
)
return self._json(start_response, HTTPStatus.ACCEPTED,
{"request_id":request_id,"accepted":True})
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},
})
# mailbox_control is channel evidence only; never an authorization decision.
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},
"authorization": False,
"evidence_ceiling": "mailbox_challenge_consumed",
},
)
def _cancel_verification(self, start_response, payload):
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,
{
"purpose": "public-registration-cancel",
"verification_id": f"fvc_{secrets.token_hex(12)}",
**evidence,
"source_system": "email-connect",
"assurance": {"mailbox_control": True},
"authorization": False,
"evidence_ceiling": "mailbox_challenge_canceled",
},
)
return self._json(start_response, HTTPStatus.OK, {
"purpose":"public-registration-cancel",
"verification_id":f"fvc_{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)))])
body = json.dumps(payload).encode()
start_response(
f"{status.value} {status.phrase}",
[("Content-Type", "application/json"), ("Content-Length", str(len(body)))],
)
return [body]
@ -197,8 +358,18 @@ def _valid_address(value: str) -> bool:
def main() -> None:
provider = SMTPProvider(os.environ["EMAIL_CONNECT_SMTP_HOST"], int(os.environ.get("EMAIL_CONNECT_SMTP_PORT", "587")),
os.environ["EMAIL_CONNECT_SMTP_USERNAME"], os.environ["EMAIL_CONNECT_SMTP_PASSWORD"], os.environ["EMAIL_CONNECT_SENDER"])
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"])
with make_server("0.0.0.0", int(os.environ.get("EMAIL_CONNECT_HTTP_PORT", "8080")), app) as server: server.serve_forever()
provider = SMTPProvider(
os.environ["EMAIL_CONNECT_SMTP_HOST"],
int(os.environ.get("EMAIL_CONNECT_SMTP_PORT", "587")),
os.environ["EMAIL_CONNECT_SMTP_USERNAME"],
os.environ["EMAIL_CONNECT_SMTP_PASSWORD"],
os.environ["EMAIL_CONNECT_SENDER"],
)
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"],
)
with make_server("0.0.0.0", int(os.environ.get("EMAIL_CONNECT_HTTP_PORT", "8080")), app) as server:
server.serve_forever()