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 []


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")
        vus_sous.append((net, nom))

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