Set-OPS-Public/scripts/eprouver_parefeu.py

#!/usr/bin/env python3
"""Activer le pare-feu Proxmox d'UNE VM de locataire, en prouvant qu'il ne coupe rien.

POURQUOI UNE PROCEDURE, ET PAS SEULEMENT `proxmox-fw-appliquer` (2026-09-28). Aucune des 26 VM
des locataires n'avait son pare-feu actif : les reconstructions clonent sans ses options.
Tout activer d'un coup ouvrait 26 pannes possibles, et chaque defaut ne s'est montre qu'a
l'activation d'UNE VM : le ping de supervision qu'aucune regle ne portait (l'ICMP etait saute
par le devis), le tunnel d'administration du locataire que nftables admettait et Proxmox non.
Une VM reconstruite perd de nouveau ses options : cette procedure servira encore.

CE QU'ELLE FAIT, POUR UNE VM :
  1. MATRICE tiree du DEVIS lui-meme : pour chaque regle des groupes de la VM, CHAQUE membre
     de la source autorisee tente VM:port. On teste ce qui est promis, pas ce qu'on croit ;
  2. OBSERVATION : les connexions entrantes ETABLIES sur la VM (trois releves), confrontees
     aux regles. La matrice ne voit que le DECLARE ; un flux reel non declare la passerait,
     puis serait coupe. Un seul flux hors regles : REFUS d'activer ;
  3. matrice AVANT. Un echec deja present : REFUS — il faut le comprendre d'abord. Sauf si
     rien n'ECOUTE sur ce port hors de 127.0.0.1 : « sans objet » (un nginx lie en local
     derriere une passerelle SSO, un frontal sans site) ;
  4. ACTIVATION par le runner du site (seul a joindre l'API du cluster) : `--vm <vmid>` ;
  5. matrice APRES, comparee ; puis sondes de sante relancees et critiques d'Icinga.

    python3 scripts/eprouver_parefeu.py --instance OPS-X --hote idm-01 --plan
    python3 scripts/eprouver_parefeu.py --instance OPS-X --hote idm-01 --activer

LIMITE CONNUE : un flux UDP est teste en TCP sur le meme port (la matrice ne sait pas ouvrir
une « connexion » UDP). Pour un resolveur, verifier une vraie resolution (`dig +notcp`).
"""
from __future__ import annotations

import argparse
import ipaddress
import json
import os
import re
import subprocess
import sys
import time
from pathlib import Path

RACINE = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(RACINE / "scripts"))

from devis_reseau import DOSSIER_INSTANCES, admin_avec_tunnel  # noqa: E402


def _inventaire(instance: str) -> Path:
    return DOSSIER_INSTANCES / instance / "inventories" / "principal" / "hosts.yml"


def _ip_par_hote(instance: str) -> dict[str, str]:
    import yaml
    table: dict[str, str] = {}

    def w(n):
        if isinstance(n, dict):
            for k, v in n.items():
                if k == "hosts" and isinstance(v, dict):
                    for h, hv in v.items():
                        if isinstance(hv, dict) and hv.get("ansible_host"):
                            table[h] = hv["ansible_host"]
                else:
                    w(v)
    w(yaml.safe_load(_inventaire(instance).read_text(encoding="utf-8")))
    return table


def _ansible(instance: str, hotes: str, commande: str) -> dict[str, str]:
    """{hote: sortie} d'une commande shell jouee en root sur des hotes du locataire."""
    r = subprocess.run(["ansible", "-i", str(_inventaire(instance)), hotes, "-b", "-m", "shell",
                        "-a", commande], capture_output=True, text=True, cwd=RACINE)
    sorties, hote = {}, None
    for ligne in r.stdout.splitlines():
        m = re.match(r"^(\S+) \| \w+ \| rc=\d+ >>$", ligne)
        if m:
            hote = m.group(1)
            sorties[hote] = ""
        elif hote:
            sorties[hote] += ligne + "\n"
    return sorties


