Set-OPS-Public/scripts/inventory_rules.py

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

from __future__ import annotations

import os
import re
from pathlib import Path

import yaml


GROUPES_OPERATIONNELS_PREFIXES = ("serveur_", "client_")
GROUPE_HOTES_ACTIFS = "hotes_actifs"
GROUPE_HOTES_PLANIFIES = "hotes_planifies"
GROUPES_ETAT_HOTE = {GROUPE_HOTES_ACTIFS, GROUPE_HOTES_PLANIFIES}

RACINE_DEPOT = Path(__file__).resolve().parents[1]


ORDRE_INVENTAIRE = ("principal", "production")
ORDRE_INVENTAIRE_MODELE = ("lab", "principal", "production")


def dependances_groupes(racine: Path | None = None) -> dict:
    """Le registre des dependances causales entre groupes (docs/dependances-groupes.yml)."""
    f = (racine or RACINE_DEPOT) / "docs" / "dependances-groupes.yml"
    if not f.is_file():
        return {}
    return (yaml.safe_load(f.read_text(encoding="utf-8")) or {}).get("groupes") or {}


def variable_effective(cle: str, dossier_group_vars: Path | None = None,
                       role_defaut: str | None = None,
                       racine: Path | None = None) -> str | None:
    """La valeur d'une variable pour une instance : ses group_vars, sinon le defaut du role.

    SOURCE UNIQUE (2026-08-22). Trois lecteurs en avaient besoin le meme jour — P35 pour
    l'interrupteur `<role>_bd`, le controle de dependances pour les clauses `sauf_si`, et
    le generateur d'inventaire. Trois copies auraient recommence exactement ce qu'on vient
    de refermer avec la resolution d'instance.

    Lecture TEXTUELLE : aucune voute n'est dechiffree, aucun Ansible n'est lance. Ces
    controles doivent rester instantanes — c'est ce qui fait qu'on les lance.
    """
    base = racine or RACINE_DEPOT
    dossier = dossier_group_vars or (dossier_inventaire() / "group_vars")
    if dossier.is_dir():
        for f in sorted(dossier.glob("**/*.yml")):
            try:
                d = yaml.safe_load(f.read_text(encoding="utf-8")) or {}
            except (OSError, yaml.YAMLError):
                continue
            if isinstance(d, dict) and cle in d:
                return str(d[cle])
    if role_defaut:
        f = base / "roles" / role_defaut / "defaults" / "main.yml"
        if f.is_file():
            d = yaml.safe_load(f.read_text(encoding="utf-8")) or {}
            if isinstance(d, dict) and cle in d:
                return str(d[cle])
    return None


def instance_courante() -> Path:
    """L'instance VISEE : `SETOPS_INSTANCE`, sinon le symlink `instance/`.

    SOURCE UNIQUE. Neuf modules en portaient chacun leur copie, et cinq defauts en sont
    sortis en cinq jours (2026-08-18 au 22) : P03 comparait chaque instance a l'inventaire
    d'une autre ; `verifier_ports` codait `principal/` en dur ; `verifier_intrants` et
    `_frontiere_absente` lisaient le symlink au lieu de la variable ; `devis_placement`
    rendait un verdict juste sur le mauvais tenant ; P35 aussi.

    Aucun n'etait une faute d'inattention : chacun etait une copie ecrite de bonne foi, a
    un moment ou le besoin semblait local. C'est le mode de panne de la duplication — pas
    l'erreur, mais la DERIVE, qui ne se voit jamais depuis l'interieur d'un fichier.
    """
    return Path(os.environ.get("SETOPS_INSTANCE") or (RACINE_DEPOT / "instance"))


def dossier_inventaire(instance: Path | None = None,
                       ordre: tuple[str, ...] = ORDRE_INVENTAIRE) -> Path:
    """Le REPERTOIRE d'inventaire de l'instance, dans l'ordre de preference donne.

    Trois niveaux, et le troisieme manquait a la moitie des copies : un `hosts.yml`
    existant, puis un REPERTOIRE existant (le cas d'une instance neuve, dont l'inventaire
    n'est pas encore genere — c'est ainsi que `make instancier` echouait sur le modele
    public), puis le premier nom par defaut.
    """
    base = Path(instance) if instance is not None else instance_courante()
    for nom in ordre:
        if (base / "inventories" / nom / "hosts.yml").is_file():
            return base / "inventories" / nom
    for nom in ordre:
        if (base / "inventories" / nom).is_dir():
            return base / "inventories" / nom
    return base / "inventories" / ordre[0]


def inventaire_de(instance: Path | None = None,
                  ordre: tuple[str, ...] = ORDRE_INVENTAIRE) -> Path:
    """Le fichier `hosts.yml` de l'instance — `SETOPS_INVENTAIRE` le force (et est garde)."""
    base = Path(instance) if instance is not None else instance_courante()
    return inventaire_force(base) or (dossier_inventaire(base, ordre) / "hosts.yml")


