73 lines
2.5 KiB
Python
73 lines
2.5 KiB
Python
"""Explicit authentication policy shared by the admin UI and internal API."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from hmac import compare_digest
|
|
from ipaddress import ip_address
|
|
from urllib.parse import urlsplit
|
|
|
|
from fastapi import HTTPException, Request
|
|
|
|
from app.config import settings
|
|
|
|
|
|
PROXY_ADMIN_USER_HEADER = "X-ClientFlow-Admin-User"
|
|
TOKEN_HEADER = "X-ClientFlow-Admin-Token"
|
|
TOKEN_COOKIE = "clientflow_admin_token"
|
|
|
|
|
|
def _is_loopback_request(request: Request) -> bool:
|
|
"""Use the transport peer, never a caller-controlled forwarded header."""
|
|
host = request.client.host if request.client else ""
|
|
try:
|
|
return ip_address(host).is_loopback
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def require_admin_auth(request: Request, *, area: str) -> None:
|
|
mode = settings.clientflow_admin_auth_mode
|
|
detail_prefix = "internal api" if area == "internal_api" else "admin"
|
|
|
|
if mode == "proxy":
|
|
admin_user = (request.headers.get(PROXY_ADMIN_USER_HEADER) or "").strip()
|
|
if not admin_user:
|
|
raise HTTPException(status_code=401, detail=f"{detail_prefix} proxy auth required")
|
|
request.state.clientflow_admin_user = admin_user
|
|
return
|
|
|
|
if mode == "token":
|
|
expected = (settings.clientflow_admin_token or "").strip()
|
|
if not expected:
|
|
raise HTTPException(status_code=503, detail=f"{detail_prefix} auth not configured")
|
|
received = (
|
|
request.headers.get(TOKEN_HEADER)
|
|
or request.cookies.get(TOKEN_COOKIE)
|
|
or ""
|
|
).strip()
|
|
if not received or not compare_digest(received, expected):
|
|
raise HTTPException(status_code=401, detail=f"{detail_prefix} auth required")
|
|
return
|
|
|
|
if mode == "local":
|
|
if not _is_loopback_request(request):
|
|
raise HTTPException(status_code=401, detail=f"{detail_prefix} local access required")
|
|
return
|
|
|
|
# Settings validates the value, but fail closed if it is mutated at runtime.
|
|
raise HTTPException(status_code=503, detail=f"{detail_prefix} auth mode invalid")
|
|
|
|
|
|
def safe_local_redirect(referer: str | None, *, fallback: str) -> str:
|
|
"""Return only an absolute-path local redirect, preserving its query."""
|
|
value = (referer or "").strip()
|
|
if not value:
|
|
return fallback
|
|
parsed = urlsplit(value)
|
|
if parsed.scheme or parsed.netloc or not parsed.path.startswith("/") or parsed.path.startswith("//"):
|
|
return fallback
|
|
target = parsed.path
|
|
if parsed.query:
|
|
target += f"?{parsed.query}"
|
|
return target
|