From 9b3db52479bb0b7a2d11c89f144dc6034bef7a76 Mon Sep 17 00:00:00 2001 From: Ephraim Anierobi Date: Tue, 14 Jul 2026 10:31:27 +0100 Subject: [PATCH 1/4] Add serialized Dag version diff engine Dag authors and operators need a reliable way to understand how two stored Dag versions differ without downloading serialized payloads or manually comparing implementation details. This adds the shared comparison engine that later API and CLI surfaces build on. Historical serialized Dags can use different schema shapes, so comparisons normalize task, schedule, dependency, and metadata state before calculating changes. Malformed legacy payloads and missing historical data degrade to unavailable results instead of failing the whole request. --- .../src/airflow/models/dag_version.py | 129 +++++- .../airflow/serialization/dag_version_diff.py | 365 ++++++++++++++++ .../tests/unit/models/test_dag_version.py | 92 +++- .../serialization/test_dag_version_diff.py | 411 ++++++++++++++++++ 4 files changed, 994 insertions(+), 3 deletions(-) create mode 100644 airflow-core/src/airflow/serialization/dag_version_diff.py create mode 100644 airflow-core/tests/unit/serialization/test_dag_version_diff.py diff --git a/airflow-core/src/airflow/models/dag_version.py b/airflow-core/src/airflow/models/dag_version.py index 8aa7efcab01e2..c8db20166d9ee 100644 --- a/airflow-core/src/airflow/models/dag_version.py +++ b/airflow-core/src/airflow/models/dag_version.py @@ -17,9 +17,10 @@ from __future__ import annotations +import hashlib import logging from datetime import datetime -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, Literal from uuid import UUID import sqlalchemy as sa @@ -40,6 +41,13 @@ log = logging.getLogger(__name__) +SourceStatus = Literal["current_stored_code", "redacted", "unavailable"] +ValuesStatus = Literal["available", "unavailable"] + + +class DagVersionNotFoundError(ValueError): + """Raised when one of the requested Dag versions does not exist.""" + class DagVersion(Base): """Model to track the versions of DAGs in the database.""" @@ -243,6 +251,125 @@ def version(self) -> str: """A human-friendly representation of the version.""" return f"{self.dag_id}-{self.version_number}" + @classmethod + @provide_session + def get_diff( + cls, + dag_id: str, + base_version_number: int, + target_version_number: int, + *, + include_values: bool = False, + include_source: bool = False, + max_changes: int | None = None, + source_status: SourceStatus | None = None, + values_status: ValuesStatus | None = None, + session: Session = NEW_SESSION, + ) -> dict[str, Any]: + """ + Compare two versions of a Dag using their currently stored state. + + ``source_status`` is supplied by callers that have an authorization context. The CLI has + operator-level authority and leaves it unset, while API callers calculate it before this + method returns any source content. + """ + from airflow.serialization.dag_version_diff import ( + DEFAULT_MAX_CHANGES, + build_serialized_dag_diff, + ) + + if max_changes is None: + max_changes = DEFAULT_MAX_CHANGES + if base_version_number < 1 or target_version_number < 1: + raise ValueError("Dag version numbers must be positive integers") + + query = ( + select(cls) + .where( + cls.dag_id == dag_id, + cls.version_number.in_((base_version_number, target_version_number)), + ) + .options(joinedload(cls.serialized_dag)) + ) + if include_source and source_status not in {"redacted", "unavailable"}: + query = query.options(joinedload(cls.dag_code)) + + versions = {version.version_number: version for version in session.scalars(query).all()} + missing_version = next( + ( + version_number + for version_number in (base_version_number, target_version_number) + if version_number not in versions + ), + None, + ) + if missing_version is not None: + raise DagVersionNotFoundError( + f"The DagVersion with dag_id: `{dag_id}` and version_number: `{missing_version}` was not found" + ) + + base_version = versions[base_version_number] + target_version = versions[target_version_number] + effective_include_values = include_values and values_status != "unavailable" + result = build_serialized_dag_diff( + base_data=base_version.serialized_dag.data if base_version.serialized_dag else None, + target_data=target_version.serialized_dag.data if target_version.serialized_dag else None, + base_provenance=_get_provenance(base_version), + target_provenance=_get_provenance(target_version), + include_values=effective_include_values, + max_changes=max_changes, + ) + if include_values: + values_available = effective_include_values and result["mode"] == "observed_state" + result["values"] = {"status": "available" if values_available else "unavailable"} + result["source"] = _get_source_diff( + base_version, + target_version, + include_source=include_source, + source_status=source_status or ("current_stored_code" if include_source else "unavailable"), + ) + return result + + +def _get_provenance(version: DagVersion) -> dict[str, Any]: + return { + "bundle_name": version.bundle_name, + "bundle_version": version.bundle_version, + "version_data": version.version_data, + } + + +def _get_source_diff( + base_version: DagVersion, + target_version: DagVersion, + *, + include_source: bool, + source_status: SourceStatus, +) -> dict[str, Any]: + if not include_source: + return {"status": "unavailable", "fidelity": "unavailable"} + if source_status != "current_stored_code": + return {"status": source_status, "fidelity": source_status} + + base_code = base_version.dag_code + target_code = target_version.dag_code + if base_code is None or target_code is None: + return {"status": "unavailable", "fidelity": "unavailable"} + + base_source = base_code.source_code + target_source = target_code.source_code + return { + "status": "current_stored_code", + "fidelity": "current_stored_code", + "changed": base_source != target_source, + "base": {"digest": _get_source_digest(base_source), "content": base_source}, + "target": {"digest": _get_source_digest(target_source), "content": target_source}, + } + + +def _get_source_digest(source: str) -> str: + return f"sha256:{hashlib.sha256(source.encode()).hexdigest()}" + def _resolve_version_data( dag_version: DagVersion | None, bundle_version: str | None diff --git a/airflow-core/src/airflow/serialization/dag_version_diff.py b/airflow-core/src/airflow/serialization/dag_version_diff.py new file mode 100644 index 0000000000000..eba5c259e1dbe --- /dev/null +++ b/airflow-core/src/airflow/serialization/dag_version_diff.py @@ -0,0 +1,365 @@ +# 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. + +"""Observed-state diffs for serialized Dag payloads.""" + +from __future__ import annotations + +import copy +import hashlib +import json +from collections.abc import Callable, Mapping +from typing import Any, Literal + +from airflow.serialization.serialized_objects import DagSerialization + +DIFF_SCHEMA_VERSION = 1 +DEFAULT_MAX_CHANGES = 500 +MAX_ALLOWED_CHANGES = 5000 +SUPPORTED_SERIALIZED_DAG_SCHEMA_VERSIONS = frozenset((1, 2, 3)) + +DiffMode = Literal["observed_state", "unavailable"] + +_ORDER_INSENSITIVE_LIST_PATHS = { + ("dag", "tags"), + ("dag", "allowed_run_types"), +} + + +def build_serialized_dag_diff( + *, + base_data: dict[str, Any] | None, + target_data: dict[str, Any] | None, + base_provenance: Mapping[str, Any] | None = None, + target_provenance: Mapping[str, Any] | None = None, + include_values: bool = False, + max_changes: int = DEFAULT_MAX_CHANGES, +) -> dict[str, Any]: + """Build a bounded, deterministic diff from two stored serialized Dag payloads.""" + _validate_max_changes(max_changes) + + base_schema_version = _get_schema_version(base_data) + target_schema_version = _get_schema_version(target_data) + result: dict[str, Any] = { + "diff_schema_version": DIFF_SCHEMA_VERSION, + "serialized_dag_schema_versions": { + "base": base_schema_version, + "target": target_schema_version, + }, + "mode": "observed_state", + "changes": [], + "truncated": False, + } + + if base_data is None or target_data is None: + return _unavailable(result, "serialized_dag_missing") + + if base_schema_version is None or target_schema_version is None: + return _unavailable(result, "serialized_dag_schema_version_missing") + + unsupported_versions = [ + version + for version in (base_schema_version, target_schema_version) + if version not in SUPPORTED_SERIALIZED_DAG_SCHEMA_VERSIONS + ] + if unsupported_versions: + return _unavailable(result, f"unsupported_serialized_dag_schema_version:{unsupported_versions[0]}") + + try: + base_document = _canonicalize_payload(base_data) + target_document = _canonicalize_payload(target_data) + except (AttributeError, KeyError, OverflowError, TypeError, ValueError): + return _unavailable(result, "serialized_dag_canonicalization_failed") + + base_document["provenance"] = _canonicalize_value(dict(base_provenance or {}), path=("provenance",)) + target_document["provenance"] = _canonicalize_value(dict(target_provenance or {}), path=("provenance",)) + + collector = _ChangeCollector(max_changes=max_changes, include_values=include_values) + _collect_changes(base_document, target_document, path=(), collector=collector) + result["changes"] = collector.changes + result["truncated"] = collector.count > max_changes + return result + + +class _ChangeCollector: + def __init__(self, *, max_changes: int, include_values: bool) -> None: + self.changes: list[dict[str, Any]] = [] + self.count = 0 + self.max_changes = max_changes + self.include_values = include_values + + def add( + self, + *, + path: tuple[str, ...], + operation: Literal["added", "removed", "changed"], + before: Any, + after: Any, + ) -> None: + self.count += 1 + if len(self.changes) >= self.max_changes: + return + + category = _get_category(path) + change = { + "path": _format_path(path), + "operation": operation, + "category": category, + "impact": _get_impact(category), + "before_digest": None if before is _MISSING else _get_digest(before), + "after_digest": None if after is _MISSING else _get_digest(after), + } + if self.include_values: + if before is not _MISSING: + change["before_value"] = before + if after is not _MISSING: + change["after_value"] = after + self.changes.append(change) + + +_MISSING = object() + + +def _validate_max_changes(max_changes: int) -> None: + if max_changes < 1: + raise ValueError("max_changes must be a positive integer") + if max_changes > MAX_ALLOWED_CHANGES: + raise ValueError(f"max_changes must not exceed {MAX_ALLOWED_CHANGES}") + + +def _get_schema_version(data: Mapping[str, Any] | None) -> int | None: + if not isinstance(data, Mapping): + return None + version = data.get("__version") + return version if isinstance(version, int) and not isinstance(version, bool) else None + + +def _unavailable(result: dict[str, Any], reason: str) -> dict[str, Any]: + result["mode"] = "unavailable" + result["unavailable_reason"] = reason + return result + + +def _canonicalize_payload(data: dict[str, Any]) -> dict[str, Any]: + payload = copy.deepcopy(data) + version = _get_schema_version(payload) + if version is None: + raise ValueError("missing or invalid __version") + if version == 1: + DagSerialization.conversion_v1_to_v2(payload) + DagSerialization.conversion_v2_to_v3(payload) + elif version == 2: + DagSerialization.conversion_v2_to_v3(payload) + if not isinstance(payload.get("dag"), Mapping): + raise ValueError("missing dag object") + _apply_client_defaults(payload) + payload.pop("__version", None) + return _canonicalize_value(payload, path=()) + + +def _apply_client_defaults(payload: dict[str, Any]) -> None: + client_defaults = payload.pop("client_defaults", None) + if client_defaults is None: + return + if not isinstance(client_defaults, Mapping): + raise ValueError("client_defaults is not an object") + + task_defaults = client_defaults.get("tasks", {}) + if not isinstance(task_defaults, Mapping): + raise ValueError("client_defaults.tasks is not an object") + + tasks = payload["dag"].get("tasks", []) + if not isinstance(tasks, list): + raise ValueError("dag.tasks is not a list") + for task in tasks: + if not isinstance(task, dict) or not isinstance(task.get("__var"), Mapping): + raise ValueError("task entry is not an object") + task_data = dict(task_defaults) + task_data.update(task["__var"]) + if isinstance(partial_kwargs := task_data.get("partial_kwargs"), Mapping): + effective_partial_kwargs = dict(task_defaults) + effective_partial_kwargs.update(partial_kwargs) + task_data["partial_kwargs"] = effective_partial_kwargs + task["__var"] = task_data + + +def _canonicalize_value(value: Any, *, path: tuple[str, ...]) -> Any: + if isinstance(value, Mapping): + return { + str(key): _canonicalize_value(item, path=path + (str(key),)) + for key, item in sorted(value.items(), key=lambda item: str(item[0])) + } + if isinstance(value, list): + canonical_values = [_canonicalize_value(item, path=path) for item in value] + if path == ("dag", "tasks"): + return _canonicalize_keyed_list(canonical_values, _get_task_id, path) + if path == ("dag", "dag_dependencies"): + return _canonicalize_keyed_list(canonical_values, _get_dependency_key, path) + if path in _ORDER_INSENSITIVE_LIST_PATHS: + return _canonicalize_keyed_list(canonical_values, _get_string_key, path) + return canonical_values + return value + + +def _canonicalize_keyed_list( + values: list[Any], key_getter: Callable[[Any], str], path: tuple[str, ...] +) -> dict[str, Any]: + keyed_values: dict[str, Any] = {} + for value in values: + key = key_getter(value) + if key in keyed_values: + if path == ("dag", "dag_dependencies") and keyed_values[key] == value: + continue + raise ValueError(f"duplicate key {key!r} in /{'/'.join(path)}") + canonical_value = value + if path == ("dag", "tasks") and isinstance(value, Mapping) and "__var" in value: + task_value = dict(value["__var"]) + if "__type" in value: + task_value["__type"] = value["__type"] + canonical_value = task_value + keyed_values[key] = canonical_value + return {key: keyed_values[key] for key in sorted(keyed_values)} + + +def _get_task_id(task: Any) -> str: + if not isinstance(task, Mapping): + raise ValueError("task entry is not an object") + task_data = task.get("__var", task) + task_id = task_data.get("task_id") if isinstance(task_data, Mapping) else None + if not isinstance(task_id, str): + raise ValueError("task entry has no task_id") + return task_id + + +def _get_string_key(value: Any) -> str: + if not isinstance(value, str): + raise ValueError("collection entry is not a string") + return value + + +def _get_dependency_key(dependency: Any) -> str: + if not isinstance(dependency, Mapping): + raise ValueError("dependency entry is not an object") + components = ( + dependency.get("dependency_type"), + dependency.get("dependency_id"), + dependency.get("source"), + dependency.get("target"), + dependency.get("label"), + ) + return json.dumps(components, ensure_ascii=False, separators=(",", ":")) + + +def _collect_changes( + before: Any, + after: Any, + *, + path: tuple[str, ...], + collector: _ChangeCollector, +) -> None: + if isinstance(before, Mapping) and isinstance(after, Mapping): + keys = sorted({str(key) for key in before} | {str(key) for key in after}) + for key in keys: + before_value = before.get(key, _MISSING) + after_value = after.get(key, _MISSING) + _collect_changes(before_value, after_value, path=path + (key,), collector=collector) + return + + if isinstance(before, list) and isinstance(after, list): + if not _json_equal(before, after): + collector.add(path=path, operation="changed", before=before, after=after) + return + + if before is _MISSING: + collector.add(path=path, operation="added", before=_MISSING, after=after) + elif after is _MISSING: + collector.add(path=path, operation="removed", before=before, after=_MISSING) + elif not _json_equal(before, after): + collector.add(path=path, operation="changed", before=before, after=after) + + +def _format_path(path: tuple[str, ...]) -> str: + return "/" + "/".join(component.replace("~", "~0").replace("/", "~1") for component in path) + + +def _canonical_json(value: Any) -> str: + return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + + +def _json_equal(before: Any, after: Any) -> bool: + # Compare canonical JSON, not Python ==, so a JSON type change (1 vs 1.0, True vs 1, + # False vs 0) registers as a change instead of collapsing under Python equality. + return _canonical_json(before) == _canonical_json(after) + + +def _get_digest(value: Any) -> str: + return f"sha256:{hashlib.sha256(_canonical_json(value).encode()).hexdigest()}" + + +def _get_category(path: tuple[str, ...]) -> str: + lowered_path = tuple(component.lower() for component in path) + if lowered_path and lowered_path[0] == "provenance": + return "provenance" + if "deadline" in lowered_path: + return "deadline" + if any("callback" in component for component in lowered_path): + return "callback" + if any(component in {"asset", "assets", "inlets", "outlets"} for component in lowered_path): + return "asset" + if any(component in {"param", "params", "default_args"} for component in lowered_path): + return "param" + if any( + component + in {"dag_dependencies", "dependencies", "edge_info", "upstream_task_ids", "downstream_task_ids"} + for component in lowered_path + ): + return "dependency" + if any( + component + in { + "schedule", + "schedule_interval", + "timetable", + "catchup", + "max_active_runs", + "max_active_tasks", + "dagrun_timeout", + "fail_fast", + "fail_stop", + "allowed_run_types", + } + for component in lowered_path + ): + return "schedule" + if len(lowered_path) >= 2 and lowered_path[:2] == ("dag", "tasks"): + return "task" + if any( + component in {"tags", "description", "doc_md", "owner", "owners", "email", "fileloc"} + for component in lowered_path + ): + return "metadata" + return "unknown" + + +def _get_impact(category: str) -> str: + if category == "provenance": + return "provenance" + if category in {"task", "dependency", "schedule", "param", "asset", "deadline", "callback"}: + return "execution" + if category == "metadata": + return "metadata" + return "unknown" diff --git a/airflow-core/tests/unit/models/test_dag_version.py b/airflow-core/tests/unit/models/test_dag_version.py index af78736826ee1..69cb084d8fb31 100644 --- a/airflow-core/tests/unit/models/test_dag_version.py +++ b/airflow-core/tests/unit/models/test_dag_version.py @@ -20,11 +20,11 @@ from unittest import mock import pytest -from sqlalchemy import func, select +from sqlalchemy import delete, func, select from airflow._shared.timezones import timezone from airflow.models.dag import DagModel -from airflow.models.dag_version import DagVersion +from airflow.models.dag_version import DagVersion, DagVersionNotFoundError from airflow.models.dagbundle import DagBundleModel from airflow.providers.standard.operators.empty import EmptyOperator @@ -209,3 +209,91 @@ def test_resolve_version_data(self, dag_version, bundle_version, expected): from airflow.models.dag_version import _resolve_version_data assert _resolve_version_data(dag_version, bundle_version) == expected + + +class TestDagVersionGetDiff: + @pytest.fixture + def dag_id(self, dag_maker, session): + import datetime + + from tests_common.test_utils.db import clear_db_serialized_dags + + clear_db_dags() + clear_db_serialized_dags() + dag_id = "version_diff_dag" + for version_number in range(1, 3): + with dag_maker(dag_id, session=session, bundle_version=f"commit{version_number}"): + for task_number in range(version_number): + EmptyOperator(task_id=f"task{task_number + 1}") + dag_maker.create_dagrun( + run_id=f"run{version_number}", + logical_date=datetime.datetime(2020, 1, version_number, tzinfo=datetime.timezone.utc), + session=session, + ) + session.commit() + # Read versions back from the DB the way the API and CLI callers do, so + # serialized payloads carry plain JSON keys rather than in-memory Encoding enums. + session.expunge_all() + return dag_id + + def test_reports_observed_changes_between_versions(self, dag_id, session): + result = DagVersion.get_diff(dag_id, 1, 2, session=session) + + assert result["mode"] == "observed_state" + paths = {change["path"]: change for change in result["changes"]} + assert paths["/dag/tasks/task2"]["operation"] == "added" + assert paths["/provenance/bundle_version"]["category"] == "provenance" + assert result["source"] == {"status": "unavailable", "fidelity": "unavailable"} + assert "values" not in result + + def test_raises_for_missing_version(self, dag_id, session): + with pytest.raises(DagVersionNotFoundError, match="version_number: `3`"): + DagVersion.get_diff(dag_id, 1, 3, session=session) + + @pytest.mark.parametrize("version_numbers", [(0, 2), (1, 0)]) + def test_rejects_non_positive_version_numbers(self, dag_id, session, version_numbers): + with pytest.raises(ValueError, match="Dag version numbers must be positive integers"): + DagVersion.get_diff(dag_id, *version_numbers, session=session) + + def test_marks_values_available_for_operator_authority(self, dag_id, session): + result = DagVersion.get_diff(dag_id, 1, 2, include_values=True, session=session) + + assert result["values"] == {"status": "available"} + assert any("after_value" in change for change in result["changes"]) + + def test_marks_values_unavailable_when_status_denied(self, dag_id, session): + result = DagVersion.get_diff( + dag_id, 1, 2, include_values=True, values_status="unavailable", session=session + ) + + assert result["values"] == {"status": "unavailable"} + assert all( + "before_value" not in change and "after_value" not in change for change in result["changes"] + ) + + def test_includes_current_stored_source(self, dag_id, session): + result = DagVersion.get_diff(dag_id, 1, 2, include_source=True, session=session) + + source = result["source"] + assert source["status"] == "current_stored_code" + assert isinstance(source["changed"], bool) + assert source["base"]["digest"].startswith("sha256:") + assert source["target"]["content"] is not None + + def test_redacts_source_when_status_denied(self, dag_id, session): + result = DagVersion.get_diff( + dag_id, 1, 2, include_source=True, source_status="redacted", session=session + ) + + assert result["source"] == {"status": "redacted", "fidelity": "redacted"} + + def test_marks_source_unavailable_when_code_missing(self, dag_id, session): + from airflow.models.dagcode import DagCode + + session.execute(delete(DagCode)) + session.commit() + session.expunge_all() + + result = DagVersion.get_diff(dag_id, 1, 2, include_source=True, session=session) + + assert result["source"] == {"status": "unavailable", "fidelity": "unavailable"} diff --git a/airflow-core/tests/unit/serialization/test_dag_version_diff.py b/airflow-core/tests/unit/serialization/test_dag_version_diff.py new file mode 100644 index 0000000000000..a98dcc101c09c --- /dev/null +++ b/airflow-core/tests/unit/serialization/test_dag_version_diff.py @@ -0,0 +1,411 @@ +# 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 pytest + +from airflow.serialization.dag_version_diff import build_serialized_dag_diff + + +def _payload( + *, + tasks: list[dict], + tags: list[str] | None = None, + schedule: str = "daily", + dependencies: list[dict] | None = None, +) -> dict: + return { + "__version": 3, + "dag": { + "dag_id": "example", + "schedule": schedule, + "tags": tags or [], + "tasks": [ + { + "__type": "airflow.providers.standard.operators.empty.EmptyOperator", + "__var": task, + } + for task in tasks + ], + "dag_dependencies": dependencies or [], + }, + } + + +def test_build_diff_is_deterministic_and_normalizes_order() -> None: + base = _payload( + tasks=[{"task_id": "extract", "retries": 1}, {"task_id": "load", "retries": 1}], + tags=["one", "two"], + dependencies=[ + { + "dependency_type": "task", + "dependency_id": "extract-load", + "source": "extract", + "target": "load", + "label": "extract-load", + } + ], + ) + target = _payload( + tasks=[{"task_id": "load", "retries": 1}, {"task_id": "extract", "retries": 1}], + tags=["two", "one"], + dependencies=[ + { + "label": "extract-load", + "target": "load", + "source": "extract", + "dependency_id": "extract-load", + "dependency_type": "task", + } + ], + ) + + result = build_serialized_dag_diff(base_data=base, target_data=target) + + assert result["mode"] == "observed_state" + assert result["serialized_dag_schema_versions"] == {"base": 3, "target": 3} + assert result["changes"] == [] + assert result["truncated"] is False + + +def test_build_diff_reports_categories_digests_and_values() -> None: + base = _payload(tasks=[{"task_id": "extract", "retries": 1}], tags=["old"]) + target = _payload( + tasks=[ + {"task_id": "extract", "retries": 2}, + {"task_id": "load", "retries": 1}, + ], + tags=["new"], + schedule="hourly", + ) + + result = build_serialized_dag_diff( + base_data=base, + target_data=target, + base_provenance={"bundle_version": "one"}, + target_provenance={"bundle_version": "two"}, + include_values=True, + ) + + changes = {change["path"]: change for change in result["changes"]} + assert changes["/dag/tasks/extract/retries"]["category"] == "task" + assert changes["/dag/tasks/extract/retries"]["impact"] == "execution" + assert changes["/dag/tasks/extract/retries"]["before_value"] == 1 + assert changes["/dag/tasks/extract/retries"]["after_value"] == 2 + assert changes["/dag/tasks/extract/retries"]["before_digest"].startswith("sha256:") + assert changes["/dag/tasks/load"]["operation"] == "added" + assert changes["/dag/schedule"]["category"] == "schedule" + assert changes["/dag/tags/old"]["operation"] == "removed" + assert changes["/dag/tags/new"]["operation"] == "added" + assert changes["/dag/tags/old"]["category"] == "metadata" + assert changes["/provenance/bundle_version"]["category"] == "provenance" + assert changes["/provenance/bundle_version"]["impact"] == "provenance" + + +def test_build_diff_bounds_changes_and_reports_truncation() -> None: + result = build_serialized_dag_diff( + base_data=_payload(tasks=[{"task_id": "extract", "retries": 1}], tags=["old"]), + target_data=_payload(tasks=[{"task_id": "extract", "retries": 2}], tags=["new"]), + max_changes=1, + ) + + assert len(result["changes"]) == 1 + assert result["truncated"] is True + + +def test_build_diff_reports_dependency_changes() -> None: + dependency = { + "dependency_type": "sensor", + "dependency_id": "upstream-task", + "source": "upstream-dag", + "target": "example", + "label": "upstream-task", + } + + result = build_serialized_dag_diff( + base_data=_payload(tasks=[], dependencies=[dependency]), + target_data=_payload(tasks=[]), + ) + + assert len(result["changes"]) == 1 + assert result["changes"][0]["category"] == "dependency" + assert result["changes"][0]["impact"] == "execution" + + +def test_build_diff_classifies_fail_fast_as_schedule() -> None: + base = _payload(tasks=[]) + target = _payload(tasks=[]) + base["dag"]["fail_fast"] = False + target["dag"]["fail_fast"] = True + + result = build_serialized_dag_diff(base_data=base, target_data=target) + + assert len(result["changes"]) == 1 + change = result["changes"][0] + assert change["path"] == "/dag/fail_fast" + assert change["operation"] == "changed" + assert change["category"] == "schedule" + assert change["impact"] == "execution" + + +@pytest.mark.parametrize( + ("base_data", "target_data", "reason"), + [ + (None, _payload(tasks=[]), "serialized_dag_missing"), + ({"__version": 99, "dag": {}}, _payload(tasks=[]), "unsupported_serialized_dag_schema_version:99"), + ({"__version": 3, "dag": []}, _payload(tasks=[]), "serialized_dag_canonicalization_failed"), + ( + { + "__version": 1, + "dag": { + "dag_id": "example", + "tasks": [], + "task_group": {}, + "schedule_interval": 1, + }, + }, + _payload(tasks=[]), + "serialized_dag_canonicalization_failed", + ), + ( + { + "__version": 1, + "dag": { + "dag_id": "example", + "tasks": [], + "task_group": {}, + "schedule_interval": {"__type": "timedelta", "__var": 10**20}, + }, + }, + _payload(tasks=[]), + "serialized_dag_canonicalization_failed", + ), + ], +) +def test_build_diff_returns_unavailable_for_unsafe_inputs(base_data, target_data, reason) -> None: + result = build_serialized_dag_diff(base_data=base_data, target_data=target_data) + + assert result["mode"] == "unavailable" + assert result["changes"] == [] + assert result["unavailable_reason"] == reason + + +def test_build_diff_rejects_non_positive_change_bound() -> None: + with pytest.raises(ValueError, match="max_changes must be a positive integer"): + build_serialized_dag_diff(base_data=_payload(tasks=[]), target_data=_payload(tasks=[]), max_changes=0) + + +def test_build_diff_rejects_unbounded_change_bound() -> None: + with pytest.raises(ValueError, match="max_changes must not exceed 5000"): + build_serialized_dag_diff( + base_data=_payload(tasks=[]), target_data=_payload(tasks=[]), max_changes=5001 + ) + + +def test_build_diff_reports_unkeyed_lists_as_one_stable_change() -> None: + base = _payload(tasks=[]) + target = _payload(tasks=[]) + base["dag"]["custom_list"] = [{"name": "old"}] + target["dag"]["custom_list"] = [{"name": "new"}] + + result = build_serialized_dag_diff(base_data=base, target_data=target) + + assert len(result["changes"]) == 1 + change = result["changes"][0] + assert change["path"] == "/dag/custom_list" + assert change["operation"] == "changed" + assert change["category"] == "unknown" + assert change["impact"] == "unknown" + assert change["before_digest"].startswith("sha256:") + assert change["after_digest"].startswith("sha256:") + + +def test_build_diff_reports_deadline_lists_as_one_stable_change() -> None: + base = _payload(tasks=[]) + target = _payload(tasks=[]) + base["dag"]["deadline"] = [{"name": "old", "interval": 60}] + target["dag"]["deadline"] = [{"name": "new", "interval": 60}] + + result = build_serialized_dag_diff(base_data=base, target_data=target) + + assert result["mode"] == "observed_state" + assert len(result["changes"]) == 1 + change = result["changes"][0] + assert change["path"] == "/dag/deadline" + assert change["operation"] == "changed" + assert change["category"] == "deadline" + assert change["impact"] == "execution" + + +def test_build_diff_preserves_order_sensitive_string_lists() -> None: + base = _payload(tasks=[]) + target = _payload(tasks=[]) + base["dag"]["template_searchpath"] = ["first", "second"] + target["dag"]["template_searchpath"] = ["second", "first"] + + result = build_serialized_dag_diff(base_data=base, target_data=target) + + assert len(result["changes"]) == 1 + change = result["changes"][0] + assert change["path"] == "/dag/template_searchpath" + assert change["operation"] == "changed" + + +@pytest.mark.parametrize( + ("base_task", "target_task"), + [ + ({"task_id": "extract", "retries": 2}, {"task_id": "extract"}), + ( + {"task_id": "extract", "retries": 2, "partial_kwargs": {"retries": 2}}, + {"task_id": "extract", "partial_kwargs": {}}, + ), + ], +) +def test_build_diff_normalizes_v3_client_defaults(base_task, target_task) -> None: + base = _payload(tasks=[base_task]) + base["__version"] = 2 + target = _payload(tasks=[target_task]) + target["client_defaults"] = {"tasks": {"retries": 2}} + + result = build_serialized_dag_diff(base_data=base, target_data=target) + + assert result["serialized_dag_schema_versions"] == {"base": 2, "target": 3} + assert result["mode"] == "observed_state" + assert result["changes"] == [] + + +@pytest.mark.parametrize( + "base_data", + [ + {"__version": 3, "dag": {"tasks": []}, "client_defaults": []}, + {"__version": 3, "dag": {"tasks": []}, "client_defaults": {"tasks": []}}, + {"__version": 3, "dag": {"tasks": {}}, "client_defaults": {"tasks": {}}}, + {"__version": 3, "dag": {"tasks": [{}]}, "client_defaults": {"tasks": {}}}, + ], +) +def test_build_diff_rejects_malformed_client_defaults(base_data) -> None: + result = build_serialized_dag_diff(base_data=base_data, target_data=_payload(tasks=[])) + + assert result["mode"] == "unavailable" + assert result["changes"] == [] + assert result["unavailable_reason"] == "serialized_dag_canonicalization_failed" + + +def test_build_diff_ignores_exact_duplicate_dependencies() -> None: + dependency = { + "dependency_type": "task", + "dependency_id": "extract-load", + "source": "extract", + "target": "load", + "label": "extract-load", + } + + result = build_serialized_dag_diff( + base_data=_payload(tasks=[], dependencies=[dependency, dependency]), + target_data=_payload(tasks=[], dependencies=[dependency]), + ) + + assert result["mode"] == "observed_state" + assert result["changes"] == [] + + +def test_build_diff_dependency_keys_do_not_collide_on_delimiters() -> None: + first_dependency = { + "dependency_type": "trigger", + "dependency_id": "id", + "source": "a", + "target": "b|c", + "label": "d", + } + second_dependency = { + "dependency_type": "trigger", + "dependency_id": "id", + "source": "a", + "target": "b", + "label": "c|d", + } + + result = build_serialized_dag_diff( + base_data=_payload(tasks=[], dependencies=[first_dependency, second_dependency]), + target_data=_payload(tasks=[], dependencies=[second_dependency, first_dependency]), + ) + + assert result["mode"] == "observed_state" + assert result["changes"] == [] + + +def test_build_diff_detects_json_type_change() -> None: + base = _payload(tasks=[]) + target = _payload(tasks=[]) + base["dag"]["flag"] = True + target["dag"]["flag"] = 1 + + result = build_serialized_dag_diff(base_data=base, target_data=target) + + assert len(result["changes"]) == 1 + change = result["changes"][0] + assert change["path"] == "/dag/flag" + assert change["operation"] == "changed" + + +def test_build_diff_classifies_callback_as_execution() -> None: + base = _payload(tasks=[]) + target = _payload(tasks=[]) + base["dag"]["on_failure_callback"] = "old" + target["dag"]["on_failure_callback"] = "new" + + result = build_serialized_dag_diff(base_data=base, target_data=target) + + assert len(result["changes"]) == 1 + change = result["changes"][0] + assert change["path"] == "/dag/on_failure_callback" + assert change["category"] == "callback" + assert change["impact"] == "execution" + + +def _v1_payload() -> dict: + return { + "__version": 1, + "dag": { + "_dag_id": "example", + "fileloc": "/dags/example.py", + "timezone": "UTC", + "task_group": {}, + "tasks": [ + { + "__type": "operator", + "__var": { + "task_id": "extract", + "_task_type": "EmptyOperator", + "_task_module": "airflow.providers.standard.operators.empty", + "template_fields": [], + "ui_color": "#e8f7e4", + }, + } + ], + "dag_dependencies": [], + }, + } + + +def test_build_diff_normalizes_valid_v1_payload() -> None: + result = build_serialized_dag_diff(base_data=_v1_payload(), target_data=_v1_payload()) + + assert result["mode"] == "observed_state" + assert result["serialized_dag_schema_versions"] == {"base": 1, "target": 1} + assert result["changes"] == [] From 71068ae6a33454ee745eaf011bbe44f3fdb16603 Mon Sep 17 00:00:00 2001 From: Ephraim Anierobi Date: Tue, 14 Jul 2026 13:16:01 +0100 Subject: [PATCH 2/4] Prevent unauthorized Dag version diff disclosure --- .../src/airflow/models/dag_version.py | 3 +- .../airflow/serialization/dag_version_diff.py | 57 ++++++++-- .../tests/unit/models/test_dag_version.py | 14 ++- .../serialization/test_dag_version_diff.py | 104 +++++++++++++++++- 4 files changed, 163 insertions(+), 15 deletions(-) diff --git a/airflow-core/src/airflow/models/dag_version.py b/airflow-core/src/airflow/models/dag_version.py index c8db20166d9ee..0ab447e03d1df 100644 --- a/airflow-core/src/airflow/models/dag_version.py +++ b/airflow-core/src/airflow/models/dag_version.py @@ -271,7 +271,8 @@ def get_diff( ``source_status`` is supplied by callers that have an authorization context. The CLI has operator-level authority and leaves it unset, while API callers calculate it before this - method returns any source content. + method returns any source content. API callers use ``values_status`` to suppress raw values + when the user cannot access Dag code; the structural diff remains available in redacted form. """ from airflow.serialization.dag_version_diff import ( DEFAULT_MAX_CHANGES, diff --git a/airflow-core/src/airflow/serialization/dag_version_diff.py b/airflow-core/src/airflow/serialization/dag_version_diff.py index eba5c259e1dbe..003c48475106c 100644 --- a/airflow-core/src/airflow/serialization/dag_version_diff.py +++ b/airflow-core/src/airflow/serialization/dag_version_diff.py @@ -23,6 +23,7 @@ import hashlib import json from collections.abc import Callable, Mapping +from enum import Enum from typing import Any, Literal from airflow.serialization.serialized_objects import DagSerialization @@ -38,6 +39,18 @@ ("dag", "tags"), ("dag", "allowed_run_types"), } +_KEYED_COLLECTION_PATHS = { + ("dag", "tasks"), + ("dag", "dag_dependencies"), + *_ORDER_INSENSITIVE_LIST_PATHS, +} +_REDACTED_RECURSIVE_MAPPING_PATHS = { + (), + ("dag",), + ("dag", "task_group"), + ("provenance",), + *_KEYED_COLLECTION_PATHS, +} def build_serialized_dag_diff( @@ -49,7 +62,12 @@ def build_serialized_dag_diff( include_values: bool = False, max_changes: int = DEFAULT_MAX_CHANGES, ) -> dict[str, Any]: - """Build a bounded, deterministic diff from two stored serialized Dag payloads.""" + """ + Build a bounded, deterministic diff from two stored serialized Dag payloads. + + Raw values, digests, and value-derived path components are returned only when + ``include_values`` is true. + """ _validate_max_changes(max_changes) base_schema_version = _get_schema_version(base_data) @@ -115,15 +133,15 @@ def add( return category = _get_category(path) - change = { - "path": _format_path(path), + change: dict[str, Any] = { + "path": _format_path(path if self.include_values else _redact_path(path)), "operation": operation, "category": category, "impact": _get_impact(category), - "before_digest": None if before is _MISSING else _get_digest(before), - "after_digest": None if after is _MISSING else _get_digest(after), } if self.include_values: + change["before_digest"] = None if before is _MISSING else _get_digest(before) + change["after_digest"] = None if after is _MISSING else _get_digest(after) if before is not _MISSING: change["before_value"] = before if after is not _MISSING: @@ -200,8 +218,11 @@ def _apply_client_defaults(payload: dict[str, Any]) -> None: def _canonicalize_value(value: Any, *, path: tuple[str, ...]) -> Any: if isinstance(value, Mapping): return { - str(key): _canonicalize_value(item, path=path + (str(key),)) - for key, item in sorted(value.items(), key=lambda item: str(item[0])) + canonical_key: _canonicalize_value(item, path=path + (canonical_key,)) + for canonical_key, item in ( + (_canonicalize_mapping_key(key), item) + for key, item in sorted(value.items(), key=lambda item: _canonicalize_mapping_key(item[0])) + ) } if isinstance(value, list): canonical_values = [_canonicalize_value(item, path=path) for item in value] @@ -215,6 +236,12 @@ def _canonicalize_value(value: Any, *, path: tuple[str, ...]) -> Any: return value +def _canonicalize_mapping_key(key: Any) -> str: + if isinstance(key, str) and isinstance(key, Enum): + return key.value + return str(key) + + def _canonicalize_keyed_list( values: list[Any], key_getter: Callable[[Any], str], path: tuple[str, ...] ) -> dict[str, Any]: @@ -272,6 +299,10 @@ def _collect_changes( collector: _ChangeCollector, ) -> None: if isinstance(before, Mapping) and isinstance(after, Mapping): + if not collector.include_values and not _should_recurse_redacted_mapping(path): + if not _json_equal(before, after): + collector.add(path=path, operation="changed", before=before, after=after) + return keys = sorted({str(key) for key in before} | {str(key) for key in after}) for key in keys: before_value = before.get(key, _MISSING) @@ -292,6 +323,18 @@ def _collect_changes( collector.add(path=path, operation="changed", before=before, after=after) +def _should_recurse_redacted_mapping(path: tuple[str, ...]) -> bool: + if path in _REDACTED_RECURSIVE_MAPPING_PATHS: + return True + return len(path) == 3 and path[:2] == ("dag", "tasks") + + +def _redact_path(path: tuple[str, ...]) -> tuple[str, ...]: + if len(path) >= 3 and path[:2] in _KEYED_COLLECTION_PATHS: + return (*path[:2], "*", *path[3:]) + return path + + def _format_path(path: tuple[str, ...]) -> str: return "/" + "/".join(component.replace("~", "~0").replace("/", "~1") for component in path) diff --git a/airflow-core/tests/unit/models/test_dag_version.py b/airflow-core/tests/unit/models/test_dag_version.py index 69cb084d8fb31..766b493c0a991 100644 --- a/airflow-core/tests/unit/models/test_dag_version.py +++ b/airflow-core/tests/unit/models/test_dag_version.py @@ -241,7 +241,8 @@ def test_reports_observed_changes_between_versions(self, dag_id, session): assert result["mode"] == "observed_state" paths = {change["path"]: change for change in result["changes"]} - assert paths["/dag/tasks/task2"]["operation"] == "added" + assert paths["/dag/tasks/*"]["operation"] == "added" + assert "after_digest" not in paths["/dag/tasks/*"] assert paths["/provenance/bundle_version"]["category"] == "provenance" assert result["source"] == {"status": "unavailable", "fidelity": "unavailable"} assert "values" not in result @@ -260,16 +261,23 @@ def test_marks_values_available_for_operator_authority(self, dag_id, session): assert result["values"] == {"status": "available"} assert any("after_value" in change for change in result["changes"]) + assert any(change["path"] == "/dag/tasks/task2" for change in result["changes"]) def test_marks_values_unavailable_when_status_denied(self, dag_id, session): result = DagVersion.get_diff( dag_id, 1, 2, include_values=True, values_status="unavailable", session=session ) - assert result["values"] == {"status": "unavailable"} + assert result["mode"] == "observed_state" + assert any(change["path"] == "/dag/tasks/*" for change in result["changes"]) assert all( - "before_value" not in change and "after_value" not in change for change in result["changes"] + "before_digest" not in change + and "after_digest" not in change + and "before_value" not in change + and "after_value" not in change + for change in result["changes"] ) + assert result["values"] == {"status": "unavailable"} def test_includes_current_stored_source(self, dag_id, session): result = DagVersion.get_diff(dag_id, 1, 2, include_source=True, session=session) diff --git a/airflow-core/tests/unit/serialization/test_dag_version_diff.py b/airflow-core/tests/unit/serialization/test_dag_version_diff.py index a98dcc101c09c..4c63905ff57ed 100644 --- a/airflow-core/tests/unit/serialization/test_dag_version_diff.py +++ b/airflow-core/tests/unit/serialization/test_dag_version_diff.py @@ -17,6 +17,8 @@ from __future__ import annotations +import json + import pytest from airflow.serialization.dag_version_diff import build_serialized_dag_diff @@ -217,6 +219,72 @@ def test_build_diff_rejects_unbounded_change_bound() -> None: ) +def test_build_diff_redacts_sensitive_data_without_values() -> None: + dependency = { + "dependency_type": "sensor", + "dependency_id": "secret-dependency", + "source": "secret-upstream", + "target": "example", + "label": "secret-label", + } + result = build_serialized_dag_diff( + base_data=_payload( + tasks=[{"task_id": "secret-task", "retries": 1}], + tags=["secret-old-tag"], + dependencies=[dependency], + ), + target_data=_payload( + tasks=[ + {"task_id": "secret-task", "retries": 2}, + {"task_id": "secret-new-task", "retries": 1}, + ], + tags=["secret-new-tag"], + ), + ) + + assert result["mode"] == "observed_state" + assert any(change["path"] == "/dag/tasks/*/retries" for change in result["changes"]) + assert any(change["path"] == "/dag/tasks/*" for change in result["changes"]) + assert any(change["path"] == "/dag/tags/*" for change in result["changes"]) + assert any(change["path"] == "/dag/dag_dependencies/*" for change in result["changes"]) + assert all( + "before_digest" not in change + and "after_digest" not in change + and "before_value" not in change + and "after_value" not in change + for change in result["changes"] + ) + encoded_result = json.dumps(result) + for sensitive_value in ( + "secret-task", + "secret-new-task", + "secret-old-tag", + "secret-new-tag", + "secret-dependency", + "secret-upstream", + "secret-label", + ): + assert sensitive_value not in encoded_result + + +def test_build_diff_collapses_arbitrary_mappings_without_values() -> None: + base = _payload(tasks=[]) + target = _payload(tasks=[]) + base["dag"]["default_args"] = {"secret-argument": "old-secret"} + target["dag"]["default_args"] = {"secret-argument": "new-secret"} + + result = build_serialized_dag_diff(base_data=base, target_data=target) + + assert result["changes"] == [ + { + "path": "/dag/default_args", + "operation": "changed", + "category": "param", + "impact": "execution", + } + ] + + def test_build_diff_reports_unkeyed_lists_as_one_stable_change() -> None: base = _payload(tasks=[]) target = _payload(tasks=[]) @@ -231,8 +299,8 @@ def test_build_diff_reports_unkeyed_lists_as_one_stable_change() -> None: assert change["operation"] == "changed" assert change["category"] == "unknown" assert change["impact"] == "unknown" - assert change["before_digest"].startswith("sha256:") - assert change["after_digest"].startswith("sha256:") + assert "before_digest" not in change + assert "after_digest" not in change def test_build_diff_reports_deadline_lists_as_one_stable_change() -> None: @@ -403,9 +471,37 @@ def _v1_payload() -> dict: } +def _equivalent_v3_payload() -> dict: + return { + "__version": 3, + "dag": { + "dag_id": "example", + "fileloc": "/dags/example.py", + "timezone": "UTC", + "timetable": { + "__type": "airflow.timetables.simple.NullTimetable", + "__var": {}, + }, + "task_group": {"group_display_name": ""}, + "tasks": [ + { + "__type": "operator", + "__var": { + "task_id": "extract", + "task_type": "EmptyOperator", + "_task_module": "airflow.providers.standard.operators.empty", + "ui_color": "#e8f7e4", + }, + } + ], + "dag_dependencies": [], + }, + } + + def test_build_diff_normalizes_valid_v1_payload() -> None: - result = build_serialized_dag_diff(base_data=_v1_payload(), target_data=_v1_payload()) + result = build_serialized_dag_diff(base_data=_v1_payload(), target_data=_equivalent_v3_payload()) assert result["mode"] == "observed_state" - assert result["serialized_dag_schema_versions"] == {"base": 1, "target": 1} + assert result["serialized_dag_schema_versions"] == {"base": 1, "target": 3} assert result["changes"] == [] From dbd4c485736276033354b0bfa1b5873a9bc4b11a Mon Sep 17 00:00:00 2001 From: Ephraim Anierobi Date: Tue, 14 Jul 2026 14:59:55 +0100 Subject: [PATCH 3/4] Keep redacted Dag version diffs non-disclosing Users with version access need a useful structural comparison without learning identifiers or values derived from Dag code. The diff contract also needs to remain stable as serializer fields evolve. --- .../airflow/serialization/dag_version_diff.py | 203 ++++++++++++++---- .../serialization/test_dag_version_diff.py | 167 +++++++++++++- 2 files changed, 327 insertions(+), 43 deletions(-) diff --git a/airflow-core/src/airflow/serialization/dag_version_diff.py b/airflow-core/src/airflow/serialization/dag_version_diff.py index 003c48475106c..e011644a89b5d 100644 --- a/airflow-core/src/airflow/serialization/dag_version_diff.py +++ b/airflow-core/src/airflow/serialization/dag_version_diff.py @@ -44,6 +44,80 @@ ("dag", "dag_dependencies"), *_ORDER_INSENSITIVE_LIST_PATHS, } +_CUSTOM_TASK_FIELDS_PATH_COMPONENT = "custom_fields" +# This allowlist is part of diff schema v1. Serializer schema changes must not +# silently change the paths visible to callers of the diff API. +_DIFF_V1_PUBLIC_TASK_FIELDS = frozenset( + { + "__type", + "_disallow_kwargs_override", + "_expand_input_attr", + "_is_mapped", + "_is_sensor", + "_logger_name", + "_needs_expansion", + "_operator_extra_links", + "_task_display_name", + "_task_module", + "allow_nested_operators", + "depends_on_past", + "do_xcom_push", + "doc", + "doc_json", + "doc_md", + "doc_rst", + "doc_yaml", + "downstream_task_ids", + "email_on_failure", + "email_on_retry", + "end_date", + "execution_timeout", + "executor", + "executor_config", + "has_on_execute_callback", + "has_on_failure_callback", + "has_on_retry_callback", + "has_on_skipped_callback", + "has_on_success_callback", + "ignore_first_depends_on_past", + "inlets", + "is_setup", + "is_teardown", + "map_index_template", + "max_active_tis_per_dag", + "max_active_tis_per_dagrun", + "max_retry_delay", + "multiple_outputs", + "on_failure_fail_dagrun", + "outlets", + "owner", + "params", + "partial_kwargs", + "pool", + "pool_slots", + "priority_weight", + "queue", + "render_template_as_native_obj", + "retries", + "retry_delay", + "retry_exponential_backoff", + "start_date", + "start_from_trigger", + "start_trigger_args", + "task_id", + "task_type", + "template_ext", + "template_fields", + "template_fields_renderers", + "trigger_rule", + "ui_color", + "ui_fgcolor", + "wait_for_downstream", + "wait_for_past_depends_before_skipping", + "weight_rule", + } +) +_DIFF_V1_REDACTED_SCHEMA_TASK_FIELDS: frozenset[str] = frozenset() _REDACTED_RECURSIVE_MAPPING_PATHS = { (), ("dag",), @@ -98,8 +172,8 @@ def build_serialized_dag_diff( return _unavailable(result, f"unsupported_serialized_dag_schema_version:{unsupported_versions[0]}") try: - base_document = _canonicalize_payload(base_data) - target_document = _canonicalize_payload(target_data) + base_document = _canonicalize_payload_v1(base_data) + target_document = _canonicalize_payload_v1(target_data) except (AttributeError, KeyError, OverflowError, TypeError, ValueError): return _unavailable(result, "serialized_dag_canonicalization_failed") @@ -132,9 +206,10 @@ def add( if len(self.changes) >= self.max_changes: return - category = _get_category(path) + public_path = _get_public_path(path) + category = _get_category(public_path) change: dict[str, Any] = { - "path": _format_path(path if self.include_values else _redact_path(path)), + "path": _format_path(path if self.include_values else public_path), "operation": operation, "category": category, "impact": _get_impact(category), @@ -172,7 +247,7 @@ def _unavailable(result: dict[str, Any], reason: str) -> dict[str, Any]: return result -def _canonicalize_payload(data: dict[str, Any]) -> dict[str, Any]: +def _canonicalize_payload_v1(data: dict[str, Any]) -> dict[str, Any]: payload = copy.deepcopy(data) version = _get_schema_version(payload) if version is None: @@ -299,6 +374,9 @@ def _collect_changes( collector: _ChangeCollector, ) -> None: if isinstance(before, Mapping) and isinstance(after, Mapping): + if not collector.include_values and _is_task_mapping_path(path): + _collect_redacted_task_changes(before, after, path=path, collector=collector) + return if not collector.include_values and not _should_recurse_redacted_mapping(path): if not _json_equal(before, after): collector.add(path=path, operation="changed", before=before, after=after) @@ -323,15 +401,47 @@ def _collect_changes( collector.add(path=path, operation="changed", before=before, after=after) -def _should_recurse_redacted_mapping(path: tuple[str, ...]) -> bool: - if path in _REDACTED_RECURSIVE_MAPPING_PATHS: - return True +def _collect_redacted_task_changes( + before: Mapping[str, Any], + after: Mapping[str, Any], + *, + path: tuple[str, ...], + collector: _ChangeCollector, +) -> None: + public_fields = _DIFF_V1_PUBLIC_TASK_FIELDS + keys = {str(key) for key in before} | {str(key) for key in after} + for key in sorted(keys & public_fields): + _collect_changes( + before.get(key, _MISSING), + after.get(key, _MISSING), + path=path + (key,), + collector=collector, + ) + + before_custom_fields = {key: before[key] for key in before if key not in public_fields} + after_custom_fields = {key: after[key] for key in after if key not in public_fields} + if not _json_equal(before_custom_fields, after_custom_fields): + collector.add( + path=path + (_CUSTOM_TASK_FIELDS_PATH_COMPONENT,), + operation="changed", + before=before_custom_fields, + after=after_custom_fields, + ) + + +def _is_task_mapping_path(path: tuple[str, ...]) -> bool: return len(path) == 3 and path[:2] == ("dag", "tasks") -def _redact_path(path: tuple[str, ...]) -> tuple[str, ...]: +def _should_recurse_redacted_mapping(path: tuple[str, ...]) -> bool: + return path in _REDACTED_RECURSIVE_MAPPING_PATHS + + +def _get_public_path(path: tuple[str, ...]) -> tuple[str, ...]: if len(path) >= 3 and path[:2] in _KEYED_COLLECTION_PATHS: - return (*path[:2], "*", *path[3:]) + path = (*path[:2], "*", *path[3:]) + if len(path) >= 4 and path[:2] == ("dag", "tasks") and path[3] not in _DIFF_V1_PUBLIC_TASK_FIELDS: + return (*path[:3], _CUSTOM_TASK_FIELDS_PATH_COMPONENT) return path @@ -357,45 +467,54 @@ def _get_category(path: tuple[str, ...]) -> str: lowered_path = tuple(component.lower() for component in path) if lowered_path and lowered_path[0] == "provenance": return "provenance" - if "deadline" in lowered_path: + if lowered_path[:2] == ("dag", "dag_dependencies"): + return "dependency" + if lowered_path[:2] == ("dag", "tags"): + return "metadata" + if lowered_path[:2] == ("dag", "allowed_run_types"): + return "schedule" + + if lowered_path[:2] == ("dag", "tasks"): + field = lowered_path[3] if len(lowered_path) >= 4 else None + default_category = "task" + elif lowered_path and lowered_path[0] == "dag": + field = lowered_path[1] if len(lowered_path) >= 2 else None + default_category = "unknown" + else: + return "unknown" + + if field == "deadline": return "deadline" - if any("callback" in component for component in lowered_path): + if field is not None and "callback" in field: return "callback" - if any(component in {"asset", "assets", "inlets", "outlets"} for component in lowered_path): + if field in {"asset", "assets", "inlets", "outlets"}: return "asset" - if any(component in {"param", "params", "default_args"} for component in lowered_path): + if field in {"param", "params", "default_args"}: return "param" - if any( - component - in {"dag_dependencies", "dependencies", "edge_info", "upstream_task_ids", "downstream_task_ids"} - for component in lowered_path - ): + if field in { + "dag_dependencies", + "dependencies", + "edge_info", + "upstream_task_ids", + "downstream_task_ids", + }: return "dependency" - if any( - component - in { - "schedule", - "schedule_interval", - "timetable", - "catchup", - "max_active_runs", - "max_active_tasks", - "dagrun_timeout", - "fail_fast", - "fail_stop", - "allowed_run_types", - } - for component in lowered_path - ): + if field in { + "schedule", + "schedule_interval", + "timetable", + "catchup", + "max_active_runs", + "max_active_tasks", + "dagrun_timeout", + "fail_fast", + "fail_stop", + "allowed_run_types", + }: return "schedule" - if len(lowered_path) >= 2 and lowered_path[:2] == ("dag", "tasks"): - return "task" - if any( - component in {"tags", "description", "doc_md", "owner", "owners", "email", "fileloc"} - for component in lowered_path - ): + if field in {"tags", "description", "doc_md", "owner", "owners", "email", "fileloc"}: return "metadata" - return "unknown" + return default_category def _get_impact(category: str) -> str: diff --git a/airflow-core/tests/unit/serialization/test_dag_version_diff.py b/airflow-core/tests/unit/serialization/test_dag_version_diff.py index 4c63905ff57ed..896e5ec892668 100644 --- a/airflow-core/tests/unit/serialization/test_dag_version_diff.py +++ b/airflow-core/tests/unit/serialization/test_dag_version_diff.py @@ -21,7 +21,12 @@ import pytest -from airflow.serialization.dag_version_diff import build_serialized_dag_diff +from airflow.serialization.dag_version_diff import ( + _DIFF_V1_PUBLIC_TASK_FIELDS, + _DIFF_V1_REDACTED_SCHEMA_TASK_FIELDS, + build_serialized_dag_diff, +) +from airflow.serialization.json_schema import load_dag_schema_dict def _payload( @@ -49,6 +54,15 @@ def _payload( } +def test_diff_v1_task_field_policy_tracks_serializer_schema() -> None: + schema_fields = frozenset(load_dag_schema_dict()["definitions"]["operator"]["properties"]) + + assert schema_fields == ( + (_DIFF_V1_PUBLIC_TASK_FIELDS - {"__type"}) | _DIFF_V1_REDACTED_SCHEMA_TASK_FIELDS + ) + assert not (_DIFF_V1_PUBLIC_TASK_FIELDS & _DIFF_V1_REDACTED_SCHEMA_TASK_FIELDS) + + def test_build_diff_is_deterministic_and_normalizes_order() -> None: base = _payload( tasks=[{"task_id": "extract", "retries": 1}, {"task_id": "load", "retries": 1}], @@ -267,6 +281,157 @@ def test_build_diff_redacts_sensitive_data_without_values() -> None: assert sensitive_value not in encoded_result +def test_build_diff_collapses_custom_task_fields_without_values() -> None: + custom_field = "secret_callback_field" + base_task = { + "task_id": "extract", + "template_fields": [custom_field], + custom_field: "old-secret", + } + target_task = { + "task_id": "extract", + "template_fields": [custom_field], + custom_field: "new-secret", + } + + result = build_serialized_dag_diff( + base_data=_payload(tasks=[base_task]), + target_data=_payload(tasks=[target_task]), + ) + + assert result["changes"] == [ + { + "path": "/dag/tasks/*/custom_fields", + "operation": "changed", + "category": "task", + "impact": "execution", + } + ] + encoded_result = json.dumps(result) + assert custom_field not in encoded_result + assert "old-secret" not in encoded_result + assert "new-secret" not in encoded_result + + +def test_build_diff_includes_custom_task_fields_with_values() -> None: + custom_field = "secret_callback_field" + base_task = { + "task_id": "extract", + "template_fields": [custom_field], + custom_field: "old-secret", + } + target_task = { + "task_id": "extract", + "template_fields": [custom_field], + custom_field: "new-secret", + } + + result = build_serialized_dag_diff( + base_data=_payload(tasks=[base_task]), + target_data=_payload(tasks=[target_task]), + include_values=True, + ) + + change = next(change for change in result["changes"] if change["path"].endswith(custom_field)) + assert change["path"] == f"/dag/tasks/extract/{custom_field}" + assert change["category"] == "task" + assert change["impact"] == "execution" + assert change["before_value"] == "old-secret" + assert change["after_value"] == "new-secret" + + +@pytest.mark.parametrize( + ( + "base_data", + "target_data", + "hidden_identifier", + "expected_path", + "expected_operation", + "expected_category", + "expected_impact", + ), + [ + pytest.param( + _payload(tasks=[{"task_id": "deadline", "retries": 1}]), + _payload(tasks=[{"task_id": "deadline", "retries": 2}]), + "deadline", + "/dag/tasks/*/retries", + "changed", + "task", + "execution", + id="task-id", + ), + pytest.param( + _payload(tasks=[]), + _payload(tasks=[], tags=["deadline"]), + "deadline", + "/dag/tags/*", + "added", + "metadata", + "metadata", + id="tag", + ), + pytest.param( + _payload(tasks=[]), + _payload( + tasks=[], + dependencies=[ + { + "dependency_type": "sensor", + "dependency_id": "upstream-task", + "source": "upstream-dag", + "target": "example", + "label": "secret_callback", + } + ], + ), + "secret_callback", + "/dag/dag_dependencies/*", + "added", + "dependency", + "execution", + id="dependency", + ), + ], +) +def test_build_diff_classification_ignores_redacted_identifiers( + base_data, + target_data, + hidden_identifier, + expected_path, + expected_operation, + expected_category, + expected_impact, +) -> None: + result = build_serialized_dag_diff(base_data=base_data, target_data=target_data) + + assert result["changes"] == [ + { + "path": expected_path, + "operation": expected_operation, + "category": expected_category, + "impact": expected_impact, + } + ] + assert hidden_identifier not in json.dumps(result) + + +def test_build_diff_preserves_known_task_field_classification_without_values() -> None: + result = build_serialized_dag_diff( + base_data=_payload(tasks=[{"task_id": "extract", "outlets": []}]), + target_data=_payload(tasks=[{"task_id": "extract", "outlets": ["asset"]}]), + ) + + assert result["changes"] == [ + { + "path": "/dag/tasks/*/outlets", + "operation": "changed", + "category": "asset", + "impact": "execution", + } + ] + + def test_build_diff_collapses_arbitrary_mappings_without_values() -> None: base = _payload(tasks=[]) target = _payload(tasks=[]) From 28351e9b8467e8d13ffec78807626ca512d9bdb0 Mon Sep 17 00:00:00 2001 From: Ephraim Anierobi Date: Tue, 14 Jul 2026 16:41:32 +0100 Subject: [PATCH 4/4] Harden Dag version diff authorization handling Authorization state crosses an internal boundary before API integration, so invalid inputs must fail closed. Consistent not-found semantics also ensure future callers can preserve the required HTTP 404 behavior. --- airflow-core/src/airflow/exceptions.py | 2 +- .../src/airflow/models/dag_version.py | 26 +++++++---- .../tests/unit/models/test_dag_version.py | 45 +++++++++++++++---- 3 files changed, 54 insertions(+), 19 deletions(-) diff --git a/airflow-core/src/airflow/exceptions.py b/airflow-core/src/airflow/exceptions.py index cb45b050fce34..fdc0408137ee0 100644 --- a/airflow-core/src/airflow/exceptions.py +++ b/airflow-core/src/airflow/exceptions.py @@ -142,7 +142,7 @@ class DagRunNotFound(AirflowNotFoundException): class DagVersionNotFound(AirflowNotFoundException): - """Raised when a DagVersion for the given dag_id / bundle_version is not found.""" + """Raised when a requested DagVersion is not found.""" class DagNotPartitionedError(ValueError): diff --git a/airflow-core/src/airflow/models/dag_version.py b/airflow-core/src/airflow/models/dag_version.py index 0ab447e03d1df..59734eb7bcb83 100644 --- a/airflow-core/src/airflow/models/dag_version.py +++ b/airflow-core/src/airflow/models/dag_version.py @@ -20,7 +20,7 @@ import hashlib import logging from datetime import datetime -from typing import TYPE_CHECKING, Any, Literal +from typing import TYPE_CHECKING, Any, Literal, get_args from uuid import UUID import sqlalchemy as sa @@ -30,6 +30,7 @@ from airflow._shared.timezones import timezone from airflow.dag_processing.bundles.manager import DagBundlesManager +from airflow.exceptions import DagVersionNotFound from airflow.models.base import Base, StringID from airflow.utils.session import NEW_SESSION, provide_session from airflow.utils.sqlalchemy import UtcDateTime, with_row_locks @@ -43,10 +44,8 @@ SourceStatus = Literal["current_stored_code", "redacted", "unavailable"] ValuesStatus = Literal["available", "unavailable"] - - -class DagVersionNotFoundError(ValueError): - """Raised when one of the requested Dag versions does not exist.""" +_VALID_SOURCE_STATUSES = get_args(SourceStatus) +_VALID_VALUES_STATUSES = get_args(ValuesStatus) class DagVersion(Base): @@ -274,16 +273,25 @@ def get_diff( method returns any source content. API callers use ``values_status`` to suppress raw values when the user cannot access Dag code; the structural diff remains available in redacted form. """ + # Keep this local to avoid the dag_version -> dag_version_diff -> serialized_objects cycle. from airflow.serialization.dag_version_diff import ( DEFAULT_MAX_CHANGES, build_serialized_dag_diff, ) + if source_status is not None and source_status not in _VALID_SOURCE_STATUSES: + raise ValueError(f"source_status must be one of {_VALID_SOURCE_STATUSES}, not {source_status!r}") + if values_status is not None and values_status not in _VALID_VALUES_STATUSES: + raise ValueError(f"values_status must be one of {_VALID_VALUES_STATUSES}, not {values_status!r}") + if max_changes is None: max_changes = DEFAULT_MAX_CHANGES if base_version_number < 1 or target_version_number < 1: raise ValueError("Dag version numbers must be positive integers") + if source_status is None: + source_status = "current_stored_code" if include_source else "unavailable" + query = ( select(cls) .where( @@ -292,7 +300,7 @@ def get_diff( ) .options(joinedload(cls.serialized_dag)) ) - if include_source and source_status not in {"redacted", "unavailable"}: + if include_source and source_status == "current_stored_code": query = query.options(joinedload(cls.dag_code)) versions = {version.version_number: version for version in session.scalars(query).all()} @@ -305,13 +313,13 @@ def get_diff( None, ) if missing_version is not None: - raise DagVersionNotFoundError( + raise DagVersionNotFound( f"The DagVersion with dag_id: `{dag_id}` and version_number: `{missing_version}` was not found" ) base_version = versions[base_version_number] target_version = versions[target_version_number] - effective_include_values = include_values and values_status != "unavailable" + effective_include_values = include_values and values_status in {None, "available"} result = build_serialized_dag_diff( base_data=base_version.serialized_dag.data if base_version.serialized_dag else None, target_data=target_version.serialized_dag.data if target_version.serialized_dag else None, @@ -327,7 +335,7 @@ def get_diff( base_version, target_version, include_source=include_source, - source_status=source_status or ("current_stored_code" if include_source else "unavailable"), + source_status=source_status, ) return result diff --git a/airflow-core/tests/unit/models/test_dag_version.py b/airflow-core/tests/unit/models/test_dag_version.py index 766b493c0a991..231c230a68687 100644 --- a/airflow-core/tests/unit/models/test_dag_version.py +++ b/airflow-core/tests/unit/models/test_dag_version.py @@ -17,14 +17,16 @@ from __future__ import annotations from datetime import timedelta +from typing import Any from unittest import mock import pytest from sqlalchemy import delete, func, select from airflow._shared.timezones import timezone +from airflow.exceptions import DagVersionNotFound from airflow.models.dag import DagModel -from airflow.models.dag_version import DagVersion, DagVersionNotFoundError +from airflow.models.dag_version import DagVersion from airflow.models.dagbundle import DagBundleModel from airflow.providers.standard.operators.empty import EmptyOperator @@ -248,7 +250,7 @@ def test_reports_observed_changes_between_versions(self, dag_id, session): assert "values" not in result def test_raises_for_missing_version(self, dag_id, session): - with pytest.raises(DagVersionNotFoundError, match="version_number: `3`"): + with pytest.raises(DagVersionNotFound, match="version_number: `3`"): DagVersion.get_diff(dag_id, 1, 3, session=session) @pytest.mark.parametrize("version_numbers", [(0, 2), (1, 0)]) @@ -256,8 +258,11 @@ def test_rejects_non_positive_version_numbers(self, dag_id, session, version_num with pytest.raises(ValueError, match="Dag version numbers must be positive integers"): DagVersion.get_diff(dag_id, *version_numbers, session=session) - def test_marks_values_available_for_operator_authority(self, dag_id, session): - result = DagVersion.get_diff(dag_id, 1, 2, include_values=True, session=session) + @pytest.mark.parametrize("values_status", [None, "available"]) + def test_marks_values_available_when_allowed(self, dag_id, session, values_status): + result = DagVersion.get_diff( + dag_id, 1, 2, include_values=True, values_status=values_status, session=session + ) assert result["values"] == {"status": "available"} assert any("after_value" in change for change in result["changes"]) @@ -279,8 +284,11 @@ def test_marks_values_unavailable_when_status_denied(self, dag_id, session): ) assert result["values"] == {"status": "unavailable"} - def test_includes_current_stored_source(self, dag_id, session): - result = DagVersion.get_diff(dag_id, 1, 2, include_source=True, session=session) + @pytest.mark.parametrize("source_status", [None, "current_stored_code"]) + def test_includes_current_stored_source(self, dag_id, session, source_status): + result = DagVersion.get_diff( + dag_id, 1, 2, include_source=True, source_status=source_status, session=session + ) source = result["source"] assert source["status"] == "current_stored_code" @@ -288,12 +296,31 @@ def test_includes_current_stored_source(self, dag_id, session): assert source["base"]["digest"].startswith("sha256:") assert source["target"]["content"] is not None - def test_redacts_source_when_status_denied(self, dag_id, session): + @pytest.mark.parametrize("source_status", ["redacted", "unavailable"]) + def test_hides_source_when_status_denied(self, dag_id, session, source_status): result = DagVersion.get_diff( - dag_id, 1, 2, include_source=True, source_status="redacted", session=session + dag_id, 1, 2, include_source=True, source_status=source_status, session=session ) - assert result["source"] == {"status": "redacted", "fidelity": "redacted"} + assert result["source"] == {"status": source_status, "fidelity": source_status} + + @pytest.mark.parametrize( + ("status_name", "invalid_status"), + [("source_status", ""), ("values_status", "redacted")], + ) + def test_rejects_invalid_authorization_status(self, dag_id, session, status_name, invalid_status): + status: dict[str, Any] = {status_name: invalid_status} + + with pytest.raises(ValueError, match=rf"{status_name} must be one of"): + DagVersion.get_diff( + dag_id, + 1, + 2, + include_source=True, + include_values=True, + session=session, + **status, + ) def test_marks_source_unavailable_when_code_missing(self, dag_id, session): from airflow.models.dagcode import DagCode