"""Cloudflare REST API v4 integration. Credentials dict keys: api_token / token / apiToken – Cloudflare API token (Bearer auth) OR global_api_key / api_key + email – Cloudflare Global API key Note: Cloudflare manages DNS zones, not domain registrations. Expiry information is only available for domains registered through Cloudflare Registrar — for others, expiry_date is None and status is 'active'. """ from __future__ import annotations import logging from datetime import date, datetime from typing import Any import httpx from integrations.base import RegistrarBase logger = logging.getLogger(__name__) _BASE_URL = "https://api.cloudflare.com/client/v4" def _parse_date(value: str | None) -> date | None: """Parse an ISO-8601 datetime string and return the date part.""" if not value: return None try: return datetime.fromisoformat(value.replace("Z", "+00:00")).date() except ValueError: return None class CloudflareIntegration(RegistrarBase): def __init__(self, credentials: dict[str, Any]) -> None: super().__init__(credentials) self.api_token: str = str( credentials.get("api_token") or credentials.get("token") or credentials.get("apiToken") or credentials.get("auth_token") or credentials.get("bearer_token") or "" ).strip() self.api_key: str = str( credentials.get("global_api_key") or credentials.get("api_key") or credentials.get("key") or credentials.get("apiKey") or "" ).strip() self.email: str = str( credentials.get("email") or credentials.get("user_email") or credentials.get("auth_email") or "" ).strip() def _headers(self) -> dict[str, str]: headers = {"Content-Type": "application/json"} if self.api_token: headers["Authorization"] = f"Bearer {self.api_token}" elif self.api_key and self.email: headers["X-Auth-Key"] = self.api_key headers["X-Auth-Email"] = self.email elif self.api_key: headers["Authorization"] = f"Bearer {self.api_key}" return headers def _raise_for_cf_errors(self, data: dict[str, Any], context: str = "") -> None: """Raise ValueError if Cloudflare returned success=False.""" if not data.get("success", True): errors = data.get("errors", []) msg = "; ".join( f"[{e.get('code')}] {e.get('message', '')}" for e in errors ) raise ValueError(f"Cloudflare API error{f' ({context})' if context else ''}: {msg}") # ------------------------------------------------------------------ # list_domains (zones) # ------------------------------------------------------------------ async def list_domains(self) -> list[dict[str, Any]]: """List all Cloudflare zones and attempt to get registrar expiry info.""" zones: list[dict[str, Any]] = [] page = 1 per_page = 50 async with httpx.AsyncClient(timeout=30, headers=self._headers()) as client: while True: resp = await client.get( f"{_BASE_URL}/zones", params={"per_page": per_page, "page": page, "status": "active"}, ) resp.raise_for_status() data = resp.json() self._raise_for_cf_errors(data, "list zones") result_info = data.get("result_info", {}) total_pages = result_info.get("total_pages", 1) for zone in data.get("result", []): zone_id: str = zone["id"] domain_name: str = zone["name"] zone_status: str = zone.get("status", "active") status = "active" if zone_status == "active" else "unknown" expiry_date: date | None = None # Attempt to get registrar domain info (Cloudflare Registrar only) try: reg_resp = await client.get( f"{_BASE_URL}/zones/{zone_id}/registrar/domains" ) if reg_resp.status_code == 200: reg_data = reg_resp.json() if reg_data.get("success") and reg_data.get("result"): reg_info = reg_data["result"] if isinstance(reg_info, list) and reg_info: reg_info = reg_info[0] if isinstance(reg_info, dict): expiry_date = _parse_date(reg_info.get("expires_at")) except Exception as e: logger.debug(f"Cloudflare Registrar info unavailable for {domain_name}: {e}") zones.append( { "domain_name": domain_name, "registrar": "cloudflare", "status": status, "expiry_date": expiry_date, "auto_renew": True, "raw": {"zone_id": zone_id, "status": zone_status}, } ) page += 1 if page > total_pages: break return zones # ------------------------------------------------------------------ # get_dns_records (Required by RegistrarBase abstract class) # ------------------------------------------------------------------ async def get_dns_records(self, domain_name: str) -> list[dict[str, Any]]: """Fetch DNS records for a given domain/zone from Cloudflare API v4.""" records: list[dict[str, Any]] = [] async with httpx.AsyncClient(timeout=30, headers=self._headers()) as client: resp = await client.get(f"{_BASE_URL}/zones", params={"name": domain_name.strip()}) if resp.status_code != 200: return records data = resp.json() zones = data.get("result", []) if not zones: return records zone_id = zones[0]["id"] rec_resp = await client.get(f"{_BASE_URL}/zones/{zone_id}/dns_records", params={"per_page": 100}) if rec_resp.status_code != 200: return records rec_data = rec_resp.json() for rec in rec_data.get("result", []): records.append( { "record_type": rec.get("type", ""), "name": rec.get("name", ""), "value": rec.get("content", ""), "ttl": rec.get("ttl"), "priority": rec.get("priority"), } ) return records # ------------------------------------------------------------------ # test_credentials / test_connection # ------------------------------------------------------------------ async def test_connection(self) -> bool: """Verify credentials by calling GET /user/tokens/verify or GET /zones.""" async with httpx.AsyncClient(timeout=15, headers=self._headers()) as client: if self.api_token: resp = await client.get(f"{_BASE_URL}/user/tokens/verify") if resp.status_code == 200 and resp.json().get("success"): return True resp = await client.get(f"{_BASE_URL}/zones", params={"per_page": 1}) if resp.status_code == 200: data = resp.json() return data.get("success") is True return False async def test_credentials(self) -> bool: return await self.test_connection() # ------------------------------------------------------------------ # purge_cache # ------------------------------------------------------------------ async def purge_cache( self, zone_id: str, purge_everything: bool = True, files: list[str] | None = None ) -> dict[str, Any]: """Purge cache for a specific Cloudflare zone.""" payload: dict[str, Any] = {} if purge_everything: payload["purge_everything"] = True elif files: payload["files"] = files else: payload["purge_everything"] = True async with httpx.AsyncClient(timeout=30, headers=self._headers()) as client: resp = await client.post( f"{_BASE_URL}/zones/{zone_id}/purge_cache", json=payload, ) resp.raise_for_status() data = resp.json() self._raise_for_cf_errors(data, "purge cache") return data.get("result", {}) # ------------------------------------------------------------------ # development_mode (get & set) # ------------------------------------------------------------------ async def get_development_mode(self, zone_id: str) -> str: """Fetch current development_mode setting ('on' or 'off') for a zone.""" async with httpx.AsyncClient(timeout=15, headers=self._headers()) as client: resp = await client.get( f"{_BASE_URL}/zones/{zone_id}/settings/development_mode" ) resp.raise_for_status() data = resp.json() self._raise_for_cf_errors(data, "get development_mode") result = data.get("result", {}) return result.get("value", "off") async def set_development_mode(self, zone_id: str, value: str) -> dict[str, Any]: """Set development_mode to 'on' or 'off' for a zone.""" valid_val = "on" if value.lower() in ("on", "true", "1") else "off" async with httpx.AsyncClient(timeout=15, headers=self._headers()) as client: resp = await client.patch( f"{_BASE_URL}/zones/{zone_id}/settings/development_mode", json={"value": valid_val}, ) resp.raise_for_status() data = resp.json() self._raise_for_cf_errors(data, "set development_mode") return data.get("result", {}) # ------------------------------------------------------------------ # search_zones # ------------------------------------------------------------------ async def search_zones(self, query: str = "") -> list[dict[str, Any]]: """Search or list zones for this account matching query fast.""" results: list[dict[str, Any]] = [] page = 1 per_page = 50 query_clean = query.strip().lower() async with httpx.AsyncClient(timeout=30, headers=self._headers()) as client: while True: params: dict[str, Any] = {"per_page": per_page, "page": page} resp = await client.get(f"{_BASE_URL}/zones", params=params) resp.raise_for_status() data = resp.json() self._raise_for_cf_errors(data, "search zones") result_info = data.get("result_info", {}) total_pages = result_info.get("total_pages", 1) for zone in data.get("result", []): zone_name = zone.get("name", "") zone_id = zone.get("id", "") if query_clean and query_clean not in zone_name.lower(): continue results.append( { "zone_id": zone_id, "domain_name": zone_name, "status": zone.get("status", "active"), "development_mode": "off", } ) page += 1 if page > total_pages or page > 5: break return results