import json import re from dataclasses import dataclass from typing import Any from urllib import error as urllib_error from urllib import parse as urllib_parse from urllib import request as urllib_request from .utils import normalize_domain, validate_selector CLOUDFLARE_API_BASE = "https://api.cloudflare.com/client/v4" CLOUDFLARE_RESPONSE_MAX_BYTES = 1024 * 1024 DEFAULT_DNS_TTL = 3600 DNS_RECORD_LABEL_RE = r"[a-z0-9_](?:[a-z0-9_-]{0,61}[a-z0-9_])?" class NoRedirectHandler(urllib_request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None CLOUDFLARE_OPENER = urllib_request.build_opener(NoRedirectHandler()) class DNSProviderError(RuntimeError): pass @dataclass class DNSRecordChange: provider: str action: str zone_id: str record_id: str name: str record_type: str content: str ttl: int def as_dict(self) -> dict[str, object]: return { "provider": self.provider, "action": self.action, "zone_id": self.zone_id, "record_id": self.record_id, "name": self.name, "type": self.record_type, "content": self.content, "ttl": self.ttl, } class CloudflareDNSProvider: provider_name = "cloudflare" def __init__(self, *, api_token: str, zone_id: str, account_id: str, api_base: str = CLOUDFLARE_API_BASE) -> None: self.api_token = api_token.strip() self.zone_id = zone_id.strip() self.account_id = account_id.strip() self.api_base = api_base.rstrip("/") parsed_api_base = urllib_parse.urlparse(self.api_base) if ( parsed_api_base.scheme != "https" or not parsed_api_base.hostname or parsed_api_base.username or parsed_api_base.password or parsed_api_base.query or parsed_api_base.fragment ): raise DNSProviderError("Cloudflare API base URL must be an HTTPS origin without credentials, query or fragment") if not self.api_token: raise DNSProviderError("Cloudflare API token is not configured for this domain") if not self.account_id: raise DNSProviderError("Cloudflare account ID is not configured for this domain") if not self.zone_id: raise DNSProviderError("Cloudflare zone ID is not configured for this domain") def _request(self, method: str, path: str, payload: dict[str, Any] | None = None, params: dict[str, str] | None = None) -> dict[str, Any]: query = f"?{urllib_parse.urlencode(params)}" if params else "" body = json.dumps(payload).encode("utf-8") if payload is not None else None request = urllib_request.Request( f"{self.api_base}{path}{query}", data=body, method=method, headers={ "Authorization": f"Bearer {self.api_token}", "Content-Type": "application/json", "Accept": "application/json", }, ) try: with CLOUDFLARE_OPENER.open(request, timeout=20) as response: final_url = response.geturl() if hasattr(response, "geturl") else request.full_url final_parsed = urllib_parse.urlparse(final_url) if ( final_parsed.scheme != "https" or not final_parsed.hostname or final_parsed.username or final_parsed.password ): raise DNSProviderError("Cloudflare API redirected to an insecure URL") response_body = response.read(CLOUDFLARE_RESPONSE_MAX_BYTES + 1) except DNSProviderError: raise except urllib_error.HTTPError as exc: detail = self._error_detail(exc) raise DNSProviderError(f"Cloudflare API error: {detail}") from exc except urllib_error.URLError as exc: raise DNSProviderError(f"Cloudflare API connection failed: {exc.reason}") from exc if len(response_body) > CLOUDFLARE_RESPONSE_MAX_BYTES: raise DNSProviderError("Cloudflare API response is too large") try: data = json.loads(response_body.decode("utf-8")) except json.JSONDecodeError as exc: raise DNSProviderError("Cloudflare API returned invalid JSON") from exc if not data.get("success"): raise DNSProviderError(f"Cloudflare API error: {cloudflare_error_message(data)}") return data @staticmethod def _error_detail(exc: urllib_error.HTTPError) -> str: try: response_body = exc.read(CLOUDFLARE_RESPONSE_MAX_BYTES + 1) if len(response_body) <= CLOUDFLARE_RESPONSE_MAX_BYTES: payload = json.loads(response_body.decode("utf-8")) return cloudflare_error_message(payload) except Exception: pass return f"HTTP {exc.code}" def _validate_zone_account(self) -> None: result = self._request("GET", f"/zones/{urllib_parse.quote(self.zone_id, safe='')}").get("result") or {} zone_account = result.get("account") if isinstance(result, dict) else {} zone_account_id = "" if isinstance(zone_account, dict): zone_account_id = str(zone_account.get("id") or "").strip() if zone_account_id and zone_account_id != self.account_id: raise DNSProviderError("Cloudflare account ID does not match the selected zone") def upsert_txt_record(self, *, name: str, content: str, ttl: int = DEFAULT_DNS_TTL) -> DNSRecordChange: record_name = normalize_dns_record_name(name) record_content = normalize_txt_content(content) ttl_value = validate_ttl(ttl) self._validate_zone_account() existing = self._request( "GET", f"/zones/{urllib_parse.quote(self.zone_id, safe='')}/dns_records", params={"type": "TXT", "name": record_name}, ).get("result", []) payload = {"type": "TXT", "name": record_name, "content": record_content, "ttl": ttl_value, "proxied": False} if existing: record_id = str(existing[0].get("id") or "") if not record_id: raise DNSProviderError("Cloudflare returned a DNS record without an ID") result = self._request( "PUT", f"/zones/{urllib_parse.quote(self.zone_id, safe='')}/dns_records/{urllib_parse.quote(record_id, safe='')}", payload, ) action = "updated" else: result = self._request("POST", f"/zones/{urllib_parse.quote(self.zone_id, safe='')}/dns_records", payload) action = "created" record = result.get("result") or {} record_id = str(record.get("id") or "") if not record_id: raise DNSProviderError("Cloudflare did not return the DNS record ID") return DNSRecordChange( provider=self.provider_name, action=action, zone_id=self.zone_id, record_id=record_id, name=record_name, record_type="TXT", content=record_content, ttl=ttl_value, ) def cloudflare_error_message(payload: dict[str, Any]) -> str: errors = payload.get("errors") if isinstance(errors, list) and errors: messages = [] for item in errors[:3]: if isinstance(item, dict): code = item.get("code") message = item.get("message") or "unknown error" messages.append(f"{code}: {message}" if code else str(message)) if messages: return "; ".join(messages) return "request failed" def normalize_dns_record_name(value: str) -> str: candidate = value.strip().rstrip(".").lower() try: candidate = candidate.encode("idna").decode("ascii") except UnicodeError as exc: raise DNSProviderError("Invalid DNS record name") from exc labels = candidate.split(".") if not labels or len(candidate) > 253: raise DNSProviderError("Invalid DNS record name") for label in labels: if not label or len(label) > 63: raise DNSProviderError("Invalid DNS record name") if not re.fullmatch(DNS_RECORD_LABEL_RE, label): raise DNSProviderError("Invalid DNS record name") return candidate def normalize_txt_content(value: str) -> str: content = value.strip() if len(content) >= 2 and content[0] == content[-1] == '"': content = content[1:-1].strip() if not content: raise DNSProviderError("DNS TXT content is empty") return content def validate_ttl(value: int) -> int: ttl = int(value) if ttl != 1 and ttl < 60: raise DNSProviderError("DNS TTL must be 1 for automatic or at least 60 seconds") return ttl def zone_relative_record_name(domain_name: str, record_name: str) -> str: safe_domain = normalize_domain(domain_name) safe_record = normalize_dns_record_name(record_name) if safe_record == safe_domain: return "@" suffix = f".{safe_domain}" if safe_record.endswith(suffix): return safe_record[: -len(suffix)] return f"{safe_record}." def dkim_record_name(domain_name: str, selector: str) -> str: safe_domain = normalize_domain(domain_name) safe_selector = validate_selector(selector) return f"{safe_selector}._domainkey.{safe_domain}" def publish_dkim_record_for_domain(domain, *, ttl: int = DEFAULT_DNS_TTL) -> DNSRecordChange: provider = (getattr(domain, "dns_provider", "") or "").strip().lower() if provider != "cloudflare": raise DNSProviderError("Only the cloudflare DNS provider is currently supported") public_key = (getattr(domain, "dkim_public_key", "") or "").strip() if not public_key: raise DNSProviderError("Generate DKIM before publishing the DNS record") # Decrypt the stored API token if it was encrypted at rest raw_token = getattr(domain, "dns_api_token", "") or "" from .crypto import decrypt_token api_token = decrypt_token(raw_token) or "" cloudflare = CloudflareDNSProvider( api_token=api_token, account_id=getattr(domain, "dns_account_id", "") or "", zone_id=getattr(domain, "dns_zone_id", "") or "", ) return cloudflare.upsert_txt_record( name=dkim_record_name(getattr(domain, "name", ""), getattr(domain, "dkim_selector", "default") or "default"), content=public_key, ttl=ttl, )