#!/usr/bin/env python3 """Gere les hotes d'un inventaire Ansible statique Set-OPS.""" from __future__ import annotations import argparse from pathlib import Path import sys import yaml from inventory_rules import ( ecriture_atomique, variable_effective, GROUPE_HOTES_ACTIFS, GROUPE_HOTES_PLANIFIES, GROUPES_ETAT_HOTE, charger_dependances, est_groupe_operationnel, ) def charger_inventaire(path: Path) -> dict: if not path.exists(): return {"all": {"children": {}}} with path.open("r", encoding="utf-8") as inventory_file: data = yaml.safe_load(inventory_file) 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_inventaire(path: Path, data: dict) -> None: with ecriture_atomique(path) as inventory_file: yaml.safe_dump( data, inventory_file, default_flow_style=False, sort_keys=False, allow_unicode=True, ) def enfants(data: dict) -> dict: return data.setdefault("all", {}).setdefault("children", {}) def assurer_groupe(data: dict, group: str) -> dict: group_data = enfants(data).setdefault(group, {}) group_data.setdefault("hosts", {}) return group_data def variables_hote(data: dict, host: str) -> dict: merged: dict = {} for group_data in enfants(data).values(): hosts = group_data.get("hosts", {}) if host in hosts and isinstance(hosts[host], dict): merged.update(hosts[host]) return merged def tous_les_hotes(data: dict) -> list[str]: hosts: set[str] = set() for group_data in enfants(data).values(): hosts.update(group_data.get("hosts", {})) return sorted(hosts) def groupes_hote(data: dict, host: str) -> list[str]: groups: list[str] = [] for group, group_data in enfants(data).items(): if host in group_data.get("hosts", {}): groups.append(group) return groups def retirer_hote_de_tous_les_groupes(data: dict, host: str) -> None: for group_data in enfants(data).values(): group_data.get("hosts", {}).pop(host, None) def ajouter_hote( data: dict, host: str, groups: list[str], ip: str | None, ansible_user: str | None, vmid: str | None, ) -> None: vars_for_host = variables_hote(data, host) if ip: vars_for_host["ansible_host"] = ip if ansible_user: vars_for_host["ansible_user"] = ansible_user if vmid: vars_for_host["proxmox_vmid"] = int(vmid) if vmid.isdigit() else vmid for group in groups: assurer_groupe(data, group)["hosts"][host] = dict(vars_for_host) if ip: assurer_groupe(data, GROUPE_HOTES_ACTIFS)["hosts"][host] = dict(vars_for_host) assurer_groupe(data, GROUPE_HOTES_PLANIFIES)["hosts"].pop(host, None) else: assurer_groupe(data, GROUPE_HOTES_PLANIFIES)["hosts"][host] = dict(vars_for_host) assurer_groupe(data, GROUPE_HOTES_ACTIFS)["hosts"].pop(host, None) def definir_groupes_hote(data: dict, host: str, groups: list[str]) -> None: vars_for_host = variables_hote(data, host) current_groups = groupes_hote(data, host) if not current_groups: raise ValueError(f"Hote absent de l'inventaire: {host}") retirer_hote_de_tous_les_groupes(data, host) for group in current_groups: if group in GROUPES_ETAT_HOTE: assurer_groupe(data, group)["hosts"][host] = dict(vars_for_host) for group in groups: assurer_groupe(data, group)["hosts"][host] = dict(vars_for_host) def analyser_groupes(raw_groups: str) -> list[str]: groups = [group.strip() for group in raw_groups.replace(",", " ").split() if group.strip()] if not groups: raise ValueError("Au moins un groupe est requis.") return groups def afficher_hote(data: dict, host: str) -> None: groups = groupes_hote(data, host) if not groups: raise ValueError(f"Hote absent de l'inventaire: {host}") yaml.safe_dump( { "hote": host, "groupes": groups, "variables": variables_hote(data, host), }, sys.stdout, default_flow_style=False, sort_keys=False, allow_unicode=True, ) CHAMPS_PROXMOX: tuple[tuple, ...] = ( ("SETOPS_VMID", "proxmox_vmid", True), ("SETOPS_IP", "ansible_host", True), ("SETOPS_CIDR", "proxmox_cidr", True), ("SETOPS_PASSERELLE", "proxmox_passerelle", True), # L'etiquette VLAN a poser sur la carte. En SDN elle est VIDE : le VNet la porte # deja, et en poser une seconde donnerait un double etiquetage. `proxmox_vlan` # sert de repli pour un inventaire pas encore regenere — mais SEULEMENT si la cle # est ABSENTE, jamais si elle est presente et vide : c'est la difference entre # « on ne sait pas » et « on a decide qu'il n'y en a pas ». ("SETOPS_VLAN", "proxmox_etiquette_vlan", False, "proxmox_vlan"), ("SETOPS_PONT", "proxmox_pont", False), ("SETOPS_STOCKAGE", "proxmox_stockage", False), ("SETOPS_DISQUE", "proxmox_disque_taille", False), ("SETOPS_NOEUD", "proxmox_noeud", False), ("SETOPS_COEURS", "proxmox_coeurs", False), ("SETOPS_MEMOIRE", "proxmox_memoire", False), # LE PLANCHER : ce que l'hote a le droit de REPRENDRE a cette machine (`balloon`). # # OUBLIE ICI LE 2026-09-13, ET C'EST LE TROISIEME MAILLON. Le moteur derivait bien # `proxmox_memoire_min` dans l'inventaire, le playbook de clonage savait bien poser # `balloon:` — et entre les deux, cette table ne transmettait rien. Quatorze machines # allaient naitre sans plancher, donc avec le ballooning DESACTIVE, sans qu'aucune # etape n'echoue : `proxmox_clone_memoire_min` serait simplement reste indefini et sa # garde `when:` aurait saute la tache en silence. # # C'est le patron d'une liste qui en suit une autre et prend du retard. La regle qui en # sort : quand une variable traverse trois modules, la troisieme s'ecrit EN MEME TEMPS # que la premiere, pas quand on s'en sert. ("SETOPS_MEMOIRE_MIN", "proxmox_memoire_min", False), # Resolveur d'AMORCAGE, pose par cloud-init. Il ne sert qu'une fois : `client_resolveur` # bascule ensuite `/etc/resolv.conf` vers 127.0.0.1. Mais sans lui, la VM nait sans # resolution de noms et `apt` ne peut meme pas installer Unbound — l'amorcage # n'aboutit jamais. Le DNS autoritatif du tenant ne convient PAS : il refuse tout ce # qui sort de la zone souveraine. ("SETOPS_DNS", "dns_amorcage", False), # Domaine de RECHERCHE de cloud-init. Sans lui, le clone herite de celui du gabarit — # `chezlepro.ca`, le domaine PUBLIC du reseau ou l'image a ete fabriquee — alors que # les machines vivent dans `chezlepro.internal`. Mesure du 2026-08-09 sur les 14 VM. # Sans consequence tant que `/etc/hosts` couvre les noms courts et que la zone # publique n'a pas de joker ; les deux tiennent aujourd'hui, aucun des deux n'est # garanti. Un nom court resolu vers une adresse publique serait silencieux et faux. ("SETOPS_DOMAINE", "domaine_interne", False), ) def _intrants_communs(inventaire: Path | None) -> dict: """Intrants de `group_vars/all/` voisins de CET inventaire, visibles au CLONAGE. L'inventaire genere ne porte que des variables d'HOTE ; un intrant commun comme `dns_amorcage` y est donc invisible. Le clonage tourne sur localhost, avant qu'Ansible ne resolve les group_vars de la cible : sans cette lecture, le parametre serait lu par les roles mais jamais par la machine qui cree la VM. Le chemin DERIVE de l'inventaire recu, jamais code en dur : lire `instance/` pendant qu'on raisonne sur un autre inventaire ferait dependre le resultat de l'environnement — et un test sur inventaire synthetique irait chercher les intrants de la production. """ if inventaire is None: return {} dossier = Path(inventaire).resolve().parent / "group_vars" / "all" if not dossier.is_dir(): return {} fusion: dict = {} for f in sorted(dossier.glob("*.yml")): try: contenu = yaml.safe_load(f.read_text(encoding="utf-8")) except (OSError, yaml.YAMLError): continue # voute chiffree ou fichier illisible : sans objet ici if isinstance(contenu, dict): fusion.update(contenu) return fusion def parametres_proxmox_hote(data: dict, host: str, inventaire: Path | None = None) -> list[str]: """Emet les parametres de clonage Proxmox d'un hote, lus depuis l'inventaire. Sortie shell-evaluable (`SETOPS_*='valeur'`) consommee par `make creer-vm`. L'inventaire genere est la seule source : aucun parametre n'est saisi a la main. """ if not groupes_hote(data, host): raise ValueError( f"Hote absent de l'inventaire: {host}. " "Declare-le dans le plan (instance/plan/serveurs.yml) puis " "`make instancier-appliquer`." ) vars_for_host = _intrants_communs(inventaire) | variables_hote(data, host) lignes: list[str] = [] manquants: list[str] = [] for champ in CHAMPS_PROXMOX: var_shell, cle, requis = champ[0], champ[1], champ[2] repli = champ[3] if len(champ) > 3 else None if cle in vars_for_host: valeur = vars_for_host.get(cle) elif repli: valeur = vars_for_host.get(repli) else: valeur = None if valeur in (None, ""): if requis: manquants.append(cle) valeur = "" valeur = str(valeur) if "'" in valeur: raise ValueError( f"Valeur Proxmox inattendue pour {cle} ({host}): apostrophe interdite." ) lignes.append(f"{var_shell}='{valeur}'") if manquants: raise ValueError( f"Parametres Proxmox manquants pour {host}: {', '.join(manquants)}. " "Regenere l'inventaire (`make instancier-appliquer`)." ) # LA CLE D'AMORCAGE — POSEE SUR LE RUNNER D'UN TENANT, ET SUR LUI SEUL. # # Le runner du SITE doit pouvoir entrer chez le runner du tenant pour l'amorcer : # c'est l'insemination, et son flux est declare dans # `roles/serveur_ops_site/meta/flux.yml`. Il lui faut donc une identite sur cette # machine — une cle publique, qui n'ouvre rien par elle-meme. # # LE CRITERE EST LE MEME QUE CELUI DU FLUX : porter `serveur_ops_tenant`. C'est ce # qui empeche le geste de deraper. `creer-vm` cree TOUTES les machines d'un tenant ; # injecter cette cle a chaque clonage donnerait au site un acces SSH a la flotte # entiere de chaque locataire, sans qu'aucune regle ne le dise — et ca deferait a la # couche IDENTITE ce que le pare-feu borne a la couche RESEAU. Deux couches qui ne # declarent pas la meme politique, c'est une politique qu'on ne peut plus lire. # # La cle vient du PLAN DU SITE, pas du disque local : on materialise depuis le runner # comme depuis le poste de l'exploitant, et un `lookup` local rendrait deux valeurs # differentes selon qui agit — la VM nee du poste n'accepterait pas le runner. lignes.append(f"SETOPS_CLES_AMORCAGE='{_cle_amorcage(data, host, inventaire)}'") return lignes GROUPE_RUNNER_TENANT = "serveur_ops_tenant" def _cle_amorcage(data: dict, host: str, inventaire: Path | None = None) -> str: """Les cles publiques a poser sur cet hote A SA NAISSANCE. Une par ligne. DEUX CLES, DEUX PORTEES, ET LA DIFFERENCE EST TOUTE L'ARCHITECTURE : la cle du SITE -> sur le SEUL runner du tenant. C'est l'insemination : le site amorce une machine, et une seule, puis s'arrete. la cle du TENANT -> sur TOUTES les machines de ce tenant. C'est son runner qui les configurera, et il ne peut pas le faire sans y entrer. POSER UNE CLE AU CLONAGE N'EST PAS ENTRER CHEZ LE TENANT. C'est un parametre de creation, au meme titre que l'adresse ou le disque : le site ecrit les conditions de NAISSANCE de la machine, il n'ouvre aucune session dedans. La distinction n'est pas rhetorique — le site n'obtient aucun acces sur ces machines, seul le runner du tenant en obtient un. SANS ELLE, LE TERRAIN EST INOCCUPABLE (mesure du 2026-08-30). Quinze VM creees, et le runner du tenant se voyait refuser l'entree sur les quinze : ops-01 -> toutes : Permission denied (publickey) Sa cle EST declaree au plan (`ssh_baseline_cles_admin`), mais c'est le SOCLE qui la depose — et le socle doit etre applique par quelqu'un qui peut deja entrer. Boucle fermee : le tenant recevait un terrain qu'il ne pouvait pas occuper. Rend '' pour ce qui n'est pas trouve, jamais une valeur devinee : un ecosysteme dont le plan ne declare rien se materialise comme avant, avec la seule cle de l'exploitant. """ cles: list[str] = [] # 1. LA CLE DU SITE — le runner du tenant, et lui seul. if GROUPE_RUNNER_TENANT in groupes_hote(data, host): try: import underlay as underlay_mod c = str((underlay_mod.lire_plan_site("10-intrants.yml") or {}) .get("runner_cle_publique") or "").strip() if c: cles.append(c) except Exception: pass # 2. LA CLE DU TENANT — toutes ses machines, y compris son runner (il se joint # lui-meme comme les autres, et l'exclure creerait un cas particulier de plus). # # ELLE VIENT DU PLAN DU TENANT, pas d'un fichier lu sur une machine : materialiser # depuis le poste et depuis le runner du site doit produire la meme VM. for entree in _cles_admin_du_plan(inventaire): if entree not in cles: cles.append(entree) for c in cles: if "'" in c: raise ValueError( f"Une cle publique declaree contient une apostrophe ({host}) : " "la valeur passe par une commande shell, elle ne peut pas en porter." ) return "\n".join(cles) def _cles_admin_du_plan(inventaire: Path | None) -> list[str]: """Les cles `ssh_baseline_cles_admin` de l'instance, celles a l'etat `present`. ON HONORE L'ETAT. Une entree passee a `absent` est une REVOCATION : la reposer a la naissance ressusciterait sur chaque VM neuve une cle qu'on vient de retirer de toute la flotte — la panne serait lente, silencieuse, et parfaitement invisible au plan. """ try: depuis = Path(inventaire).parent if inventaire else None gv = (depuis or _instance_dir()) / "group_vars" / "all" for fichier in sorted(gv.glob("*.yml")): if "vault" in fichier.name: continue data = yaml.safe_load(fichier.read_text(encoding="utf-8")) or {} if isinstance(data, dict) and data.get("ssh_baseline_cles_admin"): return [str(e.get("cle", "")).strip() for e in data["ssh_baseline_cles_admin"] if isinstance(e, dict) and str(e.get("etat", "present")) == "present" and str(e.get("cle", "")).strip()] except Exception: pass return [] def _instance_dir() -> Path: from inventory_rules import instance_courante return Path(instance_courante()) def _rang_des_couches() -> dict[str, int]: """Rang de chaque groupe dans `docs/couches-deploiement.yml`. {} si le fichier manque. SOURCE UNIQUE avec `make site` : l'ordre d'un hote seul et celui de la reconstruction complete doivent decouler du meme registre. Les tenir separement, c'est se garantir qu'ils divergeront — et la divergence ne se voit qu'au deploiement, trop tard. """ fichier = Path(__file__).resolve().parent.parent / "docs" / "couches-deploiement.yml" if not fichier.is_file(): return {} try: contenu = yaml.safe_load(fichier.read_text(encoding="utf-8")) or {} except (OSError, yaml.YAMLError): return {} couches = contenu.get("couches") if isinstance(contenu, dict) else contenu rangs: dict[str, int] = {} for rang, couche in enumerate(couches or []): for groupe in (couche or {}).get("groupes") or []: rangs.setdefault(str(groupe), rang) return rangs def afficher_playbooks_hote(data: dict, host: str, playbook_dir: Path) -> None: groups = groupes_hote(data, host) if not groups: raise ValueError(f"Hote absent de l'inventaire: {host}") # Le SOCLE passe toujours en premier : il pose /etc/hosts (resolution interne), # le durcissement, etc. Les integrations client_* et les services en dependent # (ex. client_pki doit resoudre l'AC via /etc/hosts avant de bootstrapper). # # ENSUITE, l'ordre des COUCHES (docs/couches-deploiement.yml) — le meme registre que # `make site`. Trier alphabetiquement mettait `client_metrique` avant # `serveur_step_ca` : l'integration exigeait un certificat que l'autorite, pas encore # deployee, ne pouvait pas avoir emis. Constate le 2026-08-06 sur les deux premieres # VM, dont l'hote de l'AC lui-meme. La couche « integrations » dit en toutes lettres # « deployees en dernier, quand leurs cibles sont debout » : `make deployer` l'ignorait. socle = ["serveur_debian", "serveur_durci"] couches = _rang_des_couches() groups = sorted(groups, key=lambda g: ( socle.index(g) if g in socle else len(socle), couches.get(g, len(couches)), # groupe hors registre : apres les couches connues g, )) missing_playbooks: list[str] = [] playbooks: list[Path] = [] for group in groups: if not est_groupe_operationnel(group): continue playbook = playbook_dir / f"{group}.yml" if playbook.exists(): playbooks.append(playbook) else: missing_playbooks.append(group) if missing_playbooks: missing = ", ".join(missing_playbooks) raise ValueError(f"Playbook de groupe manquant pour: {missing}") if not playbooks: raise ValueError(f"Aucun playbook de groupe opérationnel pour: {host}") for playbook in playbooks: print(playbook) def verifier_hote_actif(data: dict, host: str) -> None: hosts = enfants(data).get(GROUPE_HOTES_ACTIFS, {}).get("hosts", {}) if host not in hosts: raise ValueError(f"Hote non actif dans l'inventaire: {host}") def groupe_a_hote_actif(data: dict, group: str) -> bool: active_hosts = set(enfants(data).get(GROUPE_HOTES_ACTIFS, {}).get("hosts", {})) group_hosts = set(enfants(data).get(group, {}).get("hosts", {})) return bool(active_hosts & group_hosts) def _exigence_levee(config: dict, required_group: str, group_vars: Path | None) -> bool: """La clause `sauf_si` retire-t-elle cette exigence pour CETTE instance ? Une exigence n'est pas toujours absolue (2026-08-22) : `serveur_forgejo` exige un serveur PostgreSQL — sauf quand l'instance a choisi `serveur_forgejo_bd: sqlite`, et sa base tient alors dans un fichier. Sans cette lecture, un ecosysteme parfaitement coherent se voyait refuser le deploiement au nom d'un service dont il n'a pas besoin. La condition est lue dans les group_vars de l'instance, a defaut dans le defaut du role : meme resolution que partout ailleurs (`inventory_rules.variable_effective`). """ clause = (config.get("sauf_si") or {}).get(required_group) if not clause: return False valeur = variable_effective(str(clause.get("variable")), group_vars, role_defaut=str(clause.get("role") or "")) return valeur is not None and valeur == str(clause.get("vaut")) def verifier_dependances_groupes(data: dict, dependencies: dict, groups: list[str], group_vars: Path | None = None) -> None: missing: list[str] = [] for group in sorted(set(groups)): config = dependencies.get(group, {}) for required_group in config.get("requiert_groupes_actifs", []): if _exigence_levee(config, required_group, group_vars): continue if not groupe_a_hote_actif(data, required_group): missing.append(f"{group} requiert {required_group} actif") # `utilise_si_present` ne bloque JAMAIS : c'est un agrement, pas une exigence. # Les confondre obligeait une forge a deployer une pile courriel pour exister. if missing: raise ValueError("Dependances manquantes: " + "; ".join(missing)) def exiger_dependances(dependencies: dict, path: Path | None) -> None: if not path: raise ValueError("Fichier de dependances requis: utiliser --dependances.") if not dependencies: raise ValueError(f"Aucune dependance chargee depuis {path}.") def verifier_dependances_hote(data: dict, dependencies: dict, host: str, group_vars: Path | None = None) -> None: groups = [group for group in groupes_hote(data, host) if est_groupe_operationnel(group)] verifier_dependances_groupes(data, dependencies, groups, group_vars) def verifier_dependances_actives(data: dict, dependencies: dict, group_vars: Path | None = None) -> None: active_hosts = set(enfants(data).get(GROUPE_HOTES_ACTIFS, {}).get("hosts", {})) active_groups: set[str] = set() for group, group_data in enfants(data).items(): if not est_groupe_operationnel(group): continue if active_hosts & set(group_data.get("hosts", {})): active_groups.add(group) verifier_dependances_groupes(data, dependencies, sorted(active_groups), group_vars) def verifier_dependances_connues(data: dict, dependencies: dict, playbook_dir: Path | None) -> None: for group, config in dependencies.items(): if playbook_dir and not (playbook_dir / f"{group}.yml").exists(): raise ValueError(f"Playbook de groupe manquant pour dependance: {group}") for required_group in config.get("requiert_groupes_actifs", []): if playbook_dir and not (playbook_dir / f"{required_group}.yml").exists(): raise ValueError(f"Playbook de groupe manquant pour prerequis: {required_group}") def verifier_coherence_hotes(data: dict) -> None: vmids: dict[str, str] = {} for host in tous_les_hotes(data): groups = groupes_hote(data, host) state_groups = [group for group in groups if group in GROUPES_ETAT_HOTE] operational_groups = [group for group in groups if est_groupe_operationnel(group)] vars_for_host = variables_hote(data, host) if not state_groups and not operational_groups: continue if len(state_groups) != 1: raise ValueError( f"{host} doit appartenir exactement a un groupe d'etat: " f"{GROUPE_HOTES_ACTIFS} ou {GROUPE_HOTES_PLANIFIES}" ) if not operational_groups: raise ValueError(f"{host} doit appartenir a au moins un groupe operationnel.") if GROUPE_HOTES_ACTIFS in state_groups and not vars_for_host.get("ansible_host"): raise ValueError(f"{host} est actif mais n'a pas ansible_host.") vmid = vars_for_host.get("proxmox_vmid") if vmid in (None, ""): continue vmid_key = str(vmid) if vmid_key in vmids and vmids[vmid_key] != host: raise ValueError(f"VMID en double: {vmid_key} pour {vmids[vmid_key]} et {host}") vmids[vmid_key] = host def verifier_playbooks_groupes(data: dict, playbook_dir: Path) -> None: missing_playbooks: list[str] = [] for group, group_data in enfants(data).items(): hosts = group_data.get("hosts", {}) if not hosts: continue if not est_groupe_operationnel(group): continue playbook = playbook_dir / f"{group}.yml" if not playbook.exists(): missing_playbooks.append(group) if missing_playbooks: missing = ", ".join(sorted(missing_playbooks)) raise ValueError(f"Playbook de groupe manquant pour: {missing}") verifier_coherence_hotes(data) def main() -> int: parser = argparse.ArgumentParser(description="Gere les hotes d'un inventaire Ansible statique Set-OPS.") parser.add_argument("--inventaire", required=True, type=Path) parser.add_argument("--dependances", type=Path) subparsers = parser.add_subparsers(dest="command", required=True) add_parser = subparsers.add_parser("ajouter", help="Ajoute ou met a jour un hote dans un ou plusieurs groupes.") add_parser.add_argument("--hote", required=True) add_parser.add_argument("--groupes", required=True) add_parser.add_argument("--adresse-ip") add_parser.add_argument("--utilisateur-ansible") add_parser.add_argument("--vmid") add_parser.add_argument("--dossier-playbooks", type=Path) groups_parser = subparsers.add_parser("definir-groupes", help="Remplace l'appartenance aux groupes pour un hote.") groups_parser.add_argument("--hote", required=True) groups_parser.add_argument("--groupes", required=True) groups_parser.add_argument("--dossier-playbooks", type=Path) show_parser = subparsers.add_parser("afficher", help="Affiche les groupes et variables d'un hote.") show_parser.add_argument("--hote", required=True) proxmox_parser = subparsers.add_parser( "parametres-proxmox", help="Emet les parametres de clonage Proxmox d'un hote (lus depuis l'inventaire).", ) proxmox_parser.add_argument("--hote", required=True) playbooks_parser = subparsers.add_parser("playbooks", help="Affiche les playbooks de groupes pour un hote.") playbooks_parser.add_argument("--hote", required=True) playbooks_parser.add_argument("--dossier-playbooks", required=True, type=Path) active_parser = subparsers.add_parser("verifier-actif", help="Verifie qu'un hote est actif.") active_parser.add_argument("--hote", required=True) subparsers.add_parser("lister-actifs", help="Liste les hotes actifs, un par ligne (pour la creation/deploiement de flotte).") host_dependencies_parser = subparsers.add_parser( "verifier-dependances-hote", help="Verifie les prerequis actifs requis par les groupes d'un hote.", ) host_dependencies_parser.add_argument("--hote", required=True) group_dependencies_parser = subparsers.add_parser( "verifier-dependances-groupe", help="Verifie les prerequis actifs requis par un groupe.", ) group_dependencies_parser.add_argument("--groupe", required=True) verify_parser = subparsers.add_parser( "verifier-playbooks", help="Verifie que les groupes operationnels avec hotes ont un playbook homonyme.", ) verify_parser.add_argument("--dossier-playbooks", required=True, type=Path) dependencies_parser = subparsers.add_parser( "verifier-dependances", help="Verifie le registre de dependances et les prerequis des groupes actifs.", ) dependencies_parser.add_argument("--dossier-playbooks", required=True, type=Path) args = parser.parse_args() try: data = charger_inventaire(args.inventaire) dependencies = charger_dependances(args.dependances) if args.command == "ajouter": ajouter_hote( data, args.hote, analyser_groupes(args.groupes), args.adresse_ip, args.utilisateur_ansible, args.vmid, ) if args.dossier_playbooks: verifier_playbooks_groupes(data, args.dossier_playbooks) if args.dependances: verifier_dependances_connues(data, dependencies, args.dossier_playbooks) verifier_dependances_actives( data, dependencies, Path(args.inventaire).parent / "group_vars") ecrire_inventaire(args.inventaire, data) elif args.command == "definir-groupes": definir_groupes_hote(data, args.hote, analyser_groupes(args.groupes)) if args.dossier_playbooks: verifier_playbooks_groupes(data, args.dossier_playbooks) if args.dependances: verifier_dependances_connues(data, dependencies, args.dossier_playbooks) verifier_dependances_actives( data, dependencies, Path(args.inventaire).parent / "group_vars") ecrire_inventaire(args.inventaire, data) elif args.command == "afficher": afficher_hote(data, args.hote) elif args.command == "parametres-proxmox": for ligne in parametres_proxmox_hote(data, args.hote, args.inventaire): print(ligne) elif args.command == "playbooks": afficher_playbooks_hote(data, args.hote, args.dossier_playbooks) elif args.command == "verifier-actif": verifier_hote_actif(data, args.hote) elif args.command == "lister-actifs": for hote in sorted(enfants(data).get(GROUPE_HOTES_ACTIFS, {}).get("hosts", {}) or {}): print(hote) elif args.command == "verifier-dependances-hote": exiger_dependances(dependencies, args.dependances) verifier_dependances_hote(data, dependencies, args.hote, Path(args.inventaire).parent / 'group_vars') elif args.command == "verifier-dependances-groupe": exiger_dependances(dependencies, args.dependances) verifier_dependances_groupes( data, dependencies, [args.groupe], Path(args.inventaire).parent / "group_vars") elif args.command == "verifier-playbooks": verifier_playbooks_groupes(data, args.dossier_playbooks) elif args.command == "verifier-dependances": exiger_dependances(dependencies, args.dependances) verifier_dependances_connues(data, dependencies, args.dossier_playbooks) verifier_dependances_actives( data, dependencies, Path(args.inventaire).parent / "group_vars") except Exception as exc: print(f"erreur: {exc}", file=sys.stderr) return 2 return 0 if __name__ == "__main__": raise SystemExit(main())