def matrice(instance: str, hote: str) -> tuple[list[dict], int, dict[str, list]]:
    env = dict(os.environ, SETOPS_INSTANCE=str(DOSSIER_INSTANCES / instance))
    r = subprocess.run([sys.executable, "scripts/devis_proxmox_fw.py", "--json"],
                       capture_output=True, text=True, cwd=RACINE, env=env)
    if r.returncode != 0:
        raise SystemExit("Le devis ne se genere pas :\n" + r.stderr[:400])
    devis = json.loads(r.stdout)
    ips = _ip_par_hote(instance)
    h_par_ip = {ip: h for h, ip in ips.items()}
    # LE BLOC DE CE LOCATAIRE, PAS LE PREMIER QUI PORTE CE NOM. Les locataires partagent
    # leurs noms d'hotes (`ops-01` existe chez chacun) : prendre la premiere affectation
    # venue visait la VM d'un autre — constate au premier essai de ce script.
    for b in (x for x in devis["blocs"] if x.get("tenant") == instance):
        a = next((x for x in b["affectations"] if x["hote"] == hote), None)
        if a is None:
            continue
        groupes = {g["nom"]: g for g in b["groupes"]}
        tests = []
        # LES MEMBRES QUI SONT DES RESEAUX, PAS DES MACHINES (2026-09-30). On ne peut pas les
        # TESTER (aucune machine a qui demander d'ouvrir la connexion), mais Proxmox les
        # applique : `t23-admin` porte la zone d'administration du site. Les ignorer faisait
        # prendre le navigateur de l'exploitant, sur l'edge en 443, pour un flux « hors des
        # regles » — et `--flotte` s'arretait. `!reseau` = exclusion (`nomatch`).
        reseaux: dict[str, list] = {}
        for gn in a["groupes"]:
            for regle in groupes[gn]["regles"]:
                membres = b["ipsets"].get(regle["source"].lstrip("+"), {}).get("membres", [])
                port = str(regle.get("dport") or regle.get("icmp_type"))
                pos = [m for m in membres if "/" in str(m) and not str(m).startswith("!")]
                neg = [str(m)[1:] for m in membres if str(m).startswith("!")]
                if pos:
                    reseaux.setdefault(port, []).append(
                        ([ipaddress.ip_network(x, strict=False) for x in pos],
                         [ipaddress.ip_network(x, strict=False) for x in neg]))
                for m in membres:
                    src = h_par_ip.get(m)
                    if src and src != hote:
                        tests.append({"src": src, "ip": ips[hote], "proto": regle["proto"],
                                      "port": str(regle.get("dport") or regle.get("icmp_type")),
                                      "groupe": gn})
        return tests, int(a["vmid"]), reseaux
    raise SystemExit(f"REFUS : « {hote} » n'a aucune affectation au devis de {instance}.")


def jouer(instance: str, hote: str, tests: list[dict]) -> dict[str, str]:
    par_src: dict[str, list[dict]] = {}
    for t in tests:
        par_src.setdefault(t["src"], []).append(t)
    res: dict[str, str] = {}
    for src, xs in par_src.items():
        lignes = []
        for t in xs:
            cle = f"{t['src']}>{hote}:{t['proto']}/{t['port']}"
            essai = (f"ping -c1 -W2 {t['ip']} >/dev/null 2>&1" if t["proto"] == "icmp"
                     else f"timeout 3 bash -c '</dev/tcp/{t['ip']}/{t['port']}' 2>/dev/null")
            lignes.append(f"{essai} && echo '{cle} ok' || echo '{cle} echec'")
        for sortie in _ansible(instance, src, "; ".join(lignes)).values():
            for ligne in sortie.splitlines():
                if ligne.endswith((" ok", " echec")):
                    k, v = ligne.rsplit(" ", 1)
                    res[k] = v
    return res


def ecoutes(instance: str, hote: str) -> set[str]:
    """Ports ecoutes HORS de la boucle locale : ce qu'un voisin peut atteindre."""
    sortie = _ansible(instance, hote, "ss -Hltn").get(hote, "")
    ports = set()
    for ligne in sortie.splitlines():
        champs = ligne.split()
        if len(champs) >= 4:
            adr, _, port = champs[3].rpartition(":")
            if not adr.startswith(("127.", "[::1]")):
                ports.add(port)
    return ports


