Authorize POST /assets/events on the asset named in the body

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

@@ -396,7 +396,10 @@ def get_asset_events(
 @assets_router.post(
     "/assets/events",
     responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
-    dependencies=[Depends(requires_access_asset(method="POST")), Depends(action_logging())],
+    dependencies=[
+        Depends(requires_access_asset(method="POST", asset_id_from_body=True)),
+        Depends(action_logging()),
+    ],
 )
 def create_asset_event(
     body: CreateAssetEventsBody,

airflow-core/src/airflow/api_fastapi/core_api/security.py

@@ -56,6 +56,7 @@
 )
 from airflow.api_fastapi.common.db.common import SessionDep
 from airflow.api_fastapi.core_api.base import OrmClause
+from airflow.api_fastapi.core_api.datamodels.assets import CreateAssetEventsBody
 from airflow.api_fastapi.core_api.datamodels.common import (
     BulkAction,
     BulkActionOnExistence,
@@ -1082,19 +1083,40 @@ def _build_asset_details(asset_id: str | None) -> AssetDetails:
     return AssetDetails(id=asset_id, name=name, uri=uri)
 
 
-def requires_access_asset(method: ResourceMethod) -> Callable[[Request, BaseUser], None]:
-    def inner(
-        request: Request,
-        user: GetUserDep,
-    ) -> None:
-        details = _build_asset_details(request.path_params.get("asset_id"))
+def requires_access_asset(method: ResourceMethod, *, asset_id_from_body: bool = False) -> Callable[..., None]:
+    """
+    Authorize the caller on the asset targeted by the request.
+
+    :param method: the method to perform
+    :param asset_id_from_body: read ``asset_id`` from a ``CreateAssetEventsBody`` request body instead of
+        the path. The dependency parameter must be named ``body`` to share the route's body.
+    """
+
+    def _authorize(asset_id: str | None, user: BaseUser) -> None:
+        details = _build_asset_details(asset_id)
 
         _requires_access(
             is_authorized_callback=lambda: get_auth_manager().is_authorized_asset(
                 method=method, details=details, user=user
             ),
         )
 
+    if asset_id_from_body:
+
+        def inner_from_body(
+            body: CreateAssetEventsBody,
+            user: GetUserDep,
+        ) -> None:
+            _authorize(str(body.asset_id), user)
+
+        return inner_from_body
+
+    def inner(
+        request: Request,
+        user: GetUserDep,
+    ) -> None:
+        _authorize(request.path_params.get("asset_id"), user)
+
     return inner
 
 

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

@@ -2029,12 +2029,49 @@ def test_should_respond_200(self, test_client, session):
         }
         check_last_log(session, dag_id=None, event="create_asset_event", logical_date=None)
 
+    @mock.patch(
+        "airflow.api_fastapi.auth.managers.simple.simple_auth_manager.SimpleAuthManager.is_authorized_asset",
+        autospec=True,
+    )
+    def test_should_authorize_on_the_asset_named_in_the_body(
+        self, mock_is_authorized_asset, test_client, session
+    ):
+        """The asset id lives in the body, so the route must resolve it and authorize on the full asset."""
+        (asset,) = self.create_assets(num=1, session=session)
+        mock_is_authorized_asset.return_value = True
+
+        response = test_client.post("/assets/events", json={"asset_id": asset.id})
+
+        assert response.status_code == 200
+        mock_is_authorized_asset.assert_called_once_with(
+            mock.ANY,
+            method="POST",
+            details=AssetDetails(id=str(asset.id), name="simple1", uri="s3://bucket/key/1"),
+            user=mock.ANY,
+        )
+
+    @mock.patch(
+        "airflow.api_fastapi.auth.managers.simple.simple_auth_manager.SimpleAuthManager.is_authorized_asset",
+        autospec=True,
+    )
+    def test_should_respond_403_when_not_authorized_on_the_asset(
+        self, mock_is_authorized_asset, test_client, session
+    ):
+        (asset,) = self.create_assets(num=1, session=session)
+        mock_is_authorized_asset.return_value = False
+
+        response = test_client.post("/assets/events", json={"asset_id": asset.id})
+
+        assert response.status_code == 403
+        assert session.scalar(select(func.count()).select_from(AssetEvent)) == 0
+
     def test_should_respond_401(self, unauthenticated_test_client):
         response = unauthenticated_test_client.post("/assets/events", json={"asset_uri": "s3://bucket/key/1"})
         assert response.status_code == 401
 
-    def test_should_respond_403(self, unauthorized_test_client):
-        response = unauthorized_test_client.post("/assets/events", json={"asset_uri": "s3://bucket/key/1"})
+    def test_should_respond_403(self, unauthorized_test_client, session):
+        (asset,) = self.create_assets(num=1, session=session)
+        response = unauthorized_test_client.post("/assets/events", json={"asset_id": asset.id})
         assert response.status_code == 403
 
     def test_invalid_attr_not_allowed(self, test_client, session):