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 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 cibles_par_role(role: str) -> list[str]:
    """IP des hotes qui portent ce role, dans l'instance active (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 nom_alias(role: str) -> str:
    return f"SETOPS_{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)
    admin = _sources_admin_ssh()
    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']})",
        }
    alias["SETOPS_ADMIN"] = {
        "type": "network",
        "contenu": list(admin),
        "description": "Reseaux d'administration (intrant nftables_admin_ssh) — "
                       "source unique, partagee avec les nftables d'hote",
    }

    supernet_actif = (
        supernet_de(nomenclature_active["index"]) if nomenclature_active else None
    )
    alias_tenant_actif = next(
        (a for a in alias if a.startswith("SETOPS_TENANT_") and nom_actif
         and supernet_de(nomenclature_active["index"]) in alias[a]["contenu"]),
        None,
    )

    regles: list[dict] = []
    for role, fl in retenus:
        if role == ROLE_SOCLE:
            destination = alias_tenant_actif or "any"
        else:
            cibles = cibles_par_role(role)
            if not cibles:
                continue  # role absent de cette instance : aucune regle a poser
            a = nom_alias(role)
            alias[a] = {
                "type": "host",
                "contenu": cibles,
                "description": f"Hotes portant {role} (instance {nom_actif})",
            }
            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 ».
        source = "SETOPS_ADMIN" if (entrant and "22" in _ports(fl)) else "any"
        regles.append({
            "sens": "in" if entrant else "out",
            "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,
            "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(),
        "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,
        "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"),
        } 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"#   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 {tr['adresse_frontiere']}",
        ]
    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 (defaut : deny + journalisation)"]
    for rg in devis["regles"]:
        ports = ",".join(rg["ports"])
        out.append(
            f"pass {rg['sens']:3} 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']}")
    out += [
        "",
        "## 5. Defaut",
        "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:]))