def observer(instance: str, hote: str, tests: list[dict],
             reseaux: dict[str, list] | None = None) -> list[str]:
    """Les connexions entrantes etablies qu'AUCUNE regle ne couvre."""
    cmd = ('L=" $(ss -Hltn | awk \'{n=split($4,a,":"); print a[n]}\' | sort -u | tr "\\n" " ") "; '
           'for i in 1 2 3; do ss -Hnt state established | awk -v L="$L" '
           '\'{n=split($3,a,":"); lp=a[n]; m=split($4,b,":"); p=b[1]; for(k=2;k<m;k++) p=p":"b[k]; '
           'if (index(L," " lp " ")) print p" "lp}\'; sleep 5; done | sort -u')
    sortie = _ansible(instance, hote, cmd).get(hote, "")
    ips = _ip_par_hote(instance)
    h_par_ip = {ip: h for h, ip in ips.items()}
    permis: dict[str, set[str]] = {}
    for t in tests:
        permis.setdefault(t["port"], set()).add(t["src"])
    admin = [ipaddress.ip_network(n) for n in admin_avec_tunnel(instance)]
    hors = []
    for ligne in sortie.splitlines():
        champs = ligne.split()
        if len(champs) != 2:
            continue
        ip = champs[0].strip("[]").replace("::ffff:", "")
        port = champs[1]
        if ip.startswith("127.") or ip == "::1":
            continue
        src = h_par_ip.get(ip, ip)
        if src == hote:
            continue      # vers soi-meme par sa propre adresse : `lo`, pas le pont Proxmox
        if src in permis.get(port, set()):
            continue
        if port == "22" and any(ipaddress.ip_address(ip) in n for n in admin):
            continue
        adr = ipaddress.ip_address(ip)
        if any(any(adr in n for n in pos) and not any(adr in n for n in neg)
               for pos, neg in (reseaux or {}).get(port, [])):
            continue
        hors.append(f"{src} -> {hote}:{port}")
    return sorted(set(hors))


def activer(instance: str, vmid: int, args) -> int:
    distante = (f"cd {args.racine_runner}/Set-OPS-public && export PATH={args.racine_runner}/venv/bin:$PATH "
                f"&& SETOPS_INSTANCE={args.racine_runner}/{instance} make proxmox-fw-appliquer "
                f"CONFIRMER=true ARGS='--vm {vmid}'")
    cmd = ["ssh", "-o", "BatchMode=yes", "-J", args.rebond, args.runner,
           f"sudo -u setops bash -lc {json.dumps(distante)}"]
    r = subprocess.run(cmd, capture_output=True, text=True)
    for ligne in r.stdout.splitlines():
        if re.search(r"~ VM|ECHEC|reconcilie|INCOMPLETE|Rien a faire", ligne):
            print("   ", ligne.strip())
    return 0 if ("reconcilie" in r.stdout or "Rien a faire" in r.stdout) and "ECHEC" not in r.stdout else 1


def critiques_icinga(instance: str) -> list[str]:
    _ansible(instance, "all", "systemctl start setops-sante.service; true")
    cmd = ("P=$(sed -n '/ApiUser \"root\"/,/}/p' /etc/icinga2/conf.d/api-users.conf | "
           "grep -oP 'password = \"\\K[^\"]+'); "
           "curl -sk -u \"root:$P\" -X POST -H 'Accept: application/json' "
           "https://localhost:5665/v1/actions/reschedule-check "
           "-d '{\"type\":\"Service\",\"filter\":\"service.name==\\\"ping4\\\"\",\"force\":true}' >/dev/null; "
           "sleep 20; curl -sk -u \"root:$P\" "
           "'https://localhost:5665/v1/objects/services?filter=service.state==2&attrs=last_check_result'")
    sortie = next(iter(_ansible(instance, "serveur_icinga", cmd).values()), "{}")
    try:
        return [x["name"] for x in json.loads(sortie[sortie.index("{"):])["results"]]
    except (ValueError, KeyError):
        return ["(Icinga illisible)"]


