Daniel Allaire
58fdaa5540
Le prochain saut des routes tenants pointait le SVI du commutateur. En EVPN il ne route plus les tenants : la route arriverait sur un équipement sans chemin vers le tenant — configuration qui s'applique sans erreur et ne fonctionne pas. Le devis émet `<NOEUD-DE-SORTIE-EVPN>` et dit pourquoi. `underlay.passerelle_sortie` garde son sens : adresse du pare-feu sur le lien de transit, donc sortie de l'UNDERLAY. Deux choses distinctes. Corrigé aussi une contradiction antérieure au SDN : la section 0 demandait de router l'administration vers 10.0.4.6, le SVI du commutateur lui-même, alors que `devis-reseau` émet 10.0.4.1, l'adresse du pare-feu — tout en affirmant que l'autre devis « émet déjà ces routes ». Elles coïncident maintenant, vérifié ligne à ligne. Et deux commentaires qui affirmaient l'inverse de la décision se dérivent du mode de routage. Preuves : 24 OK, 0 échec. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
492 lines
23 KiB
Python
492 lines
23 KiB
Python
#!/usr/bin/env python3
|
|
"""Devis de configuration OPNsense — la frontiere nord/sud du reseau converge.
|
|
|
|
DERIVE de deux sources deja tenues a jour, rien n'est saisi a la main :
|
|
- le REGISTRE DES FLUX (`roles/*/meta/flux.yml`) : les flux dont le pair est
|
|
`externe` sont, par construction, ceux que le pare-feu d'hote NE traite PAS
|
|
(scripts/resoudre_flux.py les saute) parce qu'ils relevent de la frontiere.
|
|
Plusieurs `raison` le disent explicitement (« filtre a l'OPNsense »).
|
|
- les NOMENCLATURES des instances federees (`../*/plan/nomenclature.yml`) :
|
|
les supernets tenants a router, decoulant du seul seed `index`.
|
|
|
|
ROLE ASSUME : frontiere NORD/SUD uniquement. Les passerelles de zone (SVI)
|
|
restent sur les switches L3 ; OPNsense ne porte AUCUN SVI de tenant et ne voit
|
|
pas le trafic est-ouest, qui reste filtre par les ACL de `make devis-reseau`.
|
|
|
|
NON destructif : ce script n'ecrit RIEN sur le pare-feu. Il produit un devis a
|
|
relire, puis (a venir) un JSON consommable par l'API OPNsense.
|
|
|
|
Usage :
|
|
python3 scripts/devis_opnsense.py # devis lisible
|
|
python3 scripts/devis_opnsense.py --json # meme contenu, pour l'API
|
|
python3 scripts/devis_opnsense.py --verifier # garde anti-lockout (P24)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import 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>"
|
|
|
|
|
|
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.
|
|
"""
|
|
if underlay_mod.routage_tenants(underlay_mod.charger()) == "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.
|
|
return NOEUD_SORTIE
|
|
return str((transit or {}).get("passerelle") or "").strip() or PROCHAIN_SAUT
|
|
|
|
# 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)
|
|
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.
|
|
admin_par_tenant: dict[str, list[str]] = {}
|
|
for nom, pfx, n in tenants:
|
|
etiquette = f"{pfx}{n['index']}"
|
|
cidrs = admin_de(nom)
|
|
admin_par_tenant[etiquette] = cidrs
|
|
if cidrs:
|
|
alias[f"SETOPS_ADMIN_{etiquette}"] = {
|
|
"type": "network",
|
|
"contenu": list(cidrs),
|
|
"description": f"Reseaux d'administration de {nom} (intrant "
|
|
f"nftables_admin_ssh) — 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.
|
|
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 cidrs in admin_par_tenant.values() for c in cidrs
|
|
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
|
|
source = f"SETOPS_ADMIN_{etiquette}" if ssh_gestion else "any"
|
|
regles.append({
|
|
"sens": "in" if entrant else "out",
|
|
# Interface d'ARRIVEE du paquet, celle a laquelle OPNsense attache la
|
|
# regle. Entrant (depuis Internet ou le VPN d'admin, qui revient par
|
|
# l'adresse publique) : le WAN. Sortant (la flotte vers Internet) : le
|
|
# lien de transit, par ou le paquet penetre le pare-feu.
|
|
"interface": if_wan if entrant else if_transit,
|
|
"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
|
|
]
|
|
|
|
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),
|
|
"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,
|
|
"transit": {
|
|
"nom": (transit or {}).get("nom"),
|
|
"vlan": (transit or {}).get("vlan"),
|
|
"sous_reseau": (transit or {}).get("sous_reseau"),
|
|
"svi_switch": saut,
|
|
"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."),
|
|
"# ============================================================",
|
|
"",
|
|
"## 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)"]
|
|
tr = devis.get("transit")
|
|
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)",
|
|
]
|
|
if tr:
|
|
out += [
|
|
f"# lien '{tr['nom']}' {tr['sous_reseau']} — VLAN {tr['vlan']} a etiqueter",
|
|
f"# switch {tr['svi_switch']} <-> frontiere"
|
|
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 figer apres le spike EVPN.",
|
|
]
|
|
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.")
|
|
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:]))
|