#!/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())