Set-OPS-Public/scripts/prouver.py

#!/usr/bin/env python3
"""Harnais de preuve Set-OPS : rejoue les preuves automatisables du registre.

`make prouver` enchaine les commandes de preuve du registre des affirmations
(`docs/audit/affirmations.md`) et produit un rapport horodate
`docs/audit/preuve-AAAA-MM-JJ.md` — rejouable, presentable comme piece justificative.

Ce script est un ORCHESTRATEUR MINCE : il ne reimplemente AUCUNE validation. Il
appelle l'outillage existant (les memes `scripts/*.py`, `ansible-lint`, `make syntaxe`
que `make verifier`) et met en forme les resultats. Ce n'est pas un framework de test
parallele : la logique de validation vit dans les scripts appeles.

Sortie : 0 si aucune preuve automatisable n'echoue (les preuves SAUTEES — ex. voute
requise absente — ne sont pas des echecs) ; 1 sinon.

Usage :
  python3 scripts/prouver.py            # execute + ecrit docs/audit/preuve-<date>.md
  python3 scripts/prouver.py --verifier # preuves seules (verdict + code sortie), sans rapport
"""

from __future__ import annotations

import datetime as _dt
import json
import os
import subprocess
import sys
from pathlib import Path

RACINE = Path(__file__).resolve().parents[1]
INSTANCE = Path(os.environ.get("SETOPS_INSTANCE", "instance"))
DOSSIER_AUDIT = RACINE / "docs" / "audit"
GROUPES = "playbooks/groupes"
DEPENDANCES = "docs/dependances-groupes.yml"


def _inventaire() -> str:
    """Inventaire de l'instance : premier existant (principal > production > lab)."""
    for nom in ("principal", "production", "lab"):
        p = INSTANCE / "inventories" / nom / "hosts.yml"
        if (RACINE / p).exists():
            return str(p)
    return str(INSTANCE / "inventories" / "principal" / "hosts.yml")


INV = _inventaire()


# --- Preuves natives (petites verifications d'invariants, pas de logique metier) ---

def preuve_handlers() -> tuple[bool, str]:
    """Tout `notify` d'un role pointe vers un handler du meme role (AGENTS.md)."""
    import glob
    import yaml

    problemes: list[str] = []
    roles_dir = RACINE / "roles"
    for role in sorted(p.name for p in roles_dir.iterdir() if p.is_dir()):
        rp = roles_dir / role
        handlers: set[str] = set()
        for hf in glob.glob(str(rp / "handlers" / "*.yml")):
            data = yaml.safe_load(Path(hf).read_text(encoding="utf-8")) or []

            def collect(tasks):
                for t in tasks or []:
                    if not isinstance(t, dict):
                        continue
                    if "name" in t:
                        handlers.add(t["name"])
                    if "listen" in t:
                        listen = t["listen"]
                        handlers.update(listen if isinstance(listen, list) else [listen])
                    for k in ("block", "rescue", "always"):
                        if k in t:
                            collect(t[k])

            collect(data)
        notifies: set[str] = set()
        for tf in glob.glob(str(rp / "tasks" / "*.yml")):
            data = yaml.safe_load(Path(tf).read_text(encoding="utf-8")) or []

            def scan(tasks):
                for t in tasks or []:
                    if not isinstance(t, dict):
                        continue
                    if "notify" in t:
                        n = t["notify"]
                        notifies.update(n if isinstance(n, list) else [n])
                    for k in ("block", "rescue", "always"):
                        if k in t:
                            scan(t[k])

            scan(data)
        manquants = notifies - handlers
        if manquants:
            problemes.append(f"{role}: notify sans handler -> {sorted(manquants)}")
    if problemes:
        return False, "; ".join(problemes)
    return True, "Tout notify pointe vers un handler du meme role (49 roles)."


