Set-OPS-Public/scripts/tests/test_empreinte_plancher.py

#!/usr/bin/env python3
"""L'empreinte que la zone publie est celle que la sonde calcule sur le plancher rendu.

POURQUOI (2026-09-29). La zone PowerDNS publie `_plancher.<domaine> TXT "n=… sha256=…"` :
ce que le plancher `/etc/hosts` DOIT porter pour ce domaine. La sonde `plancher` calcule la
meme chose sur /etc/hosts. Les deux calculs vivent dans deux gabarits de deux roles
(`serveur_powerdns/templates/zone.db.j2`, `hosts_statiques/templates/hosts.j2`) et dans le
script de la sonde (`client_sante/templates/sonde-plancher.sh.j2`) : une liste qui en suit
une autre prend du retard. Si l'un change seul, chaque machine crierait « divergent » sur
une flotte saine.

On rend donc les deux gabarits sur UN MEME inventaire fictif — hotes actifs ET planifies
(le plancher resout un clone avant la zone), expositions de deux domaines, un hote sans
adresse — et on execute le code Python de la sonde sur le plancher rendu. Les empreintes
doivent etre identiques. Controle negatif : un nom retire du plancher doit la changer.
"""
from __future__ import annotations

import hashlib
import re
import subprocess
import sys
import tempfile
from pathlib import Path

import jinja2

ROLES = Path(__file__).resolve().parents[2] / "roles"
DOMAINE = "exemple.internal"


def environnement() -> jinja2.Environment:
    env = jinja2.Environment(undefined=jinja2.ChainableUndefined, extensions=["jinja2.ext.do"])
    env.filters["hash"] = lambda v, algo="sha1": hashlib.new(algo, str(v).encode()).hexdigest()
    env.filters["bool"] = bool
    return env


def contexte() -> dict:
    hostvars = {
        "mon-01": {"ansible_host": "10.9.20.21"},
        "edge-01": {"ansible_host": "10.9.16.11"},
        "dns-01": {"ansible_host": "10.9.19.11"},
        "neuf-01": {"ansible_host": "10.9.21.99"},   # planifie : au plancher, pas en A
        "sans-ip": {},
    }
    expositions = [
        {"fqdn": f"console.{DOMAINE}", "domaine": DOMAINE, "edge": "serveur_nginx"},
        {"fqdn": f"vigie.{DOMAINE}", "domaine": DOMAINE, "edge": "serveur_nginx"},
        {"fqdn": "www.exemple.ca", "domaine": "exemple.ca", "edge": "serveur_nginx"},
        # `resolution_interne: service` : ce nom mene au service, pas a l'edge.
        {"fqdn": f"forge.{DOMAINE}", "domaine": DOMAINE, "edge": "serveur_nginx",
         "hote": "mon-01", "resout_vers": "service"},
    ]
    return {
        "groups": {"hotes_actifs": ["mon-01", "edge-01", "dns-01", "sans-ip"],
                   "hotes_planifies": ["neuf-01"], "serveur_nginx": ["edge-01"],
                   "serveur_powerdns": ["dns-01"]},
        "hostvars": hostvars,
        # plancher
        "hosts_statiques_domaine": DOMAINE, "hosts_statiques_publier_expositions": True,
        "hosts_statiques_expositions": expositions, "hosts_statiques_externes": [],
        # zone
        "serveur_powerdns_zone": DOMAINE, "serveur_powerdns_publier_expositions": True,
        "serveur_powerdns_expositions": expositions, "serveur_powerdns_records": [],
        "serveur_powerdns_inventory_groups": ["hotes_actifs"], "serveur_powerdns_nameserver": "ns1",
    }


def code_de_la_sonde() -> str:
    gabarit = (ROLES / "client_sante/templates/sonde-plancher.sh.j2").read_text(encoding="utf-8")
    m = re.search(r"<<'PY'\n(.*?)\nPY\n", gabarit, re.S)
    assert m, "bloc Python introuvable dans sonde-plancher.sh.j2"
    return m.group(1)


def empreinte_sonde(plancher: str) -> str:
    with tempfile.NamedTemporaryFile("w", suffix=".hosts", delete=False) as f:
        f.write(plancher)
    r = subprocess.run([sys.executable, "-", f.name, DOMAINE], input=code_de_la_sonde(),
                       capture_output=True, text=True)
    Path(f.name).unlink()
    return r.stdout.splitlines()[0]


def main() -> int:
    env, ctx = environnement(), contexte()
    plancher = env.from_string((ROLES / "hosts_statiques/templates/hosts.j2")
                               .read_text(encoding="utf-8")).render(**ctx)
    zone = env.from_string((ROLES / "serveur_powerdns/templates/zone.db.j2")
                           .read_text(encoding="utf-8")).render(**ctx)
    m = re.search(r'^_plancher IN TXT "([^"]+)"', zone, re.M)
    if not m:
        print("ECHEC  la zone ne publie pas `_plancher IN TXT`.")
        return 1
    publie, calcule = m.group(1), empreinte_sonde(plancher)
    echecs = 0
    if publie != calcule:
        print(f"ECHEC  la zone publie « {publie} », la sonde calcule « {calcule} » sur le plancher "
              f"rendu du MEME inventaire : les deux gabarits ne comptent plus la meme chose.")
        echecs += 1
    if not publie.startswith("n=7 "):
        print(f"ECHEC  attendu 7 paires (4 hotes avec adresse dont 1 planifie, 3 expositions du "
              f"domaine) ; publie : {publie}")
        echecs += 1
    if f"10.9.20.21      forge.{DOMAINE}" not in plancher or f"forge IN A 10.9.20.21" not in zone:
        print("ECHEC  `resout_vers: service` : forge doit mener a mon-01 (10.9.20.21) au plancher "
              "ET dans la zone.")
        echecs += 1
    ampute = "\n".join(l for l in plancher.splitlines() if "console." not in l)
    if empreinte_sonde(ampute) == publie:
        print("ECHEC  controle negatif : un plancher sans `console` rend la meme empreinte.")
        echecs += 1
    if echecs:
        return 1
    print(f"OK — zone et plancher rendus sur le meme inventaire : meme empreinte ({publie[:18]}…) ; "
          f"un nom retire la change.")
    return 0


if __name__ == "__main__":
    sys.exit(main())