contexte : etape 2, la frontiere second temps, les entrees publiques (P88)

La face reseau porte le port de son tunnel ; verifier_entrees_publiques
confronte regles WAN, redirections et tunnel aux entrees que la face
ouvre a tous. Aucun ecart ; quatre alterations vues.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Daniel Allaire 2026-10-05 00:14:32 -04:00
parent 78c2c25761
commit 3c16ef45cd
8 changed files with 203 additions and 17 deletions

View file

@ -188,7 +188,7 @@ Si `ansible-lint` n’est pas disponible, le signaler clairement. Ne pas invente
## Écrire, puis relire (D-68) ## Écrire, puis relire (D-68)
`--syntax-check` et `ansible-lint` prouvent que le dépôt est cohérent **avec lui-même**. `--syntax-check` et `ansible-lint` prouvent que le dépôt est cohérent **avec lui-même**.
C'est aussi ce que font les 87 preuves de `make prouver` : elles lisent le dépôt, sans le C'est aussi ce que font les 88 preuves de `make prouver` : elles lisent le dépôt, sans le
moindre appel réseau. **Aucune ne demande au système déployé s'il ressemble à ce que le moindre appel réseau. **Aucune ne demande au système déployé s'il ressemble à ce que le
dépôt annonce.** dépôt annonce.**

View file

