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 e9fdcdfa737f0..f1fcd55f00125 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 @@ -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, 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 f79d3da886ac3..3a6d8b28fd7d0 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/security.py +++ b/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,12 +1083,17 @@ 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( @@ -1095,6 +1101,22 @@ def inner( ), ) + 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 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 33cc8f9fbd159..f48df57083c04 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 @@ -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):