Gestire le query WHOIS in Python
Python è una scelta naturale per gli strumenti di rete: la libreria standard offre socket di alto livello, un'implementazione completa di asyncio e tutto il necessario per gestire codifiche e domini internazionalizzati. In questo articolo costruiamo un client WHOIS completo, prima in versione sincrona con il modulo socket e poi in versione asincrona con asyncio, capace di interrogare molti domini in parallelo rispettando i limiti di frequenza dei server. Il codice richiede Python 3.11 o successivo, perché usa il gestore di contesto asyncio.timeout().
Prima di cominciare conviene inquadrare il contesto attuale. Il 28 gennaio 2025 è decaduto per registry e registrar dei domini generici (gTLD) l'obbligo, previsto dai contratti con ICANN, di offrire WHOIS sulla porta 43: per quei domini il protocollo di riferimento è ora RDAP. WHOIS però è tutt'altro che scomparso. La maggior parte dei registry nazionali, compresi quelli di .it e .de, continua a usarlo, e alcuni registry generici, tra cui quello del .com, mantengono il servizio. Il nostro client dovrà quindi funzionare bene dove WHOIS è disponibile e degradare in modo prevedibile dove non lo è.
Il protocollo
WHOIS è definito dalla RFC 3912 ed è quanto di più semplice si possa immaginare: il client si connette alla porta TCP 43, invia la query seguita da \r\n e legge fino alla chiusura della connessione da parte del server. Non esistono intestazioni, lunghezze dichiarate o codifiche dichiarate, e il formato della risposta è testo libero.
Questa semplicità sposta la complessità sul client, che deve:
- impostare timeout di connessione e di lettura, oltre a una scadenza complessiva;
- limitare la dimensione massima della risposta;
- decodificare i byte provando UTF-8 e ripiegando su Latin-1;
- validare l'input per impedire l'iniezione di righe aggiuntive nella query;
- individuare il server corretto seguendo i referral;
- interpretare il testo con un parser tollerante.
Il percorso di ricerca standard parte da whois.iana.org, che restituisce il record del TLD con il campo refer:; da lì si interroga il server del registry e, se la risposta contiene il campo Registrar WHOIS Server:, anche quello del registrar.
Una prima query
La funzione socket.create_connection() risolve il nome del server, apre la connessione e applica il timeout indicato sia alla connessione sia alle singole operazioni successive. Il primo esempio è poco più di una riga di codice significativa.
import socket
with socket.create_connection(("whois.verisign-grs.com", 43), timeout=10) as sock:
# La query è il dominio seguito da CRLF
sock.sendall(b"domain example.com\r\n")
chunks = []
while chunk := sock.recv(4096):
chunks.append(chunk)
print(b"".join(chunks).decode("utf-8"))Il ciclo con l'operatore di assegnazione := continua finché recv() restituisce byte e si ferma quando restituisce una sequenza vuota, cioè quando il server ha chiuso la connessione. La parola chiave domain davanti al nome è una particolarità del server di Verisign, che altrimenti cercherebbe la stringa anche tra i nomi dei name server registrati.
Errori, normalizzazione e decodifica
Definiamo innanzitutto un'eccezione specifica e le funzioni di supporto. L'eccezione porta con sé il nome del server quando l'errore è remoto, così chi la intercetta può distinguere un problema di input da un problema di rete.
Per i domini internazionalizzati, come caffè.it, serve la conversione in Punycode (xn--caff-8oa.it). Python include il codec idna, che però implementa la vecchia versione 2003 dello standard. Il pacchetto idna, installabile con pip install idna, implementa IDNA 2008 con la mappatura UTS #46 ed è preferibile; il codice seguente lo usa se presente e ripiega sul codec integrato altrimenti.
# whois_client/core.py
from __future__ import annotations
import re
import socket
import time
try:
import idna
except ImportError: # pragma: no cover
idna = None
WHOIS_PORT = 43
IANA_SERVER = "whois.iana.org"
DEFAULT_TIMEOUT = 10.0
MAX_RESPONSE_BYTES = 1024 * 1024
ASCII_DOMAIN = re.compile(r"^[a-z0-9.-]+$")
class WhoisError(Exception):
def __init__(self, message: str, server: str | None = None) -> None:
super().__init__(message)
self.server = server
def normalize_domain(value: str) -> str:
# Rimuove spazi, maiuscole e l'eventuale punto finale
domain = value.strip().lower().rstrip(".")
try:
if idna is not None:
ascii_domain = idna.encode(domain, uts46=True).decode("ascii")
else:
ascii_domain = domain.encode("idna").decode("ascii")
except UnicodeError as exc:
raise WhoisError(f"Dominio non valido: {value}") from exc
# Dopo la conversione sono ammessi solo caratteri ASCII da nome di dominio
if not ASCII_DOMAIN.fullmatch(ascii_domain) or len(ascii_domain) > 253:
raise WhoisError(f"Dominio non valido: {value}")
labels = ascii_domain.split(".")
if len(labels) < 2:
raise WhoisError(f"Dominio non valido: {value}")
for label in labels:
if not 0 < len(label) <= 63 or label.startswith("-") or label.endswith("-"):
raise WhoisError(f"Etichetta non valida nel dominio: {value}")
return ascii_domain
def decode_response(data: bytes) -> str:
# UTF-8 rigoroso, con ripiego su Latin-1 che decodifica qualsiasi byte
try:
return data.decode("utf-8")
except UnicodeDecodeError:
return data.decode("latin-1")Il controllo con l'espressione regolare dopo la conversione è la vera protezione contro l'iniezione: indipendentemente da ciò che contiene l'input, la stringa che finisce nella query può contenere solo lettere minuscole, cifre, punti e trattini, e quindi nessun carattere di fine riga.
Una query sincrona robusta
Il timeout passato a create_connection() si applica a ogni singola operazione: se il server tace per dieci secondi, recv() solleva TimeoutError. Non protegge però da un server che invia pochi byte a intervalli regolari, quindi aggiungiamo una scadenza complessiva controllata a ogni iterazione con time.monotonic(), l'orologio adatto a misurare intervalli perché non risente delle correzioni dell'ora di sistema.
def query_server(
server: str,
query: str,
timeout: float = DEFAULT_TIMEOUT,
deadline: float | None = None,
max_bytes: int = MAX_RESPONSE_BYTES,
) -> str:
deadline_at = time.monotonic() + (deadline or timeout * 3)
chunks: list[bytes] = []
received = 0
try:
with socket.create_connection((server, WHOIS_PORT), timeout=timeout) as sock:
sock.sendall(f"{query}\r\n".encode("ascii"))
while chunk := sock.recv(4096):
received += len(chunk)
if received > max_bytes:
raise WhoisError(f"Risposta oltre il limite di {max_bytes} byte", server)
if time.monotonic() > deadline_at:
raise WhoisError("Scadenza complessiva superata", server)
chunks.append(chunk)
except TimeoutError as exc:
raise WhoisError(f"Nessuna risposta per {timeout} secondi", server) from exc
except OSError as exc:
raise WhoisError(f"Errore di rete: {exc}", server) from exc
return decode_response(b"".join(chunks))L'ordine delle clausole except è importante: TimeoutError è una sottoclasse di OSError, quindi va intercettata per prima se vogliamo un messaggio specifico. Le nostre WhoisError sollevate all'interno del blocco non vengono intercettate da queste clausole perché non derivano da OSError, e il gestore di contesto chiude comunque il socket.
Referral e catena dei server
La logica dei referral è indipendente dal modo in cui si eseguono le query, quindi la scriviamo come funzioni pure che lavorano sul testo. Le regole sono le stesse di qualsiasi client WHOIS ben fatto: estrarre l'hostname anche quando il registrar indica un URL, evitare i cicli tenendo traccia dei server visitati, limitare il numero di passaggi e non considerare fatale il fallimento del solo registrar.
REFERRAL_PATTERNS = [
re.compile(r"^\s*refer:\s*(\S+)", re.I | re.M),
re.compile(r"^\s*whois:\s*(\S+)", re.I | re.M),
re.compile(r"^\s*Registrar WHOIS Server:\s*(\S+)", re.I | re.M),
]
HOSTNAME = re.compile(r"^[a-z0-9.-]+\.[a-z]{2,}$")
QUERY_TEMPLATES = {
"whois.verisign-grs.com": "domain {domain}",
}
MAX_HOPS = 4
def build_query(server: str, domain: str) -> str:
return QUERY_TEMPLATES.get(server, "{domain}").format(domain=domain)
def find_referral(text: str) -> str | None:
for pattern in REFERRAL_PATTERNS:
if match := pattern.search(text):
# Alcuni registrar indicano un URL: teniamo solo l'hostname
host = re.sub(r"^[a-z]+://", "", match.group(1), flags=re.I)
host = re.split(r"[/:]", host, maxsplit=1)[0].lower()
if HOSTNAME.fullmatch(host):
return host
return None
def top_level_domain(domain: str) -> str:
return domain.rsplit(".", 1)[-1]
def lookup(domain: str, tld_servers: dict[str, str] | None = None, **options) -> dict:
domain = normalize_domain(domain)
tld_servers = tld_servers if tld_servers is not None else {}
tld = top_level_domain(domain)
chain: list[dict] = []
visited: set[str] = set()
server: str | None = tld_servers.get(tld, IANA_SERVER)
while server and len(chain) < MAX_HOPS and server not in visited:
visited.add(server)
try:
text = query_server(server, build_query(server, domain), **options)
except WhoisError as exc:
# Con una risposta del registry già ottenuta, un registrar muto non è fatale
if any(step["server"] != IANA_SERVER for step in chain):
chain.append({"server": server, "error": str(exc)})
break
raise
chain.append({"server": server, "text": text})
next_server = find_referral(text)
if server == IANA_SERVER and next_server:
tld_servers[tld] = next_server
server = next_server
answers = [step for step in chain if "text" in step and step["server"] != IANA_SERVER]
if not answers:
raise WhoisError(f"Nessun server WHOIS trovato per {domain}", IANA_SERVER)
return {"domain": domain, "chain": chain, "text": answers[-1]["text"]}Il dizionario tld_servers viene passato dall'esterno e aggiornato: in questo modo chi chiama la funzione più volte può conservare l'associazione tra TLD e server del registry ed evitare di interrogare IANA a ogni ricerca. La risposta di IANA per .com, del resto, non cambia da una ricerca all'altra.
Il parser e il record
Le risposte WHOIS si dividono in due grandi famiglie: righe Chiave: valore, usate dalla maggior parte dei registry generici, e blocchi con un'intestazione senza due punti seguita da righe rientrate, usati ad esempio dal registry italiano. Il parser seguente gestisce entrambe e restituisce un dizionario di liste, perché molte chiavi compaiono più volte. Il record finale è una dataclass congelata, immutabile e comoda da serializzare.
# whois_client/parser.py
from __future__ import annotations
import re
from collections import defaultdict
from dataclasses import asdict, dataclass, field
COMMENT_PREFIXES = ("%", "#", ">>>")
FIELD_ALIASES = {
"registrar": ["registrar", "registrar.organization", "registrar.name", "sponsoring registrar"],
"created_at": ["creation date", "created", "created on", "registered on", "registration time"],
"updated_at": ["updated date", "last update", "last updated", "changed", "modified"],
"expires_at": [
"registry expiry date",
"registrar registration expiration date",
"expiry date",
"expire date",
"expire",
"expires",
"expiration time",
],
"name_servers": ["name server", "nameservers", "nserver"],
"status": ["domain status", "status", "state"],
}
NOT_FOUND_PATTERNS = [
re.compile(r"^no match for", re.I | re.M),
re.compile(r"^not found", re.I | re.M),
re.compile(r"^status:\s*(available|free)\b", re.I | re.M),
re.compile(r"no data found", re.I),
re.compile(r"no entries found", re.I),
re.compile(r"domain not found", re.I),
]
def parse_fields(text: str) -> dict[str, list[str]]:
fields: dict[str, list[str]] = defaultdict(list)
section: str | None = None
for raw_line in text.splitlines():
line = raw_line.strip()
# Una riga vuota chiude il blocco corrente
if not line:
section = None
continue
if line.startswith(COMMENT_PREFIXES):
continue
key, separator, value = line.partition(":")
if not separator or not key:
# Valore del blocco corrente oppure intestazione di un nuovo blocco
if section:
fields[section].append(line)
else:
section = line.lower()
continue
key = key.strip().lower()
value = value.strip()
if not value:
section = key
continue
fields[f"{section}.{key}" if section else key].append(value)
return dict(fields)
def is_available(text: str) -> bool:
return any(pattern.search(text) for pattern in NOT_FOUND_PATTERNS)
@dataclass(frozen=True, slots=True)
class WhoisRecord:
domain: str
available: bool
registrar: str | None = None
created_at: str | None = None
updated_at: str | None = None
expires_at: str | None = None
name_servers: list[str] = field(default_factory=list)
status: list[str] = field(default_factory=list)
servers: list[str] = field(default_factory=list)
raw: str = ""
def to_dict(self, include_raw: bool = False) -> dict:
data = asdict(self)
if not include_raw:
data.pop("raw")
return data
def _first(fields: dict[str, list[str]], aliases: list[str]) -> str | None:
for alias in aliases:
if values := fields.get(alias):
return values[0]
return None
def _all(fields: dict[str, list[str]], aliases: list[str]) -> list[str]:
# dict.fromkeys conserva l'ordine eliminando i duplicati
values = (
value.split()[0].lower()
for alias in aliases
for value in fields.get(alias, [])
)
return list(dict.fromkeys(values))
def to_record(domain: str, text: str, servers: list[str] | None = None) -> WhoisRecord:
fields = parse_fields(text)
return WhoisRecord(
domain=domain,
available=is_available(text),
registrar=_first(fields, FIELD_ALIASES["registrar"]),
created_at=_first(fields, FIELD_ALIASES["created_at"]),
updated_at=_first(fields, FIELD_ALIASES["updated_at"]),
expires_at=_first(fields, FIELD_ALIASES["expires_at"]),
name_servers=_all(fields, FIELD_ALIASES["name_servers"]),
status=_all(fields, FIELD_ALIASES["status"]),
servers=servers or [],
raw=text,
)Il metodo str.partition() è particolarmente adatto a questo tipo di parsing: divide la stringa alla prima occorrenza del separatore e restituisce sempre tre elementi, con il separatore vuoto se non è stato trovato. Così il caso della riga senza due punti si riconosce con un semplice controllo, senza gestire indici.
Il rilevamento della disponibilità, svolto sul testo grezzo, resta un'euristica: solo il registrar al momento della registrazione può confermarla. Le date restano stringhe, perché ogni registry usa un formato diverso. Se servono oggetti datetime, conviene provare una lista di formati noti con datetime.strptime() e lasciare il valore nullo quando nessuno corrisponde, invece di tentare interpretazioni creative.
La versione asincrona
La versione sincrona va bene per uno script che controlla qualche dominio. Per verificarne centinaia, attendere ogni server in sequenza è troppo lento. Con asyncio possiamo avere molte query in volo contemporaneamente in un solo thread, e il fatto che il ciclo di eventi esegua il codice Python un pezzo alla volta semplifica la gestione dello stato condiviso: non servono lock per la cache o per il limitatore di frequenza.
La funzione asyncio.open_connection() restituisce una coppia di stream per leggere e scrivere; asyncio.timeout() fornisce la scadenza complessiva in modo pulito, annullando l'operazione in corso quando il tempo scade.
# whois_client/aio.py
from __future__ import annotations
import asyncio
import time
from .core import (
IANA_SERVER,
MAX_HOPS,
MAX_RESPONSE_BYTES,
WHOIS_PORT,
WhoisError,
build_query,
decode_response,
find_referral,
normalize_domain,
top_level_domain,
)
from .parser import WhoisRecord, to_record
async def query_server_async(
server: str,
query: str,
timeout: float = 10.0,
max_bytes: int = MAX_RESPONSE_BYTES,
) -> str:
chunks: list[bytes] = []
received = 0
writer: asyncio.StreamWriter | None = None
try:
# La scadenza copre connessione, invio e lettura
async with asyncio.timeout(timeout * 3):
reader, writer = await asyncio.wait_for(
asyncio.open_connection(server, WHOIS_PORT), timeout
)
writer.write(f"{query}\r\n".encode("ascii"))
await writer.drain()
while chunk := await asyncio.wait_for(reader.read(4096), timeout):
received += len(chunk)
if received > max_bytes:
raise WhoisError(f"Risposta oltre il limite di {max_bytes} byte", server)
chunks.append(chunk)
except TimeoutError as exc:
raise WhoisError("Timeout durante la query", server) from exc
except OSError as exc:
raise WhoisError(f"Errore di rete: {exc}", server) from exc
finally:
if writer is not None:
writer.close()
return decode_response(b"".join(chunks))
class AsyncWhoisClient:
def __init__(
self,
ttl: float = 3600.0,
min_interval: float = 1.0,
concurrency: int = 10,
timeout: float = 10.0,
) -> None:
self.ttl = ttl
self.min_interval = min_interval
self.timeout = timeout
self._semaphore = asyncio.Semaphore(concurrency)
self._cache: dict[str, tuple[float, WhoisRecord]] = {}
self._in_flight: dict[str, asyncio.Task[WhoisRecord]] = {}
self._next_slot: dict[str, float] = {}
self._tld_servers: dict[str, str] = {}
async def _throttle(self, server: str) -> None:
now = time.monotonic()
slot = max(now, self._next_slot.get(server, 0.0))
# La prenotazione avviene prima dell'attesa: nessun await tra lettura e scrittura
self._next_slot[server] = slot + self.min_interval
if slot > now:
await asyncio.sleep(slot - now)
async def _query(self, server: str, query: str) -> str:
await self._throttle(server)
async with self._semaphore:
return await query_server_async(server, query, timeout=self.timeout)
async def _resolve(self, domain: str) -> WhoisRecord:
tld = top_level_domain(domain)
chain: list[dict] = []
visited: set[str] = set()
server: str | None = self._tld_servers.get(tld, IANA_SERVER)
while server and len(chain) < MAX_HOPS and server not in visited:
visited.add(server)
try:
text = await self._query(server, build_query(server, domain))
except WhoisError as exc:
if any(step["server"] != IANA_SERVER for step in chain):
chain.append({"server": server, "error": str(exc)})
break
raise
chain.append({"server": server, "text": text})
next_server = find_referral(text)
if server == IANA_SERVER and next_server:
self._tld_servers[tld] = next_server
server = next_server
answers = [step for step in chain if "text" in step and step["server"] != IANA_SERVER]
if not answers:
raise WhoisError(f"Nessun server WHOIS trovato per {domain}", IANA_SERVER)
return to_record(domain, answers[-1]["text"], [step["server"] for step in chain])
async def lookup(self, value: str, fresh: bool = False) -> WhoisRecord:
domain = normalize_domain(value)
cached = self._cache.get(domain)
if not fresh and cached and cached[0] > time.monotonic():
return cached[1]
# Le richieste concorrenti per lo stesso dominio condividono lo stesso task
task = self._in_flight.get(domain)
if task is None:
task = asyncio.create_task(self._resolve(domain))
self._in_flight[domain] = task
task.add_done_callback(lambda _: self._in_flight.pop(domain, None))
record = await task
self._cache[domain] = (time.monotonic() + self.ttl, record)
return record
async def lookup_many(self, values: list[str]) -> dict[str, WhoisRecord | WhoisError]:
results = await asyncio.gather(
*(self.lookup(value) for value in values),
return_exceptions=True,
)
output: dict[str, WhoisRecord | WhoisError] = {}
for value, result in zip(values, results):
if isinstance(result, BaseException) and not isinstance(result, WhoisError):
raise result
output[value] = result
return outputVale la pena soffermarsi su tre scelte. La prima riguarda il limitatore: il metodo _throttle() legge e aggiorna _next_slot senza alcun await in mezzo, quindi nessun'altra coroutine può intervenire tra lettura e scrittura. Le richieste concorrenti verso lo stesso server si mettono così in fila, distanziate di min_interval secondi.
La seconda riguarda il semaforo, che limita il numero di connessioni aperte contemporaneamente indipendentemente dal server. Il limitatore viene invocato prima di acquisire il semaforo: in caso contrario una coroutine in attesa del proprio turno occuperebbe un posto del semaforo senza fare nulla.
La terza riguarda la deduplicazione. Il dizionario _in_flight contiene i task in corso: se arrivano dieci richieste per lo stesso dominio, la prima crea il task e le altre attendono lo stesso. Il callback registrato con add_done_callback() rimuove il task dal dizionario quando termina, sia in caso di successo sia in caso di errore.
Infine, lookup_many() usa asyncio.gather() con return_exceptions=True, così il fallimento di un dominio non interrompe gli altri. Le eccezioni diverse da WhoisError indicano invece un errore di programmazione e vengono rilanciate.
Uno strumento a riga di comando
Concludiamo con un piccolo programma a riga di comando basato su argparse, che accetta uno o più domini e stampa i risultati in formato JSON.
# whois_client/__main__.py
from __future__ import annotations
import argparse
import asyncio
import json
import sys
from .aio import AsyncWhoisClient
from .core import WhoisError, normalize_domain
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(prog="whois-client", description="Client WHOIS asincrono")
parser.add_argument("domains", nargs="+", help="domini da interrogare")
parser.add_argument("--raw", action="store_true", help="includi il testo originale")
parser.add_argument("--interval", type=float, default=1.0, help="secondi tra due query allo stesso server")
parser.add_argument("--timeout", type=float, default=10.0, help="timeout per operazione")
return parser.parse_args()
async def main() -> int:
args = parse_args()
# La validazione avviene subito, prima di qualsiasi query
try:
for domain in args.domains:
normalize_domain(domain)
except WhoisError as exc:
print(exc, file=sys.stderr)
return 2
client = AsyncWhoisClient(min_interval=args.interval, timeout=args.timeout)
results = await client.lookup_many(args.domains)
output = {}
exit_code = 0
for domain, result in results.items():
if isinstance(result, WhoisError):
output[domain] = {"error": str(result), "server": result.server}
exit_code = 1
else:
output[domain] = result.to_dict(include_raw=args.raw)
print(json.dumps(output, indent=2, ensure_ascii=False))
return exit_code
if __name__ == "__main__":
sys.exit(asyncio.run(main()))Il programma si esegue come modulo:
python -m whois_client example.com example.org caffè.it --interval 1.5Il codice di uscita segue le convenzioni Unix: zero se tutte le ricerche sono riuscite, uno se almeno una è fallita, due se l'input non è valido. Questo lo rende facile da usare in script di monitoraggio, ad esempio per controllare periodicamente la scadenza dei propri domini.
Testare senza rete
Avendo separato la logica dei referral e il parser dalle funzioni di rete, è semplice testarli con pytest usando risposte prefabbricate. Ecco un test per il formato a blocchi del registry italiano:
# tests/test_parser.py
from whois_client.parser import to_record
IT_RESPONSE = """
Domain: example.it
Status: ok
Created: 2000-01-01 00:00:00
Expire Date: 2027-01-01
Registrar
Organization: Example Registrar S.r.l.
Name: EXAMPLE-REG
Nameservers
dns1.example.it
dns2.example.it
"""
def test_block_format_is_parsed():
record = to_record("example.it", IT_RESPONSE)
assert record.available is False
assert record.registrar == "Example Registrar S.r.l."
assert record.expires_at == "2027-01-01"
assert record.name_servers == ["dns1.example.it", "dns2.example.it"]
def test_available_domain_is_detected():
record = to_record("libero.it", "Domain: libero-test.it\nStatus: AVAILABLE\n")
assert record.available is TrueConsiderazioni finali
Il client che abbiamo costruito copre bene i casi d'uso tipici, ma ha dei confini da conoscere. Per i domini generici il protocollo di riferimento è ormai RDAP, che usa HTTPS e JSON strutturato e prevede un registro di bootstrap pubblicato da IANA per individuare il server corretto; un client completo dovrebbe tentare prima RDAP e usare WHOIS come ripiego. Dopo il GDPR molti dati degli intestatari sono oscurati, e diciture come REDACTED FOR PRIVACY vanno considerate normali. Infine, i server WHOIS vietano generalmente l'uso massivo e commerciale dei dati: la cache, il limitatore di frequenza e il semaforo non sono ottimizzazioni facoltative, ma ciò che permette al client di continuare a funzionare senza essere bloccato.