Set-OPS-Public/scripts/forge_etancheite.py

#!/usr/bin/env python3
"""La forge du site laisse-t-elle chaque runner lire ses depots, et personne d'autre ?

POURQUOI (2026-10-11, `docs/depots-prives-conception.md`). Le plan d'un ecosysteme et la carte
du site sont prives sur la forge du site ; chaque runner de locataire lit le sien avec une cle
de deploiement en lecture seule. Deux mesures, depuis deux endroits :

  SANS IDENTIFIANT, depuis ce poste (HTTPS) :
    - un depot PRIVE doit EXIGER une authentification. Un refus RESEAU ne compte pas : il ne
      dit rien de la forge, seulement du chemin (une sonde rapporte son errno) ;
    - un depot PUBLIC (le moteur, les modeles) doit repondre : un ecosysteme neuf le clone
      sans rien detenir.
  DEPUIS CHAQUE RUNNER DE LOCATAIRE (SSH, sa cle, la cle d'hote epinglee) :
    - il lit SON depot ;
    - la forge lui REFUSE les autres depots prives (« not authorized »).

Le resultat se lit ligne par ligne ; le code de sortie vaut 1 au moindre ecart.

Usage :
  python3 scripts/forge_etancheite.py           (make forge-etancheite)
"""
from __future__ import annotations

import os
import subprocess
import sys
from pathlib import Path

RACINE = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(RACINE / "scripts"))
ROLES_PRIVES = {"instance", "tenant", "hebergeur"}


def classer_https(rc: int, sortie: str) -> str:
    """'lu', 'authentification', ou 'reseau' — ce que dit un `git ls-remote` anonyme."""
    if rc == 0:
        return "lu"
    s = sortie.lower()
    if any(m in s for m in ("could not read username", "terminal prompts disabled",
                            "authentication failed", "401", "403")):
        return "authentification"
    return "reseau"


def classer_ssh(rc: int, sortie: str) -> str:
    """'lu', 'refuse' (la forge a juge), ou 'reseau' (personne n'a juge)."""
    if rc == 0:
        return "lu"
    s = sortie.lower()
    if "not authorized" in s or "permission denied" in s or "does not exist" in s:
        return "refuse"
    return "reseau"


def _site_et_depots():
    import underlay as underlay_mod
    plan = underlay_mod.lire_plan_site("serveurs.yml") or {}
    depots = []
    for srv in (plan.get("serveurs") or {}).values():
        d = (srv.get("variables") or {}).get("serveur_ops_depots")
        if d:
            depots = list(d)
    dom = str((underlay_mod.lire_plan_site("10-intrants.yml") or {}).get("domaine_interne") or "")
    racine_ac = (underlay_mod.chemin() or Path()).resolve().parent / "ac-racine-site.crt"
    return dom, depots, racine_ac


def mesurer_https(dom: str, depots: list[dict], racine_ac: Path) -> list[str]:
    ecarts = []
    env = dict(os.environ, GIT_TERMINAL_PROMPT="0", GIT_ASKPASS="/bin/false")
    if racine_ac.is_file():
        env["GIT_SSL_CAINFO"] = str(racine_ac)
    print(f"Sans identifiant, depuis ce poste (https://forge.{dom}) :")
    for d in depots:
        nom, prive = str(d["depot"]), str(d.get("role")) in ROLES_PRIVES
        r = subprocess.run(["git", "ls-remote", f"https://forge.{dom}/genome/{nom}.git", "HEAD"],
                           capture_output=True, text=True, timeout=30, env=env)
        verdict = classer_https(r.returncode, r.stdout + r.stderr)
        attendu = "authentification" if prive else "lu"
        ok = verdict == attendu
        print(f"  {'OK   ' if ok else 'ECART'} {nom:<18} {'prive ' if prive else 'public'} -> {verdict}"
              + ("" if ok else f"   (attendu : {attendu})"))
        if not ok:
            ecarts.append(f"{nom} : {verdict}, attendu {attendu}")
    return ecarts


def mesurer_runners(depots: list[dict]) -> list[str]:
    import contexte as _ctx
    import yaml
    prives = [str(d["depot"]) for d in depots if str(d.get("role")) in ROLES_PRIVES]
    ecarts = []
    for site, loc in _ctx.couples_hebergeur():
        face = loc.face_publiee() or {}
        if face.get("hebergeur") != site.nom:
            continue
        sien = str((face.get("genome") or {}).get("depot") or "")
        runners = sorted((face.get("groupes") or {}).get("serveur_ops_tenant") or [])
        gv = yaml.safe_load((loc.dossier_inventaire() / "group_vars" / "serveur_ops.yml")
                            .read_text(encoding="utf-8")) or {}
        dest = next((str(x.get("dest")) for x in gv.get("serveur_ops_depots") or []
                     if x.get("role") == "instance"), "")
        if not (sien and runners and dest):
            ecarts.append(f"{loc.nom} : face sans depot, sans runner, ou plan sans `role: instance`")
            continue
        print(f"\nDepuis le runner de {loc.nom} ({runners[0]}), par l'URL de son origin :")
        # LA COMMANDE SSH DU DEPOT LUI-MEME (`core.sshCommand`, posee par serveur_ops) : on
        # mesure ce que le runner fait vraiment, pas une commande refaite a cote.
        script = (f"cd /opt/setops/{dest} && base=$(git remote get-url origin) && base=${{base%/*}} && "
                  f"for n in {' '.join(prives)}; do "
                  "if out=$(git ls-remote \"$base/$n.git\" HEAD 2>&1 >/dev/null); then echo \"MESURE $n lu\"; "
                  "else echo \"MESURE $n $(echo $out | tr '\\n' ' ')\"; fi; done")
        r = subprocess.run(["ansible", "-i", str(loc.inventaire()), runners[0], "-b",
                            "--become-user", "setops", "-m", "shell", "-a", script],
                           capture_output=True, text=True, timeout=300, stdin=subprocess.DEVNULL)
        mesures = {}
        for ligne in r.stdout.splitlines():
            if ligne.startswith("MESURE "):
                _m, n, reste = ligne.split(" ", 2)
                mesures[n] = ("lu", "") if reste.strip() == "lu" else (classer_ssh(1, reste), reste.strip())
        if not mesures:
            ecarts.append(f"{loc.nom} : aucune mesure rendue par {runners[0]} ({(r.stderr or r.stdout)[-160:]})")
            print(f"  ECART aucune mesure rendue")
            continue
        for n in prives:
            verdict, detail = mesures.get(n, ("absent", ""))
            attendu = "lu" if n == sien else "refuse"
            ok = verdict == attendu
            print(f"  {'OK   ' if ok else 'ECART'} {n:<18} -> {verdict}"
                  + ("" if ok else f"   (attendu : {attendu}) {detail[:90]}"))
            if not ok:
                ecarts.append(f"{loc.nom} -> {n} : {verdict}, attendu {attendu}")
    return ecarts


def main() -> int:
    dom, depots, racine_ac = _site_et_depots()
    if not dom or not depots:
        print("REFUS : aucun site monte, ou son plan ne declare pas `serveur_ops_depots`.")
        return 2
    ecarts = mesurer_https(dom, depots, racine_ac) + mesurer_runners(depots)
    if ecarts:
        print(f"\nNON ETANCHE : {len(ecarts)} ecart(s).")
        return 1
    print("\nETANCHE : chaque runner lit son depot, aucun celui d'un autre ; le public reste public.")
    return 0


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