This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new b623bc6e6cd [v3-4-test] Scope asset Dag and task references to
readable Dags (#74836) (#75107)
b623bc6e6cd is described below
commit b623bc6e6cda3d819cd2ac03ca8a3414dd77051d
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Mon Oct 19 18:26:52 2026 +0200
[v3-4-test] Scope asset Dag and task references to readable Dags (#74836)
(#75107)
* 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)
Generated-by: Claude Opus 5
Co-authored-by: Jarek Potiuk <[email protected]>
---
airflow-core/newsfragments/74836.bugfix.rst | 1 +
.../api_fastapi/core_api/datamodels/assets.py | 19 ++++-
.../api_fastapi/core_api/routes/public/assets.py | 59 ++++----------
.../api_fastapi/core_api/routes/ui/assets.py | 9 ++-
.../core_api/routes/public/test_assets.py | 93 ++++++++++++++++++++--
.../api_fastapi/core_api/routes/ui/test_assets.py | 42 +++++++++-
6 files changed, 172 insertions(+), 51 deletions(-)
diff --git a/airflow-core/newsfragments/74836.bugfix.rst
b/airflow-core/newsfragments/74836.bugfix.rst
new file mode 100644
index 00000000000..bde856d6643
--- /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 f38ccc6ab94..5fca001c8d1 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 @@ class AssetResponse(BaseModel):
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 @@ class AssetResponse(BaseModel):
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 989b57ca1f2..175e119d62e 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 5a8dbefb917..25c74fae942 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.openapi.exceptions import
create_openapi_http_
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 5cf95214368..740f2c7a92f 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 @@ from airflow.models.asset import (
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 @@ class TestGetAssets(TestAssets):
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 @@ class TestGetAssets(TestAssets):
# 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 @@ class TestGetAssets(TestAssets):
"""
_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 @@ class TestGetAssetEndpoint(TestAssets):
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 @@ class TestGetAssetEndpoint(TestAssets):
"""
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 87f83aac375..f964cd7a8ed 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 @@ from airflow.models.asset import (
AssetWatcherModel,
DagScheduleAssetReference,
PartitionedAssetKeyLog,
+ TaskInletAssetReference,
TaskOutletAssetReference,
)
from airflow.models.dagbundle import DagBundleModel
@@ -707,6 +708,42 @@ class TestGetAssetsUi:
# 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 @@ class TestGetAssetsUi:
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"})
