thermograph/accounts/users.py

"""fastapi-users wiring: password hashing, cookie sessions, the auth dependency.

We use the library end-to-end (no hand-rolled crypto): pwdlib hashing under the
hood, an HttpOnly cookie transport, and a *database* session strategy (tokens live
in the access_token table) so logins survive a restart and can be revoked.
"""
import os
import secrets
import uuid

from fastapi import Depends, Request
from fastapi_users import BaseUserManager, FastAPIUsers, InvalidPasswordException, UUIDIDMixin
from fastapi_users.authentication import AuthenticationBackend, CookieTransport
from fastapi_users.authentication.strategy.db import DatabaseStrategy
from fastapi_users_db_sqlalchemy import SQLAlchemyUserDatabase
from fastapi_users_db_sqlalchemy.access_token import SQLAlchemyAccessTokenDatabase
from sqlalchemy.ext.asyncio import AsyncSession

from accounts.db import get_async_session
from accounts.models import AccessToken, User
from notifications import mailer

BASE = "/" + os.environ.get("THERMOGRAPH_BASE", "/thermograph").strip("/")
# The public origin, for links in emailed tokens — same var + default indexnow.py
# and discord.py already use for exactly this purpose (a link that has to resolve
# from wherever the recipient opens it, so it can't be relative to a request).
BASE_URL = os.environ.get("THERMOGRAPH_BASE_URL", "https://thermograph.org").rstrip("/")
# Signs verification tokens. THERMOGRAPH_AUTH_SECRET must be pinned before any
# such token is mailed — a per-boot random fallback would invalidate every
# outstanding link (and every session) on restart.
SECRET = os.environ.get("THERMOGRAPH_AUTH_SECRET") or secrets.token_urlsafe(32)
SESSION_TTL_SECONDS = int(os.environ.get("THERMOGRAPH_SESSION_TTL_DAYS", "30")) * 86400


def _cookie_secure() -> bool:
    # The app serves plain HTTP on the LAN by default; a Secure cookie would never
    # be sent over HTTP, silently breaking auth. So default OFF and let the TLS/VPS
    # deploy opt in with THERMOGRAPH_COOKIE_SECURE=1. Accepted tradeoff: LAN traffic
    # is unencrypted on a trusted network.
    v = os.environ.get("THERMOGRAPH_COOKIE_SECURE", "").strip().lower()
    return v in ("1", "true", "yes", "always", "on")


_COOKIE_SECURE = _cookie_secure()
# Default "lax" (today's exact hardcoded behavior): blocks cross-site form posts
# (CSRF), allows top-level nav, and needs no Secure. "none" is what genuine
# cross-origin fetch (frontend and backend on different origins -- repo-split
# Stage 5) requires -- browsers refuse to store a SameSite=None cookie without
# Secure at all, so that combination fails loud here instead of manifesting
# only as "login mysteriously doesn't persist" once deployed.
COOKIE_SAMESITE = os.environ.get("THERMOGRAPH_COOKIE_SAMESITE", "lax").strip().lower()
if COOKIE_SAMESITE == "none" and not _COOKIE_SECURE:
    raise RuntimeError(
        "THERMOGRAPH_COOKIE_SAMESITE=none requires THERMOGRAPH_COOKIE_SECURE=1 "
        "(browsers drop SameSite=None cookies that aren't also Secure)."
    )


async def get_user_db(session: AsyncSession = Depends(get_async_session)):
    yield SQLAlchemyUserDatabase(session, User)


async def get_access_token_db(session: AsyncSession = Depends(get_async_session)):
    yield SQLAlchemyAccessTokenDatabase(session, AccessToken)


class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]):
    reset_password_token_secret = SECRET
    verification_token_secret = SECRET

    async def validate_password(self, password: str, user) -> None:
        if len(password) < 8:
            raise InvalidPasswordException(reason="Password must be at least 8 characters.")
        if len(password) > 1024:
            # Guard against a scrypt/argon DoS from a megabyte "password".
            raise InvalidPasswordException(reason="Password is too long.")

    async def on_after_register(self, user: User, request: "Request | None" = None) -> None:
        # Kick off the same token-generation path a manual "resend verification"
        # request would use, so there is exactly one place that builds the email
        # (on_after_request_verify below) rather than two copies drifting apart.
        await self.request_verify(user, request)

    async def on_after_request_verify(
        self, user: User, token: str, request: "Request | None" = None
    ) -> None:
        # mailer.send() already no-ops on the console/disabled backends (dev,
        # tests) — the enabled() check here just skips building the message.
        if not mailer.enabled():
            return
        link = f"{BASE_URL}{BASE}/?verify_token={token}"
        mailer.send(
            user.email,
            "Confirm your Thermograph account",
            "Welcome to Thermograph! Confirm your email address to finish setting up "
            f"your account:\n\n{link}\n\n"
            "If you didn't create this account, you can safely ignore this email.",
        )


async def get_user_manager(user_db=Depends(get_user_db)):
    yield UserManager(user_db)


cookie_transport = CookieTransport(
    cookie_name="tg_session",
    cookie_max_age=SESSION_TTL_SECONDS,
    cookie_path=BASE,          # scope the cookie to /thermograph, not the whole domain
    cookie_secure=_COOKIE_SECURE,
    cookie_httponly=True,      # unreadable by JS -> XSS can't steal the session
    cookie_samesite=COOKIE_SAMESITE,
)


def get_database_strategy(access_token_db=Depends(get_access_token_db)) -> DatabaseStrategy:
    return DatabaseStrategy(access_token_db, lifetime_seconds=SESSION_TTL_SECONDS)


auth_backend = AuthenticationBackend(
    name="cookie",
    transport=cookie_transport,
    get_strategy=get_database_strategy,
)

fastapi_users = FastAPIUsers[User, uuid.UUID](get_user_manager, [auth_backend])

# Dependencies handlers use to require / peek at the logged-in user.
current_active_user = fastapi_users.current_user(active=True)
current_user_optional = fastapi_users.current_user(active=True, optional=True)