Daniel Allaire
26090e8acd
Dans OPNsense une règle est toujours `in` sur l'interface d'arrivée : posée ailleurs elle ne s'applique jamais, et le trafic est bloqué sans que la configuration paraisse anormale. Les règles n'en portaient aucune, alors que le champ est obligatoire dans l'API. L'attribution se dérive du sens du flux : `ingress`/`externe` arrive par le WAN, `egress`/`externe` par le lien de transit. Le SSH d'administration suit la première ligne — le VPN est hébergé sur le pfSense voisin et revient par l'adresse publique. Le montage parallèle décrit ce jour lève la dernière inconnue. Le rendu abandonne `pass out` pour `pass in on <interface>`, l'idiome réel d'OPNsense et ce que le client d'API devra envoyer. Ajouté aussi : - `opnsense_wan_ip`, la face publique, en section Frontière du panneau ; - invariant du dernier octet (P23) : un point de routage porte le même dernier octet sur tous ses sous-réseaux. Le chiffre vient de `reservations.passerelle`, pas d'une constante. Exemption des liens plus étroits qu'un /24 — sur le /29 de transit l'adressage est dicté par les participants. Vérifié que l'invariant tenait déjà sur les 13 sous-réseaux routés avant d'écrire la garde. Preuves : 24 OK, 0 échec. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
401 lines
17 KiB
Python
401 lines
17 KiB
Python
#!/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"
|
|
|
|
# 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")
|
|
|
|
|
|
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
|
|
|
|
|
|
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.
|
|
"""
|
|
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"]]
|
|
|
|
|
|
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 octet_passerelle() -> 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 = 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) -> 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))
|
|
|
|
# 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")
|
|
|
|
# Invariant du dernier octet : tous les points de routage doivent le partager.
|
|
# Exemption des sous-reseaux plus etroits qu'un /24 : sur un /29 de transit,
|
|
# l'adressage est dicte par les participants du lien, pas par la convention.
|
|
attendu = octet_passerelle()
|
|
if attendu is not None:
|
|
for r in reseaux(underlay):
|
|
gw = r.get("passerelle")
|
|
if not gw:
|
|
continue
|
|
try:
|
|
net = ipaddress.ip_network(r["sous_reseau"], strict=False)
|
|
except ValueError:
|
|
continue
|
|
if net.prefixlen > 24:
|
|
continue # lien etroit : exempte, cf. le /29 de transit
|
|
octet = int(str(gw).rsplit(".", 1)[1])
|
|
if octet != attendu:
|
|
erreurs.append(
|
|
f"reseau '{r.get('nom','?')}': passerelle {gw} finit par .{octet} alors que "
|
|
f"les points de routage portent .{attendu} partout "
|
|
f"(nomenclature: reservations.passerelle)")
|
|
|
|
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")
|
|
# 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()
|