Set-OPS-Public/scripts/voute.py

#!/usr/bin/env python3
"""Confronte le GABARIT de voute d'une instance aux secrets que son plan exige.

Le gabarit `group_vars/all/vault.yml.example` est la seule liste que l'operateur
voit avant de creer sa voute. S'il est incomplet, le deploiement casse au milieu,
sur une variable indefinie ; s'il est bavard, l'operateur genere des secrets morts.

Les secrets EXIGES sont recenses a trois endroits, sans rien deviner :
  1. `plan/bases-donnees.yml` -> le champ `secret` de chaque base ;
  2. les roles des groupes actifs du plan -> toute reference `vault_*` ;
  3. les `group_vars` de l'instance -> toute reference `vault_*`.

La voute REELLE (vault.yml) n'est jamais lue ni dechiffree : on ne compare que
des NOMS, et seulement ceux du gabarit en clair.

Usage :
  python3 scripts/voute.py lister
  python3 scripts/voute.py verifier          # rc=2 si le gabarit est incomplet
  python3 scripts/voute.py verifier --strict # rc=2 aussi si des cles sont inutiles
"""

from __future__ import annotations

import argparse
import os
import re
import sys
from pathlib import Path

import yaml

from inventory_rules import (charger_applications, charger_bases_donnees, charger_serveurs,
                             integrations_universelles)

RACINE = Path(__file__).resolve().parents[1]
INSTANCE = Path(os.environ.get("SETOPS_INSTANCE") or (RACINE / "instance"))
ROLES = RACINE / "roles"
MOTIF_VAULT = re.compile(r"\b(vault_[a-z0-9_]+)")

# Secrets d'acces a l'hyperviseur : attendus au gabarit, mais hors motif vault_*.
SECRETS_HORS_MOTIF = {"proxmox_api_token_id", "proxmox_api_token_secret"}


def _inventaire_dir() -> Path | None:
    for nom in ("principal", "production", "lab"):
        d = INSTANCE / "inventories" / nom
        if d.is_dir():
            return d
    return None


def gabarit() -> Path | None:
    d = _inventaire_dir()
    if not d:
        return None
    p = d / "group_vars" / "all" / "vault.yml.example"
    return p if p.is_file() else None


def _refs_vault(chemin: Path) -> set[str]:
    try:
        return set(MOTIF_VAULT.findall(chemin.read_text(encoding="utf-8", errors="ignore")))
    except OSError:
        return set()


def groupes_du_plan() -> set[str]:
    apps = charger_applications(INSTANCE / "plan/applications.yml")
    srv = charger_serveurs(INSTANCE / "plan/serveurs.yml")
    groupes = {a.get("groupe") for a in (apps.get("applications") or {}).values() if a.get("groupe")}
    serveurs = (srv.get("serveurs") or {})
    for s in serveurs.values():
        groupes.update(s.get("integrations") or [])
    # Les integrations universelles ne figurent plus dans le plan (D-33) : sans
    # cette ligne, leurs secrets cesseraient d'etre exiges — et P18 passerait au
    # vert sur une voute incomplete. L'exemption `sauf_role` ne retire le role qu'a
    # UN hote : le groupe existe des qu'il reste un serveur, donc pas de filtrage.
    if serveurs:
        groupes.update(integrations_universelles())
    return groupes


def secrets_exiges() -> dict[str, set[str]]:
    """{secret: {origines}} — d'ou vient chaque exigence, pour un message utile."""
    exiges: dict[str, set[str]] = {}

    def noter(nom: str, origine: str) -> None:
        exiges.setdefault(nom, set()).add(origine)

    bases = charger_bases_donnees(INSTANCE / "plan/bases-donnees.yml")
    for cle, base in (bases.get("bases_donnees") or {}).items():
        if base.get("secret"):
            noter(str(base["secret"]), f"base '{cle}'")

    for groupe in sorted(groupes_du_plan()):
        dossier = ROLES / groupe
        if not dossier.is_dir():
            continue
        for f in list(dossier.rglob("*.yml")) + list(dossier.rglob("*.j2")):
            for nom in _refs_vault(f):
                noter(nom, f"role {groupe}")

    d = _inventaire_dir()
    if d and (d / "group_vars").is_dir():
        for f in (d / "group_vars").rglob("*.yml"):
            if f.name.endswith(".example"):
                continue
            for nom in _refs_vault(f):
                noter(nom, f"group_vars/{f.name}")

    return exiges


def cles_gabarit(path: Path) -> set[str]:
    data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
    return set(data) if isinstance(data, dict) else set()


