#!/usr/bin/env python3
"""`contexte.py` : un tronc commun, deux classes — et le contexte actif, nomme ou devine.

Sur une arborescence FABRIQUEE (un site, deux locataires, un depot ambigu), on verifie :
  - qu'un depot dit ce qu'il est (site, locataire), et qu'un depot qui dit les deux est refuse ;
  - ce que chaque classe surcharge : index, voute, inventaire ;
  - les relations : `site.locataires()` (et les absents, nommes), `locataire.site()` ;
  - le contexte actif : le fichier `contexte` fait foi ; sans lui, les indices d'avant, et le
    locataire l'emporte quand les deux sont montes (le poste d'aujourd'hui) ;
  - un contexte nomme qui ne designe rien est un REFUS, pas un contexte vide.
Puis, si les vrais depots sont la : chaque locataire declare par un site porte l'index que ce
site lui attribue, et nomme un hebergeur qui existe ; la fiche de chaque hebergeur dit
exactement ce que son locataire porte, et une fiche alteree (sept facons) est vue en ecart ; la face
reseau de chaque locataire dit ce que les consommateurs du site en tirent, et une face alteree
(huit facons) est vue en ecart ; machine par machine, le pare-feu du locataire et celui de
Proxmox admettent les memes entrees, et des flux alteres (quatre facons) sont vus en ecart ;
la frontiere designe chaque locataire par les memes identites, et une identite alteree (cinq
facons) se voit ; ce qui entre depuis l'Internet concorde, et une entree alteree (quatre facons)
se voit ; ce que l'administration atteint concorde, et une administration alteree (quatre
facons) se voit ; ce qui sort vers l'Internet concorde, et une sortie alteree se voit.
"""
from __future__ import annotations

import os
import sys
import tempfile
from pathlib import Path

RACINE = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(RACINE / "scripts"))

import contexte as C  # noqa: E402

ECHECS: list[str] = []


def verifier(cond: bool, msg: str) -> None:
    print(("OK     " if cond else "ECHEC  ") + msg)
    if not cond:
        ECHECS.append(msg)


def refuse(fonction, msg: str) -> None:
    try:
        fonction()
    except C.ContexteInvalide:
        verifier(True, msg)
        return
    verifier(False, msg + " (aucun refus)")


def ecrire(chemin: Path, texte: str) -> None:
    chemin.parent.mkdir(parents=True, exist_ok=True)
    chemin.write_text(texte, encoding="utf-8")


def fabriquer(d: Path) -> Path:
    """Un dossier d'ecosystemes : un moteur, un site, deux locataires, un depot ambigu."""
    moteur = d / "Moteur"
    moteur.mkdir()
    ecrire(d / "SITE-Essai" / "underlay.yml",
           "underlay:\n  index: 9\n  tenants:\n    OPS-Un: 5\n    OPS-Absent: 6\n")
    ecrire(d / "SITE-Essai" / "underlay.vault.yml", "$ANSIBLE_VAULT;1.1;AES256\n")
    ecrire(d / "OPS-Un" / "plan" / "nomenclature.yml", "index: 5\n")
    ecrire(d / "OPS-Un" / "inventories" / "principal" / "hosts.yml", "all: {}\n")
    ecrire(d / "OPS-Un" / "inventories" / "principal" / "group_vars" / "all" / "vault.yml", "x\n")
    ecrire(d / "OPS-Un" / "parente.yml",
           "parente:\n  depots:\n    hebergeur:\n      nom: SITE-Essai\n")
    ecrire(d / "OPS-Deux" / "plan" / "nomenclature.yml", "index: 7\n")
    ecrire(d / "OPS-Fantome" / "plan" / "nomenclature.yml", "index: 8\n")
    ecrire(d / "OPS-Fantome" / "parente.yml",
           "parente:\n  depots:\n    hebergeur:\n      nom: SITE-Inexistant\n")
    ecrire(d / "Ambigu" / "underlay.yml", "underlay: {index: 1}\n")
    ecrire(d / "Ambigu" / "plan" / "nomenclature.yml", "index: 1\n")
    return moteur


