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 53f6af3f1a frontière : les règles couvrent tous les tenants fédérés 
Le devis était multi-tenant pour ses routes, mono-tenant pour ses règles :
il routait 10.21.0.0/16 et 10.27.0.0/16 mais ne filtrait que l'instance
active. Technolibre aurait été routé jusqu'à la bordure puis bloqué dans les
deux sens, SSH d'administration compris, sans qu'une ligne dise pourquoi.

La résolution est paramétrée par tenant : `inventaire_de()` lit le hosts.yml
de chaque instance fédérée, `cibles_par_role()` prend l'inventaire en
argument, les alias d'hôtes sont préfixés. 11 règles par tenant, 22 au total.

Cloisonnement : la première version faisait de SETOPS_ADMIN l'union des
réseaux d'administration — le plan de gestion d'un tenant serait entré chez
le voisin, la bordure rouvrant ce que les ACL de switch ferment. Corrigé
avant livraison : un alias par tenant, n'ouvrant que son propre supernet.
L'union reste pour les routes de retour et P24 : router n'est pas autoriser.

Deux omissions annoncées : tenant sans inventaire (aucune règle), tenant
sans `nftables_admin_ssh` (règle SSH omise plutôt qu'ouverte à `any`, ce qui
exposerait le SSH à Internet). Cas dégradé exercé.

Preuves : 24 OK, 0 échec.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-02 14:25:20 -04:00

434 lines
19 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 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,
)
from devis_reseau import decouvrir, 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>"
IF_TRANSIT = "<IF-TRANSIT>"
IF_WAN = "<IF-WAN>"
def intrants_frontiere() -> dict:
"""Intrants NON sensibles de la frontiere (group_vars/opnsense.yml). {} si absent."""
for nom in ("principal", "production", "lab"):
p = _inventaire().parent.parent / 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.
"""
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 inventaire_de(nom_instance: str) -> Path | None:
"""hosts.yml d'une instance FEDEREE, active ou non. None si elle n'en a pas.
La frontiere est un equipement PARTAGE : elle route vers tous les tenants, elle
doit donc aussi porter leurs regles. Se limiter a l'instance active laisserait un
tenant route mais muet — chemin present, politique absente.
"""
base = RACINE.parent / nom_instance / "inventories"
for env in ("principal", "production", "lab"):
p = base / env / "hosts.yml"
if p.is_file():
return p
return None
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 admin_de(nom_instance: str) -> list[str]:
"""Reseaux d'administration declares par une instance (intrant nftables_admin_ssh)."""
inv = inventaire_de(nom_instance)
if not inv:
return []
dossier = inv.parent / "group_vars" / "all"
for fichier in sorted(dossier.glob("*.yml")):
if "vault" in fichier.name:
continue
data = yaml.safe_load(fichier.read_text(encoding="utf-8")) or {}
if isinstance(data, dict) and data.get("nftables_admin_ssh"):
src = data["nftables_admin_ssh"]
return [str(s) for s in src] if isinstance(src, list) else [str(src)]
return []
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)
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})
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",
"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 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(),
"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)),
"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,
"reciproque_switch": [
f"ip route {cidr} -> {saut} (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 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"# 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 += [
"",
"## 2. Routes statiques (vers les supernets tenants)",
]
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."]
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"])
out.append(
f"pass {rg['sens']:3} 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:]))
Reference in a new issue View git blame Copy permalink