#!/usr/bin/env python3 """Interface web locale pour gerer un inventaire Set-OPS.""" from __future__ import annotations from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import argparse import json import os import re import secrets import socket import subprocess import tempfile import time import threading from concurrent.futures import ThreadPoolExecutor from pathlib import Path from urllib.parse import urlparse import yaml import devis_reseau from inventory_rules import instance_courante # noqa: E402 from inventory_rules import ( ecriture_atomique, GROUPE_HOTES_ACTIFS, GROUPE_HOTES_PLANIFIES, GROUPES_ETAT_HOTE, chaine_connexion, charger_applications, charger_bases_donnees, charger_dependances, charger_domaines, charger_nomenclature, charger_serveurs, est_groupe_operationnel, groupes_operationnels_connus, integrations_universelles, reconcilier_serveur, valider_applications, catalogue_liens, valider_bases, valider_domaines, valider_serveurs, ) RACINE = Path(__file__).resolve().parents[1] INSTANCE = instance_courante() def _inventaire(instance: Path, *noms: str) -> Path: """Inventaire de l'instance, détection rétro-compatible (premier existant).""" for nom in noms: p = instance / "inventories" / nom / "hosts.yml" if p.exists(): return p # Aucun hosts.yml encore généré : viser le répertoire d'inventaire déjà présent # (ses group_vars y sont), sinon le premier nom par défaut. for nom in noms: if (instance / "inventories" / nom).is_dir(): return instance / "inventories" / nom / "hosts.yml" return instance / "inventories" / noms[0] / "hosts.yml" # Un seul inventaire par instance (modèle « séparation par instance ») ; on accepte # l'ancien découpage lab/production en rétro-compatibilité. SETOPS_INVENTAIRE force. _force = os.environ.get("SETOPS_INVENTAIRE") INVENTAIRE_DEFAUT = Path(_force) if _force else _inventaire(INSTANCE, "principal", "production") INVENTAIRE_PRODUCTION = INVENTAIRE_DEFAUT.resolve() # Inventaire qui porte la config Proxmox / le clonage (modèle) : lab > principal > production. INVENTAIRE_MODELE = _inventaire(INSTANCE, "lab", "principal", "production") DOSSIER_PLAYBOOKS_GROUPES = RACINE / "playbooks/groupes" FICHIER_DEPENDANCES = RACINE / "docs/dependances-groupes.yml" FICHIER_NOMENCLATURE = INSTANCE / "plan/nomenclature.yml" FICHIER_BASES = INSTANCE / "plan/bases-donnees.yml" FICHIER_APPLICATIONS = INSTANCE / "plan/applications.yml" FICHIER_DOMAINES = INSTANCE / "plan/domaines.yml" FICHIER_SERVEURS = INSTANCE / "plan/serveurs.yml" # Intrants de base (panneau GUI). Cibles d'ecriture possedees par le GUI. # Identite : group_vars/all/10-intrants.yml de l'inventaire. Selon l'instance c'est # un fichier reel (modele « par instance ») ou un symlink vers une source partagee # (ecrire suit le symlink). Proxmox : group_vars de l'inventaire modele. Les SECRETS # n'y entrent jamais. INTRANTS_IDENTITE = INVENTAIRE_DEFAUT.parent / "group_vars/all/10-intrants.yml" INTRANTS_PROXMOX = INVENTAIRE_MODELE.parent / "group_vars/proxmox.yml" # Le seed reseau (`index`) vit dans la nomenclature (le plan reseau) — d'ou tout # l'adressage se derive. Le panneau Intrants le lit/ecrit la, chirurgicalement, sans # reformater le fichier (cf. _ecrire_index_nomenclature). # Frontiere nord/sud (OPNsense) : parametres NON sensibles du pare-feu de bordure. # La cle/secret d'API n'y entrent JAMAIS — voute uniquement (cf. INTRANTS_CLES_INTERDITES). # Fabric physique : `underlay.yml`, monte par symlink a la racine du moteur depuis le # depot de l'HEBERGEUR (ses switches, ses cables). Distinct de l'instance : il ne suit # pas `make instance-utiliser`. Imbrique sous `underlay:` et riche en commentaires : # on l'ecrit chirurgicalement, jamais par un safe_dump qui les effacerait. FICHIER_UNDERLAY = Path(os.environ.get("SETOPS_UNDERLAY") or (RACINE / "underlay.yml")) def _depot_hebergeur() -> Path | None: """Depot de l'HEBERGEUR, derive du symlink `underlay.yml` qui le designe deja. Un hebergeur sert PLUSIEURS tenants et n'a qu'une frontiere : ses intrants ne peuvent pas vivre chez le tenant actif, sinon basculer sur un invite les ferait disparaitre du panneau. None si aucun underlay n'est monte. """ if not FICHIER_UNDERLAY.exists(): return None return FICHIER_UNDERLAY.resolve().parent def _fichier_frontiere() -> Path: """Les intrants de la frontiere — a la RACINE du depot de site depuis le 2026-08-22. La frontiere est un objet du MONDE PHYSIQUE : ses reglages vivent avec `underlay.yml` et `proxmox-hebergeur.yml`, pas dans les group_vars d'un tenant. Range chez un locataire, ce fichier a garde une adresse d'API perimee sans que personne la voie — et l'applicateur restait suspendu dessus, sans message. Les anciens emplacements restent lus, pour un site pas encore migre. """ base = _depot_hebergeur() if base: racine = base / "opnsense.yml" if racine.is_file(): return racine for nom in ("principal", "production", "lab"): p = base / "inventories" / nom / "group_vars" / "opnsense.yml" if p.is_file(): return p return racine # a creer la, pas ailleurs return INVENTAIRE_DEFAUT.parent / "group_vars/opnsense.yml" def _fichier_proxmox_hebergeur() -> Path: """proxmox-hebergeur.yml de l'HEBERGEUR ; repli sur le proxmox.yml du tenant. Meme raisonnement que la frontiere et l'underlay (D-14) : un cluster appartient a l'hebergeur, pas au tenant qui y loge. Recopie chez chaque tenant, l'inventaire du cluster avait deja diverge — deux listes de stockages contradictoires pour le meme materiel. Le repli garde un depot sans underlay monte parfaitement utilisable. """ base = _depot_hebergeur() if base: return base / "proxmox-hebergeur.yml" return INTRANTS_PROXMOX INTRANTS_FRONTIERE = _fichier_frontiere() INTRANTS_PROXMOX_HEBERGEUR = _fichier_proxmox_hebergeur() FICHIERS_INTRANTS = {"identite": INTRANTS_IDENTITE, "proxmox": INTRANTS_PROXMOX, "proxmox_hebergeur": INTRANTS_PROXMOX_HEBERGEUR, "reseau": FICHIER_NOMENCLATURE, "frontiere": INTRANTS_FRONTIERE, "fabric": FICHIER_UNDERLAY} # Intrants de fabric -> chemin dans underlay.yml. Seules les valeurs PLATES sont ici : # `reseaux` et `hotes` sont des listes de tables, hors de portee du panneau (cf. la vue # dediee, a venir). Un intrant absent d'ici s'edite encore dans le YAML. CHEMINS_UNDERLAY = { "underlay_routeur": ("routeur",), "underlay_dialecte": ("dialecte",), "underlay_stp_mode": ("stp", "mode"), "underlay_stp_topologie": ("stp", "topologie"), } # (variable, fichier, classe, section, libelle, type, proprietaire) # # `proprietaire` repond a une question que le panneau posait sans y repondre : en # ouvrant l'ecran on trouvait, cote a cote et sans distinction, des valeurs du TENANT # (son domaine, son realm, son modele) et des valeurs de l'HEBERGEUR (son cluster, sa # frontiere, sa fabric). Ce sont deux proprietaires, deux depots, deux cycles de vie. INTRANTS_SCHEMA = [ ("index", "reseau", "constante", "Réseau", "Index d'instance dans la fédération (seed : dérive supernet 10., VLAN 1000+index×10+zone, VMID)", "int", "tenant"), ("domaine_interne", "identite", "constante", "Identité", "Domaine DNS interne", "str", "tenant"), ("fuseau_horaire", "identite", "defaut", "Identité", "Fuseau horaire", "str", "tenant"), ("organisation", "identite", "defaut", "Identité", "Organisation (annuaire LDAP, certificats)", "str", "tenant"), ("identite_realm", "identite", "defaut", "Identité", "Realm SSO (Keycloak)", "str", "tenant"), ("nftables_admin_ssh", "identite", "catalogue", "Sécurité", "Sources SSH d'administration toujours autorisées (garde anti-lockout du pare-feu)", "liste", "tenant"), ("proxmox_api_host", "proxmox_hebergeur", "constante", "Cluster", "Hôte API Proxmox", "str", "hebergeur"), ("proxmox_api_user", "proxmox_hebergeur", "constante", "Cluster", "Utilisateur API", "str", "hebergeur"), ("proxmox_api_port", "proxmox_hebergeur", "constante", "Cluster", "Port API", "str", "hebergeur"), ("proxmox_validate_certs", "proxmox_hebergeur", "constante", "Cluster", "Valider les certificats TLS", "bool", "hebergeur"), ("proxmox_clone_source_nom", "proxmox", "constante", "Proxmox", "Nom du golden template", "str", "tenant"), ("proxmox_clone_vmid_modele", "proxmox", "constante", "Proxmox", "VMID du golden template", "int", "tenant"), ("proxmox_clone_noeud", "proxmox", "defaut", "Proxmox", "Nœud Proxmox (défaut)", "str", "tenant"), ("proxmox_clone_stockage", "proxmox", "defaut", "Proxmox", "Stockage (défaut)", "str", "tenant"), # REPLI, pas un reglage de la flotte. Chaque VM du plan recoit le VNet DERIVE de sa # zone (`instancier` pose `proxmox_pont` par hote) ; cette valeur n'est consultee que # si l'hote n'en porte pas — c'est-a-dire jamais dans le flux normal. L'intitule le # dit, sans quoi on la corrige, on redeploie, et rien ne change : `vmbr3` y est reste # faux dix jours sans que rien ne bronche (2026-08-13). ("proxmox_clone_pont", "proxmox", "defaut", "Proxmox", "Pont réseau — clones manuels seulement (les VM du plan reçoivent le VNet de leur zone)", "str", "tenant"), ("proxmox_noeuds", "proxmox_hebergeur", "catalogue", "Cluster", "Nœuds disponibles (liste)", "liste", "hebergeur"), ("proxmox_stockages", "proxmox_hebergeur", "catalogue", "Cluster", "Stockages disponibles (liste)", "liste", "hebergeur"), ("proxmox_ponts", "proxmox_hebergeur", "catalogue", "Cluster", "Ponts réseau disponibles (liste)", "liste", "hebergeur"), ("opnsense_api_url", "frontiere", "constante", "Frontière", "URL de gestion de la frontière OPNsense (API REST)", "str", "hebergeur"), ("opnsense_api_verifier_certs", "frontiere", "defaut", "Frontière", "Valider le certificat TLS de la frontière", "bool", "hebergeur"), ("opnsense_wan_ip", "frontiere", "constante", "Frontière", "Adresse publique du WAN de la frontière (face Internet de l'écosystème)", "str", "hebergeur"), ("opnsense_if_wan", "frontiere", "defaut", "Frontière", "Interface WAN — IDENTIFIANT OPNsense (wan), pas le libellé ni le périphérique", "str", "hebergeur"), ("opnsense_if_transit", "frontiere", "defaut", "Frontière", "Interface de transit — IDENTIFIANT OPNsense (opt1, opt2…), pas « igb1 » ni le libellé affiché", "str", "hebergeur"), ("opnsense_if_gestion", "frontiere", "defaut", "Frontière", "Interface où arrivent les plans d'administration attachés à la frontière — IDENTIFIANT OPNsense (lan). Une règle SSH posée ailleurs ne peut jamais correspondre", "str", "hebergeur"), ("underlay_routeur", "fabric", "constante", "Fabric", "Switch qui porte le routage (SVI, ACL, routes) — les autres restent en L2 pur", "str", "hebergeur"), ("underlay_dialecte", "fabric", "constante", "Fabric", "Dialecte de CLI du commutateur (cisco | binardat)", "str", "hebergeur"), ("underlay_stp_mode", "fabric", "defaut", "Fabric", "Mode de spanning-tree (rstp | mstp | pvst)", "str", "hebergeur"), ("underlay_stp_topologie", "fabric", "defaut", "Fabric", "Topologie de câblage (etoile | anneau | maille)", "str", "hebergeur"), # `opnsense_prochain_saut` n'est PLUS un intrant : il derive du reseau de transit de # l'underlay (`passerelle` du reseau portant `passerelle_sortie`). Le saisir ici en # doublon ouvrirait la porte a deux valeurs contradictoires pour un seul lien. ] # Cles secretes : interdites a l'ecriture par le GUI (garde-fou). INTRANTS_CLES_INTERDITES = { "proxmox_api_token_id", "proxmox_api_token_secret", "ansible_become_pass", "ansible_password", # Frontiere : les identifiants d'API ne transitent jamais par le panneau. "opnsense_api_key", "opnsense_api_secret", } # Rappel en lecture seule (jamais de valeur) : secrets a fournir via Ansible Vault. def secrets_attendus() -> list[str]: """Noms des secrets a fournir par la voute — DERIVES, jamais tenus a la main. `voute.secrets_exiges()` les recense depuis le plan, les roles des groupes actifs et les group_vars de l'instance ; `voute.SECRETS_HORS_MOTIF` ajoute ceux qui ne portent pas le prefixe `vault_` (les jetons Proxmox). C'est la MEME source que la preuve P18. Une liste ecrite a la main vivait ici et avait diverge : elle annoncait `vault_ldap_sssd`, qu'aucun role ne consomme. Un operateur qui suivait le panneau creait donc un secret inutile. Le recensement par motif textuel a lui aussi sa limite : il a un temps reclame `vault_step_ca_fingerprint` a cause d'un defaut mort dans `client_pki`, qui derive en realite l'empreinte a chaud depuis l'AC. Depot public nu (aucune instance) : liste vide plutot qu'une erreur d'affichage. """ try: import voute return sorted(set(voute.secrets_exiges()) | voute.SECRETS_HORS_MOTIF) except Exception: return [] # Garde-fous des executions (verifier / deployer) depuis l'interface. JETON = secrets.token_urlsafe(18) VERROU = threading.Lock() VERIF_OK: dict[str, bool] = {} MOTIF_HOTE = re.compile(r"^[A-Za-z0-9._-]{1,63}$") # Champs de provisioning persistes comme variables d'hote dans l'inventaire. # (cle_gui, variable_inventaire, type) CHAMPS_PROVISION = [ ("adresse_ip", "ansible_host", "str"), ("utilisateur_ansible", "ansible_user", "str"), ("cidr", "proxmox_cidr", "int"), ("passerelle", "proxmox_passerelle", "str"), ("vlan", "proxmox_vlan", "int"), ("pont", "proxmox_pont", "str"), ("dns", "proxmox_dns", "str"), ("vmid", "proxmox_vmid", "int"), ("noeud", "proxmox_noeud", "str"), ("stockage", "proxmox_stockage", "str"), ("disque_taille", "proxmox_disque_taille", "str"), ("memoire", "proxmox_memoire", "int"), ("coeurs", "proxmox_coeurs", "int"), ] def charger_yaml(path: Path) -> dict: if not path.exists(): return {"all": {"children": {}}} with path.open("r", encoding="utf-8") as fichier: data = yaml.safe_load(fichier) or {} if not isinstance(data, dict): raise ValueError(f"{path} ne contient pas une table YAML.") data.setdefault("all", {}) data["all"].setdefault("children", {}) return data def ecrire_yaml(path: Path, data: dict) -> None: with ecriture_atomique(path) as fichier: yaml.safe_dump(data, fichier, default_flow_style=False, sort_keys=False, allow_unicode=True) def enfants(data: dict) -> dict: return data.setdefault("all", {}).setdefault("children", {}) def groupes_disponibles(data: dict) -> list[str]: return groupes_operationnels_connus(enfants(data), DOSSIER_PLAYBOOKS_GROUPES) def roles_des_groupes() -> dict: """Pour chaque playbook de groupe, extrait les roles appliques (chaine groupe -> roles).""" resultat: dict = {} for chemin in sorted(DOSSIER_PLAYBOOKS_GROUPES.glob("*.yml")): roles: list[str] = [] try: with chemin.open("r", encoding="utf-8") as fichier: docs = yaml.safe_load(fichier) or [] plays = docs if isinstance(docs, list) else [docs] for play in plays: if not isinstance(play, dict): continue for role in play.get("roles", []) or []: if isinstance(role, str): roles.append(role) elif isinstance(role, dict): nom = role.get("role") or role.get("name") if nom: roles.append(str(nom)) except (OSError, yaml.YAMLError): pass resultat[chemin.stem] = {"roles": roles, "stub": not roles} return resultat def hotes_actifs_fichier(path: Path) -> set[str]: data = charger_yaml(path) return set(enfants(data).get(GROUPE_HOTES_ACTIFS, {}).get("hosts", {}) or {}) def variables_hote(data: dict, hote: str) -> dict: resultat: dict = {} for groupe_data in enfants(data).values(): hosts = groupe_data.get("hosts", {}) valeurs = hosts.get(hote) if isinstance(valeurs, dict): resultat.update(valeurs) return resultat def liste_hotes(data: dict) -> list[dict]: noms = set() for groupe_data in enfants(data).values(): noms.update(groupe_data.get("hosts", {})) hotes = [] for nom in sorted(noms): groupes = [] etat = "planifie" visible = False for groupe, groupe_data in enfants(data).items(): if nom not in groupe_data.get("hosts", {}): continue if groupe == GROUPE_HOTES_ACTIFS: etat = "actif" visible = True elif groupe in GROUPES_ETAT_HOTE: visible = True continue elif est_groupe_operationnel(groupe): groupes.append(groupe) visible = True if not visible: continue variables = variables_hote(data, nom) hote = {"nom": nom, "etat": etat, "groupes": sorted(groupes)} for cle, var, _type in CHAMPS_PROVISION: valeur = variables.get(var, "") hote[cle] = "" if valeur is None else valeur if not str(hote.get("utilisateur_ansible", "")).strip(): hote["utilisateur_ansible"] = "ansible" hotes.append(hote) return hotes def _joignable(ip: str, port: int = 22, timeout: float = 1.5) -> bool: if not ip: return False try: with socket.create_connection((ip, port), timeout=timeout): return True except OSError: return False def sonder_hotes(path: Path) -> dict: """Etat REEL (vs plan declaratif) : la VM de chaque hote repond-elle sur SSH ?""" cibles = [(h.get("nom"), h.get("adresse_ip")) for h in liste_hotes(charger_yaml(path))] with ThreadPoolExecutor(max_workers=16) as executeur: resultats = executeur.map(lambda c: (c[0], _joignable(c[1])), cibles) return {nom: ok for nom, ok in resultats if nom} def marquer_serveur_actif(hote: str) -> None: """Auto-actif : un hote materialise (clone/deploye) passe 'actif' dans le plan.""" registre = charger_serveurs(FICHIER_SERVEURS) serveurs = registre.get("serveurs") or {} if hote in serveurs and serveurs[hote].get("etat") != "actif": serveurs[hote]["etat"] = "actif" ecrire_serveurs(FICHIER_SERVEURS, registre) def serveurs_reconcilies(data: dict) -> list: """Serveurs du plan + valeurs derivees + statut de reconciliation (lecture).""" registre = charger_serveurs(FICHIER_SERVEURS) nomenclature = charger_nomenclature(FICHIER_NOMENCLATURE) hotes = {h["nom"]: h for h in liste_hotes(data) if h.get("nom")} resultat = [] for nom, srv in (registre.get("serveurs") or {}).items(): rec = reconcilier_serveur(nom, srv, hotes.get(nom), nomenclature) d = rec["derive"] statut = "absent" if rec["absent_inventaire"] else ("divergence" if rec["divergences"] else "reconcilie") resultat.append({ "nom": nom, "fonction": srv.get("fonction", ""), "etat": srv.get("etat", ""), "noeud": srv.get("noeud", ""), "stockage": srv.get("stockage", ""), "disque": srv.get("disque", ""), "memoire": srv.get("memoire", ""), "coeurs": srv.get("coeurs", ""), "integrations": srv.get("integrations", []), "vmid": d.get("vmid", ""), "adresse_ip": d.get("adresse_ip", ""), "vlan": d.get("vlan", ""), "statut": statut, "divergences": rec["divergences"], }) return resultat def flux_api() -> list: """Matrice des flux réseau (meta/flux.yml de tous les rôles) — lecture seule.""" try: import resoudre_flux flux = resoudre_flux.charger_flux() except Exception: return [] lignes = [] for role in sorted(flux): for fl in flux[role]: pair = fl.get("pair") pair = ", ".join(str(p) for p in pair) if isinstance(pair, list) else str(pair) port = fl.get("port") port = ", ".join(str(p) for p in port) if isinstance(port, list) else str(port) lignes.append({ "role": role, "sens": fl.get("sens", ""), "port": port, "protocole": fl.get("protocole", ""), "pair": pair, "chiffrement": fl.get("chiffrement", ""), "raison": fl.get("raison", ""), }) return lignes def couches_api() -> list: """Ordre de déploiement en couches (couches-deploiement.yml + graphe) — lecture seule.""" try: import orchestrer return [{"nom": nom, "groupes": list(groupes)} for nom, groupes in orchestrer.ordre_deploiement()] except Exception: return [] def inventaire_api(path: Path) -> dict: data = charger_yaml(path) dependencies = charger_dependances(FICHIER_DEPENDANCES) return { "flux": flux_api(), "couches": couches_api(), "inventaire": os.path.relpath(path, RACINE), "production": instance_est_production(path), "groupes": groupes_disponibles(data), "dependances": dependencies, "nomenclature": charger_nomenclature(FICHIER_NOMENCLATURE), "chaine": roles_des_groupes(), "bases": charger_bases_donnees(FICHIER_BASES), "applications": charger_applications(FICHIER_APPLICATIONS), "domaines": charger_domaines(FICHIER_DOMAINES), "serveurs": serveurs_reconcilies(data), "hotes": liste_hotes(data), "intrants": intrants_api(), "integrations_disponibles": integrations_disponibles(), "integrations_universelles": integrations_universelles(RACINE), "liens_acceptes": catalogue_liens(RACINE), # LE SCHEMA VOYAGE AVEC LES DONNEES. Le navigateur genere ses formulaires depuis # lui plutot que de les porter en dur : un champ ajoute au plan apparait a l'ecran # sans toucher au JS. Genere par `make schema`, garde par P61. "schema": schema_du_plan(), } def schema_du_plan() -> dict: """Le JSON Schema des registres, tel que `make schema` l'a ecrit. Lu depuis le fichier plutot que recalcule : c'est l'artefact VERSIONNE qui fait foi, et P61 refuse qu'il soit perime. Recalculer ici laisserait l'ecran afficher une forme que personne n'a relue. """ f = RACINE / "docs" / "audit" / "schema-plan.json" if not f.is_file(): return {} try: return json.loads(f.read_text(encoding="utf-8")).get("registres", {}) except (OSError, json.JSONDecodeError): return {} def _lire_yaml_dict(path: Path) -> dict: if not path.exists(): return {} with path.open("r", encoding="utf-8") as fichier: data = yaml.safe_load(fichier) or {} return data if isinstance(data, dict) else {} def instance_est_production(path: Path) -> bool: """Cette instance est-elle la PRODUCTION ? (garde-fou de prudence, pas un blocage) Le déploiement est permis sur toute instance (lab compris, vers son infra) ; ce drapeau ne sert qu'à exiger une confirmation renforcée côté GUI quand c'est la prod. Repère explicite : `setops_production` dans `/group_vars/all/`. Rétro-compatibilité : à défaut, ancien repère (inventaire nommé « production »). """ base = path.parent / "group_vars" / "all" fichiers = sorted(base.glob("*.yml")) if base.is_dir() else [] plat = path.parent / "group_vars" / "all.yml" if plat.exists(): fichiers.append(plat) for f in fichiers: d = _lire_yaml_dict(f) if isinstance(d, dict) and "setops_production" in d: return bool(d["setops_production"]) return path.resolve() == INVENTAIRE_PRODUCTION def _lire_fabric_a_plat() -> dict: """Les intrants PLATS de underlay.yml, aplatis dans l'espace de noms du panneau.""" data = _lire_yaml_dict(FICHIER_UNDERLAY).get("underlay") or {} plat = {} for variable, chemin in CHEMINS_UNDERLAY.items(): noeud = data for clef in chemin: noeud = (noeud or {}).get(clef) if isinstance(noeud, dict) else None plat[variable] = noeud return plat def intrants_courants() -> dict: cache: dict = {} valeurs: dict = {} for variable, fcle, _classe, _section, _libelle, typ, _prop in INTRANTS_SCHEMA: if fcle not in cache: cache[fcle] = (_lire_fabric_a_plat() if fcle == "fabric" else _lire_yaml_dict(FICHIERS_INTRANTS[fcle])) v = cache[fcle].get(variable) if typ == "liste": items = v if isinstance(v, list) else [] valeurs[variable] = [str(x).strip() for x in items if str(x).strip()] else: valeurs[variable] = "" if v is None else v return valeurs # --- Couverture du schema du plan par le GUI ------------------------------------- # # Ce que les fonctions de sauvegarde JS ecrivent REELLEMENT, registre par registre. # Source unique verifiee par `scripts/couverture_gui.py` : tout champ present dans # un plan reel (instance ou modele) mais absent d'ici signale un trou de l'interface # — l'operateur devrait editer le YAML a la main, ce que la doctrine refuse. # # A TENIR A JOUR avec sauvegarderServeurs / sauvegarderApplications / # sauvegarderBases / sauvegarderDomaines. Le verificateur controle aussi que chaque # champ declare ici apparait bien dans le source du GUI (garde anti-derive). # CE QUE LE GUI SAIT ECRIRE — ET D'OU IL LE TIENT (2026-09-08). # # Cette table etait tenue A LA MAIN : quelqu'un devait penser a l'allonger quand un champ # arrivait au plan, et P19 confrontait la liste au reel. Une copie gardee reste une copie. # # LA TABLE EST VIDE, ET C'EST LE BUT. Les SIX registres ont leur formulaire GENERE depuis # `docs/audit/schema-plan.json` (`make schema`) : la liste se DEDUIT du schema au lieu # d'etre recopiee ici. `bases_donnees` est sortie la premiere, la nomenclature ensuite, # puis `serveurs_bd`, `domaines_publics`, et enfin `serveurs` et `applications`. # # On la garde plutot que de la supprimer : le mecanisme doit pouvoir accueillir un champ # qu'un formulaire ecrirait sans que le schema le decrive — et le jour ou ca arrive, il # faudra l'ECRIRE ICI, visiblement, plutot que de le laisser vivre en silence. # # P19 lit `champs_ecrits_par_gui()` et non cette constante : elle garde un mecanisme, pas # une liste. CHAMPS_ECRITS_A_LA_MAIN: dict[str, set[str]] = {} # Les registres dont le formulaire est GENERE : leurs champs viennent du schema. REGISTRES_GENERES = {"bases_donnees", "nomenclature", "serveurs_bd", "domaines_publics", "serveurs", "applications"} def champs_ecrits_par_gui() -> dict: """Ce que le GUI sait ecrire : declare a la main, plus derive du schema. Une seule fonction repond a la question, pour que P19 n'ait pas a savoir lequel des deux mecanismes couvre quel registre. """ def _noms(props: dict) -> set: """Les champs decrits, TABLES ET BLOCS IMBRIQUES COMPRIS. Les plans portent `categorie` et `service` a plat, sous la table `fonctions`. Ne rendre que les proprietes de premier niveau ferait dire au GUI qu'il ne sait pas les ecrire, alors que le formulaire genere les edite. """ vus = set() for nom, spec in (props or {}).items(): vus.add(nom) for conteneur in (spec, spec.get("additionalProperties"), spec.get("items")): if isinstance(conteneur, dict) and isinstance(conteneur.get("properties"), dict): vus |= _noms(conteneur["properties"]) return vus resultat = {nom: set(champs) for nom, champs in CHAMPS_ECRITS_A_LA_MAIN.items()} for nom, reg in (schema_du_plan() or {}).items(): if nom in REGISTRES_GENERES: resultat[nom] = _noms((reg.get("entite") or {}).get("properties") or {}) return resultat def integrations_disponibles() -> list: """Roles client_* SELECTIONNABLES : ceux qui restent un choix. Les universelles en sont exclues : on ne coche pas ce qui s'applique a tout le monde. Le GUI les affiche quand meme, en lecture seule (voir /api/etat) — sinon l'operateur lirait « aucune supervision » sur une flotte entierement supervisee. """ base = RACINE / "roles" universelles = set(integrations_universelles(RACINE)) return sorted(p.name for p in base.glob("client_*") if p.is_dir() and p.name not in universelles) def intrants_api() -> dict: schema = [{"variable": v, "fichier": f, "classe": c, "section": s, "libelle": l, "type": t, "proprietaire": p} for (v, f, c, s, l, t, p) in INTRANTS_SCHEMA] return { "schema": schema, "valeurs": intrants_courants(), "secrets_attendus": secrets_attendus(), "fichiers": {k: os.path.relpath(v, RACINE) for k, v in FICHIERS_INTRANTS.items()}, } def _coercer_intrant(typ: str, brut: object) -> object: if typ == "liste": items = brut if isinstance(brut, list) else str(brut).replace("\n", ",").split(",") # dedoublonne en preservant l'ordre vus, sortie = set(), [] for x in items: s = str(x).strip() if s and s not in vus: vus.add(s); sortie.append(s) return sortie s = str(brut).strip() if typ == "bool": return s.lower() in ("true", "1", "oui", "yes", "on") if typ == "int": return int(s) if s.lstrip("-").isdigit() else s return s _REFERENCE_VOUTE = re.compile(r"^\{\{\s*vault_[A-Za-z0-9_]+\s*\}\}$") def _est_reference_voute(valeur: object) -> bool: """Vrai si la valeur est une simple REFERENCE de voute (`{{ vault_xxx }}`). Un fichier d'intrants peut porter, a cote de ses parametres anodins, des renvois par NOM vers la voute (cas de group_vars/opnsense.yml). Ce n'est pas un secret : c'est un pointeur. La fusion « preserve les cles non gerees » les relit donc du disque, et le garde-fou ne doit pas les confondre avec une valeur saisie. """ return bool(_REFERENCE_VOUTE.match(str(valeur).strip())) _CLE_RACINE = re.compile(r"^([A-Za-z_][A-Za-z0-9_]*):") def _rendre_cle(cle: str, valeur: object) -> list[str]: """Les lignes que PyYAML ecrirait pour cette seule clef (listes en bloc incluses).""" texte = yaml.safe_dump({cle: valeur}, default_flow_style=False, allow_unicode=True, sort_keys=False) return texte.rstrip("\n").split("\n") _ENTREE_TABLE = re.compile(r"^(\s+)([^\s#:][^:]*):(\s*)(.*)$") def _fusion_table(bloc: list[str], entrees: dict) -> list[str]: """Met a jour une table clef->mapping LIGNE A LIGNE, sans la reecrire. POURQUOI PAS `_fusion_chirurgicale` SEULE (mesure du 2026-09-08). Elle remplace le BLOC entier des qu'une valeur change. Sur `fonctions:` de la nomenclature, ajouter une seule entree a produit deux degats mesures : - les quinze entrees compactes `{ categorie: 6, service: 2 }` sont devenues quarante-deux lignes — le fichier cesse d'etre une table lisible d'un coup d'oeil ; - le commentaire « LE POSTE D'EXPLOITATION », qui coiffait `ops:`, s'est retrouve en TETE du bloc. Un commentaire deplace ne se contente pas d'etre laid : il explique desormais la mauvaise ligne. Il affirmait que `collab` etait le poste d'exploitation. Ici on ne touche qu'aux LIGNES concernees. Une entree inchangee n'est pas relue, un commentaire garde son voisin, et le style compact survit parce qu'on ne le refait pas. """ tete, corps = bloc[0], bloc[1:] sortie, vues = [tete], set() flow, colonne = 0, 0 i = 0 while i < len(corps): ligne = corps[i] m = _ENTREE_TABLE.match(ligne) if not m: sortie.append(ligne) i += 1 continue indent, cle, ecart, reste = m.group(1), m.group(2).strip(), m.group(3), m.group(4) # L'entree s'etend sur les lignes PLUS indentees qui suivent immediatement. fin = i j = i + 1 while j < len(corps): suite = corps[j] if not suite.strip() or suite.strip().startswith("#"): break creux = len(suite) - len(suite.lstrip()) if creux <= len(indent): break fin = j j += 1 vues.add(cle) if reste.startswith("{"): flow += 1 colonne = max(colonne, len(indent) + len(cle) + 1 + len(ecart)) try: ancien = (yaml.safe_load("\n".join(l[len(indent):] for l in corps[i:fin + 1])) or {}).get(cle) except yaml.YAMLError: ancien = object() # illisible : on la considere changee if cle not in entrees: i = fin + 1 # entree retiree : ses lignes disparaissent, pas ses voisines continue if ancien == entrees[cle]: sortie.extend(corps[i:fin + 1]) elif reste.startswith("{"): sortie.append(f"{indent}{cle}:{ecart}{_rendre_flow(entrees[cle])}") else: # L'entree tenait sur plusieurs lignes : on la refait en bloc, indentee # comme ses voisines. sortie.extend(indent + l for l in _rendre_cle(cle, entrees[cle])) i = fin + 1 # Les entrees NEUVES vont a la fin du bloc, dans le style dominant. neuves = [c for c in entrees if c not in vues] if neuves: indent = " " for ligne in corps: m = _ENTREE_TABLE.match(ligne) if m: indent = m.group(1) break while sortie and not sortie[-1].strip(): sortie.pop() for cle in neuves: if flow: ecart = " " * max(1, colonne - len(indent) - len(str(cle)) - 1) sortie.append(f"{indent}{cle}:{ecart}{_rendre_flow(entrees[cle])}") else: sortie.extend(indent + l for l in _rendre_cle(cle, entrees[cle])) return sortie def _rendre_flow(valeur: object) -> str: """La forme compacte `{ a: 1, b: 2 }` — celle que le fichier employait deja.""" rendu = yaml.safe_dump(valeur, default_flow_style=True, allow_unicode=True, sort_keys=False, width=10 ** 6).strip() if rendu.startswith("{") and rendu.endswith("}"): rendu = "{ " + rendu[1:-1].strip() + " }" return rendu def _fusion_chirurgicale(texte: str, valeurs: dict, tables: set | None = None) -> str: """Pose les valeurs dans le texte existant SANS reformater ce qui n'a pas change. POURQUOI (mesure du 2026-08-18). Ces fichiers portent la memoire ECRITE des decisions : pourquoi `dns_amorcage` pointe sur un resolveur public, pourquoi le plan de gestion est reste en 10.0.0.0/24, pourquoi la frontiere ecoute sur `lan`. Chaque enregistrement du panneau les repassait au `safe_dump` : clefs retriees, et TOUS les commentaires effaces. Un enregistrement de 13:48 a ainsi emporte 94 lignes de commentaire dans quatre fichiers (129 -> 35, le reste etant l'entete que le panneau reecrit) — dont celle qui disait pourquoi la valeur qu'on venait de changer avait ete choisie. Le depot connaissait deja le geste juste : `_ecrire_intrants_fabric` et `_ecrire_index_nomenclature` remplacent la ligne sans toucher au reste. Ce qui manquait ici, c'est de savoir remplacer une valeur de LISTE, qui tient sur plusieurs lignes. Trois regles : - une clef dont la valeur ne change pas n'est PAS reecrite (zero bruit au diff) ; - les commentaires INTERNES a un bloc remplace sont conserves, jamais juges : une explication devenue fausse se corrige a la main, elle ne se supprime pas toute seule ; - une clef absente du fichier est ajoutee a la fin, jamais inseree au hasard. """ actuel = yaml.safe_load(texte) actuel = actuel if isinstance(actuel, dict) else {} lignes = texte.split("\n") sortie: list[str] = [] vues: set[str] = set() i = 0 while i < len(lignes): m = _CLE_RACINE.match(lignes[i]) cle = m.group(1) if m else None if cle is None or cle not in valeurs: sortie.append(lignes[i]) i += 1 continue # Le bloc de valeur : la ligne de clef, puis tout ce qui est indente ou item de # liste. Les commentaires et lignes vides qui SUIVENT la derniere ligne de valeur # appartiennent a la clef suivante — on les laisse ou ils sont. fin = i j = i + 1 while j < len(lignes): l = lignes[j] if l.startswith((" ", "\t", "- ")) or l == "-": fin = j elif l.strip().startswith("#") or not l.strip(): pass # peut etre interne au bloc : on tranche avec `fin` else: break j += 1 vues.add(cle) if cle in actuel and actuel[cle] == valeurs[cle]: sortie.extend(lignes[i:fin + 1]) # inchangee : pas un octet touche elif cle in (tables or set()) and isinstance(valeurs[cle], dict): # Table clef->mapping : on edite les lignes concernees, pas le bloc. sortie.extend(_fusion_table(lignes[i:fin + 1], valeurs[cle])) else: internes = [l for l in lignes[i + 1:fin + 1] if l.strip().startswith("#")] rendu = _rendre_cle(cle, valeurs[cle]) sortie.extend([rendu[0]] + internes + rendu[1:]) i = fin + 1 ajouts = [c for c in valeurs if c not in vues] if ajouts: while sortie and not sortie[-1].strip(): sortie.pop() for cle in ajouts: sortie.extend(_rendre_cle(cle, valeurs[cle])) sortie.append("") return "\n".join(sortie) def _ecrire_intrants_fichier(path: Path, valeurs: dict, entete: str) -> None: # Une cle sensible n'est toleree que sous forme de reference de voute preservee. # Toute VALEUR reelle sur une de ces cles fait echouer l'ecriture. interdits = sorted(cle for cle in set(valeurs) & INTRANTS_CLES_INTERDITES if not _est_reference_voute(valeurs[cle])) if interdits: raise ValueError(f"Refus d'ecrire des cles sensibles via le GUI: {', '.join(interdits)}") path.parent.mkdir(parents=True, exist_ok=True) if path.exists(): # Fichier existant : on pose les valeurs DANS son texte (commentaires preserves). path.write_text(_fusion_chirurgicale(path.read_text(encoding="utf-8"), valeurs), encoding="utf-8") return with ecriture_atomique(path) as fichier: fichier.write(entete) yaml.safe_dump(valeurs, fichier, default_flow_style=False, sort_keys=True, allow_unicode=True) def ecrire_intrants(payload: dict) -> None: valeurs = payload.get("valeurs", {}) or {} # Garde-fou en entree : le panneau n'affiche jamais ces cles, mais une requete # forgee pourrait les soumettre. On refuse bruyamment plutot que d'ignorer. soumis = sorted(set(valeurs) & INTRANTS_CLES_INTERDITES) if soumis: raise ValueError(f"Cles sensibles soumises au panneau (voute uniquement): {', '.join(soumis)}") if not str(valeurs.get("domaine_interne", "")).strip(): raise ValueError("domaine_interne ne peut pas etre vide (cle de voute de l'ecosysteme).") types = {v: t for (v, f, c, s, l, t, p) in INTRANTS_SCHEMA} fichier_de = {v: f for (v, f, c, s, l, t, p) in INTRANTS_SCHEMA} par_fichier: dict = {} for variable, valeur in valeurs.items(): if variable not in fichier_de: continue par_fichier.setdefault(fichier_de[variable], {})[variable] = _coercer_intrant(types[variable], valeur) if "identite" in par_fichier: # Fusion, comme les trois autres fichiers. C'etait la SEULE branche a ecrire # par-dessus le disque, et elle perdait donc en silence tout intrant absent du # schema du panneau. Mesure du 2026-08-10, au pre-vol de l'epreuve de # portabilite : un enregistrement du panneau a supprime `dns_amorcage` et # `amorcage_acces_courriel` de Technolibre. Le premier laisse une VM naitre sans # resolution — `apt` ne peut alors rien installer ; le second arrete le # deploiement sur la garde de `amorcage_acces` (D-72). # # Le meme geste sur Chezlepro aurait mange les memes cles. Un panneau qui ne # connait pas une valeur ne doit pas avoir le droit de la detruire. fusion = _lire_yaml_dict(INTRANTS_IDENTITE) # preserve les cles non gerees fusion.update(par_fichier["identite"]) _ecrire_intrants_fichier( INTRANTS_IDENTITE, fusion, "# Intrants d'IDENTITE de l'instance — SOURCE UNIQUE, partagee par tous les envs.\n" "# Edite par le panneau « Intrants de base » du GUI (make inventaire-ui).\n" "# Reference via symlink depuis inventories//group_vars/all/10-intrants.yml.\n---\n") if "proxmox" in par_fichier: fusion = _lire_yaml_dict(INTRANTS_PROXMOX) # preserve les cles non gerees fusion.update(par_fichier["proxmox"]) _ecrire_intrants_fichier( INTRANTS_PROXMOX, fusion, "# Parametres Proxmox non sensibles. Les secrets (jetons d'API) vont dans la\n" "# voute UNIQUE de l'instance : group_vars/all/vault.yml.\n" "# Cles d'intrants editees par le panneau « Intrants de base » du GUI.\n---\n") if "proxmox_hebergeur" in par_fichier: fusion = _lire_yaml_dict(INTRANTS_PROXMOX_HEBERGEUR) fusion.update(par_fichier["proxmox_hebergeur"]) _ecrire_intrants_fichier( INTRANTS_PROXMOX_HEBERGEUR, fusion, "# Le CLUSTER, vu par l'HEBERGEUR — pas par un tenant.\n" "#\n" "# API du cluster, noeuds, stockages, ponts : du materiel possede par\n" "# l'hebergeur. Recopiees dans le group_vars de chaque tenant, ces valeurs\n" "# avaient deja diverge — deux inventaires contradictoires du meme cluster.\n" "#\n" "# Ce fichier vit dans le depot de l'HEBERGEUR, a cote d'underlay.yml (D-14),\n" "# et se trouve par derivation du symlink qui designe deja l'hebergeur (D-17).\n" "#\n" "# Chez le TENANT restent son golden template et ses defauts de placement :\n" "# des CHOIX faits a l'interieur de ce que l'hebergeur offre ici.\n" "#\n" "# Les secrets (jeton d'API) n'entrent jamais ici : voute de l'instance.\n---\n") if "frontiere" in par_fichier: # Fusion (comme Proxmox) : preserve les cles non gerees par le panneau, # en particulier les references de voute `{{ vault_opnsense_api_* }}`. fusion = _lire_yaml_dict(INTRANTS_FRONTIERE) fusion.update(par_fichier["frontiere"]) _ecrire_intrants_fichier( INTRANTS_FRONTIERE, fusion, "# Frontiere nord/sud (OPNsense) — parametres NON sensibles.\n" "# Cles d'intrants editees par le panneau « Intrants de base » du GUI.\n" "#\n" "# Les identifiants d'API ne figurent ici que par REFERENCE de nom : leurs\n" "# valeurs vivent dans la voute unique de l'instance (ansible-vault edit\n" "# group_vars/all/vault.yml), sous vault_opnsense_api_key / _api_secret.\n" "# Les obtenir dans OPNsense : System > Access > Users > >\n" "# API keys > « + ». Le secret n'est affiche qu'a la creation.\n" "#\n" "# Le boitier est HORS FLOTTE Ansible : on ne fait que piloter son API REST\n" "# depuis le controleur. Cf. docs/frontiere-opnsense.md.\n" "#\n" "# Les interfaces se nomment par leur IDENTIFIANT OPNsense (wan, opt1...), celui\n" "# que parle l'API REST — ni le peripherique FreeBSD (igb1), ni le libelle affiche.\n" "# `opnsense_prochain_saut` n'est pas ici : il DERIVE du reseau de transit de\n" "# l'underlay (underlay.yml, cle `passerelle_sortie`). Source unique.\n---\n") if "fabric" in par_fichier: _ecrire_intrants_fabric(par_fichier["fabric"]) if "reseau" in par_fichier and "index" in par_fichier["reseau"]: _ecrire_index_nomenclature(int(par_fichier["reseau"]["index"])) def _ecrire_intrants_fabric(valeurs: dict) -> None: """Met a jour les cles PLATES de underlay.yml SANS reformater le fichier. Meme motif que `_ecrire_index_nomenclature` : ce fichier porte des commentaires qui expliquent des decisions d'architecture (pourquoi un seul routeur, pourquoi le transit vit dans l'underlay...). Un `safe_dump` les effacerait toutes. On remplace donc la ligne existante, en respectant son indentation, et on n'invente rien : une clef absente du fichier n'est PAS creee — elle n'a pas de place evidente ou aller. """ if not FICHIER_UNDERLAY.exists(): return texte = FICHIER_UNDERLAY.read_text(encoding="utf-8") for variable, chemin in CHEMINS_UNDERLAY.items(): if variable not in valeurs: continue val = str(valeurs[variable]).strip() if not val: continue feuille = chemin[-1] # Indentation attendue : 2 espaces a la racine du bloc `underlay`, 4 si imbrique. indent = " " * len(chemin) motif = re.compile(rf"(?m)^{indent}{re.escape(feuille)}:[ \t]*\S.*$") remplacement = f"{indent}{feuille}: {val}" texte, n = motif.subn(remplacement, texte, count=1) if not n: raise ValueError(f"underlay.yml : clef '{feuille}' introuvable — " f"l'ajouter au fichier avant de la regler ici.") FICHIER_UNDERLAY.write_text(texte, encoding="utf-8") def valider_nomenclature(donnees: dict, serveurs: dict) -> None: """Refuse une nomenclature qui casserait la derivation, AVANT d'ecrire. La nomenclature n'est pas un registre comme les autres : elle ne decrit pas des objets, elle decrit la REGLE dont VMID, VLAN, adresse et passerelle se derivent. Une fonction retiree ne laisse pas un trou local — elle prive de tout adressage chaque VM qui la porte, et `deriver_nomenclature` rend `None` SANS RIEN DIRE. Trois refus, dans l'ordre du degat : - une fonction encore portee par une VM ne se retire pas ; - une fonction ne peut designer qu'une zone declaree ; - `index` ne s'ecrit pas ici (il est ALLOUE par le site, cf. l'entete du fichier). """ categories = donnees.get("categories") or {} fonctions = donnees.get("fonctions") or {} if not isinstance(categories, dict) or not isinstance(fonctions, dict): raise ValueError("`categories` et `fonctions` doivent etre des tables.") connues = {str(c) for c in categories} for nom, f in fonctions.items(): if not isinstance(f, dict): raise ValueError(f"Fonction '{nom}': entree invalide.") cat = f.get("categorie") if cat in (None, ""): raise ValueError(f"Fonction '{nom}': 'categorie' requise.") if str(cat) not in connues: raise ValueError(f"Fonction '{nom}': zone {cat} non declaree dans `categories` " f"— la VM n'aurait ni VLAN ni adresse.") if f.get("service") in (None, ""): raise ValueError(f"Fonction '{nom}': 'service' requis.") portees = {} for nom_vm, srv in (serveurs or {}).items(): if isinstance(srv, dict) and srv.get("fonction"): portees.setdefault(str(srv["fonction"]), []).append(nom_vm) for fonction, vms in sorted(portees.items()): if fonction not in fonctions: raise ValueError(f"Fonction '{fonction}' retiree, mais encore portee par " f"{len(vms)} VM ({', '.join(sorted(vms)[:4])}) — " f"deplacer ces VM d'abord.") zones_utilisees = {str(f.get("categorie")) for f in fonctions.values() if isinstance(f, dict)} for zone in sorted(zones_utilisees - connues): raise ValueError(f"Zone {zone} utilisee mais non declaree.") def ecrire_nomenclature(path: Path, donnees: dict) -> None: """Ecrit la nomenclature SANS effacer ses commentaires. POURQUOI PAS `safe_dump`. Ce fichier porte dix-neuf lignes qui disent pourquoi `index` EST RECU ET PAS DECIDE, et pourquoi le modele des zones est commun a toute la flotte. Un `safe_dump` les emporterait toutes — c'est exactement l'incident du 2026-08-18 (94 lignes de commentaire perdues en un enregistrement). On repasse donc par `_fusion_chirurgicale`, qui pose les valeurs dans le texte existant. `index` n'est jamais du lot : il s'ecrit par `_ecrire_index_nomenclature`, depuis le panneau Intrants, parce qu'il vient du site et non de cette vue. """ texte = path.read_text(encoding="utf-8") if path.exists() else "---\n" valeurs = {cle: donnees[cle] for cle in ("cidr_hote", "reservations", "categories", "fonctions") if cle in donnees} with ecriture_atomique(path) as fichier: fichier.write(_fusion_chirurgicale(texte, valeurs, tables={"categories", "fonctions"})) def _ecrire_index_nomenclature(index: int) -> None: """Met a jour `index:` dans plan/nomenclature.yml SANS reformater le reste. Le fichier garde ses commentaires et son formatage compact (l'adressage se derive de ce seul seed). Remplacement de la ligne existante, ou insertion en tete a defaut. """ path = FICHIER_NOMENCLATURE texte = path.read_text(encoding="utf-8") if path.exists() else "---\n" nouveau, remplace = re.subn(r"(?m)^index:.*$", f"index: {index}", texte) if not remplace: nouveau = re.sub(r"(?m)^(---\n)", rf"\g<1>index: {index}\n", texte, count=1) or texte # Atomique comme les autres : cette ecriture-ci avait ete oubliee au passage du # 2026-09-06, et une coupure ici laisserait le seed a moitie ecrit — c'est-a-dire # tout l'adressage de l'ecosysteme. with ecriture_atomique(path) as fichier: fichier.write(nouveau) def _ecrire_registre(path: Path, entete: str, contenu: dict) -> None: """Ecrit un registre du plan SANS effacer les commentaires qu'il porte. MESURE DU 2026-09-08, sur les fichiers reels de cet ecosysteme : domaines.yml 6 lignes de commentaire -> 3 (PERD 3) applications.yml 27 -> 5 (PERD 22) serveurs.yml 18 -> 3 (PERD 15) bases-donnees.yml 4 -> 4 (intact, il n'en a pas) Quarante lignes, detruites par n'importe quel « Sauvegarder » depuis les vues Serveurs, Applications ou Domaines. Parmi elles, celle qui explique pourquoi `backup-01` a ete RETIRE : « une supervision creuse est pire qu'aucune — elle est verte ». Et celle qui dit dans quel ORDRE les deux roles du runner s'appliquent. C'est l'incident du 2026-08-18 (94 lignes perdues dans les fichiers d'intrants), jamais corrige pour les registres du plan : `_fusion_chirurgicale` avait ete ecrite pour les intrants seuls, et les quatre `ecrire_*` sont restes au `safe_dump`. Les registres du plan ont tous la meme forme — une clef racine, une table d'entites — donc chaque clef racine est declaree comme TABLE : `_fusion_table` edite alors les lignes concernees, une entite retiree disparait, et le reste n'est pas touche. Un fichier qui n'existe pas encore recoit son entete et un `safe_dump` : il n'y a rien a preserver. """ if path.exists(): texte = path.read_text(encoding="utf-8") if texte.strip(): with ecriture_atomique(path) as fichier: fichier.write(_fusion_chirurgicale(texte, contenu, tables=set(contenu))) return with ecriture_atomique(path) as fichier: fichier.write(entete) yaml.safe_dump(contenu, fichier, default_flow_style=False, sort_keys=False, allow_unicode=True) def ecrire_bases(path: Path, registre: dict) -> None: entete = ( "# Registre des bases de donnees Set-OPS (serveurs_bd + bases_donnees).\n" "# Edite par make inventaire-ui ou scripts/bases_donnees.py.\n" "# 'portee' (application|groupe|hote) interprete 'consommateur' ; defaut groupe.\n" "# 'secret' nomme une variable Ansible Vault (jamais le mot de passe).\n" "---\n" ) _ecrire_registre(path, entete, { "serveurs_bd": registre.get("serveurs_bd", {}) or {}, "bases_donnees": registre.get("bases_donnees", {}) or {}, }) def ecrire_applications(path: Path, registre: dict) -> None: entete = ( "# Registre des applications Set-OPS (application = entite pivot).\n" "# Edite par make inventaire-ui ou scripts/applications.py.\n" "# Une VM peut porter plusieurs applications ; une base se lie a une application.\n" "# 'liens' = bindings declaratifs [{vers, role}] resolus par scripts/instancier.py ;\n" "# les roles acceptes sont ceux de roles//meta/liens.yml.\n" "---\n" ) _ecrire_registre(path, entete, {"applications": registre.get("applications", {}) or {}}) def ecrire_serveurs(path: Path, registre: dict) -> None: entete = ( "# Registre des serveurs (VM) du plan Set-OPS.\n" "# Edite par make inventaire-ui ou scripts/serveurs.py.\n" "# VMID/IP/VLAN/passerelle sont DERIVES de la fonction via instance/plan/nomenclature.yml.\n" "---\n" ) _ecrire_registre(path, entete, {"serveurs": registre.get("serveurs", {}) or {}}) def ecrire_domaines(path: Path, registre: dict) -> None: entete = ( "# Registre des domaines publics (zones DNS de l'instance).\n" "# Edite par make inventaire-ui ou scripts/domaines.py.\n" "# autorite : primaire-cache | auto-heberge | delegue (label descriptif).\n" "---\n" ) _ecrire_registre(path, entete, {"domaines_publics": registre.get("domaines_publics", {}) or {}}) def _extraire_echec(ligne: str, tache: str) -> dict | None: """D'une ligne Ansible fatal/unreachable, tire un échec en LANGAGE CLAIR (tâche · hôte · type · message) — pour communiquer entre humains.""" m = re.search(r'(?:fatal|failed):\s*\[([^\]]+)\]:\s*(FAILED!|UNREACHABLE!)?\s*=>\s*(\{.*\})', ligne) if not m: return None hote = m.group(1) typ = "injoignable (SSH)" if m.group(2) == "UNREACHABLE!" else "échec" detail: dict = {} try: detail = json.loads(m.group(3)) except Exception: pass if detail.get("censored"): message = "sortie masquée (no_log) — souvent une tâche qui manipule un secret" else: msg = str(detail.get("msg", "")).strip() # Pour un échec de commande, msg est générique ; la vraie cause est dans stderr. stderr = "" for cle in ("stderr", "module_stderr", "stdout"): valeur = str(detail.get(cle, "")).strip() if valeur: lignes = [x for x in valeur.splitlines() if x.strip()] stderr = lignes[0] if lignes else valeur # l'erreur mène, l'indice suit break if msg and msg.lower() not in ("non-zero return code", "module failure", ""): message = msg elif stderr: message = stderr else: message = msg or "échec sans message explicite (voir le journal complet)" message = " ".join(str(message).split()) # une ligne propre return {"tache": (tache or "?").strip(), "hote": hote, "type": typ, "message": message[:400]} def donnees_hote(hote: dict) -> dict: variables: dict = {} for cle, var, typ in CHAMPS_PROVISION: brut = str(hote.get(cle, "")).strip() if not brut: continue if typ == "int" and brut.lstrip("-").isdigit(): variables[var] = int(brut) else: variables[var] = brut return variables def construire_inventaire(payload: dict, groupes_connus: list[str]) -> dict: enfants_data: dict = { "modeles_vm": {"hosts": {}}, GROUPE_HOTES_ACTIFS: {"hosts": {}}, GROUPE_HOTES_PLANIFIES: {"hosts": {}}, } for groupe in groupes_connus: enfants_data.setdefault(groupe, {"hosts": {}}) enfants_data.setdefault("hotes_proxmox", {"hosts": {}}) for hote in payload.get("hotes", []): nom = str(hote.get("nom", "")).strip() if not nom: continue variables = donnees_hote(hote) groupe_etat = GROUPE_HOTES_ACTIFS if hote.get("etat") == "actif" else GROUPE_HOTES_PLANIFIES enfants_data[groupe_etat]["hosts"][nom] = variables for groupe in hote.get("groupes", []): if groupe not in groupes_connus: continue enfants_data[groupe]["hosts"][nom] = {} return {"all": {"children": enfants_data}} def valider_dependances_payload(payload: dict, dependencies: dict) -> None: hotes = payload.get("hotes", []) active_groups: set[str] = set() for hote in hotes: if hote.get("etat") != "actif": continue active_groups.update(hote.get("groupes", [])) missing: list[str] = [] for hote in hotes: if hote.get("etat") != "actif": continue nom = str(hote.get("nom", "")).strip() for groupe in hote.get("groupes", []): for required_group in dependencies.get(groupe, {}).get("requiert_groupes_actifs", []): if required_group not in active_groups: missing.append(f"{nom}: {groupe} requiert {required_group} actif") if missing: raise ValueError("Dependances manquantes: " + "; ".join(missing)) def valider_payload(payload: dict, groupes_connus: list[str], dependencies: dict) -> None: noms: set[str] = set() vmids: dict[str, str] = {} for hote in payload.get("hotes", []): nom = str(hote.get("nom", "")).strip() if not nom: raise ValueError("Un hote a un nom vide.") if nom in noms: raise ValueError(f"Hote en double: {nom}") noms.add(nom) if hote.get("etat") == "actif" and not str(hote.get("adresse_ip", "")).strip(): raise ValueError(f"Hote actif sans adresse IP: {nom}") groupes = hote.get("groupes", []) if not groupes: raise ValueError(f"Aucun groupe operationnel pour: {nom}") inconnus = sorted(set(groupes) - set(groupes_connus)) if inconnus: raise ValueError(f"Groupes inconnus pour {nom}: {', '.join(inconnus)}") vlan = str(hote.get("vlan", "")).strip() if vlan and (not vlan.isdigit() or not 1 <= int(vlan) <= 4094): raise ValueError(f"VLAN invalide pour {nom}: {vlan} (attendu 1-4094)") cidr = str(hote.get("cidr", "")).strip() if cidr and (not cidr.isdigit() or not 0 <= int(cidr) <= 32): raise ValueError(f"CIDR invalide pour {nom}: {cidr} (attendu 0-32)") for cle, libelle in (("memoire", "Memoire"), ("coeurs", "Coeurs")): valeur = str(hote.get(cle, "")).strip() if valeur and (not valeur.isdigit() or int(valeur) <= 0): raise ValueError(f"{libelle} invalide pour {nom}: {valeur} (entier positif attendu)") vmid = str(hote.get("vmid", "")).strip() if not vmid: continue if vmid in vmids: raise ValueError(f"VMID en double: {vmid} pour {vmids[vmid]} et {nom}") vmids[vmid] = nom valider_dependances_payload(payload, dependencies) HTML = r""" Set-OPS · Votre artisan numérique

