Set-OPS-Public/scripts/resoudre_flux.py

#!/usr/bin/env python3
"""Resolveur des flux reseau Set-OPS : meta/flux.yml -> registre d'audit + apercus nftables.

Agrege les `roles/<role>/meta/flux.yml`, resout les `pair`, et produit :
  - docs/registre-flux.md : la MATRICE D'AUDIT (niveau moteur, role->role : port,
    sens, chiffrement, raison). Deterministe, sans donnee d'instance.
  - des APERCUS nftables par hote (IP resolues, `ip saddr` = moindre privilege,
    policy drop) sous <instance>/flux-genere/<hote>.nft. Fichiers GENERES et
    inspectables — NON actives (l'activation reste un geste dedie, teste par noeud).

NON destructif : n'active aucun pare-feu. Voir docs/flux-conception.md.

Usage :
  python3 scripts/resoudre_flux.py registre    # (re)genere docs/registre-flux.md
  python3 scripts/resoudre_flux.py nftables     # apercus par hote (requiert l'inventaire)
  python3 scripts/resoudre_flux.py verifier      # valide le schema de tous les flux.yml
"""

from __future__ import annotations

import glob
import os
import sys
from pathlib import Path

import yaml

from inventory_rules import instance_courante, est_groupe_operationnel

RACINE = Path(__file__).resolve().parents[1]
ROLES = RACINE / "roles"
REGISTRE = RACINE / "docs" / "registre-flux.md"
INSTANCE = instance_courante()

SENS = {"ingress", "egress"}
# `icmp` n'a pas de port : le champ `port` porte alors le TYPE (ex. `frag-needed`).
# Sans lui, la decouverte de MTU de chemin ne peut pas etre DECLAREE — et une bordure
# en default-deny la casse en silence des que l'overlay descend sous 1500.
PROTO = {"tcp", "udp", "icmp"}
CHIFFREMENT = {"tls-requis", "tls", "starttls", "ssh", "tls-cible", "clair", "n-a"}
# `admin` : les RESEAUX d'administration (intrant `nftables_admin_ssh`), pas des
# hotes. C'est le troisieme chemin d'arrivee, distinct de `flotte` (le tenant) et
# d'`externe` (Internet) — l'exploitant n'est ni l'un ni l'autre.
# `voisins_site` : les autres tenants FEDERES de la meme fabric. Ni `flotte` (mon
# ecosysteme), ni `externe` (l'Internet) — un quatrieme chemin d'arrivee, celui du
# VOISINAGE.
#
# POURQUOI CE MOT EXISTE (2026-08-24). Le cache d'artefacts se CHAINE : chaque ecosysteme
# garde le sien, et celui-ci prend le cache de l'hebergeur comme amont. Debian n'est alors
# telecharge qu'une fois pour tout le site. Sans ce mot, la seule facon de declarer ce flux
# aurait ete `ingress` + `externe` — qui signifie « depuis l'Internet » et aurait PUBLIE le
# cache au monde entier.
#
# Il rend des CIDR (les supernets des voisins), pas des adresses d'hotes : il est donc
# traite a part, comme `admin`.
# `fabric` — LE MATERIEL DE L'HEBERGEUR : hyperviseurs, frontiere, commutateurs.
#
# Il a fallu ce mot le 2026-08-25. Le runner de SITE declarait son API Proxmox en
# `externe`, et « externe » se rend par « tout sauf les espaces prives ». Or les
# hyperviseurs SONT en RFC 1918 : la regle sortante les aurait exclus, tout en ayant
# l'air d'ouvrir le flux. Un flux qui a l'air ouvert et qui ne l'est pas est pire qu'un
# flux ferme — il ne se cherche pas.
#
# Ce n'est pas non plus `flotte` (les machines d'un ecosysteme) ni `voisins_site` (les
# tenants d'a cote) : c'est le SOCLE sur lequel les uns et les autres reposent. Le seul
# a s'y adresser est le runner du site, et c'est tout son objet — materialiser.
MOTS_PAIR = {"edge", "flotte", "externe", "localhost", "expositions", "derive", "admin",
             "voisins_site", "fabric", "runner_site"}
GROUPE_EDGE = "serveur_nginx"


class ErreurFlux(Exception):
    pass


def charger_flux() -> dict[str, list[dict]]:
    """{role: [flux, ...]} pour chaque role portant un meta/flux.yml."""
    resultat: dict[str, list[dict]] = {}
    for chemin in sorted(glob.glob(str(ROLES / "*" / "meta" / "flux.yml"))):
        role = Path(chemin).parents[1].name
        data = yaml.safe_load(Path(chemin).read_text(encoding="utf-8")) or {}
        flux = data.get("flux")
        if not isinstance(flux, list):
            raise ErreurFlux(f"{chemin} : cle 'flux' (liste) attendue.")
        resultat[role] = flux
    return resultat


