Set-OPS-Public/scripts/expositions_etat.py

#!/usr/bin/env python3
"""Chaque exposition du plan est-elle VRAIMENT servie, et par le bon certificat ?
Ne corrige rien.

POURQUOI CE CONTROLE EXISTE (2026-09-11).

Renommer une exposition touche cinq choses, et une seule ne suit pas toute seule. Mesure
deux fois le 2026-09-10, sur `grafana -> observatoire` puis sur `icinga -> vigie` :

    serveur_powerdns   la zone publie le nouveau nom          ✓ au deploiement
    serveur_keycloak   le client OIDC accepte le retour       ✓ au deploiement
    le service         il fabrique ses URL avec le bon nom    ✓ (P67)
    serveur_nginx      le vhost repond sur le nouveau nom     ✓ au deploiement
    client_pki         le SAN du certificat porte le nom      ✗ PAS au deploiement

`serveur_nginx` pose le vhost et laisse le SAN en arriere : c'est `client_pki` qui reemet,
et rien ne le reclame. Le symptome est trompeur — le site repond, la page s'affiche, et
c'est le NAVIGATEUR qui refuse, avec une erreur de certificat que personne ne relie a un
renommage fait la veille.

CE QU'ON COMPARE, ET DANS LES DEUX SENS :

  1. les expositions DECLAREES par `plan/applications.yml` (la verite) ;
  2. les noms PORTES par le certificat reellement SERVI (pas celui sur le disque) ;
  3. le code que le vhost REND pour ce nom.

Le sens inverse compte autant : un nom qui reste dans le SAN apres avoir disparu du plan
est un nom que le certificat continue d'authentifier. C'est exactement ce qu'a laisse le
premier renommage, jusqu'au passage de `client_pki`.

POURQUOI CE N'EST PAS UNE PREUVE DU HARNAIS. `make prouver` est STATIQUE : il lit le
depot, zero appel reseau, et c'est ce qui le rend rejouable partout. Rien de statique ne
peut lire un certificat SERVI — il faut ouvrir la connexion. C'est donc un controle a la
demande, comme `make gabarit-etat` et `make routes-fabric-etat`.

Usage :
  python3 scripts/expositions_etat.py           # l'instance montee
  python3 scripts/expositions_etat.py --site    # l'ecosysteme du SITE
"""
from __future__ import annotations

import re
import subprocess
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

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

# LE PORT EST UNE CONVENTION, PAS UNE DERIVATION. Le champ `port:` du plan est celui que
# le service ECOUTE en local (Forgejo 3000), pas celui sous lequel il est EXPOSE. Toutes
# les expositions de Set-OPS sont servies en TLS sur 443 ; le jour ou ce ne sera plus vrai,
# c'est ici qu'il faudra le declarer.
PORT = 443


def _san_servi(adresse: str, fqdn: str) -> tuple[list[str], str]:
    """Les noms du certificat que le serveur PRESENTE pour ce SNI. ([], raison) si muet."""
    try:
        poignee = subprocess.run(
            ["openssl", "s_client", "-connect", f"{adresse}:{PORT}",
             "-servername", fqdn, "-verify_quiet"],
            stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=20)
    except subprocess.TimeoutExpired:
        return [], "silence (aucune reponse TLS)"
    if "BEGIN CERTIFICATE" not in poignee.stdout:
        premiere = (poignee.stderr or "").strip().splitlines()
        return [], (premiere[0][:70] if premiere else "aucun certificat presente")
    texte = subprocess.run(
        ["openssl", "x509", "-noout", "-ext", "subjectAltName"],
        input=poignee.stdout, capture_output=True, text=True, timeout=20).stdout
    return sorted(set(re.findall(r"DNS:([^\s,]+)", texte))), ""


def _code_http(adresse: str, fqdn: str) -> str:
    r = subprocess.run(
        ["curl", "-sk", "-o", "/dev/null", "-w", "%{http_code}", "--max-time", "12",
         "--resolve", f"{fqdn}:{PORT}:{adresse}", f"https://{fqdn}/"],
        capture_output=True, text=True, timeout=25)
    return (r.stdout or "000").strip()


