#!/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 # --- CE QUE CETTE CONSOLE PEUT, DERIVE DE CE QUE SA MACHINE PORTE -------------------- # # LE DEFAUT QUI A RENDU CECI NECESSAIRE (2026-09-16). Servie par le runner d'un SITE, # cette console affichait ZERO serveur, ZERO application, ZERO base — sans une erreur. # `serveur_ops` RETIRE le lien `instance` sur un runner d'hebergeur (a juste titre : un # lien perime vers un tenant serait pire), `charger_yaml` rend `{"all": {"children": {}}}` # sur un fichier absent, et la page dessinait ce vide comme un plan vide. # # C'est la forme exacte que P79 garde partout ailleurs : une derivation qui ne trouve rien # ne se distingue pas d'une derivation qui n'a rien a trouver. Ici elle avait le pire # visage possible — celui d'un ecosysteme qu'on croit sans machines. # # ON DERIVE, ON NE DECLARE PAS. Un `serveur_ops_gui_mode: site|tenant` au plan serait une # SECONDE liste, et une liste qui suit une autre prend du retard : il suffirait de # reconfigurer un runner sans y penser. Les deux symlinks, eux, SONT le pouvoir : # # instance/ -> je configure CET ecosysteme (j'ai sa voute) # underlay.yml -> je materialise sur CETTE fabric (j'ai celle du site) # # C'est la meme coupure que `roles/serveur_ops` exige deja dans son assertion, et que # `docs/responsabilites-locataire-hebergeur.md` nomme : calculer, configurer, materialiser. def contexte() -> dict: """La portee de cette console — DELEGUEE a la classe qui la sert. Ce que cette fonction faisait a la main vit desormais dans `Console` et ses deux branches. Elle reste parce que le harnais et les preuves l'appellent par ce nom, et parce qu'un appelant n'a pas a savoir laquelle des deux consoles lui repond. """ return console().contexte() def inventaire_du_site() -> dict: """Les machines du SITE, mises dans la forme d'un `hosts.yml`. L'inventaire d'un site est un SCRIPT (`site_inventaire.py`) : on l'EXECUTE — en l'important, sans sous-processus a chaque page — et l'on traduit son resultat dans la forme que le reste de ce fichier sait lire. Juger sa source reviendrait a relire le raisonnement au lieu du resultat. """ import site_inventaire brut = site_inventaire.inventaire() or {} hostvars = ((brut.get("_meta") or {}).get("hostvars") or {}) enfants_site: dict = {} for groupe, valeur in brut.items(): if groupe == "_meta" or not isinstance(valeur, dict): continue hotes = valeur.get("hosts") or [] if not hotes: continue enfants_site[groupe] = {"hosts": {h: dict(hostvars.get(h) or {}) for h in hotes}} return {"all": {"children": enfants_site}} def sonder_donnees(data: dict) -> dict: """Meme sonde que `sonder_hotes`, sur un inventaire DEJA charge (le site).""" cibles = [(h.get("nom"), h.get("adresse_ip")) for h in liste_hotes(data)] 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} # CE QUE CHAQUE POINT D'ECRITURE EXIGE. Toute route POST y figure — P81 refuse qu'une # route nouvelle echappe a la garde en silence, parce qu'une garde qui suit une liste # prend du retard sur elle. POUVOIR_REQUIS = { "/api/inventaire": "editer_le_plan", "/api/bases": "editer_le_plan", "/api/applications": "editer_le_plan", "/api/serveurs": "editer_le_plan", "/api/nomenclature": "editer_le_plan", "/api/domaines": "editer_le_plan", "/api/intrants": "editer_le_plan", "/api/instancier": "editer_le_plan", "/api/instance-creer": "editer_le_plan", "/api/instance-utiliser": "editer_le_plan", "/api/verifier": "configurer", "/api/deployer": "configurer", "/api/pousser": "configurer", "/api/creer": "materialiser", # LA GARDE DE CETTE ROUTE EST PLUS FINE QUE LES AUTRES, ET C'EST VOULU. Un runbook de # site et un runbook de locataire n'exigent pas le meme pouvoir : un seul mot ici # serait soit trop laxiste (un site lancerait un geste de locataire), soit trop # severe (un locataire ne pourrait plus rien). La portee est DECLAREE au registre, # et c'est elle qu'on resout — jamais celle que le navigateur annonce. "/api/runbook-etape": "selon-le-runbook", } # CE QU'UNE PORTEE DE RUNBOOK EXIGE DE LA CONSOLE. Les deux pouvoirs de `contexte()`, et # rien d'autre : `poste` porte les deux liens, il doit donc les deux. POUVOIR_PAR_PORTEE = { "tenant": ("configurer",), "site": ("materialiser",), "poste": ("configurer", "materialiser"), "toute": (), } # Ce qu'une valeur d'etape a le droit d'etre. Les commandes partent en LISTE (jamais par # un shell), donc l'injection n'est pas le risque ; ce qui l'est, c'est qu'une valeur se # fasse lire par `make` comme une option. On refuse donc ce qui commence par un tiret. MOTIF_VALEUR = re.compile(r"^[^-\s][\w .@:/+=,-]{0,199}$", re.UNICODE) 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 [] # ═══ LES DEUX CONSOLES, ET LE TRONC QU'ELLES PARTAGENT ═══════════════════════════════ # # POURQUOI DEUX CLASSES, ET PAS DEUX FONCTIONS (2026-09-20). # # Il y avait deux fonctions : `inventaire_api` pour un locataire, `inventaire_api_du_site` # pour un site. Elles assemblaient LA MEME charge, chacune de son cote — et elles ont # derive, deux fois le meme jour : # # en FORME `serveurs` valait `[]` d'un cote et `{}` de l'autre. Les deux « vides », # et la page ne les lit pas pareil : `charger()` LEVAIT, et la console d'un # site mourait avant sa premiere vue. # en CONTENU cinq registres etaient servis vides « pour ne pas fabriquer un faux plan # de tenant » — alors que le site a SON plan, au meme format, que les # lecteurs du moteur lisent sans broncher : 9 serveurs, 21 applications. # # Deux fonctions qui doivent rendre la meme forme finissent toujours par ne plus la # rendre. Un tronc qui assemble, et des branches qui ne decident que de CE QUI LEUR # APPARTIENT, rendent cette derive impossible : il n'y a plus qu'un seul endroit ou la # charge se compose. # # CE QUE CHAQUE BRANCHE EST, au sens de `docs/responsabilites-locataire-hebergeur.md` : # # ConsoleLocataire configure SES services — voute du tenant, plan derive d'un seed # ConsoleSite materialise le terrain — voute du site, plan DECLARE (ip, vmid) # ConsoleOrpheline ne pilote rien, et c'est ce qu'elle doit dire # # Aucune n'est omnipotente, et la classe le porte au lieu d'un `if portee == ...` seme # dans le code. class Console: """Le tronc : ce qu'une console EST, quelle que soit la machine qui la sert.""" # LE ROLE est une propriete de la CLASSE — ce pour quoi cette console existe. # LA PORTEE se CALCULE — ce qu'elle peut vraiment, ici et maintenant. Les deux ne se # confondent pas : un poste prive de la voute du site reste l'atelier du mainteneur, # et n'engendre pourtant rien. Figer la portee sur la classe aurait rendu a ce # poste-la un pouvoir que la mesure lui refuse. role = "aucun" @property def portee(self) -> str: p = self.pouvoirs() return ("poste" if p["configurer"] and p["materialiser"] else "tenant" if p["configurer"] else "site" if p["materialiser"] else "orphelin") @property def titre(self) -> str: return { "poste": f"Poste du mainteneur — {self.ecosysteme()} sur {self.fabric()}", "tenant": f"Console du locataire — {self.ecosysteme()}", "site": f"Console du SITE — {self.fabric()}", }.get(self.portee, "Console sans pouvoir — elle ne pilote rien") # LES REGISTRES DU PLAN, declares UNE fois : leur nom a l'ecran, leur fichier, leur # lecteur. Une branche qui n'a pas de plan les rend vides — de la BONNE forme, parce # que c'est le tronc qui decide de la forme, pas la branche. REGISTRES = ( ("nomenclature", "nomenclature.yml", charger_nomenclature), ("bases", "bases-donnees.yml", charger_bases_donnees), ("applications", "applications.yml", charger_applications), ("domaines", "domaines.yml", charger_domaines), ) # --- CE QUE CHAQUE BRANCHE DECIDE POUR ELLE --------------------------------------- def racine_du_plan(self) -> Path | None: """Le dossier `plan/` que cette console lit. None = elle n'en a pas.""" return None def inventaire_brut(self) -> dict: """L'inventaire, dans la forme d'un `hosts.yml`. Vide par defaut.""" return {"all": {"children": {}}} def nom_inventaire(self) -> str: return "" def est_production(self) -> bool: return False def serveurs(self, data: dict) -> list: """Les serveurs, TOUJOURS en liste. D'ou ils viennent regarde la branche.""" return [] def intrants(self): return None def pouvoirs(self) -> dict: return {"editer_le_plan": False, "configurer": False, "materialiser": False, "fabric": False} def raisons(self) -> dict: return { "configurer": "Cette console ne porte la voute d'aucun ecosysteme.", "editer_le_plan": "Aucun plan n'est monte ici.", "materialiser": "Cette console ne sait sur quelle fabric poser une VM.", "fabric": "Aucune fabric n'est montee ici.", } def ecosysteme(self) -> str | None: return None def fabric(self) -> str | None: return None # --- LE TRONC : UN SEUL ENDROIT OU LA CHARGE SE COMPOSE --------------------------- def contexte(self) -> dict: return { "portee": self.portee, "role": self.role, "titre": self.titre, "ecosysteme": self.ecosysteme(), "fabric": self.fabric(), "peut": self.pouvoirs(), "raison": self.raisons(), # QUELLE SOURCE D'INVENTAIRE : un site n'a pas de `hosts.yml`, son inventaire # est un SCRIPT. Une console qui a quelque chose a piloter le NOMME. "inventaire": self.nom_inventaire() or None, } def peut_conduire(self, portee_etape: str) -> tuple[bool, str]: """Cette console peut-elle conduire une etape de CETTE portee — et sinon, pourquoi ? LE JUGEMENT VIT ICI, PAS DANS LES ROUTES. Il etait ecrit deux fois : dans la route qui LISTE les assistants et dans celle qui EXECUTE une etape. Deux copies d'une meme regle finissent par ne plus dire la meme chose — et celle qui se trompe est toujours celle qui execute, parce que c'est la moins relue. """ peut = self.pouvoirs() manquants = [p for p in POUVOIR_PAR_PORTEE.get(portee_etape, ()) if not peut.get(p)] if not manquants: return True, "" return False, self.raisons().get(manquants[0], "Geste hors de la portee de cette console.") def charge(self) -> dict: """LA charge servie a la page — la seule, pour les deux consoles. Les clefs et leurs FORMES sont fixees ici. Une branche choisit ce qu'elle met dedans, jamais la forme de ce qu'elle met : c'est exactement ce qui a manque. """ data = self.inventaire_brut() plan = self.racine_du_plan() registres = {} for clef, fichier, lecteur in self.REGISTRES: chemin = (plan / fichier) if plan else None registres[clef] = lecteur(chemin) if (chemin and chemin.is_file()) else {} return { "flux": flux_api(), "couches": couches_api(), "inventaire": self.nom_inventaire(), "production": self.est_production(), "groupes": groupes_disponibles(data), "dependances": charger_dependances(FICHIER_DEPENDANCES), "chaine": roles_des_groupes(), **registres, "serveurs": self.serveurs(data), "hotes": liste_hotes(data), "intrants": self.intrants(), "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. Genere par `make schema`, P61. "schema": schema_du_plan(), "contexte": self.contexte(), } class ConsoleLocataire(Console): """Elle CONFIGURE : des roles sur ses machines, avec la voute du tenant. Son plan DERIVE tout d'un seed `index` — VMID, adresse, VLAN. Elle edite donc le plan, et l'inventaire qu'elle lit est un artefact genere depuis lui. """ role = "locataire" def __init__(self, instance: Path, inventaire: Path, a_voute: bool): self._instance = instance self._inventaire = inventaire self._a_voute = a_voute def ecosysteme(self): return self._instance.resolve().name def racine_du_plan(self): return self._instance / "plan" def inventaire_brut(self): return charger_yaml(self._inventaire) def nom_inventaire(self): return os.path.relpath(self._inventaire, RACINE) def est_production(self): return instance_est_production(self._inventaire) def serveurs(self, data): return serveurs_reconcilies(data) def intrants(self): return intrants_api() def pouvoirs(self): return {"editer_le_plan": True, "configurer": self._a_voute, "materialiser": False, "fabric": False} def raisons(self): r = super().raisons() r["materialiser"] = ("Cette console configure un ecosysteme ; elle ne sait pas " "sur quelle fabric poser une VM, et c'est voulu : exploiter " "n'est pas engendrer.") r["configurer"] = ("Un ecosysteme est monte, mais sa voute n'est pas sur cette " "machine : calculer un inventaire ne demande aucun secret, " "configurer un service si.") r["fabric"] = ("La fabric appartient a l'hebergeur. Cette console n'en monte " "aucune carte.") return r class ConsoleSite(Console): """Elle MATERIALISE : cree et detruit des VM, pose le reseau, avec la voute du site. SES ZONES DERIVENT, SES MACHINES SONT ENCORE ECRITES. Un site a son propre index depuis le 2026-09-12 — le 37 pour celui-ci — et ses zones en descendent comme celles d'un locataire descendent du sien (P20 le garde). Ce qui reste ecrit dans son plan, ce sont les `ip`, `vmid` et `noeud` de ses MACHINES : une dette nommee, pas une doctrine. La docstring disait le contraire : elle recopiait un commentaire perime de huit jours au lieu de mesurer. Son inventaire, lui, n'est pas un fichier mais un SCRIPT. ELLE A UN PLAN, ET ON LE LUI SERT. La charge du site servait cinq registres vides « pour ne pas fabriquer un faux plan de tenant ». Le site n'a pas de faux plan : il a LE SIEN, au meme format, que les lecteurs du moteur lisent sans broncher — mesure du 2026-09-20 : 9 serveurs, 21 applications, 2 bases, 1 domaine, tous invisibles. """ role = "site" def __init__(self, depot: Path, a_voute: bool): self._depot = depot self._a_voute = a_voute def fabric(self): return self._depot.name def racine_du_plan(self): return self._depot / "plan" def inventaire_brut(self): return inventaire_du_site() def nom_inventaire(self): return f"{self.fabric()} (inventaire dynamique)" def est_production(self): return True def serveurs(self, data): """Les serveurs DECLARES du site, dans la forme que la page sait lire. On ne les reconcilie pas contre une nomenclature : un site n'en a pas, ses valeurs sont ecrites. Le statut se lit donc contre l'inventaire REEL — declare ici, present la-bas, ou pas. """ chemin = self.racine_du_plan() / "serveurs.yml" if not chemin.is_file(): return [] registre = (charger_serveurs(chemin) or {}).get("serveurs") or {} vivants = {h.get("nom") for h in liste_hotes(data)} sortie = [] for nom, srv in registre.items(): gabarit = srv.get("gabarit") or {} sortie.append({ "nom": nom, # Un site ne derive d'aucune fonction : il nomme le RESEAU qui le porte. "fonction": srv.get("reseau", ""), "etat": srv.get("etat", ""), "noeud": srv.get("noeud", ""), "stockage": srv.get("stockage", ""), "disque": gabarit.get("disque", ""), "memoire": gabarit.get("memoire", ""), "coeurs": gabarit.get("vcpu", ""), "integrations": srv.get("integrations", []), "vmid": srv.get("vmid", ""), "adresse_ip": srv.get("ip", ""), "vlan": "", "statut": "reconcilie" if nom in vivants else "absent", "divergences": [], }) return sortie def pouvoirs(self): # EDITER : pas encore. Les formulaires sont generes depuis le schema du plan d'un # LOCATAIRE, ou l'adressage se derive ; le plan d'un site l'ECRIT. Offrir un # formulaire qui ecrirait de travers serait pire que ne rien offrir. On LIT. return {"editer_le_plan": False, "configurer": False, "materialiser": self._a_voute, "fabric": True} def raisons(self): r = super().raisons() r["configurer"] = ("Cette console materialise le terrain ; elle n'a la voute " "d'aucun locataire, donc elle ne configure aucun de leurs " "services.") r["editer_le_plan"] = ("Le plan de ce site se LIT ici. Ses adresses sont ECRITES, " "pas derivees d'un seed, et les formulaires sont generes " "depuis le schema d'un plan de locataire : ils ecriraient " "de travers.") r["materialiser"] = ("Cette console LIT la carte d'une fabric dont elle n'a pas " "la voute. Creer ou detruire une machine demande le secret " "du site, et il n'est pas ici.") r["fabric"] = ("La fabric appartient a l'hebergeur. Ce qui s'affiche ici vient " "d'une copie locale, et une copie avait deja diverge.") return r class ConsolePoste(ConsoleLocataire): """L'atelier du mainteneur : les deux voutes, donc les deux pouvoirs. Elle EST une console de locataire — meme plan, meme inventaire — qui sait en plus sur quelle fabric poser une VM. L'heritage dit exactement ca, et evite une troisieme copie de la meme charge. """ role = "poste" def __init__(self, instance, inventaire, depot, a_voute_tenant, a_voute_site): super().__init__(instance, inventaire, a_voute_tenant) self._depot = depot self._a_voute_site = a_voute_site def fabric(self): return self._depot.name def pouvoirs(self): p = super().pouvoirs() p["materialiser"] = self._a_voute_site p["fabric"] = True return p def raisons(self): # CE POSTE SAIT SUR QUELLE FABRIC POSER — il en monte la carte. Ce qui lui manque, # le cas echeant, c'est la VOUTE. Heriter du refus d'un locataire (« elle ne sait # pas sur quelle fabric ») enverrait chercher une carte qui est pourtant la, et # ferait perdre le temps qu'un message juste economise. r = super().raisons() r["materialiser"] = ("Cette console LIT la carte d'une fabric dont elle n'a pas " "la voute. Creer ou detruire une machine demande le secret " "du site, et il n'est pas ici.") r["fabric"] = ("La fabric appartient a l'hebergeur. Ce qui s'affiche ici vient " "d'une copie locale, et une copie avait deja diverge.") return r def console(inventaire: Path | None = None) -> Console: """QUELLE console sert cette machine — derive de ce qu'elle peut OUVRIR. `docs/responsabilites-locataire-hebergeur.md` attache chaque pouvoir a une voute : calculer n'en demande aucune, configurer celle du tenant, materialiser celle du site. On ne prouve pas qu'une voute s'ouvre — le mot de passe se tape a l'execution — mais son ABSENCE est decisive et se mesure sans rien ouvrir. """ a_instance = INSTANCE.exists() depot = _depot_hebergeur() inv = inventaire or INVENTAIRE_DEFAUT voute_tenant = (INSTANCE / "inventories" / "principal" / "group_vars" / "all" / "vault.yml") if a_instance else None a_voute_tenant = bool(voute_tenant and voute_tenant.is_file()) a_voute_site = bool(depot and (depot / "underlay.vault.yml").is_file()) # LA BRANCHE SUIT LES LIENS — ce que cette machine est faite pour faire. La PORTEE, # elle, suivra les voutes : une console de locataire sans sa voute reste une console # de locataire (elle calcule son inventaire), et se dira « orpheline » parce qu'elle # ne peut rien configurer. Le role et le pouvoir sont deux questions distinctes. if a_instance and depot: return ConsolePoste(INSTANCE, inv, depot, a_voute_tenant, a_voute_site) if a_instance: return ConsoleLocataire(INSTANCE, inv, a_voute_tenant) if depot: return ConsoleSite(depot, a_voute_site) return Console() def inventaire_api(path: Path) -> dict: """La charge d'une console de LOCATAIRE — deleguee au tronc. Elle reste parce que le harnais et les tests l'appellent par ce nom. Elle n'assemble plus rien : assembler deux fois est exactement ce qui a fait deriver la charge du site de celle du locataire. """ return ConsoleLocataire(INSTANCE, path, True).charge() 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 executer_runbook(self, runbook: str, cible: str, valeurs: dict, vault: str | None = None, index: int | None = None) -> None: """Une etape DECLAREE d'un runbook — et rien d'autre. LE NAVIGATEUR NE NOMME PAS UNE COMMANDE, IL NOMME UNE PLACE. On ne lance pas « la cible que la page demande » : on lance ce que le registre declare a cette place-la, avec les seules variables qu'il declare. Une page compromise, ou simplement perimee, ne peut donc pas reclamer `raser` depuis un runbook de mesure. C'est la meme discipline que `POUVOIR_REQUIS` : ce qui protege n'est pas un bouton grise, c'est un serveur qui refuse — et qui DIT pourquoi. """ try: import runbooks as _rb etape = _rb.etape_declaree(runbook, cible, RACINE, index) except Exception as exc: # noqa: BLE001 self.repondre_json(500, {"erreur": f"Registre des runbooks illisible : {exc}"}) return if etape is None: self.repondre_json(409, {"erreur": f"« {cible} » n'est pas une etape declaree " f"du runbook « {runbook} ». Rien n'a ete lance."}) return # Les variables : celles du registre, dans l'ordre du registre. args: list[str] = [] for v in etape["variables"]: brut = str(valeurs.get(v["nom"], "")).strip() if not brut: if v.get("facultatif"): continue self.repondre_json(400, {"erreur": f"{v['nom']} est requis pour cette etape."}) return if not MOTIF_VALEUR.match(brut): self.repondre_json(400, {"erreur": f"{v['nom']} : valeur refusee (caracteres " f"interdits, ou commence par un tiret)."}) return if v.get("valeurs") and brut not in v["valeurs"]: self.repondre_json(400, {"erreur": f"{v['nom']} : « {brut} » n'est pas une " f"valeur admise ({', '.join(v['valeurs'])})."}) return args.append(f"{v['nom']}={brut}") # Les valeurs IMPOSEES par l'etape (CONFIRMER=true et compagnie) ne viennent # jamais du navigateur : une confirmation qu'on peut envoyer soi-meme n'en est pas # une. Le garde-fou reel est la confirmation ECRITE exigee cote page, plus la # garde propre a la cible. for clef, val in (etape["fixes"] or {}).items(): args.append(f"{clef}={val}") if not VERROU.acquire(blocking=False): self.repondre_json(409, {"erreur": "Une execution est deja en cours."}) return fichier_vault = None journal = None try: env = dict(os.environ, PYTHONUNBUFFERED="1", ANSIBLE_FORCE_COLOR="0") if vault: fd, fichier_vault = tempfile.mkstemp(prefix="setops-vault-") 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() # Un journal par execution. Chez un locataire il vit dans l'instance ; sur un # runner de site il n'y a pas d'instance, et `tmp/` est deja hors depot. dossier_logs = (INSTANCE / "logs") if INSTANCE.exists() else (RACINE / "tmp" / "logs") dossier_logs.mkdir(parents=True, exist_ok=True) chemin_log = dossier_logs / f"{runbook}-{cible}-{time.strftime('%Y%m%dT%H%M%S')}.log" journal = open(chemin_log, "w", encoding="utf-8") entete = f"$ make {cible} {' '.join(args)}\n\n" journal.write(entete) self._ecrire_flux(entete) tache_courante = "" echecs: list[dict] = [] proc = subprocess.Popen( ["make", cible, *args], 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): for reste in proc.stdout: journal.write(reste) break finally: rc = proc.wait() 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/contexte": self.repondre_json(200, contexte()) elif chemin == "/api/inventaire": # LE GESTIONNAIRE NE SAIT PLUS LAQUELLE LUI REPOND, et c'est le but : il # demandait « est-ce un site ? » pour choisir entre deux assembleurs, et # les deux ont derive. Il demande maintenant sa charge a sa console. self.repondre_json(200, console(self.inventaire).charge()) elif chemin == "/api/sondes": self.repondre_json(200, sonder_donnees(console(self.inventaire).inventaire_brut())) 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/runbooks": # Import paresseux, comme les devis : un registre casse ne doit pas # emporter le reste de la console avec lui. try: import runbooks as _rb ctx = contexte() _c = console(self.inventaire) _juger = _c.peut_conduire liste = [] for rb in _rb.assembler(RACINE): # CHAQUE ETAPE SE PESE SEULE (2026-09-20, mesure sur TechnoLibre). # Juge au seul runbook, une unique etape qui materialise fermait # toute la sequence a un locataire — qui perdait le droit de # deployer SA flotte. Il la conduit donc, et bute precisement la # ou il faut : sur la machine a engendrer. etapes = [] for etape in rb["etapes"]: ok, pourquoi = _juger(etape["portee"]) etapes.append({**etape, "conduisible": ok, "raison": pourquoi}) ok_rb, pourquoi_rb = _juger(rb["portee"]) liste.append({ **rb, "etapes": etapes, # ON NE CACHE PAS CE QU'ON NE PEUT PAS FAIRE. Un runbook hors # de portee reste VISIBLE et se lit : savoir que le geste # existe, et chez qui il se fait, fait partie du metier. "conduisible": any(e["conduisible"] for e in etapes), "raison": pourquoi_rb if not ok_rb else "", }) self.repondre_json(200, {"runbooks": liste, "contexte": ctx, "ecarts": _rb.verifier(RACINE)}) except Exception as exc: # noqa: BLE001 self.repondre_json(200, {"runbooks": [], "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 # LA GARDE EST ICI, PAS DANS LE NAVIGATEUR. Un bouton grise est une politesse ; # ce qui protege, c'est un serveur qui refuse. Et il DIT pourquoi : « interdit » # sans raison envoie chercher une panne la ou il n'y en a pas. ctx = contexte() besoin = POUVOIR_REQUIS.get(chemin) if besoin == "selon-le-runbook": # LA PORTEE VIENT DU REGISTRE, PAS DU CORPS DE LA REQUETE. Le navigateur nomme # un runbook ; c'est le fichier declare qui dit ce que ce runbook exige. besoin = None try: import runbooks as _rb vise = str(donnees.get("runbook", "")) _i = donnees.get("index") _etape = _rb.etape_declaree(vise, str(donnees.get("cible", "")), RACINE, _i if isinstance(_i, int) else None) # La portee de L'ETAPE. Celle du runbook n'est que son defaut, et juger # sur elle fermerait a un locataire une sequence qui est la sienne. portee_etape = _etape["portee"] if _etape else None connu = any(r["id"] == vise for r in _rb.assembler(RACINE)) except Exception: # noqa: BLE001 portee_etape, connu = None, False if not connu: self.repondre_json(409, {"erreur": f"Runbook inconnu : « {donnees.get('runbook', '')} »."}) return # Etape introuvable a cette place : `executer_runbook` le dira mieux que nous, # avec le nom et la place. On le laisse refuser, sans rien lancer. if portee_etape is not None: ok_etape, pourquoi = console(self.inventaire).peut_conduire(portee_etape) if not ok_etape: self.repondre_json(409, {"erreur": pourquoi, "portee": ctx["portee"]}) return if besoin and not ctx["peut"].get(besoin): self.repondre_json(409, {"erreur": ctx["raison"].get(besoin, "Geste hors de " "la portee de cette console."), "portee": ctx["portee"], "pouvoir": besoin}) 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/runbook-etape": _idx = donnees.get("index") self.executer_runbook(str(donnees.get("runbook", "")), str(donnees.get("cible", "")), donnees.get("valeurs") or {}, donnees.get("vault"), int(_idx) if isinstance(_idx, int) else None) 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())