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 4ae8da9d53c [v3-4-test] Scope upstream asset events of a Dag run to
assets and Dags the caller may read (#72924) (#75108)
4ae8da9d53c is described below
commit 4ae8da9d53c9b881874e35257a38bcc39fab039c
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Mon Oct 19 22:41:32 2026 +0200
[v3-4-test] Scope upstream asset events of a Dag run to assets and Dags the
caller may read (#72924) (#75108)
* [v3-4-test] Scope upstream asset events of a Dag run to assets and Dags
the caller may read (#72924)
GET /dags/{dag_id}/dagRuns/{dag_run_id}/upstreamAssetEvents returned every
asset event a run consumed as long as the caller could read the consuming
Dag.
Events of assets, or produced by source Dags, the caller is not allowed to
read
were exposed, including their name, uri, extra and source Dag/task ids, even
though GET /assets/events hides the same events.
Query the consumed events through the dagrun_asset_event table and apply the
same per-asset and per-source-Dag readability filters as GET /assets/events,
so hidden events are excluded from both the response and total_entries.
(cherry picked from commit 6f408e1715b411525e2c8395b4a3c39121c5d97d)
Co-authored-by: Henry Chen <[email protected]>
* Mock the simple auth manager in the upstream asset events permission test
(#75126)
SimpleAuthManager overrides get_authorized_assets since #72934, so patching
the BaseAuthManager method left the real one in place, every asset stayed
readable, and the test added in #72924 failed on main.
Generated-by: Claude Opus 5
(cherry picked from commit 33e6b5f1fa9f0e0a96fd52075e89000d47b781ee)
---------
Co-authored-by: Henry Chen <[email protected]>
Co-authored-by: Jarek Potiuk <[email protected]>
---
.../api_fastapi/core_api/routes/public/dag_run.py | 34 ++++++---
.../core_api/routes/public/test_dag_run.py | 88 +++++++++++++++++++++-
2 files changed, 108 insertions(+), 14 deletions(-)
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 9eef9661799..974f086a059 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
@@ -26,7 +26,7 @@ from fastapi.exceptions import RequestValidationError
from fastapi.responses import StreamingResponse
from pydantic import ValidationError
from sqlalchemy import select
-from sqlalchemy.orm import joinedload
+from sqlalchemy.orm import joinedload, subqueryload
from airflow.api_fastapi.app import get_auth_manager
from airflow.api_fastapi.auth.managers.models.resource_details import
DagAccessEntity, DagDetails
@@ -102,6 +102,8 @@ from airflow.api_fastapi.core_api.datamodels.task_instances
import (
from airflow.api_fastapi.core_api.openapi.exceptions import
create_openapi_http_exception_doc
from airflow.api_fastapi.core_api.security import (
GetUserDep,
+ ReadableAssetEventsByAssetFilterDep,
+ ReadableAssetEventsFilterDep,
ReadableDagRunsFilterDep,
requires_access_asset,
requires_access_dag,
@@ -122,7 +124,7 @@ from airflow.api_fastapi.core_api.services.public.dag_run
import (
from airflow.api_fastapi.logging.decorators import action_logging
from airflow.exceptions import DagVersionNotFound, ParamValidationError
from airflow.models import DagModel, DagRun
-from airflow.models.asset import AssetEvent
+from airflow.models.asset import AssetEvent, association_table as
dagrun_asset_event_table
from airflow.models.dag_version import DagVersion
from airflow.utils.state import DagRunState
from airflow.utils.types import DagRunTriggeredByType, DagRunType
@@ -282,26 +284,34 @@ def bulk_dag_runs(
],
)
def get_upstream_asset_events(
- dag_id: str, dag_run_id: str, session: SessionDep
+ dag_id: str,
+ dag_run_id: str,
+ readable_asset_events_filter: ReadableAssetEventsFilterDep,
+ readable_asset_events_by_asset_filter: ReadableAssetEventsByAssetFilterDep,
+ session: SessionDep,
) -> AssetEventCollectionResponse:
"""If dag run is asset-triggered, return the asset events that triggered
it."""
- dag_run: DagRun | None = session.scalar(
- select(DagRun)
- .where(
+ dag_run_pk = session.scalar(
+ select(DagRun.id).where(
DagRun.dag_id == dag_id,
DagRun.run_id == dag_run_id,
)
- .options(
-
joinedload(DagRun.consumed_asset_events).joinedload(AssetEvent.asset),
-
joinedload(DagRun.consumed_asset_events).subqueryload(AssetEvent.created_dagruns),
- )
)
- if dag_run is None:
+ if dag_run_pk is None:
raise HTTPException(
status.HTTP_404_NOT_FOUND,
f"The DagRun with dag_id: `{dag_id}` and run_id: `{dag_run_id}`
was not found",
)
- events = dag_run.consumed_asset_events
+ # Reading the consuming run must not expose events of assets or source
Dags the caller may not read.
+ events_select = apply_filters_to_select(
+ statement=select(AssetEvent)
+ .join(dagrun_asset_event_table, dagrun_asset_event_table.c.event_id ==
AssetEvent.id)
+ .where(dagrun_asset_event_table.c.dag_run_id == dag_run_pk)
+ .options(joinedload(AssetEvent.asset),
subqueryload(AssetEvent.created_dagruns))
+ .order_by(AssetEvent.id),
+ filters=[readable_asset_events_filter,
readable_asset_events_by_asset_filter],
+ )
+ events = session.scalars(events_select).unique().all()
return AssetEventCollectionResponse(
asset_events=serialize_asset_events(events, session=session),
total_entries=len(events),
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 36b9e4b6822..0fec3f0fe15 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
@@ -2019,7 +2019,7 @@ class TestGetDagRunAssetTriggerEvents:
session.commit()
assert event.timestamp
- with assert_queries_count(4):
+ with assert_queries_count(7):
response = test_client.get(
"/dags/TEST_DAG_ID/dagRuns/TEST_DAG_RUN_ID/upstreamAssetEvents",
)
@@ -2102,7 +2102,7 @@ class TestGetDagRunAssetTriggerEvents:
session.commit()
# Constant regardless of the number of consumed events (parametrized 1
vs 5).
- with assert_queries_count(4):
+ with assert_queries_count(7):
response =
test_client.get("/dags/TEST_DAG_ID/dagRuns/TEST_DAG_RUN_ID/upstreamAssetEvents")
assert response.status_code == 200
@@ -2116,6 +2116,90 @@ class TestGetDagRunAssetTriggerEvents:
assert triggering_by_event[newest_event_id] is True
assert all(value is False for eid, value in
triggering_by_event.items() if eid != newest_event_id)
+ def _create_consuming_run_with_events(self, dag_maker, session):
+ """
+ Create a run consuming three events: one per source Dag, one with no
source Dag.
+
+ Returns the asset ids keyed by index and the event ids keyed by
scenario.
+ """
+ asset_ids = {}
+ for index in (1, 2):
+ asset = AssetModel(name=f"ds{index}", uri=f"file:///da{index}",
group="asset")
+ session.add(asset)
+ session.flush()
+ asset_ids[index] = asset.id
+ events = {
+ "dag_a": AssetEvent(asset_id=asset_ids[1],
source_dag_id="source_dag_a", timestamp=START_DATE1),
+ "dag_b": AssetEvent(asset_id=asset_ids[2],
source_dag_id="source_dag_b", timestamp=START_DATE1),
+ "dagless": AssetEvent(asset_id=asset_ids[1],
timestamp=START_DATE1),
+ }
+ session.add_all(events.values())
+ 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.extend(events.values())
+ session.commit()
+ return asset_ids, {key: event.id for key, event in events.items()}
+
+ @pytest.mark.parametrize(
+ ("readable_dags", "expected_events"),
+ [
+ pytest.param(["source_dag_a"], ["dag_a", "dagless"],
id="one-readable-source-dag-plus-dagless"),
+ pytest.param(
+ ["source_dag_a", "source_dag_b"],
+ ["dag_a", "dag_b", "dagless"],
+ id="both-source-dags-readable",
+ ),
+ pytest.param([], ["dagless"],
id="no-readable-source-dags-still-sees-dagless"),
+ ],
+ )
+ @mock.patch(
+
"airflow.api_fastapi.auth.managers.base_auth_manager.BaseAuthManager.get_authorized_dag_ids",
+ autospec=True,
+ )
+ def test_returns_only_events_from_source_dags_the_caller_may_read(
+ self, mock_get_authorized_dag_ids, test_client, dag_maker, session,
readable_dags, expected_events
+ ):
+ """Reading the consuming run must not expose events produced by Dags
the caller cannot read."""
+ mock_get_authorized_dag_ids.return_value = set(readable_dags)
+ _, event_ids = self._create_consuming_run_with_events(dag_maker,
session)
+ expected_ids = [event_ids[key] for key in expected_events]
+
+ response =
test_client.get("/dags/TEST_DAG_ID/dagRuns/TEST_DAG_RUN_ID/upstreamAssetEvents")
+
+ assert response.status_code == 200
+ body = response.json()
+ assert [event["id"] for event in body["asset_events"]] == expected_ids
+ # The count must be scoped too, so the existence of hidden events does
not leak.
+ assert body["total_entries"] == len(expected_ids)
+
+ @mock.patch(
+
"airflow.api_fastapi.auth.managers.base_auth_manager.BaseAuthManager.get_authorized_dag_ids",
+ autospec=True,
+ )
+ @mock.patch(
+
"airflow.api_fastapi.auth.managers.simple.simple_auth_manager.SimpleAuthManager.get_authorized_assets",
+ autospec=True,
+ )
+ def test_returns_only_events_of_assets_the_caller_may_read(
+ self, mock_get_authorized_assets, mock_get_authorized_dag_ids,
test_client, dag_maker, session
+ ):
+ """An event of an asset the caller cannot read is hidden even when its
source Dag is readable."""
+ asset_ids, event_ids =
self._create_consuming_run_with_events(dag_maker, session)
+ mock_get_authorized_dag_ids.return_value = {"source_dag_a",
"source_dag_b"}
+ mock_get_authorized_assets.return_value = {asset_ids[2]}
+
+ response =
test_client.get("/dags/TEST_DAG_ID/dagRuns/TEST_DAG_RUN_ID/upstreamAssetEvents")
+
+ assert response.status_code == 200
+ body = response.json()
+ assert [event["id"] for event in body["asset_events"]] ==
[event_ids["dag_b"]]
+ assert {event["uri"] for event in body["asset_events"]} ==
{"file:///da2"}
+ assert body["total_entries"] == 1
+
def test_should_respond_401(self, unauthenticated_test_client):
response = unauthenticated_test_client.get(
"/dags/TEST_DAG_ID/dagRuns/TEST_DAG_RUN_ID/upstreamAssetEvents",
