thermograph/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
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 db import get_async_session
from models import AccessToken, User

BASE = "/" + os.environ.get("THERMOGRAPH_BASE", "/thermograph").strip("/")
# Only used to sign reset/verification tokens — features that stay dormant until
# email is wired up. A per-boot random secret is fine while they're unused.
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")


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 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="lax",     # blocks cross-site form posts (CSRF), allows top-level nav
)


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)