def cles_voute_reelle() -> set[str] | None:
    """Noms des cles de la VRAIE voute, si elle est dechiffrable ici. Sinon None.

    Le gabarit dit ce qu'il FAUDRAIT ; il ne dit pas ce qui EST. Un tenant a vecu
    avec son jeton d'hyperviseur dans `proxmox.vault.yml` — un fichier gitignore,
    hors de la voute unique — pendant que le gabarit affichait « complet ». La
    verification du gabarit ne pouvait pas le voir.

    On ne lit QUE les noms de cles ; aucune valeur n'est retournee ni affichee.
    None (et non un ensemble vide) quand la voute est absente ou le mot de passe
    indisponible : c'est une verification qui se SAUTE, jamais une qui echoue —
    le harnais doit rester utilisable sans acces aux secrets.
    """
    chemin = None
    for nom in ("principal", "production", "lab"):
        p = INSTANCE / "inventories" / nom / "group_vars" / "all" / "vault.yml"
        if p.is_file():
            chemin = p
            break
    if chemin is None:
        return None
    brut = chemin.read_bytes()
    if not brut.startswith(b"$ANSIBLE_VAULT"):
        data = yaml.safe_load(brut.decode("utf-8", "ignore")) or {}
        return set(data) if isinstance(data, dict) else set()
    mdp_fichier = os.environ.get("ANSIBLE_VAULT_PASSWORD_FILE")
    if not mdp_fichier or not Path(os.path.expanduser(mdp_fichier)).is_file():
        return None
    try:
        from ansible.parsing.vault import VaultLib, VaultSecret
        mdp = Path(os.path.expanduser(mdp_fichier)).read_bytes().strip()
        clair = VaultLib([("default", VaultSecret(mdp))]).decrypt(brut)
        data = yaml.safe_load(clair.decode("utf-8")) or {}
        return set(data) if isinstance(data, dict) else set()
    except Exception:
        return None


def main() -> int:
    parser = argparse.ArgumentParser(description="Couverture du gabarit de voute Set-OPS.")
    sub = parser.add_subparsers(dest="commande", required=True)
    sub.add_parser("lister", help="Liste les secrets exiges et leur origine.")
    pv = sub.add_parser("verifier", help="Le gabarit couvre-t-il les secrets exiges ?")
    pv.add_argument("--strict", action="store_true",
                    help="Echouer aussi si le gabarit porte des cles inutiles.")
    args = parser.parse_args()

    g = gabarit()
    if not g:
        print("Aucun gabarit vault.yml.example : rien a verifier.")
        return 0

    exiges = secrets_exiges()
    presentes = cles_gabarit(g)

    if args.commande == "lister":
        for nom in sorted(exiges):
            etat = "au gabarit" if nom in presentes else "ABSENT du gabarit"
            print(f"{nom:<34} {etat:<18} <- {', '.join(sorted(exiges[nom]))}")
        return 0

    manquants = sorted(set(exiges) - presentes)
    inutiles = sorted(presentes - set(exiges) - SECRETS_HORS_MOTIF)

    if manquants:
        print(f"erreur: {len(manquants)} secret(s) exige(s) par le plan mais ABSENT(s) "
              f"du gabarit {g.relative_to(RACINE) if g.is_relative_to(RACINE) else g} :",
              file=sys.stderr)
        for nom in manquants:
            print(f"  - {nom}  <- {', '.join(sorted(exiges[nom]))}", file=sys.stderr)
    if inutiles:
        flux = sys.stderr if args.strict else sys.stdout
        print(f"{'erreur' if args.strict else 'note'}: {len(inutiles)} cle(s) du gabarit "
              f"que le plan n'exige plus : {', '.join(inutiles)}", file=flux)

    # Deuxieme volet : la voute REELLE, quand elle est lisible ici. Le gabarit dit
    # ce qu'il faudrait ; lui seul ne prouve rien sur ce qui est reellement en place.
    reelles = cles_voute_reelle()
    absentes_reelles: list[str] = []
    if reelles is not None:
        attendues = set(exiges) | SECRETS_HORS_MOTIF
        absentes_reelles = sorted(attendues - reelles)
        if absentes_reelles:
            print(f"erreur: {len(absentes_reelles)} secret(s) absent(s) de la VOUTE REELLE "
                  f"(group_vars/all/vault.yml) : {', '.join(absentes_reelles)}", file=sys.stderr)
            print("  Un secret servi depuis un autre fichier (proxmox.vault.yml, "
                  "proxmox.local.yml) n'est pas dans la voute unique — cf. D-19.", file=sys.stderr)

    if manquants or absentes_reelles or (inutiles and args.strict):
        return 2
    if reelles is None:
        print(f"Gabarit de voute complet : {len(exiges)} secret(s) exige(s), tous presents. "
              f"(Voute reelle non lisible ici : verification sautee.)")
    else:
        print(f"Gabarit de voute complet : {len(exiges)} secret(s) exige(s), tous presents. "
              f"Voute reelle : {len(reelles)} cle(s), aucun manque.")
    return 0


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