[v3-3-test] Decode deadline alert interval and callback without generic deserialization (#72651)

airflow-core/src/airflow/serialization/decoders.py

@@ -56,6 +56,7 @@
     from airflow.partition_mappers.base import PartitionMapper
     from airflow.partition_mappers.wait_policy import WaitPolicy
     from airflow.partition_mappers.window import Window
+    from airflow.sdk.definitions.callback import AsyncCallback, SyncCallback  # noqa: SDK001
     from airflow.timetables.base import Timetable as CoreTimetable
 
 R = TypeVar("R")
@@ -181,15 +182,138 @@ def decode_deadline_reference(reference_data: dict):
     return reference_class.deserialize_reference(reference_data)
 
 
+_TIMEDELTA_CLASSNAME = "datetime.timedelta"
+_VARIABLE_INTERVAL_CLASSNAMES = frozenset(
+    {
+        "airflow.sdk.definitions.deadline.VariableInterval",
+        "airflow.serialization.definitions.deadline.SerializedVariableInterval",
+    }
+)
+# The callback classes lived here until 3.2 moved them to ``...definitions.callback``,
+# so alerts serialized by an earlier version name this module instead.
+_LEGACY_CALLBACK_MODULE = "airflow.sdk.definitions.deadline"
+
+
+def _normalised_payload(encoded: Any, field: str) -> tuple[str, Any]:
+    """
+    Return ``(classname, data)`` for a serde payload, in either encoding.
+
+    ``serde`` accepts a legacy ``{"__type": ..., "__var": ...}`` shape and rewrites it
+    into the current one *inside* ``deserialize``. Anything inspecting the payload before
+    that call therefore has to normalise it first, or the legacy spelling carries no
+    ``__classname__`` at the moment it is looked at and slips past unexamined.
+    """
+    from airflow.sdk.serde import CLASSNAME, DATA, _convert
+
+    if not isinstance(encoded, dict):
+        raise ValueError(f"Deadline {field} is not a serialized object.")
+    converted = _convert(encoded)
+    if not isinstance(converted, dict):
+        raise ValueError(f"Deadline {field} is not a serialized object.")
+    if CLASSNAME not in converted:
+        raise ValueError(f"Deadline {field} does not name a class.")
+    return converted[CLASSNAME], converted.get(DATA)
+
+
+def _decode_deadline_interval(raw_interval: Any) -> datetime.timedelta | SerializedVariableInterval:
+    """
+    Build the interval from its encoded form without importing what the payload names.
+
+    Only three shapes are legitimate, and each is reconstructed from primitives directly.
+    Nothing here reaches ``serde.deserialize``, so no class named by a Dag author is
+    imported or instantiated in the scheduler or the API server.
+    """
+    # Backward compatibility: previously stored as total_seconds().
+    if isinstance(raw_interval, (int, float)) and not isinstance(raw_interval, bool):
+        return datetime.timedelta(seconds=raw_interval)
+
+    classname, data = _normalised_payload(raw_interval, "interval")
+
+    if classname == _TIMEDELTA_CLASSNAME:
+        if isinstance(data, (int, float)) and not isinstance(data, bool):
+            return datetime.timedelta(seconds=data)
+        raise ValueError("Deadline interval timedelta payload is not a number.")
+
+    if classname in _VARIABLE_INTERVAL_CLASSNAMES:
+        key = data.get("key") if isinstance(data, dict) else None
+        if not isinstance(key, str):
+            raise ValueError("Deadline interval variable payload has no string key.")
+        return SerializedVariableInterval(key=key)
+
+    raise ValueError(
+        f"Refusing to deserialize {classname!r} as a deadline interval. "
+        f"Permitted: {_TIMEDELTA_CLASSNAME}, {', '.join(sorted(_VARIABLE_INTERVAL_CLASSNAMES))}."
+    )
+
+
+def _decode_deadline_callback(raw_callback: Any) -> AsyncCallback | SyncCallback:
+    """
+    Build the callback from its encoded form without importing what the payload names.
+
+    ``kwargs`` is still passed through generic deserialization, and that is a deliberate,
+    documented limit rather than an oversight. Leaving it encoded would close a real
+    residual -- a legitimate callback can carry an arbitrary allow-listed class under its
+    kwargs, which serde constructs while the outer payload looks entirely valid -- but the
+    kwargs are consumed through two different paths that use two different encodings
+    (``BaseSerialization`` in the triggerer, serde here), and deferring the decode without
+    getting both exactly right silently hands user code an encoded dict in place of its
+    argument. That residual is not specific to deadlines: it is the general property of
+    deserializing Dag-author data, shared with every other serde call site. Closing it
+    belongs with that broader work, not smuggled in here.
+    """
+    from airflow.sdk.definitions.callback import (
+        AsyncCallback,
+        SyncCallback,
+        _SerializedCallbackPath,
+    )
+
+    permitted = {
+        f"{module}.{cls.__qualname__}": cls
+        for cls in (AsyncCallback, SyncCallback)
+        for module in (cls.__module__, _LEGACY_CALLBACK_MODULE)
+    }
+    classname, data = _normalised_payload(raw_callback, "callback")
+    callback_cls = permitted.get(classname)
+    if callback_cls is None:
+        raise ValueError(
+            f"Refusing to deserialize {classname!r} as a deadline callback. "
+            f"Permitted: {', '.join(sorted(permitted))}."
+        )
+    if not isinstance(data, dict):
+        raise ValueError("Deadline callback payload is not a mapping.")
+
+    # The fields each class declares are the fields it can be rebuilt from, so an unknown
+    # one means the payload does not describe this callback and nothing below should run.
+    # ``queue`` and ``executor`` belong to one subclass each, never to both.
+    permitted_fields = set(callback_cls.serialized_fields())
+    unexpected = set(data) - permitted_fields
+    if unexpected:
+        raise ValueError(f"Unexpected deadline callback fields: {', '.join(sorted(unexpected))}.")
+
+    path = data.get("path")
+    if not isinstance(path, str):
+        raise ValueError("Deadline callback payload has no string path.")
+
+    from airflow.sdk.serde import deserialize
+
+    raw_kwargs = data.get("kwargs") or {}
+    fields: dict[str, Any] = {"kwargs": deserialize(raw_kwargs) if raw_kwargs else {}}
+    for optional in sorted(permitted_fields - {"path", "kwargs"}):
+        if optional in data:
+            value = data[optional]
+            if value is not None and not isinstance(value, str):
+                raise ValueError(f"Deadline callback {optional} is not a string.")
+            fields[optional] = value
+
+    return callback_cls(callback_callable=_SerializedCallbackPath(path), **fields)
+
+
 def decode_deadline_alert(encoded_data: dict):
     """
     Decode a previously serialized deadline alert.
 
     :meta private:
     """
-    from airflow.sdk.definitions.deadline import VariableInterval
-    from airflow.sdk.serde import deserialize
-
     data = encoded_data.get(Encoding.VAR, encoded_data)
 
     reference_data = data[DeadlineAlertFields.REFERENCE]
@@ -203,28 +327,12 @@ def decode_deadline_alert(encoded_data: dict):
             "from a version that supports VariableInterval. Downgrade is not fully reversible."
         )
 
