Set-OPS-Public/scripts/devis_proxmox_fw.py

#!/usr/bin/env python3
"""Devis de pare-feu Proxmox — le filtrage EST-OUEST intra-tenant.

DERIVE du meme registre des flux que les nftables d'hote et que la frontiere. Les trois
lisent `roles/*/meta/flux.yml` par les MEMES fonctions : ils ne peuvent pas diverger.

PARTAGE DES ROLES (cf. docs/sdn-evpn.md) :
  - ce devis            : est-ouest INTRA-tenant, applique par l'hyperviseur ;
  - `make flux`         : la meme politique, rejouee sur l'hote lui-meme (defense en
                          profondeur — une VM compromise franchit deux barrieres) ;
  - `make devis-opnsense` : tout ce qui ENTRE ou SORT d'un tenant.

Les flux `pair: externe` sont donc SAUTES ici : ils relevent de la frontiere.

NON destructif : ce script n'ecrit rien sur le cluster. Il produit un devis a relire,
puis (a venir) un JSON pour l'API Proxmox.

Usage :
  python3 scripts/devis_proxmox_fw.py            # devis lisible
  python3 scripts/devis_proxmox_fw.py --json     # meme contenu, pour l'API
  python3 scripts/devis_proxmox_fw.py --verifier # garde : aucun role sans source
"""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

import yaml

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

from resoudre_flux import (  # noqa: E402
    GROUPE_EDGE,
    charger_flux,
    est_groupe_operationnel,
    _enfants,
    _hotes_du_groupe,
    _ip_par_hote,
    _pairs,
    _resoudre_sources,
)
from devis_reseau import admin_de, decouvrir, inventaire_de, prefixe  # noqa: E402


LONGUEUR_MAX_GROUPE = 18  # limite de Proxmox pour un nom de groupe de securite


def _role_court(role: str) -> str:
    """Forme abregee d'un role, commune aux IPSets et aux groupes."""
    return role.replace("serveur_", "srv-").replace("client_", "cli-").replace("_", "-")


def nom_ipset(role: str, index: int) -> str:
    """Nom d'IPSet. MEME prefixe que les groupes — `t17-` — pour qu'un seul schema de
    nommage se lise dans tout le devis.

    Pas de troncature ici : Proxmox est large sur les IPSets, etroit sur les groupes.
    Un nom peut donc etre entier cote IPSet et abrege cote groupe ; chacun respecte sa
    contrainte, et le prefixe reste commun.
    """
    return f"t{index}-{_role_court(role)}".lower()


def nom_groupe(role: str, index: int) -> str:
    """Nom de groupe de securite, sous la limite de 18 caracteres de Proxmox.

    Le prefixe porte l'INDEX du tenant plutot que son etiquette : `t17-` au lieu de
    `chez17-`, ce qui laisse trois caracteres de plus au role et rend la troncature
    beaucoup plus rare. Les groupes sont a l'echelle du datacenter, d'ou le prefixe.
    """
    return f"t{index}-{_role_court(role)}".lower()[:LONGUEUR_MAX_GROUPE]


def _verifier_unicite(noms: list[str], tenant: str) -> None:
    """Une troncature qui fait collisionner deux roles FUSIONNERAIT leurs regles.

    Silencieusement, et dans le mauvais sens : une VM recevrait les autorisations d'un
    role qu'elle ne porte pas. On echoue plutot que d'emettre un devis pareil.
    """
    vus: dict[str, str] = {}
    for role, nom in noms:
        if nom in vus:
            raise SystemExit(
                f"erreur: {tenant} — les roles '{vus[nom]}' et '{role}' donnent le meme "
                f"groupe '{nom}' apres troncature a {LONGUEUR_MAX_GROUPE} caracteres. "
                f"Leurs regles fusionneraient. Renommer un role ou raccourcir le prefixe.")
        vus[nom] = role


def _ports(fl: dict) -> list[str]:
    p = fl["port"]
    return [str(x) for x in (p if isinstance(p, list) else [p])]