def preuve_runbooks() -> tuple[bool, str]:
    """Les runbooks / registres cites par les surfaces publiques existent."""
    attendus = [
        "docs/plan-et-generation.md", "docs/catalogue-services.md",
        "docs/dependances-groupes.yml", "docs/dns-interne.md",
        "docs/nomenclature-vm.md", "docs/config-proxmox.md", "docs/vm-lifecycle.md",
        "docs/procedure-template-debian13-proxmox.md", "docs/positionnement.md",
        "docs/bindings-conception.md", "docs/flux-conception.md",
        "docs/couches-deploiement.yml", "docs/registre-flux.md",
        "docs/carte-set-ops.md", "docs/runbooks-exploitation.md",
        "exemples/vault.exemple.yml", "exemples/instance.exemple.yml",
    ]
    manquants = [f for f in attendus if not (RACINE / f).exists()]
    if manquants:
        return False, "manquants : " + ", ".join(manquants)
    return True, f"{len(attendus)}/{len(attendus)} runbooks/registres cites presents."


def preuve_structurels() -> tuple[bool, str]:
    """Invariants de structure/doctrine verifiables par le contenu du depot."""
    echecs: list[str] = []

    def lire(rel: str) -> str:
        p = RACINE / rel
        return p.read_text(encoding="utf-8") if p.exists() else ""

    # AFF-015 : LICENSE present
    if not (RACINE / "LICENSE").exists():
        echecs.append("LICENSE absent")
    # AFF-022 : modele socle en forme dossier (intrants charges, pas de shadowing)
    if not (RACINE / "exemples/modeles/socle/inventories/production/group_vars/all/10-intrants.yml").exists():
        echecs.append("socle: all/10-intrants.yml absent")
    if (RACINE / "exemples/modeles/socle/inventories/production/group_vars/all.yml").exists():
        echecs.append("socle: all.yml (forme fichier) masquerait all/")
    # AFF-038 : pas de playbooks de couches paralleles
    for couche in ("socle", "durcissement"):
        if (RACINE / "playbooks" / couche).exists():
            echecs.append(f"playbooks/{couche}/ ne doit pas exister")
    # AFF-037 : SSH clef-only des la construction
    if 'ssh_baseline_password_authentication: "no"' not in lire("roles/ssh_baseline/defaults/main.yml"):
        echecs.append("ssh_baseline: PasswordAuthentication != no")
    if "AuthenticationMethods publickey" not in lire("roles/ssh_baseline/templates/10-setops.conf.j2"):
        echecs.append("ssh_baseline: AuthenticationMethods publickey absent")
    # AFF-062 : nftables installe mais desactive par defaut
    if "nftables_baseline_enabled: false" not in lire("roles/nftables_baseline/defaults/main.yml"):
        echecs.append("nftables_baseline_enabled != false")
    if echecs:
        return False, "; ".join(echecs)
    return True, "LICENSE, socle dossier, pas de couches paralleles, SSH clef-only, nftables off : OK."


def preuve_modele_socle() -> tuple[bool, str]:
    """Le modele public `socle` valide (registres) — garde contre une regression d'exemple."""
    socle = "exemples/modeles/socle"
    env = {**os.environ, "SETOPS_INSTANCE": socle}
    for s in ("domaines", "serveurs", "applications", "bases_donnees"):
        r = subprocess.run([sys.executable, f"scripts/{s}.py", "verifier"],
                           cwd=RACINE, env=env, capture_output=True, text=True)
        if r.returncode != 0:
            sortie = (r.stdout + r.stderr).strip().splitlines()
            return False, f"{s}: {sortie[-1] if sortie else 'rc!=0'}"
    return True, "Modele public socle : domaines/serveurs/applications/bases valides."


