#!/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.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() def __call__(self, chemin: str, corps: dict | None = None) -> dict: data = json.dumps(corps).encode() if corps is not None else None h = {"Authorization": self.auth} if data: h["Content-Type"] = "application/json" 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 _corps_regle(r: dict, k: str) -> dict: 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": r["destination"], "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()} survivants |= {r["destination"] for r in devis["regles"]} return { "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, 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 n in sorted(p["alias_retirer"]): print(f" - alias ORPHELIN {n}") print(f"\n a creer : {len(p['alias_creer']) + len(p['regles_creer'])}" f" | a retirer : {len(p['regles_retirer']) + len(p['alias_retirer'])}" f" | inchange : {len(p['regles_garder'])}") return any(p[c] for c in ("alias_creer", "alias_majer", "alias_retirer", "regles_creer", "regles_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}") # 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 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 print(" alias :", api("/api/firewall/alias/reconfigure/", {}).get("status", "?")) print(" regles :", str(api("/api/firewall/filter/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"))) sortie = subprocess.run([sys.executable, str(RACINE / "scripts" / "devis_opnsense.py"), "--json"], cwd=RACINE, capture_output=True, text=True, env={**os.environ, "SETOPS_INSTANCE": str(base)}) 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())