diff --git a/app_new/core/authorization.py b/app_new/core/authorization.py index 2ade088..40c7162 100644 --- a/app_new/core/authorization.py +++ b/app_new/core/authorization.py @@ -77,6 +77,14 @@ STOCK_BLUEPRINTS = {"parts", "meters", "cleaning"} CONTRACT_BLUEPRINTS = {"companies", "contracts", "services"} PREVENTION_BLUEPRINTS = {"trainings", "constraints", "prevention"} +# Catalogue affiché dans la matrice RBAC. Les anciennes permissions +# ``*.manage`` restent compatibles et impliquent les actions ci-dessous. +GRANULAR_PERMISSION_CODES = { + f"{module}.{action}" + for module in ("dashboard", "patrimoine", "intervention", "planning", "stock", "contract", "prevention", "user", "system") + for action in ("view", "create", "edit", "delete", "validate", "reject", "export", "configure", "manage") +} + def canonical_role(role): return ROLE_ALIASES.get((role or "").strip().lower(), "lecture") @@ -86,8 +94,38 @@ def has_permission(permission, user=None): user = user or current_user if not getattr(user, "is_authenticated", False): return False + # Les rôles configurables sont prioritaires lorsqu'ils existent. Le + # fallback statique conserve la compatibilité avec les anciennes bases + # avant application de la migration RBAC. + try: + from .models.rbac import Role, RolePermission, Permission, UserPermission + from datetime import datetime, timezone + roles = [link.role for link in user.role_links if link.role and link.role.is_active] + legacy_role = Role.query.filter_by(slug=user.role, is_active=True).first() + if legacy_role and legacy_role not in roles: + roles.append(legacy_role) + if roles: + granted = {link.permission.code for role in roles for link in role.permission_links + if link.effect == "allow" and link.permission.is_active} + denied = set() + for override in user.permission_links: + if override.expires_at and override.expires_at < datetime.now(timezone.utc): + continue + (granted if override.effect == "allow" else denied).add(override.permission.code) + if permission in denied: + return False + if permission in granted or "*" in granted: + return True + module = permission.rsplit(".", 1)[0] if "." in permission else permission + return f"{module}.manage" in granted and f"{module}.manage" not in denied + except Exception: + # Une base non migrée ne doit pas rendre toutes les routes indisponibles. + pass permissions = PERMISSIONS_BY_ROLE.get(canonical_role(user.role), set()) - return "*" in permissions or permission in permissions + if "*" in permissions or permission in permissions: + return True + module = permission.rsplit(".", 1)[0] if "." in permission else permission + return f"{module}.manage" in permissions def permission_required(permission): @@ -108,6 +146,16 @@ def required_permission(endpoint, method): blueprint = endpoint.split(".", 1)[0] mutating = method not in {"GET", "HEAD", "OPTIONS"} + if blueprint == "cleaning": + action = "view" + if mutating: + action = "create" if any(token in endpoint for token in ("new", "receive", "issue", "transfer", "transvasement", "inventory")) else "edit" + if "delete" in endpoint: + action = "delete" + if "validate" in endpoint: + action = "validate" + return f"stock.{action}" + if blueprint in ADMIN_BLUEPRINTS: return "system.admin" if blueprint == "auth": diff --git a/app_new/core/models/__init__.py b/app_new/core/models/__init__.py index b18f5b4..7396c75 100644 --- a/app_new/core/models/__init__.py +++ b/app_new/core/models/__init__.py @@ -26,6 +26,7 @@ from .cleaning import ( RequestProfile, RequestProfileItem, StockInventory, StockInventoryLine, CleaningForecastConfig, ) +from .rbac import Role, Permission, UserRole, RolePermission, UserPermission __all__ = [ 'User', 'Staff', @@ -45,4 +46,5 @@ __all__ = [ 'MaterialAssignment', 'ReusableContainer', 'ContainerFill', 'ProductDocument', 'RequestProfile', 'RequestProfileItem', 'StockInventory', 'StockInventoryLine', 'CleaningForecastConfig', + 'Role', 'Permission', 'UserRole', 'RolePermission', 'UserPermission', ] diff --git a/app_new/core/models/rbac.py b/app_new/core/models/rbac.py new file mode 100644 index 0000000..4eda5b4 --- /dev/null +++ b/app_new/core/models/rbac.py @@ -0,0 +1,66 @@ +"""RBAC configurable : rôles, permissions et exceptions utilisateur.""" +from datetime import datetime, timezone +from ...extensions import db + +def utcnow(): + return datetime.now(timezone.utc) + +class Role(db.Model): + __tablename__ = "roles" + id = db.Column(db.Integer, primary_key=True) + slug = db.Column(db.String(80), nullable=False, unique=True) + name = db.Column(db.String(120), nullable=False) + description = db.Column(db.Text) + is_system = db.Column(db.Boolean, nullable=False, default=False) + is_active = db.Column(db.Boolean, nullable=False, default=True) + created_at = db.Column(db.DateTime, nullable=False, default=utcnow) + updated_at = db.Column(db.DateTime, nullable=False, default=utcnow, onupdate=utcnow) + user_links = db.relationship("UserRole", back_populates="role", cascade="all, delete-orphan") + permission_links = db.relationship("RolePermission", back_populates="role", cascade="all, delete-orphan") + +class Permission(db.Model): + __tablename__ = "permissions" + id = db.Column(db.Integer, primary_key=True) + code = db.Column(db.String(120), nullable=False, unique=True) + name = db.Column(db.String(160), nullable=False) + module = db.Column(db.String(80), nullable=False, index=True) + action = db.Column(db.String(40), nullable=False) + description = db.Column(db.Text) + is_active = db.Column(db.Boolean, nullable=False, default=True) + role_links = db.relationship("RolePermission", back_populates="permission", cascade="all, delete-orphan") + user_links = db.relationship("UserPermission", back_populates="permission", cascade="all, delete-orphan") + +class UserRole(db.Model): + __tablename__ = "user_roles" + __table_args__ = (db.UniqueConstraint("user_id", "role_id", name="uq_user_role"),) + id = db.Column(db.Integer, primary_key=True) + user_id = db.Column(db.Integer, db.ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True) + role_id = db.Column(db.Integer, db.ForeignKey("roles.id", ondelete="CASCADE"), nullable=False, index=True) + assigned_at = db.Column(db.DateTime, nullable=False, default=utcnow) + assigned_by_id = db.Column(db.Integer, db.ForeignKey("users.id", ondelete="SET NULL")) + user = db.relationship("User", foreign_keys=[user_id], back_populates="role_links") + role = db.relationship("Role", back_populates="user_links") + assigned_by = db.relationship("User", foreign_keys=[assigned_by_id]) + +class RolePermission(db.Model): + __tablename__ = "role_permissions" + __table_args__ = (db.UniqueConstraint("role_id", "permission_id", name="uq_role_permission"),) + id = db.Column(db.Integer, primary_key=True) + role_id = db.Column(db.Integer, db.ForeignKey("roles.id", ondelete="CASCADE"), nullable=False, index=True) + permission_id = db.Column(db.Integer, db.ForeignKey("permissions.id", ondelete="CASCADE"), nullable=False, index=True) + effect = db.Column(db.String(10), nullable=False, default="allow") + role = db.relationship("Role", back_populates="permission_links") + permission = db.relationship("Permission", back_populates="role_links") + +class UserPermission(db.Model): + __tablename__ = "user_permissions" + __table_args__ = (db.UniqueConstraint("user_id", "permission_id", name="uq_user_permission"),) + id = db.Column(db.Integer, primary_key=True) + user_id = db.Column(db.Integer, db.ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True) + permission_id = db.Column(db.Integer, db.ForeignKey("permissions.id", ondelete="CASCADE"), nullable=False, index=True) + effect = db.Column(db.String(10), nullable=False, default="allow") + expires_at = db.Column(db.DateTime) + note = db.Column(db.String(255)) + user = db.relationship("User", back_populates="permission_links") + permission = db.relationship("Permission", back_populates="user_links") + diff --git a/app_new/core/models/user.py b/app_new/core/models/user.py index ff0b937..81e7404 100644 --- a/app_new/core/models/user.py +++ b/app_new/core/models/user.py @@ -22,6 +22,8 @@ class User(UserMixin, db.Model): is_active = db.Column(db.Boolean, default=True) created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc)) last_login = db.Column(db.DateTime, nullable=True) + role_links = db.relationship("UserRole", foreign_keys="UserRole.user_id", back_populates="user", cascade="all, delete-orphan") + permission_links = db.relationship("UserPermission", foreign_keys="UserPermission.user_id", back_populates="user", cascade="all, delete-orphan") def set_password(self, password): """Hash et stocke le mot de passe.""" diff --git a/app_new/core/routes/admin.py b/app_new/core/routes/admin.py index 66fdeda..4b03923 100644 --- a/app_new/core/routes/admin.py +++ b/app_new/core/routes/admin.py @@ -12,6 +12,7 @@ from ...extensions import db from ..models.user import User, Staff from ..models.college import College, Building, Room, RoomType from ..models.company import Company, Service +from ..models.rbac import Role, Permission, UserRole, RolePermission, UserPermission from app_new.constants import INTERVENTION_STATUSES, EQUIPMENT_STATUSES, ROLES from ..authorization import PERMISSIONS_BY_ROLE, ROLE_LABELS, canonical_role @@ -54,7 +55,7 @@ def users(): """Liste des utilisateurs.""" page = request.args.get('page', 1, type=int) users = User.query.order_by(User.created_at.desc()).paginate(page=page, per_page=20) - return render_template('admin/users.html', users=users) + return render_template('admin/users.html', users=users, rbac_roles=Role.query.filter_by(is_active=True).order_by(Role.name).all()) @admin_bp.route('/users/new', methods=['GET', 'POST']) @@ -68,6 +69,7 @@ def user_new(): password = request.form.get('password') or '' password_confirm = request.form.get('password_confirm') or '' role = request.form.get('role', 'demandeur') + valid_roles = set(ROLES) | {r.slug for r in Role.query.filter_by(is_active=True).all()} if len(username) < 3: flash('Le nom d’utilisateur doit contenir au moins 3 caractères.', 'danger') @@ -77,7 +79,7 @@ def user_new(): flash('Le mot de passe doit contenir au moins 12 caractères.', 'danger') elif password != password_confirm: flash('Les mots de passe ne correspondent pas.', 'danger') - elif role not in ROLES: + elif role not in valid_roles: flash('Rôle utilisateur invalide.', 'danger') elif User.query.filter_by(username=username).first(): flash('Ce nom d\'utilisateur existe déjà.', 'danger') @@ -87,11 +89,16 @@ def user_new(): user = User(username=username, email=email, role=role) user.set_password(password) db.session.add(user) + db.session.flush() + selected_role = Role.query.filter_by(slug=role, is_active=True).first() + if selected_role: + db.session.add(UserRole(user_id=user.id, role_id=selected_role.id, assigned_by_id=current_user.id)) db.session.commit() flash('Utilisateur créé avec succès.', 'success') return redirect(url_for('admin.users')) - return render_template('admin/user_form.html', roles=ROLES) + roles = {**ROLES, **{r.slug: r.name for r in Role.query.filter_by(is_active=True).all()}} + return render_template('admin/user_form.html', roles=roles) @admin_bp.route('/users//edit', methods=['GET', 'POST']) @@ -103,7 +110,8 @@ def user_edit(user_id): if request.method == 'POST': role = request.form.get('role') - if role not in ROLES: + valid_roles = set(ROLES) | {r.slug for r in Role.query.filter_by(is_active=True).all()} + if role not in valid_roles: flash('Rôle utilisateur invalide.', 'danger') return render_template('admin/user_form.html', user=user, roles=ROLES), 400 user.email = request.form.get('email') @@ -117,11 +125,15 @@ def user_edit(user_id): return render_template('admin/user_form.html', user=user), 400 user.set_password(new_password) + selected_role = Role.query.filter_by(slug=role, is_active=True).first() + if selected_role and not UserRole.query.filter_by(user_id=user.id, role_id=selected_role.id).first(): + db.session.add(UserRole(user_id=user.id, role_id=selected_role.id, assigned_by_id=current_user.id)) db.session.commit() flash('Utilisateur modifié avec succès.', 'success') return redirect(url_for('admin.users')) - return render_template('admin/user_form.html', user=user, roles=ROLES) + roles = {**ROLES, **{r.slug: r.name for r in Role.query.filter_by(is_active=True).all()}} + return render_template('admin/user_form.html', user=user, roles=roles) @admin_bp.route('/permissions') @@ -139,6 +151,89 @@ def permissions(): ) +@admin_bp.route('/roles', methods=['GET', 'POST']) +@login_required +@admin_required +def roles(): + """Gérer les rôles configurables sans supprimer les rôles historiques.""" + if request.method == 'POST': + name = (request.form.get('name') or '').strip() + slug = (request.form.get('slug') or '').strip().lower().replace(' ', '_') + if not name or len(slug) < 2: + flash('Le nom et le code du rôle sont obligatoires.', 'danger') + elif Role.query.filter_by(slug=slug).first(): + flash('Ce code de rôle existe déjà.', 'danger') + else: + db.session.add(Role(slug=slug, name=name, description=request.form.get('description'), is_system=False)) + db.session.commit() + flash('Rôle créé. Configurez maintenant ses permissions.', 'success') + return redirect(url_for('admin.role_permissions', role_id=Role.query.filter_by(slug=slug).first().id)) + return render_template('admin/roles.html', roles_list=Role.query.order_by(Role.is_system.desc(), Role.name).all()) + + +@admin_bp.route('/roles//permissions', methods=['GET', 'POST']) +@login_required +@admin_required +def role_permissions(role_id): + role = Role.query.get_or_404(role_id) + permissions_list = Permission.query.filter_by(is_active=True).order_by(Permission.module, Permission.action, Permission.code).all() + if request.method == 'POST': + selected = {int(value) for value in request.form.getlist('permission_ids')} + RolePermission.query.filter_by(role_id=role.id).delete(synchronize_session=False) + for permission in permissions_list: + if permission.id in selected: + db.session.add(RolePermission(role_id=role.id, permission_id=permission.id, effect='allow')) + db.session.commit() + flash(f'Permissions du rôle « {role.name} » enregistrées.', 'success') + return redirect(url_for('admin.roles')) + selected = {link.permission_id for link in role.permission_links if link.effect == 'allow'} + return render_template('admin/role_permissions.html', role=role, permissions_list=permissions_list, selected=selected) + + +@admin_bp.route('/roles//archive', methods=['POST']) +@login_required +@admin_required +def role_archive(role_id): + role = Role.query.get_or_404(role_id) + if role.is_system: + flash('Les rôles système ne peuvent pas être supprimés.', 'danger') + else: + role.is_active = False + db.session.commit() + flash(f'Rôle « {role.name} » archivé. Son historique est conservé.', 'success') + return redirect(url_for('admin.roles')) + + +@admin_bp.route('/users//access', methods=['GET', 'POST']) +@login_required +@admin_required +def user_access(user_id): + user = User.query.get_or_404(user_id) + roles_list = Role.query.filter_by(is_active=True).order_by(Role.name).all() + permissions_list = Permission.query.filter_by(is_active=True).order_by(Permission.module, Permission.action, Permission.code).all() + if request.method == 'POST': + selected_roles = {int(value) for value in request.form.getlist('role_ids')} + UserRole.query.filter_by(user_id=user.id).delete(synchronize_session=False) + for role in roles_list: + if role.id in selected_roles: + db.session.add(UserRole(user_id=user.id, role_id=role.id, assigned_by_id=current_user.id)) + # Les exceptions explicites sont optionnelles et tracées ; une case + # « refuser » est prioritaire sur les rôles. + UserPermission.query.filter_by(user_id=user.id).delete(synchronize_session=False) + for permission in permissions_list: + effect = request.form.get(f'permission_{permission.id}') + if effect in ('allow', 'deny'): + db.session.add(UserPermission(user_id=user.id, permission_id=permission.id, effect=effect, note='Exception administrateur')) + if selected_roles: + user.role = next((r.slug for r in roles_list if r.id in selected_roles), user.role) + db.session.commit() + flash(f'Accès de {user.username} enregistrés.', 'success') + return redirect(url_for('admin.users')) + selected_roles = {link.role_id for link in user.role_links} + overrides = {link.permission_id: link.effect for link in user.permission_links} + return render_template('admin/user_access.html', user=user, roles_list=roles_list, permissions_list=permissions_list, selected_roles=selected_roles, overrides=overrides) + + @admin_bp.route('/audit') @login_required @admin_required diff --git a/app_new/templates/admin/role_permissions.html b/app_new/templates/admin/role_permissions.html new file mode 100644 index 0000000..a4b822b --- /dev/null +++ b/app_new/templates/admin/role_permissions.html @@ -0,0 +1 @@ +{% extends "base.html" %}{% block title %}Permissions — {{ role.name }}{% endblock %}{% block content %}

