This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new c654696c2fb Resolve the Dag's team when authorizing a Dag found by
lookup (#70893) (#71107)
c654696c2fb is described below
commit c654696c2fb7b07dfbd759b6e9ff4164267eb711
Author: Rahul Vats <[email protected]>
AuthorDate: Wed Aug 5 07:34:32 2026 +0530
Resolve the Dag's team when authorizing a Dag found by lookup (#70893)
(#71107)
* 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.
* 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.
* Update
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
* Update
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
* Fix docstring indentation from the applied review suggestions
---------
(cherry picked from commit f81459835adb111e57b2a87f586d856395660f8a)
Co-authored-by: Jarek Potiuk <[email protected]>
Co-authored-by: Amogh Desai <[email protected]>
---
.../api_fastapi/core_api/routes/public/assets.py | 6 ++++-
.../api_fastapi/core_api/routes/public/dag_run.py | 5 +++-
.../core_api/routes/public/test_assets.py | 28 ++++++++++++++++++++++
.../core_api/routes/public/test_dag_run.py | 26 ++++++++++++++++++++
4 files changed, 63 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 a19f4c7da77..9db8e1bffbd 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
@@ -82,6 +82,7 @@ from airflow.models.asset import (
AssetWatcherModel,
TaskOutletAssetReference,
)
+from airflow.models.dag import DagModel
from airflow.typing_compat import Unpack
from airflow.utils.state import DagRunState
from airflow.utils.types import DagRunTriggeredByType, DagRunType
@@ -435,7 +436,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 be2a247f748..19bffb1015c 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
@@ -834,7 +834,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 f28915973af..b4a4e697f62 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 @@ import time_machine
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 (
@@ -1742,6 +1743,33 @@ class TestPostAssetMaterialize(TestAssets):
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 — see the
+ call site's comment for why an unresolved team asks about the wrong
resource."""
+ recorded = []
+
+ auth_manager = mock.Mock(spec=BaseAuthManager)
+ 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,
autospec=True
+ ) as mock_get_team_name,
+ ):
+ test_client.post("/assets/1/materialize")
+
+ 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):
@pytest.mark.usefixtures("time_freezer")
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 7b672ab8134..57fb84cd6b2 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
@@ -4085,6 +4085,32 @@ class TestWaitDagRun:
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, 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",
+ 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(
