Set-OPS-Public/scripts/devis_opnsense.py

#!/usr/bin/env python3
"""Devis de configuration OPNsense — la frontiere nord/sud du reseau converge.

DERIVE de deux sources deja tenues a jour, rien n'est saisi a la main :
  - le REGISTRE DES FLUX (`roles/*/meta/flux.yml`) : les flux dont le pair est
    `externe` sont, par construction, ceux que le pare-feu d'hote NE traite PAS
    (scripts/resoudre_flux.py les saute) parce qu'ils relevent de la frontiere.
    Plusieurs `raison` le disent explicitement (« filtre a l'OPNsense »).
  - les NOMENCLATURES des instances federees (`../*/plan/nomenclature.yml`) :
    les supernets tenants a router, decoulant du seul seed `index`.

ROLE ASSUME : frontiere NORD/SUD uniquement. Les passerelles de zone (SVI)
restent sur les switches L3 ; OPNsense ne porte AUCUN SVI de tenant et ne voit
pas le trafic est-ouest, qui reste filtre par les ACL de `make devis-reseau`.

NON destructif : ce script n'ecrit RIEN sur le pare-feu. Il produit un devis a
relire, puis (a venir) un JSON consommable par l'API OPNsense.

Usage :
  python3 scripts/devis_opnsense.py              # devis lisible
  python3 scripts/devis_opnsense.py --json       # meme contenu, pour l'API
  python3 scripts/devis_opnsense.py --verifier   # garde anti-lockout (P24)
"""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

import yaml

RACINE = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(RACINE / "scripts"))

from inventory_rules import supernet_de  # noqa: E402
# Helpers internes du resolveur de flux : reutilises VOLONTAIREMENT plutot que
# redupliques ici — le registre des flux et l'inventaire doivent avoir une seule
# lecture, sinon le devis et les nftables divergeraient en silence.
from resoudre_flux import (  # noqa: E402
    charger_flux,
    _enfants,
    _hotes_du_groupe,
    _inventaire,
    _ip_par_hote,
    _pairs,
    _sources_admin_ssh,
)
from devis_reseau import decouvrir, prefixe  # noqa: E402
# Le lien de transit (et donc le prochain saut) vit dans l'underlay : il est PARTAGE
# par tous les tenants, il ne peut deriver d'aucun index. Source unique, pas un intrant.
import underlay as underlay_mod  # noqa: E402

# Marqueurs de repli, tant que les intrants de frontiere ne sont pas renseignes
# (meme idiome que devis_reseau). Le panneau « Intrants de base » du GUI les ecrit
# dans group_vars/opnsense.yml : le devis se complete tout seul ensuite.
PROCHAIN_SAUT = "<PROCHAIN-SAUT-SWITCH>"
IF_TRANSIT = "<IF-TRANSIT>"
IF_WAN = "<IF-WAN>"


def intrants_frontiere() -> dict:
    """Intrants NON sensibles de la frontiere (group_vars/opnsense.yml). {} si absent."""
    for nom in ("principal", "production", "lab"):
        p = _inventaire().parent.parent / nom / "group_vars" / "opnsense.yml"
        if p.is_file():
            return yaml.safe_load(p.read_text(encoding="utf-8")) or {}
    return {}


def _ou_marqueur(intrants: dict, cle: str, marqueur: str) -> str:
    """La valeur d'intrant si elle est renseignee, sinon le marqueur a completer."""
    valeur = str(intrants.get(cle) or "").strip()
    return valeur or marqueur


def transit_underlay() -> dict | None:
    """Le reseau de transit declare dans l'underlay, ou None."""
    return underlay_mod.reseau_transit(underlay_mod.charger())


def nom_frontiere(transit: dict | None) -> str:
    """Nom du boitier de bordure, s'il est declare comme hote sur le lien de transit."""
    if not transit:
        return ""
    sortie = str(transit.get("passerelle_sortie") or "")
    for h in underlay_mod.hotes(underlay_mod.charger()):
        if h.get("reseau") == transit.get("nom") and str(h.get("ip")) == sortie:
            return str(h.get("nom") or "")
    return ""


