#!/usr/bin/env python3 """Générateur de plan `User Permission` · RBAC 50 rôles → row-level (Sprint 2). Complément du générateur `../fixtures_gen` (Role + Custom DocPerm) : il couvre la dimension ROW-LEVEL (`scope_donnees`) que les DocPerm ne portent pas seuls. Cible de portage VPS : agent ERPNext Backend (SPEC §7). Ce worker n'écrit JAMAIS sur le VPS — il produit en-repo un PLAN que l'agent applique côté serveur en matérialisant, par utilisateur assigné à un rôle, les `User Permission` du template (champ `user` = sentinelle `__ASSIGN_PER_USER__`). Transforme `rbac_50_roles.json` (contrat validé par `rbac.schema.json`) en : - user_permission_plan.json → 1 entrée/rôle : mécanisme d'enforcement + template - MANIFEST.json → traçabilité, comptes par mécanisme, Companies + rôles `equipe` à confirmer VPS Sous-commandes : build [-o OUT] → écrit user_permission_plan.json / MANIFEST.json validate [-o OUT] → (re)génère en mémoire, valide vs userperm.schema.json + invariants row-level — sort en erreur si un invariant casse. Sortie déterministe (tri stable, aucun horodatage) → diffable + re-générable. Anti-invention (#6) : 100 % des données proviennent du contrat ; aucun utilisateur n'est inventé ; les portées sans mécanisme natif (`equipe`) sont signalées, pas mappées arbitrairement. """ from __future__ import annotations import argparse import json import os import sys _HERE = os.path.dirname(os.path.abspath(__file__)) _RBAC_DIR = os.path.normpath(os.path.join(_HERE, "..")) # rbac/ _DELIVERABLES = os.path.normpath(os.path.join(_RBAC_DIR, "..")) # 05_deliverables_mvp/ sys.path.insert(0, _HERE) # Réutilisation (workflow #5 : zéro duplication) du validateur maison Publiciste. sys.path.insert(0, os.path.join(_DELIVERABLES, "publiciste")) from permlib import builder, frappe # noqa: E402 from lib import validator as maison # type: ignore # noqa: E402 _CONTRACT_PATH = os.path.join(_RBAC_DIR, "rbac_50_roles.json") _SCHEMA_PATH = os.path.join(_HERE, "userperm.schema.json") _DEFAULT_OUT = os.path.join(_HERE, "out") def _eprint(*args) -> None: print(*args, file=sys.stderr) def _load(path: str) -> dict: with open(path, encoding="utf-8") as fh: return json.load(fh) def _write_json(path: str, data) -> None: with open(path, "w", encoding="utf-8") as fh: json.dump(data, fh, ensure_ascii=False, indent=2) fh.write("\n") def _build_bundle() -> dict: return builder.build_plan(_load(_CONTRACT_PATH)) def _validate_bundle(bundle: dict) -> list[str]: """Valide le plan contre userperm.schema.json + invariants row-level RBAC.""" schema = _load(_SCHEMA_PATH) errors = list(maison.validate(bundle, schema)) manifest = bundle["manifest"] plan = bundle["plan"] contract = _load(_CONTRACT_PATH) scope_by_role = {r["erpnext_role_name"]: r["scope_donnees"] for r in contract["roles"]} entite_by_role = {r["erpnext_role_name"]: r["entite_principale"] for r in contract["roles"]} # Invariant 1 · cible non-négociable « RBAC 50 rôles » : 1 entrée par rôle. if manifest["cible_rbac_roles"] != len(plan): errors.append( f"cible_rbac_roles={manifest['cible_rbac_roles']} ≠ plan entries={len(plan)}" ) # Invariant 2 · couverture exhaustive & unique des 50 rôles du contrat. plan_roles = [e["erpnext_role_name"] for e in plan] if len(plan_roles) != len(set(plan_roles)): errors.append("Rôle dupliqué dans le plan.") if set(plan_roles) != set(scope_by_role): missing = set(scope_by_role) - set(plan_roles) extra = set(plan_roles) - set(scope_by_role) errors.append(f"Plan ≠ contrat (manquants={sorted(missing)}, en trop={sorted(extra)}).") seen_companies: set[str] = set() for e in plan: role = e["erpnext_role_name"] scope = e["scope_donnees"] tmpl = e["user_permission_template"] # Invariant 3 · fidélité au contrat (scope + entité, aucune dérive). if scope != scope_by_role.get(role): errors.append(f"{role}: scope {scope!r} ≠ contrat {scope_by_role.get(role)!r}.") if e["entite_principale"] != entite_by_role.get(role): errors.append(f"{role}: entite_principale divergente du contrat.") # Invariant 4 · mécanisme déterministe = mapping natif du scope. expected_mech = frappe.SCOPE_MECHANISM.get(scope) if e["mechanism"] != expected_mech: errors.append(f"{role}: mechanism {e['mechanism']!r} ≠ attendu {expected_mech!r}.") # Invariant 5 · un template existe SSI portée `entite` (aucune # restriction fabriquée pour own/groupe/equipe). if scope == "entite": if tmpl is None: errors.append(f"{role}: portée `entite` sans template User Permission.") else: # Invariant 6 · le template est fidèle (Company + entité du contrat). if tmpl.get("allow") != frappe.ALLOW_COMPANY: errors.append(f"{role}: template allow ≠ Company.") if tmpl.get("for_value") != e["entite_principale"]: errors.append(f"{role}: for_value ≠ entite_principale du contrat.") if tmpl.get("user") != frappe.USER_PLACEHOLDER: errors.append(f"{role}: user template n'est pas la sentinelle (invention ?).") if tmpl.get("apply_to_all_doctypes") != 1: errors.append(f"{role}: apply_to_all_doctypes ≠ 1.") if tmpl["for_value"] in frappe.NON_COMPANY_ENTITES: errors.append(f"{role}: for_value consolidé ({tmpl['for_value']}) ne peut être une Company.") seen_companies.add(tmpl["for_value"]) elif tmpl is not None: errors.append(f"{role}: portée {scope!r} ne doit porter aucun template (a un template).") # Invariant 7 · le manifeste liste EXACTEMENT les Companies des templates. if set(manifest["companies_a_confirmer"]) != seen_companies: errors.append("companies_a_confirmer ≠ Companies réellement ciblées par les templates.") # Invariant 8 · les rôles `equipe` du contrat sont TOUS signalés (jamais mappés en douce). equipe_contract = sorted(r for r, s in scope_by_role.items() if s == "equipe") if manifest["roles_scope_equipe_a_confirmer"] != equipe_contract: errors.append("roles_scope_equipe_a_confirmer ≠ rôles `equipe` du contrat.") # Invariant 9 · cohérence des comptages du manifeste. counts = manifest["counts"] if counts["plan_entries"] != len(plan): errors.append("counts.plan_entries incohérent.") tmpl_count = sum(1 for e in plan if e["user_permission_template"] is not None) if counts["user_permission_templates"] != tmpl_count: errors.append("counts.user_permission_templates incohérent avec le plan.") if sum(counts["by_mechanism"].values()) != len(plan): errors.append("counts.by_mechanism ne somme pas à 50.") return errors def cmd_build(args: argparse.Namespace) -> int: bundle = _build_bundle() errors = _validate_bundle(bundle) if errors: _eprint("❌ Plan invalide — génération refusée (anti-régression) :") for e in errors: _eprint(f" - {e}") return 1 out = os.path.abspath(args.out) os.makedirs(out, exist_ok=True) _write_json(os.path.join(out, "user_permission_plan.json"), bundle["plan"]) _write_json(os.path.join(out, "MANIFEST.json"), bundle["manifest"]) m = bundle["manifest"] bm = m["counts"]["by_mechanism"] print(f"✅ Plan User Permission généré dans {out}") print(f" user_permission_plan.json : {m['counts']['plan_entries']} entrées (1/rôle)") print( f" templates Company : {m['counts']['user_permission_templates']} " f"(portée `entite`)" ) print( " mécanismes : " f"entite={bm['user_permission_company']} · " f"consolidé={bm['none_consolidated']} · " f"own={bm['docperm_if_owner']} · " f"equipe(à confirmer)={bm['vps_confirm_team']}" ) if m["companies_a_confirmer"]: print( " ⚠ Companies à créer/confirmer VPS avant application : " + ", ".join(m["companies_a_confirmer"]) ) if m["roles_scope_equipe_a_confirmer"]: print( " ⚠ Rôles `equipe` sans dimension row-level native — mécanisme à " "décider VPS : " + ", ".join(m["roles_scope_equipe_a_confirmer"]) ) return 0 def cmd_validate(args: argparse.Namespace) -> int: bundle = _build_bundle() errors = _validate_bundle(bundle) if errors: _eprint("❌ Validation KO :") for e in errors: _eprint(f" - {e}") return 1 m = bundle["manifest"] print( f"✅ Validation OK — {m['counts']['plan_entries']} entrées, " f"{m['counts']['user_permission_templates']} templates Company, " f"schéma + invariants row-level verts." ) return 0 def main(argv: list[str] | None = None) -> int: p = argparse.ArgumentParser( description="Générateur de plan User Permission depuis le contrat RBAC." ) sub = p.add_subparsers(dest="cmd", required=True) pb = sub.add_parser("build", help="génère user_permission_plan.json / MANIFEST.json") pb.add_argument("-o", "--out", default=_DEFAULT_OUT, help="dossier de sortie (défaut: ./out)") pb.set_defaults(func=cmd_build) pv = sub.add_parser("validate", help="valide le plan (schéma + invariants) sans écrire") pv.add_argument("-o", "--out", default=_DEFAULT_OUT, help="ignoré (compat)") pv.set_defaults(func=cmd_validate) args = p.parse_args(argv) return args.func(args) if __name__ == "__main__": raise SystemExit(main())