def classes(d: Path) -> None:
    site = C.Ecosysteme.depuis(d / "SITE-Essai")
    un = C.Ecosysteme.depuis(d / "OPS-Un")
    verifier(isinstance(site, C.Site) and site.nature == "site", "un depot a `underlay.yml` est un Site")
    verifier(isinstance(un, C.Locataire) and un.nature == "locataire",
             "un depot a nomenclature est un Locataire")
    refuse(lambda: C.Ecosysteme.depuis(d / "Ambigu"),
           "un depot qui porte `underlay.yml` ET une nomenclature est refuse")
    refuse(lambda: C.Ecosysteme.depuis(d / "Moteur"), "un depot qui ne declare rien est refuse")

    verifier(site.index == 9 and un.index == 5, "chaque classe lit SON index (underlay / nomenclature)")
    verifier(site.voute == d / "SITE-Essai" / "underlay.vault.yml" and site.a_sa_voute(),
             "la voute d'un site est `underlay.vault.yml`")
    verifier(un.voute == d / "OPS-Un" / "inventories" / "principal" / "group_vars" / "all" / "vault.yml"
             and un.a_sa_voute(), "la voute d'un locataire est dans son inventaire")
    verifier(un.inventaire() == d / "OPS-Un" / "inventories" / "principal" / "hosts.yml",
             "l'inventaire d'un locataire est son `hosts.yml` genere")
    verifier(site.inventaire().name == "site_inventaire.py",
             "l'inventaire d'un site est un script, pas un `hosts.yml`")
    verifier(un.cle_voute.name == "setops-vault-ops-un", "la cle de voute est nommee d'apres le depot")

    verifier(site.allocations() == {"OPS-Un": 5, "OPS-Absent": 6}, "le site rend les index qu'il attribue")
    verifier([l.nom for l in site.locataires()] == ["OPS-Un"],
             "`locataires()` rend les locataires presents")
    verifier(site.locataires_absents() == ["OPS-Absent"], "un locataire declare sans depot est NOMME absent")
    verifier(un.site() == site, "`locataire.site()` rend l'hebergeur que nomme `parente.yml`")
    verifier(C.Locataire(d / "OPS-Deux").site() is None, "sans hebergeur nomme, `site()` rend None")
    refuse(lambda: C.Locataire(d / "OPS-Fantome").site(), "un hebergeur nomme mais absent est refuse")
    refuse(lambda: C.Site.charger("OPS-Un", d), "`Site.charger` refuse un locataire")
    refuse(lambda: C.Locataire.charger("SITE-Essai", d), "`Locataire.charger` refuse un site")


def contexte_actif(d: Path, moteur: Path) -> None:
    sauve = {k: os.environ.pop(k) for k in ("SETOPS_INSTANCE", "SETOPS_UNDERLAY") if k in os.environ}
    try:
        verifier(C.actif(moteur) is None, "rien de monte, rien de nomme : aucun contexte (None)")

        (moteur / "underlay.yml").symlink_to(d / "SITE-Essai" / "underlay.yml")
        verifier(C.actif(moteur) == C.Site(d / "SITE-Essai"), "le lien `underlay.yml` seul designe le site")

        (moteur / "instance").symlink_to(d / "OPS-Un")
        verifier(len(C.montes(moteur)) == 2, "les deux liens montes : deux ecosystemes designes")
        verifier(C.actif(moteur) == C.Locataire(d / "OPS-Un"),
                 "les deux montes : le locataire l'emporte, comme la console d'aujourd'hui")

        os.environ["SETOPS_INSTANCE"] = str(d / "OPS-Deux")
        verifier(C.actif(moteur) == C.Locataire(d / "OPS-Deux"), "`SETOPS_INSTANCE` passe avant le lien")
        del os.environ["SETOPS_INSTANCE"]

        fichier = moteur / C.NOM_FICHIER_CONTEXTE
        fichier.write_text("# le contexte de ce poste\nsite:SITE-Essai\n", encoding="utf-8")
        verifier(C.actif(moteur) == C.Site(d / "SITE-Essai"),
                 "le fichier `contexte` fait foi, meme quand les liens disent autre chose")
        fichier.write_text("locataire: OPS-Un\n", encoding="utf-8")
        verifier(C.actif(moteur) == C.Locataire(d / "OPS-Un"), "`locataire:<depot>` designe le locataire")

        for contenu, msg in (("site:SITE-Inexistant\n", "un site nomme qui n'existe pas"),
                             ("locataire:SITE-Essai\n", "une nature qui ne correspond pas au depot"),
                             ("hebergeur:SITE-Essai\n", "une nature inconnue"),
                             ("site:SITE-Essai\nlocataire:OPS-Un\n", "deux valeurs"),
                             ("\n# rien\n", "un fichier vide")):
            fichier.write_text(contenu, encoding="utf-8")
            refuse(lambda: C.actif(moteur), f"fichier `contexte` refuse : {msg}")
    finally:
        os.environ.pop("SETOPS_INSTANCE", None)
        os.environ.update(sauve)


