#!/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" # Intrants de base (panneau GUI). Cibles d'ecriture possedees par le GUI. # Identite : source UNIQUE partagee par tous les envs (symlink depuis chaque # group_vars/all/). Proxmox : fichier lab (le clonage le lit en dur), cles gerees # mises a jour, les autres cles preservees. Les SECRETS n'y entrent jamais. INTRANTS_IDENTITE = INSTANCE / "inventories/partage/intrants-identite.yml" INTRANTS_PROXMOX = INSTANCE / "inventories/lab/group_vars/proxmox.yml" FICHIERS_INTRANTS = {"identite": INTRANTS_IDENTITE, "proxmox": INTRANTS_PROXMOX} # (variable, fichier, classe, section, libelle, type) INTRANTS_SCHEMA = [ ("domaine_interne", "identite", "constante", "Identité", "Domaine DNS interne", "str"), ("chezlepro_timezone", "identite", "defaut", "Identité", "Fuseau horaire", "str"), ("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 = [ "proxmox_api_token_id", "proxmox_api_token_secret", "vault_step_ca_password", "vault_step_ca_fingerprint", "vault_step_ca_provisioner_password", "vault_openldap_admin", "vault_ldap_sssd", "vault_keycloak_admin", "vault_postgresql_keycloak", "vault_bd_keycloak", "vault_forgejo_admin", "vault_forgejo_secret_key", "vault_forgejo_internal_token", "vault_bd_forgejo", "vault_grafana_admin", "vault_redis", "vault_bd_icingadb", ] # 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), "intrants": intrants_api(), "integrations_disponibles": integrations_disponibles(), } 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 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 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") 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/intrants": ecrire_intrants(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())