-    interval: datetime.timedelta | SerializedVariableInterval
-
-    # Backward compatibility: previously interval was stored as total_seconds() (float/int).
-    # Handle numeric values by converting to timedelta.
-    if isinstance(raw_interval, (int, float)):
-        interval = datetime.timedelta(seconds=raw_interval)
-    else:
-        deserialized = deserialize(raw_interval)
-
-        if isinstance(deserialized, datetime.timedelta):
-            interval = deserialized
-        elif isinstance(deserialized, SerializedVariableInterval):
-            interval = deserialized
-        elif isinstance(deserialized, VariableInterval):
-            interval = SerializedVariableInterval(key=deserialized.key)
-        else:
-            raise TypeError(f"Invalid interval type: {type(deserialized).__name__}")
+    interval = _decode_deadline_interval(raw_interval)
 
     return SerializedDeadlineAlert(
         reference=reference,
         interval=interval,
-        callback=deserialize(data[DeadlineAlertFields.CALLBACK]),
+        callback=_decode_deadline_callback(data[DeadlineAlertFields.CALLBACK]),
         name=data.get(DeadlineAlertFields.NAME),
     )
 

airflow-core/tests/unit/serialization/test_serialized_objects.py

@@ -79,6 +79,7 @@
     AsyncCallback,
     DeadlineAlert,
     DeadlineReference,
+    SyncCallback,
     VariableInterval,
 )
 from airflow.sdk.definitions.decorators import task
@@ -1602,3 +1603,188 @@ def test_serialized_dag_getitem_returns_task_group(dag_maker):
     ser_dag = DagSerialization.from_dict(ser_dict)
     assert isinstance(ser_dag, SerializedDAG)
     assert ser_dag["section"].group_id == tg.group_id
