Daniel Allaire
fa69ec7c0f
Deux obstacles, aucun n'etait celui qu'on croyait. Proxmox n'installe AUCUN defaut dans le VRF du tenant : `default-originate` annonce une route aux autres noeuds, il n'en pose pas chez lui. Les deux zones etaient dans cet etat. La sortie vient d'une strophe frr.conf.local, que Proxmox fusionne a chaque regeneration (verifie : survit a `pvesh set /cluster/sdn` et a un redemarrage de FRR). `nexthop-vrf default` emprunte UNE adresse au lieu d'importer la table principale : la route par defaut des hyperviseurs ne gouverne pas la sortie des tenants. `import vrf default` l'aurait fait contourner la frontiere et aurait fuite le transport VXLAN, la gestion et les VLAN herites dans le VRF. Le NAT sortant en mode automatique ne couvre que les reseaux directement attaches ; un supernet joint par route statique en sort en silence. L'etat montrait `nat_addr` absent : le filtre passait, la traduction manquait. `devis_opnsense` emet le NAT (section 2bis), le reconciliateur l'applique et le retire, et P24 refuse tout supernet route mais non traduit. Mesure : tenant -> frontiere 3/3, -> passerelle FAI 3/3, -> Internet 2/2 pour les deux tenants. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
706 lines
34 KiB
Python
706 lines
34 KiB
Python
#!/usr/bin/env python3
|
|
"""Devis de configuration OPNsense — la frontiere nord/sud du reseau converge.
|
|
|
|
DERIVE de deux sources deja tenues a jour, rien n'est saisi a la main :
|
|
- le REGISTRE DES FLUX (`roles/*/meta/flux.yml`) : les flux dont le pair est
|
|
`externe` sont, par construction, ceux que le pare-feu d'hote NE traite PAS
|
|
(scripts/resoudre_flux.py les saute) parce qu'ils relevent de la frontiere.
|
|
Plusieurs `raison` le disent explicitement (« filtre a l'OPNsense »).
|
|
- les NOMENCLATURES des instances federees (`../*/plan/nomenclature.yml`) :
|
|
les supernets tenants a router, decoulant du seul seed `index`.
|
|
|
|
ROLE ASSUME : frontiere NORD/SUD uniquement. Les passerelles de zone (SVI)
|
|
restent sur les switches L3 ; OPNsense ne porte AUCUN SVI de tenant et ne voit
|
|
pas le trafic est-ouest, qui reste filtre par les ACL de `make devis-reseau`.
|
|
|
|
NON destructif : ce script n'ecrit RIEN sur le pare-feu. Il produit un devis a
|
|
relire, puis (a venir) un JSON consommable par l'API OPNsense.
|
|
|
|
Usage :
|
|
python3 scripts/devis_opnsense.py # devis lisible
|
|
python3 scripts/devis_opnsense.py --json # meme contenu, pour l'API
|
|
python3 scripts/devis_opnsense.py --verifier # garde anti-lockout (P24)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import os
|
|
import ipaddress
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
RACINE = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(RACINE / "scripts"))
|
|
|
|
from inventory_rules import supernet_de # noqa: E402
|
|
# Helpers internes du resolveur de flux : reutilises VOLONTAIREMENT plutot que
|
|
# redupliques ici — le registre des flux et l'inventaire doivent avoir une seule
|
|
# lecture, sinon le devis et les nftables divergeraient en silence.
|
|
from resoudre_flux import ( # noqa: E402
|
|
charger_flux,
|
|
_enfants,
|
|
_hotes_du_groupe,
|
|
_inventaire,
|
|
_ip_par_hote,
|
|
_pairs,
|
|
_sources_admin_ssh,
|
|
)
|
|
# `inventaire_de` / `admin_de` viennent de devis_reseau : les routes de retour du
|
|
# switch et les regles de la frontiere doivent lire les MEMES tenants, sinon l'un
|
|
# certifie des routes que l'autre n'emet pas.
|
|
from devis_reseau import admin_de, decouvrir, inventaire_de, prefixe # noqa: E402
|
|
# Le lien de transit (et donc le prochain saut) vit dans l'underlay : il est PARTAGE
|
|
# par tous les tenants, il ne peut deriver d'aucun index. Source unique, pas un intrant.
|
|
import underlay as underlay_mod # noqa: E402
|
|
|
|
# Marqueurs de repli, tant que les intrants de frontiere ne sont pas renseignes
|
|
# (meme idiome que devis_reseau). Le panneau « Intrants de base » du GUI les ecrit
|
|
# dans group_vars/opnsense.yml : le devis se complete tout seul ensuite.
|
|
PROCHAIN_SAUT = "<PROCHAIN-SAUT-SWITCH>"
|
|
NOEUD_SORTIE = "<NOEUD-DE-SORTIE-EVPN>"
|
|
IF_TRANSIT = "<IF-TRANSIT>"
|
|
IF_WAN = "<IF-WAN>"
|
|
IF_GESTION = "<IF-GESTION>"
|
|
|
|
|
|
def depot_hebergeur() -> Path | None:
|
|
"""Depot de l'HEBERGEUR — celui qui possede le materiel, donc la frontiere.
|
|
|
|
Derive du symlink `underlay.yml` : il pointe vers le depot de l'hebergeur, ce qui
|
|
le DESIGNE deja. Le declarer une seconde fois ouvrirait la porte a deux valeurs
|
|
contradictoires. None si aucun underlay (site sans fabric declaree).
|
|
"""
|
|
c = underlay_mod.chemin()
|
|
return c.resolve().parent if c else None
|
|
|
|
|
|
def intrants_frontiere() -> dict:
|
|
"""Intrants NON sensibles de la frontiere (group_vars/opnsense.yml). {} si absent.
|
|
|
|
Lus chez l'HEBERGEUR, pas chez le tenant actif : un hebergeur sert plusieurs
|
|
tenants et n'a qu'une frontiere. Basculer l'instance active sur un invite ne doit
|
|
pas faire perdre au devis l'URL de gestion, l'adresse publique et les interfaces.
|
|
Repli sur l'instance active quand aucun underlay ne designe d'hebergeur.
|
|
"""
|
|
for base in (depot_hebergeur(), _inventaire().parent.parent.parent):
|
|
if base is None:
|
|
continue
|
|
for nom in ("principal", "production", "lab"):
|
|
p = base / "inventories" / nom / "group_vars" / "opnsense.yml"
|
|
if p.is_file():
|
|
return yaml.safe_load(p.read_text(encoding="utf-8")) or {}
|
|
return {}
|
|
|
|
|
|
def _ou_marqueur(intrants: dict, cle: str, marqueur: str) -> str:
|
|
"""La valeur d'intrant si elle est renseignee, sinon le marqueur a completer."""
|
|
valeur = str(intrants.get(cle) or "").strip()
|
|
return valeur or marqueur
|
|
|
|
|
|
def transit_underlay() -> dict | None:
|
|
"""Le reseau de transit declare dans l'underlay, ou None."""
|
|
return underlay_mod.reseau_transit(underlay_mod.charger())
|
|
|
|
|
|
def nom_frontiere(transit: dict | None) -> str:
|
|
"""Nom du boitier de bordure, s'il est declare comme hote sur le lien de transit."""
|
|
if not transit:
|
|
return ""
|
|
sortie = str(transit.get("passerelle_sortie") or "")
|
|
for h in underlay_mod.hotes(underlay_mod.charger()):
|
|
if h.get("reseau") == transit.get("nom") and str(h.get("ip")) == sortie:
|
|
return str(h.get("nom") or "")
|
|
return ""
|
|
|
|
|
|
def prochain_saut(transit: dict | None) -> str:
|
|
"""Prochain saut des routes tenants = le SVI du switch sur le lien de transit.
|
|
|
|
DERIVE, jamais saisi : c'est le meme lien que celui d'ou `devis_reseau` tire la
|
|
route par defaut et les routes de retour. Le declarer deux fois (une fois ici, une
|
|
fois dans l'underlay) ouvrirait la porte a deux valeurs contradictoires — c'est
|
|
exactement le mode de panne qu'on ferme. Marqueur tant qu'aucun transit n'existe.
|
|
"""
|
|
underlay = underlay_mod.charger()
|
|
if underlay_mod.routage_tenants(underlay) == "sdn":
|
|
# En EVPN, le commutateur ne route plus les tenants : le voisin de la frontiere
|
|
# est un NOEUD DE SORTIE (hyperviseur), pas le SVI. Emettre le SVI produirait
|
|
# des routes vers un equipement qui n'a aucun chemin vers le tenant.
|
|
#
|
|
# C'est le noeud PRIMAIRE, et son adresse SUR LE LIEN DE TRANSIT — pas son
|
|
# adresse d'underlay ni de gestion : la frontiere n'atteint que celle-la.
|
|
# Une seule route statique par tenant, donc un seul saut : deux noeuds actifs
|
|
# en sortie avec une seule route en entree donneraient un chemin asymetrique.
|
|
return adresse_noeud_de_sortie(underlay, transit) or NOEUD_SORTIE
|
|
return str((transit or {}).get("passerelle") or "").strip() or PROCHAIN_SAUT
|
|
|
|
|
|
def reseaux_locaux_frontiere(underlay: dict | None, transit: dict | None) -> list[str]:
|
|
"""Sous-reseaux DIRECTEMENT ATTACHES a la frontiere, hors lien de transit.
|
|
|
|
C'est le plan de gestion vu du pare-feu : les reseaux ou la frontiere porte elle-meme
|
|
une adresse. Un paquet venu de la n'arrive donc NI par le WAN NI par le transit, mais
|
|
par l'interface de gestion — et c'est a cette interface que sa regle doit etre posee.
|
|
|
|
Derive de l'underlay, jamais saisi : la frontiere y declare deja ses adresses, une par
|
|
reseau. Un intrant supplementaire ouvrirait la porte a deux valeurs contradictoires.
|
|
"""
|
|
nom = nom_frontiere(transit)
|
|
if not nom:
|
|
return []
|
|
sur = {str(h.get("reseau")) for h in underlay_mod.hotes(underlay)
|
|
if str(h.get("nom") or "") == nom}
|
|
sur.discard(str((transit or {}).get("nom")))
|
|
cidrs = []
|
|
for r in underlay_mod.reseaux(underlay):
|
|
if str(r.get("nom")) in sur and r.get("sous_reseau"):
|
|
cidrs.append(str(r["sous_reseau"]))
|
|
return sorted(cidrs)
|
|
|
|
|
|
def _porte_par(cidr: str, reseaux: list[str]) -> bool:
|
|
"""Vrai si `cidr` est contenu dans l'un des `reseaux` (ou lui est egal)."""
|
|
try:
|
|
r = ipaddress.ip_network(cidr, strict=False)
|
|
except ValueError:
|
|
return False
|
|
for n in reseaux:
|
|
try:
|
|
if r.subnet_of(ipaddress.ip_network(n, strict=False)):
|
|
return True
|
|
except (ValueError, TypeError):
|
|
continue
|
|
return False
|
|
|
|
|
|
def adresse_noeud_de_sortie(underlay: dict | None, transit: dict | None) -> str:
|
|
"""Adresse du noeud de sortie PRIMAIRE sur le lien de frontiere. '' si indeterminee.
|
|
|
|
Deux declarations doivent concorder, et c'est voulu : l'hebergeur nomme le noeud
|
|
(`proxmox_sdn.sortie_primaire`, propriete du cluster), l'underlay dit son adresse
|
|
sur ce lien. Nommer un noeud absent du lien rend le devis muet plutot que faux.
|
|
"""
|
|
primaire = str((_sdn_hebergeur() or {}).get("sortie_primaire") or "").strip()
|
|
if not primaire or not transit:
|
|
return ""
|
|
for h in underlay_mod.hotes(underlay):
|
|
if (h.get("nom") == primaire and h.get("reseau") == transit.get("nom")
|
|
and h.get("ip")):
|
|
return str(h["ip"])
|
|
return ""
|
|
|
|
|
|
def _sdn_hebergeur() -> dict:
|
|
"""`proxmox_sdn` du depot de l'HEBERGEUR — meme derivation que devis_sdn (D-35)."""
|
|
lien = Path(os.environ.get("SETOPS_UNDERLAY") or (RACINE / "underlay.yml"))
|
|
if not lien.exists():
|
|
return {}
|
|
fichier = lien.resolve().parent / "proxmox-hebergeur.yml"
|
|
if not fichier.is_file():
|
|
return {}
|
|
return (yaml.safe_load(fichier.read_text(encoding="utf-8")) or {}).get("proxmox_sdn") or {}
|
|
|
|
# Le socle est porte par TOUS les hotes : sa destination n'est pas une poignee
|
|
# de machines mais le supernet du tenant. Evite un alias de 14 adresses.
|
|
ROLE_SOCLE = "serveur_debian"
|
|
|
|
|
|
def flux_frontiere(flux_par_role: dict[str, list[dict]]) -> list[tuple[str, dict]]:
|
|
"""[(role, flux)] pour tout flux dont le pair inclut `externe`, trie."""
|
|
retenus = [
|
|
(role, fl)
|
|
for role, flux in sorted(flux_par_role.items())
|
|
for fl in flux
|
|
if "externe" in _pairs(fl)
|
|
]
|
|
retenus.sort(key=lambda rf: (rf[1].get("sens", ""), rf[0], str(rf[1].get("port"))))
|
|
return retenus
|
|
|
|
|
|
def _ports(flux: dict) -> list[str]:
|
|
port = flux["port"]
|
|
return [str(p) for p in (port if isinstance(port, list) else [port])]
|
|
|
|
|
|
def cibles_par_role(role: str, inventaire: Path) -> list[str]:
|
|
"""IP des hotes qui portent ce role dans cette instance (actifs ET planifies).
|
|
|
|
On inclut les planifies a dessein : la frontiere se prepare avant que les VM
|
|
existent, sinon le devis changerait a chaque materialisation d'hote.
|
|
"""
|
|
data = yaml.safe_load(inventaire.read_text(encoding="utf-8")) or {}
|
|
ips = _ip_par_hote(data)
|
|
return sorted({ips[h] for h in _hotes_du_groupe(data, role) if h in ips})
|
|
|
|
|
|
def nom_alias(role: str, etiquette: str) -> str:
|
|
"""Alias d'hotes, PREFIXE par le tenant : deux tenants ont chacun leur nginx."""
|
|
return f"SETOPS_{etiquette}_{role.upper()}"
|
|
|
|
|
|
def _instance_active() -> tuple[str, dict] | tuple[None, None]:
|
|
"""(nom, nomenclature) de l'instance vers laquelle pointe le symlink `instance`."""
|
|
cible = (RACINE / "instance").resolve().name
|
|
for nom, _pfx, n in decouvrir():
|
|
if nom == cible:
|
|
return nom, n
|
|
return None, None
|
|
|
|
|
|
def construire(tenants: list[tuple[str, str, dict]]) -> dict:
|
|
"""Le devis, en structure — rendue ensuite en texte ou en JSON."""
|
|
flux_par_role = charger_flux()
|
|
retenus = flux_frontiere(flux_par_role)
|
|
_intr = intrants_frontiere()
|
|
if_wan = _ou_marqueur(_intr, "opnsense_if_wan", IF_WAN)
|
|
if_transit = _ou_marqueur(_intr, "opnsense_if_transit", IF_TRANSIT)
|
|
if_gestion = _ou_marqueur(_intr, "opnsense_if_gestion", IF_GESTION)
|
|
nom_actif, nomenclature_active = _instance_active()
|
|
|
|
alias: dict[str, dict] = {}
|
|
for nom, pfx, n in tenants:
|
|
alias[f"SETOPS_TENANT_{pfx}{n['index']}"] = {
|
|
"type": "network",
|
|
"contenu": [supernet_de(n["index"])],
|
|
"description": f"Supernet du tenant {nom} (index {n['index']})",
|
|
}
|
|
|
|
# Reseaux d'administration : UN ALIAS PAR TENANT, jamais une union. Chaque tenant
|
|
# declare les siens (intrant `nftables_admin_ssh`) et ils n'ouvrent QUE son supernet :
|
|
# une union laisserait le plan de gestion d'un tenant entrer chez le voisin, ce que
|
|
# les ACL de switch interdisent par ailleurs. La frontiere ne doit pas rouvrir ce que
|
|
# l'isolation inter-tenant ferme.
|
|
#
|
|
# Chaque CIDR est en outre RANGE PAR INTERFACE D'ARRIVEE. Une regle OPNsense est
|
|
# toujours `in` sur l'interface par ou le paquet entre : posee ailleurs, elle ne peut
|
|
# tout simplement jamais correspondre. Un plan de gestion directement attache a la
|
|
# frontiere (`10.0.0.0/24` ici) arrive par l'interface de GESTION ; un plan distant,
|
|
# qui revient par l'adresse publique ou un VPN, arrive par le WAN.
|
|
reseaux_gestion = reseaux_locaux_frontiere(underlay_mod.charger(), transit_underlay())
|
|
admin_par_tenant: dict[str, list[str]] = {}
|
|
admin_par_if: dict[str, dict[str, list[str]]] = {}
|
|
for nom, pfx, n in tenants:
|
|
etiquette = f"{pfx}{n['index']}"
|
|
cidrs = admin_de(nom)
|
|
admin_par_tenant[etiquette] = cidrs
|
|
locaux = [c for c in cidrs if _porte_par(c, reseaux_gestion)]
|
|
distants = [c for c in cidrs if c not in locaux]
|
|
admin_par_if[etiquette] = {"gestion": locaux, "wan": distants}
|
|
# UN ALIAS PAR INTERFACE, et le nom le dit. Un alias unique couvrant les deux
|
|
# cotes forcerait a poser la meme regle deux fois avec des sources qui ne peuvent
|
|
# pas arriver la : la moitie serait morte, sans que rien ne le signale.
|
|
for cle, portee in (("gestion", "GESTION"), ("wan", "WAN")):
|
|
part = admin_par_if[etiquette][cle]
|
|
if part:
|
|
alias[f"SETOPS_ADMIN_{etiquette}_{portee}"] = {
|
|
"type": "network",
|
|
"contenu": list(part),
|
|
"description": f"Reseaux d'administration de {nom} arrivant par "
|
|
f"{portee.lower()} (intrant nftables_admin_ssh) — "
|
|
f"source unique, partagee avec ses nftables",
|
|
}
|
|
# Union, pour les routes de RETOUR cote switch et la garde P24 : router n'est pas
|
|
# autoriser. Le switch doit savoir revenir vers tous les plans de gestion.
|
|
admin = sorted({c for cidrs in admin_par_tenant.values() for c in cidrs})
|
|
|
|
# Sources privees arrivant par le WAN : OPNsense active par defaut « Block private
|
|
# networks » sur cette interface, et ce filtre s'applique AVANT les regles. Une regle
|
|
# correcte ne serait donc jamais atteinte — panne muette, invisible dans le devis.
|
|
# RFC1918 exactement — c'est le perimetre de la case « Block private networks ».
|
|
# `ipaddress.is_private` serait trop large : il inclut les plages de documentation
|
|
# et le CGNAT, que la case voisine « Block bogon networks » traite separement.
|
|
#
|
|
# SEULS les CIDR ranges cote WAN comptent : conseiller de decocher le filtre pour un
|
|
# reseau qui arrive par la gestion affaiblirait l'interface publique sans rien ouvrir.
|
|
rfc1918 = [ipaddress.ip_network(b) for b in ("10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16")]
|
|
admin_prives = sorted({
|
|
c for parts in admin_par_if.values() for c in parts["wan"]
|
|
if any(ipaddress.ip_network(c, strict=False).subnet_of(b) for b in rfc1918)
|
|
})
|
|
|
|
supernet_actif = (
|
|
supernet_de(nomenclature_active["index"]) if nomenclature_active else None
|
|
)
|
|
|
|
# Les regles sont emises POUR CHAQUE TENANT FEDERE, pas seulement l'actif : la
|
|
# frontiere route deja vers eux tous (section 2). N'en filtrer qu'un laisserait les
|
|
# autres routes menant a un `block` muet.
|
|
regles: list[dict] = []
|
|
tenants_sans_inventaire: list[str] = []
|
|
tenants_sans_admin: list[str] = []
|
|
for nom, pfx, n in tenants:
|
|
etiquette = f"{pfx}{n['index']}"
|
|
alias_tenant = f"SETOPS_TENANT_{etiquette}"
|
|
inv = inventaire_de(nom)
|
|
if inv is None:
|
|
tenants_sans_inventaire.append(nom)
|
|
continue
|
|
for role, fl in retenus:
|
|
if role == ROLE_SOCLE:
|
|
destination = alias_tenant
|
|
else:
|
|
cibles = cibles_par_role(role, inv)
|
|
if not cibles:
|
|
continue # role absent de ce tenant : aucune regle a poser
|
|
a = nom_alias(role, etiquette)
|
|
alias[a] = {
|
|
"type": "host",
|
|
"contenu": cibles,
|
|
"description": f"Hotes portant {role} (instance {nom})",
|
|
}
|
|
destination = a
|
|
entrant = fl.get("sens") == "ingress"
|
|
# Le SSH de gestion est le SEUL flux `externe` a ne pas venir d'Internet :
|
|
# sa `raison` dit « l'acces depuis l'exterieur est filtre a l'OPNsense ».
|
|
ssh_gestion = entrant and "22" in _ports(fl)
|
|
if ssh_gestion and not admin_par_tenant.get(etiquette):
|
|
# Sans reseau d'administration declare, la regle n'aurait pas de source :
|
|
# l'omettre ferme l'acces, l'ouvrir a `any` exposerait le SSH a Internet.
|
|
tenants_sans_admin.append(nom)
|
|
continue
|
|
# Interface d'ARRIVEE du paquet, celle a laquelle OPNsense attache la regle.
|
|
# Sortant (la flotte vers Internet) : le lien de transit, par ou le paquet
|
|
# penetre le pare-feu. Entrant depuis Internet : le WAN. Entrant depuis un
|
|
# plan d'administration : l'interface ou ce plan est REELLEMENT attache —
|
|
# une regle posee sur la mauvaise interface ne correspond jamais.
|
|
if ssh_gestion:
|
|
portees = []
|
|
if admin_par_if[etiquette]["gestion"]:
|
|
portees.append((f"SETOPS_ADMIN_{etiquette}_GESTION", if_gestion))
|
|
if admin_par_if[etiquette]["wan"]:
|
|
portees.append((f"SETOPS_ADMIN_{etiquette}_WAN", if_wan))
|
|
else:
|
|
portees = [("any", if_wan if entrant else if_transit)]
|
|
for source, interface in portees:
|
|
regles.append({
|
|
"sens": "in" if entrant else "out",
|
|
"interface": interface,
|
|
"protocole": fl.get("protocole", "tcp"),
|
|
"source": source if entrant else destination,
|
|
"destination": destination if entrant else "any",
|
|
"ports": _ports(fl),
|
|
"chiffrement": fl.get("chiffrement"),
|
|
"role": role,
|
|
"tenant": nom,
|
|
"raison": fl.get("raison", ""),
|
|
})
|
|
|
|
intrants = intrants_frontiere()
|
|
transit = transit_underlay()
|
|
saut = prochain_saut(transit)
|
|
routes = [
|
|
{
|
|
"reseau": supernet_de(n["index"]),
|
|
"prochain_saut": saut,
|
|
"description": f"{nom} (index {n['index']}) — routage inter-zone "
|
|
+ ("dans la zone EVPN du tenant"
|
|
if underlay_mod.routage_tenants(underlay_mod.charger()) == "sdn"
|
|
else "sur les switches L3"),
|
|
}
|
|
for nom, _pfx, n in tenants
|
|
]
|
|
|
|
# NAT sortant : une regle par tenant, source = son alias de supernet, cible = l'adresse
|
|
# du WAN. Le mode automatique d'OPNsense ne couvre QUE les reseaux directement attaches ;
|
|
# un supernet joint par route statique en sort sans avertissement. Mesure du 2026-08-06 :
|
|
# le filtre laissait passer, un etat s'ouvrait sans `nat_addr`, et rien ne revenait.
|
|
nat = [
|
|
{
|
|
"interface": if_wan,
|
|
"source": f"SETOPS_TENANT_{pfx}{n['index']}",
|
|
"destination": "any",
|
|
"cible": "wanip",
|
|
"tenant": nom,
|
|
}
|
|
for nom, pfx, n in tenants
|
|
]
|
|
|
|
# Adresse du noeud de sortie primaire sur le lien de frontiere, calculee une fois :
|
|
# elle sert au prochain saut des routes ET au commentaire qui l'explique.
|
|
_ip_sortie = adresse_noeud_de_sortie(underlay_mod.charger(), transit)
|
|
|
|
return {
|
|
"role_frontiere": "nord-sud",
|
|
"instance_active": nom_actif,
|
|
"supernet_actif": supernet_actif,
|
|
"api_url": str(intrants.get("opnsense_api_url") or "").strip(),
|
|
"wan_ip": str(intrants.get("opnsense_wan_ip") or "").strip(),
|
|
"if_wan": _ou_marqueur(intrants, "opnsense_if_wan", IF_WAN),
|
|
"if_transit": _ou_marqueur(intrants, "opnsense_if_transit", IF_TRANSIT),
|
|
"if_gestion": _ou_marqueur(intrants, "opnsense_if_gestion", IF_GESTION),
|
|
"reseaux_gestion": reseaux_gestion,
|
|
"alias": alias,
|
|
"routes": routes,
|
|
"nat": nat,
|
|
"regles": regles,
|
|
"admin": admin,
|
|
"tenants_sans_inventaire": tenants_sans_inventaire,
|
|
"tenants_sans_admin": sorted(set(tenants_sans_admin)),
|
|
"admin_prives": admin_prives,
|
|
"noeud_sortie": ({"nom": str((_sdn_hebergeur() or {}).get("sortie_primaire") or ""),
|
|
"ip": _ip_sortie}
|
|
if _ip_sortie else None),
|
|
"transit": {
|
|
"nom": (transit or {}).get("nom"),
|
|
"vlan": (transit or {}).get("vlan"),
|
|
"sous_reseau": (transit or {}).get("sous_reseau"),
|
|
# L'adresse du commutateur SUR LE LIEN, pas le prochain saut des tenants :
|
|
# les deux ont diverge avec le SDN, ou le voisin de la frontiere devient un
|
|
# noeud de sortie. Ce champ decrit le cablage, pas le routage.
|
|
"svi_switch": (transit or {}).get("passerelle"),
|
|
"adresse_frontiere": (transit or {}).get("passerelle_sortie"),
|
|
"nom_frontiere": nom_frontiere(transit),
|
|
} if transit else None,
|
|
# Le switch route vers l'ADRESSE DU PARE-FEU sur le lien de transit
|
|
# (`passerelle_sortie`), pas vers son propre SVI ni vers un noeud de sortie :
|
|
# c'est la sortie de l'underlay. `devis_reseau` emet exactement ces lignes.
|
|
"reciproque_switch": [
|
|
f"ip route {cidr} -> {(transit or {}).get('passerelle_sortie') or PROCHAIN_SAUT}"
|
|
f" (retour vers l'administration)"
|
|
for cidr in admin
|
|
],
|
|
}
|
|
|
|
|
|
def rendre(devis: dict) -> str:
|
|
out = [
|
|
"# ============================================================",
|
|
"# DEVIS OPNSENSE — frontiere nord/sud (Set-OPS)",
|
|
"# Genere par scripts/devis_opnsense.py. NE PAS EDITER A LA MAIN.",
|
|
"# Derive du registre des flux (pair: externe) + des nomenclatures.",
|
|
("# Les passerelles de zone sont anycast sur les hyperviseurs (SDN EVPN) : aucun SVI ici."
|
|
if underlay_mod.routage_tenants(underlay_mod.charger()) == "sdn"
|
|
else "# Les passerelles de zone restent sur les switches L3 : aucun SVI ici."),
|
|
"# ============================================================",
|
|
"",
|
|
]
|
|
tr = devis.get("transit")
|
|
if tr and not tr.get("svi_switch"):
|
|
# Aucun SVI sur le lien : les switches ne routent pas, donc ils n'ont aucune
|
|
# route de retour a porter. Emettre la section quand meme laisserait croire
|
|
# qu'il reste un prerequis a satisfaire ailleurs.
|
|
out += [
|
|
"## 0. Prerequis reciproques — AUCUN",
|
|
"# Les switches ne routent pas : ils n'ont ni SVI sur ce lien, ni route de",
|
|
"# retour a porter. La frontiere est le seul equipement L3, et les noeuds de",
|
|
"# sortie EVPN lui sont ADJACENTS sur ce meme VLAN — rien ne s'interpose.",
|
|
]
|
|
else:
|
|
out += [
|
|
"## 0. Prerequis reciproques — COTE SWITCHES",
|
|
"# Sans ces routes de retour, le trafic part mais ne revient pas : la",
|
|
"# reponse sort par une autre interface et le pare-feu la jette (hors etat).",
|
|
]
|
|
out += [f"# {r}" for r in devis["reciproque_switch"]] or ["# (aucun reseau d'administration declare)"]
|
|
if tr:
|
|
out.append(f"# DEJA EMIS par `make devis-reseau` (section 5), derive du transit"
|
|
f" '{tr['nom']}' de l'underlay.")
|
|
else:
|
|
out.append("# A EMETTRE : aucun reseau de transit n'est declare dans underlay.yml")
|
|
out.append("# (cle `passerelle_sortie`). Sans lui, ni sortie ni retour ne sont derivables.")
|
|
out += [
|
|
"",
|
|
"## 1. Interfaces",
|
|
f"# Gestion : {devis['api_url'] or '(intrant opnsense_api_url a renseigner)'}",
|
|
f"# WAN : {devis['if_wan']}"
|
|
+ (f" — {devis['wan_ip']} (face publique de l'ecosysteme)" if devis.get("wan_ip") else ""),
|
|
f"# TRANSIT : {devis['if_transit']} (trunk 802.1Q vers les switches)",
|
|
f"# GESTION : {devis['if_gestion']}"
|
|
+ (f" — plan(s) attache(s) : {', '.join(devis['reseaux_gestion'])}"
|
|
if devis.get("reseaux_gestion") else " — aucun plan attache a la frontiere"),
|
|
]
|
|
if tr:
|
|
out += [
|
|
f"# lien '{tr['nom']}' {tr['sous_reseau']} — VLAN {tr['vlan']} a etiqueter",
|
|
(f"# switch {tr['svi_switch']} <-> frontiere" if tr.get("svi_switch")
|
|
else "# segment partage — aucun SVI de switch ; la frontiere y est")
|
|
+ f" {tr['adresse_frontiere']}"
|
|
+ (f" ({tr['nom_frontiere']})" if tr.get("nom_frontiere") else ""),
|
|
]
|
|
out += [
|
|
"",
|
|
]
|
|
if devis.get("admin_prives"):
|
|
out += [
|
|
f"# /!\\ DECOCHER « Block private networks » SUR {devis['if_wan'].upper()}.",
|
|
"# Des sources PRIVEES entrent par cette interface : "
|
|
+ ", ".join(devis["admin_prives"]) + ".",
|
|
"# Ce filtre d'interface s'applique AVANT les regles : coche, il jette le",
|
|
"# paquet d'administration sans qu'aucune regle ne soit consultee. La config",
|
|
"# parait juste, le SSH ne passe pas, et rien ne dit pourquoi.",
|
|
"",
|
|
]
|
|
out += [
|
|
"## 2. Routes statiques (vers les supernets tenants)",
|
|
]
|
|
if devis["routes"] and devis["routes"][0]["prochain_saut"] == NOEUD_SORTIE:
|
|
out += [
|
|
"# /!\\ LE PROCHAIN SAUT N'EST PAS LE COMMUTATEUR. En SDN EVPN, il ne route plus",
|
|
"# les tenants : une route pointee vers son SVI arriverait sur un equipement sans",
|
|
"# chemin vers le tenant. Le voisin de la frontiere est un NOEUD DE SORTIE — un",
|
|
"# hyperviseur qui porte la sortie du VRF.",
|
|
"# A DECLARER : `proxmox_sdn.sortie_primaire` chez l'hebergeur, et l'adresse de",
|
|
"# ce noeud sur le lien de transit dans underlay.yml. Les deux doivent concorder.",
|
|
]
|
|
elif devis.get("noeud_sortie"):
|
|
out += [
|
|
f"# Prochain saut = {devis['noeud_sortie']['nom']}, noeud de sortie PRIMAIRE, a son",
|
|
f"# adresse SUR LE LIEN DE FRONTIERE ({devis['noeud_sortie']['ip']}) — ni son underlay,",
|
|
"# ni sa gestion : la frontiere n'atteint que celle-la.",
|
|
"#",
|
|
"# Un seul saut, parce qu'une route statique n'en porte qu'un. Deux noeuds actifs",
|
|
"# en sortie avec une seule route en entree donneraient un chemin asymetrique : la",
|
|
"# reponse reviendrait par une interface ou l'etat n'a pas ete cree, et serait jetee.",
|
|
]
|
|
out += [
|
|
]
|
|
for r in devis["routes"]:
|
|
out.append(f"route add {r['reseau']} via {r['prochain_saut']} # {r['description']}")
|
|
out += ["", "## 2bis. NAT sortant (traduction des supernets tenants)"]
|
|
if devis.get("nat"):
|
|
out += [
|
|
"# Une route ne suffit pas : sans traduction, le paquet tenant quitte le WAN avec",
|
|
"# sa source PRIVEE et rien ne peut lui repondre. La panne est muette — le filtre",
|
|
"# laisse passer, un etat s'ouvre, et aucune reponse n'arrive jamais.",
|
|
"#",
|
|
"# Le mode « automatique » d'OPNsense ne traduit que les reseaux DIRECTEMENT",
|
|
"# ATTACHES. Un supernet tenant est joint par une ROUTE STATIQUE : il tombe hors",
|
|
"# de ce perimetre sans que rien ne le signale.",
|
|
]
|
|
for n in devis["nat"]:
|
|
out.append(f"nat on {n['interface']:8} from {n['source']:28} to any -> {n['cible']}"
|
|
f" # {n['tenant']}")
|
|
else:
|
|
out.append("# Aucun tenant federe : rien a traduire.")
|
|
out += ["", "## 3. Alias"]
|
|
for nom, a in devis["alias"].items():
|
|
contenu = ", ".join(a["contenu"]) if a["contenu"] else "(VIDE)"
|
|
out.append(f"alias {nom:32} {a['type']:8} {contenu}")
|
|
out.append(f"# {a['description']}")
|
|
out += ["", "## 4. Regles — PAR TENANT FEDERE (defaut : deny + journalisation)",
|
|
"# La frontiere route vers tous les tenants (section 2) : elle porte donc aussi",
|
|
"# leurs regles. Chacun n'est joignable en SSH que depuis SES PROPRES reseaux",
|
|
"# d'administration — jamais ceux du voisin, que l'isolation inter-tenant ferme.",
|
|
"# Chaque regle est `in` sur son interface d'ARRIVEE : le WAN pour ce qui vient",
|
|
"# d'Internet (et du VPN d'administration, qui revient par l'adresse publique),",
|
|
"# le lien de transit pour ce que la flotte emet vers l'exterieur."]
|
|
for nom in devis.get("tenants_sans_inventaire", []):
|
|
out.append(f"# /!\\ {nom} : aucun inventaire genere — AUCUNE regle. Route, puis bloque.")
|
|
for nom in devis.get("tenants_sans_admin", []):
|
|
out.append(f"# /!\\ {nom} : `nftables_admin_ssh` vide — regle SSH OMISE. L'ouvrir a "
|
|
f"`any` exposerait le SSH a Internet ; renseigner l'intrant.")
|
|
for rg in devis["regles"]:
|
|
ports = ",".join(rg["ports"])
|
|
# Dans OPNsense une regle est TOUJOURS `in` sur l'interface d'arrivee : c'est
|
|
# la que le paquet penetre le pare-feu. Le sens du flux (entrant/sortant vis-a-vis
|
|
# de l'ecosysteme) se lit alors dans l'interface, pas dans un mot-cle.
|
|
out.append(
|
|
f"pass in on {rg['interface']:8} proto {rg['protocole']:3} "
|
|
f"from {rg['source']:24} to {rg['destination']:24} port {ports:7} "
|
|
f"# {rg['role']} ({rg['chiffrement']})"
|
|
)
|
|
out.append(f"# {rg['raison']}")
|
|
sortants = sum(1 for r in devis["regles"] if r["sens"] == "out")
|
|
out += [
|
|
"",
|
|
"## 5. Defaut — DENY DANS LES DEUX SENS",
|
|
"# `block out` est un vrai default-deny : ce qui n'est pas declare en section 4",
|
|
f"# ne sort PAS. {sortants} regle(s) sortante(s) y figurent, toutes derivees du",
|
|
"# registre des flux (`sens: egress`, `pair: externe`).",
|
|
"#",
|
|
"# Un besoin sortant oublie ne se manifeste pas par un refus clair : apt expire,",
|
|
"# l'horloge derive et les certificats deviennent invalides des semaines plus tard.",
|
|
"# Le declarer dans `roles/<role>/meta/flux.yml` — ne jamais l'ouvrir ici a la main,",
|
|
"# la regle serait perdue a la prochaine generation.",
|
|
"block in log all # tout ce qui n'est pas ci-dessus",
|
|
"block out log all",
|
|
]
|
|
return "\n".join(out)
|
|
|
|
|
|
def verifier(devis: dict) -> tuple[bool, list[str]]:
|
|
"""Garde anti-lockout : la frontiere ne doit pas etre generee a l'aveugle."""
|
|
erreurs: list[str] = []
|
|
if not devis["admin"]:
|
|
erreurs.append(
|
|
"Aucun reseau d'administration declare (intrant `nftables_admin_ssh` vide) : "
|
|
"la regle SSH entrante n'aurait aucune source et le defaut deny fermerait "
|
|
"l'acces d'administration. Declarer le(s) CIDR d'admin (VPN compris)."
|
|
)
|
|
if not devis["routes"]:
|
|
erreurs.append("Aucun tenant federe decouvert : rien a router.")
|
|
for nom, a in devis["alias"].items():
|
|
if not a["contenu"]:
|
|
erreurs.append(f"Alias {nom} vide : une regle le referencerait dans le vide.")
|
|
|
|
# Une regle est TOUJOURS `in` sur l'interface d'arrivee du paquet. Posee ailleurs elle
|
|
# ne correspond jamais : le devis affirme un acces qui n'existe pas, et la panne ne se
|
|
# voit qu'a l'usage. On confronte donc chaque source d'administration a l'endroit ou
|
|
# son reseau est reellement attache.
|
|
# Une route vers un supernet sans NAT correspondant est le defaut MUET du 2026-08-06 :
|
|
# le filtre autorise, un etat s'ouvre, et aucune reponse ne revient jamais. On exige donc
|
|
# que tout supernet route soit aussi traduit.
|
|
traduits = {n["source"] for n in devis.get("nat") or []}
|
|
for nom, a in devis["alias"].items():
|
|
if nom.startswith("SETOPS_TENANT_") and nom not in traduits:
|
|
erreurs.append(
|
|
f"Supernet {nom} route mais NON traduit : ses paquets quitteraient le WAN "
|
|
"avec une source privee. Le filtre laisserait passer et rien ne reviendrait."
|
|
)
|
|
|
|
gestion = devis.get("reseaux_gestion") or []
|
|
for r in devis["regles"]:
|
|
src = str(r.get("source") or "")
|
|
if not src.startswith("SETOPS_ADMIN_"):
|
|
continue
|
|
cidrs = (devis["alias"].get(src) or {}).get("contenu") or []
|
|
locaux = [c for c in cidrs if _porte_par(c, gestion)]
|
|
if r["interface"] == devis.get("if_gestion") and len(locaux) != len(cidrs):
|
|
erreurs.append(
|
|
f"Regle sur {r['interface']} (gestion) avec la source {src} : "
|
|
+ ", ".join(c for c in cidrs if c not in locaux)
|
|
+ " n'est attache a aucun plan local de la frontiere — le paquet "
|
|
"arriverait par une autre interface et la regle ne matcherait jamais."
|
|
)
|
|
if r["interface"] == devis.get("if_wan") and locaux:
|
|
erreurs.append(
|
|
f"Regle sur {r['interface']} (WAN) avec la source {src} : "
|
|
+ ", ".join(locaux)
|
|
+ " est directement attache a la frontiere, donc arrive par "
|
|
f"{devis.get('if_gestion')} — cette regle est morte."
|
|
)
|
|
return (not erreurs), erreurs
|
|
|
|
|
|
def main(argv: list[str]) -> int:
|
|
ap = argparse.ArgumentParser(description=__doc__,
|
|
formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
ap.add_argument("--json", action="store_true", help="Sortie structuree (future API OPNsense).")
|
|
ap.add_argument("--verifier", action="store_true", help="Valide le devis sans l'afficher.")
|
|
args = ap.parse_args(argv)
|
|
|
|
tenants = decouvrir()
|
|
if not tenants:
|
|
print("Aucune instance federee trouvee (../*/plan/nomenclature.yml).", file=sys.stderr)
|
|
return 1
|
|
devis = construire(tenants)
|
|
|
|
if args.verifier:
|
|
ok, erreurs = verifier(devis)
|
|
for e in erreurs:
|
|
print(f"ECHEC : {e}", file=sys.stderr)
|
|
if ok:
|
|
print(f"CONFORME : frontiere nord/sud, {len(devis['regles'])} regles, "
|
|
f"{len(devis['routes'])} routes, admin={','.join(devis['admin'])}.")
|
|
return 0 if ok else 1
|
|
|
|
print(json.dumps(devis, indent=2, ensure_ascii=False) if args.json else rendre(devis))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv[1:]))
|