This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new e26a4cfe458 [v3-3-test] Scope /assets/events to the Dags the caller
may read (#71741) (#71785)
e26a4cfe458 is described below
commit e26a4cfe458def722555d38568c97bbe7b7442d7
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Tue Aug 18 20:58:00 2026 +0200
[v3-3-test] Scope /assets/events to the Dags the caller may read (#71741)
(#71785)
GET /api/v2/assets/events returned AssetEvent rows for every Dag. It is
gated
on requires_access_asset(method="GET"), which under the FAB auth manager
checks
the global "Assets" resource and does not consider which Dag produced the
event, and the query applied no per-Dag row filter.
A caller with read on a single Dag plus the global "Assets" resource could
therefore read the source Dag, task and run identifiers, the created dag
runs,
and the task-authored "extra" payload of events belonging to every other
Dag,
and could target a specific one with ?source_dag_id=.
The six sibling queued-events routes in the same file already apply
ReadableDagsFilterDep; only this one did not.
Add PermittedAssetEventFilter and apply it to the query. Events produced by
a
Dag's task are scoped to that Dag's readability. Events with no source Dag —
created through the API, or emitted by a watcher — carry no per-Dag key to
authorize on and stay visible to any caller who may read assets.
The filter is applied inside paginated_select rather than after the fact, so
total_entries and pagination are scoped too and the existence of hidden
events
does not leak either.
Test fixtures that create events with a source_dag_id now register the
corresponding Dag, since the scoping resolves against DagModel and the
fixtures
previously referenced Dags that did not exist. The query-count assertion
moves
from 4 to 5: resolving the caller's readable Dags costs one query, the same
cost the queued-events routes already pay.
(cherry picked from commit f01520cbd1b20ac6beef80e244452adfa39566c9)
Generated-by: Claude Opus 5 (1M context) following the guidelines at
https:
//github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions
Co-authored-by: Jarek Potiuk <[email protected]>
---
.../api_fastapi/core_api/routes/public/assets.py | 3 +
.../src/airflow/api_fastapi/core_api/security.py | 21 ++++
.../core_api/routes/public/test_assets.py | 112 ++++++++++++++++++++-
3 files changed, 135 insertions(+), 1 deletion(-)
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 b3c41514b7d..b5aaa18cef3 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
@@ -65,6 +65,7 @@ from airflow.api_fastapi.core_api.datamodels.dag_run import
DAGRunResponse
from airflow.api_fastapi.core_api.openapi.exceptions import
create_openapi_http_exception_doc
from airflow.api_fastapi.core_api.security import (
GetUserDep,
+ ReadableAssetEventsFilterDep,
ReadableDagsFilterDep,
requires_access_asset,
requires_access_asset_alias,
@@ -324,6 +325,7 @@ def get_asset_events(
name_pattern: QueryAssetNamePatternSearch,
name_prefix_pattern: QueryAssetNamePrefixPatternSearch,
timestamp_range: Annotated[RangeFilter,
Depends(datetime_range_filter_factory("timestamp", AssetEvent))],
+ readable_asset_events_filter: ReadableAssetEventsFilterDep,
session: SessionDep,
) -> AssetEventCollectionResponse:
"""Get asset events."""
@@ -342,6 +344,7 @@ def get_asset_events(
name_pattern,
name_prefix_pattern,
timestamp_range,
+ readable_asset_events_filter,
],
order_by=order_by,
offset=offset,
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/security.py
b/airflow-core/src/airflow/api_fastapi/core_api/security.py
index f5c86a5d6a9..c814aec5956 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/security.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/security.py
@@ -70,6 +70,7 @@ from airflow.api_fastapi.core_api.datamodels.pools import
PoolBody
from airflow.api_fastapi.core_api.datamodels.variables import VariableBody
from airflow.configuration import conf
from airflow.models import Connection, Pool, Variable
+from airflow.models.asset import AssetEvent
from airflow.models.backfill import Backfill
from airflow.models.dag import DagModel, DagRun, DagTag
from airflow.models.dag_version import DagVersion
@@ -274,6 +275,23 @@ class PermittedEventLogFilter(PermittedDagFilter):
return select.where(or_(Log.dag_id.in_(self.value or set()),
Log.dag_id.is_(None)))
+class PermittedAssetEventFilter(PermittedDagFilter):
+ """A parameter that filters asset events to those produced by Dags the
user may read."""
+
+ def to_orm(self, select: Select) -> Select:
+ # Asset events created through the API, or emitted by a watcher, have
no source Dag.
+ # They carry no per-Dag key to authorize on, so they stay visible to
any caller who
+ # may read assets; only events produced by a Dag's task are scoped to
that Dag's
+ # readability. Filtering here rather than after the fact keeps
unauthorized rows out
+ # of the count and pagination too, so their existence does not leak
either.
+ return select.where(
+ or_(
+ AssetEvent.source_dag_id.in_(self.value or set()),
+ AssetEvent.source_dag_id.is_(None),
+ )
+ )
+
+
class PermittedTIFilter(PermittedDagFilter):
"""A parameter that filters the permitted task instances for the user."""
@@ -334,6 +352,9 @@ ReadableDagsFilterDep = Annotated[PermittedDagFilter,
Depends(permitted_dag_filt
ReadableDagRunsFilterDep = Annotated[
PermittedDagRunFilter, Depends(permitted_dag_filter_factory("GET",
PermittedDagRunFilter))
]
+ReadableAssetEventsFilterDep = Annotated[
+ PermittedAssetEventFilter, Depends(permitted_dag_filter_factory("GET",
PermittedAssetEventFilter))
+]
ReadableDagWarningsFilterDep = Annotated[
PermittedDagWarningFilter, Depends(permitted_dag_filter_factory("GET",
PermittedDagWarningFilter))
]
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 29327d1d404..20801eecd14 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
@@ -27,6 +27,7 @@ from sqlalchemy import delete, func, select, update
from airflow._shared.timezones import timezone
from airflow.api_fastapi.auth.managers.base_auth_manager import BaseAuthManager
from airflow.api_fastapi.auth.managers.models.resource_details import
DagAccessEntity, DagDetails
+from airflow.api_fastapi.core_api.security import PermittedAssetEventFilter
from airflow.models import DagModel
from airflow.models.asset import (
AssetActive,
@@ -68,6 +69,10 @@ pytestmark = pytest.mark.db_test
def _create_assets(session, num: int = 2) -> list[AssetModel]:
+ # Event fixtures in this module attribute their events to these Dag ids.
/assets/events
+ # scopes results to the Dags the caller may read, so the Dags have to
exist for the
+ # fixtures to represent a real deployment.
+ _ensure_dags(session, "source_dag_id", "d", "d1", "d2")
assets = [
AssetModel(
id=i,
@@ -173,7 +178,29 @@ def _create_provided_asset_alias(session, asset_alias:
AssetAliasModel) -> None:
session.commit()
+def _ensure_dags(session, *dag_ids: str) -> None:
+ """Register the Dags that asset-event fixtures attribute their events to.
+
+ ``/assets/events`` scopes events to the Dags the caller may read, and that
scoping
+ resolves against ``DagModel``. Fixtures that create events with a
``source_dag_id``
+ therefore need the corresponding Dag to exist, as it would in a real
deployment.
+ """
+ from airflow.models.dagbundle import DagBundleModel
+
+ session.merge(DagBundleModel(name="testing"))
+ session.flush()
+ for dag_id in dag_ids:
+ if session.get(DagModel, dag_id) is None:
+ session.add(DagModel(dag_id=dag_id, bundle_name="testing"))
+ session.commit()
+
+
+def _ensure_source_dag(session) -> None:
+ _ensure_dags(session, "source_dag_id")
+
+
def _create_assets_events(session, num: int = 2, varying_timestamps=False) ->
None:
+ _ensure_source_dag(session)
assets_events = [
AssetEvent(
id=i,
@@ -191,6 +218,7 @@ def _create_assets_events(session, num: int = 2,
varying_timestamps=False) -> No
def _create_assets_events_with_sensitive_extra(session, num: int = 2) -> None:
+ _ensure_source_dag(session)
assets_events = [
AssetEvent(
id=i,
@@ -208,11 +236,13 @@ def _create_assets_events_with_sensitive_extra(session,
num: int = 2) -> None:
def _create_provided_asset_event(session, asset_event: AssetEvent) -> None:
+ _ensure_source_dag(session)
session.add(asset_event)
session.commit()
def _create_dag_run(session, num: int = 2):
+ _ensure_source_dag(session)
dag_runs = [
DagRun(
dag_id="source_dag_id",
@@ -814,6 +844,84 @@ class
TestGetAssetAliasesEndpointPagination(TestAssetAliases):
assert len(response.json()["asset_aliases"]) == 50
+class TestGetAssetEventsPerDagScoping(TestAssets):
+ """``/assets/events`` returns only events the caller is entitled to see.
+
+ An event produced by a Dag's task is scoped to that Dag's readability. An
event with no
+ source Dag — created through the API, or emitted by a watcher — carries no
per-Dag key to
+ authorize on and stays visible.
+ """
+
+ def test_filter_scopes_to_source_dag_and_keeps_dagless_events(self):
+ """The clause admits the readable Dags and rows with no source Dag."""
+ rendered =
str(PermittedAssetEventFilter({"readable_dag"}).to_orm(select(AssetEvent)))
+
+ assert "source_dag_id IN" in rendered
+ assert "source_dag_id IS NULL" in rendered
+
+ def test_filter_with_no_readable_dags_still_admits_dagless_events(self,
session):
+ """A caller who may read no Dag at all still sees events that belong
to no Dag."""
+ self.create_assets(session=session, num=1)
+ session.add(AssetEvent(id=1, asset_id=1, extra={},
timestamp=DEFAULT_DATE))
+ session.add(
+ AssetEvent(id=2, asset_id=1, extra={},
source_dag_id="source_dag_id", timestamp=DEFAULT_DATE)
+ )
+ session.commit()
+
+ statement = PermittedAssetEventFilter(set()).to_orm(select(AssetEvent))
+ visible = session.scalars(statement).all()
+
+ assert [event.id for event in visible] == [1]
+
+ def test_filter_admits_only_events_from_readable_dags(self, session):
+ """An event produced by a Dag the caller cannot read is not
returned."""
+ self.create_assets(session=session, num=1)
+ session.add(
+ AssetEvent(id=1, asset_id=1, extra={},
source_dag_id="source_dag_id", timestamp=DEFAULT_DATE)
+ )
+ session.add(AssetEvent(id=2, asset_id=1, extra={},
source_dag_id="other_dag", timestamp=DEFAULT_DATE))
+ session.commit()
+
+ statement =
PermittedAssetEventFilter({"source_dag_id"}).to_orm(select(AssetEvent))
+ visible = session.scalars(statement).all()
+
+ assert [event.id for event in visible] == [1]
+
+ @pytest.mark.parametrize(
+ ("readable_dags", "expected_ids"),
+ [
+ pytest.param(["source_dag_id"], [1, 3],
id="one-readable-dag-plus-dagless"),
+ pytest.param(["source_dag_id", "other_dag"], [1, 2, 3],
id="both-dags-readable"),
+ pytest.param([], [3], id="no-readable-dags-still-sees-dagless"),
+ ],
+ )
+
@mock.patch("airflow.api_fastapi.auth.managers.base_auth_manager.BaseAuthManager.get_authorized_dag_ids")
+ def test_endpoint_returns_only_events_the_caller_may_read(
+ self, mock_get_authorized_dag_ids, test_client, session,
readable_dags, expected_ids
+ ):
+ """End-to-end: the route itself scopes the response, not just the
filter class."""
+ mock_get_authorized_dag_ids.return_value = set(readable_dags)
+
+ self.create_assets(session=session, num=1)
+ session.add_all(
+ [
+ AssetEvent(id=1, asset_id=1, extra={},
source_dag_id="source_dag_id", timestamp=DEFAULT_DATE),
+ AssetEvent(id=2, asset_id=1, extra={},
source_dag_id="other_dag", timestamp=DEFAULT_DATE),
+ # No source Dag: created through the API or emitted by a
watcher.
+ AssetEvent(id=3, asset_id=1, extra={}, timestamp=DEFAULT_DATE),
+ ]
+ )
+ session.commit()
+
+ response = test_client.get("/assets/events")
+
+ assert response.status_code == 200
+ body = response.json()
+ assert sorted(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)
+
+
class TestGetAssetEvents(TestAssets):
def test_should_respond_200(self, test_client, session):
asset1, asset2 = self.create_assets(session=session)
@@ -824,7 +932,9 @@ class TestGetAssetEvents(TestAssets):
session.commit()
assert len(assets) == 2
- with assert_queries_count(4):
+ # 5 rather than 4: resolving the caller's readable Dags, so events can
be scoped
+ # to them, costs one additional query — the same cost the
queued-events routes pay.
+ with assert_queries_count(5):
response = test_client.get("/assets/events")
assert response.status_code == 200
