Set-OPS-Public/scripts/underlay.py

#!/usr/bin/env python3
"""Underlay = la fabric physique partagee par les instances federees.

Cluster-global, HORS de la derivation par-tenant (aucun index) : management des
switches et du mgmt Proxmox/OOB, stockage iSCSI, reseaux Ceph (public + cluster).
Ce sont des reseaux du *sous-sol* : ils portent la flotte, ils n'appartiennent a
aucun tenant.

Convention : VLAN < 1000, franchement SOUS la plage tenant (VLAN = 1000+index*10+zone).
Les sous-reseaux ne doivent chevaucher aucun supernet tenant (10.(10+index).0.0/16).

EMPLACEMENT : l'underlay appartient a l'HEBERGEUR — ses switches, ses cables. Il vit
donc dans SON depot, et le moteur le monte par symlink :

    Set-OPS-public/underlay.yml -> ../OPS-<hebergeur>/underlay.yml

Ce symlink NE SUIT PAS `make instance-utiliser` : la fabric reste celle de l'hebergeur,
quel que soit le tenant actif. Surchargeable par `SETOPS_UNDERLAY`. Absent (ou lien
brise, cas d'un clone sans depot frere) => le devis omet simplement la section.

CLI : `python3 scripts/underlay.py` (affiche + valide), `--verifier` (P23, code de sortie).
"""
from __future__ import annotations
import argparse
import ipaddress
import os
import sys
from pathlib import Path

import yaml

RACINE = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(RACINE / "scripts"))
from inventory_rules import supernet_de, vlan_de  # noqa: E402

# Les VLAN tenant commencent a 1000+index*10+zone (>= 1011). L'underlay reste en dessous.
SEUIL_VLAN_TENANT = 1000


def chemin() -> Path | None:
    """Emplacement du underlay.yml : SETOPS_UNDERLAY, sinon le symlink a la racine.

    `exists()` suit le lien : un symlink brise (clone sans le depot de l'hebergeur)
    rend None, et tout degrade proprement au lieu d'echouer.
    """
    env = os.environ.get("SETOPS_UNDERLAY")
    if env:
        p = Path(env).expanduser()
        return p if p.exists() else None
    p = RACINE / "underlay.yml"
    return p if p.exists() else None


def charger() -> dict | None:
    """Le bloc `underlay` du fichier, ou None si absent/vide."""
    c = chemin()
    if not c:
        return None
    data = yaml.safe_load(c.read_text(encoding="utf-8")) or {}
    return data.get("underlay") or None


def reseaux(underlay: dict | None) -> list[dict]:
    return (underlay or {}).get("reseaux", []) or []


def hotes(underlay: dict | None) -> list[dict]:
    return (underlay or {}).get("hotes", []) or []


FABRIC_DEFAUT = "principal"

# Priorites de pont : le routeur est la racine (basse), les switches d'acces ne
# doivent jamais l'etre. 4096 laisse 0 disponible pour une reprise d'urgence.
STP_PRIORITE_RACINE = 4096
STP_PRIORITE_ACCES = 32768
# Dialecte de CLI du commutateur : c'est une propriete du MATERIEL, donc de la fabric.
# Source unique (devis_reseau y lit ses choix d'argparse).
DIALECTES = ("cisco", "binardat")
STP_MODES = ("rstp", "mstp", "pvst")
STP_TOPOLOGIES = ("etoile", "anneau", "maille")


ROUTAGE_TENANTS = ("switch", "sdn")
# VXLAN coute 50 octets d'encapsulation : le TRANSPORT doit valoir au moins l'overlay
# plus cette surcharge. Sous ce seuil, il casse PARTIELLEMENT — le ping passe, les
# transferts echouent — la panne la plus couteuse a diagnostiquer.
SURCOUT_VXLAN = 50
MTU_OVERLAY_DEFAUT = 1450  # ce que voit une VM ; regle par zone dans Proxmox SDN


def mtu_overlay(underlay: dict | None) -> int:
    """MTU annonce aux VM par les VNet. Declare, parce qu'il commande le minimum du
    transport : un overlay a 1450 tient dans un transport a 1500, pas un overlay a 1500."""
    v = (underlay or {}).get("mtu_overlay")
    return int(v) if isinstance(v, int) else MTU_OVERLAY_DEFAUT


