administration : la frontiere garde l intrant, l est-ouest y ajoute le tunnel

Ajouter le tunnel a admin_de l avait fait classer WAN par le devis de la
frontiere : une regle SSH sur le WAN qui ne correspondrait jamais.
admin_avec_tunnel pour Proxmox et l epreuve ; admin_de redevient l intrant.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Daniel Allaire 2026-09-28 16:54:32 -04:00
parent 22fffdbb6f
commit a401481ae9
3 changed files with 23 additions and 14 deletions

View file

@ -42,7 +42,7 @@ from resoudre_flux import ( # noqa: E402
_pairs,
_resoudre_sources,
)
from devis_reseau import admin_de, decouvrir_du_site, inventaire_de, prefixe # noqa: E402
from devis_reseau import admin_avec_tunnel, decouvrir_du_site, inventaire_de, prefixe # noqa: E402
LONGUEUR_MAX_GROUPE = 18 # limite de Proxmox pour un nom de groupe de securite
@ -158,10 +158,10 @@ def construire(tenants: list[tuple[str, str, dict]]) -> dict:
# Le reseau d'administration a son PROPRE IPSet. L'ajouter a `flotte` ouvrirait
# aussi tous les autres flux qui s'y referent — LDAP, SQL, metriques — a un
# reseau qui n'a rien a y faire. Un ensemble, un sens.
if admin_de(nom):
if admin_avec_tunnel(nom):
ipsets[f"t{n['index']}-admin"] = {
"role": "reseaux d'administration (intrant nftables_admin_ssh)",
"membres": sorted(admin_de(nom)),
"membres": sorted(admin_avec_tunnel(nom)),
}
for groupe in sorted(_enfants(data)):
if not est_groupe_operationnel(groupe):
@ -213,7 +213,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_de(nom):
if not admin_avec_tunnel(nom):
continue
utilises.add(cle)
for cible in _cibles(fl):
@ -248,7 +248,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_de(nom)):
and "22" in _ports(fl) and admin_avec_tunnel(nom)):
utilises.add(f"t{n['index']}-admin")
regles.append({
"action": "ACCEPT", "sens": "IN",
@ -309,7 +309,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_de(nom),
"admin": admin_avec_tunnel(nom),
})
return {"blocs": blocs}

View file

@ -477,15 +477,24 @@ def admin_de(nom_instance: str) -> list[str]:
data = yaml.safe_load(fichier.read_text(encoding="utf-8")) or {}
if isinstance(data, dict) and data.get("nftables_admin_ssh"):
src = data["nftables_admin_ssh"]
recus = [str(s) for s in src] if isinstance(src, list) else [str(src)]
return sorted(set(recus) | set(_tunnel_de(nom_instance)))
return sorted(set(_tunnel_de(nom_instance)))
return [str(s) for s in src] if isinstance(src, list) else [str(src)]
return []
def _tunnel_de(nom_instance: str) -> list[str]:
"""Le tunnel du locataire : MEME derivation que nftables (`tunnel_admin_de`)."""
def admin_avec_tunnel(nom_instance: str) -> list[str]:
"""Les sources d'administration EST-OUEST : l'intrant PLUS le tunnel du locataire.
C'est ce que nftables admet dans chaque VM (`resoudre_flux._sources_admin_ssh`), et ce
que le pare-feu de Proxmox, devant elles, doit admettre aussi (2026-09-28).
PAS `admin_de`, ET C'EST VOULU. La frontiere lit `admin_de` et range chaque reseau par
INTERFACE (gestion, WAN, VPN du site). Le tunnel d'un locataire n'est aucune des
trois : l'y ajouter l'a fait classer WAN — une regle « SSH depuis 10.17.29.0/24 sur le
WAN » qui ne correspondrait jamais, ce trafic arrivant par l'interface WireGuard du
locataire. Deux usages, deux fonctions ; une seule derivation du tunnel.
"""
from inventory_rules import tunnel_admin_de
return tunnel_admin_de(DOSSIER_INSTANCES / nom_instance)
return sorted(set(admin_de(nom_instance)) | set(tunnel_admin_de(DOSSIER_INSTANCES / nom_instance)))
def admin_tous_tenants() -> list[str]:

View file

@ -41,7 +41,7 @@ from pathlib import Path
RACINE = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(RACINE / "scripts"))
from devis_reseau import DOSSIER_INSTANCES, admin_de # noqa: E402
from devis_reseau import DOSSIER_INSTANCES, admin_avec_tunnel # noqa: E402
def _inventaire(instance: str) -> Path:
@ -156,7 +156,7 @@ def observer(instance: str, hote: str, tests: list[dict]) -> list[str]:
permis: dict[str, set[str]] = {}
for t in tests:
permis.setdefault(t["port"], set()).add(t["src"])
admin = [ipaddress.ip_network(n) for n in admin_de(instance)]
admin = [ipaddress.ip_network(n) for n in admin_avec_tunnel(instance)]
hors = []
for ligne in sortie.splitlines():
champs = ligne.split()