github-advanced-security[bot] commented on code in PR #69864: URL: https://github.com/apache/airflow/pull/69864#discussion_r4005422661
########## airflow-core/src/airflow/serialization/dag_version_diff.py: ########## @@ -0,0 +1,896 @@ +# 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 datetime import timedelta +from enum import Enum +from typing import Any, Literal + +import structlog + +from airflow.serialization.definitions.baseoperator import SerializedBaseOperator +from airflow.serialization.definitions.mappedoperator import SerializedMappedOperator +from airflow.serialization.serialized_objects import ( + _DAG_CALLBACK_FIELDS, + _OPERATOR_TIMEDELTA_FIELDS, + DagSerialization, + OperatorSerialization, +) + +log = structlog.get_logger(__name__) + +DIFF_SCHEMA_VERSION = 1 +DEFAULT_MAX_CHANGES = 500 +MAX_ALLOWED_CHANGES = 5000 +SUPPORTED_SERIALIZED_DAG_SCHEMA_VERSIONS = frozenset((1, 2, 3)) + +_ORDER_INSENSITIVE_LIST_PATHS = { + ("dag", "tags"), + ("dag", "allowed_run_types"), +} +_KEYED_COLLECTION_PATHS = { + ("dag", "tasks"), + ("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({"_arg_bindings"}) +# _get_category classifies task fields with these; every name must be a public task field or the +# entry is unreachable, since a non-public field is aggregated under custom_fields before lookup. +_DIFF_V1_TASK_ASSET_FIELDS = frozenset({"inlets", "outlets"}) +_DIFF_V1_TASK_PARAM_FIELDS = frozenset({"params"}) +_DIFF_V1_TASK_DEPENDENCY_FIELDS = frozenset({"downstream_task_ids"}) +_DIFF_V1_TASK_METADATA_FIELDS = frozenset( + { + "doc", + "doc_json", + "doc_md", + "doc_rst", + "doc_yaml", + "owner", + "ui_color", + "ui_fgcolor", + "_task_display_name", + "task_display_name", + } +) +_DIFF_V1_PUBLIC_PARTIAL_TASK_FIELDS = _DIFF_V1_PUBLIC_TASK_FIELDS | {"task_display_name"} +# Classify every Dag schema field explicitly so new fields require a policy decision. +_DIFF_V1_DAG_FIELD_CATEGORIES = { + "_concurrency": "schedule", + "_processor_dags_folder": "provenance", + "access_control": "authorization", + "allowed_run_types": "schedule", + "bundle_name": "provenance", + "catchup": "schedule", + "dag_dependencies": "dependency", + "dag_display_name": "metadata", + "dag_id": "metadata", + "dagrun_timeout": "schedule", + "deadline": "deadline", + "default_args": "param", + "description": "metadata", + "disable_bundle_versioning": "task", + "doc_md": "metadata", + "edge_info": "metadata", + "end_date": "schedule", + "fail_fast": "schedule", + "fileloc": "provenance", + "has_on_failure_callback": "callback", + "has_on_success_callback": "callback", + "is_paused_upon_creation": "schedule", + "max_active_runs": "schedule", + "max_active_tasks": "schedule", + "max_consecutive_failed_dag_runs": "schedule", + "owner_links": "metadata", + "params": "param", + "relative_fileloc": "provenance", + "render_template_as_native_obj": "task", + "rerun_with_latest_version": "task", + "start_date": "schedule", + "tags": "metadata", + "task_group": "task", + "tasks": "task", + "timetable": "schedule", + "timezone": "schedule", +} +_DIFF_V1_LEGACY_DAG_FIELD_CATEGORIES = { + "fail_stop": "schedule", + "on_failure_callback": "callback", + "on_success_callback": "callback", + "schedule": "schedule", + "schedule_interval": "schedule", +} +_RECURSIVE_MAPPING_PATHS = { + (), + ("dag",), + ("provenance",), + *_KEYED_COLLECTION_PATHS, +} +_DIFF_V1_TASK_GROUP_METADATA_FIELDS = frozenset( + {"group_display_name", "tooltip", "doc_md", "ui_color", "ui_fgcolor"} +) +_DIFF_V1_PUBLIC_TASK_GROUP_FIELDS = _DIFF_V1_TASK_GROUP_METADATA_FIELDS | { + "_group_id", + "prefix_group_id", + "children", + "upstream_group_ids", + "downstream_group_ids", + "upstream_task_ids", + "downstream_task_ids", + "expand_input", + "is_mapped", +} + + +def build_unavailable_dag_diff( + *, + base_data: dict[str, Any] | None, + target_data: dict[str, Any] | None, + reason: str, +) -> dict[str, Any]: + """Report a known unavailable reason without comparing the stored payloads.""" + return _mark_unavailable( + _build_diff_result(_get_schema_version(base_data), _get_schema_version(target_data)), reason + ) + + +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. + + Raw values, digests, and value-derived path components are returned only when + ``include_values`` is true. Callers must authorize disclosure of the entire + serialized payload, including access-control role names and permission mappings, + before enabling it. + """ + validate_max_changes(max_changes) + + base_schema_version = _get_schema_version(base_data) + target_schema_version = _get_schema_version(target_data) + result = _build_diff_result(base_schema_version, target_schema_version) + + if base_data is None or target_data is None: + return _mark_unavailable(result, "serialized_dag_missing") + + if base_schema_version is None or target_schema_version is None: + return _mark_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 _mark_unavailable( + result, f"unsupported_serialized_dag_schema_version:{unsupported_versions[0]}" + ) + + try: + base_document = _canonicalize_payload_v1(base_data) + target_document = _canonicalize_payload_v1(target_data) + base_document["provenance"] = _canonicalize_value(dict(base_provenance or {}), path=("provenance",)) + target_document["provenance"] = _canonicalize_value( + dict(target_provenance or {}), path=("provenance",) + ) + except (AttributeError, KeyError, OverflowError, TypeError, ValueError) as error: + log.warning( + "Serialized Dag diff canonicalization failed", + error_type=type(error).__name__, + base_schema_version=base_schema_version, + target_schema_version=target_schema_version, + ) + return _mark_unavailable(result, "serialized_dag_canonicalization_failed") + + collector = _ChangeCollector(max_changes=max_changes, include_values=include_values) + try: + _collect_changes(base_document, target_document, path=(), collector=collector) + except _JsonEncodingError: + log.warning( + "Serialized Dag diff JSON encoding failed", + base_schema_version=base_schema_version, + target_schema_version=target_schema_version, + ) + return _mark_unavailable(result, "serialized_dag_json_encoding_failed") + + result["changes"] = collector.changes + result["truncated"] = collector.is_truncated + if include_values: + result["values"] = {"status": "available"} + 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 + + @property + def is_truncated(self) -> bool: + return self.count > self.max_changes + + 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 + + public_path = _get_public_path(path) + category = _get_category(public_path) + change: dict[str, Any] = { + "path": _format_path(path if self.include_values else public_path), + "operation": operation, + "category": category, + "impact": _get_impact(category), + } + 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: + change["after_value"] = after + self.changes.append(change) + + +_MISSING = object() + + +def validate_max_changes(max_changes: int) -> None: + if not isinstance(max_changes, int) or isinstance(max_changes, bool) or 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 _mark_unavailable(result: dict[str, Any], reason: str) -> dict[str, Any]: + result["mode"] = "unavailable" + result["unavailable_reason"] = reason + return result + + +def _build_diff_result(base_schema_version: int | None, target_schema_version: int | None) -> dict[str, Any]: + return { + "diff_schema_version": DIFF_SCHEMA_VERSION, + "serialized_dag_schema_versions": { + "base": base_schema_version, + "target": target_schema_version, + }, + "mode": "observed_state", + "changes": [], + "truncated": False, + # Always present so that reading result["values"]["status"] is safe for every caller + # of every entry point, whatever the outcome. + "values": {"status": "unavailable"}, + } + + +def _canonicalize_payload_v1(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") + dag_defaults = { + field: value + for field, value in DagSerialization.get_schema_defaults("dag").items() + # Dag callback flags are enabled by their presence, even when their value is false. + if field not in _DAG_CALLBACK_FIELDS + } + payload["dag"] = {**dag_defaults, **payload["dag"]} + for field in _DAG_CALLBACK_FIELDS & payload["dag"].keys(): + payload["dag"][field] = True + if "params" in payload["dag"]: + payload["dag"]["params"] = _normalize_params(payload["dag"]["params"]) + _apply_task_defaults(payload) + payload.pop("__version", None) + return _canonicalize_value(payload, path=()) + + +def _apply_task_defaults(payload: dict[str, Any]) -> None: + client_defaults = payload.pop("client_defaults", None) + if client_defaults is None: + client_defaults = {} + if not isinstance(client_defaults, Mapping): + raise ValueError("client_defaults is not an object") + + # Fail loudly on a section this version cannot fold in: dropping it would silently + # compare two payloads as equal when the unhandled defaults actually differ. + if unknown_sections := client_defaults.keys() - {"tasks"}: + raise ValueError(f"unsupported client_defaults sections: {sorted(unknown_sections)}") + + task_defaults = client_defaults.get("tasks", {}) + if not isinstance(task_defaults, Mapping): + raise ValueError("client_defaults.tasks is not an object") + + schema_defaults = DagSerialization.get_schema_defaults("operator") + partial_fields = ( + SerializedBaseOperator.get_serialized_fields() - SerializedMappedOperator.get_serialized_fields() + ) + # A mapped task resolves these through partial_kwargs, so their defaults have to be applied + # there and encoded like any other partial value rather than left at the outer level. + partial_schema_defaults = { + field: value for field, value in schema_defaults.items() if field in partial_fields + } + outer_schema_defaults = { + field: value for field, value in schema_defaults.items() if field not in partial_fields + } + # set_task_dag_references falls back to the Dag's dates, and the serializer elides a task date + # that already matches, so an absent task date means the Dag's date rather than a change. + inherited_dates = {field: payload["dag"].get(field) for field in ("start_date", "end_date")} + 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") + encoded_task = OperatorSerialization._apply_defaults_to_encoded_op( + dict(task["__var"]), dict(client_defaults) + ) + upgraded_task = OperatorSerialization._upgrade_encoded_operator(encoded_task) + # Operator identity is selected before client defaults are applied during hydration. + upgraded_task["_is_mapped"] = bool(task["__var"].get("_is_mapped", False)) + template_fields = upgraded_task.get("template_fields", []) + if upgraded_task.get("_is_mapped"): + task_data = {**outer_schema_defaults, **upgraded_task} + partial_kwargs = task_data.get("partial_kwargs", {}) + if not isinstance(partial_kwargs, Mapping): + raise ValueError("partial_kwargs is not an object") + # populate_operator only folds client defaults into partial_kwargs when the payload + # carries the key, so an absent one leaves the top-level value as the effective value. + effective_partial_kwargs = ( + {field: _encode_partial_field_value(field, value) for field, value in task_defaults.items()} + if "partial_kwargs" in task_data + else {} + ) + for field, value in partial_kwargs.items(): + if isinstance(value, Mapping) and "__type" in value and "__var" in value: + effective_partial_kwargs[field] = value + else: + effective_partial_kwargs[field] = _encode_partial_field_value(field, value) + # Match populate_operator: partial values take precedence over outer task defaults. + for field in partial_fields & task_data.keys(): + value = task_data.pop(field) + # Outer fields are already encoded unless template handling bypasses deserialization. + if field in template_fields: + value = _encode_json_value(value) + effective_partial_kwargs.setdefault(field, value) + for field, value in partial_schema_defaults.items(): + effective_partial_kwargs.setdefault(field, _encode_partial_field_value(field, value)) + task_data["partial_kwargs"] = effective_partial_kwargs + _normalize_retry_backoff(effective_partial_kwargs) + else: + task_data = {**schema_defaults, **upgraded_task} + _normalize_retry_backoff(task_data) + for field in _OPERATOR_TIMEDELTA_FIELDS & task_data.keys(): + value = task_data[field] + task_data[field] = ( + _encode_json_value(value) + if field in template_fields and field in upgraded_task + else _encode_partial_field_value(field, value) + ) + for field, dag_date in inherited_dates.items(): + if task_data.get(field) is None: + task_data[field] = _encode_partial_field_value(field, dag_date) + elif field in template_fields: + task_data[field] = _encode_json_value(task_data[field]) + else: + task_data[field] = _encode_partial_field_value(field, task_data[field]) + if "params" in task_data and "params" not in template_fields: + task_data["params"] = _normalize_params(task_data["params"]) + task["__var"] = task_data + + +def _normalize_params(params: Any) -> Any: + """Normalize legacy Params without losing the order used by the trigger form.""" + if isinstance(params, Mapping): + # 2.9.2 and earlier stored params as a JSON object instead of ordered pairs. + pairs: Any = params.items() + elif isinstance(params, list): + pairs = params + else: + return params + + normalized: dict[str, Any] = {} + for pair in pairs: + if not isinstance(pair, (list, tuple)) or len(pair) != 2: + # Leave an unrecognised shape to the raw comparison rather than guessing at it. + return params + name, value = pair + if isinstance(value, Mapping) and "__class" in value: + value = { + "default": _normalize_param_attribute(value.get("default")), + "description": _normalize_param_attribute(value.get("description")), + "schema": _normalize_param_attribute(value.get("schema") or {}), + "source": _normalize_param_attribute(value.get("source")), + } + else: + value = { + "default": _encode_json_value(value), + "description": None, + "schema": _encode_json_value({}), + "source": None, + } + normalized[str(name)] = value + return [[name, value] for name, value in normalized.items()] + + +def _normalize_param_attribute(value: Any) -> Any: + # Match _deserialize_param's legacy-encoding detection without hydrating user objects. + if isinstance(value, Mapping) and "__type" in value: + return value + if isinstance(value, list) and all(isinstance(item, Mapping) and "__type" in item for item in value): + return value + return _encode_json_value(value) + + +def _encode_partial_field_value(field: str, value: Any) -> Any: + if field in _OPERATOR_TIMEDELTA_FIELDS and value is not None: + return {"__type": "timedelta", "__var": value} + if field.endswith("_date") and value is not None and not isinstance(value, str): + return {"__type": "datetime", "__var": value} + if field == "resources": + # Resources serialize as a raw mapping; a plain dictionary has a dict envelope instead. + return value + return _encode_json_value(value) + + +def _encode_json_value(value: Any) -> Any: + """Encode plain JSON without invoking object serializers.""" + if isinstance(value, Mapping): + return {"__type": "dict", "__var": {key: _encode_json_value(item) for key, item in value.items()}} + if isinstance(value, list): + return [_encode_json_value(item) for item in value] + return value + + +def _normalize_retry_backoff(task_fields: dict[str, Any]) -> None: + if "retry_exponential_backoff" in task_fields: + value = task_fields["retry_exponential_backoff"] + task_fields["retry_exponential_backoff"] = 2.0 if value is True else float(value) + + +def _canonicalize_value(value: Any, *, path: tuple[str, ...]) -> Any: + if isinstance(value, Mapping): + if path == ("dag", "deadline"): + if value.get("__type") == "deadline_alert": + value = value["__var"] + value = {"name": None, **value} + interval = value.get("interval") + if isinstance(interval, (int, float)) and not isinstance(interval, bool): + # Diff schema v1 uses the SDK's version-2 timedelta encoding for legacy seconds. + value["interval"] = { + "__classname__": "datetime.timedelta", + "__version__": 2, + "__data__": timedelta(seconds=interval).total_seconds(), + } + return { + 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, tuple)): + 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_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]: + keyed_values: dict[str, Any] = {} + for value in values: + key = key_getter(value) + if key in keyed_values: + if path in _ORDER_INSENSITIVE_LIST_PATHS or ( + path == ("dag", "dag_dependencies") and keyed_values[key] == value + ): + continue + raise ValueError(f"duplicate key {key!r} in /{'/'.join(path)}") + keyed_values[key] = 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 before is _MISSING and after is _MISSING: + return + if isinstance(before, Mapping) and isinstance(after, Mapping): + # The walk shape never depends on ``include_values``: authorization decides what each + # change carries, not which changes exist, so ``max_changes`` means one thing and a + # caller allowed values never receives a less complete change set than a redacted one. + if _is_task_mapping_path(path): + _collect_task_changes(before, after, path=path, collector=collector) + return + if _is_task_group_mapping_path(path): + _collect_public_field_changes( + before, + after, + path=path, + public_fields=_DIFF_V1_PUBLIC_TASK_GROUP_FIELDS, + collector=collector, + ) + return + if not _should_recurse_mapping(path): + if not _is_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) + after_value = after.get(key, _MISSING) + _collect_changes(before_value, after_value, path=path + (key,), collector=collector) + if collector.is_truncated: + return + return + + if isinstance(before, list) and isinstance(after, list): + if ( + _is_task_group_child_path(path) + and len(before) == len(after) == 2 + and before[0] == after[0] == "taskgroup" + and isinstance(before[1], Mapping) + and isinstance(after[1], Mapping) + ): + _collect_changes(before[1], after[1], path=path + ("1",), collector=collector) + return + if not _is_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 _is_json_equal(before, after): + collector.add(path=path, operation="changed", before=before, after=after) + + +def _collect_task_changes( + before: Mapping[str, Any], + after: Mapping[str, Any], + *, + path: tuple[str, ...], + collector: _ChangeCollector, +) -> None: + if len(path) == 3: + _collect_changes( + before.get("__type", _MISSING), + after.get("__type", _MISSING), + path=path + ("__type",), + collector=collector, + ) + if collector.is_truncated: + return + before, after = before["__var"], after["__var"] + public_fields = _DIFF_V1_PUBLIC_TASK_FIELDS - {"__type"} + else: + public_fields = _DIFF_V1_PUBLIC_PARTIAL_TASK_FIELDS + _collect_public_field_changes(before, after, path=path, public_fields=public_fields, collector=collector) + + +def _collect_public_field_changes( + before: Mapping[str, Any], + after: Mapping[str, Any], + *, + path: tuple[str, ...], + public_fields: frozenset[str] | set[str], + collector: _ChangeCollector, +) -> None: + """Walk allowlisted fields one by one and report everything else as a single change.""" + 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, + ) + if collector.is_truncated: + return + + 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 _is_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 path[:2] == ("dag", "tasks") and ( + len(path) == 3 or (len(path) == 4 and path[3] == "partial_kwargs") + ) + + +def _is_task_group_mapping_path(path: tuple[str, ...]) -> bool: + index = _get_task_group_field_index(path) + return index is not None and len(path) == index + + +def _should_recurse_mapping(path: tuple[str, ...]) -> bool: + if path in _RECURSIVE_MAPPING_PATHS: + return True + index = _get_task_group_field_index(path) + return index is not None and path[index:] == ("children",) + + +def _get_task_group_field_index(path: tuple[str, ...]) -> int | None: + if path[:2] != ("dag", "task_group"): + return None + index = 2 + while len(path) >= index + 3 and path[index] == "children" and path[index + 2] == "1": + index += 3 + return index + + +def _is_task_group_child_path(path: tuple[str, ...]) -> bool: + index = _get_task_group_field_index(path) + return index is not None and len(path) == index + 2 and path[index] == "children" + + +def _get_public_path(path: tuple[str, ...]) -> tuple[str, ...]: + if path[:2] == ("dag", "task_group"): + public_path = list(path) + index = 2 + while len(path) >= index + 2 and path[index] == "children": + public_path[index + 1] = "*" + if len(path) < index + 3 or path[index + 2] != "1": + return tuple(public_path) + index += 3 + if len(path) > index and path[index] not in _DIFF_V1_PUBLIC_TASK_GROUP_FIELDS: + public_path[index] = "custom_fields" + return tuple(public_path) + if len(path) >= 3 and path[:2] in _KEYED_COLLECTION_PATHS: + path = (*path[:2], "*", *path[3:]) + if len(path) >= 4 and path[:2] == ("dag", "tasks"): + field_index = 4 if path[3] == "partial_kwargs" and len(path) >= 5 else 3 + public_fields = ( + _DIFF_V1_PUBLIC_PARTIAL_TASK_FIELDS if field_index == 4 else _DIFF_V1_PUBLIC_TASK_FIELDS + ) + if path[field_index] not in public_fields: + return (*path[:field_index], _CUSTOM_TASK_FIELDS_PATH_COMPONENT) + return path + + +def _format_path(path: tuple[str, ...]) -> str: + return "/" + "/".join(component.replace("~", "~0").replace("/", "~1") for component in path) + + +class _JsonEncodingError(TypeError): + """Distinguish unsupported JSON values from comparison implementation errors.""" + + +def _serialize_canonical_json(value: Any) -> str: + try: + return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + except TypeError as error: + raise _JsonEncodingError from error + + +def _is_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 _serialize_canonical_json(before) == _serialize_canonical_json(after) + + +def _get_digest(value: Any) -> str: + return f"sha256:{hashlib.sha256(_serialize_canonical_json(value).encode()).hexdigest()}" Review Comment: ## CodeQL / Use of a broken or weak cryptographic hashing algorithm on sensitive data [Sensitive data (password)](1) is used in a hashing algorithm (SHA256) that is insecure for password hashing, since it is not a computationally expensive hash function. [Show more details](https://github.com/apache/airflow/security/code-scanning/655) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
