Set-OPS/scripts/inventory_rules.py

#!/usr/bin/env python3
"""Regles partagees pour les inventaires statiques Set-OPS."""

from __future__ import annotations

from pathlib import Path

import yaml


GROUPES_OPERATIONNELS_PREFIXES = ("serveurs_", "clients_")
GROUPE_HOTES_ACTIFS = "hotes_actifs"
GROUPE_HOTES_PLANIFIES = "hotes_planifies"
GROUPES_ETAT_HOTE = {GROUPE_HOTES_ACTIFS, GROUPE_HOTES_PLANIFIES}


def est_groupe_operationnel(groupe: str) -> bool:
    return groupe.startswith(GROUPES_OPERATIONNELS_PREFIXES)


def groupes_operationnels_connus(enfants: dict, dossier_playbooks: Path) -> list[str]:
    groupes = set(enfants)
    groupes.update(playbook.stem for playbook in dossier_playbooks.glob("*.yml"))
    return sorted(groupe for groupe in groupes if est_groupe_operationnel(groupe))


def charger_bases_donnees(path: Path | None) -> dict:
    """Charge le registre des bases de donnees (serveurs_bd + bases_donnees)."""
    if not path or not path.exists():
        return {"serveurs_bd": {}, "bases_donnees": {}}
    with path.open("r", encoding="utf-8") as fichier:
        data = yaml.safe_load(fichier) or {}
    if not isinstance(data, dict):
        raise ValueError(f"{path} ne contient pas une table YAML.")
    data.setdefault("serveurs_bd", {})
    data.setdefault("bases_donnees", {})
    return data


def valider_bases(registre: dict) -> None:
    """Valide la coherence du registre des bases de donnees."""
    serveurs = registre.get("serveurs_bd") or {}
    bases = registre.get("bases_donnees") or {}
    if not isinstance(serveurs, dict) or not isinstance(bases, dict):
        raise ValueError("serveurs_bd et bases_donnees doivent etre des tables.")
    for nom, srv in serveurs.items():
        if not isinstance(srv, dict):
            raise ValueError(f"Serveur BD '{nom}': table attendue.")
        for champ in ("type", "hote", "port"):
            if not str(srv.get(champ, "")).strip():
                raise ValueError(f"Serveur BD '{nom}': champ '{champ}' requis.")
    noms_bases: set[str] = set()
    for cle, base in bases.items():
        if not isinstance(base, dict):
            raise ValueError(f"Base '{cle}': table attendue.")
        for champ in ("serveur", "base", "proprietaire", "secret"):
            if not str(base.get(champ, "")).strip():
                raise ValueError(f"Base '{cle}': champ '{champ}' requis.")
        if base["serveur"] not in serveurs:
            raise ValueError(f"Base '{cle}': serveur '{base['serveur']}' inconnu.")
        couple = f"{base['serveur']}/{base['base']}"
        if couple in noms_bases:
            raise ValueError(f"Base en double sur un meme serveur: {couple}")
        noms_bases.add(couple)


def chaine_connexion(entree: dict, serveur: dict, secret: str = "***") -> str:
    """Derive une chaine de connexion (mot de passe masque par defaut)."""
    type_bd = (serveur or {}).get("type", "postgres")
    hote = (serveur or {}).get("hote", "?")
    port = (serveur or {}).get("port", "?")
    return f"{type_bd}://{entree.get('proprietaire', '?')}:{secret}@{hote}:{port}/{entree.get('base', '?')}"


def charger_nomenclature(path: Path | None) -> dict:
    """Charge le registre de nomenclature (domaines, categories, adressage)."""
    if not path or not path.exists():
        return {}
    with path.open("r", encoding="utf-8") as fichier:
        data = yaml.safe_load(fichier) or {}
    if not isinstance(data, dict):
        raise ValueError(f"{path} ne contient pas une table YAML.")
    return data


def charger_dependances(path: Path | None) -> dict:
    if not path:
        return {}
    if not path.exists():
        raise ValueError(f"Fichier de dependances introuvable: {path}")

    with path.open("r", encoding="utf-8") as dependencies_file:
        data = yaml.safe_load(dependencies_file) or {}

    if not isinstance(data, dict):
        raise ValueError(f"{path} ne contient pas une table YAML.")

    groups = data.get("groupes", {})
    if not isinstance(groups, dict):
        raise ValueError(f"{path} doit contenir une cle 'groupes' de type table.")

    for group, config in groups.items():
        if not est_groupe_operationnel(group):
            raise ValueError(f"Groupe de dependance non operationnel: {group}")
        if not isinstance(config, dict):
            raise ValueError(f"Dependance invalide pour {group}: table attendue.")
        required = config.get("requiert_groupes_actifs", [])
        if required is None:
            required = []
        if not isinstance(required, list) or not all(isinstance(item, str) for item in required):
            raise ValueError(f"requiert_groupes_actifs doit etre une liste de groupes pour {group}.")
        for required_group in required:
            if not est_groupe_operationnel(required_group):
                raise ValueError(f"Prerequis non operationnel pour {group}: {required_group}")

    return groups