read.markets/app/routers/email.py
Giorgio Gilestro 355593c4f7 css: split cassandra.css into per-section files
Splits the 2571-line cassandra.css into ten focused stylesheets:
tokens (palette + fonts), layout (chrome), panels, dashboard,
portfolio, log-chat, auth, settings, news, public. base.html and
public_base.html load only what they need; auth pages (login,
verify, unsubscribe confirm) load tokens + layout + auth.

Brand drift-detection test repointed at tokens.css (where the
palette now lives). 291 tests still pass.
2026-05-28 12:31:29 +02:00

104 lines
3.3 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 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 = """\
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<title>Unsubscribed — {brand}</title>
<link rel="stylesheet" href="/static/css/tokens.css">
<link rel="stylesheet" href="/static/css/layout.css">
<link rel="stylesheet" href="/static/css/auth.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),
):
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))