Set-OPS-Public/scripts/sonder.py

#!/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())