87 lines
3.6 KiB
Python
87 lines
3.6 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, UserRole
|
|
from ..models.user import User
|
|
|
|
|
|
class LastSuperAdminError(ValueError):
|
|
"""Opération qui supprimerait le dernier super administrateur actif."""
|
|
|
|
|
|
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)]
|