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 getpass
import os
import re
import sys
from pathlib import Path

import yaml

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

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

# Secrets du MONDE PHYSIQUE : ils n'ont pas le prefixe `vault_`, et depuis le 2026-08-22
# ils ne vivent PLUS dans la voute d'un tenant mais dans celle de l'underlay, chez
# l'hebergeur (docs/frontiere-physique-virtuel.md). Un tenant qui les porte encore est un
# reste de migration, pas une exigence — d'ou leur retrait des attentes du gabarit.
SECRETS_MONDE_PHYSIQUE = {"proxmox_api_token_id", "proxmox_api_token_secret",
                          "vault_opnsense_api_key", "vault_opnsense_api_secret"}
SECRETS_HORS_MOTIF: set[str] = set()


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]:
    """Les `vault_*` qu'un fichier UTILISE — pas ceux dont il PARLE.

    ON IGNORE LES COMMENTAIRES (2026-08-24). Le scanner lisait le fichier entier, si bien
    qu'expliquer en commentaire « pour ouvrir cette option, fournir vault_x » SUFFISAIT a
    rendre `vault_x` obligatoire dans le gabarit et dans la voute reelle.

    Constate sur `serveur_collabora` : sa console d'administration est FERMEE par defaut,
    et le role se contentait de documenter comment l'ouvrir. Trois phrases de commentaire
    imposaient donc a tout ecosysteme installant Collabora d'inventer, chiffrer et faire
    tourner un secret pour une porte que personne n'ouvre.

    Documenter le nom d'une clef ne doit pas l'exiger. Une clef reellement utilisee
    apparait dans du code — c'est ce qu'on lit ici.
    """
    try:
        texte = chemin.read_text(encoding="utf-8", errors="ignore")
    except OSError:
        return set()
    # Suffisant pour YAML, Jinja (`{# … #}` compris) et Python : ce qui suit un `#` hors
    # guillemets est de la prose. On ne cherche pas a analyser finement — un `#` dans une
    # chaine rendrait au pire une reference ignoree, et une reference reellement utilisee
    # apparait ailleurs dans du code.
    utile = "\n".join(ligne.split("#", 1)[0] for ligne in texte.splitlines())
    return set(MOTIF_VAULT.findall(utile))


def _cles_dnssec() -> list[str]:
    """Une cle par zone `primaire-cache` signee du plan. Meme nom que `dnssec.py` et le role."""
    import re as _re
    try:
        doms = (yaml.safe_load((INSTANCE / "plan" / "domaines.yml").read_text(encoding="utf-8"))
                or {}).get("domaines_publics") or {}
    except OSError:
        return []
    return sorted("vault_dnssec_" + _re.sub(r"[^a-z0-9]", "_", str(z).lower().rstrip("."))
                  for z, c in doms.items()
                  if isinstance(c, dict) and c.get("autorite") == "primaire-cache"
                  and c.get("dnssec") is True)


FAMILLES_DERIVEES = {"vault_dnssec_": _cles_dnssec}


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}")

    # Roles appliques PAR UN PLAYBOOK sans etre eux-memes un groupe (ex.
    # `amorcage_acces`, porte par le playbook de l'annuaire). Sans ce passage,
    # leurs secrets echappaient au recensement — ce que D-20 interdit : « la
    # liste des secrets se recense, elle ne s'ecrit pas ». Constate le 2026-08-07.
    for pb in sorted((RACINE / "playbooks" / "groupes").glob("*.yml")):
        try:
            docs = yaml.safe_load(pb.read_text(encoding="utf-8")) or []
        except yaml.YAMLError:
            continue
        for jeu in docs if isinstance(docs, list) else [docs]:
            for r in (jeu or {}).get("roles") or []:
                nom_role = r if isinstance(r, str) else (r or {}).get("role", "")
                dossier = ROLES / str(nom_role)
                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 {nom_role} (playbook {pb.name})")

    # LES FAMILLES DE SECRETS DONT LE NOM SE DERIVE DU PLAN (2026-09-16). Un role qui compose
    # `'vault_dnssec_' ~ zone` laisse ici un PREFIXE : le lire comme un nom exigeait une clef
    # `vault_dnssec_` que personne ne peut fournir. Le taire aurait ete pire — la voute d'un
    # locataire aurait pu manquer la cle d'une zone signee sans qu'aucune preuve le dise.
    # Chaque famille connue est donc DEPLIEE depuis le plan ; une famille inconnue reste
    # exigee telle quelle, et l'echec qui en resulte la nomme.
    for prefixe, derive in FAMILLES_DERIVEES.items():
        origines = exiges.pop(prefixe, None)
        if origines is None:
            continue
        for nom in derive():
            for origine in origines:
                noter(nom, f"{origine}, derive du plan")

    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 _voute_reelle() -> Path | None:
    for nom in ("principal", "production", "lab"):
        p = INSTANCE / "inventories" / nom / "group_vars" / "all" / "vault.yml"
        if p.is_file():
            return p
    return None


