contexte : etape 3, le pare-feu Proxmox lit la face reseau

La face publie le verdict des flux conditionnels ; devis_proxmox_fw y lit
inventaire, verdicts et administration. Devis identique, octet pour octet.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

CHANGELOG.md

@@ -1,5 +1,35 @@
 # CHANGELOG — Set-OPS
 
+## 2026-10-05 (92) — Étape 3 : le pare-feu Proxmox lit la face réseau
+
+**Le chemin** (`docs/conception-contextes.md` §6), étape 3. Troisième consommateur du site :
+le devis du pare-feu Proxmox. Il lisait chez chaque locataire son inventaire (adresses,
+groupes, VMID), ses variables (pour les conditions `seulement_si`) et ses réseaux
+d'administration (intrant, `plan/acces.yml`).
+
+**Fait** :
+- la face réseau publie le **verdict de chaque flux conditionnel**, groupe par groupe
+  (`conditions`), évalué chez le locataire par la même `flux_applicable` : le site n'a plus à
+  lire ses variables. Chez Technolibre : la soumission de Postfix (465, 587) et le DNS public
+  (5300) actifs ; les ports locaux de la vigie (8080) et de la console (8090) non ;
+- `contexte.inventaire_depuis_face()` reconstruit l'inventaire minimal que lisent les
+  fonctions de flux (appartenance, adresse, VMID — rien d'autre n'y est lu) ;
+- `devis_proxmox_fw.py` lit la face publiée : inventaire reconstruit, verdicts, administration.
+  Sans face, l'ancienne lecture, et il le dit ;
+- les faces des deux locataires sont republiées (le champ nouveau).
+
+**Le critère** : le devis Proxmox complet (63 Ko, les deux locataires) est **identique
+octet pour octet**, depuis la face ou par l'ancienne lecture.
+
+**Éprouvé** :
+- `test_contexte.py`, 101 contrôles : un verdict basculé dans la face publiée retire sa règle
+  du devis (Postfix 587 : 3 règles, puis 0) — le devis lit bien la face, il ne la contourne pas.
+- `make verifier` conforme, 92/92 (P92 garde la face publiée à jour, verdicts compris).
+
+**Restent à basculer** : la frontière, le SDN, les pools, le placement et le clonage.
+
+**À commiter** : `face-reseau.yml` republiée chez les deux locataires.
+
 ## 2026-10-05 (91) — Étape 3 : le locataire publie sa face réseau, et le site la lit
 
 **Le chemin** (`docs/conception-contextes.md` §6), étape 3, sens locataire → site. Par

docs/conception-contextes.md

@@ -182,7 +182,7 @@ les publie désormais (`sources_declarees`). La frontière, en trois temps : les
 (P87), les entrées publiques (P88), l'administration (P89) et les sorties (P90) sont faites.
 **L'étape 2 est terminée** (2026-10-05). Étape 3 : l'instancier lit la fiche déposée par le site, et
 l'inventaire d'un locataire se génère sans son site, à l'octet près (P91). Le locataire publie sa face
-réseau (P92) ; les comptes de sauvegarde et le DNS public du site la lisent.
+réseau (P92) ; les comptes de sauvegarde, le DNS public et le pare-feu Proxmox du site la lisent.
 
 Méthodes du contrat : `site.fiche_pour(locataire)`, `locataire.face_reseau()`. Une classe
 n'ouvre jamais les fichiers de l'autre ; une preuve vérifiera la règle.

scripts/contexte.py

@@ -65,6 +65,27 @@ def _yaml(chemin: Path) -> dict:
     return yaml.safe_load(chemin.read_text(encoding="utf-8")) or {}
 
 
+def cle_condition(fl: dict) -> str:
+    """L'identite stable d'un flux conditionnel : sens, protocole, port(s), variable lue."""
+    ports = fl.get("port") if isinstance(fl.get("port"), list) else [fl.get("port")]
+    return "|".join([str(fl.get("sens")), str(fl.get("protocole", "tcp")),
+                     ",".join(str(p) for p in ports),
+                     str((fl.get("seulement_si") or {}).get("variable"))])
+
+
+def inventaire_depuis_face(face: dict) -> dict:
+    """L'inventaire minimal que les fonctions de flux lisent, reconstruit depuis une face reseau :
+    l'appartenance aux groupes, l'adresse (`ansible_host`) et le VMID. Rien d'autre n'y est lu."""
+    enfants: dict = {g: {"hosts": {h: None for h in membres}}
+                     for g, membres in (face.get("groupes") or {}).items()}
+    enfants["hotes_actifs"] = {"hosts": {
+        h: {"ansible_host": m.get("ip"),
+            **({"proxmox_vmid": m["materialisation"]["proxmox_vmid"]}
+               if (m.get("materialisation") or {}).get("proxmox_vmid") else {})}
+        for h, m in (face.get("machines") or {}).items()}}
+    return {"all": {"children": enfants}}
+
+
 def _empreinte(chemin: Path) -> str:
     return hashlib.sha256(chemin.read_bytes()).hexdigest() if chemin.is_file() else ""
 