def _expositions_et_porteurs(site: bool) -> tuple[dict[str, tuple[str, str]], str]:
    """{fqdn expose: (adresse qui le sert, nom court du porteur)}. L'adresse vient de `sans_exposition`, pose par
    le generateur sur l'hote dont les groupes contiennent l'edge de cette exposition —
    la MEME derivation que celle qui alimente le certificat. Interroger une autre source
    reviendrait a fabriquer une troisieme liste."""
    import yaml
    if site:
        # L'INVENTAIRE DU SITE EST DYNAMIQUE, PAS STATIQUE : il rend
        # `{groupe: {hosts: [noms]}, _meta: {hostvars: {...}}}` — la forme qu'Ansible
        # attend d'un script, et non l'arbre `all.children` d'un `hosts.yml`.
        import site_inventaire
        inv = site_inventaire.inventaire()
        hotes = dict((inv.get("_meta") or {}).get("hostvars") or {})
        etiquette = "SITE"
    else:
        from inventory_rules import instance_courante
        chemin = instance_courante().resolve()
        fichier = next(iter(sorted((chemin / "inventories").glob("*/hosts.yml"))), None)
        if not fichier:
            return {}, ""
        data = yaml.safe_load(fichier.read_text(encoding="utf-8")) or {}
        enfants = (data.get("all") or {}).get("children") or {}
        hotes = {}
        for etat in ("hotes_actifs", "hotes_planifies"):
            for nom, v in ((enfants.get(etat) or {}).get("hosts") or {}).items():
                hotes[nom] = v or {}
        etiquette = chemin.name

    porteurs: dict[str, tuple[str, str]] = {}
    for nom, v in hotes.items():
        adresse = str(v.get("ansible_host") or "")
        for fqdn in (v.get("sans_exposition") or []):
            if adresse:
                porteurs[str(fqdn)] = (adresse, str(nom))
    if porteurs:
        return porteurs, etiquette

    # PAS D'EDGE : LA MACHINE SERT SON PROPRE TLS (le cas du SITE).
    #
    # `sans_exposition` est pose sur l'hote dont les groupes contiennent l'EDGE de
    # l'exposition. Le site n'en a aucun — `site-forge-01` ecoute lui-meme sur 443. Le
    # porteur se derive alors de l'application : son `groupe` dit quel hote la sert.
    # Deux formes de terminaison, une seule question posee.
    import underlay as underlay_mod
    apps = (underlay_mod.lire_plan_site("applications.yml") or {}).get("applications") or {}
    hote_du_groupe: dict[str, tuple[str, str]] = {}
    for nom, v in hotes.items():
        adresse = str(v.get("ansible_host") or "")
        if adresse:
            hote_du_groupe.setdefault(str(nom), (adresse, str(nom)))
    for app in apps.values():
        if not isinstance(app, dict):
            continue
        cible = hote_du_groupe.get(str(app.get("hote") or ""))
        if not cible:
            continue
        for fqdn in (app.get("expose") or []):
            porteurs[str(fqdn)] = cible
    return porteurs, etiquette


def main() -> int:
    site = "--site" in sys.argv[1:]
    porteurs, etiquette = _expositions_et_porteurs(site)
    if not porteurs:
        print("Aucune exposition derivee : rien a comparer. "
              "(`make instancier` a-t-il tourne ?)")
        return 0

    print(f"{etiquette} : {len(porteurs)} exposition(s) declaree(s), "
          f"servies sur le port {PORT}.\n")

    ecarts: list[str] = []
    # Un meme porteur sert souvent plusieurs noms : on ne lit son certificat qu'une fois
    # par (adresse, sni) et on garde le SAN pour la comparaison inverse.
    san_par_adresse: dict[str, list[str]] = {}

    noms_propres: dict[str, str] = {a: n for a, n in porteurs.values()}
    for fqdn, (adresse, _porteur) in sorted(porteurs.items()):
        san, raison = _san_servi(adresse, fqdn)
        if not san:
            print(f"  {fqdn:<34} {adresse:<14} INJOIGNABLE — {raison}")
            ecarts.append(f"{fqdn} : {raison} (depuis CE poste ; "
                          f"la frontiere peut etre le mur, pas le service)")
            continue
        san_par_adresse.setdefault(adresse, san)
        code = _code_http(adresse, fqdn)
        dans_le_san = fqdn in san
        etat = "OK" if dans_le_san and code.startswith(("2", "3")) else "ECART"
        print(f"  {fqdn:<34} {adresse:<14} cert={'oui' if dans_le_san else 'NON'} "
              f"http={code}  {etat}")
        if not dans_le_san:
            ecarts.append(f"{fqdn} : le certificat SERVI ne porte pas ce nom — "
                          f"rejouer `client_pki` sur le porteur ; `serveur_nginx` "
                          f"seul ne reemet pas le SAN")
        elif not code.startswith(("2", "3")):
            ecarts.append(f"{fqdn} : le certificat porte le nom mais le vhost rend "
                          f"{code} — le nom est authentifie, rien ne le sert")

    # LE SENS INVERSE : un nom reste dans le SAN apres avoir quitte le plan.
    declares = set(porteurs)
    for adresse, san in sorted(san_par_adresse.items()):
        propre = noms_propres.get(adresse, "")
        for nom in san:
            if nom in declares:
                continue
            # LE PORTEUR S'AUTHENTIFIE SOUS SON PROPRE NOM, et ce n'est pas un reste :
            # `client_pki` met toujours le FQDN et le nom court de la machine dans le SAN.
            # Les confondre avec un vestige de renommage ferait crier ce controle a chaque
            # execution — un controle qui crie toujours ne se lit plus.
            if propre and nom.split(".")[0] == propre:
                continue
            ecarts.append(f"{nom} : porte par le certificat de {adresse} alors "
                          f"qu'aucune application ne l'expose — reste d'un "
                          f"renommage, encore authentifie")

    print()
    if ecarts:
        print(f"ECART — {len(ecarts)} :")
        for e in ecarts:
            print(f"  - {e}")
        return 1
    print("Conforme : chaque exposition declaree est servie, sous un certificat qui la "
          "porte, et le certificat ne porte rien d'autre.")
    return 0


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