admin_security.py

"""Access checks for administrative API endpoints."""
from __future__ import annotations

from typing import Mapping, Optional

from app_config import AppSettings, load_settings


LOOPBACK_HOSTS = {"127.0.0.1", "::1", "localhost"}


def _normalize_headers(headers: Mapping[str, str]) -> dict:
    return {str(key).lower(): str(value) for key, value in headers.items()}


def _bearer_token(headers: Mapping[str, str]) -> Optional[str]:
    auth = _normalize_headers(headers).get("authorization", "")
    prefix = "bearer "
    if auth.lower().startswith(prefix):
        return auth[len(prefix):].strip()
    return None


def is_loopback_host(host: Optional[str]) -> bool:
    if not host:
        return False
    return host.split("%", 1)[0] in LOOPBACK_HOSTS


def is_admin_request_allowed(
    headers: Mapping[str, str],
    client_host: Optional[str],
    settings: Optional[AppSettings] = None,
) -> bool:
    settings = settings or load_settings()

    if settings.admin_local_only and not settings.public_tunnel_enabled:
        return is_loopback_host(client_host)

    token = _bearer_token(headers)
    return bool(settings.admin_token and token == settings.admin_token)