From c3cd25865a142586615237ddda148c242f726650 Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Sat, 1 Aug 2026 05:18:03 +0200 Subject: [PATCH 1/5] Resolve the Dag's team when authorizing a Dag found by lookup Two authorization checks build DagDetails(id=dag_id) without team_name: materialize_asset, where the Dag is resolved from the asset, and the XCom-specific check in wait_dag_run_until_finished. Every other call site passes the team, resolved with DagModel.get_team_name. A team-aware auth manager distinguishes a team-scoped Dag from a global one by that field, so omitting it asks about a differently-scoped resource than the one being acted on. In wait_dag_run_until_finished the route dependency already resolves the team for its RUN check, so the two checks in the same handler disagreed. Resolve the team at both sites, reusing the request session. --- .../core_api/routes/public/assets.py | 6 ++++- .../core_api/routes/public/dag_run.py | 5 +++- .../core_api/routes/public/test_assets.py | 26 +++++++++++++++++++ 3 files changed, 35 insertions(+), 2 deletions(-) diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py index 0a6818648cd68..830d4edaeedac 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py @@ -86,6 +86,7 @@ AssetWatcherModel, TaskOutletAssetReference, ) +from airflow.models.dag import DagModel from airflow.models.dag_version import DagVersion from airflow.typing_compat import Unpack from airflow.utils.state import DagRunState @@ -459,7 +460,10 @@ def materialize_asset( if not get_auth_manager().is_authorized_dag( method="POST", access_entity=DagAccessEntity.RUN, - details=DagDetails(id=dag_id), + # The Dag is resolved from the asset here rather than named by the caller, so its team has + # to be looked up too. A team-aware auth manager distinguishes a team-scoped Dag from a + # global one by this field, so leaving it None asks the wrong question. + details=DagDetails(id=dag_id, team_name=DagModel.get_team_name(dag_id, session=session)), user=user, ): raise HTTPException( diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py index 1b857df06ce57..e5d06587cb550 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dag_run.py @@ -869,7 +869,10 @@ def wait_dag_run_until_finished( if not get_auth_manager().is_authorized_dag( method="GET", access_entity=DagAccessEntity.XCOM, - details=DagDetails(id=dag_id), + # The route dependency above already authorizes RUN access with the Dag's team resolved; + # this second, XCom-specific check has to resolve it the same way, or the two checks ask + # a team-aware auth manager about differently-scoped resources. + details=DagDetails(id=dag_id, team_name=DagModel.get_team_name(dag_id, session=session)), user=user, ): if result_task_ids: diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py index fcf411628723a..a007e5cac30aa 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py @@ -2151,6 +2151,32 @@ def test_should_respond_200_with_partition_date_for_partitioned_dag( assert dag_run.partition_key == "2025-06-01T00:00:00" assert dag_run.partition_date == timezone.datetime(2025, 6, 1) + @pytest.mark.parametrize("team_name", ["team_b", None]) + def test_authorizes_against_the_dags_team(self, test_client, session, team_name): + """The Dag is resolved from the asset, so its team must be resolved and passed too. + + A team-aware auth manager distinguishes a team-scoped Dag from a global one by this field, + so leaving it ``None`` asks about a differently-scoped resource than the one being acted on. + """ + recorded = [] + + auth_manager = mock.Mock() + auth_manager.is_authorized_dag.side_effect = lambda **kw: recorded.append(kw) or True + + with ( + mock.patch( + "airflow.api_fastapi.core_api.routes.public.assets.get_auth_manager", + return_value=auth_manager, + ), + mock.patch.object(DagModel, "get_team_name", return_value=team_name), + ): + test_client.post("/assets/1/materialize") + + assert recorded, "the authorization check did not run" + details = recorded[0]["details"] + assert details.id == self.DAG_ASSET1_ID + assert details.team_name == team_name + class TestGetAssetQueuedEvents(TestQueuedEventEndpoint): @pytest.mark.usefixtures("time_freezer") From 9ebd025b85ea9a648bfe27ac926c2b27b05d7dff Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Tue, 4 Aug 2026 01:11:55 +0200 Subject: [PATCH 2/5] Cover the XCom authorization check with a team-scoped Dag The existing wait-endpoint test uses a Dag with no team, where the resolved and unresolved forms are indistinguishable, so nothing caught the second check asking about a differently-scoped resource than the route dependency did. --- .../core_api/routes/public/test_assets.py | 11 +++++-- .../core_api/routes/public/test_dag_run.py | 29 +++++++++++++++++++ 2 files changed, 37 insertions(+), 3 deletions(-) diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py index a007e5cac30aa..3e3b299620251 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py @@ -25,6 +25,7 @@ from sqlalchemy import delete, func, select, update from airflow._shared.timezones import timezone +from airflow.api_fastapi.auth.managers.base_auth_manager import BaseAuthManager from airflow.api_fastapi.auth.managers.models.resource_details import DagAccessEntity, DagDetails from airflow.models import DagModel from airflow.models.asset import ( @@ -2160,7 +2161,7 @@ def test_authorizes_against_the_dags_team(self, test_client, session, team_name) """ recorded = [] - auth_manager = mock.Mock() + auth_manager = mock.Mock(spec=BaseAuthManager) auth_manager.is_authorized_dag.side_effect = lambda **kw: recorded.append(kw) or True with ( @@ -2168,14 +2169,18 @@ def test_authorizes_against_the_dags_team(self, test_client, session, team_name) "airflow.api_fastapi.core_api.routes.public.assets.get_auth_manager", return_value=auth_manager, ), - mock.patch.object(DagModel, "get_team_name", return_value=team_name), + mock.patch.object( + DagModel, "get_team_name", return_value=team_name, autospec=True + ) as mock_get_team_name, ): test_client.post("/assets/1/materialize") - assert recorded, "the authorization check did not run" + assert len(recorded) == 1, "expected exactly one authorization check" details = recorded[0]["details"] assert details.id == self.DAG_ASSET1_ID assert details.team_name == team_name + # resolved for the Dag the asset led to, not for some other Dag + mock_get_team_name.assert_called_once_with(self.DAG_ASSET1_ID, session=mock.ANY) class TestGetAssetQueuedEvents(TestQueuedEventEndpoint): diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py index 97a54d1ef572f..8d27e8ff49aa5 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py @@ -4394,6 +4394,35 @@ def test_should_respond_403_when_user_lacks_xcom_permission(self, test_client): user=mock.ANY, ) + @pytest.mark.parametrize("team_name", ["team_b", None]) + def test_authorizes_xcom_against_the_dags_team(self, test_client, team_name): + """The XCom check must carry the Dag's team, as the route dependency above already does. + + A team-aware auth manager distinguishes a team-scoped Dag from a global one by this field, + so leaving it ``None`` asks about a differently-scoped resource than the one being read. + """ + with ( + mock.patch( + "airflow.api_fastapi.core_api.routes.public.dag_run.get_auth_manager", + autospec=True, + ) as mock_get_auth_manager, + mock.patch.object(DagModel, "get_team_name", return_value=team_name, autospec=True), + ): + mock_get_auth_manager.return_value.is_authorized_dag.return_value = True + + response = test_client.get( + f"/dags/{DAG1_ID}/dagRuns/{DAG1_RUN1_ID}/wait", + params={"interval": "1", "result": "task_1"}, + ) + + assert response.status_code == 200 + mock_get_auth_manager.return_value.is_authorized_dag.assert_called_once_with( + method="GET", + access_entity=DagAccessEntity.XCOM, + details=DagDetails(id=DAG1_ID, team_name=team_name), + user=mock.ANY, + ) + def test_should_respond_200_without_result_when_user_lacks_xcom_permission(self, test_client): """Waiting without result parameter should not require XCom permissions.""" with mock.patch( From 90558f7f0b82ec3084b2e660de530564a15b66a3 Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Tue, 4 Aug 2026 10:43:09 +0200 Subject: [PATCH 3/5] Update airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py Co-authored-by: Amogh Desai --- .../api_fastapi/core_api/routes/public/test_dag_run.py | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py index 8d27e8ff49aa5..c7fe17ceee9ff 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py @@ -4396,11 +4396,8 @@ def test_should_respond_403_when_user_lacks_xcom_permission(self, test_client): @pytest.mark.parametrize("team_name", ["team_b", None]) def test_authorizes_xcom_against_the_dags_team(self, test_client, team_name): - """The XCom check must carry the Dag's team, as the route dependency above already does. - - A team-aware auth manager distinguishes a team-scoped Dag from a global one by this field, - so leaving it ``None`` asks about a differently-scoped resource than the one being read. - """ + """The XCom check must carry the Dag's team, matching what the route dependency above already + resolves — see the call site's comment for why.""" with ( mock.patch( "airflow.api_fastapi.core_api.routes.public.dag_run.get_auth_manager", From 3287b65e598c272e91033fcc5e7be563390991ab Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Tue, 4 Aug 2026 10:43:19 +0200 Subject: [PATCH 4/5] Update airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py Co-authored-by: Amogh Desai --- .../unit/api_fastapi/core_api/routes/public/test_assets.py | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py index 3e3b299620251..a6537e3ae3819 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py @@ -2154,11 +2154,8 @@ def test_should_respond_200_with_partition_date_for_partitioned_dag( @pytest.mark.parametrize("team_name", ["team_b", None]) def test_authorizes_against_the_dags_team(self, test_client, session, team_name): - """The Dag is resolved from the asset, so its team must be resolved and passed too. - - A team-aware auth manager distinguishes a team-scoped Dag from a global one by this field, - so leaving it ``None`` asks about a differently-scoped resource than the one being acted on. - """ +"""The Dag is resolved from the asset, so its team must be resolved and passed too — see the + call site's comment for why an unresolved team asks about the wrong resource.""" recorded = [] auth_manager = mock.Mock(spec=BaseAuthManager) From fceff9a9a12a4673ff38e70bdfc1107faefdd81c Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Tue, 4 Aug 2026 12:26:30 +0200 Subject: [PATCH 5/5] Fix docstring indentation from the applied review suggestions --- .../unit/api_fastapi/core_api/routes/public/test_assets.py | 4 ++-- .../unit/api_fastapi/core_api/routes/public/test_dag_run.py | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py index a6537e3ae3819..39054a28d7024 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py @@ -2154,8 +2154,8 @@ def test_should_respond_200_with_partition_date_for_partitioned_dag( @pytest.mark.parametrize("team_name", ["team_b", None]) def test_authorizes_against_the_dags_team(self, test_client, session, team_name): -"""The Dag is resolved from the asset, so its team must be resolved and passed too — see the - call site's comment for why an unresolved team asks about the wrong resource.""" + """The Dag is resolved from the asset, so its team must be resolved and passed too — see the + call site's comment for why an unresolved team asks about the wrong resource.""" recorded = [] auth_manager = mock.Mock(spec=BaseAuthManager) diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py index c7fe17ceee9ff..0a14f69bcbeac 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py @@ -4396,8 +4396,8 @@ def test_should_respond_403_when_user_lacks_xcom_permission(self, test_client): @pytest.mark.parametrize("team_name", ["team_b", None]) def test_authorizes_xcom_against_the_dags_team(self, test_client, team_name): - """The XCom check must carry the Dag's team, matching what the route dependency above already - resolves — see the call site's comment for why.""" + """The XCom check must carry the Dag's team, matching what the route dependency above already + resolves — see the call site's comment for why.""" with ( mock.patch( "airflow.api_fastapi.core_api.routes.public.dag_run.get_auth_manager",