Archived
101 lines
4.2 KiB
Python
101 lines
4.2 KiB
Python
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")
|
|
parsed = _read_cloudflare_json(request)
|
|
_raise_for_cloudflare_error(parsed)
|
|
return parsed
|
|
|
|
|
|
def _read_cloudflare_json(request: urllib.request.Request) -> dict[str, Any]:
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=10) as response:
|
|
return cast(dict[str, Any], 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
|
|
|
|
|
|
def _raise_for_cloudflare_error(parsed: dict[str, Any]) -> None:
|
|
if not parsed.get("success", False):
|
|
raise DnsProviderError("Cloudflare API returned an unsuccessful response")
|