diff --git a/README.md b/README.md index 00a999f..29d2e00 100644 --- a/README.md +++ b/README.md @@ -427,7 +427,9 @@ Backend layout: | `karldns/dns/` | Cloudflare and RFC 2136 DNS publishing. | ```bash +black --check karldns tests ruff check karldns tests +mypy karldns pytest python -m compileall karldns ``` diff --git a/karldns/dns/cloudflare.py b/karldns/dns/cloudflare.py index b3d1ecc..6ee2600 100644 --- a/karldns/dns/cloudflare.py +++ b/karldns/dns/cloudflare.py @@ -4,6 +4,7 @@ import json import urllib.error import urllib.parse import urllib.request +from typing import Any, cast from ..models import UpdateResult from ..settings import DdnsSettings @@ -49,11 +50,13 @@ def _cloudflare_upsert_record(settings: DdnsSettings, hostname: str, record_type _cloudflare_reconcile_record(settings, hostname, record_type, value) -def _cloudflare_find_record(settings: DdnsSettings, hostname: str, record_type: str) -> dict[str, str] | None: +def _cloudflare_find_record(settings: DdnsSettings, hostname: str, record_type: str) -> dict[str, Any] | None: query = urllib.parse.urlencode({"type": record_type, "name": hostname}) response = _cloudflare_request(settings, f"/dns_records?{query}") records = response.get("result", []) - return records[0] if records else None + if not isinstance(records, list) or not records: + return None + return cast(dict[str, Any], records[0]) def _cloudflare_reconcile_record(settings: DdnsSettings, hostname: str, record_type: str, value: str) -> None: @@ -69,7 +72,7 @@ def _cloudflare_request( path: str, method: str = "GET", payload: dict[str, object] | None = None, -) -> dict[str, object]: +) -> dict[str, Any]: url = f"https://api.cloudflare.com/client/v4/zones/{settings.cloudflare_zone_id}{path}" body = json.dumps(payload).encode("utf-8") if payload is not None else None request = urllib.request.Request(url, data=body, method=method) diff --git a/karldns/routers/admin.py b/karldns/routers/admin.py index 471b9e3..fc2a7ca 100644 --- a/karldns/routers/admin.py +++ b/karldns/routers/admin.py @@ -67,7 +67,9 @@ def make_router(admin_path: str = "/admin") -> APIRouter: cleanup_events: int = Query(0, ge=0), ) -> str: notice = ( - _admin_notice("success", f"Cleanup finished. Removed {cleanup_unused} hostnames and {cleanup_events} old events.") + _admin_notice( + "success", f"Cleanup finished. Removed {cleanup_unused} hostnames and {cleanup_events} old events." + ) if cleanup_done else None ) @@ -138,7 +140,13 @@ def make_router(admin_path: str = "/admin") -> APIRouter: "router update URL rotated by admin", client_ip(request, settings), ) - notice = _admin_secret_panel("Update URL changed", [_copy_row("Update-URL:", service.fritz_update_url(rotated), "Secret")]) if rotated else None + notice = ( + _admin_secret_panel( + "Update URL changed", [_copy_row("Update-URL:", service.fritz_update_url(rotated), "Secret")] + ) + if rotated + else None + ) elif action == "management": rotated = await run_in_threadpool(store.rotate_management_slug, hostname) await run_in_threadpool( @@ -150,7 +158,14 @@ def make_router(admin_path: str = "/admin") -> APIRouter: "dashboard link rotated by admin", client_ip(request, settings), ) - notice = _admin_secret_panel("Dashboard link changed", [_copy_row("Private dashboard:", service.dashboard_url(rotated), "Secret")]) if rotated else None + notice = ( + _admin_secret_panel( + "Dashboard link changed", + [_copy_row("Private dashboard:", service.dashboard_url(rotated), "Secret")], + ) + if rotated + else None + ) else: raise HTTPException(status_code=400, detail="invalid rotation action") page = await run_in_threadpool(render_admin, service, settings, store, None, notice) diff --git a/karldns/routers/api.py b/karldns/routers/api.py index 1a2e4ab..ea1f186 100644 --- a/karldns/routers/api.py +++ b/karldns/routers/api.py @@ -16,8 +16,12 @@ from ..schemas import ( 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, @@ -62,7 +66,11 @@ async def read_hostname( 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 [] + 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) @@ -149,9 +157,9 @@ async def read_admin( ), metrics=_admin_metrics(metrics), hostnames=[_admin_hostname(account) for account in accounts], - update_events=events, - cleanup_runs=cleanup_runs, - scheduled_tasks=scheduled_tasks, + 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, ) @@ -204,7 +212,13 @@ async def list_admin_events( ) 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=items, limit=limit, offset=offset, next_before=next_before, total=total) + return AdminEventsResponse( + items=[_admin_update_event(item) for item in items], + limit=limit, + offset=offset, + next_before=next_before, + total=total, + ) @router.get( @@ -224,4 +238,9 @@ async def list_admin_records( ) -> AdminRecordsResponse: items = await run_in_threadpool(store.list_records_page, limit, offset) total = await run_in_threadpool(store.count_records) - return AdminRecordsResponse(items=items, limit=limit, offset=offset, total=total) + return AdminRecordsResponse( + items=[_admin_record(item) for item in items], + limit=limit, + offset=offset, + total=total, + ) diff --git a/karldns/routers/assets.py b/karldns/routers/assets.py index 4413d4f..c623ee3 100644 --- a/karldns/routers/assets.py +++ b/karldns/routers/assets.py @@ -14,7 +14,16 @@ ASSET_CACHE_HEADERS = {"Cache-Control": "public, max-age=31536000, immutable"} def _minify_css(css: str) -> str: - return " ".join(css.split()).replace(" {", "{").replace("{ ", "{").replace("; ", ";").replace(": ", ":").replace(", ", ",").replace(" }", "}").replace(";}", "}") + return ( + " ".join(css.split()) + .replace(" {", "{") + .replace("{ ", "{") + .replace("; ", ";") + .replace(": ", ":") + .replace(", ", ",") + .replace(" }", "}") + .replace(";}", "}") + ) MINIFIED_BASE_CSS = _minify_css(BASE_CSS) @@ -25,50 +34,62 @@ MINIFIED_ADMIN_CSS = _minify_css(ADMIN_CSS) def favicon() -> FileResponse: return FileResponse(ASSET_DIR / "favicon.ico", media_type="image/x-icon", headers=ASSET_CACHE_HEADERS) + @router.get("/bootstrap-subset.css") def bootstrap_subset_css() -> FileResponse: return FileResponse(ASSET_DIR / "bootstrap-subset.css", media_type="text/css", headers=ASSET_CACHE_HEADERS) + @router.get("/favicon-16x16.png") def favicon_16_png() -> FileResponse: return FileResponse(ASSET_DIR / "favicon-16x16.png", media_type="image/png", headers=ASSET_CACHE_HEADERS) + @router.get("/favicon-32x32.png") def favicon_32_png() -> FileResponse: return FileResponse(ASSET_DIR / "favicon-32x32.png", media_type="image/png", headers=ASSET_CACHE_HEADERS) + @router.get("/apple-touch-icon.png") def apple_touch_icon_png() -> FileResponse: return FileResponse(ASSET_DIR / "apple-touch-icon.png", media_type="image/png", headers=ASSET_CACHE_HEADERS) + @router.get("/logo.png") def logo_png() -> FileResponse: return FileResponse(ASSET_DIR / "logo.png", media_type="image/png", headers=ASSET_CACHE_HEADERS) + @router.get("/logo.webp") def logo_webp() -> FileResponse: return FileResponse(ASSET_DIR / "logo.webp", media_type="image/webp", headers=ASSET_CACHE_HEADERS) + @router.get("/logo-nav.webp") def logo_nav_webp() -> FileResponse: return FileResponse(ASSET_DIR / "logo-nav.webp", media_type="image/webp", headers=ASSET_CACHE_HEADERS) + @router.get("/logo-192.png") def logo_192_png() -> FileResponse: return FileResponse(ASSET_DIR / "logo-192.png", media_type="image/png", headers=ASSET_CACHE_HEADERS) + @router.get("/logo-512.png") def logo_512_png() -> FileResponse: return FileResponse(ASSET_DIR / "logo-512.png", media_type="image/png", headers=ASSET_CACHE_HEADERS) + @router.get("/social-preview.png") def social_preview_png() -> FileResponse: return FileResponse(ASSET_DIR / "social-preview.png", media_type="image/png", headers=ASSET_CACHE_HEADERS) + @router.get("/social-preview.webp") def social_preview_webp() -> FileResponse: return FileResponse(ASSET_DIR / "social-preview.webp", media_type="image/webp", headers=ASSET_CACHE_HEADERS) + @router.get("/site.webmanifest") def site_webmanifest() -> Response: return Response( @@ -92,6 +113,7 @@ def site_webmanifest() -> Response: headers=ASSET_CACHE_HEADERS, ) + @router.get("/theme.js") def theme_js() -> Response: return Response(THEME_JS, media_type="application/javascript", headers=ASSET_CACHE_HEADERS) diff --git a/karldns/routers/management.py b/karldns/routers/management.py index 3f6d361..bb60a01 100644 --- a/karldns/routers/management.py +++ b/karldns/routers/management.py @@ -50,14 +50,24 @@ def manage_dashboard( account = get_active_management_account(store, management_slug) csrf_token = dashboard_csrf_token(management_slug, account) if not management_account_is_accepted(account): - return _render_pending_management_page(service, settings, management_slug, account, csrf_token, request_ip=client_ip(request, settings)) + return _render_pending_management_page( + service, settings, management_slug, account, csrf_token, request_ip=client_ip(request, settings) + ) if not has_management_access(request, settings, management_slug, account): return _render_management_unlock_page(management_slug, account, csrf_token) message = "Hostname registered. Add the update URL to your router." if activated else None - return _render_management_page(service, settings, management_slug, account, csrf_token, message=message, request_ip=client_ip(request, settings)) + return _render_management_page( + service, + settings, + management_slug, + account, + csrf_token, + message=message, + request_ip=client_ip(request, settings), + ) -@router.post("/m/{management_slug}/activate", include_in_schema=False) +@router.post("/m/{management_slug}/activate", include_in_schema=False, response_model=None) async def activate_dashboard( management_slug: ManagementSlugPath, request: Request, @@ -68,7 +78,7 @@ async def activate_dashboard( ipaddr: str | None = None, myipv6: str | None = None, ip6addr: str | None = None, -) -> RedirectResponse: +) -> RedirectResponse | HTMLResponse: account = await get_active_management_account_async(store, management_slug) form = await request_form(request, settings) require_dashboard_csrf(management_slug, account, first_form_value(form, "csrf")) @@ -102,8 +112,8 @@ async def activate_dashboard( ), status_code=500, ) - account = await run_in_threadpool(store.activate_account, hostname) - if not account: + activated_account = await run_in_threadpool(lambda: store.activate_account(hostname)) + if not activated_account: return _not_found() await run_in_threadpool( store.log_update_event, @@ -117,14 +127,14 @@ async def activate_dashboard( return RedirectResponse(f"/m/{management_slug}?activated=1", status_code=303) -@router.post("/m/{management_slug}/rotate", response_class=HTMLResponse, include_in_schema=False) +@router.post("/m/{management_slug}/rotate", response_class=HTMLResponse, include_in_schema=False, response_model=None) async def rotate_dashboard_secrets( management_slug: ManagementSlugPath, request: Request, settings: SettingsDep, store: StoreDep, service: ServiceDep, -) -> str: +) -> str | HTMLResponse: account = await get_active_management_account_async(store, management_slug) if not management_account_is_accepted(account): return HTMLResponse( @@ -172,10 +182,15 @@ async def rotate_dashboard_secrets( @router.post("/m/{management_slug}/delete", include_in_schema=False) -async def delete_dashboard_hostname(management_slug: ManagementSlugPath, request: Request, settings: SettingsDep, store: StoreDep, service: ServiceDep) -> Response: +async def delete_dashboard_hostname( + 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) + 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"]) @@ -195,8 +210,7 @@ async def delete_dashboard_hostname(management_slug: ManagementSlugPath, request return HTMLResponse( _page( "Hostname deleted", - trusted( - """ + trusted("""
@@ -207,8 +221,7 @@ async def delete_dashboard_hostname(management_slug: ManagementSlugPath, request
- """ - ), + """), ) ) diff --git a/karldns/routers/management_passwords.py b/karldns/routers/management_passwords.py index 12cf679..ef57ad4 100644 --- a/karldns/routers/management_passwords.py +++ b/karldns/routers/management_passwords.py @@ -24,7 +24,9 @@ router = APIRouter() @router.post("/m/{management_slug}/unlock", include_in_schema=False) -async def unlock_dashboard(management_slug: ManagementSlugPath, request: Request, settings: SettingsDep, store: StoreDep) -> Response: +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) @@ -35,7 +37,9 @@ async def unlock_dashboard(management_slug: ManagementSlugPath, request: Request 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."), + _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) @@ -59,7 +63,10 @@ async def set_dashboard_password( ) -> 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) + 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"]) @@ -129,7 +136,10 @@ async def remove_dashboard_password( ) -> 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) + 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"]) diff --git a/karldns/routers/management_shared.py b/karldns/routers/management_shared.py index cc3e62b..49586b3 100644 --- a/karldns/routers/management_shared.py +++ b/karldns/routers/management_shared.py @@ -35,7 +35,9 @@ 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: +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 @@ -49,7 +51,10 @@ def has_management_access(request: Request, settings: DdnsSettings, management_s 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) + return HTMLResponse( + _render_management_unlock_page(management_slug, account, dashboard_csrf_token(management_slug, account)), + status_code=401, + ) def management_error_response( @@ -82,7 +87,9 @@ async def management_error_response_async( message: str, request_ip: str | None = None, ) -> HTMLResponse: - return await run_in_threadpool(management_error_response, service, settings, management_slug, account, message, request_ip) + return await run_in_threadpool( + management_error_response, service, settings, management_slug, account, message, request_ip + ) async def render_management_page_async( diff --git a/karldns/routers/updates.py b/karldns/routers/updates.py index 387873b..284cf2a 100644 --- a/karldns/routers/updates.py +++ b/karldns/routers/updates.py @@ -152,7 +152,9 @@ async def _update_hostname( raise HTTPException(status_code=404, detail="not found") if not service.hostname_is_managed(str(account["hostname"])): raise HTTPException(status_code=404, detail="not found") - return await run_in_threadpool(apply_update_response, service, request, str(account["hostname"]), myip, ipaddr, myipv6, ip6addr) + return await run_in_threadpool( + apply_update_response, service, request, str(account["hostname"]), myip, ipaddr, myipv6, ip6addr + ) async def _ip_params_from_request( diff --git a/karldns/schemas.py b/karldns/schemas.py index 6265627..1a59d12 100644 --- a/karldns/schemas.py +++ b/karldns/schemas.py @@ -1,3 +1,5 @@ +from collections.abc import Mapping, Sequence + from pydantic import BaseModel, Field from .services.ddns import DdnsService @@ -12,7 +14,9 @@ class HostnameCreateResponse(BaseModel): status: str = Field(description="Lifecycle state. Newly generated hostnames start as pending until registered.") update_url: str = Field(description="Secret update URL for router DynDNS updates.") dashboard_url: str = Field(description="Secret dashboard URL for reading, rotating, or deleting this hostname.") - registration_url: str = Field(description="Secret URL used to register the pending hostname after secrets are saved.") + registration_url: str = Field( + description="Secret URL used to register the pending hostname after secrets are saved." + ) class HostnameStatusResponse(BaseModel): @@ -134,9 +138,9 @@ class AdminEventsResponse(BaseModel): total: int -def _create_response(service: DdnsService, account: dict[str, str]) -> HostnameCreateResponse: +def _create_response(service: DdnsService, account: Mapping[str, object]) -> HostnameCreateResponse: return HostnameCreateResponse( - hostname=account["hostname"], + hostname=str(account["hostname"]), status="active" if account.get("accepted_at") else "pending", update_url=service.fritz_update_url(account), dashboard_url=service.dashboard_url(account), @@ -146,8 +150,8 @@ def _create_response(service: DdnsService, account: dict[str, str]) -> HostnameC def _hostname_status( service: DdnsService, - account: dict[str, str | int | None], - history: list[dict[str, str | int | None]] | None = None, + account: Mapping[str, object], + history: Sequence[Mapping[str, object]] | None = None, ) -> HostnameStatusResponse: account_for_url = {"update_slug": str(account["update_slug"])} account_for_dashboard = {"management_slug": str(account.get("management_slug", ""))} @@ -163,7 +167,9 @@ def _hostname_status( ) -def _hostname_rotate_update_response(service: DdnsService, account: dict[str, str | int | None], dashboard_secret: str) -> HostnameRotateUpdateResponse: +def _hostname_rotate_update_response( + service: DdnsService, account: Mapping[str, object], dashboard_secret: str +) -> HostnameRotateUpdateResponse: return HostnameRotateUpdateResponse( hostname=str(account["hostname"]), status="active" if account.get("accepted_at") else "pending", @@ -172,7 +178,7 @@ def _hostname_rotate_update_response(service: DdnsService, account: dict[str, st ) -def _hostname_update_event(row: dict[str, str | int | None]) -> HostnameUpdateEvent: +def _hostname_update_event(row: Mapping[str, object]) -> HostnameUpdateEvent: return HostnameUpdateEvent( hostname=str(row["hostname"]), ipv4=str(row["ipv4"]) if row.get("ipv4") else None, @@ -180,11 +186,11 @@ def _hostname_update_event(row: dict[str, str | int | None]) -> HostnameUpdateEv status=str(row["status"]), detail=str(row["detail"]), created_at=str(row["created_at"]), - count=int(row.get("count") or 1), + count=_int_value(row.get("count"), 1), ) -def _admin_hostname(account: dict[str, str | int | None]) -> AdminHostname: +def _admin_hostname(account: Mapping[str, object]) -> AdminHostname: return AdminHostname( hostname=str(account["hostname"]), status="disabled" if account.get("disabled") else ("active" if account.get("accepted_at") else "pending"), @@ -201,3 +207,59 @@ def _admin_metrics(metrics: dict[str, float]) -> AdminMetrics: samples = metrics.get("update_latency_samples_total", 0) average_latency = metrics.get("update_latency_ms_total", 0) / samples if samples else 0 return AdminMetrics(values=metrics, average_update_latency_ms=average_latency) + + +def _admin_record(row: Mapping[str, object]) -> AdminRecord: + return AdminRecord( + hostname=str(row["hostname"]), + ipv4=str(row["ipv4"]) if row.get("ipv4") else None, + ipv6=str(row["ipv6"]) if row.get("ipv6") else None, + updated_at=str(row["updated_at"]), + ) + + +def _admin_update_event(row: Mapping[str, object]) -> AdminUpdateEvent: + return AdminUpdateEvent( + hostname=str(row["hostname"]), + ipv4=str(row["ipv4"]) if row.get("ipv4") else None, + ipv6=str(row["ipv6"]) if row.get("ipv6") else None, + status=str(row["status"]), + detail=str(row["detail"]), + source_ip=str(row["source_ip"]), + created_at=str(row["created_at"]), + count=_int_value(row.get("count"), 1), + ) + + +def _admin_cleanup_run(row: Mapping[str, object]) -> AdminCleanupRun: + return AdminCleanupRun( + created_at=str(row["created_at"]), + unused_accounts_deleted=_int_value(row["unused_accounts_deleted"], 0), + ) + + +def _admin_scheduled_task(row: Mapping[str, object]) -> AdminScheduledTask: + return AdminScheduledTask( + name=str(row["name"]), + interval_seconds=_int_value(row["interval_seconds"], 0), + next_run_at=str(row["next_run_at"]), + lease_owner=str(row["lease_owner"]) if row.get("lease_owner") else None, + lease_expires_at=str(row["lease_expires_at"]) if row.get("lease_expires_at") else None, + last_started_at=str(row["last_started_at"]) if row.get("last_started_at") else None, + last_finished_at=str(row["last_finished_at"]) if row.get("last_finished_at") else None, + last_status=str(row["last_status"]), + last_error=str(row["last_error"]), + updated_at=str(row["updated_at"]), + ) + + +def _int_value(value: object, default: int) -> int: + if value is None: + return default + if isinstance(value, int): + return value + if isinstance(value, float): + return int(value) + if isinstance(value, str): + return int(value) + return default diff --git a/karldns/security.py b/karldns/security.py index bccc14d..89c027e 100644 --- a/karldns/security.py +++ b/karldns/security.py @@ -5,6 +5,7 @@ import hashlib import hmac import secrets import time +from collections.abc import Mapping from datetime import UTC, datetime from ipaddress import ip_address, ip_network @@ -135,7 +136,7 @@ def dashboard_cookie_signing_secret(settings: DdnsSettings) -> str: return settings.dashboard_cookie_secret or settings.admin_password -def dashboard_csrf_token(management_slug: str, account: dict[str, str | int | None]) -> str: +def dashboard_csrf_token(management_slug: str, account: Mapping[str, object]) -> str: hostname = str(account["hostname"]) password_hash = str(account.get("dashboard_password_hash") or "") day = datetime.now(UTC).strftime("%Y-%m-%d") @@ -143,7 +144,7 @@ def dashboard_csrf_token(management_slug: str, account: dict[str, str | int | No return hmac.new(key, f"dashboard:{hostname}:{day}".encode(), hashlib.sha256).hexdigest() -def require_dashboard_csrf(management_slug: str, account: dict[str, str | int | None], supplied: str) -> None: +def require_dashboard_csrf(management_slug: str, account: Mapping[str, object], supplied: str) -> None: if not supplied or not hmac.compare_digest(supplied, dashboard_csrf_token(management_slug, account)): raise HTTPException(status_code=403, detail="invalid csrf token") diff --git a/karldns/services/ddns.py b/karldns/services/ddns.py index 8d36141..df77eec 100644 --- a/karldns/services/ddns.py +++ b/karldns/services/ddns.py @@ -8,6 +8,7 @@ import sqlite3 import threading import time import urllib.parse +from collections.abc import Mapping from dataclasses import dataclass from fastapi import HTTPException, Request @@ -18,6 +19,7 @@ from ..models import UpdateResult from ..security import client_ip from ..settings import DdnsSettings from ..store import DdnsStore +from ..store.types import AccountRow LOGGER = logging.getLogger(__name__) INITIAL_PLACEHOLDER_IPV4 = "192.0.2.1" @@ -41,7 +43,7 @@ class DdnsService: self.store = store self._locks = [threading.Lock() for _ in range(self._LOCK_STRIPES)] - def create_managed_account(self) -> dict[str, str]: + def create_managed_account(self) -> AccountRow: if not self.settings.hostname_suffix: raise HTTPException(status_code=500, detail="DDNS_HOSTNAME_SUFFIX is required") for _ in range(5): @@ -79,6 +81,8 @@ class DdnsService: source_ip = client_ip(request, self.settings) source_fallback = source_ip if _is_global_ip(source_ip) else "" supplied_ip = any(value and value.strip() for value in (myip, ipaddr, myipv6, ip6addr)) + ipv4: str | None + ipv6: str | None if not supplied_ip and not source_fallback: ipv4 = INITIAL_PLACEHOLDER_IPV4 ipv6 = None @@ -108,7 +112,9 @@ class DdnsService: planned.update_ipv6, type(exc).__name__, ) - self.store.log_update_event(hostname, ipv4, ipv6, DNS_RECONCILE_FAILED, f"{type(exc).__name__}: {exc}", source_ip) + self.store.log_update_event( + hostname, ipv4, ipv6, DNS_RECONCILE_FAILED, f"{type(exc).__name__}: {exc}", source_ip + ) raise except DnsProviderError as exc: LOGGER.warning( @@ -118,7 +124,9 @@ class DdnsService: planned.update_ipv6, type(exc).__name__, ) - self.store.log_update_event(hostname, ipv4, ipv6, DNS_FAILED, f"{type(exc).__name__}: {exc}", source_ip) + self.store.log_update_event( + hostname, ipv4, ipv6, DNS_FAILED, f"{type(exc).__name__}: {exc}", source_ip + ) raise result = self.store.commit_update_result(planned) self.store.log_update_event(hostname, result.ipv4, result.ipv6, UPDATED, detail, source_ip) @@ -149,7 +157,9 @@ class DdnsService: ) lock = self._lock_for(hostname) with lock: - planned = self.store.preview_update(hostname, ipv4, ipv6, update_ipv4=update_ipv4, update_ipv6=update_ipv6) + planned = self.store.preview_update( + hostname, ipv4, ipv6, update_ipv4=update_ipv4, update_ipv6=update_ipv6 + ) if planned.changed: try: publish_dns(planned, self.settings) @@ -180,7 +190,9 @@ class DdnsService: planned.update_ipv6, type(exc).__name__, ) - self.store.log_update_event(hostname, ipv4, ipv6, DNS_FAILED, f"{type(exc).__name__}: {exc}", source_ip) + self.store.log_update_event( + hostname, ipv4, ipv6, DNS_FAILED, f"{type(exc).__name__}: {exc}", source_ip + ) raise result = self.store.commit_update_result(planned) is_success = True @@ -203,19 +215,19 @@ class DdnsService: is_bad_ip=is_bad_ip, ) - def fritz_update_url(self, account: dict[str, str]) -> str: + def fritz_update_url(self, account: Mapping[str, object]) -> str: base = self.settings.public_base_url.rstrip("/") - slug = urllib.parse.quote(account["update_slug"]) + slug = urllib.parse.quote(str(account["update_slug"])) return f"{base}/api/v1/router/update/{slug}?myip=&myipv6=" - def dashboard_url(self, account: dict[str, str]) -> str: + def dashboard_url(self, account: Mapping[str, object]) -> str: base = self.settings.public_base_url.rstrip("/") - slug = urllib.parse.quote(account["management_slug"]) + slug = urllib.parse.quote(str(account["management_slug"])) return f"{base}/m/{slug}" - def registration_url(self, account: dict[str, str]) -> str: + def registration_url(self, account: Mapping[str, object]) -> str: base = self.settings.public_base_url.rstrip("/") - slug = urllib.parse.quote(account["management_slug"]) + slug = urllib.parse.quote(str(account["management_slug"])) return f"{base}/api/v1/router/register/{slug}" def random_managed_hostname(self) -> str: diff --git a/karldns/store/__init__.py b/karldns/store/__init__.py index 908a082..68d5b0c 100644 --- a/karldns/store/__init__.py +++ b/karldns/store/__init__.py @@ -1,6 +1,7 @@ from __future__ import annotations import threading +from collections.abc import Iterator from contextlib import contextmanager from pathlib import Path from sqlite3 import Connection @@ -31,7 +32,7 @@ class DdnsStore(AccountsMixin, MetricsMixin, RecordsEventsMixin, RateLimitCleanu self._init_db() @contextmanager - def connect(self): + def connect(self) -> Iterator[Connection]: conn = self._thread_connection() try: yield conn diff --git a/karldns/store/accounts.py b/karldns/store/accounts.py index d2127be..0ababbd 100644 --- a/karldns/store/accounts.py +++ b/karldns/store/accounts.py @@ -3,10 +3,11 @@ from __future__ import annotations from datetime import UTC, datetime from ..security import hash_lookup_token, hash_secret, new_bearer_secret, verify_secret -from .types import AccountRow +from .base import StoreConnectionMixin +from .types import AccountRow, account_row -class AccountsMixin: +class AccountsMixin(StoreConnectionMixin): def create_account(self, hostname: str, username: str | None = None, *, accepted: bool = True) -> AccountRow: now = datetime.now(UTC).isoformat() update_slug = new_bearer_secret() @@ -41,7 +42,7 @@ class AccountsMixin: """, (limit, offset), ).fetchall() - return [dict(row) for row in rows] + return [account_row(row) for row in rows] def count_accounts(self) -> int: with self.connect() as conn: @@ -60,7 +61,7 @@ class AccountsMixin: """, (hostname,), ).fetchone() - return dict(account) if account else None + return account_row(account) if account else None def get_account_by_slug(self, update_slug: str) -> AccountRow | None: with self.connect() as conn: @@ -71,7 +72,7 @@ class AccountsMixin: """, (update_slug,), ).fetchone() - return dict(account) if account else None + return account_row(account) if account else None def get_account_by_management_slug(self, management_slug: str) -> AccountRow | None: with self.connect() as conn: @@ -85,7 +86,7 @@ class AccountsMixin: """, (hash_lookup_token(management_slug),), ).fetchone() - return dict(account) if account else None + return account_row(account) if account else None def activate_account(self, hostname: str) -> AccountRow | None: now = datetime.now(UTC).isoformat() diff --git a/karldns/store/base.py b/karldns/store/base.py new file mode 100644 index 0000000..c54d4d1 --- /dev/null +++ b/karldns/store/base.py @@ -0,0 +1,9 @@ +from __future__ import annotations + +import sqlite3 +from contextlib import AbstractContextManager + + +class StoreConnectionMixin: + def connect(self) -> AbstractContextManager[sqlite3.Connection]: + raise NotImplementedError diff --git a/karldns/store/metrics.py b/karldns/store/metrics.py index 8a00994..e321baf 100644 --- a/karldns/store/metrics.py +++ b/karldns/store/metrics.py @@ -2,8 +2,10 @@ from __future__ import annotations from datetime import UTC, datetime +from .base import StoreConnectionMixin -class MetricsMixin: + +class MetricsMixin(StoreConnectionMixin): def increment_metric(self, name: str, amount: float = 1.0) -> None: now = datetime.now(UTC).isoformat() with self.connect() as conn: diff --git a/karldns/store/rate_limit.py b/karldns/store/rate_limit.py index 86903b3..300b9dc 100644 --- a/karldns/store/rate_limit.py +++ b/karldns/store/rate_limit.py @@ -1,14 +1,23 @@ from __future__ import annotations +import threading import time from datetime import UTC, datetime, timedelta -from .types import CleanupRunRow +from .base import StoreConnectionMixin +from .types import CleanupRunRow, cleanup_run_row RATE_LIMIT_FLUSH_SECONDS = 10.0 -class RateLimitCleanupMixin: +class RateLimitCleanupMixin(StoreConnectionMixin): + _rate_limit_lock: threading.Lock + _rate_limit_prune_lock: threading.Lock + _rate_limit_buckets: dict[str, dict[str, float | int]] + _rate_limit_dirty_counts: dict[tuple[str, int], int] + _last_rate_limit_flush: float + _last_rate_limit_prune_window: int + def list_cleanup_runs(self, limit: int = 20, offset: int = 0) -> list[CleanupRunRow]: with self.connect() as conn: rows = conn.execute( @@ -20,7 +29,7 @@ class RateLimitCleanupMixin: """, (limit, offset), ).fetchall() - return [CleanupRunRow(**dict(row)) for row in rows] + return [cleanup_run_row(row) for row in rows] def count_cleanup_runs(self) -> int: with self.connect() as conn: diff --git a/karldns/store/records.py b/karldns/store/records.py index a414724..aeedfa3 100644 --- a/karldns/store/records.py +++ b/karldns/store/records.py @@ -4,10 +4,11 @@ import sqlite3 from datetime import UTC, datetime from ..models import UpdateResult -from .types import UpdateEventRow +from .base import StoreConnectionMixin +from .types import RecordRow, UpdateEventRow, record_row, update_event_row -class RecordsEventsMixin: +class RecordsEventsMixin(StoreConnectionMixin): def upsert( self, hostname: str, @@ -84,10 +85,10 @@ class RecordsEventsMixin: update_ipv6=update_ipv6, ) - def list_records(self) -> list[dict[str, str | None]]: + def list_records(self) -> list[RecordRow]: return self.list_records_page(limit=10_000, offset=0) - def list_records_page(self, limit: int = 100, offset: int = 0) -> list[dict[str, str | None]]: + def list_records_page(self, limit: int = 100, offset: int = 0) -> list[RecordRow]: with self.connect() as conn: rows = conn.execute( """ @@ -98,7 +99,7 @@ class RecordsEventsMixin: """, (limit, offset), ).fetchall() - return [dict(row) for row in rows] + return [record_row(row) for row in rows] def count_records(self) -> int: with self.connect() as conn: @@ -167,7 +168,7 @@ class RecordsEventsMixin: """, (limit, offset), ).fetchall() - return [dict(row) for row in rows] + return [update_event_row(row) for row in rows] def list_update_events_before(self, limit: int = 100, before: str | None = None) -> list[UpdateEventRow]: where_clause = "WHERE created_at < ?" if before else "" @@ -189,7 +190,7 @@ class RecordsEventsMixin: """, params, ).fetchall() - return [dict(row) for row in rows] + return [update_event_row(row) for row in rows] def count_update_events(self) -> int: with self.connect() as conn: @@ -228,7 +229,7 @@ def _merge_event_rows( limit: int, offset: int, ) -> list[UpdateEventRow]: - rows = [dict(row) for row in event_rows] - rows.extend(dict(row) for row in rollup_rows) + rows = [update_event_row(row) for row in event_rows] + rows.extend(update_event_row(row) for row in rollup_rows) rows.sort(key=lambda row: str(row["created_at"]), reverse=True) return rows[offset : offset + limit] diff --git a/karldns/store/scheduler.py b/karldns/store/scheduler.py index c742c56..6dadb2b 100644 --- a/karldns/store/scheduler.py +++ b/karldns/store/scheduler.py @@ -2,7 +2,8 @@ from __future__ import annotations from datetime import UTC, datetime, timedelta -from .types import ScheduledTaskRow +from .base import StoreConnectionMixin +from .types import ScheduledTaskRow, scheduled_task_row def _now() -> datetime: @@ -13,7 +14,7 @@ def _iso(value: datetime) -> str: return value.isoformat() -class SchedulerMixin: +class SchedulerMixin(StoreConnectionMixin): def ensure_scheduled_task(self, name: str, interval_seconds: int) -> None: now = _now() with self.connect() as conn: @@ -86,12 +87,10 @@ class SchedulerMixin: def list_scheduled_tasks(self) -> list[ScheduledTaskRow]: with self.connect() as conn: - rows = conn.execute( - """ + rows = conn.execute(""" SELECT name, interval_seconds, next_run_at, lease_owner, lease_expires_at, last_started_at, last_finished_at, last_status, last_error, updated_at FROM scheduled_tasks ORDER BY name - """ - ).fetchall() - return [ScheduledTaskRow(**dict(row)) for row in rows] + """).fetchall() + return [scheduled_task_row(row) for row in rows] diff --git a/karldns/store/schema.py b/karldns/store/schema.py index e93d00a..dc96a1b 100644 --- a/karldns/store/schema.py +++ b/karldns/store/schema.py @@ -35,18 +35,15 @@ def initialize_schema(conn: sqlite3.Connection) -> None: configure_connection(conn) schema_version = conn.execute("PRAGMA user_version").fetchone()[0] conn.execute("PRAGMA journal_mode = WAL") - conn.execute( - """ + conn.execute(""" CREATE TABLE IF NOT EXISTS records ( hostname TEXT PRIMARY KEY, ipv4 TEXT, ipv6 TEXT, updated_at TEXT NOT NULL ) - """ - ) - conn.execute( - """ + """) + conn.execute(""" CREATE TABLE IF NOT EXISTS update_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, hostname TEXT NOT NULL, @@ -57,12 +54,12 @@ def initialize_schema(conn: sqlite3.Connection) -> None: source_ip TEXT NOT NULL, created_at TEXT NOT NULL ) - """ - ) + """) conn.execute("CREATE INDEX IF NOT EXISTS idx_update_events_created_at ON update_events(created_at)") - conn.execute("CREATE INDEX IF NOT EXISTS idx_update_events_hostname_created_at ON update_events(hostname, created_at DESC)") conn.execute( - """ + "CREATE INDEX IF NOT EXISTS idx_update_events_hostname_created_at ON update_events(hostname, created_at DESC)" + ) + conn.execute(""" CREATE TABLE IF NOT EXISTS update_event_rollups ( id INTEGER PRIMARY KEY AUTOINCREMENT, hostname TEXT NOT NULL, @@ -76,28 +73,24 @@ def initialize_schema(conn: sqlite3.Connection) -> None: count INTEGER NOT NULL, UNIQUE(hostname, ipv4, ipv6, status, detail, source_ip) ) - """ - ) - conn.execute("CREATE INDEX IF NOT EXISTS idx_update_event_rollups_last_seen_at ON update_event_rollups(last_seen_at)") + """) conn.execute( - """ + "CREATE INDEX IF NOT EXISTS idx_update_event_rollups_last_seen_at ON update_event_rollups(last_seen_at)" + ) + conn.execute(""" CREATE INDEX IF NOT EXISTS idx_update_event_rollups_hostname_last_seen_at ON update_event_rollups(hostname, last_seen_at DESC) - """ - ) - conn.execute( - """ + """) + conn.execute(""" CREATE TABLE IF NOT EXISTS cleanup_runs ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at TEXT NOT NULL, unused_accounts_deleted INTEGER NOT NULL ) - """ - ) + """) _migrate_cleanup_runs_schema(conn) conn.execute("CREATE INDEX IF NOT EXISTS idx_cleanup_runs_created_at ON cleanup_runs(created_at)") - conn.execute( - """ + conn.execute(""" CREATE TABLE IF NOT EXISTS scheduled_tasks ( name TEXT PRIMARY KEY, interval_seconds INTEGER NOT NULL, @@ -110,37 +103,29 @@ def initialize_schema(conn: sqlite3.Connection) -> None: last_error TEXT NOT NULL DEFAULT '', updated_at TEXT NOT NULL ) - """ - ) + """) conn.execute("CREATE INDEX IF NOT EXISTS idx_scheduled_tasks_next_run_at ON scheduled_tasks(next_run_at)") - conn.execute( - """ + conn.execute(""" CREATE INDEX IF NOT EXISTS idx_scheduled_tasks_due ON scheduled_tasks(next_run_at, lease_expires_at) - """ - ) - conn.execute( - """ + """) + conn.execute(""" CREATE TABLE IF NOT EXISTS rate_limits ( key TEXT NOT NULL, window INTEGER NOT NULL, count INTEGER NOT NULL, PRIMARY KEY(key, window) ) - """ - ) + """) conn.execute("CREATE INDEX IF NOT EXISTS idx_rate_limits_window ON rate_limits(window)") - conn.execute( - """ + conn.execute(""" CREATE TABLE IF NOT EXISTS service_metrics ( name TEXT PRIMARY KEY, value REAL NOT NULL, updated_at TEXT NOT NULL ) - """ - ) - conn.execute( - """ + """) + conn.execute(""" CREATE TABLE IF NOT EXISTS accounts ( hostname TEXT PRIMARY KEY, update_slug TEXT NOT NULL UNIQUE, @@ -150,8 +135,7 @@ def initialize_schema(conn: sqlite3.Connection) -> None: accepted_at TEXT, disabled INTEGER NOT NULL DEFAULT 0 ) - """ - ) + """) _ensure_column(conn, "accounts", "update_slug", "TEXT") _ensure_column(conn, "accounts", "management_hash", "TEXT") _ensure_column(conn, "accounts", "dashboard_password_hash", "TEXT") @@ -190,8 +174,7 @@ def _migrate_accounts_schema(conn: sqlite3.Connection) -> None: conn.execute("DROP INDEX IF EXISTS idx_accounts_accepted_at") conn.execute("DROP INDEX IF EXISTS idx_accounts_created_at") conn.execute("DROP INDEX IF EXISTS idx_accounts_hostname_username") - conn.execute( - """ + conn.execute(""" CREATE TABLE accounts_new ( hostname TEXT PRIMARY KEY, update_slug TEXT NOT NULL UNIQUE, @@ -201,15 +184,12 @@ def _migrate_accounts_schema(conn: sqlite3.Connection) -> None: accepted_at TEXT, disabled INTEGER NOT NULL DEFAULT 0 ) - """ - ) - conn.execute( - """ + """) + conn.execute(""" INSERT INTO accounts_new(hostname, update_slug, management_hash, dashboard_password_hash, created_at, accepted_at, disabled) SELECT hostname, update_slug, management_hash, dashboard_password_hash, created_at, accepted_at, disabled FROM accounts - """ - ) + """) conn.execute("DROP TABLE accounts") conn.execute("ALTER TABLE accounts_new RENAME TO accounts") @@ -219,21 +199,17 @@ def _migrate_cleanup_runs_schema(conn: sqlite3.Connection) -> None: if columns <= {"id", "created_at", "unused_accounts_deleted"}: return conn.execute("DROP INDEX IF EXISTS idx_cleanup_runs_created_at") - conn.execute( - """ + conn.execute(""" CREATE TABLE cleanup_runs_new ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at TEXT NOT NULL, unused_accounts_deleted INTEGER NOT NULL ) - """ - ) - conn.execute( - """ + """) + conn.execute(""" INSERT INTO cleanup_runs_new(id, created_at, unused_accounts_deleted) SELECT id, created_at, unused_accounts_deleted FROM cleanup_runs - """ - ) + """) conn.execute("DROP TABLE cleanup_runs") conn.execute("ALTER TABLE cleanup_runs_new RENAME TO cleanup_runs") diff --git a/karldns/store/types.py b/karldns/store/types.py index b840853..a9f01ce 100644 --- a/karldns/store/types.py +++ b/karldns/store/types.py @@ -1,6 +1,6 @@ from __future__ import annotations -from typing import TypedDict +from typing import Any, TypedDict, cast class AccountRow(TypedDict, total=False): @@ -43,3 +43,30 @@ class ScheduledTaskRow(TypedDict, total=False): last_status: str last_error: str updated_at: str + + +class RecordRow(TypedDict): + hostname: str + ipv4: str | None + ipv6: str | None + updated_at: str + + +def account_row(row: Any) -> AccountRow: + return cast(AccountRow, dict(row)) + + +def cleanup_run_row(row: Any) -> CleanupRunRow: + return cast(CleanupRunRow, dict(row)) + + +def record_row(row: Any) -> RecordRow: + return cast(RecordRow, dict(row)) + + +def scheduled_task_row(row: Any) -> ScheduledTaskRow: + return cast(ScheduledTaskRow, dict(row)) + + +def update_event_row(row: Any) -> UpdateEventRow: + return cast(UpdateEventRow, dict(row)) diff --git a/karldns/web/components.py b/karldns/web/components.py index ed20c80..c1b99e5 100644 --- a/karldns/web/components.py +++ b/karldns/web/components.py @@ -46,8 +46,7 @@ def _copy_row(label: str, value: str, sensitivity: str | None = None, *, mask_se else "" ) row_class = "copy-field copy-field-secret" if is_secret else "copy-field copy-field-public" - return trusted( - f""" + return trusted(f"""
{safe_label} @@ -58,8 +57,7 @@ def _copy_row(label: str, value: str, sensitivity: str | None = None, *, mask_se {_svg_icon("clipboard")}Copy{_svg_icon("check", "check-icon")}
- """ - ) + """) def _copy_button(label: str, value: str, *, style: str = "outline-primary") -> SafeHtml: @@ -89,8 +87,7 @@ def _copy_menu(label: str, items: list[tuple[str, str]]) -> SafeHtml: f'' ) - return trusted( - f""" + return trusted(f"""
{_svg_icon("more", "more-icon")} @@ -99,22 +96,21 @@ def _copy_menu(label: str, items: list[tuple[str, str]]) -> SafeHtml: {''.join(buttons)}
- """ - ) + """) -def _code_window(commands: str, *, copy_label: str = "code", class_name: str = "", copy_value: str | None = None) -> SafeHtml: +def _code_window( + commands: str, *, copy_label: str = "code", class_name: str = "", copy_value: str | None = None +) -> SafeHtml: safe_class = f" {attr(class_name)}" if class_name else "" code_lines = "\n".join(f"{escape(line)}" for line in commands.splitlines()) copy_value = copy_value if copy_value is not None else _code_copy_value(commands) - return trusted( - f""" + return trusted(f"""
{_copy_icon_button(copy_label, copy_value)} {code_lines}
- """ - ) + """) def _code_copy_value(commands: str) -> str: @@ -167,8 +163,7 @@ def _known_request_ip(request_ip: str | None = None) -> tuple[int | None, str]: def _manual_update_panel(commands: str) -> SafeHtml: - return trusted( - f""" + return trusted(f"""

Manual updates

@@ -176,8 +171,7 @@ def _manual_update_panel(commands: str) -> SafeHtml:
{_code_window(commands, copy_label="manual update commands", class_name="manual-code")}
- """ - ) + """) def _fritzbox_fields( @@ -196,25 +190,23 @@ def _fritzbox_fields( def _secret_notice(*, include_registration: bool = False) -> SafeHtml: - secret_items = "Update URL, register URL, and dashboard link" if include_registration else "Update URL and dashboard link" - return trusted( - f""" + secret_items = ( + "Update URL, register URL, and dashboard link" if include_registration else "Update URL and dashboard link" + ) + return trusted(f"""
Public: Hostname and CNAME target can be shared. Secret: {escape(secret_items)} can update or manage this hostname. Secrets are hidden on screen. Copy still copies the real value.
- """ - ) + """) def _message_band(message: str | None) -> SafeHtml: if not message: return trusted("") - return trusted( - f""" + return trusted(f"""
{escape(message)}
- """ - ) + """) diff --git a/karldns/web/forms.py b/karldns/web/forms.py index 4486ef9..23a60ea 100644 --- a/karldns/web/forms.py +++ b/karldns/web/forms.py @@ -17,7 +17,9 @@ def parse_form(body: bytes) -> dict[str, list[str]]: try: return urllib.parse.parse_qs(decoded, keep_blank_values=True, max_num_fields=MAX_FORM_FIELDS) except ValueError as exc: - raise HTTPException(status_code=400, detail=f"form must not contain more than {MAX_FORM_FIELDS} fields") from exc + raise HTTPException( + status_code=400, detail=f"form must not contain more than {MAX_FORM_FIELDS} fields" + ) from exc async def request_form(request: Request, settings: DdnsSettings) -> dict[str, list[str]]: diff --git a/karldns/web/layout.py b/karldns/web/layout.py index f54693e..9fc1d84 100644 --- a/karldns/web/layout.py +++ b/karldns/web/layout.py @@ -18,8 +18,7 @@ def _top_nav(label: str | None = None, links: SafeHtml | None = None) -> SafeHtm {_svg_icon("sun", "theme-icon theme-icon-sun")} """ - return trusted( - f""" + return trusted(f""" - """ - ) + """) def _site_footer() -> SafeHtml: - return trusted( - f""" + return trusted(f"""
Made with ❤️ by karl
@@ -65,8 +62,7 @@ def _site_footer() -> SafeHtml:
- """ - ) + """) def _page( @@ -85,7 +81,11 @@ def _page( raise TypeError("body must be a SafeHtml fragment built by a template helper") extra_styles_html = extra_styles or trusted("") extra_scripts_html = extra_scripts or trusted("") - preload_logo_html = trusted('') if preload_logo else trusted("") + preload_logo_html = ( + trusted('') + if preload_logo + else trusted("") + ) canonical_html = trusted(f'') if canonical_url else trusted("") social_url = canonical_url or "" social_image = image_url or "" @@ -93,14 +93,18 @@ def _page( if social_url: social_card = "summary_large_image" if social_image else "summary" social_image_alt = image_alt or title - social_image_html = f""" + social_image_html = ( + f""" - """ if social_image else "" + """ + if social_image + else "" + ) social_html = trusted(f""" diff --git a/karldns/web/security_headers.py b/karldns/web/security_headers.py index f7f6563..6c026eb 100644 --- a/karldns/web/security_headers.py +++ b/karldns/web/security_headers.py @@ -47,7 +47,6 @@ class RequestBodyLimitMiddleware: await response(scope, receive, send) - def install_security_middleware(app: FastAPI, settings: DdnsSettings, store: DdnsStore) -> None: @app.middleware("http") async def security_middleware(request: Request, call_next): diff --git a/karldns/web/templates_admin.py b/karldns/web/templates_admin.py index 23a9ffc..f6c8098 100644 --- a/karldns/web/templates_admin.py +++ b/karldns/web/templates_admin.py @@ -1,5 +1,6 @@ from __future__ import annotations +from collections.abc import Mapping, Sequence from datetime import datetime from urllib.parse import urlencode @@ -30,13 +31,14 @@ def _admin_secret_panel(title: str, rows: list[SafeHtml]) -> SafeHtml: """) + def _render_admin_page( service: DdnsService, settings: DdnsSettings, - accounts: list[dict[str, str | int | None]], - cleanup_runs: list[dict[str, str | int]], - scheduled_tasks: list[dict[str, str | int | None]], - update_events: list[dict[str, str | int | None]], + accounts: Sequence[Mapping[str, object]], + cleanup_runs: Sequence[Mapping[str, object]], + scheduled_tasks: Sequence[Mapping[str, object]], + update_events: Sequence[Mapping[str, object]], service_metrics: dict[str, float], table_counts: dict[str, int], table_offsets: dict[str, int], @@ -55,8 +57,16 @@ def _render_admin_page( accounts_count = escape(table_counts.get("accounts", len(accounts))) event_count = escape(table_counts.get("update_events", len(update_events))) metrics_html = _metrics_panel(service_metrics) - account_pager = _server_page_links(admin_path, "accounts_offset", table_offsets.get("accounts", 0), table_limit, table_counts.get("accounts", 0)) - event_pager = _server_page_links(admin_path, "update_events_offset", table_offsets.get("update_events", 0), table_limit, table_counts.get("update_events", 0)) + account_pager = _server_page_links( + admin_path, "accounts_offset", table_offsets.get("accounts", 0), table_limit, table_counts.get("accounts", 0) + ) + event_pager = _server_page_links( + admin_path, + "update_events_offset", + table_offsets.get("update_events", 0), + table_limit, + table_counts.get("update_events", 0), + ) cleanup_panel = _cleanup_panel(settings, cleanup_runs, scheduled_tasks, safe_admin_path, csrf) totals_html = _totals_strip(table_counts, service_metrics) return _page( @@ -142,10 +152,7 @@ def _totals_strip(table_counts: dict[str, int], metrics: dict[str, float]) -> Sa ] return trusted( '
' - + "".join( - f'{escape(value)}{escape(label)}' - for label, value in items - ) + + "".join(f"{escape(value)}{escape(label)}" for label, value in items) + "
" ) @@ -181,12 +188,15 @@ def _server_page_links(admin_path: str, offset_name: str, offset: int, limit: in def _cleanup_panel( settings: DdnsSettings, - cleanup_runs: list[dict[str, str | int]], - scheduled_tasks: list[dict[str, str | int | None]], + cleanup_runs: Sequence[Mapping[str, object]], + scheduled_tasks: Sequence[Mapping[str, object]], safe_admin_path: str, csrf: str, ) -> SafeHtml: - task = next((item for item in scheduled_tasks if item.get("name") == "cleanup"), scheduled_tasks[0] if scheduled_tasks else {}) + task = next( + (item for item in scheduled_tasks if item.get("name") == "cleanup"), + scheduled_tasks[0] if scheduled_tasks else {}, + ) last_run = task.get("last_finished_at") or (cleanup_runs[0].get("created_at") if cleanup_runs else None) next_run = task.get("next_run_at") status = task.get("last_status") or "pending" @@ -235,14 +245,14 @@ def _metrics_panel(metrics: dict[str, float]) -> SafeHtml: ("Blocked requests", _metric_int(metrics.get("rate_limit_hits_total", 0))), ("Average update time", f"{average_latency:.1f} ms"), ("Last update time", f"{metrics.get('last_update_latency_ms', 0):.1f} ms"), - ("Cleanup ok/errors", f"{_metric_int(metrics.get('cleanup_success_total', 0))}/{_metric_int(metrics.get('cleanup_errors_total', 0))}"), + ( + "Cleanup ok/errors", + f"{_metric_int(metrics.get('cleanup_success_total', 0))}/{_metric_int(metrics.get('cleanup_errors_total', 0))}", + ), ] - cards = "\n".join( - f""" + cards = "\n".join(f""" {escape(value)}{escape(label)} - """ - for label, value in items - ) + """ for label, value in items) return trusted(f"""
{cards} @@ -250,13 +260,16 @@ def _metrics_panel(metrics: dict[str, float]) -> SafeHtml: """) -def _metric_int(value: float) -> str: - return str(int(value)) +def _metric_int(value: object) -> str: + try: + return str(int(float(str(value)))) + except ValueError: + return "0" def _format_interval(seconds: object) -> str: try: - value = int(seconds) + value = int(str(seconds)) except (TypeError, ValueError): return str(seconds) if value % 3600 == 0: diff --git a/karldns/web/templates_admin_tables.py b/karldns/web/templates_admin_tables.py index 2dfae33..b489e35 100644 --- a/karldns/web/templates_admin_tables.py +++ b/karldns/web/templates_admin_tables.py @@ -1,12 +1,13 @@ from __future__ import annotations +from collections.abc import Mapping from datetime import datetime from .components import _event_label from .html_utils import attr, escape -def _account_row(account: dict[str, str | int | None], csrf: str = "", admin_path: str = "/admin") -> str: +def _account_row(account: Mapping[str, object], csrf: str = "", admin_path: str = "/admin") -> str: raw_hostname = str(account["hostname"]) hostname = escape(raw_hostname) hostname_attr = attr(raw_hostname) @@ -77,7 +78,7 @@ def _admin_table_pagination() -> str: """ -def _event_row(event: dict[str, str | int | None]) -> str: +def _event_row(event: Mapping[str, object]) -> str: created_raw = str(event.get("created_at") or "-") hostname_raw = str(event.get("hostname") or "-") detail_raw = str(event.get("detail") or "-") diff --git a/karldns/web/templates_management.py b/karldns/web/templates_management.py index d9843f5..70166cd 100644 --- a/karldns/web/templates_management.py +++ b/karldns/web/templates_management.py @@ -1,5 +1,7 @@ from __future__ import annotations +from collections.abc import Mapping + from ..services.ddns import DdnsService from ..settings import DdnsSettings from .components import ( @@ -19,7 +21,7 @@ def _render_management_page( service: DdnsService, settings: DdnsSettings, management_slug: str, - account: dict[str, str | int | None], + account: Mapping[str, object], csrf_token: str, message: str | None = None, request_ip: str | None = None, @@ -37,11 +39,19 @@ def _render_management_page( fritzbox_fields = _fritzbox_fields(update_url, hostname) cname_record = f"your-subdomain.example.com. CNAME {hostname}." manual_commands = _manual_update_commands(update_url, request_ip) - history_rows = "\n".join(_user_event_row(event) for event in service.store.list_update_events_for_hostname(str(account["hostname"]), 10)) or """ + history_rows = ( + "\n".join( + _user_event_row(event) + for event in service.store.list_update_events_for_hostname(str(account["hostname"]), 10) + ) + or """ No router updates yet. """ + ) message_html = _management_message(message) - dashboard_password_panel = _dashboard_password_panel(management_slug, csrf_token, bool(account.get("dashboard_password_hash"))) + dashboard_password_panel = _dashboard_password_panel( + management_slug, csrf_token, bool(account.get("dashboard_password_hash")) + ) return _page( "KarlDNS management", trusted(f""" @@ -143,11 +153,12 @@ def _render_management_page( extra_styles=trusted(''), ) + def _render_pending_management_page( service: DdnsService, settings: DdnsSettings, management_slug: str, - account: dict[str, str | int | None], + account: Mapping[str, object], csrf_token: str, message: str | None = None, request_ip: str | None = None, @@ -224,13 +235,11 @@ def _render_pending_management_page( def _management_message(message: str | None) -> SafeHtml: if not message: return trusted("") - return trusted( - f""" + return trusted(f"""
{escape(message)}
- """ - ) + """) def _dashboard_password_panel(management_slug: str, csrf_token: str, password_enabled: bool) -> SafeHtml: @@ -254,8 +263,7 @@ def _dashboard_password_panel(management_slug: str, csrf_token: str, password_en
""" - return trusted( - f""" + return trusted(f"""
@@ -285,13 +293,12 @@ def _dashboard_password_panel(management_slug: str, csrf_token: str, password_en
- """ - ) + """) def _render_management_unlock_page( management_slug: str, - account: dict[str, str | int | None], + account: Mapping[str, object], csrf_token: str, message: str | None = None, ) -> str: @@ -333,7 +340,7 @@ def _render_management_unlock_page( ) -def _user_event_row(event: dict[str, str | int | None]) -> str: +def _user_event_row(event: Mapping[str, object]) -> str: created = escape(event.get("created_at") or "-") raw_status = str(event.get("status") or "") status = escape(_event_label(raw_status)) diff --git a/karldns/web/templates_public.py b/karldns/web/templates_public.py index d431ce0..709bc40 100644 --- a/karldns/web/templates_public.py +++ b/karldns/web/templates_public.py @@ -99,6 +99,7 @@ def _render_public_home( }, ) + def _render_privacy_page(settings: DdnsSettings) -> str: base_url = settings.public_base_url.rstrip("/") description = "Privacy and legal information for KarlDNS, including stored hostnames, update links, IP history, and cleanup behavior." @@ -159,7 +160,9 @@ def _render_api_docs_page(settings: DdnsSettings, request_ip: str | None = None) base_url = settings.public_base_url.rstrip("/") example_ipv4, _ = _example_update_ips(request_ip) create_example = f"curl -sS -X POST {base_url}/api/v1/hostnames" - register_example = f"curl -sS -X POST '{base_url}/api/v1/router/register/?myip={example_ipv4}'" + register_example = ( + f"curl -sS -X POST '{base_url}/api/v1/router/register/?myip={example_ipv4}'" + ) update_example = f"curl -sS '{base_url}/api/v1/router/update/?myip={example_ipv4}'" status_example = f"curl -sS '{base_url}/api/v1/hostnames/?history_limit=10'" rotate_example = f"curl -sS -X POST {base_url}/api/v1/hostnames//rotate-update-url" diff --git a/karldns/web/templates_public_sections.py b/karldns/web/templates_public_sections.py index f6e1f21..3cf535e 100644 --- a/karldns/web/templates_public_sections.py +++ b/karldns/web/templates_public_sections.py @@ -8,8 +8,7 @@ def _linux_installer_section(base_url: str) -> SafeHtml: base_url = base_url.rstrip("/") system_install = f"curl -fsSL {base_url}/install.sh | sudo sh" system_uninstall = f"curl -fsSL {base_url}/uninstall.sh | sudo sh" - return trusted( - f""" + return trusted(f"""
@@ -31,5 +30,4 @@ def _linux_installer_section(base_url: str) -> SafeHtml:
- """ - ) + """) diff --git a/pyproject.toml b/pyproject.toml index 7a77af1..fb6abdd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -33,7 +33,9 @@ Issues = "https://github.com/StasonJatham/karldns.de/issues" [project.optional-dependencies] dev = [ + "black>=26.5.1", "httpx>=0.28", + "mypy>=2.1.0", "pytest>=8.2", "ruff>=0.8", ] @@ -55,3 +57,15 @@ target-version = "py311" [tool.ruff.lint] select = ["E", "F", "I", "UP", "B"] ignore = ["E501"] + +[tool.black] +line-length = 120 +target-version = ["py311"] + +[tool.mypy] +python_version = "3.11" +files = ["karldns"] +check_untyped_defs = true +warn_unused_ignores = true +warn_redundant_casts = true +show_error_codes = true diff --git a/tests/test_admin.py b/tests/test_admin.py index 4382f7c..3f0e572 100644 --- a/tests/test_admin.py +++ b/tests/test_admin.py @@ -138,8 +138,14 @@ def test_update_events_page_orders_raw_and_rollup_events_in_sql(tmp_path: Path) store.log_update_rollup("rollup-mid.example.net", "203.0.113.3", None, "unchanged", "ok", "198.51.100.3") with store.connect() as conn: - conn.execute("UPDATE update_events SET created_at = ? WHERE hostname = ?", ("2026-01-01T00:00:00+00:00", "raw-old.example.net")) - conn.execute("UPDATE update_events SET created_at = ? WHERE hostname = ?", ("2026-01-03T00:00:00+00:00", "raw-new.example.net")) + conn.execute( + "UPDATE update_events SET created_at = ? WHERE hostname = ?", + ("2026-01-01T00:00:00+00:00", "raw-old.example.net"), + ) + conn.execute( + "UPDATE update_events SET created_at = ? WHERE hostname = ?", + ("2026-01-03T00:00:00+00:00", "raw-new.example.net"), + ) conn.execute( "UPDATE update_event_rollups SET last_seen_at = ? WHERE hostname = ?", ("2026-01-02T00:00:00+00:00", "rollup-mid.example.net"), @@ -156,8 +162,14 @@ def test_update_events_support_keyset_pagination(tmp_path: Path) -> None: store.log_update_event("raw-old.example.net", "203.0.113.1", None, "updated", "ok", "198.51.100.1") store.log_update_event("raw-new.example.net", "203.0.113.2", None, "updated", "ok", "198.51.100.2") with store.connect() as conn: - conn.execute("UPDATE update_events SET created_at = ? WHERE hostname = ?", ("2026-01-01T00:00:00+00:00", "raw-old.example.net")) - conn.execute("UPDATE update_events SET created_at = ? WHERE hostname = ?", ("2026-01-02T00:00:00+00:00", "raw-new.example.net")) + conn.execute( + "UPDATE update_events SET created_at = ? WHERE hostname = ?", + ("2026-01-01T00:00:00+00:00", "raw-old.example.net"), + ) + conn.execute( + "UPDATE update_events SET created_at = ? WHERE hostname = ?", + ("2026-01-02T00:00:00+00:00", "raw-new.example.net"), + ) app = make_app(DdnsSettings(database_path=database_path, admin_password="admin")) client = TestClient(app) @@ -247,7 +259,9 @@ def test_admin_can_rotate_links_and_password(tmp_path: Path) -> None: ) assert update.status_code == 200 assert "Update URL changed" in update.text - assert client.get(f"/api/v1/router/update/{account['update_slug']}", params={"myip": "203.0.113.9"}).status_code == 404 + assert ( + client.get(f"/api/v1/router/update/{account['update_slug']}", params={"myip": "203.0.113.9"}).status_code == 404 + ) management = client.post( "/admin/hostnames/rotate", diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index 5cdceb3..bd4a628 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -116,8 +116,7 @@ def test_first_run_creates_schema_and_repeated_startup_is_idempotent(tmp_path: P def test_startup_migrates_legacy_cleanup_run_columns(tmp_path: Path) -> None: database_path = tmp_path / "ddns.sqlite3" with sqlite3.connect(database_path) as conn: - conn.execute( - """ + conn.execute(""" CREATE TABLE cleanup_runs ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at TEXT NOT NULL, @@ -125,8 +124,7 @@ def test_startup_migrates_legacy_cleanup_run_columns(tmp_path: Path) -> None: abandoned_setups_deleted INTEGER NOT NULL, unused_accounts_deleted INTEGER NOT NULL ) - """ - ) + """) conn.execute( """ INSERT INTO cleanup_runs( @@ -144,12 +142,10 @@ def test_startup_migrates_legacy_cleanup_run_columns(tmp_path: Path) -> None: with store.connect() as conn: columns = {row[1] for row in conn.execute("PRAGMA table_info(cleanup_runs)")} - row = conn.execute( - """ + row = conn.execute(""" SELECT unused_accounts_deleted FROM cleanup_runs - """ - ).fetchone() + """).fetchone() assert columns == {"id", "created_at", "unused_accounts_deleted"} assert tuple(row) == (1,) @@ -225,16 +221,11 @@ def test_sqlite_startup_tuning_is_idempotent(tmp_path: Path) -> None: temp_store = conn.execute("PRAGMA temp_store").fetchone()[0] mmap_size = conn.execute("PRAGMA mmap_size").fetchone()[0] user_version = conn.execute("PRAGMA user_version").fetchone()[0] - index_names = { - row[1] - for row in conn.execute( - """ + index_names = {row[1] for row in conn.execute(""" SELECT type, name FROM sqlite_master WHERE type = 'index' - """ - ) - } + """)} assert busy_timeout == 10_000 assert journal_mode == "wal" @@ -255,8 +246,7 @@ def test_accounts_schema_migration_removes_legacy_router_credentials(tmp_path: P database_path = tmp_path / "ddns.sqlite3" with sqlite3.connect(database_path) as conn: conn.execute("PRAGMA user_version = 2") - conn.execute( - """ + conn.execute(""" CREATE TABLE accounts ( hostname TEXT PRIMARY KEY, username TEXT NOT NULL, @@ -268,8 +258,7 @@ def test_accounts_schema_migration_removes_legacy_router_credentials(tmp_path: P accepted_at TEXT, disabled INTEGER NOT NULL DEFAULT 0 ) - """ - ) + """) conn.execute( """ INSERT INTO accounts( @@ -303,7 +292,9 @@ def test_accounts_schema_migration_removes_legacy_router_credentials(tmp_path: P assert account["hostname"] == "home.ddns.example.net" -def test_rate_limit_flush_requeues_counts_after_database_failure(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: +def test_rate_limit_flush_requeues_counts_after_database_failure( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: store = DdnsStore(tmp_path / "ddns.sqlite3") assert store.allow_rate_limit("198.51.100.1:/api", 60) is True diff --git a/tests/test_scripts.py b/tests/test_scripts.py index 05c5544..86fd0a2 100644 --- a/tests/test_scripts.py +++ b/tests/test_scripts.py @@ -140,7 +140,10 @@ esac assert "KarlDNS: registering hostname" in result.stdout config = home_dir / ".config" / "karldns" / "updater.env" config_text = config.read_text(encoding="utf-8") - assert "KARLDNS_UPDATE_URL='https://karldns.test/api/v1/router/update/karl_update_secret_12345678901234567890'" in config_text + assert ( + "KARLDNS_UPDATE_URL='https://karldns.test/api/v1/router/update/karl_update_secret_12345678901234567890'" + in config_text + ) assert "KARLDNS_HOSTNAME='abc.karldns.test'" in config_text diff --git a/tests/test_security.py b/tests/test_security.py index e1291cb..726fd2b 100644 --- a/tests/test_security.py +++ b/tests/test_security.py @@ -136,7 +136,7 @@ def test_dashboard_unlock_cookie_rejects_password_hash_signed_forgery(tmp_path: def test_settings_reject_unknown_constructor_fields() -> None: with pytest.raises(ValueError, match="Extra inputs are not permitted"): - DdnsSettings(admn_password="typo") + DdnsSettings(**{"admn_password": "typo"}) def test_trusted_host_middleware_rejects_unlisted_host(tmp_path: Path) -> None: diff --git a/tests/test_ui.py b/tests/test_ui.py index 24b7194..70aa0fd 100644 --- a/tests/test_ui.py +++ b/tests/test_ui.py @@ -58,7 +58,9 @@ def test_legacy_custom_account_form_is_rejected(tmp_path: Path) -> None: ) client = TestClient(app) - response = client.post("/accounts", data={"mode": "custom", "hostname": "home.example.net", "claim_secret": "legacy"}) + response = client.post( + "/accounts", data={"mode": "custom", "hostname": "home.example.net", "claim_secret": "legacy"} + ) assert response.status_code == 404 assert app.state.store.count_accounts() == 0 @@ -174,14 +176,18 @@ def test_generated_hostname_must_be_activated_before_updates_work(tmp_path: Path account = app.state.store.get_account_by_management_slug(management_slug) assert account["accepted_at"] is None - assert client.get(f"/api/v1/router/update/{account['update_slug']}", params={"myip": "203.0.113.9"}).status_code == 404 + assert ( + client.get(f"/api/v1/router/update/{account['update_slug']}", params={"myip": "203.0.113.9"}).status_code == 404 + ) activated = client.post(f"/m/{management_slug}/activate", data={"csrf": _csrf(created.text)}) assert activated.status_code == 200 assert "Hostname registered" in activated.text assert app.state.store.get_account_by_management_slug(management_slug)["accepted_at"] - assert client.get(f"/api/v1/router/update/{account['update_slug']}", params={"myip": "203.0.113.9"}).status_code == 200 + assert ( + client.get(f"/api/v1/router/update/{account['update_slug']}", params={"myip": "203.0.113.9"}).status_code == 200 + ) def test_old_magic_route_is_gone(tmp_path: Path) -> None: @@ -322,7 +328,10 @@ def test_management_dashboard_password_can_be_removed(tmp_path: Path) -> None: unlock = client.post( f"/m/{account['management_slug']}/unlock", - data={"csrf": _csrf(client.get(f"/m/{account['management_slug']}").text), "dashboard_password": "correct horse"}, + data={ + "csrf": _csrf(client.get(f"/m/{account['management_slug']}").text), + "dashboard_password": "correct horse", + }, ) assert unlock.status_code == 200 @@ -430,12 +439,18 @@ def test_management_dashboard_rotates_update_url(tmp_path: Path) -> None: assert "data-secret-toggle" in dashboard.text assert 'type="password"' in dashboard.text - rotated_update = client.post(f"/m/{account['management_slug']}/rotate", data={"csrf": _csrf(dashboard.text), "action": "update"}) + rotated_update = client.post( + f"/m/{account['management_slug']}/rotate", data={"csrf": _csrf(dashboard.text), "action": "update"} + ) assert rotated_update.status_code == 200 assert "Update URL changed. Update your router now." in rotated_update.text assert "data-secret-input" in rotated_update.text - assert client.get(f"/api/v1/router/update/{account['update_slug']}", params={"myip": "203.0.113.9"}).status_code == 404 - statuses = {event["status"] for event in app.state.store.list_update_events_for_hostname("home.ddns.example.net", 10)} + assert ( + client.get(f"/api/v1/router/update/{account['update_slug']}", params={"myip": "203.0.113.9"}).status_code == 404 + ) + statuses = { + event["status"] for event in app.state.store.list_update_events_for_hostname("home.ddns.example.net", 10) + } assert "dashboard_update_url_rotated" in statuses assert "dashboard_router_credentials_rotated" not in statuses @@ -449,13 +464,19 @@ def test_management_dashboard_actions_are_scoped_to_slug_owner(tmp_path: Path) - client = TestClient(app) dashboard = client.get(f"/m/{first['management_slug']}") - rotated = client.post(f"/m/{first['management_slug']}/rotate", data={"csrf": _csrf(dashboard.text), "action": "update"}) + rotated = client.post( + f"/m/{first['management_slug']}/rotate", data={"csrf": _csrf(dashboard.text), "action": "update"} + ) assert rotated.status_code == 200 assert store.get_account_by_hostname("first.ddns.example.net")["update_slug"] != first["update_slug"] assert store.get_account_by_hostname("second.ddns.example.net")["update_slug"] == second["update_slug"] - assert client.get(f"/api/v1/router/update/{first['update_slug']}", params={"myip": "203.0.113.1"}).status_code == 404 - assert client.get(f"/api/v1/router/update/{second['update_slug']}", params={"myip": "203.0.113.2"}).status_code == 200 + assert ( + client.get(f"/api/v1/router/update/{first['update_slug']}", params={"myip": "203.0.113.1"}).status_code == 404 + ) + assert ( + client.get(f"/api/v1/router/update/{second['update_slug']}", params={"myip": "203.0.113.2"}).status_code == 200 + ) def test_locked_dashboard_rejects_mutating_actions_without_unlock_cookie(tmp_path: Path) -> None: @@ -558,7 +579,10 @@ def test_public_examples_use_karldns_domain_and_no_shell_prompt(tmp_path: Path) assert "https://karldns.de/api/v1/router/update/<secret>" in response.text assert "http://127.0.0.1:8099/api/v1/router/update/<secret>" not in response.text assert "$ curl" not in response.text - assert 'data-copy="curl -sS 'https://karldns.de/api/v1/router/update/<secret>?myip=203.0.113.10'"' in response.text + assert ( + 'data-copy="curl -sS 'https://karldns.de/api/v1/router/update/<secret>?myip=203.0.113.10'"' + in response.text + ) def test_application_pages_use_stricter_script_csp(tmp_path: Path) -> None: @@ -579,7 +603,10 @@ def test_application_pages_use_stricter_script_csp(tmp_path: Path) -> None: assert "