def vrais_depots() -> None:
    """Sur les depots reels, s'ils sont la : les deux sens de la relation concordent."""
    sites = [C.Site(p.parent) for p in sorted(C.DOSSIER_ECOSYSTEMES.glob("SITE-*/underlay.yml"))]
    if not sites:
        print("SAUTE  aucun site frere : rien de reel a confronter")
        return
    for site in sites:
        for loc in site.locataires():
            attendu = site.allocations().get(loc.nom)
            verifier(attendu is None or loc.index == attendu,
                     f"{site.nom} attribue {attendu} a {loc.nom}, qui porte l'index {loc.index}")
            heb = loc.nom_hebergeur()
            if heb:
                verifier((C.DOSSIER_ECOSYSTEMES / heb / "underlay.yml").is_file(),
                         f"{loc.nom} nomme l'hebergeur {heb}, qui existe")
        verifier(not site.locataires_absents(),
                 f"{site.nom} : aucun locataire declare sans depot ({site.locataires_absents() or 'ok'})")


def monter_restaure() -> None:
    avant = {k: os.environ.get(k) for k in ("SETOPS_INSTANCE", "SETOPS_UNDERLAY")}
    try:
        with C._monter(C.Site(Path("/nulle/SITE")), C.Locataire(Path("/nulle/OPS"))):
            dedans = (os.environ.get("SETOPS_UNDERLAY"), os.environ.get("SETOPS_INSTANCE"))
            raise RuntimeError("sortie par exception")
    except RuntimeError:
        pass
    apres = {k: os.environ.get(k) for k in ("SETOPS_INSTANCE", "SETOPS_UNDERLAY")}
    verifier(dedans == ("/nulle/SITE/underlay.yml", "/nulle/OPS") and apres == avant,
             "`_monter` designe le site et le locataire voulus, puis remet l'environnement, meme sur exception")


def fiches() -> None:
    """La fiche du site dit ce que chaque locataire porte — et `verifier_fiche` voit chaque ecart."""
    import copy
    couples = C.couples_hebergeur()
    if not couples:
        print("SAUTE  aucun site frere : aucune fiche a confronter")
        return
    for site, loc in couples:
        e = C.verifier_fiche(site, loc)
        verifier(not e, f"fiche {site.nom} -> {loc.nom} : identique a ce que le locataire porte"
                 + (f" ({e[0]})" if e else ""))
    site, loc = couples[0]
    juste = site.fiche_pour(loc)
    alterations = {
        "index": lambda f: f["attribue"].__setitem__("index", (f["attribue"]["index"] or 0) + 1),
        "adresse publique": lambda f: f["attribue"].__setitem__("ip_publique", "192.0.2.99"),
        "intrant de service": lambda f: f["offre"].__setitem__("dns_amorcage", "192.0.2.53"),
        "racine de confiance": lambda f: f["racine_confiance"].__setitem__("sha256", "0" * 64),
        "serveurs de temps": lambda f: f["generation"].__setitem__("serveurs_temps", ["192.0.2.123"]),
        "delegation DNS": lambda f: f["generation"].__setitem__("resolveurs_site", ["192.0.2.1"]),
        "mode de routage": lambda f: f["generation"].__setitem__(
            "routage_tenants", "switch" if f["generation"]["routage_tenants"] == "sdn" else "sdn"),
    }
    for nom, alterer in alterations.items():
        f = copy.deepcopy(juste)
        alterer(f)
        verifier(bool(C.verifier_fiche(site, loc, f)),
                 f"une fiche alteree ({nom}) est vue en ecart")


def faces() -> None:
    """La face reseau dit ce que les consommateurs du site en tirent — et chaque ecart se voit."""
    import copy
    couples = C.couples_hebergeur()
    if not couples:
        print("SAUTE  aucun site frere : aucune face reseau a confronter")
        return
    for site, loc in couples:
        e = C.verifier_face(site, loc)
        verifier(not e, f"face reseau {loc.nom} -> {site.nom} : identique a ce que le site en tire"
                 + (f" ({e[0]})" if e else ""))
    site, loc = couples[0]
    juste = loc.face_reseau()
    groupe = next(g for g in ("serveur_postfix", "serveur_nginx", "client_backup") if juste["groupes"].get(g))
    hote = juste["groupes"][groupe][0]

    def deplacer_machine(f):
        f["machines"][hote]["ip"] = "192.0.2.10"

    def retirer_du_groupe(f):
        f["groupes"][groupe] = [h for h in f["groupes"][groupe] if h != hote]

    alterations = {
        "index": lambda f: f.__setitem__("index", (f["index"] or 0) + 1),
        "adresse d'une machine": deplacer_machine,
        f"membres du groupe {groupe}": retirer_du_groupe,
        "administration declaree": lambda f: f["administration"].__setitem__("intrant", ["192.0.2.0/24"]),
        "tunnel d'administration": lambda f: f["administration"].__setitem__("tunnel", ["192.0.2.128/25"]),
        "zones publiques": lambda f: f["dns_public"].__setitem__("zones", ["exemple.invalid"]),
        "primaire DNS": lambda f: f["dns_public"].__setitem__("primaire", "192.0.2.53"),
        "cle de sauvegarde": lambda f: f["sauvegarde"].__setitem__("cle_publique", "ssh-ed25519 AAAA-fausse"),
    }
    for nom, alterer in alterations.items():
        f = copy.deepcopy(juste)
        alterer(f)
        verifier(bool(C.verifier_face(site, loc, f)), f"une face reseau alteree ({nom}) est vue en ecart")


