Daniel Allaire
36a882b125
Principe : les valeurs de configuration se derivent des intrants, elles ne se reecrivent pas a la main. La nomenclature dupliquait ce qu'index determine deja (supernet, sous-reseaux, passerelles, VLAN). Corrige en rupture nette. - inventory_rules : source unique de derivation — supernet_de, base3_de, sous_reseau_de, passerelle_de, vlan_de. Modele 6 zones encode une fois (2e octet = 10+index, 3e octet zone = 15+categorie, VLAN = 1000+index*10+zone). deriver_nomenclature ne lit plus aucun adressage stocke ; mode compact supprime. - devis_reseau : importe ces helpers (fin de la duplication) ; decouvre les tenants sur `index` present (filtre vmid_schema retire). - GUI : `index` devient un INTRANT (section Reseau). Il vit dans la nomenclature (plan reseau uniforme, contrairement aux intrants des modeles heterogenes) et le GUI l'ecrit chirurgicalement (une ligne, sans reformater). Le miroir JS derive le VLAN du seed (fin de la lecture de c.vlan stocke). - socle public : nomenclature au format maigre. Preuve P20 (preuve_nomenclature_derivee) : aucune nomenclature ne stocke d'adressage — garde-fou permanent, teste en negatif. Valide : DIFF VIDE sur les 3 instances (la derivation reproduit exactement l'adressage stocke), 7 modeles valident, devis_reseau genere les memes VLAN (1011-1016 derives), make verifier rc=0 CONFORME 20/20, node --check du GUI OK. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
605 lines
26 KiB
Python
605 lines
26 KiB
Python
#!/usr/bin/env python3
|
|
"""Regles partagees pour les inventaires statiques Set-OPS."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
|
|
GROUPES_OPERATIONNELS_PREFIXES = ("serveur_", "client_")
|
|
GROUPE_HOTES_ACTIFS = "hotes_actifs"
|
|
GROUPE_HOTES_PLANIFIES = "hotes_planifies"
|
|
GROUPES_ETAT_HOTE = {GROUPE_HOTES_ACTIFS, GROUPE_HOTES_PLANIFIES}
|
|
|
|
RACINE_DEPOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def liens_acceptes(groupe: str, racine: Path | None = None) -> dict:
|
|
"""Liens qu'un role accepte : roles/<groupe>/meta/liens.yml -> setops_liens.accepte.
|
|
|
|
Source unique partagee par le validateur, le GUI et scripts/instancier.py.
|
|
Retourne {} si le role n'en declare aucun.
|
|
"""
|
|
fichier = (racine or RACINE_DEPOT) / "roles" / groupe / "meta" / "liens.yml"
|
|
if not fichier.is_file():
|
|
return {}
|
|
data = yaml.safe_load(fichier.read_text(encoding="utf-8")) or {}
|
|
return ((data.get("setops_liens") or {}).get("accepte")) or {}
|
|
|
|
|
|
def catalogue_liens(racine: Path | None = None) -> dict:
|
|
"""{groupe: {role_de_lien: {cible, injecte_sur, variables}}} pour tous les roles.
|
|
|
|
Alimente la liste deroulante « rôle du lien » du GUI : on ne propose que ce
|
|
que le role porteur declare accepter.
|
|
"""
|
|
base = (racine or RACINE_DEPOT) / "roles"
|
|
if not base.is_dir():
|
|
return {}
|
|
catalogue = {}
|
|
for meta in sorted(base.glob("*/meta/liens.yml")):
|
|
accepte = liens_acceptes(meta.parents[1].name, racine)
|
|
if accepte:
|
|
catalogue[meta.parents[1].name] = accepte
|
|
return catalogue
|
|
|
|
|
|
def est_groupe_operationnel(groupe: str) -> bool:
|
|
return groupe.startswith(GROUPES_OPERATIONNELS_PREFIXES)
|
|
|
|
|
|
def groupes_operationnels_connus(enfants: dict, dossier_playbooks: Path) -> list[str]:
|
|
groupes = set(enfants)
|
|
groupes.update(playbook.stem for playbook in dossier_playbooks.glob("*.yml"))
|
|
return sorted(groupe for groupe in groupes if est_groupe_operationnel(groupe))
|
|
|
|
|
|
def charger_bases_donnees(path: Path | None) -> dict:
|
|
"""Charge le registre des bases de donnees (serveurs_bd + bases_donnees)."""
|
|
if not path or not path.exists():
|
|
return {"serveurs_bd": {}, "bases_donnees": {}}
|
|
with path.open("r", encoding="utf-8") as fichier:
|
|
data = yaml.safe_load(fichier) or {}
|
|
if not isinstance(data, dict):
|
|
raise ValueError(f"{path} ne contient pas une table YAML.")
|
|
data.setdefault("serveurs_bd", {})
|
|
data.setdefault("bases_donnees", {})
|
|
return data
|
|
|
|
|
|
PORTEES_BD = {"groupe", "hote", "application"}
|
|
|
|
|
|
def valider_bases(registre: dict, applications: dict | None = None) -> None:
|
|
"""Valide la coherence du registre des bases de donnees.
|
|
|
|
Si `applications` (le registre des applications) est fourni, les bases de
|
|
portee 'application' sont verifiees : leur consommateur doit etre une
|
|
application declaree.
|
|
"""
|
|
serveurs = registre.get("serveurs_bd") or {}
|
|
bases = registre.get("bases_donnees") or {}
|
|
apps = (applications or {}).get("applications") or {}
|
|
if not isinstance(serveurs, dict) or not isinstance(bases, dict):
|
|
raise ValueError("serveurs_bd et bases_donnees doivent etre des tables.")
|
|
for nom, srv in serveurs.items():
|
|
if not isinstance(srv, dict):
|
|
raise ValueError(f"Serveur BD '{nom}': table attendue.")
|
|
for champ in ("type", "hote", "port"):
|
|
if not str(srv.get(champ, "")).strip():
|
|
raise ValueError(f"Serveur BD '{nom}': champ '{champ}' requis.")
|
|
noms_bases: set[str] = set()
|
|
for cle, base in bases.items():
|
|
if not isinstance(base, dict):
|
|
raise ValueError(f"Base '{cle}': table attendue.")
|
|
for champ in ("serveur", "base", "proprietaire", "secret", "consommateur"):
|
|
if not str(base.get(champ, "")).strip():
|
|
raise ValueError(f"Base '{cle}': champ '{champ}' requis.")
|
|
if base["serveur"] not in serveurs:
|
|
raise ValueError(f"Base '{cle}': serveur '{base['serveur']}' inconnu.")
|
|
portee = str(base.get("portee", "groupe")).strip()
|
|
if portee not in PORTEES_BD:
|
|
attendu = ", ".join(sorted(PORTEES_BD))
|
|
raise ValueError(f"Base '{cle}': portee '{portee}' inconnue (attendu: {attendu}).")
|
|
if portee == "groupe" and not est_groupe_operationnel(base["consommateur"]):
|
|
raise ValueError(f"Base '{cle}': portee groupe mais consommateur '{base['consommateur']}' non operationnel.")
|
|
if portee == "application" and apps and base["consommateur"] not in apps:
|
|
raise ValueError(f"Base '{cle}': application consommatrice '{base['consommateur']}' inconnue (instance/plan/applications.yml).")
|
|
couple = f"{base['serveur']}/{base['base']}"
|
|
if couple in noms_bases:
|
|
raise ValueError(f"Base en double sur un meme serveur: {couple}")
|
|
noms_bases.add(couple)
|
|
|
|
|
|
def _base_resolue(cle: str, base: dict, serveurs: dict) -> dict:
|
|
return {"cle": cle, "base": base, "serveur": serveurs.get(base.get("serveur"), {})}
|
|
|
|
|
|
def bases_du_groupe(registre: dict, groupe: str) -> list[dict]:
|
|
"""Bases de portee 'groupe' consommees par un groupe applicatif."""
|
|
serveurs = registre.get("serveurs_bd") or {}
|
|
resultat = []
|
|
for cle, base in (registre.get("bases_donnees") or {}).items():
|
|
if str(base.get("portee", "groupe")) == "groupe" and base.get("consommateur") == groupe:
|
|
resultat.append(_base_resolue(cle, base, serveurs))
|
|
return resultat
|
|
|
|
|
|
def bases_de_application(registre: dict, application: str | None = None,
|
|
groupe: str | None = None, hote: str | None = None) -> list[dict]:
|
|
"""Bases vues par une application instanciee, toutes portees confondues.
|
|
|
|
Une application `A` (du groupe `G`, sur l'hote `H`) recoit les bases ou :
|
|
(portee=application ET consommateur=A) OU (portee=groupe ET consommateur=G)
|
|
OU (portee=hote ET consommateur=H).
|
|
"""
|
|
serveurs = registre.get("serveurs_bd") or {}
|
|
resultat = []
|
|
for cle, base in (registre.get("bases_donnees") or {}).items():
|
|
portee = str(base.get("portee", "groupe"))
|
|
conso = base.get("consommateur")
|
|
if ((portee == "application" and application is not None and conso == application)
|
|
or (portee == "groupe" and groupe is not None and conso == groupe)
|
|
or (portee == "hote" and hote is not None and conso == hote)):
|
|
resultat.append(_base_resolue(cle, base, serveurs))
|
|
return resultat
|
|
|
|
|
|
def charger_applications(path: Path | None) -> dict:
|
|
"""Charge le registre des applications (applications -> groupe, hote)."""
|
|
if not path or not path.exists():
|
|
return {"applications": {}}
|
|
with path.open("r", encoding="utf-8") as fichier:
|
|
data = yaml.safe_load(fichier) or {}
|
|
if not isinstance(data, dict):
|
|
raise ValueError(f"{path} ne contient pas une table YAML.")
|
|
data.setdefault("applications", {})
|
|
return data
|
|
|
|
|
|
def valider_applications(registre: dict, domaines: dict | None = None,
|
|
serveurs: dict | None = None) -> None:
|
|
"""Valide la coherence du registre des applications.
|
|
|
|
Si `domaines` (le registre des domaines publics) est fourni, chaque FQDN
|
|
de `expose` doit avoir un domaine parent declare.
|
|
|
|
Si `serveurs` (le registre des serveurs) est fourni, l'hote porteur de chaque
|
|
application doit y etre declare — c'est le garde-fou contre l'hote FANTOME
|
|
(une application posee sur une VM qui n'existe pas). Sans lui, le plan passe
|
|
la validation et casse au deploiement.
|
|
"""
|
|
apps = registre.get("applications") or {}
|
|
doms = (domaines or {}).get("domaines_publics") or {}
|
|
if not isinstance(apps, dict):
|
|
raise ValueError("applications doit etre une table.")
|
|
for nom, app in apps.items():
|
|
if not isinstance(app, dict):
|
|
raise ValueError(f"Application '{nom}': table attendue.")
|
|
for champ in ("groupe", "hote"):
|
|
if not str(app.get(champ, "")).strip():
|
|
raise ValueError(f"Application '{nom}': champ '{champ}' requis.")
|
|
hotes_connus = (serveurs or {}).get("serveurs")
|
|
if hotes_connus and app["hote"] not in hotes_connus:
|
|
raise ValueError(
|
|
f"Application '{nom}': hote '{app['hote']}' absent du registre des "
|
|
f"serveurs (hote fantome). Declare-le dans plan/serveurs.yml, ou "
|
|
f"deplace l'application sur un hote existant.")
|
|
if not est_groupe_operationnel(app["groupe"]):
|
|
raise ValueError(f"Application '{nom}': groupe '{app['groupe']}' non operationnel.")
|
|
requiert = app.get("requiert") or []
|
|
if not isinstance(requiert, list):
|
|
raise ValueError(f"Application '{nom}': 'requiert' doit etre une liste.")
|
|
for dep in requiert:
|
|
if dep not in apps:
|
|
raise ValueError(f"Application '{nom}': requiert '{dep}' inconnue.")
|
|
expose = app.get("expose") or []
|
|
if not isinstance(expose, list):
|
|
raise ValueError(f"Application '{nom}': 'expose' doit etre une liste.")
|
|
for fqdn in expose:
|
|
if doms and domaine_parent(fqdn, doms) is None:
|
|
raise ValueError(f"Application '{nom}': FQDN expose '{fqdn}' sans domaine parent declare.")
|
|
port = app.get("port")
|
|
if port is not None and not str(port).strip().isdigit():
|
|
raise ValueError(f"Application '{nom}': 'port' doit etre un entier.")
|
|
liens = app.get("liens") or []
|
|
if not isinstance(liens, list):
|
|
raise ValueError(f"Application '{nom}': 'liens' doit etre une liste.")
|
|
accepte = None
|
|
for lien in liens:
|
|
if not isinstance(lien, dict):
|
|
raise ValueError(f"Application '{nom}': chaque lien doit etre une table {{vers, role}}.")
|
|
vers = str(lien.get("vers", "")).strip()
|
|
role = str(lien.get("role", "")).strip()
|
|
if not vers or not role:
|
|
raise ValueError(f"Application '{nom}': lien incomplet (champs 'vers' et 'role' requis).")
|
|
if vers == nom:
|
|
raise ValueError(f"Application '{nom}': lien vers elle-meme.")
|
|
if vers not in apps:
|
|
raise ValueError(f"Application '{nom}': cible de lien '{vers}' inconnue.")
|
|
if accepte is None:
|
|
accepte = liens_acceptes(str(app.get("groupe", "")))
|
|
# Un role sans meta/liens.yml n'accepte aucun lien : on ne bloque que
|
|
# s'il en declare (sinon instancier.py tranchera avec le meme message).
|
|
if accepte and role not in accepte:
|
|
raise ValueError(
|
|
f"Application '{nom}': role de lien '{role}' non accepte par "
|
|
f"'{app.get('groupe')}' (roles/{app.get('groupe')}/meta/liens.yml). "
|
|
f"Acceptes : {', '.join(sorted(accepte))}.")
|
|
|
|
|
|
def applications_de_hote(registre: dict, hote: str) -> list[dict]:
|
|
"""Applications instanciees sur un hote donne (plusieurs possibles)."""
|
|
return [
|
|
{"id": nom, **app}
|
|
for nom, app in (registre.get("applications") or {}).items()
|
|
if app.get("hote") == hote
|
|
]
|
|
|
|
|
|
def domaine_parent(fqdn: str, domaines_publics: dict) -> str | None:
|
|
"""Domaine declare le plus long dont `fqdn` releve (apex inclus)."""
|
|
fqdn = str(fqdn).strip()
|
|
candidats = [d for d in domaines_publics if fqdn == d or fqdn.endswith("." + d)]
|
|
return max(candidats, key=len) if candidats else None
|
|
|
|
|
|
def expositions_des_applications(applications: dict, domaines: dict, edge: str | None = None) -> list[dict]:
|
|
"""Expositions publiques derivees des applications (champ 'expose').
|
|
|
|
Pour chaque FQDN qu'une application expose, on resout son domaine parent et
|
|
l'edge de ce domaine. Si `edge` est fourni, ne retient que les expositions
|
|
de cet edge. Le rendu de l'amont (IP interne) reste a la charge de l'appelant.
|
|
"""
|
|
apps = (applications or {}).get("applications") or {}
|
|
doms = (domaines or {}).get("domaines_publics") or {}
|
|
resultat = []
|
|
for app_id, app in apps.items():
|
|
for fqdn in (app.get("expose") or []):
|
|
dom = domaine_parent(fqdn, doms)
|
|
conf = doms.get(dom, {}) if dom else {}
|
|
edge_dom = conf.get("edge")
|
|
if edge is not None and edge_dom != edge:
|
|
continue
|
|
resultat.append({
|
|
"fqdn": fqdn, "application": app_id, "groupe": app.get("groupe"),
|
|
"hote": app.get("hote"), "port": app.get("port"),
|
|
"edge": edge_dom, "domaine": dom,
|
|
# Exposition WebSocket-aware (ex. Collabora) : le vhost ajoute
|
|
# alors l'upgrade Upgrade/Connection. Declare par l'app (websocket: true).
|
|
"websocket": bool(app.get("websocket")),
|
|
})
|
|
return resultat
|
|
|
|
|
|
def chaine_connexion(entree: dict, serveur: dict, secret: str = "***") -> str:
|
|
"""Derive une chaine de connexion (mot de passe masque par defaut)."""
|
|
type_bd = (serveur or {}).get("type", "postgres")
|
|
hote = (serveur or {}).get("hote", "?")
|
|
port = (serveur or {}).get("port", "?")
|
|
return f"{type_bd}://{entree.get('proprietaire', '?')}:{secret}@{hote}:{port}/{entree.get('base', '?')}"
|
|
|
|
|
|
AUTORITES_DNS = {"primaire-cache", "auto-heberge", "delegue"}
|
|
|
|
|
|
def charger_domaines(path: Path | None) -> dict:
|
|
"""Charge le registre des domaines publics (domaines_publics)."""
|
|
if not path or not path.exists():
|
|
return {"domaines_publics": {}}
|
|
with path.open("r", encoding="utf-8") as fichier:
|
|
data = yaml.safe_load(fichier) or {}
|
|
if not isinstance(data, dict):
|
|
raise ValueError(f"{path} ne contient pas une table YAML.")
|
|
data.setdefault("domaines_publics", {})
|
|
return data
|
|
|
|
|
|
def fqdn_exposition(domaine: str, nom: str) -> str:
|
|
"""FQDN publie pour une etiquette ('@' ou le domaine lui-meme = apex)."""
|
|
nom = str(nom).strip()
|
|
if nom in ("@", "", domaine):
|
|
return domaine
|
|
return f"{nom}.{domaine}"
|
|
|
|
|
|
def valider_domaines(registre: dict) -> None:
|
|
"""Valide la coherence du registre des domaines publics."""
|
|
domaines = registre.get("domaines_publics") or {}
|
|
if not isinstance(domaines, dict):
|
|
raise ValueError("domaines_publics doit etre une table.")
|
|
fqdns: set[str] = set()
|
|
for nom_domaine, conf in domaines.items():
|
|
if not isinstance(conf, dict):
|
|
raise ValueError(f"Domaine '{nom_domaine}': table attendue.")
|
|
autorite = str(conf.get("autorite", "")).strip()
|
|
if autorite not in AUTORITES_DNS:
|
|
attendu = ", ".join(sorted(AUTORITES_DNS))
|
|
raise ValueError(f"Domaine '{nom_domaine}': autorite '{autorite}' inconnue (attendu: {attendu}).")
|
|
if not str(conf.get("edge", "")).strip():
|
|
raise ValueError(f"Domaine '{nom_domaine}': champ 'edge' requis.")
|
|
secondaires = conf.get("secondaires") or []
|
|
if not isinstance(secondaires, list):
|
|
raise ValueError(f"Domaine '{nom_domaine}': 'secondaires' doit etre une liste.")
|
|
exposition = conf.get("exposition") or []
|
|
if not isinstance(exposition, list):
|
|
raise ValueError(f"Domaine '{nom_domaine}': 'exposition' doit etre une liste.")
|
|
for entree in exposition:
|
|
if not isinstance(entree, dict):
|
|
raise ValueError(f"Domaine '{nom_domaine}': entree d'exposition invalide (table attendue).")
|
|
for champ in ("nom", "cible"):
|
|
if not str(entree.get(champ, "")).strip():
|
|
raise ValueError(f"Domaine '{nom_domaine}': exposition, champ '{champ}' requis.")
|
|
fqdn = fqdn_exposition(nom_domaine, entree["nom"])
|
|
if fqdn in fqdns:
|
|
raise ValueError(f"Exposition en double: {fqdn}")
|
|
fqdns.add(fqdn)
|
|
|
|
|
|
def expositions_du_groupe(registre: dict, groupe: str) -> list[dict]:
|
|
"""Expositions publiques dont la cible est un groupe interne donne."""
|
|
resultat = []
|
|
for nom_domaine, conf in (registre.get("domaines_publics") or {}).items():
|
|
for entree in (conf.get("exposition") or []):
|
|
if entree.get("cible") == groupe:
|
|
resultat.append({
|
|
"domaine": nom_domaine,
|
|
"fqdn": fqdn_exposition(nom_domaine, entree.get("nom", "")),
|
|
"type": entree.get("type", "web"),
|
|
"edge": conf.get("edge"),
|
|
})
|
|
return resultat
|
|
|
|
|
|
def charger_nomenclature(path: Path | None) -> dict:
|
|
"""Charge le registre de nomenclature (fonctions, categories, adressage)."""
|
|
if not path or not path.exists():
|
|
return {}
|
|
with path.open("r", encoding="utf-8") as fichier:
|
|
data = yaml.safe_load(fichier) or {}
|
|
if not isinstance(data, dict):
|
|
raise ValueError(f"{path} ne contient pas une table YAML.")
|
|
return data
|
|
|
|
|
|
ETATS_SERVEUR = {"actif", "planifie"}
|
|
|
|
|
|
def fonction_seq(nom: str) -> tuple[str, int | None]:
|
|
"""Decompose un hostname 'fonction-NN' en (fonction, NN)."""
|
|
m = re.match(r"^(.+)-(\d+)$", str(nom or ""))
|
|
if m:
|
|
return m.group(1), int(m.group(2))
|
|
return str(nom or ""), None
|
|
|
|
|
|
# --- Derivation de l'ADRESSAGE depuis le seul seed `index` --------------------
|
|
#
|
|
# Principe : TOUT l'adressage (supernet, sous-reseaux, passerelles, VLAN, VMID) se
|
|
# DERIVE de l'index d'instance + du numero de zone. Rien n'est ecrit a la main dans
|
|
# la nomenclature (elle ne garde que le MODELE : libelles de zones + placement des
|
|
# fonctions). Source unique, ci-dessous, partagee par la generation d'inventaire et
|
|
# le devis reseau (scripts/devis_reseau.py).
|
|
#
|
|
# Conventions du modele a 6 zones (reference integral) :
|
|
# - 2e octet du supernet = OCTET_TENANT + index (index 1 -> 10.11.0.0/16)
|
|
# - 3e octet de la zone = OCTET_ZONE + categorie (zone 1 -> .16, dans .16.0/20)
|
|
# - VLAN sur le trunk convergE = SOCLE_VLAN + index x 10 + zone (4 chiffres, unique)
|
|
# - VMID ip-miroir = VLAN(4) . octet-hote(3) . sequence(2) = 9 chiffres mnemotechniques
|
|
OCTET_TENANT = 10
|
|
OCTET_ZONE = 15
|
|
SOCLE_VLAN = 1000
|
|
INDEX_BAC_A_SABLE = 9 # index par defaut d'une nomenclature sans seed (jetable)
|
|
|
|
|
|
def supernet_de(index: int) -> str:
|
|
"""/16 du tenant : 10.(10+index).0.0/16."""
|
|
return f"10.{OCTET_TENANT + int(index)}.0.0/16"
|
|
|
|
|
|
def base3_de(index: int, categorie: int) -> str:
|
|
"""Trois premiers octets d'une zone : 10.(10+index).(15+categorie)."""
|
|
return f"10.{OCTET_TENANT + int(index)}.{OCTET_ZONE + int(categorie)}"
|
|
|
|
|
|
def sous_reseau_de(index: int, categorie: int, cidr: int = 24) -> str:
|
|
"""/24 d'une zone : 10.(10+index).(15+categorie).0/24."""
|
|
return f"{base3_de(index, categorie)}.0/{int(cidr)}"
|
|
|
|
|
|
def passerelle_de(index: int, categorie: int) -> str:
|
|
"""Passerelle d'une zone (SVI) : premier hote du /24."""
|
|
return f"{base3_de(index, categorie)}.1"
|
|
|
|
|
|
def vlan_de(index: int, zone: int) -> int:
|
|
"""VLAN 802.1Q sur le trunk convergE : 1000 + index x 10 + zone (unique globalement)."""
|
|
return SOCLE_VLAN + int(index) * 10 + int(zone)
|
|
|
|
|
|
def deriver_nomenclature(fonction: str, seq: int, nomenclature: dict) -> dict | None:
|
|
"""Derive VMID / VLAN / IP / passerelle / CIDR depuis la fonction et le rang.
|
|
|
|
Source unique de la derivation (miroir Python de l'auto-proposition du GUI).
|
|
Tout vient de l'index + de la zone : aucune lecture d'adressage stocke.
|
|
"""
|
|
fonctions = nomenclature.get("fonctions") or {}
|
|
categories = nomenclature.get("categories") or {}
|
|
f = fonctions.get(fonction)
|
|
if not f or seq is None:
|
|
return None
|
|
cat = f.get("categorie")
|
|
svc = f.get("service")
|
|
c = categories.get(cat) or categories.get(str(cat))
|
|
if not c:
|
|
return None
|
|
index = nomenclature.get("index", INDEX_BAC_A_SABLE)
|
|
hote = svc * 10 + int(seq)
|
|
vlan = vlan_de(index, int(cat))
|
|
return {
|
|
"vmid": f"{vlan}{hote:03d}{int(seq):02d}",
|
|
"vlan": vlan,
|
|
"adresse_ip": f"{base3_de(index, int(cat))}.{hote}",
|
|
"passerelle": passerelle_de(index, int(cat)),
|
|
"cidr": nomenclature.get("cidr_hote", 24),
|
|
}
|
|
|
|
|
|
def charger_serveurs(path: Path | None) -> dict:
|
|
"""Charge le registre des serveurs (VM) du plan."""
|
|
if not path or not path.exists():
|
|
return {"serveurs": {}}
|
|
with path.open("r", encoding="utf-8") as fichier:
|
|
data = yaml.safe_load(fichier) or {}
|
|
if not isinstance(data, dict):
|
|
raise ValueError(f"{path} ne contient pas une table YAML.")
|
|
data.setdefault("serveurs", {})
|
|
return data
|
|
|
|
|
|
def valider_serveurs(registre: dict, nomenclature: dict | None = None) -> None:
|
|
"""Valide la coherence du registre des serveurs."""
|
|
serveurs = registre.get("serveurs") or {}
|
|
fonctions = (nomenclature or {}).get("fonctions") or {}
|
|
if not isinstance(serveurs, dict):
|
|
raise ValueError("serveurs doit etre une table.")
|
|
for nom, srv in serveurs.items():
|
|
if not isinstance(srv, dict):
|
|
raise ValueError(f"Serveur '{nom}': table attendue.")
|
|
fonction = str(srv.get("fonction", "")).strip()
|
|
if not fonction:
|
|
raise ValueError(f"Serveur '{nom}': champ 'fonction' requis.")
|
|
if fonctions and fonction not in fonctions:
|
|
raise ValueError(f"Serveur '{nom}': fonction '{fonction}' inconnue de la nomenclature.")
|
|
etat = str(srv.get("etat", "")).strip()
|
|
if etat and etat not in ETATS_SERVEUR:
|
|
raise ValueError(f"Serveur '{nom}': etat '{etat}' inconnu (attendu: {', '.join(sorted(ETATS_SERVEUR))}).")
|
|
integrations = srv.get("integrations")
|
|
if integrations is not None and not isinstance(integrations, list):
|
|
raise ValueError(f"Serveur '{nom}': 'integrations' doit etre une liste.")
|
|
|
|
|
|
def reconcilier_serveur(nom: str, srv: dict, hote_inventaire: dict | None, nomenclature: dict) -> dict:
|
|
"""Compare les valeurs derivees (fonction+rang) aux valeurs de l'inventaire.
|
|
|
|
Renvoie {derive, inventaire, divergences} ; divergences vide = reconcilie.
|
|
"""
|
|
_, seq = fonction_seq(nom)
|
|
derive = deriver_nomenclature(str(srv.get("fonction", "")), seq, nomenclature) or {}
|
|
inv = hote_inventaire or {}
|
|
champs = {"vmid": "vmid", "adresse_ip": "adresse_ip", "vlan": "vlan", "passerelle": "passerelle"}
|
|
divergences = []
|
|
for cle_derivee, cle_inv in champs.items():
|
|
if cle_derivee not in derive:
|
|
continue
|
|
attendu = str(derive[cle_derivee])
|
|
present = str(inv.get(cle_inv, "")).strip()
|
|
if present and present != attendu:
|
|
divergences.append(f"{cle_derivee}: inventaire={present} != derive={attendu}")
|
|
return {"derive": derive, "inventaire": inv, "divergences": divergences,
|
|
"absent_inventaire": hote_inventaire is None}
|
|
|
|
|
|
# --- Dimensionnement derive des ressources VM -------------------------------
|
|
# ressources(VM) = socle_SE + somme(empreinte des logiciels) -> marge -> arrondi.
|
|
# Les empreintes vivent dans roles/<groupe>/meta/empreinte.yml (propriete du
|
|
# logiciel). Les groupes sans role dedie ont un repli ci-dessous. Le plan peut
|
|
# tout surcharger par hote (serveurs.yml: coeurs/memoire/disque).
|
|
|
|
SOCLE_SE = {"coeurs": 1, "memoire_mo": 512, "disque_go": 8}
|
|
EMPREINTE_DEFAUT = {"coeurs": 1, "memoire_mo": 512, "disque_go": 5}
|
|
EMPREINTES_SANS_ROLE = {
|
|
"serveur_web_frontal": {"coeurs": 1, "memoire_mo": 512, "disque_go": 5},
|
|
"serveur_web_dorsal": {"coeurs": 1, "memoire_mo": 1024, "disque_go": 10},
|
|
}
|
|
MARGE_MEMOIRE = 1.20
|
|
MARGE_DISQUE = 1.25
|
|
PALIER_MEMOIRE_MO = 512
|
|
MIN_MEMOIRE_MO = 1024
|
|
PALIER_DISQUE_GO = 5
|
|
MIN_DISQUE_GO = 12
|
|
MIN_COEURS = 1
|
|
MAX_COEURS = 8
|
|
CHAMPS_EMPREINTE = ("coeurs", "memoire_mo", "disque_go")
|
|
|
|
|
|
def charger_empreinte_role(racine_roles: Path, groupe: str) -> dict | None:
|
|
"""Lit roles/<groupe>/meta/empreinte.yml -> empreinte normalisee, ou None."""
|
|
fichier = racine_roles / groupe / "meta" / "empreinte.yml"
|
|
if not fichier.exists():
|
|
return None
|
|
with fichier.open("r", encoding="utf-8") as fh:
|
|
data = yaml.safe_load(fh) or {}
|
|
emp = (data or {}).get("setops_empreinte") or {}
|
|
if not isinstance(emp, dict):
|
|
raise ValueError(f"{fichier}: 'setops_empreinte' doit etre une table.")
|
|
return {cle: int(emp.get(cle, 0)) for cle in CHAMPS_EMPREINTE}
|
|
|
|
|
|
def empreinte_groupe(racine_roles: Path, groupe: str) -> dict:
|
|
"""Empreinte d'un groupe de service : role > repli connu > defaut."""
|
|
return (charger_empreinte_role(racine_roles, groupe)
|
|
or EMPREINTES_SANS_ROLE.get(groupe)
|
|
or EMPREINTE_DEFAUT)
|
|
|
|
|
|
def _arrondi_superieur(valeur: float, palier: int) -> int:
|
|
import math
|
|
return int(math.ceil(valeur / palier) * palier)
|
|
|
|
|
|
def deriver_ressources(groupes_service, racine_roles: Path, socle: dict | None = None) -> dict:
|
|
"""Estime coeurs/memoire(Mo)/disque(Go) d'une VM depuis ses logiciels + le SE.
|
|
|
|
`groupes_service` : groupes applicatifs portes par l'hote (hors socle/etat).
|
|
RAM et disque additifs (contraintes dures) ; coeurs en max (burstable).
|
|
"""
|
|
socle = socle or SOCLE_SE
|
|
coeurs_somme = 0
|
|
memoire = socle["memoire_mo"]
|
|
disque = socle["disque_go"]
|
|
for groupe in groupes_service:
|
|
emp = empreinte_groupe(racine_roles, groupe)
|
|
coeurs_somme += emp["coeurs"]
|
|
memoire += emp["memoire_mo"]
|
|
disque += emp["disque_go"]
|
|
coeurs = max(MIN_COEURS, min(MAX_COEURS, max(socle["coeurs"], coeurs_somme)))
|
|
memoire = max(MIN_MEMOIRE_MO, _arrondi_superieur(memoire * MARGE_MEMOIRE, PALIER_MEMOIRE_MO))
|
|
disque = max(MIN_DISQUE_GO, _arrondi_superieur(disque * MARGE_DISQUE, PALIER_DISQUE_GO))
|
|
return {"coeurs": coeurs, "memoire": memoire, "disque": f"{disque}G"}
|
|
|
|
|
|
def charger_dependances(path: Path | None) -> dict:
|
|
if not path:
|
|
return {}
|
|
if not path.exists():
|
|
raise ValueError(f"Fichier de dependances introuvable: {path}")
|
|
|
|
with path.open("r", encoding="utf-8") as dependencies_file:
|
|
data = yaml.safe_load(dependencies_file) or {}
|
|
|
|
if not isinstance(data, dict):
|
|
raise ValueError(f"{path} ne contient pas une table YAML.")
|
|
|
|
groups = data.get("groupes", {})
|
|
if not isinstance(groups, dict):
|
|
raise ValueError(f"{path} doit contenir une cle 'groupes' de type table.")
|
|
|
|
for group, config in groups.items():
|
|
if not est_groupe_operationnel(group):
|
|
raise ValueError(f"Groupe de dependance non operationnel: {group}")
|
|
if not isinstance(config, dict):
|
|
raise ValueError(f"Dependance invalide pour {group}: table attendue.")
|
|
required = config.get("requiert_groupes_actifs", [])
|
|
if required is None:
|
|
required = []
|
|
if not isinstance(required, list) or not all(isinstance(item, str) for item in required):
|
|
raise ValueError(f"requiert_groupes_actifs doit etre une liste de groupes pour {group}.")
|
|
for required_group in required:
|
|
if not est_groupe_operationnel(required_group):
|
|
raise ValueError(f"Prerequis non operationnel pour {group}: {required_group}")
|
|
|
|
return groups
|