kadence/backend/app/api/routes_auth.py

"""Signing in and out of the browser UI.

Mounted directly on the app, not through `api_router` — these endpoints have
to be reachable *before* a session exists, so they never carry the
`require_auth` dependency. `change_password` checks for a live session itself.
"""

from __future__ import annotations

from typing import Annotated, Any

from fastapi import APIRouter, Body, Cookie, HTTPException, Request, Response, status

from app.api.deps import DbSession
from app.services import auth_service, settings_service

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

_COOKIE = auth_service.SESSION_COOKIE


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


def _is_https(request: Request) -> bool:
    # TLS usually terminates at a reverse proxy in front of this container, so
    # `request.url.scheme` alone would read "http" even on an HTTPS deployment.
    return request.url.scheme == "https" or request.headers.get("x-forwarded-proto") == "https"


def _set_cookie(response: Response, request: Request, token: str, max_age: int) -> None:
    response.set_cookie(
        _COOKIE,
        token,
        max_age=max_age,
        httponly=True,
        secure=_is_https(request),
        samesite="lax",
        path="/",
    )


@router.get("/me")
def me(
    session: DbSession,
    kadence_session: Annotated[str | None, Cookie()] = None,
) -> dict:
    settings = settings_service.load_settings(session)
    row = auth_service.valid_session(session, kadence_session)
    login_required = bool(settings.auth.password_hash) or (
        settings.auth.telegram_login_enabled and bool(settings.auth.telegram_owner_id)
    )
    return {
        "authenticated": row is not None or not login_required,
        "login_required": login_required,
        "username": settings.auth.username,
        "password_login_enabled": bool(settings.auth.password_hash),
        "telegram_login_enabled": settings.auth.telegram_login_enabled
        and bool(settings.auth.telegram_owner_id),
        "telegram_bot_username": settings.auth.telegram_bot_username,
    }


@router.post("/login")
def login(
    session: DbSession,
    request: Request,
    response: Response,
    body: dict[str, Any] = Body(...),
) -> dict:
    key = f"password:{_client_key(request)}"
    if auth_service.rate_limited(key):
        raise HTTPException(status.HTTP_429_TOO_MANY_REQUESTS, "Too many attempts. Wait a few minutes.")

    settings = settings_service.load_settings(session)
    username = str(body.get("username", ""))
    password = str(body.get("password", ""))
    ok = (
        bool(settings.auth.password_hash)
        and username == settings.auth.username
        and auth_service.verify_password(password, settings.auth.password_hash)
    )
    if not ok:
        auth_service.record_failure(key)
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Wrong username or password.")

    auth_service.record_success(key)
    row = auth_service.create_session(session, "password", settings.auth.session_days)
    _set_cookie(response, request, row.token, settings.auth.session_days * 86400)
    return {"ok": True}


@router.post("/login/telegram")
def login_telegram(
    session: DbSession,
    request: Request,
    response: Response,
    body: dict[str, Any] = Body(...),
) -> dict:
    key = f"telegram:{_client_key(request)}"
    if auth_service.rate_limited(key):
        raise HTTPException(status.HTTP_429_TOO_MANY_REQUESTS, "Too many attempts. Wait a few minutes.")

    settings = settings_service.load_settings(session)
    telegram = settings.notifications.telegram
    auth = settings.auth
    ok = (
        auth.telegram_login_enabled
        and bool(auth.telegram_owner_id)
        and bool(telegram.bot_token)
        and auth_service.verify_telegram_payload(body, telegram.bot_token)
        and str(body.get("id")) == auth.telegram_owner_id
    )
    if not ok:
        auth_service.record_failure(key)
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Telegram sign-in did not check out.")

    auth_service.record_success(key)
    row = auth_service.create_session(session, "telegram", auth.session_days)
    _set_cookie(response, request, row.token, auth.session_days * 86400)
    return {"ok": True}


@router.post("/logout")
def logout(
    session: DbSession,
    response: Response,
    kadence_session: Annotated[str | None, Cookie()] = None,
) -> dict:
    auth_service.delete_session(session, kadence_session)
    response.delete_cookie(_COOKIE, path="/")
    return {"ok": True}


@router.post("/password")
def change_password(
    session: DbSession,
    body: dict[str, Any] = Body(...),
    kadence_session: Annotated[str | None, Cookie()] = None,
) -> dict:
    """Requires a live session — set the first password via `AUTH_PASSWORD` or
    `make set-password` instead, since there is nothing to authenticate with
    yet on a fresh install."""
    if auth_service.valid_session(session, kadence_session) is None:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Sign in first.")

    settings = settings_service.load_settings(session)
    current = str(body.get("current_password", ""))
    new = str(body.get("new_password", ""))
    if settings.auth.password_hash and not auth_service.verify_password(
        current, settings.auth.password_hash
    ):
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Current password is wrong.")
    if len(new) < 8:
        raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "Use at least 8 characters.")

    settings.auth.password_hash = auth_service.hash_password(new)
    settings_service.save_settings(session, settings)
    return {"ok": True}