"""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}