Scope /assets/events to the Dags the caller may read

[potiuk]

GET /api/v2/assets/events returns 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 applies no per-Dag row filter.

A caller with read on a single Dag plus the global Assets resource can therefore read the source Dag/task/run identifiers, the created dag runs, and the task-authored extra payload of events belonging to every other Dag — and can target a specific one with ?source_dag_id=.

The six sibling queued-events routes in the same file already apply ReadableDagsFilterDep. Only this one does not.

Change

Adds PermittedAssetEventFilter, following the existing Permitted*Filter family, and applies it to the query.

The filter goes into paginated_select rather than being applied after the fact, so total_entries and pagination are scoped too — otherwise the existence of hidden events still leaks through the count. That mirrors the reasoning already written into PermittedEventLogFilter.

The design question I'd most like reviewed

What should happen to events with no source Dag?

AssetEvent.source_dag_id is nullable — events created through the API, or emitted by a watcher, have no producing Dag. I've made those visible to any caller who may read assets, on the grounds that they carry no per-Dag key to authorize on:

or_(
    AssetEvent.source_dag_id.in_(self.value or set()),
    AssetEvent.source_dag_id.is_(None),
)

PermittedEventLogFilter faces the same question for audit rows not tied to a Dag, and answers it differently: it gates them behind a separate permission (AccessView.AUDIT_LOGS_ALL) rather than showing them unconditionally. If you want symmetry with that, this should grow an equivalent gate.

There is a second-order case I have not handled: source_dag_id is denormalized with no FK, so events survive deletion of the Dag that produced them. Those rows have a non-null source_dag_id that matches no readable Dag, so under this patch they become invisible to everyone, permanently. Alternatives are to treat "source Dag no longer exists" like the null case (preserves audit history, but leaks the existence of events from a Dag you could never read, once it is deleted) or to leave it as-is. I went with as-is because it is the conservative default, but it is a real behaviour change for anyone who deletes Dags and later reads asset events.

Test fixture change worth a look

The existing fixtures created events with source_dag_id="source_dag_id" (and "d", "d1", "d2") without ever creating the corresponding DagModel rows, so those Dags did not exist. Once events are scoped against readable Dags, that made 41 tests fail — not because the scoping was wrong, but because the fixtures did not represent a real deployment. They now register the Dags they attribute events to.

I want to flag that explicitly rather than bury it: a large fixture change landing alongside a behaviour change is exactly the shape that hides a bug, so please sanity-check that the fixtures now describe something realistic rather than merely something green.

The query-count assertion moves 4 → 5. The extra query is get_authorized_dag_ids, i.e. the authorization lookup itself — the same cost the queued-events routes already pay.

Testing

TestGetAssetEventsPerDagScoping covers the clause shape, the filter in isolation, and — the part that actually matters — an end-to-end test that mocks get_authorized_dag_ids and asserts the endpoint returns only permitted events, including total_entries.

I checked that these are not vacuous: unwiring the filter from the route makes the end-to-end cases fail while the isolated-filter cases still pass. An earlier version of these tests exercised only the filter class and passed even with the route unwired, which is why the end-to-end case exists.

186 tests in test_assets.py pass, as do the 34 in test_dependencies.py (the other consumer of this filter family).

---

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

[github-actions]

Hi maintainer, this PR was merged without a milestone set.
We've automatically set the milestone to Airflow 3.3.2 based on: backport label targeting v3-3-test
If this milestone is not correct, please update it to the appropriate milestone.

This comment was generated by Milestone Tag Assistant.

[github-actions]

Backport successfully created: v3-3-test

Note: As of Merging PRs targeted for Airflow 3.X
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.

In matter of doubt please ask in #release-management Slack channel.

Status	Branch	Result
✅	v3-3-test

[vincbeck]

This one broke main. Fix: #71799