From 076d7f57577d8891a189561d80e7d9aa690fac79 Mon Sep 17 00:00:00 2001 From: henry3260 Date: Tue, 20 Oct 2026 11:12:26 +0800 Subject: [PATCH 1/2] Scope asset expressions to the assets the caller may read `next_run_assets` scopes the asset scheduling expression it serves to the assets the caller may read (#72864), and the dependencies graph followed (#73008). Four endpoints still served `DagModel.asset_expression` verbatim: the public Dag details endpoint, the UI Dag list, and both partitioned Dag run endpoints. A caller authorized on a Dag but not on its upstream assets could read every name and uri out of them, and the public endpoint needs no UI to reach. Asset names and uris commonly encode bucket, table or dataset names, which is why the rest of the asset API already scopes them. The sibling `assets` array of the pending partitioned Dag run response carries the same names; scoping that is a separate change with its own UI consumers. --- .../api_fastapi/core_api/datamodels/common.py | 16 +++ .../core_api/routes/public/dags.py | 25 ++++- .../api_fastapi/core_api/routes/ui/dags.py | 8 +- .../routes/ui/partitioned_dag_runs.py | 18 +++- .../core_api/routes/public/test_dags.py | 41 +++++++ .../core_api/routes/ui/test_dags.py | 36 +++++++ .../routes/ui/test_partitioned_dag_runs.py | 101 +++++++++++++++++- 7 files changed, 235 insertions(+), 10 deletions(-) diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/common.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/common.py index 5bae6c5888cf6..1710a661aaa6b 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/common.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/common.py @@ -220,6 +220,22 @@ def _coerce_unrecognized_expression_to_none(value: Any) -> Any: ] """``AssetExpression | None`` that degrades a legacy/unrecognized stored shape to ``None``.""" + +_maybe_asset_expression_adapter: TypeAdapter[Union[AssetExpression, None]] = TypeAdapter(MaybeAssetExpression) + + +def validate_asset_expression(value: Any) -> AssetExpression | None: + """ + Validate a raw ``asset_expression`` tree into the typed model, the way a response field does. + + A route that rewrites the tree after its response model is built -- redacting the assets the + caller may not read, for instance -- must assign the result of this rather than the raw tree. + FastAPI does not validate the nested models of an already-built response, so a raw tree would + bypass ``MaybeAssetExpression`` and reach the client verbatim, legacy shapes included. + """ + return _maybe_asset_expression_adapter.validate_python(value) + + # Common Bulk Data Models T = TypeVar("T") K = TypeVar("K") diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py index 8d3c8e083873e..11dbc1792fc87 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py @@ -25,6 +25,7 @@ from sqlalchemy import delete, func, insert, select, update from airflow.api.common import delete_dag as delete_dag_module +from airflow.api_fastapi.common.asset_expression import redact_asset_expression from airflow.api_fastapi.common.dagbag import DagBagDep, get_latest_version_of_dag from airflow.api_fastapi.common.db.common import SessionDep, apply_filters_to_select, paginated_select from airflow.api_fastapi.common.db.dags import eager_load_teams, generate_dag_with_latest_run_query @@ -58,7 +59,11 @@ ) from airflow.api_fastapi.common.router import AirflowRouter from airflow.api_fastapi.compat import HTTP_422_UNPROCESSABLE_CONTENT -from airflow.api_fastapi.core_api.datamodels.common import BulkBody, BulkResponse +from airflow.api_fastapi.core_api.datamodels.common import ( + BulkBody, + BulkResponse, + validate_asset_expression, +) from airflow.api_fastapi.core_api.datamodels.dags import ( BulkDAGBody, DAGCollectionResponse, @@ -71,6 +76,7 @@ from airflow.api_fastapi.core_api.security import ( EditableDagsFilterDep, GetUserDep, + ReadableAssetsFilterDep, ReadableDagsFilterDep, requires_access_dag, requires_access_dag_bulk, @@ -239,7 +245,11 @@ def get_dag( dependencies=[Depends(requires_access_dag(method="GET"))], ) def get_dag_details( - dag_id: str, session: SessionDep, dag_bag: DagBagDep, user: GetUserDep + dag_id: str, + session: SessionDep, + dag_bag: DagBagDep, + user: GetUserDep, + readable_assets_filter: ReadableAssetsFilterDep, ) -> DAGDetailsResponse: """Get details of Dag.""" dag = get_latest_version_of_dag(dag_bag, dag_id, session) @@ -280,7 +290,16 @@ def get_dag_details( setattr(dag_model, "active_runs_count", active_runs_count) setattr(dag_model, "latest_dag_version", latest_dag_version) - return DAGDetailsResponse.model_validate(dag_model) + details = DAGDetailsResponse.model_validate(dag_model) + # The redaction lands on the response, not on `dag_model`: that is a live ORM instance whose + # session commits when the request ends, so assigning the blanked leaves back onto it would + # persist them. + details.asset_expression = validate_asset_expression( + redact_asset_expression( + dag_model.asset_expression, readable_asset_ids=readable_assets_filter.value or set() + ) + ) + return details @dags_router.patch( diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dags.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dags.py index d011fa4c2ffb8..c0c747befcdab 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dags.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dags.py @@ -26,6 +26,7 @@ from sqlalchemy.orm import defaultload from airflow.api_fastapi.auth.managers.models.resource_details import DagAccessEntity +from airflow.api_fastapi.common.asset_expression import redact_asset_expression from airflow.api_fastapi.common.dagbag import DagBagDep from airflow.api_fastapi.common.db.common import ( SessionDep, @@ -79,6 +80,7 @@ from airflow.api_fastapi.core_api.routes.public.hitl import HITLDetailView from airflow.api_fastapi.core_api.security import ( GetUserDep, + ReadableAssetsFilterDep, ReadableDagsFilterDep, requires_access_dag, ) @@ -155,6 +157,7 @@ def get_dags( ], has_pending_actions: QueryPendingActionsFilter, readable_dags_filter: ReadableDagsFilterDep, + readable_assets_filter: ReadableAssetsFilterDep, session: SessionDep, dag_bag: DagBagDep, user: GetUserDep, @@ -311,6 +314,7 @@ def get_dags( # Build the dict dynamically from DAGResponse.model_fields so that new fields # added to DAGResponse are picked up automatically without code changes here. dag_runs_by_dag_id: dict[str, DAGWithLatestDagRunsResponse] = {} + readable_asset_ids = readable_assets_filter.value or set() for dag in dags: dag_data = { DAG_ALIAS_MAPPING.get(field_name, field_name): getattr( @@ -320,7 +324,9 @@ def get_dags( } dag_data.update( { - "asset_expression": dag.asset_expression, + "asset_expression": redact_asset_expression( + dag.asset_expression, readable_asset_ids=readable_asset_ids + ), "latest_dag_runs": [], "has_unfinished_runs": has_unfinished_runs_by_dag_id[dag.dag_id], "pending_actions": pending_actions_by_dag_id[dag.dag_id], diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/partitioned_dag_runs.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/partitioned_dag_runs.py index f87e4cd12b260..143cae00bb224 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/partitioned_dag_runs.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/partitioned_dag_runs.py @@ -22,6 +22,7 @@ from fastapi import Depends, HTTPException, status from sqlalchemy import and_, select +from airflow.api_fastapi.common.asset_expression import redact_asset_expression from airflow.api_fastapi.common.db.common import SessionDep, apply_filters_to_select, paginated_select from airflow.api_fastapi.common.parameters import ( QueryLimit, @@ -42,6 +43,7 @@ ) from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc from airflow.api_fastapi.core_api.security import ( + ReadableAssetsFilterDep, ReadableDagsFilterDep, requires_access_asset, requires_access_dag, @@ -251,6 +253,7 @@ def get_partitioned_dag_runs( limit: QueryLimit, offset: QueryOffset, readable_dags_filter: ReadableDagsFilterDep, + readable_assets_filter: ReadableAssetsFilterDep, dag_id: QueryPartitionedDagRunDagIdFilter, has_created_dag_run_id: QueryPartitionedDagRunHasCreatedDagRunIdFilter, ) -> PartitionedDagRunCollectionResponse: @@ -351,7 +354,11 @@ def get_partitioned_dag_runs( asset_expressions: dict[str, dict | None] | None = None if dag_id.value is None: - asset_expressions = {dm.dag_id: dm.asset_expression for dm in dag_models.values()} + readable_asset_ids = readable_assets_filter.value or set() + asset_expressions = { + dm.dag_id: redact_asset_expression(dm.asset_expression, readable_asset_ids=readable_asset_ids) + for dm in dag_models.values() + } model_data: dict[str, Any] = { "partitioned_dag_runs": results, @@ -370,6 +377,7 @@ def get_pending_partitioned_dag_run( dag_id: str, partition_key: str, session: SessionDep, + readable_assets_filter: ReadableAssetsFilterDep, partitioned_dag_run_id: int | None = None, ) -> PartitionedDagRunDetailResponse: """Return full details for pending PartitionedDagRun.""" @@ -496,7 +504,13 @@ def get_pending_partitioned_dag_run( total_received = sum(a.received_count for a in assets) total_required = sum(a.required_count for a in assets) - asset_expression = dag_model.asset_expression if dag_model is not None else None + asset_expression = ( + redact_asset_expression( + dag_model.asset_expression, readable_asset_ids=readable_assets_filter.value or set() + ) + if dag_model is not None + else None + ) model_data: dict[str, Any] = { "id": partitioned_dag_run.id, diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dags.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dags.py index 5c184bfb784c2..186c75e42cc6f 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dags.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dags.py @@ -1776,6 +1776,47 @@ def test_dag_details_includes_is_favorite_field(self, session, test_client): assert isinstance(body["is_favorite"], bool) assert body["is_favorite"] is False + @mock.patch( + "airflow.api_fastapi.auth.managers.simple.simple_auth_manager.SimpleAuthManager.get_authorized_assets", + autospec=True, + ) + def test_dag_details_hides_assets_the_caller_may_not_read( + self, mock_get_authorized_assets, session, test_client + ): + dag_model = session.get(DagModel, DAG2_ID) + dag_model.asset_expression = { + "all": [ + {"asset": {"uri": "s3://bucket/visible", "name": "visible", "group": "asset", "id": 1}}, + {"asset": {"uri": "s3://bucket/secret", "name": "secret", "group": "asset", "id": 2}}, + ] + } + session.commit() + mock_get_authorized_assets.return_value = {1} + + response = test_client.get(f"/dags/{DAG2_ID}/details") + + assert response.status_code == 200 + assert response.json()["asset_expression"] == { + "all": [ + { + "asset": { + "uri": "s3://bucket/visible", + "name": "visible", + "group": "asset", + "id": 1, + "hidden": False, + } + }, + {"asset": {"uri": None, "name": None, "group": "asset", "id": None, "hidden": True}}, + ] + } + assert "s3://bucket/secret" not in response.text + + # Only the response is redacted: writing the blanked leaves back through the request + # session would overwrite the stored expression when that session commits. + session.expire_all() + assert session.get(DagModel, DAG2_ID).asset_expression["all"][1]["asset"]["name"] == "secret" + def test_dag_details_serves_legacy_asset_expression_as_null(self, session, test_client): """A pre-3.0 dataset-format ``asset_expression`` that the typed model cannot describe is served as ``null`` instead of 500ing the endpoint (handled by ``MaybeAssetExpression``).""" diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dags.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dags.py index ce504955e7e33..cc33d43f5407b 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dags.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dags.py @@ -559,6 +559,42 @@ def test_rejects_sort_by_dag_run_column_names(self, test_client, order_by): f"the attribute does not exist on the model" ) + @mock.patch( + "airflow.api_fastapi.auth.managers.simple.simple_auth_manager.SimpleAuthManager.get_authorized_assets", + autospec=True, + ) + def test_hides_assets_the_caller_may_not_read(self, mock_get_authorized_assets, test_client, session): + dag_model = session.get(DagModel, DAG1_ID) + dag_model.asset_expression = { + "any": [ + {"asset": {"uri": "s3://bucket/visible", "name": "visible", "group": "asset", "id": 1}}, + {"asset": {"uri": "s3://bucket/secret", "name": "secret", "group": "asset", "id": 2}}, + ] + } + session.commit() + mock_get_authorized_assets.return_value = {1} + + response = test_client.get("/dags", params={"dag_ids": [DAG1_ID]}) + + assert response.status_code == 200 + assert response.json()["dags"][0]["asset_expression"] == { + "any": [ + { + "asset": { + "uri": "s3://bucket/visible", + "name": "visible", + "group": "asset", + "id": 1, + "hidden": False, + } + }, + # The route is served with ``response_model_exclude_none``, so the blanked fields + # drop out of the payload entirely. + {"asset": {"group": "asset", "hidden": True}}, + ] + } + assert "s3://bucket/secret" not in response.text + def test_get_dags_no_n_plus_one_queries(self, session, test_client): """Test that fetching DAGs with tags doesn't trigger n+1 queries.""" num_dags = 5 diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_partitioned_dag_runs.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_partitioned_dag_runs.py index 1fe5df0b9a389..4b987a4d74ab4 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_partitioned_dag_runs.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_partitioned_dag_runs.py @@ -59,6 +59,51 @@ def test_should_response_403(self, unauthorized_test_client): response = unauthorized_test_client.get("/partitioned_dag_runs?dag_id=any") assert response.status_code == 403 + @mock.patch( + "airflow.api_fastapi.auth.managers.simple.simple_auth_manager.SimpleAuthManager.get_authorized_assets", + autospec=True, + ) + def test_hides_assets_the_caller_may_not_read( + self, mock_get_authorized_assets, test_client, dag_maker, session + ): + visible = Asset(uri="s3://bucket/visible", name="visible") + secret = Asset(uri="s3://bucket/secret", name="secret") + with dag_maker( + dag_id="redacted_dag", + schedule=PartitionedAssetTimetable(assets=visible & secret), + serialized=True, + ): + EmptyOperator(task_id="t") + dag_maker.create_dagrun() + dag_maker.sync_dagbag_to_db() + + session.add( + AssetPartitionDagRun( + target_dag_id="redacted_dag", partition_key="2024-06-01", created_dag_run_id=None + ) + ) + session.commit() + visible_id = session.scalar(select(AssetModel.id).where(AssetModel.name == "visible")) + mock_get_authorized_assets.return_value = {visible_id} + + resp = test_client.get("/partitioned_dag_runs") + + assert resp.status_code == 200 + assert resp.json()["asset_expressions"]["redacted_dag"] == { + "all": [ + { + "asset": { + "uri": "s3://bucket/visible", + "name": "visible", + "group": "asset", + "id": visible_id, + "hidden": False, + } + }, + {"asset": {"uri": None, "name": None, "group": "asset", "id": None, "hidden": True}}, + ] + } + def test_should_response_404(self, test_client): assert test_client.get("/partitioned_dag_runs?dag_id=no_such_dag").status_code == 404 @@ -68,7 +113,7 @@ def test_should_response_200_non_partitioned_dag_returns_empty(self, test_client dag_maker.create_dagrun() dag_maker.sync_dagbag_to_db() - with assert_queries_count(4): + with assert_queries_count(5): resp = test_client.get("/partitioned_dag_runs?dag_id=normal&has_created_dag_run_id=false") assert resp.status_code == 200 assert resp.json() == {"partitioned_dag_runs": [], "total": 0, "asset_expressions": None} @@ -155,17 +200,18 @@ def test_should_response_200( ) session.commit() - # Six batch queries, each independent of len(rows) / num_assets: + # Seven batch queries, each independent of len(rows) / num_assets: # 1. SELECT COUNT of matching AssetPartitionDagRun rows # 2. outer SELECT of paginated AssetPartitionDagRun rows # 3. SELECT DagModel WHERE dag_id IN (unique_dag_ids) # 4. _fetch_active_assets_per_dag — joined SELECT on AssetModel + DagScheduleAssetReference # 5. SELECT PartitionedAssetKeyLog WHERE asset_partition_dag_run_id IN (apdr_ids) # 6. SELECT timetable for asset_expressions (cached per response) + # 7. SELECT AssetModel.id for the assets the caller may read # The count does not scale with num_assets or len(rows), so the bump - # from main's baseline of 3 (3 → 4 → 5 → 6 across this branch) is constant + # from main's baseline of 3 (3 → 4 → 5 → 6 → 7 across this branch) is constant # per request, not per row. - with assert_queries_count(6): + with assert_queries_count(7): resp = test_client.get( f"/partitioned_dag_runs?dag_id=list_dag" f"&has_created_dag_run_id={str(has_created_dag_run_id).lower()}" @@ -740,6 +786,53 @@ def test_should_response_200(self, test_client, dag_maker, session, num_assets, received_uris = {a["asset_uri"] for a in body["assets"] if a["received"]} assert received_uris == set(uris[:received_count]) + @mock.patch( + "airflow.api_fastapi.auth.managers.simple.simple_auth_manager.SimpleAuthManager.get_authorized_assets", + autospec=True, + ) + def test_hides_assets_the_caller_may_not_read( + self, mock_get_authorized_assets, test_client, dag_maker, session + ): + visible = Asset(uri="s3://bucket/visible", name="visible") + secret = Asset(uri="s3://bucket/secret", name="secret") + with dag_maker( + dag_id="redacted_detail_dag", + schedule=PartitionedAssetTimetable(assets=visible & secret), + serialized=True, + ): + EmptyOperator(task_id="t") + dag_maker.create_dagrun() + dag_maker.sync_dagbag_to_db() + + session.add( + AssetPartitionDagRun( + target_dag_id="redacted_detail_dag", partition_key="2024-07-01", created_dag_run_id=None + ) + ) + session.commit() + visible_id = session.scalar(select(AssetModel.id).where(AssetModel.name == "visible")) + mock_get_authorized_assets.return_value = {visible_id} + + resp = test_client.get("/pending_partitioned_dag_run/redacted_detail_dag?partition_key=2024-07-01") + + assert resp.status_code == 200 + # The sibling `assets` array still carries every name and uri; scoping that is a separate + # change with its own UI consumers. + assert resp.json()["asset_expression"] == { + "all": [ + { + "asset": { + "uri": "s3://bucket/visible", + "name": "visible", + "group": "asset", + "id": visible_id, + "hidden": False, + } + }, + {"asset": {"uri": None, "name": None, "group": "asset", "id": None, "hidden": True}}, + ] + } + def test_is_rollup_false_for_non_rollup_asset(self, test_client, dag_maker, session): """is_rollup is False for every asset when the schedule uses identity (non-rollup) mappers.""" asset_defs = [ From e0a3f0db48a745c5e8fdc9eb187aa0ccea11bcb9 Mon Sep 17 00:00:00 2001 From: henry3260 Date: Tue, 20 Oct 2026 19:31:42 +0800 Subject: [PATCH 2/2] UI: Scope partitioned Dag run asset list to readable assets The pending partitioned Dag run detail response carries two asset-bearing fields. `asset_expression` is scoped to the assets the caller may read; the sibling `assets` array was not, so one payload could blank an asset inside the boolean tree while spelling its name and uri out one key over. That defeated the redaction on this route, and the route's own asset dependency only checks whether the caller may read assets at all -- there is no asset id in its path for it to authorize against. The counts and partition keys of a blanked asset are kept. They describe the Dag's own schedule, which a caller who reached this endpoint can already read, and dropping them would leave a stuck run with no explanation at all. This endpoint is served only to the UI: the public v2 spec regenerates byte-identical, and no airflow-ctl model references the response. --- .../datamodels/ui/partitioned_dag_runs.py | 12 +++- .../core_api/openapi/_private_ui.yaml | 16 ++++- .../routes/ui/partitioned_dag_runs.py | 16 +++-- .../ui/openapi-gen/requests/schemas.gen.ts | 32 +++++++++- .../ui/openapi-gen/requests/types.gen.ts | 7 ++- .../ui/public/i18n/locales/en/common.json | 1 + .../src/components/AssetProgressCell.test.tsx | 63 ++++++++++++++++++- .../ui/src/components/AssetProgressCell.tsx | 54 ++++++++++------ .../routes/ui/test_partitioned_dag_runs.py | 27 +++++++- 9 files changed, 187 insertions(+), 41 deletions(-) diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/partitioned_dag_runs.py b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/partitioned_dag_runs.py index 79a284b3f3f01..e7126cff107a0 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/partitioned_dag_runs.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/partitioned_dag_runs.py @@ -44,9 +44,15 @@ class PartitionedDagRunCollectionResponse(BaseModel): class PartitionedDagRunAssetResponse(BaseModel): """Asset info within a partitioned Dag run detail.""" - asset_id: int - asset_name: str - asset_uri: str + asset_id: int | None + asset_name: str | None + asset_uri: str | None + hidden: bool = False + """True when the caller may not read this asset: ``asset_id``, ``asset_name`` and ``asset_uri`` + are blanked and the UI must not link to the asset. The counts and partition keys stay, so the + run's progress is still readable -- they describe the Dag's own schedule, which the caller can + already read, rather than the asset's identity. Mirrors the redaction + ``redact_asset_expression`` applies to the sibling ``asset_expression`` field.""" received: bool received_count: int required_count: int diff --git a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml index e7bc1eccb2c75..05e013d623cfe 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml +++ b/airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml @@ -5458,14 +5458,24 @@ components: PartitionedDagRunAssetResponse: properties: asset_id: - type: integer + anyOf: + - type: integer + - type: 'null' title: Asset Id asset_name: - type: string + anyOf: + - type: string + - type: 'null' title: Asset Name asset_uri: - type: string + anyOf: + - type: string + - type: 'null' title: Asset Uri + hidden: + type: boolean + title: Hidden + default: false received: type: boolean title: Received diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/partitioned_dag_runs.py b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/partitioned_dag_runs.py index 143cae00bb224..899f6aca081fc 100644 --- a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/partitioned_dag_runs.py +++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/partitioned_dag_runs.py @@ -445,8 +445,13 @@ def get_pending_partitioned_dag_run( has_rollup_mappers = dag_model is not None and dag_model.has_rollup_mappers rollup_timetable = load_partitioned_timetable(dag_id, session) if has_rollup_mappers else None + readable_asset_ids = readable_assets_filter.value or set() + assets = [] for asset_row in asset_rows: + # The row itself is still needed in full below: the rollup mapper is keyed by the asset's + # real name and uri, and the received-key lookup by its real id. Only the response is blanked. + is_hidden = asset_row.id not in readable_asset_ids received_keys = sorted(received_keys_by_asset.get(asset_row.id, set())) required_keys: list[str] = [partition_key] is_rollup = ( @@ -488,9 +493,10 @@ def get_pending_partitioned_dag_run( received_count = 1 if received_keys else 0 assets.append( PartitionedDagRunAssetResponse( - asset_id=asset_row.id, - asset_name=asset_row.name, - asset_uri=asset_row.uri, + asset_id=None if is_hidden else asset_row.id, + asset_name=None if is_hidden else asset_row.name, + asset_uri=None if is_hidden else asset_row.uri, + hidden=is_hidden, received=received_count >= required_count and required_count > 0, received_count=received_count, required_count=required_count, @@ -505,9 +511,7 @@ def get_pending_partitioned_dag_run( total_received = sum(a.received_count for a in assets) total_required = sum(a.required_count for a in assets) asset_expression = ( - redact_asset_expression( - dag_model.asset_expression, readable_asset_ids=readable_assets_filter.value or set() - ) + redact_asset_expression(dag_model.asset_expression, readable_asset_ids=readable_asset_ids) if dag_model is not None else None ) diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts index b1fb2e977304e..37adf54db5bd6 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts @@ -12814,17 +12814,43 @@ export const $OklchColor = { export const $PartitionedDagRunAssetResponse = { properties: { asset_id: { - type: 'integer', + anyOf: [ + { + type: 'integer' + }, + { + type: 'null' + } + ], title: 'Asset Id' }, asset_name: { - type: 'string', + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], title: 'Asset Name' }, asset_uri: { - type: 'string', + anyOf: [ + { + type: 'string' + }, + { + type: 'null' + } + ], title: 'Asset Uri' }, + hidden: { + type: 'boolean', + title: 'Hidden', + default: false + }, received: { type: 'boolean', title: 'Received' diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts index 5321f56062b61..715681c9613a6 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts @@ -3280,9 +3280,10 @@ export type OklchColor = string; * Asset info within a partitioned Dag run detail. */ export type PartitionedDagRunAssetResponse = { - asset_id: number; - asset_name: string; - asset_uri: string; + asset_id: number | null; + asset_name: string | null; + asset_uri: string | null; + hidden?: boolean; received: boolean; received_count: number; required_count: number; diff --git a/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json b/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json index 6d1cdb9962202..7df9e8bc34c08 100644 --- a/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json +++ b/airflow-core/src/airflow/ui/public/i18n/locales/en/common.json @@ -139,6 +139,7 @@ "all": "All", "and": "AND", "any": "Any", + "hiddenAsset": "Hidden asset", "or": "OR" }, "filter": "Filter", diff --git a/airflow-core/src/airflow/ui/src/components/AssetProgressCell.test.tsx b/airflow-core/src/airflow/ui/src/components/AssetProgressCell.test.tsx index 291f3198a9b32..f8b8cc8d75895 100644 --- a/airflow-core/src/airflow/ui/src/components/AssetProgressCell.test.tsx +++ b/airflow-core/src/airflow/ui/src/components/AssetProgressCell.test.tsx @@ -16,11 +16,13 @@ * specific language governing permissions and limitations * under the License. */ -import { render, waitFor } from "@testing-library/react"; +import "@testing-library/jest-dom/vitest"; +import { fireEvent, render, screen, waitFor } from "@testing-library/react"; import { expect, it, vi } from "vitest"; import { PartitionedDagRunService } from "openapi/requests/services.gen"; +import i18n from "src/i18n/config"; import { Wrapper } from "src/utils/Wrapper"; import { AssetProgressCell } from "./AssetProgressCell"; @@ -68,3 +70,62 @@ it("fetches separate details for pending runs sharing a partition key", async () }); }); }); + +it("labels a hidden asset instead of linking to it, leaving readable assets linked", async () => { + // The shape the pending-detail endpoint serves for an asset the caller may not read. + vi.spyOn(PartitionedDagRunService, "getPendingPartitionedDagRun").mockResolvedValue({ + assets: [ + { + asset_id: 1, + asset_name: "sales_daily", + asset_uri: "s3://bucket/sales", + hidden: false, + is_rollup: true, + received: false, + received_count: 0, + received_keys: [], + required_count: 1, + required_keys: ["2024-07-01"], + }, + { + asset_id: null, + asset_name: null, + asset_uri: null, + hidden: true, + is_rollup: true, + received: false, + received_count: 0, + received_keys: [], + required_count: 1, + required_keys: ["2024-07-01"], + }, + ], + dag_id: "consumer", + id: 1, + partition_key: "2024-07-01", + total_received: 0, + total_required: 2, + }); + + render( + + + , + ); + + const trigger = screen.getByRole("button"); + + // The trigger is disabled while the detail query is in flight, which swallows the click. + await waitFor(() => expect(trigger).not.toBeDisabled()); + fireEvent.click(trigger); + + await waitFor(() => expect(screen.getByText(i18n.t("common:expression.hiddenAsset"))).toBeInTheDocument()); + expect(screen.getByRole("link", { name: "sales_daily" })).toHaveAttribute("href", "/assets/1"); + expect(screen.getAllByRole("link")).toHaveLength(1); +}); diff --git a/airflow-core/src/airflow/ui/src/components/AssetProgressCell.tsx b/airflow-core/src/airflow/ui/src/components/AssetProgressCell.tsx index 799499b6bcfbb..fba23ea9f2611 100644 --- a/airflow-core/src/airflow/ui/src/components/AssetProgressCell.tsx +++ b/airflow-core/src/airflow/ui/src/components/AssetProgressCell.tsx @@ -16,9 +16,9 @@ * specific language governing permissions and limitations * under the License. */ -import { Button, HStack, VStack } from "@chakra-ui/react"; +import { Button, HStack, Text, VStack } from "@chakra-ui/react"; import { useTranslation } from "react-i18next"; -import { FiDatabase } from "react-icons/fi"; +import { FiDatabase, FiLock } from "react-icons/fi"; import { usePartitionedDagRunServiceGetPendingPartitionedDagRun } from "openapi/queries"; import type { NextRunAssetEventResponse, PartitionedDagRunAssetResponse } from "openapi/requests/types.gen"; @@ -45,7 +45,7 @@ export const AssetProgressCell = ({ totalReceived, totalRequired, }: Props) => { - const { i18n } = useTranslation(); + const { i18n, t: translate } = useTranslation(); const { data, isLoading } = usePartitionedDagRunServiceGetPendingPartitionedDagRun({ dagId, partitionedDagRunId, @@ -57,17 +57,24 @@ export const AssetProgressCell = ({ const hasRollup = assets.some((ak) => ak.is_rollup); - const events: Array = assets - .filter((ak: PartitionedDagRunAssetResponse) => ak.received_count > 0) - .map((ak: PartitionedDagRunAssetResponse) => ({ - id: ak.asset_id, - is_rollup: ak.is_rollup, - last_update: ak.received ? "received" : null, - name: ak.asset_name, - received_count: ak.received_count, - required_count: ak.required_count, - uri: ak.asset_uri, - })); + // An asset the caller may not read arrives with its id, name and uri blanked, so it has nothing + // to match an expression leaf on -- and the leaf it would have matched is redacted to `id: null` + // too. Dropping it here keeps the event list to the assets that can actually line up with one. + const events: Array = assets.flatMap((ak: PartitionedDagRunAssetResponse) => + ak.received_count > 0 && ak.asset_id !== null && ak.asset_name !== null && ak.asset_uri !== null + ? [ + { + id: ak.asset_id, + is_rollup: ak.is_rollup, + last_update: ak.received ? "received" : null, + name: ak.asset_name, + received_count: ak.received_count, + required_count: ak.required_count, + uri: ak.asset_uri, + }, + ] + : [], + ); return ( // eslint-disable-next-line jsx-a11y/no-autofocus @@ -85,11 +92,20 @@ export const AssetProgressCell = ({ {assets .filter((ak) => ak.is_rollup) - .map((ak) => ( - - - {ak.asset_name} - + .map((ak, index) => ( + // A hidden asset has no id to key on; the rows come back ordered by the asset's + // uri, so the position within one response is stable. + + {ak.hidden ? ( + + + {translate("expression.hiddenAsset")} + + ) : ( + + {ak.asset_name} + + )} ))} diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_partitioned_dag_runs.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_partitioned_dag_runs.py index 4b987a4d74ab4..4749b68e5fca0 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_partitioned_dag_runs.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_partitioned_dag_runs.py @@ -816,9 +816,30 @@ def test_hides_assets_the_caller_may_not_read( resp = test_client.get("/pending_partitioned_dag_run/redacted_detail_dag?partition_key=2024-07-01") assert resp.status_code == 200 - # The sibling `assets` array still carries every name and uri; scoping that is a separate - # change with its own UI consumers. - assert resp.json()["asset_expression"] == { + body = resp.json() + # Rows come back ordered by the asset's uri, so "secret" precedes "visible". + assert [ + {k: a[k] for k in ("asset_id", "asset_name", "asset_uri", "hidden", "required_count")} + for a in body["assets"] + ] == [ + { + "asset_id": None, + "asset_name": None, + "asset_uri": None, + "hidden": True, + "required_count": 1, + }, + { + "asset_id": visible_id, + "asset_name": "visible", + "asset_uri": "s3://bucket/visible", + "hidden": False, + "required_count": 1, + }, + ] + # Progress stays readable: a blanked asset still counts toward the run's totals. + assert body["total_required"] == 2 + assert body["asset_expression"] == { "all": [ { "asset": {