Alliance-Boreale/Set-OPS-Public
2
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity
Set-OPS-Public/scripts/devis_opnsense.py
Daniel Allaire d1db332ed3
Some checks are pending
verifier / verifier (push) Waiting to run
Details
frontiere : lire ses reglages a la racine du depot de site 
`opnsense.yml` decrivait le monde physique depuis les group_vars d'un tenant. Le devis et
le GUI le cherchent desormais d'abord a la racine du depot de site, comme underlay.yml et
proxmox-hebergeur.yml, par la meme derivation depuis le symlink.

CE QUE LE MAUVAIS RANGEMENT A COUTE : l'adresse d'API de la frontiere y etait restee a
10.0.0.1 apres migration vers 10.17.0.1. `make frontiere-appliquer` restait suspendu sur
une adresse morte, sans aucun message — trouve par l'exploitant en lancant la commande
dans son terminal, apres que j'aie moi-meme conclu deux fois a tort.

DEUX LECONS, ecrites plutot que corrigees en silence :
- un objet range chez celui qui n'en est pas responsable derive sans que personne le voie ;
- mes commandes s'executent dans ma session : l'exploitant ne voit pas leur sortie. Une
  commande lente ressemble alors a un blocage, et un blocage a une commande lente. Pour
  toute ecriture longue sur du materiel, c'est a lui de la lancer.

Les anciens emplacements restent lus : un site pas encore migre continue de fonctionner.
make verifier 41/41.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-23 00:47:32 -04:00

831 lines
42 KiB
Python
Raw Blame History