def preuve_nomenclature_derivee() -> tuple[bool, str]:
    """Aucune nomenclature ne STOCKE d'adressage : tout se derive du seed `index`.

    Interdit `supernet` au niveau racine et `sous_reseau`/`passerelle`/`vlan` par zone.
    Attrape toute rechute vers l'ecriture manuelle de l'adressage (rupture du 2026-07-23).
    Portee : l'instance liee + tous les modeles decouverts (SETOPS_MODELES inclus).
    """
    import yaml
    sys.path.insert(0, str(RACINE / "scripts"))
    import modeles as _mod

    cibles = []
    inst_nom = RACINE / INSTANCE / "plan" / "nomenclature.yml"
    if inst_nom.exists():
        cibles.append(inst_nom)
    cibles += [m / "plan" / "nomenclature.yml" for m in _mod.decouvrir()
               if (m / "plan" / "nomenclature.yml").exists()]

    interdits_zone = {"sous_reseau", "passerelle", "vlan"}
    fautes = []
    for p in cibles:
        n = yaml.safe_load(p.read_text(encoding="utf-8")) or {}
        rel = p.relative_to(RACINE) if p.is_relative_to(RACINE) else p
        if "supernet" in n:
            fautes.append(f"{rel}: supernet stocke")
        if "vmid_schema" in n:
            fautes.append(f"{rel}: vmid_schema obsolete (mode compact retire)")
        for z, c in (n.get("categories") or {}).items():
            for k in interdits_zone & set(c or {}):
                fautes.append(f"{rel}: categorie {z}.{k} stocke")
    if fautes:
        return False, f"{len(fautes)} champ(s) d'adressage stockes : " + " | ".join(fautes[:4])
    return True, f"{len(cibles)} nomenclature(s) : adressage 100% derive du seed index."


def preuve_pas_lab_code_en_dur() -> tuple[bool, str]:
    """Aucun chemin `instance/inventories/lab/group_vars` code en dur (AFF-097).

    On vise le vrai chemin (prefixe `instance/`) et on ignore les fichiers meta qui
    *parlent* de l'audit (CHANGELOG, docs/audit) sans etre des chemins d'exploitation.
    """
    r = subprocess.run(
        ["git", "grep", "-n", "instance/inventories/lab/group_vars", "--", "*.md", "*.yml"],
        cwd=RACINE, capture_output=True, text=True,
    )
    exclus = ("docs/audit/", "CHANGELOG.md:")
    lignes = [l for l in r.stdout.splitlines() if not l.startswith(exclus)]
    if lignes:
        return False, f"{len(lignes)} occurrence(s) : " + " | ".join(lignes[:3])
    return True, "Aucun chemin instance/inventories/lab/group_vars code en dur."


def preuve_inventaire_ansible() -> tuple[bool, str]:
    """`ansible-inventory --list` : l'inventaire (voute dechiffree) se parse (AFF-030)."""
    r = subprocess.run(["ansible-inventory", "-i", INV, "--list"],
                       cwd=RACINE, capture_output=True, text=True, timeout=600)
    if r.returncode != 0:
        sortie = (r.stdout + r.stderr).strip().splitlines()
        return False, f"rc={r.returncode} : {sortie[-1] if sortie else ''}"
    try:
        data = json.loads(r.stdout)
    except ValueError:
        return True, "inventaire liste (JSON non parse)."
    n_hotes = len((data.get("_meta") or {}).get("hostvars") or {})
    n_groupes = len([k for k in data if k not in ("_meta", "all")])
    return True, f"{n_hotes} hotes, {n_groupes} groupes (inventaire dechiffre et parse)."


# --- Registre des preuves : (id, titre, refs AFF, executeur) -----------------------
#
# executeur = liste de commandes argv (toutes doivent renvoyer 0), ou callable -> (ok, detail).

