Daniel Allaire
4af1fcd0e2
Some checks are pending
verifier / verifier (push) Waiting to run
L'exploitant lance `make frontiere-appliquer` dans son terminal : « a creer 0, inchange
53 + 15 routes — la frontiere dit deja ce que le devis dit ». Elle etait DEJA conforme.
J'avais annonce quelques heures plus tot « 89 a creer, 0 inchange, donc les regles
heritees sont invisibles a l'API ». FAUX. L'intrant `opnsense_api_url` pointait sur
10.0.0.1, l'adresse d'avant la migration du boitier : chaque lecture rendait
{"_erreur": ...}, le plan lisait `.get("rows")`, n'y trouvait rien, et concluait au vide.
Avec CONFIRMER, on aurait pousse une politique entiere EN DOUBLE.
TROISIEME FOIS EN UNE SOIREE, meme confusion — « pas de reponse » pris pour « rien » :
devis_placement iterait un dict d'erreur comme une liste
devis_underlay declarait morts les reseaux qu'il ne joignait pas
appliquer_opnsense lisait un boitier injoignable comme un boitier vide
Toutes les lectures de la frontiere passent desormais par une garde qui refuse en nommant
l'hote, la cause et le piege evite. Eprouvee contre l'ancienne adresse : elle refuse.
CE QUE CA DIT DE LA METHODE : ce n'est ni le harnais ni moi qui avons trouve, c'est
l'exploitant, en lancant la commande la ou il VOIT la sortie. Mes commandes s'executent
dans ma session ; il n'en voit rien. Une commande lente ressemble a un blocage, et un
blocage a une commande lente — j'ai conclu deux fois a tort avant qu'il ne regarde.
make verifier 41/41 ; le plan reel reste « rien a faire ».
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
446 lines
21 KiB
Python
446 lines
21 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
|
|
from proxmox_api import voute as voute_proxmox # noqa: E402
|
|
|
|
|
|
def _voute(base: Path) -> dict:
|
|
"""Secrets d'API de la frontiere — SOURCE UNIQUE, partagee avec l'acces au cluster.
|
|
|
|
La frontiere appartient au MONDE PHYSIQUE : sa cle d'API vit dans la voute de
|
|
l'underlay, chez l'hebergeur, pas chez un locataire (2026-08-22, cf.
|
|
docs/frontiere-physique-virtuel.md). `proxmox_api.voute` porte deja cette regle et son
|
|
repli sur la voute du tenant pendant la migration : la reecrire ici aurait fait une
|
|
dixieme copie le jour meme ou l'on refermait les neuf autres.
|
|
"""
|
|
return voute_proxmox(base)
|
|
|
|
|
|
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 (_lire(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 _lire(api: Frontiere, chemin: str, corps: dict | None = None) -> dict:
|
|
"""Une lecture du boitier — ou un REFUS. Jamais un silence pris pour du vide.
|
|
|
|
POURQUOI (mesure du 2026-08-22, trouvee par l'exploitant). L'intrant `opnsense_api_url`
|
|
pointait encore sur `10.0.0.1`, l'adresse d'avant la migration du boitier. Chaque
|
|
lecture echouait donc et rendait `{"_erreur": ...}` ; le plan lisait `.get("rows")`,
|
|
n'y trouvait rien, et concluait que la frontiere etait VIDE : « 89 objets a creer ».
|
|
|
|
J'en avais tire une conclusion fausse — « les regles heritees sont invisibles a
|
|
l'API » — et, avec CONFIRMER, on aurait pousse une politique entiere en double sur un
|
|
boitier qui la portait deja. La verite, une fois la bonne adresse posee : 53 objets et
|
|
15 routes, tous conformes, rien a faire.
|
|
|
|
UN BOITIER INJOIGNABLE N'EST PAS UN BOITIER VIDE. C'est la troisieme fois en une
|
|
soiree que cette confusion se paie — apres le devis de placement qui iterait une
|
|
erreur comme une liste, et la reconnaissance de l'underlay qui declarait morts les
|
|
reseaux qu'elle ne joignait pas. On refuse donc, en nommant l'hote et la cause.
|
|
"""
|
|
rep = api(chemin, corps)
|
|
if isinstance(rep, dict) and rep.get("_erreur"):
|
|
raise SystemExit(
|
|
f"La frontiere ne repond pas sur `{chemin}` : {rep['_erreur']}\n"
|
|
f" boitier interroge : {api.base if hasattr(api, 'base') else '?'}\n"
|
|
f" a verifier : `opnsense_api_url` designe-t-il l'adresse ACTUELLE du "
|
|
f"boitier, le VPN est-il monte, la cle d'API est-elle encore valide ?\n"
|
|
f" (un boitier injoignable serait lu comme un boitier VIDE, et le plan "
|
|
f"proposerait de tout recreer)")
|
|
return rep if isinstance(rep, dict) else {}
|
|
|
|
|
|
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 (_lire(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 (_lire(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 (_lire(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 (_lire(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")))
|
|
|
|
# NE PAS FORCER SETOPS_INSTANCE VERS L'HEBERGEUR (2026-08-22). Le devis a besoin d'un
|
|
# TENANT — il lit son inventaire pour resoudre les flux — et l'hebergeur n'en est pas
|
|
# un : depuis que le monde physique a son propre depot (SITE-<nom>), il ne porte ni
|
|
# plan ni inventaire. Ce qui vient de l'hebergeur (underlay, proxmox-hebergeur, voute)
|
|
# est DERIVE du symlink `underlay.yml`, jamais de cette variable.
|
|
#
|
|
# L'hypothese « le depot de l'hebergeur est aussi une instance » n'a tenu que tant que
|
|
# les deux etaient confondus. La separation l'a revelee, ce qui est son interet.
|
|
env_devis = {k: v for k, v in os.environ.items() if k != "SETOPS_INVENTAIRE"}
|
|
sortie = subprocess.run([sys.executable, str(RACINE / "scripts" / "devis_opnsense.py"),
|
|
"--json"], cwd=RACINE, capture_output=True, text=True,
|
|
env=env_devis)
|
|
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
|
|
(_lire(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())
|