This is an automated email from the ASF dual-hosted git repository.
bbovenzi 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 a462d683469 UI: Scope partitioned Dag run asset list to readable
assets (#75213)
a462d683469 is described below
commit a462d683469bb9f55646b0515ef84c41e16650fb
Author: Henry Chen <[email protected]>
AuthorDate: Wed Oct 28 02:58:34 2026 +0800
UI: Scope partitioned Dag run asset list to readable assets (#75213)
* 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.
* 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.
---
.../api_fastapi/core_api/datamodels/common.py | 16 +++
.../core_api/datamodels/ui/partitioned_dag_runs.py | 12 +-
.../api_fastapi/core_api/openapi/_private_ui.yaml | 16 ++-
.../api_fastapi/core_api/routes/public/dags.py | 25 ++++-
.../airflow/api_fastapi/core_api/routes/ui/dags.py | 8 +-
.../core_api/routes/ui/partitioned_dag_runs.py | 28 ++++-
.../airflow/ui/openapi-gen/requests/schemas.gen.ts | 32 +++++-
.../airflow/ui/openapi-gen/requests/types.gen.ts | 7 +-
.../ui/src/components/AssetProgressCell.test.tsx | 63 ++++++++++-
.../ui/src/components/AssetProgressCell.tsx | 54 +++++----
.../core_api/routes/public/test_dags.py | 41 +++++++
.../api_fastapi/core_api/routes/ui/test_dags.py | 36 ++++++
.../routes/ui/test_partitioned_dag_runs.py | 122 ++++++++++++++++++++-
13 files changed, 415 insertions(+), 45 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 5bae6c5888c..1710a661aaa 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 @@ MaybeAssetExpression = Annotated[
]
"""``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 79a284b3f3f..e7126cff107 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 ecf8b565089..d89efe70a34 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 8d3c8e08387..11dbc1792fc 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 pydantic import ValidationError
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.parameters import (
)
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.openapi.exceptions import
create_openapi_http_
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 d011fa4c2ff..c0c747befcd 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 import and_, false, func, literal, select,
union_all
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.openapi.exceptions import
create_openapi_http_
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 f87e4cd12b2..899f6aca081 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 @@ import structlog
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.datamodels.ui.partitioned_dag_runs import (
)
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 c3d8c0f2da7..8c151dcf36f 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 86a1680d2c5..c34a9a1dfee 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 291f3198a9b..f8b8cc8d758 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(
+ <Wrapper>
+ <AssetProgressCell
+ dagId="consumer"
+ partitionedDagRunId={1}
+ partitionKey="2024-07-01"
+ totalReceived={0}
+ totalRequired={2}
+ />
+ </Wrapper>,
+ );
+
+ 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 799499b6bcf..fba23ea9f26 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<NextRunAssetEventResponse> = 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<NextRunAssetEventResponse> = 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 = ({
<HStack align="start" gap={4} maxH="300px" overflowX="auto"
overflowY="auto">
{assets
.filter((ak) => ak.is_rollup)
- .map((ak) => (
- <VStack align="start" gap={1} key={ak.asset_id}>
- <RouterLink fontSize="xs" fontWeight="semibold"
to={`/assets/${ak.asset_id}`}>
- {ak.asset_name}
- </RouterLink>
+ .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.
+ <VStack align="start" gap={1} key={ak.asset_id ??
`hidden-${index}`}>
+ {ak.hidden ? (
+ <HStack color="fg.muted" fontSize="xs"
fontWeight="semibold" gap={1}>
+ <FiLock />
+ <Text>{translate("expression.hiddenAsset")}</Text>
+ </HStack>
+ ) : (
+ <RouterLink fontSize="xs" fontWeight="semibold"
to={`/assets/${ak.asset_id}`}>
+ {ak.asset_name}
+ </RouterLink>
+ )}
<RollupKeyChecklist receivedKeys={ak.received_keys}
requiredKeys={ak.required_keys} />
</VStack>
))}
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 5c184bfb784..186c75e42cc 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 @@ class TestDagDetails(TestDagEndpoint):
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 ce504955e7e..cc33d43f540 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 @@ class TestGetDagRuns(TestPublicDagEndpoint):
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 1fe5df0b9a3..4749b68e5fc 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 @@ class TestGetPartitionedDagRuns:
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 @@ class TestGetPartitionedDagRuns:
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 @@ class TestGetPartitionedDagRuns:
)
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 @@ class TestGetPendingPartitionedDagRun:
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 = [
