add auth to tasks endpoint (apache#47684)

Author: vatsrahul1001

airflow/api_fastapi/core_api/openapi/v1-generated.yaml

@@ -6394,6 +6394,8 @@ paths:
       summary: Get Tasks
       description: Get tasks for DAG.
       operationId: get_tasks
+      security:
+      - OAuth2PasswordBearer: []
       parameters:
       - name: dag_id
         in: path
@@ -6452,6 +6454,8 @@ paths:
       summary: Get Task
       description: Get simplified representation of a task.
       operationId: get_task
+      security:
+      - OAuth2PasswordBearer: []
       parameters:
       - name: dag_id
         in: path

airflow/api_fastapi/core_api/routes/public/tasks.py

@@ -20,11 +20,13 @@
 from operator import attrgetter
 from typing import cast
 
-from fastapi import HTTPException, Request, status
+from fastapi import Depends, HTTPException, Request, status
 
+from airflow.api_fastapi.auth.managers.models.resource_details import DagAccessEntity
 from airflow.api_fastapi.common.router import AirflowRouter
 from airflow.api_fastapi.core_api.datamodels.tasks import TaskCollectionResponse, TaskResponse
 from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.security import requires_access_dag
 from airflow.exceptions import TaskNotFound
 from airflow.models import DAG
 
@@ -39,6 +41,7 @@
             status.HTTP_404_NOT_FOUND,
         ]
     ),
+    dependencies=[Depends(requires_access_dag(method="GET", access_entity=DagAccessEntity.TASK))],
 )
 def get_tasks(
     dag_id: str,
@@ -67,6 +70,7 @@ def get_tasks(
             status.HTTP_404_NOT_FOUND,
         ]
     ),
+    dependencies=[Depends(requires_access_dag(method="GET", access_entity=DagAccessEntity.TASK))],
 )
 def get_task(dag_id: str, task_id, request: Request) -> TaskResponse:
     """Get simplified representation of a task."""

tests/api_fastapi/core_api/routes/public/test_tasks.py

@@ -297,6 +297,14 @@ def test_should_respond_404_when_dag_not_found(self, test_client):
         )
         assert response.status_code == 404
 
+    def test_should_respond_401(self, unauthenticated_test_client):
+        response = unauthenticated_test_client.get(f"{self.api_prefix}/{self.dag_id}/tasks/{self.task_id}")
+        assert response.status_code == 401
+
+    def test_should_respond_403(self, unauthorized_test_client):
+        response = unauthorized_test_client.get(f"{self.api_prefix}/{self.dag_id}/tasks/{self.task_id}")
+        assert response.status_code == 403
+
 
 class TestGetTasks(TestTaskEndpoint):
     def test_should_respond_200(self, test_client):
@@ -540,3 +548,11 @@ def test_should_respond_404(self, test_client):
         dag_id = "xxxx_not_existing"
         response = test_client.get(f"{self.api_prefix}/{dag_id}/tasks")
         assert response.status_code == 404
+
+    def test_should_respond_401(self, unauthenticated_test_client):
+        response = unauthenticated_test_client.get(f"{self.api_prefix}/{self.dag_id}/tasks")
+        assert response.status_code == 401
+
+    def test_should_respond_403(self, unauthorized_test_client):
+        response = unauthorized_test_client.get(f"{self.api_prefix}/{self.dag_id}/tasks")
+        assert response.status_code == 403