#!/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 sous_reseau_de, 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, decouvrir_du_site, 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.
A LA RACINE DU DEPOT DE SITE depuis le 2026-08-22, comme `underlay.yml` et
`proxmox-hebergeur.yml` : la frontiere est un objet du MONDE PHYSIQUE. Tant que ce
fichier vivait dans les `group_vars` d'un tenant, son adresse d'API a pu rester
perimee sans que personne la voie — rangee chez quelqu'un qui n'en est pas
responsable. L'applicateur est reste suspendu sur `10.0.0.1`, une adresse morte, sans
message : la panne la plus couteuse a diagnostiquer.
Les deux anciens emplacements restent lus, pour qu'un site non encore migre continue
de fonctionner.
"""
heb = depot_hebergeur()
if heb is not None and (heb / "opnsense.yml").is_file():
return yaml.safe_load((heb / "opnsense.yml").read_text(encoding="utf-8")) or {}
for base in (heb, _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()
# L'alias d'un tenant vaut ses SOUS-RESEAUX ATTRIBUES, pas son supernet.
#
# Tant qu'il valait `10.27.0.0/16`, nos propres regles autorisaient
# `admin -> 10.27.0.0/16:22` : le filtre laissait donc passer une connexion vers
# n'importe quelle adresse du /16, y compris celles ou aucune machine n'existe.
# Mesure du 2026-08-09, apres avoir retire les routes /16 : un `connect()` vers
# 10.27.99.99:22 aboutissait encore, et l'etat pf portait la description de NOTRE
# regle. Le symptome que nous avons attribue deux jours durant a une fonction
# d'anti-usurpation de la frontiere venait d'abord de nos declarations trop larges.
#
# Meme geste que pour les routes, et pour la meme raison : ne declarer que ce qui
# existe. L'alias sert a la fois de DESTINATION aux regles de filtrage et de SOURCE
# au NAT sortant ; le retrecir resserre correctement les deux.
alias: dict[str, dict] = {}
for nom, pfx, n in tenants:
alias[f"SETOPS_TENANT_{pfx}{n['index']}"] = {
"type": "network",
"contenu": [
sous_reseau_de(n["index"], int(z))
for z in sorted((n.get("categories") or {}), key=int)
],
"description": f"Sous-reseaux attribues du tenant {nom} (index {n['index']})",
}
# « Vers Internet » ne veut pas dire « vers n'importe ou ». Les flux sortants d'un
# tenant (DNS, SMTP, NTP, HTTP/HTTPS) visaient `any` : la destination n'excluait donc
# ni le plan de gestion, ni la frontiere elle-meme, ni le supernet du VOISIN. Mesure du
# 2026-08-09 depuis une VM du tenant : `https://10.0.0.1/` — la console d'administration
# du pare-feu — repondait. Autorise par notre propre regle, pas par une regle heritee.
#
# Cet alias dit ce qui est INTERNE ; les regles sortantes le prennent en destination
# NIEE. Il vaut les TROIS BLOCS PRIVES (RFC 1918) et non la liste de nos reseaux : une
# exclusion incomplete ne protege rien, et une liste derivee de l'underlay laissait
# dehors `192.168.11.0/24` — le plan de gestion herite, celui-la meme ou fuyaient les
# paquets du 2026-08-09. Un reseau interne ajoute demain est couvert sans rien changer.
#
# C'est aussi la definition honnete de « vers Internet » : tout ce qui n'est pas prive.
alias["SETOPS_INTERNES"] = {
"type": "network",
"contenu": ["10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16"],
"description": "Espaces prives RFC 1918 — sert en destination NIEE aux flux "
"sortants : « vers Internet » n'est pas « vers n'importe ou »",
}
# 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))
elif entrant:
# Un service publie est joint DEPUIS INTERNET par le WAN, et le plus
# souvent aussi depuis le VLAN d'administration — le poste de
# l'exploitant y est.
#
# Sauf quand le flux declare `poste: false`. Tous les services publies ne
# s'adressent pas a un humain : le `25` entrant de Postfix est un flux
# SERVEUR A SERVEUR (les MX distants). Le declarer joignable depuis le
# poste faisait dire a la frontiere ce qu'aucune autre couche n'honorait —
# mesure du 2026-08-09 : la bordure l'autorisait, le nftables de l'hote le
# refusait. Deux couches qui ne declarent pas la meme politique, c'est
# une politique qu'on ne peut plus lire.
#
# Le mot-cle vit dans `meta/flux.yml`, avec le role qui sait ce que son
# port veut dire. Le generateur, lui, ne connait aucun numero de port.
# Ce second chemin fonctionnait jusqu'ici par la regle d'usine
# `Default allow LAN to any`, qui autorise TOUT depuis le LAN : notre
# defaut-deny n'y avait donc aucun effet. Mesure du 2026-08-09 : depuis
# le poste, le 443 d'un nginx repondait alors que seul le 22 est declare.
#
# Cette regle heritee n'a AUCUNE API (verifie : zero regle non-Set-OPS
# visible) — elle se desactive a la main. La declarer ici est ce qui rend
# cette desactivation possible SANS couper l'exploitant de ses propres
# services. Sans elle, fermer le LAN fermerait aussi ses consoles web.
portees = [("any", if_wan)]
if admin_par_if[etiquette]["gestion"] and fl.get("poste", True):
portees.append((f"SETOPS_ADMIN_{etiquette}_GESTION", if_gestion))
else:
portees = [("any", 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,
# Sortant : « tout sauf l'interne ». Le `!` est porte par le devis
# lui-meme pour qu'il apparaisse dans l'identite de la regle — sans
# quoi la version niee et la version ouverte auraient la meme cle et
# la seconde ne remplacerait jamais la premiere.
"destination": destination if entrant else "!SETOPS_INTERNES",
"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)
# UNE ROUTE PAR SOUS-RESEAU REELLEMENT ATTRIBUE, et non une par supernet.
#
# Router le /16 entier faisait porter a la frontiere des destinations qui n'existent
# nulle part. Ces paquets atteignaient le noeud de sortie, y arrivaient dans la table
# PRINCIPALE — pas dans le VRF, qui n'est atteint que par les /24 annonces en BGP — et
# repartaient vers la passerelle du reseau d'ADMINISTRATION. Mesure du 2026-08-09 :
# `ip route get 10.27.99.99` rendait `via 192.168.11.254`.
#
# C'est aussi ce qui faisait reussir tout `connect()` depuis le VLAN d'administration,
# y compris vers des adresses inexistantes — symptome que nous avons attribue pendant
# deux jours a une fonction d'anti-usurpation de la frontiere, alors que c'etait un
# routage trop large.
#
# La frontiere ne route desormais QUE ce qui existe. L'alias `SETOPS_TENANT_*`, qui
# porte le NAT sortant et sert de destination aux regles, enumere exactement les memes
# sous-reseaux : la garde P24 compare ces deux listes et reste donc satisfaite.
_sdn = underlay_mod.routage_tenants(underlay_mod.charger()) == "sdn"
routes = [
{
"reseau": sous_reseau_de(n["index"], int(z)),
"prochain_saut": saut,
"tenant": nom,
"description": f"{nom} — {cat.get('libelle', f'zone{z}')} (zone {z}, index "
f"{n['index']}) — " + ("zone EVPN du tenant" if _sdn
else "switches L3"),
}
for nom, _pfx, n in tenants
for z, cat in sorted((n.get("categories") or {}).items(), key=lambda x: int(x[0]))
]
# 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."
)
# Ce qui est ROUTE et ce qui est AUTORISE doivent designer exactement les memes
# reseaux. Un alias plus large que les routes laisse le filtre approuver des
# destinations qui n'existent nulle part — defaut mesure le 2026-08-09 : l'alias valait
# le supernet /16, nos regles autorisaient donc `admin -> tout le /16:22`, et un
# `connect()` vers une adresse inexistante aboutissait toujours. Un alias plus etroit
# que les routes est la panne symetrique : la frontiere achemine vers un reseau que
# le filtre refuse. Les deux se voient ici, pas a l'usage.
routes_declarees = {str(r["reseau"]) for r in devis.get("routes") or []}
autorises: set[str] = set()
for nom, a in devis["alias"].items():
if nom.startswith("SETOPS_TENANT_"):
autorises |= {str(c) for c in a["contenu"]}
if routes_declarees != autorises:
for large in sorted(autorises - routes_declarees):
erreurs.append(
f"{large} est autorise par un alias de tenant mais n'est pas route : le "
"filtre approuverait une destination que la frontiere ne sait pas joindre."
)
for etroit in sorted(routes_declarees - autorises):
erreurs.append(
f"{etroit} est route mais absent des alias de tenant : la frontiere y "
"acheminerait des paquets que le filtre refuserait."
)
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)
# UNE FRONTIERE NE POLICE QUE LES TENANTS DE SON SITE (2026-08-14). Le filtre est
# devenu la source unique des trois devis qui equipent un site — commutateur, SDN et
# frontiere : cf. `devis_reseau.decouvrir_du_site`.
tenants = decouvrir_du_site()
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:]))
Reference in a new issue View git blame Copy permalink