From d58b5aed0b7346ec36b25514455112849b0ebb74 Mon Sep 17 00:00:00 2001 From: root Date: Fri, 21 Aug 2026 22:05:04 +0000 Subject: [PATCH] Initialiser le catalogue RBAC au setup --- app_new/core/routes/setup_wizard.py | 29 +++-------- app_new/core/services/rbac.py | 75 ++++++++++++++++++++++++++++- tests/integration/test_rbac.py | 35 +++++++++++++- 3 files changed, 116 insertions(+), 23 deletions(-) diff --git a/app_new/core/routes/setup_wizard.py b/app_new/core/routes/setup_wizard.py index 9ab4bbb..8ebed29 100644 --- a/app_new/core/routes/setup_wizard.py +++ b/app_new/core/routes/setup_wizard.py @@ -14,8 +14,8 @@ from ..models.user import User from ..models.college import College, Building, Zone, Room from ..models.equipment import EquipmentCategory from ..models.maintenance import Lot -from ..models.rbac import Role, Permission, RolePermission, UserRole -from ..services.rbac import replace_user_roles +from ..models.rbac import Role +from ..services.rbac import ensure_rbac_catalog, replace_user_roles from ..authorization import has_permission from ..models.planning import WorkSchedule, CollegeClosure, ClosureWorkDay from ..setup_catalog import EQUIPMENT_CATEGORIES, LOTS, grouped_catalog @@ -271,25 +271,12 @@ def api_save_step(step): db.session.add(user) db.session.flush() - # Le premier compte est le compte de récupération de l'installation - # et doit recevoir explicitement le rôle protégé super_admin. - super_role = Role.query.filter_by(slug='super_admin', is_active=True).first() - if not super_role: - super_role = Role(slug='super_admin', name='Super administrateur', - description="Rôle système protégé donnant un accès complet à l'application.", - is_system=True, is_active=True) - db.session.add(super_role) - db.session.flush() - wildcard = Permission.query.filter_by(code='*').first() - if not wildcard: - wildcard = Permission(code='*', name="Accès complet à l'application", - module='system', action='all', - description="Accès complet réservé au super administrateur.", - is_active=True) - db.session.add(wildcard) - db.session.flush() - if not RolePermission.query.filter_by(role_id=super_role.id, permission_id=wildcard.id).first(): - db.session.add(RolePermission(role_id=super_role.id, permission_id=wildcard.id, effect='allow')) + # Le premier compte est le compte de récupération de l'installation. + # Le bootstrap crée aussi tout le catalogue RBAC (rôles système, + # permissions et associations), afin qu'une installation neuve + # soit immédiatement administrable après le wizard. + role_map, _permission_map = ensure_rbac_catalog() + super_role = role_map['super_admin'] replace_user_roles(user, [super_role.id]) progress.admin_username = username diff --git a/app_new/core/services/rbac.py b/app_new/core/services/rbac.py index 53f5e8d..f094b14 100644 --- a/app_new/core/services/rbac.py +++ b/app_new/core/services/rbac.py @@ -5,14 +5,87 @@ Cela permet aux routes d'enregistrer simultanément l'utilisateur et ses accès. """ from ...extensions import db -from ..models.rbac import Role, UserRole +from ..models.rbac import Role, Permission, UserRole, RolePermission from ..models.user import User +from ..authorization import PERMISSIONS_BY_ROLE, ROLE_LABELS, GRANULAR_PERMISSION_CODES, permission_metadata class LastSuperAdminError(ValueError): """Opération qui supprimerait le dernier super administrateur actif.""" +def ensure_rbac_catalog(): + """Initialise idempotemment le catalogue RBAC pour une installation neuve. + + Cette fonction ne supprime aucun rôle, permission ou association existante. + Elle complète uniquement les rôles système et les permissions manquantes. + La transaction appelante reste responsable du commit. + """ + role_map = {} + for slug, codes in PERMISSIONS_BY_ROLE.items(): + role = Role.query.filter_by(slug=slug).first() + if not role: + role = Role( + slug=slug, + name=ROLE_LABELS.get(slug, slug), + description={ + 'super_admin': "Rôle système protégé donnant un accès complet à l'application.", + 'admin': "Rôle système destiné à l'administration courante de la GMAO. Ses permissions sont configurables.", + 'responsable_gmao': "Rôle système destiné à la supervision et à la gestion générale de la GMAO.", + 'technicien': "Rôle système destiné aux opérations de maintenance.", + 'assistant_prevention': "Rôle système donnant accès aux fonctions liées à la prévention.", + 'demandeur': "Rôle système destiné aux utilisateurs pouvant créer et suivre des demandes.", + 'lecture': "Rôle système destiné aux utilisateurs autorisés uniquement à consulter certaines informations.", + }.get(slug, 'Rôle système fourni par l’application.'), + is_system=True, + is_active=True, + ) + db.session.add(role) + db.session.flush() + else: + role.is_system = True + role.is_active = True + if not role.name or role.name in {slug, 'Administrateur système'}: + role.name = ROLE_LABELS.get(slug, role.name or slug) + role_map[slug] = role + + codes = set(GRANULAR_PERMISSION_CODES) + codes.update(code for values in PERMISSIONS_BY_ROLE.values() for code in values) + codes.add('*') + permission_map = {} + for code in sorted(codes): + permission = Permission.query.filter_by(code=code).first() + if not permission: + module, _, action = code.partition('.') + name, description = permission_metadata(code) + permission = Permission( + code=code, + name=name, + module=module or 'system', + action=action or 'all', + description=description, + is_active=True, + ) + db.session.add(permission) + db.session.flush() + else: + permission.is_active = True + name, description = permission_metadata(code) + permission.name = name + permission.description = description + permission_map[code] = permission + + for slug, codes_for_role in PERMISSIONS_BY_ROLE.items(): + role = role_map[slug] + for code in codes_for_role: + permission = permission_map[code] + link = RolePermission.query.filter_by(role_id=role.id, permission_id=permission.id).first() + if not link: + db.session.add(RolePermission(role_id=role.id, permission_id=permission.id, effect='allow')) + + return role_map, permission_map + + def ensure_super_admin_survives(user, *, will_be_active=None, deleting=False): """Bloque la désactivation/suppression du dernier super-admin actif.""" if not any(link.role and link.role.slug == "super_admin" and link.role.is_active for link in user.role_links): diff --git a/tests/integration/test_rbac.py b/tests/integration/test_rbac.py index 713ab72..b014533 100644 --- a/tests/integration/test_rbac.py +++ b/tests/integration/test_rbac.py @@ -3,10 +3,43 @@ import pytest from app_new.extensions import db from app_new.core.models import User, Role, Permission, RolePermission, UserPermission, UserRole from app_new.core.authorization import has_permission -from app_new.core.services.rbac import replace_user_roles +from app_new.core.services.rbac import replace_user_roles, ensure_rbac_catalog from app_new.core.services.rbac import ensure_super_admin_survives, LastSuperAdminError +def test_rbac_catalog_bootstrap_is_complete_and_idempotent(app): + """Le wizard doit pouvoir amorcer une base RBAC vierge sans doublons.""" + with app.app_context(): + role_map, permission_map = ensure_rbac_catalog() + db.session.commit() + + expected_roles = { + 'super_admin', 'admin', 'responsable_gmao', 'technicien', + 'assistant_prevention', 'demandeur', 'lecture', + } + assert expected_roles.issubset(role_map) + assert '*' in permission_map + assert RolePermission.query.filter_by( + role_id=role_map['super_admin'].id, + permission_id=permission_map['*'].id, + ).first() is not None + assert RolePermission.query.filter_by( + role_id=role_map['admin'].id, + permission_id=permission_map['*'].id, + ).first() is None + + counts_before = { + 'roles': Role.query.filter(Role.slug.in_(expected_roles)).count(), + 'permissions': Permission.query.count(), + 'links': RolePermission.query.count(), + } + ensure_rbac_catalog() + db.session.commit() + assert Role.query.filter(Role.slug.in_(expected_roles)).count() == counts_before['roles'] + assert Permission.query.count() == counts_before['permissions'] + assert RolePermission.query.count() == counts_before['links'] + + def _role_with_permissions(slug, permissions): role = Role(slug=slug, name=slug, is_system=False, is_active=True) db.session.add(role)