karldns.de/router_dyndns/ddns_models.py

from __future__ import annotations

import re
import urllib.parse
from dataclasses import dataclass
from pathlib import Path

from pydantic import BaseModel, ConfigDict, Field, field_validator


class DdnsSettings(BaseModel):
    model_config = ConfigDict(extra="forbid")

    database_path: Path = Field(default=Path("ddns.sqlite3"))
    shared_secret: str = Field(default="")
    admin_password: str = Field(default="")
    admin_path: str = Field(default="/admin")
    allowed_hosts: set[str] = Field(default_factory=set)
    public_base_url: str = Field(default="http://localhost:8080")
    trusted_hosts: set[str] = Field(default_factory=set)
    hostname_suffix: str = Field(default="")
    verification_prefix: str = Field(default="_karldns")
    rate_limit_per_minute: int = Field(default=60, ge=1, le=10_000)
    admin_rate_limit_per_minute: int = Field(default=20, ge=1, le=10_000)
    max_request_body_bytes: int = Field(default=16_384, ge=1024, le=1_048_576)
    trusted_proxy_ips: set[str] = Field(default_factory=set)
    rfc2136_server: str | None = None
    rfc2136_zone: str | None = None
    rfc2136_key_name: str | None = None
    rfc2136_key_secret: str | None = None
    dns_provider: str = Field(default="")
    dns_zones: set[str] = Field(default_factory=set)
    require_dns_provider: bool = Field(default=False)
    cloudflare_api_token: str = Field(default="")
    cloudflare_zone_id: str = Field(default="")
    cleanup_challenge_hours: int = Field(default=72, ge=1, le=24 * 365)
    cleanup_unused_account_hours: int = Field(default=1080, ge=1, le=24 * 365)
    cleanup_event_history_hours: int = Field(default=24 * 180, ge=1, le=24 * 3650)
    cleanup_interval_seconds: int = Field(default=3600, ge=1, le=24 * 60 * 60)
    dashboard_unlock_seconds: int = Field(default=24 * 60 * 60, ge=60, le=30 * 24 * 60 * 60)
    dashboard_stay_unlocked_seconds: int = Field(default=30 * 24 * 60 * 60, ge=60, le=30 * 24 * 60 * 60)
    dashboard_cookie_secret: str = Field(default="")
    redacted_access_log: bool = Field(default=False)
    ttl: int = Field(default=60, ge=60, le=86400)

    @field_validator("public_base_url")
    @classmethod
    def validate_public_base_url(cls, value: str) -> str:
        if not value.startswith(("http://", "https://")):
            raise ValueError("public_base_url must start with http:// or https://")
        return value.rstrip("/")

    @field_validator("trusted_hosts")
    @classmethod
    def validate_trusted_hosts(cls, value: set[str]) -> set[str]:
        return {host.strip().lower() for host in value if host.strip()}

    @field_validator("admin_path")
    @classmethod
    def validate_admin_path(cls, value: str) -> str:
        path = value.strip()
        if not path.startswith("/") or path == "/" or path.endswith("/"):
            raise ValueError("admin_path must start with / and must not be / or end with /")
        if not re.fullmatch(r"/[A-Za-z0-9._~/-]+", path):
            raise ValueError("admin_path may only contain URL-safe path characters")
        reserved_paths = {
            "/accounts",
            "/admin.css",
            "/admin-tables.js",
            "/api",
            "/app.js",
            "/claim",
            "/clipboard.js",
            "/d",
            "/docs",
            "/favicon.ico",
            "/healthz",
            "/install.sh",
            "/ip",
            "/bootstrap-subset.css",
            "/logo.png",
            "/logo-nav.webp",
            "/logo.webp",
            "/m",
            "/magic",
            "/nic",
            "/openapi.json",
            "/privacy",
            "/redoc",
            "/robots.txt",
            "/sitemap.xml",
            "/site.webmanifest",
            "/social-preview.png",
            "/social-preview.webp",
            "/theme.js",
            "/u",
            "/uninstall.sh",
        }
        if any(path == reserved or path.startswith(f"{reserved}/") for reserved in reserved_paths):
            raise ValueError("admin_path must not conflict with a public route")
        return path

    @property
    def effective_trusted_hosts(self) -> list[str]:
        hosts = set(self.trusted_hosts)
        public_host = urllib.parse.urlparse(self.public_base_url).hostname
        if public_host:
            hosts.add(public_host.lower())
        if not self.require_dns_provider:
            hosts.update({"testserver", "localhost", "127.0.0.1", "::1"})
        return sorted(hosts)

    def validate_launch_ready(self) -> None:
        from .dns import validate_dns_provider_settings

        validate_dns_provider_settings(self)
        if not self.require_dns_provider:
            return
        if len(self.admin_password) < 16:
            raise RuntimeError("DDNS_ADMIN_PASSWORD must be at least 16 characters when DDNS_REQUIRE_DNS_PROVIDER=1")
        if len(self.dashboard_cookie_secret or self.admin_password) < 32:
            raise RuntimeError(
                "DDNS_DASHBOARD_COOKIE_SECRET or DDNS_ADMIN_PASSWORD must be at least 32 characters for dashboard cookie signing when DDNS_REQUIRE_DNS_PROVIDER=1"
            )
        if not self.hostname_suffix:
            raise RuntimeError("DDNS_HOSTNAME_SUFFIX is required when DDNS_REQUIRE_DNS_PROVIDER=1")
        if not self.effective_trusted_hosts:
            raise RuntimeError("At least one trusted host is required for production")

    @classmethod
    def from_env(cls) -> DdnsSettings:
        from .config import settings_from_env

        return settings_from_env()


@dataclass(slots=True)
class UpdateResult:
    hostname: str
    ipv4: str | None
    ipv6: str | None
    changed: bool
    update_ipv4: bool = True
    update_ipv6: bool = True

    @property
    def dyndns_response(self) -> str:
        ip = self.ipv4 or self.ipv6 or "0.0.0.0"
        return f"{'good' if self.changed else 'nochg'} {ip}"