app/api/api_keys.py

from __future__ import annotations

from fastapi import APIRouter, HTTPException, Query, Request, Response, status

from app.security.errors import APIKeyConflictError, APIKeyNotFoundError
from app.security.schemas import (
    APIKeyCreate,
    APIKeyCreated,
    APIKeyList,
    APIKeyPatch,
    APIKeyRotate,
    APIKeyView,
    AuthContextView,
    AuditLogView,
)
from app.security.scopes import ALL_SCOPES

router = APIRouter(prefix="/v1", tags=["authentication"])


def _view(record: object) -> APIKeyView:
    return APIKeyView.model_validate(record)


def _not_found() -> HTTPException:
    return HTTPException(status_code=404, detail="API key was not found")


@router.get("/auth/context", response_model=AuthContextView)
async def current_auth_context(request: Request) -> AuthContextView:
    """Validate a key and return only its safe, non-secret authorization context."""
    context = request.state.auth
    return AuthContextView(
        id=context.api_key_id,
        name=context.key_name,
        key_prefix=context.key_prefix,
        environment=context.environment,
        role=context.role,
        scopes=sorted(context.scopes),
        expires_at=context.expires_at,
        workspace_id=context.workspace_id,
        user_id=context.user_id,
        membership_role=context.membership_role,
    )


@router.get("/api-keys/capabilities")
async def api_key_capabilities(request: Request) -> dict[str, object]:
    container = request.app.state.container
    return {
        "scopes": sorted(ALL_SCOPES),
        "roles": {
            role: sorted(scopes)
            for role, scopes in container.api_keys.roles.items()
        },
        "defaults": {
            "requests_per_minute": container.settings.auth_default_requests_per_minute,
            "concurrent_jobs": container.settings.auth_default_concurrent_jobs,
            "uploads_per_hour": container.settings.auth_default_uploads_per_hour,
            "processing_bytes_per_day": (
                container.settings.auth_default_processing_bytes_per_day
            ),
        },
    }


@router.get("/api-keys", response_model=APIKeyList)
async def list_api_keys(
    request: Request,
    offset: int = Query(default=0, ge=0),
    limit: int = Query(default=100, ge=1, le=500),
) -> APIKeyList:
    records, total = await request.app.state.container.api_keys.list(
        offset=offset, limit=limit
    )
    return APIKeyList(items=[_view(record) for record in records], total=total)


@router.post(
    "/api-keys", response_model=APIKeyCreated, status_code=status.HTTP_201_CREATED
)
async def create_api_key(request: Request, payload: APIKeyCreate) -> APIKeyCreated:
    context = request.state.auth
    try:
        record, secret = await request.app.state.container.api_keys.create(
            payload,
            created_by=context.api_key_id,
            workspace_id=context.workspace_id,
            user_id=context.user_id,
        )
    except APIKeyConflictError as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc
    return APIKeyCreated(**_view(record).model_dump(), api_key=secret)


@router.get("/api-keys/{key_id}", response_model=APIKeyView)
async def get_api_key(request: Request, key_id: str) -> APIKeyView:
    try:
        return _view(await request.app.state.container.api_keys.get(key_id))
    except APIKeyNotFoundError as exc:
        raise _not_found() from exc


@router.patch("/api-keys/{key_id}", response_model=APIKeyView)
async def patch_api_key(
    request: Request, key_id: str, payload: APIKeyPatch
) -> APIKeyView:
    try:
        return _view(await request.app.state.container.api_keys.patch(key_id, payload))
    except APIKeyNotFoundError as exc:
        raise _not_found() from exc
    except APIKeyConflictError as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc


@router.delete("/api-keys/{key_id}", status_code=status.HTTP_204_NO_CONTENT)
async def revoke_api_key(request: Request, key_id: str) -> Response:
    try:
        await request.app.state.container.api_keys.set_status(key_id, "revoked")
    except APIKeyNotFoundError as exc:
        raise _not_found() from exc
    return Response(status_code=status.HTTP_204_NO_CONTENT)


@router.post(
    "/api-keys/{key_id}/rotate",
    response_model=APIKeyCreated,
    status_code=status.HTTP_201_CREATED,
)
async def rotate_api_key(
    request: Request, key_id: str, payload: APIKeyRotate
) -> APIKeyCreated:
    try:
        record, secret = await request.app.state.container.api_keys.rotate(
            key_id,
            payload.grace_period_seconds,
            created_by=request.state.auth.api_key_id,
        )
    except APIKeyNotFoundError as exc:
        raise _not_found() from exc
    except APIKeyConflictError as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc
    return APIKeyCreated(**_view(record).model_dump(), api_key=secret)


@router.post("/api-keys/{key_id}/disable", response_model=APIKeyView)
async def disable_api_key(request: Request, key_id: str) -> APIKeyView:
    try:
        return _view(
            await request.app.state.container.api_keys.set_status(key_id, "disabled")
        )
    except APIKeyNotFoundError as exc:
        raise _not_found() from exc
    except APIKeyConflictError as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc


@router.post("/api-keys/{key_id}/enable", response_model=APIKeyView)
async def enable_api_key(request: Request, key_id: str) -> APIKeyView:
    try:
        return _view(
            await request.app.state.container.api_keys.set_status(key_id, "active")
        )
    except APIKeyNotFoundError as exc:
        raise _not_found() from exc
    except APIKeyConflictError as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc


@router.get("/audit-logs", response_model=list[AuditLogView])
async def list_audit_logs(
    request: Request,
    offset: int = Query(default=0, ge=0),
    limit: int = Query(default=100, ge=1, le=500),
) -> list[AuditLogView]:
    records = await request.app.state.container.audit.list(offset=offset, limit=limit)
    return [AuditLogView.model_validate(record) for record in records]