diff --git a/app/main.py b/app/main.py index a9dcb50..efca3ae 100644 --- a/app/main.py +++ b/app/main.py @@ -19,6 +19,7 @@ from app.db import get_session_factory from app.logging import configure_logging, get_logger from app.routers import api as api_router from app.routers import auth as auth_router +from app.routers import email as email_router from app.routers import pages as pages_router from app.routers import public as public_router from app.routers import sync as sync_router @@ -83,6 +84,7 @@ app.mount( ) app.include_router(auth_router.router, tags=["auth"]) +app.include_router(email_router.router, tags=["email"]) app.include_router(api_router.router, prefix="/api", tags=["api"]) app.include_router(universe_router.router, prefix="/api", tags=["universe"]) app.include_router(sync_router.router, tags=["portfolio-sync"]) diff --git a/app/routers/email.py b/app/routers/email.py new file mode 100644 index 0000000..c6aa80f --- /dev/null +++ b/app/routers/email.py @@ -0,0 +1,101 @@ +"""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 = """\ + + + + + Unsubscribed — {brand} + + + +
+
{brand}
+
email preferences
+

You're unsubscribed from email digests.

+

+ You can re-enable digests any time from + Settings. +

+
+ + +""" + + +@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)) diff --git a/tests/test_email_unsubscribe.py b/tests/test_email_unsubscribe.py new file mode 100644 index 0000000..22b5f90 --- /dev/null +++ b/tests/test_email_unsubscribe.py @@ -0,0 +1,83 @@ +"""Unsubscribe token roundtrip + endpoint.""" +from __future__ import annotations + +import asyncio + + +def _build_app(tmp_path, secret="rt-secret-32-bytes-or-so-padding-here"): + import os + os.environ["CASSANDRA_SESSION_SECRET"] = secret + + from fastapi import FastAPI + from fastapi.testclient import TestClient + from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + + from app import db as db_mod + from app.db import Base + from app.config import get_settings + from app.models import User + from app.routers import email as email_router + get_settings.cache_clear() # pick up the new env var + + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/u.db") + factory = async_sessionmaker(engine, expire_on_commit=False) + db_mod._engine = engine + db_mod._session_factory = factory + + async def _seed(): + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + async with factory() as s: + s.add(User(id=42, email="u@x", tier="paid", email_digest_opt_in=True)) + await s.commit() + + asyncio.run(_seed()) + + app = FastAPI() + app.include_router(email_router.router) + return TestClient(app) + + +def test_sign_and_verify_token_roundtrip(monkeypatch): + monkeypatch.setenv("CASSANDRA_SESSION_SECRET", "rt-secret-32-bytes-or-so-padding-here") + from app.config import get_settings + get_settings.cache_clear() + from app.routers.email import sign_unsubscribe_token, verify_unsubscribe_token + tok = sign_unsubscribe_token(42) + assert verify_unsubscribe_token(tok) == 42 + assert verify_unsubscribe_token("garbage") is None + + +def test_get_unsubscribe_flips_flag(tmp_path): + client = _build_app(tmp_path) + from app.routers.email import sign_unsubscribe_token + tok = sign_unsubscribe_token(42) + r = client.get(f"/email/unsubscribe?token={tok}") + assert r.status_code == 200 + assert "unsubscribed" in r.text.lower() + + async def _check(): + from app import db as db_mod + from app.models import User + async with db_mod._session_factory() as s: + u = await s.get(User, 42) + assert u.email_digest_opt_in is False + asyncio.run(_check()) + + +def test_get_unsubscribe_invalid_token_returns_generic_page(tmp_path): + client = _build_app(tmp_path) + r = client.get("/email/unsubscribe?token=garbage") + # We don't 4xx — that would leak token validity. Show the generic page. + assert r.status_code == 200 + assert "unsubscribed" in r.text.lower() or "preferences" in r.text.lower() + + +def test_replay_is_idempotent(tmp_path): + client = _build_app(tmp_path) + from app.routers.email import sign_unsubscribe_token + tok = sign_unsubscribe_token(42) + r1 = client.get(f"/email/unsubscribe?token={tok}") + r2 = client.get(f"/email/unsubscribe?token={tok}") + assert r1.status_code == 200 + assert r2.status_code == 200