#!/usr/bin/env python3 """Generateur d'inventaire Set-OPS (Phase 3) : plan -> hosts.yml. NON destructif : ecrit hosts.genere.yml a cote de l'inventaire de l'instance et compare SEMANTIQUEMENT (via ansible-inventory --list) avec l'inventaire actuel. Aucune bascule tant que la comparaison n'est pas vide et validee. Derivation : - host vars : ansible_host/ansible_user + proxmox_* (IP/VMID/VLAN/passerelle derives de la nomenclature ; placement/taille depuis instance/plan/serveurs.yml) ; - groupes : socle (serveur_debian/durcis) + groupes de service derives des applications de l'hote + groupe d'etat (hotes_actifs/hotes_planifies). - Les integrations clients_* ne sont PAS derivables du plan a ce stade : la comparaison les revele (lacune a combler en Phase 3b). """ from __future__ import annotations import os import json import subprocess import sys from pathlib import Path import yaml from inventory_rules import ( charger_applications, charger_domaines, charger_nomenclature, charger_serveurs, deriver_nomenclature, deriver_ressources, expositions_des_applications, fonction_seq, liens_acceptes, ) RACINE = Path(__file__).resolve().parents[1] ROLES = RACINE / "roles" INSTANCE = Path(os.environ.get("SETOPS_INSTANCE") or (RACINE / "instance")) def _inventaire(instance: Path, *noms: str) -> Path: for nom in noms: p = instance / "inventories" / nom / "hosts.yml" if p.exists(): return p # Aucun hosts.yml encore genere : viser le repertoire d'inventaire deja present # (ses group_vars y sont), sinon le premier nom par defaut. Evite d'ecrire dans # 'principal/' alors que l'instance (ex. modele socle) est en 'production/'. for nom in noms: if (instance / "inventories" / nom).is_dir(): return instance / "inventories" / nom / "hosts.yml" return instance / "inventories" / noms[0] / "hosts.yml" # Inventaire unique de l'instance (rétro-compat : principal > production). # SETOPS_INVENTAIRE force la cible. _force = os.environ.get("SETOPS_INVENTAIRE") INVENTAIRE = Path(_force) if _force else _inventaire(INSTANCE, "principal", "production") GENERE = INVENTAIRE.with_name("hosts.genere.yml") FICHIER_SERVEURS = INSTANCE / "plan/serveurs.yml" FICHIER_APPLICATIONS = INSTANCE / "plan/applications.yml" FICHIER_NOMENCLATURE = INSTANCE / "plan/nomenclature.yml" FICHIER_DOMAINES = INSTANCE / "plan/domaines.yml" GROUPES_SOCLE = ["serveur_debian", "serveur_durci"] PLACEMENT = [("noeud", "proxmox_noeud"), ("stockage", "proxmox_stockage"), ("disque", "proxmox_disque_taille"), ("memoire", "proxmox_memoire"), ("coeurs", "proxmox_coeurs")] # --- Bindings (liens) : relations declaratives entre applications ----------------- # Une application declare ses 'liens' ([{vers, role}]) dans plan/applications.yml ; # chaque role decrit les liens qu'il accepte dans roles//meta/liens.yml # (setops_liens.accepte). instancier resout la cible (FQDN interne derive) et injecte # les variables en host_vars du consommateur. Cf. docs/bindings-conception.md. def _domaine_interne() -> str: """Lit domaine_interne dans group_vars/all de l'instance (keystone, non chiffre).""" base = INVENTAIRE.parent / "group_vars" / "all" if base.is_dir(): for fichier in sorted(base.glob("*.yml")): try: data = yaml.safe_load(fichier.read_text(encoding="utf-8")) or {} except yaml.YAMLError: continue if isinstance(data, dict) and str(data.get("domaine_interne", "")).strip(): return str(data["domaine_interne"]).strip() return "" def _attributs_cible(vers: str, apps: dict, serveurs: dict, nomenclature: dict, domaine: str) -> dict | None: """Resout une cible de lien en attributs substituables. Phase 1 : applications.""" if vers in apps: hote = apps[vers].get("hote") srv = serveurs.get(hote, {}) _, seq = fonction_seq(str(hote)) d = deriver_nomenclature(str(srv.get("fonction", "")), seq, nomenclature) or {} return { "genre": "application", "hote": hote, "fqdn": f"{hote}.{domaine}" if domaine else str(hote), "ip": d.get("adresse_ip"), } return None def _substituer(valeur, attrs: dict): """Remplace {cible.X} dans une chaine ; laisse bool/int tels quels.""" if not isinstance(valeur, str): return valeur for cle, val in attrs.items(): valeur = valeur.replace("{cible." + cle + "}", str(val)) return valeur def resoudre_liens(apps: dict, serveurs: dict, nomenclature: dict) -> dict: """Derive {hote: {var: valeur}} depuis les 'liens' des applications (bindings).""" domaine = _domaine_interne() injections: dict = {} for nom_app, app in apps.items(): accepte = None for lien in (app.get("liens") or []): vers, role = lien.get("vers"), lien.get("role") if accepte is None: accepte = liens_acceptes(str(app.get("groupe", "")), RACINE) spec = accepte.get(role) if not spec: raise ValueError( f"application '{nom_app}': lien role '{role}' non accepte par " f"'{app.get('groupe')}' (roles/{app.get('groupe')}/meta/liens.yml).") attrs = _attributs_cible(str(vers), apps, serveurs, nomenclature, domaine) if attrs is None: raise ValueError(f"application '{nom_app}': cible de lien '{vers}' introuvable.") attendu = spec.get("cible") if attendu and attendu != attrs["genre"]: raise ValueError( f"application '{nom_app}': lien '{role}' attend une cible '{attendu}', " f"'{vers}' est '{attrs['genre']}'.") sur_conso = spec.get("injecte_sur", "consommateur") == "consommateur" cible_hote = app.get("hote") if sur_conso else attrs["hote"] for var, gabarit in (spec.get("variables") or {}).items(): injections.setdefault(cible_hote, {})[var] = _substituer(gabarit, attrs) return injections def generer() -> dict: serveurs = charger_serveurs(FICHIER_SERVEURS).get("serveurs", {}) apps = charger_applications(FICHIER_APPLICATIONS).get("applications", {}) nomenclature = charger_nomenclature(FICHIER_NOMENCLATURE) injections = resoudre_liens(apps, serveurs, nomenclature) # Expositions du plan (champ 'expose') -> pour deriver les SANs de chaque edge. domaines = charger_domaines(FICHIER_DOMAINES) expositions = expositions_des_applications({"applications": apps}, domaines) services_par_hote: dict = {} for app in apps.values(): services_par_hote.setdefault(app.get("hote"), set()).add(app.get("groupe")) children: dict = { "modeles_vm": {"hosts": {}}, "hotes_actifs": {"hosts": {}}, "hotes_planifies": {"hosts": {}}, } for nom, srv in serveurs.items(): _, seq = fonction_seq(nom) d = deriver_nomenclature(str(srv.get("fonction", "")), seq, nomenclature) or {} hostvars = { "ansible_host": d.get("adresse_ip"), "ansible_user": "ansible", "proxmox_cidr": d.get("cidr"), "proxmox_passerelle": d.get("passerelle"), "proxmox_vlan": d.get("vlan"), "proxmox_vmid": int(d["vmid"]) if d.get("vmid") else None, } for cle_srv, cle_var in PLACEMENT: if str(srv.get(cle_srv, "")).strip(): hostvars[cle_var] = srv[cle_srv] # Ressources derivees des logiciels (groupes de service) + socle SE. # setdefault : une valeur explicite du plan (PLACEMENT ci-dessus) gagne. ressources = deriver_ressources(sorted(services_par_hote.get(nom, set())), ROLES) hostvars.setdefault("proxmox_coeurs", ressources["coeurs"]) hostvars.setdefault("proxmox_memoire", ressources["memoire"]) hostvars.setdefault("proxmox_disque_taille", ressources["disque"]) # Variables derivees des liens (bindings) declares par les applications. for var, val in injections.get(nom, {}).items(): hostvars[var] = val groupes = set(GROUPES_SOCLE) | services_par_hote.get(nom, set()) | set(srv.get("integrations") or []) # SANs d'exposition : les FQDN que cet edge sert (derives du plan) -> cert edge # (client_pki_sans), au lieu d'une liste manuelle. Vide si l'hote n'est pas un edge. sans = sorted({e["fqdn"] for e in expositions if e.get("edge") in groupes}) if sans: hostvars["sans_exposition"] = sans etat = "hotes_actifs" if srv.get("etat") == "actif" else "hotes_planifies" children[etat]["hosts"][nom] = hostvars for groupe in sorted(groupes): children.setdefault(groupe, {"hosts": {}})["hosts"][nom] = None return {"all": {"children": children}} def ecrire(path: Path = GENERE) -> None: entete = ("# Inventaire GENERE depuis le plan (make instancier / instancier-appliquer).\n" "# NE PAS editer a la main : edite instance/plan/serveurs.yml + instance/plan/applications.yml.\n" "# Source : instance/plan/serveurs.yml + instance/plan/applications.yml + instance/plan/nomenclature.yml.\n") with path.open("w", encoding="utf-8") as fichier: fichier.write(entete) yaml.safe_dump(generer(), fichier, default_flow_style=False, sort_keys=True, allow_unicode=True) def _resolu(fichier: Path) -> tuple[dict, dict]: cmd = ["ansible-inventory", "-i", str(fichier), "--list"] # Une voute chiffree (group_vars/all/vault.yml) fait echouer ansible-inventory sans # mot de passe (exit 4). Si ANSIBLE_VAULT_PASSWORD_FILE n'est pas deja fourni, on # tente le fichier conventionnel Set-OPS ~/.config/setops-vault-pass. if not os.environ.get("ANSIBLE_VAULT_PASSWORD_FILE"): conv = Path.home() / ".config" / "setops-vault-pass" if conv.is_file(): cmd += ["--vault-password-file", str(conv)] sortie = subprocess.run(cmd, capture_output=True, text=True, check=True).stdout data = json.loads(sortie) hostvars = data.get("_meta", {}).get("hostvars", {}) groupes: dict = {} for groupe, info in data.items(): if groupe in ("_meta", "all", "ungrouped") or not isinstance(info, dict): continue for hote in (info.get("hosts") or []): groupes.setdefault(hote, set()).add(groupe) return hostvars, groupes def compter_ecarts(reference: Path, genere: Path) -> int: hv_act, grp_act = _resolu(reference) hv_gen, grp_gen = _resolu(genere) hotes = sorted(set(hv_act) | set(hv_gen)) ecarts = 0 for hote in hotes: if hote not in hv_gen: print(f" - {hote} : ABSENT du genere"); ecarts += 1; continue if hote not in hv_act: print(f" - {hote} : EN TROP dans le genere"); ecarts += 1; continue g_manquants = grp_act.get(hote, set()) - grp_gen.get(hote, set()) g_surplus = grp_gen.get(hote, set()) - grp_act.get(hote, set()) vars_diff = [k for k in set(hv_act[hote]) | set(hv_gen[hote]) if hv_act[hote].get(k) != hv_gen[hote].get(k)] if g_manquants or g_surplus or vars_diff: ecarts += 1 print(f" - {hote} :") if g_manquants: print(f" groupes non reproduits : {', '.join(sorted(g_manquants))}") if g_surplus: print(f" groupes en trop : {', '.join(sorted(g_surplus))}") if vars_diff: details = ", ".join(f"{k} (inv={hv_act[hote].get(k)!r} gen={hv_gen[hote].get(k)!r})" for k in sorted(vars_diff)) print(f" vars differentes : {details}") return ecarts def comparer() -> int: ecrire(GENERE) ecarts = compter_ecarts(INVENTAIRE, GENERE) if ecarts == 0: print("DIFF VIDE : le plan reproduit exactement l'inventaire actuel. Bascule possible.") else: print(f"\n{ecarts} hote(s) avec ecart.") return 0 def appliquer(force: bool = False) -> int: """Ecrit hosts.yml depuis le plan. Refuse si le diff n'est pas vide (sauf --force).""" ecrire(GENERE) ecarts = compter_ecarts(INVENTAIRE, GENERE) if ecarts and not force: print(f"\nDiff NON vide ({ecarts} hote(s)). Revois 'make instancier'. " "Utilise FORCE=1 pour appliquer un changement intentionnel du plan.") return 1 ecrire(INVENTAIRE) print(f"\n{os.path.relpath(INVENTAIRE, RACINE)} (re)genere depuis le plan. " "git est le filet : git diff / git checkout pour revenir.") return 0 def main() -> int: import argparse parser = argparse.ArgumentParser(description="Generateur d'inventaire Set-OPS (Phase 3).") sub = parser.add_subparsers(dest="commande", required=True) sub.add_parser("generer", help="Ecrit hosts.genere.yml depuis le plan.") sub.add_parser("comparer", help="Compare le genere a l'inventaire actuel (semantique).") pa = sub.add_parser("appliquer", help="Ecrit hosts.yml depuis le plan (refuse si diff non vide).") pa.add_argument("--force", action="store_true", help="Applique meme si le diff n'est pas vide.") args = parser.parse_args() try: if args.commande == "generer": ecrire() print(f"{os.path.relpath(GENERE, RACINE)} genere depuis le plan.") elif args.commande == "comparer": return comparer() elif args.commande == "appliquer": return appliquer(force=args.force) except Exception as exc: print(f"erreur: {exc}", file=sys.stderr) return 2 return 0 if __name__ == "__main__": raise SystemExit(main())