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 788c5073dc DNS public, phase 1 : le locataire ecrit, le site sert 
Role serveur_dns_public (secondaire public, transfert signe TSIG, aucune zone interne) ;
serveur_powerdns exerce enfin autorite primaire-cache dans une instance pdns@public a part,
pour que le site ne puisse jamais interroger la zone .internal. Relations derivees des plans
des locataires, mots de flux dns_public_site et primaires_dns_locataires. Eprouve avant
d ecrire : allow-axfr-ips et TSIG sont alternatifs, le primaire notifie aussi ses NS, le
serial fige aurait gele le secondaire. P82 refuse l exposition sans DNSSEC.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-16 18:08:20 -04:00

1648 lines
91 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]
# LES ROLES QUE LE DEPOT CONNAIT — pour distinguer un PAIR QUI NOMME UN ROLE d'un mot-cle
# de portee (`externe`, `flotte`, `admin`, `edge`, `voisins_site`, `frontiere`, `fabric`,
# `localhost`). La distinction decide de ce qu'on fait d'une destination introuvable :
# ignorer le flux, ou l'ouvrir vers l'Internet. Les deux n'ont rien a voir.
_ROLES_CONNUS = {d.name for d in (RACINE / "roles").iterdir() if d.is_dir()}
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>"
# L'interface ou arrive le trafic des MACHINES DU SITE (vlan 30 / vmbr3). Une regle
# posee sur la mauvaise interface ne correspond jamais : le trafic du site ne penetre pas
# le pare-feu par le lien de transit des tenants, mais par sa propre patte.
IF_SITE = "<IF-SITE>"
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 que la FRONTIERE doit porter, trie.
Deux familles, et pas une :
`externe` — le nord-sud : l'Internet d'un cote, un tenant de l'autre.
`voisins_site` — l'est-ouest INTER-TENANT. Deux tenants de la meme fabric vivent
sur des VLAN distincts, routes par la frontiere : leur trafic la
traverse, donc elle doit le porter.
ET UNE TROISIEME QUI N'EN EST PAS UNE : `runner_site`, l'INSEMINATION. Ce trafic
traverse bien la frontiere — le runner du SITE et celui du tenant vivent dans deux
VRF — mais le paquet y PENETRE PAR LA PATTE DU SITE, pas par le lien de transit. La
regle qui lui correspond est donc emise plus bas, avec les machines du site, depuis
la declaration `egress` de `serveur_ops_site`.
L'emettre ici AUSSI produirait une seconde regle attachee a la mauvaise interface :
jamais evaluee, impossible a distinguer d'une regle utile, et comptee comme telle par
tout ce qui audite ce devis. La declaration `ingress` du tenant n'est pas perdue pour
autant — c'est elle qui pose la regle nftables sur la machine visee, et elle seule.
OUBLIER LA SECONDE, C'ETAIT DECLARER UN FLUX QUE PERSONNE N'APPLIQUE (2026-08-24). Le
mot `voisins_site` etait accepte par la validation, le devis rendait zero regle, et
rien ne le signalait — le chainage des caches n'aurait simplement jamais fonctionne.
"""
retenus = [
(role, fl)
for role, flux in sorted(flux_par_role.items())
for fl in flux
if ("externe" in _pairs(fl) or "voisins_site" in _pairs(fl)
or "frontiere" 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})
# OPNsense refuse un nom d'alias de 32 caracteres ou plus. La contrainte n'etait nulle
# part, et elle s'est manifestee a l'APPLICATION — pas au devis :
#
# SETOPS_PATI29_SERVEUR_ARTEFACTS_SITE (36)
# « The name must start with a letter [...] and be less than 32 characters »
#
# L'outil a bien refuse d'appliquer quoi que ce soit, mais le devis, lui, annoncait un
# objet impossible. Un devis qui promet ce que la cible refusera n'est pas un devis.
LONGUEUR_MAX_ALIAS = 31
# Meme abreviation que le pare-feu est-ouest (`devis_proxmox_fw` : serveur_ -> srv-), pour
# qu'un seul schema de lecture serve aux deux couches.
ABREGES = (("SERVEUR_", "SRV_"), ("CLIENT_", "CLI_"))
def nom_alias(role: str, etiquette: str) -> str:
"""Alias d'hotes, PREFIXE par le tenant : deux tenants ont chacun leur nginx.
Abrege SEULEMENT si le nom complet depasse — un nom court reste lisible tel quel, et
la flotte existante n'est pas renommee pour la commodite d'un seul role.
"""
nom = f"SETOPS_{etiquette}_{role.upper()}"
if len(nom) <= LONGUEUR_MAX_ALIAS:
return nom
court = role.upper()
for long, bref in ABREGES:
court = court.replace(long, bref)
nom = f"SETOPS_{etiquette}_{court}"
if len(nom) > LONGUEUR_MAX_ALIAS:
# Mieux vaut un devis qui refuse qu'un devis qui promet l'impossible.
raise SystemExit(
f"Nom d'alias trop long pour OPNsense ({len(nom)} > {LONGUEUR_MAX_ALIAS}) : {nom}.\n"
f"Raccourcir le nom du role `{role}`.")
return nom
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
# CES TROIS LECTEURS ONT DEMENAGE DANS `underlay` (2026-08-28). `resoudre_flux` a eu
# besoin du plan du site a son tour, pour resoudre le pair `runner_site` ; une seconde
# copie privee aurait fini par diverger de celle-ci sans que rien ne le signale. Lecon de
# P41 appliquee au plan du site : une seule resolution, partagee. Les noms locaux restent
# pour ne pas reecrire trente appels.
_lire_plan_site = underlay_mod.lire_plan_site
_machines_du_plan_site = underlay_mod.machines_site
def _services_du_plan_site() -> dict[str, list[str]]:
"""{machine du site: [groupes qu'elle porte]} — tels que l'INVENTAIRE les donne.
LA SOURCE EST L'INVENTAIRE, PAS `applications.yml` SEUL (2026-09-10).
`underlay.services_site()` lit le registre des applications : il voit `serveur_loki`,
`serveur_icinga`, `serveur_forgejo` — ce qu'une machine du site heberge COMME SERVICE.
Il ne voit pas les INTEGRATIONS UNIVERSELLES, que `site_inventaire` ajoute a toutes les
machines sans qu'aucune application ne les declare : `client_metrique`, `client_journal`,
`client_pki`, `client_sante`.
Ces groupes-la portent pourtant des flux, et des flux qui ouvrent des PORTS D'ECOUTE :
`client_metrique` declare `ingress 9100`. Invisible au devis, la regle d'entree n'etait
jamais emise — et comme `serveur_prometheus` declare l'`egress` en face, sa regle de
sortie tombait sur `!SETOPS_INTERNES`. Le devis autorisait donc Prometheus a scruter le
monde entier sur 9100 SAUF les six machines qu'il devait scruter.
Meme forme que le defaut du 2026-09-02 sur `serveur_icinga` : les deux declarations
etaient justes, c'est leur RENCONTRE au site qui manquait.
Ne retient que les groupes qui sont de vrais roles — un groupe de commodite
(`site`, la flotte, le socle) n'a pas de flux et n'a pas a produire d'alias.
"""
try:
import site_inventaire # noqa: PLC0415 — le site peut ne pas etre monte
inv = site_inventaire.inventaire()
except Exception:
return underlay_mod.services_site()
hotes = set((inv.get("_meta") or {}).get("hostvars") or {})
if not hotes:
return underlay_mod.services_site()
out: dict[str, list[str]] = {h: [] for h in hotes}
for groupe, corps in inv.items():
if groupe == "_meta" or not isinstance(corps, dict):
continue
if not (RACINE / "roles" / str(groupe)).is_dir():
continue
for h in (corps.get("hosts") or []):
if h in out:
out[h].append(str(groupe))
return {h: sorted(set(g)) for h, g in out.items()}
def _ports_du_plan_site() -> dict[str, str]:
"""Port declare au plan pour chaque role du site — ce que `port: derive` vaut ici.
UN PORT NON NUMERIQUE N'EST PAS UNE ECOUTE (voir `resoudre_flux`). `derive` dit que le
port depend du deploiement : la forge ecoute 3000 derriere un edge, 443 quand elle
sert son propre TLS. Le rendre tel quel produirait une regle que le boitier refuse —
on l'a deja paye sur le pare-feu est-ouest.
"""
apps = (_lire_plan_site("applications.yml") or {}).get("applications") or {}
out: dict[str, str] = {}
for app in apps.values():
if app.get("groupe") and app.get("port"):
out[str(app["groupe"])] = str(app["port"])
return out
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)
if_site = _ou_marqueur(_intr, "opnsense_if_site", IF_SITE)
# UNE PATTE PAR ZONE DU SITE (2026-08-25). Le site n'est plus un `/24` plat : chaque
# nature d'autorite a son VLAN, et le trafic penetre le pare-feu par SA patte. Une
# regle posee sur la mauvaise interface ne correspond jamais.
if_zones = dict(_intr.get("opnsense_if_zones") or {})
def _if_de(reseau: str | None) -> str:
"""L'interface ou arrive le trafic de cette zone ; l'ancienne patte a defaut."""
return if_zones.get(str(reseau or ""), if_site)
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] = []
# PAS D'ALIAS POUR LA SOURCE DE L'INSEMINATION ICI. La regle est emise du cote SITE
# (voir `flux_frontiere`), ou la source est deja l'alias de role du runner —
# `SETOPS_SITE_SERVEUR_OPS_SITE`. En poser un second ici aurait laisse sur la
# frontiere un alias qu'aucune regle ne reference : `appliquer_opnsense` l'aurait
# cree, et personne n'aurait su a quoi il sert.
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"
if "frontiere" in _pairs(fl):
# TRAITE PAR LE BLOC DEDIE, PLUS BAS — et il faut sauter ICI, sinon le
# meme flux sort DEUX FOIS : une regle nommee vers `SETOPS_FRONTIERE`, et
# une seconde par le chemin generique vers `!SETOPS_INTERNES`. La seconde
# annulerait tout le benefice — elle rouvre le port vers l'Internet entier.
continue
# 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 and "voisins_site" in _pairs(fl):
# UN FLUX ENTRE TENANTS DU MEME SITE.
#
# Ce n'est ni `flotte` (mon ecosysteme) ni `externe` (l'Internet) : le
# paquet vient d'un VOISIN pose sur la meme fabric. Il arrive par le meme
# lien de transit que tout le reste — les tenants ne se distinguent pas
# par une interface mais par leur ALIAS SOURCE, deja construit plus haut
# (`SETOPS_TENANT_<etiquette>`).
#
# LA REGLE APPARTIENT AU SITE, PAS A L'UN DES DEUX TENANTS. C'est le
# runner de site qui prepare le terrain : il configure les VNets, les
# routes et les flux pour que les plans des tenants tiennent. Aucun
# ecosysteme n'ouvre de porte chez un autre.
portees = [
(f"SETOPS_TENANT_{p2}{n2['index']}", if_transit)
for nom2, p2, n2 in tenants if nom2 != nom
]
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))
elif "voisins_site" in _pairs(fl):
# SORTANT VERS UN VOISIN : la destination est NOMMEE, pas « tout sauf
# l'interne ». Rendre `!SETOPS_INTERNES` ici aurait ouvert le port vers
# l'INTERNET — mesure du 2026-08-24, visible au devis avant application.
#
# Une regle par voisin : une regle porte une destination.
portees = [
(f"SETOPS_TENANT_{p2}{n2['index']}", if_transit)
for nom2, p2, n2 in tenants if nom2 != nom
]
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.
# Sortant vers un voisin : la destination est le voisin lui-meme,
# portee par `source` puisque `portees` la transporte. Sortant vers
# l'Internet : « tout sauf l'interne ».
"destination": destination if entrant
else (source if "voisins_site" in _pairs(fl) else "!SETOPS_INTERNES"),
"ports": _ports(fl),
"chiffrement": fl.get("chiffrement"),
"role": role,
"tenant": nom,
"raison": fl.get("raison", ""),
})
intrants = intrants_frontiere()
# --- LES MACHINES DU SITE -------------------------------------------------
#
# Elles ne sont dans AUCUN plan de tenant : la boucle ci-dessus ne pouvait donc pas
# les voir, et le devis rendait zero regle pour elles sans rien signaler. Un devis
# muet sur une machine qui existe n'est pas un devis.
#
# Leur trafic arrive par la patte du site (`if_site`), pas par le lien de transit.
# Une regle posee sur la mauvaise interface ne correspond jamais — c'est la panne la
# plus silencieuse de cette couche.
_u = underlay_mod.charger()
# LES MACHINES DU SITE VIENNENT DE SON PLAN, plus de l'underlay (2026-08-25).
#
# En deplacant le plan hors de `underlay.yml`, j'ai vide `machines()` sans reprendre
# ici : le devis a cesse de voir le site et a propose de RETIRER ses 36 objets — tous
# ses alias et toutes ses regles. Aucune preuve ne couvre le devis de la frontiere du
# site, donc `make prouver` restait vert. C'est le plan avant application qui l'a
# attrape, et rien d'autre ne l'aurait fait.
_machines_site = _machines_du_plan_site()
_services_site = _services_du_plan_site()
_ports_site = _ports_du_plan_site()
# CE QU'UN EDGE DOIT JOINDRE POUR RELAYER (2026-09-15).
#
# `serveur_nginx` declare `egress port: derive, pair: expositions` — « je sors vers
# ce que je publie ». Ce mot n'etait resolu NULLE PART : ni par le pare-feu d'hote
# (qui ne filtre pas l'egress), ni ici. La declaration existait, aucun moteur ne la
# consommait.
#
# ELLE N'A JAMAIS EU BESOIN DE L'ETRE. Chez un locataire, les passerelles de zone
# sont tenues par le SDN de Proxmox : le trafic edge -> amont ne traverse pas le
# boitier. Au SITE, elles sont tenues par la frontiere — et c'est le premier endroit
# ou un edge doit franchir la bordure pour atteindre ce qu'il relaie.
#
# Mesure du 2026-09-15 : les trois noms du site repondaient en TLS verifie, puis
# `http=000`. Le devis le disait, et c'est la seule raison qu'on l'ait vu :
#
# note : serveur_nginx declare un port `derive` que le plan du site ne resout
# pas — aucune regle emise.
#
# UNE REGLE PAR AMONT, jamais une regle large : un edge qui publie trois services ne
# doit pouvoir joindre que ces trois-la, sur LEUR port. `expositions` rendu comme
# « tout le site » aurait ouvert l'edge sur ses voisins — l'exact contraire de ce
# qu'une zone de publication separee cherche a obtenir.
#
# `underlay_mod`, ET PAS `U` — LA MEME ERREUR QU'HIER, AVALEE DE LA MEME FACON.
# Le module s'importe sous ce nom ICI ; `U` est l'alias d'un AUTRE fichier. Un
# `except Exception` large a transforme le `NameError` en « aucune exposition », et
# le devis a rendu une note plausible au lieu d'une erreur. On attrape donc ce que
# la lecture d'un fichier peut vraiment lever, et rien de plus : une faute de frappe
# doit faire du bruit.
_amonts_expositions: list[tuple[str, int]] = []
try:
_plan_apps = (underlay_mod.lire_plan_site("applications.yml")
or {}).get("applications") or {}
except (OSError, ValueError) as _e:
print(f"note : plan du site illisible ({_e.__class__.__name__}) — aucune "
f"exposition derivee.", file=sys.stderr)
_plan_apps = {}
for _app in _plan_apps.values():
if not isinstance(_app, dict) or not _app.get("expose"):
continue
_g, _port = str(_app.get("groupe") or ""), _app.get("port")
if _g and _g != "serveur_nginx" and str(_port or "").isdigit():
_amonts_expositions.append((_g, int(_port)))
_amonts_expositions = sorted(set(_amonts_expositions))
if _machines_site:
_res_site = {r.get("nom"): r for r in underlay_mod.reseaux(_u)}
_sous_site = sorted({str(_res_site[m["reseau"]]["sous_reseau"])
for m in _machines_site if m.get("reseau") in _res_site})
alias["SETOPS_SITE"] = {
"type": "network",
"contenu": _sous_site,
"description": "Reseau des machines de l'hebergeur (runner de SITE, cache racine)",
}
# LA FABRIC — ce que le runner du site doit joindre pour MATERIALISER : l'API des
# hyperviseurs, leur shell, l'API de la frontiere. C'est du RFC 1918 : le rendre
# par `!SETOPS_INTERNES` (la forme de « vers Internet ») l'aurait exclu tout en
# ayant l'air d'ouvrir le flux. D'ou le mot `fabric`, et cet alias-ci.
_fabric = sorted({str(h["ip"]) for h in underlay_mod.hotes(_u)
if h.get("role") in ("hyperviseur", "frontiere") and h.get("ip")})
alias["SETOPS_FABRIC"] = {
"type": "host",
"contenu": _fabric,
"description": "Materiel de l'hebergeur : hyperviseurs et frontiere — "
"ce que le runner du SITE pilote pour materialiser",
}
# LE PLAN D'ADMINISTRATION, source du SSH vers le site. Sans lui, la regle
# n'aurait pas de source : l'omettre ferme l'acces, l'ouvrir a `any` exposerait
# le SSH du runner — la machine la plus puissante du site — a l'Internet.
_admin_site = [str(r["sous_reseau"]) for r in underlay_mod.reseaux(_u)
if r.get("nom") == "management" and r.get("sous_reseau")]
if _admin_site:
alias["SETOPS_ADMIN_SITE"] = {
"type": "network",
"contenu": _admin_site,
"description": "Plan d'administration — seule source du SSH vers le site",
}
# Le socle vaut aussi pour le site : ses machines sont des Debian de la flotte,
# et c'est lui qui porte leur SSH. `site_inventaire.py` les y range.
# LE SITE RESOUT CHEZ SA PASSERELLE, ET IL FAUT LE DIRE (2026-08-25).
#
# Un tenant a son propre resolveur DANS son reseau : son DNS ne traverse jamais la
# frontiere, et aucun flux ne le declare. Le site n'en a pas — trois machines ne
# justifient pas un service de plus — il interroge donc la frontiere elle-meme.
#
# Sans cette regle, le default-deny bloque la requete, et la panne ne ressemble pas
# a un pare-feu : `/etc/resolv.conf` est correct, Unbound ecoute bien sur toutes les
# interfaces, le port repond au test TCP — et `apt update` echoue quand meme. On a
# cherche du cote du resolveur pendant que c'etait le filtre.
#
# La destination est l'adresse DECLAREE dans `site.dns_amorcage`, pas `any` : ouvrir
# le 53 vers le monde depuis le site serait une sortie DNS non policee.
_resolveur = str((_u.get("site") or {}).get("dns_amorcage") or "").strip()
if _resolveur:
alias["SETOPS_RESOLVEUR_SITE"] = {
"type": "host",
"contenu": [_resolveur],
"description": "Resolveur des machines du site — leur passerelle, "
"declaree dans `site.dns_amorcage`",
}
for _proto in ("udp", "tcp"):
regles.append({
"sens": "out",
"interface": if_site,
"protocole": _proto,
"source": "SETOPS_SITE",
"destination": "SETOPS_RESOLVEUR_SITE",
"ports": ["53"],
"chiffrement": "clair",
"role": "site",
"tenant": "SITE",
"raison": "Resolution DNS des machines du site aupres de leur "
"passerelle : le site n'a pas de resolveur a lui.",
})
# LES RUNNERS DES TENANTS QUE CE SITE HEBERGE — destination de l'insemination.
#
# Un seul alias pour tous : ce que la regle dit, c'est « le runner du site peut
# amorcer le runner d'un tenant de CE site ». Le perimetre est celui du registre
# `underlay.tenants`, deja la source du reste du devis.
#
# Ce sont des adresses d'hotes, pas des reseaux : ouvrir le supernet d'un tenant
# au SSH du site rendrait l'insemination indiscernable d'un droit d'entree
# general — exactement ce que l'isolation refuse.
_runners_tenants = sorted({ip for _n2, _p2, _x2 in tenants
for ip in (cibles_par_role("serveur_ops_tenant",
inventaire_de(_n2))
if inventaire_de(_n2) else [])})
if _runners_tenants:
alias["SETOPS_RUNNERS_TENANTS"] = {
"type": "host",
"contenu": _runners_tenants,
"description": "Runners des tenants heberges — cible de l'insemination",
}
# LES AUTORITATIFS DES LOCATAIRES — la source du DNS public du site (2026-09-16).
#
# Meme forme que les runners : des HOTES, pas des reseaux. Le serveur public tire la
# zone d'un locataire ; lui ouvrir le supernet entier ferait d'un transfert de zone
# un droit d'entree general. Et la regle ne suffit pas a lire : sans la cle TSIG de
# sa relation, le primaire refuse le transfert.
# SEULEMENT LES LOCATAIRES QUI PUBLIENT (resserre le 2026-09-16). La premiere
# version prenait l'autoritatif de TOUS les locataires du site : Patient0, qui n'a
# aucune zone publique, recevait une ouverture 5300 depuis le serveur public. Rien
# n'y ecoutait — mais un chemin ouvert vers une machine qui n'en a pas besoin est
# exactement ce que le moindre privilege refuse.
def _publie(_nom_tenant: str) -> bool:
try:
import devis_reseau as _dr
_f = _dr.DOSSIER_INSTANCES / _nom_tenant / "plan" / "domaines.yml"
_d = (yaml.safe_load(_f.read_text(encoding="utf-8")) or {}).get("domaines_publics") or {}
return any(isinstance(_c, dict) and str(_c.get("autorite")) == "primaire-cache"
for _c in _d.values())
except (OSError, ImportError):
return False
_primaires_dns = sorted({ip for _n2, _p2, _x2 in tenants if _publie(_n2)
for ip in (cibles_par_role("serveur_powerdns",
inventaire_de(_n2))
if inventaire_de(_n2) else [])})
if _primaires_dns:
alias["SETOPS_DNS_PRIMAIRES_TENANTS"] = {
"type": "host",
"contenu": _primaires_dns,
"description": "Autoritatifs des tenants heberges — source du DNS public",
}
_roles_site = sorted({g for gs in _services_site.values() for g in gs}
| {"serveur_debian"})
for _role in _roles_site:
_ips = sorted({str(m["ip"]) for m in _machines_site
if _role == "serveur_debian"
or _role in _services_site.get(m["nom"], [])})
_a = nom_alias(_role, "SITE")
alias[_a] = {
"type": "host",
"contenu": _ips,
"description": f"Machines du site portant {_role}",
}
for _fl in flux_par_role.get(_role, []):
_entrant = _fl.get("sens") == "ingress"
_pairs_fl = _pairs(_fl)
if "frontiere" in _pairs_fl:
continue # bloc dedie, plus bas (voir la boucle des tenants)
if _entrant and "voisins_site" in _pairs_fl:
# Un tenant qui s'adresse au site : le paquet penetre le pare-feu par
# le lien de TRANSIT, pas par la patte du site. Une regle par voisin,
# parce qu'une regle porte une source.
_portees = [(f"SETOPS_TENANT_{p2}{n2['index']}", if_transit)
for _n2, p2, n2 in tenants]
elif _entrant and "flotte" in _pairs_fl:
# CE QUI ETAIT GRATUIT DEVIENT POLICE (2026-08-25).
#
# `flotte` veut dire « les machines de mon ecosysteme ». Dans un `/24`
# plat, ce trafic ne traversait rien : le cache, la forge, le
# resolveur et l'AC se joignaient en L2, sans qu'aucune regle
# n'existe — ni ne manque.
#
# Une zone par autorite change cela : une machine qui interroge le
# resolveur ou tire un paquet du cache passe desormais par la
# frontiere. Ce qui n'avait pas besoin d'etre dit doit l'etre.
#
# Une regle PAR ZONE SOURCE : le paquet penetre le pare-feu par la
# patte de la zone d'ou il part, et une regle posee ailleurs ne
# correspondrait jamais. La destination reste nommee — le role vise,
# jamais la zone entiere.
_portees = [("SETOPS_SITE", z) for z in
sorted({_if_de(m.get("reseau")) for m in _machines_site})]
# UN FLUX A DEUX PAIRES N'OBTIENT QU'UNE BRANCHE (2026-09-12).
#
# La chaine de `elif` range chaque flux dans UN cas. La forge du site
# declare `pair: [flotte, admin]` : `flotte` capte la premiere, et la
# part administrative disparait sans que rien ne le dise. Mesure a la
# premiere reconstruction — la machine acceptait l'exploitant, la
# frontiere le refusait, et les deux couches se croyaient d'accord.
#
# On AJOUTE la portee administrative au lieu d'en faire un cas exclusif.
if "admin" in _pairs_fl and _admin_site:
_portees += [("SETOPS_ADMIN_SITE", if_gestion)]
elif _entrant and "fabric" in _pairs_fl:
# LA FABRIC PARLE AUSSI AU SITE (2026-09-10).
#
# `fabric` n'etait connu qu'en DESTINATION — « le runner du site pilote
# les hyperviseurs ». Un flux ENTRANT depuis la fabric tombait donc dans
# la branche « rien d'autre n'entre » et n'emettait AUCUNE regle, sans
# rien dire. Meme forme que le trou des integrations universelles,
# trouve le matin meme : le generateur ne voyait qu'une moitie du monde.
#
# Le cas concret : les hyperviseurs POUSSENT leurs journaux vers Loki.
# Ils ne peuvent pas etre scrutes (route par defaut gelee, D-57) mais ils
# peuvent emettre — un push n'attend qu'un accuse, la ou un scrape exige
# un chemin symetrique.
#
# Une regle par ZONE du site ou vit le role vise, comme pour `flotte` :
# le paquet penetre le pare-feu par la patte de la zone DESTINATAIRE.
# L'INTERFACE EST CELLE PAR OU LE PAQUET ARRIVE (D-61), donc celle
# qui fait face a la FABRIC — pas celle de la destination. Derivee du
# reseau ou vivent les hyperviseurs, jamais ecrite : le jour ou la
# fabric change de patte, la regle suit.
# SEULEMENT LES RESEAUX DONT ON CONNAIT LA PATTE. Un hyperviseur a
# trois adresses (gestion, transport, transit) ; deux de ces reseaux ne
# touchent pas la frontiere et n'ont pas d'interface declaree. Sans ce
# filtre, `_if_de` retombe sur la patte par DEFAUT et pose une regle sur
# une interface que ce trafic n'emprunte jamais — une regle qui a l'air
# d'ouvrir un flux et ne correspond a rien.
_res_fabric = sorted({str(h.get("reseau")) for h in underlay_mod.hotes(_u)
if h.get("role") == "hyperviseur"
and str(h.get("reseau")) in if_zones})
_portees = [("SETOPS_FABRIC", _if_de(r)) for r in _res_fabric]
if not _portees:
continue
elif _entrant and ("admin" in _pairs_fl or "22" in _ports(_fl)) and _admin_site:
# `admin` N'EST PAS QUE LE SSH (2026-09-12). Cette branche testait le
# PORT — « si c'est du 22, ça vient de l'administration ». Juste tant
# que le seul flux administratif etait SSH.
#
# Mesure a la premiere reconstruction du site : l'exploitant declare
# `admin` sur le 443 de sa forge pour pouvoir l'amorcer, la machine
# accepte, et la frontiere continue de refuser — parce que le devis ne
# reconnaissait l'administration qu'a son port. Deux couches, une seule
# au courant : c'est la panne la plus silencieuse de cette pile.
#
# On teste desormais la PAIRE DECLAREE, et le port reste ce qu'il est.
# Le SSH de gestion arrive par la patte d'ADMINISTRATION de la
# frontiere, pas par celle du site : c'est la que le paquet penetre
# le pare-feu. Une regle posee ailleurs ne correspondrait jamais.
_portees = [("SETOPS_ADMIN_SITE", if_gestion)]
elif _entrant and (set(_pairs_fl) & set(_roles_site)):
# UN ROLE DU SITE QUI EN APPELLE UN AUTRE — ET CE N'EST PAS
# « L'EXTERIEUR » (mesure du 2026-09-02).
#
# La branche precedente disait « rien d'autre n'entre chez le site
# depuis l'exterieur » et sautait le flux. C'etait vrai tant qu'un
# `pair` nomme designait quelque chose de lointain. Mais un `pair`
# peut nommer un ROLE DU SITE LUI-MEME, pose dans une AUTRE ZONE — et
# depuis le decoupage en zones, deux machines du site ne se parlent
# qu'a travers la frontiere.
#
# CE QUE CA A COUTE : `serveur_icinga` declare `ingress 5665` depuis
# `serveur_backup`, et `serveur_backup` declare l'`egress` en face.
# Les deux etaient justes. La regle d'entree n'existait pas, celle de
# sortie visait `!SETOPS_INTERNES` — le depot pouvait donc parler a
# un Icinga du monde entier SAUF a celui de son propre site. Le
# rapport passif expirait en 20 s, et la supervision qu'on venait de
# monter ne recevait rien.
#
# Une regle par ZONE SOURCE, comme pour `flotte` : le paquet penetre
# par la patte de la zone d'ou il part. La source reste le ROLE
# appelant, jamais sa zone entiere — c'est la difference entre
# « le depot peut joindre le temoin » et « la zone du depot le peut ».
_appelants = sorted(set(_pairs_fl) & set(_roles_site))
_portees = []
for _ap in _appelants:
_zones_ap = sorted({_if_de(m.get("reseau")) for m in _machines_site
if _ap == "serveur_debian"
or _ap in _services_site.get(m["nom"], [])})
_portees += [(nom_alias(_ap, "SITE"), z) for z in _zones_ap]
if not _portees:
continue
elif _entrant:
continue # rien d'autre n'entre chez le site depuis l'exterieur
else:
# Une portee par ZONE ou ce role est present : deux machines d'un meme
# role dans deux zones differentes ne partagent pas leur patte.
_zones = sorted({_if_de(m.get("reseau")) for m in _machines_site
if _role == "serveur_debian"
or _role in _services_site.get(m["nom"], [])})
_portees = [(_a, z) for z in _zones] or [(_a, if_site)]
# `derive` se resout par le port declare au plan ; sans lui, on n'emet
# rien plutot qu'une regle que le boitier refusera.
# Une destination qui n'existe pas ne produit pas de regle : aucun
# tenant de ce site ne declare encore de runner.
if "serveur_ops_tenant" in _pairs_fl and not _runners_tenants:
continue
# UN EDGE QUI SORT VERS SES EXPOSITIONS : une regle par amont, chacune
# avec SON port. Traite ici et pas plus bas, parce que ni le port ni la
# destination ne sont ceux du role qui declare le flux — ils sont ceux
# des services qu'il relaie.
if not _entrant and "expositions" in _pairs_fl:
if not _amonts_expositions:
print(f"note : {_role} sort vers ses expositions, mais le plan du "
f"site n'en declare aucune avec un port — aucune regle "
f"emise.", file=sys.stderr)
continue
for _g_amont, _port_amont in _amonts_expositions:
if _g_amont not in _roles_site:
continue
for _src, _itf in _portees:
regles.append({
"sens": "out",
"interface": _itf,
"protocole": _fl.get("protocole", "tcp"),
"source": _a,
"destination": nom_alias(_g_amont, "SITE"),
"ports": [str(_port_amont)],
"chiffrement": _fl.get("chiffrement"),
"role": _role,
"tenant": "SITE",
"raison": f"{_fl.get('raison', '')} (amont {_g_amont})",
})
continue
_p = _ports(_fl)
# L'ICMP N'A PAS DE PORT, ET SON `port` N'EST PAS SYMBOLIQUE (2026-09-09).
#
# Le test ci-dessous dit « non numerique = a resoudre par le plan », ce qui
# est vrai pour `derive` — le port d'un service que seul le plan connait.
# C'est FAUX pour un flux ICMP : `echo-request` et `frag-needed` sont des
# litteraux, pas des inconnues. Ils tombaient donc dans la branche
# « le plan ne resout pas » et AUCUNE regle n'etait emise.
#
# CE QUE CA A COUTE : l'Icinga du site tenait 6 de ses 7 machines pour
# MORTES. `hostalive` est un ping ; le site est decoupe en zones ; la
# frontiere avalait l'ICMP inter-zones faute de regle. Et Icinga SUPPRIME
# les notifications des services d'un hote DOWN — la supervision n'alertait
# plus de rien tout en ayant l'air de fonctionner.
#
# LE PMTUD, LUI, N'ETAIT PAS CASSE — et il faut le dire pour ne pas gonfler
# la portee de ce correctif. Les memes six regles manquaient, mais mesure :
#
# site-mon-01 -> autre zone du site : MTU de chemin 1500
# site-mon-01 -> Internet : MTU de chemin 1500
# ops-01 (tenant, overlay EVPN) : MTU 1450
#
# Le site est a 1500 de bout en bout : aucun lien ne plafonne plus bas,
# donc aucun « fragmentation necessaire » n'avait lieu d'etre emis. La
# contrainte a 1450 vit chez les TENANTS. C'etait une protection ABSENTE
# (elle compte des que le site parle a un correspondant qui plafonne plus
# bas, ou si une zone passe un jour sur l'overlay), pas une panne active.
#
# CE QUE LA REGLE EMISE AUTORISE, EXACTEMENT. `appliquer_opnsense` n'envoie
# `destination_port` que pour TCP et UDP : une regle ICMP ouvre le
# PROTOCOLE entre les deux pairs, pas le seul type declare. Le type reste
# porte par la regle d'HOTE, que `resoudre_flux` emet precisement
# (`icmp type echo-request accept`). La frontiere dit qui peut parler a
# qui ; l'hote dit ce qu'il accepte d'entendre. On ne pretend donc pas ici
# une finesse qu'on n'applique pas — c'est le sens de la ligne « (type
# filtre a l'hote) » ajoutee au devis.
if str(_fl.get("protocole", "")).lower() in ("icmp", "icmpv6"):
pass
elif any(not str(x).isdigit() for x in _p):
_reel = _ports_site.get(_role)
if not _reel:
# SUR STDERR, JAMAIS STDOUT : ce script rend du JSON, et une
# note au milieu le rend illisible a `appliquer_opnsense`.
print(f"note : {_role} declare un port `{'/'.join(_p)}` que le "
f"plan du site ne resout pas — aucune regle emise.",
file=sys.stderr)
continue
_p = [_reel]
# UNE SORTIE VERS UN ROLE DU SITE VISE CE ROLE, PAS « L'EXTERIEUR ».
#
# Symetrique du correctif du 2026-09-02, qui n'avait traite que l'entree.
# `!SETOPS_INTERNES` est la forme de « vers Internet » : c'est le bon
# rendu quand le pair est lointain (un depot Debian, un serveur NTP). Quand
# le pair NOMME UN ROLE DU SITE, cette destination dit exactement l'inverse
# de ce que le flux declare — elle exclut la seule machine visee.
#
# Mesure du 2026-09-10 : `client_journal` sortait vers `!SETOPS_INTERNES`
# sur 3100. Le Loki du site etant interne, la regle de sortie ne le couvrait
# pas ; seule la regle d'ENTREE, emise en face, faisait passer les journaux.
# Le devis autorisait donc a expedier les journaux du site a n'importe quel
# Loki du monde, et a nul autre endroit qu'a celui-la.
if _entrant:
_dests = [_a]
elif "serveur_ops_tenant" in _pairs_fl:
_dests = ["SETOPS_RUNNERS_TENANTS"]
elif "primaires_dns_locataires" in _pairs_fl:
# AUCUN PRIMAIRE, AUCUNE REGLE — jamais le repli. Tomber dans le cas
# general rendrait `!SETOPS_INTERNES` : le serveur public du site
# pourrait alors ouvrir un transfert vers N'IMPORTE QUELLE machine
# de l'Internet. C'est « une source vide ouvre le port », cote
# destination, que P79 garde deja pour les roles nommes.
if "SETOPS_DNS_PRIMAIRES_TENANTS" not in alias:
print(f"note : {_role} declare une sortie vers les primaires DNS "
f"des locataires, et aucun locataire de ce site n'en a — "
f"aucune regle emise.", file=sys.stderr)
continue
_dests = ["SETOPS_DNS_PRIMAIRES_TENANTS"]
elif "fabric" in _pairs_fl:
_dests = ["SETOPS_FABRIC"]
else:
_internes = sorted(set(_pairs_fl) & set(_roles_site))
if _internes:
_dests = [nom_alias(_x, "SITE") for _x in _internes]
elif set(_pairs_fl) & _ROLES_CONNUS:
# UN PAIR QUI NOMME UN ROLE ABSENT N'EST PAS « L'INTERNET ».
#
# Mesure du 2026-09-14, au devis et avant toute ecriture :
# `serveur_icingaweb2` declare une sortie LDAPS vers
# `serveur_openldap`. Un SITE n'a pas d'annuaire — ce sont des
# services d'ecosysteme. `_internes` etait donc vide, et le repli
# rendait `!SETOPS_INTERNES` : la frontiere aurait autorise la
# console a parler LDAPS a N'IMPORTE QUELLE machine du monde,
# pour joindre un annuaire qui n'existe pas.
#
# C'est « une source vide ouvre le port », cote DESTINATION. Le
# repli est juste quand le pair est un mot-cle lointain
# (`externe`, un depot Debian, un serveur NTP) ; il est faux des
# que le pair NOMME un role, parce qu'alors le flux ne parle pas
# de l'Internet — il parle d'une machine, et elle n'est pas la.
print(f"note : {_role} declare une sortie vers "
f"{'/'.join(sorted(set(_pairs_fl) & _ROLES_CONNUS))}, "
f"absent de ce site — aucune regle emise (le repli aurait "
f"ouvert le port vers l'Internet).", file=sys.stderr)
continue
else:
_dests = ["!SETOPS_INTERNES"]
for _src, _itf in _portees:
for _dst in _dests:
regles.append({
"sens": "in" if _entrant else "out",
"interface": _itf,
"protocole": _fl.get("protocole", "tcp"),
"source": _src if _entrant else _a,
"destination": _dst,
"ports": _p,
"chiffrement": _fl.get("chiffrement"),
"role": _role,
"tenant": "SITE",
"raison": _fl.get("raison", ""),
})
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
]
# LE SITE AUSSI SORT. Son reseau est directement attache a la frontiere (OPT2), donc
# le mode automatique d'OPNsense le couvrirait — mais nous ne sommes plus en mode
# automatique : declarer les tenants a la main l'a bascule. Compter sur un mode qu'on
# a soi-meme quitte, c'est le genre de supposition qui se paie au premier `apt update`
# du cache racine, quand plus rien ne se telecharge et que tout a l'air correct.
if _machines_site and "SETOPS_SITE" in alias:
nat.append({
"interface": if_wan,
"source": "SETOPS_SITE",
"destination": "any",
"cible": "wanip",
"tenant": "SITE",
})
# LES SILENCES — refuser sans CONSIGNER (2026-08-27).
#
# Un pare-feu refuse deja tout ce qui n'est pas declare ; le probleme n'est pas ce
# qu'il bloque, c'est ce qu'il ECRIT. Mesure du jour : 982 000 entrees par jour, dont
# 82 % un balayage Internet contre le port VNC et le reste du bavardage de decouverte
# (mDNS, SSDP, NetBIOS) du reseau local. La consequence n'est pas cosmetique — la
# fenetre utile du journal tombait a QUARANTE-QUATRE SECONDES, moins que le temps de
# reproduire un defaut. Un journal noye ne dit rien, exactement comme un journal mort.
#
# Une regle de silence ne change AUCUN comportement : ce qu'elle bloque etait deja
# refuse par le defaut. Elle ne fait que taire une trace que personne ne lira. C'est
# pourquoi elle se declare ici, avec son MOTIF : sans lui, une regle `block` muette
# dans un pare-feu est indiscernable d'un oubli.
for s in (intrants.get("opnsense_silences") or []):
_dest = str(s.get("destination") or "any")
# Une liste d'adresses devient un ALIAS, pas une regle par adresse : l'alias
# nomme l'intention, et le boitier montre alors POURQUOI c'est tu.
if s.get("destinations"):
_nom_alias = f"SETOPS_SILENCE_{str(s['nom']).upper()}"
alias[_nom_alias] = {
"type": "network",
"contenu": [str(x) for x in s["destinations"]],
"description": f"Silence — {s.get('motif') or s['nom']}",
}
_dest = _nom_alias
regles.append({
"sens": "in",
"interface": str(s.get("interface") or if_wan),
"protocole": str(s.get("protocole") or "any"),
"source": str(s.get("source") or "any"),
"destination": _dest,
"ports": [str(p) for p in (s.get("ports") or [])],
"role": f"silence:{s.get('nom') or 'bruit'}",
"tenant": "SILENCE",
"action": "block",
"journaliser": False,
# 900 : APRES tous les `pass` (sequence 1) et Wireguard (101). OPNsense evalue
# en `quick` — un blocage large place avant eux fermerait le courrier, le web
# et l'acces distant d'un seul coup.
"sequence": str(s.get("sequence") or 900),
"chiffrement": "sans objet",
"raison": str(s.get("motif") or "bruit de journal, deja refuse par defaut"),
})
# DEUX DECLARATIONS JUSTES NE FONT QU'UNE REGLE (2026-09-10).
#
# Un flux interne au site est declare DES DEUX COTES : `client_journal` sort vers
# Loki, `serveur_loki` accepte les journaux. C'est voulu — chaque role dit ce qui le
# concerne, sans savoir qui est en face. Depuis que les deux resolvent la meme paire
# d'alias, ils produisent le meme quintuplet, et la frontiere se retrouvait avec deux
# regles identiques : le boitier les applique toutes les deux, l'exploitant se
# demande laquelle compte, et la suivante qu'on retire ne change rien.
#
# LE `sens` N'EST PAS DANS LA CLE, et c'est le coeur de l'affaire : `appliquer_opnsense`
# pose TOUTES les regles en `direction: in` (D-61 — le devis raisonne en ARRIVEE sur
# l'interface). Deux regles qui ne different que par leur `sens` sont donc, sur le
# boitier, le meme filtre pose deux fois. Seul leur marqueur differait.
#
# On dedoublonne sur ce que la frontiere applique VRAIMENT : interface, protocole,
# source, destination, ports, sequence. Pas sur la raison, qui differe selon le role
# qui l'a dite — la premiere rencontree la garde, et c'est la seule chose qu'on perd.
# C'EST LA FORME `ingress` QUI SURVIT, et ce n'est pas indifferent : le marqueur pose
# sur le boitier contient le `sens`. Laisser gagner l'un ou l'autre au hasard de
# l'ordre alphabetique des roles ferait RETIRER puis RECREER des regles deja justes —
# `serveur_forgejo -> serveur_postgresql` disparaissait pour renaitre a l'identique
# sous un autre marqueur. Une reconciliation qui remue ce qui va bien devient une
# reconciliation qu'on n'ose plus lancer.
def _cle_regle(r: dict) -> tuple:
return (r.get("interface"), r.get("protocole"),
r.get("source"), r.get("destination"),
tuple(r.get("ports") or ()), r.get("sequence"),
r.get("action"), bool(r.get("journaliser")))
_entrantes = {_cle_regle(_r) for _r in regles if _r.get("sens") == "in"}
_vues: set[tuple] = set()
_uniques = []
for _r in regles:
_cle = _cle_regle(_r)
if _cle in _vues:
continue
if _cle in _entrantes and _r.get("sens") != "in":
continue
_vues.add(_cle)
_uniques.append(_r)
regles = _uniques
# --- LA FRONTIERE COMME PAIR : L'HEURE (2026-09-11) ----------------------------
#
# `pair: frontiere` designe la frontiere ELLE-MEME comme destination — ni l'Internet,
# ni un voisin, ni un role du site. Le seul usage aujourd'hui est l'horloge : le socle
# declare `egress 123/udp` vers elle, et elle repond en `stratum 3`.
#
# POURQUOI UN BLOC A PART plutot qu'une branche de plus dans les deux boucles
# ci-dessus : la regle a exactement la MEME forme pour un locataire et pour une
# machine du site — seule change la patte d'arrivee. L'enfiler dans deux conditionnels
# deja longs aurait duplique la meme intention a deux endroits, et c'est ainsi qu'une
# moitie finit par prendre du retard sur l'autre.
#
# RIEN N'EST ECRIT EN DUR : le port, le protocole et la raison viennent de la
# declaration. Le jour ou un second flux vise la frontiere, il sort d'ici sans qu'on
# y touche.
_flux_frontiere = [(r, f) for r, flux in sorted(flux_par_role.items()) for f in flux
if "frontiere" in _pairs(f) and f.get("sens") != "ingress"]
if _flux_frontiere:
_pattes_frontiere = sorted({str(h["ip"]) for h in underlay_mod.hotes(_u)
if h.get("role") == "frontiere" and h.get("ip")
and str(h.get("etat", "actif")) != "reserve"})
if _pattes_frontiere:
alias["SETOPS_FRONTIERE"] = {
"type": "host",
"contenu": _pattes_frontiere,
"description": "Pattes de la frontiere active (autorite de temps)",
}
# LES PATTES D'ARRIVEE. Un locataire entre par le lien de transit — ses
# passerelles de zone sont tenues par le SDN, pas par la frontiere. Une
# machine du site entre par la patte de SA zone. D-61 : on raisonne en
# ARRIVEE, donc `in` dans les deux cas.
_portees_f = [(f"SETOPS_TENANT_{p2}{n2['index']}", if_transit)
for _n2, p2, n2 in tenants]
_portees_f += [("SETOPS_SITE", z) for z in
sorted({_if_de(m.get("reseau")) for m in _machines_site})]
for _role_f, _fl_f in _flux_frontiere:
for _src_f, _if_f in _portees_f:
regles.append({
"sens": "in",
"interface": _if_f,
"protocole": _fl_f.get("protocole", "udp"),
"source": _src_f,
"destination": "SETOPS_FRONTIERE",
"ports": _ports(_fl_f),
"chiffrement": _fl_f.get("chiffrement"),
"role": _role_f,
"tenant": "",
"raison": _fl_f.get("raison", ""),
})
# 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.
# L'ACTION S'AFFICHE (2026-08-27) : le rendu disait « pass » pour tout, ce qui
# etait vrai tant que l'outil ne savait qu'autoriser. Un devis qui affiche
# « pass » devant une regle de blocage serait pire qu'un devis muet.
_act = str(rg.get("action") or "pass")
_mut = "" if rg.get("journaliser", False) or _act == "pass" else " [non consigne]"
out.append(
f"{_act:4} 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']}){_mut}"
)
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