def plan_de(instance: Path | None = None) -> Path:
    """Le repertoire `plan/` de l'instance visee."""
    base = Path(instance) if instance is not None else instance_courante()
    return base / "plan"


def inventaire_force(instance: Path) -> Path | None:
    """Inventaire impose par SETOPS_INVENTAIRE, ou None. REFUSE s'il vise AILLEURS.

    POURQUOI CETTE GARDE (mesure du 2026-08-18). Deux variables designent la cible, et
    la seconde gagne : `SETOPS_INSTANCE` dit QUELLE instance, `SETOPS_INVENTAIRE` FORCE
    le fichier. Le Makefile exporte la seconde (ligne 13), derivee de l'instance ACTIVE.
    Tout code qui redirige la premiere vers une AUTRE instance sans retirer la seconde
    lit donc le plan de l'une et l'inventaire de l'autre.

    C'est arrive a P03, la preuve du diff-vide : sous `make prouver` elle generait le
    plan de chaque instance federee et le comparait a l'inventaire applique de la seule
    instance active. Elle annoncait « 17 hotes avec ecart » sur un lab qui n'en avait
    aucun — et, plus grave, l'inventaire applique des autres tenants n'etait JAMAIS lu,
    l'angle meme que cette preuve existe pour couvrir (D-77, tenant perime).

    Une contradiction entre les deux ne peut pas se deviner : on refuse, en nommant les
    deux valeurs. Le geste correct au site d'appel est `env.pop("SETOPS_INVENTAIRE")`.
    """
    force = os.environ.get("SETOPS_INVENTAIRE")
    if not force:
        return None
    cible = Path(force)
    base = Path(instance).resolve()
    if base != cible.resolve() and base not in cible.resolve().parents:
        raise SystemExit(
            "REFUS : SETOPS_INVENTAIRE designe un inventaire HORS de l'instance demandee.\n"
            f"  SETOPS_INSTANCE    {base}\n"
            f"  SETOPS_INVENTAIRE  {cible.resolve()}\n"
            "Comparer le plan d'une instance a l'inventaire d'une AUTRE rend un verdict\n"
            "qui ne veut rien dire. Retirer SETOPS_INVENTAIRE de l'environnement fils\n"
            "(env.pop) partout ou l'on redirige SETOPS_INSTANCE."
        )
    return cible


def liens_acceptes(groupe: str, racine: Path | None = None) -> dict:
    """Liens qu'un role accepte : roles/<groupe>/meta/liens.yml -> setops_liens.accepte.

    Source unique partagee par le validateur, le GUI et scripts/instancier.py.
    Retourne {} si le role n'en declare aucun.
    """
    fichier = (racine or RACINE_DEPOT) / "roles" / groupe / "meta" / "liens.yml"
    if not fichier.is_file():
        return {}
    data = yaml.safe_load(fichier.read_text(encoding="utf-8")) or {}
    return ((data.get("setops_liens") or {}).get("accepte")) or {}


def catalogue_liens(racine: Path | None = None) -> dict:
    """{groupe: {role_de_lien: {cible, injecte_sur, variables}}} pour tous les roles.

    Alimente la liste deroulante « rôle du lien » du GUI : on ne propose que ce
    que le role porteur declare accepter.
    """
    base = (racine or RACINE_DEPOT) / "roles"
    if not base.is_dir():
        return {}
    catalogue = {}
    for meta in sorted(base.glob("*/meta/liens.yml")):
        accepte = liens_acceptes(meta.parents[1].name, racine)
        if accepte:
            catalogue[meta.parents[1].name] = accepte
    return catalogue


def integrations_universelles(racine: Path | None = None) -> dict[str, dict]:
    """{role: {raison, sauf_role}} — les integrations declarees universelles.

    Une integration universelle n'est PAS recopiee sur chaque serveur du plan : le
    role la declare une fois dans roles/<role>/meta/integration.yml, et tout hote
    la recoit. Le plan ne porte plus que les integrations reellement facultatives
    (client_backup, client_smtp, client_resolveur) et les exemptions.

    Pourquoi : recopiee 14 fois, la meme ligne finit par manquer une quinzieme —
    et un hote non supervise, non journalise ou sans certificat ne proteste pas.
    Le defaut est desormais « oui », et c'est l'exception qui doit se justifier.
    """
    base = (racine or RACINE_DEPOT) / "roles"
    if not base.is_dir():
        return {}
    politiques = {}
    for meta in sorted(base.glob("client_*/meta/integration.yml")):
        data = yaml.safe_load(meta.read_text(encoding="utf-8")) or {}
        politique = data.get("integration") or {}
        if politique.get("universelle"):
            politiques[meta.parents[1].name] = {
                "raison": str(politique.get("raison", "")).strip(),
                "sauf_role": politique.get("sauf_role") or None,
            }
    return politiques


