Daniel Allaire
3b65384b0f
L'outil ne savait qu'AUTORISER — `"action": "pass"` etait en dur dans l'emetteur. Une regle de silence ne pouvait donc pas naitre du depot, et j'en avais pose deux a la main sur le boitier : exactement ce que ce projet refuse. CE QUI L'A MOTIVE. Le journal de la frontiere ecrivait 982 000 entrees par jour, dont 82 % un balayage Internet contre le port VNC et le reste du bavardage de decouverte du reseau local. Sa fenetre utile etait tombee a QUARANTE-QUATRE SECONDES. J'y ai cherche la trace d'un flux du site vers les hyperviseurs, je n'ai rien trouve, et j'en ai conclu a tort qu'aucune regle ne bloquait. Un journal noye ment aussi surement qu'un journal mort. Mesure apres declaration : ~20 700/jour. Une regle de silence ne change AUCUN comportement : ce qu'elle vise etait deja refuse par le defaut. Elle ne supprime qu'une trace que personne ne lira. TROIS PIECES. `cle_regle` accepte une action sans changer d'un octet la cle des regles `pass` deja posees. L'ajout naif d'un champ les aurait toutes detruites pour les recreer a l'identique, sur la frontiere, en production. Le plan l'a confirme : 7 a creer, 0 a retirer, 121 inchangees. `_corps_regle` lit l'action, la consignation et la SEQUENCE depuis le devis. La sequence est ce qui rend un `block` sur : OPNsense evalue en `quick`, donc un blocage large emis avant les `pass` fermerait courrier, web et acces distant. `devis_opnsense` lit `opnsense_silences` et en fabrique regles et alias, motif compris — une regle `block` muette sans raison ecrite est indiscernable d'un oubli. P50 garde ces deux dangers. Controles negatifs verifies : un silence en sequence 1 echoue, un silence sans motif echoue. Carte : 28 pieces d'audit (P48 l'avait vu juste). make prouver : CONFORME, 50 OK, 0 echec. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
468 lines
22 KiB
Python
468 lines
22 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 ["*"]))
|
|
# L'ACTION N'ENTRE DANS LA CLE QUE SI ELLE N'EST PAS `pass` (2026-08-27). Les regles
|
|
# de SILENCE — des `block` non consignes — ont besoin d'une identite distincte, mais
|
|
# l'ajout naif d'un champ aurait change la cle des 117 regles deja posees : le
|
|
# rapprochement les aurait toutes detruites pour les recreer a l'identique, sur la
|
|
# frontiere, en production. Une marque vide pour le cas majoritaire garde ces cles
|
|
# octet pour octet, et n'en donne une nouvelle qu'a ce qui est nouveau.
|
|
marque = "" if str(r.get("action") or "pass") == "pass" else f"{r['action']}:"
|
|
return (f"setops:{marque}{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("!")
|
|
# L'ACTION ET LA CONSIGNATION VIENNENT DU DEVIS (2026-08-27). Elles etaient en dur :
|
|
# l'outil ne savait qu'AUTORISER, et une regle qui fait taire du bruit deja refuse
|
|
# n'avait donc aucun moyen de naitre du depot. Les defauts reproduisent exactement
|
|
# l'ancien comportement — `pass`, non consigne, sequence 1 — pour que les regles
|
|
# existantes soient emises a l'identique.
|
|
#
|
|
# LA SEQUENCE EST CE QUI REND UN `block` SUR : OPNsense evalue en `quick`, donc la
|
|
# PREMIERE regle qui correspond gagne. Un blocage large pose avant les `pass` du
|
|
# devis fermerait le courrier, le web et l'acces distant. Les silences se declarent
|
|
# a 900, apres tout le reste.
|
|
corps = {
|
|
"enabled": "1",
|
|
"sequence": str(r.get("sequence") or "1"),
|
|
"action": str(r.get("action") or "pass"),
|
|
"log": "1" if r.get("journaliser") else "0",
|
|
"interface": r["interface"],
|
|
"direction": "in", # le devis raisonne en ARRIVEE sur l'interface (D-61)
|
|
"ipprotocol": "inet",
|
|
# `any` reste EN MINUSCULES : OPNsense attend ses protocoles en majuscules mais
|
|
# nomme « tout protocole » `any`. Un `ANY` majuscule est rejete a la validation.
|
|
"protocol": ("any" if r["protocole"] == "any"
|
|
else "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())
|