160 lines
7 KiB
Python
160 lines
7 KiB
Python
"""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)]
|