def _pairs(flux: dict) -> list:
    pair = flux.get("pair")
    return pair if isinstance(pair, list) else [pair]


def valider(flux_par_role: dict[str, list[dict]]) -> tuple[int, int]:
    """Valide le schema + la coherence de matrice. Leve ErreurFlux sinon."""
    roles_existants = {p.name for p in ROLES.iterdir() if p.is_dir()}
    erreurs: list[str] = []
    ingress: dict[str, set] = {}
    egress: list[tuple[str, str, object]] = []
    total = 0
    for role, flux in flux_par_role.items():
        for i, fl in enumerate(flux):
            total += 1
            ref = f"{role}/meta/flux.yml[{i}]"
            for cle in ("sens", "port", "protocole", "pair", "chiffrement", "raison"):
                if cle not in fl:
                    erreurs.append(f"{ref} : cle manquante {cle}")
            if fl.get("sens") not in SENS:
                erreurs.append(f"{ref} : sens invalide {fl.get('sens')!r}")
            if fl.get("protocole") not in PROTO:
                erreurs.append(f"{ref} : protocole invalide {fl.get('protocole')!r}")
            if fl.get("chiffrement") not in CHIFFREMENT:
                erreurs.append(f"{ref} : chiffrement invalide {fl.get('chiffrement')!r}")
            for p in _pairs(fl):
                if p not in MOTS_PAIR and p not in roles_existants and not str(p).startswith(("serveur_", "client_")):
                    erreurs.append(f"{ref} : pair inconnu {p!r}")
            if fl.get("sens") == "ingress":
                ingress.setdefault(role, set()).add(fl.get("port"))
            elif fl.get("sens") == "egress":
                for p in _pairs(fl):
                    if str(p).startswith(("serveur_", "client_")):
                        egress.append((role, p, fl.get("port")))
    for role, pair, port in egress:
        if port not in ingress.get(pair, set()):
            erreurs.append(f"matrice : {role} -> {pair}:{port} sans ingress correspondant sur {pair}")
    if erreurs:
        raise ErreurFlux("\n  - ".join(["Flux incoherents :"] + erreurs))
    return len(flux_par_role), total


# --- Registre d'audit (niveau moteur, sans instance) ---------------------------

def _cle_port(port) -> tuple:
    """Cle de tri stable pour un port, qu'il soit numerique ou symbolique.

    Tous les ports ne sont pas des nombres : `derive` (serveur_nginx) et
    `frag-needed` (ICMP, serveur_debian) en sont. Comparer directement
    laisserait `int < str` lever une TypeError des que deux ports de types
    differents se retrouvent dans le meme sens pour un meme role.
    """
    if isinstance(port, list):
        port = port[0] if port else 0
    try:
        return (0, int(port), "")
    except (TypeError, ValueError):
        return (1, 0, str(port))


def generer_registre(flux_par_role: dict[str, list[dict]]) -> str:
    lignes = [
        "# Registre des flux réseau — matrice d'audit (GÉNÉRÉ)",
        "",
        "> Généré par `scripts/resoudre_flux.py` (`make flux`) depuis les `roles/*/meta/flux.yml`.",
        "> **Ne pas éditer à la main.** Matrice source→destination pour l'audit de sécurité et le",
        "> label de certification. `ingress` = le rôle écoute ; `egress` = le rôle se connecte.",
        "",
        "| Rôle (propriétaire) | Sens | Port | Proto | Pair | Chiffrement | Raison |",
        "| --- | --- | --- | --- | --- | --- | --- |",
    ]
    rangs = {s: i for i, s in enumerate(("ingress", "egress"))}
    for role in sorted(flux_par_role):
        for fl in sorted(flux_par_role[role], key=lambda f: (rangs.get(f.get("sens"), 9), _cle_port(f.get("port")))):
            pair = ", ".join(str(p) for p in _pairs(fl))
            port = fl.get("port")
            port = ", ".join(str(x) for x in port) if isinstance(port, list) else str(port)
            lignes.append(
                f"| `{role}` | {fl.get('sens')} | {port} | {fl.get('protocole')} "
                f"| {pair} | {fl.get('chiffrement')} | {fl.get('raison', '')} |"
            )
    # Synthese chiffrement
    compte: dict[str, int] = {}
    for flux in flux_par_role.values():
        for fl in flux:
            compte[fl.get("chiffrement")] = compte.get(fl.get("chiffrement"), 0) + 1
    lignes += ["", "## Synthèse chiffrement", ""]
    for chi in sorted(compte):
        lignes.append(f"- **{chi}** : {compte[chi]} flux")
    return "\n".join(lignes) + "\n"