def routage_tenants(underlay: dict | None) -> str:
    """Qui route entre les zones d'un tenant : `switch` (SVI) ou `sdn` (EVPN).

    En `sdn`, une zone EVPN par tenant porte le VRF : le routage ET le filtrage
    inter-zone d'un meme tenant vivent sur les hyperviseurs. Aucun VLAN de tenant ne
    circule alors sur le fil — seulement du VXLAN encapsule dans de l'IP — et le devis
    switch cesse d'emettre VLAN tenants, SVI et ACL.

    L'inter-tenant, lui, sort du VRF et passe par la frontiere, qui le police.
    """
    v = str((underlay or {}).get("routage_tenants") or "switch")
    return v if v in ROUTAGE_TENANTS else "switch"


def acl_inter_tenant(underlay: dict | None) -> bool:
    """La fabric porte-t-elle l'isolation inter-tenant par ACL ? Vrai par defaut.

    A mettre a `false` quand le materiel ne sait pas lier une ACL a une interface de
    routage — cas du Binardat, dont les interfaces VLAN n'offrent aucun `access-group`.
    Le devis cesse alors d'emettre des ACL qui ne seraient jamais liees : mieux vaut
    aucune regle qu'une regle inoperante qu'on croit active.

    Conséquence assumee, et ecrite dans le devis : l'isolation repose entierement sur
    les nftables d'hote, et le plan de gestion de la fabric n'a plus de protection
    RESEAU contre les tenants.
    """
    v = (underlay or {}).get("acl_inter_tenant")
    return True if v is None else bool(v)


def dialecte(underlay: dict | None) -> str:
    """Dialecte de CLI declare pour la fabric. '' si absent."""
    return str((underlay or {}).get("dialecte") or "")


def stp(underlay: dict | None) -> dict:
    """Parametres de spanning-tree de la fabric principale. {} si non declares."""
    return (underlay or {}).get("stp") or {}


def fabric_de(reseau: dict) -> str:
    """Fabric physique qui porte ce reseau. `principal` par defaut."""
    return str(reseau.get("fabric") or FABRIC_DEFAUT)


def fabriques(underlay: dict | None) -> list[str]:
    """Les fabrics declarees, `principal` en tete."""
    vues = {fabric_de(r) for r in reseaux(underlay)}
    return [FABRIC_DEFAUT] * (FABRIC_DEFAUT in vues) + sorted(vues - {FABRIC_DEFAUT})


def reseaux_de_fabric(underlay: dict | None, fabric: str) -> list[dict]:
    return [r for r in reseaux(underlay) if fabric_de(r) == fabric]


def fabric_du_routeur(underlay: dict | None) -> str:
    """Fabric du switch routeur : celle du reseau ou il est declare comme hote."""
    r_nom = routeur(underlay)
    par_nom = {r.get("nom"): r for r in reseaux(underlay)}
    for h in hotes(underlay):
        if h.get("nom") == r_nom and h.get("reseau") in par_nom:
            return fabric_de(par_nom[h["reseau"]])
    return FABRIC_DEFAUT


def routeur(underlay: dict | None) -> str | None:
    """Nom du switch qui porte les SVI, ou None si aucun n'est designe.

    Sans MLAG, UN SEUL switch route : les autres restent en L2 pur. Appliquer le
    meme devis sur les trois creerait autant de conflits d'adresses que de SVI.
    """
    return (underlay or {}).get("routeur") or None


def ports_de(hote: dict) -> dict:
    """Ports physiques declares d'un equipement. {} si aucun.

    Trois clefs, qui correspondent aux trois natures de lien du devis :
      `hyperviseurs` (liste)  ports terminaux vers Proxmox ;
      `frontiere`    (liste)  port(s) vers le pare-feu de bordure ;
      `rayons`       (table)  {nom du switch d'acces: port}, cote routeur ;
      `montante`     (str)    port vers le routeur, cote switch d'acces.
    Non declares, le devis retombe sur ses marqueurs : il reste lisible, il n'est
    simplement pas applicable tel quel.
    """
    return (hote or {}).get("ports") or {}


