Daniel Allaire
3c16ef45cd
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>
1038 lines
51 KiB
Python
1038 lines
51 KiB
Python
#!/usr/bin/env python3
|
|
"""Le contexte d'un ecosysteme : un tronc commun, deux classes (SITE et LOCATAIRE).
|
|
|
|
Conception : `docs/conception-contextes.md` (arretee avec l'exploitant le 2026-10-04).
|
|
|
|
POURQUOI CE MODULE. Le 2026-10-04, 33 scripts DEVINAIENT leur contexte, chacun a sa facon :
|
|
le lien `instance/`, le lien `underlay.yml`, les variables `SETOPS_*`, les dossiers freres,
|
|
les `SITE-*`. Les indices ne concordaient pas toujours, et chaque desaccord avait deja produit
|
|
un defaut silencieux (la frontiere d'un site recevant les regles de l'autre, la console d'un
|
|
site affichant zero machine, `make ci` melangeant le modele public et les ecosystemes reels).
|
|
|
|
Ce module est le SEUL endroit ou se lisent ces indices. Les scripts lui demandent « ou
|
|
suis-je ? » et recoivent un objet : un `Site` ou un `Locataire`, qui heritent d'`Ecosysteme`.
|
|
|
|
ETAPES 1 ET 2 DU CHEMIN (§6) : le module existe, il est teste, et RIEN NE L'UTILISE ENCORE.
|
|
La fiche du site (`site.fiche_pour(locataire)`) est calculee A COTE de l'existant, et
|
|
`verifier_fiche` prouve qu'elle dit exactement ce que le locataire porte aujourd'hui. La face
|
|
reseau du locataire suit. Les consommateurs basculeront a l'etape 3, a resultat identique.
|
|
|
|
Usage :
|
|
python3 scripts/contexte.py # le contexte actif, et ce que chaque indice dit
|
|
python3 scripts/contexte.py --json
|
|
python3 scripts/contexte.py --fiche OPS-Technolibre # la fiche que son hebergeur lui destine
|
|
python3 scripts/contexte.py --verifier-fiches # chaque fiche dit-elle ce que le locataire porte ?
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import contextlib
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
import yaml # noqa: E402
|
|
|
|
import underlay as underlay_mod # noqa: E402
|
|
import voutes as voutes_mod # noqa: E402
|
|
from inventory_rules import ORDRE_INVENTAIRE, dossier_inventaire, instance_courante # noqa: E402
|
|
|
|
RACINE = Path(__file__).resolve().parents[1]
|
|
# Les ecosystemes vivent a cote du moteur : des dossiers freres, pas de registre.
|
|
DOSSIER_ECOSYSTEMES = RACINE.parent
|
|
# Le contexte actif, nomme (decision du 2026-10-04) : une seule valeur, `site:<depot>` ou
|
|
# `locataire:<depot>`. Absent, on retombe sur les liens d'avant, le temps de la bascule.
|
|
NOM_FICHIER_CONTEXTE = "contexte"
|
|
NATURES = ("site", "locataire")
|
|
|
|
|
|
class ContexteInvalide(Exception):
|
|
"""Un contexte nomme qui ne designe rien, ou un depot qui ne dit pas ce qu'il est."""
|
|
|
|
|
|
def _yaml(chemin: Path) -> dict:
|
|
if not chemin.is_file():
|
|
return {}
|
|
return yaml.safe_load(chemin.read_text(encoding="utf-8")) or {}
|
|
|
|
|
|
def _empreinte(chemin: Path) -> str:
|
|
return hashlib.sha256(chemin.read_bytes()).hexdigest() if chemin.is_file() else ""
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def _monter(site: "Site | None" = None, locataire: "Locataire | None" = None):
|
|
"""Monte CE site et CE locataire le temps d'interroger les modules d'avant.
|
|
|
|
`underlay`, `site_intrants`, `instancier` lisent « le site monte » et « l'instance
|
|
courante » (variables `SETOPS_*`, sinon les liens). Plutot que de recopier leurs calculs,
|
|
on leur designe l'ecosysteme voulu, puis on remet l'environnement tel quel. C'est une
|
|
passerelle de transition : a l'etape 3, ces modules recevront leurs objets.
|
|
"""
|
|
clefs = ("SETOPS_UNDERLAY", "SETOPS_INSTANCE")
|
|
avant = {k: os.environ.get(k) for k in clefs}
|
|
try:
|
|
if site is not None:
|
|
os.environ["SETOPS_UNDERLAY"] = str(site.underlay)
|
|
if locataire is not None:
|
|
os.environ["SETOPS_INSTANCE"] = str(locataire.depot)
|
|
yield
|
|
finally:
|
|
for k, v in avant.items():
|
|
if v is None:
|
|
os.environ.pop(k, None)
|
|
else:
|
|
os.environ[k] = v
|
|
|
|
|
|
# --- LE TRONC COMMUN ----------------------------------------------------------------------
|
|
|
|
class Ecosysteme:
|
|
"""Ce que tout ecosysteme possede, quel que soit son contexte (§2.1).
|
|
|
|
Les methodes qui dependent du contexte sont SURCHARGEES par `Site` et `Locataire` ; ici
|
|
elles refusent, pour qu'un oubli de surcharge se voie au premier appel.
|
|
"""
|
|
|
|
nature = "ecosysteme"
|
|
|
|
def __init__(self, depot: Path):
|
|
self.depot = Path(depot)
|
|
|
|
# Identite -------------------------------------------------------------------------
|
|
@property
|
|
def nom(self) -> str:
|
|
return self.depot.name
|
|
|
|
@property
|
|
def plan(self) -> Path:
|
|
return self.depot / "plan"
|
|
|
|
def lire_plan(self, fichier: str) -> dict:
|
|
"""Un fichier du plan, ou {} s'il n'existe pas."""
|
|
return _yaml(self.plan / fichier)
|
|
|
|
# Voute : un depot, une cle (2026-08-28) ---------------------------------------------
|
|
@property
|
|
def cle_voute(self) -> Path:
|
|
"""La cle qui ouvre SA voute, nommee d'apres le depot (`voutes.cle_de`)."""
|
|
return voutes_mod.cle_de(self.nom)
|
|
|
|
@property
|
|
def voute(self) -> Path:
|
|
raise NotImplementedError(f"{type(self).__name__} : `voute` non surchargee")
|
|
|
|
def a_sa_voute(self) -> bool:
|
|
return self.voute.is_file()
|
|
|
|
# Filiation ----------------------------------------------------------------------------
|
|
@property
|
|
def filiation(self) -> dict:
|
|
"""`parente.yml` : de quels depots (moteur, hebergeur, modeles) il descend."""
|
|
return (_yaml(self.depot / "parente.yml").get("parente")) or {}
|
|
|
|
# Ce que chaque contexte surcharge ------------------------------------------------------
|
|
@property
|
|
def index(self) -> int | None:
|
|
raise NotImplementedError(f"{type(self).__name__} : `index` non surcharge")
|
|
|
|
def inventaire(self) -> Path:
|
|
raise NotImplementedError(f"{type(self).__name__} : `inventaire` non surcharge")
|
|
|
|
# Commodites ----------------------------------------------------------------------------
|
|
def __repr__(self) -> str:
|
|
return f"{type(self).__name__}({self.nom!r})"
|
|
|
|
def __eq__(self, autre: object) -> bool:
|
|
return (isinstance(autre, Ecosysteme) and type(autre) is type(self)
|
|
and autre.depot.resolve() == self.depot.resolve())
|
|
|
|
def __hash__(self) -> int:
|
|
return hash((type(self).__name__, str(self.depot.resolve())))
|
|
|
|
def decrire(self) -> dict:
|
|
return {"nature": self.nature, "nom": self.nom, "depot": str(self.depot),
|
|
"index": self.index, "voute": str(self.voute),
|
|
"a_sa_voute": self.a_sa_voute(), "cle_voute": str(self.cle_voute)}
|
|
|
|
# Reconnaitre un depot ------------------------------------------------------------------
|
|
@staticmethod
|
|
def depuis(depot: Path) -> "Ecosysteme":
|
|
"""Le `Site` ou le `Locataire` que porte ce depot, d'apres CE QU'IL DECLARE.
|
|
|
|
Un site declare un `underlay.yml` ; un locataire, une nomenclature. Un depot qui
|
|
declare les deux (l'ancien modele `socle`) ne dit pas ce qu'il est : refuse, plutot
|
|
que de choisir pour lui.
|
|
"""
|
|
depot = Path(depot)
|
|
est_site = (depot / "underlay.yml").is_file()
|
|
est_locataire = (depot / "plan" / "nomenclature.yml").is_file()
|
|
if est_site and est_locataire:
|
|
raise ContexteInvalide(
|
|
f"{depot.name} porte un `underlay.yml` ET une nomenclature : un site ou un "
|
|
f"locataire, pas les deux.")
|
|
if est_site:
|
|
return Site(depot)
|
|
if est_locataire:
|
|
return Locataire(depot)
|
|
raise ContexteInvalide(
|
|
f"{depot} n'est ni un site (pas d'`underlay.yml`) ni un locataire (pas de "
|
|
f"`plan/nomenclature.yml`).")
|
|
|
|
|
|
# --- SITE ---------------------------------------------------------------------------------
|
|
|
|
class Site(Ecosysteme):
|
|
"""L'hebergeur : le materiel, la fabric, la frontiere, et ses locataires (§2.2)."""
|
|
|
|
nature = "site"
|
|
|
|
@classmethod
|
|
def charger(cls, nom: str, dossier: Path = DOSSIER_ECOSYSTEMES) -> "Site":
|
|
depot = Path(dossier) / nom
|
|
if not (depot / "underlay.yml").is_file():
|
|
raise ContexteInvalide(f"site « {nom} » introuvable : pas de {depot}/underlay.yml")
|
|
return cls(depot)
|
|
|
|
@property
|
|
def underlay(self) -> Path:
|
|
return self.depot / "underlay.yml"
|
|
|
|
def carte(self) -> dict:
|
|
"""Le contenu de `underlay:`."""
|
|
return _yaml(self.underlay).get("underlay") or {}
|
|
|
|
@property
|
|
def voute(self) -> Path:
|
|
return self.depot / "underlay.vault.yml"
|
|
|
|
@property
|
|
def index(self) -> int | None:
|
|
i = self.carte().get("index")
|
|
return i if isinstance(i, int) and not isinstance(i, bool) else None
|
|
|
|
def inventaire(self) -> Path:
|
|
"""Un site n'a pas de `hosts.yml` : son inventaire est un SCRIPT (`site_inventaire.py`)."""
|
|
return RACINE / "scripts" / "site_inventaire.py"
|
|
|
|
def allocations(self) -> dict[str, int]:
|
|
"""L'index que CE site attribue a chacun de ses locataires (`underlay.tenants`)."""
|
|
return underlay_mod.allocations({"underlay": self.carte()})
|
|
|
|
def noms_locataires(self) -> list[str]:
|
|
"""Les locataires que ce site declare, sous l'une ou l'autre forme de `tenants`."""
|
|
v = self.carte().get("tenants")
|
|
if isinstance(v, dict):
|
|
return sorted(str(n) for n in v)
|
|
if isinstance(v, list):
|
|
return sorted(str(n) for n in v)
|
|
return []
|
|
|
|
def locataires(self) -> list["Locataire"]:
|
|
"""Ses locataires, resolus en objets. Un nom sans depot frere n'est PAS rendu ici :
|
|
`locataires_absents()` le dit — un filtre qui oublie en silence est un defaut."""
|
|
out = []
|
|
for nom in self.noms_locataires():
|
|
depot = self.depot.parent / nom
|
|
if (depot / "plan" / "nomenclature.yml").is_file():
|
|
out.append(Locataire(depot))
|
|
return out
|
|
|
|
def locataires_absents(self) -> list[str]:
|
|
presents = {l.nom for l in self.locataires()}
|
|
return [n for n in self.noms_locataires() if n not in presents]
|
|
|
|
# LE CONTRAT, SENS SITE -> LOCATAIRE (§2.4) ---------------------------------------------
|
|
def fiche_pour(self, locataire: "Locataire") -> dict:
|
|
"""Ce que CE site attribue et offre a CE locataire, en une fiche.
|
|
|
|
Trois canaux le portaient jusqu'ici, sans fiche (relevé du 2026-10-04) :
|
|
- des COPIES a la main : index, adresse publique, intrants de service, racine ;
|
|
- une LECTURE DIRECTE du site par `instancier.py` au moment de generer l'inventaire :
|
|
delegation DNS, serveurs de temps, mode de routage ;
|
|
- le genome (hors fiche : il passe par la forge).
|
|
La fiche les reunit, calcules par LES MEMES fonctions qu'aujourd'hui. Elle porte
|
|
l'empreinte de ses sources : une fiche qui ne correspond plus a son site se voit.
|
|
"""
|
|
import site_intrants as si
|
|
with _monter(self, locataire):
|
|
contrat = si.contrat()
|
|
carte = underlay_mod.charger() or {}
|
|
intrants = underlay_mod.lire_plan_site("10-intrants.yml")
|
|
# Les trois lectures d'`instancier.py`, a l'identique (`_zones_deleguees`, la patte
|
|
# de transit de la frontiere, `routage_tenants`).
|
|
zone = str(intrants.get("domaine_interne") or "").strip()
|
|
resolveurs = [a for a in underlay_mod.adresses_site_portant("serveur_resolveur") if a]
|
|
transit = underlay_mod.reseau_transit(carte) or {}
|
|
reseaux_transit = {str(transit.get("nom"))} if transit.get("nom") else set()
|
|
temps = sorted({str(h["ip"]) for h in underlay_mod.hotes(carte)
|
|
if h.get("role") == "frontiere" and h.get("ip")
|
|
and str(h.get("reseau")) in reseaux_transit
|
|
and str(h.get("etat", "actif")) != "reserve"})
|
|
routage = underlay_mod.routage_tenants(carte)
|
|
ip_publique = si.ip_publique_de(locataire.nom)
|
|
racine = self.depot / "ac-racine-site.crt"
|
|
offre = {k: v for k, v in contrat.items() if k not in ("ip_publique", "forge_amont_ac")}
|
|
return {
|
|
"site": self.nom,
|
|
"locataire": locataire.nom,
|
|
"sources": {f: _empreinte(self.depot / f) for f in
|
|
("underlay.yml", "plan/10-intrants.yml", "opnsense.yml", "ac-racine-site.crt")},
|
|
"attribue": {"index": self.allocations().get(locataire.nom), "ip_publique": ip_publique},
|
|
"offre": offre,
|
|
"racine_confiance": {"fichier": racine.name, "sha256": _empreinte(racine),
|
|
"pem": racine.read_text(encoding="utf-8") if racine.is_file() else ""},
|
|
"generation": {"zone_site": zone, "resolveurs_site": resolveurs,
|
|
"serveurs_temps": temps, "routage_tenants": routage},
|
|
}
|
|
|
|
def decrire(self) -> dict:
|
|
d = super().decrire()
|
|
d.update({"locataires": self.noms_locataires(),
|
|
"locataires_absents": self.locataires_absents()})
|
|
return d
|
|
|
|
|
|
# --- LOCATAIRE ----------------------------------------------------------------------------
|
|
|
|
class Locataire(Ecosysteme):
|
|
"""Une organisation hebergee : son plan de services, son inventaire genere (§2.3).
|
|
|
|
UN LOCATAIRE PEUT FIGURER CHEZ DEUX SITES : son hebergeur actif, que nomme `parente.yml`,
|
|
et un site de reprise qui le declare aussi (SITE-Technolibre declare les deux locataires de
|
|
SITE-Chezlepro). `site()` rend l'hebergeur ACTIF ; les autres le connaissent par leur
|
|
propre `locataires()`.
|
|
"""
|
|
|
|
nature = "locataire"
|
|
|
|
@classmethod
|
|
def charger(cls, nom: str, dossier: Path = DOSSIER_ECOSYSTEMES) -> "Locataire":
|
|
depot = Path(dossier) / nom
|
|
if not (depot / "plan" / "nomenclature.yml").is_file():
|
|
raise ContexteInvalide(
|
|
f"locataire « {nom} » introuvable : pas de {depot}/plan/nomenclature.yml")
|
|
return cls(depot)
|
|
|
|
def nomenclature(self) -> dict:
|
|
return self.lire_plan("nomenclature.yml")
|
|
|
|
@property
|
|
def index(self) -> int | None:
|
|
i = self.nomenclature().get("index")
|
|
return i if isinstance(i, int) and not isinstance(i, bool) else None
|
|
|
|
def dossier_inventaire(self) -> Path:
|
|
return dossier_inventaire(self.depot, ORDRE_INVENTAIRE)
|
|
|
|
def inventaire(self) -> Path:
|
|
"""Son `hosts.yml` GENERE du plan. Le chemin est rendu meme s'il n'existe pas encore
|
|
(instance neuve) : c'est `instancier` qui l'ecrira."""
|
|
return self.dossier_inventaire() / "hosts.yml"
|
|
|
|
@property
|
|
def voute(self) -> Path:
|
|
return self.dossier_inventaire() / "group_vars" / "all" / "vault.yml"
|
|
|
|
def domaine_interne(self) -> str:
|
|
"""Son domaine interne : le premier `domaine_interne` non vide de `group_vars/all`,
|
|
dans l'ordre des fichiers — la lecture meme d'`instancier._domaine_interne`."""
|
|
base = self.dossier_inventaire() / "group_vars" / "all"
|
|
for fichier in sorted(base.glob("*.yml")) if base.is_dir() else []:
|
|
try:
|
|
d = yaml.safe_load(fichier.read_text(encoding="utf-8")) or {}
|
|
except yaml.YAMLError:
|
|
continue
|
|
if isinstance(d, dict) and str(d.get("domaine_interne", "")).strip():
|
|
return str(d["domaine_interne"]).strip()
|
|
return ""
|
|
|
|
def nom_hebergeur(self) -> str | None:
|
|
h = ((self.filiation.get("depots") or {}).get("hebergeur") or {}).get("nom")
|
|
return str(h) if h else None
|
|
|
|
def site(self) -> "Site | None":
|
|
"""Son hebergeur ACTIF. None s'il n'en nomme aucun ; refus s'il en nomme un absent."""
|
|
nom = self.nom_hebergeur()
|
|
if not nom:
|
|
return None
|
|
return Site.charger(nom, self.depot.parent)
|
|
|
|
# LE CONTRAT, SENS LOCATAIRE -> SITE (§2.4) ---------------------------------------------
|
|
def face_reseau(self) -> dict:
|
|
"""Ce que CE locataire demande a son site, en une fiche — les faits, pas encore les flux.
|
|
|
|
Le site lisait jusqu'ici six fichiers internes du locataire (relevé du 2026-10-04) :
|
|
sa nomenclature, son inventaire genere, ses `group_vars` (`nftables_admin_ssh`,
|
|
`serveur_backup_pubkey`), son `plan/acces.yml`, son `plan/domaines.yml` et ce qui
|
|
l'accompagne. La face reseau les publie, calcules ICI, chez le locataire, qui seul sait
|
|
lire ses fichiers : l'adresse de son primaire DNS vient de son inventaire (le site la
|
|
re-derivait, sequence 1 par defaut), sa cle de sauvegarde de SON dossier d'inventaire
|
|
(le site ecrivait `principal` en dur).
|
|
|
|
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.
|
|
"""
|
|
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
|
|
data = _yaml(self.inventaire())
|
|
hv = _hostvars(self.inventaire())
|
|
ips = _ip_par_hote(data) if data else {}
|
|
groupes = {g: sorted(h for h in _hotes_du_groupe(data, g) if h in ips)
|
|
for g in sorted(_enfants(data)) if est_groupe_operationnel(g)} if data else {}
|
|
machines = {
|
|
h: {"ip": ip,
|
|
"groupes": sorted(g for g, membres in groupes.items() if h in membres),
|
|
"materialisation": {k: v for k, v in sorted((hv.get(h) or {}).items())
|
|
if k.startswith("proxmox_")}}
|
|
for h, ip in sorted(ips.items())}
|
|
# Les reseaux d'administration que le locataire DECLARE (l'intrant), et son tunnel.
|
|
intrant: list[str] = []
|
|
gv_all = self.dossier_inventaire() / "group_vars" / "all"
|
|
for fichier in sorted(gv_all.glob("*.yml")) if gv_all.is_dir() else []:
|
|
if "vault" in fichier.name:
|
|
continue
|
|
d = _yaml(fichier)
|
|
if isinstance(d, dict) and d.get("nftables_admin_ssh"):
|
|
src = d["nftables_admin_ssh"]
|
|
intrant = [str(s) for s in src] if isinstance(src, list) else [str(src)]
|
|
break
|
|
try:
|
|
pairs = charger_acces(self.plan / "acces.yml").get("acces_admin_vpn") or {}
|
|
except (OSError, ValueError):
|
|
pairs = {}
|
|
# Ses zones publiques : celles qu'il fait servir par le site (`primaire-cache`).
|
|
doms = self.lire_plan("domaines.yml").get("domaines_publics") or {}
|
|
apps = self.lire_plan("applications.yml").get("applications") or {}
|
|
zones = sorted(z for z, c in doms.items()
|
|
if isinstance(c, dict) and str(c.get("autorite")) == "primaire-cache")
|
|
hote_dns = next((str(a.get("hote")) for a in apps.values()
|
|
if str(a.get("groupe")) == "serveur_powerdns"), "")
|
|
colocalise = any(str(a.get("groupe")) == "serveur_resolveur" and str(a.get("hote")) == hote_dns
|
|
for a in apps.values())
|
|
dns_public = ({"zones": zones,
|
|
"signees": sorted(z for z in zones if doms[z].get("dnssec") is True),
|
|
"primaire": ips.get(hote_dns, ""),
|
|
"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()
|
|
# 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.
|
|
flux = {}
|
|
for h in sorted(ips):
|
|
c = self.depot / "flux-genere" / f"{h}.connectivite.json"
|
|
if c.is_file():
|
|
d = json.loads(c.read_text(encoding="utf-8"))
|
|
flux[h] = {"entrees": d.get("entrees") or [], "administration": d.get("admin_ssh") or []}
|
|
sources = {str(Path(*f.relative_to(self.depot).parts)): _empreinte(f) for f in (
|
|
self.plan / "nomenclature.yml", self.inventaire(), self.plan / "acces.yml",
|
|
self.plan / "domaines.yml", self.plan / "applications.yml",
|
|
self.dossier_inventaire() / "group_vars" / "serveur_backup.yml")}
|
|
sources.update({f"flux-genere/{h}.connectivite.json":
|
|
_empreinte(self.depot / "flux-genere" / f"{h}.connectivite.json") for h in flux})
|
|
# Ses zones : un sous-reseau par categorie de sa nomenclature, derive de son index
|
|
# (`sous_reseau_de`, comme la frontiere et le SDN le derivent).
|
|
from inventory_rules import sous_reseau_de
|
|
cats = self.nomenclature().get("categories") or {}
|
|
zones = ([sous_reseau_de(self.index, int(z)) for z in sorted(cats, key=int)]
|
|
if self.index is not None else [])
|
|
return {
|
|
"locataire": self.nom,
|
|
"hebergeur": self.nom_hebergeur(),
|
|
"sources": sources,
|
|
"index": self.index,
|
|
"zones": zones,
|
|
"machines": machines,
|
|
"groupes": groupes,
|
|
"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")}
|
|
for n, c in sorted(pairs.items()) if isinstance(c, dict)}},
|
|
"dns_public": dns_public,
|
|
"sauvegarde": {"cle_publique": cle_sauvegarde},
|
|
"flux": flux,
|
|
}
|
|
|
|
def decrire(self) -> dict:
|
|
d = super().decrire()
|
|
d.update({"inventaire": str(self.inventaire()), "hebergeur": self.nom_hebergeur()})
|
|
return d
|
|
|
|
|
|
# --- LA FICHE DIT-ELLE CE QUE LE LOCATAIRE PORTE ? ----------------------------------------
|
|
|
|
def _hostvars(inventaire: Path) -> dict:
|
|
d = _yaml(inventaire)
|
|
hv: dict = {}
|
|
|
|
def parcourir(g: dict) -> None:
|
|
for h, v in (g.get("hosts") or {}).items():
|
|
hv.setdefault(h, {}).update(v or {})
|
|
for c in (g.get("children") or {}).values():
|
|
parcourir(c or {})
|
|
parcourir(d.get("all") or {})
|
|
return hv
|
|
|
|
|
|
def verifier_fiche(site: "Site", locataire: "Locataire", fiche: dict | None = None) -> list[str]:
|
|
"""Les ecarts entre la fiche du site et ce que le locataire porte AUJOURD'HUI.
|
|
|
|
Vide = la fiche dit exactement ce que les trois canaux d'avant ont depose chez le locataire :
|
|
ses copies (intrants, index, adresse publique, racine) et son inventaire genere (temps,
|
|
delegation DNS, rattachement SDN). C'est la condition pour que l'instancier puisse, a
|
|
l'etape 3, lire la fiche au lieu du site sans que l'inventaire change d'un octet.
|
|
"""
|
|
import site_intrants as si
|
|
f = fiche if fiche is not None else site.fiche_pour(locataire)
|
|
ecarts: list[str] = []
|
|
|
|
# 1. L'attribution.
|
|
if f["attribue"]["index"] != locataire.index:
|
|
ecarts.append(f"index : le site attribue {f['attribue']['index']}, "
|
|
f"le locataire porte {locataire.index}")
|
|
|
|
# 2. L'offre et l'adresse publique, contre les copies du locataire (la table de `site_intrants`).
|
|
with _monter(site, locataire):
|
|
vu = si._declare_par_le_locataire()
|
|
valeurs = dict(f["offre"], ip_publique=f["attribue"]["ip_publique"])
|
|
for clef, copie in vu.items():
|
|
attendu = valeurs.get(clef)
|
|
if not attendu or "{{" in str(copie):
|
|
continue
|
|
if si._normaliser(copie) != si._normaliser(attendu):
|
|
fichier, variable = si.OU_LE_LOCATAIRE_LE_DIT[clef]
|
|
ecarts.append(f"{variable} ({fichier}) : le locataire dit « {copie} », la fiche « {attendu} »")
|
|
|
|
# 3. La racine de confiance.
|
|
copie_ac = locataire.depot / "ac-racine-site.crt"
|
|
if copie_ac.is_file() and _empreinte(copie_ac) != f["racine_confiance"]["sha256"]:
|
|
ecarts.append("ac-racine-site.crt : la copie du locataire n'est pas la racine du site")
|
|
|
|
# 4. Ce que l'instancier a lu dans le site, tel qu'il l'a ecrit dans l'inventaire.
|
|
hv = _hostvars(locataire.inventaire())
|
|
if not hv:
|
|
ecarts.append(f"inventaire introuvable ou vide : {locataire.inventaire()}")
|
|
return ecarts
|
|
gen = f["generation"]
|
|
temps = gen["serveurs_temps"] or None
|
|
for h, v in sorted(hv.items()):
|
|
if v.get("chrony_serveurs") != temps:
|
|
ecarts.append(f"{h} : chrony_serveurs = {v.get('chrony_serveurs')}, la fiche dit {temps}")
|
|
break
|
|
domaine_loc = locataire.domaine_interne()
|
|
attendu_zones = ([{"adresses": gen["resolveurs_site"], "nom": gen["zone_site"]}]
|
|
if gen["zone_site"] and gen["resolveurs_site"] and gen["zone_site"] != domaine_loc
|
|
else [])
|
|
portees = [v["serveur_resolveur_zones_deleguees"] for v in hv.values()
|
|
if "serveur_resolveur_zones_deleguees" in v]
|
|
normal = [sorted(({"adresses": list(z.get("adresses") or []), "nom": z.get("nom")} for z in p),
|
|
key=lambda z: str(z["nom"])) for p in portees]
|
|
if attendu_zones and not normal:
|
|
ecarts.append("delegation DNS : la fiche en offre une, l'inventaire n'en porte aucune")
|
|
for p in normal:
|
|
if p != attendu_zones and (attendu_zones or p):
|
|
ecarts.append(f"delegation DNS : l'inventaire porte {p}, la fiche dit {attendu_zones}")
|
|
break
|
|
sdn = gen["routage_tenants"] == "sdn"
|
|
for h, v in sorted(hv.items()):
|
|
if sdn and (not v.get("proxmox_pont") or v.get("proxmox_etiquette_vlan")):
|
|
ecarts.append(f"{h} : routage SDN selon la fiche, mais pont « {v.get('proxmox_pont')} », "
|
|
f"etiquette « {v.get('proxmox_etiquette_vlan')} »")
|
|
break
|
|
if not sdn and v.get("proxmox_pont"):
|
|
ecarts.append(f"{h} : routage « {gen['routage_tenants']} » selon la fiche, mais un pont SDN")
|
|
break
|
|
return ecarts
|
|
|
|
|
|
def verifier_face(site: "Site", locataire: "Locataire", face: dict | None = None) -> list[str]:
|
|
"""Les ecarts entre la face reseau du locataire et ce que les CONSOMMATEURS DU SITE en tirent
|
|
aujourd'hui, en lisant ses fichiers internes.
|
|
|
|
On interroge les consommateurs eux-memes, le site monte : la decouverte des locataires
|
|
(`devis_reseau`), le devis du pare-feu Proxmox (ses ensembles d'adresses par groupe et
|
|
d'administration), la frontiere (`admin_de`), et l'inventaire du site (relations du DNS
|
|
public, comptes de sauvegarde). Vide = chacun lirait la face reseau sans rien changer.
|
|
"""
|
|
import devis_proxmox_fw
|
|
import devis_reseau
|
|
import site_inventaire
|
|
f = face if face is not None else locataire.face_reseau()
|
|
ecarts: list[str] = []
|
|
with _monter(site):
|
|
decouverts = {nom: n for nom, _p, n in devis_reseau.decouvrir_du_site()}
|
|
bloc = next((b for b in devis_proxmox_fw.construire(devis_reseau.decouvrir_du_site())["blocs"]
|
|
if b.get("tenant") == locataire.nom), None)
|
|
admin_frontiere = sorted(devis_reseau.admin_de(locataire.nom))
|
|
inv = site_inventaire.inventaire() or {}
|
|
hv_site = (inv.get("_meta") or {}).get("hostvars") or {}
|
|
|
|
# 1. L'index, tel que le site le decouvre.
|
|
n = decouverts.get(locataire.nom)
|
|
if n is None:
|
|
ecarts.append("le site ne decouvre pas ce locataire")
|
|
elif n.get("index") != f["index"]:
|
|
ecarts.append(f"index : le site decouvre {n.get('index')}, la face dit {f['index']}")
|
|
|
|
# 2. Les machines et leurs groupes, tels que le pare-feu Proxmox les range.
|
|
ips = {h: m["ip"] for h, m in f["machines"].items()}
|
|
if bloc is None:
|
|
ecarts.append("le devis Proxmox n'a aucun bloc pour ce locataire")
|
|
else:
|
|
ens = bloc.get("ipsets") or {}
|
|
flotte = next((v["membres"] for v in ens.values() if str(v.get("role", "")).startswith("flotte")), [])
|
|
if sorted(flotte) != sorted(set(ips.values())):
|
|
ecarts.append(f"flotte : le devis Proxmox range {sorted(flotte)}, la face {sorted(set(ips.values()))}")
|
|
# Le devis ne garde que les ensembles que ses regles UTILISENT : chaque groupe qu'il range
|
|
# doit exister dans la face avec les memes adresses ; un groupe inutilise n'est pas un ecart.
|
|
speciaux = ("flotte", "Internet", "reseaux d'admin")
|
|
par_groupe = {str(v.get("role")): sorted(v.get("membres") or []) for v in ens.values()
|
|
if not str(v.get("role", "")).startswith(speciaux)}
|
|
for g, membres in sorted(par_groupe.items()):
|
|
if g not in f["groupes"]:
|
|
ecarts.append(f"groupe {g} : range par le devis Proxmox, absent de la face")
|
|
elif membres != sorted({ips[h] for h in f["groupes"][g]}):
|
|
ecarts.append(f"groupe {g} : le devis Proxmox range {membres}, "
|
|
f"la face {sorted({ips[h] for h in f['groupes'][g]})}")
|
|
adm = next((v["membres"] for v in ens.values() if str(v.get("role", "")).startswith("reseaux d'admin")), [])
|
|
attendu_adm = sorted(set(f["administration"]["intrant"]) | set(f["administration"]["tunnel"]))
|
|
if sorted(adm) != attendu_adm:
|
|
ecarts.append(f"administration (Proxmox) : le devis admet {sorted(adm)}, la face {attendu_adm}")
|
|
|
|
# 3. L'administration, telle que la frontiere la lit.
|
|
if admin_frontiere != sorted(f["administration"]["intrant"]):
|
|
ecarts.append(f"administration (frontiere) : le site lit {admin_frontiere}, "
|
|
f"la face {sorted(f['administration']['intrant'])}")
|
|
|
|
# 4. Le DNS public et la sauvegarde, tels que l'inventaire du site les porte.
|
|
relations = [r for v in hv_site.values() for r in (v.get("serveur_dns_public_relations") or [])
|
|
if r.get("ecosysteme") == locataire.nom]
|
|
rel = relations[0] if relations else {}
|
|
attendu_dns = f["dns_public"]
|
|
vu_dns = ({k: rel.get(k) for k in ("zones", "signees", "primaire", "port")} if rel else {})
|
|
if vu_dns != attendu_dns:
|
|
ecarts.append(f"DNS public : l'inventaire du site porte {vu_dns}, la face {attendu_dns}")
|
|
comptes = [c for v in hv_site.values() for c in (v.get("serveur_backup_site_locataires") or [])
|
|
if c.get("nom") == locataire.nom]
|
|
cle_site = comptes[0].get("pubkey") if comptes else ""
|
|
if (cle_site or "") != f["sauvegarde"]["cle_publique"]:
|
|
ecarts.append("sauvegarde : la cle que le site autorise n'est pas celle que le locataire publie")
|
|
return ecarts
|
|
|
|
|
|
# LES DEUX PARE-FEUX ECRIVENT LA MEME POLITIQUE AUTREMENT. On les ramene a une forme commune :
|
|
# pour chaque machine, pour chaque (protocole, port), l'ENSEMBLE des sources admises, chacune
|
|
# etant une adresse, `EXTERNE` ou `ADMIN`. Les differences d'ecriture, toutes neutralisees ici :
|
|
# - Proxmox ecrit une regle par groupe source ; le locataire reunit les sources d'un port ;
|
|
# - Proxmox nomme un ensemble (`+t23-srv-nginx`) ou ecrit les adresses en clair, separees
|
|
# par des virgules (la fabric vers Loki, le DNS public du site) ;
|
|
# - « Internet » : `0.0.0.0/0` chez le locataire, l'ensemble « tout sauf RFC 1918 » chez
|
|
# Proxmox, double d'une regle « administration (poste) » sur le meme port ;
|
|
# - l'administration : une liste a part (`admin_ssh`) et des entrees ordinaires chez le
|
|
# locataire, l'ensemble « reseaux d'administration » chez Proxmox ;
|
|
# - l'ICMP : `frag-needed` d'un cote, `fragmentation-needed` de l'autre.
|
|
# Un port ouvert a tous ET a des clients nommes porte ces derniers dans `sources_declarees`
|
|
# (depuis le 2026-10-04) : sans eux, Proxmox ne saurait pas qui admettre en plus d'Internet.
|
|
EXTERNE, ADMIN = "EXTERNE", "ADMIN"
|
|
_ICMP = {"frag-needed": "fragmentation-needed"}
|
|
|
|
|
|
def _entrees_locataire(face: dict) -> dict[str, dict]:
|
|
out: dict[str, dict] = {}
|
|
for h, f in (face.get("flux") or {}).items():
|
|
admin = set(f.get("administration") or [])
|
|
par_port = out.setdefault(h, {})
|
|
for e in f.get("entrees") or []:
|
|
cle = (str(e.get("proto")), _ICMP.get(str(e.get("port")), str(e.get("port"))))
|
|
atomes = par_port.setdefault(cle, set())
|
|
# `sources_declarees` : les clients nommes d'un port aussi ouvert a tous (`0.0.0.0/0`).
|
|
for s in list(e.get("sources") or []) + list(e.get("sources_declarees") or []):
|
|
atomes.add(EXTERNE if s == "0.0.0.0/0" else ADMIN if s in admin else str(s))
|
|
if admin:
|
|
par_port.setdefault(("tcp", "22"), set()).add(ADMIN)
|
|
return out
|
|
|
|
|
|
def _entrees_proxmox(bloc: dict) -> dict[str, dict]:
|
|
ens = bloc.get("ipsets") or {}
|
|
regles = {g["nom"]: g.get("regles") or [] for g in bloc.get("groupes") or []}
|
|
out: dict[str, dict] = {}
|
|
for a in bloc.get("affectations") or []:
|
|
par_port = out.setdefault(a["hote"], {})
|
|
for g in a.get("groupes") or []:
|
|
for r in regles.get(g, []):
|
|
if r.get("action") != "ACCEPT" or r.get("sens") != "IN":
|
|
continue
|
|
if str(r.get("via", "")) == "administration (poste)":
|
|
continue # le double d'une regle `internet` sur le meme port
|
|
source = str(r.get("source", ""))
|
|
if source.startswith("+"):
|
|
e = ens.get(source[1:]) or {}
|
|
role = str(e.get("role", ""))
|
|
atomes = ({EXTERNE} if role.startswith("Internet") else
|
|
{ADMIN} if role.startswith("reseaux d'admin") else
|
|
{str(m) for m in e.get("membres") or []})
|
|
else:
|
|
atomes = {s.strip() for s in source.split(",") if s.strip()}
|
|
for p in str(r.get("dport") or r.get("icmp_type")).split(","):
|
|
cle = (str(r.get("proto")), _ICMP.get(p.strip(), p.strip()))
|
|
par_port.setdefault(cle, set()).update(atomes)
|
|
return out
|
|
|
|
|
|
def verifier_flux(site: "Site", locataire: "Locataire", face: dict | None = None) -> list[str]:
|
|
"""Machine par machine : le pare-feu du locataire (nftables, resolu par lui) et celui de
|
|
Proxmox (devise par le site) admettent-ils les memes entrees ?
|
|
|
|
C'est le constat 2 de la conception : les flux etaient calcules DEUX FOIS, chacun avec sa
|
|
version du moteur. Vide = les deux pare-feux disent la meme chose, et le site pourra, a
|
|
l'etape 3, deviser Proxmox depuis la face reseau sans rien recalculer.
|
|
"""
|
|
import devis_proxmox_fw
|
|
import devis_reseau
|
|
f = face if face is not None else locataire.face_reseau()
|
|
with _monter(site):
|
|
bloc = next((b for b in devis_proxmox_fw.construire(devis_reseau.decouvrir_du_site())["blocs"]
|
|
if b.get("tenant") == locataire.nom), None)
|
|
if bloc is None:
|
|
return ["le devis Proxmox n'a aucun bloc pour ce locataire"]
|
|
loc, pve = _entrees_locataire(f), _entrees_proxmox(bloc)
|
|
ecarts: list[str] = []
|
|
for h in sorted(set(loc) | set(pve)):
|
|
if h not in loc:
|
|
ecarts.append(f"{h} : filtre par Proxmox, sans flux publie par le locataire")
|
|
continue
|
|
if h not in pve:
|
|
ecarts.append(f"{h} : flux publies par le locataire, aucune affectation Proxmox")
|
|
continue
|
|
|
|
for cle in sorted(set(loc[h]) | set(pve[h])):
|
|
l, p = loc[h].get(cle, set()), pve[h].get(cle, set())
|
|
if l == p:
|
|
continue
|
|
port = f"{cle[0]}/{cle[1]}"
|
|
if l - p:
|
|
ecarts.append(f"{h} {port} : le locataire admet {sorted(l - p)}, Proxmox non")
|
|
if p - l:
|
|
ecarts.append(f"{h} {port} : Proxmox admet {sorted(p - l)}, le locataire non")
|
|
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,
|
|
fiche: dict | None = None) -> list[str]:
|
|
"""La frontiere, premier temps : les IDENTITES par lesquelles elle designe ce locataire.
|
|
|
|
Ses alias (supernet, administration par interface, tunnel, un alias par groupe qui a des
|
|
flux de bordure), ses routes, et sa traduction sortante. Chacun doit dire ce que la face
|
|
reseau du locataire (zones, groupes, administration) et la fiche du site (adresse publique
|
|
attribuee) portent deja. Les entrees publiques et les sorties sont les temps suivants.
|
|
"""
|
|
import re
|
|
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"]
|
|
ecarts: list[str] = []
|
|
alias = devis.get("alias") or {}
|
|
ips = {h: m["ip"] for h, m in f["machines"].items()}
|
|
|
|
# Le supernet : ses zones.
|
|
sup = (alias.get(f"SETOPS_TENANT_{etq}") or {}).get("contenu")
|
|
if sorted(sup or []) != sorted(f["zones"]):
|
|
ecarts.append(f"SETOPS_TENANT_{etq} : la frontiere porte {sup}, la face {f['zones']}")
|
|
# L'administration : l'intrant, que la frontiere range par interface, et le tunnel.
|
|
adm = sorted({c for k, v in alias.items() if re.fullmatch(rf"SETOPS_ADMIN_{etq}_(?!TUNNEL)\w+", k)
|
|
for c in v.get("contenu") or []})
|
|
if adm != sorted(f["administration"]["intrant"]):
|
|
ecarts.append(f"administration : la frontiere range {adm}, la face {sorted(f['administration']['intrant'])}")
|
|
tun = sorted((alias.get(f"SETOPS_ADMIN_{etq}_TUNNEL") or {}).get("contenu") or [])
|
|
if tun != sorted(f["administration"]["tunnel"]):
|
|
ecarts.append(f"tunnel : la frontiere porte {tun}, la face {sorted(f['administration']['tunnel'])}")
|
|
# Un alias par groupe : ses membres sont ceux du groupe dans la face.
|
|
for k, v in sorted(alias.items()):
|
|
m = re.search(r"portant (\S+) \(instance (\S+)\)", str(v.get("description", "")))
|
|
if not m or m.group(2) != locataire.nom:
|
|
continue
|
|
role = m.group(1)
|
|
attendu = sorted({ips[h] for h in f["groupes"].get(role, [])})
|
|
if role not in f["groupes"]:
|
|
ecarts.append(f"{k} : groupe {role} inconnu de la face")
|
|
elif sorted(v.get("contenu") or []) != attendu:
|
|
ecarts.append(f"{k} : la frontiere porte {sorted(v.get('contenu') or [])}, la face {attendu}")
|
|
# Les routes : une par zone.
|
|
routes = sorted(r["reseau"] for r in devis.get("routes") or [] if r.get("tenant") == locataire.nom)
|
|
if routes != sorted(f["zones"]):
|
|
ecarts.append(f"routes : la frontiere route {routes}, la face a pour zones {sorted(f['zones'])}")
|
|
# La traduction sortante : sa flotte sort par l'adresse que le site lui attribue.
|
|
nat = [x for x in devis.get("nat") or [] if x.get("tenant") == locataire.nom]
|
|
cibles = sorted(str(x.get("cible")) for x in nat)
|
|
if cibles != [str(fi["attribue"]["ip_publique"])]:
|
|
ecarts.append(f"traduction sortante : la frontiere fait sortir par {cibles}, "
|
|
f"la fiche attribue {fi['attribue']['ip_publique']}")
|
|
if any(x.get("source") != f"SETOPS_TENANT_{etq}" for x in nat):
|
|
ecarts.append("traduction sortante : la source n'est pas le supernet du locataire")
|
|
return ecarts
|
|
|
|
|
|
def couples_hebergeur() -> list[tuple["Site", "Locataire"]]:
|
|
"""Chaque (site, locataire) dont ce site est l'hebergeur ACTIF — les seuls qui ont une fiche
|
|
a verifier : un site de reprise ne depose rien tant que la bascule n'a pas eu lieu."""
|
|
out = []
|
|
for underlay in sorted(DOSSIER_ECOSYSTEMES.glob("*/underlay.yml")):
|
|
site = Site(underlay.parent)
|
|
out += [(site, loc) for loc in site.locataires() if loc.nom_hebergeur() == site.nom]
|
|
return out
|
|
|
|
|
|
# --- LE CONTEXTE ACTIF --------------------------------------------------------------------
|
|
|
|
def _lire_fichier_contexte(fichier: Path, dossier: Path) -> Ecosysteme:
|
|
lignes = [l.split("#", 1)[0].strip()
|
|
for l in fichier.read_text(encoding="utf-8").splitlines()]
|
|
valeurs = [l for l in lignes if l]
|
|
if len(valeurs) != 1 or ":" not in valeurs[0]:
|
|
raise ContexteInvalide(
|
|
f"{fichier} doit porter UNE valeur, `site:<depot>` ou `locataire:<depot>` "
|
|
f"(lu : {valeurs or 'rien'}).")
|
|
nature, nom = (x.strip() for x in valeurs[0].split(":", 1))
|
|
if nature == "site":
|
|
return Site.charger(nom, dossier)
|
|
if nature == "locataire":
|
|
return Locataire.charger(nom, dossier)
|
|
raise ContexteInvalide(f"{fichier} : nature « {nature} » inconnue ({' ou '.join(NATURES)}).")
|
|
|
|
|
|
def montes(racine: Path = RACINE) -> list[Ecosysteme]:
|
|
"""Ce que les INDICES D'AVANT designent : variables `SETOPS_*`, puis liens a la racine.
|
|
|
|
Garde le temps de la bascule, et pour la dire : sur le poste d'aujourd'hui, cette liste
|
|
rend DEUX ecosystemes, ce qu'un selecteur ne doit plus faire.
|
|
"""
|
|
# LES DEUX RESOLUTIONS QUI FONT DEJA FOI, pas une troisieme copie (P41) :
|
|
# `instance_courante()` lit `SETOPS_INSTANCE`, puis le lien `instance/` ; `underlay.chemin()`
|
|
# lit `SETOPS_UNDERLAY`, puis le lien `underlay.yml`. Quand on interroge une AUTRE racine que
|
|
# celle du moteur (les tests), leurs liens par defaut sont remplaces par ceux de cette racine.
|
|
racine = Path(racine)
|
|
autre_racine = racine.resolve() != RACINE.resolve()
|
|
out: list[Ecosysteme] = []
|
|
inst_p = instance_courante()
|
|
if autre_racine and inst_p == RACINE.joinpath("instance"):
|
|
inst_p = racine.joinpath("instance")
|
|
if not inst_p.is_absolute():
|
|
inst_p = racine / inst_p
|
|
if (inst_p / "plan" / "nomenclature.yml").is_file():
|
|
out.append(Locataire(inst_p.resolve()))
|
|
und_p = underlay_mod.chemin()
|
|
if autre_racine and (und_p is None or und_p == RACINE.joinpath("underlay.yml")):
|
|
und_p = racine.joinpath("underlay.yml")
|
|
if und_p is not None and not und_p.is_absolute():
|
|
und_p = racine / und_p
|
|
if und_p is not None and und_p.is_file():
|
|
out.append(Site(und_p.resolve().parent))
|
|
return out
|
|
|
|
|
|
def actif(racine: Path = RACINE, dossier: Path | None = None) -> Ecosysteme | None:
|
|
"""LE contexte actif de cette machine.
|
|
|
|
1. Le fichier `contexte` a la racine du moteur, s'il existe : il fait foi.
|
|
2. Sinon, les indices d'avant (`montes`). S'ils designent un locataire ET un site — le
|
|
poste d'aujourd'hui —, le locataire l'emporte : c'est ce que fait deja la console
|
|
(`ConsolePoste` herite de `ConsoleLocataire`). Le fichier `contexte` leve l'ambiguite.
|
|
3. Rien : None. A l'appelant de le dire, jamais de le dessiner comme un plan vide.
|
|
"""
|
|
dossier = Path(dossier) if dossier is not None else Path(racine).parent
|
|
fichier = Path(racine) / NOM_FICHIER_CONTEXTE
|
|
if fichier.is_file():
|
|
return _lire_fichier_contexte(fichier, dossier)
|
|
m = montes(racine)
|
|
return m[0] if m else None
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
|
ap.add_argument("--json", action="store_true")
|
|
ap.add_argument("--fiche", metavar="LOCATAIRE", help="la fiche que son hebergeur lui destine")
|
|
ap.add_argument("--verifier-fiches", action="store_true",
|
|
help="chaque fiche dit-elle ce que le locataire porte ?")
|
|
ap.add_argument("--face", metavar="LOCATAIRE", help="la face reseau qu'il publie pour son site")
|
|
ap.add_argument("--verifier-faces", action="store_true",
|
|
help="chaque face reseau dit-elle ce que le site en tire ?")
|
|
args = ap.parse_args(argv)
|
|
if args.face:
|
|
print(yaml.safe_dump(Locataire.charger(args.face).face_reseau(), allow_unicode=True,
|
|
sort_keys=False), end="")
|
|
return 0
|
|
if args.verifier_faces:
|
|
couples = couples_hebergeur()
|
|
if not couples:
|
|
print("Aucun site frere : aucune face reseau a verifier.")
|
|
return 0
|
|
total = 0
|
|
for site, loc in couples:
|
|
face = loc.face_reseau()
|
|
for quoi, e in (("faits", verifier_face(site, loc, face)),
|
|
("flux ", verifier_flux(site, loc, face)),
|
|
("frontiere (identites)", verifier_frontiere(site, loc, face)),
|
|
("frontiere (entrees publiques)", verifier_entrees_publiques(site, loc, face))):
|
|
total += len(e)
|
|
print(f"{'OK ' if not e else 'ECART'} {loc.nom} -> {site.nom} ({quoi})"
|
|
+ "".join(f"\n - {x}" for x in e))
|
|
return 1 if total else 0
|
|
if args.fiche:
|
|
loc = Locataire.charger(args.fiche)
|
|
site = loc.site()
|
|
if site is None:
|
|
print(f"{loc.nom} ne nomme aucun hebergeur.")
|
|
return 1
|
|
print(yaml.safe_dump(site.fiche_pour(loc), allow_unicode=True, sort_keys=False), end="")
|
|
return 0
|
|
if args.verifier_fiches:
|
|
couples = couples_hebergeur()
|
|
if not couples:
|
|
print("Aucun site frere : aucune fiche a verifier.")
|
|
return 0
|
|
total = 0
|
|
for site, loc in couples:
|
|
e = verifier_fiche(site, loc)
|
|
total += len(e)
|
|
print(f"{'OK ' if not e else 'ECART'} {site.nom} -> {loc.nom}"
|
|
+ "".join(f"\n - {x}" for x in e))
|
|
return 1 if total else 0
|
|
try:
|
|
eco = actif()
|
|
erreur = None
|
|
except ContexteInvalide as e:
|
|
eco, erreur = None, str(e)
|
|
fichier = RACINE / NOM_FICHIER_CONTEXTE
|
|
rapport = {
|
|
"source": "fichier `contexte`" if fichier.is_file() else "indices d'avant (liens, variables)",
|
|
"actif": eco.decrire() if eco else None,
|
|
"erreur": erreur,
|
|
"indices": [e.decrire() for e in montes()],
|
|
}
|
|
if args.json:
|
|
print(json.dumps(rapport, indent=2, ensure_ascii=False))
|
|
return 0 if eco else 1
|
|
print(f"Source : {rapport['source']}")
|
|
if erreur:
|
|
print(f"REFUS : {erreur}")
|
|
elif eco:
|
|
d = eco.decrire()
|
|
print(f"Contexte actif : {d['nature']} {d['nom']} (index {d['index']})")
|
|
if isinstance(eco, Site):
|
|
print(f" locataires : {', '.join(d['locataires']) or 'aucun'}"
|
|
+ (f" ; ABSENTS : {', '.join(d['locataires_absents'])}" if d['locataires_absents'] else ""))
|
|
else:
|
|
print(f" hebergeur actif : {d['hebergeur'] or 'aucun nomme'}")
|
|
else:
|
|
print("Aucun contexte : ni fichier `contexte`, ni locataire, ni site monte.")
|
|
if len(rapport["indices"]) > 1:
|
|
print(" NOTE : les indices d'avant designent "
|
|
+ " ET ".join(f"{i['nature']} {i['nom']}" for i in rapport["indices"])
|
|
+ " — un selecteur n'en garde qu'un (fichier `contexte`).")
|
|
return 0 if eco else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|