def prochain_saut(transit: dict | None) -> str:
    """Prochain saut des routes tenants = le SVI du switch sur le lien de transit.

    DERIVE, jamais saisi : c'est le meme lien que celui d'ou `devis_reseau` tire la
    route par defaut et les routes de retour. Le declarer deux fois (une fois ici, une
    fois dans l'underlay) ouvrirait la porte a deux valeurs contradictoires — c'est
    exactement le mode de panne qu'on ferme. Marqueur tant qu'aucun transit n'existe.
    """
    return str((transit or {}).get("passerelle") or "").strip() or PROCHAIN_SAUT

# Le socle est porte par TOUS les hotes : sa destination n'est pas une poignee
# de machines mais le supernet du tenant. Evite un alias de 14 adresses.
ROLE_SOCLE = "serveur_debian"


def flux_frontiere(flux_par_role: dict[str, list[dict]]) -> list[tuple[str, dict]]:
    """[(role, flux)] pour tout flux dont le pair inclut `externe`, trie."""
    retenus = [
        (role, fl)
        for role, flux in sorted(flux_par_role.items())
        for fl in flux
        if "externe" in _pairs(fl)
    ]
    retenus.sort(key=lambda rf: (rf[1].get("sens", ""), rf[0], str(rf[1].get("port"))))
    return retenus


def _ports(flux: dict) -> list[str]:
    port = flux["port"]
    return [str(p) for p in (port if isinstance(port, list) else [port])]


def inventaire_de(nom_instance: str) -> Path | None:
    """hosts.yml d'une instance FEDEREE, active ou non. None si elle n'en a pas.

    La frontiere est un equipement PARTAGE : elle route vers tous les tenants, elle
    doit donc aussi porter leurs regles. Se limiter a l'instance active laisserait un
    tenant route mais muet — chemin present, politique absente.
    """
    base = RACINE.parent / nom_instance / "inventories"
    for env in ("principal", "production", "lab"):
        p = base / env / "hosts.yml"
        if p.is_file():
            return p
    return None


def cibles_par_role(role: str, inventaire: Path) -> list[str]:
    """IP des hotes qui portent ce role dans cette instance (actifs ET planifies).

    On inclut les planifies a dessein : la frontiere se prepare avant que les VM
    existent, sinon le devis changerait a chaque materialisation d'hote.
    """
    data = yaml.safe_load(inventaire.read_text(encoding="utf-8")) or {}
    ips = _ip_par_hote(data)
    return sorted({ips[h] for h in _hotes_du_groupe(data, role) if h in ips})


def admin_de(nom_instance: str) -> list[str]:
    """Reseaux d'administration declares par une instance (intrant nftables_admin_ssh)."""
    inv = inventaire_de(nom_instance)
    if not inv:
        return []
    dossier = inv.parent / "group_vars" / "all"
    for fichier in sorted(dossier.glob("*.yml")):
        if "vault" in fichier.name:
            continue
        data = yaml.safe_load(fichier.read_text(encoding="utf-8")) or {}
        if isinstance(data, dict) and data.get("nftables_admin_ssh"):
            src = data["nftables_admin_ssh"]
            return [str(s) for s in src] if isinstance(src, list) else [str(src)]
    return []


def nom_alias(role: str, etiquette: str) -> str:
    """Alias d'hotes, PREFIXE par le tenant : deux tenants ont chacun leur nginx."""
    return f"SETOPS_{etiquette}_{role.upper()}"


def _instance_active() -> tuple[str, dict] | tuple[None, None]:
    """(nom, nomenclature) de l'instance vers laquelle pointe le symlink `instance`."""
    cible = (RACINE / "instance").resolve().name
    for nom, _pfx, n in decouvrir():
        if nom == cible:
            return nom, n
    return None, None


