This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new c614c57062a Decode deadline alert interval and callback without
generic deserialization (#72651)
c614c57062a is described below
commit c614c57062a95aade62cd5df774ae3c33bd57003
Author: Jarek Potiuk <[email protected]>
AuthorDate: Fri Sep 18 06:47:01 2026 +0200
Decode deadline alert interval and callback without generic deserialization
(#72651)
* Decode deadline alert interval and callback without generic
deserialization
decode_deadline_alert passed the Dag-author controlled interval and
callback to
airflow.sdk.serde.deserialize, which imports the class named in the payload
and
instantiates it with the encoded arguments. These decoders run in the
scheduler
and the API server whenever a serialized Dag is loaded, so any class under
the
airflow.* allow list could be constructed there.
The security model says a Dag author reaches those processes only through
registered plugins and providers, and the codebase enforces that at decode
time
for timetables, priority weight strategies and operator extra links.
Deadline
fields had no equivalent gate.
Both fields are now rebuilt from their encoded form directly:
* interval accepts a number, a timedelta payload, or a variable-interval
payload
carrying a key, each reconstructed from primitives.
* callback accepts AsyncCallback or SyncCallback, selected from a fixed map
rather than imported by name, with path as a string, queue/executor as
optional strings, and unexpected fields refused.
Neither reaches serde.deserialize, so the class a Dag author names in either
field is never imported.
Filtering in front of deserialize was tried first and was not sufficient.
serde
normalises the legacy {__type, __var} shape into __classname__ *inside*
deserialize, so a payload inspected beforehand carries no class name to
reject.
Payloads are normalised before inspection here, and that case is tested.
Known residual, deliberately not closed here: callback kwargs are still
passed
through generic deserialization, so a legitimate callback can carry an
arbitrary
allow-listed class under its kwargs. Deferring that decode to the process
that
runs the callback would close it, but the kwargs are consumed through two
paths
using two different encodings, and getting either wrong hands user code an
encoded dict in place of its argument. The residual is not specific to
deadlines -- it is the general property of deserializing Dag-author data,
shared
with every other serde call site. A test asserts the current behaviour so
the
gap stays visible and any change to it has to be deliberate.
Tests assert the class is never constructed rather than that an error is
raised.
Against unpatched sources the callback case reports DID NOT RAISE and the
interval case names the instance that had already been built.
* Accept pre-3.2 callback paths and validate fields per callback class
Review feedback on the deadline decoding gate.
Callbacks moved out of airflow.sdk.definitions.deadline in 3.2, so alerts
serialized by an earlier version name the old module. The allow list only
held
the current path, which would have made those rows undecodable on upgrade --
the same backward-compatibility case the interval allow list already covers.
The permitted callback fields were a hardcoded set covering both subclasses
at
once, so a payload could carry queue on a SyncCallback or executor on an
AsyncCallback. The set is per class, and each class already declares its own
via serialized_fields(), so ask it rather than restating the answer here and
letting the two drift. That also turns a TypeError raised from inside the
rebuild into the intended refusal, and the check now runs before anything is
reconstructed from the payload.
Generated-by: Claude Opus 5
Claude-Session: https://claude.ai/code/session_012zrnHJHPchB83FtwrYRf5q
---
airflow-core/src/airflow/serialization/decoders.py | 150 ++++++++++++++---
.../unit/serialization/test_serialized_objects.py | 186 +++++++++++++++++++++
generated/known_sdk_imports_in_core.txt | 2 +-
3 files changed, 316 insertions(+), 22 deletions(-)
diff --git a/airflow-core/src/airflow/serialization/decoders.py
b/airflow-core/src/airflow/serialization/decoders.py
index be5812ac24b..ca5db04200d 100644
--- a/airflow-core/src/airflow/serialization/decoders.py
+++ b/airflow-core/src/airflow/serialization/decoders.py
@@ -56,6 +56,7 @@ if TYPE_CHECKING:
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")
@@ -182,15 +183,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]
@@ -204,28 +328,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),
)
diff --git a/airflow-core/tests/unit/serialization/test_serialized_objects.py
b/airflow-core/tests/unit/serialization/test_serialized_objects.py
index 3a84c70d29b..3e0812cfcb9 100644
--- a/airflow-core/tests/unit/serialization/test_serialized_objects.py
+++ b/airflow-core/tests/unit/serialization/test_serialized_objects.py
@@ -79,6 +79,7 @@ from airflow.sdk.definitions.deadline import (
AsyncCallback,
DeadlineAlert,
DeadlineReference,
+ SyncCallback,
VariableInterval,
)
from airflow.sdk.definitions.decorators import task
@@ -1632,3 +1633,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: {}}
+
+
[email protected]("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"
+ )
+
+
[email protected]("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
+
+
[email protected]("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
+
+
[email protected](
+ ("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)
diff --git a/generated/known_sdk_imports_in_core.txt
b/generated/known_sdk_imports_in_core.txt
index 85bd4863e64..6c212ad6f61 100644
--- a/generated/known_sdk_imports_in_core.txt
+++ b/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
