from uuid import uuid4 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 def _role_with_permissions(slug, permissions): role = Role(slug=slug, name=slug, is_system=False, is_active=True) db.session.add(role) db.session.flush() for code in permissions: permission = Permission.query.filter_by(code=code).first() if permission is None: permission = Permission(code=code, name=code, module=code.split('.', 1)[0], action=code.split('.', 1)[1]) db.session.add(permission) db.session.flush() db.session.add(RolePermission(role_id=role.id, permission_id=permission.id, effect='allow')) return role def _user(): user = User(username='multi_' + uuid4().hex[:10], email=uuid4().hex + '@test.local', full_name='Multi', role='demandeur') user.set_password('test-password-strong') db.session.add(user) db.session.flush() return user def test_multiple_roles_union_and_removal(app): with app.app_context(): role_a = _role_with_permissions('role_a_' + uuid4().hex[:8], ['patrimoine.view']) role_b = _role_with_permissions('role_b_' + uuid4().hex[:8], ['prevention.view']) role_c = _role_with_permissions('role_c_' + uuid4().hex[:8], ['stock.manage']) user = _user() replace_user_roles(user, [role_a.id, role_b.id, role_c.id]) db.session.commit() assert has_permission('patrimoine.view', user) assert has_permission('prevention.view', user) assert has_permission('stock.manage', user) replace_user_roles(user, [role_b.id, role_c.id]) db.session.commit() assert not has_permission('patrimoine.view', user) assert has_permission('prevention.view', user) assert has_permission('stock.manage', user) def test_admin_removed_keeps_other_roles_without_wildcard(app): with app.app_context(): admin = Role.query.filter_by(slug='admin', is_active=True).first() tech = Role.query.filter_by(slug='technicien', is_active=True).first() prevention = Role.query.filter_by(slug='assistant_prevention', is_active=True).first() user = _user() replace_user_roles(user, [admin.id, tech.id, prevention.id]) db.session.commit() assert has_permission('*', user) replace_user_roles(user, [tech.id, prevention.id]) db.session.commit() assert not has_permission('*', user) assert has_permission('patrimoine.view', user) assert has_permission('prevention.view', user) def test_individual_deny_overrides_all_roles_and_allow_adds(app): with app.app_context(): role_a = _role_with_permissions('deny_a_' + uuid4().hex[:8], ['stock.manage']) role_b = _role_with_permissions('deny_b_' + uuid4().hex[:8], ['stock.view']) user = _user() replace_user_roles(user, [role_a.id, role_b.id]) db.session.commit() permission = Permission.query.filter_by(code='stock.manage').first() db.session.add(UserPermission(user_id=user.id, permission_id=permission.id, effect='deny')) extra = Permission(code='audit.test.view_' + uuid4().hex[:6], name='audit test', module='audit', action='view') db.session.add(extra) db.session.flush() db.session.add(UserPermission(user_id=user.id, permission_id=extra.id, effect='allow')) db.session.commit() assert not has_permission('stock.manage', user) assert has_permission(extra.code, user) def test_inactive_role_is_ignored_and_no_duplicate_links(app): with app.app_context(): code = 'system.test.view_' + uuid4().hex[:6] role = _role_with_permissions('inactive_' + uuid4().hex[:8], [code]) role.is_active = False user = _user() replace_user_roles(user, [role.id, role.id]) replace_user_roles(user, [role.id]) db.session.commit() assert not has_permission(code, user) assert UserRole.query.filter_by(user_id=user.id, role_id=role.id).count() == 0 def test_rbac_errors_fail_closed(app, monkeypatch): with app.app_context(): class BrokenUser: is_authenticated = True id = 999999 @property def role_links(self): raise RuntimeError('database unavailable') @property def permission_links(self): return [] assert not has_permission('dashboard.view', BrokenUser()) 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.flush() db.session.add(UserRole(user_id=user.id, role_id=role.id)); 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