def vmids(instance: str) -> dict[str, int]:
    """{hote: vmid} du devis de ce locataire — la meme source que l'activation."""
    env = dict(os.environ, SETOPS_INSTANCE=str(DOSSIER_INSTANCES / instance))
    r = subprocess.run([sys.executable, "scripts/devis_proxmox_fw.py", "--json"],
                       capture_output=True, text=True, cwd=RACINE, env=env)
    if r.returncode != 0:
        raise SystemExit("Le devis ne se genere pas :\n" + r.stderr[:400])
    table = {}
    for b in (x for x in json.loads(r.stdout)["blocs"] if x.get("tenant") == instance):
        for a in b["affectations"]:
            table[a["hote"]] = int(a["vmid"])
    return table


def etat_sonde(instance: str, nom: str) -> dict[str, tuple[int, str, float]]:
    """{hote: (etat, texte, fin du dernier rapport)} du service `nom`, lu dans Icinga."""
    cmd = ("P=$(sed -n '/ApiUser \"root\"/,/}/p' /etc/icinga2/conf.d/api-users.conf | "
           "grep -oP 'password = \"\\K[^\"]+'); "
           "curl -sk -u \"root:$P\" -H 'Accept: application/json' "
           f"'https://localhost:5665/v1/objects/services?filter=service.name==%22{nom}%22"
           "&attrs=host_name&attrs=state&attrs=last_check_result'")
    sortie = next(iter(_ansible(instance, "serveur_icinga", cmd).values()), "{}")
    table = {}
    try:
        for x in json.loads(sortie[sortie.index("{"):])["results"]:
            a = x["attrs"]
            r = a.get("last_check_result") or {}
            table[a["host_name"]] = (int(a.get("state", 3)), str(r.get("output", "")),
                                     float(r.get("execution_end") or 0))
    except (ValueError, KeyError):
        pass
    return table


def activer_par_sondes(instance: str, args) -> int:
    """La flotte entiere d'un coup, jugee par la sonde `connectivite` (2026-09-30).

    La matrice VM par VM prenait une vingtaine de minutes en fin de reconstruction, pour une
    preuve valable a l'instant de l'activation seulement. Ici : chaque VM teste deja, chaque
    minute, les flux que le registre lui promet, et le dit a Icinga. On verifie que tout est
    sain AVANT, on active tout, on attend les rapports suivants, et seul ce qui CHANGE compte.
    Perdu, en connaissance de cause : l'isolement d'une faute a une seule VM — la sonde, elle,
    nomme le couple exact qui ne passe plus.
    """
    table = vmids(instance)
    avant = etat_sonde(instance, "connectivite")
    manquent = sorted(set(table) - set(avant))
    if manquent:
        print(f"  REFUS : aucune sonde `connectivite` dans Icinga pour {', '.join(manquent)} — "
              "redeployer `serveur_durci` et `client_sante`, puis `serveur_icinga`.")
        return 2
    deja = {h: v for h, v in avant.items() if h in table and v[0] == 2}
    if deja:
        print("  REFUS : des flux sont DEJA coupes, avant toute activation — les comprendre d'abord :")
        for h, (_, texte, _) in sorted(deja.items()):
            print(f"    {h} : {texte[:180]}")
        return 2
    print(f"  avant : {len(table)} VM, `connectivite` saine partout "
          f"({sum(1 for h in table if avant[h][0] == 1)} avertissement(s))")
    crit_avant = set(critiques_icinga(instance))
    if args.plan:
        print("  PLAN SEUL — rien n'a ete active.")
        return 0
    t0 = time.time()
    print(f"  activation de {len(table)} VM d'un coup a {time.strftime('%H:%M:%S')}")
    distante = (f"cd {args.racine_runner}/Set-OPS-public && export PATH={args.racine_runner}/venv/bin:$PATH "
                f"&& SETOPS_INSTANCE={args.racine_runner}/{instance} make proxmox-fw-appliquer "
                f"CONFIRMER=true ARGS='{' '.join(f'--vm {v}' for v in sorted(table.values()))}'")
    r = subprocess.run(["ssh", "-o", "BatchMode=yes", "-J", args.rebond, args.runner,
                        f"sudo -u setops bash -lc {json.dumps(distante)}"], capture_output=True, text=True)
    if not (("reconcilie" in r.stdout or "Rien a faire" in r.stdout) and "ECHEC" not in r.stdout):
        print("  ECHEC de l'activation :")
        for ligne in (r.stdout + r.stderr).splitlines()[-8:]:
            print("    " + ligne)
        return 3
    # On attend que CHAQUE VM ait rapporte APRES l'activation (rapport a la minute, `ttl` 3 min).
    limite = t0 + 300
    while True:
        time.sleep(20)
        apres = etat_sonde(instance, "connectivite")
        frais = [h for h in table if h in apres and apres[h][2] > t0 + 5]
        print(f"  {time.strftime('%H:%M:%S')} : {len(frais)}/{len(table)} VM ont rapporte depuis l'activation")
        if len(frais) == len(table) or time.time() > limite:
            break
    absents = sorted(set(table) - set(frais))
    coupes = {h: apres[h] for h in frais if apres[h][0] == 2}
    for h, (_, texte, _) in sorted(coupes.items()):
        print(f"    COUPE {h} : {texte[:200]}")
    crit = sorted(set(critiques_icinga(instance)) - crit_avant)
    print(f"  Icinga : {len(crit)} critique(s) apparu(s)" + (f" — {', '.join(crit)}" if crit else ""))
    if absents:
        print(f"  SANS NOUVELLE de {', '.join(absents)} apres 5 min — muette, ou coupee d'Icinga.")
    if coupes or crit or absents:
        print("  RETOUR ARRIERE d'une VM : `enable=0` sur /nodes/<noeud>/qemu/<vmid>/firewall/options "
              f"(vmid : {', '.join(f'{h}={table[h]}' for h in sorted(set(coupes) | set(absents)))})")
        return 4
    print(f"  {len(table)}/{len(table)} VM : pare-feu Proxmox actif, `connectivite` saine partout.")
    return 0


