#!/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, instance_courante # 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 _opnsense() -> dict: """La carte de la frontiere du site (`opnsense.yml`), ou {}.""" try: chemin = underlay_mod.chemin() f = chemin.resolve().parent / "opnsense.yml" if chemin else None if f and f.is_file(): return yaml.safe_load(f.read_text(encoding="utf-8")) or {} except Exception: pass return {} def _ip_publique_site() -> str: """L'adresse publique de la frontiere du site, lue dans sa carte (`opnsense.yml`).""" return str(_opnsense().get("opnsense_wan_ip") or "") LIBRE = "libre" def pool_ips_publiques(carte: dict | None = None) -> dict[str, str]: """{adresse : locataire | 'libre'} — le pool que le site attribue (`opnsense.yml`).""" carte = _opnsense() if carte is None else carte return {str(k): str(v) for k, v in (carte.get("opnsense_ips_publiques") or {}).items()} def valider_pool(pool: dict[str, str], wan: str, locataires: list[str] | None) -> list[str]: """Les fautes du pool : adresse invalide, adresse du site, locataire inconnu ou servi deux fois. Une adresse attribuee a deux locataires est impossible par construction (la cle est l'adresse) ; un locataire qui en recoit deux, lui, ne l'est pas.""" import ipaddress fautes: list[str] = [] vus: dict[str, str] = {} for ip, qui in sorted(pool.items()): try: a = ipaddress.ip_address(ip) except ValueError: fautes.append(f"`{ip}` n'est pas une adresse IP") continue if a.is_private or a.is_loopback: fautes.append(f"`{ip}` n'est pas une adresse publique") if ip == wan: fautes.append(f"`{ip}` est l'adresse du site (opnsense_wan_ip) : elle n'est pas a attribuer") if qui == LIBRE: continue if locataires is not None and qui not in locataires: fautes.append(f"`{ip}` attribuee a « {qui} », que le site n'heberge pas " f"(underlay : {', '.join(locataires) or 'aucun'})") if qui in vus: fautes.append(f"« {qui} » recoit deux adresses ({vus[qui]}, {ip}) : une par locataire") vus[qui] = ip return fautes def ip_publique_de(locataire: str, pool: dict[str, str] | None = None) -> str: """L'adresse publique que le site attribue a ce locataire, ou ''.""" pool = pool_ips_publiques() if pool is None else pool return next((ip for ip, qui in sorted(pool.items()) if qui == locataire), "") 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(), # L'ADRESSE PUBLIQUE DE CE LOCATAIRE — tiree du pool du site, attribuee par lui # (`opnsense_ips_publiques`). Celle de son web frontal et de son courriel. "ip_publique": ip_publique_de(instance_courante().name), # SOUS QUEL NOM LE SITE FAIT AUTORITE — les NS et le SOA de chaque zone publique. "dns_public_nom": str(intrants.get("dns_public_nom") or ""), # QUI A LE DROIT D'ENTRER EN SSH # QUI A LE DROIT D'ENTRER EN SSH : le plan d'administration du site ET le tunnel # WireGuard des admins, derive de sa declaration (jamais recopie). "nftables_admin_ssh": ([gestion] if gestion else []) + ([str((intrants.get("acces_admin_vpn") or {}).get("reseau"))] if (intrants.get("acces_admin_vpn") or {}).get("reseau") 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"), "ip_publique": ("all/10-intrants.yml", "ip_publique"), "dns_public_nom": ("all/10-intrants.yml", "dns_public_nom"), # 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 = [], [] # LE POOL D'ABORD : une attribution fautive rendrait fausse la valeur meme a comparer. ecarts += [f"pool d'adresses publiques du site : {f}" for f in valider_pool( pool_ips_publiques(), _ip_publique_site(), underlay_mod.tenants_du_site(underlay_mod.charger()))] 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:]))