@ -1,5 +1,38 @@
# CHANGELOG — Set-OPS # CHANGELOG — Set-OPS
## 2026-10-04 (84) — Étape 2 : la frontière, second temps (les entrées publiques)
**Le chemin** (`docs/conception-contextes.md` §6), étape 2, frontière, second des trois temps :
ce qui entre chez un locataire depuis l'Internet.
**Fait** (`scripts/contexte.py`) :
- la face réseau porte le **port de son tunnel d'administration**, dérivé de son index par
`inventory_rules.port_vpn_locataire()` ;
- `verifier_entrees_publiques(site, locataire)` confronte, machine par machine, les entrées
que la face ouvre à tous (`0.0.0.0/0`) avec les **règles du WAN** (alias de destination
développés en adresses, le supernet en toutes ses machines), les **redirections** (depuis
l'adresse que la fiche du site attribue, vers la machine et son port) et la règle du
**tunnel** ;
- la construction du devis de la frontière est partagée (`_devis_frontiere`) entre les deux
temps déjà faits.
**Le port public d'un locataire** est son port local : seul `serveur_dns_public`, rôle du
site, déclare un `port_public`. La face ne porte donc pas de port public ; la vérification le
dirait si une redirection d'un locataire en avait un autre.
**Relevé en chemin, non corrigé ici** : `devis_opnsense.py` recalcule le port du tunnel
(`52000 + index`) au lieu d'appeler `port_vpn_locataire()`. Une copie de plus de la même
formule ; elle disparaîtra quand la frontière lira la face réseau (étape 3).
**Éprouvé** :
- **P88**, « Ce qui entre depuis l'Internet chez un locataire, la frontière le tient de lui » :
aucun écart chez les deux locataires. Comparés chez Technolibre : 19 entrées ouvertes à tous
(l'ICMP « fragmentation nécessaire » sur ses 13 machines, et 6 ports publics), 6
redirections, 8 règles WAN dont celle du tunnel (52023).
- `test_contexte.py`, 79 contrôles : une entrée publique altérée de **quatre façons** (un port
fermé, un port ouvert en trop, l'adresse publique, le port du tunnel) est vue à la frontière.
- `make verifier` conforme, **88/88**. Les trois documents comptent 88 preuves.
## 2026-10-04 (83) — Étape 2 : la frontière, premier temps (les identités) ## 2026-10-04 (83) — Étape 2 : la frontière, premier temps (les identités)
**Le chemin** (`docs/conception-contextes.md` §6), étape 2, dernier consommateur : la **Le chemin** (`docs/conception-contextes.md` §6), étape 2, dernier consommateur : la

View file

@ -179,7 +179,7 @@ croisée, plus aucun recalcul.
croisées produisent. La première mesure des flux a trouvé une information que le locataire croisées produisent. La première mesure des flux a trouvé une information que le locataire
jetait : les clients nommés d'un port aussi public, que Proxmox doit admettre nommément. Il jetait : les clients nommés d'un port aussi public, que Proxmox doit admettre nommément. Il
les publie désormais (`sources_declarees`). La frontière, en trois temps : les identités les publie désormais (`sources_declarees`). La frontière, en trois temps : les identités
(P87) sont faites ; restent les entrées publiques, puis les sorties et l'administration. (P87) et les entrées publiques (P88) sont faites ; restent les sorties et l'administration.
Méthodes du contrat : `site.fiche_pour(locataire)`, `locataire.face_reseau()`. Une classe 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. n'ouvre jamais les fichiers de l'autre ; une preuve vérifiera la règle.

View file

@ -30,7 +30,7 @@ make placement-plan # chaque VM est-elle là où le plan la met
## Le trou qu'il comble ## Le trou qu'il comble
`scripts/prouver.py` porte 87 preuves (dont une conditionnelle, sautée sans la clé de la voûte). Elles sont toutes **statiques** : elles lisent le `scripts/prouver.py` porte 88 preuves (dont une conditionnelle, sautée sans la clé de la voûte). Elles sont toutes **statiques** : elles lisent le
dépôt. Zéro appel réseau, zéro SSH, zéro `ansible`. Elles établissent que le dépôt est dépôt. Zéro appel réseau, zéro SSH, zéro `ansible`. Elles établissent que le dépôt est
cohérent **avec lui-même** — que les handlers existent, que les intrants ont un cohérent **avec lui-même** — que les handlers existent, que les intrants ont un
propriétaire, que rien n'est codé en dur. propriétaire, que rien n'est codé en dur.

View file

@ -99,7 +99,7 @@ Nommer un point ouvert vaut mieux qu'une ligne rassurante sans mécanisme derri
Aucune de ces responsabilités n'est laissée à la parole : Aucune de ces responsabilités n'est laissée à la parole :
``` ```
make prouver le dépôt est-il cohérent avec lui-même (87 preuves, zéro réseau) make prouver le dépôt est-il cohérent avec lui-même (88 preuves, zéro réseau)
make remise-verifier le second temps de la remise est-il fait, ou en retard ? (P80) make remise-verifier le second temps de la remise est-il fait, ou en retard ? (P80)
make certificats-plan ce que le disque porte contre ce que la mémoire sert make certificats-plan ce que le disque porte contre ce que la mémoire sert
make expositions-plan chaque service publié répond-il, et depuis où make expositions-plan chaque service publié répond-il, et depuis où

View file

@ -376,7 +376,8 @@ class Locataire(Ecosysteme):
LES FLUX de chaque machine viennent de ce que le locataire a deja resolu pour ses LES FLUX de chaque machine viennent de ce que le locataire a deja resolu pour ses
propres pare-feux (`flux-genere/*.connectivite.json`). Ceux de la frontiere suivront. propres pare-feux (`flux-genere/*.connectivite.json`). Ceux de la frontiere suivront.
""" """
from inventory_rules import charger_acces, est_groupe_operationnel, tunnel_admin_de from inventory_rules import (charger_acces, est_groupe_operationnel, port_vpn_locataire,
tunnel_admin_de)
from resoudre_flux import _enfants, _hotes_du_groupe, _ip_par_hote from resoudre_flux import _enfants, _hotes_du_groupe, _ip_par_hote
data = _yaml(self.inventaire()) data = _yaml(self.inventaire())
hv = _hostvars(self.inventaire()) hv = _hostvars(self.inventaire())
@ -449,6 +450,8 @@ class Locataire(Ecosysteme):
"machines": machines, "machines": machines,
"groupes": groupes, "groupes": groupes,
"administration": {"intrant": intrant, "tunnel": tunnel_admin_de(self.depot), "administration": {"intrant": intrant, "tunnel": tunnel_admin_de(self.depot),
"port_tunnel": (port_vpn_locataire(self.index)
if tunnel_admin_de(self.depot) else None),
"pairs": {n: {k: c.get(k) for k in ("cle_publique", "adresse", "etat")} "pairs": {n: {k: c.get(k) for k in ("cle_publique", "adresse", "etat")}
for n, c in sorted(pairs.items()) if isinstance(c, dict)}}, for n, c in sorted(pairs.items()) if isinstance(c, dict)}},
"dns_public": dns_public, "dns_public": dns_public,
@ -721,6 +724,99 @@ def verifier_flux(site: "Site", locataire: "Locataire", face: dict | None = None
return ecarts return ecarts
def _devis_frontiere(site: "Site", locataire: "Locataire") -> tuple[str | None, dict]:
"""L'etiquette du locataire a la frontiere (`TECH23`) et le devis, CE site monte.
Le devis imprime ses notes (flux sans `port_public`, sorties vers un absent) : utiles a qui
le lit, du bruit dans une preuve. On les retient ici.
"""
import io
import devis_opnsense
import devis_reseau
with _monter(site):
tenants = devis_reseau.decouvrir_du_site()
etq = next((p + str(n["index"]) for nom, p, n in tenants if nom == locataire.nom), None)
with contextlib.redirect_stderr(io.StringIO()):
devis = devis_opnsense.construire(tenants) if etq else {}
return etq, devis
def verifier_entrees_publiques(site: "Site", locataire: "Locataire", face: dict | None = None,
fiche: dict | None = None) -> list[str]:
"""La frontiere, second temps : ce qui ENTRE chez le locataire depuis l'Internet.
Machine par machine, les regles du WAN (destination developpee en adresses) et les
redirections doivent dire ce que la face reseau porte comme entrees ouvertes a tous
(`0.0.0.0/0`) : les memes (machine, protocole, port). Une redirection part de l'adresse que
la fiche du site attribue, et son port public est le port local — aucun role de locataire
n'en declare d'autre ; s'il le faisait, la face ne le porterait pas, et ce serait un ecart.
Le tunnel d'administration entre par l'adresse WAN de la frontiere, sur son port.
"""
import ipaddress
f = face if face is not None else locataire.face_reseau()
fi = fiche if fiche is not None else site.fiche_pour(locataire)
etq, devis = _devis_frontiere(site, locataire)
if not etq:
return ["la frontiere ne decouvre pas ce locataire"]
alias = devis.get("alias") or {}
ips = sorted({m["ip"] for m in f["machines"].values()})
def adresses(nom: str) -> set[str]:
a = alias.get(nom) or {}
if a.get("type") == "network":
reseaux = [ipaddress.ip_network(str(c), strict=False) for c in a.get("contenu") or []]
return {ip for ip in ips if any(ipaddress.ip_address(ip) in r for r in reseaux)}
return {str(c) for c in a.get("contenu") or []}
# La face : chaque entree ouverte a tous, par machine.
publiques: set[tuple] = set()
for h, fl in (f.get("flux") or {}).items():
ip = f["machines"][h]["ip"]
for e in fl.get("entrees") or []:
if "0.0.0.0/0" in (e.get("sources") or []):
publiques.add((ip, str(e.get("proto")), str(e.get("port"))))
ecarts: list[str] = []
# Les regles du WAN, hors tunnel.
wan = {(ip, str(r.get("protocole")), str(p))
for r in devis.get("regles") or []
if r.get("tenant") == locataire.nom and r.get("interface") == devis.get("if_wan")
and r.get("sens") == "in" and r.get("role") != "acces_admin_vpn"
for ip in adresses(str(r.get("destination"))) for p in r.get("ports") or []}
norm = {(i, pr, _ICMP.get(po, po)) for i, pr, po in publiques}
wan_n = {(i, pr, _ICMP.get(po, po)) for i, pr, po in wan}
for x in sorted(norm - wan_n):
ecarts.append(f"{x[0]} {x[1]}/{x[2]} : ouvert a tous selon la face, aucune regle WAN")
for x in sorted(wan_n - norm):
ecarts.append(f"{x[0]} {x[1]}/{x[2]} : regle WAN, la face ne l'ouvre pas a tous")
# Les redirections : celles de la face qui ont un port (TCP/UDP), depuis l'adresse attribuee.
redir = [d for d in devis.get("redirections") or [] if d.get("tenant") == locataire.nom]
vues = {(str(d.get("cible")), str(d.get("protocole")), str(d.get("port_local"))) for d in redir}
attendues = {x for x in publiques if x[1] in ("tcp", "udp")}
for x in sorted(attendues - vues):
ecarts.append(f"{x[0]} {x[1]}/{x[2]} : ouvert a tous selon la face, aucune redirection")
for x in sorted(vues - attendues):
ecarts.append(f"{x[0]} {x[1]}/{x[2]} : redirige par la frontiere, la face ne l'ouvre pas a tous")
for d in redir:
if str(d.get("destination")) != str(fi["attribue"]["ip_publique"]):
ecarts.append(f"redirection {d.get('protocole')}/{d.get('port_public')} : depuis "
f"{d.get('destination')}, la fiche attribue {fi['attribue']['ip_publique']}")
if str(d.get("port_public")) != str(d.get("port_local")):
ecarts.append(f"redirection {d.get('protocole')}/{d.get('port_public')} -> {d.get('port_local')} : "
f"un port public distinct, que la face ne porte pas")
# Le tunnel d'administration.
tun = [r for r in devis.get("regles") or []
if r.get("tenant") == locataire.nom and r.get("role") == "acces_admin_vpn"
and r.get("interface") == devis.get("if_wan")]
ports_tun = sorted(str(p) for r in tun for p in r.get("ports") or [])
attendu_tun = [str(f["administration"]["port_tunnel"])] if f["administration"].get("port_tunnel") else []
if ports_tun != attendu_tun:
ecarts.append(f"tunnel d'administration : la frontiere ouvre {ports_tun}, la face {attendu_tun}")
return ecarts
def verifier_frontiere(site: "Site", locataire: "Locataire", face: dict | None = None, def verifier_frontiere(site: "Site", locataire: "Locataire", face: dict | None = None,
fiche: dict | None = None) -> list[str]: fiche: dict | None = None) -> list[str]:
"""La frontiere, premier temps : les IDENTITES par lesquelles elle designe ce locataire. """La frontiere, premier temps : les IDENTITES par lesquelles elle designe ce locataire.
@ -731,18 +827,9 @@ def verifier_frontiere(site: "Site", locataire: "Locataire", face: dict | None =
attribuee) portent deja. Les entrees publiques et les sorties sont les temps suivants. attribuee) portent deja. Les entrees publiques et les sorties sont les temps suivants.
""" """
import re import re
import devis_opnsense
import devis_reseau
f = face if face is not None else locataire.face_reseau() f = face if face is not None else locataire.face_reseau()
fi = fiche if fiche is not None else site.fiche_pour(locataire) fi = fiche if fiche is not None else site.fiche_pour(locataire)
with _monter(site): etq, devis = _devis_frontiere(site, locataire)
tenants = devis_reseau.decouvrir_du_site()
etq = next((p + str(n["index"]) for nom, p, n in tenants if nom == locataire.nom), None)
# Le devis imprime ses notes (flux sans `port_public`, sorties vers un absent) : utiles a
# qui le lit, du bruit dans une preuve. On les retient ici.
import io
with contextlib.redirect_stderr(io.StringIO()):
devis = devis_opnsense.construire(tenants) if etq else {}
if not etq: if not etq:
return ["la frontiere ne decouvre pas ce locataire"] return ["la frontiere ne decouvre pas ce locataire"]
ecarts: list[str] = [] ecarts: list[str] = []
@ -886,7 +973,8 @@ def main(argv: list[str] | None = None) -> int:
face = loc.face_reseau() face = loc.face_reseau()
for quoi, e in (("faits", verifier_face(site, loc, face)), for quoi, e in (("faits", verifier_face(site, loc, face)),
("flux ", verifier_flux(site, loc, face)), ("flux ", verifier_flux(site, loc, face)),
("frontiere (identites)", verifier_frontiere(site, loc, face))): ("frontiere (identites)", verifier_frontiere(site, loc, face)),
("frontiere (entrees publiques)", verifier_entrees_publiques(site, loc, face))):
total += len(e) total += len(e)
print(f"{'OK ' if not e else 'ECART'} {loc.nom} -> {site.nom} ({quoi})" print(f"{'OK ' if not e else 'ECART'} {loc.nom} -> {site.nom} ({quoi})"
+ "".join(f"\n - {x}" for x in e)) + "".join(f"\n - {x}" for x in e))

View file

@ -4763,6 +4763,29 @@ def preuve_frontiere_identites() -> tuple[bool, str]:
f"publient (supernet, administration, tunnel, groupes, routes, sortie).") f"publient (supernet, administration, tunnel, groupes, routes, sortie).")
def preuve_frontiere_entrees_publiques() -> tuple[bool, str]:
"""Les regles du WAN, les redirections et le tunnel d'administration d'un locataire disent-ils
ce que sa face reseau ouvre a tous ?
POURQUOI (2026-10-04, `docs/conception-contextes.md` §2.4). Second temps de la frontiere.
Machine par machine, chaque entree ouverte a tous (`0.0.0.0/0`) dans la face doit avoir sa
regle WAN et, si elle a un port, sa redirection depuis l'adresse que la fiche du site
attribue — et rien de plus. Le port public d'un locataire est son port local ; s'il en
differait, la face ne le porterait pas, et la preuve le dirait.
"""
sys.path.insert(0, str(RACINE / "scripts"))
import contexte as _ctx
couples = _ctx.couples_hebergeur()
if not couples:
return True, "Aucun site frere : aucune entree publique a confronter."
ecarts = [f"{l.nom} : {e}" for s, l in couples for e in _ctx.verifier_entrees_publiques(s, l)]
if ecarts:
return False, (f"{len(ecarts)} ecart(s) entre la frontiere et ce que le locataire ouvre a tous : "
+ " | ".join(ecarts[:4]))
return True, (f"{len(couples)} locataire(s) : regles WAN, redirections et tunnel disent exactement "
f"ce que leur face reseau ouvre a tous.")
PREUVES: list[dict] = [ PREUVES: list[dict] = [
{"id": "P01", "titre": "Lint (ansible-lint)", "refs": ["AFF-006"], {"id": "P01", "titre": "Lint (ansible-lint)", "refs": ["AFF-006"],
"cmds": [["ansible-lint", "-q"]]}, "cmds": [["ansible-lint", "-q"]]},
@ -4957,6 +4980,8 @@ PREUVES: list[dict] = [
"refs": [], "func": preuve_flux_concordent}, "refs": [], "func": preuve_flux_concordent},
{"id": "P87", "titre": "La frontiere designe chaque locataire par les identites qu'il publie", {"id": "P87", "titre": "La frontiere designe chaque locataire par les identites qu'il publie",
"refs": [], "func": preuve_frontiere_identites}, "refs": [], "func": preuve_frontiere_identites},
{"id": "P88", "titre": "Ce qui entre depuis l'Internet chez un locataire, la frontiere le tient de lui",
"refs": [], "func": preuve_frontiere_entrees_publiques},
] ]

View file

@ -15,7 +15,8 @@ reseau de chaque locataire dit ce que les consommateurs du site en tirent, et un
(huit facons) est vue en ecart ; machine par machine, le pare-feu du locataire et celui de (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 ; 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 la frontiere designe chaque locataire par les memes identites, et une identite alteree (cinq
facons) se voit. facons) se voit ; ce qui entre depuis l'Internet concorde, et une entree alteree (quatre facons)
se voit.
""" """
from __future__ import annotations from __future__ import annotations
@ -312,6 +313,44 @@ def frontiere() -> None:
verifier(bool(C.verifier_frontiere(site, loc, f, fi)), f"une identite alteree ({nom}) est vue a la frontiere") 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 main() -> int: def main() -> int:
with tempfile.TemporaryDirectory() as tmp: with tempfile.TemporaryDirectory() as tmp:
d = Path(tmp) d = Path(tmp)
@ -324,6 +363,7 @@ def main() -> int:
faces() faces()
flux() flux()
frontiere() frontiere()
entrees_publiques()
if ECHECS: if ECHECS:
print(f"\n{len(ECHECS)} echec(s).") print(f"\n{len(ECHECS)} echec(s).")
return 1 return 1