karldns.de/karldns/dns/cloudflare.py

from __future__ import annotations

import json
import urllib.error
import urllib.parse
import urllib.request
from typing import Any, cast

from ..models import UpdateResult
from ..settings import DdnsSettings
from .base import DnsProviderError, DnsReconciliationError


def publish_cloudflare(result: UpdateResult, settings: DdnsSettings) -> None:
    if not settings.cloudflare_api_token or not settings.cloudflare_zone_id:
        raise DnsProviderError("Cloudflare DNS provider is missing required credentials")
    if result.update_ipv4:
        _cloudflare_upsert_record(settings, result.hostname, "A", result.ipv4)
    if result.update_ipv6:
        _cloudflare_upsert_record(settings, result.hostname, "AAAA", result.ipv6)


def delete_cloudflare_records(hostname: str, settings: DdnsSettings) -> None:
    if not settings.cloudflare_api_token or not settings.cloudflare_zone_id:
        raise DnsProviderError("Cloudflare DNS provider is missing required credentials")
    for record_type in ("A", "AAAA"):
        record = _cloudflare_find_record(settings, hostname, record_type)
        if record:
            _cloudflare_request(settings, f"/dns_records/{record['id']}", method="DELETE")


def _cloudflare_upsert_record(settings: DdnsSettings, hostname: str, record_type: str, value: str | None) -> None:
    record = _cloudflare_find_record(settings, hostname, record_type)
    if value is None:
        if record:
            _cloudflare_request(settings, f"/dns_records/{record['id']}", method="DELETE")
        return

    payload = {
        "type": record_type,
        "name": hostname,
        "content": value,
        "ttl": settings.ttl,
        "proxied": False,
    }
    if record:
        _cloudflare_request(settings, f"/dns_records/{record['id']}", method="PATCH", payload=payload)
    else:
        _cloudflare_request(settings, "/dns_records", method="POST", payload=payload)
    _cloudflare_reconcile_record(settings, hostname, record_type, value)


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", [])
    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:
    record = _cloudflare_find_record(settings, hostname, record_type)
    if not record:
        raise DnsReconciliationError(f"Cloudflare {record_type} record was not visible after publish")
    if record.get("content") != value:
        raise DnsReconciliationError(f"Cloudflare {record_type} record did not match requested content after publish")


def _cloudflare_request(
    settings: DdnsSettings,
    path: str,
    method: str = "GET",
    payload: dict[str, object] | None = None,
) -> 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)
    request.add_header("Authorization", f"Bearer {settings.cloudflare_api_token}")
    request.add_header("Content-Type", "application/json")
    try:
        with urllib.request.urlopen(request, timeout=10) as response:
            parsed = json.loads(response.read().decode("utf-8"))
    except urllib.error.HTTPError as exc:
        raise DnsProviderError(f"Cloudflare API returned HTTP {exc.code}") from exc
    except urllib.error.URLError as exc:
        raise DnsProviderError("Cloudflare API request failed") from exc
    except json.JSONDecodeError as exc:
        raise DnsProviderError("Cloudflare API returned invalid JSON") from exc
    if not parsed.get("success", False):
        raise DnsProviderError("Cloudflare API returned an unsuccessful response")
    return parsed