def construire(tenants: list[tuple[str, str, dict]]) -> dict:
    """Le devis, en structure — rendue ensuite en texte ou en JSON."""
    flux_par_role = charger_flux()
    retenus = flux_frontiere(flux_par_role)
    _intr = intrants_frontiere()
    if_wan = _ou_marqueur(_intr, "opnsense_if_wan", IF_WAN)
    if_transit = _ou_marqueur(_intr, "opnsense_if_transit", IF_TRANSIT)
    nom_actif, nomenclature_active = _instance_active()

    alias: dict[str, dict] = {}
    for nom, pfx, n in tenants:
        alias[f"SETOPS_TENANT_{pfx}{n['index']}"] = {
            "type": "network",
            "contenu": [supernet_de(n["index"])],
            "description": f"Supernet du tenant {nom} (index {n['index']})",
        }

    # Reseaux d'administration : UN ALIAS PAR TENANT, jamais une union. Chaque tenant
    # declare les siens (intrant `nftables_admin_ssh`) et ils n'ouvrent QUE son supernet :
    # une union laisserait le plan de gestion d'un tenant entrer chez le voisin, ce que
    # les ACL de switch interdisent par ailleurs. La frontiere ne doit pas rouvrir ce que
    # l'isolation inter-tenant ferme.
    admin_par_tenant: dict[str, list[str]] = {}
    for nom, pfx, n in tenants:
        etiquette = f"{pfx}{n['index']}"
        cidrs = admin_de(nom)
        admin_par_tenant[etiquette] = cidrs
        if cidrs:
            alias[f"SETOPS_ADMIN_{etiquette}"] = {
                "type": "network",
                "contenu": list(cidrs),
                "description": f"Reseaux d'administration de {nom} (intrant "
                               f"nftables_admin_ssh) — source unique, partagee avec ses nftables",
            }
    # Union, pour les routes de RETOUR cote switch et la garde P24 : router n'est pas
    # autoriser. Le switch doit savoir revenir vers tous les plans de gestion.
    admin = sorted({c for cidrs in admin_par_tenant.values() for c in cidrs})

    supernet_actif = (
        supernet_de(nomenclature_active["index"]) if nomenclature_active else None
    )

    # Les regles sont emises POUR CHAQUE TENANT FEDERE, pas seulement l'actif : la
    # frontiere route deja vers eux tous (section 2). N'en filtrer qu'un laisserait les
    # autres routes menant a un `block` muet.
    regles: list[dict] = []
    tenants_sans_inventaire: list[str] = []
    tenants_sans_admin: list[str] = []
    for nom, pfx, n in tenants:
        etiquette = f"{pfx}{n['index']}"
        alias_tenant = f"SETOPS_TENANT_{etiquette}"
        inv = inventaire_de(nom)
        if inv is None:
            tenants_sans_inventaire.append(nom)
            continue
        for role, fl in retenus:
            if role == ROLE_SOCLE:
                destination = alias_tenant
            else:
                cibles = cibles_par_role(role, inv)
                if not cibles:
                    continue  # role absent de ce tenant : aucune regle a poser
                a = nom_alias(role, etiquette)
                alias[a] = {
                    "type": "host",
                    "contenu": cibles,
                    "description": f"Hotes portant {role} (instance {nom})",
                }
                destination = a
            entrant = fl.get("sens") == "ingress"
            # Le SSH de gestion est le SEUL flux `externe` a ne pas venir d'Internet :
            # sa `raison` dit « l'acces depuis l'exterieur est filtre a l'OPNsense ».
            ssh_gestion = entrant and "22" in _ports(fl)
            if ssh_gestion and not admin_par_tenant.get(etiquette):
                # Sans reseau d'administration declare, la regle n'aurait pas de source :
                # l'omettre ferme l'acces, l'ouvrir a `any` exposerait le SSH a Internet.
                tenants_sans_admin.append(nom)
                continue
            source = f"SETOPS_ADMIN_{etiquette}" if ssh_gestion else "any"
            regles.append({
                "sens": "in" if entrant else "out",
                # Interface d'ARRIVEE du paquet, celle a laquelle OPNsense attache la
                # regle. Entrant (depuis Internet ou le VPN d'admin, qui revient par
                # l'adresse publique) : le WAN. Sortant (la flotte vers Internet) : le
                # lien de transit, par ou le paquet penetre le pare-feu.
                "interface": if_wan if entrant else if_transit,
                "protocole": fl.get("protocole", "tcp"),
                "source": source if entrant else destination,
                "destination": destination if entrant else "any",
                "ports": _ports(fl),
                "chiffrement": fl.get("chiffrement"),
                "role": role,
                "tenant": nom,
                "raison": fl.get("raison", ""),
            })

    intrants = intrants_frontiere()
    transit = transit_underlay()
    saut = prochain_saut(transit)
    routes = [
        {
            "reseau": supernet_de(n["index"]),
            "prochain_saut": saut,
            "description": f"{nom} (index {n['index']}) — routage inter-zone sur les switches L3",
        }
        for nom, _pfx, n in tenants
    ]

    return {
        "role_frontiere": "nord-sud",
        "instance_active": nom_actif,
        "supernet_actif": supernet_actif,
        "api_url": str(intrants.get("opnsense_api_url") or "").strip(),
        "wan_ip": str(intrants.get("opnsense_wan_ip") or "").strip(),
        "if_wan": _ou_marqueur(intrants, "opnsense_if_wan", IF_WAN),
        "if_transit": _ou_marqueur(intrants, "opnsense_if_transit", IF_TRANSIT),
        "alias": alias,
        "routes": routes,
        "regles": regles,
        "admin": admin,
        "tenants_sans_inventaire": tenants_sans_inventaire,
        "tenants_sans_admin": sorted(set(tenants_sans_admin)),
        "transit": {
            "nom": (transit or {}).get("nom"),
            "vlan": (transit or {}).get("vlan"),
            "sous_reseau": (transit or {}).get("sous_reseau"),
            "svi_switch": saut,
            "adresse_frontiere": (transit or {}).get("passerelle_sortie"),
            "nom_frontiere": nom_frontiere(transit),
        } if transit else None,
        "reciproque_switch": [
            f"ip route {cidr} -> {saut} (retour vers l'administration)"
            for cidr in admin
        ],
    }