def saisir(noms: list[str], remplacer: bool = False) -> int:
    """Ecrit dans la voute des secrets EMIS PAR UN TIERS, saisis sans echo.

    Pendant de la generation, et sa contrepartie exacte : on GENERE un secret dont
    le depot est la source (Ansible configure les deux cotes depuis la meme
    variable) ; on SAISIT un secret dont un tiers est la source. Inventer une cle
    d'API OPNsense donnerait une valeur syntaxiquement correcte et refusee a la
    premiere requete — pire qu'une absence, puisque la garde passerait au vert.

    Rien n'est affiche, rien n'est ecrit en clair sur disque, rien ne passe par la
    ligne de commande (donc ni par l'historique du shell, ni par la liste des
    processus). Le resultat est reparse et redechiffre avant d'etre pose.
    """
    from ansible.parsing.vault import VaultLib, VaultSecret

    cible = _voute_reelle()
    if cible is None:
        print("erreur: aucune voute group_vars/all/vault.yml dans cette instance.",
              file=sys.stderr)
        return 2
    fichier_mdp = os.environ.get("ANSIBLE_VAULT_PASSWORD_FILE")
    if not fichier_mdp or not Path(os.path.expanduser(fichier_mdp)).is_file():
        print("erreur: ANSIBLE_VAULT_PASSWORD_FILE absent ou introuvable.", file=sys.stderr)
        return 2
    mdp = Path(os.path.expanduser(fichier_mdp)).read_bytes().strip()
    vault = VaultLib([("default", VaultSecret(mdp))])

    brut = cible.read_bytes()
    texte = (vault.decrypt(brut) if brut.startswith(b"$ANSIBLE_VAULT") else brut).decode("utf-8")
    data = yaml.safe_load(texte) or {}

    a_poser: dict[str, str] = {}
    for nom in noms:
        if str(data.get(nom, "")).strip() and not remplacer:
            print(f"  {nom} : deja renseigne, ignore (--remplacer pour ecraser).")
            continue
        while True:
            v = getpass.getpass(f"  {nom} : ")
            if not v.strip():
                print("    valeur vide refusee.")
                continue
            if getpass.getpass(f"  {nom} (confirmer) : ") != v:
                print("    les deux saisies different, on recommence.")
                continue
            break
        a_poser[nom] = v
    if not a_poser:
        print("Rien a ecrire.")
        return 0

    nouveau = (texte.rstrip("\n") + "\n\n"
               + "# --- Saisis a la main (secrets emis par un tiers) ---\n"
               + yaml.safe_dump(a_poser, default_flow_style=False,
                                allow_unicode=True, sort_keys=True))
    attendu = {**data, **a_poser}
    if yaml.safe_load(nouveau) != attendu:
        print("erreur: le YAML fusionne ne reproduit pas l'union attendue.", file=sys.stderr)
        return 2
    chiffre = vault.encrypt(nouveau.encode("utf-8"))
    if not chiffre.endswith(b"\n"):
        chiffre += b"\n"
    if yaml.safe_load(vault.decrypt(chiffre).decode()) != attendu:
        print("erreur: aller-retour de chiffrement infidele, rien n'a ete ecrit.", file=sys.stderr)
        return 2
    cible.write_bytes(chiffre)
    print(f"Ecrit et verifie : {len(a_poser)} secret(s) -> {cible} "
          f"({len(attendu)} cle(s) au total).")
    return 0


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.")
    ps = sub.add_parser("saisir", help="Saisir sans echo des secrets emis par un tiers.")
    ps.add_argument("noms", nargs="+", help="Noms des cles a saisir (ex. vault_opnsense_api_key).")
    ps.add_argument("--remplacer", action="store_true",
                    help="Ecraser une cle deja renseignee.")
    args = parser.parse_args()

    if args.commande == "saisir":
        return saisir(args.noms, args.remplacer)

    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
                      - SECRETS_MONDE_PHYSIQUE)

    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) - SECRETS_MONDE_PHYSIQUE
        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())