""" Access Control Handler - Roles, Permissions, Policies Phase J.3: 12 tools - Roles: list, get, create, update, delete (5) - Permissions: list, get, create, update, delete, get_my (6) - Policies: list (1) Note: Directus v10+ uses policies for permissions. The 'policy' parameter is required in create_permission. """ import json from typing import Any from plugins.directus.client import DirectusClient def _parse_json_param(value: Any, param_name: str = "parameter") -> Any: """Parse a parameter that may be a JSON string or already a native type.""" if value is None: return None if isinstance(value, (dict, list)): return value if isinstance(value, str): stripped = value.strip() if stripped.startswith("{") or stripped.startswith("["): try: return json.loads(stripped) except json.JSONDecodeError as e: raise ValueError(f"Invalid JSON in '{param_name}': {e}") return value def get_tool_specifications() -> list[dict[str, Any]]: """Return tool specifications for ToolGenerator (12 tools)""" return [ # ===================== # ROLES (5) # ===================== { "name": "list_roles", "method_name": "list_roles", "description": "List all roles in Directus.", "schema": { "type": "object", "properties": { "filter": { "anyOf": [{"type": "object"}, {"type": "null"}], "description": "Filter object", }, "sort": { "anyOf": [{"type": "array", "items": {"type": "string"}}, {"type": "null"}], "description": "Sort fields", }, "limit": { "type": "integer", "description": "Maximum roles to return", "default": 100, }, }, "required": [], }, "scope": "read", }, { "name": "get_role", "method_name": "get_role", "description": "Get role details by ID.", "schema": { "type": "object", "properties": {"id": {"type": "string", "description": "Role UUID"}}, "required": ["id"], }, "scope": "read", }, { "name": "create_role", "method_name": "create_role", "description": "Create a new role.", "schema": { "type": "object", "properties": { "name": {"type": "string", "description": "Role name"}, "icon": { "type": "string", "description": "Material icon name", "default": "supervised_user_circle", }, "description": { "anyOf": [{"type": "string"}, {"type": "null"}], "description": "Role description", }, "admin_access": { "type": "boolean", "description": "Full admin access", "default": False, }, "app_access": { "type": "boolean", "description": "Access to admin app", "default": True, }, }, "required": ["name"], }, "scope": "admin", }, { "name": "update_role", "method_name": "update_role", "description": "Update role settings.", "schema": { "type": "object", "properties": { "id": {"type": "string", "description": "Role UUID"}, "data": { "type": "object", "description": "Fields to update (name, icon, description, admin_access, etc.)", }, }, "required": ["id", "data"], }, "scope": "admin", }, { "name": "delete_role", "method_name": "delete_role", "description": "Delete a role. Users with this role will have no role.", "schema": { "type": "object", "properties": {"id": {"type": "string", "description": "Role UUID to delete"}}, "required": ["id"], }, "scope": "admin", }, # ===================== # PERMISSIONS (6) # ===================== { "name": "list_permissions", "method_name": "list_permissions", "description": "List all permissions.", "schema": { "type": "object", "properties": { "filter": { "anyOf": [{"type": "object"}, {"type": "null"}], "description": 'Filter object (e.g., {"role": {"_eq": "role-uuid"}})', }, "limit": { "type": "integer", "description": "Maximum permissions to return", "default": 100, }, }, "required": [], }, "scope": "read", }, { "name": "get_permission", "method_name": "get_permission", "description": "Get permission details by ID.", "schema": { "type": "object", "properties": {"id": {"type": "string", "description": "Permission ID"}}, "required": ["id"], }, "scope": "read", }, { "name": "create_permission", "method_name": "create_permission", "description": "Create a new permission rule. NOTE: In Directus v10+, 'policy' is required.", "schema": { "type": "object", "properties": { "collection": {"type": "string", "description": "Collection name"}, "action": { "type": "string", "enum": ["create", "read", "update", "delete", "share"], "description": "Permission action", }, "policy": { "type": "string", "description": "Policy UUID (REQUIRED in Directus v10+). Use list_policies to get available policies.", }, "role": { "anyOf": [{"type": "string"}, {"type": "null"}], "description": "Role UUID (null for public) - deprecated in v10+, use policy instead", }, "permissions": { "anyOf": [{"type": "object"}, {"type": "null"}], "description": "Filter rules for this permission", }, "validation": { "anyOf": [{"type": "object"}, {"type": "null"}], "description": "Validation rules for create/update", }, "presets": { "anyOf": [{"type": "object"}, {"type": "null"}], "description": "Default values for create", }, "fields": { "anyOf": [{"type": "array", "items": {"type": "string"}}, {"type": "null"}], "description": "Allowed fields (* for all)", }, }, "required": ["collection", "action", "policy"], }, "scope": "admin", }, { "name": "update_permission", "method_name": "update_permission", "description": "Update a permission rule.", "schema": { "type": "object", "properties": { "id": {"type": "string", "description": "Permission ID"}, "data": {"type": "object", "description": "Fields to update"}, }, "required": ["id", "data"], }, "scope": "admin", }, { "name": "delete_permission", "method_name": "delete_permission", "description": "Delete a permission rule.", "schema": { "type": "object", "properties": {"id": {"type": "string", "description": "Permission ID to delete"}}, "required": ["id"], }, "scope": "admin", }, { "name": "get_my_permissions", "method_name": "get_my_permissions", "description": "Get the current user's effective permissions.", "schema": {"type": "object", "properties": {}, "required": []}, "scope": "read", }, # ===================== # POLICIES (1) # ===================== { "name": "list_policies", "method_name": "list_policies", "description": "List all access policies.", "schema": { "type": "object", "properties": { "filter": { "anyOf": [{"type": "object"}, {"type": "null"}], "description": "Filter object", }, "limit": { "type": "integer", "description": "Maximum policies to return", "default": 100, }, }, "required": [], }, "scope": "read", }, ] # ===================== # HANDLER FUNCTIONS # ===================== async def list_roles( client: DirectusClient, filter: dict | None = None, sort: list[str] | None = None, limit: int = 100, ) -> str: """List roles.""" try: # Parse JSON string parameters parsed_filter = _parse_json_param(filter, "filter") parsed_sort = _parse_json_param(sort, "sort") result = await client.list_roles(filter=parsed_filter, sort=parsed_sort, limit=limit) roles = result.get("data", []) return json.dumps( {"success": True, "total": len(roles), "roles": roles}, indent=2, ensure_ascii=False ) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def get_role(client: DirectusClient, id: str) -> str: """Get role by ID.""" try: result = await client.get_role(id) return json.dumps( {"success": True, "role": result.get("data")}, indent=2, ensure_ascii=False ) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def create_role( client: DirectusClient, name: str, icon: str = "supervised_user_circle", description: str | None = None, admin_access: bool = False, app_access: bool = True, ) -> str: """Create a new role.""" try: result = await client.create_role( name=name, icon=icon, description=description, admin_access=admin_access, app_access=app_access, ) return json.dumps( {"success": True, "message": f"Role '{name}' created", "role": result.get("data")}, indent=2, ensure_ascii=False, ) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def update_role(client: DirectusClient, id: str, data: dict[str, Any]) -> str: """Update role.""" try: # Parse JSON string parameter parsed_data = _parse_json_param(data, "data") result = await client.update_role(id, parsed_data) return json.dumps( {"success": True, "message": "Role updated", "role": result.get("data")}, indent=2, ensure_ascii=False, ) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def delete_role(client: DirectusClient, id: str) -> str: """Delete a role.""" try: await client.delete_role(id) return json.dumps({"success": True, "message": f"Role {id} deleted"}, indent=2) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def list_permissions( client: DirectusClient, filter: dict | None = None, limit: int = 100 ) -> str: """List permissions.""" try: # Parse JSON string parameter parsed_filter = _parse_json_param(filter, "filter") result = await client.list_permissions(filter=parsed_filter, limit=limit) permissions = result.get("data", []) return json.dumps( {"success": True, "total": len(permissions), "permissions": permissions}, indent=2, ensure_ascii=False, ) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def get_permission(client: DirectusClient, id: str) -> str: """Get permission by ID.""" try: result = await client.get_permission(id) return json.dumps( {"success": True, "permission": result.get("data")}, indent=2, ensure_ascii=False ) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def create_permission( client: DirectusClient, collection: str, action: str, policy: str, role: str | None = None, permissions: dict | None = None, validation: dict | None = None, presets: dict | None = None, fields: list[str] | None = None, ) -> str: """Create a new permission. Requires 'policy' in Directus v10+.""" try: # Parse JSON string parameters parsed_permissions = _parse_json_param(permissions, "permissions") parsed_validation = _parse_json_param(validation, "validation") parsed_presets = _parse_json_param(presets, "presets") parsed_fields = _parse_json_param(fields, "fields") result = await client.create_permission( collection=collection, action=action, policy=policy, role=role, permissions=parsed_permissions, validation=parsed_validation, presets=parsed_presets, fields=parsed_fields, ) return json.dumps( { "success": True, "message": f"Permission created for {collection}.{action}", "permission": result.get("data"), }, indent=2, ensure_ascii=False, ) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def update_permission(client: DirectusClient, id: str, data: dict[str, Any]) -> str: """Update permission.""" try: # Parse JSON string parameter parsed_data = _parse_json_param(data, "data") result = await client.update_permission(id, parsed_data) return json.dumps( {"success": True, "message": "Permission updated", "permission": result.get("data")}, indent=2, ensure_ascii=False, ) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def delete_permission(client: DirectusClient, id: str) -> str: """Delete a permission.""" try: await client.delete_permission(id) return json.dumps({"success": True, "message": f"Permission {id} deleted"}, indent=2) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def get_my_permissions(client: DirectusClient) -> str: """Get current user's permissions.""" try: result = await client.get_my_permissions() return json.dumps( {"success": True, "permissions": result.get("data")}, indent=2, ensure_ascii=False ) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def list_policies( client: DirectusClient, filter: dict | None = None, limit: int = 100 ) -> str: """List policies.""" try: # Parse JSON string parameter parsed_filter = _parse_json_param(filter, "filter") result = await client.list_policies(filter=parsed_filter, limit=limit) policies = result.get("data", []) return json.dumps( {"success": True, "total": len(policies), "policies": policies}, indent=2, ensure_ascii=False, ) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2)