Set-OPS-Public/scripts/devis_underlay.py

#!/usr/bin/env python3
"""Devis de l'UNDERLAY — ce que le fichier declare, confronte a ce qui existe.

`underlay.yml` decrit le monde PHYSIQUE d'un site : commutateurs, hyperviseurs,
frontiere, reseaux de transport. Rien ne verifiait qu'il decrive encore quelque chose.

POURQUOI CE DEVIS EXISTE (mesure du 2026-08-22). Le fichier de Chezlepro declarait un
plan de gestion `10.0.0.0/24` avec les hyperviseurs en `10.0.0.41/43/47`. Mesure : ni
`10.0.0.1`, ni `10.0.0.41` ne repondent, et les hyperviseurs vivent en `192.168.11.x`. La
frontiere, elle, avait deja migre vers `10.17.0.1`. Le fichier decrivait un monde disparu.

CE QUE CA COUTAIT, et ce n'etait pas theorique : `devis_opnsense` derive l'interface d'une
regle de l'ATTACHEMENT REEL de sa source, et cet attachement se lit ici. Un plan
d'administration absent du fichier est classe « distant », et sa regle atterrit sur `wan`
ou elle ne peut JAMAIS correspondre. On aurait pose 89 objets sur la frontiere pour se
voir refuser quand meme.

CE QU'IL INTERROGE : l'API du cluster pour les adresses REELLES des noeuds, et une sonde
TCP pour ce qui repond. N'ECRIT RIEN — ni sur le cluster, ni dans le fichier.

  python3 scripts/devis_underlay.py             # tableau lisible
  python3 scripts/devis_underlay.py --verifier  # code de sortie (0 = le fichier dit vrai)
"""
from __future__ import annotations

import argparse
import ipaddress
import socket
import sys
from pathlib import Path

RACINE = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(RACINE / "scripts"))

import underlay as underlay_mod  # noqa: E402
from proxmox_api import Cluster  # noqa: E402

PORTS_SONDE = (22, 8006, 443, 80)


def repond(adresse: str, ports: tuple[int, ...] = PORTS_SONDE, delai: float = 1.5) -> int | None:
    """Le premier port qui accepte une connexion, ou None. Sonde TCP, jamais ICMP.

    Un ping refuse ne prouve rien — beaucoup d'equipements le filtrent par defaut. Une
    poignee TCP acceptee prouve qu'il y a quelqu'un.
    """
    for port in ports:
        try:
            with socket.create_connection((adresse, port), timeout=delai):
                return port
        except OSError:
            continue
    return None


def adresses_reelles(api: Cluster, noeuds: list[str]) -> dict[str, list[str]]:
    """Les adresses que PORTENT reellement les noeuds, vues par l'API du cluster."""
    trouve: dict[str, list[str]] = {}
    for n in noeuds:
        rep = api(f"/nodes/{n}/network")
        if not isinstance(rep, list):
            continue
        adresses = []
        for i in rep:
            cidr = i.get("cidr") or i.get("address")
            if cidr and not str(cidr).startswith("127."):
                adresses.append(str(cidr))
        trouve[n] = sorted(adresses)
    return trouve


