Set-OPS-Public/scripts/appliquer_opnsense.py

#!/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 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 _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:
    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"]}
    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

    return {
        "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, 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 n in sorted(p["alias_retirer"]):
        print(f"  - alias ORPHELIN        {n}")
    creer = len(p["alias_creer"]) + len(p["regles_creer"]) + len(p["nat_creer"])
    retirer = len(p["regles_retirer"]) + len(p["alias_retirer"]) + len(p["nat_retirer"])
    print(f"\n  a creer : {creer}   |  a retirer : {retirer}"
          f"   |  inchange : {len(p['regles_garder']) + len(p['nat_garder'])}")
    return any(p[c] for c in ("alias_creer", "alias_majer", "alias_retirer",
                              "regles_creer", "regles_retirer",
                              "nat_creer", "nat_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}")

    # 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
    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")))

    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())