PREUVES: list[dict] = [
    {"id": "P01", "titre": "Lint (ansible-lint)", "refs": ["AFF-006"],
     "cmds": [["ansible-lint", "-q"]]},
    {"id": "P02", "titre": "Tests unitaires (inventory_host)", "refs": [],
     "cmds": [[sys.executable, "scripts/tests/test_inventory_host.py"]]},
    {"id": "P03", "titre": "Diff-vide du plan (inventaire genere)", "refs": ["AFF-001", "AFF-004", "AFF-030", "AFF-031", "AFF-032"],
     "refs": ["AFF-001", "AFF-004", "AFF-030", "AFF-031", "AFF-032"],
     "cmds": [[sys.executable, "scripts/instancier.py", "generer"],
              [sys.executable, "scripts/instancier.py", "comparer"]]},
    {"id": "P04", "titre": "Groupes <-> playbooks homonymes", "refs": ["AFF-008"],
     "cmds": [[sys.executable, "scripts/inventory_host.py", "--inventaire", INV,
               "verifier-playbooks", "--dossier-playbooks", GROUPES]]},
    {"id": "P05", "titre": "Dependances causales de groupes", "refs": ["AFF-009", "AFF-084"],
     "cmds": [[sys.executable, "scripts/inventory_host.py", "--inventaire", INV,
               "--dependances", DEPENDANCES, "verifier-dependances",
               "--dossier-playbooks", GROUPES]]},
    {"id": "P06", "titre": "Validateurs de registres (serveurs/apps/bases/domaines)", "refs": ["AFF-003"],
     "refs": ["AFF-003"],
     "cmds": [[sys.executable, "scripts/serveurs.py", "verifier"],
              [sys.executable, "scripts/applications.py", "verifier"],
              [sys.executable, "scripts/bases_donnees.py", "verifier"],
              [sys.executable, "scripts/domaines.py", "verifier"]]},
    {"id": "P07", "titre": "GUI (node --check)", "refs": ["AFF-033"],
     "cmds": [[sys.executable, "scripts/verifier_gui.py"]]},
    {"id": "P08", "titre": "Orchestration (couches + graphe)", "refs": ["AFF-070"],
     "cmds": [[sys.executable, "scripts/orchestrer.py", "verifier"]]},
    {"id": "P09", "titre": "Flux reseau (schema + matrice)", "refs": ["AFF-071"],
     "cmds": [[sys.executable, "scripts/resoudre_flux.py", "verifier"]]},
    {"id": "P10", "titre": "Handlers <-> notify", "refs": ["AFF-034", "AFF-035"],
     "func": preuve_handlers},
    {"id": "P11", "titre": "Syntaxe des playbooks (--syntax-check)", "refs": ["AFF-083"],
     "cmds": [["make", "--no-print-directory", "syntaxe"]]},
    {"id": "P12", "titre": "Existence des runbooks cites", "refs": ["AFF-010", "AFF-011", "AFF-012"],
     "refs": ["AFF-010", "AFF-011", "AFF-012", "AFF-083"], "func": preuve_runbooks},
    {"id": "P13", "titre": "Invariants structurels/doctrinaux", "refs": ["AFF-015", "AFF-022", "AFF-037", "AFF-038", "AFF-062"],
     "refs": ["AFF-015", "AFF-022", "AFF-037", "AFF-038", "AFF-062"], "func": preuve_structurels},
    {"id": "P14", "titre": "Pas de chemin lab/ code en dur", "refs": ["AFF-097"],
     "func": preuve_pas_lab_code_en_dur},
    {"id": "P15", "titre": "Modele public socle valide", "refs": ["AFF-022", "AFF-099"],
     "func": preuve_modele_socle},
    {"id": "P17", "titre": "Tous les modeles valident (registres + underlay)", "refs": ["AFF-105"],
     "refs": ["AFF-022", "AFF-099"],
     "cmds": [[sys.executable, "scripts/modeles.py", "verifier"]]},
    {"id": "P18", "titre": "Gabarit de voute complet", "refs": ["AFF-026"],
     "cmds": [[sys.executable, "scripts/voute.py", "verifier"]]},
    {"id": "P19", "titre": "Le GUI couvre le schema du plan", "refs": ["AFF-106"],
     "refs": ["AFF-002", "AFF-095"],
     "cmds": [[sys.executable, "scripts/couverture_gui.py", "verifier",
               "--tolerer", "nomenclature"]]},
    {"id": "P20", "titre": "Adressage 100% derive du seed (aucun stocke)", "refs": ["AFF-101"],
     "refs": ["AFF-001", "AFF-003"], "func": preuve_nomenclature_derivee},
    {"id": "P21", "titre": "Federation : aucun index en collision", "refs": ["AFF-102"],
     "cmds": [[sys.executable, "scripts/instances.py", "--verifier"]]},
    {"id": "P22", "titre": "Plan de recette a jour (genere du wiki)", "refs": ["AFF-002"],
     "cmds": [[sys.executable, "scripts/plan_recette.py", "--verifier"]]},
    {"id": "P23", "titre": "Underlay sans collision avec la plage tenant", "refs": ["AFF-103"],
     "cmds": [[sys.executable, "scripts/underlay.py", "--verifier"]]},
    {"id": "P25", "titre": "Pare-feu Proxmox : est-ouest intra-tenant derive", "refs": ["AFF-107"],
     "cmds": [[sys.executable, "scripts/devis_proxmox_fw.py", "--verifier"]]},
    {"id": "P24", "titre": "Frontiere nord/sud : acces d'administration declare", "refs": ["AFF-104"],
     "cmds": [[sys.executable, "scripts/devis_opnsense.py", "--verifier"]]},
]