def ordre_flotte(instance: str) -> list[str]:
    """Toutes les VM du locataire, son RUNNER EN DERNIER.

    C'est depuis le poste qu'on joue la matrice, mais c'est le runner qui conduit le reste
    de la vie du locataire : s'il devait etre coupe, qu'il le soit apres que tout le reste
    a ete prouve — et qu'on le voie seul en cause.
    """
    import yaml
    enfants = (yaml.safe_load(_inventaire(instance).read_text(encoding="utf-8"))
               .get("all", {}).get("children", {}))
    runners = set()
    for g in ("serveur_ops_tenant", "serveur_ops"):
        runners |= set(((enfants.get(g) or {}).get("hosts") or {}))
    hotes = sorted(_ip_par_hote(instance))
    return [h for h in hotes if h not in runners] + [h for h in hotes if h in runners]


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    ap.add_argument("--instance", required=True, help="le locataire, ex. OPS-Technolibre")
    cible = ap.add_mutually_exclusive_group(required=True)
    cible.add_argument("--hote", help="la VM, ex. idm-01")
    cible.add_argument("--flotte", action="store_true",
                       help="toutes les VM, une a une, le runner en dernier ; arret au premier refus")
    ap.add_argument("--sondes", action="store_true",
                    help="avec --flotte : tout activer d'un coup, juge par la sonde `connectivite` d'Icinga")
    mode = ap.add_mutually_exclusive_group(required=True)
    mode.add_argument("--plan", action="store_true", help="verifier, sans rien activer")
    mode.add_argument("--activer", action="store_true", help="verifier, activer, reverifier")
    ap.add_argument("--runner", default="ansible@10.37.31.11", help="runner du site (API du cluster)")
    ap.add_argument("--rebond", default="ansible@10.37.0.1", help="rebond vers le runner")
    ap.add_argument("--racine-runner", default="/opt/setops")
    args = ap.parse_args()
    if args.hote:
        return eprouver(args.instance, args.hote, args)
    if args.sondes:
        return activer_par_sondes(args.instance, args)
    # UNE A UNE, ET ON S'ARRETE AU PREMIER REFUS (2026-09-30). C'etait une boucle tapee a
    # la main apres chaque reconstruction ; la regle qui la rend sure — ne pas continuer
    # apres un ecart — doit vivre avec elle, pas dans la memoire de celui qui la tape.
    hotes = ordre_flotte(args.instance)
    for n, h in enumerate(hotes, 1):
        rc = eprouver(args.instance, h, args)
        if rc != 0:
            print(f"\nARRET sur {h} (code {rc}) : {n - 1}/{len(hotes)} VM traitees avant lui. "
                  "Rien d'autre n'a ete touche.")
            return rc
    print(f"\n{len(hotes)}/{len(hotes)} VM : pare-feu Proxmox "
          + ("verifiable" if args.plan else "actif") + ", aucun flux perdu.")
    return 0