+
+
+class _DeadlineGadget:
+    """Stands in for any allow-listed class a Dag author could name in a deadline field.
+
+    Deliberately constructible by serde -- it carries the ``serialize`` /
+    ``deserialize`` pair serde requires. A gadget serde cannot build would make the
+    test pass for the wrong reason: deserialization would fail on its own and the
+    "was it constructed" assertion would never be exercised.
+    """
+
+    instantiated = False
+
+    def __init__(self, **kwargs):
+        type(self).instantiated = True
+        self.kwargs = kwargs
+
+    def serialize(self):
+        return {}
+
+    @staticmethod
+    def deserialize(data, version):
+        return _DeadlineGadget(**(data or {}))
+
+
+def _encode_as(classname: str) -> dict:
+    """Build a serde payload naming ``classname``."""
+    from airflow.sdk.serde import CLASSNAME, DATA, VERSION
+
+    return {CLASSNAME: classname, VERSION: 1, DATA: {}}
+
+
+@pytest.mark.parametrize("field", [DeadlineAlertFields.CALLBACK, DeadlineAlertFields.INTERVAL])
+def test_deadline_fields_refuse_unexpected_classes_without_constructing_them(field):
+    """A Dag author must not be able to name an arbitrary class in a deadline field.
+
+    These decoders run in the scheduler and API server whenever a serialized Dag is
+    loaded, and the Security Model says a Dag author reaches those processes only
+    through registered plugins. The assertion that matters is that the class is never
+    *constructed*: `deserialize()` instantiates before returning, so a check on the
+    returned object would already be too late for a class with side effects in
+    `__init__`.
+    """
+    from unittest import mock
+
+    from airflow.sdk import serde
+
+    gadget_name = f"{_DeadlineGadget.__module__}.{_DeadlineGadget.__qualname__}"
+    _DeadlineGadget.instantiated = False
+
+    valid = DeadlineAlert(
+        reference=DeadlineReference.DAGRUN_QUEUED_AT,
+        interval=timedelta(hours=1),
+        callback=AsyncCallback(TEST_CALLBACK_PATH, kwargs=TEST_CALLBACK_KWARGS),
+    )
+    serialized = BaseSerialization.serialize(valid)
+    serialized[Encoding.VAR][field] = _encode_as(gadget_name)
+
+    # Make the gadget acceptable to serde itself, so the refusal comes from the
+    # deadline gate rather than from the general deserialization allow list.
+    with mock.patch.object(serde, "_extra_allowed", serde._extra_allowed | {gadget_name}):
+        with pytest.raises(ValueError, match="Refusing to deserialize"):
+            BaseSerialization.deserialize(serialized)
+
+    assert _DeadlineGadget.instantiated is False, "the class was constructed before being rejected"
+
+
+def test_deadline_callback_kwargs_still_construct_nested_classes():
+    """Documents a known residual: kwargs are still generically deserialized.
+
+    The outer payload is a valid AsyncCallback and the class sits under ``kwargs``,
+    which serde deserializes recursively. This is NOT closed here, and the test asserts
+    the current behaviour rather than the desired one so the gap is visible and a future
+    change has to update it deliberately.
+
+    It is not specific to deadlines -- it is the general property of deserializing
+    Dag-author data, shared with every other serde call site. Closing it means deferring
+    the kwargs decode to the process that runs the callback, which spans two consumption
+    paths using two different encodings and belongs with that broader work.
+    """
+    from unittest import mock
+
+    from airflow.sdk import serde
+
+    gadget_name = f"{_DeadlineGadget.__module__}.{_DeadlineGadget.__qualname__}"
+    _DeadlineGadget.instantiated = False
+
+    valid = DeadlineAlert(
+        reference=DeadlineReference.DAGRUN_QUEUED_AT,
+        interval=timedelta(hours=1),
+        callback=AsyncCallback(TEST_CALLBACK_PATH, kwargs=TEST_CALLBACK_KWARGS),
+    )
+    serialized = BaseSerialization.serialize(valid)
+    callback_payload = serialized[Encoding.VAR][DeadlineAlertFields.CALLBACK]
+    callback_payload[serde.DATA]["kwargs"] = {"evil": _encode_as(gadget_name)}
+
+    with mock.patch.object(serde, "_extra_allowed", serde._extra_allowed | {gadget_name}):
+        BaseSerialization.deserialize(serialized)
+
+    assert _DeadlineGadget.instantiated is True, (
+        "kwargs deserialization behaviour changed -- if this is now closed, update the "
+        "docstring on _decode_deadline_callback and this test together"
+    )
+
+
+@pytest.mark.parametrize("field", [DeadlineAlertFields.CALLBACK, DeadlineAlertFields.INTERVAL])
+def test_deadline_fields_refuse_legacy_encoded_classes(field):
+    """The legacy ``{"__type", "__var"}`` spelling must not slip past the check.
+
+    serde rewrites that shape into ``__classname__`` *inside* deserialize, so a payload
+    inspected beforehand carries no ``__classname__`` at all and an unnormalised check
+    sees nothing to reject.
+    """
+    from unittest import mock
+
+    from airflow.sdk import serde
+    from airflow.sdk._shared.serialization import OLD_DATA, OLD_TYPE
+
+    gadget_name = f"{_DeadlineGadget.__module__}.{_DeadlineGadget.__qualname__}"
+    _DeadlineGadget.instantiated = False
+
+    valid = DeadlineAlert(
+        reference=DeadlineReference.DAGRUN_QUEUED_AT,
+        interval=timedelta(hours=1),
+        callback=AsyncCallback(TEST_CALLBACK_PATH, kwargs=TEST_CALLBACK_KWARGS),
+    )
+    serialized = BaseSerialization.serialize(valid)
+    serialized[Encoding.VAR][field] = {OLD_TYPE: gadget_name, OLD_DATA: {}}
+
+    with mock.patch.object(serde, "_extra_allowed", serde._extra_allowed | {gadget_name}):
+        with pytest.raises(ValueError, match="Refusing to deserialize"):
+            BaseSerialization.deserialize(serialized)
+
+    assert _DeadlineGadget.instantiated is False
+
+
+@pytest.mark.parametrize("callback_cls", [AsyncCallback, SyncCallback])
+def test_deadline_callback_accepts_pre_3_2_module_path(callback_cls):
+    """Callbacks serialized before 3.2 name the module they were defined in back then.
+
+    3.2 moved them out of ``airflow.sdk.definitions.deadline`` into
+    ``...definitions.callback``, so an alert stored by an earlier version carries the old
+    path. Rejecting it would make those rows undecodable on upgrade.
+    """
+    from airflow.sdk import serde
+
+    valid = DeadlineAlert(
+        reference=DeadlineReference.DAGRUN_QUEUED_AT,
+        interval=timedelta(hours=1),
+        callback=callback_cls(TEST_CALLBACK_PATH, kwargs=TEST_CALLBACK_KWARGS),
+    )
+    serialized = BaseSerialization.serialize(valid)
+    payload = serialized[Encoding.VAR][DeadlineAlertFields.CALLBACK]
+    payload[serde.CLASSNAME] = f"airflow.sdk.definitions.deadline.{callback_cls.__qualname__}"
+
+    decoded = BaseSerialization.deserialize(serialized)
+
+    assert isinstance(decoded.callback, callback_cls)
+    assert decoded.callback.path == TEST_CALLBACK_PATH
+    assert decoded.callback.kwargs == TEST_CALLBACK_KWARGS
+
+
+@pytest.mark.parametrize(
+    ("callback_cls", "foreign_field"),
+    [(AsyncCallback, "executor"), (SyncCallback, "queue")],
+    ids=["async-rejects-executor", "sync-rejects-queue"],
+)
+def test_deadline_callback_rejects_field_belonging_to_the_other_subclass(callback_cls, foreign_field):
+    """``queue`` and ``executor`` belong to one subclass each, not to both.
+
+    Accepting either for both classes would hand the constructor an argument it does not
+    take, turning a malformed payload into a TypeError from deep inside the rebuild.
+    """
+    from airflow.sdk import serde
+
+    valid = DeadlineAlert(
+        reference=DeadlineReference.DAGRUN_QUEUED_AT,
+        interval=timedelta(hours=1),
+        callback=callback_cls(TEST_CALLBACK_PATH, kwargs=TEST_CALLBACK_KWARGS),
+    )
+    serialized = BaseSerialization.serialize(valid)
+    serialized[Encoding.VAR][DeadlineAlertFields.CALLBACK][serde.DATA][foreign_field] = "something"
+
+    with pytest.raises(ValueError, match=f"Unexpected deadline callback fields: {foreign_field}"):
+        BaseSerialization.deserialize(serialized)

generated/known_sdk_imports_in_core.txt

@@ -23,7 +23,7 @@ airflow-core/src/airflow/models/xcom_arg.py::1
 airflow-core/src/airflow/plugins_manager.py::1
 airflow-core/src/airflow/providers_manager.py::5
 airflow-core/src/airflow/secrets/__init__.py::1
-airflow-core/src/airflow/serialization/decoders.py::2
+airflow-core/src/airflow/serialization/decoders.py::3
 airflow-core/src/airflow/serialization/definitions/baseoperator.py::1
 airflow-core/src/airflow/serialization/definitions/dag.py::1
 airflow-core/src/airflow/serialization/definitions/mappedoperator.py::5