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/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 ecf8b56508995..d89efe70a34bc 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
@@ -5531,14 +5531,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/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..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
@@ -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."""
@@ -437,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 = (
@@ -480,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,
@@ -496,7 +510,11 @@ 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_asset_ids)
+ if dag_model is not None
+ else None
+ )
model_data: dict[str, Any] = {
"id": partitioned_dag_run.id,
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 c3d8c0f2da733..8c151dcf36f88 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
@@ -12954,17 +12954,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 86a1680d2c592..c34a9a1dfeeb7 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
@@ -3317,9 +3317,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/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/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..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
@@ -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,74 @@ 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
+ 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": {
+ "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 = [