Set-OPS-Public/scripts/inventory_host.py
Daniel Allaire 6077b179af
Some checks are pending
verifier / verifier (push) Waiting to run
plan : l ecriture des registres devient atomique — tout, ou rien
`path.open("w")` TRONQUE avant d ecrire : entre les deux, le fichier est vide.
Une exception dans yaml.safe_dump, un disque plein, un Ctrl-C, et
instance/plan/serveurs.yml reste mutile.

L asymetrie fait la gravite : hosts.yml se regenere d un
make instancier-appliquer, le PLAN ne se regenere de rien. C est la source
unique de verite. Git est le filet, mais encore faut-il savoir qu on est tombe.

TREIZE SITES, UNE SEULE FONCTION

Le defaut n etait pas dans le GUI seul : douze sites dans sept fichiers, dont
les miroirs CLI des MEMES registres. Corriger le GUI seul aurait recree la
divergence que P41 garde depuis les neuf resolutions d instance. La fonction
vit donc dans inventory_rules.py, que les sept importaient deja. Une source,
pas douze.

TROIS DETAILS QUI FONT LA DIFFERENCE ENTRE « CA MARCHE » ET « CA TIENT »

  temporaire dans le MEME dossier   os.replace n est atomique qu au sein d un
                                    meme systeme de fichiers ; un /tmp sur une
                                    autre partition casserait la garantie sans
                                    rien dire
  fsync AVANT le rename             sinon le renommage peut atteindre le disque
                                    avant le contenu : au retour d une coupure
                                    brutale, un fichier neuf et VIDE — le defaut
                                    qu on ferme, deplace d un cran
  report des droits                 mkstemp cree en 0600, le plan est en 0664 et
                                    doit rester lisible par le groupe sur les
                                    runners

LE TEST PORTE SON PROPRE CONTROLE NEGATIF

scripts/tests/test_ecriture_atomique.py rejoue D ABORD l ancienne forme et
verifie qu elle DETRUIT. Sans ce controle, « le fichier est intact » ne
prouverait rien — il pourrait l etre parce que rien n a ete ecrit du tout. Une
garantie qu on n a jamais vue echouer n est pas une garantie, c est une
habitude.

Branche sur P02, dont le titre annoncait « inventory_host » alors qu il lance
maintenant trois tests. Corrige au passage.

LA VOUTE DU GUI : VERIFIEE, PAS DE DEFAUT

Le soupcon etait qu executer_flux pose ANSIBLE_VAULT_PASSWORD_FILE (un seul mot
de passe) alors que creer une VM ouvre DEUX voutes depuis « une voute, une cle ».

Eprouve contre deux voutes JETABLES a mots de passe distincts — jamais les
vraies. Les deux variables se CUMULENT : Ansible essaie tous les secrets, et un
PASSWORD_FILE errone n empeche rien. Confirme en sondant l environnement qu une
recette make recoit reellement : le mot de passe saisi ET les cinq cles
calculees par voutes.py.

Ce qui sauve ce chemin n est donc pas le mot de passe saisi, c est
l IDENTITY_LIST que make pose par-dessus. Chacun couvre ce que l autre ne
couvre pas — le PASSWORD_FILE sert le runner qui n a que sa cle, l IDENTITY_LIST
le poste qui les a toutes. Ecrit au-dessus du code, pour que personne ne
« simplifie » en retirant l un des deux.

make prouver : CONFORME, 59 OK, 0 echec, 1 saute.
make instancier : DIFF VIDE, quatre registres relus, droits 664 preserves.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Crgis8CxCWkAGFA1ecBz3q
2026-09-08 14:57:57 -04:00

704 lines
29 KiB
Python

#!/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),
# 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())