# Copyright (c) 2025 Patrick Motsch # All rights reserved. """ RBAC routes for the backend API. Implements endpoints for role-based access control permissions. MULTI-TENANT: - Permission queries use RequestContext (mandateId from header) - AccessRule management is Admin-aware (SysAdmin sees all, MandateAdmin sees own mandate's rules) - Role management is Admin-aware (SysAdmin sees all, MandateAdmin sees own mandate's roles) - Catalog stats and cleanup remain SysAdmin-only """ from fastapi import APIRouter, HTTPException, Depends, Query, Body, Path, Request from typing import Optional, List, Dict, Any import logging import json import math from modules.auth import limiter, getRequestContext, requireSysAdminRole, RequestContext from modules.datamodels.datamodelUam import User, UserPermissions, AccessLevel from modules.datamodels.datamodelRbac import AccessRuleContext, AccessRule, Role from modules.datamodels.datamodelMembership import UserMandate from modules.datamodels.datamodelPagination import PaginationParams, PaginatedResponse, PaginationMetadata, normalize_pagination_dict from modules.interfaces.interfaceDbApp import getInterface, getRootInterface from modules.shared.i18nRegistry import apiRouteContext, t, resolveText routeApiMsg = apiRouteContext("routeAdminRbacRules") # Configure logger logger = logging.getLogger(__name__) router = APIRouter( prefix="/api/rbac", tags=["RBAC"], responses={404: {"description": "Not found"}} ) def _getAdminMandateIds(context: RequestContext) -> List[str]: """Get mandate IDs where the user has an admin role.""" mandateIds = [] try: from modules.interfaces.interfaceDbApp import getRootInterface rootInterface = getRootInterface() userMandates = rootInterface.getUserMandates(str(context.user.id)) for um in userMandates: if not getattr(um, 'enabled', True): continue umId = getattr(um, 'id', None) mandateId = getattr(um, 'mandateId', None) if not umId or not mandateId: continue roleIds = rootInterface.getRoleIdsForUserMandate(str(umId)) for roleId in roleIds: role = rootInterface.getRole(roleId) if role and role.roleLabel == "admin" and not role.featureInstanceId: mandateIds.append(str(mandateId)) break except Exception as e: logger.error(f"Error getting admin mandate IDs: {e}") return mandateIds def _isRoleInAdminMandates(roleId: str, adminMandateIds: List[str]) -> bool: """Check if a role belongs to one of the admin's mandates.""" try: from modules.interfaces.interfaceDbApp import getRootInterface rootInterface = getRootInterface() role = rootInterface.getRole(roleId) if not role: return False return str(role.mandateId) in adminMandateIds if role.mandateId else False except Exception: return False @router.get("/permissions", response_model=UserPermissions) @limiter.limit("300/minute") # Raised from 60 - sidebar checks many pages individually def get_permissions( request: Request, context: str = Query(..., description="Context type: DATA, UI, or RESOURCE"), item: Optional[str] = Query(None, description="Item identifier (table name, UI path, or resource path)"), reqContext: RequestContext = Depends(getRequestContext) ) -> UserPermissions: """ Get RBAC permissions for the current user for a specific context and item. MULTI-TENANT: Uses RequestContext (mandateId/featureInstanceId from headers). Query Parameters: - context: Context type (DATA, UI, or RESOURCE) - item: Optional item identifier. For DATA: table name (e.g., "UserInDB"), For UI: cascading string (e.g., "playground.voice.settings"), For RESOURCE: cascading string (e.g., "ai.model.anthropic") Returns: - UserPermissions object with view, read, create, update, delete permissions Examples: - GET /api/rbac/permissions?context=DATA&item=UserInDB - GET /api/rbac/permissions?context=UI&item=playground.voice.settings - GET /api/rbac/permissions?context=RESOURCE&item=ai.model.anthropic """ try: # Validate context try: accessContext = AccessRuleContext(context.upper()) except ValueError: raise HTTPException( status_code=400, detail=f"Invalid context '{context}'. Must be one of: DATA, UI, RESOURCE" ) # Get interface and RBAC permissions interface = getInterface(reqContext.user) if not interface.rbac: raise HTTPException( status_code=500, detail=routeApiMsg("RBAC interface not available") ) # MULTI-TENANT: Get permissions using context (mandateId/featureInstanceId) # For DATA context, resolve short model names to full objectKeys # e.g., "ChatWorkflow" → "data.chat.ChatWorkflow" resolvedItem = item or "" if accessContext == AccessRuleContext.DATA and resolvedItem and "." not in resolvedItem: from modules.interfaces.interfaceRbac import buildDataObjectKey resolvedItem = buildDataObjectKey(resolvedItem) # Pass mandateId and featureInstanceId to load Feature-Instance roles permissions = interface.rbac.getUserPermissions( reqContext.user, accessContext, resolvedItem, mandateId=reqContext.mandateId, featureInstanceId=reqContext.featureInstanceId ) return permissions except HTTPException: raise except Exception as e: logger.error(f"Error getting RBAC permissions: {str(e)}") raise HTTPException( status_code=500, detail=f"Failed to get permissions: {str(e)}" ) @router.get("/permissions/all", response_model=Dict[str, Any]) @limiter.limit("120/minute") # Raised from 30 - optimized endpoint for bulk permission fetch def get_all_permissions( request: Request, context: Optional[str] = Query(None, description="Context type: UI or RESOURCE (if not provided, returns both)"), reqContext: RequestContext = Depends(getRequestContext) ) -> Dict[str, Any]: """ Get all RBAC permissions for the current user for UI and/or RESOURCE contexts. MULTI-TENANT: Uses RequestContext (mandateId/featureInstanceId from headers). This endpoint is optimized for UI initialization to avoid multiple API calls. Query Parameters: - context: Optional context filter. If "UI", returns only UI permissions. If "RESOURCE", returns only RESOURCE permissions. If not provided, returns both UI and RESOURCE permissions. Returns: - Dictionary with structure: { "ui": { "item1": UserPermissions, "item2": UserPermissions, ... }, "resource": { "item1": UserPermissions, "item2": UserPermissions, ... } } If context is specified, only that context is returned. Example: - GET /api/rbac/permissions/all - GET /api/rbac/permissions/all?context=UI - GET /api/rbac/permissions/all?context=RESOURCE """ try: # Get interface and RBAC permissions interface = getInterface(reqContext.user) if not interface.rbac: raise HTTPException( status_code=500, detail=routeApiMsg("RBAC interface not available") ) # Determine which contexts to fetch contextsToFetch = [] if context: try: accessContext = AccessRuleContext(context.upper()) if accessContext in [AccessRuleContext.UI, AccessRuleContext.RESOURCE]: contextsToFetch = [accessContext] else: raise HTTPException( status_code=400, detail=f"Context '{context}' must be UI or RESOURCE for this endpoint" ) except ValueError: raise HTTPException( status_code=400, detail=f"Invalid context '{context}'. Must be UI or RESOURCE" ) else: # Return both UI and RESOURCE if no context specified contextsToFetch = [AccessRuleContext.UI, AccessRuleContext.RESOURCE] result: Dict[str, Any] = {} # For UI/RESOURCE permissions: These are GLOBAL (not mandate-specific) # System roles (admin, user, viewer) have global UI rules that apply without mandate context rootInterface = getRootInterface() # Start with roleIds from current mandate context (if any) roleIds = list(reqContext.roleIds or []) # For UI/RESOURCE: Load system roles the user has across ALL their mandates # This allows users to access system UI elements without needing a specific mandate header allUserMandates = rootInterface.getUserMandates(str(reqContext.user.id)) userMandates = [um for um in allUserMandates if um.enabled] logger.debug(f"UI/RESOURCE permissions: Found {len(userMandates)} UserMandates for user {reqContext.user.id}") # Collect all role IDs the user has across all mandates for userMandate in userMandates: mandateRoleIds = rootInterface.getRoleIdsForUserMandate(userMandate.id) logger.debug(f"UI/RESOURCE permissions: UserMandate {userMandate.id} (mandate {userMandate.mandateId}) has {len(mandateRoleIds)} roles: {mandateRoleIds}") for rid in mandateRoleIds: if rid not in roleIds: roleIds.append(rid) logger.debug(f"UI/RESOURCE permissions: User has {len(roleIds)} roles across all mandates") if not roleIds and not reqContext.hasSysAdminRole: # No roles at all, return empty permissions for ctx in contextsToFetch: result[ctx.value.lower()] = {} return result # Get all access rules for user's roles and requested contexts # IMPORTANT: Use direct DB access without RBAC filtering! # Otherwise we have a chicken-and-egg problem: need AccessRule read permission to calculate permissions allRules: Dict[AccessRuleContext, List[AccessRule]] = {} for ctx in contextsToFetch: allRules[ctx] = [] # Get all rules for user's roles - bypass RBAC filtering for roleId in roleIds: # Use interface method and filter by context rules = rootInterface.getAccessRulesByRole(str(roleId)) for rule in rules: if rule.context == ctx.value: allRules[ctx].append(rule) # Build result: for each context, collect all unique items and calculate permissions for ctx in contextsToFetch: result[ctx.value.lower()] = {} # Check for global rule (item=None) first - this grants access to ALL UI/RESOURCE items # Calculate permissions directly from loaded rules (don't call getUserPermissions - it requires mandateId) hasGlobalRule = False globalView = False globalRead = None globalCreate = None globalUpdate = None globalDelete = None for rule in allRules[ctx]: if rule.item is None: hasGlobalRule = True if rule.view: globalView = True if rule.read: globalRead = rule.read.value if hasattr(rule.read, 'value') else rule.read if rule.create: globalCreate = rule.create.value if hasattr(rule.create, 'value') else rule.create if rule.update: globalUpdate = rule.update.value if hasattr(rule.update, 'value') else rule.update if rule.delete: globalDelete = rule.delete.value if hasattr(rule.delete, 'value') else rule.delete # If there's a global rule with view permission, add "_global" key if hasGlobalRule and globalView: logger.debug(f"Adding _global key for context {ctx.value} with view={globalView}") result[ctx.value.lower()]["_global"] = { "view": globalView, "read": globalRead, "create": globalCreate, "update": globalUpdate, "delete": globalDelete } # Collect all unique items from rules (specific rules) items = set() for rule in allRules[ctx]: if rule.item: items.add(rule.item) # For each item, calculate user permissions # For UI/RESOURCE context: Calculate permissions directly from the collected rules # (Don't use getUserPermissions with mandateId - that would limit to one mandate's roles) for item in sorted(items): # Find matching rules for this item from the already-collected rules itemView = False for rule in allRules[ctx]: if rule.item == item and rule.view: itemView = True break # Only include if user has view permission if itemView: result[ctx.value.lower()][item] = { "view": True, "read": None, # UI context doesn't use CRUD permissions "create": None, "update": None, "delete": None } return result except HTTPException: raise except Exception as e: logger.error(f"Error getting all RBAC permissions: {str(e)}") raise HTTPException( status_code=500, detail=f"Failed to get all permissions: {str(e)}" ) @router.get("/rules", response_model=PaginatedResponse) @limiter.limit("30/minute") def get_access_rules( request: Request, roleLabel: Optional[str] = Query(None, description="Filter by role label"), context: Optional[str] = Query(None, description="Filter by context (DATA, UI, RESOURCE)"), item: Optional[str] = Query(None, description="Filter by item identifier"), pagination: Optional[str] = Query(None, description="JSON-encoded PaginationParams object"), reqContext: RequestContext = Depends(getRequestContext) ) -> PaginatedResponse: """ Get access rules with optional filters. MULTI-TENANT: Admin-aware (SysAdmin sees all, MandateAdmin sees own mandate's rules). Query Parameters: - roleLabel: Optional role label filter - context: Optional context filter (DATA, UI, RESOURCE) - item: Optional item filter Returns: - List of AccessRule objects """ try: isSysAdmin = reqContext.hasSysAdminRole adminMandateIds = [] if isSysAdmin else _getAdminMandateIds(reqContext) if not isSysAdmin and not adminMandateIds: raise HTTPException(status_code=403, detail=routeApiMsg("Admin role required")) # Get interface - uses root interface for admin access interface = getRootInterface() # Parse context if provided accessContext = None if context: try: accessContext = AccessRuleContext(context.upper()) except ValueError: raise HTTPException( status_code=400, detail=f"Invalid context '{context}'. Must be one of: DATA, UI, RESOURCE" ) # Parse pagination parameter paginationParams = None if pagination: try: paginationDict = json.loads(pagination) if paginationDict: paginationDict = normalize_pagination_dict(paginationDict) paginationParams = PaginationParams(**paginationDict) except (json.JSONDecodeError, ValueError) as e: raise HTTPException( status_code=400, detail=f"Invalid pagination parameter: {str(e)}" ) # Get rules with optional pagination # MandateAdmin: fetch all then filter by admin's mandates. # NOTE: Cannot use DB-level pagination for MandateAdmin because # _isRoleInAdminMandates requires joining Role → mandateId which # isn't expressible via getRecordsetPaginated's recordFilter. if not isSysAdmin: allRules = interface.getAccessRules( roleLabel=roleLabel, context=accessContext, item=item, pagination=None ) filteredRules = [rule for rule in allRules if _isRoleInAdminMandates(str(rule.roleId), adminMandateIds)] if paginationParams: totalItems = len(filteredRules) totalPages = math.ceil(totalItems / paginationParams.pageSize) if totalItems > 0 else 0 startIdx = (paginationParams.page - 1) * paginationParams.pageSize endIdx = startIdx + paginationParams.pageSize return PaginatedResponse( items=[rule.model_dump() for rule in filteredRules[startIdx:endIdx]], pagination=PaginationMetadata( currentPage=paginationParams.page, pageSize=paginationParams.pageSize, totalItems=totalItems, totalPages=totalPages, sort=paginationParams.sort, filters=paginationParams.filters ) ) else: return PaginatedResponse( items=[rule.model_dump() for rule in filteredRules], pagination=None ) # SysAdmin: use server-side pagination result = interface.getAccessRules( roleLabel=roleLabel, context=accessContext, item=item, pagination=paginationParams ) # If pagination was requested, result is PaginatedResult # If no pagination, result is List[AccessRule] if paginationParams: return PaginatedResponse( items=[rule.model_dump() for rule in result.items], pagination=PaginationMetadata( currentPage=paginationParams.page, pageSize=paginationParams.pageSize, totalItems=result.totalItems, totalPages=result.totalPages, sort=paginationParams.sort, filters=paginationParams.filters ) ) else: return PaginatedResponse( items=[rule.model_dump() for rule in result], pagination=None ) except HTTPException: raise except Exception as e: logger.error(f"Error getting access rules: {str(e)}") raise HTTPException( status_code=500, detail=f"Failed to get access rules: {str(e)}" ) @router.get("/rules/by-role/{roleId}", response_model=PaginatedResponse) @limiter.limit("30/minute") def get_access_rules_by_role( request: Request, roleId: str = Path(..., description="Role ID to get rules for"), reqContext: RequestContext = Depends(getRequestContext) ) -> PaginatedResponse: """ Get all access rules for a specific role. MULTI-TENANT: Admin-aware (SysAdmin sees all, MandateAdmin sees own mandate's roles). Path Parameters: - roleId: The role ID to get rules for Returns: - List of AccessRule objects for the specified role """ try: isSysAdmin = reqContext.hasSysAdminRole adminMandateIds = [] if isSysAdmin else _getAdminMandateIds(reqContext) if not isSysAdmin and not adminMandateIds: raise HTTPException(status_code=403, detail=routeApiMsg("Admin role required")) # MandateAdmin: verify role belongs to their mandates if not isSysAdmin and not _isRoleInAdminMandates(roleId, adminMandateIds): raise HTTPException(status_code=403, detail=routeApiMsg("Access denied: role not in your mandates")) interface = getRootInterface() # Get rules from database using interface method ruleObjects = interface.getAccessRulesByRole(roleId) return PaginatedResponse( items=[rule.model_dump() for rule in ruleObjects], pagination=None ) except HTTPException: raise except Exception as e: logger.error(f"Error getting access rules for role {roleId}: {str(e)}") raise HTTPException( status_code=500, detail=f"Failed to get access rules for role: {str(e)}" ) @router.get("/rules/{ruleId}", response_model=dict) @limiter.limit("30/minute") def get_access_rule( request: Request, ruleId: str = Path(..., description="Access rule ID"), reqContext: RequestContext = Depends(getRequestContext) ) -> dict: """ Get a specific access rule by ID. MULTI-TENANT: Admin-aware (SysAdmin sees all, MandateAdmin sees own mandate's rules). Path Parameters: - ruleId: Access rule ID Returns: - AccessRule object """ try: isSysAdmin = reqContext.hasSysAdminRole adminMandateIds = [] if isSysAdmin else _getAdminMandateIds(reqContext) if not isSysAdmin and not adminMandateIds: raise HTTPException(status_code=403, detail=routeApiMsg("Admin role required")) # Get interface - uses root interface for admin access interface = getRootInterface() # Get rule rule = interface.getAccessRule(ruleId) if not rule: raise HTTPException( status_code=404, detail=f"Access rule {ruleId} not found" ) # MandateAdmin: verify rule's role belongs to their mandates if not isSysAdmin and not _isRoleInAdminMandates(str(rule.roleId), adminMandateIds): raise HTTPException(status_code=403, detail=routeApiMsg("Access denied: rule's role not in your mandates")) # Convert to dict for JSON serialization return rule.model_dump() except HTTPException: raise except Exception as e: logger.error(f"Error getting access rule {ruleId}: {str(e)}") raise HTTPException( status_code=500, detail=f"Failed to get access rule: {str(e)}" ) @router.post("/rules", response_model=dict) @limiter.limit("30/minute") def create_access_rule( request: Request, accessRuleData: dict = Body(..., description="Access rule data"), reqContext: RequestContext = Depends(getRequestContext) ) -> dict: """ Create a new access rule. MULTI-TENANT: Admin-aware (SysAdmin sees all, MandateAdmin creates for own mandate's roles). Request Body: - AccessRule object data (roleLabel, context, item, view, read, create, update, delete) Returns: - Created AccessRule object """ try: isSysAdmin = reqContext.hasSysAdminRole adminMandateIds = [] if isSysAdmin else _getAdminMandateIds(reqContext) if not isSysAdmin and not adminMandateIds: raise HTTPException(status_code=403, detail=routeApiMsg("Admin role required")) # Get interface - uses root interface for admin access interface = getRootInterface() # Validate and parse access rule data try: logger.debug(f"Creating access rule with data: {accessRuleData}") # Parse context if provided as string if "context" in accessRuleData and isinstance(accessRuleData["context"], str): accessRuleData["context"] = AccessRuleContext(accessRuleData["context"].upper()) # Parse AccessLevel fields if provided as strings # Handle empty strings by converting to None for field in ["read", "create", "update", "delete"]: if field in accessRuleData: value = accessRuleData[field] if value == "" or value is None: accessRuleData[field] = None elif isinstance(value, str): accessRuleData[field] = AccessLevel(value) # Create AccessRule object accessRule = AccessRule(**accessRuleData) except ValueError as e: logger.error(f"Invalid access rule data: {accessRuleData} - Error: {str(e)}") raise HTTPException( status_code=400, detail=f"Invalid access rule data: {str(e)}" ) # MandateAdmin: verify the rule's role belongs to their mandates if not isSysAdmin and accessRule.roleId: if not _isRoleInAdminMandates(str(accessRule.roleId), adminMandateIds): raise HTTPException(status_code=403, detail=routeApiMsg("Access denied: role not in your mandates")) # Create rule createdRule = interface.createAccessRule(accessRule) logger.info(f"Created access rule {createdRule.id} by admin {reqContext.user.id}") # Convert to dict for JSON serialization return createdRule.model_dump() except HTTPException: raise except Exception as e: logger.error(f"Error creating access rule: {str(e)}") raise HTTPException( status_code=500, detail=f"Failed to create access rule: {str(e)}" ) @router.put("/rules/{ruleId}", response_model=dict) @limiter.limit("30/minute") def update_access_rule( request: Request, ruleId: str = Path(..., description="Access rule ID"), accessRuleData: dict = Body(..., description="Updated access rule data"), reqContext: RequestContext = Depends(getRequestContext) ) -> dict: """ Update an existing access rule. MULTI-TENANT: Admin-aware (SysAdmin sees all, MandateAdmin updates own mandate's rules). Path Parameters: - ruleId: Access rule ID Request Body: - AccessRule object data (roleLabel, context, item, view, read, create, update, delete) Returns: - Updated AccessRule object """ try: isSysAdmin = reqContext.hasSysAdminRole adminMandateIds = [] if isSysAdmin else _getAdminMandateIds(reqContext) if not isSysAdmin and not adminMandateIds: raise HTTPException(status_code=403, detail=routeApiMsg("Admin role required")) # Get interface - uses root interface for admin access interface = getRootInterface() # Get existing rule to ensure it exists existingRule = interface.getAccessRule(ruleId) if not existingRule: raise HTTPException( status_code=404, detail=f"Access rule {ruleId} not found" ) # MandateAdmin: verify existing rule's role belongs to their mandates if not isSysAdmin and not _isRoleInAdminMandates(str(existingRule.roleId), adminMandateIds): raise HTTPException(status_code=403, detail=routeApiMsg("Access denied: rule's role not in your mandates")) # Validate and parse access rule data try: # Merge with existing rule data updateData = existingRule.model_dump() updateData.update(accessRuleData) # Parse context if provided as string if "context" in updateData and isinstance(updateData["context"], str): updateData["context"] = AccessRuleContext(updateData["context"].upper()) # Parse AccessLevel fields if provided as strings # Handle empty strings by converting to None for field in ["read", "create", "update", "delete"]: if field in updateData: value = updateData[field] if value == "" or value is None: updateData[field] = None elif isinstance(value, str): updateData[field] = AccessLevel(value) # Ensure ID is set correctly updateData["id"] = ruleId # Create AccessRule object accessRule = AccessRule(**updateData) except ValueError as e: logger.error(f"Invalid access rule update data: {updateData} - Error: {str(e)}") raise HTTPException( status_code=400, detail=f"Invalid access rule data: {str(e)}" ) # Update rule updatedRule = interface.updateAccessRule(ruleId, accessRule) logger.info(f"Updated access rule {ruleId} by admin {reqContext.user.id}") # Convert to dict for JSON serialization return updatedRule.model_dump() except HTTPException: raise except Exception as e: logger.error(f"Error updating access rule {ruleId}: {str(e)}") raise HTTPException( status_code=500, detail=f"Failed to update access rule: {str(e)}" ) @router.delete("/rules/{ruleId}") @limiter.limit("30/minute") def delete_access_rule( request: Request, ruleId: str = Path(..., description="Access rule ID"), reqContext: RequestContext = Depends(getRequestContext) ) -> dict: """ Delete an access rule. MULTI-TENANT: Admin-aware (SysAdmin sees all, MandateAdmin deletes own mandate's rules). Path Parameters: - ruleId: Access rule ID Returns: - Success message """ try: isSysAdmin = reqContext.hasSysAdminRole adminMandateIds = [] if isSysAdmin else _getAdminMandateIds(reqContext) if not isSysAdmin and not adminMandateIds: raise HTTPException(status_code=403, detail=routeApiMsg("Admin role required")) # Get interface - uses root interface for admin access interface = getRootInterface() # Get existing rule to ensure it exists existingRule = interface.getAccessRule(ruleId) if not existingRule: raise HTTPException( status_code=404, detail=f"Access rule {ruleId} not found" ) # MandateAdmin: verify rule's role belongs to their mandates if not isSysAdmin and not _isRoleInAdminMandates(str(existingRule.roleId), adminMandateIds): raise HTTPException(status_code=403, detail=routeApiMsg("Access denied: rule's role not in your mandates")) # Delete rule success = interface.deleteAccessRule(ruleId) if not success: raise HTTPException( status_code=500, detail=f"Failed to delete access rule {ruleId}" ) logger.info(f"Deleted access rule {ruleId} by admin {reqContext.user.id}") return {"success": True, "message": f"Access rule {ruleId} deleted successfully"} except HTTPException: raise except Exception as e: logger.error(f"Error deleting access rule {ruleId}: {str(e)}") raise HTTPException( status_code=500, detail=f"Failed to delete access rule: {str(e)}" ) # ============================================================================ # Role Management Endpoints # MULTI-TENANT: Admin-aware (SysAdmin sees all, MandateAdmin sees own mandate's roles) # ============================================================================ @router.get("/roles", response_model=PaginatedResponse) @limiter.limit("60/minute") def list_roles( request: Request, pagination: Optional[str] = Query(None, description="JSON-encoded PaginationParams object"), includeTemplates: bool = Query(False, description="Include feature template roles"), mandateId: Optional[str] = Query(None, description="Include mandate-specific roles for this mandate"), scopeFilter: Optional[str] = Query(None, description="Filter by scope: 'all', 'mandate', 'global', 'system'"), reqContext: RequestContext = Depends(getRequestContext) ) -> PaginatedResponse: """ Get list of roles with metadata. MULTI-TENANT: Admin-aware (SysAdmin sees all, MandateAdmin sees own mandate's roles). By default, only returns true global roles (mandateId=None, featureInstanceId=None, featureCode=None). Feature template roles are managed via /api/features/templates/roles. NOTE: Base query (getAllRoles) already uses db.getRecordsetPaginated() internally. However pagination=None is passed here because post-processing adds computed fields (userCount, scopeType) and applies scope/mandate/template filtering that cannot run at the DB level. In-memory pagination is applied after all transformations. Args: pagination: Optional pagination parameters (includes search, filters, sort) includeTemplates: If True, also include feature template roles (featureCode != None) mandateId: If provided, also include mandate-specific roles for this mandate scopeFilter: Filter by scope type: 'all', 'mandate', 'global', 'system' Returns: - List of role dictionaries with role label, description, user count, and computed scopeType """ try: isSysAdmin = reqContext.hasSysAdminRole adminMandateIds = [] if isSysAdmin else _getAdminMandateIds(reqContext) if not isSysAdmin and not adminMandateIds: raise HTTPException(status_code=403, detail=routeApiMsg("Admin role required")) interface = getRootInterface() # Parse pagination parameter paginationParams = None if pagination: try: paginationDict = json.loads(pagination) if paginationDict: paginationDict = normalize_pagination_dict(paginationDict) paginationParams = PaginationParams(**paginationDict) except (json.JSONDecodeError, ValueError) as e: raise HTTPException( status_code=400, detail=f"Invalid pagination parameter: {str(e)}" ) # Get all roles from database dbRoles = interface.getAllRoles(pagination=None) # Count role assignments from UserMandateRole table roleCounts = interface.countRoleAssignments() # Helper function to compute scopeType # Note: mandateId takes precedence — a role with mandateId is always "mandate" scope, # even if isSystemRole=True (which just means it was copied from a system template) def _computeScopeType(role) -> str: if role.mandateId: return "mandate" if role.isSystemRole: return "system" return "global" # Convert Role objects to dictionaries and add user counts # Filter logic: # - Always include global roles (mandateId=None, featureInstanceId=None) # - If mandateId provided, also include roles for that specific mandate # - Unless includeTemplates=True, exclude feature template roles (featureCode != None) result = [] for role in dbRoles: # Always exclude feature-instance level roles if role.featureInstanceId is not None: continue # When mandateId requested: only mandate-scoped roles # When no mandateId: only global roles (mandateId=None) if mandateId: if role.mandateId != mandateId: continue else: if role.mandateId is not None: continue # Filter: Exclude feature template roles unless includeTemplates=True if not includeTemplates and role.featureCode is not None: continue # Compute scopeType (system, global, mandate) scopeType = _computeScopeType(role) # Apply scopeFilter if provided if scopeFilter and scopeFilter != 'all': if scopeFilter == 'mandate' and scopeType != 'mandate': continue if scopeFilter == 'global' and scopeType not in ('global', 'system'): continue if scopeFilter == 'system' and scopeType != 'system': continue result.append({ "id": role.id, "roleLabel": role.roleLabel, "description": resolveText(role.description), "mandateId": role.mandateId, "featureInstanceId": role.featureInstanceId, "featureCode": role.featureCode, "userCount": roleCounts.get(str(role.id), 0), "isSystemRole": role.isSystemRole, "scopeType": scopeType # Computed field for frontend display }) # MandateAdmin: filter to only roles in admin's mandates if not isSysAdmin: result = [r for r in result if r.get("mandateId") and str(r["mandateId"]) in adminMandateIds] # Apply search, filtering and sorting if pagination requested if paginationParams: # Apply search (if search term provided in filters) searchTerm = paginationParams.filters.get("search", "").lower() if paginationParams.filters else "" if searchTerm: searchedResult = [] for item in result: roleLabel = (item.get("roleLabel") or "").lower() descText = (item.get("description") or "").lower() scopeType = (item.get("scopeType") or "").lower() if searchTerm in roleLabel or searchTerm in descText or searchTerm in scopeType: searchedResult.append(item) result = searchedResult # Apply filtering (if filters provided) if paginationParams.filters: # Use the interface's filter method filteredResult = interface._applyFilters(result, paginationParams.filters) else: filteredResult = result # Apply sorting (in order of sortFields) if paginationParams.sort: sortedResult = interface._applySorting(filteredResult, paginationParams.sort) else: sortedResult = filteredResult # Apply pagination totalItems = len(sortedResult) totalPages = math.ceil(totalItems / paginationParams.pageSize) if totalItems > 0 else 0 startIdx = (paginationParams.page - 1) * paginationParams.pageSize endIdx = startIdx + paginationParams.pageSize paginatedResult = sortedResult[startIdx:endIdx] return PaginatedResponse( items=paginatedResult, pagination=PaginationMetadata( currentPage=paginationParams.page, pageSize=paginationParams.pageSize, totalItems=totalItems, totalPages=totalPages, sort=paginationParams.sort, filters=paginationParams.filters ) ) else: # No pagination - return all roles return PaginatedResponse( items=result, pagination=None ) except HTTPException: raise except Exception as e: logger.error(f"Error listing roles: {str(e)}") raise HTTPException( status_code=500, detail=f"Failed to list roles: {str(e)}" ) @router.get("/roles/filter-values") @limiter.limit("60/minute") def get_roles_filter_values( request: Request, column: str = Query(..., description="Column key"), pagination: Optional[str] = Query(None, description="JSON-encoded current filters"), includeTemplates: bool = Query(False, description="Include feature template roles"), mandateId: Optional[str] = Query(None, description="Include mandate-specific roles for this mandate"), scopeFilter: Optional[str] = Query(None, description="Filter by scope: 'all', 'mandate', 'global', 'system'"), reqContext: RequestContext = Depends(getRequestContext) ) -> list: """Return distinct filter values for a column in roles.""" try: from modules.routes.routeDataUsers import _handleFilterValuesRequest isSysAdmin = reqContext.hasSysAdminRole adminMandateIds = [] if isSysAdmin else _getAdminMandateIds(reqContext) if not isSysAdmin and not adminMandateIds: raise HTTPException(status_code=403, detail=routeApiMsg("Admin role required")) interface = getRootInterface() dbRoles = interface.getAllRoles(pagination=None) roleCounts = interface.countRoleAssignments() def _computeScopeType(role) -> str: if role.mandateId: return "mandate" if role.isSystemRole: return "system" return "global" result = [] for role in dbRoles: if role.featureInstanceId is not None: continue if mandateId: if role.mandateId != mandateId: continue else: if role.mandateId is not None: continue if not includeTemplates and role.featureCode is not None: continue scopeType = _computeScopeType(role) if scopeFilter and scopeFilter != 'all': if scopeFilter == 'mandate' and scopeType != 'mandate': continue if scopeFilter == 'global' and scopeType not in ('global', 'system'): continue if scopeFilter == 'system' and scopeType != 'system': continue result.append({ "id": role.id, "roleLabel": role.roleLabel, "description": resolveText(role.description), "mandateId": role.mandateId, "featureInstanceId": role.featureInstanceId, "featureCode": role.featureCode, "userCount": roleCounts.get(str(role.id), 0), "isSystemRole": role.isSystemRole, "scopeType": scopeType }) if not isSysAdmin: result = [r for r in result if r.get("mandateId") and str(r["mandateId"]) in adminMandateIds] return _handleFilterValuesRequest(result, column, pagination) except HTTPException: raise except Exception as e: logger.error(f"Error getting filter values for roles: {str(e)}") raise HTTPException(status_code=500, detail=str(e)) @router.post("/roles", response_model=Dict[str, Any]) @limiter.limit("30/minute") def create_role( request: Request, role: Role = Body(...), reqContext: RequestContext = Depends(getRequestContext) ) -> Dict[str, Any]: """ Create a new role. MULTI-TENANT: Admin-aware (SysAdmin sees all, MandateAdmin creates in own mandate). Request Body: - role: Role object to create Returns: - Created role dictionary """ try: isSysAdmin = reqContext.hasSysAdminRole adminMandateIds = [] if isSysAdmin else _getAdminMandateIds(reqContext) if not isSysAdmin and not adminMandateIds: raise HTTPException(status_code=403, detail=routeApiMsg("Admin role required")) # MandateAdmin: can only create roles in their own mandates if not isSysAdmin: if not role.mandateId or str(role.mandateId) not in adminMandateIds: raise HTTPException(status_code=403, detail=routeApiMsg("Access denied: can only create roles in your own mandates")) interface = getRootInterface() createdRole = interface.createRole(role) logger.info(f"Created role {createdRole.roleLabel} by admin {reqContext.user.id}") return { "id": createdRole.id, "roleLabel": createdRole.roleLabel, "description": createdRole.description, "mandateId": createdRole.mandateId, "featureInstanceId": createdRole.featureInstanceId, "featureCode": createdRole.featureCode, "isSystemRole": createdRole.isSystemRole } except HTTPException: raise except ValueError as e: raise HTTPException( status_code=400, detail=str(e) ) except Exception as e: logger.error(f"Error creating role: {str(e)}") raise HTTPException( status_code=500, detail=f"Failed to create role: {str(e)}" ) @router.get("/roles/{roleId}", response_model=Dict[str, Any]) @limiter.limit("60/minute") def get_role( request: Request, roleId: str = Path(..., description="Role ID"), reqContext: RequestContext = Depends(getRequestContext) ) -> Dict[str, Any]: """ Get a role by ID. MULTI-TENANT: Admin-aware (SysAdmin sees all, MandateAdmin sees own mandate's roles). Path Parameters: - roleId: Role ID Returns: - Role dictionary """ try: isSysAdmin = reqContext.hasSysAdminRole adminMandateIds = [] if isSysAdmin else _getAdminMandateIds(reqContext) if not isSysAdmin and not adminMandateIds: raise HTTPException(status_code=403, detail=routeApiMsg("Admin role required")) interface = getRootInterface() role = interface.getRole(roleId) if not role: raise HTTPException( status_code=404, detail=f"Role {roleId} not found" ) # MandateAdmin: verify role belongs to their mandates if not isSysAdmin: if not role.mandateId or str(role.mandateId) not in adminMandateIds: raise HTTPException(status_code=403, detail=routeApiMsg("Access denied: role not in your mandates")) return { "id": role.id, "roleLabel": role.roleLabel, "description": resolveText(role.description), "mandateId": role.mandateId, "featureInstanceId": role.featureInstanceId, "featureCode": role.featureCode, "isSystemRole": role.isSystemRole } except HTTPException: raise except Exception as e: logger.error(f"Error getting role: {str(e)}") raise HTTPException( status_code=500, detail=f"Failed to get role: {str(e)}" ) @router.put("/roles/{roleId}", response_model=Dict[str, Any]) @limiter.limit("30/minute") def update_role( request: Request, roleId: str = Path(..., description="Role ID"), role: Role = Body(...), reqContext: RequestContext = Depends(getRequestContext) ) -> Dict[str, Any]: """ Update an existing role. MULTI-TENANT: Admin-aware (SysAdmin sees all, MandateAdmin updates own mandate's roles, not template/system). Path Parameters: - roleId: Role ID Request Body: - role: Updated Role object Returns: - Updated role dictionary """ try: isSysAdmin = reqContext.hasSysAdminRole adminMandateIds = [] if isSysAdmin else _getAdminMandateIds(reqContext) if not isSysAdmin and not adminMandateIds: raise HTTPException(status_code=403, detail=routeApiMsg("Admin role required")) interface = getRootInterface() # MandateAdmin: verify role belongs to their mandates and is not a template/system role if not isSysAdmin: existingRole = interface.getRole(roleId) if not existingRole: raise HTTPException(status_code=404, detail=f"Role {roleId} not found") if existingRole.isSystemRole and not existingRole.mandateId: raise HTTPException(status_code=403, detail=routeApiMsg("Access denied: cannot modify template/system roles")) if not existingRole.mandateId or str(existingRole.mandateId) not in adminMandateIds: raise HTTPException(status_code=403, detail=routeApiMsg("Access denied: role not in your mandates")) updatedRole = interface.updateRole(roleId, role) logger.info(f"Updated role {roleId} by admin {reqContext.user.id}") return { "id": updatedRole.id, "roleLabel": updatedRole.roleLabel, "description": updatedRole.description, "mandateId": updatedRole.mandateId, "featureInstanceId": updatedRole.featureInstanceId, "featureCode": updatedRole.featureCode, "isSystemRole": updatedRole.isSystemRole } except HTTPException: raise except ValueError as e: raise HTTPException( status_code=400, detail=str(e) ) except Exception as e: logger.error(f"Error updating role: {str(e)}") raise HTTPException( status_code=500, detail=f"Failed to update role: {str(e)}" ) @router.delete("/roles/{roleId}", response_model=Dict[str, str]) @limiter.limit("30/minute") def delete_role( request: Request, roleId: str = Path(..., description="Role ID"), reqContext: RequestContext = Depends(getRequestContext) ) -> Dict[str, str]: """ Delete a role. MULTI-TENANT: Admin-aware (SysAdmin sees all, MandateAdmin deletes own mandate's roles, not template/system). Path Parameters: - roleId: Role ID Returns: - Success message """ try: isSysAdmin = reqContext.hasSysAdminRole adminMandateIds = [] if isSysAdmin else _getAdminMandateIds(reqContext) if not isSysAdmin and not adminMandateIds: raise HTTPException(status_code=403, detail=routeApiMsg("Admin role required")) interface = getRootInterface() # MandateAdmin: verify role belongs to their mandates and is not a template/system role if not isSysAdmin: existingRole = interface.getRole(roleId) if not existingRole: raise HTTPException(status_code=404, detail=f"Role {roleId} not found") if existingRole.isSystemRole and not existingRole.mandateId: raise HTTPException(status_code=403, detail=routeApiMsg("Access denied: cannot delete template/system roles")) if not existingRole.mandateId or str(existingRole.mandateId) not in adminMandateIds: raise HTTPException(status_code=403, detail=routeApiMsg("Access denied: role not in your mandates")) success = interface.deleteRole(roleId) if not success: raise HTTPException( status_code=404, detail=f"Role {roleId} not found" ) logger.info(f"Deleted role {roleId} by admin {reqContext.user.id}") return {"message": f"Role {roleId} deleted successfully"} except HTTPException: raise except ValueError as e: raise HTTPException( status_code=400, detail=str(e) ) except Exception as e: logger.error(f"Error deleting role: {str(e)}") raise HTTPException( status_code=500, detail=f"Failed to delete role: {str(e)}" ) # ============================================================================ # RBAC Catalog Endpoints # ============================================================================ @router.get("/catalog/objects", response_model=Dict[str, Any]) @limiter.limit("60/minute") def getCatalogObjects( request: Request, context: Optional[str] = Query(None, description="Filter by context (DATA, UI, RESOURCE)"), featureCode: Optional[str] = Query(None, description="Filter by feature code"), mandateId: Optional[str] = Query(None, description="Filter by mandate's active features"), reqContext: RequestContext = Depends(getRequestContext) # Available to all authenticated users ) -> Dict[str, Any]: """ Get available RBAC catalog objects. Returns all registered DATA, UI and RESOURCE objects that can be used in AccessRules. Query Parameters: - context: Optional filter by context type (DATA, UI, RESOURCE) - featureCode: Optional filter by feature (e.g., "trustee") - mandateId: Optional filter to only include objects from features active in this mandate Returns: - Dictionary with objects grouped by context type, each with: - objectKey: Dot-notation identifier (e.g., "data.feature.trustee.TrusteeContract") - label: Multilingual label - featureCode: Owning feature - meta: Additional metadata (table name, fields, etc.) Examples: - GET /api/rbac/catalog/objects → all objects - GET /api/rbac/catalog/objects?context=DATA → only DATA objects - GET /api/rbac/catalog/objects?featureCode=trustee → only trustee objects - GET /api/rbac/catalog/objects?mandateId=xxx → objects from mandate's active features """ try: from modules.security.rbacCatalog import getCatalogService catalog = getCatalogService() # If mandateId is provided, get active features for that mandate activeFeatures = None if mandateId: try: interface = getRootInterface() # Get all feature instances for this mandate using interface method instances = interface.getFeatureInstancesByMandate(mandateId, enabledOnly=True) activeFeatures = set(inst.featureCode for inst in instances) # Always include "system" feature activeFeatures.add("system") except Exception as e: logger.warning(f"Could not get active features for mandate {mandateId}: {e}") def _resolveLabels(objects: list) -> list: for obj in objects: resolved = resolveText(obj.get("label")) obj["label"] = resolved if resolved else f"[{obj.get('objectKey', '?')}]" return objects if context: # Single context filter try: accessContext = AccessRuleContext(context.upper()) except ValueError: raise HTTPException( status_code=400, detail=f"Invalid context '{context}'. Must be one of: DATA, UI, RESOURCE" ) if accessContext == AccessRuleContext.DATA: objects = catalog.getDataObjects(featureCode) elif accessContext == AccessRuleContext.UI: objects = catalog.getUiObjects(featureCode) else: objects = catalog.getResourceObjects(featureCode) # Filter by active features if mandateId was provided if activeFeatures: objects = [obj for obj in objects if obj.get("featureCode") in activeFeatures] return {context.upper(): _resolveLabels(objects)} else: # All contexts result = catalog.getAllCatalogObjects(featureCode) # Filter by active features if mandateId was provided if activeFeatures: for ctxKey in result: result[ctxKey] = [obj for obj in result[ctxKey] if obj.get("featureCode") in activeFeatures] for ctxKey in result: _resolveLabels(result[ctxKey]) return result except HTTPException: raise except Exception as e: logger.error(f"Error getting catalog objects: {str(e)}") raise HTTPException( status_code=500, detail=f"Failed to get catalog objects: {str(e)}" ) @router.get("/catalog/stats", response_model=Dict[str, Any]) @limiter.limit("60/minute") def getCatalogStats( request: Request, currentUser: User = Depends(requireSysAdminRole) ) -> Dict[str, Any]: """ Get statistics about the RBAC catalog. Returns: - Statistics about registered features, objects, and roles """ try: from modules.security.rbacCatalog import getCatalogService catalog = getCatalogService() return catalog.getCatalogStats() except Exception as e: logger.error(f"Error getting catalog stats: {str(e)}") raise HTTPException( status_code=500, detail=f"Failed to get catalog stats: {str(e)}" ) # ============================================================================= # CLEANUP: Remove duplicate AccessRules # ============================================================================= @router.post("/cleanup/duplicate-rules", response_model=dict) @limiter.limit("5/minute") def cleanup_duplicate_access_rules( request: Request, dryRun: bool = Query(True, description="If true, only report duplicates without deleting"), currentUser: User = Depends(requireSysAdminRole) ) -> dict: """ Find and remove duplicate AccessRules. Duplicates are rules with the same (roleId, context, item) signature. Only the first rule (oldest) is kept, all others are deleted. Query Parameters: - dryRun: If true (default), only report what would be deleted. Set to false to actually delete. Returns: - Summary with counts and details of duplicates found/removed """ try: rootInterface = getRootInterface() # Get ALL AccessRules from DB allRules = rootInterface.db.getRecordset(AccessRule) # Group by signature (roleId, context, item) rulesBySignature: Dict[tuple, list] = {} for rule in allRules: context = rule.get("context", "") # Normalize context enum value if hasattr(context, 'value'): context = context.value sig = (rule.get("roleId"), str(context), rule.get("item")) if sig not in rulesBySignature: rulesBySignature[sig] = [] rulesBySignature[sig].append(rule) # Find duplicates and collect IDs to delete duplicateGroups = [] idsToDelete = [] for sig, rules in rulesBySignature.items(): if len(rules) > 1: # Sort by creation time (keep oldest) rules.sort(key=lambda r: r.get("sysCreatedAt", 0)) keepRule = rules[0] deleteRules = rules[1:] duplicateGroups.append({ "roleId": sig[0], "context": sig[1], "item": sig[2] or "(global)", "totalCount": len(rules), "keepId": keepRule.get("id"), "deleteCount": len(deleteRules), "deleteIds": [r.get("id") for r in deleteRules] }) idsToDelete.extend([r.get("id") for r in deleteRules]) # Perform deletion if not dry run deletedCount = 0 if not dryRun and idsToDelete: for ruleId in idsToDelete: try: rootInterface.db.recordDelete(AccessRule, ruleId) deletedCount += 1 except Exception as e: logger.warning(f"Failed to delete rule {ruleId}: {e}") # ===================================================================== # Phase 2: Fix template role assignments # UserMandateRole should reference mandate-instance roles, not templates # ===================================================================== from modules.datamodels.datamodelMembership import UserMandate, UserMandateRole allUserMandateRoles = rootInterface.db.getRecordset(UserMandateRole) templateFixDetails = [] templateFixedCount = 0 for umr in allUserMandateRoles: roleId = umr.get("roleId") userMandateId = umr.get("userMandateId") umrId = umr.get("id") if not roleId or not userMandateId: continue # Check if assigned role is a template role = rootInterface.getRole(roleId) if not role or role.mandateId is not None: continue # Not a template role, OK if not role.isSystemRole: continue # Not a system template, skip # Template role assigned! Find the UserMandate to get the mandateId userMandateRecords = rootInterface.db.getRecordset( UserMandate, recordFilter={"id": userMandateId} ) if not userMandateRecords: continue mandateId = userMandateRecords[0].get("mandateId") if not mandateId: continue # Find the correct mandate-instance role mandateRoles = rootInterface.db.getRecordset( Role, recordFilter={"roleLabel": role.roleLabel, "mandateId": mandateId, "featureInstanceId": None} ) detail = { "userMandateRoleId": umrId, "userMandateId": userMandateId, "mandateId": mandateId, "templateRoleId": roleId, "templateRoleLabel": role.roleLabel, "action": "none" } if mandateRoles: instanceRoleId = mandateRoles[0].get("id") detail["instanceRoleId"] = instanceRoleId detail["action"] = "replace" if not dryRun else "would_replace" if not dryRun: try: rootInterface.db.recordModify(UserMandateRole, umrId, {"roleId": instanceRoleId}) templateFixedCount += 1 logger.info(f"Fixed template role assignment: {umrId} → {role.roleLabel} template → instance {instanceRoleId}") except Exception as e: detail["action"] = f"error: {e}" logger.warning(f"Failed to fix role assignment {umrId}: {e}") else: detail["action"] = "delete_inconsistent" if not dryRun else "would_delete_inconsistent" if not dryRun: try: rootInterface.db.recordDelete(UserMandateRole, umrId) templateFixedCount += 1 logger.info(f"Deleted inconsistent template role assignment: {umrId} (template '{role.roleLabel}' in mandate {mandateId}, no instance role found)") except Exception as e: detail["action"] = f"error: {e}" logger.warning(f"Failed to delete inconsistent assignment {umrId}: {e}") templateFixDetails.append(detail) result = { "dryRun": dryRun, "duplicateRules": { "totalRules": len(allRules), "uniqueSignatures": len(rulesBySignature), "duplicateGroups": len(duplicateGroups), "duplicateRulesToDelete": len(idsToDelete), "deletedCount": deletedCount, "details": duplicateGroups[:50] }, "templateRoleAssignments": { "totalUserMandateRoles": len(allUserMandateRoles), "invalidAssignments": len(templateFixDetails), "fixedCount": templateFixedCount, "details": templateFixDetails[:50] } } logger.info(f"RBAC cleanup: dryRun={dryRun}, " f"duplicates={len(duplicateGroups)}/{deletedCount} deleted, " f"templateFixes={len(templateFixDetails)}/{templateFixedCount} fixed") return result except Exception as e: logger.error(f"Error during AccessRule cleanup: {str(e)}") raise HTTPException( status_code=500, detail=f"Failed to cleanup duplicate rules: {str(e)}" )