Permissions du rôle « {{ role.name }} »

Les cases cochées autorisent l'action. Les permissions sont regroupées par module.

{% set ns = namespace(module='') %}{% for permission in permissions_list %}{% if permission.module != ns.module %}{% if not loop.first %}
{% endif %}{% set ns.module = permission.module %}
{{ permission.module|title }}
{% endif %}
{% if loop.last %}
{% endif %}{% else %}

Aucune permission. Exécutez la migration RBAC.

{% endfor %}Annuler{% endblock %} diff --git a/app_new/templates/admin/roles.html b/app_new/templates/admin/roles.html new file mode 100644 index 0000000..b9bdef5 --- /dev/null +++ b/app_new/templates/admin/roles.html @@ -0,0 +1 @@ +{% extends "base.html" %}{% block title %}Rôles configurables{% endblock %}{% block content %}

Rôles et permissions

Vue historique
Créer un rôle
{% for role in roles_list %}{% else %}{% endfor %}
RôleCodeTypeUtilisateurs
{{ role.name }}
{{ role.description or '' }}		{{ role.slug }}{{ 'Système' if role.is_system else 'Personnalisé' }}{{ role.user_links|length }}Configurer les droits{% if not role.is_system and role.is_active %}
{% endif %}
Aucun rôle RBAC migré.
{% endblock %} diff --git a/app_new/templates/admin/user_access.html b/app_new/templates/admin/user_access.html new file mode 100644 index 0000000..1c466b8 --- /dev/null +++ b/app_new/templates/admin/user_access.html @@ -0,0 +1 @@ +{% extends "base.html" %}{% block title %}Accès — {{ user.username }}{% endblock %}{% block content %}

