#!/usr/bin/env python3 """Les intrants qu'un SITE expose a ses locataires — derives, jamais ecrits. UN SITE EST UNE INTERFACE, PAS UN DECOR. Un locataire qui veut l'habiter a besoin d'un petit nombre de valeurs : ou resoudre, ou prendre ses paquets, ou cloner le genome, ou deposer son etat, quel plan d'administration laisser entrer. Aujourd'hui il les ECRIT A LA MAIN dans ses intrants, et le depot s'en remet a une garde pour verifier qu'elles n'ont pas vieilli. CE QUE CA COUTE, MESURE DEUX FOIS EN DEUX JOURS : - `serveur_ops_forge_amont: https://10.0.33.11` chez Chezlepro — la forge avait demenage en 10.37.33.11 avec le renumerotage du site. Le commentaire juste au-dessus expliquait encore pourquoi l'ancienne adresse figurait dans les SAN du certificat : le raisonnement etait intact, la valeur non. - `OPS-Technolibre/underlay.yml` decrivait un site d'index 23 quand le vrai disait 31. Les deux avaient la meme forme : une copie qui a cesse de designer ce qu'elle nommait, et que rien ne relisait. Une valeur DERIVEE ne peut pas avoir ce defaut. CE SCRIPT NE MODIFIE RIEN. Il lit le site monte (`underlay.yml` et son `plan/`) et rend le contrat. `--verifier` le confronte a ce que le locataire monte declare, et c'est la que le retard se voit. Usage : python3 scripts/site_intrants.py # le contrat, lisible python3 scripts/site_intrants.py --json # le meme, pour l'API et les gabarits python3 scripts/site_intrants.py --verifier # garde : le locataire suit-il son site ? """ from __future__ import annotations import argparse import json import sys from pathlib import Path RACINE = Path(__file__).resolve().parent.parent sys.path.insert(0, str(RACINE / "scripts")) import underlay as underlay_mod # noqa: E402 from inventory_rules import dossier_inventaire # noqa: E402 import yaml # noqa: E402 PORT_CACHE = 3142 PREFIXE_BINAIRES = "setops-binaires" def _une(groupe: str) -> str: """L'adresse du site qui porte ce groupe, ou '' s'il n'y en a pas. PLUSIEURS ADRESSES = AUCUNE REPONSE. Un service porte par deux machines demande un choix (lequel amorce ?) que ce script n'a pas a faire silencieusement. """ a = underlay_mod.adresses_site_portant(groupe) return a[0] if len(a) == 1 else "" def _nom_expose(groupe: str, domaine: str) -> str: """Le nom de SERVICE que le site publie pour ce groupe, si son plan en declare un.""" apps = underlay_mod.lire_plan_site("applications.yml").get("applications") or {} for a in apps.values(): if a.get("groupe") == groupe: for e in (a.get("expose") or []): return str(e) hote = next((m["nom"] for m in underlay_mod.machines_site() if groupe in (underlay_mod.services_site().get(m["nom"]) or [])), "") return f"{hote}.{domaine}" if hote and domaine else "" def _ip_publique_site() -> str: """L'adresse publique de la frontiere du site, lue dans sa carte (`opnsense.yml`).""" try: chemin = underlay_mod.chemin() f = chemin.resolve().parent / "opnsense.yml" if chemin else None if f and f.is_file(): return str((yaml.safe_load(f.read_text(encoding="utf-8")) or {}) .get("opnsense_wan_ip") or "") except Exception: pass return "" def contrat() -> dict: u = underlay_mod.charger() if not u: return {} intrants = underlay_mod.lire_plan_site("10-intrants.yml") domaine = str(intrants.get("domaine_interne") or "") gestion = next((r.get("sous_reseau") for r in underlay_mod.reseaux(u) if r.get("nom") == "management"), "") transit = underlay_mod.reseau_transit(u) or {} cache = _une("serveur_artefacts") forge = _une("serveur_forgejo") return { # QUI EST CE SITE "domaine_site": domaine, "index_site": underlay_mod.index_site(u), # OU RESOUDRE, AVANT QUE LE LOCATAIRE AIT SON PROPRE RESOLVEUR "dns_amorcage": _une("serveur_resolveur"), # OU PRENDRE SES PAQUETS, ET SES BINAIRES DE PUBLICATION "artefacts_amorcage": f"{cache}:{PORT_CACHE}" if cache else "", "setops_depot_binaires": (f"http://{cache}:{PORT_CACHE}/{PREFIXE_BINAIRES}" if cache else ""), # OU CLONER LE GENOME, ET AVEC QUELLE RACINE LUI FAIRE CONFIANCE "serveur_ops_forge_amont": f"https://{forge}" if forge else "", "forge_amont_ac": str((underlay_mod.chemin() or Path()).resolve().parent / "ac-racine-site.crt"), # OU DEPOSER SON ETAT — par NOM : le depot sert plusieurs locataires et peut bouger "client_backup_cible": _nom_expose("serveur_backup_site", domaine), # QUI TIRE SES ZONES PUBLIQUES — le seul serveur a qui le primaire notifie, et le # seul que son pare-feu laisse frapper. Une ADRESSE : c'est un pair de flux. "dns_public_site": _une("serveur_dns_public"), # VERS OU POINTENT SES NOMS PUBLICS — l'adresse publique de la frontiere. Un fait # du site : le locataire la recoit, il ne la recopie pas. "ip_publique_site": _ip_publique_site(), # QUI A LE DROIT D'ENTRER EN SSH "nftables_admin_ssh": [gestion] if gestion else [], # PAR OU SORTIR, ET A QUI DEMANDER L'HEURE "passerelle_sortie": str(transit.get("passerelle_sortie") or ""), } # Ce que le locataire monte declare, et sous quel nom, pour chaque clef du contrat. # Vide = le locataire ne declare rien de tel, et c'est legitime (il derive deja). OU_LE_LOCATAIRE_LE_DIT = { "dns_amorcage": ("all/10-intrants.yml", "dns_amorcage"), "artefacts_amorcage": ("all/10-intrants.yml", "artefacts_amorcage"), "setops_depot_binaires": ("all/10-intrants.yml", "setops_depot_binaires"), "serveur_ops_forge_amont": ("serveur_ops.yml", "serveur_ops_forge_amont"), "client_backup_cible": ("client_backup.yml", "client_backup_cible"), "dns_public_site": ("all/10-intrants.yml", "dns_public_site"), "ip_publique_site": ("all/10-intrants.yml", "ip_publique_site"), # CES DEUX-LA MANQUAIENT, ET LE SILENCE A COUTE (2026-09-13). Le site en expose SEPT ; # cette table n'en comparait que CINQ. Le controle rapportait donc « CONFORME » avec # un aplomb complet, sur un locataire dont le plan d'administration pointait encore # les bouts de WAN d'un AUTRE site. # # CE QUE CA A COUTE : quatorze machines materialisees, le runner insemine — et # l'exploitant incapable d'entrer sur son propre runner pour l'armer. `Connection # timed out`, sans qu'aucune garde n'ait rien eu a dire. # # Une liste qui en suit une autre prend du retard : la regle est connue ici, et c'est # sa cinquieme occurrence. La garde s'ecrit EN MEME TEMPS que la seconde liste. "nftables_admin_ssh": ("all/10-intrants.yml", "nftables_admin_ssh"), "passerelle_sortie": ("all/10-intrants.yml", "passerelle_sortie"), } def _normaliser(v) -> str: """Une forme comparable, que la valeur soit une chaine ou une liste. `nftables_admin_ssh` est une LISTE de CIDR ; les cinq intrants d'origine etaient tous des chaines, et `str()` suffisait. Comparer `str(['a','b'])` a `str(['b','a'])` ferait crier une difference qui n'en est pas une — l'ordre d'une liste de sources ne porte aucun sens ici. """ if isinstance(v, (list, tuple)): return " ".join(sorted(str(x).strip().rstrip("/") for x in v)) return str(v).strip().rstrip("/") def _declare_par_le_locataire() -> dict[str, str]: """Lit, sans Ansible, ce que le locataire monte ecrit a propos de son site.""" # LA RESOLUTION D'INSTANCE EST PARTAGEE, PAS RECOPIEE (P41). La premiere version # ecrivait `RACINE / "instance" / "inventories" / "principal"` — le symlink et le nom # du dossier en dur. Neuf modules avaient deja porte chacun leur copie, et cinq # defauts en etaient sortis en cinq jours. La preuve a mordu sur la dixieme. base = dossier_inventaire() / "group_vars" vu: dict[str, str] = {} for clef, (fichier, variable) in OU_LE_LOCATAIRE_LE_DIT.items(): f = base / fichier if not f.is_file(): continue d = yaml.safe_load(f.read_text(encoding="utf-8")) or {} if variable in d: vu[clef] = d[variable] return vu def verifier() -> int: c = contrat() if not c: print("Aucun site monte : rien a exposer.") return 0 vu = _declare_par_le_locataire() ecarts, absents = [], [] for clef, attendu in c.items(): if clef not in OU_LE_LOCATAIRE_LE_DIT or not attendu: continue fichier, variable = OU_LE_LOCATAIRE_LE_DIT[clef] if clef not in vu: absents.append(f"{variable} ({fichier})") continue # `setops_depot_binaires` se DERIVE legitimement d'`artefacts_amorcage` en Jinja : # une expression n'est pas un retard, c'est le contraire. if "{{" in str(vu[clef]): continue if _normaliser(vu[clef]) != _normaliser(attendu): ecarts.append(f"{variable} : le locataire dit « {vu[clef]} », " f"le site expose « {attendu} »") if ecarts: print("NON CONFORME : le locataire ne suit plus son site.") for e in ecarts: print(" -", e) return 1 note = f" ({len(absents)} non declare(s), donc derive(s) ou non utilise(s))" if absents else "" print(f"CONFORME : {len(vu)} intrant(s) du locataire concordent avec ce que le site expose{note}.") return 0 def main(argv: list[str]) -> int: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--json", action="store_true") ap.add_argument("--verifier", action="store_true") args = ap.parse_args(argv) if args.verifier: return verifier() c = contrat() if not c: print("Aucun site monte (`underlay.yml` absent ou brise).") return 0 if args.json: print(json.dumps(c, indent=2, ensure_ascii=False)) return 0 print(f"# Intrants exposes par le site « {c['domaine_site'] or '?'} » (index {c['index_site']})") print("#") print("# A recopier dans les intrants d'un locataire qui veut l'habiter — ou mieux :") print("# a comparer aux siens par `--verifier`, qui dit lesquels ont pris du retard.") for clef, valeur in c.items(): if clef in ("domaine_site", "index_site", "forge_amont_ac"): continue v = json.dumps(valeur, ensure_ascii=False) if isinstance(valeur, list) else valeur print(f"{clef}: {v}" if valeur else f"# {clef}: (le site ne porte pas ce service)") print(f"#\n# racine de confiance de la forge : {c['forge_amont_ac']}") return 0 if __name__ == "__main__": raise SystemExit(main(sys.argv[1:]))