This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 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 d22b2fb0e8c [v3-4-test] Scope asset-wide queued event deletion to
editable Dags (#74839) (#75089)
d22b2fb0e8c is described below
commit d22b2fb0e8cf3693f5b444b71d2dc5d64e9db1ce
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Mon Oct 19 18:44:19 2026 +0530
[v3-4-test] Scope asset-wide queued event deletion to editable Dags
(#74839) (#75089)
* Scope asset-wide queued event deletion to editable Dags
DELETE /assets/{asset_id}/queuedEvents filtered the rows it removed by the
Dags the user can read, and checked Dag edit against an optional ?dag_id=
query parameter unrelated to those rows. The delete is now limited to the
queued events of Dags the user may edit, and the route no longer makes a
Dag-level check against a request-supplied dag_id.
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
* Regenerate API permission docs
(cherry picked from commit 8942dd498313e79b7ff3d6766f5ec253e3dff926)
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
Co-authored-by: Jarek Potiuk <[email protected]>
---
airflow-core/docs/security/api_permissions_ref.rst | 4 --
airflow-core/newsfragments/74839.bugfix.rst | 1 +
.../api_fastapi/core_api/routes/public/assets.py | 8 ++-
.../core_api/routes/public/test_assets.py | 69 +++++++++++++++++++++-
providers/fab/docs/auth-manager/access-control.rst | 4 --
5 files changed, 74 insertions(+), 12 deletions(-)
diff --git a/airflow-core/docs/security/api_permissions_ref.rst
b/airflow-core/docs/security/api_permissions_ref.rst
index 0e9af0696c5..cf5e188ff15 100644
--- a/airflow-core/docs/security/api_permissions_ref.rst
+++ b/airflow-core/docs/security/api_permissions_ref.rst
@@ -86,10 +86,6 @@ source code so it stays up to date as endpoints are added or
changed.
- ``/api/v2/assets/{asset_id}/queuedEvents``
- ``Asset``
- ``DELETE``
- * - ``DELETE``
- - ``/api/v2/assets/{asset_id}/queuedEvents``
- - ``DAG``
- - ``PUT``
* - ``GET``
- ``/api/v2/assets/{asset_id}/queuedEvents``
- ``Asset``
diff --git a/airflow-core/newsfragments/74839.bugfix.rst
b/airflow-core/newsfragments/74839.bugfix.rst
new file mode 100644
index 00000000000..8c20449bae6
--- /dev/null
+++ b/airflow-core/newsfragments/74839.bugfix.rst
@@ -0,0 +1 @@
+Fix ``DELETE /assets/{asset_id}/queuedEvents`` deleting queued asset events of
Dags the user can only read; it now deletes only the queued events of Dags the
user may edit.
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 abd1d1ae70f..989b57ca1f2 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
@@ -70,6 +70,7 @@ from airflow.api_fastapi.core_api.datamodels.assets import (
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 (
+ EditableDagsFilterDep,
GetUserDep,
ReadableAssetEventsByAssetFilterDep,
ReadableAssetEventsFilterDep,
@@ -719,19 +720,20 @@ def get_dag_asset_queued_event(
responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
dependencies=[
Depends(requires_access_asset(method="DELETE")),
- Depends(requires_access_dag(method="PUT")),
Depends(action_logging()),
],
)
def delete_asset_queued_events(
asset_id: int,
- readable_dags_filter: ReadableDagsFilterDep,
+ editable_dags_filter: EditableDagsFilterDep,
session: SessionDep,
before: OptionalDateTimeQuery = None,
):
"""Delete queued asset events for an asset."""
+ # This route is not tied to a single Dag, so the Dag axis is enforced per
row: only the
+ # queued events of Dags the user may edit are deleted, and the others are
left in place.
where_clause = _generate_queued_event_where_clause(
- asset_id=asset_id, before=before,
permitted_dag_ids=readable_dags_filter.value
+ asset_id=asset_id, before=before,
permitted_dag_ids=editable_dags_filter.value
)
delete_stmt = delete(AssetDagRunQueue).where(*where_clause)
result = cast("CursorResult", session.execute(delete_stmt))
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 ac902c693af..5cf95214368 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
@@ -1985,7 +1985,6 @@ class TestQueuedEventsDagAxisAuthorization:
@pytest.mark.parametrize(
"path",
[
- "/assets/{asset_id}/queuedEvents",
"/dags/{dag_id}/assets/queuedEvents",
"/dags/{dag_id}/assets/{asset_id}/queuedEvents",
],
@@ -2013,6 +2012,24 @@ class TestQueuedEventsDagAxisAuthorization:
for route in matches:
assert self._dag_axis_methods(route) == ["GET"]
+ @pytest.mark.parametrize("method", ["GET", "DELETE"])
+ def test_asset_wide_queued_events_do_not_gate_on_a_request_dag_id(self,
routes_by_path, method):
+ """The asset-wide routes have no Dag in the path; they scope rows to
permitted Dags instead.
+
+ A Dag check there would be made against whatever ``?dag_id=`` the
caller sends, which says
+ nothing about the Dags whose rows the route returns or deletes.
+ """
+ path = "/assets/{asset_id}/queuedEvents"
+ # Exclude the per-Dag routes, whose paths end with the same suffix.
+ matches = [
+ r
+ for (p, methods), r in routes_by_path.items()
+ if p.endswith(path) and "/dags/" not in p and method in methods
+ ]
+ assert matches, f"no {method} route registered for {path}"
+ for route in matches:
+ assert self._dag_axis_methods(route) == []
+
class TestDeleteDagDatasetQueuedEvents(TestQueuedEventEndpoint):
@pytest.mark.usefixtures("time_freezer")
@@ -2816,6 +2833,56 @@ class
TestDeleteAssetQueuedEvents(TestQueuedEventEndpoint):
assert response.status_code == 404
assert response.json()["detail"] == "Queue event with asset_id: `1`
was not found"
+ def _create_queued_events_for_dags(self, session, dag_ids: list[str]) ->
AssetModel:
+ (asset,) = self.create_assets(session=session, num=1)
+ _ensure_dags(session, *dag_ids)
+ for dag_id in dag_ids:
+ event = AssetEvent(asset_id=asset.id, timestamp=timezone.utcnow())
+ session.add(event)
+ session.flush()
+ session.add(AssetDagRunQueue(target_dag_id=dag_id,
asset_id=asset.id, asset_event_id=event.id))
+ session.commit()
+ return asset
+
+ @staticmethod
+ def _authorized_dag_ids_by_method(*, readable: set[str], editable:
set[str]):
+ def get_authorized_dag_ids(*, user, method="GET", **kwargs):
+ return set(editable) if method == "PUT" else set(readable)
+
+ return get_authorized_dag_ids
+
+ @pytest.mark.parametrize("query_string", ["", "?dag_id=dag_a"],
ids=["no-dag-id", "editable-dag-id"])
+ def test_deletes_only_queued_events_of_editable_dags(self, test_client,
session, query_string):
+ """A user who may edit dag_a but only read dag_b must leave dag_b's
queued events in place."""
+ asset = self._create_queued_events_for_dags(session, ["dag_a",
"dag_b"])
+
+ with mock.patch.object(
+ BaseAuthManager,
+ "get_authorized_dag_ids",
+ side_effect=self._authorized_dag_ids_by_method(readable={"dag_a",
"dag_b"}, editable={"dag_a"}),
+ ):
+ response =
test_client.delete(f"/assets/{asset.id}/queuedEvents{query_string}")
+
+ assert response.status_code == 204
+ session.expire_all()
+ remaining =
session.scalars(select(AssetDagRunQueue.target_dag_id)).all()
+ assert remaining == ["dag_b"]
+
+ def test_responds_404_when_only_readable_dags_have_queued_events(self,
test_client, session):
+ """Naming an editable Dag in the query string does not widen the
delete to readable Dags."""
+ asset = self._create_queued_events_for_dags(session, ["dag_b"])
+
+ with mock.patch.object(
+ BaseAuthManager,
+ "get_authorized_dag_ids",
+ side_effect=self._authorized_dag_ids_by_method(readable={"dag_a",
"dag_b"}, editable={"dag_a"}),
+ ):
+ response =
test_client.delete(f"/assets/{asset.id}/queuedEvents?dag_id=dag_a")
+
+ assert response.status_code == 404
+ session.expire_all()
+ assert session.scalars(select(AssetDagRunQueue.target_dag_id)).all()
== ["dag_b"]
+
def test_delete_does_not_read_back_deleted_row_keys(self, test_client,
session, create_dummy_dag):
from sqlalchemy import event
diff --git a/providers/fab/docs/auth-manager/access-control.rst
b/providers/fab/docs/auth-manager/access-control.rst
index 8d73e648bb9..87567352400 100644
--- a/providers/fab/docs/auth-manager/access-control.rst
+++ b/providers/fab/docs/auth-manager/access-control.rst
@@ -260,10 +260,6 @@ Stable API Permissions
- DELETE
- Assets.can_delete
- Op
- * - ``/api/v2/assets/{asset_id}/queuedEvents``
- - DELETE
- - DAGs.can_edit
- - User
* - ``/api/v2/assets/{asset_id}/queuedEvents``
- GET
- Assets.can_read