def flux() -> None:
    """Machine par machine, le pare-feu du locataire et celui de Proxmox admettent les memes
    entrees — et une face dont les flux sont alteres se voit."""
    import copy
    couples = C.couples_hebergeur()
    if not couples:
        print("SAUTE  aucun site frere : aucun flux a confronter")
        return
    for site, loc in couples:
        e = C.verifier_flux(site, loc)
        verifier(not e, f"flux {loc.nom} : le locataire et Proxmox admettent les memes entrees"
                 + (f" ({e[0]})" if e else ""))
    site, loc = couples[0]
    juste = loc.face_reseau()
    mta = next(h for h, f in juste["flux"].items()
               if any(x.get("sources_declarees") for x in f["entrees"]))
    autre = next(h for h, f in juste["flux"].items()
                 if any(len(x.get("sources") or []) > 1 for x in f["entrees"]))

    def retirer_une_source(f):
        e = next(x for x in f["flux"][autre]["entrees"] if len(x.get("sources") or []) > 1)
        e["sources"] = e["sources"][1:]

    def ajouter_un_port(f):
        f["flux"][autre]["entrees"].append({"proto": "tcp", "port": "6666", "role": "x",
                                            "sources": ["192.0.2.66"]})

    def effacer_declarees(f):
        for x in f["flux"][mta]["entrees"]:
            x.pop("sources_declarees", None)

    def retirer_admin(f):
        f["flux"][autre]["administration"] = []

    for nom, alterer in (("une source retiree", retirer_une_source),
                         ("un port ajoute", ajouter_un_port),
                         (f"sources declarees effacees sur {mta}", effacer_declarees),
                         ("administration retiree", retirer_admin)):
        f = copy.deepcopy(juste)
        alterer(f)
        verifier(bool(C.verifier_flux(site, loc, f)), f"des flux alteres ({nom}) sont vus en ecart")


def frontiere() -> None:
    """La frontiere designe chaque locataire par les memes identites que sa face reseau et la
    fiche du site — et chaque identite alteree se voit."""
    import copy
    couples = C.couples_hebergeur()
    if not couples:
        print("SAUTE  aucun site frere : aucune frontiere a confronter")
        return
    for site, loc in couples:
        e = C.verifier_frontiere(site, loc)
        verifier(not e, f"frontiere {loc.nom} : alias, routes et traduction sortante concordent"
                 + (f" ({e[0]})" if e else ""))
    site, loc = couples[0]
    face, fiche = loc.face_reseau(), site.fiche_pour(loc)
    groupe = next(g for g in ("serveur_postfix", "serveur_dovecot", "serveur_nginx") if face["groupes"].get(g))
    alterations = {
        "zones": lambda f, fi: f.__setitem__("zones", f["zones"][:-1]),
        "administration declaree": lambda f, fi: f["administration"].__setitem__("intrant", ["192.0.2.0/24"]),
        "tunnel": lambda f, fi: f["administration"].__setitem__("tunnel", ["192.0.2.128/25"]),
        f"membres du groupe {groupe}": lambda f, fi: f["groupes"].__setitem__(groupe, []),
        "adresse publique attribuee": lambda f, fi: fi["attribue"].__setitem__("ip_publique", "192.0.2.60"),
    }
    for nom, alterer in alterations.items():
        f, fi = copy.deepcopy(face), copy.deepcopy(fiche)
        alterer(f, fi)
        verifier(bool(C.verifier_frontiere(site, loc, f, fi)), f"une identite alteree ({nom}) est vue a la frontiere")


