my_voice_assistant_v2/app/api/admin.py

from fastapi import APIRouter, Depends, HTTPException, Request

from app.auth import is_admin_user, require_admin, require_admin_or_user
from app.config import settings
from app.dependencies import get_store
from app.schemas import UserCreate, UserCreated

router = APIRouter()


@router.get("/admin/request-headers")
async def request_headers(request: Request, key: str | None = None):
    """Discovery: zeigt die eingehenden HTTP-Header + Quell-IP.

    Hilft, hinter dem Reverse-Proxy/SSO den richtigen Identitaets-Header
    (TRUSTED_AUTH_HEADER) festzustellen.

    Henne-Ei: Beim Einrichten kennt das Gateway den SSO-Admin noch nicht. Daher
    ist der Endpoint auch per Admin-Key aufrufbar - als Query (`?key=...`, im Browser
    durchs SSO bequem) oder X-Admin-Key-Header. Nach dem Setup wieder meiden bzw.
    den Key rotieren (er landet sonst in Proxy-Logs).
    """
    expected = settings.admin_api_key.strip()
    provided = key or request.headers.get("x-admin-key")
    ok = bool(expected) and provided is not None and provided.strip() == expected
    if not ok and not is_admin_user(_current_user_or_none(request)):
        raise HTTPException(status_code=403, detail="Admin privileges required")
    return {
        "client": request.client.host if request.client else None,
        "headers": dict(request.headers),
    }


def _current_user_or_none(request: Request):
    from app.auth import authenticate, _bearer_token

    client_host = request.client.host if request.client else ""
    token = _bearer_token(request.headers.get("authorization"))
    return authenticate(request.headers, client_host, token)


@router.post("/admin/users", response_model=UserCreated, dependencies=[Depends(require_admin)])
async def create_user(payload: UserCreate):
    """Legt einen Nutzer an und gibt das Bearer-Token EINMALIG zurueck."""
    user, token = get_store().create_user(payload.display_name)
    return UserCreated(user_id=user.id, display_name=user.display_name, token=token)


@router.get("/admin/users", dependencies=[Depends(require_admin_or_user)])
async def list_users():
    """Listet die Nutzer (ohne Secrets). Fuer Admins (SSO/ADMIN_USERS) oder ADMIN_API_KEY."""
    return [
        {
            "user_id": u.id,
            "display_name": u.display_name,
            "external_id": u.external_id,
            "created_at": u.created_at,
        }
        for u in get_store().list_users()
    ]