#!/usr/bin/env python3 """Sonder une cible, et RAPPORTER CE QUI DISTINGUE — pas un « echec ». POURQUOI CET OUTIL EXISTE (2026-08-28). Trois faux diagnostics en une journee, tous dus a l'instrument et aucun au composant : 1. « BLOQUE » vers le resolveur, le cache et la PKI d'un tenant, lu comme un pare-feu. C'etait une ABSENCE — une seule VM existait. Un pare-feu a ete applique pour reparer un vide. 2. « 3,05 s » presente comme le signal d'un refus. C'etait l'ARP qui abandonne, pas un REJECT — lequel repond en microsecondes. 3. Un REJECT declare muet. Il emettait bien quatre RST (compteur 17 -> 21) ; la sonde partait d'une passerelle ANYCAST, donc la reponse revenait ailleurs. `curl`, `bash /dev/tcp` et un `ping` nu ecrasent quatre causes incompatibles dans le meme mot. L'information etait la a chaque fois — l'outil la jetait. CE QU'IL MESURE, ET CE QU'IL REFUSE DE CONCLURE : ouvert, ~0 s le flux est declare et le service ecoute refuse (RST/ECONNREFUSED) une POLITIQUE interne refuse — reject, immediat EHOSTUNREACH, ~3 s LA MACHINE N'EXISTE PAS (l'ARP a abandonne) silence jusqu'au delai la FRONTIERE, muette par conception (block) Et deux gardes sur l'instrument lui-meme : - il dit par quelle SOURCE le paquet part (`ip route get`), parce qu'une passerelle anycast rend la mesure NON CONCLUANTE : la reponse revient au noeud qui porte l'adresse, pas a l'emetteur ; - il ne conclut pas quand il ne peut pas. « Non concluant » est un resultat. TCP SEULEMENT. UDP n'a pas de poignee : un silence y confond « ouvert » et « filtre », et pretendre trancher serait exactement le defaut qu'on corrige. Usage : python3 scripts/sonder.py 10.17.19.41 22 443 python3 scripts/sonder.py infra-dns-01 53 # nom resolu par l'inventaire """ from __future__ import annotations import argparse import errno as _errno import socket import subprocess import sys import time from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) # Seuil au-dela duquel un echec « injoignable » porte la signature de l'ARP qui abandonne # (trois retransmissions). En-dessous, c'est un refus immediat de la couche 3. SEUIL_ARP = 1.5 VERDICTS = { "ouvert": "le flux est declare et le service ecoute", "refuse": "une POLITIQUE interne refuse (reject) — immediat", "absent": "LA MACHINE N'EXISTE PAS : l'ARP a abandonne", "injoignable": "pas de route vers cette destination", "silence": "refus MUET — la frontiere (block), ou un drop en chemin", } def interpreter(issue: str, err: int | None, duree: float) -> tuple[str, str]: """(cle, phrase) — la lecture d'une sonde. Fonction PURE, donc testable. ELLE NE DEVINE PAS. `EHOSTUNREACH` rapide n'est pas la meme chose que le meme code apres trois secondes : le premier est un refus de la couche 3, le second la signature de l'ARP qui renonce. Confondre les deux, c'est refaire l'erreur du 2026-08-28. """ if issue == "ouvert": return "ouvert", VERDICTS["ouvert"] if err == _errno.ECONNREFUSED: return "refuse", VERDICTS["refuse"] if err == _errno.EHOSTUNREACH: cle = "absent" if duree >= SEUIL_ARP else "injoignable" return cle, VERDICTS[cle] if err in (_errno.ENETUNREACH,): return "injoignable", VERDICTS["injoignable"] return "silence", VERDICTS["silence"] def sonder(cible: str, port: int, delai: float) -> dict: debut = time.time() s = socket.socket() s.settimeout(delai) try: s.connect((cible, port)) issue, err = "ouvert", None except OSError as e: issue, err = "echec", getattr(e, "errno", None) finally: s.close() duree = time.time() - debut cle, phrase = interpreter(issue, err, duree) return {"port": port, "cle": cle, "phrase": phrase, "duree": duree, "errno": err, "nom_errno": _errno.errorcode.get(err, "") if err else ""} def source_vers(cible: str) -> tuple[str, str]: """(adresse source, note) telle que le noyau la choisira. ('', '') si indisponible. La note vaut '', 'anycast' (derive du plan) ou 'anycast?' (heuristique, module absent). """ try: sortie = subprocess.run(["ip", "route", "get", cible], capture_output=True, text=True, timeout=5).stdout except (OSError, subprocess.SubprocessError): return "", "" jetons = sortie.split() if "src" not in jetons: return "", "" src = jetons[jetons.index("src") + 1] return src, _est_passerelle(src) def _est_passerelle(adresse: str) -> str: """'anycast', 'anycast?' ou '' — cette adresse est-elle une passerelle de zone ? DERIVE, jamais compare a une liste : les passerelles sont le premier hote de chaque /24 de zone (`passerelle_de`). Une liste ecrite ici vieillirait au premier tenant ajoute, et la garde se tairait exactement quand elle devrait parler. ET ELLE NE TOMBE PAS DU COTE SILENCIEUX. Ma premiere version rendait « pas anycast » quand `inventory_rules` etait introuvable — ce qui arrive des qu'on copie ce seul fichier sur un hyperviseur pour diagnostiquer, c'est-a-dire exactement la ou le piege se produit. Constate le 2026-08-28 : l'avertissement s'est taise sur `10.17.21.1`, qui EST une passerelle anycast. Une garde qui echoue doit crier, pas se taire ; sans le module, on retombe donc sur une heuristique franchement etiquetee. """ morceaux = adresse.split(".") if len(morceaux) != 4 or morceaux[3] != "1" or morceaux[0] != "10": return "" try: index, troisieme = int(morceaux[1]), int(morceaux[2]) except ValueError: return "" try: from inventory_rules import passerelle_de except Exception: # Sans le moteur, on ne PROUVE pas — mais un `10.x.y.1` est un premier hote, # donc une passerelle dans toute nomenclature de ce depot. On le dit au # conditionnel plutot que de laisser croire que la mesure est concluante. return "anycast?" return "anycast" if any(passerelle_de(index, cat) == adresse for cat in range(0, 12)) else "" def _adresse_de(nom: str) -> str: """Le nom tel que l'inventaire le connait, sinon le nom tel quel.""" try: import yaml from inventory_rules import inventaire_de inv = inventaire_de() if inv and Path(inv).is_file(): data = yaml.safe_load(Path(inv).read_text(encoding="utf-8")) or {} pile = [data.get("all", {})] while pile: n = pile.pop() for h, v in (n.get("hosts") or {}).items(): if h == nom and isinstance(v, dict) and v.get("ansible_host"): return str(v["ansible_host"]) pile.extend((n.get("children") or {}).values()) except Exception: pass return nom def main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("cible", help="adresse, ou nom d'hote connu de l'inventaire") ap.add_argument("ports", nargs="*", type=int, default=[22], help="ports TCP (defaut 22)") ap.add_argument("--delai", type=float, default=6.0, help="patience, en secondes") args = ap.parse_args(argv) adresse = _adresse_de(args.cible) etiquette = args.cible if adresse == args.cible else f"{args.cible} ({adresse})" print(f"Sonde de {etiquette} — TCP, patience {args.delai:g} s") src, note = source_vers(adresse) if src: avertissement = "" if note: certain = note == "anycast" avertissement = ( f" /!\\ PASSERELLE {'ANYCAST' if certain else 'PROBABLE (module absent)'}" " — mesure NON CONCLUANTE en cas de silence :\n" " la reponse revient au noeud qui porte cette adresse, pas ici.") print(f" source choisie par le noyau : {src}{avertissement}") print() resultats = [sonder(adresse, p, args.delai) for p in args.ports] for r in resultats: code = f" [{r['nom_errno']}]" if r["nom_errno"] else "" print(f" {r['port']:<7} {r['cle']:<12} {r['duree']:6.3f} s{code}") print(f" {r['phrase']}") if note and any(r["cle"] == "silence" for r in resultats): print("\n Un silence mesure depuis une passerelle anycast ne prouve RIEN :" "\n compter les paquets du cote qui refuse, ou sonder depuis une VM.") return 0 if __name__ == "__main__": raise SystemExit(main())