#!/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 = "" NOEUD_SORTIE = "" IF_TRANSIT = "" IF_WAN = "" 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 = "" 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) # `admin` AUSSI (2026-09-29). Le poste de l'exploitant est sur le plan de # gestion, derriere la frontiere : un flux `admin` la traverse. Le devis ne le # rendait qu'au travers des flux `externe` (option `poste`) — si bien qu'en # retirant `externe` de l'edge, le chemin du poste vers les interfaces des # locataires disparaissait avec lui. Proxmox et nftables, eux, l'appliquaient. or (fl.get("sens") == "ingress" and "admin" 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()) # LE TUNNEL D'ADMINISTRATION (2026-09-17). Un CIDR d'administration qui tombe dedans # n'arrive NI par la gestion NI par le WAN : il arrive par l'interface WireGuard. Range # ailleurs, sa regle serait posee sur une patte que ce trafic n'emprunte jamais — la # panne la plus silencieuse de cette couche, et la troisieme fois qu'elle se presente. _vpn_admin = (underlay_mod.lire_plan_site("10-intrants.yml") or {}).get("acces_admin_vpn") or {} _reseau_vpn = str(_vpn_admin.get("reseau") or "") if_vpn = _ou_marqueur(intrants_frontiere(), "opnsense_if_admin_vpn", "wireguard") def _par_le_vpn(cidr: str) -> bool: if not _reseau_vpn: return False try: return ipaddress.ip_network(cidr, strict=False).subnet_of( ipaddress.ip_network(_reseau_vpn, strict=False)) except ValueError: return False 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 par_vpn = [c for c in cidrs if _par_le_vpn(c)] locaux = [c for c in cidrs if c not in par_vpn and _porte_par(c, reseaux_gestion)] distants = [c for c in cidrs if c not in par_vpn and c not in locaux] admin_par_if[etiquette] = {"gestion": locaux, "wan": distants, "vpn": par_vpn} # 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"), ("vpn", "VPN")): 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] = [] # Ce que l'Internet frappe sur l'adresse publique, et ou la frontiere le renvoie. redirections: 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. # L'ADRESSE PUBLIQUE DE CHAQUE LOCATAIRE (2026-09-29) — attribuee par le site dans son # pool (`opnsense_ips_publiques`). Elle traduit sa sortie ET porte ses services publics. ips_publiques = {str(qui): str(ip) for ip, qui in (_intr.get("opnsense_ips_publiques") or {}).items() if str(qui) != "libre"} 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: # AVANT L'ALIAS (2026-09-17). Le saut plus bas arrivait APRES la creation de # l'alias du role : un flux vers la frontiere declare par `serveur_prometheus` # posait `SETOPS__SRV_PROMETHEUS` chez chaque locataire, qu'aucune regle # ne referencait. Le plan l'a montre avant toute ecriture. if "frontiere" in _pairs(fl): continue 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 ». # LE CHEMIN DE L'ADMINISTRATION : le SSH de gestion, et tout flux `admin`. Memes # portes (gestion, WAN d'administration, VPN), memes alias. par_admin = entrant and "admin" in _pairs(fl) and "externe" not in _pairs(fl) if par_admin and not all(p.isdigit() for p in _ports(fl)): print(f"note : {role} declare un flux `admin` sur un port non numerique " f"({', '.join(_ports(fl))}) — aucune regle a la frontiere de {nom}.", file=sys.stderr) continue ssh_gestion = entrant and ("22" in _ports(fl) or par_admin) 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)) if admin_par_if[etiquette]["vpn"]: portees.append((f"SETOPS_ADMIN_{etiquette}_VPN", if_vpn)) 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_`). # # 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)) if admin_par_if[etiquette]["vpn"] and fl.get("poste", True): portees.append((f"SETOPS_ADMIN_{etiquette}_VPN", if_vpn)) 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", ""), }) # UN SERVICE PUBLIC D'UN LOCATAIRE EST JOINT PAR SON ADRESSE PUBLIQUE (2026-09-29). # # La regle WAN ci-dessus vise une adresse privee : sans redirection devant elle, # elle ne recoit rien. La redirection part de l'adresse que le site attribue au # locataire — jamais de celle du WAN, que tous partagent : deux locataires qui # publient le 25 ne se disputent rien. Le SSH de gestion n'est pas publie ; l'ICMP # n'a pas de port. Une cible, ou rien : une redirection porte UNE adresse. if (entrant and "externe" in _pairs(fl) and not ssh_gestion and role != ROLE_SOCLE and str(fl.get("protocole", "tcp")).lower() in ("tcp", "udp")): ip_pub = ips_publiques.get(nom) if not ip_pub: print(f"note : {nom} n'a pas d'adresse publique au pool du site — " f"{role} ({', '.join(_ports(fl))}) n'est pas redirige.", file=sys.stderr) elif len(cibles) != 1: print(f"note : {role} est porte par {len(cibles)} machine(s) de {nom} — une " f"redirection exige une cible unique ; aucune.", file=sys.stderr) else: _pub = fl.get("port_public") for _pl in _ports(fl): if not str(_pl).isdigit(): continue redirections.append({ "interface": if_wan, "protocole": fl.get("protocole", "tcp"), "destination": ip_pub, "port_public": int(_pub) if str(_pub or "").isdigit() else int(_pl), "cible": cibles[0], "port_local": int(_pl), "role": role, "tenant": nom, }) 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", } if _reseau_vpn: alias["SETOPS_ADMIN_SITE_VPN"] = { "type": "network", "contenu": [_reseau_vpn], "description": "Tunnel WireGuard des administrateurs (plan du site, " "`acces_admin_vpn`) — arrive par l'interface WireGuard", } # 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 : un locataire sans # 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 or _reseau_vpn): # `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. # DEUX CHEMINS, DEUX PATTES. Le plan d'administration entre par la # gestion ; le tunnel des admins par l'interface WireGuard. Une seule # regle pour les deux serait juste pour l'un et morte pour l'autre. _portees = ([("SETOPS_ADMIN_SITE", if_gestion)] if _admin_site else []) if _reseau_vpn: _portees.append(("SETOPS_ADMIN_SITE_VPN", if_vpn)) 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 and "externe" in _pairs_fl: # UN SERVICE DU SITE PUBLIE SUR L'INTERNET (2026-09-16, DNS public). # # L'adresse publique est celle de la frontiere : sans redirection, un # filtre sur le WAN vers une adresse privee ne correspond a rien. Le # role dit DEUX ports — `port_public`, ce que l'Internet frappe, et # `port`, ce que la machine ecoute — parce qu'ils different : le 53 # public aboutit au frontal dnsdist (1053), jamais a PowerDNS. # # UNE CIBLE, OU RIEN. Une redirection porte une adresse ; deux machines # portant le role rendraient un choix que personne n'a fait. _pub = _fl.get("port_public") if not str(_pub or "").isdigit(): print(f"note : {_role} declare une entree `externe` sans " f"`port_public` — aucune redirection, aucune regle.", file=sys.stderr) continue if len(_ips) != 1: print(f"note : {_role} est porte par {len(_ips)} machine(s) du site " f"— une redirection publique exige une cible unique ; aucune " f"regle emise.", file=sys.stderr) continue for _pl in _ports(_fl): redirections.append({ "interface": if_wan, "protocole": _fl.get("protocole", "tcp"), "port_public": int(_pub), "cible": _ips[0], "port_local": int(_pl), "role": _role, }) # Le filtre voit le paquet APRES traduction : destination = la machine, # port = celui qu'elle ecoute. Source : tout l'Internet. _portees = [("any", if_wan)] 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. # # ET CHAQUE LOCATAIRE SORT PAR SA PROPRE ADRESSE (2026-09-29) — celle que le site lui # attribue. Tous sortaient par l'adresse du WAN : vu de l'Internet, les deux locataires # et le site etaient une seule machine (reputation du courriel, listes d'acces, journaux # de ceux qu'on joint). Sans attribution, le repli reste l'adresse du WAN. nat = [ { "interface": if_wan, "source": f"SETOPS_TENANT_{pfx}{n['index']}", "destination": "any", "cible": ips_publiques.get(nom, "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 (heure, metriques)", } # 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: # LE SOCLE PARLE A LA FRONTIERE DEPUIS PARTOUT ; UN ROLE, DEPUIS SES MACHINES # (2026-09-17). L'heure est demandee par toutes les machines : les portees # ci-dessus sont justes pour elle. La scrutation des metriques de la # frontiere, elle, n'est l'affaire que du Prometheus du SITE — l'ouvrir a # chaque locataire et a chaque zone aurait donne la carte de charge du # boitier a qui la demande. if _role_f == "serveur_debian": _portees_role = _portees_f else: _portees_role = [(nom_alias(_role_f, "SITE"), z) for z in sorted({_if_de(m.get("reseau")) for m in _machines_site if _role_f in _services_site.get(m["nom"], [])})] if not _portees_role: print(f"note : {_role_f} declare un flux vers la frontiere et aucune " f"machine du site ne le porte — aucune regle emise.", file=sys.stderr) for _src_f, _if_f in _portees_role: 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", ""), }) # --- CE QUE LE TUNNEL D'ADMINISTRATION ATTEINT (2026-09-17, mesure en production) --- # # DEUX TROUS QUE SEUL L'USAGE A MONTRES, une heure apres la pose du tunnel : # # 1. LE SSH DES MACHINES DU SITE. Le socle declare son 22 en `pair: [flotte, externe]`, # jamais `admin` : la boucle des roles du site le range donc dans « flotte », dont la # source est le site lui-meme. L'exploitant y arrivait par REBOND sur la frontiere — # la connexion partait alors du boitier, que pf laisse sortir sans regle. Par le # tunnel, il arrive comme une source exterieure, et plus rien ne l'autorise. Les # locataires, eux, marchaient : leur regle SSH nait de `nftables_admin_ssh`, qui # contient deja le reseau du tunnel. # # 2. LA FRONTIERE ELLE-MEME. Aucun flux ne la designe comme destination # d'administration : monter le tunnel revenait a perdre sa console et son API — donc # le moyen meme de reparer la regle manquante. C'est la panne qui se referme sur # celui qui la repare. if _reseau_vpn: _ports_frontiere = [str(p) for p in (_vpn_admin.get("ports_frontiere") or [22, 443])] _cibles_vpn = [("SETOPS_SITE_SERVEUR_DEBIAN", ["22"], "SSH d'administration vers TOUTES les machines du site, par le tunnel " "nominatif. Le socle declare son 22 en `flotte` : sans cette regle, " "l'acces ne survit qu'au rebond par la frontiere.")] if _ports_frontiere: _cibles_vpn.append(("SETOPS_FRONTIERE", _ports_frontiere, "Console et API de la frontiere depuis le tunnel : sans elles, " "monter le tunnel fait perdre le moyen de le corriger.")) # UNE REGLE PAR PORT (mesure du 2026-09-17). OPNsense refuse « 22,443 » dans le champ # `destination_port` : « Please specify a valid portnumber, name, alias or range ». # Ailleurs, chaque flux ne porte qu'un port et le cas ne s'etait jamais presente. for _dst_vpn, _ports_vpn, _raison_vpn in _cibles_vpn: if _dst_vpn not in alias: continue for _port_vpn in _ports_vpn: regles.append({ "sens": "in", "interface": if_vpn, "protocole": "tcp", "source": "SETOPS_ADMIN_SITE_VPN", "destination": _dst_vpn, "ports": [_port_vpn], "chiffrement": "ssh" if _port_vpn == "22" else "tls", "role": "acces_admin_vpn", "tenant": "SITE", "raison": _raison_vpn, }) # --- LES TUNNELS DES LOCATAIRES (2026-09-17) ----------------------------------- # # UN LOCATAIRE DECLARE QUI ENTRE CHEZ LUI (`plan/acces.yml`), le site porte la route. # Reseau et port derivent de son index ; l'instance WireGuard est posee par # `scripts/vpn_admin.py`. # # L'ISOLEMENT EST LA REGLE, PAS UNE INTENTION : la destination est le supernet DU MEME # locataire, jamais le site ni un voisin. P83 refuse toute regle qui s'en ecarterait. import devis_reseau as _dr for _nom_t, _pfx_t, _n_t in tenants: _idx_t = _n_t.get("index") if _idx_t is None: continue try: _acces_t = (yaml.safe_load( (_dr.DOSSIER_INSTANCES / _nom_t / "plan" / "acces.yml").read_text(encoding="utf-8")) or {}).get("acces_admin_vpn") or {} except (OSError, yaml.YAMLError): continue if not [n for n, c in _acces_t.items() if isinstance(c, dict) and str(c.get("etat", "present")) == "present"]: continue # aucun pair actif : aucune route, aucune regle _etiq_t = f"{_pfx_t}{_idx_t}" _alias_tunnel = f"SETOPS_ADMIN_{_etiq_t}_TUNNEL" alias[_alias_tunnel] = { "type": "network", "contenu": [f"10.{int(_idx_t)}.29.0/24"], "description": f"Tunnel d'administration de {_nom_t} (plan/acces.yml) — " f"n'atteint que son propre ecosysteme", } for _port_t, _chiffre_t, _pourquoi_t in ( ("22", "ssh", "SSH d'administration du locataire, par SON tunnel nominatif."), ("443", "tls", "Consoles web de son ecosysteme, par SON tunnel nominatif.")): regles.append({ "sens": "in", "interface": if_vpn, "protocole": "tcp", "source": _alias_tunnel, "destination": f"SETOPS_TENANT_{_etiq_t}", "ports": [_port_t], "chiffrement": _chiffre_t, "role": "acces_admin_vpn", "tenant": _nom_t, "raison": _pourquoi_t, }) if str(intrants.get("opnsense_wan_ip") or ""): alias.setdefault("SETOPS_FRONTIERE_WAN", { "type": "host", "contenu": [str(intrants["opnsense_wan_ip"])], "description": "Adresse publique de la frontiere — porte des tunnels d'administration", }) regles.append({ "sens": "in", "interface": if_wan, "protocole": "udp", "source": "any", "destination": "SETOPS_FRONTIERE_WAN", "ports": [str(52000 + int(_idx_t))], "chiffrement": "wireguard", "role": "acces_admin_vpn", "tenant": _nom_t, "raison": f"Poignee de main WireGuard des administrateurs de {_nom_t} " f"(port derive de son index).", }) # --- LA PORTE DU TUNNEL D'ADMINISTRATION (2026-09-17) -------------------------- # # Le tunnel ne peut pas s'etablir si son port n'est pas ouvert sur le WAN. C'est le SEUL # flux entrant depuis l'Internet que cette voie demande — tout le reste voyage dedans. # # LA DESTINATION EST L'ADRESSE PUBLIQUE, PAS `any`. WireGuard ne repond qu'a une poignee # de main correctement signee : ouvrir large ne l'affaiblirait pas, mais une frontiere # dont la lecture dit « ce port est ouvert vers le boitier » est plus juste qu'une qui # dit « vers n'importe quoi ». if _reseau_vpn and _vpn_admin.get("port") and str(intrants.get("opnsense_wan_ip") or ""): alias["SETOPS_FRONTIERE_WAN"] = { "type": "host", "contenu": [str(intrants["opnsense_wan_ip"])], "description": "Adresse publique de la frontiere — porte du tunnel d'administration", } regles.append({ "sens": "in", "interface": if_wan, "protocole": "udp", "source": "any", "destination": "SETOPS_FRONTIERE_WAN", "ports": [str(_vpn_admin["port"])], "chiffrement": "wireguard", "role": "acces_admin_vpn", "tenant": "", "raison": ("Poignee de main WireGuard des administrateurs du site. Le reste de " "l'administration voyage DANS le tunnel : c'est le seul port que cette " "voie demande sur l'Internet."), }) # 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, "redirections": redirections, "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//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." ) # UNE REDIRECTION ET SA REGLE VONT ENSEMBLE. Redirigee sans regle, la requete est jetee # apres traduction ; la regle sans redirection vise une adresse privee que l'Internet ne # peut pas nommer. L'une sans l'autre est une publication qui a l'air faite. wan = devis.get("if_wan") for d in devis.get("redirections") or []: if not any(r["interface"] == wan and r["sens"] == "in" and r["source"] == "any" and r["protocole"] == d["protocole"] and str(d["port_local"]) in map(str, r["ports"]) and d["cible"] in ((devis["alias"].get(r["destination"]) or {}).get("contenu") or []) for r in devis["regles"]): erreurs.append(f"Redirection {d['protocole']}/{d['port_public']} -> {d['cible']}:" f"{d['port_local']} sans regle WAN correspondante.") if not ipaddress.ip_address(d["cible"]).is_private: erreurs.append(f"Redirection vers {d['cible']} : une cible de redirection est interne.") # UNE ADRESSE, UN PROTOCOLE, UN PORT PUBLIC : UNE SEULE DESTINATION. Deux redirections # identiques cote Internet, la premiere gagnerait sans que rien ne le dise. _vues: dict[tuple, str] = {} for d in devis.get("redirections") or []: c = (d.get("destination", "wanip"), d["protocole"], str(d["port_public"])) if c in _vues: erreurs.append(f"Redirection en double sur {c[0]} {c[1]}/{c[2]} : " f"{_vues[c]} et {d['cible']}:{d['port_local']}.") _vues[c] = f"{d['cible']}:{d['port_local']}" # L'ADRESSE D'UN LOCATAIRE EST LA SIENNE : celle que le pool du site lui attribue. _pool = {str(ip): str(q) for ip, q in (intrants_frontiere().get("opnsense_ips_publiques") or {}).items()} for d in devis.get("redirections") or []: if d.get("tenant") and _pool.get(str(d.get("destination"))) != d["tenant"]: erreurs.append(f"Redirection de {d['tenant']} depuis {d.get('destination')} : " f"le pool du site ne lui attribue pas cette adresse.") # UN TUNNEL DE LOCATAIRE NE VOIT QUE SON LOCATAIRE (2026-09-17). # # Chaque ecosysteme declare qui entre chez lui ; le site porte la route. Si cette route # menait ailleurs — au site, a un voisin — le locataire aurait obtenu, en ajoutant un # pair dans SON plan, un acces chez quelqu'un d'autre. C'est la seule chose que cette # delegation ne doit jamais permettre, et c'est pourquoi elle se garde ici, avant toute # ecriture, plutot que de se relire sur le boitier. for r in devis["regles"]: src = str(r.get("source") or "") if not (src.startswith("SETOPS_ADMIN_") and src.endswith("_TUNNEL")): continue attendu = "SETOPS_TENANT_" + src[len("SETOPS_ADMIN_"):-len("_TUNNEL")] if str(r.get("destination")) != attendu: erreurs.append( f"Le tunnel {src} vise {r.get('destination')} : un tunnel de locataire ne " f"doit atteindre que {attendu}. Un ecosysteme obtiendrait un acces chez un " f"autre en ajoutant un pair dans son propre plan.") 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:]))