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

    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 _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())