def integrations_de(srv: dict, services_hote: set[str] | None = None,
                    universelles: dict[str, dict] | None = None,
                    racine: Path | None = None,
                    services_du_plan: set[str] | None = None) -> set[str]:
    """Integrations effectives d'un serveur : les universelles + celles du plan.

    `services_hote` = groupes de service que cet hote rend (derives des
    applications). Il sert aux exemptions : `sauf_role` retire l'integration a
    l'hote qui rend le service lui-meme — l'AC ne s'enrole pas aupres d'elle-meme.
    L'exemption suit donc le service, jamais un nom d'hote.

    `services_du_plan` = TOUS les groupes de service de cet ecosysteme. UNE INTEGRATION
    UNIVERSELLE N'EST POSEE QUE SI SON SERVICE CENTRAL EXISTE ICI (2026-08-22).

    Pourquoi cette regle a manque. « Tout hote est mesure » est vrai dans un ecosysteme qui
    porte un Prometheus. Dans un ecosysteme qui n'en a pas — un patient 0 minimal, une
    petite offre — la meme phrase pose sur chaque machine un client qui n'a personne a qui
    parler, et le controle de dependances refuse le deploiement. Le moteur supposait
    l'ecosysteme COMPLET ; c'est la troisieme fois que cette hypothese se paie, apres les
    intrants (P32) et les bases (P35).

    La regle est DERIVEE, pas declaree : le service central d'une integration est celui que
    le registre des dependances lui donne deja (`requiert_groupes_actifs`). Rien de neuf a
    tenir a jour, donc rien de neuf a oublier.
    """
    if universelles is None:
        universelles = integrations_universelles(racine)
    rendus = services_hote or set()
    effectives = {role for role, p in universelles.items()
                  if not (p["sauf_role"] and p["sauf_role"] in rendus)}
    if services_du_plan is not None:
        deps = dependances_groupes(racine)
        effectives = {
            role for role in effectives
            if all(exige in services_du_plan
                   for exige in (deps.get(role, {}).get("requiert_groupes_actifs") or []))
        }
    return effectives | set(srv.get("integrations") or [])


def pool_de(nom_instance: str, index: int) -> str:
    """Nom du pool Proxmox d'un tenant. DERIVE, jamais saisi.

    Pourquoi le pool. Onze des quatorze serveurs portent le meme nom court chez deux
    tenants : c'est la preuve que la nomenclature est un vrai gabarit, et il ne faut
    pas la casser. Mais la console Proxmox affiche le NOM — deux `infra-pki-01` y
    sont indiscernables a l'oeil, alors que tout le reste (IP, VMID, VLAN, FQDN)
    differe. Le pool restitue l'appartenance sans rien renommer.

    Pourquoi DERIVE malgre les pools existants. Le cluster porte deja une convention
    `Env.Tenant` (`Prod.Chezlepro`, `Lab.KBR`...) qui regroupe des VM anterieures a
    Set-OPS — ZIMBRA, erplibre, eregion... Ces pools-la sont l'ANCIEN MONDE : on n'y
    touche pas, et on n'y verse pas la flotte generee. Melanger les deux effacerait
    justement la frontiere que Set-OPS etablit. La flotte derivee a donc ses propres
    pools, nommes comme tout le reste : depuis le seed.

    `index` est deja garanti unique dans la federation par P21 : le nom l'est donc
    aussi, sans registre supplementaire. Forme : `Chezlepro-17`, `Technolibre-11`.
    """
    base = re.sub(r"^OPS-", "", nom_instance)
    base = re.sub(r"[^A-Za-z0-9._-]", "", base) or "tenant"
    return f"{base}-{int(index)}"


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


PORTEES_BD = {"groupe", "hote", "application"}


def valider_bases(registre: dict, applications: dict | None = None) -> None:
    """Valide la coherence du registre des bases de donnees.

    Si `applications` (le registre des applications) est fourni, les bases de
    portee 'application' sont verifiees : leur consommateur doit etre une
    application declaree.
    """
    serveurs = registre.get("serveurs_bd") or {}
    bases = registre.get("bases_donnees") or {}
    apps = (applications or {}).get("applications") 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.")
        portee = str(base.get("portee", "groupe")).strip()
        if portee not in PORTEES_BD:
            attendu = ", ".join(sorted(PORTEES_BD))
            raise ValueError(f"Base '{cle}': portee '{portee}' inconnue (attendu: {attendu}).")
        if portee == "groupe" and not est_groupe_operationnel(base["consommateur"]):
            raise ValueError(f"Base '{cle}': portee groupe mais consommateur '{base['consommateur']}' non operationnel.")
        if portee == "application" and apps and base["consommateur"] not in apps:
            raise ValueError(f"Base '{cle}': application consommatrice '{base['consommateur']}' inconnue (instance/plan/applications.yml).")
        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 _base_resolue(cle: str, base: dict, serveurs: dict) -> dict:
    return {"cle": cle, "base": base, "serveur": serveurs.get(base.get("serveur"), {})}


