"""Services transactionnels pour la gestion des rôles RBAC. Les fonctions de ce module ne valident ni ne commitent la transaction appelante. Cela permet aux routes d'enregistrer simultanément l'utilisateur et ses accès. """ from ...extensions import db 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): return if not deleting and (will_be_active is None or will_be_active): return count = UserRole.query.join(Role).join(User, User.id == UserRole.user_id).filter( Role.slug == "super_admin", Role.is_active.is_(True), db.or_(User.is_active.is_(True), User.is_active.is_(None)) ).count() if count <= 1: raise LastSuperAdminError("Impossible de désactiver ou supprimer le dernier super administrateur actif.") def _is_last_super_admin(user, selected_ids): super_role = Role.query.filter_by(slug="super_admin").first() if not super_role or super_role.id not in {int(value) for value in selected_ids}: if super_role and any(link.role_id == super_role.id for link in user.role_links): count = UserRole.query.join(Role).join(User, User.id == UserRole.user_id).filter( Role.slug == "super_admin", Role.is_active.is_(True), db.or_(User.is_active.is_(True), User.is_active.is_(None)) ).count() return count <= 1 return False def replace_user_roles(user, role_ids, assigned_by=None): """Remplace l'ensemble des rôles actifs d'un utilisateur. Les liens déjà sélectionnés sont conservés (et donc leur traçabilité), les liens absents sont retirés et les nouveaux sont ajoutés sans doublon. ``role_ids`` peut être vide. Les rôles inactifs ou inconnus sont ignorés. La fonction travaille dans la transaction courante et ne committe jamais. """ normalized_ids = set() for value in role_ids or (): try: normalized_ids.add(int(value)) except (TypeError, ValueError): continue active_roles = { role.id: role for role in Role.query.filter(Role.id.in_(normalized_ids), Role.is_active.is_(True)).all() } if normalized_ids else {} selected_ids = set(active_roles) super_role = Role.query.filter_by(slug="super_admin").first() if super_role and super_role.id in selected_ids and assigned_by is not None: actor_is_super = any( link.role and link.role.slug == "super_admin" and link.role.is_active for link in getattr(assigned_by, "role_links", []) ) if not actor_is_super: raise LastSuperAdminError("Seul un super administrateur peut attribuer le rôle super_admin.") if _is_last_super_admin(user, selected_ids): raise LastSuperAdminError("Impossible de retirer le dernier super administrateur actif.") existing = {link.role_id: link for link in list(user.role_links)} for role_id, link in existing.items(): if role_id not in selected_ids: db.session.delete(link) for role_id in selected_ids - set(existing): db.session.add(UserRole( user_id=user.id, role_id=role_id, assigned_by_id=getattr(assigned_by, "id", assigned_by), )) return [active_roles[role_id] for role_id in sorted(selected_ids)]