Gan, Jimmy
4fa84fce8a
- 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
185 lines
6.4 KiB
Python
185 lines
6.4 KiB
Python
"""
|
|
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")
|