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 1246e653357 [v3-4-test] Scope asset event created Dag runs to readable
Dags (#76234) (#76236)
1246e653357 is described below
commit 1246e653357904ed5475a0d619be964bd087eb42
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Sun Nov 1 19:55:02 2026 +0100
[v3-4-test] Scope asset event created Dag runs to readable Dags (#76234)
(#76236)
* Scope asset event created Dag runs to readable Dags
Generated-by: Claude Opus 5
* Rename newsfragment to the PR number
(cherry picked from commit e54a6af12d6006721e946f5bff11c8bec739cd1a)
Generated-by: Claude Opus 5
Co-authored-by: Jarek Potiuk <[email protected]>
---
airflow-core/newsfragments/76234.bugfix.rst | 1 +
.../api_fastapi/core_api/routes/public/assets.py | 4 +-
.../api_fastapi/core_api/routes/public/dag_run.py | 4 +-
.../api_fastapi/core_api/services/public/assets.py | 17 ++++++-
.../core_api/routes/public/test_assets.py | 42 ++++++++++++++++-
.../core_api/routes/public/test_dag_run.py | 54 ++++++++++++++++++++++
6 files changed, 117 insertions(+), 5 deletions(-)
diff --git a/airflow-core/newsfragments/76234.bugfix.rst
b/airflow-core/newsfragments/76234.bugfix.rst
new file mode 100644
index 00000000000..d733ae3816a
--- /dev/null
+++ b/airflow-core/newsfragments/76234.bugfix.rst
@@ -0,0 +1 @@
+The ``created_dagruns`` list in asset events returned by the asset events and
upstream asset events API endpoints now only includes Dag runs of Dags the
requesting user can read.
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 175e119d62e..003e880b81d 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
@@ -381,7 +381,9 @@ def get_asset_events(
assets_events = session.scalars(assets_event_select).all()
return AssetEventCollectionResponse(
- asset_events=serialize_asset_events(assets_events, session=session),
+ asset_events=serialize_asset_events(
+ assets_events,
readable_dag_ids=readable_asset_events_filter.value, session=session
+ ),
total_entries=total_entries,
)
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 d77dbe4148f..d7bb5a70887 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
@@ -313,7 +313,9 @@ def get_upstream_asset_events(
)
events = session.scalars(events_select).unique().all()
return AssetEventCollectionResponse(
- asset_events=serialize_asset_events(events, session=session),
+ asset_events=serialize_asset_events(
+ events, readable_dag_ids=readable_asset_events_filter.value,
session=session
+ ),
total_entries=len(events),
)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/services/public/assets.py
b/airflow-core/src/airflow/api_fastapi/core_api/services/public/assets.py
index f110f15e1fd..c7816ebc577 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/services/public/assets.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/services/public/assets.py
@@ -28,18 +28,31 @@ if TYPE_CHECKING:
from airflow.models.asset import AssetEvent
-def serialize_asset_events(events: Sequence[AssetEvent], *, session: Session)
-> list[AssetEventResponse]:
+def serialize_asset_events(
+ events: Sequence[AssetEvent], *, readable_dag_ids: set[str] | None,
session: Session
+) -> list[AssetEventResponse]:
"""
Serialize asset events, flagging on each created dag run whether this
event triggered it.
Only a run's most recent consumed event triggers it; the rest were merely
included. The flag is
resolved per ``(event_id, dag_id, run_id)`` tuple and set on the run
before validation so the response
carries it.
+
+ The created dag runs are narrowed to the Dags in ``readable_dag_ids``.
``None`` is treated as no
+ readable Dag, as the asset events filter the ids come from does. The
argument has no default so that a
+ new caller has to decide, rather than return every run by omission. The
narrowing is applied to the
+ response, never to the ``created_dagruns`` relationship, so the session
does not see a changed
+ collection.
"""
+ readable = readable_dag_ids or set()
triggering_refs = resolve_triggering_event_refs(events, session=session)
asset_events = []
for event in events:
for run in event.created_dagruns:
run.triggering = (event.id, run.dag_id, run.run_id) in
triggering_refs
- asset_events.append(AssetEventResponse.model_validate(event))
+ response = AssetEventResponse.model_validate(event)
+ response = response.model_copy(
+ update={"created_dagruns": [run for run in
response.created_dagruns if run.dag_id in readable]}
+ )
+ asset_events.append(response)
return asset_events
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 6ab18c19990..d9ba3d5b9b6 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
@@ -1111,6 +1111,46 @@ class TestGetAssetEventsPerDagScoping(TestAssets):
assert [event["id"] for event in body["asset_events"]] == [2, 4]
assert body["total_entries"] == 2
+ @pytest.mark.parametrize(
+ ("readable_dags", "expected_run_dags"),
+ [
+ pytest.param({"dag_a"}, ["dag_a"], id="one-readable-consumer-dag"),
+ pytest.param({"dag_a", "dag_b"}, ["dag_a", "dag_b"],
id="both-consumer-dags-readable"),
+ pytest.param(set(), [], id="no-readable-consumer-dags"),
+ ],
+ )
+
@mock.patch("airflow.api_fastapi.auth.managers.base_auth_manager.BaseAuthManager.get_authorized_dag_ids")
+ def test_endpoint_returns_only_created_dagruns_of_readable_dags(
+ self, mock_get_authorized_dag_ids, test_client, session,
readable_dags, expected_run_dags
+ ):
+ """The Dag runs an event created are narrowed to the Dags the caller
may read."""
+ mock_get_authorized_dag_ids.return_value = readable_dags
+ _ensure_dags(session, "dag_a", "dag_b")
+ self.create_assets(session=session, num=1)
+ # No source Dag, so the event itself is visible to every caller.
+ event = AssetEvent(id=1, asset_id=1, extra={}, timestamp=DEFAULT_DATE)
+ session.add(event)
+ session.flush()
+ for dag_id in ("dag_a", "dag_b"):
+ dag_run = DagRun(
+ dag_id=dag_id,
+ run_id=f"{dag_id}_run",
+ run_type=DagRunType.ASSET_TRIGGERED,
+ logical_date=DEFAULT_DATE,
+ start_date=DEFAULT_DATE,
+ data_interval=(DEFAULT_DATE, DEFAULT_DATE),
+ state=DagRunState.SUCCESS,
+ )
+ dag_run.consumed_asset_events.append(event)
+ session.add(dag_run)
+ session.commit()
+
+ response = test_client.get("/assets/events")
+
+ assert response.status_code == 200
+ [returned_event] = response.json()["asset_events"]
+ assert sorted(run["dag_id"] for run in
returned_event["created_dagruns"]) == expected_run_dags
+
class TestGetAssetEvents(TestAssets):
@pytest.mark.parametrize("unknown", [False, True])
@@ -1278,7 +1318,7 @@ class TestGetAssetEvents(TestAssets):
def test_should_return_created_dag_run_without_start_date(self,
test_client, session):
self.create_assets(num=1, session=session)
- _ensure_dags(session, "producer_dag")
+ _ensure_dags(session, "producer_dag", "consumer_dag")
asset_event = AssetEvent(
asset_id=1,
source_dag_id="producer_dag",
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 6c6aa57990d..9610949dded 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
@@ -2217,6 +2217,60 @@ class TestGetDagRunAssetTriggerEvents:
assert {event["uri"] for event in body["asset_events"]} ==
{"file:///da2"}
assert body["total_entries"] == 1
+ @pytest.mark.parametrize(
+ ("readable_dags", "expected_run_dags"),
+ [
+ pytest.param({"TEST_DAG_ID"}, ["TEST_DAG_ID"],
id="other-consumer-dag-not-readable"),
+ pytest.param(
+ {"TEST_DAG_ID", "other_consumer"}, ["TEST_DAG_ID",
"other_consumer"], id="both-readable"
+ ),
+ ],
+ )
+ @mock.patch(
+
"airflow.api_fastapi.auth.managers.base_auth_manager.BaseAuthManager.get_authorized_dag_ids",
+ autospec=True,
+ )
+ def test_returns_only_created_dagruns_of_readable_dags(
+ self, mock_get_authorized_dag_ids, test_client, dag_maker, session,
readable_dags, expected_run_dags
+ ):
+ """An event's created Dag runs are narrowed to the Dags the caller may
read."""
+ mock_get_authorized_dag_ids.return_value = readable_dags
+ asset = AssetModel(name="ds1", uri="file:///da1", group="asset")
+ session.add(asset)
+ session.flush()
+ # No source Dag, so the event itself is visible to every caller.
+ event = AssetEvent(asset_id=asset.id, timestamp=START_DATE1)
+ session.add(event)
+ session.flush()
+ with dag_maker(
+ dag_id="TEST_DAG_ID", start_date=START_DATE1,
schedule=timedelta(days=1), session=session
+ ):
+ pass
+ dag_run = dag_maker.create_dagrun(run_id="TEST_DAG_RUN_ID",
run_type=DagRunType.ASSET_TRIGGERED)
+ dag_run.consumed_asset_events.append(event)
+ session.merge(DagBundleModel(name="testing"))
+ session.flush()
+ session.add(DagModel(dag_id="other_consumer", bundle_name="testing"))
+ session.flush()
+ other_run = DagRun(
+ dag_id="other_consumer",
+ run_id="other_consumer_run",
+ run_type=DagRunType.ASSET_TRIGGERED,
+ logical_date=START_DATE1,
+ start_date=START_DATE1,
+ data_interval=(START_DATE1, START_DATE1),
+ state=DagRunState.SUCCESS,
+ )
+ other_run.consumed_asset_events.append(event)
+ session.add(other_run)
+ session.commit()
+
+ response =
test_client.get("/dags/TEST_DAG_ID/dagRuns/TEST_DAG_RUN_ID/upstreamAssetEvents")
+
+ assert response.status_code == 200
+ [returned_event] = response.json()["asset_events"]
+ assert sorted(run["dag_id"] for run in
returned_event["created_dagruns"]) == expected_run_dags
+
def test_should_respond_401(self, unauthenticated_test_client):
response = unauthenticated_test_client.get(
"/dags/TEST_DAG_ID/dagRuns/TEST_DAG_RUN_ID/upstreamAssetEvents",
