Set-OPS-Public/scripts/sonde_tcp.py

#!/usr/bin/env python3
"""Sonde TCP honnete : distingue LIVRE, BLOQUE et AMBIGU — et refuse de conclure.

Pourquoi ce fichier existe. Pendant deux jours, `connect()` a menti : a travers la
frontiere, toute connexion aboutissait, y compris vers `172.31.99.99` — une adresse qui
n'appartient a personne. La cause etait un `synproxy` porte par une regle heritee : le
pare-feu repondait au SYN a la place de la destination. Trois faux diagnostics en sont
sortis avant qu'un controle ne le revele.

D'ou les deux principes de cette sonde :

1. **Un CONTROLE avant toute conclusion.** On sonde d'abord une adresse ou personne
   n'ecoute. Si elle « repond », l'instrument est en cause et le releve est declare NUL —
   aucun verdict n'est rendu. Mieux vaut pas de mesure qu'une mesure fausse.

2. **On ne conclut jamais sur `connect()` seul.** Etablir n'est pas livrer. La sonde
   essaie de faire PARLER le service, dans cet ordre : lire une banniere (SSH, SMTP,
   IMAP parlent en premier), sinon envoyer une requete HTTP minimale, sinon tenter une
   poignee TLS. Si rien ne revient, le verdict est **AMBIGU**, pas « ouvert » : certains
   protocoles (LDAP, PostgreSQL) attendent un message bien forme qu'on ne fabrique pas
   ici, et un synproxy se comporte exactement pareil.

Usage : la liste des cibles arrive en JSON sur l'entree standard, les verdicts sortent en
JSON sur la sortie standard. Aucun etat, aucun fichier, aucun privilege.

  echo '{"controle": ["172.31.99.99", 22], "cibles": [["10.27.18.21", 22, "ssh"]]}' \
    | python3 scripts/sonde_tcp.py
"""
from __future__ import annotations

import json
import socket
import ssl
import sys
import time
from concurrent.futures import ThreadPoolExecutor

DELAI_CONNEXION = 4.0
# Genereux a dessein : Postfix retarde sa banniere de plusieurs secondes (postscreen le
# fait expres, pour pieger les spambots). A 2 s, un port parfaitement sain ressortait
# AMBIGU. On compense le cout par le parallelisme plutot qu'en raccourcissant l'attente —
# raccourcir aurait fabrique de faux ecarts, ce que cette sonde existe pour eviter.
DELAI_LECTURE = 8.0
PARALLELE = 12


def _sonder(hote: str, port: int) -> dict:
    """Un verdict pour une cible : BLOQUE, LIVRE ou AMBIGU, avec la preuve."""
    debut = time.monotonic()
    try:
        s = socket.create_connection((hote, port), timeout=DELAI_CONNEXION)
    except OSError as e:
        return {"verdict": "BLOQUE", "ms": int((time.monotonic() - debut) * 1000),
                "preuve": type(e).__name__}
    ms = int((time.monotonic() - debut) * 1000)
    try:
        s.settimeout(DELAI_LECTURE)
        # 1. Le serveur parle-t-il en premier ? (SSH, SMTP, IMAP, POP3, redis apres coup)
        try:
            data = s.recv(96)
            if data:
                return {"verdict": "LIVRE", "ms": ms,
                        "preuve": data.decode("utf-8", "replace").strip()[:60]}
        except (TimeoutError, socket.timeout, OSError):
            pass
        # 2. Sinon, le faire parler. Une requete HTTP minimale suffit pour tout ce qui
        #    parle HTTP, et arrache une erreur lisible a plusieurs qui ne le parlent pas.
        try:
            s.sendall(b"GET / HTTP/1.0\r\nHost: sonde\r\n\r\n")
            data = s.recv(96)
            if data:
                return {"verdict": "LIVRE", "ms": ms,
                        "preuve": data.decode("utf-8", "replace").strip()[:60]}
        except (TimeoutError, socket.timeout, OSError):
            pass
    finally:
        try:
            s.close()
        except OSError:
            pass
    # 3. Dernier essai : une vraie poignee TLS. En TLS c'est le CLIENT qui parle en
    #    premier — le silence des deux etapes precedentes ne prouve donc rien ici.
    try:
        ctx = ssl.create_default_context()
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        with socket.create_connection((hote, port), timeout=DELAI_CONNEXION) as brut:
            with ctx.wrap_socket(brut) as tls:
                cert = tls.getpeercert(binary_form=True)
                if cert:
                    return {"verdict": "LIVRE", "ms": ms,
                            "preuve": f"TLS {tls.version()}, certificat servi"}
    except (OSError, ssl.SSLError):
        pass
    return {"verdict": "AMBIGU", "ms": ms,
            "preuve": "connexion etablie, aucune donnee applicative"}


def main() -> int:
    # En argument quand la sonde est deposee sur un hote distant (`ansible.builtin.script`
    # ne sait pas alimenter l'entree standard), sur stdin en local.
    entree = json.loads(sys.argv[1]) if len(sys.argv) > 1 else json.load(sys.stdin)
    ctrl_hote, ctrl_port = entree["controle"]
    controle = _sonder(str(ctrl_hote), int(ctrl_port))

    # Le controle ne doit rien LIVRER. Qu'il ressorte AMBIGU est tolerable et meme
    # attendu ici : la frontiere repond elle-meme a la poignee TCP sans jamais relayer
    # (verifie a la destination le 2026-08-09 — `idm-01` et `collab-01` ne voyaient
    # aucune connexion pendant que le poste en tenait une « etablie »). Ce que le controle
    # garantit, c'est que la LIVRAISON n'est pas simulee. S'il livre des donnees, quelque
    # chose parle a la place des destinations et plus aucun verdict n'a de sens.
    if controle["verdict"] == "LIVRE":
        json.dump({"valide": False, "controle": controle, "resultats": []}, sys.stdout)
        return 0

    def _un(cible) -> dict:
        hote, port = str(cible[0]), int(cible[1])
        return {"hote": hote, "port": port,
                "etiquette": cible[2] if len(cible) > 2 else "",
                **_sonder(hote, port)}

    liste = entree.get("cibles") or []
    with ThreadPoolExecutor(max_workers=PARALLELE) as pool:
        resultats = list(pool.map(_un, liste))
    json.dump({"valide": True, "controle": controle, "resultats": resultats}, sys.stdout)
    return 0


if __name__ == "__main__":
    sys.exit(main())