feat: add authz dependency injections

.vscode/launch.json

@@ -34,7 +34,10 @@
             "module": "blueapi",
             "args": "--config ${input:config_path} serve",
             "env": {
-                "OTLP_EXPORT_ENABLED": "false"
+                "OTLP_EXPORT_ENABLED": "false",
+                "EPICS_CA_NAME_SERVERS": "127.0.0.1:9064",
+                "EPICS_PVA_NAME_SERVERS": "127.0.0.1:9075",
+                "EPICS_CA_ADDR_LIST": "127.0.0.1:9064"
             },
         },
         {

src/blueapi/config.py

@@ -303,7 +303,7 @@ class ApplicationConfig(BlueapiBaseModel):
     """
 
     #: API version to publish in OpenAPI schema
-    REST_API_VERSION: ClassVar[str] = "1.3.0"
+    REST_API_VERSION: ClassVar[str] = "1.3.1"
 
     LICENSE_INFO: ClassVar[dict[str, str]] = {
         "name": "Apache 2.0",

src/blueapi/service/authentication.py

@@ -272,3 +272,12 @@ def get_access_token(self):
     def sync_auth_flow(self, request):
         request.headers["Authorization"] = f"Bearer {self.get_access_token()}"
         yield request
+
+
+class OPAClient:  # placeholder until https://jira.diamond.ac.uk/browse/ACQP-550 is done
+    def do_some_checks(self, task_request) -> bool:
+        return True
+
+
+def get_opa_client() -> OPAClient:  # placeholder
+    return OPAClient()

src/blueapi/service/main.py

@@ -37,6 +37,7 @@
 from blueapi import __version__
 from blueapi.config import ApplicationConfig, OIDCConfig, Tag
 from blueapi.service import interface
+from blueapi.service.authentication import OPAClient, get_opa_client
 from blueapi.worker import TrackableTask, WorkerState
 from blueapi.worker.event import TaskStatusEnum
 
@@ -166,6 +167,51 @@ def inner(request: Request, access_token: str = Depends(oauth_scheme)):
 TRACER = get_tracer("interface")
 
 
+def submit_permission(
+    opa: Annotated[OPAClient, Depends(get_opa_client)],
+    task_request: TaskRequest,
+):
+    allowed = opa.do_some_checks(task_request)
+
+    if not allowed:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
+
+
+def access_task_permission(
+    request: Request,
+    task_id: str,
+    runner: Annotated[WorkerDispatcher, Depends(_runner)],
+):
+    access_token: dict[str, Any] | None = getattr(
+        request.state, "decoded_access_token", None
+    )
+    try:
+        task = runner.run(interface.get_task_by_id, task_id)
+    except KeyError:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED) from None
+
+    if (
+        access_token
+        and task
+        and access_token.get("fedid") != task.task.metadata.get("user")
+    ):
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
+
+
+# start_task_permission is used when there is WorkerTask
+def start_task_permission(
+    request: Request,
+    task: WorkerTask,
+    runner: Annotated[WorkerDispatcher, Depends(_runner)],
+):
+    if not task.task_id:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail="No task id provided",
+        )
+    access_task_permission(request, task.task_id, runner)
+
+
 async def on_key_error_404(_: Request, __: Exception):
     return JSONResponse(
         status_code=status.HTTP_404_NOT_FOUND,
@@ -291,6 +337,7 @@ def submit_task(
     request: Request,
     response: Response,
     task_request: Annotated[TaskRequest, Body(..., examples=[example_task_request])],
+    _: Annotated[None, Depends(submit_permission)],
     runner: Annotated[WorkerDispatcher, Depends(_runner)],
 ) -> TaskResponse:
     """Submit a task to the worker."""
@@ -336,6 +383,7 @@ def submit_task(
 @start_as_current_span(TRACER, "task_id")
 def delete_submitted_task(
     task_id: str,
+    _: Annotated[None, Depends(access_task_permission)],
     runner: Annotated[WorkerDispatcher, Depends(_runner)],
 ) -> TaskResponse:
     return TaskResponse(task_id=runner.run(interface.clear_task, task_id))
@@ -353,6 +401,7 @@ def validate_task_status(v: str) -> TaskStatusEnum:
 @secure_router.get("/tasks", status_code=status.HTTP_200_OK, tags=[Tag.TASK])
 @start_as_current_span(TRACER)
 def get_tasks(
+    request: Request,
     runner: Annotated[WorkerDispatcher, Depends(_runner)],
     task_status: str | SkipJsonSchema[None] = None,
 ) -> TasksListResponse:
@@ -373,6 +422,15 @@ def get_tasks(
         tasks = runner.run(interface.get_tasks_by_status, desired_status)
     else:
         tasks = runner.run(interface.get_tasks)
+
+    access_token: dict[str, Any] | None = getattr(
+        request.state, "decoded_access_token", None
+    )
+    user = access_token.get("fedid") if access_token else None
+
+    if user:
+        tasks = [t for t in tasks if t.task.metadata.get("user") == user]
+
     return TasksListResponse(tasks=tasks)
 
 
@@ -390,6 +448,7 @@ def get_tasks(
 def set_active_task(
     request: Request,
     task: WorkerTask,
+    _: Annotated[None, Depends(start_task_permission)],
     runner: Annotated[WorkerDispatcher, Depends(_runner)],
 ) -> WorkerTask:
     """Set a task to active status, the worker should begin it as soon as possible.
@@ -420,6 +479,7 @@ def get_passthrough_headers(request: Request) -> dict[str, str]:
 @start_as_current_span(TRACER, "task_id")
 def get_task(
     task_id: str,
+    _: Annotated[None, Depends(access_task_permission)],
     runner: Annotated[WorkerDispatcher, Depends(_runner)],
 ) -> TrackableTask:
     """Retrieve a task"""
@@ -497,6 +557,7 @@ def get_state(runner: Annotated[WorkerDispatcher, Depends(_runner)]) -> WorkerSt
 def set_state(
     state_change_request: StateChangeRequest,
     response: Response,
+    _: Annotated[None, Depends(access_task_permission)],
     runner: Annotated[WorkerDispatcher, Depends(_runner)],
 ) -> WorkerState:
     """
@@ -528,6 +589,9 @@ def set_state(
         elif new_state == WorkerState.RUNNING:
             runner.run(interface.resume_worker)
         elif new_state in {WorkerState.ABORTING, WorkerState.STOPPING}:
+            # active = runner.run(interface.get_active_task)
+            # if active.task.metadata.get("user"):
+
             try:
                 runner.run(
                     interface.cancel_active_task,