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).

Fichier : `underlay.yml` a la racine du moteur (gitignore ; voir `underlay.yml.example`),
surchargeable par `SETOPS_UNDERLAY`. Absent => le devis omet simplement la section underlay.

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 racine du moteur. None si absent."""
    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"


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 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 _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) -> 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 not r.get("passerelle"):
                    erreurs.append(f"reseau '{nom}': passerelle_sortie exige aussi 'passerelle' "
                                   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))

    # 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")
    # Le routeur PORTE le SVI : son adresse de gestion est la passerelle du reseau, pas
    # une adresse d'hote a part. Deux valeurs differentes se contrediraient — une
    # interface VLAN n'a qu'une adresse primaire.
    for h in hotes(underlay):
        if h.get("nom") != r_nom:
            continue
        r = {x.get("nom"): x for x in reseaux(underlay)}.get(h.get("reseau")) or {}
        gw = r.get("passerelle")
        if gw and str(h.get("ip")) != str(gw):
            erreurs.append(f"routeur '{r_nom}': ip {h.get('ip')} != passerelle {gw} du reseau "
                           f"'{h.get('reseau')}' — le routeur porte ce SVI, les deux doivent coincider")

    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()