def _vault_requis_absent() -> bool:
    """Vrai si l'inventaire chiffre un group_vars et qu'aucun mot de passe n'est fourni."""
    if os.environ.get("ANSIBLE_VAULT_PASSWORD_FILE"):
        return False
    gv = (RACINE / INV).parent / "group_vars"
    if not gv.exists():
        return False
    for f in gv.rglob("*"):
        if f.is_file():
            try:
                if f.read_text(encoding="utf-8", errors="ignore").startswith("$ANSIBLE_VAULT"):
                    return True
            except OSError:
                continue
    return False


def _underlay_absent() -> bool:
    """Vrai si aucun underlay.yml n'est defini (fabric non decrite : rien a prouver)."""
    import underlay
    return underlay.chemin() is None


def _frontiere_absente() -> bool:
    """Vrai si aucune instance n'est active (depot public nu : aucune frontiere a prouver)."""
    return not (RACINE / "instance").exists()


def _executer(preuve: dict) -> tuple[str, str]:
    """Renvoie (statut, detail). statut in {OK, ECHEC, SAUTE}."""
    if "func" in preuve:
        ok, detail = preuve["func"]()
        return ("OK" if ok else "ECHEC"), detail
    dernier = ""
    for cmd in preuve["cmds"]:
        r = subprocess.run(cmd, cwd=RACINE, capture_output=True, text=True, timeout=600)
        sortie = [l for l in (r.stdout + r.stderr).strip().splitlines()
                  if l.strip() and not l.lstrip().startswith("[WARNING]")]
        dernier = sortie[-1] if sortie else ""
        if r.returncode != 0:
            return "ECHEC", f"rc={r.returncode} : {dernier}"
    return "OK", dernier


