Alliance-Boreale/Set-OPS-Public
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
Set-OPS-Public/scripts/devis_opnsense.py
Daniel Allaire 510568e3b4 frontiere : l'interface d'une regle se derive de l'attachement (D-61) 
Le devis rangeait tout flux entrant sur le WAN, en supposant que
l'administration revenait par l'adresse publique. Vrai pour 192.168.255.0/24,
faux depuis que c'est 10.0.0.0/24 — directement attache sur `lan`. Les deux
regles SSH etaient mortes deux fois : mauvaise interface, et « Block private
networks » les aurait filtrees. Le devis conseillait meme de decocher ce filtre
sur le WAN, ce qui aurait affaibli l'interface publique pour rien.

`reseaux_locaux_frontiere()` derive de l'underlay les sous-reseaux ou la
frontiere porte une adresse, hors transit. Un alias par interface : Technolibre
a les deux cotes, Chezlepro seulement la gestion. Sans underlay, tout retombe
sur le WAN — comportement inchange.

P24 confronte desormais chaque regle d'administration a l'attachement de sa
source, et refuse l'ancien comportement.

Nouvel intrant `opnsense_if_gestion`, au catalogue du GUI.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-06 14:39:51 -04:00

663 lines
32 KiB
Python
Raw Blame History

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