def entrees_publiques() -> None:
    """Ce qui entre chez chaque locataire depuis l'Internet : la frontiere dit ce que la face
    ouvre a tous — et chaque alteration se voit."""
    import copy
    couples = C.couples_hebergeur()
    if not couples:
        print("SAUTE  aucun site frere : aucune entree publique a confronter")
        return
    for site, loc in couples:
        e = C.verifier_entrees_publiques(site, loc)
        verifier(not e, f"entrees publiques {loc.nom} : regles WAN, redirections et tunnel concordent"
                 + (f" ({e[0]})" if e else ""))
    site, loc = couples[0]
    face, fiche = loc.face_reseau(), site.fiche_pour(loc)
    hote, entree = next((h, e) for h, f in face["flux"].items() for e in f["entrees"]
                        if "0.0.0.0/0" in (e.get("sources") or []) and e.get("proto") == "tcp")
    autre = next(h for h in face["flux"] if h != hote)

    def fermer(f, fi):
        for e in f["flux"][hote]["entrees"]:
            if e.get("port") == entree["port"] and e.get("proto") == "tcp":
                e["sources"] = ["192.0.2.1"]
                e.pop("sources_declarees", None)

    alterations = {
        f"port {entree['port']} de {hote} ferme": fermer,
        "port public ajoute": lambda f, fi: f["flux"][autre]["entrees"].append(
            {"proto": "tcp", "port": "8443", "role": "x", "sources": ["0.0.0.0/0"]}),
        "adresse publique attribuee": lambda f, fi: fi["attribue"].__setitem__("ip_publique", "192.0.2.60"),
        "port du tunnel": lambda f, fi: f["administration"].__setitem__("port_tunnel", 51999),
    }
    for nom, alterer in alterations.items():
        f, fi = copy.deepcopy(face), copy.deepcopy(fiche)
        alterer(f, fi)
        verifier(bool(C.verifier_entrees_publiques(site, loc, f, fi)),
                 f"une entree publique alteree ({nom}) est vue a la frontiere")


def administration() -> None:
    """Ce que l'administration atteint par la gestion, le VPN et le tunnel : la frontiere dit ce
    que la face ouvre — et chaque alteration se voit."""
    import copy
    couples = C.couples_hebergeur()
    if not couples:
        print("SAUTE  aucun site frere : aucune administration a confronter")
        return
    for site, loc in couples:
        e = C.verifier_administration(site, loc)
        verifier(not e, f"administration {loc.nom} : gestion, VPN et tunnel concordent"
                 + (f" ({e[0]})" if e else ""))
    site, loc = couples[0]
    face = loc.face_reseau()
    ferme = [(h, e) for h, f in face["flux"].items() for e in f["entrees"] if e.get("poste") is False]
    verifier(bool(ferme), f"la face porte les entrees publiques fermees au poste ({[(h, e['port']) for h, e in ferme]})")
    hote_adm = next(h for h, f in face["flux"].items()
                    if any(set(e.get("sources") or []) & set(face["administration"]["intrant"])
                           and "0.0.0.0/0" not in (e.get("sources") or []) for e in f["entrees"]))

    def ouvrir_au_poste(f):
        h, e0 = ferme[0]
        for e in f["flux"][h]["entrees"]:
            if e.get("port") == e0.get("port"):
                e["poste"] = True

    def retirer_sources_admin(f):
        for e in f["flux"][hote_adm]["entrees"]:
            e["sources"] = [s for s in e.get("sources") or [] if s not in face["administration"]["intrant"]]

    alterations = {
        "une entree fermee au poste, ouverte": ouvrir_au_poste,
        f"sources d'administration retirees sur {hote_adm}": retirer_sources_admin,
        "SSH d'administration retire": lambda f: f["flux"][hote_adm].__setitem__("administration", []),
        "tunnel retire": lambda f: f["administration"].__setitem__("tunnel", []),
    }
    for nom, alterer in alterations.items():
        f = copy.deepcopy(face)
        alterer(f)
        verifier(bool(C.verifier_administration(site, loc, f)),
                 f"une administration alteree ({nom}) est vue a la frontiere")


def sorties() -> None:
    """Ce qui sort de chez chaque locataire vers l'Internet : la frontiere dit ce que la face
    publie — et chaque alteration se voit."""
    import copy
    couples = C.couples_hebergeur()
    if not couples:
        print("SAUTE  aucun site frere : aucune sortie a confronter")
        return
    for site, loc in couples:
        e = C.verifier_sorties(site, loc)
        verifier(not e, f"sorties {loc.nom} : la frontiere laisse sortir exactement ce que la face publie"
                 + (f" ({e[0]})" if e else ""))
    site, loc = couples[0]
    face = loc.face_reseau()
    hote = next(h for h, f in face["flux"].items() if f.get("sorties_externes"))
    verifier(sum(len(f.get("sorties_externes") or []) for f in face["flux"].values()) > 0,
             "la face publie les sorties vers l'Internet de chaque machine")

    def retirer(f):
        f["flux"][hote]["sorties_externes"] = f["flux"][hote]["sorties_externes"][1:]

    def ajouter(f):
        f["flux"][hote]["sorties_externes"].append({"proto": "tcp", "port": "6667", "role": "x"})

    for nom, alterer in (("une sortie retiree", retirer), ("une sortie ajoutee", ajouter)):
        f = copy.deepcopy(face)
        alterer(f)
        verifier(bool(C.verifier_sorties(site, loc, f)), f"une sortie alteree ({nom}) est vue a la frontiere")


