"""Email-related public routes.
Currently:
- GET /email/unsubscribe?token=...
The token is `itsdangerous.URLSafeSerializer` over a small payload,
signed with CASSANDRA_SESSION_SECRET. No auth dependency: the whole
point of one-click unsubscribe is that the user does not have to
sign in.
"""
from __future__ import annotations
from fastapi import APIRouter, Depends, Query, Request
from fastapi.responses import HTMLResponse
from itsdangerous import BadSignature, URLSafeSerializer
from sqlalchemy.ext.asyncio import AsyncSession
from app import branding
from app.config import get_settings
from app.db import get_session
from app.logging import get_logger
from app.models import User
router = APIRouter()
log = get_logger("email_router")
_SALT = "digest-unsubscribe-v1"
def _serializer() -> URLSafeSerializer:
s = get_settings()
if not s.CASSANDRA_SESSION_SECRET:
# In tests with no secret configured, fall back to a constant.
# An empty CASSANDRA_SESSION_SECRET in prod would also break login,
# so this branch is "best-effort dev fallback", not a real prod path.
return URLSafeSerializer("dev-only-empty-secret", salt=_SALT)
return URLSafeSerializer(s.CASSANDRA_SESSION_SECRET, salt=_SALT)
def sign_unsubscribe_token(user_id: int) -> str:
return _serializer().dumps({"uid": int(user_id), "purpose": "digest_optout"})
def verify_unsubscribe_token(token: str) -> int | None:
try:
data = _serializer().loads(token)
except BadSignature:
return None
if not isinstance(data, dict):
return None
if data.get("purpose") != "digest_optout":
return None
try:
return int(data["uid"])
except (KeyError, TypeError, ValueError):
return None
_CONFIRM_PAGE = """\
Unsubscribed — {brand}