email-connect/src/email_connect/transactional.py

"""Narrow authenticated transactional mail receiver for user-engine."""

from __future__ import annotations

import hmac
import json
import os
import smtplib
import sqlite3
from email.message import EmailMessage
from email.utils import parseaddr
from http import HTTPStatus
from wsgiref.simple_server import make_server

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.db.execute("CREATE TABLE IF NOT EXISTS deliveries (event_id TEXT PRIMARY KEY, provider_ref TEXT NOT NULL)")
        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()
        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))


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

    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)
        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:
        self.store, self.provider, self.token = store, provider, bearer_token
        self.portal_url = portal_url.rstrip("/")

    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":
            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")
            payload = json.loads(environ["wsgi.input"].read(length))
            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)
        except (KeyError, TypeError, ValueError, json.JSONDecodeError) as 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})

    @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)))])
        return [body]


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


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