#!/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 ( 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, reconcilier_serveur, valider_applications, catalogue_liens, valider_bases, valider_domaines, valider_serveurs, ) RACINE = Path(__file__).resolve().parents[1] INSTANCE = Path(os.environ.get("SETOPS_INSTANCE") or (RACINE / "instance")) 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). FICHIERS_INTRANTS = {"identite": INTRANTS_IDENTITE, "proxmox": INTRANTS_PROXMOX, "reseau": FICHIER_NOMENCLATURE} # (variable, fichier, classe, section, libelle, type) INTRANTS_SCHEMA = [ ("index", "reseau", "constante", "Réseau", "Index d'instance dans la fédération (seed : dérive supernet 10.(10+index), VLAN 1000+index×10+zone, VMID)", "int"), ("domaine_interne", "identite", "constante", "Identité", "Domaine DNS interne", "str"), ("fuseau_horaire", "identite", "defaut", "Identité", "Fuseau horaire", "str"), ("organisation", "identite", "defaut", "Identité", "Organisation (annuaire LDAP, certificats)", "str"), ("identite_realm", "identite", "defaut", "Identité", "Realm SSO (Keycloak)", "str"), ("nftables_admin_ssh", "identite", "catalogue", "Sécurité", "Sources SSH d'administration toujours autorisées (garde anti-lockout du pare-feu)", "liste"), ("proxmox_api_host", "proxmox", "constante", "Proxmox", "Hôte API Proxmox", "str"), ("proxmox_api_user", "proxmox", "constante", "Proxmox", "Utilisateur API", "str"), ("proxmox_api_port", "proxmox", "constante", "Proxmox", "Port API", "str"), ("proxmox_validate_certs", "proxmox", "constante", "Proxmox", "Valider les certificats TLS", "bool"), ("proxmox_clone_source_nom", "proxmox", "constante", "Proxmox", "Nom du golden template", "str"), ("proxmox_clone_vmid_modele", "proxmox", "constante", "Proxmox", "VMID du golden template", "int"), ("proxmox_clone_noeud", "proxmox", "defaut", "Proxmox", "Nœud Proxmox (défaut)", "str"), ("proxmox_clone_stockage", "proxmox", "defaut", "Proxmox", "Stockage (défaut)", "str"), ("proxmox_clone_pont", "proxmox", "defaut", "Proxmox", "Pont réseau (défaut)", "str"), ("proxmox_noeuds", "proxmox", "catalogue", "Proxmox", "Nœuds disponibles (liste)", "liste"), ("proxmox_stockages", "proxmox", "catalogue", "Proxmox", "Stockages disponibles (liste)", "liste"), ("proxmox_ponts", "proxmox", "catalogue", "Proxmox", "Ponts réseau disponibles (liste)", "liste"), ] # 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", } # Rappel en lecture seule (jamais de valeur) : secrets a fournir via Ansible Vault. SECRETS_ATTENDUS = [ # Accès Proxmox (proxmox.vault.yml) "proxmox_api_token_id", "proxmox_api_token_secret", # PKI (step-ca) "vault_step_ca_password", "vault_step_ca_provisioner_password", # Identité (LDAP / Keycloak) "vault_openldap_admin", "vault_ldap_sssd", "vault_keycloak_admin", # Bases de données "vault_postgresql_keycloak", "vault_bd_keycloak", "vault_bd_forgejo", "vault_bd_icingadb", "vault_bd_nextcloud", # Forge (Forgejo) "vault_forgejo_admin", "vault_forgejo_secret_key", "vault_forgejo_internal_token", # Observabilité / cache "vault_grafana_admin", "vault_redis", # Applications (admin) "vault_nextcloud_admin", # Secrets OIDC des clients SSO (un par app fédérée) "vault_grafana_oidc", "vault_forgejo_oidc", "vault_icingaweb2_oidc", "vault_nextcloud_oidc", "vault_oauth2_cookie", # Sauvegarde (restic hors-nœud) "vault_restic_password", "vault_backup_ssh_privkey", ] # 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 path.open("w", encoding="utf-8") 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(), "liens_acceptes": catalogue_liens(RACINE), } 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 intrants_courants() -> dict: cache: dict = {} valeurs: dict = {} for variable, fcle, _classe, _section, _libelle, typ in INTRANTS_SCHEMA: if fcle not in cache: cache[fcle] = _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). CHAMPS_ECRITS_PAR_GUI = { "serveurs": {"fonction", "etat", "noeud", "stockage", "disque", "memoire", "coeurs", "integrations"}, "applications": {"groupe", "hote", "port", "requiert", "expose", "liens", "websocket"}, "bases_donnees": {"serveur", "base", "proprietaire", "secret", "consommateur", "portee", "usage"}, "serveurs_bd": {"type", "hote", "port", "groupe"}, "domaines_publics": {"autorite", "edge", "secondaires", "dnssec", "mail"}, # nomenclature : AUCUN champ. Le GUI la LIT (liste des fonctions) sans jamais # l'ecrire — ajouter une fonction exige d'editer le YAML. Trou connu et assume ; # `couverture_gui.py` le signale tant qu'il dure. "nomenclature": set(), } def integrations_disponibles() -> list: """Roles client_* presents dans roles/ : source des integrations selectionnables.""" base = RACINE / "roles" return sorted(p.name for p in base.glob("client_*") if p.is_dir()) def intrants_api() -> dict: schema = [{"variable": v, "fichier": f, "classe": c, "section": s, "libelle": l, "type": t} for (v, f, c, s, l, t) 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 def _ecrire_intrants_fichier(path: Path, valeurs: dict, entete: str) -> None: interdits = sorted(set(valeurs) & INTRANTS_CLES_INTERDITES) if interdits: raise ValueError(f"Refus d'ecrire des cles sensibles via le GUI: {', '.join(interdits)}") path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", encoding="utf-8") 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 {} 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) in INTRANTS_SCHEMA} fichier_de = {v: f for (v, f, c, s, l, t) 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: _ecrire_intrants_fichier( INTRANTS_IDENTITE, par_fichier["identite"], "# 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 vont dans proxmox.vault.yml).\n" "# Cles d'intrants editees par le panneau « Intrants de base » du GUI.\n---\n") if "reseau" in par_fichier and "index" in par_fichier["reseau"]: _ecrire_index_nomenclature(int(par_fichier["reseau"]["index"])) 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 path.write_text(nouveau, encoding="utf-8") 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" ) contenu = { "serveurs_bd": registre.get("serveurs_bd", {}) or {}, "bases_donnees": registre.get("bases_donnees", {}) or {}, } with path.open("w", encoding="utf-8") as fichier: fichier.write(entete) yaml.safe_dump(contenu, fichier, default_flow_style=False, sort_keys=False, allow_unicode=True) 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" ) with path.open("w", encoding="utf-8") as fichier: fichier.write(entete) yaml.safe_dump({"applications": registre.get("applications", {}) or {}}, fichier, default_flow_style=False, sort_keys=False, allow_unicode=True) 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" ) with path.open("w", encoding="utf-8") as fichier: fichier.write(entete) yaml.safe_dump({"serveurs": registre.get("serveurs", {}) or {}}, fichier, default_flow_style=False, sort_keys=False, allow_unicode=True) 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" ) with path.open("w", encoding="utf-8") as fichier: fichier.write(entete) yaml.safe_dump({"domaines_publics": registre.get("domaines_publics", {}) or {}}, fichier, default_flow_style=False, sort_keys=False, allow_unicode=True) 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: 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/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/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())