# --- Apercus nftables par hote (requiert l'inventaire) -------------------------

def _inventaire() -> Path:
    for nom in ("principal", "production"):
        p = INSTANCE / "inventories" / nom / "hosts.yml"
        if p.exists():
            return p
    raise ErreurFlux(f"Inventaire introuvable sous {INSTANCE}/inventories/(principal|production).")


def _enfants(data: dict) -> dict:
    return data.get("all", {}).get("children", {})


def _hotes_du_groupe(data: dict, groupe: str) -> dict:
    return _enfants(data).get(groupe, {}).get("hosts", {}) or {}


def _sources_admin_ssh() -> list[str]:
    """CIDR d'administration SSH toujours autorises (intrant nftables_admin_ssh).

    Garde anti-lockout : le contexte d'ou l'on administre (controleur/VPN) doit
    rester joignable en SSH quelle que soit la resolution des flux.
    """
    dossier = _inventaire().parent / "group_vars" / "all"
    for fichier in sorted(dossier.glob("*.yml")):
        if "vault" in fichier.name:
            continue
        data = yaml.safe_load(fichier.read_text(encoding="utf-8")) or {}
        if isinstance(data, dict) and data.get("nftables_admin_ssh"):
            sources = data["nftables_admin_ssh"]
            return [str(s) for s in sources] if isinstance(sources, list) else [str(sources)]
    return []


def _ip_par_hote(data: dict) -> dict[str, str]:
    """{hote: ansible_host} — l'IP vit dans hotes_actifs ; l'appartenance ailleurs est null."""
    table: dict[str, str] = {}
    for membres in _enfants(data).values():
        for hote, v in (membres.get("hosts") or {}).items():
            if isinstance(v, dict) and v.get("ansible_host"):
                table[hote] = v["ansible_host"]
    return table


def _resoudre_sources(data: dict, pair, actifs_noms: set[str], ip_par_hote: dict[str, str]) -> list[str]:
    """pair -> IP sources concretes (pour une regle ingress). [] = pas de regle inter-noeud."""
    noms: set[str] = set()
    for p in (pair if isinstance(pair, list) else [pair]):
        if p in ("localhost", "externe", "expositions", "derive"):
            continue  # lo (auto), frontiere OPNsense, ou hors perimetre noeud
        if p in ("admin", "voisins_site", "runner_site"):
            # Aucun des trois n'est un groupe d'hotes DE CET ECOSYSTEME : ce sont des
            # sources qui lui sont EXTERIEURES. `admin` vient de l'intrant
            # `nftables_admin_ssh` ; `voisins_site` des supernets des autres tenants
            # federes ; `runner_site` du plan du SITE. Traites a part, plus bas.
            continue
        if p == "flotte":
            noms |= actifs_noms
        else:
            groupe = GROUPE_EDGE if p == "edge" else p
            noms |= {h for h in _hotes_du_groupe(data, groupe) if h in actifs_noms}
    srcs = {ip_par_hote[h] for h in noms if h in ip_par_hote}
    demandes = pair if isinstance(pair, list) else [pair]
    if "admin" in demandes:
        srcs |= set(_sources_admin_ssh())
    if "voisins_site" in demandes:
        srcs |= set(_supernets_voisins())
    if "runner_site" in demandes:
        srcs |= set(_adresses_runner_site())
    return sorted(srcs)


def _adresses_runner_site() -> list[str]:
    """L'adresse du runner du SITE — la source du flux d'insemination.

    DERIVEE DU PLAN DU SITE, jamais ecrite ici. Le site est le seul a savoir ou vit son
    runner, et il l'a deja deplace une fois (decoupage en zones du 2026-08-25). Une
    adresse recopiee ici aurait survecu a ce deplacement en silence, et le tenant se
    serait retrouve avec une regle qui n'autorise personne — la panne la plus muette de
    cette couche.

    Rend [] quand le depot de l'hebergeur n'est pas monte : un tenant clone sans lui
    degrade proprement, il ne casse pas.
    """
    try:
        import underlay as underlay_mod
        return underlay_mod.adresses_site_portant("serveur_ops_site")
    except Exception:
        return []


