Set-OPS-Public/scripts/tests/test_frontiere_pki.py

#!/usr/bin/env python3
"""L'autorite de la frontiere est bornee, et ses cles ne quittent pas la boite.

POURQUOI (2026-10-12). La frontiere porte desormais une autorite subordonnee signee par la
racine du site (scripts/frontiere_pki.py). Ce test garde ce qui la rend sure :
  1. le gabarit de signature : aucune autorite sous elle, signature seulement, contrainte
     de noms CRITIQUE sur le domaine et le supernet du site ;
  2. la lecture du magasin ne passe pas par `/api/trust/*/search`, qui renvoie les cles
     privees (mesure du 2026-10-12 : la liste rapatriait la cle de l'autorite sur le poste) ;
  3. les noms du certificat sont ELIGIBLES : FQDN et adresses du supernet seulement ;
  4. l'interface web (R3) reste un geste a part, derriere sa propre confirmation ;
  5. le registre dit `tls` pour le syslog (1514) et l'exportateur de la frontiere.
"""
from __future__ import annotations

import ipaddress
import json
import re
import sys
from pathlib import Path

import yaml

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

import frontiere_pki as fp  # noqa: E402


def main() -> int:
    echecs: list[str] = []

    # 1. Le gabarit, rendu comme le script le rend.
    rendu = fp.GABARIT_SOUS_AC.format(domaine=json.dumps("exemple.internal"), supernet=json.dumps("10.99.0.0/16"))
    gabarit = json.loads(rendu.replace("{{ toJson .Subject }}", "{}"))
    if gabarit.get("basicConstraints") != {"isCA": True, "maxPathLen": 0}:
        echecs.append(f"basicConstraints = {gabarit.get('basicConstraints')}, attendu isCA + maxPathLen 0")
    if sorted(gabarit.get("keyUsage") or []) != ["certSign", "crlSign"]:
        echecs.append(f"keyUsage = {gabarit.get('keyUsage')}, attendu certSign + crlSign seulement")
    nc = gabarit.get("nameConstraints") or {}
    if not nc.get("critical") or nc.get("permittedDNSDomains") != ["exemple.internal"] \
            or nc.get("permittedIPRanges") != ["10.99.0.0/16"]:
        echecs.append(f"nameConstraints = {nc}, attendu critique, domaine et supernet du site")

    # 2. Aucune lecture du magasin par la recherche de l'API (elle rend les cles).
    source = (RACINE / "scripts/frontiere_pki.py").read_text(encoding="utf-8")
    code = "\n".join(l for l in source.splitlines() if not l.lstrip().startswith("#"))
    if re.search(r"/api/trust/[^\"']*/search", code.replace("`/api/trust/{ca,cert}/search`", "")):
        echecs.append("frontiere_pki.py lit le magasin par /api/trust/*/search, qui renvoie les cles privees")
    if "/conf/config.xml" not in code or "'crt'" not in code:
        echecs.append("la lecture du magasin ne se fait plus sur la frontiere (config.xml)")

    # 3. Les noms du certificat, tels que l'inventaire du site les derive.
    try:
        import site_inventaire as si
        hv = si.inventaire()["_meta"]["hostvars"]
        f = (hv.get(fp.HOTE_PKI) or {}).get("setops_frontiere") or {}
    except Exception:  # noqa: BLE001 — aucun site monte
        f = {}
    if f:
        reseau = ipaddress.ip_network(f["supernet"])
        hors = [a for a in f["adresses"] if ipaddress.ip_address(a) not in reseau]
        if hors:
            echecs.append(f"adresses hors du supernet {reseau} dans le certificat : {hors}")
        if not f.get("admin") or f["admin"] not in f["adresses"]:
            echecs.append("l'adresse d'administration de la frontiere n'est pas dans ses adresses du site")

    # 4. L'interface web : un geste a part, confirme a part.
    if "def basculer_interface" not in code or 'action == "interface"' not in code:
        echecs.append("la bascule de l'interface web n'est plus une commande separee")
    if re.search(r"if action == .appliquer.[\s\S]*basculer_interface\(", code.split("def main")[1].split('if action == "interface"')[0]):
        echecs.append("`appliquer` bascule l'interface web (R3) sans sa propre confirmation")

    # 5. Le registre.
    def chiffrement(role: str, port: int, pair: str) -> str:
        flux = yaml.safe_load((RACINE / f"roles/{role}/meta/flux.yml").read_text(encoding="utf-8"))["flux"]
        return next((x.get("chiffrement") for x in flux
                     if x.get("port") == port and pair in (x.get("pair") if isinstance(x.get("pair"), list) else [x.get("pair")])), "")
    if chiffrement("serveur_loki", 1514, "fabric") != "tls-requis":
        echecs.append("le registre ne dit pas le syslog de la frontiere (1514) en tls-requis")
    if chiffrement("serveur_prometheus", 9100, "frontiere") != "tls":
        echecs.append("le registre ne dit pas l'exportateur de la frontiere en tls")

    for e in echecs:
        print(f"ECHEC : {e}")
    if not echecs:
        print("OK : autorite de la frontiere bornee, magasin lu sans ses cles, noms eligibles, interface a part.")
    return 1 if echecs else 0


if __name__ == "__main__":
    raise SystemExit(main())