[v3-4-test] Scope asset Dag and task references to readable Dags (#74836)

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.

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": {

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,
     )
 
 

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)
     ]
 

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

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"})