def _supernets_voisins() -> list[str]:
    """Les supernets des AUTRES tenants que CE SITE heberge.

    ON REUTILISE `devis_reseau.decouvrir_du_site()`, on n'en ecrit pas une deuxieme
    version. C'est la lecon de P41 appliquee ici : deux recensements de tenants finiraient
    par diverger, et la divergence se lirait « tout va bien ».

    Elle porte aussi la bonne semantique. « Voisin » veut dire *sur ma fabric*, pas
    *quelque part dans la federation* : un tenant pose sur un autre site n'est joignable
    par aucune regle de cette frontiere-ci, et lui ouvrir un flux serait un chemin qui ne
    correspondra jamais a un paquet.

    Elle porte enfin la regle du `federe` ABSENT = federe (devis_reseau, ligne 568). Ma
    premiere version lisait l'absence comme « non federe » et excluait Chezlepro et
    Technolibre en silence — leur nomenclature est anterieure a cette cle.
    """
    try:
        import devis_reseau
        from inventory_rules import supernet_de
        tenants = devis_reseau.decouvrir_du_site()
    except Exception:
        # Sans underlay designe (ex. `make flux` hors site), il n'y a pas de voisinage a
        # resoudre. Rendre [] plutot qu'echouer : l'absence de voisins n'est pas une faute.
        return []
    moi = INSTANCE.resolve().name
    return sorted({supernet_de(int(n["index"])) for nom, _p, n in tenants if nom != moi})


# Codes ICMP de destination inatteignable qu'un flux peut nommer dans `port`. L'ICMP n'a
# PAS de port : il a un type et un code. Emettre `icmp dport frag-needed` produit un jeu
# que `nft` refuse — et le service ne demarre plus du tout, donc l'hote perd sa barriere
# au lieu d'en gagner une. Constate le 2026-08-06 sur les 14 hotes a la fois.
CODES_ICMP = {
    "frag-needed": "icmp type destination-unreachable icmp code frag-needed",
}


def _selecteur_nft(protocole: str, port) -> str:
    """Selecteur nftables pour un flux : `tcp dport 22`, ou l'equivalent ICMP.

    Un `port` non numerique sur de l'ICMP designe un code, jamais un port. Un code
    inconnu est refuse ici plutot que de produire un fichier que `nft` rejettera —
    l'erreur se lit au moment de la generation, pas trois deploiements plus tard.
    """
    p = str(protocole).lower()
    if p in ("icmp", "icmpv6"):
        cle = str(port).strip()
        if cle in CODES_ICMP:
            return CODES_ICMP[cle]
        raise ErreurFlux(
            f"Flux ICMP avec `port: {port}` : l'ICMP n'a pas de port. Nommer un code "
            f"connu ({', '.join(sorted(CODES_ICMP))}) ou etendre CODES_ICMP."
        )
    return f"{p} dport {port}"


def generer_nftables(flux_par_role: dict[str, list[dict]]) -> list[Path]:
    data = yaml.safe_load(_inventaire().read_text(encoding="utf-8")) or {}
    actifs_noms = set(_hotes_du_groupe(data, "hotes_actifs"))
    ip_par_hote = _ip_par_hote(data)
    admin_ssh = _sources_admin_ssh()
    regles_admin = [
        f"    ip saddr {{ {cidr} }} tcp dport 22 accept  # administration (garde anti-lockout, intrant nftables_admin_ssh)"
        for cidr in admin_ssh
    ]
    sortie_dir = _inventaire().parent.parent.parent / "flux-genere"
    sortie_dir.mkdir(parents=True, exist_ok=True)
    ecrits: list[Path] = []
    for hote in sorted(actifs_noms):
        groupes = [g for g, membres in _enfants(data).items()
                   if hote in (membres.get("hosts") or {}) and est_groupe_operationnel(g)]
        regles: list[str] = list(regles_admin)
        for g in sorted(groupes):
            for fl in flux_par_role.get(g, []):
                if fl.get("sens") != "ingress":
                    continue
                ports = fl["port"] if isinstance(fl["port"], list) else [fl["port"]]
                srcs = _resoudre_sources(data, fl.get("pair"), actifs_noms, ip_par_hote)
                saddr = "" if not srcs else "ip saddr { " + ", ".join(srcs) + " } "
                local = "localhost" in (fl.get("pair") if isinstance(fl.get("pair"), list) else [fl.get("pair")])
                for port in ports:
                    if local and not srcs:
                        continue  # lo deja accepte
                    regles.append(f"    {saddr}{_selecteur_nft(fl['protocole'], port)} accept"
                                  f"  # {g}: {fl.get('raison','')}")
        contenu = _rendre_nft(hote, regles)
        chemin = sortie_dir / f"{hote}.nft"
        chemin.write_text(contenu, encoding="utf-8")
        ecrits.append(chemin)
    return ecrits


