feat: add resource-level authorization for OPC

- Add OPCAuthz module with fine-grained access control - Users can only view/modify tasks they created or are assigned to - Admins have full access to all resources - Members can trigger PM agent, only admins can trigger CTO/COO - Only admins can approve CxO-level agent executions - Track task creator in metadata for authorization - Add metadata column with default '{}' to tasks table - Filter task and execution lists based on user permissions - Add 10 integration tests for authorization logic
2026-04-08 01:28:12 +08:00
parent 69cc62a931
commit 4fa84fce8a
5 changed files with 474 additions and 9 deletions
@@ -0,0 +1,184 @@
+"""
+OPC Resource-Level Authorization
+
+Implements fine-grained access control for OPC tasks, agents, and executions.
+Page-level RBAC (require_page("opc")) controls who can access OPC pages.
+This module controls which specific resources users can view/modify.
+"""
+
+from typing import Any
+
+from fastapi import HTTPException
+from rbac import User
+
+
+class OPCAuthz:
+    """Authorization logic for OPC resources"""
+
+    @staticmethod
+    def can_view_task(user: User, task: dict[str, Any]) -> bool:
+        """Check if user can view a task"""
+        # Admins see everything
+        if user.role == "admin":
+            return True
+
+        # Users can see tasks they created or are assigned to
+        if task.get("assigned_to") == user.username:
+            return True
+
+        # Check metadata for created_by (will be added in migration)
+        metadata = task.get("metadata", {})
+        if isinstance(metadata, dict) and metadata.get("created_by") == user.username:
+            return True
+
+        # Viewers can see all tasks (read-only enforced elsewhere)
+        if user.role == "viewer":
+            return True
+
+        # Members can see unassigned tasks (parking lot)
+        if task.get("status") == "parking_lot" and not task.get("assigned_to"):
+            return True
+
+        return False
+
+    @staticmethod
+    def can_modify_task(user: User, task: dict[str, Any]) -> bool:
+        """Check if user can modify a task"""
+        # Admins can modify everything
+        if user.role == "admin":
+            return True
+
+        # Viewers cannot modify anything
+        if user.role == "viewer":
+            return False
+
+        # Users can modify tasks they created or are assigned to
+        if task.get("assigned_to") == user.username:
+            return True
+
+        metadata = task.get("metadata", {})
+        if isinstance(metadata, dict) and metadata.get("created_by") == user.username:
+            return True
+
+        # Members can modify unassigned tasks
+        if user.role == "member" and task.get("status") == "parking_lot" and not task.get("assigned_to"):
+            return True
+
+        return False
+
+    @staticmethod
+    def can_delete_task(user: User, task: dict[str, Any]) -> bool:
+        """Check if user can delete a task"""
+        # Only admins and task creators can delete
+        if user.role == "admin":
+            return True
+
+        metadata = task.get("metadata", {})
+        if isinstance(metadata, dict) and metadata.get("created_by") == user.username:
+            return True
+
+        return False
+
+    @staticmethod
+    def can_trigger_agent(user: User, agent_id: str) -> bool:
+        """Check if user can manually trigger an agent"""
+        # Admins can trigger any agent
+        if user.role == "admin":
+            return True
+
+        # Viewers cannot trigger agents
+        if user.role == "viewer":
+            return False
+
+        # Members can trigger PM agent (auto-approval)
+        # CTO/COO agents require admin (CxO approval level)
+        if agent_id == "pm":
+            return True
+
+        return False
+
+    @staticmethod
+    def can_approve_execution(user: User, execution: dict[str, Any]) -> bool:
+        """Check if user can approve an agent execution"""
+        # Only admins can approve CxO-level executions
+        if user.role == "admin":
+            return True
+
+        return False
+
+    @staticmethod
+    def can_view_agent_execution(user: User, execution: dict[str, Any], task: dict[str, Any] | None = None) -> bool:
+        """Check if user can view an agent execution"""
+        # Admins see everything
+        if user.role == "admin":
+            return True
+
+        # Users can see executions for tasks they have access to
+        if task and OPCAuthz.can_view_task(user, task):
+            return True
+
+        return False
+
+    @staticmethod
+    def filter_tasks(user: User, tasks: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        """Filter tasks to only those user can view"""
+        if user.role == "admin":
+            return tasks
+
+        return [task for task in tasks if OPCAuthz.can_view_task(user, task)]
+
+    @staticmethod
+    def filter_executions(
+        user: User, executions: list[dict[str, Any]], tasks_by_id: dict[int, dict[str, Any]] | None = None
+    ) -> list[dict[str, Any]]:
+        """Filter executions to only those user can view"""
+        if user.role == "admin":
+            return executions
+
+        if tasks_by_id is None:
+            # Without task context, only show executions for tasks user is assigned to
+            # This is a conservative filter - caller should provide tasks_by_id for accurate filtering
+            return []
+
+        return [
+            execution
+            for execution in executions
+            if execution.get("task_id") in tasks_by_id
+            and OPCAuthz.can_view_task(user, tasks_by_id[execution["task_id"]])
+        ]
+
+    @staticmethod
+    def require_task_view(user: User, task: dict[str, Any]):
+        """Raise 403 if user cannot view task"""
+        if not OPCAuthz.can_view_task(user, task):
+            raise HTTPException(status_code=403, detail="Access denied")
+
+    @staticmethod
+    def require_task_modify(user: User, task: dict[str, Any]):
+        """Raise 403 if user cannot modify task"""
+        if not OPCAuthz.can_modify_task(user, task):
+            raise HTTPException(status_code=403, detail="Access denied")
+
+    @staticmethod
+    def require_task_delete(user: User, task: dict[str, Any]):
+        """Raise 403 if user cannot delete task"""
+        if not OPCAuthz.can_delete_task(user, task):
+            raise HTTPException(status_code=403, detail="Access denied")
+
+    @staticmethod
+    def require_agent_trigger(user: User, agent_id: str):
+        """Raise 403 if user cannot trigger agent"""
+        if not OPCAuthz.can_trigger_agent(user, agent_id):
+            raise HTTPException(status_code=403, detail="Access denied")
+
+    @staticmethod
+    def require_execution_approve(user: User, execution: dict[str, Any]):
+        """Raise 403 if user cannot approve execution"""
+        if not OPCAuthz.can_approve_execution(user, execution):
+            raise HTTPException(status_code=403, detail="Admin approval required")
+
+    @staticmethod
+    def require_execution_view(user: User, execution: dict[str, Any], task: dict[str, Any] | None = None):
+        """Raise 403 if user cannot view execution"""
+        if not OPCAuthz.can_view_agent_execution(user, execution, task):
+            raise HTTPException(status_code=403, detail="Access denied")