Set-OPS-Public/scripts/verifier_intrants.py

#!/usr/bin/env python3
"""Tout intrant qu'un rôle EXIGE est-il fourni par l'instance ?

Un rôle qui `assert` une variable non vide déclare un **contrat** : sans cette valeur, le
déploiement s'arrête. Rien ne vérifiait que l'instance honore ces contrats — et le défaut
ne se voit qu'au moment où la tâche gardée s'exécute vraiment.

C'est ainsi que la première reconstruction from-zero s'est arrêtée sur
`amorcage_acces_courriel` (2026-08-08) : la valeur était obligatoire depuis le matin, aucun
tenant ne la déclarait, et personne ne pouvait le savoir parce que le compte existait déjà
— donc la garde n'avait jamais eu l'occasion de se déclencher.

**Un intrant est considéré comme SATISFAIT s'il est** :
  - doté d'un défaut non vide dans `<role>/defaults/main.yml` (y compris un `{{ vault_* }}`,
    dont la présence réelle est gardée par P18) ; ou
  - posé à l'exécution par un `set_fact` d'un rôle (les résolveurs `resoudre_*`) ; ou
  - déclaré en clair par l'inventaire de l'instance (`group_vars`, `host_vars`).

**Ce qu'il ne fait pas, et il faut le savoir** : il ne déchiffre aucune voûte, et il ne
tient pas compte des `when:` qui rendent une assertion conditionnelle. Un intrant exigé
seulement quand une option est active sera donc signalé même si l'option est éteinte.
Signaler à tort coûte une ligne de déclaration ; ne pas signaler coûte un déploiement.
"""

from __future__ import annotations

import argparse
import re
import sys
from pathlib import Path

import yaml

RACINE = Path(__file__).resolve().parent.parent

# Les deux idiomes d'exigence du depot. `| bool` n'en est pas un : une valeur fausse est
# une reponse valable, pas une absence.
MOTIFS = (
    re.compile(r"^([a-z][a-z0-9_]*)\s*\|\s*length\s*>\s*0\s*$"),
    re.compile(r"^([a-z][a-z0-9_]*)\s+is\s+defined\s*$"),
)


def _assertions(chemin: Path) -> set[str]:
    """Variables exigees par les `assert` d'un fichier de taches."""
    try:
        taches = yaml.safe_load(chemin.read_text(encoding="utf-8")) or []
    except yaml.YAMLError:
        return set()
    trouvees: set[str] = set()
    for tache in taches if isinstance(taches, list) else []:
        if not isinstance(tache, dict):
            continue
        bloc = tache.get("ansible.builtin.assert") or tache.get("assert")
        if not isinstance(bloc, dict):
            continue
        conditions = bloc.get("that") or []
        for c in [conditions] if isinstance(conditions, str) else conditions:
            for motif in MOTIFS:
                m = motif.match(str(c).strip())
                if m:
                    trouvees.add(m.group(1))
    return trouvees


def _defauts_non_vides(role: Path) -> set[str]:
    f = role / "defaults" / "main.yml"
    if not f.is_file():
        return set()
    try:
        d = yaml.safe_load(f.read_text(encoding="utf-8")) or {}
    except yaml.YAMLError:
        return set()
    return {k for k, v in d.items() if v not in (None, "", [], {})}


def _poses_par_set_fact() -> set[str]:
    """Variables qu'un role FABRIQUE a l'execution (resolveurs)."""
    poses: set[str] = set()
    for f in RACINE.glob("roles/*/tasks/*.yml"):
        try:
            taches = yaml.safe_load(f.read_text(encoding="utf-8")) or []
        except yaml.YAMLError:
            continue
        for tache in taches if isinstance(taches, list) else []:
            if not isinstance(tache, dict):
                continue
            bloc = tache.get("ansible.builtin.set_fact") or tache.get("set_fact")
            if isinstance(bloc, dict):
                poses |= {k for k in bloc if k != "cacheable"}
    return poses


