LLeMbas/src/lembas/api/auth.py

"""Registration, sign-in and sign-out."""

from __future__ import annotations

import logging

from fastapi import APIRouter, Form, Request, Response, status
from fastapi.responses import RedirectResponse
from sqlalchemy import func, select

from lembas.api.deps import CurrentUser, Db
from lembas.config import settings
from lembas.db.models import ROLE_ADMIN, ROLE_USER, User
from lembas.security.passwords import hash_password, validate_password, verify_password
from lembas.security.sessions import COOKIE_NAME, create_session, revoke_session
from lembas.services import settings_store
from lembas.web.templating import render

log = logging.getLogger(__name__)

router = APIRouter(prefix="/auth", tags=["auth"])


def _no_users_yet(db: Db) -> bool:
    return db.scalar(select(func.count()).select_from(User)) == 0


def _set_session_cookie(response: Response, token: str) -> None:
    response.set_cookie(
        COOKIE_NAME,
        token,
        max_age=settings.session_ttl,
        httponly=True,
        # Lax is what makes this application CSRF-safe without tokens: the
        # cookie is not sent on cross-site POSTs, and every mutating route here
        # is a POST. Do not relax to "none".
        samesite="lax",
        # Only over HTTPS when the deployment is not plain local http. Marking
        # it secure on http would silently break sign-in for a LAN install.
        secure=False,
        path="/",
    )


def _safe_next(raw: str | None) -> str:
    """Reject open redirects: only same-origin absolute paths are allowed."""
    if not raw or not raw.startswith("/") or raw.startswith("//"):
        return "/"
    return raw


def _login_page(request: Request, db: Db, *, status_code: int = 200, **context):
    """Render the sign-in page.

    Always goes through here so `allow_signup` reflects the *stored* setting
    rather than the environment default baked in by render(). Otherwise the
    "Create one" link would keep appearing after an administrator closed
    registration, offering a link that only leads to a refusal.
    """
    context.setdefault("next", "/")
    context["allow_signup"] = settings_store.signup_allowed(db)
    return render(request, "auth/login.html", context, status_code=status_code)


@router.get("/login")
async def login_form(request: Request, db: Db, user: CurrentUser, next: str = "/"):
    if user is not None:
        return RedirectResponse(_safe_next(next), status_code=status.HTTP_303_SEE_OTHER)
    # An empty database means this install has never been set up. Send the
    # first visitor straight to registration rather than to a login form they
    # cannot possibly satisfy.
    if _no_users_yet(db):
        return RedirectResponse("/auth/register", status_code=status.HTTP_303_SEE_OTHER)
    return _login_page(request, db, next=_safe_next(next))


@router.post("/login")
async def login(
    request: Request,
    db: Db,
    email: str = Form(...),
    password: str = Form(...),
    next: str = Form("/"),
):
    email = email.strip().lower()
    user = db.scalar(select(User).where(User.email == email))

    # One message for "no such account" and "wrong password" alike, so the form
    # cannot be used to discover which addresses are registered.
    if user is None or not verify_password(password, user.password_hash):
        log.info("failed sign-in for %s", email)
        return _login_page(
            request,
            db,
            status_code=status.HTTP_401_UNAUTHORIZED,
            error="That email and password do not match.",
            email=email,
            next=_safe_next(next),
        )

    if not user.active:
        return _login_page(
            request,
            db,
            status_code=status.HTTP_403_FORBIDDEN,
            error="This account has been deactivated. Ask an administrator.",
            email=email,
            next=_safe_next(next),
        )

    token = create_session(
        db,
        user,
        user_agent=request.headers.get("user-agent", ""),
        ip_address=request.client.host if request.client else "",
    )
    response = RedirectResponse(_safe_next(next), status_code=status.HTTP_303_SEE_OTHER)
    _set_session_cookie(response, token)
    return response


@router.get("/register")
async def register_form(request: Request, db: Db, user: CurrentUser):
    if user is not None:
        return RedirectResponse("/", status_code=status.HTTP_303_SEE_OTHER)
    first_run = _no_users_yet(db)
    if not first_run and not settings_store.signup_allowed(db):
        return _login_page(
            request,
            db,
            status_code=status.HTTP_403_FORBIDDEN,
            error="Registration is closed. Ask an administrator for an account.",
        )
    return render(request, "auth/register.html", {"first_run": first_run})


@router.post("/register")
async def register(
    request: Request,
    db: Db,
    name: str = Form(...),
    email: str = Form(...),
    password: str = Form(...),
):
    first_run = _no_users_yet(db)
    if not first_run and not settings_store.signup_allowed(db):
        return _login_page(
            request,
            db,
            status_code=status.HTTP_403_FORBIDDEN,
            error="Registration is closed. Ask an administrator for an account.",
        )

    name = name.strip()
    email = email.strip().lower()

    def fail(message: str) -> Response:
        return render(
            request,
            "auth/register.html",
            {"error": message, "name": name, "email": email, "first_run": first_run},
            status_code=status.HTTP_400_BAD_REQUEST,
        )

    if not name:
        return fail("Please enter a name.")
    if "@" not in email or "." not in email.split("@")[-1]:
        return fail("Please enter a valid email address.")
    if (problem := validate_password(password)) is not None:
        return fail(problem)
    if db.scalar(select(User).where(User.email == email)) is not None:
        return fail("An account with that email already exists.")

    # Whoever sets the instance up owns it. Everyone after that is a plain user
    # until an admin says otherwise.
    user = User(
        name=name,
        email=email,
        password_hash=hash_password(password),
        role=ROLE_ADMIN if first_run else ROLE_USER,
    )
    db.add(user)
    db.commit()
    log.info("registered %s as %s", email, user.role)

    token = create_session(
        db,
        user,
        user_agent=request.headers.get("user-agent", ""),
        ip_address=request.client.host if request.client else "",
    )
    response = RedirectResponse("/", status_code=status.HTTP_303_SEE_OTHER)
    _set_session_cookie(response, token)
    return response


@router.post("/logout")
async def logout(request: Request, db: Db):
    revoke_session(db, request.cookies.get(COOKIE_NAME))
    response = RedirectResponse("/auth/login", status_code=status.HTTP_303_SEE_OTHER)
    response.delete_cookie(COOKIE_NAME, path="/")
    return response