def _ports_a_plat(hote: dict) -> list[str]:
    """Tous les ports d'un equipement, a plat — pour detecter un doublon."""
    p = ports_de(hote)
    plat = [str(x) for x in (p.get("hyperviseurs") or [])]
    plat += [str(x) for x in (p.get("frontiere") or [])]
    plat += [str(v) for v in (p.get("rayons") or {}).values()]
    if p.get("montante"):
        plat.append(str(p["montante"]))
    return plat


ROLES_HOTE = ("switch", "hyperviseur", "frontiere")


def role_hote(hote: dict) -> str:
    """Nature d'un equipement declare. `switch` par defaut (compatibilite).

    Le reseau ne suffit pas a la deduire : un hyperviseur partage le reseau de
    management avec les commutateurs, et recevrait sinon leur configuration.
    """
    return str((hote or {}).get("role") or "switch")


def switches_acces(underlay: dict | None) -> list[dict]:
    """Les switches d'acces : hotes du reseau de management, hors routeur.

    Source unique du « qui est un switch d'acces » — utilisee pour leur devis (partie B)
    ET pour les rayons de l'etoile emis cote routeur. Les deux ne peuvent pas diverger.

    Filtre sur le ROLE autant que sur le reseau : un hyperviseur declare sur le meme
    reseau ne doit recevoir aucune ligne de configuration de commutateur.
    """
    r_nom = routeur(underlay)
    if not r_nom:
        return []
    fabric = fabric_du_routeur(underlay)
    mgmt = next((r for r in reseaux_de_fabric(underlay, fabric) if r.get("passerelle")), None)
    if not mgmt:
        return []
    return [h for h in hotes(underlay)
            if h.get("nom") != r_nom and h.get("reseau") == mgmt["nom"]
            and role_hote(h) == "switch"]


def reseau_transit(underlay: dict | None) -> dict | None:
    """Le reseau de TRANSIT vers la frontiere nord/sud, s'il est declare.

    Convention : c'est le reseau underlay qui porte `passerelle_sortie` — l'adresse
    du pare-feu de bordure sur ce lien. Il vit dans l'underlay et non dans un tenant
    parce qu'il est PARTAGE : la frontiere route vers tous les supernets tenants par
    ce meme saut, il ne peut donc deriver d'aucun `index`.
    """
    for r in reseaux(underlay):
        if r.get("passerelle_sortie"):
            return r
    return None


def charger_depuis(chemin_fichier: Path) -> dict | None:
    """Le bloc `underlay` d'un fichier DESIGNE — pour valider un modele, pas le site."""
    if not chemin_fichier.is_file():
        return None
    data = yaml.safe_load(chemin_fichier.read_text(encoding="utf-8")) or {}
    return data.get("underlay") or None


def octet_passerelle(plan_nomenclature: Path | None = None) -> int | None:
    """Dernier octet reserve aux points de routage, lu dans la nomenclature active.

    Convention d'exploitation : un routeur porte LE MEME dernier octet sur tous les
    sous-reseaux ou il participe — on retient une adresse, pas treize. Declare une
    seule fois (`reservations.passerelle`), jamais code en dur ici.
    """
    p = plan_nomenclature or (RACINE / "instance" / "plan" / "nomenclature.yml")
    if not p.is_file():
        return None
    n = yaml.safe_load(p.read_text(encoding="utf-8")) or {}
    v = (n.get("reservations") or {}).get("passerelle")
    return int(v) if isinstance(v, int) else None


def _tenants_reperes() -> tuple[set[int], list[str]]:
    """VLAN et supernets des tenants federes (import paresseux : evite le cycle avec devis)."""
    import devis_reseau
    tenants = devis_reseau.decouvrir()
    vlans = {vlan_de(n["index"], z) for _, _, n in tenants for z in n["categories"]}
    supernets = [supernet_de(n["index"]) for _, _, n in tenants]
    return vlans, supernets