def bases_du_groupe(registre: dict, groupe: str) -> list[dict]:
    """Bases de portee 'groupe' consommees par un groupe applicatif."""
    serveurs = registre.get("serveurs_bd") or {}
    resultat = []
    for cle, base in (registre.get("bases_donnees") or {}).items():
        if str(base.get("portee", "groupe")) == "groupe" and base.get("consommateur") == groupe:
            resultat.append(_base_resolue(cle, base, serveurs))
    return resultat


def bases_de_application(registre: dict, application: str | None = None,
                         groupe: str | None = None, hote: str | None = None) -> list[dict]:
    """Bases vues par une application instanciee, toutes portees confondues.

    Une application `A` (du groupe `G`, sur l'hote `H`) recoit les bases ou :
      (portee=application ET consommateur=A) OU (portee=groupe ET consommateur=G)
      OU (portee=hote ET consommateur=H).
    """
    serveurs = registre.get("serveurs_bd") or {}
    resultat = []
    for cle, base in (registre.get("bases_donnees") or {}).items():
        portee = str(base.get("portee", "groupe"))
        conso = base.get("consommateur")
        if ((portee == "application" and application is not None and conso == application)
                or (portee == "groupe" and groupe is not None and conso == groupe)
                or (portee == "hote" and hote is not None and conso == hote)):
            resultat.append(_base_resolue(cle, base, serveurs))
    return resultat


def charger_applications(path: Path | None) -> dict:
    """Charge le registre des applications (applications -> groupe, hote)."""
    if not path or not path.exists():
        return {"applications": {}}
    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("applications", {})
    return data


def valider_applications(registre: dict, domaines: dict | None = None,
                         serveurs: dict | None = None) -> None:
    """Valide la coherence du registre des applications.

    Si `domaines` (le registre des domaines publics) est fourni, chaque FQDN
    de `expose` doit avoir un domaine parent declare.

    Si `serveurs` (le registre des serveurs) est fourni, l'hote porteur de chaque
    application doit y etre declare — c'est le garde-fou contre l'hote FANTOME
    (une application posee sur une VM qui n'existe pas). Sans lui, le plan passe
    la validation et casse au deploiement.
    """
    apps = registre.get("applications") or {}
    doms = (domaines or {}).get("domaines_publics") or {}
    if not isinstance(apps, dict):
        raise ValueError("applications doit etre une table.")
    for nom, app in apps.items():
        if not isinstance(app, dict):
            raise ValueError(f"Application '{nom}': table attendue.")
        for champ in ("groupe", "hote"):
            if not str(app.get(champ, "")).strip():
                raise ValueError(f"Application '{nom}': champ '{champ}' requis.")
        hotes_connus = (serveurs or {}).get("serveurs")
        if hotes_connus and app["hote"] not in hotes_connus:
            raise ValueError(
                f"Application '{nom}': hote '{app['hote']}' absent du registre des "
                f"serveurs (hote fantome). Declare-le dans plan/serveurs.yml, ou "
                f"deplace l'application sur un hote existant.")
        if not est_groupe_operationnel(app["groupe"]):
            raise ValueError(f"Application '{nom}': groupe '{app['groupe']}' non operationnel.")
        requiert = app.get("requiert") or []
        if not isinstance(requiert, list):
            raise ValueError(f"Application '{nom}': 'requiert' doit etre une liste.")
        for dep in requiert:
            if dep not in apps:
                raise ValueError(f"Application '{nom}': requiert '{dep}' inconnue.")
        expose = app.get("expose") or []
        if not isinstance(expose, list):
            raise ValueError(f"Application '{nom}': 'expose' doit etre une liste.")
        for fqdn in expose:
            if doms and domaine_parent(fqdn, doms) is None:
                raise ValueError(f"Application '{nom}': FQDN expose '{fqdn}' sans domaine parent declare.")
        port = app.get("port")
        if port is not None and not str(port).strip().isdigit():
            raise ValueError(f"Application '{nom}': 'port' doit etre un entier.")
        liens = app.get("liens") or []
        if not isinstance(liens, list):
            raise ValueError(f"Application '{nom}': 'liens' doit etre une liste.")
        accepte = None
        for lien in liens:
            if not isinstance(lien, dict):
                raise ValueError(f"Application '{nom}': chaque lien doit etre une table {{vers, role}}.")
            vers = str(lien.get("vers", "")).strip()
            role = str(lien.get("role", "")).strip()
            if not vers or not role:
                raise ValueError(f"Application '{nom}': lien incomplet (champs 'vers' et 'role' requis).")
            if vers == nom:
                raise ValueError(f"Application '{nom}': lien vers elle-meme.")
            if vers not in apps:
                raise ValueError(f"Application '{nom}': cible de lien '{vers}' inconnue.")
            if accepte is None:
                accepte = liens_acceptes(str(app.get("groupe", "")))
            # Un role sans meta/liens.yml n'accepte aucun lien : on ne bloque que
            # s'il en declare (sinon instancier.py tranchera avec le meme message).
            if accepte and role not in accepte:
                raise ValueError(
                    f"Application '{nom}': role de lien '{role}' non accepte par "
                    f"'{app.get('groupe')}' (roles/{app.get('groupe')}/meta/liens.yml). "
                    f"Acceptes : {', '.join(sorted(accepte))}.")


