#!/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())