karldns.de/karldns/routers/api.py

from __future__ import annotations

from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response, status
from starlette.concurrency import run_in_threadpool

from ..dependencies import ServiceDep, StoreDep, authenticate_admin
from ..dns import DnsProviderError
from ..event_statuses import DASHBOARD_UPDATE_URL_ROTATED
from ..schemas import (
    AdminDashboardResponse,
    AdminEventsResponse,
    AdminHostnamesResponse,
    AdminRecordsResponse,
    AdminTotals,
    ErrorResponse,
    HostnameCreateResponse,
    HostnameRotateUpdateResponse,
    HostnameStatusResponse,
    _admin_cleanup_run,
    _admin_hostname,
    _admin_metrics,
    _admin_record,
    _admin_scheduled_task,
    _admin_update_event,
    _create_response,
    _hostname_rotate_update_response,
    _hostname_status,
)
from ..security import client_ip
from .path_params import ManagementSlugPath

router = APIRouter(prefix="/api/v1")
ADMIN_API_LIMIT_CAP = 500
ADMIN_API_OFFSET_CAP = 10_000
HOSTNAME_HISTORY_LIMIT_CAP = 100
AdminAuthDep = Annotated[None, Depends(authenticate_admin)]
HostnameHistoryLimitQuery = Annotated[int, Query(ge=0, le=HOSTNAME_HISTORY_LIMIT_CAP)]
AdminLimitQuery = Annotated[int, Query(ge=1, le=ADMIN_API_LIMIT_CAP)]
AdminOffsetQuery = Annotated[int, Query(ge=0, le=ADMIN_API_OFFSET_CAP)]
AdminDashboardLimitQuery = Annotated[int, Query(ge=1, le=ADMIN_API_LIMIT_CAP)]
AdminBeforeQuery = Annotated[str | None, Query(max_length=64)]


@router.post(
    "/hostnames",
    response_model=HostnameCreateResponse,
    status_code=status.HTTP_201_CREATED,
    summary="Create a KarlDNS hostname",
    tags=["hostnames"],
    responses={500: {"model": ErrorResponse}},
)
async def create_hostname(request: Request, service: ServiceDep) -> HostnameCreateResponse:
    account = await run_in_threadpool(service.create_managed_account)
    request.state.account_created = True
    return _create_response(service, account)


@router.get(
    "/hostnames/{dashboard_secret}",
    response_model=HostnameStatusResponse,
    summary="Read a hostname by private dashboard link",
    description=(
        "Uses the private dashboard secret. Dashboard passwords protect only the web dashboard, "
        "not this API endpoint."
    ),
    tags=["hostnames"],
    responses={404: {"model": ErrorResponse}},
)
async def read_hostname(
    dashboard_secret: ManagementSlugPath,
    store: StoreDep,
    service: ServiceDep,
    history_limit: HostnameHistoryLimitQuery = 10,
) -> HostnameStatusResponse:
    account = await run_in_threadpool(store.get_account_by_management_slug, dashboard_secret)
    if not account or account["disabled"]:
        raise HTTPException(status_code=404, detail="not found")
    account["management_slug"] = dashboard_secret
    history = (
        await run_in_threadpool(store.list_update_events_for_hostname, str(account["hostname"]), history_limit)
        if history_limit
        else []
    )
    return _hostname_status(service, account, history)


@router.post(
    "/hostnames/{dashboard_secret}/rotate-update-url",
    response_model=HostnameRotateUpdateResponse,
    summary="Rotate a hostname update URL",
    description="Uses the private dashboard secret. The old router update URL stops working immediately.",
    tags=["hostnames"],
    responses={400: {"model": ErrorResponse}, 404: {"model": ErrorResponse}},
)
async def rotate_hostname_update_url(
    dashboard_secret: ManagementSlugPath,
    request: Request,
    store: StoreDep,
    service: ServiceDep,
) -> HostnameRotateUpdateResponse:
    account = await run_in_threadpool(store.get_account_by_management_slug, dashboard_secret)
    if not account or account["disabled"]:
        raise HTTPException(status_code=404, detail="not found")
    if not account["accepted_at"]:
        raise HTTPException(status_code=400, detail="register hostname before rotating update URL")
    hostname = str(account["hostname"])
    rotated = await run_in_threadpool(store.rotate_update_slug, hostname)
    if not rotated:
        raise HTTPException(status_code=404, detail="not found")
    await run_in_threadpool(
        store.log_update_event,
        hostname,
        None,
        None,
        DASHBOARD_UPDATE_URL_ROTATED,
        "router update URL rotated by API",
        client_ip(request, service.settings),
    )
    return _hostname_rotate_update_response(service, rotated, dashboard_secret)


