This is an automated email from the ASF dual-hosted git repository.
vincbeck pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 9636fcbbb01 Authorize POST /assets/events on the asset named in the
body (#73007)
9636fcbbb01 is described below
commit 9636fcbbb015e933772393769378c2895b7abcfb
Author: Henry Chen <[email protected]>
AuthorDate: Tue Sep 15 00:42:50 2026 +0800
Authorize POST /assets/events on the asset named in the body (#73007)
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.
---
.../api_fastapi/core_api/routes/public/assets.py | 5 ++-
.../src/airflow/api_fastapi/core_api/security.py | 34 ++++++++++++++----
.../core_api/routes/public/test_assets.py | 41 ++++++++++++++++++++--
3 files changed, 71 insertions(+), 9 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 e9fdcdfa737..f1fcd55f001 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 f79d3da886a..3a6d8b28fd7 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,
@@ -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 requires_access_asset(method: ResourceMethod) ->
Callable[[Request, BaseUser
),
)
+ 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 33cc8f9fbd1..f48df57083c 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 @@ 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 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):
