#!/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 subprocess import tempfile import threading from pathlib import Path from urllib.parse import urlparse import yaml 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, valider_bases, valider_serveurs, ) RACINE = Path(__file__).resolve().parents[1] INSTANCE = Path(os.environ.get("SETOPS_INSTANCE") or (RACINE / "instance")) INVENTAIRE_DEFAUT = INSTANCE / "inventories/production/hosts.yml" INVENTAIRE_PRODUCTION = (INSTANCE / "inventories/production/hosts.yml").resolve() 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" # 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 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 inventaire_api(path: Path) -> dict: data = charger_yaml(path) dependencies = charger_dependances(FICHIER_DEPENDANCES) return { "inventaire": os.path.relpath(path, RACINE), "production": path.resolve() == INVENTAIRE_PRODUCTION, "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), } 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" "---\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 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
""" class Gestionnaire(BaseHTTPRequestHandler): inventaire: Path = INVENTAIRE_DEFAUT 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 if self.inventaire != INVENTAIRE_PRODUCTION: self.repondre_json(409, {"erreur": "Deploiement reserve a l'inventaire production."}) return if 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 try: cible = "deployer" if mode == "deployer" else "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() self._ecrire_flux(f"$ make {cible} HOTE={hote}\n\n") 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: if not self._ecrire_flux(ligne): proc.terminate() break finally: rc = proc.wait() if mode == "verifier" and rc == 0: VERIF_OK[hote] = True elif mode == "deployer": VERIF_OK.pop(hote, None) self._ecrire_flux(f"\n__FIN__ rc={rc}\n") finally: 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)) 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)) 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/instancier": resultat = subprocess.run( ["python3", str(RACINE / "scripts/instancier.py"), "appliquer"], 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")) 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() Gestionnaire.inventaire = args.inventaire.resolve() 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())