Set-OPS-Public/scripts/appliquer_opnsense.py
Daniel Allaire 4462c13b3d P03 : la preuve comparait chaque instance a l'inventaire d'UNE SEULE
Trouve en validant une mise a jour du CHANGELOG. Deux invocations de la meme preuve,
deux verdicts : `make prouver` -> NON CONFORME (« lab : 17 hotes avec ecart »),
`python3 scripts/prouver.py` -> CONFORME 37/37. Le lab n'avait aucun ecart.

DEUX VARIABLES DESIGNENT LA CIBLE, ET LA SECONDE GAGNE. Le Makefile exporte
SETOPS_INVENTAIRE (ligne 13), derive de l'instance ACTIVE ; instancier.py:68 lui fait
FORCER la cible par-dessus SETOPS_INSTANCE. P03 (prouver.py:505) ne redirigeait que
SETOPS_INSTANCE : elle generait le plan de CHAQUE instance federee et le comparait a
l'inventaire applique de la SEULE instance active.

LE ROUGE N'ETAIT PAS LE PROBLEME, LE VERT L'ETAIT. Sous `make`, l'inventaire applique
de lab et de Technolibre n'etait JAMAIS lu — l'angle meme pour lequel P03 a ete ecrite
le 2026-08-12 (un tenant qu'on ne regarde pas imposant ses vieilles adresses au pare-feu
partage). La preuve etait aveugle a son propre cas, par l'invocation documentee. Les
rapports du 13 et du 14 sortent de cette invocation-la. Signature visible sans lire le
code : les hosts.genere.yml de lab et de Technolibre ne bougeaient pas.

CORRECTIF, cinq sites : env.pop("SETOPS_INVENTAIRE") partout ou l'on redirige
SETOPS_INSTANCE — P03 et P15, plus les trois applicateurs (opnsense, proxmox_fw, sdn) qui
pointent vers l'HEBERGEUR. Ces trois sont sans effet tant qu'hebergeur et tenant actif
coincident, c'est-a-dire jusqu'au second site. Le geste existait deja (modeles.py:96).

