""" Access control module for Management interface. Handles user access management and permission checks. """ import logging from typing import Dict, Any, List, Optional from modules.interfaces.interfaceAppModel import User # Configure logger logger = logging.getLogger(__name__) class ComponentAccess: """ Access control class for Management interface. Handles user access management and permission checks. """ def __init__(self, currentUser: User, db): """Initialize with user context.""" self.currentUser = currentUser self.userId = currentUser.id self.mandateId = currentUser.mandateId self.privilege = currentUser.privilege self.db = db def getInitialUserid(self): return "----" # return self.db.getInitialUserId() --> to get from AdminDB ! def canModifyAttribute(self, table: str, attribute: str) -> bool: """ Checks if the current user can modify a specific attribute in a table. Args: table: Name of the table attribute: Name of the attribute Returns: Boolean indicating permission """ userPrivilege = self.privilege # Special case for mandateId in prompts table if table == "prompts" and attribute == "mandateId": return userPrivilege == "sysadmin" return True def uam(self, table: str, recordset: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """ Unified user access management function that filters data based on user privileges and adds access control attributes. Args: table: Name of the table recordset: Recordset to filter based on access rules Returns: Filtered recordset with access control attributes """ userPrivilege = self.privilege filtered_records = [] initialid = self.getInitialUserid() # Apply filtering based on privilege if userPrivilege == "sysadmin": filtered_records = recordset # System admins see all records elif userPrivilege == "admin": # Admins see records in their mandate filtered_records = [r for r in recordset if r.get("mandateId") == self.mandateId] else: # Regular users # For prompts, users can see all prompts from their mandate if table == "prompts": filtered_records = [r for r in recordset if r.get("mandateId") == self.mandateId] elif table == "users": # For users table, users can only see their own record filtered_records = [r for r in recordset if r.get("id") == self.userId] else: # Users see only their records for other tables filtered_records = [ r for r in recordset if r.get("mandateId") == self.mandateId and r.get("_createdBy") == self.userId ] # Add access control attributes to each record for record in filtered_records: record_id = record.get("id") # Set access control flags based on user permissions if table == "prompts": record["_hideView"] = False # Everyone can view record["_hideEdit"] = not self.canModify("prompts", record_id) record["_hideDelete"] = not self.canModify("prompts", record_id) # Add attribute-level permissions for mandateId if "mandateId" in record: record["_hideEdit_mandateId"] = not self.canModifyAttribute("prompts", "mandateId") elif table == "files": record["_hideView"] = False # Everyone can view record["_hideEdit"] = not self.canModify("files", record_id) record["_hideDelete"] = not self.canModify("files", record_id) record["_hideDownload"] = not self.canModify("files", record_id) elif table == "workflows": record["_hideView"] = False # Everyone can view record["_hideEdit"] = not self.canModify("workflows", record_id) record["_hideDelete"] = not self.canModify("workflows", record_id) elif table == "workflowMessages": record["_hideView"] = False # Everyone can view record["_hideEdit"] = not self.canModify("workflows", record.get("workflowId")) record["_hideDelete"] = not self.canModify("workflows", record.get("workflowId")) elif table == "workflowLogs": record["_hideView"] = False # Everyone can view record["_hideEdit"] = not self.canModify("workflows", record.get("workflowId")) record["_hideDelete"] = not self.canModify("workflows", record.get("workflowId")) elif table == "users": # For users table, users can only modify their own connections record["_hideView"] = False record["_hideEdit"] = record_id != self.userId record["_hideDelete"] = record_id != self.userId # Add connection-specific permissions if "connections" in record: for conn in record["connections"]: conn["_hideEdit"] = record_id != self.userId conn["_hideDelete"] = record_id != self.userId else: # Default access control for other tables record["_hideView"] = False record["_hideEdit"] = not self.canModify(table, record_id) record["_hideDelete"] = not self.canModify(table, record_id) return filtered_records def canModify(self, table: str, recordId: Optional[int] = None) -> bool: """ Checks if the current user can modify (create/update/delete) records in a table. Args: table: Name of the table recordId: Optional record ID for specific record check Returns: Boolean indicating permission """ userPrivilege = self.privilege # System admins can modify anything if userPrivilege == "sysadmin": return True # For regular users and admins, check specific cases if recordId is not None: # Get the record to check ownership records: List[Dict[str, Any]] = self.db.getRecordset(table, recordFilter={"id": recordId}) if not records: return False record = records[0] # Special case for users table - users can modify their own connections if table == "users": if record.get("id") == self.userId: return True return False # Admins can modify anything in their mandate, if mandate is specified for a record if userPrivilege == "admin" and record.get("mandateId","-") == self.mandateId: return True # Regular users can only modify their own records if (record.get("mandateId","-") == self.mandateId and record.get("_createdBy") == self.userId): return True return False else: # For general modification permission (e.g., create) # Admins can create anything in their mandate if userPrivilege == "admin": return True # Regular users can create in most tables return True