Set-OPS-Public/scripts/devis_frontiere.py

#!/usr/bin/env python3
"""Devis d'etancheite de la frontiere : ce qui n'est pas declare est-il refuse ?

`devis_expositions.py` pose la question POSITIVE — chaque exposition declaree repond-elle.
Celui-ci pose la NEGATIVE, et c'est une autre question : la frontiere laisse-t-elle passer
ce que la politique n'autorise pas. Un pare-feu peut tres bien servir tout ce qu'on lui
demande *et* laisser passer tout le reste.

La politique attendue n'est pas saisie : elle est LUE dans `devis_opnsense.py`, la meme
source qui produit les regles posees sur le boitier. Les cibles ne sont pas saisies non
plus : ce sont les ports REELLEMENT EN ECOUTE dans la flotte, releves par le playbook.
Sonder un port ferme ne prouverait rien du pare-feu.

Deux modes :
  --cibles   lit le releve des ports en ecoute (stdin), emet les listes a sonder
  (defaut)   lit le releve complet et rend le verdict, code 1 s'il y a un ecart

Aucun acces reseau ici — c'est le playbook qui sonde (D-23).

  make frontiere-mesurer
"""
from __future__ import annotations

import argparse
import ipaddress
import json
import subprocess
import sys
from pathlib import Path

RACINE = Path(__file__).resolve().parent.parent

# Adresse de controle : documentee comme non routable et attribuee a personne. Si elle
# repond, c'est l'instrument qui ment — voir l'en-tete de `sonde_tcp.py`.
CONTROLE = ["172.31.99.99", 22]


def _politique() -> dict:
    """Ce que la frontiere est CENSEE autoriser, lu a la source qui la configure."""
    r = subprocess.run([sys.executable, str(RACINE / "scripts" / "devis_opnsense.py"),
                        "--json"], cwd=RACINE, capture_output=True, text=True)
    if r.returncode != 0:
        raise SystemExit("Le devis de la frontiere ne se genere pas :\n"
                         + r.stderr.strip()[:400])
    return json.loads(r.stdout)


def _adresses(devis: dict, nom_alias: str) -> list[str]:
    """Les adresses derriere un alias — hotes tels quels, reseaux developpes en prefixe."""
    a = devis["alias"].get(nom_alias.lstrip("!"))
    return list(a["contenu"]) if a else []


def autorise_depuis_gestion(devis: dict) -> set[tuple[str, int]]:
    """Couples (adresse, port) que le VLAN d'administration a le DROIT d'atteindre."""
    permis: set[tuple[str, int]] = set()
    for r in devis["regles"]:
        if r["interface"] != devis.get("if_gestion") or r["sens"] != "in":
            continue
        if r["protocole"] not in ("tcp",):
            continue
        for cible in _adresses(devis, r["destination"]):
            for p in r["ports"] or []:
                if str(p).isdigit():
                    permis.add((cible, int(p)))
    return permis


def _couvert(ip: str, permis: set[tuple[str, int]], port: int) -> bool:
    """Un permis peut etre un hote OU un reseau (le `/24` d'un tenant, par exemple)."""
    for cible, p in permis:
        if p != port:
            continue
        if cible == ip:
            return True
        if "/" in cible:
            try:
                if ipaddress.ip_address(ip) in ipaddress.ip_network(cible, strict=False):
                    return True
            except ValueError:
                continue
    return False


def cibles(ports_par_hote: dict) -> dict:
    """Les listes a sonder, derivees de la politique et de ce qui ecoute vraiment."""
    devis = _politique()
    permis = autorise_depuis_gestion(devis)

    admin = []
    for hote, info in sorted(ports_par_hote.items()):
        ip = str(info["ip"])
        for port in sorted(set(info["ports"])):
            attendu = "LIVRE" if _couvert(ip, permis, int(port)) else "REFUSE"
            admin.append([ip, int(port), f"{hote}:{port} attendu={attendu}"])

    # Sens sortant : la frontiere elle-meme, qu'aucun tenant ne doit joindre. C'est le
    # trou mesure le 2026-08-09 — `https://10.0.0.1/` repondait depuis une VM.
    interne = []
    for r in devis["regles"]:
        if r["interface"] == devis.get("if_gestion") and r["sens"] == "in":
            for c in _adresses(devis, r["source"]):
                if "/" in c:
                    reseau = ipaddress.ip_network(c, strict=False)
                    interne.append([str(next(reseau.hosts())), 443,
                                    "frontiere depuis le tenant attendu=REFUSE"])
    vus, tenant = set(), []
    for c in interne:
        if c[0] not in vus:
            vus.add(c[0])
            tenant.append(c)
    return {"controle": CONTROLE, "admin": admin, "tenant": tenant}