def constater() -> tuple[list[dict], list[str]]:
    """Confronte chaque reseau et chaque hote declares a ce qui repond."""
    u = underlay_mod.charger()
    if u is None:
        raise SystemExit("Aucun underlay monte (underlay.yml absent) — rien a confronter.")

    lignes: list[dict] = []
    notes: list[str] = []

    # 2. Les noeuds du cluster : ou sont-ils VRAIMENT ? (l'AUTORITE)
    try:
        api, _ = Cluster.depuis_hebergeur()
        rep = api("/nodes")
        noeuds = sorted(n.get("node") for n in rep) if isinstance(rep, list) else []
        reelles = adresses_reelles(api, noeuds)
    except Exception as e:  # noqa: BLE001 — l'API est un tiers ; on rend la cause lisible
        notes.append(f"Cluster injoignable ({str(e)[:80]}) : adresses reelles non relevees.")
        reelles = {}

    portees = {a.split("/")[0] for ads in reelles.values() for a in ads}

    # 1. Les hotes declares : le CLUSTER porte-t-il cette adresse ?
    #
    # L'autorite est l'API, pas ma sonde TCP. Le premier jet de ce devis declarait « MUET »
    # tout ce qu'il ne joignait pas — et disait donc faux de `10.0.4.41`, une adresse bien
    # reelle sur un reseau de CHEMIN, que rien ne joint jamais de l'exterieur par
    # construction (D-78). Un instrument qui confond « absent » et « pas joignable d'ici »
    # accuse le monde de ce qu'il ne sait pas voir. La sonde TCP reste, en second rang :
    # elle dit « joignable d'ici », ce qui est une autre question, utile pour le plan
    # d'administration.
    for h in underlay_mod.hotes(u):
        ip = str(h.get("ip") or "")
        if not ip:
            continue
        # L'AUTORITE DEPEND DU ROLE, et c'est la deuxieme fois que ce devis l'apprend :
        # l'API de Proxmox connait ses hyperviseurs, et EUX SEULS. Un commutateur ou une
        # frontiere n'y figurent pas — les declarer « portes par personne » parce que le
        # cluster les ignore serait accuser le monde de ce que l'instrument ne voit pas.
        role = str(h.get("role") or "")
        port = repond(ip)
        if role == "hyperviseur":
            if not reelles:
                verdict = "cluster injoignable — non verifie"
            elif ip in portees:
                verdict = f"portee par le cluster{f', joignable (tcp/{port})' if port else ''}"
            else:
                verdict = "PORTEE PAR AUCUN NOEUD"
        else:
            verdict = (f"repond (tcp/{port})" if port
                       else "non joignable d'ici (equipement hors cluster)")
        lignes.append({"objet": f"{h.get('nom','?')} ({h.get('reseau','?')})",
                       "declare": ip, "verdict": verdict})

    declarees = {str(h.get("ip")) for h in underlay_mod.hotes(u) if h.get("ip")}
    for noeud, adresses in sorted(reelles.items()):
        nues = [a.split("/")[0] for a in adresses]
        connues = [a for a in nues if a in declarees]
        lignes.append({"objet": f"{noeud} — adresses reelles",
                       "declare": ", ".join(a for a in nues if not a.startswith("169.254"))[:60],
                       "verdict": "declaree" if connues else "AUCUNE N'EST DECLAREE"})

    # 3. Les reseaux declares : au moins une adresse reelle y tombe-t-elle ?
    toutes_reelles = [a.split("/")[0] for ads in reelles.values() for a in ads]
    for r in underlay_mod.reseaux(u):
        try:
            reseau = ipaddress.ip_network(str(r.get("sous_reseau")), strict=False)
        except ValueError:
            continue
        dedans = [a for a in toutes_reelles
                  if ipaddress.ip_address(a) in reseau] if toutes_reelles else []
        if not toutes_reelles:
            verdict = "cluster injoignable — non verifie"
        elif dedans:
            verdict = f"occupe ({len(dedans)} adresse(s) reelle(s))"
        elif repond(str(r.get("passerelle") or "")) is not None:
            verdict = "passerelle repond, aucun noeud dessus"
        else:
            verdict = "PERSONNE"
        lignes.append({"objet": f"reseau {r.get('nom')}",
                       "declare": str(r.get("sous_reseau")), "verdict": verdict})

    # 4. L'inverse, et c'est celui qu'on oublie : ce que les noeuds PORTENT et que
    #    l'underlay ne declare nulle part. Un reseau non declare est un reseau que le
    #    moteur ne sait pas classer — donc dont les regles atterrissent au mauvais endroit.
    reseaux_declares = []
    for r in underlay_mod.reseaux(u):
        try:
            reseaux_declares.append(ipaddress.ip_network(str(r.get("sous_reseau")), strict=False))
        except ValueError:
            continue
    inconnues = sorted({a for a in toutes_reelles
                        if not a.startswith(("127.", "169.254."))
                        and not any(ipaddress.ip_address(a) in r for r in reseaux_declares)})
    for a in inconnues:
        lignes.append({"objet": "porte par le cluster", "declare": a,
                       "verdict": "NON DECLARE dans underlay.yml"})
    return lignes, notes


def afficher(lignes: list[dict], notes: list[str]) -> bool:
    print(f"Devis de l'underlay — source : {underlay_mod.chemin()}\n")
    print(f"  {'OBJET':<34} {'DECLARE':<40} VERDICT")
    faux = 0
    for l in lignes:
        alerte = l["verdict"].isupper() or l["verdict"] in ("MUET", "PERSONNE")
        faux += 1 if alerte else 0
        print(f"  {l['objet']:<34} {l['declare']:<40} {l['verdict']}")
    for n in notes:
        print(f"\n  /!\\ {n}")
    print("\n" + (f"ECART : {faux} ligne(s) que le fichier declare et que rien n'occupe. "
                  f"L'underlay decrit un monde qui n'est plus la."
                  if faux else
                  "CONFORME : tout ce que l'underlay declare existe et repond."))
    return faux == 0


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    ap.add_argument("--verifier", action="store_true", help="code de sortie seulement")
    a = ap.parse_args()
    lignes, notes = constater()
    if a.verifier:
        faux = [l for l in lignes if l["verdict"].isupper()
                or l["verdict"] in ("MUET", "PERSONNE")]
        for l in faux:
            print(f"underlay : {l['objet']} declare « {l['declare']} » — {l['verdict']}",
                  file=sys.stderr)
        return 1 if faux else 0
    return 0 if afficher(lignes, notes) else 1


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