From 6089862880ccde70acfec197de5027a5824fee77 Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Mon, 19 Oct 2026 16:32:23 +0200 Subject: [PATCH] [v3-4-test] Scope asset Dag and task references to readable Dags (#74836) * Scope asset Dag and task references to readable Dags The asset list and detail endpoints (public and UI) returned every scheduled Dag, producing task and consuming task reference of an asset, including ones that belong to Dags the requesting user cannot read. Those nested references are now narrowed to the user's readable Dags. Generated-by: Claude Opus 5 following the guidelines at https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions * Rename newsfragment to the PR number Generated-by: Claude Opus 5 following the guidelines at https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions * Require readable_dag_ids when building an asset response With a None default, a caller that forgot the argument got every Dag and task reference back unfiltered. Without a default, each caller has to say whether the references are narrowed, so a new endpoint fails closed instead of leaking references to Dags the user cannot read. (cherry picked from commit 99f2b8618df8727c9e58a5524b0eb9a7ed8f518b) Co-authored-by: Jarek Potiuk Generated-by: Claude Opus 5 --- airflow-core/newsfragments/74836.bugfix.rst | 1 + .../api_fastapi/core_api/datamodels/assets.py | 19 +++- .../core_api/routes/public/assets.py | 59 ++++-------- .../api_fastapi/core_api/routes/ui/assets.py | 9 +- .../core_api/routes/public/test_assets.py | 93 ++++++++++++++++++- .../core_api/routes/ui/test_assets.py | 42 ++++++++- 6 files changed, 172 insertions(+), 51 deletions(-) create mode 100644 airflow-core/newsfragments/74836.bugfix.rst diff --git a/airflow-core/newsfragments/74836.bugfix.rst b/airflow-core/newsfragments/74836.bugfix.rst new file mode 100644 index 0000000000000..bde856d6643bc --- /dev/null +++ b/airflow-core/newsfragments/74836.bugfix.rst @@ -0,0 +1 @@ +The ``scheduled_dags``, ``producing_tasks`` and ``consuming_tasks`` lists returned by the asset list and asset detail API endpoints now only include references to Dags the requesting user can read. diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/assets.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/assets.py index f38ccc6ab94bd..5fca001c8d138 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/assets.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/assets.py @@ -116,8 +116,22 @@ def from_asset_row( asset: AssetModel, last_asset_event_id: int | None, last_asset_event_timestamp: datetime | None, + *, + readable_dag_ids: set[str] | None, ) -> AssetResponse: - """Build a response from an ``AssetModel`` row joined with its last AssetEvent id/timestamp.""" + """ + Build a response from an ``AssetModel`` row joined with its last AssetEvent id/timestamp. + + The scheduled Dag, producing task and consuming task references are narrowed to the Dags in + ``readable_dag_ids``; ``None`` means the caller may read every Dag. It has no default so that + a new caller has to decide, rather than return every reference by omission. + """ + + def _readable(refs): + if readable_dag_ids is None: + return refs + return [ref for ref in refs if ref.dag_id in readable_dag_ids] + watchers_data = [ { "name": watcher.name, @@ -129,6 +143,9 @@ def from_asset_row( return cls.model_validate( { **asset.__dict__, + "scheduled_dags": _readable(asset.scheduled_dags), + "producing_tasks": _readable(asset.producing_tasks), + "consuming_tasks": _readable(asset.consuming_tasks), "aliases": asset.aliases, "watchers": watchers_data, "last_asset_event": { 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 989b57ca1f240..175e119d62e3d 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 @@ -161,6 +161,7 @@ def get_assets( Depends(SortParam(["id", "name", "uri", "created_at", "updated_at"], AssetModel).dynamic_depends()), ], readable_assets_filter: ReadableAssetsFilterDep, + readable_dags_filter: ReadableDagsFilterDep, session: SessionDep, ) -> AssetCollectionResponse: """Get assets.""" @@ -224,30 +225,16 @@ def get_assets( ) ) - assets = [] - - for asset, last_asset_event_id, last_asset_event_timestamp in assets_rows: - watchers_data = [ - { - "name": watcher.name, - "trigger_id": watcher.trigger_id, - "created_date": watcher.trigger.created_date, - } - for watcher in asset.watchers - ] - - asset_response = AssetResponse.model_validate( - { - **asset.__dict__, - "aliases": asset.aliases, - "watchers": watchers_data, - "last_asset_event": { - "id": last_asset_event_id, - "timestamp": last_asset_event_timestamp, - }, - } + # Nested Dag and task references are narrowed to the Dags the caller may read. + assets = [ + AssetResponse.from_asset_row( + asset, + last_asset_event_id, + last_asset_event_timestamp, + readable_dag_ids=readable_dags_filter.value, ) - assets.append(asset_response) + for asset, last_asset_event_id, last_asset_event_timestamp in assets_rows + ] return AssetCollectionResponse( assets=assets, @@ -593,6 +580,7 @@ def get_asset_queued_events( ) def get_asset( asset_id: int, + readable_dags_filter: ReadableDagsFilterDep, session: SessionDep, ) -> AssetResponse: """Get an asset.""" @@ -625,25 +613,12 @@ def get_asset( if asset is None: raise HTTPException(status.HTTP_404_NOT_FOUND, f"The Asset with ID: `{asset_id}` was not found") - watchers_data = [ - { - "name": watcher.name, - "trigger_id": watcher.trigger_id, - "created_date": watcher.trigger.created_date, - } - for watcher in asset.watchers - ] - - return AssetResponse.model_validate( - { - **asset.__dict__, - "aliases": asset.aliases, - "watchers": watchers_data, - "last_asset_event": { - "id": last_asset_event_id, - "timestamp": last_asset_event_timestamp, - }, - } + # Nested Dag and task references are narrowed to the Dags the caller may read. + return AssetResponse.from_asset_row( + asset, + last_asset_event_id, + last_asset_event_timestamp, + readable_dag_ids=readable_dags_filter.value, ) diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/assets.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/assets.py index 5a8dbefb91724..25c74fae94293 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/assets.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/assets.py @@ -52,6 +52,7 @@ from airflow.api_fastapi.core_api.routes.public.assets import OnlyActiveFilter from airflow.api_fastapi.core_api.security import ( ReadableAssetsFilterDep, + ReadableDagsFilterDep, requires_access_asset, requires_access_asset_alias, requires_access_dag, @@ -115,6 +116,7 @@ def get_assets( ), ], readable_assets_filter: ReadableAssetsFilterDep, + readable_dags_filter: ReadableDagsFilterDep, session: SessionDep, ) -> AssetCollectionResponse: """Get assets. Like the public endpoint, but also supports sorting by group and last asset event timestamp.""" @@ -141,7 +143,12 @@ def get_assets( ) assets = [ - AssetResponse.from_asset_row(asset, last_asset_event_id, last_asset_event_timestamp) + AssetResponse.from_asset_row( + asset, + last_asset_event_id, + last_asset_event_timestamp, + readable_dag_ids=readable_dags_filter.value, + ) for asset, last_asset_event_id, last_asset_event_timestamp in session.execute(assets_select) ] 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 5cf95214368bd..740f2c7a92ffa 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 @@ -42,6 +42,7 @@ AssetModel, AssetWatcherModel, DagScheduleAssetReference, + TaskInletAssetReference, TaskOutletAssetReference, ) from airflow.models.base import ID_LEN @@ -168,6 +169,35 @@ def _create_assets_with_team_references(session, num: int = 2, refs_per_asset: i return assets +def _create_asset_referenced_by_two_dags(session) -> AssetModel: + """Create one asset that is scheduling, produced by and consumed by both ``dag_a`` and ``dag_b``.""" + _ensure_dags(session, "dag_a", "dag_b") + asset = AssetModel(id=1, name="shared", uri="s3://bucket/shared", group="asset") + session.add(asset) + session.add(AssetActive.for_asset(asset)) + session.flush() + for dag_id in ("dag_a", "dag_b"): + session.add_all( + [ + DagScheduleAssetReference(dag_id=dag_id, asset=asset), + TaskOutletAssetReference(dag_id=dag_id, task_id=f"{dag_id}_producer", asset=asset), + TaskInletAssetReference(dag_id=dag_id, task_id=f"{dag_id}_consumer", asset=asset), + ] + ) + session.commit() + return asset + + +def _assert_references_scoped_to_dag_a(asset: dict) -> None: + assert [ref["dag_id"] for ref in asset["scheduled_dags"]] == ["dag_a"] + assert [(ref["dag_id"], ref["task_id"]) for ref in asset["producing_tasks"]] == [ + ("dag_a", "dag_a_producer") + ] + assert [(ref["dag_id"], ref["task_id"]) for ref in asset["consuming_tasks"]] == [ + ("dag_a", "dag_a_consumer") + ] + + def _create_assets_with_sensitive_extra(session, num: int = 2) -> None: assets = [ AssetModel( @@ -374,9 +404,10 @@ def test_should_respond_200(self, test_client, session): assert len(session.scalars(select(AssetModel)).all()) == 3 assert len(session.scalars(select(AssetActive)).all()) == 2 - # 8 rather than 7: resolving the caller's readable assets, so the list can be scoped - # to them, costs one additional query. - with assert_queries_count(8): + # 9 rather than 7: resolving the caller's readable assets, so the list can be scoped + # to them, and the caller's readable Dags, so the nested Dag and task references can + # be scoped to them, each cost one additional query. + with assert_queries_count(9): response = test_client.get("/assets") assert response.status_code == 200 @@ -437,6 +468,31 @@ def test_should_return_only_assets_the_caller_may_read( # The count must be scoped too, so the existence of hidden assets does not leak. assert body["total_entries"] == 2 + @mock.patch("airflow.api_fastapi.auth.managers.base_auth_manager.BaseAuthManager.get_authorized_dag_ids") + def test_should_return_only_references_of_readable_dags( + self, mock_get_authorized_dag_ids, test_client, session + ): + _create_asset_referenced_by_two_dags(session) + mock_get_authorized_dag_ids.return_value = {"dag_a"} + + response = test_client.get("/assets") + + assert response.status_code == 200 + body = response.json() + assert body["total_entries"] == 1 + _assert_references_scoped_to_dag_a(body["assets"][0]) + + def test_should_return_all_references_when_all_dags_are_readable(self, test_client, session): + _create_asset_referenced_by_two_dags(session) + + response = test_client.get("/assets") + + assert response.status_code == 200 + asset = response.json()["assets"][0] + assert {ref["dag_id"] for ref in asset["scheduled_dags"]} == {"dag_a", "dag_b"} + assert {ref["dag_id"] for ref in asset["producing_tasks"]} == {"dag_a", "dag_b"} + assert {ref["dag_id"] for ref in asset["consuming_tasks"]} == {"dag_a", "dag_b"} + def test_should_respond_200_with_watchers(self, test_client, session): """Test that assets with watchers return the watcher information in the API response.""" asset1, asset2 = self.create_assets_with_watchers(session=session, num=2) @@ -612,7 +668,7 @@ def test_query_count_with_multi_team(self, test_client, session): """ _create_assets_with_team_references(session, num=5) - with assert_queries_count(10): + with assert_queries_count(11): response = test_client.get("/assets") assert response.status_code == 200 @@ -1785,6 +1841,33 @@ def test_should_authorize_with_asset_name_and_uri(self, mock_is_authorized_asset user=mock.ANY, ) + @mock.patch("airflow.api_fastapi.auth.managers.base_auth_manager.BaseAuthManager.get_authorized_dag_ids") + def test_should_return_only_references_of_readable_dags( + self, mock_get_authorized_dag_ids, test_client, session + ): + asset = _create_asset_referenced_by_two_dags(session) + mock_get_authorized_dag_ids.return_value = {"dag_a"} + + response = test_client.get(f"/assets/{asset.id}") + + assert response.status_code == 200 + _assert_references_scoped_to_dag_a(response.json()) + + @mock.patch("airflow.api_fastapi.auth.managers.base_auth_manager.BaseAuthManager.get_authorized_dag_ids") + def test_should_return_no_references_without_readable_dags( + self, mock_get_authorized_dag_ids, test_client, session + ): + asset = _create_asset_referenced_by_two_dags(session) + mock_get_authorized_dag_ids.return_value = set() + + response = test_client.get(f"/assets/{asset.id}") + + assert response.status_code == 200 + body = response.json() + assert body["scheduled_dags"] == [] + assert body["producing_tasks"] == [] + assert body["consuming_tasks"] == [] + @provide_session def test_should_respond_200_with_watchers(self, test_client, *, session): """Test that single asset endpoint returns watcher information.""" @@ -1828,7 +1911,7 @@ def test_query_count_with_multi_team(self, test_client, session): """ asset = _create_assets_with_team_references(session, num=1, refs_per_asset=5)[0] - with assert_queries_count(8): + with assert_queries_count(9): response = test_client.get(f"/assets/{asset.id}") assert response.status_code == 200 diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_assets.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_assets.py index 87f83aac3754b..f964cd7a8ed61 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_assets.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_assets.py @@ -35,6 +35,7 @@ AssetWatcherModel, DagScheduleAssetReference, PartitionedAssetKeyLog, + TaskInletAssetReference, TaskOutletAssetReference, ) from airflow.models.dagbundle import DagBundleModel @@ -707,6 +708,42 @@ def test_should_return_only_assets_the_caller_may_read( # The count must be scoped too, so the existence of hidden assets does not leak. assert body["total_entries"] == 1 + @mock.patch("airflow.api_fastapi.auth.managers.base_auth_manager.BaseAuthManager.get_authorized_dag_ids") + def test_should_return_only_references_of_readable_dags( + self, mock_get_authorized_dag_ids, test_client, session + ): + bundle = DagBundleModel(name="testing") + session.add(bundle) + asset = AssetModel(name="shared", uri="s3://bucket/shared", group="asset") + session.add(asset) + session.add(AssetActive.for_asset(asset)) + session.flush() + for dag_id in ("dag_a", "dag_b"): + session.add_all( + [ + DagModel(dag_id=dag_id, bundle_name="testing"), + DagScheduleAssetReference(dag_id=dag_id, asset=asset), + TaskOutletAssetReference(dag_id=dag_id, task_id=f"{dag_id}_producer", asset=asset), + TaskInletAssetReference(dag_id=dag_id, task_id=f"{dag_id}_consumer", asset=asset), + ] + ) + session.commit() + mock_get_authorized_dag_ids.return_value = {"dag_a"} + + response = test_client.get("/assets") + + assert response.status_code == 200 + body = response.json() + assert body["total_entries"] == 1 + result = body["assets"][0] + assert [ref["dag_id"] for ref in result["scheduled_dags"]] == ["dag_a"] + assert [(ref["dag_id"], ref["task_id"]) for ref in result["producing_tasks"]] == [ + ("dag_a", "dag_a_producer") + ] + assert [(ref["dag_id"], ref["task_id"]) for ref in result["consuming_tasks"]] == [ + ("dag_a", "dag_a_consumer") + ] + def test_sort_by_last_asset_event_timestamp(self, test_client, session): older = AssetModel(name="older", uri="s3://bucket/older", group="asset") newer = AssetModel(name="newer", uri="s3://bucket/newer", group="asset") @@ -1027,8 +1064,9 @@ def test_query_count(self, test_client, session): assets[i].aliases.append(AssetAliasModel(name=f"alias{i}", group="")) session.commit() - # One of these queries resolves the caller's readable assets so the list can be scoped to them. - with assert_queries_count(9): + # Two of these queries resolve the caller's readable assets and readable Dags, so the list + # and its nested Dag and task references can be scoped to them. + with assert_queries_count(10): assert test_client.get("/assets").status_code == 200 @conf_vars({("core", "multi_team"): "True"})