def analyser(releve: dict) -> tuple[list[str], list[str]]:
    """Confronte les verdicts sondes a la politique. Rend (ecarts, lignes lisibles)."""
    ecarts: list[str] = []
    lignes: list[str] = []

    for vantage, cle in (("VLAN d'administration", "admin"), ("depuis un tenant", "tenant")):
        bloc = releve.get(cle) or {}
        if not bloc:
            continue
        if not bloc.get("valide", False):
            ecarts.append(
                f"Releve {vantage} NUL : le controle {CONTROLE[0]} a repondu — quelque "
                "chose sur le chemin repond a la place des destinations. Aucun verdict "
                "de ce point de vue n'a de sens tant que ce n'est pas corrige."
            )
            continue
        lignes.append(f"  --- {vantage} ---")
        for r in bloc.get("resultats") or []:
            etiq = str(r.get("etiquette") or "")
            attendu = etiq.rsplit("attendu=", 1)[-1] if "attendu=" in etiq else "?"
            obtenu = str(r.get("verdict"))
            # AMBIGU vaut « la connexion s'etablit sans rien livrer » : pour un port qui
            # doit etre REFUSE, c'est un ecart — la frontiere a relaye. Pour un port qui
            # doit etre LIVRE, c'est aussi un ecart : on n'a pas la preuve qu'il sert.
            conforme = (attendu == obtenu)
            lignes.append(f"    {'ok ' if conforme else 'ECART'} {etiq[:52]:<52} "
                          f"{obtenu:<7} {str(r.get('preuve'))[:40]}")
            if conforme:
                continue
            # Deux natures d'ecart, qui n'appellent pas le meme geste :
            #  - REFUSE attendu, connexion etablie  -> la bordure a RELAYE. C'est le trou.
            #  - LIVRE attendu, rien livre          -> la bordure autorise, l'HOTE refuse.
            #    Rien ne fuit ; ce sont les deux couches qui ne disent pas la meme chose.
            if attendu == "LIVRE":
                ecarts.append(f"{etiq} — la frontiere l'autorise mais l'hote le refuse : "
                              "les deux couches ne declarent pas la meme politique")
            else:
                ecarts.append(f"{etiq} — obtenu {obtenu} : la frontiere a relaye "
                              f"({r.get('preuve')})")
    return ecarts, lignes


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--cibles", action="store_true",
                    help="emet les listes a sonder (lit les ports en ecoute sur stdin)")
    ap.add_argument("--releve", default=None, help="fichier de releve a analyser")
    a = ap.parse_args(argv)

    if a.cibles:
        json.dump(cibles(json.load(sys.stdin)), sys.stdout)
        return 0

    chemin = Path(a.releve) if a.releve else RACINE / "instance" / "devis-frontiere.json"
    if not chemin.is_file():
        raise SystemExit(f"Aucun releve : {chemin}\nLancer d'abord `make frontiere-mesurer`.")
    ecarts, lignes = analyser(json.loads(chemin.read_text()))
    print("\n".join(lignes))
    if ecarts:
        print(f"\nNON CONFORME : {len(ecarts)} ecart(s) — la frontiere ne dit pas ce que "
              "la politique dit.")
        for e in ecarts:
            print("  -", e)
        return 1
    print("\nCONFORME : tout ce qui est declare est livre, tout le reste est refuse.")
    return 0


if __name__ == "__main__":
    sys.exit(main())