"""Namecheap XML API integration. Credentials dict keys: api_user – Namecheap API username (same as account username) api_key – Namecheap API key from dashboard username – Namecheap account username (usually same as api_user) client_ip – Whitelisted IP address for API access """ from __future__ import annotations import logging import math import xml.etree.ElementTree as ET from datetime import date, datetime from typing import Any import httpx from integrations.base import RegistrarBase logger = logging.getLogger(__name__) _BASE_URL = "https://api.namecheap.com/xml.response" _NS = "http://api.namecheap.com/xml.response" # XML namespace def _ns(tag: str) -> str: """Wrap *tag* with the Namecheap XML namespace.""" return f"{{{_NS}}}{tag}" def _parse_date(value: str | None) -> date | None: """Parse an ISO-8601 or MM/DD/YYYY date string; return None on failure.""" if not value: return None for fmt in ("%m/%d/%Y", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%d"): try: return datetime.strptime(value.split()[0], fmt).date() except ValueError: continue return None class NamecheapIntegration(RegistrarBase): def __init__(self, credentials: dict[str, Any]) -> None: super().__init__(credentials) self._api_user: str = credentials["api_user"] self._api_key: str = credentials["api_key"] self._username: str = credentials.get("username", credentials["api_user"]) self._client_ip: str = credentials["client_ip"] def _base_params(self) -> dict[str, str]: return { "ApiUser": self._api_user, "ApiKey": self._api_key, "UserName": self._username, "ClientIp": self._client_ip, } def _check_errors(self, root: ET.Element) -> None: """Raise ValueError if the API response contains an error.""" errors = root.find(_ns("Errors")) if errors is not None: for error in errors: code = error.get("Number", "") msg = (error.text or "").strip() raise ValueError(f"Namecheap API error [{code}]: {msg}") status = root.get("Status", "") if status.upper() == "ERROR": raise ValueError("Namecheap API returned ERROR status") # ------------------------------------------------------------------ # list_domains # ------------------------------------------------------------------ async def list_domains(self) -> list[dict[str, Any]]: """Paginate through namecheap.domains.getList and return all domains.""" domains: list[dict[str, Any]] = [] page = 1 page_size = 100 async with httpx.AsyncClient(timeout=30) as client: while True: params = { **self._base_params(), "Command": "namecheap.domains.getList", "Page": str(page), "PageSize": str(page_size), "ListType": "ALL", } resp = await client.get(_BASE_URL, params=params) resp.raise_for_status() root = ET.fromstring(resp.text) self._check_errors(root) cmd_response = root.find(_ns("CommandResponse")) if cmd_response is None: break domain_list_result = cmd_response.find(_ns("DomainGetListResult")) if domain_list_result is None: break for domain_el in domain_list_result.findall(_ns("Domain")): name = domain_el.get("Name", "") expires = domain_el.get("Expires", "") is_auto_renew = domain_el.get("AutoRenew", "false").lower() == "true" is_locked = domain_el.get("IsLocked", "false").lower() == "true" is_expired = domain_el.get("IsExpired", "false").lower() == "true" if is_expired: status = "expired" elif is_locked: status = "locked" else: status = "active" domains.append( { "domain_name": name, "expiry_date": _parse_date(expires), "auto_renew": is_auto_renew, "status": status, "registrar_domain_id": domain_el.get("ID", None), } ) # Check pagination paging = cmd_response.find(_ns("Paging")) if paging is None: break total_items_el = paging.find(_ns("TotalItems")) if total_items_el is None or not total_items_el.text: break total_items = int(total_items_el.text) total_pages = math.ceil(total_items / page_size) if page >= total_pages: break page += 1 return domains # ------------------------------------------------------------------ # get_dns_records # ------------------------------------------------------------------ async def get_dns_records(self, domain_name: str) -> list[dict[str, Any]]: """Fetch all DNS host records for *domain_name* via namecheap.domains.dns.getHosts.""" parts = domain_name.rsplit(".", 2) if len(parts) >= 2: sld = parts[-2] tld = parts[-1] else: raise ValueError(f"Cannot split domain name: {domain_name!r}") params = { **self._base_params(), "Command": "namecheap.domains.dns.getHosts", "SLD": sld, "TLD": tld, } async with httpx.AsyncClient(timeout=30) as client: resp = await client.get(_BASE_URL, params=params) resp.raise_for_status() root = ET.fromstring(resp.text) self._check_errors(root) cmd_response = root.find(_ns("CommandResponse")) if cmd_response is None: return [] dns_hosts_result = cmd_response.find(_ns("DomainDNSGetHostsResult")) if dns_hosts_result is None: return [] records: list[dict[str, Any]] = [] for host in dns_hosts_result.findall(_ns("host")): record_type = host.get("Type", "") name = host.get("Name", "") value = host.get("Address", "") ttl_str = host.get("TTL", None) mx_pref_str = host.get("MXPref", None) records.append( { "record_type": record_type, "name": name, "value": value, "ttl": int(ttl_str) if ttl_str else None, "priority": int(mx_pref_str) if mx_pref_str else None, } ) return records # ------------------------------------------------------------------ # test_connection # ------------------------------------------------------------------ async def test_connection(self) -> bool: """Verify credentials by calling namecheap.users.getBalances.""" params = { **self._base_params(), "Command": "namecheap.users.getBalances", } try: async with httpx.AsyncClient(timeout=15) as client: resp = await client.get(_BASE_URL, params=params) resp.raise_for_status() root = ET.fromstring(resp.text) self._check_errors(root) return True except Exception as exc: logger.warning("Namecheap test_connection failed: %s", exc) return False