def fiche_deposee() -> None:
    """La fiche deposee est a jour, l'inventaire se genere sans le site — et une fiche deposee
    alteree se voit."""
    couples = C.couples_hebergeur()
    if not couples:
        print("SAUTE  aucun site frere : aucune fiche deposee a confronter")
        return
    for site, loc in couples:
        e = C.verifier_fiche_deposee(site, loc)
        verifier(not e, f"{loc.nom} porte la fiche que {site.nom} lui destine" + (f" ({e[0]})" if e else ""))
        e = C.verifier_inventaire_sans_site(loc)
        verifier(not e, f"{loc.nom} : l'inventaire se genere sans le site, a l'octet pres"
                 + (f" ({e[0]})" if e else ""))
    site, loc = couples[0]
    chemin = loc.depot / C.NOM_FICHIER_FICHE_SITE
    original = chemin.read_text(encoding="utf-8")
    try:
        chemin.write_text(original.replace("10.0.4.1", "192.0.2.1"), encoding="utf-8")
        verifier(bool(C.verifier_fiche_deposee(site, loc)), "une fiche deposee alteree est vue perimee")
        verifier(bool(C.verifier_inventaire_sans_site(loc)),
                 "une fiche deposee alteree change l'inventaire genere : la preuve le voit")
    finally:
        chemin.write_text(original, encoding="utf-8")


def face_publiee() -> None:
    """Chaque locataire a publie sa face reseau a jour — et une face publiee alteree se voit."""
    couples = C.couples_hebergeur()
    if not couples:
        print("SAUTE  aucun site frere : aucune face publiee a confronter")
        return
    for _site, loc in sorted({(s.nom, l) for s, l in couples}, key=lambda x: x[1].nom):
        e = C.verifier_face_publiee(loc)
        verifier(not e, f"{loc.nom} a publie sa face reseau a jour" + (f" ({e[0]})" if e else ""))
    loc = couples[0][1]
    chemin = loc.depot / C.NOM_FICHIER_FACE_RESEAU
    original = chemin.read_text(encoding="utf-8")
    try:
        chemin.write_text(original.replace("cle_publique: ssh-ed25519", "cle_publique: ssh-rsa", 1),
                          encoding="utf-8")
        verifier(bool(C.verifier_face_publiee(loc)), "une face publiee alteree est vue perimee")
    finally:
        chemin.write_text(original, encoding="utf-8")


def proxmox_lit_la_face() -> None:
    """Le devis Proxmox lit la face publiee : un verdict bascule dans la face retire sa regle."""
    import devis_proxmox_fw
    import devis_reseau
    import yaml as _yaml
    couples = C.couples_hebergeur()
    if not couples:
        print("SAUTE  aucun site frere : aucun devis Proxmox a confronter")
        return
    site, loc = couples[0]
    face = loc.face_publiee() or {}
    cible = next(((g, k) for g, v in (face.get("conditions") or {}).items() for k, ok in v.items()
                  if ok and k.startswith("ingress|tcp|")), None)
    if cible is None:
        print("SAUTE  aucune condition vraie a basculer")
        return
    g, cle = cible
    port = cle.split("|")[2]

    def regles_du_port() -> int:
        with C._monter(site):
            bloc = next(b for b in devis_proxmox_fw.construire(devis_reseau.decouvrir_du_site())["blocs"]
                        if b["tenant"] == loc.nom)
        return sum(1 for gr in bloc["groupes"] if gr["role"] == g
                   for r in gr["regles"] if str(r.get("dport")) == port)

    chemin = loc.depot / C.NOM_FICHIER_FACE_RESEAU
    original = chemin.read_text(encoding="utf-8")
    avant = regles_du_port()
    try:
        f = _yaml.safe_load(original)
        f["conditions"][g][cle] = False
        chemin.write_text(_yaml.safe_dump(f, allow_unicode=True, sort_keys=False), encoding="utf-8")
        apres = regles_du_port()
    finally:
        chemin.write_text(original, encoding="utf-8")
    verifier(avant > 0 and apres == 0,
             f"le devis Proxmox lit la face publiee : {g} {port}/tcp, {avant} regle(s), 0 une fois le verdict bascule")


