feat(dashboard): persist administrator login

dashboard/auth_server.py

@@ -0,0 +1,144 @@
+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 _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("/")
+    response.delete_cookie(
+        authentication.AUTH_COOKIE_NAME,
+        path="/",
+        secure=True,
+        httponly=True,
+        samesite="lax",
+    )
+    authentication.log_auth_event("logout", 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"]),
+    ]