Scope asset-wide queued event deletion to editable Dags

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``

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.

airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py

@@ -70,6 +70,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 (
+    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))

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py

@@ -1985,7 +1985,6 @@ def routes_by_path(self, test_client):
     @pytest.mark.parametrize(
         "path",
         [
-            "/assets/{asset_id}/queuedEvents",
             "/dags/{dag_id}/assets/queuedEvents",
             "/dags/{dag_id}/assets/{asset_id}/queuedEvents",
         ],
@@ -2013,6 +2012,24 @@ def test_get_queued_events_requires_only_dag_read(self, routes_by_path, path):
         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 @@ def test_should_respond_404(self, test_client):
         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
 

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