karldns.de/karldns/routers/management_passwords.py

from __future__ import annotations

from collections.abc import Mapping
from dataclasses import dataclass

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

from ..dependencies import ServiceDep, SettingsDep, StoreDep
from ..event_statuses import DASHBOARD_PASSWORD_CHANGED, DASHBOARD_PASSWORD_ENABLED, DASHBOARD_PASSWORD_REMOVED
from ..security import client_ip, dashboard_csrf_token, require_dashboard_csrf
from ..store import DdnsStore
from ..web.forms import first_form_value, request_form
from ..web.templates_management import ManagementPageContext, _render_management_unlock_page
from .management_shared import (
    ManagementAccount,
    delete_unlock_cookie,
    get_active_management_account_async,
    has_management_access,
    management_error_response_async,
    render_management_page_async,
    set_unlock_cookie,
    unlock_cookie_seconds,
)
from .path_params import ManagementSlugPath

router = APIRouter()


@dataclass(frozen=True, slots=True)
class _DashboardPasswordErrorContext:
    management_slug: str
    account: ManagementAccount
    form: dict[str, list[str]]
    request_ip: str
    current_password_error: str | None


@router.post("/m/{management_slug}/unlock", include_in_schema=False)
async def unlock_dashboard(
    management_slug: ManagementSlugPath, request: Request, settings: SettingsDep, store: StoreDep
) -> Response:
    account = await get_active_management_account_async(store, management_slug)
    if not account.get("dashboard_password_hash"):
        return RedirectResponse(f"/m/{management_slug}", status_code=303)
    form = await request_form(request, settings)
    require_dashboard_csrf(management_slug, account, first_form_value(form, "csrf"))
    password = first_form_value(form, "dashboard_password")
    hostname = str(account["hostname"])
    is_valid = await run_in_threadpool(store.verify_dashboard_password, hostname, password)
    if not is_valid:
        return HTMLResponse(
            _render_management_unlock_page(
                management_slug, account, dashboard_csrf_token(management_slug, account), "Incorrect password."
            ),
            status_code=401,
        )
    response = RedirectResponse(f"/m/{management_slug}", status_code=303)
    set_unlock_cookie(
        response,
        settings,
        management_slug,
        account,
        unlock_cookie_seconds(settings, first_form_value(form, "stay_unlocked")),
    )
    return response


@router.post("/m/{management_slug}/password", include_in_schema=False)
async def set_dashboard_password(
    management_slug: ManagementSlugPath,
    request: Request,
    settings: SettingsDep,
    store: StoreDep,
    service: ServiceDep,
) -> Response:
    account = await get_active_management_account_async(store, management_slug)
    if not has_management_access(request, settings, management_slug, account):
        return HTMLResponse(
            _render_management_unlock_page(management_slug, account, dashboard_csrf_token(management_slug, account)),
            status_code=401,
        )
    form = await request_form(request, settings)
    require_dashboard_csrf(management_slug, account, first_form_value(form, "csrf"))
    hostname = str(account["hostname"])
    current_password_error = await _verify_current_dashboard_password(store, hostname, account, form)
    request_ip = client_ip(request, settings)
    error_response = await _dashboard_password_error_response(
        service,
        _DashboardPasswordErrorContext(
            management_slug=management_slug,
            account=account,
            form=form,
            request_ip=request_ip,
            current_password_error=current_password_error,
        ),
    )
    if error_response:
        return error_response
    password = first_form_value(form, "dashboard_password")
    had_password = bool(account.get("dashboard_password_hash"))
    await run_in_threadpool(store.set_dashboard_password, hostname, password)
    account = await get_active_management_account_async(store, management_slug)
    await run_in_threadpool(
        store.log_update_event,
        hostname,
        None,
        None,
        DASHBOARD_PASSWORD_CHANGED if had_password else DASHBOARD_PASSWORD_ENABLED,
        "dashboard password changed" if had_password else "dashboard password enabled",
        client_ip(request, settings),
    )
    response = RedirectResponse(f"/m/{management_slug}", status_code=303)
    set_unlock_cookie(response, settings, management_slug, account, settings.dashboard_unlock_seconds)
    return response


def _dashboard_password_error(password: str, confirm: str) -> str | None:
    if len(password) < 8:
        return "Password must be at least 8 characters."
    if password != confirm:
        return "The passwords do not match."
    return None


async def _dashboard_password_error_response(
    service: ServiceDep,
    context: _DashboardPasswordErrorContext,
) -> HTMLResponse | None:
    if context.current_password_error:
        return await management_error_response_async(
            service,
            context.management_slug,
            context.account,
            context.current_password_error,
            context.request_ip,
        )
    password_error = _dashboard_password_error(
        first_form_value(context.form, "dashboard_password"),
        first_form_value(context.form, "dashboard_password_confirm"),
    )
    if password_error is None:
        return None
    page = await render_management_page_async(
        service,
        ManagementPageContext(
            management_slug=context.management_slug,
            account=context.account,
            csrf_token=dashboard_csrf_token(context.management_slug, context.account),
            message=password_error,
            request_ip=context.request_ip,
        ),
    )
    return HTMLResponse(page, status_code=400)


async def _verify_current_dashboard_password(
    store: DdnsStore,
    hostname: str,
    account: Mapping[str, object],
    form: dict[str, list[str]],
) -> str | None:
    if not account.get("dashboard_password_hash"):
        return None
    is_current = await run_in_threadpool(
        store.verify_dashboard_password,
        hostname,
        first_form_value(form, "current_dashboard_password"),
    )
    return None if is_current else "Current password is incorrect."


@router.post("/m/{management_slug}/password/remove", include_in_schema=False)
async def remove_dashboard_password(
    management_slug: ManagementSlugPath,
    request: Request,
    settings: SettingsDep,
    store: StoreDep,
    service: ServiceDep,
) -> Response:
    account = await get_active_management_account_async(store, management_slug)
    if not has_management_access(request, settings, management_slug, account):
        return HTMLResponse(
            _render_management_unlock_page(management_slug, account, dashboard_csrf_token(management_slug, account)),
            status_code=401,
        )
    form = await request_form(request, settings)
    require_dashboard_csrf(management_slug, account, first_form_value(form, "csrf"))
    hostname = str(account["hostname"])
    if account.get("dashboard_password_hash"):
        is_current = await run_in_threadpool(
            store.verify_dashboard_password,
            hostname,
            first_form_value(form, "current_dashboard_password"),
        )
        if not is_current:
            return await management_error_response_async(
                service,
                management_slug,
                account,
                "Current password is incorrect.",
                client_ip(request, settings),
            )
    await run_in_threadpool(store.clear_dashboard_password, hostname)
    await run_in_threadpool(
        store.log_update_event,
        hostname,
        None,
        None,
        DASHBOARD_PASSWORD_REMOVED,
        "dashboard password removed",
        client_ip(request, settings),
    )
    response = RedirectResponse(f"/m/{management_slug}", status_code=303)
    delete_unlock_cookie(response, management_slug)
    return response