def rendre(devis: dict) -> str:
    out = [
        "# ============================================================",
        "# DEVIS OPNSENSE — frontiere nord/sud (Set-OPS)",
        "# Genere par scripts/devis_opnsense.py. NE PAS EDITER A LA MAIN.",
        "# Derive du registre des flux (pair: externe) + des nomenclatures.",
        "# Les passerelles de zone restent sur les switches L3 : aucun SVI ici.",
        "# ============================================================",
        "",
        "## 0. Prerequis reciproques — COTE SWITCHES",
        "# Sans ces routes de retour, le trafic part mais ne revient pas : la",
        "# reponse sort par une autre interface et le pare-feu la jette (hors etat).",
    ]
    out += [f"#   {r}" for r in devis["reciproque_switch"]] or ["#   (aucun reseau d'administration declare)"]
    tr = devis.get("transit")
    if tr:
        out.append(f"# DEJA EMIS par `make devis-reseau` (section 5), derive du transit"
                   f" '{tr['nom']}' de l'underlay.")
    else:
        out.append("# A EMETTRE : aucun reseau de transit n'est declare dans underlay.yml")
        out.append("# (cle `passerelle_sortie`). Sans lui, ni sortie ni retour ne sont derivables.")
    out += [
        "",
        "## 1. Interfaces",
        f"#   Gestion : {devis['api_url'] or '(intrant opnsense_api_url a renseigner)'}",
        f"#   WAN     : {devis['if_wan']}"
        + (f"  —  {devis['wan_ip']} (face publique de l'ecosysteme)" if devis.get("wan_ip") else ""),
        f"#   TRANSIT : {devis['if_transit']} (trunk 802.1Q vers les switches)",
    ]
    if tr:
        out += [
            f"#             lien '{tr['nom']}' {tr['sous_reseau']} — VLAN {tr['vlan']} a etiqueter",
            f"#             switch {tr['svi_switch']}  <->  frontiere"
            f" {tr['adresse_frontiere']}"
            + (f" ({tr['nom_frontiere']})" if tr.get("nom_frontiere") else ""),
        ]
    out += [
        "",
        "## 2. Routes statiques (vers les supernets tenants)",
    ]
    for r in devis["routes"]:
        out.append(f"route add {r['reseau']} via {r['prochain_saut']}    # {r['description']}")
    out += ["", "## 3. Alias"]
    for nom, a in devis["alias"].items():
        contenu = ", ".join(a["contenu"]) if a["contenu"] else "(VIDE)"
        out.append(f"alias {nom:32} {a['type']:8} {contenu}")
        out.append(f"#     {a['description']}")
    out += ["", "## 4. Regles — PAR TENANT FEDERE (defaut : deny + journalisation)",
            "# La frontiere route vers tous les tenants (section 2) : elle porte donc aussi",
            "# leurs regles. Chacun n'est joignable en SSH que depuis SES PROPRES reseaux",
            "# d'administration — jamais ceux du voisin, que l'isolation inter-tenant ferme.",
            "# Chaque regle est `in` sur son interface d'ARRIVEE : le WAN pour ce qui vient",
            "# d'Internet (et du VPN d'administration, qui revient par l'adresse publique),",
            "# le lien de transit pour ce que la flotte emet vers l'exterieur."]
    for nom in devis.get("tenants_sans_inventaire", []):
        out.append(f"# /!\\ {nom} : aucun inventaire genere — AUCUNE regle. Route, puis bloque.")
    for nom in devis.get("tenants_sans_admin", []):
        out.append(f"# /!\\ {nom} : `nftables_admin_ssh` vide — regle SSH OMISE. L'ouvrir a "
                   f"`any` exposerait le SSH a Internet ; renseigner l'intrant.")
    for rg in devis["regles"]:
        ports = ",".join(rg["ports"])
        # Dans OPNsense une regle est TOUJOURS `in` sur l'interface d'arrivee : c'est
        # la que le paquet penetre le pare-feu. Le sens du flux (entrant/sortant vis-a-vis
        # de l'ecosysteme) se lit alors dans l'interface, pas dans un mot-cle.
        out.append(
            f"pass in on {rg['interface']:8} proto {rg['protocole']:3} "
            f"from {rg['source']:24} to {rg['destination']:24} port {ports:7} "
            f"# {rg['role']} ({rg['chiffrement']})"
        )
        out.append(f"#     {rg['raison']}")
    sortants = sum(1 for r in devis["regles"] if r["sens"] == "out")
    out += [
        "",
        "## 5. Defaut — DENY DANS LES DEUX SENS",
        "# `block out` est un vrai default-deny : ce qui n'est pas declare en section 4",
        f"# ne sort PAS. {sortants} regle(s) sortante(s) y figurent, toutes derivees du",
        "# registre des flux (`sens: egress`, `pair: externe`).",
        "#",
        "# Un besoin sortant oublie ne se manifeste pas par un refus clair : apt expire,",
        "# l'horloge derive et les certificats deviennent invalides des semaines plus tard.",
        "# Le declarer dans `roles/<role>/meta/flux.yml` — ne jamais l'ouvrir ici a la main,",
        "# la regle serait perdue a la prochaine generation.",
        "block in  log all    # tout ce qui n'est pas ci-dessus",
        "block out log all",
    ]
    return "\n".join(out)


