This is an automated email from the ASF dual-hosted git repository.
vincbeck pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new edcd4a50335 Hide assets the caller may not read from the UI
dependencies graph (#73008)
edcd4a50335 is described below
commit edcd4a50335e87fa6e7a0b3fc1ab72d42c7cbca7
Author: Henry Chen <[email protected]>
AuthorDate: Tue Oct 20 02:13:14 2026 +0800
Hide assets the caller may not read from the UI dependencies graph (#73008)
* Hide assets the caller may not read from the UI dependencies graph
The dependencies graph scoped its nodes by which Dags the caller may read,
but never by which assets. A caller who can read a Dag saw every asset that
Dag produces or consumes, name included, even when the auth manager scopes
that asset away from them; the data graph would also expand outward through
such an asset. Asset names commonly encode bucket, table or dataset names,
so this is the same class of gap the rest of the asset API closed.
* Hide an unenriched asset leaf from the data dependencies graph
An asset_expression leaf whose id the Dag processor has not filled in yet
cannot be authorized, so it must be hidden like any other asset the caller
may not read. The data graph instead parsed the id out of the node, which
failed the request outright rather than dropping the node.
---
.../api_fastapi/core_api/routes/ui/dependencies.py | 20 ++-
.../core_api/services/ui/dependencies.py | 74 ++++++--
.../core_api/routes/ui/test_dependencies.py | 189 ++++++++++++++++++++-
3 files changed, 262 insertions(+), 21 deletions(-)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dependencies.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dependencies.py
index e1e44616704..25129a280d2 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dependencies.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dependencies.py
@@ -27,7 +27,11 @@ from airflow.api_fastapi.common.db.common import SessionDep
from airflow.api_fastapi.common.router import AirflowRouter
from airflow.api_fastapi.core_api.datamodels.ui.common import BaseGraphResponse
from airflow.api_fastapi.core_api.openapi.exceptions import
create_openapi_http_exception_doc
-from airflow.api_fastapi.core_api.security import ReadableDagsFilterDep,
requires_access_dag
+from airflow.api_fastapi.core_api.security import (
+ ReadableAssetsFilterDep,
+ ReadableDagsFilterDep,
+ requires_access_dag,
+)
from airflow.api_fastapi.core_api.services.ui.dependencies import (
extract_single_connected_component,
get_data_dependencies,
@@ -50,6 +54,7 @@ dependencies_router = AirflowRouter(tags=["Dependencies"])
def get_dependencies(
session: SessionDep,
readable_dags_filter: ReadableDagsFilterDep,
+ readable_assets_filter: ReadableAssetsFilterDep,
node_id: str | None = None,
dependency_type: Literal["scheduling", "data"] = "scheduling",
) -> BaseGraphResponse:
@@ -65,12 +70,21 @@ def get_dependencies(
except ValueError:
raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Invalid asset
node_id: {node_id}")
- data = get_data_dependencies(asset_id, session,
readable_dags_filter.value)
+ data = get_data_dependencies(
+ asset_id,
+ session,
+ readable_dag_ids=readable_dags_filter.value,
+ readable_asset_ids=readable_assets_filter.value or set(),
+ )
if not data["nodes"]:
raise HTTPException(status.HTTP_404_NOT_FOUND, f"Asset with id
{asset_id} was not found")
return BaseGraphResponse(**data)
- data = get_scheduling_dependencies(readable_dags_filter.value, session)
+ data = get_scheduling_dependencies(
+ session,
+ readable_dag_ids=readable_dags_filter.value,
+ readable_asset_ids=readable_assets_filter.value or set(),
+ )
if node_id is not None:
try:
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/services/ui/dependencies.py
b/airflow-core/src/airflow/api_fastapi/core_api/services/ui/dependencies.py
index eb2e665e889..e9b8e4c9076 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/services/ui/dependencies.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/services/ui/dependencies.py
@@ -39,7 +39,8 @@ def _asset_node_id_and_label(asset: dict) -> tuple[str, str]:
asset_type = asset["type"]
if asset_type == "asset":
- return f"asset:{asset['id']}", asset["name"]
+ # A leaf the Dag processor has not enriched yet has no id;
hide_unreadable_assets drops it.
+ return f"asset:{asset.get('id')}", asset["name"]
if asset_type in ("asset-alias", "asset-name-ref"):
return f"{asset_type}:{asset['name']}", asset["name"]
if asset_type == "asset-uri-ref":
@@ -189,7 +190,41 @@ def extract_single_connected_component(
return {"nodes": nodes, "edges": edges}
-def get_scheduling_dependencies(readable_dag_ids: set[str] | None, session:
Session) -> dict[str, list[dict]]:
+def _is_readable_asset_node(node_id: str, readable_asset_ids: set[int]) ->
bool:
+ suffix = node_id.removeprefix("asset:")
+ return suffix.isdigit() and int(suffix) in readable_asset_ids
+
+
+def hide_unreadable_assets(
+ graph: dict[str, list[dict]], readable_asset_ids: set[int]
+) -> dict[str, list[dict]]:
+ """
+ Drop asset nodes the caller may not read, and every edge touching them,
from a graph.
+
+ Alias and ref nodes carry no asset id to authorize on, so they are left in
place. An asset node
+ without a numeric id, from an asset_expression the Dag processor has not
re-enriched yet, cannot
+ be authorized and is hidden as well.
+ """
+ hidden = {
+ node["id"]
+ for node in graph["nodes"]
+ if node["type"] == "asset" and not _is_readable_asset_node(node["id"],
readable_asset_ids)
+ }
+ if not hidden:
+ return graph
+ return {
+ "nodes": [node for node in graph["nodes"] if node["id"] not in hidden],
+ "edges": [
+ edge
+ for edge in graph["edges"]
+ if edge["source_id"] not in hidden and edge["target_id"] not in
hidden
+ ],
+ }
+
+
+def get_scheduling_dependencies(
+ session: Session, *, readable_dag_ids: set[str] | None,
readable_asset_ids: set[int]
+) -> dict[str, list[dict]]:
"""Get scheduling dependencies between Dags."""
from airflow.models.serialized_dag import SerializedDagModel
@@ -307,10 +342,13 @@ def get_scheduling_dependencies(readable_dag_ids:
set[str] | None, session: Sess
if team_name:
node["team"] = team_name
- return {
- "nodes": list(nodes_dict.values()),
- "edges": [{"source_id": source, "target_id": target} for source,
target in sorted(edge_tuples)],
- }
+ return hide_unreadable_assets(
+ {
+ "nodes": list(nodes_dict.values()),
+ "edges": [{"source_id": source, "target_id": target} for source,
target in sorted(edge_tuples)],
+ },
+ readable_asset_ids,
+ )
def _entry_task_id_from_serialized_data(data: dict | None) -> str | None:
@@ -391,7 +429,11 @@ def _get_dag_entry_points(dag_ids: set[str], session:
Session) -> dict[str, tupl
def get_data_dependencies(
- asset_id: int, session: Session, readable_dag_ids: set[str] | None = None
+ asset_id: int,
+ session: Session,
+ *,
+ readable_dag_ids: set[str] | None = None,
+ readable_asset_ids: set[int],
) -> dict[str, list[dict]]:
"""Get full task dependencies for an asset."""
from sqlalchemy import select, tuple_, union_all
@@ -411,6 +453,10 @@ def get_data_dependencies(
# reachable through other readable dags) even though the user has no
legitimate
# lineage connection to it. A readable_dag_ids value of None means no
filter is
# applied (the user has unrestricted dag read access).
+ # Same empty result as an unrelated asset, so an unreadable asset id is
not distinguishable
+ # from a nonexistent one.
+ if asset_id not in readable_asset_ids:
+ return {"nodes": [], "edges": []}
if readable_dag_ids is not None:
connected_dag_ids_query = union_all(
select(TaskOutletAssetReference.dag_id).where(TaskOutletAssetReference.asset_id
== asset_id),
@@ -442,6 +488,7 @@ def get_data_dependencies(
while frontier:
round_asset_ids = frontier - processed_assets
processed_assets |= round_asset_ids
+ round_asset_ids &= readable_asset_ids
if not round_asset_ids:
break
@@ -572,7 +619,7 @@ def get_data_dependencies(
nodes_dict.setdefault(node["id"], node)
# Continue the BFS through any other concrete assets this
gate combines
# with, so their own producers are traced too.
- if node["id"].startswith("asset:"):
+ if _is_readable_asset_node(node["id"], readable_asset_ids):
sibling_asset_id =
int(node["id"].removeprefix("asset:"))
if sibling_asset_id not in processed_assets:
next_frontier.add(sibling_asset_id)
@@ -661,7 +708,10 @@ def get_data_dependencies(
if team_name:
node["team"] = team_name
- return {
- "nodes": list(nodes_dict.values()),
- "edges": [{"source_id": source, "target_id": target} for source,
target in edge_set],
- }
+ return hide_unreadable_assets(
+ {
+ "nodes": list(nodes_dict.values()),
+ "edges": [{"source_id": source, "target_id": target} for source,
target in edge_set],
+ },
+ readable_asset_ids,
+ )
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dependencies.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dependencies.py
index 426c11134d0..1c678857abb 100644
---
a/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dependencies.py
+++
b/airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dependencies.py
@@ -234,7 +234,7 @@ def expected_secondary_component_response(asset2_id):
class TestGetDependencies:
@pytest.mark.usefixtures("make_primary_connected_component")
def test_should_response_200(self, test_client,
expected_primary_component_response):
- with assert_queries_count(8):
+ with assert_queries_count(9):
response = test_client.get("/dependencies")
assert response.status_code == 200
@@ -270,7 +270,7 @@ class TestGetDependencies:
@pytest.mark.usefixtures("make_primary_connected_component",
"make_secondary_connected_component")
def test_with_node_id_filter(self, test_client, node_id,
expected_response_fixture, request):
expected_response = request.getfixturevalue(expected_response_fixture)
- with assert_queries_count(8):
+ with assert_queries_count(9):
response = test_client.get("/dependencies", params={"node_id":
node_id})
assert response.status_code == 200
@@ -288,7 +288,7 @@ class TestGetDependencies:
(asset1_id, expected_primary_component_response),
(asset2_id, expected_secondary_component_response),
):
- with assert_queries_count(8):
+ with assert_queries_count(9):
response = test_client.get("/dependencies", params={"node_id":
f"asset:{asset_id}"})
assert response.status_code == 200
@@ -502,6 +502,183 @@ class TestGetDependencies:
)
assert response.status_code == 404
+ @mock.patch(
+
"airflow.api_fastapi.auth.managers.base_auth_manager.BaseAuthManager.get_authorized_assets",
+ autospec=True,
+ return_value=set(),
+ )
+ @pytest.mark.usefixtures("make_primary_connected_component")
+ def
test_scheduling_dependencies_hides_assets_the_caller_may_not_read(self, _,
test_client, asset1_id):
+ """Reading the Dags around an asset must not expose the asset itself,
nor its edges."""
+ response = test_client.get("/dependencies")
+
+ assert response.status_code == 200
+ result = response.json()
+ node_ids = {node["id"] for node in result["nodes"]}
+ assert f"asset:{asset1_id}" not in node_ids
+ assert {"dag:upstream", "dag:downstream"} <= node_ids
+ edge_endpoints = {edge["source_id"] for edge in result["edges"]} | {
+ edge["target_id"] for edge in result["edges"]
+ }
+ assert f"asset:{asset1_id}" not in edge_endpoints
+
+ @mock.patch(
+
"airflow.api_fastapi.auth.managers.base_auth_manager.BaseAuthManager.get_authorized_assets",
+ autospec=True,
+ )
+ def test_scheduling_dependencies_hides_unreadable_gate_leaf(
+ self, mock_get_authorized_assets, dag_maker, test_client, session
+ ):
+ asset_a = Asset(uri="s3://gate-bucket/a", name="gate_asset_a")
+ asset_b = Asset(uri="s3://gate-bucket/b", name="gate_asset_b")
+ with dag_maker(
+ dag_id="gate_downstream", schedule=(asset_a & asset_b),
serialized=True, session=session
+ ):
+ EmptyOperator(task_id="consume")
+ dag_maker.sync_dagbag_to_db()
+ asset_a_id =
session.scalar(select(AssetModel.id).where(AssetModel.name == "gate_asset_a"))
+ asset_b_id =
session.scalar(select(AssetModel.id).where(AssetModel.name == "gate_asset_b"))
+ mock_get_authorized_assets.return_value = {asset_a_id}
+
+ response = test_client.get("/dependencies", params={"node_id":
"dag:gate_downstream"})
+
+ assert response.status_code == 200
+ result = response.json()
+ node_ids = {node["id"] for node in result["nodes"]}
+ assert f"asset:{asset_a_id}" in node_ids
+ assert f"asset:{asset_b_id}" not in node_ids
+ assert "gate_asset_b" not in {node["label"] for node in
result["nodes"]}
+ gate_node = next(node for node in result["nodes"] if node["type"] ==
"asset-condition")
+ edge_tuples = {(edge["source_id"], edge["target_id"]) for edge in
result["edges"]}
+ assert (f"asset:{asset_a_id}", gate_node["id"]) in edge_tuples
+ assert (f"asset:{asset_b_id}", gate_node["id"]) not in edge_tuples
+
+ @mock.patch(
+
"airflow.api_fastapi.auth.managers.base_auth_manager.BaseAuthManager.get_authorized_assets",
+ autospec=True,
+ return_value=set(),
+ )
+ @pytest.mark.usefixtures("make_primary_connected_component")
+ def
test_scheduling_dependencies_with_unreadable_asset_node_id_responds_404(
+ self, _, test_client, asset1_id
+ ):
+ """Selecting an unreadable asset is served like a nonexistent node, so
its existence does not leak."""
+ response = test_client.get("/dependencies", params={"node_id":
f"asset:{asset1_id}"})
+
+ assert response.status_code == 404
+ assert "asset1" not in response.text
+
+ def test_scheduling_dependencies_hides_gate_leaf_without_id(self,
dag_maker, test_client, session):
+ """A leaf the Dag processor has not enriched with an id yet cannot be
authorized, so it is hidden."""
+ asset_a = Asset(uri="s3://gate-bucket/a", name="gate_asset_a")
+ asset_b = Asset(uri="s3://gate-bucket/b", name="gate_asset_b")
+ with dag_maker(
+ dag_id="gate_downstream", schedule=(asset_a & asset_b),
serialized=True, session=session
+ ):
+ EmptyOperator(task_id="consume")
+ dag_maker.sync_dagbag_to_db()
+ dag_model = session.get(DagModel, "gate_downstream")
+ stale_expression = {
+ "all": [
+ {"asset": {"name": "gate_asset_a", "uri":
"s3://gate-bucket/a", "id": None}},
+ {"asset": {"name": "gate_asset_b", "uri":
"s3://gate-bucket/b"}},
+ ]
+ }
+ dag_model.asset_expression = stale_expression
+ session.commit()
+
+ response = test_client.get("/dependencies", params={"node_id":
"dag:gate_downstream"})
+
+ assert response.status_code == 200
+ result = response.json()
+ assert not any(node["type"] == "asset" for node in result["nodes"])
+ assert {"gate_asset_a", "gate_asset_b"}.isdisjoint(node["label"] for
node in result["nodes"])
+
+ @mock.patch(
+
"airflow.api_fastapi.auth.managers.base_auth_manager.BaseAuthManager.get_authorized_assets",
+ autospec=True,
+ return_value=set(),
+ )
+ @pytest.mark.usefixtures("make_primary_connected_component")
+ def test_data_dependencies_hides_asset_the_caller_may_not_read(self, _,
test_client, asset1_id):
+ """An unreadable root asset is served like a nonexistent one, even
when its Dags are readable."""
+ response = test_client.get(
+ "/dependencies", params={"node_id": f"asset:{asset1_id}",
"dependency_type": "data"}
+ )
+
+ assert response.status_code == 404
+
+ @mock.patch(
+
"airflow.api_fastapi.auth.managers.base_auth_manager.BaseAuthManager.get_authorized_assets",
+ autospec=True,
+ )
+ def test_data_dependencies_hides_unreadable_sibling_asset(
+ self, mock_get_authorized_assets, dag_maker, test_client, session
+ ):
+ """The BFS through a gate must neither name an unreadable sibling
asset nor expand past it."""
+ asset_a = Asset(uri="s3://gate-bucket/a", name="gate_asset_a")
+ asset_b = Asset(uri="s3://gate-bucket/b", name="gate_asset_b")
+ with dag_maker(dag_id="gate_upstream_b", serialized=True,
session=session):
+ EmptyOperator(task_id="produce_b", outlets=[asset_b])
+ with dag_maker(
+ dag_id="gate_downstream", schedule=(asset_a & asset_b),
serialized=True, session=session
+ ):
+ EmptyOperator(task_id="consume")
+ dag_maker.sync_dagbag_to_db()
+ asset_a_id =
session.scalar(select(AssetModel.id).where(AssetModel.name == "gate_asset_a"))
+ asset_b_id =
session.scalar(select(AssetModel.id).where(AssetModel.name == "gate_asset_b"))
+ mock_get_authorized_assets.return_value = {asset_a_id}
+
+ response = test_client.get(
+ "/dependencies", params={"node_id": f"asset:{asset_a_id}",
"dependency_type": "data"}
+ )
+
+ assert response.status_code == 200
+ result = response.json()
+ node_ids = {node["id"] for node in result["nodes"]}
+ assert f"asset:{asset_a_id}" in node_ids
+ assert f"asset:{asset_b_id}" not in node_ids
+ assert "gate_asset_b" not in {node["label"] for node in
result["nodes"]}
+ assert not any(node_id.startswith("task:gate_upstream_b") for node_id
in node_ids)
+
+ @mock.patch(
+
"airflow.api_fastapi.auth.managers.base_auth_manager.BaseAuthManager.get_authorized_assets",
+ autospec=True,
+ )
+ def test_data_dependencies_hides_gate_leaf_without_id(
+ self, mock_get_authorized_assets, dag_maker, test_client, session
+ ):
+ """A gate leaf the Dag processor has not enriched with an id yet is
hidden, not a 500."""
+ asset_a = Asset(uri="s3://gate-bucket/a", name="gate_asset_a")
+ asset_b = Asset(uri="s3://gate-bucket/b", name="gate_asset_b")
+ with dag_maker(
+ dag_id="gate_downstream", schedule=(asset_a & asset_b),
serialized=True, session=session
+ ):
+ EmptyOperator(task_id="consume")
+ dag_maker.sync_dagbag_to_db()
+ asset_a_id =
session.scalar(select(AssetModel.id).where(AssetModel.name == "gate_asset_a"))
+ asset_b_id =
session.scalar(select(AssetModel.id).where(AssetModel.name == "gate_asset_b"))
+ mock_get_authorized_assets.return_value = {asset_a_id, asset_b_id}
+ dag_model = session.get(DagModel, "gate_downstream")
+ dag_model.asset_expression = {
+ "all": [
+ {"asset": {"name": "gate_asset_a", "uri":
"s3://gate-bucket/a", "id": asset_a_id}},
+ {"asset": {"name": "gate_asset_b", "uri":
"s3://gate-bucket/b", "id": None}},
+ ]
+ }
+ session.commit()
+
+ response = test_client.get(
+ "/dependencies", params={"node_id": f"asset:{asset_a_id}",
"dependency_type": "data"}
+ )
+
+ assert response.status_code == 200
+ result = response.json()
+ node_ids = {node["id"] for node in result["nodes"]}
+ assert f"asset:{asset_a_id}" in node_ids
+ assert "asset:None" not in node_ids
+ assert "gate_asset_b" not in {node["label"] for node in
result["nodes"]}
+
def test_scheduling_dependencies_include_team_name(self, dag_maker,
test_client, session):
team = Team(name="my-team")
session.add(team)
@@ -856,7 +1033,7 @@ class TestGetDependencies:
asset_a_id =
session.scalar(select(AssetModel.id).where(AssetModel.name ==
"data_dep_batch_asset_a"))
- with assert_queries_count(12):
+ with assert_queries_count(13):
response = test_client.get(
"/dependencies", params={"node_id": f"asset:{asset_a_id}",
"dependency_type": "data"}
)
@@ -891,7 +1068,7 @@ class TestGetDependencies:
select(AssetModel.id).where(AssetModel.name ==
"data_dep_inlet_batch_asset_a")
)
- expected_query_count = 9
+ expected_query_count = 10
with assert_queries_count(expected_query_count):
response = test_client.get(
"/dependencies", params={"node_id": f"asset:{asset_a_id}",
"dependency_type": "data"}
@@ -927,7 +1104,7 @@ class TestGetDependencies:
select(AssetModel.id).where(AssetModel.name ==
"data_dep_outlet_batch_asset_a")
)
- expected_query_count = 9
+ expected_query_count = 10
with assert_queries_count(expected_query_count):
response = test_client.get(
"/dependencies", params={"node_id": f"asset:{asset_a_id}",
"dependency_type": "data"}
