jarca-ledger-mcp/dashboard/auth_server.py

from __future__ import annotations

import secrets
import time
from urllib.parse import parse_qs, urlsplit

from starlette.requests import Request
from starlette.responses import PlainTextResponse, RedirectResponse, Response
from starlette.routing import Route

from dashboard import authentication

_MAX_LOGIN_BODY_BYTES = 4096
_MAX_CREDENTIAL_CHARS = 1024


def _redirect(location: str) -> RedirectResponse:
    response = RedirectResponse(location, status_code=303)
    response.headers["Cache-Control"] = "no-store"
    response.headers["Pragma"] = "no-cache"
    return response


def _same_origin(request: Request) -> bool:
    origin = request.headers.get("origin")
    host = request.headers.get("host")
    if not origin or not host:
        return False
    try:
        parsed = urlsplit(origin)
        authority = urlsplit(f"//{host}")
        origin_port = parsed.port or 443
        request_port = authority.port or 443
        return (
            request.url.scheme == "https"
            and parsed.scheme == "https"
            and parsed.hostname is not None
            and authority.hostname is not None
            and parsed.hostname.lower() == authority.hostname.lower()
            and origin_port == request_port
            and parsed.username is None
            and parsed.password is None
            and parsed.path in {"", "/"}
            and not parsed.query
            and not parsed.fragment
        )
    except ValueError:
        return False


def _same_origin_navigation(request: Request) -> bool:
    referer = request.headers.get("referer")
    host = request.headers.get("host")
    if not referer or not host:
        return False
    try:
        parsed = urlsplit(referer)
        authority = urlsplit(f"//{host}")
        referer_port = parsed.port or 443
        request_port = authority.port or 443
        return (
            request.url.scheme == "https"
            and parsed.scheme == "https"
            and parsed.hostname is not None
            and authority.hostname is not None
            and parsed.hostname.lower() == authority.hostname.lower()
            and referer_port == request_port
            and parsed.username is None
            and parsed.password is None
        )
    except ValueError:
        return False


def _one(values: dict[str, list[str]], name: str) -> str | None:
    candidates = values.get(name)
    if candidates is None or len(candidates) != 1:
        return None
    value = candidates[0]
    if len(value) > _MAX_CREDENTIAL_CHARS:
        return None
    return value


async def login(request: Request) -> Response:
    correlation_id = secrets.token_hex(8)
    if not _same_origin(request):
        authentication.log_auth_event("login_origin_rejected", correlation_id)
        return PlainTextResponse("Forbidden", status_code=403)
    content_type = request.headers.get("content-type", "").split(";", 1)[0].strip().lower()
    if content_type != "application/x-www-form-urlencoded":
        authentication.log_auth_event("login_request_rejected", correlation_id)
        return PlainTextResponse("Bad Request", status_code=400)
    body = await request.body()
    if len(body) > _MAX_LOGIN_BODY_BYTES:
        authentication.log_auth_event("login_request_rejected", correlation_id)
        return PlainTextResponse("Bad Request", status_code=400)
    try:
        values = parse_qs(
            body.decode("utf-8"),
            keep_blank_values=True,
            strict_parsing=True,
            max_num_fields=3,
        )
    except (UnicodeDecodeError, ValueError):
        authentication.log_auth_event("login_request_rejected", correlation_id)
        return PlainTextResponse("Bad Request", status_code=400)

    settings = authentication.load_auth_settings()
    if settings is None:
        authentication.log_auth_event("configuration_error", correlation_id)
        return _redirect("/?auth_error=unavailable")

    submitted_id = _one(values, "admin_id")
    submitted_password = _one(values, "admin_password")
    if submitted_id is None or submitted_password is None:
        authentication.log_auth_event("login_request_rejected", correlation_id)
        return _redirect("/?auth_error=invalid")

    started_at = time.monotonic()
    if not authentication.verify_credentials(
        settings.credentials, submitted_id, submitted_password
    ):
        authentication.enforce_failed_login_delay(started_at=started_at)
        authentication.log_auth_event("login_failed", correlation_id)
        return _redirect("/?auth_error=invalid")

    remember_me = _one(values, "remember_me") == "on"
    issued = authentication.issue_persistent_token(
        settings,
        remember_me=remember_me,
    )
    response = _redirect("/")
    response.set_cookie(
        authentication.AUTH_COOKIE_NAME,
        issued.value,
        max_age=issued.max_age,
        path="/",
        secure=True,
        httponly=True,
        samesite="lax",
    )
    authentication.log_auth_event("login_succeeded", correlation_id)
    return response


async def logout(request: Request) -> Response:
    correlation_id = secrets.token_hex(8)
    if not _same_origin(request):
        authentication.log_auth_event("logout_origin_rejected", correlation_id)
        return PlainTextResponse("Forbidden", status_code=403)
    response = _redirect("/")
    _delete_auth_cookie(response)
    authentication.log_auth_event("logout", correlation_id)
    return response


def _delete_auth_cookie(response: Response) -> None:
    response.delete_cookie(
        authentication.AUTH_COOKIE_NAME,
        path="/",
        secure=True,
        httponly=True,
        samesite="lax",
    )


async def cleanup(request: Request) -> Response:
    correlation_id = secrets.token_hex(8)
    if not _same_origin_navigation(request):
        authentication.log_auth_event("cleanup_origin_rejected", correlation_id)
        return PlainTextResponse("Forbidden", status_code=403)
    response = _redirect("/")
    _delete_auth_cookie(response)
    authentication.log_auth_event("cleanup", correlation_id)
    return response


def create_auth_routes() -> list[Route]:
    return [
        Route("/api/auth/login", login, methods=["POST"]),
        Route("/api/auth/logout", logout, methods=["POST"]),
        Route("/api/auth/cleanup", cleanup, methods=["GET"]),
    ]