"""Cloudflare-backed cfmail subdomain provisioner.""" from __future__ import annotations from dataclasses import dataclass from datetime import datetime, timezone import json import os from pathlib import Path import secrets import subprocess import tempfile import time from typing import Any from curl_cffi import requests as cffi_requests from .cfmail import DEFAULT_CFMAIL_CONFIG_PATH, load_cfmail_accounts_from_file MX_RECORDS = ( ("route1.mx.cloudflare.net", 20), ("route2.mx.cloudflare.net", 85), ("route3.mx.cloudflare.net", 36), ) def _utc_stamp() -> str: return datetime.now(timezone.utc).strftime("%m%d%H%M%S") def _normalize_host(value: str) -> str: candidate = str(value or "").strip() if candidate.startswith("https://"): candidate = candidate[len("https://") :] elif candidate.startswith("http://"): candidate = candidate[len("http://") :] return candidate.strip().strip("/") @dataclass(frozen=True) class ProvisioningSettings: auth_email: str auth_key: str account_id: str zone_id: str worker_name: str zone_name: str @classmethod def from_env(cls) -> "ProvisioningSettings": return cls( auth_email=str(os.getenv("ZHUCE6_CFMAIL_CF_AUTH_EMAIL", "")).strip(), auth_key=str(os.getenv("ZHUCE6_CFMAIL_CF_AUTH_KEY", "")).strip(), account_id=str(os.getenv("ZHUCE6_CFMAIL_CF_ACCOUNT_ID", "")).strip(), zone_id=str(os.getenv("ZHUCE6_CFMAIL_CF_ZONE_ID", "")).strip(), worker_name=str(os.getenv("ZHUCE6_CFMAIL_WORKER_NAME", "")).strip(), zone_name=str(os.getenv("ZHUCE6_CFMAIL_ZONE_NAME", "")).strip(), ) def validate(self) -> None: missing = [ name for name, value in ( ("ZHUCE6_CFMAIL_CF_AUTH_EMAIL", self.auth_email), ("ZHUCE6_CFMAIL_CF_AUTH_KEY", self.auth_key), ("ZHUCE6_CFMAIL_CF_ACCOUNT_ID", self.account_id), ("ZHUCE6_CFMAIL_CF_ZONE_ID", self.zone_id), ("ZHUCE6_CFMAIL_WORKER_NAME", self.worker_name), ("ZHUCE6_CFMAIL_ZONE_NAME", self.zone_name), ) if not value ] if missing: raise RuntimeError(f"missing cfmail provisioning env: {', '.join(missing)}") @dataclass(frozen=True) class ProvisionResult: success: bool step: str old_domain: str = "" new_domain: str = "" error: str = "" class CfmailProvisioner: def __init__( self, *, config_path: str | Path | None = None, proxy_url: str | None = None, settings: ProvisioningSettings | None = None, ) -> None: self.config_path = Path(config_path or DEFAULT_CFMAIL_CONFIG_PATH) self.settings = settings or ProvisioningSettings.from_env() self.proxies = {"http": proxy_url, "https": proxy_url} if proxy_url else None def _headers(self) -> dict[str, str]: self.settings.validate() return { "X-Auth-Email": self.settings.auth_email, "X-Auth-Key": self.settings.auth_key, } def _request( self, method: str, url: str, *, json_body: dict[str, Any] | None = None, ) -> dict[str, Any]: response = cffi_requests.request( method.upper(), url, headers={ **self._headers(), "Content-Type": "application/json", }, json=json_body, proxies=self.proxies, timeout=30, impersonate="chrome", ) data = response.json() if response.content else {} if response.status_code >= 400 or not data.get("success", False): raise RuntimeError(f"{method.upper()} {url} failed: HTTP {response.status_code} {data}") return data def _request_paginated(self, url: str) -> list[dict[str, Any]]: page = 1 results: list[dict[str, Any]] = [] while True: separator = "&" if "?" in url else "?" payload = self._request("GET", f"{url}{separator}page={page}&per_page=100") items = payload.get("result") or [] if isinstance(items, list): results.extend(item for item in items if isinstance(item, dict)) info = payload.get("result_info") or {} total_pages = int(info.get("total_pages") or 1) if page >= total_pages: break page += 1 return results def _get_worker_settings(self) -> dict[str, Any]: url = ( f"https://api.cloudflare.com/client/v4/accounts/{self.settings.account_id}/workers/scripts/" f"{self.settings.worker_name}/settings" ) return self._request("GET", url).get("result") or {} def _patch_worker_settings(self, bindings: list[dict[str, Any]]) -> None: url = ( f"https://api.cloudflare.com/client/v4/accounts/{self.settings.account_id}/workers/scripts/" f"{self.settings.worker_name}/settings" ) self.settings.validate() body = json.dumps({"bindings": bindings}, ensure_ascii=False, separators=(",", ":")) boundary = secrets.token_hex(16) multipart_body = ( f"--{boundary}\r\n" f'Content-Disposition: form-data; name="settings"\r\n' f"Content-Type: application/json\r\n" f"\r\n" f"{body}\r\n" f"--{boundary}--\r\n" ).encode("utf-8") last_error = "" for attempt in range(3): try: response = cffi_requests.patch( url, headers={ "X-Auth-Email": self.settings.auth_email, "X-Auth-Key": self.settings.auth_key, "Content-Type": f"multipart/form-data; boundary={boundary}", }, data=multipart_body, proxies=self.proxies, timeout=30, impersonate="chrome", ) data = response.json() if response.content else {} if response.status_code >= 400 or not data.get("success", False): last_error = f"HTTP {response.status_code} {data}" if attempt < 2: time.sleep(2) continue raise RuntimeError(f"PATCH worker settings failed: {last_error}") return except RuntimeError: raise except Exception as exc: last_error = str(exc) if attempt < 2: time.sleep(2) continue raise RuntimeError(f"PATCH worker settings failed after {attempt + 1} attempts: {last_error}") def _make_new_label(self) -> str: return f"auto{_utc_stamp()}{secrets.token_hex(2)}" def _new_domain(self, label: str) -> str: return f"{label}.{self.settings.zone_name}" def _create_email_routing_rule(self, domain: str, label: str) -> None: url = f"https://api.cloudflare.com/client/v4/zones/{self.settings.zone_id}/email/routing/rules" self._request( "POST", url, json_body={ "name": f"{label} subdomain catch-all", "enabled": True, "matchers": [{"type": "literal", "field": "to", "value": f"*@" + domain}], "actions": [{"type": "worker", "value": [self.settings.worker_name]}], }, ) def _create_dns_records(self, domain: str) -> None: url = f"https://api.cloudflare.com/client/v4/zones/{self.settings.zone_id}/dns_records" for content, priority in MX_RECORDS: self._request( "POST", url, json_body={ "type": "MX", "name": domain, "content": content, "priority": priority, "ttl": 1, }, ) self._request( "POST", url, json_body={ "type": "TXT", "name": domain, "content": "v=spf1 include:_spf.mx.cloudflare.net ~all", "ttl": 1, }, ) def _list_dns_records(self) -> list[dict[str, Any]]: url = f"https://api.cloudflare.com/client/v4/zones/{self.settings.zone_id}/dns_records" return self._request_paginated(url) def _delete_dns_record(self, record_id: str) -> None: url = f"https://api.cloudflare.com/client/v4/zones/{self.settings.zone_id}/dns_records/{record_id}" self._request("DELETE", url) def _list_email_routing_rules(self) -> list[dict[str, Any]]: url = f"https://api.cloudflare.com/client/v4/zones/{self.settings.zone_id}/email/routing/rules" return self._request_paginated(url) def _delete_email_routing_rule(self, rule_id: str) -> None: url = f"https://api.cloudflare.com/client/v4/zones/{self.settings.zone_id}/email/routing/rules/{rule_id}" self._request("DELETE", url) def _routing_rule_domains(self, rule: dict[str, Any]) -> set[str]: domains: set[str] = set() for matcher in rule.get("matchers") or []: if not isinstance(matcher, dict): continue value = str(matcher.get("value") or "").strip().lower() if "*@" in value: domains.add(value.split("*@", 1)[-1]) return domains def _normalize_domain_name(self, value: str) -> str: return str(value or "").strip().lower().rstrip(".") def _is_managed_auto_domain(self, domain: str) -> bool: domain_key = self._normalize_domain_name(domain) zone_suffix = f".{self.settings.zone_name.lower()}" return bool(domain_key) and domain_key.startswith("auto") and domain_key.endswith(zone_suffix) def _delete_domain_artifacts(self, domain: str) -> None: domain_key = str(domain or "").strip().lower() if not domain_key: return for record in self._list_dns_records(): if str(record.get("name") or "").strip().lower() == domain_key: record_id = str(record.get("id") or "").strip() if record_id: try: self._delete_dns_record(record_id) except Exception: pass # skip read-only or protected records for rule in self._list_email_routing_rules(): if domain_key in self._routing_rule_domains(rule): rule_id = str(rule.get("id") or "").strip() if rule_id: try: self._delete_email_routing_rule(rule_id) except Exception: pass def _managed_auto_domains(self, accounts: list[dict[str, Any]]) -> list[str]: return [ self._normalize_domain_name(str(item.get("email_domain") or "")) for item in accounts if self._is_managed_auto_domain(str(item.get("email_domain") or "")) ] def current_active_accounts(self) -> list[dict[str, Any]]: accounts = self._load_all_accounts() return [ dict(item) for item in accounts if bool(item.get("enabled", True)) and str(item.get("email_domain") or "").strip() ] def current_active_domains(self) -> list[str]: return [ self._normalize_domain_name(str(item.get("email_domain") or "")) for item in self.current_active_accounts() if self._normalize_domain_name(str(item.get("email_domain") or "")) ] def cleanup_stale_domains(self, keep_domains: set[str] | list[str] | None = None) -> dict[str, Any]: return self.cleanup_stale_cf_resources(keep_domains=keep_domains) def cleanup_stale_cf_resources(self, keep_domains: set[str] | list[str] | None = None) -> dict[str, Any]: accounts = self._load_all_accounts() active_domain = self._normalize_domain_name(str(self.current_active_account().get("email_domain") or "")) keep_set = { self._normalize_domain_name(domain) for domain in (keep_domains or []) if self._normalize_domain_name(domain) } keep_set.discard(active_domain) stale_domains: set[str] = set() removed_dns_records: list[str] = [] removed_routing_rules: list[str] = [] errors: list[str] = [] for rule in self._list_email_routing_rules(): rule_domains = { domain for domain in self._routing_rule_domains(rule) if self._is_managed_auto_domain(domain) and domain != active_domain and domain not in keep_set } stale_domains.update(rule_domains) if not rule_domains: continue rule_id = str(rule.get("id") or "").strip() if not rule_id: continue try: self._delete_email_routing_rule(rule_id) removed_routing_rules.append(rule_id) except Exception as exc: errors.append(f"routing_rule:{rule_id}: {exc}") for record in self._list_dns_records(): record_type = str(record.get("type") or "").strip().upper() if record_type not in {"MX", "TXT"}: continue domain = self._normalize_domain_name(str(record.get("name") or "")) if not self._is_managed_auto_domain(domain) or domain == active_domain or domain in keep_set: continue stale_domains.add(domain) record_id = str(record.get("id") or "").strip() if not record_id: continue try: self._delete_dns_record(record_id) removed_dns_records.append(record_id) except Exception as exc: errors.append(f"dns_record:{record_id}: {exc}") stale_account_domains = set(self._managed_auto_domains(accounts)).intersection(stale_domains) if stale_account_domains: pruned_accounts = [ item for item in accounts if self._normalize_domain_name(str(item.get("email_domain") or "")) not in stale_account_domains ] if len(pruned_accounts) != len(accounts): self._write_accounts(pruned_accounts) return { "removed_domains": sorted(stale_domains), "removed_dns_records": removed_dns_records, "removed_routing_rules": removed_routing_rules, "errors": errors, } def _is_record_quota_error(self, exc: Exception) -> bool: message = str(exc or "") return "81045" in message or "Record quota exceeded" in message def _set_worker_domains(self, domains: list[str]) -> None: settings = self._get_worker_settings() bindings = list(settings.get("bindings") or []) normalized_domains: list[str] = [] seen_domains: set[str] = set() for domain in domains: domain_key = self._normalize_domain_name(domain) if not domain_key or domain_key in seen_domains: continue normalized_domains.append(domain_key) seen_domains.add(domain_key) updated = False for binding in bindings: if binding.get("name") not in {"DOMAINS", "DEFAULT_DOMAINS"} or binding.get("type") != "json": continue binding["json"] = list(normalized_domains) updated = True if not updated: raise RuntimeError("worker DOMAINS bindings missing") self._patch_worker_settings(bindings) def _update_worker_domains(self, domain: str | list[str], old_domain: str | None = None) -> None: if isinstance(domain, list): domains = list(domain) else: domains = [domain] old_domain_key = self._normalize_domain_name(old_domain or "") if old_domain_key and old_domain_key != self._normalize_domain_name(domain): domains.append(old_domain_key) self._set_worker_domains(domains) def smoke_test(self, worker_domain: str, admin_password: str, email_domain: str) -> None: test_name = f"smoke{secrets.token_hex(3)}" required_successes = 3 success_streak = 0 last_error = "smoke test did not run" for attempt in range(1, 10): response = cffi_requests.post( f"https://{_normalize_host(worker_domain)}/admin/new_address", headers={ "x-admin-auth": admin_password, "Content-Type": "application/json", }, json={"enablePrefix": True, "name": f"{test_name}{attempt}", "domain": email_domain}, proxies=self.proxies, timeout=20, impersonate="chrome", ) if response.status_code != 200: success_streak = 0 last_error = f"HTTP {response.status_code} {response.text[:240]}" time.sleep(min(8, attempt * 2)) continue try: data = response.json() if response.content else {} except Exception: success_streak = 0 last_error = f"non-json response: {response.text[:240]}" time.sleep(min(8, attempt * 2)) continue if str(data.get("address") or "").strip() and str(data.get("jwt") or "").strip(): success_streak += 1 if success_streak >= required_successes: return last_error = f"smoke success streak={success_streak}/{required_successes}" time.sleep(1) continue success_streak = 0 last_error = f"incomplete payload: {json.dumps(data, ensure_ascii=False)[:240]}" time.sleep(min(8, attempt * 2)) raise RuntimeError(f"smoke test failed: {last_error}") def _load_all_accounts(self) -> list[dict[str, Any]]: data = load_cfmail_accounts_from_file(self.config_path, silent=False) return [item for item in data if isinstance(item, dict)] def _write_accounts(self, accounts: list[dict[str, Any]]) -> None: payload = {"accounts": accounts} tmp_path = self.config_path.with_suffix(self.config_path.suffix + ".tmp") tmp_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") tmp_path.replace(self.config_path) def _pick_active_domain(self, accounts: list[dict[str, Any]]) -> str: for item in reversed(accounts): domain = str(item.get("email_domain") or "").strip().lower() if domain and item.get("enabled", True): return domain return "" def normalize_accounts_to_single_active_domain(self) -> dict[str, Any]: accounts = self._load_all_accounts() active_domain = self._pick_active_domain(accounts) if not active_domain: return {"active_domain": "", "removed_domains": []} managed_domains = set(self._managed_auto_domains(accounts)) previous_managed_domain = "" for item in reversed(accounts): domain = str(item.get("email_domain") or "").strip().lower() if not domain or domain == active_domain or domain not in managed_domains: continue previous_managed_domain = domain break normalized_accounts: list[dict[str, Any]] = [] removed_domains: list[str] = [] changed = False for item in accounts: domain = str(item.get("email_domain") or "").strip().lower() if domain == active_domain: if item.get("enabled") is not True: changed = True item["enabled"] = True normalized_accounts.append(item) continue if domain == previous_managed_domain: if item.get("enabled") is not False: changed = True item["enabled"] = False normalized_accounts.append(item) continue if domain in managed_domains: removed_domains.append(domain) changed = True continue if item.get("enabled") is not False: changed = True item["enabled"] = False normalized_accounts.append(item) if changed or len(normalized_accounts) != len(accounts): self._write_accounts(normalized_accounts) for domain in removed_domains: try: self._delete_domain_artifacts(domain) except Exception: pass return {"active_domain": active_domain, "removed_domains": removed_domains} def provision_additional_domain(self) -> ProvisionResult: current = self.current_active_account() worker_domain = str(current.get("worker_domain") or "").strip() admin_password = str(current.get("admin_password") or "").strip() if not worker_domain or not admin_password: return ProvisionResult(success=False, step="load_active_account", error="active cfmail account incomplete") label = self._make_new_label() new_domain = self._new_domain(label) try: self._create_email_routing_rule(new_domain, label) self._create_dns_records(new_domain) existing_domains = self.current_active_domains() self._set_worker_domains([*existing_domains, new_domain]) self.smoke_test(worker_domain, admin_password, new_domain) accounts = self._load_all_accounts() accounts.append( { "name": f"cfmail-{new_domain.split('.', 1)[0]}", "worker_domain": _normalize_host(worker_domain), "email_domain": new_domain, "admin_password": admin_password, "enabled": True, } ) self._write_accounts(accounts) return ProvisionResult(success=True, step="provision_additional_domain", new_domain=new_domain) except Exception as exc: try: self._delete_domain_artifacts(new_domain) except Exception: pass return ProvisionResult( success=False, step="provision_additional_domain", new_domain=new_domain, error=str(exc), ) def retire_domain(self, domain: str) -> ProvisionResult: domain_key = self._normalize_domain_name(domain) if not domain_key: return ProvisionResult(success=False, step="retire_domain", error="missing domain") accounts = self._load_all_accounts() active_before = self.current_active_domains() matched = False updated_accounts: list[dict[str, Any]] = [] for item in accounts: item_domain = self._normalize_domain_name(str(item.get("email_domain") or "")) if item_domain != domain_key: updated_accounts.append(item) continue matched = True if self._is_managed_auto_domain(domain_key): continue item["enabled"] = False updated_accounts.append(item) if not matched: return ProvisionResult(success=False, step="retire_domain", old_domain=domain_key, error="domain not found") self._write_accounts(updated_accounts) active_after = [ self._normalize_domain_name(str(item.get("email_domain") or "")) for item in updated_accounts if bool(item.get("enabled", True)) ] if active_after: self._set_worker_domains(active_after) elif active_before: self._set_worker_domains([d for d in active_before if d != domain_key]) if self._is_managed_auto_domain(domain_key): try: self._delete_domain_artifacts(domain_key) except Exception: pass return ProvisionResult(success=True, step="retire_domain", old_domain=domain_key) def normalize_to_domain_pool(self, target_count: int) -> dict[str, Any]: desired = max(1, int(target_count)) accounts = self._load_all_accounts() enabled_accounts = [ dict(item) for item in accounts if bool(item.get("enabled", True)) and str(item.get("email_domain") or "").strip() ] changed = False provisioned_domains: list[str] = [] retired_domains: list[str] = [] if not enabled_accounts: latest_index = -1 for idx in range(len(accounts) - 1, -1, -1): domain = self._normalize_domain_name(str(accounts[idx].get("email_domain") or "")) if domain: latest_index = idx break if latest_index >= 0: accounts[latest_index]["enabled"] = True enabled_accounts = [dict(accounts[latest_index])] changed = True self._write_accounts(accounts) if len(enabled_accounts) > desired: keep = enabled_accounts[-desired:] keep_domains = { self._normalize_domain_name(str(item.get("email_domain") or "")) for item in keep } for item in enabled_accounts[:-desired]: domain = self._normalize_domain_name(str(item.get("email_domain") or "")) if not domain: continue result = self.retire_domain(domain) if result.success: retired_domains.append(domain) enabled_accounts = self.current_active_accounts() changed = True while len(enabled_accounts) < desired: result = self.provision_additional_domain() if not result.success: break provisioned_domains.append(result.new_domain) enabled_accounts = self.current_active_accounts() changed = True active_domains = [ self._normalize_domain_name(str(item.get("email_domain") or "")) for item in enabled_accounts if self._normalize_domain_name(str(item.get("email_domain") or "")) ] if active_domains: self._set_worker_domains(active_domains) return { "active_domains": active_domains, "provisioned_domains": provisioned_domains, "retired_domains": retired_domains, "changed": changed, } def switch_active_domain(self, *, old_domain: str, new_domain: str, worker_domain: str, admin_password: str) -> list[str]: accounts = self._load_all_accounts() managed_domains = set(self._managed_auto_domains(accounts)) old_domain_key = str(old_domain or "").strip().lower() replacement = { "name": f"cfmail-{new_domain.split('.', 1)[0]}", "worker_domain": _normalize_host(worker_domain), "email_domain": new_domain, "admin_password": admin_password, "enabled": True, } normalized_accounts: list[dict[str, Any]] = [] removed_domains: list[str] = [] matched = False for item in accounts: domain = str(item.get("email_domain") or "").strip().lower() if domain == new_domain.lower(): item.update(replacement) item["enabled"] = True normalized_accounts.append(item) matched = True continue if domain == old_domain_key: item["enabled"] = False normalized_accounts.append(item) continue if domain in managed_domains: removed_domains.append(domain) continue item["enabled"] = False normalized_accounts.append(item) if not matched: normalized_accounts.append(replacement) self._write_accounts(normalized_accounts) return sorted(set(domain for domain in removed_domains if domain and domain != new_domain.lower())) def current_active_account(self) -> dict[str, Any]: accounts = self._load_all_accounts() active_domain = self._pick_active_domain(accounts) for item in reversed(accounts): if str(item.get("email_domain") or "").strip().lower() == active_domain: return item raise RuntimeError("no active cfmail account found") def rotate_active_domain(self) -> ProvisionResult: current = self.current_active_account() old_domain = str(current.get("email_domain") or "").strip().lower() worker_domain = str(current.get("worker_domain") or "").strip() admin_password = str(current.get("admin_password") or "").strip() if not old_domain or not worker_domain or not admin_password: return ProvisionResult(success=False, step="load_active_account", error="active cfmail account incomplete") last_error = "" last_new_domain = "" for attempt in range(2): label = self._make_new_label() new_domain = self._new_domain(label) last_new_domain = new_domain try: self._create_email_routing_rule(new_domain, label) self._create_dns_records(new_domain) self._update_worker_domains(new_domain, old_domain=old_domain) self.smoke_test(worker_domain, admin_password, new_domain) self.switch_active_domain( old_domain=old_domain, new_domain=new_domain, worker_domain=worker_domain, admin_password=admin_password, ) try: self.cleanup_stale_cf_resources(keep_domains={old_domain}) except Exception: pass # cleanup is best-effort, must not abort rotation return ProvisionResult( success=True, step="completed", old_domain=old_domain, new_domain=new_domain, ) except Exception as exc: last_error = str(exc) try: self._delete_domain_artifacts(new_domain) except Exception: pass if attempt == 0 and self._is_record_quota_error(exc): cleanup_result = self.cleanup_stale_cf_resources() if ( cleanup_result.get("removed_domains") or cleanup_result.get("removed_dns_records") or cleanup_result.get("removed_routing_rules") ): continue break return ProvisionResult( success=False, step="failed", old_domain=old_domain, new_domain=last_new_domain, error=last_error, )