def _cles_vars(noeud) -> set[str]:
    """Variables d'un inventaire YAML, a n'importe quelle profondeur.

    DEUX emplacements, et oublier le second faisait crier au manque a tort : les blocs
    `vars:` d'un groupe, ET les cles posees directement sous un nom d'hote dans `hosts:`.
    C'est la seconde forme que `instancier` emploie pour les valeurs derivees du plan.
    """
    trouvees: set[str] = set()
    if isinstance(noeud, dict):
        v = noeud.get("vars")
        if isinstance(v, dict):
            trouvees |= {k for k, val in v.items() if val not in (None, "", [], {})}
        h = noeud.get("hosts")
        if isinstance(h, dict):
            for attributs in h.values():
                if isinstance(attributs, dict):
                    trouvees |= {k for k, val in attributs.items()
                                 if val not in (None, "", [], {})}
        for sous in noeud.values():
            trouvees |= _cles_vars(sous)
    elif isinstance(noeud, list):
        for sous in noeud:
            trouvees |= _cles_vars(sous)
    return trouvees


def _declares_par_instance(base: Path) -> set[str]:
    """Cles declarees EN CLAIR par l'inventaire. Les voutes ne sont jamais dechiffrees.

    Le FICHIER d'inventaire compte autant que `group_vars/` : `instancier` y ecrit les
    valeurs derivees du plan — les liens entre applications, notamment. Ne regarder que
    les repertoires faisait crier au manque sur `serveur_postfix_mailstore_hote`, qui
    etait bel et bien fourni. Mesure du 2026-08-09, avant meme la premiere execution
    utile : un verificateur incomplet accuse le composant a la place de l'instrument.
    """
    declares: set[str] = set()
    for f in base.glob("inventories/*/hosts.yml"):
        try:
            declares |= _cles_vars(yaml.safe_load(f.read_text(encoding="utf-8")))
        except yaml.YAMLError:
            continue
    for f in list(base.glob("inventories/*/group_vars/**/*.yml")) + \
             list(base.glob("inventories/*/host_vars/**/*.yml")):
        texte = f.read_text(encoding="utf-8", errors="ignore")
        if texte.lstrip().startswith("$ANSIBLE_VAULT"):
            continue
        try:
            d = yaml.safe_load(texte) or {}
        except yaml.YAMLError:
            continue
        if isinstance(d, dict):
            declares |= {k for k, v in d.items() if v not in (None, "", [], {})}
    return declares


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    ap.add_argument("--instance", default=None, help="racine de l'instance (defaut : instance/)")
    a = ap.parse_args(argv)

    base = Path(a.instance) if a.instance else (RACINE / "instance")
    if not base.exists():
        print("Aucune instance montee : rien a verifier.", file=sys.stderr)
        return 2

    poses = _poses_par_set_fact()
    declares = _declares_par_instance(base)

    manquants: list[tuple[str, str]] = []
    exiges = 0
    for role in sorted(RACINE.glob("roles/*")):
        if not role.is_dir():
            continue
        defauts = _defauts_non_vides(role)
        vus: set[str] = set()
        for f in sorted(role.glob("tasks/*.yml")):
            vus |= _assertions(f)
        for var in sorted(vus):
            exiges += 1
            if var in defauts or var in poses or var in declares:
                continue
            manquants.append((role.name, var))

    if manquants:
        print(f"{len(manquants)} intrant(s) EXIGE(s) que l'instance ne fournit pas :\n")
        for r, v in manquants:
            print(f"  {r:26} exige `{v}` — sans defaut, sans set_fact, absent de l'inventaire")
        print("\nLe deploiement s'arretera sur la premiere garde atteinte.")
        return 1

    print(f"CONFORME : {exiges} exigence(s) de role, toutes satisfaites "
          f"({len(declares)} cle(s) declaree(s) par l'instance).")
    return 0


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