#!/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//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//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//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