def applications_de_hote(registre: dict, hote: str) -> list[dict]:
    """Applications instanciees sur un hote donne (plusieurs possibles)."""
    return [
        {"id": nom, **app}
        for nom, app in (registre.get("applications") or {}).items()
        if app.get("hote") == hote
    ]


def domaine_parent(fqdn: str, domaines_publics: dict) -> str | None:
    """Domaine declare le plus long dont `fqdn` releve (apex inclus)."""
    fqdn = str(fqdn).strip()
    candidats = [d for d in domaines_publics if fqdn == d or fqdn.endswith("." + d)]
    return max(candidats, key=len) if candidats else None


def expositions_des_applications(applications: dict, domaines: dict, edge: str | None = None) -> list[dict]:
    """Expositions publiques derivees des applications (champ 'expose').

    Pour chaque FQDN qu'une application expose, on resout son domaine parent et
    l'edge de ce domaine. Si `edge` est fourni, ne retient que les expositions
    de cet edge. Le rendu de l'amont (IP interne) reste a la charge de l'appelant.
    """
    apps = (applications or {}).get("applications") or {}
    doms = (domaines or {}).get("domaines_publics") or {}
    resultat = []
    for app_id, app in apps.items():
        for fqdn in (app.get("expose") or []):
            dom = domaine_parent(fqdn, doms)
            conf = doms.get(dom, {}) if dom else {}
            # SANS EDGE DECLARE, LE SERVICE SE SERT LUI-MEME (2026-08-25).
            #
            # `edge` se resout par le DOMAINE : chez un tenant, `domaines.yml` dit quel
            # nginx front telle zone, et tout FQDN de cette zone y aboutit. Le SITE n'a
            # pas d'edge — chaque service ecoute lui-meme, avec son propre certificat.
            #
            # Sans ce repli, la derivation rendait `edge: None` : le gabarit de
            # `hosts_statiques` cherchait alors `groups[None]`, ne trouvait rien, et
            # ecrivait un plancher SANS alias. Le certificat de la forge portait
            # `forge.genese.internal`, et ce nom ne resolvait nulle part — du TLS correct
            # sur un nom que personne ne pouvait appeler.
            edge_dom = conf.get("edge") or app.get("groupe")
            if edge is not None and edge_dom != edge:
                continue
            # SANS DOMAINE PUBLIC DECLARE, LE DOMAINE EST CELUI QUE PORTE LE FQDN
            # (2026-08-25). Meme repli que pour `edge` juste au-dessus, et il manquait.
            #
            # `domaine_parent` cherche le FQDN parmi les domaines PUBLICS. Le SITE n'en a
            # aucun — tous ses noms sont internes — donc `dom` valait None, et le modele
            # de zone PowerDNS, qui ne retient que les expositions dont le domaine EST sa
            # zone, les ecartait toutes.
            #
            # Consequence mesuree le 2026-08-25 : `forge.genese.internal` et
            # `pki.genese.internal` n'avaient aucun enregistrement A. Seul le plancher
            # /etc/hosts savait les resoudre — or un service ne doit pas dependre d'un
            # plancher pour etre joignable ; le plancher est un filet, pas le sol.
            if not dom and "." in str(fqdn):
                dom = str(fqdn).split(".", 1)[1]
            resultat.append({
                "fqdn": fqdn, "application": app_id, "groupe": app.get("groupe"),
                "hote": app.get("hote"), "port": app.get("port"),
                "edge": edge_dom, "domaine": dom,
                # Exposition WebSocket-aware (ex. Collabora) : le vhost ajoute
                # alors l'upgrade Upgrade/Connection. Declare par l'app (websocket: true).
                "websocket": bool(app.get("websocket")),
            })
    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 (fonctions, 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


ETATS_SERVEUR = {"actif", "planifie"}


def fonction_seq(nom: str) -> tuple[str, int | None]:
    """Decompose un hostname 'fonction-NN' en (fonction, NN)."""
    m = re.match(r"^(.+)-(\d+)$", str(nom or ""))
    if m:
        return m.group(1), int(m.group(2))
    return str(nom or ""), None


# --- Derivation de l'ADRESSAGE depuis le seul seed `index` --------------------
#
# Principe : TOUT l'adressage (supernet, sous-reseaux, passerelles, VLAN, VMID) se
# DERIVE de l'index d'instance + du numero de zone. Rien n'est ecrit a la main dans
# la nomenclature (elle ne garde que le MODELE : libelles de zones + placement des
# fonctions). Source unique, ci-dessous, partagee par la generation d'inventaire et
# le devis reseau (scripts/devis_reseau.py).
#
# Conventions du modele a 6 zones (reference integral) :
#   - 2e octet du supernet = index TEL QUEL          (index 1 -> 10.1.0.0/16)
#   - 3e octet de la zone   = OCTET_ZONE + categorie  (zone 1 -> .16, dans .16.0/20)
#   - VLAN sur le trunk convergE = SOCLE_VLAN + index x 10 + zone (4 chiffres, unique)
#   - VMID ip-miroir = VLAN(4) . octet-hote(3) . sequence(2) = 9 chiffres mnemotechniques
#
# LE DECALAGE DE +10 A ETE RETIRE le 2026-08-12. Il produisait `10.(10+index)`, et
# personne ne savait plus pourquoi : ni le commentaire de la constante, ni le wiki, ni
# le commit fondateur (36a882b) ne le justifiaient. Ses deux effets constates :
#   - il reservait 10.0-10.9 sous la plage tenant — utile tant que l'underlay vivait
#     LA ; D-77 l'a fait entrer dans la bande basse de son propre /16, ce qui a vide
#     cette reserve de son role ;
#   - il eloignait le premier tenant de 10.0.0.0/16, la plage la plus repandue en
#     reseau domestique. Ce risque revient donc aux INDEX BAS, et c'est assume :
#     choisir un index, c'est choisir sa plage — autant que ce soit lisible.
# En echange, l'index se lit DIRECTEMENT dans l'adresse (index 17 -> 10.17.x.x) et la
# plage utile passe de 0-245 a 0-255.
OCTET_ZONE = 15
SOCLE_VLAN = 1000

# BORNES DE L'INDEX. Le 2e octet d'une IPv4 vaut 0-255 : au-dela, la derivation ne
# produit plus une adresse mais une CHAINE qui y ressemble — `10.300.0.0/16` passe
# silencieusement tant que personne ne la donne a `ip_network()`, et l'erreur remonte
# alors tres loin de sa cause. Depuis le retrait du decalage de +10 (2026-08-12), la
# plage utile est exactement celle de l'octet.
INDEX_MIN, INDEX_MAX = 0, 255
INDEX_BAC_A_SABLE = 9  # index par defaut d'une nomenclature sans seed (jetable)


def valider_index(index) -> int:
    """L'index, verifie et rendu en entier. Leve `ValueError` hors bornes.

    Pose a la SOURCE de la derivation plutot que dans un validateur de plan : toutes
    les fonctions derivees passent par ici, donc aucune ne peut fabriquer une adresse
    invalide, d'ou qu'on l'appelle — plan, GUI, devis ou test.
    """
    try:
        i = int(index)
    except (TypeError, ValueError, OverflowError):
        # OverflowError : `int(float('inf'))`. Trouve par le test, pas par relecture —
        # un infini flottant n'est ni un entier valide ni une TypeError.
        raise ValueError(f"index invalide : {index!r} n'est pas un entier") from None
    if not (INDEX_MIN <= i <= INDEX_MAX):
        raise ValueError(
            f"index {i} hors bornes [{INDEX_MIN}-{INDEX_MAX}] : le 2e octet d'une IPv4 "
            f"ne peut pas le porter. La derivation produirait « 10.{i}.0.0/16 », qui "
            f"ressemble a un reseau sans en etre un.")
    return i


def supernet_de(index: int) -> str:
    """/16 du tenant : 10.<index>.0.0/16 — l'index se lit dans l'adresse."""
    return f"10.{valider_index(index)}.0.0/16"


def base3_de(index: int, categorie: int) -> str:
    """Trois premiers octets d'une zone : 10.<index>.(15+categorie)."""
    return f"10.{valider_index(index)}.{OCTET_ZONE + int(categorie)}"


def sous_reseau_de(index: int, categorie: int, cidr: int = 24) -> str:
    """/24 d'une zone : 10.<index>.(15+categorie).0/24."""
    return f"{base3_de(index, categorie)}.0/{int(cidr)}"


def passerelle_de(index: int, categorie: int) -> str:
    """Passerelle d'une zone (SVI) : premier hote du /24."""
    return f"{base3_de(index, categorie)}.1"


def vlan_de(index: int, zone: int) -> int:
    """VLAN 802.1Q sur le trunk convergE : 1000 + index x 10 + zone (unique globalement).

    La meme borne y protege un second plafond : a l'index 255 le VLAN vaut 3550+zone,
    sous les 4094 du 802.1Q. Sans elle, un index de trois chiffres deborderait aussi la.
    """
    return SOCLE_VLAN + valider_index(index) * 10 + int(zone)


def deriver_nomenclature(fonction: str, seq: int, nomenclature: dict) -> dict | None:
    """Derive VMID / VLAN / IP / passerelle / CIDR depuis la fonction et le rang.

    Source unique de la derivation (miroir Python de l'auto-proposition du GUI).
    Tout vient de l'index + de la zone : aucune lecture d'adressage stocke.
    """
    fonctions = nomenclature.get("fonctions") or {}
    categories = nomenclature.get("categories") or {}
    f = fonctions.get(fonction)
    if not f or seq is None:
        return None
    cat = f.get("categorie")
    svc = f.get("service")
    c = categories.get(cat) or categories.get(str(cat))
    if not c:
        return None
    index = nomenclature.get("index", INDEX_BAC_A_SABLE)
    hote = svc * 10 + int(seq)
    vlan = vlan_de(index, int(cat))
    return {
        "vmid": f"{vlan}{hote:03d}{int(seq):02d}",
        "vlan": vlan,
        "adresse_ip": f"{base3_de(index, int(cat))}.{hote}",
        "passerelle": passerelle_de(index, int(cat)),
        "cidr": nomenclature.get("cidr_hote", 24),
        # La ZONE de securite, exposee : c'est d'elle que se derive le VNet du SDN
        # (`t17serv`). La recalculer ailleurs demanderait de relire fonctions +
        # categories — deux lectures pour une valeur deja connue ici.
        "zone": int(cat),
        "zone_libelle": str(c.get("libelle", f"zone{cat}")),
    }


def charger_serveurs(path: Path | None) -> dict:
    """Charge le registre des serveurs (VM) du plan."""
    if not path or not path.exists():
        return {"serveurs": {}}
    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", {})
    return data


def valider_serveurs(registre: dict, nomenclature: dict | None = None) -> None:
    """Valide la coherence du registre des serveurs."""
    serveurs = registre.get("serveurs") or {}
    fonctions = (nomenclature or {}).get("fonctions") or {}
    if not isinstance(serveurs, dict):
        raise ValueError("serveurs doit etre une table.")
    universelles = integrations_universelles()
    for nom, srv in serveurs.items():
        if not isinstance(srv, dict):
            raise ValueError(f"Serveur '{nom}': table attendue.")
        fonction = str(srv.get("fonction", "")).strip()
        if not fonction:
            raise ValueError(f"Serveur '{nom}': champ 'fonction' requis.")
        if fonctions and fonction not in fonctions:
            raise ValueError(f"Serveur '{nom}': fonction '{fonction}' inconnue de la nomenclature.")
        etat = str(srv.get("etat", "")).strip()
        if etat and etat not in ETATS_SERVEUR:
            raise ValueError(f"Serveur '{nom}': etat '{etat}' inconnu (attendu: {', '.join(sorted(ETATS_SERVEUR))}).")
        integrations = srv.get("integrations")
        if integrations is not None and not isinstance(integrations, list):
            raise ValueError(f"Serveur '{nom}': 'integrations' doit etre une liste.")
        # Une integration universelle recopiee dans le plan a deux torts : elle
        # laisse croire que son ABSENCE chez un voisin signifie « non integre »,
        # et elle offre une valeur a oublier. La politique du role est la seule
        # source ; on refuse la copie plutot que de la tolerer silencieusement.
        redites = sorted(set(integrations or []) & set(universelles))
        if redites:
            raise ValueError(
                f"Serveur '{nom}': integration(s) {', '.join(redites)} universelle(s) — "
                f"declaree(s) dans roles/<role>/meta/integration.yml, elles s'appliquent "
                f"a tout hote. Retirer ces lignes du plan.")


def reconcilier_serveur(nom: str, srv: dict, hote_inventaire: dict | None, nomenclature: dict) -> dict:
    """Compare les valeurs derivees (fonction+rang) aux valeurs de l'inventaire.

    Renvoie {derive, inventaire, divergences} ; divergences vide = reconcilie.
    """
    _, seq = fonction_seq(nom)
    derive = deriver_nomenclature(str(srv.get("fonction", "")), seq, nomenclature) or {}
    inv = hote_inventaire or {}
    champs = {"vmid": "vmid", "adresse_ip": "adresse_ip", "vlan": "vlan", "passerelle": "passerelle"}
    divergences = []
    for cle_derivee, cle_inv in champs.items():
        if cle_derivee not in derive:
            continue
        attendu = str(derive[cle_derivee])
        present = str(inv.get(cle_inv, "")).strip()
        if present and present != attendu:
            divergences.append(f"{cle_derivee}: inventaire={present} != derive={attendu}")
    return {"derive": derive, "inventaire": inv, "divergences": divergences,
            "absent_inventaire": hote_inventaire is None}


# --- Dimensionnement derive des ressources VM -------------------------------
# ressources(VM) = socle_SE + somme(empreinte des logiciels) -> marge -> arrondi.
# Les empreintes vivent dans roles/<groupe>/meta/empreinte.yml (propriete du
# logiciel). Les groupes sans role dedie ont un repli ci-dessous. Le plan peut
# tout surcharger par hote (serveurs.yml: coeurs/memoire/disque).

SOCLE_SE = {"coeurs": 1, "memoire_mo": 512, "disque_go": 8}
EMPREINTE_DEFAUT = {"coeurs": 1, "memoire_mo": 512, "disque_go": 5}
EMPREINTES_SANS_ROLE = {
    "serveur_web_frontal": {"coeurs": 1, "memoire_mo": 512, "disque_go": 5},
    "serveur_web_dorsal": {"coeurs": 1, "memoire_mo": 1024, "disque_go": 10},
}
MARGE_MEMOIRE = 1.20
MARGE_DISQUE = 1.25
PALIER_MEMOIRE_MO = 512
MIN_MEMOIRE_MO = 1024
PALIER_DISQUE_GO = 5
MIN_DISQUE_GO = 12
MIN_COEURS = 1
MAX_COEURS = 8
CHAMPS_EMPREINTE = ("coeurs", "memoire_mo", "disque_go")


def charger_empreinte_role(racine_roles: Path, groupe: str) -> dict | None:
    """Lit roles/<groupe>/meta/empreinte.yml -> empreinte normalisee, ou None."""
    fichier = racine_roles / groupe / "meta" / "empreinte.yml"
    if not fichier.exists():
        return None
    with fichier.open("r", encoding="utf-8") as fh:
        data = yaml.safe_load(fh) or {}
    emp = (data or {}).get("setops_empreinte") or {}
    if not isinstance(emp, dict):
        raise ValueError(f"{fichier}: 'setops_empreinte' doit etre une table.")
    # Un fichier qui declare les valeurs A LA RACINE au lieu de sous
    # `setops_empreinte:` etait lu comme VIDE : la VM recevait le minimum, en
    # silence. Quatre roles etaient dans ce cas, et `collab-01` s'est retrouvee
    # avec 1 coeur / 1 Go pour porter Nextcloud ET Collabora — elle a cesse de
    # repondre en SSH (2026-08-07). Un fichier qui existe mais ne dit rien est
    # pire qu'un fichier absent : le repli, lui, aurait donne des valeurs sensees.
    if not emp and any(c in (data or {}) for c in CHAMPS_EMPREINTE):
        raise ValueError(
            f"{fichier}: les valeurs sont declarees a la racine ; elles doivent "
            f"etre sous la cle `setops_empreinte:`. Sans elle, l'empreinte est lue "
            f"comme nulle et la VM recoit le minimum."
        )
    return {cle: int(emp.get(cle, 0)) for cle in CHAMPS_EMPREINTE}


def empreinte_groupe(racine_roles: Path, groupe: str) -> dict:
    """Empreinte d'un groupe de service : role > repli connu > defaut."""
    return (charger_empreinte_role(racine_roles, groupe)
            or EMPREINTES_SANS_ROLE.get(groupe)
            or EMPREINTE_DEFAUT)


def _arrondi_superieur(valeur: float, palier: int) -> int:
    import math
    return int(math.ceil(valeur / palier) * palier)


def deriver_ressources(groupes_service, racine_roles: Path, socle: dict | None = None) -> dict:
    """Estime coeurs/memoire(Mo)/disque(Go) d'une VM depuis ses logiciels + le SE.

    `groupes_service` : groupes applicatifs portes par l'hote (hors socle/etat).
    RAM et disque additifs (contraintes dures) ; coeurs en max (burstable).
    """
    socle = socle or SOCLE_SE
    coeurs_somme = 0
    memoire = socle["memoire_mo"]
    disque = socle["disque_go"]
    for groupe in groupes_service:
        emp = empreinte_groupe(racine_roles, groupe)
        coeurs_somme += emp["coeurs"]
        memoire += emp["memoire_mo"]
        disque += emp["disque_go"]
    coeurs = max(MIN_COEURS, min(MAX_COEURS, max(socle["coeurs"], coeurs_somme)))
    memoire = max(MIN_MEMOIRE_MO, _arrondi_superieur(memoire * MARGE_MEMOIRE, PALIER_MEMOIRE_MO))
    disque = max(MIN_DISQUE_GO, _arrondi_superieur(disque * MARGE_DISQUE, PALIER_DISQUE_GO))
    return {"coeurs": coeurs, "memoire": memoire, "disque": f"{disque}G"}


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