def construire(tenants: list[tuple[str, str, dict]]) -> dict:
    """Le devis, en structure. Un bloc par tenant federe."""
    flux_par_role = charger_flux()
    blocs: list[dict] = []

    for nom, pfx, n in tenants:
        inv = inventaire_de(nom)
        if inv is None:
            continue
        etiquette = f"{pfx}{n['index']}"
        data = yaml.safe_load(inv.read_text(encoding="utf-8")) or {}
        # Les PLANIFIES comptent : le pare-feu se prepare avant que la VM existe,
        # sinon le devis changerait a chaque materialisation d'hote.
        tous = set(_ip_par_hote(data))
        ips = _ip_par_hote(data)

        # Un IPSet par role porte dans ce tenant : les membres sont ses hotes.
        # `flotte` en recoit un aussi : sans lui, six regles par tenant portaient
        # quatorze adresses en dur, illisibles et impossibles a mettre a jour d'un geste.
        ipsets: dict[str, dict] = {}
        if ips:
            ipsets[f"t{n['index']}-flotte"] = {
                "role": "flotte (tous les hotes du tenant)", "membres": sorted(set(ips.values())),
            }
        for groupe in sorted(_enfants(data)):
            if not est_groupe_operationnel(groupe):
                continue
            membres = sorted({ips[h] for h in _hotes_du_groupe(data, groupe) if h in ips})
            if membres:
                ipsets[nom_ipset(groupe, n["index"])] = {
                    "role": groupe, "membres": membres,
                }

        groupes: list[dict] = []
        sans_source: list[str] = []
        utilises: set[str] = set()
        for groupe in sorted(_enfants(data)):
            if not est_groupe_operationnel(groupe):
                continue
            if not _hotes_du_groupe(data, groupe):
                continue
            regles: list[dict] = []
            for fl in flux_par_role.get(groupe, []):
                if fl.get("sens") != "ingress":
                    continue
                pairs = _pairs(fl)
                if "externe" in pairs:
                    continue          # la frontiere s'en charge
                if pairs == ["localhost"]:
                    continue          # boucle locale : aucune regle inter-noeud
                srcs = _resoudre_sources(data, fl.get("pair"), tous, ips)
                if not srcs:
                    sans_source.append(f"{groupe}:{','.join(_ports(fl))}")
                    continue
                # Source par IPSet des que possible : une seule liste a tenir, et une
                # regle qui se lit. Les mots-cles `flotte` et `edge` y ont droit comme
                # les roles — `edge` designe le role nginx.
                source = ",".join(srcs)
                if len(pairs) == 1:
                    cle = {"flotte": f"t{n['index']}-flotte",
                           "edge": nom_ipset(GROUPE_EDGE, n["index"])}.get(
                               pairs[0], nom_ipset(pairs[0], n["index"]))
                    if cle in ipsets:
                        source = f"+{cle}"
                        utilises.add(cle)
                for port in _ports(fl):
                    regles.append({
                        "action": "ACCEPT", "sens": "IN",
                        "source": source,
                        "proto": fl.get("protocole", "tcp"),
                        "dport": port,
                        "raison": fl.get("raison", ""),
                    })
            if regles:
                groupes.append({
                    "nom": nom_groupe(groupe, n["index"]), "role": groupe, "regles": regles,
                })
        _verifier_unicite([(g["role"], g["nom"]) for g in groupes], nom)

        # Affectation : quelle VM recoit quels groupes.
        affect: list[dict] = []
        par_nom_groupe = {g["role"]: g["nom"] for g in groupes}
        # TOUS les hotes du plan, actifs comme planifies — meme regle que les IPSets.
        # Se limiter aux actifs donnait un devis different selon l'etat du tenant : un
        # operateur aurait lu qu'une seule VM avait besoin de regles.
        for hote in sorted(ips):
            vmid = None
            for g, membres in _enfants(data).items():
                h = (membres.get("hosts") or {}).get(hote) or {}
                if h.get("proxmox_vmid"):
                    vmid = h["proxmox_vmid"]
                    break
            porte = [par_nom_groupe[g] for g in sorted(_enfants(data))
                     if hote in (_enfants(data)[g].get("hosts") or {}) and g in par_nom_groupe]
            if porte:
                affect.append({"hote": hote, "vmid": vmid, "groupes": porte})

        blocs.append({
            "tenant": nom, "etiquette": etiquette,
            "ipsets": {k: v for k, v in ipsets.items() if k in utilises},
            "groupes": groupes,
            "affectations": affect, "sans_source": sorted(set(sans_source)),
            "admin": admin_de(nom),
        })

    return {"blocs": blocs}