def _rendre_nft(hote: str, regles: list[str]) -> str:
    corps = "\n".join(regles) if regles else "    # (aucun flux ingress inter-noeud)"
    return (
        f"#!/usr/sbin/nft -f\n"
        f"# GENERE par scripts/resoudre_flux.py depuis les meta/flux.yml — hote {hote}.\n"
        f"# Deploye par le role `nftables_baseline` en tant que /etc/nftables.conf.\n\n"
        f"# Remplace UNIQUEMENT les tables setops (pas de `flush ruleset` : préserve les\n"
        f"# tables etrangeres — DNAT/forward de conteneurs, par exemple).\n"
        f"#\n"
        f"# `setops_filter` est le gabarit de REPLI, pose quand aucun ruleset resolu\n"
        f"# n'existe encore. Les deux fichiers sont des ALTERNATIVES, jamais des couches —\n"
        f"# mais `nft -f` AJOUTE sans purger, et le repli survivait donc a la bascule.\n"
        f"# Deux chaines `input` sur le meme hook, toutes deux en `policy drop` : le\n"
        f"# paquet traverse les DEUX, et seule l'intersection de leurs `accept` passait.\n"
        f"# Mesure du 2026-08-10 sur Technolibre : l'AC etait debout, sa regle 8443 posee\n"
        f"# et acceptante, l'ICMP passait en 0,15 ms — et le TCP expirait, arrete par le\n"
        f"# `policy drop` du repli. Chaque fichier retire donc desormais la table de\n"
        f"# l'autre, et la bascule converge dans les deux sens.\n"
        f"table inet setops_flux {{}}\n"
        f"delete table inet setops_flux\n"
        f"table inet setops_filter {{}}\n"
        f"delete table inet setops_filter\n\n"
        f"table inet setops_flux {{\n"
        f"  chain input {{\n"
        f"    type filter hook input priority 0; policy drop;\n\n"
        f"    iif \"lo\" accept\n"
        f"    ct state established,related accept\n"
        f"    ct state invalid drop\n"
        f"    ip protocol icmp accept\n"
        f"    ip6 nexthdr icmpv6 accept\n\n"
        f"{corps}\n"
        f"  }}\n\n"
        f"  chain forward {{\n"
        f"    type filter hook forward priority 0; policy drop;\n"
        f"    ct state established,related accept\n"
        f"    iifname \"docker0\" accept\n"
        f"    oifname \"docker0\" accept\n"
        f"  }}\n"
        f"  chain output {{ type filter hook output priority 0; policy accept; }}\n"
        f"}}\n"
    )


def main(argv: list[str]) -> int:
    commande = argv[0] if argv else "registre"
    try:
        flux = charger_flux()
        if commande == "verifier":
            r, t = valider(flux)
            # Chaque flux doit produire un selecteur nftables VALIDE. Le verifier ici, et
            # pas seulement a la generation, ferme le cas ou un role declare un flux que
            # personne ne porte encore : le devis passerait, et la panne arriverait le
            # jour ou un hote prend ce role. `nft -c` local demanderait des privileges
            # (netlink) ; construire le selecteur ne coute rien et attrape la meme faute.
            for role, liste in sorted(flux.items()):
                for fl in liste:
                    ports = fl["port"] if isinstance(fl["port"], list) else [fl["port"]]
                    for port in ports:
                        try:
                            _selecteur_nft(fl.get("protocole", "tcp"), port)
                        except ErreurFlux as e:
                            raise ErreurFlux(f"{role} : {e}") from e
            print(f"Flux coherents : {r} rôles, {t} flux, schéma + matrice OK.")
        elif commande == "registre":
            valider(flux)
            REGISTRE.write_text(generer_registre(flux), encoding="utf-8")
            print(f"{REGISTRE.relative_to(RACINE)} (re)généré depuis les meta/flux.yml.")
        elif commande == "nftables":
            valider(flux)
            ecrits = generer_nftables(flux)
            print(f"{len(ecrits)} aperçu(s) nftables générés (NON activés) :")
            for p in ecrits:
                print(f"  {p}")
        else:
            print(f"Commande inconnue : {commande} (registre | nftables | verifier)", file=sys.stderr)
            return 2
    except ErreurFlux as e:
        print(f"ERREUR flux :\n{e}", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))