def frontiere_lit_la_face() -> None:
    """La frontiere lit la face publiee : un verdict bascule dans la face retire sa regle WAN."""
    import yaml as _yaml
    couples = C.couples_hebergeur()
    if not couples:
        print("SAUTE  aucun site frere : aucune frontiere a confronter")
        return
    site, loc = couples[0]
    face = loc.face_publiee() or {}
    cible = next(((g, k) for g, v in (face.get("conditions") or {}).items() for k, ok in v.items()
                  if ok and k.startswith("ingress|tcp|") and g == "serveur_postfix"), None)
    if cible is None:
        print("SAUTE  aucune soumission Postfix active a basculer")
        return
    g, cle = cible
    port = cle.split("|")[2]

    def regles_wan() -> int:
        import devis_opnsense
        devis_opnsense._FACES.clear()
        _e, devis = C._devis_frontiere(site, loc)
        return sum(1 for r in devis.get("regles") or []
                   if r.get("tenant") == loc.nom and r.get("role") == g and port in (r.get("ports") or [])
                   and r.get("interface") == devis.get("if_wan"))

    chemin = loc.depot / C.NOM_FICHIER_FACE_RESEAU
    original = chemin.read_text(encoding="utf-8")
    avant = regles_wan()
    try:
        f = _yaml.safe_load(original)
        f["conditions"][g][cle] = False
        chemin.write_text(_yaml.safe_dump(f, allow_unicode=True, sort_keys=False), encoding="utf-8")
        apres = regles_wan()
    finally:
        chemin.write_text(original, encoding="utf-8")
        import devis_opnsense
        devis_opnsense._FACES.clear()
    verifier(avant > 0 and apres == 0,
             f"la frontiere lit la face publiee : {g} {port}/tcp sur le WAN, {avant} regle(s), 0 une fois le verdict bascule")


def decouverte_lit_la_face() -> None:
    """La decouverte du site lit l'index que le locataire publie, pas sa nomenclature."""
    import devis_reseau
    import yaml as _yaml
    couples = C.couples_hebergeur()
    if not couples:
        print("SAUTE  aucun site frere : aucune decouverte a confronter")
        return
    site, loc = couples[0]
    chemin = loc.depot / C.NOM_FICHIER_FACE_RESEAU
    original = chemin.read_text(encoding="utf-8")

    def index_decouvert():
        with C._monter(site):
            return next((n["index"] for nom, _p, n in devis_reseau.decouvrir_du_site() if nom == loc.nom), None)

    avant = index_decouvert()
    try:
        f = _yaml.safe_load(original)
        f["nomenclature"]["index"] = 99
        chemin.write_text(_yaml.safe_dump(f, allow_unicode=True, sort_keys=False), encoding="utf-8")
        apres = index_decouvert()
    finally:
        chemin.write_text(original, encoding="utf-8")
    verifier(avant == loc.index and apres == 99,
             f"la decouverte du site lit l'index publie : {avant}, puis 99 une fois la face alteree")


def pools_complets() -> None:
    """Chaque pool porte toutes les VM que le locataire publie (regression du 2026-10-05 : la
    face ne publiait pas `fonctions`, et le devis des pools s'est retrouve vide, sans erreur)."""
    import devis_proxmox_pools
    import devis_reseau
    couples = C.couples_hebergeur()
    if not couples:
        print("SAUTE  aucun site frere : aucun pool a confronter")
        return
    site = couples[0][0]
    with C._monter(site):
        pools = {b["tenant"]: b for b in devis_proxmox_pools.construire(devis_reseau.decouvrir_du_site())["pools"]}
    for _s, loc in couples:
        b = pools.get(loc.nom) or {}
        attendu = len((loc.face_publiee() or {}).get("machines") or {})
        verifier(len(b.get("membres") or []) == attendu and not b.get("sans_vmid"),
                 f"pool {loc.nom} : {len(b.get('membres') or [])} VM sur {attendu} publiees, aucune sans VMID")