def main(argv: list[str] | None = None) -> int:
    argv = sys.argv[1:] if argv is None else argv
    # --verifier : execute les preuves (verdict + code de sortie) sans ecrire de rapport,
    # pour un appel depuis `make verifier` (evite d'ecraser la piece justificative committee).
    ecrire_rapport = "--verifier" not in argv
    date = _dt.date.today().isoformat()
    resultats: list[tuple[dict, str, str]] = []

    # Preuve conditionnelle : inventaire Ansible complet (necessite la voute).
    preuve_inv = {"id": "P16", "titre": "Inventaire Ansible complet (--list)",
                  "refs": ["AFF-030"], "func": preuve_inventaire_ansible}
    liste = list(PREUVES)
    if _vault_requis_absent():
        resultats.append((preuve_inv, "SAUTE",
                          "Voute chiffree sans ANSIBLE_VAULT_PASSWORD_FILE (prerequis AFF-026)."))
    else:
        liste.append(preuve_inv)

    # Preuve conditionnelle : underlay (fabric physique). Sautee si non definie.
    if _underlay_absent():
        p23 = next((p for p in liste if p["id"] == "P23"), None)
        if p23:
            liste.remove(p23)
            resultats.append((p23, "SAUTE", "Aucun underlay.yml (fabric non definie) — rien a prouver."))

    # Preuve conditionnelle : frontiere nord/sud. Sautee sans instance active.
    if _frontiere_absente():
        p24 = next((p for p in liste if p["id"] == "P24"), None)
        if p24:
            liste.remove(p24)
            resultats.append((p24, "SAUTE", "Aucune instance active — aucune frontiere a prouver."))

    for preuve in liste:
        statut, detail = _executer(preuve)
        resultats.append((preuve, statut, detail))

    # Tri : garder l'ordre d'execution (P01..P19) ; la preuve sautee a ete inseree tot.
    resultats.sort(key=lambda r: r[0]["id"])

    n_ok = sum(1 for _, s, _ in resultats if s == "OK")
    n_echec = sum(1 for _, s, _ in resultats if s == "ECHEC")
    n_saute = sum(1 for _, s, _ in resultats if s == "SAUTE")
    conforme = n_echec == 0
    print(f"{'CONFORME' if conforme else 'NON CONFORME'} : "
          f"{n_ok} OK, {n_echec} echec, {n_saute} saute.")
    if not ecrire_rapport:
        return 0 if conforme else 1

    icone = {"OK": "✅", "ECHEC": "❌", "SAUTE": "⚪"}

    lignes = [
        f"# Preuve de conformite — Set-OPS — {date}",
        "",
        "> Genere par `make prouver` (`scripts/prouver.py`). **Rejouable** : relancer",
        "> reproduit ce rapport. Chaque preuve rejoue l'outillage existant du depot ;",
        "> aucune validation n'est reimplementee ici. Voir le mode d'emploi :",
        "> [`docs/audit/README.md`](README.md), et le registre trace :",
        "> [`docs/audit/affirmations.md`](affirmations.md).",
        "",
        f"- **Instance** : `{INSTANCE}` — inventaire `{INV}`",
        f"- **Verdict** : {'✅ CONFORME' if conforme else '❌ NON CONFORME'} "
        f"({n_ok} OK · {n_echec} echec · {n_saute} saute)",
        "",
        "## Preuves",
        "",
        "| # | Preuve | Affirmations | Statut | Detail |",
        "|---|---|---|---|---|",
    ]
    for preuve, statut, detail in resultats:
        refs = ", ".join(preuve["refs"]) or "—"
        det = detail.replace("|", "\\|")[:160]
        lignes.append(f"| {preuve['id']} | {preuve['titre']} | {refs} | {icone[statut]} {statut} | {det} |")

    lignes += [
        "",
        "## Couverture des affirmations ✅ du registre",
        "",
        "Chaque affirmation ✅ automatisable est couverte par la preuve indiquee ci-dessus.",
        "Les ✅ **structurelles/doctrinales** non rejouables par une commande (ex. AFF-005",
        "`make`=aide, AFF-014 ciblage groupe, AFF-024 `instancier-appliquer`, AFF-051 autorite",
        "d'AGENTS.md, AFF-073/075 gardes `make`, AFF-090 wiki) ont ete verifiees a l'audit ;",
        "elles restent hors du harnais recurrent (rien d'executable a rejouer).",
        "",
        "## Declarations d'intention (⚪ invérifiables localement — assumees)",
        "",
        "Ces affirmations ne sont pas rejouables hors production ; elles sont **assumees**",
        "comme declarations d'intention, non comme preuves :",
        "",
        "- **AFF-036** — « testables avec `--check` autant que possible » : verifiable seulement",
        "  contre une flotte vivante.",
        "- **AFF-091** — contenu pedagogique du wiki : affirmations conceptuelles.",
        "- **AFF-096** — « GUI 100 % francais » : revue exhaustive des libelles rendus, non automatisee.",
        "- **AFF-007** — hote d'exemple `web-frontal-01` : placeholder assume.",
        "",
        f"_Rapport genere le {date}._",
        "",
    ]

    DOSSIER_AUDIT.mkdir(parents=True, exist_ok=True)
    rapport = DOSSIER_AUDIT / f"preuve-{date}.md"
    rapport.write_text("\n".join(lignes), encoding="utf-8")
    print(f"Rapport : {rapport.relative_to(RACINE)}")
    return 0 if conforme else 1


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))