Alliance-Boreale/Set-OPS-Public
2
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity
Set-OPS-Public/scripts/tests/test_contexte.py
Daniel Allaire 0d5ec924cc materialisation : le site pose lui-meme la cle de son runner en clonant 
La cle du runner du site transitait par la face du locataire : lue dans le plan du
site, publiee dans la face, relue par le site. Le runner d'un locataire, sans ce
plan, recalculait une face sans elle. La face ne la porte plus ; le site l'ajoute en
clonant la machine qui porte serveur_ops_tenant, et elle seule, en tete des cles
(clonage_rendu). Le clonage par la face reste identique a l'octet pres au chemin du
poste (P93 le compare cle comprise). La console juge de nouveau la face partout.

94 preuves.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-11 18:43:58 -04:00

759 lines
35 KiB
Python
Raw Blame History

#!/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 cle_du_site_posee_par_le_site() -> None:
"""LA CLE DU RUNNER DU SITE NE TRANSITE PLUS PAR LA FACE (2026-10-11).
Le site l'ajoute en clonant, sur le runner du locataire et sur lui seul, en tete — et
une face d'avant qui la porte n'est pas doublee. Puis, sur les vrais depots : un
locataire sans le plan de son site calcule exactement la face publiee. C'est le defaut
mesure sur le runner de Technolibre : il la recalculait sans la cle, et la disait perimee.
"""
runner = {"groupes": ["serveur_ops", "serveur_ops_tenant"],
"clonage": {"SETOPS_VMID": "1", "SETOPS_CLES_AMORCAGE": "ssh-ed25519 LOC loc"}}
autre = {"groupes": ["serveur_debian"],
"clonage": {"SETOPS_VMID": "2", "SETOPS_CLES_AMORCAGE": "ssh-ed25519 LOC loc"}}
cle = "ssh-ed25519 SITE site"
verifier(C.clonage_rendu(runner, cle)["SETOPS_CLES_AMORCAGE"] == cle + "\nssh-ed25519 LOC loc",
"le site pose sa cle en tete sur le runner du locataire")
verifier(C.clonage_rendu(autre, cle) == autre["clonage"], "une machine ordinaire ne la recoit pas")
deja = {"groupes": runner["groupes"],
"clonage": {"SETOPS_CLES_AMORCAGE": cle + "\nssh-ed25519 LOC loc"}}
verifier(C.clonage_rendu(deja, cle) == deja["clonage"], "une face d'avant n'est pas doublee")
verifier(C.clonage_rendu(runner, "") == runner["clonage"], "un site sans runner declare n'ajoute rien")
couples = C.couples_hebergeur()
if not couples:
print("SAUTE aucun site frere : face sans le site non confrontee")
return
import underlay as _u
avant = os.environ.get("SETOPS_UNDERLAY")
os.environ["SETOPS_UNDERLAY"] = "/inexistant/underlay.yml"
try:
verifier(not _u.lire_plan_site("10-intrants.yml"), "aucun plan de site n'est lu (temoin)")
for _s, loc in couples:
verifier(loc.face_reseau() == loc.face_publiee(),
f"{loc.nom} : sans le plan du site, la face calculee est la face publiee")
finally:
if avant is None:
os.environ.pop("SETOPS_UNDERLAY", None)
else:
os.environ["SETOPS_UNDERLAY"] = avant
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()
cle_du_site_posee_par_le_site()
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())
Reference in a new issue View git blame Copy permalink