def valider(underlay: dict | None,
            tenants_vlans: set[int] | None = None,
            tenants_supernets: list[str] | None = None,
            plan_nomenclature: Path | None = None) -> list[str]:
    """Liste des erreurs (vide = conforme). Verifie champs requis, unicite, et
    surtout l'absence de collision avec la plage tenant (VLAN et sous-reseaux)."""
    if underlay is None:
        return []  # absent = rien a valider (le devis omettra la section)
    if tenants_vlans is None or tenants_supernets is None:
        tenants_vlans, tenants_supernets = _tenants_reperes()
    t_supernets = [ipaddress.ip_network(s, strict=False) for s in tenants_supernets]

    erreurs: list[str] = []
    vus_vlan: dict[int, str] = {}
    vus_sous: list[tuple[ipaddress.IPv4Network, str]] = []
    noms_reseaux: set[str] = set()

    for r in reseaux(underlay):
        nom = r.get("nom", "?")
        noms_reseaux.add(nom)
        vlan, sous = r.get("vlan"), r.get("sous_reseau")
        if vlan is None or sous is None:
            erreurs.append(f"reseau '{nom}': 'vlan' et 'sous_reseau' sont requis")
            continue
        if not isinstance(vlan, int) or vlan >= SEUIL_VLAN_TENANT:
            erreurs.append(f"reseau '{nom}': VLAN {vlan} doit etre un entier < {SEUIL_VLAN_TENANT} "
                           f"(sinon empiete sur la plage tenant)")
        if vlan in tenants_vlans:
            erreurs.append(f"reseau '{nom}': VLAN {vlan} deja porte par un tenant federe")
        if vlan in vus_vlan:
            erreurs.append(f"reseau '{nom}': VLAN {vlan} en double avec '{vus_vlan[vlan]}'")
        vus_vlan[vlan] = nom
        try:
            net = ipaddress.ip_network(sous, strict=False)
        except ValueError:
            erreurs.append(f"reseau '{nom}': sous_reseau '{sous}' invalide")
            continue
        for ts in t_supernets:
            if net.overlaps(ts):
                erreurs.append(f"reseau '{nom}': {sous} chevauche le supernet tenant {ts}")
        for autre, autre_nom in vus_sous:
            if net.overlaps(autre):
                erreurs.append(f"reseau '{nom}': {sous} chevauche le reseau underlay '{autre_nom}'")
        if r.get("passerelle"):
            try:
                if ipaddress.ip_address(r["passerelle"]) not in net:
                    erreurs.append(f"reseau '{nom}': passerelle {r['passerelle']} hors de {sous}")
            except ValueError:
                erreurs.append(f"reseau '{nom}': passerelle '{r['passerelle']}' invalide")
        # Transit vers la frontiere : la sortie doit etre SUR le lien, et distincte du SVI.
        if r.get("passerelle_sortie"):
            try:
                sortie = ipaddress.ip_address(r["passerelle_sortie"])
                if sortie not in net:
                    erreurs.append(f"reseau '{nom}': passerelle_sortie {sortie} hors de {sous}")
                elif r.get("passerelle") and str(sortie) == str(r["passerelle"]):
                    erreurs.append(f"reseau '{nom}': passerelle_sortie {sortie} identique au SVI "
                                   f"(le switch ne peut pas etre sa propre sortie)")
                elif False:
                    erreurs.append(f"reseau '{nom}': (inatteignable) "
                                   f"(le SVI du switch sur le lien de transit)")
            except ValueError:
                erreurs.append(f"reseau '{nom}': passerelle_sortie '{r['passerelle_sortie']}' invalide")
        vus_sous.append((net, nom))

    for h in hotes(underlay):
        r = (h or {}).get("role")
        if r is not None and str(r) not in ROLES_HOTE:
            erreurs.append(f"hote '{h.get('nom','?')}': role '{r}' inconnu "
                           f"(attendu : {', '.join(ROLES_HOTE)})")

    # Ports physiques : un meme port ne peut pas servir deux fois sur un equipement,
    # et un rayon ne peut viser qu'un switch d'acces reellement declare.
    noms_acces = {h.get("nom") for h in switches_acces(underlay)}
    r_courant = routeur(underlay)
    for h in hotes(underlay):
        hn = h.get("nom", "?")
        plat = _ports_a_plat(h)
        doublons = sorted({p for p in plat if plat.count(p) > 1})
        if doublons:
            erreurs.append(f"hote '{hn}': port(s) declare(s) deux fois : {', '.join(doublons)}")
        p = ports_de(h)
        for cible in (p.get("rayons") or {}):
            if cible not in noms_acces:
                erreurs.append(f"hote '{hn}': rayon vers '{cible}', qui n'est pas un switch d'acces")
        if p.get("rayons") and hn != r_courant:
            erreurs.append(f"hote '{hn}': des `rayons` mais ce n'est pas le routeur")
        if p.get("montante") and hn == r_courant:
            erreurs.append(f"hote '{hn}': une `montante` alors qu'il EST le routeur")

    # L'INVARIANT DU DERNIER OCTET NE S'APPLIQUE PAS ICI (decide le 2026-08-04).
    #
    # Il vaut pour l'adressage DERIVE des tenants, ou `passerelle_de(index, zone)`
    # produit le meme `.1` dans les treize sous-reseaux d'un tenant : on retient une
    # adresse, pas treize. C'est une propriete de la derivation, pas une regle
    # universelle.
    #
    # Dans l'underlay, l'adressage est dicte par les PARTICIPANTS d'un lien physique.
    # La regle y produisait deux effets pervers : un seuil arbitraire (les sous-reseaux
    # plus etroits qu'un /24 etaient exemptes, donc elargir un /29 changeait la
    # validite du fichier), et une couture entre proprietaires — l'octet attendu venait
    # de la nomenclature d'un TENANT, appliquee a la fabric de l'HEBERGEUR. La validite
    # de l'underlay aurait dependu du tenant actif si l'un d'eux reservait autre chose.
    #
    # Ce qui reste, et qui est plus fort : une passerelle doit etre l'adresse d'un hote
    # DECLARE sur ce reseau (verifie plus bas). Elle attrape les passerelles fantomes,
    # ce que le comptage d'octets ne faisait pas.

    v = (underlay or {}).get("acl_inter_tenant")
    if v is not None and not isinstance(v, bool):
        erreurs.append(f"acl_inter_tenant doit etre un booleen (recu : {v!r})")

    rt = (underlay or {}).get("routage_tenants")
    if rt is not None and str(rt) not in ROUTAGE_TENANTS:
        erreurs.append(f"routage_tenants '{rt}' inconnu (attendu : {', '.join(ROUTAGE_TENANTS)})")
    if routage_tenants(underlay) == "sdn":
        # Le transport doit valoir l'overlay DECLARE plus la surcharge d'encapsulation.
        minimal = mtu_overlay(underlay) + SURCOUT_VXLAN
        fab = fabric_du_routeur(underlay)
        for r in reseaux_de_fabric(underlay, fab):
            mtu = int(r.get("mtu") or 1500)
            if mtu < minimal:
                erreurs.append(
                    f"reseau '{r.get('nom','?')}': MTU {mtu} < {minimal} "
                    f"(overlay {mtu_overlay(underlay)} + {SURCOUT_VXLAN} de VXLAN) — sous ce "
                    f"seuil le ping passe et les transferts echouent")

    d = dialecte(underlay)
    if d and d not in DIALECTES:
        erreurs.append(f"dialecte '{d}' inconnu (attendu : {', '.join(DIALECTES)})")

    s = stp(underlay)
    if s:
        if str(s.get("mode", "")) not in STP_MODES:
            erreurs.append(f"stp.mode '{s.get('mode')}' inconnu (attendu : {', '.join(STP_MODES)})")
        if str(s.get("topologie", "")) not in STP_TOPOLOGIES:
            erreurs.append(f"stp.topologie '{s.get('topologie')}' inconnue "
                           f"(attendu : {', '.join(STP_TOPOLOGIES)})")
        if not routeur(underlay):
            erreurs.append("stp declare sans `routeur` : impossible de designer le pont racine")

    # Une seule sortie possible : deux transits donneraient deux routes par defaut.
    transits = [r.get("nom", "?") for r in reseaux(underlay) if r.get("passerelle_sortie")]
    if len(transits) > 1:
        erreurs.append(f"un seul reseau de transit autorise (passerelle_sortie), "
                       f"trouves : {', '.join(transits)}")

    # Le routeur designe doit etre un hote declare : sinon le devis ne saurait pas
    # quelle adresse de gestion donner aux switches d'acces.
    r_nom = routeur(underlay)
    if r_nom and r_nom not in {h.get("nom") for h in hotes(underlay)}:
        erreurs.append(f"routeur '{r_nom}': aucun hote de ce nom n'est declare")
    # La passerelle d'un reseau doit APPARTENIR A UN HOTE DECLARE de ce reseau.
    #
    # Cette regle en remplace une plus etroite, qui exigeait que la passerelle soit
    # l'adresse du switch routeur. Elle etait vraie tant qu'un switch routait ; depuis
    # que la frontiere est le seul equipement L3 (2026-08-04), elle refusait la seule
    # configuration correcte. La nouvelle est PLUS forte : elle attrape aussi une
    # passerelle qui ne correspond a rien — une adresse inventee, un octet de trop.
    par_reseau: dict[str, list[dict]] = {}
    for h in hotes(underlay):
        if h.get("reseau"):
            par_reseau.setdefault(h["reseau"], []).append(h)
    for r in reseaux(underlay):
        for cle, sens in (("passerelle", "passerelle"), ("passerelle_sortie", "sortie")):
            adr = r.get(cle)
            if not adr:
                continue
            porteurs = [h for h in par_reseau.get(r.get("nom"), []) if str(h.get("ip")) == str(adr)]
            if not porteurs:
                erreurs.append(
                    f"reseau '{r.get('nom','?')}': {sens} {adr} n'est l'adresse d'aucun hote "
                    f"declare sur ce reseau — passerelle fantome")

    par_nom = {r.get("nom"): r for r in reseaux(underlay)}
    for h in hotes(underlay):
        hn = h.get("nom", "?")
        r = par_nom.get(h.get("reseau"))
        if not r:
            erreurs.append(f"hote '{hn}': reseau '{h.get('reseau')}' inconnu")
            continue
        try:
            if ipaddress.ip_address(h["ip"]) not in ipaddress.ip_network(r["sous_reseau"], strict=False):
                erreurs.append(f"hote '{hn}': IP {h.get('ip')} hors de {r['sous_reseau']}")
        except (ValueError, KeyError):
            erreurs.append(f"hote '{hn}': IP absente ou invalide")
    return erreurs