Accès de {{ user.full_name or user.username }}

Rôles
{% for role in roles_list %}
{% else %}

Aucun rôle RBAC disponible.

{% endfor %}
Exceptions personnellesAccorder ou refuser ici un droit en dehors des rôles. Un refus est prioritaire.
{% for permission in permissions_list %}
{% else %}

Aucune permission RBAC disponible.

{% endfor %}
Annuler
{% endblock %} diff --git a/app_new/templates/admin/users.html b/app_new/templates/admin/users.html index 4a4f2d5..52db313 100644 --- a/app_new/templates/admin/users.html +++ b/app_new/templates/admin/users.html @@ -37,6 +37,7 @@ + Accès {% endfor %} @@ -49,4 +50,4 @@ -{% endblock %} \ No newline at end of file +{% endblock %} diff --git a/app_new/templates/base.html b/app_new/templates/base.html index a04e673..297dc04 100644 --- a/app_new/templates/base.html +++ b/app_new/templates/base.html @@ -276,6 +276,7 @@
  • Chatbot
  • Utilisateurs
    • +
  • Rôles et permissions
  • Gestion des données
  • Audit des templates
  • Gestion bases
    • diff --git a/migrations/versions/9b06c7d8e9f0_ajouter_rbac_configurable.py b/migrations/versions/9b06c7d8e9f0_ajouter_rbac_configurable.py new file mode 100644 index 0000000..c1e90b3 --- /dev/null +++ b/migrations/versions/9b06c7d8e9f0_ajouter_rbac_configurable.py @@ -0,0 +1,40 @@ +"""Ajouter le RBAC configurable (rôles, permissions et exceptions). +Revision ID: 9b06c7d8e9f0 +Revises: 8a95b1c2d3e4 +""" +from alembic import op +import sqlalchemy as sa +from app_new.core.models.rbac import Role, Permission, UserRole, RolePermission, UserPermission +from app_new.core.authorization import PERMISSIONS_BY_ROLE, ROLE_LABELS, GRANULAR_PERMISSION_CODES +revision = "9b06c7d8e9f0" +down_revision = "8a95b1c2d3e4" +branch_labels = None +depends_on = None +_TABLES = [Role.__table__, Permission.__table__, UserRole.__table__, RolePermission.__table__, UserPermission.__table__] + +def upgrade(): + bind = op.get_bind() + for table in _TABLES: + table.create(bind=bind, checkfirst=True) + roles = {slug: ROLE_LABELS.get(slug, slug) for slug in PERMISSIONS_BY_ROLE} + for slug, name in roles.items(): + bind.execute(sa.text("INSERT INTO roles (slug,name,description,is_system,is_active,created_at,updated_at) SELECT :slug,:name,:description,1,1,NOW(),NOW() FROM DUAL WHERE NOT EXISTS (SELECT 1 FROM roles WHERE slug=:slug)"), {"slug": slug, "name": name, "description": "Rôle système historique"}) + permission_codes = sorted({code for values in PERMISSIONS_BY_ROLE.values() for code in values} | GRANULAR_PERMISSION_CODES) + for code in permission_codes: + if code == "*": + module, action, name = "system", "all", "Accès complet" + else: + bits = code.split(".", 1); module = bits[0]; action = bits[1] if len(bits) > 1 else "manage"; name = f"{module.title()} — {action}" + bind.execute(sa.text("INSERT INTO permissions (code,name,module,action,description,is_active) SELECT :code,:name,:module,:action,:description,1 FROM DUAL WHERE NOT EXISTS (SELECT 1 FROM permissions WHERE code=:code)"), {"code": code, "name": name, "module": module, "action": action, "description": "Permission initiale compatible avec l'ancienne matrice"}) + for slug, codes in PERMISSIONS_BY_ROLE.items(): + role_id = bind.execute(sa.text("SELECT id FROM roles WHERE slug=:slug"), {"slug": slug}).scalar() + for code in codes: + permission_id = bind.execute(sa.text("SELECT id FROM permissions WHERE code=:code"), {"code": code}).scalar() + bind.execute(sa.text("INSERT INTO role_permissions (role_id,permission_id,effect) SELECT :role_id,:permission_id,'allow' FROM DUAL WHERE NOT EXISTS (SELECT 1 FROM role_permissions WHERE role_id=:role_id AND permission_id=:permission_id)"), {"role_id": role_id, "permission_id": permission_id}) + # Rattacher les comptes existants à leur rôle historique sans modifier leur rôle. + bind.execute(sa.text("INSERT INTO user_roles (user_id,role_id,assigned_at) SELECT u.id,r.id,NOW() FROM users u JOIN roles r ON r.slug=u.role WHERE NOT EXISTS (SELECT 1 FROM user_roles ur WHERE ur.user_id=u.id AND ur.role_id=r.id)")) + +def downgrade(): + bind = op.get_bind() + for table in reversed(_TABLES): + table.drop(bind=bind, checkfirst=True) diff --git a/migrations/versions/9c17d8e9f0a1_ajouter_actions_rbac.py b/migrations/versions/9c17d8e9f0a1_ajouter_actions_rbac.py new file mode 100644 index 0000000..909f155 --- /dev/null +++ b/migrations/versions/9c17d8e9f0a1_ajouter_actions_rbac.py @@ -0,0 +1,23 @@ +"""Ajouter les actions granulaires au catalogue RBAC. +Revision ID: 9c17d8e9f0a1 +Revises: 9b06c7d8e9f0 +""" +from alembic import op +import sqlalchemy as sa +from app_new.core.authorization import GRANULAR_PERMISSION_CODES +revision = "9c17d8e9f0a1" +down_revision = "9b06c7d8e9f0" +branch_labels = None +depends_on = None +def upgrade(): + bind = op.get_bind() + for code in sorted(GRANULAR_PERMISSION_CODES): + bits = code.split(".", 1) + module = bits[0] + action = bits[1] if len(bits) > 1 else "manage" + bind.execute(sa.text("INSERT INTO permissions (code,name,module,action,description,is_active) SELECT :code,:name,:module,:action,:description,1 FROM DUAL WHERE NOT EXISTS (SELECT 1 FROM permissions WHERE code=:code)"), {"code": code, "name": f"{module.title()} — {action}", "module": module, "action": action, "description": "Action granulaire configurable"}) +def downgrade(): + bind = op.get_bind() + for code in sorted(GRANULAR_PERMISSION_CODES): + bind.execute(sa.text("DELETE FROM permissions WHERE code=:code"), {"code": code}) + diff --git a/tests/integration/test_rbac.py b/tests/integration/test_rbac.py new file mode 100644 index 0000000..bf0b902 --- /dev/null +++ b/tests/integration/test_rbac.py @@ -0,0 +1,26 @@ +from uuid import uuid4 +from app_new.extensions import db +from app_new.core.models import User, Role, Permission, RolePermission, UserPermission +from app_new.core.authorization import has_permission + +def test_custom_role_permission_and_override(app): + with app.app_context(): + slug = "custom_" + uuid4().hex[:8] + role = Role(slug=slug, name="Gestionnaire test", is_system=False) + permission = Permission(code="stock.test.view", name="Stock test", module="stock", action="view") + db.session.add_all([role, permission]); db.session.flush() + db.session.add(RolePermission(role_id=role.id, permission_id=permission.id, effect="allow")) + user = User(username="rbac_" + uuid4().hex[:8], email=uuid4().hex + "@test.local", full_name="RBAC", role=slug) + user.set_password("test-password-strong") + db.session.add(user); db.session.commit() + assert has_permission("stock.test.view", user) + assert not has_permission("stock.test.edit", user) + db.session.add(UserPermission(user_id=user.id, permission_id=permission.id, effect="deny")) + db.session.commit() + assert not has_permission("stock.test.view", user) + +def test_admin_rbac_page_requires_authentication(client): + response = client.get("/admin/roles") + assert response.status_code == 302 + assert "/auth/login" in response.location +