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 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

LOGGER = logging.getLogger(__name__)


@dataclass(slots=True)
class UpdateOutcome:
    result: UpdateResult
    source_ip: 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) -> dict[str, str]:
        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,
        myip: str | None,
        ipaddr: str | None,
        myipv6: str | None,
        ip6addr: str | None,
    ) -> None:
        if not dns_backend_configured(self.settings):
            return
        source_ip = client_ip(request, self.settings)
        source_fallback = source_ip if _is_global_ip(source_ip) else ""
        try:
            ipv4, ipv6, update_ipv4, update_ipv6 = parse_update_ips(
                source_fallback,
                myip,
                ipaddr,
                myipv6,
                ip6addr,
            )
        except BadIpError as exc:
            raise BadIpError("registration needs a public source IP or myip/myipv6 parameter") from exc
        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:
                    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, ipv4, 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, 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, "initial dns registered", source_ip)

    def apply_update(
        self,
        request: Request,
        hostname: str,
        myip: str | None,
        ipaddr: str | None,
        myipv6: str | None,
        ip6addr: str | None,
    ) -> 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,
                myip,
                ipaddr,
                myipv6,
                ip6addr,
            )
            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: dict[str, str]) -> str:
        base = self.settings.public_base_url.rstrip("/")
        slug = urllib.parse.quote(account["update_slug"])
        return f"{base}/api/v1/router/update/{slug}?myip=<ipaddr>&myipv6=<ip6addr>"

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

    def registration_url(self, account: dict[str, str]) -> str:
        base = self.settings.public_base_url.rstrip("/")
        slug = urllib.parse.quote(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:
    if not value or not value.strip():
        return None
    try:
        ip = ipaddress.ip_address(value.strip())
    except ValueError:
        return None
    if 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 _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
        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")
        if selected is None:
            selected = str(ip)
    return selected


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))