def verifier(devis: dict) -> tuple[bool, list[str]]:
    """Garde anti-lockout : la frontiere ne doit pas etre generee a l'aveugle."""
    erreurs: list[str] = []
    if not devis["admin"]:
        erreurs.append(
            "Aucun reseau d'administration declare (intrant `nftables_admin_ssh` vide) : "
            "la regle SSH entrante n'aurait aucune source et le defaut deny fermerait "
            "l'acces d'administration. Declarer le(s) CIDR d'admin (VPN compris)."
        )
    if not devis["routes"]:
        erreurs.append("Aucun tenant federe decouvert : rien a router.")
    for nom, a in devis["alias"].items():
        if not a["contenu"]:
            erreurs.append(f"Alias {nom} vide : une regle le referencerait dans le vide.")
    return (not erreurs), erreurs


def main(argv: list[str]) -> int:
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--json", action="store_true", help="Sortie structuree (future API OPNsense).")
    ap.add_argument("--verifier", action="store_true", help="Valide le devis sans l'afficher.")
    args = ap.parse_args(argv)

    tenants = decouvrir()
    if not tenants:
        print("Aucune instance federee trouvee (../*/plan/nomenclature.yml).", file=sys.stderr)
        return 1
    devis = construire(tenants)

    if args.verifier:
        ok, erreurs = verifier(devis)
        for e in erreurs:
            print(f"ECHEC : {e}", file=sys.stderr)
        if ok:
            print(f"CONFORME : frontiere nord/sud, {len(devis['regles'])} regles, "
                  f"{len(devis['routes'])} routes, admin={','.join(devis['admin'])}.")
        return 0 if ok else 1

    print(json.dumps(devis, indent=2, ensure_ascii=False) if args.json else rendre(devis))
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))