This is an automated email from the ASF dual-hosted git repository. vatsrahul1001 pushed a commit to branch v3-3-test in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 1c0d24cede6e7862a5c5f6e8ba8013d99e6a7ef9 Author: Vincent <[email protected]> AuthorDate: Mon Sep 14 14:52:51 2026 -0400 Authorize POST /assets/events on the asset named in the body (#73007) (#73139) The route dependency reads the asset id from the path, but this endpoint carries it in the request body, so the auth manager was only asked whether the caller may post to any asset at all. An auth manager that scopes assets by id, name, or uri could not deny an event for an asset the caller may not touch, and the response still returned that asset's name and uri. Co-authored-by: Henry Chen <[email protected]> --- .../api_fastapi/core_api/routes/public/assets.py | 5 ++- .../src/airflow/api_fastapi/core_api/security.py | 32 ++++++++++++--- .../core_api/routes/public/test_assets.py | 47 ++++++++++++++++++++-- 3 files changed, 74 insertions(+), 10 deletions(-) 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 fd359fd86bc..de780f24dba 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 @@ -366,7 +366,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 a78187be582..7d180d59067 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.auth.managers.models.resource_details import ( ) 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, @@ -967,19 +968,38 @@ def requires_access_dag_run_clear_bulk() -> Callable[[BulkDAGRunClearBody, BaseU return inner -def requires_access_asset(method: ResourceMethod) -> Callable[[Request, BaseUser], None]: - def inner( - request: Request, - user: GetUserDep, - ) -> None: - asset_id = 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: _requires_access( is_authorized_callback=lambda: get_auth_manager().is_authorized_asset( method=method, details=AssetDetails(id=asset_id), 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 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 e1b66fb7a65..e5a19448302 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 @@ -26,7 +26,11 @@ from sqlalchemy import delete, func, select, update from airflow._shared.timezones import timezone from airflow.api_fastapi.auth.managers.base_auth_manager import BaseAuthManager -from airflow.api_fastapi.auth.managers.models.resource_details import DagAccessEntity, DagDetails +from airflow.api_fastapi.auth.managers.models.resource_details import ( + AssetDetails, + DagAccessEntity, + DagDetails, +) from airflow.api_fastapi.core_api.security import PermittedAssetEventFilter from airflow.models import DagModel from airflow.models.asset import ( @@ -1532,12 +1536,49 @@ class TestPostAssetEvents(TestAssets): } 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 read it from there to authorize.""" + (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)), + 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):
