karldns.de/karldns/services/ddns.py

from __future__ import annotations

import ipaddress
import logging
import re
import secrets
import sqlite3
import threading
import time
import urllib.parse
from collections.abc import Mapping
from dataclasses import dataclass

from fastapi import HTTPException, Request

from ..dns import DnsProviderError, DnsReconciliationError, delete_dns_records, dns_backend_configured, publish_dns
from ..event_statuses import DNS_DELETE_FAILED, DNS_FAILED, DNS_RECONCILE_FAILED, UNCHANGED, UPDATED
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"


@dataclass(slots=True)
class UpdateOutcome:
    result: UpdateResult
    source_ip: str


@dataclass(frozen=True, slots=True)
class UpdateIpParams:
    myip: str | None = None
    ipaddr: str | None = None
    myipv6: str | None = None
    ip6addr: str | None = None

    @property
    def initial_plan_args(self) -> tuple[str | None, str | None, str | None, str | None]:
        return self.myip, self.ipaddr, self.myipv6, self.ip6addr

    def with_body_values(self, body_values: Mapping[str, str]) -> UpdateIpParams:
        return UpdateIpParams(
            myip=self.myip or body_values.get("myip"),
            ipaddr=self.ipaddr or body_values.get("ipaddr"),
            myipv6=self.myipv6 or body_values.get("myipv6"),
            ip6addr=self.ip6addr or body_values.get("ip6addr"),
        )


@dataclass(frozen=True, slots=True)
class InitialDnsPlan:
    ipv4: str | None
    ipv6: str | None
    update_ipv4: bool
    update_ipv6: bool
    detail: str


class BadIpError(ValueError):
    """Raised when a supplied DynDNS IP parameter is invalid."""