def eprouver(I: str, H: str, args: argparse.Namespace) -> int:
    """La procedure pour UNE VM (voir l'en-tete). 0 = conforme."""
    tests, vmid, reseaux = matrice(I, H)
    print(f"=== {H} ({I}, VM {vmid}) — {len(tests)} test(s) tires du devis")
    hors = observer(I, H, tests, reseaux)
    print(f"  observation : {len(hors)} flux etabli(s) hors des regles")
    for h in hors:
        print(f"    HORS : {h}")
    avant = jouer(I, H, tests)
    ecoute = ecoutes(I, H)
    def _sans_objet(cle: str) -> bool:
        proto, port = cle.rsplit(":", 1)[1].split("/", 1)
        return proto != "icmp" and port not in ecoute
    sans_objet = {k for k, v in avant.items() if v == "echec" and _sans_objet(k)}
    echecs = sorted(k for k, v in avant.items() if v == "echec" and k not in sans_objet)
    print(f"  avant : {sum(v == 'ok' for v in avant.values())}/{len(avant)} ok"
          + (f", {len(sans_objet)} sans objet (rien n'ecoute hors de 127.0.0.1)" if sans_objet else ""))
    for k in sorted(sans_objet):
        print(f"    sans objet : {k}")
    for k in echecs:
        print(f"    DEJA EN ECHEC : {k}")
    if args.plan:
        print("  PLAN SEUL — rien n'a ete active.")
        return 1 if (hors or echecs) else 0
    if hors or echecs:
        print("  REFUS D'ACTIVER : un flux reel hors des regles, ou un echec a comprendre d'abord.")
        return 2
    # LES CRITIQUES D'AVANT NE SONT PAS LES NOTRES (2026-09-30). Tout critique d'Icinga
    # arretait la procedure — y compris ceux qui existaient avant l'activation : a la
    # reconstruction de Chezlepro, des `restauration` rouges (rien a voir avec le pare-feu)
    # l'ont arretee deux fois. On releve donc AVANT, et seul ce qui apparait apres compte.
    crit_avant = set(critiques_icinga(I))
    if activer(I, vmid, args) != 0:
        print("  ECHEC de l'activation.")
        return 3
    print(f"  active a {time.strftime('%H:%M:%S')}")
    apres = jouer(I, H, tests)
    change = sorted(k for k in avant if avant[k] != apres.get(k))
    print(f"  apres : {sum(v == 'ok' for v in apres.values())}/{len(apres)} ok ; "
          f"changements : {', '.join(change) or 'aucun'}")
    crit_apres = critiques_icinga(I)
    crit = sorted(set(crit_apres) - crit_avant)
    print(f"  Icinga : {len(crit)} critique(s) apparu(s)" + (f" — {', '.join(crit)}" if crit else "")
          + (f" ; {len(crit_avant)} deja present(s) avant, sans rapport avec le pare-feu"
             if crit_avant else ""))
    if change:
        print(f"  RETOUR ARRIERE : remettre `enable=0` sur la VM {vmid} "
              f"(/nodes/<noeud>/qemu/{vmid}/firewall/options).")
    return 4 if (change or crit) else 0


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