@@ -464,6 +485,17 @@ class Locataire(Ecosysteme):
                        "port": 5300 if colocalise else 53} if zones else {})
         cle_sauvegarde = str(_yaml(self.dossier_inventaire() / "group_vars" / "serveur_backup.yml")
                              .get("serveur_backup_pubkey") or "").strip()
+        # LE VERDICT DES FLUX CONDITIONNELS (`seulement_si`), groupe par groupe. La condition lit
+        # des variables du locataire (group_vars, variables d'hote, defauts des roles) : le site
+        # ne doit plus les ouvrir. Le locataire l'evalue ici, par la meme `flux_applicable`.
+        from resoudre_flux import charger_flux as _catalogue, flux_applicable
+        conditions: dict = {}
+        _cat = _catalogue()
+        for g in sorted(groupes):
+            for fl in _cat.get(g, []):
+                if fl.get("seulement_si"):
+                    conditions.setdefault(g, {})[cle_condition(fl)] = bool(
+                        flux_applicable(fl, g, self.depot))
         # LES FLUX DE CHAQUE MACHINE, tels que le locataire les a deja resolus pour ses propres
         # pare-feux (`make flux` -> `flux-genere/<hote>.connectivite.json`) : chaque entree
         # admise (protocole, port, sources), plus ses reseaux d'administration.
@@ -518,6 +550,7 @@ class Locataire(Ecosysteme):
             "dns_public": dns_public,
             "sauvegarde": {"cle_publique": cle_sauvegarde},
             "flux": flux,
+            "conditions": conditions,
         }
 
     def decrire(self) -> dict:

scripts/devis_proxmox_fw.py

@@ -144,17 +144,57 @@ def _cibles(fl: dict) -> list[dict]:
     return [{"proto": fl.get("protocole", "tcp"), "dport": port} for port in _ports(fl)]
 
 
+def _face_publiee(nom: str) -> dict | None:
+    """La face reseau que ce locataire a publiee, ou None (et on le dit)."""
+    try:
+        import contexte
+        from devis_reseau import DOSSIER_INSTANCES
+        face = contexte.Locataire(DOSSIER_INSTANCES / nom).face_publiee()
+    except Exception:
+        face = None
+    if face is None:
+        print(f"note : {nom} n'a pas publie sa face reseau — lecture de son inventaire et de "
+              f"ses variables (transition).", file=sys.stderr)
+    return face
+
+
 def construire(tenants: list[tuple[str, str, dict]]) -> dict:
     """Le devis, en structure. Un bloc par tenant federe."""
     flux_par_role = charger_flux()
     blocs: list[dict] = []
 
     for nom, pfx, n in tenants:
+        # LA FACE RESEAU D'ABORD (2026-10-05, docs/conception-contextes.md §6, etape 3). Le
+        # locataire publie ses machines, ses groupes, son administration et le verdict de ses
+        # flux conditionnels (`face-reseau.yml`) ; le devis ne lit plus son inventaire ni ses
+        # variables. Sans face publiee (transition), l'ancienne lecture, et on le dit.
+        face = _face_publiee(nom)
         inv = inventaire_de(nom)
-        if inv is None:
+        if face is None and inv is None:
             continue
         etiquette = f"{pfx}{n['index']}"