class DdnsService:
    _LOCK_STRIPES = 256

    def __init__(self, settings: DdnsSettings, store: DdnsStore) -> None:
        self.settings = settings
        self.store = store
        self._locks = [threading.Lock() for _ in range(self._LOCK_STRIPES)]

    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):
            try:
                return self.store.create_account(self.random_managed_hostname(), accepted=False)
            except sqlite3.IntegrityError:
                continue
        raise HTTPException(status_code=500, detail="could not allocate hostname")

    def delete_account(self, hostname: str, source: str) -> None:
        try:
            delete_dns_records(hostname, self.settings)
        except DnsProviderError as exc:
            LOGGER.warning(
                "DNS delete failed for hostname=%s source=%s error_type=%s",
                hostname,
                source,
                type(exc).__name__,
            )
            self.store.log_update_event(hostname, None, None, DNS_DELETE_FAILED, f"{type(exc).__name__}: {exc}", source)
            raise
        self.store.delete_account(hostname)

    def publish_initial_dns(
        self,
        request: Request,
        hostname: str,
        ip_params: UpdateIpParams,
    ) -> bool:
        if not dns_backend_configured(self.settings):
            return False
        source_ip = client_ip(request, self.settings)
        plan = _initial_dns_plan(source_ip, *ip_params.initial_plan_args)
        lock = self._lock_for(hostname)
        with lock:
            planned = self.store.preview_update(
                hostname,
                plan.ipv4,
                plan.ipv6,
                update_ipv4=plan.update_ipv4,
                update_ipv6=plan.update_ipv6,
            )
            if planned.changed:
                try:
                    publish_dns(planned, self.settings)
                except DnsReconciliationError as exc:
                    LOGGER.warning(
                        "Initial DNS reconciliation failed for hostname=%s update_ipv4=%s update_ipv6=%s error_type=%s",
                        hostname,
                        planned.update_ipv4,
                        planned.update_ipv6,
                        type(exc).__name__,
                    )
                    self.store.log_update_event(
                        hostname,
                        plan.ipv4,
                        plan.ipv6,
                        DNS_RECONCILE_FAILED,
                        f"{type(exc).__name__}: {exc}",
                        source_ip,
                    )
                    raise
                except DnsProviderError as exc:
                    LOGGER.warning(
                        "Initial DNS publish failed for hostname=%s update_ipv4=%s update_ipv6=%s error_type=%s",
                        hostname,
                        planned.update_ipv4,
                        planned.update_ipv6,
                        type(exc).__name__,
                    )
                    self.store.log_update_event(
                        hostname,
                        plan.ipv4,
                        plan.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, plan.detail, source_ip)
        return True

    def apply_update(
        self,
        request: Request,
        hostname: str,
        ip_params: UpdateIpParams,
    ) -> UpdateOutcome:
        started_at = time.perf_counter()
        is_success = False
        is_dns_failure = False
        is_reconcile_failure = False
        is_bad_ip = False
        source_ip = client_ip(request, self.settings)
        try:
            ipv4, ipv6, update_ipv4, update_ipv6 = parse_update_ips(
                source_ip,
                *ip_params.initial_plan_args,
            )
            lock = self._lock_for(hostname)
            with lock:
                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)
                    except DnsReconciliationError as exc:
                        is_reconcile_failure = True
                        LOGGER.warning(
                            "DNS reconciliation failed for hostname=%s update_ipv4=%s update_ipv6=%s error_type=%s",
                            hostname,
                            planned.update_ipv4,
                            planned.update_ipv6,
                            type(exc).__name__,
                        )
                        self.store.log_update_event(
                            hostname,
                            ipv4,
                            ipv6,
                            DNS_RECONCILE_FAILED,
                            f"{type(exc).__name__}: {exc}",
                            source_ip,
                        )
                        raise
                    except DnsProviderError as exc:
                        is_dns_failure = True
                        LOGGER.warning(
                            "DNS publish failed for hostname=%s update_ipv4=%s update_ipv6=%s error_type=%s",
                            hostname,
                            planned.update_ipv4,
                            planned.update_ipv6,
                            type(exc).__name__,
                        )
                        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
                    self.store.log_update_event(hostname, result.ipv4, result.ipv6, UPDATED, "ok", source_ip)
                else:
                    result = planned
                    is_success = True
                    self.store.log_update_rollup(hostname, result.ipv4, result.ipv6, UNCHANGED, "ok", source_ip)
            return UpdateOutcome(result=result, source_ip=source_ip)
        except BadIpError:
            is_bad_ip = True
            raise
        finally:
            elapsed_ms = (time.perf_counter() - started_at) * 1000
            self.store.record_update_metrics(
                elapsed_ms,
                is_success=is_success,
                is_dns_failure=is_dns_failure,
                is_reconcile_failure=is_reconcile_failure,
                is_bad_ip=is_bad_ip,
            )

    def fritz_update_url(self, account: Mapping[str, object]) -> str:
        base = self.settings.public_base_url.rstrip("/")
        slug = urllib.parse.quote(str(account["update_slug"]))
        return f"{base}/api/v1/router/update/{slug}?myip=<ipaddr>&myipv6=<ip6addr>"

    def dashboard_url(self, account: Mapping[str, object]) -> str:
        base = self.settings.public_base_url.rstrip("/")
        slug = urllib.parse.quote(str(account["management_slug"]))
        return f"{base}/m/{slug}"

    def registration_url(self, account: Mapping[str, object]) -> str:
        base = self.settings.public_base_url.rstrip("/")
        slug = urllib.parse.quote(str(account["management_slug"]))
        return f"{base}/api/v1/router/register/{slug}"

    def random_managed_hostname(self) -> str:
        suffix = self.settings.hostname_suffix.strip().lower().strip(".")
        if not suffix:
            raise HTTPException(status_code=500, detail="DDNS_HOSTNAME_SUFFIX is required for managed hostnames")
        return f"{secrets.token_hex(6)}.{suffix}"

    def hostname_is_managed(self, hostname: str) -> bool:
        suffix = self.settings.hostname_suffix.strip().lower().strip(".")
        if not suffix:
            return True
        normalized = hostname.strip().lower().rstrip(".")
        return normalized == suffix or normalized.endswith(f".{suffix}")

    def _lock_for(self, hostname: str) -> threading.Lock:
        return self._locks[hash(hostname) % self._LOCK_STRIPES]