Set-OPS

Votre artisan numérique
Dépendances causales
""" def basculer_instance(nom: str) -> str: """Repointe le symlink `instance/` vers un depot frere. Renvoie la nouvelle cible. Memes garde-fous que `make instance-utiliser`, plus une validation STRICTE : `nom` doit etre une instance DECOUVERTE (dossier frere avec plan/nomenclature.yml), ce qui interdit toute traversee de chemin (pas de '/', '..', cible arbitraire). """ import instances as _inst connues = {i["nom"] for i in _inst.decouvrir()} if nom not in connues: raise ValueError(f"Instance inconnue : {nom!r} (attendu : {', '.join(sorted(connues)) or 'aucune'}).") cible = RACINE.parent / nom if not cible.is_dir(): raise ValueError(f"Dossier introuvable : ../{nom}") lien = RACINE / "instance" if lien.exists() and not lien.is_symlink(): raise ValueError("'instance' existe et n'est pas un symlink — refus.") if lien.is_symlink() or lien.exists(): lien.unlink() lien.symlink_to(f"../{nom}") return os.readlink(lien) class Gestionnaire(BaseHTTPRequestHandler): # Inventaire de l'instance ACTIVE, resolu DYNAMIQUEMENT depuis le symlink # `instance/` a chaque acces : ainsi la bascule d'instance (POST # /api/instance-utiliser) prend effet sans redemarrer le serveur. Les chemins # de plan (FICHIER_*) sont deja relatifs au symlink et suivent de meme. # SETOPS_INVENTAIRE force un inventaire fixe (CI / usage explicite). @property def inventaire(self) -> Path: force = os.environ.get("SETOPS_INVENTAIRE") return Path(force) if force else _inventaire(INSTANCE, "principal", "production") def log_message(self, *args) -> None: # silence par defaut pass def repondre(self, status: int, contenu: bytes, content_type: str) -> None: try: self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(contenu))) self.end_headers() self.wfile.write(contenu) except (BrokenPipeError, ConnectionResetError): pass # le client a ferme la connexion : rien a faire def repondre_json(self, status: int, data: dict) -> None: self.repondre(status, json.dumps(data, ensure_ascii=False).encode("utf-8"), "application/json; charset=utf-8") def _ecrire_flux(self, texte: str) -> bool: try: self.wfile.write(texte.encode("utf-8", "replace")) self.wfile.flush() return True except (BrokenPipeError, ConnectionResetError): return False def executer_flux(self, hote: str, mode: str, vault: str | None = None) -> None: if not MOTIF_HOTE.match(hote or ""): self.repondre_json(400, {"erreur": "Nom d'hote invalide."}) return tous_hotes = {h.get("nom") for h in liste_hotes(charger_yaml(self.inventaire))} if hote not in tous_hotes: self.repondre_json(409, {"erreur": f"{hote} est inconnu de l'inventaire (applique le plan d'abord)."}) return # 'creer' (clone) est permis meme sur un hote planifie ; les autres exigent actif. if mode in ("verifier", "deployer", "pousser") and hote not in hotes_actifs_fichier(self.inventaire): self.repondre_json(409, {"erreur": f"{hote} n'est pas un hote actif sauvegarde."}) return if mode == "deployer" and not VERIF_OK.get(hote): self.repondre_json(409, {"erreur": "Verification (dry-run) requise avant de deployer."}) return if not VERROU.acquire(blocking=False): self.repondre_json(409, {"erreur": "Une execution est deja en cours."}) return fichier_vault = None journal = None try: cible = {"deployer": "deployer", "pousser": "deployer", "creer": "creer-vm"}.get(mode, "verifier-deploiement") env = dict(os.environ, PYTHONUNBUFFERED="1", ANSIBLE_FORCE_COLOR="0") if vault: # LE MOT DE PASSE SAISI S'AJOUTE AUX CLES, IL NE LES REMPLACE PAS. # # Verifie empiriquement le 2026-09-08, contre deux voutes jetables a mots # de passe distincts : `ANSIBLE_VAULT_PASSWORD_FILE` et # `ANSIBLE_VAULT_IDENTITY_LIST` se CUMULENT — Ansible essaie tous les # secrets, et un PASSWORD_FILE errone n'empeche rien. # # Ca compte depuis « une voute, une cle » (2026-08-28) : creer une VM # ouvre DEUX voutes dans la meme execution — celle du tenant et # `underlay.vault.yml`. Un seul mot de passe ne pourrait pas les ouvrir # toutes les deux. Ce qui sauve ce chemin, c'est que `make` calcule # `ANSIBLE_VAULT_IDENTITY_LIST` par-dessus (Makefile, `?=`), et que les # deux coexistent dans l'environnement de la recette. # # NE PAS « SIMPLIFIER » EN RETIRANT L'UN DES DEUX. Le PASSWORD_FILE sert # le cas ou la cle n'est pas sur la machine (un runner qui n'a que la # sienne) ; l'IDENTITY_LIST sert le poste de l'exploitant, qui les a # toutes. Chacun couvre ce que l'autre ne couvre pas. fd, fichier_vault = tempfile.mkstemp(prefix="setops-vault-") # 0600 par defaut os.write(fd, str(vault).encode("utf-8")) os.close(fd) env["ANSIBLE_VAULT_PASSWORD_FILE"] = fichier_vault self.send_response(200) self.send_header("Content-Type", "text/plain; charset=utf-8") self.send_header("Cache-Control", "no-cache") self.end_headers() # Persistance : un journal horodaté par exécution, en plus du streaming live. dossier_logs = self.inventaire.parents[2] / "logs" dossier_logs.mkdir(parents=True, exist_ok=True) chemin_log = dossier_logs / f"{hote}-{cible}-{time.strftime('%Y%m%dT%H%M%S')}.log" journal = open(chemin_log, "w", encoding="utf-8") entete = f"$ make {cible} HOTE={hote}\n\n" journal.write(entete) self._ecrire_flux(entete) tache_courante = "" echecs: list[dict] = [] proc = subprocess.Popen( ["make", cible, f"HOTE={hote}"], cwd=str(RACINE), env=env, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, ) try: for ligne in proc.stdout: journal.write(ligne) if ligne.startswith("TASK ["): deb, fin = ligne.find("["), ligne.find("]") if -1 < deb < fin: tache_courante = ligne[deb + 1:fin] elif ("FAILED!" in ligne or "UNREACHABLE!" in ligne) and \ ("fatal:" in ligne or ligne.lstrip().startswith("failed:")): echec = _extraire_echec(ligne, tache_courante) if echec: echecs.append(echec) if not self._ecrire_flux(ligne): # Navigateur déconnecté : on continue à journaliser jusqu'au bout. for reste in proc.stdout: journal.write(reste) break finally: rc = proc.wait() if mode == "verifier" and rc == 0: VERIF_OK[hote] = True elif mode in ("deployer", "pousser"): VERIF_OK.pop(hote, None) # Auto-actif : un hote qu'on vient de materialiser (clone reussi) ou de # deployer passe 'actif' dans le plan. Reflet fidele : materialise = actif. if rc == 0 and mode in ("creer", "deployer", "pousser"): try: marquer_serveur_actif(hote) except Exception: pass if rc != 0 and echecs: self._ecrire_flux("__RESUME__ " + json.dumps({"echecs": echecs[:6]}, ensure_ascii=False) + "\n") journal.write(f"\n__FIN__ rc={rc}\n") self._ecrire_flux(f"\n[journal : {os.path.relpath(chemin_log, RACINE)}]\n__FIN__ rc={rc}\n") finally: if journal: try: journal.close() except OSError: pass if fichier_vault: try: os.remove(fichier_vault) except OSError: pass VERROU.release() def do_GET(self) -> None: chemin = urlparse(self.path).path try: if chemin == "/": page = HTML.replace("__JETON__", JETON) self.repondre(200, page.encode("utf-8"), "text/html; charset=utf-8") elif chemin == "/favicon.ico": self.repondre(204, b"", "image/x-icon") elif chemin == "/api/inventaire": self.repondre_json(200, inventaire_api(self.inventaire)) elif chemin == "/api/sondes": self.repondre_json(200, sonder_hotes(self.inventaire)) elif chemin == "/api/devis-reseau": tenants = devis_reseau.decouvrir() self.repondre_json(200, { "config": devis_reseau.generer(tenants) if tenants else "", "tenants": [{"nom": nom, "index": n["index"]} for nom, _, n in tenants], }) elif chemin == "/api/devis-opnsense": # Import paresseux : ce module lit l'underlay et les inventaires de tous # les tenants ; une erreur ici ne doit pas empecher le reste du panneau. try: import devis_opnsense as _dop tenants = _dop.decouvrir() self.repondre_json(200, { "config": _dop.rendre(_dop.construire(tenants)) if tenants else "", }) except Exception as exc: self.repondre_json(200, {"config": "", "erreur": str(exc)}) elif chemin == "/api/instances": import instances as _inst import instance_creer as _ic liste = _inst.decouvrir() self.repondre_json(200, { "instances": liste, "collisions": {str(k): v for k, v in _inst.collisions(liste).items()}, "modeles": sorted(_ic.modeles_disponibles()), "index_pris": sorted(_ic.index_federes_utilises()), }) else: self.repondre_json(404, {"erreur": "Introuvable"}) except Exception as exc: self.repondre_json(500, {"erreur": str(exc)}) def do_POST(self) -> None: chemin = urlparse(self.path).path if self.headers.get("X-Jeton") != JETON: self.repondre_json(403, {"erreur": "Jeton manquant ou invalide."}) return longueur = int(self.headers.get("Content-Length", "0") or 0) corps = self.rfile.read(longueur).decode("utf-8") if longueur else "" try: donnees = json.loads(corps) if corps else {} except json.JSONDecodeError: self.repondre_json(400, {"erreur": "Corps JSON invalide."}) return try: if chemin == "/api/inventaire": self.repondre_json(409, {"erreur": "hosts.yml est genere depuis le plan. " "Edite instance/plan/serveurs.yml et instance/plan/applications.yml " "(vues Serveurs / Applications), puis « Appliquer le plan »."}) elif chemin == "/api/bases": valider_bases(donnees, charger_applications(FICHIER_APPLICATIONS)) ecrire_bases(FICHIER_BASES, donnees) self.repondre_json(200, inventaire_api(self.inventaire)) elif chemin == "/api/applications": valider_applications(donnees, charger_domaines(FICHIER_DOMAINES), charger_serveurs(FICHIER_SERVEURS)) ecrire_applications(FICHIER_APPLICATIONS, donnees) self.repondre_json(200, inventaire_api(self.inventaire)) elif chemin == "/api/serveurs": valider_serveurs(donnees, charger_nomenclature(FICHIER_NOMENCLATURE)) ecrire_serveurs(FICHIER_SERVEURS, donnees) self.repondre_json(200, inventaire_api(self.inventaire)) elif chemin == "/api/nomenclature": valider_nomenclature(donnees, charger_serveurs(FICHIER_SERVEURS)) ecrire_nomenclature(FICHIER_NOMENCLATURE, donnees) self.repondre_json(200, inventaire_api(self.inventaire)) elif chemin == "/api/domaines": valider_domaines(donnees) ecrire_domaines(FICHIER_DOMAINES, donnees) self.repondre_json(200, inventaire_api(self.inventaire)) elif chemin == "/api/intrants": ecrire_intrants(donnees) self.repondre_json(200, inventaire_api(self.inventaire)) elif chemin == "/api/instance-utiliser": cible = basculer_instance(str(donnees.get("nom", "")).strip()) self.repondre_json(200, {"ok": True, "cible": cible}) elif chemin == "/api/instance-creer": import instance_creer as _ic brut_idx = str(donnees.get("index", "")).strip() _ic.creer(str(donnees.get("nom", "")).strip(), str(donnees.get("modele", "")).strip(), int(brut_idx) if brut_idx else None) self.repondre_json(200, {"ok": True, "nom": str(donnees.get("nom", "")).strip()}) elif chemin == "/api/instancier": # Cliquer « Appliquer le plan » EST l'intention explicite : on force # (le diff-refus est un garde-fou CLI ; ici le clic le confirme). git = filet. resultat = subprocess.run( ["python3", str(RACINE / "scripts/instancier.py"), "appliquer", "--force"], capture_output=True, text=True, cwd=str(RACINE)) self.repondre_json(200, {"ok": resultat.returncode == 0, "sortie": (resultat.stdout + resultat.stderr).strip()}) elif chemin == "/api/verifier": self.executer_flux(str(donnees.get("hote", "")), "verifier", donnees.get("vault")) elif chemin == "/api/deployer": self.executer_flux(str(donnees.get("hote", "")), "deployer", donnees.get("vault")) elif chemin == "/api/creer": self.executer_flux(str(donnees.get("hote", "")), "creer", donnees.get("vault")) elif chemin == "/api/pousser": self.executer_flux(str(donnees.get("hote", "")), "pousser", donnees.get("vault")) else: self.repondre_json(404, {"erreur": "Introuvable"}) except Exception as exc: self.repondre_json(400, {"erreur": str(exc)}) def main() -> int: parser = argparse.ArgumentParser(description="Interface web locale pour gerer un inventaire Set-OPS.") parser.add_argument("--inventaire", type=Path, default=INVENTAIRE_DEFAUT) parser.add_argument("--hote", default="127.0.0.1") parser.add_argument("--port", type=int, default=8765) args = parser.parse_args() serveur = ThreadingHTTPServer((args.hote, args.port), Gestionnaire) print(f"Interface inventaire: http://{args.hote}:{args.port}/") print(f"Inventaire: {Gestionnaire.inventaire}") serveur.serve_forever() return 0 if __name__ == "__main__": raise SystemExit(main())