This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new f81459835ad Resolve the Dag's team when authorizing a Dag found by
lookup (#70893)
f81459835ad is described below
commit f81459835adb111e57b2a87f586d856395660f8a
Author: Jarek Potiuk <[email protected]>
AuthorDate: Tue Aug 4 14:23:50 2026 +0200
Resolve the Dag's team when authorizing a Dag found by lookup (#70893)
* 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
Co-authored-by: Amogh Desai <[email protected]>
* Update
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
Co-authored-by: Amogh Desai <[email protected]>
* Fix docstring indentation from the applied review suggestions
---------
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 0a6818648cd..830d4edaeed 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 @@ from airflow.models.asset import (
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 1b857df06ce..e5d06587cb5 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 fcf41162872..39054a28d70 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 (
@@ -2151,6 +2152,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 97a54d1ef57..0a14f69bcbe 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,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(