def pools_et_tunnels_lisent_la_face() -> None:
    """Les pools et les tunnels d'administration lisent la face publiee : l'alterer les change."""
    import devis_proxmox_pools
    import devis_reseau
    import vpn_admin
    import yaml as _yaml
    couples = C.couples_hebergeur()
    if not couples:
        print("SAUTE  aucun site frere : ni pool ni tunnel a confronter")
        return
    site = couples[0][0]
    loc = next((l for _s, l in couples if (l.face_publiee() or {}).get("administration", {}).get("pairs")),
               couples[0][1])
    chemin = loc.depot / C.NOM_FICHIER_FACE_RESEAU
    original = chemin.read_text(encoding="utf-8")

    def releve():
        with C._monter(site):
            pool = next(b for b in devis_proxmox_pools.construire(devis_reseau.decouvrir_du_site())["pools"]
                        if b["tenant"] == loc.nom)
            tun = next((x for x in vpn_admin.tunnels(True) if x.get("portee") == loc.nom), {})
        return sorted(m["vmid"] for m in pool["membres"]), sorted(tun.get("pairs") or {})

    vmids, pairs = releve()
    try:
        f = _yaml.safe_load(original)
        premier = sorted(f["machines"])[0]
        f["machines"][premier]["materialisation"]["proxmox_vmid"] = 999999999
        f["administration"]["pairs"] = {}
        chemin.write_text(_yaml.safe_dump(f, allow_unicode=True, sort_keys=False), encoding="utf-8")
        vmids2, pairs2 = releve()
    finally:
        chemin.write_text(original, encoding="utf-8")
    verifier(999999999 in vmids2 and 999999999 not in vmids, "le devis des pools lit la face publiee (VMID altere repris)")
    verifier(bool(pairs) and not pairs2,
             f"les tunnels d'administration lisent la face publiee ({len(pairs)} pair(s), puis 0)")


def clonage_publie() -> None:
    """La face publie, machine par machine, les parametres de clonage de `parametres-proxmox`
    — et une valeur alteree se voit."""
    import copy
    couples = C.couples_hebergeur()
    if not couples:
        print("SAUTE  aucun site frere : aucun parametre de clonage a confronter")
        return
    for _s, loc in couples:
        e = C.verifier_clonage(loc)
        verifier(not e, f"{loc.nom} : la face porte les parametres de clonage de chaque machine"
                 + (f" ({e[0]})" if e else ""))
    loc = couples[0][1]
    f = copy.deepcopy(loc.face_publiee())
    h = sorted(f["machines"])[0]
    f["machines"][h]["clonage"]["SETOPS_MEMOIRE"] = "1"
    verifier(bool(C.verifier_clonage(loc, f)), f"un parametre de clonage altere ({h}, memoire) se voit")


def materialisation_par_la_face() -> None:
    """Creer, raser et placer par la face visent les memes intrants que par l'instance montee
    — et une face alteree (VMID, etat, pont, placement) se voit, chacune a part."""
    import copy
    couples = C.couples_hebergeur()
    if not couples:
        print("SAUTE  aucun site frere : aucune materialisation a confronter")
        return
    for _s, loc in couples:
        e = C.verifier_materialisation(loc)
        verifier(not e, f"{loc.nom} : creer, raser et placer par la face = par l'instance"
                 + (f" ({e[0]})" if e else ""))
    loc = couples[0][1]
    h = sorted(loc.face_publiee()["machines"])[0]
    def altere(fn):
        f = copy.deepcopy(loc.face_publiee())
        fn(f)
        return C.verifier_materialisation(loc, f)
    e = altere(lambda f: f["machines"][h]["materialisation"].__setitem__("proxmox_vmid", 999999999))
    verifier(any("raser" in x for x in e), f"un VMID altere ({h}) se voit cote raser")
    e = altere(lambda f: f["machines"][h].__setitem__("etat", "retire"))
    verifier(any("creer" in x for x in e), f"un etat altere ({h}) se voit cote creation")
    e = altere(lambda f: f["machines"][h]["materialisation"].__setitem__("proxmox_pont", "zzpont"))
    verifier(any("ponts" in x for x in e), f"un pont altere ({h}) se voit cote placement")
    e = altere(lambda f: f.__setitem__("placement", {**(f.get("placement") or {}), "proxmox_noeud": "zz"}))
    verifier(any("placement" in x for x in e), "une valeur de placement alteree se voit")


def main() -> int:
    with tempfile.TemporaryDirectory() as tmp:
        d = Path(tmp)
        moteur = fabriquer(d)
        classes(d)
        contexte_actif(d, moteur)
    vrais_depots()
    monter_restaure()
    fiches()
    faces()
    flux()
    frontiere()
    entrees_publiques()
    administration()
    sorties()
    fiche_deposee()
    face_publiee()
    proxmox_lit_la_face()
    frontiere_lit_la_face()
    decouverte_lit_la_face()
    pools_complets()
    pools_et_tunnels_lisent_la_face()
    clonage_publie()
    materialisation_par_la_face()
    if ECHECS:
        print(f"\n{len(ECHECS)} echec(s).")
        return 1
    print("\nLe contexte se nomme, se reconnait, et refuse ce qui ne designe rien.")
    return 0


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