101 lines
3.1 KiB
Python
101 lines
3.1 KiB
Python
"""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.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 —
|
|
# NEVER reach production; settings validation should catch this.
|
|
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 = """\
|
|
<!DOCTYPE html>
|
|
<html lang="en">
|
|
<head>
|
|
<meta charset="utf-8">
|
|
<title>Unsubscribed — {brand}</title>
|
|
<link rel="stylesheet" href="/static/css/cassandra.css">
|
|
</head>
|
|
<body class="auth-shell">
|
|
<div class="auth-card" style="max-width:480px;">
|
|
<div class="auth-card__brand">{brand}</div>
|
|
<div class="auth-card__hint">email preferences</div>
|
|
<p class="auth-card__lede">You're unsubscribed from email digests.</p>
|
|
<p style="font-size:13px; color:var(--muted); line-height:1.6;">
|
|
You can re-enable digests any time from
|
|
<a href="/settings" style="color:var(--accent);">Settings</a>.
|
|
</p>
|
|
</div>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
|
|
@router.get("/email/unsubscribe", response_class=HTMLResponse)
|
|
async def unsubscribe(
|
|
request: Request,
|
|
token: str = Query(...),
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
from app import branding
|
|
uid = verify_unsubscribe_token(token)
|
|
if uid is not None:
|
|
user = await session.get(User, uid)
|
|
if user is not None and user.email_digest_opt_in:
|
|
user.email_digest_opt_in = False
|
|
await session.commit()
|
|
log.info("email.unsubscribe.ok", user_id=uid)
|
|
else:
|
|
log.info("email.unsubscribe.noop_or_unknown", user_id=uid)
|
|
else:
|
|
log.info("email.unsubscribe.bad_token")
|
|
|
|
# Same confirmation page regardless — don't leak token validity.
|
|
return HTMLResponse(_CONFIRM_PAGE.format(brand=branding.BRAND_NAME))
|