Clarium/backend/auth.py

"""Auth slice: server-side sessions (ADR-0002) and auth routes."""

import hashlib
import re
import secrets
from collections.abc import Callable
from pathlib import Path
from typing import ParamSpec, TypedDict, TypeVar, cast

import psycopg2.errors
from flask import (
    Blueprint,
    Flask,
    Response,
    current_app,
    g,
    jsonify,
    make_response,
    request,
)

from db import query, query_one, query_row
from passwords import (
    dummy_hash,
    hash_password,
    load_pepper,
    needs_rehash,
    verify_password,
)
from validation import ApiError, JsonObject, json_body, require_utf8

bp = Blueprint("auth", __name__, url_prefix="/api/auth")

SESSION_COOKIE = "sid"
USERNAME_PATTERN = re.compile(r"[A-Za-z0-9_.-]{3,64}")
MIN_PASSWORD, MAX_PASSWORD = 12, 1024
INVALID_LOGIN = "Invalid username or password"

P = ParamSpec("P")
R = TypeVar("R")


class UserRow(TypedDict):
    id: int
    username: str


class UserWithHash(UserRow):
    password_hash: str


def init_app(app: Flask) -> None:
    app.extensions["password_pepper"] = load_pepper(Path(app.root_path) / ".pepper")
    dummy_hash()
    app.before_request(_require_session)
    app.register_blueprint(bp)


# --- Sessions ------------------------------------------------------------------


def allow_anonymous(view: Callable[P, R]) -> Callable[P, R]:
    view.allow_anonymous = True  # type: ignore[attr-defined]
    return view


def _require_session() -> None:
    view = current_app.view_functions.get(request.endpoint or "")
    if view is None or getattr(view, "allow_anonymous", False):
        return
    token = request.cookies.get(SESSION_COOKIE)
    row = None
    if token:
        row = query_one(
            """
            UPDATE sessions
               SET expires_at = LEAST(now() + interval '30 minutes', created_at + interval '12 hours')
             WHERE token_hash = %s
               AND expires_at > now()
               AND created_at > now() - interval '12 hours'
            RETURNING user_id
            """,
            (_token_hash(token),),
        )
    if row is None:
        raise ApiError(401, "Not signed in or session expired; log in again")
    g.user_id = row["user_id"]


def _token_hash(token: str) -> bytes:
    return hashlib.sha256(token.encode("utf-8")).digest()


def _pepper() -> bytes:
    pepper: bytes = current_app.extensions["password_pepper"]
    return pepper


def _start_session(user: UserRow, status: int) -> Response:
    token = secrets.token_urlsafe(32)
    query(
        "DELETE FROM sessions WHERE user_id = %s AND expires_at <= now()", (user["id"],)
    )
    query(
        "INSERT INTO sessions (token_hash, user_id, expires_at) VALUES (%s, %s, now() + interval '30 minutes')",
        (_token_hash(token), user["id"]),
    )
    response = jsonify(id=user["id"], username=user["username"])
    response.status_code = status
    response.set_cookie(
        SESSION_COOKIE, token, httponly=True, secure=True, samesite="Lax", path="/api"
    )
    return response


# --- Routes ----------------------------------------------------------------------


def _password(body: JsonObject) -> str:
    password = body.get("password")
    if not isinstance(password, str):
        raise ApiError(400, "password is required and must be a string", "password")
    if len(password) > MAX_PASSWORD:
        raise ApiError(
            400, f"password must be at most {MAX_PASSWORD} characters", "password"
        )
    require_utf8(password, "password")
    return password


@bp.post("/register")
@allow_anonymous
def register() -> Response:
    body = json_body({"username", "password"})
    username = body.get("username")
    if not isinstance(username, str) or not USERNAME_PATTERN.fullmatch(username):
        raise ApiError(
            400,
            "username must be 3-64 characters: letters, digits, '.', '_' or '-'",
            "username",
        )
    password = _password(body)
    if len(password) < MIN_PASSWORD:
        raise ApiError(
            400, f"password must be at least {MIN_PASSWORD} characters", "password"
        )
    try:
        user = query_row(
            "INSERT INTO users (username, password_hash) VALUES (%s, %s) RETURNING id, username",
            (username, hash_password(password, _pepper())),
        )
    except psycopg2.errors.UniqueViolation:
        raise ApiError(
            409, "Username already taken; choose another", "username"
        ) from None
    return _start_session(cast(UserRow, user), 201)


@bp.post("/login")
@allow_anonymous
def login() -> Response:
    body = json_body({"username", "password"})
    password = _password(body)
    username = body.get("username")
    user = None
    if isinstance(username, str) and USERNAME_PATTERN.fullmatch(username):
        user = cast(
            UserWithHash | None,
            query_one(
                "SELECT id, username, password_hash FROM users WHERE lower(username) = lower(%s)",
                (username,),
            ),
        )
    if user is None:
        verify_password(password, dummy_hash(), _pepper())
        raise ApiError(401, INVALID_LOGIN)
    if not verify_password(password, user["password_hash"], _pepper()):
        raise ApiError(401, INVALID_LOGIN)
    if needs_rehash(user["password_hash"]):
        query(
            "UPDATE users SET password_hash = %s WHERE id = %s",
            (hash_password(password, _pepper()), user["id"]),
        )
    return _start_session(user, 200)


@bp.post("/logout")
@allow_anonymous
def logout() -> Response:
    token = request.cookies.get(SESSION_COOKIE)
    if token:
        query("DELETE FROM sessions WHERE token_hash = %s", (_token_hash(token),))
    response = make_response("", 204)
    response.delete_cookie(
        SESSION_COOKIE, path="/api", secure=True, httponly=True, samesite="Lax"
    )
    return response


@bp.get("/me")
def me() -> Response:
    return jsonify(
        query_one("SELECT id, username FROM users WHERE id = %s", (g.user_id,))
    )