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 9dcd6a0a2ee Hide unreadable assets in the asset expression served by
next_run_assets (#72864)
9dcd6a0a2ee is described below
commit 9dcd6a0a2eeca35168a5ba69363dbb7253c5701d
Author: Henry Chen <[email protected]>
AuthorDate: Sat Sep 19 02:29:31 2026 +0800
Hide unreadable assets in the asset expression served by next_run_assets
(#72864)
---
.../airflow/api_fastapi/common/asset_expression.py | 62 ++++++++++++++++++
.../api_fastapi/core_api/datamodels/common.py | 9 ++-
.../api_fastapi/core_api/openapi/_private_ui.yaml | 23 ++++++-
.../core_api/openapi/v2-rest-api-generated.yaml | 23 ++++++-
.../api_fastapi/core_api/routes/ui/assets.py | 12 +++-
.../airflow/ui/openapi-gen/requests/schemas.gen.ts | 29 ++++++++-
.../airflow/ui/openapi-gen/requests/types.gen.ts | 9 ++-
.../api_fastapi/common/test_asset_expression.py | 73 ++++++++++++++++++++++
.../api_fastapi/core_api/datamodels/test_common.py | 14 ++++-
.../api_fastapi/core_api/routes/ui/test_assets.py | 48 +++++++++++++-
.../src/airflowctl/api/datamodels/generated.py | 9 ++-
11 files changed, 290 insertions(+), 21 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/common/asset_expression.py
b/airflow-core/src/airflow/api_fastapi/common/asset_expression.py
new file mode 100644
index 00000000000..b926677a758
--- /dev/null
+++ b/airflow-core/src/airflow/api_fastapi/common/asset_expression.py
@@ -0,0 +1,62 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Authorization-aware handling of the ``DagModel.asset_expression`` tree
served by the API."""
+
+from __future__ import annotations
+
+from collections.abc import Collection
+from typing import Any
+
+
+def redact_asset_expression(
+ expression: dict[str, Any] | None, *, readable_asset_ids: Collection[int]
+) -> dict[str, Any] | None:
+ """
+ Return a copy of an asset scheduling expression with the assets the caller
may not read hidden.
+
+ ``DagModel.asset_expression`` names every upstream asset of a Dag, so any
endpoint that serves it
+ must scope it to the caller the same way the asset list endpoints do. An
``asset`` leaf whose id is
+ not in ``readable_asset_ids`` keeps its place in the boolean tree (so the
shape of the schedule is
+ still honest) but has its identifying fields blanked and ``hidden`` set. A
leaf without an id, such
+ as a row not yet re-enriched by the Dag processor, cannot be authorized
and is hidden as well.
+
+ ``alias`` and ``asset_ref`` leaves are returned unchanged: the auth
manager exposes no batch
+ authorization for aliases, and references are unresolved names by design.
+
+ The input is never mutated: the value lives on an ORM instance whose
session commits on exit.
+ """
+ if expression is None:
+ return None
+ return _redact_node(expression, readable_asset_ids)
+
+
+def _redact_node(node: Any, readable_asset_ids: Collection[int]) -> Any:
+ if not isinstance(node, dict):
+ # Legacy pre-3.0 shapes hold bare strings; ``MaybeAssetExpression``
drops those later.
+ return node
+ if "asset" in node:
+ asset = node["asset"]
+ if not isinstance(asset, dict):
+ return node
+ if asset.get("id") in readable_asset_ids:
+ return {"asset": dict(asset)}
+ return {"asset": {"uri": None, "name": None, "group":
asset.get("group"), "id": None, "hidden": True}}
+ for key in ("all", "any"):
+ if key in node and isinstance(node[key], list):
+ return {key: [_redact_node(child, readable_asset_ids) for child in
node[key]]}
+ return node
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 75fd131d2a2..3caad7b34a6 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
@@ -50,12 +50,17 @@ class AssetExpressionAssetInfo(BaseModel):
persisted; ``BaseAsset.as_expression()`` itself only emits
``uri``/``name``/``group``. It is left
optional so a row persisted before id-enrichment (or migrated from the
pre-3.0 dataset format)
degrades gracefully instead of failing response validation.
+
+ A leaf the caller is not authorized to read is served with ``hidden`` set
and ``uri``, ``name``
+ and ``id`` blanked (see ``airflow.api_fastapi.common.asset_expression``),
so the shape of the
+ schedule stays visible without revealing which asset it waits on.
"""
- uri: str
- name: str
+ uri: str | None
+ name: str | None
group: str
id: int | None = None
+ hidden: bool = False
class AssetExpressionAliasInfo(BaseModel):
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 732c9202acf..7af3e1262f4 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
@@ -2450,10 +2450,14 @@ components:
AssetExpressionAssetInfo:
properties:
uri:
- type: string
+ anyOf:
+ - type: string
+ - type: 'null'
title: Uri
name:
- type: string
+ anyOf:
+ - type: string
+ - type: 'null'
title: Name
group:
type: string
@@ -2463,6 +2467,10 @@ components:
- type: integer
- type: 'null'
title: Id
+ hidden:
+ type: boolean
+ title: Hidden
+ default: false
type: object
required:
- uri
@@ -2481,7 +2489,16 @@ components:
optional so a row persisted before id-enrichment (or migrated from the
pre-3.0
dataset format)
- degrades gracefully instead of failing response validation.'
+ degrades gracefully instead of failing response validation.
+
+
+ A leaf the caller is not authorized to read is served with ``hidden``
set
+ and ``uri``, ``name``
+
+ and ``id`` blanked (see
``airflow.api_fastapi.common.asset_expression``),
+ so the shape of the
+
+ schedule stays visible without revealing which asset it waits on.'
AssetExpressionRef:
properties:
asset_ref:
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
index 6b6bae9b1dc..a9e9be01f4a 100644
---
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
+++
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
@@ -11366,10 +11366,14 @@ components:
AssetExpressionAssetInfo:
properties:
uri:
- type: string
+ anyOf:
+ - type: string
+ - type: 'null'
title: Uri
name:
- type: string
+ anyOf:
+ - type: string
+ - type: 'null'
title: Name
group:
type: string
@@ -11379,6 +11383,10 @@ components:
- type: integer
- type: 'null'
title: Id
+ hidden:
+ type: boolean
+ title: Hidden
+ default: false
type: object
required:
- uri
@@ -11397,7 +11405,16 @@ components:
optional so a row persisted before id-enrichment (or migrated from the
pre-3.0
dataset format)
- degrades gracefully instead of failing response validation.'
+ degrades gracefully instead of failing response validation.
+
+
+ A leaf the caller is not authorized to read is served with ``hidden``
set
+ and ``uri``, ``name``
+
+ and ``id`` blanked (see
``airflow.api_fastapi.common.asset_expression``),
+ so the shape of the
+
+ schedule stays visible without revealing which asset it waits on.'
AssetExpressionRef:
properties:
asset_ref:
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/assets.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/assets.py
index dd5feb55783..9fb73471300 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/assets.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/assets.py
@@ -22,6 +22,7 @@ import structlog
from fastapi import Depends, HTTPException, status
from sqlalchemy import ColumnElement, and_, case, exists, func, select, true
+from airflow.api_fastapi.common.asset_expression import redact_asset_expression
from airflow.api_fastapi.common.db.assets import
generate_assets_with_last_event_query
from airflow.api_fastapi.common.db.common import SessionDep, paginated_select
from airflow.api_fastapi.common.parameters import (
@@ -149,12 +150,17 @@ def get_assets(
)
def next_run_assets(
dag_id: str,
+ readable_assets_filter: ReadableAssetsFilterDep,
session: SessionDep,
) -> NextRunAssetsResponse:
dag_model = DagModel.get_dagmodel(dag_id, session=session)
if dag_model is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, f"Dag with id {dag_id}
was not found")
+ asset_expression = redact_asset_expression(
+ dag_model.asset_expression,
readable_asset_ids=readable_assets_filter.value or set()
+ )
+
latest_run = dag_model.get_last_dagrun(session=session)
event_filter = (
AssetEvent.timestamp >= latest_run.logical_date if latest_run and
latest_run.logical_date else true()
@@ -241,7 +247,7 @@ def next_run_assets(
)
for row in raw_rows
]
- model_data: dict[str, Any] = {"asset_expression":
dag_model.asset_expression, "events": events}
+ model_data: dict[str, Any] = {"asset_expression": asset_expression,
"events": events}
return NextRunAssetsResponse.model_validate(model_data)
# Partitioned Dags: enrich with per-asset received/required counts and
rollup flag.
@@ -276,7 +282,7 @@ def next_run_assets(
for row in raw_rows
]
model_data = {
- "asset_expression": dag_model.asset_expression,
+ "asset_expression": asset_expression,
"events": events,
"pending_partition_count": pending_partition_count,
}
@@ -350,7 +356,7 @@ def next_run_assets(
)
model_data = {
- "asset_expression": dag_model.asset_expression,
+ "asset_expression": asset_expression,
"events": events,
"pending_partition_count": pending_partition_count,
}
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 4771e7de12e..566cd1dba51 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
@@ -423,11 +423,25 @@ export const $AssetExpressionAsset = {
export const $AssetExpressionAssetInfo = {
properties: {
uri: {
- type: 'string',
+ anyOf: [
+ {
+ type: 'string'
+ },
+ {
+ type: 'null'
+ }
+ ],
title: 'Uri'
},
name: {
- type: 'string',
+ anyOf: [
+ {
+ type: 'string'
+ },
+ {
+ type: 'null'
+ }
+ ],
title: 'Name'
},
group: {
@@ -444,6 +458,11 @@ export const $AssetExpressionAssetInfo = {
}
],
title: 'Id'
+ },
+ hidden: {
+ type: 'boolean',
+ title: 'Hidden',
+ default: false
}
},
type: 'object',
@@ -454,7 +473,11 @@ export const $AssetExpressionAssetInfo = {
\`\`id\`\` is injected by
\`\`DagModelOperation.update_dag_asset_expression\`\` when the expression is
persisted; \`\`BaseAsset.as_expression()\`\` itself only emits
\`\`uri\`\`/\`\`name\`\`/\`\`group\`\`. It is left
optional so a row persisted before id-enrichment (or migrated from the pre-3.0
dataset format)
-degrades gracefully instead of failing response validation.`
+degrades gracefully instead of failing response validation.
+
+A leaf the caller is not authorized to read is served with \`\`hidden\`\` set
and \`\`uri\`\`, \`\`name\`\`
+and \`\`id\`\` blanked (see
\`\`airflow.api_fastapi.common.asset_expression\`\`), so the shape of the
+schedule stays visible without revealing which asset it waits on.`
} as const;
export const $AssetExpressionRef = {
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 dc08fe4d80e..91ddaf68f7d 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
@@ -126,12 +126,17 @@ export type AssetExpressionAsset = {
* persisted; ``BaseAsset.as_expression()`` itself only emits
``uri``/``name``/``group``. It is left
* optional so a row persisted before id-enrichment (or migrated from the
pre-3.0 dataset format)
* degrades gracefully instead of failing response validation.
+ *
+ * A leaf the caller is not authorized to read is served with ``hidden`` set
and ``uri``, ``name``
+ * and ``id`` blanked (see ``airflow.api_fastapi.common.asset_expression``),
so the shape of the
+ * schedule stays visible without revealing which asset it waits on.
*/
export type AssetExpressionAssetInfo = {
- uri: string;
- name: string;
+ uri: string | null;
+ name: string | null;
group: string;
id?: number | null;
+ hidden?: boolean;
};
/**
diff --git
a/airflow-core/tests/unit/api_fastapi/common/test_asset_expression.py
b/airflow-core/tests/unit/api_fastapi/common/test_asset_expression.py
new file mode 100644
index 00000000000..8f7f7d150e7
--- /dev/null
+++ b/airflow-core/tests/unit/api_fastapi/common/test_asset_expression.py
@@ -0,0 +1,73 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import copy
+
+import pytest
+
+from airflow.api_fastapi.common.asset_expression import redact_asset_expression
+
+_VISIBLE = {"asset": {"uri": "s3://bucket/visible", "name": "visible",
"group": "asset", "id": 1}}
+_HIDDEN = {"asset": {"uri": "s3://bucket/hidden", "name": "hidden", "group":
"asset", "id": 2}}
+_REDACTED = {"asset": {"uri": None, "name": None, "group": "asset", "id":
None, "hidden": True}}
+_ALIAS = {"alias": {"name": "my_alias", "group": "asset"}}
+_REF = {"asset_ref": {"name": "by_name"}}
+
+
+def test_none_passes_through():
+ assert redact_asset_expression(None, readable_asset_ids={1}) is None
+
+
[email protected](
+ ("expression", "expected"),
+ [
+ pytest.param(_VISIBLE, _VISIBLE, id="readable_leaf_unchanged"),
+ pytest.param(_HIDDEN, _REDACTED, id="unreadable_leaf_redacted"),
+ pytest.param(
+ {"asset": {"uri": "s3://b", "name": "n", "group": "asset"}},
+ _REDACTED,
+ id="leaf_without_id_fails_closed",
+ ),
+ pytest.param(_ALIAS, _ALIAS, id="alias_untouched"),
+ pytest.param(_REF, _REF, id="asset_ref_untouched"),
+ pytest.param(
+ {"all": [_VISIBLE, {"any": [_HIDDEN, _ALIAS]}, _HIDDEN]},
+ {"all": [_VISIBLE, {"any": [_REDACTED, _ALIAS]}, _REDACTED]},
+ id="nested_keeps_shape",
+ ),
+ pytest.param(
+ {"any": ["s3://legacy-a", "s3://legacy-b"]},
+ {"any": ["s3://legacy-a", "s3://legacy-b"]},
+ id="legacy_string_leaves_left_for_field_coercion",
+ ),
+ ],
+)
+def test_redact(expression, expected):
+ assert redact_asset_expression(expression, readable_asset_ids={1}) ==
expected
+
+
+def test_does_not_mutate_input():
+ expression = {"all": [_VISIBLE, _HIDDEN]}
+ snapshot = copy.deepcopy(expression)
+
+ redacted = redact_asset_expression(expression, readable_asset_ids={1})
+
+ assert expression == snapshot
+ assert redacted is not expression
+ assert redacted["all"][0] is not expression["all"][0]
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/datamodels/test_common.py
b/airflow-core/tests/unit/api_fastapi/core_api/datamodels/test_common.py
index 2bf6fddb122..d1c1515d6a9 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/datamodels/test_common.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/datamodels/test_common.py
@@ -51,7 +51,17 @@ _REF_BY_URI = {"asset_ref": {"uri": "s3://bucket/key"}}
def test_asset_expression_round_trips_unchanged(expression: dict):
"""The typed model must accept and re-serialize each stored expression
byte-identically."""
validated = _adapter.validate_python(expression)
- assert _adapter.dump_python(validated, by_alias=True) == expression
+ # ``hidden`` is an API-only marker that stored rows never carry, so it is
excluded when unset.
+ assert _adapter.dump_python(validated, by_alias=True, exclude_unset=True)
== expression
+
+
+def test_asset_expression_accepts_redacted_leaf():
+ """A leaf redacted for an unauthorized caller keeps its place in the tree
with blanked identity."""
+ redacted = {"asset": {"uri": None, "name": None, "group": "asset", "id":
None, "hidden": True}}
+ validated = _adapter.validate_python({"all": [_ASSET, redacted]})
+ assert validated.all[1].asset.hidden is True
+ assert validated.all[0].asset.hidden is False
+ assert _adapter.dump_python(validated, by_alias=True)["all"][1] == redacted
def test_asset_expression_tolerates_legacy_asset_leaf_without_id():
@@ -114,4 +124,4 @@ def test_field_preserves_current_shapes(expression):
if expression is None:
assert validated is None
else:
- assert _field_adapter.dump_python(validated, by_alias=True) ==
expression
+ assert _field_adapter.dump_python(validated, by_alias=True,
exclude_unset=True) == expression
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_assets.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_assets.py
index ef4ca97fb2b..e0a87013c4d 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_assets.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_assets.py
@@ -16,6 +16,7 @@
# under the License.
from __future__ import annotations
+import json
from unittest import mock
from unittest.mock import MagicMock
@@ -83,7 +84,8 @@ class TestNextRunAssets:
dag_maker.create_dagrun()
dag_maker.sync_dagbag_to_db()
- with assert_queries_count(4):
+ # 4 queries for the endpoint plus 1 to resolve the assets the caller
may read.
+ with assert_queries_count(5):
response = test_client.get("/next_run_assets/upstream")
assert response.status_code == 200
@@ -96,6 +98,7 @@ class TestNextRunAssets:
"name": "asset1",
"group": "asset",
"id": mock.ANY,
+ "hidden": False,
}
}
]
@@ -118,6 +121,47 @@ class TestNextRunAssets:
"pending_partition_count": None,
}
+ @mock.patch(
+
"airflow.api_fastapi.auth.managers.base_auth_manager.BaseAuthManager.get_authorized_assets",
+ autospec=True,
+ )
+ def test_asset_expression_hides_assets_the_caller_may_not_read(
+ self, mock_get_authorized_assets, test_client, dag_maker, session
+ ):
+ with dag_maker(
+ dag_id="hidden_upstream",
+ schedule=[
+ Asset(uri="s3://bucket/visible", name="visible_asset"),
+ Asset(uri="s3://bucket/hidden", name="hidden_asset"),
+ ],
+ serialized=True,
+ ):
+ EmptyOperator(task_id="task1")
+ dag_maker.sync_dagbag_to_db()
+ visible_id =
session.scalar(select(AssetModel.id).where(AssetModel.name == "visible_asset"))
+ mock_get_authorized_assets.return_value = {visible_id}
+
+ response = test_client.get("/next_run_assets/hidden_upstream")
+
+ assert response.status_code == 200
+ assert response.json()["asset_expression"] == {
+ "all": [
+ {
+ "asset": {
+ "uri": "s3://bucket/visible",
+ "name": "visible_asset",
+ "group": "asset",
+ "id": visible_id,
+ "hidden": False,
+ }
+ },
+ {"asset": {"uri": None, "name": None, "group": "asset", "id":
None, "hidden": True}},
+ ]
+ }
+ redacted = json.dumps(response.json()["asset_expression"])
+ assert "hidden_asset" not in redacted
+ assert "s3://bucket/hidden" not in redacted
+
def test_should_respond_401(self, unauthenticated_test_client):
response = unauthenticated_test_client.get("/next_run_assets/upstream")
assert response.status_code == 401
@@ -170,6 +214,7 @@ class TestNextRunAssets:
"name": "A",
"group": "asset",
"id": mock.ANY,
+ "hidden": False,
}
},
{
@@ -178,6 +223,7 @@ class TestNextRunAssets:
"name": "B",
"group": "asset",
"id": mock.ANY,
+ "hidden": False,
}
},
]
diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
index 5e340d7667d..cd3210e6699 100644
--- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
+++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
@@ -78,12 +78,17 @@ class AssetExpressionAssetInfo(BaseModel):
persisted; ``BaseAsset.as_expression()`` itself only emits
``uri``/``name``/``group``. It is left
optional so a row persisted before id-enrichment (or migrated from the
pre-3.0 dataset format)
degrades gracefully instead of failing response validation.
+
+ A leaf the caller is not authorized to read is served with ``hidden`` set
and ``uri``, ``name``
+ and ``id`` blanked (see ``airflow.api_fastapi.common.asset_expression``),
so the shape of the
+ schedule stays visible without revealing which asset it waits on.
"""
- uri: Annotated[str, Field(title="Uri")]
- name: Annotated[str, Field(title="Name")]
+ uri: Annotated[str | None, Field(title="Uri")]
+ name: Annotated[str | None, Field(title="Name")]
group: Annotated[str, Field(title="Group")]
id: Annotated[int | None, Field(title="Id")] = None
+ hidden: Annotated[bool | None, Field(title="Hidden")] = False
class AssetExpressionRef(BaseModel):