GARDE, pour que la classe cesse d'etre silencieuse : inventory_rules.inventaire_force()
REFUSE une cible hors de l'instance visee, en nommant les deux valeurs. Eprouvee dans les
deux sens (contradiction -> code 1 ; cible legitime dans l'instance -> passe). Branchee
sur les quatre resolutions de _inventaire (instancier, serveurs, applications,
config_proxmox). Le GUI garde la sienne : il ne redirige jamais SETOPS_INSTANCE pour un
fils et resout par symlink a chaque requete.

make prouver : 37 OK, 0 echec, 0 saute — et les hosts.genere.yml des TROIS instances
portent l'horodatage du passage. make test 0.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-18 14:35:01 -04:00

415 lines
19 KiB
Python

#!/usr/bin/env python3
"""Reconcilie la frontiere OPNsense avec le devis (`devis_opnsense.py`).
DEUX SENS, et c'est tout l'objet de ce script :
- ce que le devis demande et qui manque -> cree
- ce que le devis ne demande plus -> RETIRE
Sans le second sens, un devis qui change laisse derriere lui des regles mortes :
elles n'ouvrent rien, mais elles decrivent une politique qui n'est plus la notre.
Une bordure dont la lecture ment est pire qu'une bordure vide.
PERIMETRE STRICT : seuls les objets marques `setops:` (regles) ou prefixes
`SETOPS_` (alias) sont touches. Tout ce qu'un humain a pose a la main dans
l'interface reste intact — le script ne connait meme pas son existence.
NON DESTRUCTIF PAR DEFAUT : sans `CONFIRMER=true`, il n'ecrit RIEN et se contente
d'afficher le plan. C'est la regle 4 du depot.
Usage :
python3 scripts/appliquer_opnsense.py # plan seul, aucune ecriture
CONFIRMER=true python3 scripts/appliquer_opnsense.py # applique le plan
"""
from __future__ import annotations
import base64
import json
import os
import ssl
import subprocess
import sys
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
import yaml
RACINE = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(RACINE / "scripts"))
import devis_opnsense as devis_mod # noqa: E402
def _voute(base: Path) -> dict:
"""Secrets d'API de la frontiere, lus par ansible-vault. Rien n'est journalise."""
for nom in ("principal", "production", "lab"):
p = base / "inventories" / nom / "group_vars" / "all" / "vault.yml"
if p.is_file():
r = subprocess.run(["ansible-vault", "view", str(p)],
capture_output=True, text=True)
if r.returncode != 0:
raise SystemExit(
"Voute illisible : renseigner ANSIBLE_VAULT_PASSWORD_FILE.\n"
+ r.stderr.strip()[:300])
return yaml.safe_load(r.stdout) or {}
raise SystemExit(f"Aucune voute trouvee sous {base}/inventories/*/group_vars/all/.")
class Frontiere:
"""Le minimum d'API pour reconcilier — et rien de plus."""
def __init__(self, url: str, cle: str, secret: str, verifier: bool):
self.base = url.rstrip("/")
self.auth = "Basic " + base64.b64encode(f"{cle}:{secret}".encode()).decode()
self.ctx = None if verifier else ssl._create_unverified_context()
@staticmethod
def _formulaire(corps: dict) -> bytes:
"""Encode `{racine: {champ: valeur}}` en `racine[champ]=valeur`.
POURQUOI PAS DU JSON (mesure du 2026-08-13, sur un OPNsense 24.7 neuf).
Le controleur d'OPNsense lit ses champs avec `hasPost(<racine>)`. Quand le
corps arrive en `application/json` et que le boitier ne le decompose pas en
variables de POST, ce test est FAUX : le controleur renvoie `{"result":
"failed"}` NU — sans la moindre validation, donc sans dire ce qui manque.
Toute ecriture echouait ainsi : alias, regles, NAT, routes. Le meme corps en
formulaire encode passe du premier coup.
C'est la forme que poste l'interface web elle-meme : aucune version
d'OPNsense ne peut la refuser, alors que le JSON depend du boitier.
"""
champs: dict[str, str] = {}
for racine, valeurs in corps.items():
if isinstance(valeurs, dict):
for k, v in valeurs.items():
champs[f"{racine}[{k}]"] = "" if v is None else str(v)
else:
champs[racine] = "" if valeurs is None else str(valeurs)
return urllib.parse.urlencode(champs).encode()
def __call__(self, chemin: str, corps: dict | None = None) -> dict:
data = self._formulaire(corps) if corps is not None else None
h = {"Authorization": self.auth}
if data is not None:
h["Content-Type"] = "application/x-www-form-urlencoded"
req = urllib.request.Request(self.base + chemin, data=data, headers=h,
method="POST" if data is not None else "GET")
try:
with urllib.request.urlopen(req, context=self.ctx, timeout=45) as rep:
return json.loads(rep.read())
except urllib.error.HTTPError as e:
return {"_erreur": f"{e.code} {e.read().decode()[:200]}"}
except OSError as e:
return {"_erreur": str(e)[:200]}
def cle_regle(r: dict) -> str:
"""Identite d'une regle, portee par sa DESCRIPTION sur le boitier.
C'est ce qui rend la reconciliation possible sans tenir un etat local : le
boitier porte lui-meme de quoi se comparer au devis. Tout changement de source,
de port ou d'interface produit une cle differente — donc une regle a creer et
une regle a retirer, ce qui est exactement la verite.
"""
ports = ",".join(map(str, r["ports"] or ["*"]))
return (f"setops:{r['tenant']}:{r['interface']}:{r['sens']}:{r['protocole']}:"
f"{r['source']}->{r['destination']}:{ports}")
def cle_nat(n: dict) -> str:
"""Identite d'une regle de NAT sortant, meme principe que `cle_regle`."""
return f"setopsnat:{n['tenant']}:{n['interface']}:{n['source']}->{n['destination']}:{n['cible']}"
def cle_route(r: dict) -> str:
"""Identite d'une route statique, meme principe que `cle_regle`.
Les routes ont longtemps ete posees A LA MAIN a cote de ce script. Elles
fonctionnaient, mais rien ne les reconciliait : leur disparition n'aurait ete vue
par personne, et un devis qui change les laissait derriere lui. C'est exactement le
defaut que ce script existe pour empecher — il ne pouvait pas s'appliquer a
lui-meme tant qu'il ignorait les routes.
"""
return f"setopsroute:{r['tenant']}:{r['reseau']}->{r['prochain_saut']}"
def _corps_route(r: dict, k: str, passerelle: str) -> dict:
return {
"network": r["reseau"],
"gateway": passerelle,
"disabled": "0",
"descr": k,
}
def _passerelle_pour(api: Frontiere, saut: str) -> str | None:
"""Nom de la passerelle OPNsense portant cette adresse.
Une route se declare par NOM de passerelle, le devis raisonne en ADRESSE de
prochain saut. On resout ici plutot que d'exiger un intrant de plus : le nom est
deja sur le boitier, et le demander deux fois serait une occasion de divergence.
"""
for x in (api("/api/routes/gateway/status").get("items") or []):
if str(x.get("address") or "").strip() == str(saut).strip():
return str(x.get("name"))
return None
def _corps_nat(n: dict, k: str) -> dict:
return {
"enabled": "1",
"sequence": "100",
"interface": n["interface"],
"ipprotocol": "inet",
"protocol": "any",
"source_net": n["source"],
"destination_net": n["destination"],
"target": n["cible"],
"description": f"{k} — sortie tenant",
}
def _corps_regle(r: dict, k: str) -> dict:
# Une destination prefixee de `!` est une destination NIEE (« tout sauf »). Le devis
# porte le `!` pour qu'il compte dans l'identite de la regle ; c'est ici qu'il devient
# le champ `destination_not` d'OPNsense.
dst = str(r["destination"])
nie = dst.startswith("!")
corps = {
"enabled": "1",
"sequence": "1",
"action": "pass",
"interface": r["interface"],
"direction": "in", # le devis raisonne en ARRIVEE sur l'interface (D-61)
"ipprotocol": "inet",
"protocol": "ICMP" if r["protocole"] == "icmp" else r["protocole"].upper(),
"source_net": r["source"],
"destination_net": dst[1:] if nie else dst,
"destination_not": "1" if nie else "0",
"description": f"{k} — {r['role']}",
}
if r["protocole"] in ("tcp", "udp") and r["ports"]:
corps["destination_port"] = ",".join(map(str, r["ports"]))
return corps
def _contenu(x) -> set[str]:
if isinstance(x, (list, tuple)):
return {str(v).strip() for v in x if str(v).strip()}
return {v.strip() for v in str(x or "").replace(",", "\n").split("\n") if v.strip()}
def plan(api: Frontiere, devis: dict) -> dict:
"""Ce qu'il faudrait faire pour que le boitier dise ce que le devis dit."""
voulues = {cle_regle(r): r for r in devis["regles"]}
posees = {}
for x in (api("/api/firewall/filter/search_rule/",
{"current": 1, "rowCount": 1000}).get("rows") or []):
d = str(x.get("description") or "")
if d.startswith("setops:"):
posees[d.split(" — ")[0]] = x
al_voulus = devis["alias"]
al_poses = {x["name"]: x for x in (api("/api/firewall/alias/searchItem/",
{"rowCount": 1000}).get("rows") or [])
if str(x.get("name", "")).startswith("SETOPS_")}
# Un alias encore reference par une regle qui SURVIT ne doit pas partir : la
# suppression echouerait, et le boitier resterait a moitie reconcilie.
survivants = {r["source"] for k, r in voulues.items()}
# `lstrip('!')` : une destination niee reference le MEME alias. L'oublier ferait
# passer `SETOPS_INTERNES` pour un orphelin, et le script tenterait de supprimer
# l'alias que ses propres regles utilisent.
survivants |= {str(r["destination"]).lstrip("!") for r in devis["regles"]}
survivants |= {n["source"] for n in devis.get("nat") or []} # references par le NAT
nat_voulus = {cle_nat(n): n for n in devis.get("nat") or []}
nat_poses = {}
for x in (api("/api/firewall/source_nat/search_rule/",
{"current": 1, "rowCount": 1000}).get("rows") or []):
d = str(x.get("description") or "")
if d.startswith("setopsnat:"):
nat_poses[d.split(" — ")[0]] = x
rt_voulues = {cle_route(r): r for r in devis.get("routes") or []}
rt_posees = {}
for x in (api("/api/routes/routes/searchroute/",
{"current": 1, "rowCount": 1000}).get("rows") or []):
d = str(x.get("descr") or "")
if d.startswith("setopsroute:"):
rt_posees[d] = x
return {
"routes_creer": {k: r for k, r in rt_voulues.items() if k not in rt_posees},
"routes_garder": {k for k in rt_voulues if k in rt_posees},
"routes_retirer": {k: x for k, x in rt_posees.items() if k not in rt_voulues},
"nat_creer": {k: n for k, n in nat_voulus.items() if k not in nat_poses},
"nat_garder": {k for k in nat_voulus if k in nat_poses},
"nat_retirer": {k: x for k, x in nat_poses.items() if k not in nat_voulus},
"alias_creer": {n: a for n, a in al_voulus.items() if n not in al_poses},
"alias_majer": {n: a for n, a in al_voulus.items()
if n in al_poses
and _contenu(a["contenu"]) != _contenu(al_poses[n].get("content"))},
"alias_retirer": {n: x for n, x in al_poses.items()
if n not in al_voulus and n not in survivants},
"regles_creer": {k: r for k, r in voulues.items() if k not in posees},
"regles_garder": {k for k in voulues if k in posees},
"regles_retirer": {k: x for k, x in posees.items() if k not in voulues},
}
def afficher(p: dict) -> bool:
"""Rend le plan lisible. Retourne True s'il y a quelque chose a faire."""
for nom, contenu in (("alias a creer", p["alias_creer"]),
("alias a mettre a jour", p["alias_majer"])):
for n, a in sorted(contenu.items()):
print(f" + {nom:<22} {n:<32} {', '.join(a['contenu'])[:60]}")
for k, r in sorted(p["regles_creer"].items()):
print(f" + regle {r['interface']:<5} {r['protocole']:<4} "
f"{str(r['ports'] or ''):<9} {r['source'][:30]:<30} -> {r['destination'][:24]}")
for k, n in sorted(p["nat_creer"].items()):
print(f" + NAT sortant {n['interface']:<5} {n['source'][:30]:<30} "
f"-> {n['cible']}")
for k, r in sorted(p["routes_creer"].items()):
print(f" + route {r['reseau']:<18} -> {r['prochain_saut']:<14} "
f"{r['tenant']}")
for k, x in sorted(p["regles_retirer"].items()):
print(f" - regle PERIMEE {str(x.get('interface')):<5} "
f"{str(x.get('protocol')):<4} {str(x.get('destination_port') or ''):<9} "
f"{str(x.get('source_net'))[:30]:<30} -> {str(x.get('destination_net'))[:24]}")
for k, x in sorted(p["nat_retirer"].items()):
print(f" - NAT PERIME {str(x.get('interface')):<5} "
f"{str(x.get('source_net'))[:30]:<30} -> {str(x.get('target'))}")
for k, x in sorted(p["routes_retirer"].items()):
print(f" - route PERIMEE {str(x.get('network')):<18} "
f"-> {str(x.get('gateway'))}")
for n in sorted(p["alias_retirer"]):
print(f" - alias ORPHELIN {n}")
creer = len(p["alias_creer"]) + len(p["regles_creer"]) + len(p["nat_creer"]) \
+ len(p["routes_creer"])
retirer = len(p["regles_retirer"]) + len(p["alias_retirer"]) + len(p["nat_retirer"]) \
+ len(p["routes_retirer"])
print(f"\n a creer : {creer} | a retirer : {retirer}"
f" | inchange : {len(p['regles_garder']) + len(p['nat_garder'])}"
f" + {len(p['routes_garder'])} routes")
return any(p[c] for c in ("alias_creer", "alias_majer", "alias_retirer",
"regles_creer", "regles_retirer",
"nat_creer", "nat_retirer",
"routes_creer", "routes_retirer"))
def appliquer(api: Frontiere, p: dict) -> int:
"""Ordre impose par les dependances : on n'enleve jamais un objet encore utilise."""
echecs = 0
def _fait(rep, quoi):
nonlocal echecs
if rep.get("result") in ("saved", "deleted") or rep.get("status") == "ok":
return True
echecs += 1
print(f" ! ECHEC {quoi} : {json.dumps(rep)[:160]}")
return False
# 1. Les alias d'abord : une regle qui reference un alias absent est refusee.
for n, a in sorted(p["alias_creer"].items()):
_fait(api("/api/firewall/alias/addItem/",
{"alias": {"name": n, "type": a["type"], "enabled": "1",
"content": "\n".join(a["contenu"]),
"description": a.get("description", "")[:255]}}), f"alias {n}")
for n, a in sorted(p["alias_majer"].items()):
uuid = p["_uuid_alias"][n]
_fait(api(f"/api/firewall/alias/setItem/{uuid}",
{"alias": {"name": n, "type": a["type"], "enabled": "1",
"content": "\n".join(a["contenu"]),
"description": a.get("description", "")[:255]}}), f"alias {n}")
# 2. Creer avant de retirer : a aucun instant la politique n'est plus permissive
# qu'avant, et si le retrait echoue on reste en surcouverture, jamais en trou.
for k, r in sorted(p["regles_creer"].items()):
_fait(api("/api/firewall/filter/add_rule/", {"rule": _corps_regle(r, k)}), f"regle {k}")
for k, n in sorted(p["nat_creer"].items()):
_fait(api("/api/firewall/source_nat/add_rule/", {"rule": _corps_nat(n, k)}), f"nat {k}")
# 2bis. Routes, meme ordre et pour une raison plus forte encore : une route
# manquante coupe la flotte, une route en trop ne fait qu'acheminer vers un
# VRF qui la jettera. En cas de doute, on reste large.
for k, r in sorted(p["routes_creer"].items()):
gw = _passerelle_pour(api, r["prochain_saut"])
if gw is None:
echecs += 1
print(f" ! ECHEC route {k} : aucune passerelle ne porte {r['prochain_saut']}")
continue
_fait(api("/api/routes/routes/addroute/", {"route": _corps_route(r, k, gw)}),
f"route {k}")
for k, x in sorted(p["routes_retirer"].items()):
_fait(api(f"/api/routes/routes/delroute/{x['uuid']}", {}), f"retrait route {k}")
# 3. Retrait des perimees, puis des alias devenus orphelins.
for k, x in sorted(p["regles_retirer"].items()):
_fait(api(f"/api/firewall/filter/del_rule/{x['uuid']}", {}), f"retrait {k}")
for k, x in sorted(p["nat_retirer"].items()):
_fait(api(f"/api/firewall/source_nat/del_rule/{x['uuid']}", {}), f"retrait nat {k}")
for n, x in sorted(p["alias_retirer"].items()):
_fait(api(f"/api/firewall/alias/delItem/{x['uuid']}", {}), f"retrait alias {n}")
if echecs:
print(f"\n {echecs} echec(s) — RIEN N'EST APPLIQUE, la config reste en attente.")
return 1
if p["routes_creer"] or p["routes_retirer"]:
print(" routes :", api("/api/routes/routes/reconfigure/", {}).get("status", "?"))
print(" alias :", api("/api/firewall/alias/reconfigure/", {}).get("status", "?"))
print(" regles :", str(api("/api/firewall/filter/apply/", {}).get("status", "?")).strip())
print(" nat :", str(api("/api/firewall/source_nat/apply/", {}).get("status", "?")).strip())
return 0
def main() -> int:
base = devis_mod.depot_hebergeur()
if base is None:
raise SystemExit("Aucun underlay ne designe d'hebergeur : pas de frontiere a piloter.")
intr = devis_mod.intrants_frontiere()
url = str(intr.get("opnsense_api_url") or "").strip()
if not url:
raise SystemExit("Intrant `opnsense_api_url` absent : rien a joindre.")
v = _voute(base)
api = Frontiere(url, v["vault_opnsense_api_key"], v["vault_opnsense_api_secret"],
bool(intr.get("opnsense_api_verifier_certs")))
# SETOPS_INSTANCE designe ici l'HEBERGEUR, pas le tenant actif. SETOPS_INVENTAIRE,
# exporte par le Makefile, gagnerait sur lui (instancier.py) : le devis d'un site
# lirait l'inventaire d'un tenant d'ailleurs. Sans effet tant qu'hebergeur et tenant
# actif coincident — c'est-a-dire jusqu'au second site (mesure du 2026-08-18).
env_heb = {**os.environ, "SETOPS_INSTANCE": str(base)}
env_heb.pop("SETOPS_INVENTAIRE", None)
sortie = subprocess.run([sys.executable, str(RACINE / "scripts" / "devis_opnsense.py"),
"--json"], cwd=RACINE, capture_output=True, text=True,
env=env_heb)
if sortie.returncode != 0:
raise SystemExit("Le devis ne se genere pas :\n" + sortie.stderr.strip()[:400])
devis = json.loads(sortie.stdout)
ok, erreurs = devis_mod.verifier(devis)
if not ok:
print("Le devis ne passe pas sa propre garde — rien ne sera pousse :")
for e in erreurs:
print(" -", e)
return 2
print(f"Frontiere {url} — {len(devis['regles'])} regles au devis\n")
p = plan(api, devis)
p["_uuid_alias"] = {x["name"]: x["uuid"] for x in
(api("/api/firewall/alias/searchItem/", {"rowCount": 1000}).get("rows") or [])}
if not afficher(p):
print("\n La frontiere dit deja ce que le devis dit. Rien a faire.")
return 0
if os.environ.get("CONFIRMER") != "true":
print("\n PLAN SEUL — aucune ecriture. Rejouer avec CONFIRMER=true pour appliquer.")
return 0
print()
return appliquer(api, p)
if __name__ == "__main__":
sys.exit(main())