Usare le API di Cloudflare per la gestione DNS con Python
Cloudflare espone quasi tutte le funzionalità del proprio pannello di controllo tramite una API REST (la versione 4, raggiungibile all'indirizzo https://api.cloudflare.com/client/v4). La gestione dei record DNS è uno dei casi d'uso più comuni: aggiornare un record A quando cambia l'IP pubblico di un server domestico, creare in automatico i sottodomini di un nuovo ambiente di staging, mantenere sotto controllo di versione la configurazione DNS di decine di zone.
In questo articolo vedremo come interagire con queste API usando Python, partendo dalle richieste HTTP di base fino ad arrivare a un piccolo client riutilizzabile, a uno script di DNS dinamico e a una sincronizzazione dichiarativa dei record a partire da un file JSON.
Prerequisiti e autenticazione
Per seguire gli esempi servono Python 3.10 o superiore e la libreria requests:
python3 -m venv .venv
source .venv/bin/activate
pip install requestsCloudflare supporta due metodi di autenticazione: la vecchia Global API Key (associata all'account, con accesso completo) e gli API Token, che permettono di limitare i permessi a specifiche risorse. Il secondo metodo è quello da preferire sempre: un token con il solo permesso Zone → DNS → Edit su una singola zona riduce drasticamente i danni in caso di compromissione.
Il token si crea dal pannello di Cloudflare nella sezione My Profile → API Tokens → Create Token, partendo dal modello Edit zone DNS. Una volta generato, conviene esporlo al codice tramite una variabile d'ambiente, mai scriverlo nel sorgente:
export CLOUDFLARE_API_TOKEN="il-tuo-token"Il token va inviato in ogni richiesta nell'header Authorization con lo schema Bearer. Per verificare che sia valido esiste un endpoint dedicato:
import os
import requests
API_BASE = "https://api.cloudflare.com/client/v4"
token = os.environ["CLOUDFLARE_API_TOKEN"]
response = requests.get(
f"{API_BASE}/user/tokens/verify",
headers={"Authorization": f"Bearer {token}"},
timeout=10,
)
data = response.json()
# Lo stato atteso per un token utilizzabile è "active"
print(data["success"], data["result"]["status"])La struttura delle risposte
Tutte le risposte della API v4 condividono lo stesso involucro JSON, il che rende semplice gestirle in modo uniforme:
{
"success": true,
"errors": [],
"messages": [],
"result": [ ... ],
"result_info": {
"page": 1,
"per_page": 100,
"count": 12,
"total_count": 12,
"total_pages": 1
}
}Il campo success indica l'esito dell'operazione; in caso di errore l'array errors contiene oggetti con un code numerico e un message descrittivo. Il campo result contiene il dato richiesto (un oggetto o un array), mentre result_info è presente solo negli endpoint paginati.
Individuare l'ID della zona
Gli endpoint DNS non lavorano con il nome del dominio ma con l'identificativo della zona (zone ID). Lo si può copiare dal pannello, ma è più comodo ricavarlo via API filtrando per nome:
def get_zone_id(token: str, zone_name: str) -> str:
response = requests.get(
f"{API_BASE}/zones",
headers={"Authorization": f"Bearer {token}"},
params={"name": zone_name},
timeout=10,
)
response.raise_for_status()
zones = response.json()["result"]
if not zones:
raise LookupError(f"Zona non trovata: {zone_name}")
return zones[0]["id"]
zone_id = get_zone_id(token, "example.com")
print(zone_id)Le operazioni CRUD sui record DNS
Gli endpoint per i record DNS seguono uno schema REST classico, tutti sotto il percorso /zones/{zone_id}/dns_records:
GET /zones/{zone_id}/dns_records: elenco dei record, con filtri e paginazione;GET /zones/{zone_id}/dns_records/{record_id}: dettaglio di un singolo record;POST /zones/{zone_id}/dns_records: creazione di un record;PATCH /zones/{zone_id}/dns_records/{record_id}: modifica parziale;PUT /zones/{zone_id}/dns_records/{record_id}: sostituzione completa;DELETE /zones/{zone_id}/dns_records/{record_id}: eliminazione.
Un record è descritto da pochi campi principali: type (A, AAAA, CNAME, MX, TXT, CAA, SRV e altri), name (il nome completo, ad esempio www.example.com), content (il valore), ttl (in secondi, dove il valore 1 significa "automatico") e proxied, che indica se il traffico deve passare attraverso il proxy di Cloudflare (la cosiddetta "nuvola arancione"). Sono disponibili anche i campi facoltativi comment e tags, molto utili per documentare l'origine di un record creato da uno script. I record MX e SRV richiedono inoltre il campo priority.
Elencare i record
headers = {"Authorization": f"Bearer {token}"}
response = requests.get(
f"{API_BASE}/zones/{zone_id}/dns_records",
headers=headers,
params={"type": "A", "per_page": 100},
timeout=10,
)
response.raise_for_status()
for record in response.json()["result"]:
print(f'{record["type"]:6} {record["name"]:30} {record["content"]:20} proxied={record["proxied"]}')I parametri di query type, name e content permettono di filtrare i risultati lato server, evitando di scaricare l'intera zona per cercare un singolo record.
Creare un record
payload = {
"type": "A",
"name": "staging.example.com",
"content": "203.0.113.10",
"ttl": 1,
"proxied": True,
"comment": "Creato da script Python",
}
response = requests.post(
f"{API_BASE}/zones/{zone_id}/dns_records",
headers=headers,
json=payload,
timeout=10,
)
data = response.json()
if not data["success"]:
# Un record identico già esistente produce un errore specifico
for error in data["errors"]:
print(error["code"], error["message"])
else:
print("Creato record con ID", data["result"]["id"])Modificare ed eliminare un record
Con PATCH è sufficiente inviare solo i campi da modificare; con PUT occorre invece inviare la rappresentazione completa del record:
record_id = data["result"]["id"]
# Modifica parziale: cambiamo solo l'indirizzo IP
requests.patch(
f"{API_BASE}/zones/{zone_id}/dns_records/{record_id}",
headers=headers,
json={"content": "203.0.113.20"},
timeout=10,
).raise_for_status()
# Eliminazione
requests.delete(
f"{API_BASE}/zones/{zone_id}/dns_records/{record_id}",
headers=headers,
timeout=10,
).raise_for_status()Un client riutilizzabile
Ripetere headers, timeout e controllo degli errori in ogni chiamata diventa presto scomodo. Conviene incapsulare tutto in una classe che usi una requests.Session (per riutilizzare le connessioni TCP), gestisca in modo uniforme gli errori, segua la paginazione in automatico e rispetti i limiti di frequenza.
Cloudflare applica un limite globale di richieste per utente (nell'ordine di 1200 richieste ogni cinque minuti); superato il limite, la API risponde con lo stato 429. Il client che segue ritenta automaticamente in questi casi, rispettando l'header Retry-After quando presente.
# cloudflare_dns.py
from __future__ import annotations
import os
import time
from dataclasses import dataclass, field
from typing import Any, Iterator
import requests
API_BASE = "https://api.cloudflare.com/client/v4"
class CloudflareError(Exception):
"""Errore restituito dalla API di Cloudflare."""
def __init__(self, status: int, errors: list[dict[str, Any]]):
self.status = status
self.errors = errors
details = "; ".join(f'[{e.get("code")}] {e.get("message")}' for e in errors)
super().__init__(f"HTTP {status}: {details or 'errore sconosciuto'}")
@dataclass
class DnsRecord:
type: str
name: str
content: str
ttl: int = 1
proxied: bool = False
priority: int | None = None
comment: str | None = None
id: str | None = field(default=None, compare=False)
@classmethod
def from_api(cls, data: dict[str, Any]) -> "DnsRecord":
return cls(
type=data["type"],
name=data["name"],
content=data["content"],
ttl=data.get("ttl", 1),
proxied=data.get("proxied", False),
priority=data.get("priority"),
comment=data.get("comment"),
id=data.get("id"),
)
def to_payload(self) -> dict[str, Any]:
payload: dict[str, Any] = {
"type": self.type,
"name": self.name,
"content": self.content,
"ttl": self.ttl,
}
# Solo alcuni tipi di record possono passare dal proxy di Cloudflare
if self.type in {"A", "AAAA", "CNAME"}:
payload["proxied"] = self.proxied
if self.priority is not None:
payload["priority"] = self.priority
if self.comment:
payload["comment"] = self.comment
return payload
class CloudflareDNS:
def __init__(self, token: str | None = None, max_retries: int = 5, timeout: float = 15.0):
token = token or os.environ.get("CLOUDFLARE_API_TOKEN")
if not token:
raise ValueError("Token API mancante: impostare CLOUDFLARE_API_TOKEN")
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
})
self.max_retries = max_retries
self.timeout = timeout
# --- Livello di trasporto -------------------------------------------
def _request(self, method: str, path: str, **kwargs: Any) -> dict[str, Any]:
url = f"{API_BASE}{path}"
for attempt in range(self.max_retries + 1):
response = self.session.request(method, url, timeout=self.timeout, **kwargs)
# Limite di frequenza o errore temporaneo lato server: attesa e nuovo tentativo
if response.status_code in (429, 502, 503, 504) and attempt < self.max_retries:
retry_after = response.headers.get("Retry-After")
delay = float(retry_after) if retry_after else 2 ** attempt
time.sleep(delay)
continue
try:
data = response.json()
except ValueError:
raise CloudflareError(response.status_code, [{"message": response.text[:200]}])
if not response.ok or not data.get("success", False):
raise CloudflareError(response.status_code, data.get("errors", []))
return data
raise CloudflareError(429, [{"message": "Numero massimo di tentativi superato"}])
def _paginate(self, path: str, params: dict[str, Any] | None = None) -> Iterator[dict[str, Any]]:
params = dict(params or {})
params.setdefault("per_page", 100)
page = 1
while True:
params["page"] = page
data = self._request("GET", path, params=params)
yield from data["result"]
info = data.get("result_info") or {}
if page >= info.get("total_pages", 1):
break
page += 1
# --- Zone -----------------------------------------------------------
def zone_id(self, zone_name: str) -> str:
data = self._request("GET", "/zones", params={"name": zone_name})
if not data["result"]:
raise LookupError(f"Zona non trovata: {zone_name}")
return data["result"][0]["id"]
# --- Record DNS -----------------------------------------------------
def list_records(self, zone_id: str, **filters: Any) -> list[DnsRecord]:
path = f"/zones/{zone_id}/dns_records"
return [DnsRecord.from_api(r) for r in self._paginate(path, filters)]
def find_record(self, zone_id: str, record_type: str, name: str) -> DnsRecord | None:
records = self.list_records(zone_id, type=record_type, name=name)
return records[0] if records else None
def create_record(self, zone_id: str, record: DnsRecord) -> DnsRecord:
data = self._request("POST", f"/zones/{zone_id}/dns_records", json=record.to_payload())
return DnsRecord.from_api(data["result"])
def update_record(self, zone_id: str, record_id: str, **changes: Any) -> DnsRecord:
data = self._request("PATCH", f"/zones/{zone_id}/dns_records/{record_id}", json=changes)
return DnsRecord.from_api(data["result"])
def replace_record(self, zone_id: str, record_id: str, record: DnsRecord) -> DnsRecord:
data = self._request("PUT", f"/zones/{zone_id}/dns_records/{record_id}", json=record.to_payload())
return DnsRecord.from_api(data["result"])
def delete_record(self, zone_id: str, record_id: str) -> None:
self._request("DELETE", f"/zones/{zone_id}/dns_records/{record_id}")
def upsert_record(self, zone_id: str, record: DnsRecord) -> tuple[str, DnsRecord]:
"""Crea il record se non esiste, altrimenti lo aggiorna solo se è cambiato."""
existing = self.find_record(zone_id, record.type, record.name)
if existing is None:
return "created", self.create_record(zone_id, record)
if existing == record:
return "unchanged", existing
return "updated", self.replace_record(zone_id, existing.id, record)
# --- Import / export in formato BIND --------------------------------
def export_zone(self, zone_id: str) -> str:
response = self.session.get(
f"{API_BASE}/zones/{zone_id}/dns_records/export", timeout=self.timeout
)
response.raise_for_status()
return response.textAlcune osservazioni sul codice:
- la
dataclassDnsRecordesclude il campoiddal confronto (compare=False), così due record con gli stessi valori risultano uguali indipendentemente dall'identificativo: è ciò che permette aupsert_record()di evitare aggiornamenti inutili; - il campo
proxiedviene inviato solo per i tipiA,AAAAeCNAME, gli unici che Cloudflare può mettere dietro il proprio proxy; - il metodo
_paginate()è un generatore: i record vengono prodotti man mano che le pagine arrivano, senza dover conoscere in anticipo il numero totale; - l'endpoint di esportazione restituisce testo in formato zone file BIND e non JSON, per cui viene gestito a parte senza passare da
_request().
Un esempio d'uso:
from cloudflare_dns import CloudflareDNS, DnsRecord
cf = CloudflareDNS()
zone_id = cf.zone_id("example.com")
action, record = cf.upsert_record(
zone_id,
DnsRecord(type="CNAME", name="docs.example.com", content="example.github.io", proxied=True),
)
print(action, record.name, "->", record.content)
# Backup della zona in formato BIND
with open("example.com.zone", "w", encoding="utf-8") as fh:
fh.write(cf.export_zone(zone_id))Caso pratico: DNS dinamico per un server domestico
Chi ospita servizi su una connessione con IP pubblico non statico ha bisogno di aggiornare il record DNS ogni volta che l'indirizzo cambia. Con il client appena scritto, uno script di DNS dinamico richiede poche righe:
#!/usr/bin/env python3
# ddns.py - aggiorna un record A con l'IP pubblico corrente
import logging
import sys
import requests
from cloudflare_dns import CloudflareDNS, DnsRecord, CloudflareError
ZONE = "example.com"
HOSTNAME = "home.example.com"
IP_SERVICES = [
"https://api.ipify.org",
"https://ifconfig.me/ip",
"https://icanhazip.com",
]
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("ddns")
def current_public_ip() -> str:
# Si provano più servizi in sequenza per non dipendere da uno solo
for url in IP_SERVICES:
try:
response = requests.get(url, timeout=5)
response.raise_for_status()
return response.text.strip()
except requests.RequestException as exc:
log.warning("Servizio %s non disponibile: %s", url, exc)
raise RuntimeError("Impossibile determinare l'IP pubblico")
def main() -> int:
try:
ip = current_public_ip()
cf = CloudflareDNS()
zone_id = cf.zone_id(ZONE)
existing = cf.find_record(zone_id, "A", HOSTNAME)
if existing and existing.content == ip:
log.info("Nessuna modifica: %s punta già a %s", HOSTNAME, ip)
return 0
record = DnsRecord(
type="A",
name=HOSTNAME,
content=ip,
ttl=300,
proxied=False,
comment="Gestito da ddns.py",
)
action, _ = cf.upsert_record(zone_id, record)
log.info("Record %s: %s -> %s", action, HOSTNAME, ip)
return 0
except (CloudflareError, RuntimeError, LookupError) as exc:
log.error("Aggiornamento fallito: %s", exc)
return 1
if __name__ == "__main__":
sys.exit(main())Lo script può essere eseguito periodicamente con cron, ad esempio ogni cinque minuti:
*/5 * * * * CLOUDFLARE_API_TOKEN="il-tuo-token" /opt/ddns/.venv/bin/python /opt/ddns/ddns.py >> /var/log/ddns.log 2>&1Il TTL basso (300 secondi) fa sì che i resolver aggiornino rapidamente la cache dopo un cambio di indirizzo. Il record non è in modalità proxy perché servizi come SSH o VPN non passano attraverso il proxy HTTP di Cloudflare; per un semplice sito web si può invece impostare proxied=True e nascondere così l'IP reale.
Caso pratico: sincronizzazione dichiarativa
Un approccio più maturo consiste nel descrivere lo stato desiderato della zona in un file versionato e lasciare che uno script calcoli le differenze e le applichi, in modo simile a quanto fanno strumenti come Terraform. Partiamo da un file dns.json:
{
"zone": "example.com",
"records": [
{ "type": "A", "name": "example.com", "content": "203.0.113.10", "proxied": true },
{ "type": "CNAME", "name": "www.example.com", "content": "example.com", "proxied": true },
{ "type": "MX", "name": "example.com", "content": "mail.example.com", "priority": 10 },
{ "type": "TXT", "name": "example.com", "content": "\"v=spf1 mx -all\"" }
]
}Lo script confronta i record desiderati con quelli presenti. Per sicurezza, di default mostra soltanto il piano delle modifiche; le applica solo con l'opzione --apply, e cancella i record non dichiarati solo con l'opzione --prune:
#!/usr/bin/env python3
# dns_sync.py - sincronizza una zona Cloudflare con un file JSON
import argparse
import json
from collections import defaultdict
from cloudflare_dns import CloudflareDNS, DnsRecord
# Tipi di record gestiti dallo script: gli altri vengono ignorati
MANAGED_TYPES = {"A", "AAAA", "CNAME", "MX", "TXT", "CAA"}
def key(record: DnsRecord) -> tuple[str, str]:
return (record.type, record.name)
def plan(desired: list[DnsRecord], current: list[DnsRecord], prune: bool):
to_create, to_update, to_delete = [], [], []
current_by_key = defaultdict(list)
for rec in current:
if rec.type in MANAGED_TYPES:
current_by_key[key(rec)].append(rec)
for rec in desired:
candidates = current_by_key.get(key(rec), [])
# Record identico già presente: nulla da fare
match = next((c for c in candidates if c == rec), None)
if match:
candidates.remove(match)
continue
# Stesso tipo e nome ma valori diversi: aggiornamento
# (per MX e TXT possono esistere più record con lo stesso nome)
same_content = next((c for c in candidates if c.content == rec.content), None)
target = same_content or (candidates[0] if candidates and rec.type in {"A", "AAAA", "CNAME"} else None)
if target:
candidates.remove(target)
to_update.append((target, rec))
else:
to_create.append(rec)
if prune:
for leftovers in current_by_key.values():
to_delete.extend(leftovers)
return to_create, to_update, to_delete
def main() -> None:
parser = argparse.ArgumentParser(description="Sincronizza i record DNS con Cloudflare")
parser.add_argument("config", help="File JSON con lo stato desiderato")
parser.add_argument("--apply", action="store_true", help="Applica le modifiche")
parser.add_argument("--prune", action="store_true", help="Elimina i record non dichiarati")
args = parser.parse_args()
with open(args.config, encoding="utf-8") as fh:
config = json.load(fh)
desired = [DnsRecord(**r) for r in config["records"]]
cf = CloudflareDNS()
zone_id = cf.zone_id(config["zone"])
current = cf.list_records(zone_id)
to_create, to_update, to_delete = plan(desired, current, args.prune)
for rec in to_create:
print(f"+ {rec.type:5} {rec.name} {rec.content}")
for old, new in to_update:
print(f"~ {new.type:5} {new.name} {old.content} -> {new.content}")
for rec in to_delete:
print(f"- {rec.type:5} {rec.name} {rec.content}")
if not (to_create or to_update or to_delete):
print("La zona è già allineata.")
return
if not args.apply:
print("\nPiano generato. Eseguire con --apply per applicarlo.")
return
for rec in to_create:
cf.create_record(zone_id, rec)
for old, new in to_update:
cf.replace_record(zone_id, old.id, new)
for rec in to_delete:
cf.delete_record(zone_id, rec.id)
print("Modifiche applicate.")
if __name__ == "__main__":
main()Un'esecuzione tipica produce un output di questo tipo:
$ python dns_sync.py dns.json
+ MX example.com mail.example.com
~ A example.com 198.51.100.7 -> 203.0.113.10
Piano generato. Eseguire con --apply per applicarlo.Mantenendo dns.json in un repository Git e lanciando lo script da una pipeline CI/CD (prima senza --apply nelle merge request, poi con --apply sul branch principale), ogni modifica al DNS diventa tracciabile, revisionabile e reversibile.
Modifiche atomiche con l'endpoint batch
Lo script precedente esegue una richiesta per ogni record: se una chiamata fallisce a metà, la zona resta in uno stato intermedio. Per questi casi Cloudflare mette a disposizione l'endpoint POST /zones/{zone_id}/dns_records/batch, che accetta in un'unica richiesta quattro liste (deletes, patches, puts e posts) e le applica in modo transazionale, in quest'ordine: o vanno a buon fine tutte, o nessuna.
def apply_batch(cf: CloudflareDNS, zone_id: str, to_create, to_update, to_delete) -> dict:
body = {
"deletes": [{"id": rec.id} for rec in to_delete],
"puts": [{"id": old.id, **new.to_payload()} for old, new in to_update],
"posts": [rec.to_payload() for rec in to_create],
}
# Le liste vuote vengono omesse dalla richiesta
body = {k: v for k, v in body.items() if v}
data = cf._request("POST", f"/zones/{zone_id}/dns_records/batch", json=body)
return data["result"]Sostituendo il ciclo finale di dns_sync.py con una chiamata ad apply_batch() si ottiene una sincronizzazione atomica, con il vantaggio aggiuntivo di consumare una sola richiesta rispetto al limite di frequenza.
L'SDK ufficiale
In alternativa alle chiamate HTTP dirette, Cloudflare pubblica un SDK Python ufficiale, installabile con pip install cloudflare. Le versioni recenti sono generate automaticamente dalla specifica OpenAPI e offrono tipizzazione completa, paginazione automatica e gestione dei tentativi integrata:
import os
from cloudflare import Cloudflare
client = Cloudflare(api_token=os.environ["CLOUDFLARE_API_TOKEN"])
zone = client.zones.list(name="example.com").result[0]
# L'iterazione sulla lista gestisce la paginazione in automatico
for record in client.dns.records.list(zone_id=zone.id, type="A"):
print(record.name, record.content)
client.dns.records.create(
zone_id=zone.id,
type="A",
name="api.example.com",
content="203.0.113.30",
ttl=1,
proxied=True,
)L'SDK è la scelta più comoda per progetti che usano molte aree della API di Cloudflare. Occorre però tenere presente che le versioni maggiori hanno introdotto cambiamenti incompatibili nell'interfaccia: conviene fissare la versione nel file dei requisiti e consultare la documentazione relativa alla versione installata. Per uno script mirato come quelli visti in questo articolo, un client minimale basato su requests ha il vantaggio di non avere dipendenze pesanti e di rendere esplicito ciò che accade a livello HTTP.
Buone pratiche di sicurezza
- Privilegi minimi: creare un token per ciascuno script, con il solo permesso
DNS → Edit(oDNS → Readper il solo export) limitato alle zone necessarie. - Restrizioni sul token: in fase di creazione si possono limitare gli indirizzi IP da cui il token è utilizzabile e impostare una data di scadenza.
- Gestione dei segreti: leggere il token da variabili d'ambiente, da un file con permessi
600o da un gestore di segreti; non inserirlo mai nel repository. - Modalità di prova: negli script che modificano più record, prevedere sempre un'esecuzione senza effetti che mostri il piano delle modifiche, come fatto in
dns_sync.py. - Backup: esportare periodicamente le zone in formato BIND con l'endpoint
/dns_records/export, così da poterle ripristinare tramite l'endpoint di import in caso di errori.
Conclusioni
Le API di Cloudflare rendono la gestione DNS pienamente automatizzabile, e Python, grazie a requests e a poche righe di codice ben organizzate, è uno strumento ideale per farlo. Partendo dalle singole chiamate CRUD abbiamo costruito un client con paginazione, gestione degli errori e dei limiti di frequenza, e lo abbiamo usato per due scenari concreti: un aggiornatore di DNS dinamico e una sincronizzazione dichiarativa della zona, resa atomica grazie all'endpoint batch. Da qui è semplice estendere il client ad altre aree della API, come le regole del firewall, la cache o i certificati.