import re from uuid import UUID from sqlalchemy import delete, func, select from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import ( BadRequestException, ConflictException, ForbiddenException, NotFoundException, ) from app.core.permissions import ROLE_DEFAULT_PERMISSIONS, PermissionCode, resolve_user_permissions from app.models.audit import AuditCategory from app.models.user import Permission, Role, RolePermission, User, UserPermissionOverride, UserRole from app.repositories.permission_repository import PermissionRepository from app.repositories.user_repository import UserRepository from app.schemas.rbac import ( PermissionDefinitionResponse, PermissionSyncResponse, RoleCreateRequest, RolePermissionEntryResponse, RolePermissionsResponse, RolePermissionsUpdateRequest, RoleResponse, RoleUpdateRequest, UserPermissionDetailResponse, UserPermissionOverrideItem, UserPermissionOverridesUpdateRequest, ) from app.services.audit_service import AuditService _MANAGEABLE_ROLES: list[UserRole] = [r for r in UserRole if r != UserRole.GENERAL_USER] class RbacService: def __init__( self, permission_repo: PermissionRepository | None = None, user_repo: UserRepository | None = None, audit_service: AuditService | None = None, ) -> None: self._permissions = permission_repo or PermissionRepository() self._users = user_repo or UserRepository() self._audit = audit_service or AuditService() async def _record_permission_event( self, db: AsyncSession, *, actor: User, action: str, description: str, target_id: str, details: dict[str, object] | None = None, ) -> None: role = getattr(actor.role, "value", actor.role) await self._audit.record_event( db, action=action, description=description, category=AuditCategory.ADMIN.value, actor_name=actor.name, actor_role=str(role), actor_id=actor.id, target_type="permission_catalog", target_id=target_id, ip_address=None, details=details, lab_id=actor.lab_id, ) async def list_permission_definitions(self, db: AsyncSession) -> list[PermissionDefinitionResponse]: stmt = select(Permission).order_by(Permission.code) res = await db.execute(stmt) rows = res.scalars().all() return [ PermissionDefinitionResponse( id=str(p.id), code=p.code, description=p.description, is_active=bool(p.is_active), ) for p in rows ] async def toggle_permission( self, db: AsyncSession, permission_id: UUID, is_active: bool, actor: User ) -> PermissionDefinitionResponse: """Enable or disable a system capability/feature globally.""" stmt = select(Permission).where(Permission.id == permission_id) res = await db.execute(stmt) perm = res.scalars().first() if not perm: raise NotFoundException(message=f"Permission with ID '{permission_id}' not found.") previous = bool(perm.is_active) perm.is_active = is_active if previous != is_active: await self._record_permission_event( db, actor=actor, action="PERMISSION_FEATURE_TOGGLED", description="Changed global feature availability.", target_id=str(perm.id), details={"permission_code": perm.code, "is_active": is_active}, ) await db.commit() await db.refresh(perm) return PermissionDefinitionResponse( id=str(perm.id), code=perm.code, description=perm.description, is_active=bool(perm.is_active), ) async def sync_permissions(self, db: AsyncSession, actor: User) -> PermissionSyncResponse: """ Synchronize database permission definitions with system PermissionCode enum. Adds any newly created feature capabilities, seeds role defaults, and ensures all platform features are present. """ added_count = 0 all_codes = [c.value for c in PermissionCode] # Fetch existing definitions existing_stmt = select(Permission) existing_rows = (await db.execute(existing_stmt)).scalars().all() existing_map: dict[str, Permission] = {p.code: p for p in existing_rows} # 1. Add any missing permission definitions for code_enum in PermissionCode: code_str = code_enum.value if code_str not in existing_map: parts = code_str.split(":") action = parts[1] if len(parts) > 1 else "access" domain = parts[0].capitalize() desc = f"Allows {action} operations across {domain} modules." new_perm = Permission(code=code_str, description=desc, is_active=True) db.add(new_perm) await db.flush() existing_map[code_str] = new_perm added_count += 1 # 2. Ensure role default assignments exist for role, allowed_codes in ROLE_DEFAULT_PERMISSIONS.items(): allowed_values = {c.value for c in allowed_codes} for code_str, perm_obj in existing_map.items(): rp_stmt = select(RolePermission).where( RolePermission.role == role, RolePermission.permission_id == perm_obj.id, ) existing_rp = (await db.execute(rp_stmt)).scalars().first() if existing_rp is None: db.add( RolePermission( role=role, permission_id=perm_obj.id, default_granted=code_str in allowed_values, ) ) await self._record_permission_event( db, actor=actor, action="PERMISSION_CATALOG_SYNCED", description="Synchronized the permission catalog with registered application capabilities.", target_id="catalog", details={"added_count": added_count, "registered_count": len(all_codes)}, ) await db.commit() # Total count total_stmt = select(Permission) total_perms = (await db.execute(total_stmt)).scalars().all() return PermissionSyncResponse( synced=len(all_codes), added=added_count, total=len(total_perms), codes=all_codes, ) async def list_roles(self, db: AsyncSession) -> list[RoleResponse]: roles_stmt = select(Role).order_by(Role.is_system.desc(), Role.name.asc()) roles_res = await db.execute(roles_stmt) roles = roles_res.scalars().all() user_counts_stmt = ( select(User.role, func.count(User.id)).where(User.deleted_at.is_(None)).group_by(User.role) ) user_counts: dict[str, int] = dict((await db.execute(user_counts_stmt)).tuples().all()) perm_counts_stmt = ( select(RolePermission.role, func.count(RolePermission.permission_id)) .where(RolePermission.default_granted.is_(True)) .group_by(RolePermission.role) ) perm_counts: dict[str, int] = dict((await db.execute(perm_counts_stmt)).tuples().all()) return [ RoleResponse( id=str(r.id), code=r.code, name=r.name, description=r.description, is_system=bool(r.is_system), user_count=int(user_counts.get(r.code, 0)), permission_count=int(perm_counts.get(r.code, 0)), created_at=r.created_at.isoformat() if r.created_at else None, ) for r in roles ] async def create_role(self, db: AsyncSession, payload: RoleCreateRequest, actor: User) -> RoleResponse: code = payload.code if not code: code = re.sub(r"[^a-z0-9_]+", "_", payload.name.lower().strip()).strip("_") if not code or len(code) < 2: raise BadRequestException(message="Role code must be at least 2 alphanumeric characters.") stmt = select(Role).where(Role.code == code) existing = (await db.execute(stmt)).scalars().first() if existing: raise ConflictException(message=f"Role with identifier '{code}' already exists.") new_role = Role( code=code, name=payload.name.strip(), description=payload.description.strip() if payload.description else None, is_system=False, ) db.add(new_role) await db.flush() all_perms_stmt = select(Permission) all_perms = (await db.execute(all_perms_stmt)).scalars().all() granted_set = set(payload.permissions) for perm in all_perms: db.add( RolePermission( role=code, permission_id=perm.id, default_granted=(perm.code in granted_set), ) ) await db.commit() await db.refresh(new_role) return RoleResponse( id=str(new_role.id), code=new_role.code, name=new_role.name, description=new_role.description, is_system=new_role.is_system, user_count=0, permission_count=len([p for p in all_perms if p.code in granted_set]), created_at=new_role.created_at.isoformat() if new_role.created_at else None, ) async def delete_role(self, db: AsyncSession, role_code: str, actor: User) -> None: stmt = select(Role).where(Role.code == role_code) role = (await db.execute(stmt)).scalars().first() if not role: raise NotFoundException(message=f"Role '{role_code}' was not found.") if role.is_system: raise ForbiddenException(message=f"System built-in role '{role.name}' cannot be deleted.") user_stmt = select(func.count(User.id)).where( User.role == role_code, User.deleted_at.is_(None), ) assigned_users = (await db.execute(user_stmt)).scalar_one() or 0 if assigned_users > 0: raise BadRequestException( message=( f"Cannot delete role '{role.name}' because {assigned_users}" " active staff member(s) are assigned to it. Please reassign them first." ) ) await db.execute(delete(RolePermission).where(RolePermission.role == role_code)) await db.delete(role) await db.commit() async def update_role( self, db: AsyncSession, role_code: str, payload: RoleUpdateRequest, actor: User ) -> RoleResponse: stmt = select(Role).where(Role.code == role_code) role = (await db.execute(stmt)).scalars().first() if not role: raise NotFoundException(message=f"Role '{role_code}' was not found.") if payload.name is not None: name_clean = payload.name.strip() if not name_clean: raise BadRequestException(message="Role name cannot be empty.") role.name = name_clean if payload.description is not None: role.description = payload.description.strip() or None await db.commit() await db.refresh(role) user_stmt = select(func.count(User.id)).where( User.role == role_code, User.deleted_at.is_(None), ) assigned_users = (await db.execute(user_stmt)).scalar_one() or 0 perm_stmt = select(func.count(RolePermission.permission_id)).where( RolePermission.role == role_code, RolePermission.default_granted.is_(True), ) perm_count = (await db.execute(perm_stmt)).scalar_one() or 0 return RoleResponse( id=str(role.id), code=role.code, name=role.name, description=role.description, is_system=role.is_system, user_count=assigned_users, permission_count=perm_count, created_at=role.created_at.isoformat() if role.created_at else None, ) async def get_role_permissions(self, db: AsyncSession, role: str) -> RolePermissionsResponse: all_perms_stmt = select(Permission).order_by(Permission.code) all_perms = (await db.execute(all_perms_stmt)).scalars().all() existing_rp_stmt = select(RolePermission).where(RolePermission.role == role) existing_rps = (await db.execute(existing_rp_stmt)).scalars().all() rp_map = {rp.permission_id: rp.default_granted for rp in existing_rps} missing_rps = [] for p in all_perms: if p.id not in rp_map: default_val = role == UserRole.SUPERADMIN.value missing_rps.append( RolePermission( role=role, permission_id=p.id, default_granted=default_val, ) ) rp_map[p.id] = default_val if missing_rps: db.add_all(missing_rps) await db.commit() permissions = [ RolePermissionEntryResponse( code=p.code, description=p.description, granted=bool(rp_map.get(p.id, False)), ) for p in all_perms ] return RolePermissionsResponse(role=role, permissions=permissions) async def update_role_permissions( self, db: AsyncSession, role: str, payload: RolePermissionsUpdateRequest, actor: User, ) -> RolePermissionsResponse: if role == UserRole.SUPERADMIN.value and not actor.is_superadmin: raise ForbiddenException(message="Only superadmins can modify superadmin role permissions.") for item in payload.permissions: perm = await self._permissions.get_by_code(db, item.code) if not perm: raise NotFoundException(message=f"Permission '{item.code}' was not found.") stmt = select(RolePermission).where( RolePermission.role == role, RolePermission.permission_id == perm.id, ) row = (await db.execute(stmt)).scalars().first() if row is None: db.add( RolePermission( role=role, permission_id=perm.id, default_granted=item.granted, ) ) else: row.default_granted = item.granted await db.commit() return await self.get_role_permissions(db, role) async def get_user_permission_detail( self, db: AsyncSession, user_id: UUID ) -> UserPermissionDetailResponse: user = await self._users.get_by_id(db, user_id) if not user: raise NotFoundException(message="User not found.") role_matrix = await self.get_role_permissions(db, user.role) overrides = await self._permissions.get_user_overrides(db, user_id) override_items = [ UserPermissionOverrideItem(code=o.permission.code, is_granted=o.is_granted) for o in overrides if o.permission is not None ] effective = sorted(await resolve_user_permissions(db, user)) return UserPermissionDetailResponse( user_id=str(user.id), role=user.role, effective_permissions=effective, role_defaults=role_matrix.permissions, overrides=override_items, ) async def update_user_overrides( self, db: AsyncSession, user_id: UUID, payload: UserPermissionOverridesUpdateRequest, actor: User, ) -> UserPermissionDetailResponse: user = await self._users.get_by_id(db, user_id) if not user: raise NotFoundException(message="User not found.") if user.is_superadmin and not actor.is_superadmin: raise ForbiddenException(message="Only superadmins can modify superadmin permission overrides.") role_matrix = { p.code: p.granted for p in (await self.get_role_permissions(db, user.role)).permissions } existing_rows = await self._permissions.get_user_overrides(db, user_id) existing_by_code = {o.permission.code: o for o in existing_rows if o.permission} requested = {item.code: item.is_granted for item in payload.overrides} for code, role_default in role_matrix.items(): if code not in requested: continue desired = requested[code] if desired == role_default: row = existing_by_code.get(code) if row: await db.delete(row) continue perm = await self._permissions.get_by_code(db, code) if not perm: continue row = existing_by_code.get(code) if row: row.is_granted = desired row.set_by = actor.id else: db.add( UserPermissionOverride( user_id=user_id, permission_id=perm.id, is_granted=desired, set_by=actor.id, ) ) await db.commit() return await self.get_user_permission_detail(db, user_id)