Set-OPS-Public/scripts/inventory_host.py

#!/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 (
    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 path.open("w", encoding="utf-8") 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_unbound`
    # 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),
)


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`)."
        )

    return lignes


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 verifier_dependances_groupes(data: dict, dependencies: dict, groups: list[str]) -> None:
    missing: list[str] = []

    for group in sorted(set(groups)):
        for required_group in dependencies.get(group, {}).get("requiert_groupes_actifs", []):
            if not groupe_a_hote_actif(data, required_group):
                missing.append(f"{group} requiert {required_group} actif")

    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) -> None:
    groups = [group for group in groupes_hote(data, host) if est_groupe_operationnel(group)]
    verifier_dependances_groupes(data, dependencies, groups)


def verifier_dependances_actives(data: dict, dependencies: dict) -> 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))


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)
            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)
            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)
        elif args.command == "verifier-dependances-groupe":
            exiger_dependances(dependencies, args.dependances)
            verifier_dependances_groupes(data, dependencies, [args.groupe])
        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)
    except Exception as exc:
        print(f"erreur: {exc}", file=sys.stderr)
        return 2

    return 0


if __name__ == "__main__":
    raise SystemExit(main())