app/routers/auth.py

"""Implement password login and signed-session lifecycle routes.

Login attempts are rate-limited in memory by the direct client address.
"""

import time
from collections import defaultdict, deque
from threading import Lock

from fastapi import APIRouter, Depends, Request, status

from app.deps import require_login
from app.models import (
    ApiEnvelope,
    AuthenticatedData,
    ChangePasswordRequest,
    LoginRequest,
    MeData,
    StoredConfig,
    err,
    ok,
)

router = APIRouter(prefix="/api/auth", tags=["auth"])
_attempts: dict[str, deque[float]] = defaultdict(deque)
_attempts_lock = Lock()


def _client_key(request: Request) -> str:
    return request.client.host if request.client else "unknown"


def _rate_limited(request: Request) -> bool:
    settings = request.app.state.settings
    now = time.monotonic()
    cutoff = now - settings.login_rate_window_sec
    key = _client_key(request)
    with _attempts_lock:
        attempts = _attempts[key]
        while attempts and attempts[0] <= cutoff:
            attempts.popleft()
        if len(attempts) >= settings.login_rate_limit:
            return True
        attempts.append(now)
        return False


def _clear_attempts(request: Request) -> None:
    with _attempts_lock:
        _attempts.pop(_client_key(request), None)


def _set_session(request: Request, config: StoredConfig) -> None:
    max_age = request.app.state.settings.session_max_age_sec
    request.session.clear()
    request.session.update(
        {
            "auth": True,
            "sv": config.session_version,
            "exp": int(time.time()) + max_age,
        }
    )


def _session_is_current(request: Request, config: StoredConfig | None) -> bool:
    if config is None:
        return False
    session = request.session
    expiry = session.get("exp")
    return (
        session.get("auth") is True
        and session.get("sv") == config.session_version
        and isinstance(expiry, (int, float))
        and expiry >= time.time()
    )


@router.post("/login", response_model=ApiEnvelope)
def login(request: Request, body: LoginRequest) -> object:
    """Authenticate and establish a signed cookie session."""

    config = request.app.state.config_store.load()
    if config is None:
        return err(
            "setup_required",
            "Application setup is required",
            status.HTTP_503_SERVICE_UNAVAILABLE,
        )
    if _rate_limited(request):
        return err(
            "rate_limited",
            "Too many login attempts",
            status.HTTP_429_TOO_MANY_REQUESTS,
        )
    if not request.app.state.config_store.password_matches(body.password):
        return err(
            "unauthorized",
            "Invalid credentials",
            status.HTTP_401_UNAUTHORIZED,
        )

    _clear_attempts(request)
    _set_session(request, config)
    return ok(AuthenticatedData(authenticated=True))


@router.post("/logout", response_model=ApiEnvelope)
def logout(
    request: Request,
    _config: StoredConfig = Depends(require_login),
) -> dict[str, object]:
    """Clear the current signed session."""

    request.session.clear()
    return ok(AuthenticatedData(authenticated=False))


@router.get("/me", response_model=ApiEnvelope)
def me(request: Request) -> dict[str, object]:
    """Return public app identity and current session state."""

    config = request.app.state.config_store.load()
    authenticated = _session_is_current(request, config)
    if not authenticated:
        request.session.clear()
    return ok(
        MeData(
            authenticated=authenticated,
            app_name=request.app.state.settings.app_name,
        )
    )


@router.post("/change-password", response_model=ApiEnvelope)
def change_password(
    request: Request,
    body: ChangePasswordRequest,
    _config: StoredConfig = Depends(require_login),
) -> object:
    """Rotate the shared password and invalidate older sessions."""

    changed = request.app.state.config_store.change_password(
        body.old_password,
        body.new_password,
    )
    if not changed:
        return err(
            "unauthorized",
            "Invalid credentials",
            status.HTTP_401_UNAUTHORIZED,
        )
    config = request.app.state.config_store.load()
    if config is None:
        return err(
            "internal",
            "Configuration unavailable",
            status.HTTP_500_INTERNAL_SERVER_ERROR,
        )
    _set_session(request, config)
    return ok(AuthenticatedData(authenticated=True))