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", "consommateur"):
            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 bases_du_groupe(registre: dict, groupe: str) -> list[dict]:
    """Bases consommees par un groupe applicatif, avec leur serveur resolu."""
    serveurs = registre.get("serveurs_bd") or {}
    resultat = []
    for cle, base in (registre.get("bases_donnees") or {}).items():
        if base.get("consommateur") == groupe:
            resultat.append({"cle": cle, "base": base, "serveur": serveurs.get(base.get("serveur"), {})})
    return resultat


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', '?')}"


AUTORITES_DNS = {"primaire-cache", "auto-heberge", "delegue"}


def charger_domaines(path: Path | None) -> dict:
    """Charge le registre des domaines publics (domaines_publics)."""
    if not path or not path.exists():
        return {"domaines_publics": {}}
    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("domaines_publics", {})
    return data


def fqdn_exposition(domaine: str, nom: str) -> str:
    """FQDN publie pour une etiquette ('@' ou le domaine lui-meme = apex)."""
    nom = str(nom).strip()
    if nom in ("@", "", domaine):
        return domaine
    return f"{nom}.{domaine}"


def valider_domaines(registre: dict) -> None:
    """Valide la coherence du registre des domaines publics."""
    domaines = registre.get("domaines_publics") or {}
    if not isinstance(domaines, dict):
        raise ValueError("domaines_publics doit etre une table.")
    fqdns: set[str] = set()
    for nom_domaine, conf in domaines.items():
        if not isinstance(conf, dict):
            raise ValueError(f"Domaine '{nom_domaine}': table attendue.")
        autorite = str(conf.get("autorite", "")).strip()
        if autorite not in AUTORITES_DNS:
            attendu = ", ".join(sorted(AUTORITES_DNS))
            raise ValueError(f"Domaine '{nom_domaine}': autorite '{autorite}' inconnue (attendu: {attendu}).")
        if not str(conf.get("edge", "")).strip():
            raise ValueError(f"Domaine '{nom_domaine}': champ 'edge' requis.")
        secondaires = conf.get("secondaires") or []
        if not isinstance(secondaires, list):
            raise ValueError(f"Domaine '{nom_domaine}': 'secondaires' doit etre une liste.")
        exposition = conf.get("exposition") or []
        if not isinstance(exposition, list):
            raise ValueError(f"Domaine '{nom_domaine}': 'exposition' doit etre une liste.")
        for entree in exposition:
            if not isinstance(entree, dict):
                raise ValueError(f"Domaine '{nom_domaine}': entree d'exposition invalide (table attendue).")
            for champ in ("nom", "cible"):
                if not str(entree.get(champ, "")).strip():
                    raise ValueError(f"Domaine '{nom_domaine}': exposition, champ '{champ}' requis.")
            fqdn = fqdn_exposition(nom_domaine, entree["nom"])
            if fqdn in fqdns:
                raise ValueError(f"Exposition en double: {fqdn}")
            fqdns.add(fqdn)


def expositions_du_groupe(registre: dict, groupe: str) -> list[dict]:
    """Expositions publiques dont la cible est un groupe interne donne."""
    resultat = []
    for nom_domaine, conf in (registre.get("domaines_publics") or {}).items():
        for entree in (conf.get("exposition") or []):
            if entree.get("cible") == groupe:
                resultat.append({
                    "domaine": nom_domaine,
                    "fqdn": fqdn_exposition(nom_domaine, entree.get("nom", "")),
                    "type": entree.get("type", "web"),
                    "edge": conf.get("edge"),
                })
    return resultat


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