def verifier() -> int:
    underlay = charger()
    if underlay is None:
        print("Underlay : aucun fichier (underlay.yml absent) — section omise, rien a prouver.")
        return 0
    erreurs = valider(underlay)
    if erreurs:
        print("Underlay NON conforme :")
        for e in erreurs:
            print(f"  - {e}")
        return 1
    n = len(reseaux(underlay))
    print(f"Underlay conforme : {n} reseau(x), aucune collision avec la plage tenant.")
    return 0


def afficher() -> None:
    underlay = charger()
    if underlay is None:
        print("Aucun underlay defini (underlay.yml absent). Copier underlay.yml.example pour commencer.")
        return
    print(f"Underlay (fabric physique, cluster-global) — source : {chemin()}")
    print(f"{'RESEAU':<16}{'VLAN':>6}  {'SOUS-RESEAU':<18}{'PASSERELLE':<15}{'MTU':>6}  DESCRIPTION")
    for r in reseaux(underlay):
        print(f"{r.get('nom','?'):<16}{str(r.get('vlan','?')):>6}  "
              f"{r.get('sous_reseau','?'):<18}{str(r.get('passerelle') or '-'):<15}"
              f"{str(r.get('mtu') or '-'):>6}  {r.get('description','')}")
    if hotes(underlay):
        print("\nHotes fixes documentes :")
        for h in hotes(underlay):
            print(f"  {h.get('nom','?'):<16} {h.get('ip','?'):<15} ({h.get('reseau','?')})")
    erreurs = valider(underlay)
    print("\n" + ("Conforme (aucune collision tenant)." if not erreurs
                   else "ERREURS :\n  - " + "\n  - ".join(erreurs)))


def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--verifier", action="store_true", help="valide l'underlay (P23), code de sortie")
    args = ap.parse_args()
    if args.verifier:
        sys.exit(verifier())
    afficher()


if __name__ == "__main__":
    main()