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

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