@router.delete(
    "/hostnames/{dashboard_secret}",
    status_code=status.HTTP_204_NO_CONTENT,
    summary="Delete a hostname by private dashboard link",
    description=(
        "Uses the private dashboard secret. Dashboard passwords protect only the web dashboard, "
        "not this API endpoint."
    ),
    tags=["hostnames"],
    responses={404: {"model": ErrorResponse}, 500: {"model": ErrorResponse}},
)
async def delete_hostname(dashboard_secret: ManagementSlugPath, store: StoreDep, service: ServiceDep) -> Response:
    account = await run_in_threadpool(store.get_account_by_management_slug, dashboard_secret)
    if not account or account["disabled"]:
        raise HTTPException(status_code=404, detail="not found")
    hostname = str(account["hostname"])
    try:
        await run_in_threadpool(service.delete_account, hostname, "api")
    except DnsProviderError as exc:
        raise HTTPException(status_code=500, detail="DNS delete failed. Hostname was not deleted.") from exc
    return Response(status_code=status.HTTP_204_NO_CONTENT)


@router.get(
    "/admin/dashboard",
    response_model=AdminDashboardResponse,
    summary="Read operator dashboard summary",
    description="Requires admin HTTP Basic auth. Returns the same high-level data shown in the operator console.",
    tags=["admin"],
    include_in_schema=False,
    responses={401: {"model": ErrorResponse}},
)
async def read_admin(
    store: StoreDep,
    _: AdminAuthDep,
    limit: AdminDashboardLimitQuery = 25,
) -> AdminDashboardResponse:
    metrics = await run_in_threadpool(store.get_service_metrics)
    accounts = await run_in_threadpool(store.list_accounts, limit, 0)
    cleanup_runs = await run_in_threadpool(store.list_cleanup_runs, limit, 0)
    scheduled_tasks = await run_in_threadpool(store.list_scheduled_tasks)
    events = await run_in_threadpool(store.list_update_events, limit, 0)
    return AdminDashboardResponse(
        totals=AdminTotals(
            hostnames=await run_in_threadpool(store.count_accounts),
            records=await run_in_threadpool(store.count_records),
            update_events=await run_in_threadpool(store.count_update_events),
            cleanup_runs=await run_in_threadpool(store.count_cleanup_runs),
        ),
        metrics=_admin_metrics(metrics),
        hostnames=[_admin_hostname(account) for account in accounts],
        update_events=[_admin_update_event(event) for event in events],
        cleanup_runs=[_admin_cleanup_run(cleanup_run) for cleanup_run in cleanup_runs],
        scheduled_tasks=[_admin_scheduled_task(task) for task in scheduled_tasks],
        limit=limit,
    )


@router.get(
    "/admin/hostnames",
    response_model=AdminHostnamesResponse,
    summary="List operator hostnames",
    description="Requires admin HTTP Basic auth.",
    tags=["admin"],
    include_in_schema=False,
    responses={401: {"model": ErrorResponse}},
)
async def list_admin_hostnames(
    store: StoreDep,
    _: AdminAuthDep,
    limit: AdminLimitQuery = 100,
    offset: AdminOffsetQuery = 0,
) -> AdminHostnamesResponse:
    accounts = await run_in_threadpool(store.list_accounts, limit, offset)
    total = await run_in_threadpool(store.count_accounts)
    return AdminHostnamesResponse(
        items=[_admin_hostname(account) for account in accounts],
        limit=limit,
        offset=offset,
        total=total,
    )


@router.get(
    "/admin/events",
    response_model=AdminEventsResponse,
    summary="List operator update history",
    description="Requires admin HTTP Basic auth. Supports offset pagination or keyset pagination with before.",
    tags=["admin"],
    include_in_schema=False,
    responses={401: {"model": ErrorResponse}},
)
async def list_admin_events(
    store: StoreDep,
    _: AdminAuthDep,
    limit: AdminLimitQuery = 100,
    offset: AdminOffsetQuery = 0,
    before: AdminBeforeQuery = None,
) -> AdminEventsResponse:
    items = (
        await run_in_threadpool(store.list_update_events_before, limit, before)
        if before
        else await run_in_threadpool(store.list_update_events_page, limit, offset)
    )
    next_before = str(items[-1]["created_at"]) if len(items) == limit and items[-1].get("created_at") else None
    total = await run_in_threadpool(store.count_update_events)
    return AdminEventsResponse(
        items=[_admin_update_event(item) for item in items],
        limit=limit,
        offset=offset,
        next_before=next_before,
        total=total,
    )


@router.get(
    "/admin/records",
    response_model=AdminRecordsResponse,
    summary="List operator DNS records",
    description="Requires admin HTTP Basic auth.",
    tags=["admin"],
    include_in_schema=False,
    responses={401: {"model": ErrorResponse}},
)
async def list_admin_records(
    store: StoreDep,
    _: AdminAuthDep,
    limit: AdminLimitQuery = 100,
    offset: AdminOffsetQuery = 0,
) -> AdminRecordsResponse:
    items = await run_in_threadpool(store.list_records_page, limit, offset)
    total = await run_in_threadpool(store.count_records)
    return AdminRecordsResponse(
        items=[_admin_record(item) for item in items],
        limit=limit,
        offset=offset,
        total=total,
    )