def rendre(devis: dict) -> str:
    out = [
        "# ============================================================",
        "# DEVIS PARE-FEU PROXMOX — est-ouest INTRA-tenant",
        "# Genere par scripts/devis_proxmox_fw.py. NE PAS EDITER A LA MAIN.",
        "# Derive du MEME registre des flux que les nftables d'hote : les deux",
        "# barrieres disent la meme chose, et ne peuvent pas diverger.",
        "#",
        "# Ce qui ENTRE ou SORT d'un tenant n'est PAS ici : c'est la frontiere",
        "# (`make devis-opnsense`). Les flux `pair: externe` sont sautes.",
        "# ============================================================",
        "",
        "## 0. Prerequis, au niveau du datacenter",
        "#   Pare-feu active, politique d'entree DROP. Sans cela, ces regles",
        "#   s'ajoutent a un tout-permis et n'interdisent rien.",
        "",
    ]
    for b in devis["blocs"]:
        out += [f"## Tenant {b['tenant']} ({b['etiquette']})", ""]
        out.append(f"### IPSets — {len(b['ipsets'])} (les membres d'un role, en un seul endroit)")
        for nom, s in sorted(b["ipsets"].items()):
            out.append(f"ipset {nom}    # {s['role']}")
            for ip in s["membres"]:
                out.append(f"  {ip}")
        out += ["", f"### Groupes de securite — {len(b['groupes'])} (les flux entrants d'un role)"]
        for g in b["groupes"]:
            out.append(f"securitygroup {g['nom']}    # {g['role']}")
            for r in g["regles"]:
                out.append(f"  {r['sens']} {r['action']} -source {r['source']} "
                           f"-p {r['proto']} -dport {r['dport']}")
                if r["raison"]:
                    out.append(f"  #   {r['raison']}")
        out += ["", f"### Affectation aux VM — {len(b['affectations'])}"]
        for a in b["affectations"]:
            out.append(f"  vmid {a['vmid'] or '(non materialise)'}  {a['hote']:<18} "
                       f"-> {', '.join(a['groupes'])}")
        if b["sans_source"]:
            out += ["", "# /!\\ Flux ingress sans source resolue dans ce tenant (aucune regle emise) :"]
            out += [f"#     {x}" for x in b["sans_source"]]
        out.append("")
    out += [
        "## Defaut",
        "# Politique d'entree DROP au datacenter : ce qui n'est pas ci-dessus",
        "# n'entre pas. Un flux oublie se declare dans `roles/<role>/meta/flux.yml`,",
        "# jamais a la main dans Proxmox — la regle serait perdue a la generation.",
    ]
    return "\n".join(out)


def main(argv: list[str]) -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--json", action="store_true")
    ap.add_argument("--verifier", action="store_true")
    args = ap.parse_args(argv)

    devis = construire(decouvrir())
    if args.verifier:
        total_g = sum(len(b["groupes"]) for b in devis["blocs"])
        total_r = sum(len(g["regles"]) for b in devis["blocs"] for g in b["groupes"])
        if not total_g:
            print("erreur: aucun groupe de securite derive — le registre ou l'inventaire "
                  "est vide.", file=sys.stderr)
            return 2
        print(f"CONFORME : pare-feu Proxmox, {len(devis['blocs'])} tenant(s), "
              f"{total_g} groupe(s), {total_r} regle(s).")
        return 0
    print(json.dumps(devis, indent=2, ensure_ascii=False) if args.json else rendre(devis))
    return 0


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