"""Message providers (task 4.1-4.2, design D9, spec messaging). v1 sends email only. The interface stays channel-shaped so v2 adds WhatsApp and SMS as new implementations rather than as a rewrite of the pipeline. `DryRunProvider` wraps whatever is configured whenever DRY_RUN is set. It renders and logs exactly what would be sent without calling anyone's API, which is what makes it safe to rehearse a full reminder cycle against real guest data. """ import base64 import binascii import hashlib import hmac import logging import time from dataclasses import dataclass from typing import Any, Protocol import httpx from app.config import get_settings from app.models.enums import Channel logger = logging.getLogger(__name__) class ProviderNotConfiguredError(RuntimeError): """Raised when a real send is attempted with no credentials. Named separately so the failure reaches an admin as "EMAIL_PROVIDER_API_KEY is not set" rather than as a 401 from a vendor they have never heard of. Only reachable with dry-run off — a dev stack rehearses without a key by design. """ def __init__(self, setting: str = "EMAIL_PROVIDER_API_KEY") -> None: super().__init__( f"Email sending is not configured: {setting} is not set. " "Set it, or enable DRY_RUN to rehearse without sending." ) self.setting = setting @dataclass(frozen=True) class SendRequest: to: str subject: str | None html: str text: str idempotency_key: str #: Where a reply goes — the couple, when they have given an address (D9). The `From` #: stays the platform sender either way, because that is the domain SPF and DKIM #: authorise; putting a Gmail address there fails authentication and lands invitations #: in spam. reply_to: str | None = None @dataclass(frozen=True) class SendResult: provider_message_id: str status: str # "sent" | "queued" @dataclass(frozen=True) class DeliveryUpdate: """Normalised webhook event, so callers never parse vendor payloads themselves.""" provider_message_id: str status: str # delivered | failed | bounced | complained reason: str | None = None hard_bounce: bool = False class MessageProvider(Protocol): channel: Channel async def send(self, request: SendRequest) -> SendResult: ... def parse_webhook(self, payload: Any) -> list[DeliveryUpdate]: ... def verify_signature(self, body: bytes, headers: dict[str, str]) -> bool: ... class ResendEmailProvider: """Resend's HTTP API. Chosen for setup speed; SES swaps in behind this interface.""" channel = Channel.EMAIL BASE_URL = "https://api.resend.com" def __init__(self, api_key: str, sender: str) -> None: self._api_key = api_key self._sender = sender async def send(self, request: SendRequest) -> SendResult: if not self._api_key: raise ProviderNotConfiguredError() payload: dict[str, Any] = { "from": self._sender, "to": [request.to], "subject": request.subject or "", "html": request.html, "text": request.text, } if request.reply_to: payload["reply_to"] = request.reply_to async with httpx.AsyncClient(timeout=20) as client: response = await client.post( f"{self.BASE_URL}/emails", headers={ "Authorization": f"Bearer {self._api_key}", # Lets the vendor collapse a retry into the original send. "Idempotency-Key": request.idempotency_key, }, json=payload, ) response.raise_for_status() return SendResult(provider_message_id=response.json().get("id", ""), status="sent") def verify_signature(self, body: bytes, headers: dict[str, str]) -> bool: """Reject anything unsigned, misdated, or signed with the wrong key. Resend delivers through Svix, which signs `{id}.{timestamp}.{body}` rather than the body alone. Signing the body alone would be a working signature check that is still replayable forever: an attacker who captures one legitimate `email.bounced` webhook could re-post it whenever they liked and permanently silence that guest. Binding the timestamp into the signed payload — and then bounding it — is what closes that. """ settings = get_settings() secret = settings.email_webhook_secret if not secret: # No secret configured means no way to tell a real webhook from a forged one. logger.warning("EMAIL_WEBHOOK_SECRET is unset; rejecting the delivery webhook") return False # Header lookup is case-insensitive because Starlette lower-cases but a direct # dict from a test or another caller may not. lowered = {k.lower(): v for k, v in headers.items()} message_id = lowered.get("svix-id", "") timestamp = lowered.get("svix-timestamp", "") signature_header = lowered.get("svix-signature", "") if not (message_id and timestamp and signature_header): return False try: sent_at = int(timestamp) except ValueError: return False drift = abs(time.time() - sent_at) if drift > settings.webhook_tolerance_seconds: logger.warning("rejected a delivery webhook %.0fs outside the replay window", drift) return False expected = hmac.new( self._webhook_key(secret), f"{message_id}.{timestamp}.".encode() + body, hashlib.sha256, ).digest() expected_b64 = base64.b64encode(expected).decode() # Svix sends a space-separated list so a secret can be rotated without downtime; # any one matching is a valid signature. for candidate in signature_header.split(" "): _, _, value = candidate.partition(",") if value and hmac.compare_digest(expected_b64, value): return True return False @staticmethod def _webhook_key(secret: str) -> bytes: """Svix secrets are `whsec_` + base64. A raw string is accepted too, so a self-hosted or test signer does not need the vendor's prefix.""" raw = secret.removeprefix("whsec_") try: return base64.b64decode(raw, validate=True) except (ValueError, binascii.Error): return secret.encode() def parse_webhook(self, payload: Any) -> list[DeliveryUpdate]: if not isinstance(payload, dict): return [] event = str(payload.get("type", "")) data = payload.get("data") or {} message_id = str(data.get("email_id") or data.get("id") or "") if not message_id: return [] mapping = { "email.delivered": ("delivered", False), "email.bounced": ("failed", True), "email.complained": ("failed", True), "email.delivery_delayed": ("failed", False), } if event not in mapping: return [] status, hard = mapping[event] return [ DeliveryUpdate( provider_message_id=message_id, status=status, reason=event, hard_bounce=hard, ) ] class DryRunProvider: """Renders and logs; never calls out. Wraps the real provider rather than replacing it, so the code path under test is the same one production uses.""" def __init__(self, inner: MessageProvider) -> None: self._inner = inner self.channel = inner.channel self.sent: list[SendRequest] = [] async def send(self, request: SendRequest) -> SendResult: self.sent.append(request) logger.info( "DRY RUN — would send to %s | subject=%r | %d chars", request.to, request.subject, len(request.text), ) return SendResult(provider_message_id=f"dryrun-{request.idempotency_key}", status="sent") def verify_signature(self, body: bytes, headers: dict[str, str]) -> bool: return self._inner.verify_signature(body, headers) def parse_webhook(self, payload: Any) -> list[DeliveryUpdate]: return self._inner.parse_webhook(payload) _registry: dict[Channel, MessageProvider] = {} def get_provider(channel: Channel = Channel.EMAIL) -> MessageProvider: """Resolve the provider for a channel, honouring dry-run.""" if channel in _registry: return _registry[channel] settings = get_settings() if channel is not Channel.EMAIL: # v2 registers WhatsApp and SMS here. Failing loudly beats sending nowhere. raise NotImplementedError(f"{channel} is not available in v1 (email only)") provider: MessageProvider = ResendEmailProvider( api_key=settings.email_provider_api_key, sender=settings.email_from ) if settings.dry_run: provider = DryRunProvider(provider) _registry[channel] = provider return provider def reset_registry() -> None: """Test hook — the registry caches by channel and settings are cached too.""" _registry.clear()