-        data = yaml.safe_load(inv.read_text(encoding="utf-8")) or {}
+        if face is not None:
+            import contexte
+            data = contexte.inventaire_depuis_face(face)
+            _verdicts = face.get("conditions") or {}
+            _admin = sorted(set((face.get("administration") or {}).get("intrant") or [])
+                            | set((face.get("administration") or {}).get("tunnel") or []))
+
+            def applicable(fl, groupe, _v=_verdicts):
+                if not fl.get("seulement_si"):
+                    return True
+                return bool((_v.get(groupe) or {}).get(contexte.cle_condition(fl), True))
+
+            def admin_du_tenant(_nom, _a=_admin):
+                return _a
+        else:
+            data = yaml.safe_load(inv.read_text(encoding="utf-8")) or {}
+            _dossier = inv.parent.parent.parent
+
+            def applicable(fl, groupe, _d=_dossier):
+                return flux_applicable(fl, groupe, _d)
+
+            admin_du_tenant = admin_avec_tunnel
         # Les PLANIFIES comptent : le pare-feu se prepare avant que la VM existe,
         # sinon le devis changerait a chaque materialisation d'hote.
         tous = set(_ip_par_hote(data))
@@ -179,10 +219,10 @@ def construire(tenants: list[tuple[str, str, dict]]) -> dict:
             "role": "Internet (tout sauf RFC 1918) — flux `externe`",
             "membres": list(INTERNET),
         }
-        if admin_avec_tunnel(nom):
+        if admin_du_tenant(nom):
             ipsets[f"t{n['index']}-admin"] = {
                 "role": "reseaux d'administration (intrant nftables_admin_ssh)",
-                "membres": sorted(admin_avec_tunnel(nom)),
+                "membres": sorted(admin_du_tenant(nom)),
             }
         for groupe in sorted(_enfants(data)):
             if not est_groupe_operationnel(groupe):
@@ -205,7 +245,7 @@ def construire(tenants: list[tuple[str, str, dict]]) -> dict:
             for fl in flux_par_role.get(groupe, []):
                 if fl.get("sens") != "ingress":
                     continue
-                if not flux_applicable(fl, groupe, inv.parent.parent.parent):
+                if not applicable(fl, groupe):
                     continue
                 pairs = _pairs(fl)
                 # `externe` est saute PAIR PAR PAIR, jamais le flux entier : le SSH du
@@ -221,7 +261,7 @@ def construire(tenants: list[tuple[str, str, dict]]) -> dict:
                 if "externe" in pairs and not (
                         str(fl.get("protocole", "tcp")).lower() != "icmp" and "22" in _ports(fl)):
                     sources = [(f"t{n['index']}-internet", "internet")]
-                    if fl.get("poste", True) and admin_avec_tunnel(nom):
+                    if fl.get("poste", True) and admin_du_tenant(nom):
                         sources.append((f"t{n['index']}-admin", "administration (poste)"))
                     for cle_src, via in sources:
                         utilises.add(cle_src)
@@ -255,7 +295,7 @@ def construire(tenants: list[tuple[str, str, dict]]) -> dict:
                     # rendrait une liste vide et la regle serait sautee. Son IPSet
                     # existe deja (garde anti-lockout SSH) — on s'y refere directement.
                     if pair == "admin":
-                        if not admin_avec_tunnel(nom):
+                        if not admin_du_tenant(nom):
                             continue
                         utilises.add(cle)
                         for cible in _cibles(fl):
@@ -290,7 +330,7 @@ def construire(tenants: list[tuple[str, str, dict]]) -> dict:
                 # `_ports` seulement pour TCP/UDP : appele sur un flux ICMP, il rangeait le
                 # type parmi les « ports derives sautes » — un flux pourtant rendu.
                 if (str(fl.get("protocole", "tcp")).lower() != "icmp"
-                        and "22" in _ports(fl) and admin_avec_tunnel(nom)):
+                        and "22" in _ports(fl) and admin_du_tenant(nom)):
                     utilises.add(f"t{n['index']}-admin")
                     regles.append({
                         "action": "ACCEPT", "sens": "IN",
@@ -351,7 +391,7 @@ def construire(tenants: list[tuple[str, str, dict]]) -> dict:
             "ipsets": {k: v for k, v in ipsets.items() if k in utilises},
             "groupes": groupes,
             "affectations": affect, "sans_source": sorted(set(sans_source)),
-            "admin": admin_avec_tunnel(nom),
+            "admin": admin_du_tenant(nom),
         })
 
     return {"blocs": blocs}

scripts/tests/test_contexte.py

@@ -470,6 +470,46 @@ def face_publiee() -> None:
         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 main() -> int:
     with tempfile.TemporaryDirectory() as tmp:
         d = Path(tmp)
@@ -487,6 +527,7 @@ def main() -> int:
     sorties()
     fiche_deposee()
     face_publiee()
+    proxmox_lit_la_face()
     if ECHECS:
         print(f"\n{len(ECHECS)} echec(s).")
         return 1