def normalize_ip(value: str | None, version: int | None = None) -> str | None:
    ip = _optional_ip(value)
    if ip is None or (version and ip.version != version):
        return None
    return str(ip)


def parse_update_ips(
    source_ip: str,
    myip: str | None,
    ipaddr: str | None,
    myipv6: str | None,
    ip6addr: str | None,
) -> tuple[str | None, str | None, bool, bool]:
    ipv4 = _first_valid_ip((("myip", myip), ("ipaddr", ipaddr)), version=4)
    ipv6 = _first_valid_ip((("myipv6", myipv6), ("ip6addr", ip6addr)), version=6)
    update_ipv4 = ipv4 is not None
    update_ipv6 = ipv6 is not None
    if not update_ipv4 and not update_ipv6:
        ipv4 = normalize_ip(source_ip, version=4)
        ipv6 = normalize_ip(source_ip, version=6)
        update_ipv4 = ipv4 is not None
        update_ipv6 = ipv6 is not None
    if not update_ipv4 and not update_ipv6:
        raise BadIpError("source IP is not a valid IPv4 or IPv6 address")
    return ipv4, ipv6, update_ipv4, update_ipv6


def _initial_dns_plan(
    source_ip: str,
    myip: str | None,
    ipaddr: str | None,
    myipv6: str | None,
    ip6addr: str | None,
) -> InitialDnsPlan:
    source_fallback = source_ip if _is_global_ip(source_ip) else ""
    if not _has_supplied_ip(myip, ipaddr, myipv6, ip6addr) and not source_fallback:
        return InitialDnsPlan(
            ipv4=INITIAL_PLACEHOLDER_IPV4,
            ipv6=None,
            update_ipv4=True,
            update_ipv6=False,
            detail="initial placeholder dns registered",
        )
    ipv4, ipv6, update_ipv4, update_ipv6 = parse_update_ips(
        source_fallback,
        myip,
        ipaddr,
        myipv6,
        ip6addr,
    )
    return InitialDnsPlan(
        ipv4=ipv4,
        ipv6=ipv6,
        update_ipv4=update_ipv4,
        update_ipv6=update_ipv6,
        detail="initial dns registered",
    )


def _has_supplied_ip(*values: str | None) -> bool:
    return any(value and value.strip() for value in values)


def _is_global_ip(value: str | None) -> bool:
    if not value:
        return False
    try:
        return ipaddress.ip_address(value.strip()).is_global
    except ValueError:
        return False


def _first_valid_ip(params: tuple[tuple[str, str | None], ...], version: int) -> str | None:
    selected = None
    for name, value in params:
        if value is None or not value.strip():
            continue
        ip_value = _required_ip_param(name, value, version)
        if selected is None:
            selected = ip_value
    return selected


def _optional_ip(value: str | None) -> ipaddress.IPv4Address | ipaddress.IPv6Address | None:
    if not value or not value.strip():
        return None
    try:
        return ipaddress.ip_address(value.strip())
    except ValueError:
        return None


def _required_ip_param(name: str, value: str, version: int) -> str:
    try:
        ip = ipaddress.ip_address(value.strip())
    except ValueError as exc:
        raise BadIpError(f"{name} must be a valid IPv{version} address") from exc
    if ip.version != version:
        raise BadIpError(f"{name} must be an IPv{version} address")
    return str(ip)


def normalize_hostname(value: str, settings: DdnsSettings, expand_suffix: bool = True) -> str:
    hostname = value.strip().lower().rstrip(".")
    suffix = settings.hostname_suffix.strip().lower().strip(".")
    if expand_suffix and suffix and "." not in hostname:
        hostname = f"{hostname}.{suffix}"
    if not _valid_domain_name(hostname):
        return ""
    return hostname


def _valid_domain_name(value: str) -> bool:
    label = r"[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?"
    return bool(re.fullmatch(rf"{label}(\.{label})+", value))