CVE-2026-40690

@@ -65,6 +65,8 @@ def get_dependencies(
             raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Invalid asset node_id: {node_id}")
 
         data = get_data_dependencies(asset_id, session, readable_dags_filter.value)
+        if not data["nodes"]:
+            raise HTTPException(status.HTTP_404_NOT_FOUND, f"Asset with id {asset_id} was not found")
         return BaseGraphResponse(**data)
 
     data = get_scheduling_dependencies(readable_dags_filter.value)

@@ -130,13 +130,33 @@ def get_data_dependencies(
     asset_id: int, session: Session, readable_dag_ids: set[str] | None = None
 ) -> dict[str, list[dict]]:
     """Get full task dependencies for an asset."""
-    from sqlalchemy import select
+    from sqlalchemy import select, union_all
     from sqlalchemy.orm import selectinload
 
-    from airflow.models.asset import TaskInletAssetReference, TaskOutletAssetReference
+    from airflow.models.asset import (
+        DagScheduleAssetReference,
+        TaskInletAssetReference,
+        TaskOutletAssetReference,
+    )
 
     SEPARATOR = "__SEPARATOR__"
 
+    # Hide the asset entirely if the user has no read access to any dag that produces,
+    # consumes, or is scheduled by it. Without this check, visiting the asset graph page
+    # for an unrelated asset would leak its existence and name (and of connected nodes
+    # reachable through other readable dags) even though the user has no legitimate
+    # lineage connection to it. A readable_dag_ids value of None means no filter is
+    # applied (the user has unrestricted dag read access).
+    if readable_dag_ids is not None:
+        connected_dag_ids_query = union_all(
+            select(TaskOutletAssetReference.dag_id).where(TaskOutletAssetReference.asset_id == asset_id),
+            select(TaskInletAssetReference.dag_id).where(TaskInletAssetReference.asset_id == asset_id),
+            select(DagScheduleAssetReference.dag_id).where(DagScheduleAssetReference.asset_id == asset_id),
+        )
+        connected_dag_ids = set(session.scalars(select(connected_dag_ids_query.subquery().c.dag_id)))
+        if not connected_dag_ids & readable_dag_ids:
+            return {"nodes": [], "edges": []}
+
     nodes_dict: dict[str, dict] = {}
     edge_set: set[tuple[str, str]] = set()
 

@@ -399,3 +399,31 @@ def test_data_dependencies_respects_readable_dags_filter(
             assert node_id in nodes_by_id
         for node_id in expected_absent:
             assert node_id not in nodes_by_id
+
+    @mock.patch("airflow.api_fastapi.auth.managers.base_auth_manager.BaseAuthManager.get_authorized_dag_ids")
+    @pytest.mark.usefixtures("make_primary_connected_component", "make_secondary_connected_component")
+    def test_data_dependencies_hides_unrelated_asset(
+        self, mock_get_authorized_dag_ids, test_client, asset1_id, asset2_id
+    ):
+        # User only has read access to dags in the primary component, but asks for an asset
+        # belonging exclusively to a disjoint (secondary) component. The graph must not
+        # leak the asset's existence, name, or topology.
+        mock_get_authorized_dag_ids.return_value = {"upstream", "downstream"}
+
+        response = test_client.get(
+            "/dependencies",
+            params={"node_id": f"asset:{asset2_id}", "dependency_type": "data"},
+        )
+        assert response.status_code == 404
+
+    @mock.patch(
+        "airflow.api_fastapi.auth.managers.base_auth_manager.BaseAuthManager.get_authorized_dag_ids",
+        return_value=set(),
+    )
+    @pytest.mark.usefixtures("make_primary_connected_component")
+    def test_data_dependencies_hides_asset_when_user_has_no_dag_access(self, _, test_client, asset1_id):
+        response = test_client.get(
+            "/dependencies",
+            params={"node_id": f"asset:{asset1_id}", "dependency_type": "data"},
+        )
+        assert response.status_code == 404