""" Access control module for Gateway interface. Handles user access management and permission checks. """ from typing import Dict, Any, List, Optional class GatewayAccess: """ Access control class for Gateway interface. Handles user access management and permission checks. """ def __init__(self, currentUser: Dict[str, Any], db): """Initialize with user context.""" self.currentUser = currentUser self.mandateId = currentUser.get("mandateId") self.userId = currentUser.get("id") if not self.mandateId or not self.userId: raise ValueError("Invalid user context: mandateId and userId are required") self.db = db 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.currentUser.get("privilege", "user") filtered_records = [] # 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 # Users only see records they own within their mandate 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 == "mandates": record["_hideView"] = False # Everyone can view record["_hideEdit"] = not self.canModify("mandates", record_id) record["_hideDelete"] = not self.canModify("mandates", record_id) elif table == "users": record["_hideView"] = False # Everyone can view record["_hideEdit"] = not self.canModify("users", record_id) record["_hideDelete"] = not self.canModify("users", record_id) 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.currentUser.get("privilege", "user") # System admins can modify anything if userPrivilege == "sysadmin": return True # Check specific record permissions if recordId is not None: # Get the record to check ownership records = self.db.getRecordset(table, recordFilter={"id": recordId}) if not records: return False record = records[0] # Admins can modify anything in their mandate if userPrivilege == "admin" and record.get("mandateId","-") == self.mandateId: # Exception: Can't modify Root mandate unless you are a sysadmin if table == "mandates" and record.get("initialid") and userPrivilege != "sysadmin": return False return True # 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 table modify permission (e.g., create) # Admins can create anything in their mandate if userPrivilege == "admin": return True # Regular users can create most entities if table == "mandates": return False # Regular users can't create mandates return True