karldns.de/karldns/routers/management_shared.py

from __future__ import annotations

from fastapi import HTTPException, Request, Response
from fastapi.responses import HTMLResponse
from starlette.concurrency import run_in_threadpool

from ..security import (
    DASHBOARD_UNLOCK_COOKIE,
    DashboardUnlockIdentity,
    dashboard_cookie_signing_secret,
    dashboard_csrf_token,
    sign_dashboard_unlock_cookie,
    verify_dashboard_unlock_cookie,
)
from ..services.ddns import DdnsService
from ..settings import DdnsSettings
from ..store import DdnsStore
from ..store.types import AccountRow
from ..web.templates_management import ManagementPageContext, _render_management_page, _render_management_unlock_page

ManagementAccount = AccountRow


def get_active_management_account(store: DdnsStore, management_slug: str) -> ManagementAccount:
    account = store.get_account_by_management_slug(management_slug)
    if not account or account["disabled"]:
        raise HTTPException(status_code=404, detail="not found")
    return account


async def get_active_management_account_async(store: DdnsStore, management_slug: str) -> ManagementAccount:
    return await run_in_threadpool(get_active_management_account, store, management_slug)


def management_account_is_accepted(account: ManagementAccount) -> bool:
    return bool(account.get("accepted_at"))


def has_management_access(
    request: Request, settings: DdnsSettings, management_slug: str, account: ManagementAccount
) -> bool:
    password_hash = account.get("dashboard_password_hash")
    if not password_hash:
        return True
    return verify_dashboard_unlock_cookie(
        str(account["hostname"]),
        management_slug,
        str(password_hash),
        dashboard_cookie_signing_secret(settings),
        request.cookies.get(DASHBOARD_UNLOCK_COOKIE),
    )


def locked_management_response(management_slug: str, account: ManagementAccount) -> HTMLResponse:
    return HTMLResponse(
        _render_management_unlock_page(management_slug, account, dashboard_csrf_token(management_slug, account)),
        status_code=401,
    )


def management_error_response(
    service: DdnsService,
    management_slug: str,
    account: ManagementAccount,
    message: str,
    request_ip: str | None = None,
) -> HTMLResponse:
    return HTMLResponse(
        _render_management_page(
            service,
            ManagementPageContext(
                management_slug=management_slug,
                account=account,
                csrf_token=dashboard_csrf_token(management_slug, account),
                message=message,
                request_ip=request_ip,
            ),
        ),
        status_code=400,
    )


async def management_error_response_async(
    service: DdnsService,
    management_slug: str,
    account: ManagementAccount,
    message: str,
    request_ip: str | None = None,
) -> HTMLResponse:
    return await run_in_threadpool(management_error_response, service, management_slug, account, message, request_ip)


async def render_management_page_async(
    service: DdnsService,
    context: ManagementPageContext,
) -> str:
    return await run_in_threadpool(
        _render_management_page,
        service,
        context,
    )


def set_unlock_cookie(
    response: Response,
    settings: DdnsSettings,
    management_slug: str,
    account: ManagementAccount,
    max_age_seconds: int,
) -> None:
    password_hash = account.get("dashboard_password_hash")
    signing_secret = dashboard_cookie_signing_secret(settings)
    if not password_hash or not signing_secret:
        return
    response.set_cookie(
        DASHBOARD_UNLOCK_COOKIE,
        sign_dashboard_unlock_cookie(
            DashboardUnlockIdentity(str(account["hostname"]), management_slug, str(password_hash)),
            signing_secret,
            max_age_seconds,
        ),
        max_age=max_age_seconds,
        path=f"/m/{management_slug}",
        httponly=True,
        samesite="lax",
        secure=settings.public_base_url.startswith("https://"),
    )


def delete_unlock_cookie(response: Response, management_slug: str) -> None:
    response.delete_cookie(DASHBOARD_UNLOCK_COOKIE, path=f"/m/{management_slug}", httponly=True, samesite="lax")


def unlock_cookie_seconds(settings: DdnsSettings, stay_unlocked: str) -> int:
    if stay_unlocked:
        return settings.dashboard_stay_unlocked_seconds
    return settings.dashboard_unlock_seconds