From cb92726e87599586689502cbcf24adfdbe5c5a57 Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Sun, 14 Jun 2026 22:40:57 -0400 Subject: [PATCH 1/3] Resolve serialized exception nodes without importing the stored class name When deserializing AIRFLOW_EXC_SER / BASE_EXC_SER nodes, BaseSerialization resolved the exception class with import_string() on a name taken from the serialized blob. Resolve it against in-memory classes instead, so a stored DAG never imports a class named in the blob: - AIRFLOW_EXC_SER: look the name up in a map of loaded AirflowException subclasses, built once from the in-memory subclass tree; a name that is not a registered AirflowException subclass is rejected. - BASE_EXC_SER: resolve against the fixed {KeyError, AttributeError} set that the encoder is the only producer of. Unknown or disallowed names raise DeserializationError instead of being imported. The trigger-node branch is handled separately. Generated-by: Claude Opus 4.8 following the guidelines at https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions --- .../serialization/serialized_objects.py | 61 +++++++++++++++++-- .../serialization/test_dag_serialization.py | 58 ++++++++++++++++++ 2 files changed, 115 insertions(+), 4 deletions(-) diff --git a/airflow-core/src/airflow/serialization/serialized_objects.py b/airflow-core/src/airflow/serialization/serialized_objects.py index 54bc3389c64ce..5aec9f666c038 100644 --- a/airflow-core/src/airflow/serialization/serialized_objects.py +++ b/airflow-core/src/airflow/serialization/serialized_objects.py @@ -29,7 +29,7 @@ import math import sys import weakref -from collections.abc import Collection, Iterable, Mapping +from collections.abc import Collection, Iterable, Iterator, Mapping from functools import cache, cached_property, lru_cache from inspect import signature from textwrap import dedent @@ -41,7 +41,7 @@ from dateutil import relativedelta from pendulum.tz.timezone import FixedTimezone, Timezone -from airflow._shared.module_loading import import_string, qualname +from airflow._shared.module_loading import qualname from airflow._shared.timezones.timezone import from_timestamp, parse_timezone, utcnow from airflow.callbacks.callback_requests import DagCallbackRequest, TaskCallbackRequest from airflow.exceptions import AirflowException, DeserializationError, SerializationError @@ -242,6 +242,54 @@ def _decode_priority_weight_strategy(var: str) -> PriorityWeightStrategy: return priority_weight_strategy_class() +# Builtin exceptions the serializer emits as ``BASE_EXC_SER``. Only these are ever +# serialized (see the encode side), so deserialization resolves the stored name against +# this fixed map instead of importing it -- ``builtins.eval`` / ``builtins.exec`` and any +# other name are rejected without importing anything. +_DESERIALIZABLE_BUILTIN_EXCEPTIONS: dict[str, type[BaseException]] = { + "KeyError": KeyError, + "AttributeError": AttributeError, +} + + +def _iter_subclasses(cls: type) -> Iterator[type]: + """Yield every (transitive) subclass of ``cls``.""" + for sub in cls.__subclasses__(): + yield sub + yield from _iter_subclasses(sub) + + +@cache +def _serializable_airflow_exceptions() -> dict[str, type[AirflowException]]: + """ + Map ``"." -> AirflowException subclass``, used to resolve ``AIRFLOW_EXC_SER`` nodes. + + Built once, from the in-memory ``AirflowException`` subclass tree (never from the + attacker-controlled stored name), and never rebuilt -- a name absent from it is rejected, not + imported. ``airflow.exceptions`` is imported by this module, so every built-in Airflow exception + is registered by the time this is first called; exceptions defined later are not added. + """ + return { + f"{cls.__module__}.{cls.__name__}": cls + for cls in (AirflowException, *_iter_subclasses(AirflowException)) + } + + +def _resolve_airflow_exception(exc_cls_name: str) -> type[AirflowException]: + """ + Resolve a serialized ``AirflowException`` class name to the loaded class, without importing it. + + The name is matched against the once-built ``AirflowException`` subclass map, so deserializing a + stored DAG never runs the top-level code of a module named in the blob. A name that is not a + registered ``AirflowException`` subclass -- e.g. an attacker's ``subprocess.check_output`` -- is + rejected rather than imported. + """ + exc_cls = _serializable_airflow_exceptions().get(exc_cls_name) + if exc_cls is None: + raise DeserializationError(f"Refusing to deserialize unknown exception class {exc_cls_name!r}") + return exc_cls + + def _encode_start_trigger_args(var: StartTriggerArgs) -> dict[str, Any]: """Encode a StartTriggerArgs.""" @@ -664,9 +712,14 @@ def deserialize(cls, encoded_var: Any) -> Any: kwargs = deser["kwargs"] del deser if type_ == DAT.AIRFLOW_EXC_SER: - exc_cls = import_string(exc_cls_name) + exc_cls: type[BaseException] = _resolve_airflow_exception(exc_cls_name) else: - exc_cls = import_string(f"builtins.{exc_cls_name}") + builtin_exc_cls = _DESERIALIZABLE_BUILTIN_EXCEPTIONS.get(exc_cls_name) + if builtin_exc_cls is None: + raise DeserializationError( + f"Refusing to deserialize disallowed builtin exception {exc_cls_name!r}" + ) + exc_cls = builtin_exc_cls return exc_cls(*args, **kwargs) elif type_ == DAT.SET: return {cls.deserialize(v) for v in var} diff --git a/airflow-core/tests/unit/serialization/test_dag_serialization.py b/airflow-core/tests/unit/serialization/test_dag_serialization.py index 4708365e846e2..aeb8522a10934 100644 --- a/airflow-core/tests/unit/serialization/test_dag_serialization.py +++ b/airflow-core/tests/unit/serialization/test_dag_serialization.py @@ -2806,6 +2806,64 @@ def test_create_dagrun_accepts_partition_key_for_partitioned_at_runtime_dag(self dr = dag_maker.create_dagrun(partition_key="runtime-key") assert dr.partition_key == "runtime-key" + def test_airflow_exc_deserialization_rejects_unknown_class(self): + """An AIRFLOW_EXC_SER name that is not a loaded AirflowException subclass is rejected. + + The name is resolved against the in-memory subclass tree, so an attacker-controlled + ``subprocess.check_output`` is never imported. + """ + from airflow.exceptions import DeserializationError + from airflow.serialization.enums import DagAttributeTypes + + encoded = BaseSerialization._encode( + BaseSerialization.serialize( + {"exc_cls_name": "subprocess.check_output", "args": [], "kwargs": {}} + ), + type_=DagAttributeTypes.AIRFLOW_EXC_SER, + ) + with pytest.raises(DeserializationError, match="Refusing to deserialize unknown exception class"): + BaseSerialization.deserialize(encoded) + + def test_airflow_exc_deserialization_roundtrips_airflow_exception(self): + """A genuine AirflowException subclass round-trips via the registry, without importing.""" + from airflow.exceptions import AirflowException + + result = BaseSerialization.deserialize(BaseSerialization.serialize(AirflowException("boom"))) + assert isinstance(result, AirflowException) + assert result.args == ("boom",) + + def test_base_exc_deserialization_rejects_non_allowlisted_builtin(self): + """A BASE_EXC_SER name outside the {KeyError, AttributeError} the encoder emits is rejected.""" + from airflow.exceptions import DeserializationError + from airflow.serialization.enums import DagAttributeTypes + + # ``eval`` is the weaponisable case; ``ValueError`` is a harmless builtin the encoder + # never emits as BASE_EXC_SER -- both must be rejected. + for name in ("eval", "ValueError"): + encoded = BaseSerialization._encode( + BaseSerialization.serialize({"exc_cls_name": name, "args": ["1"], "kwargs": {}}), + type_=DagAttributeTypes.BASE_EXC_SER, + ) + with pytest.raises( + DeserializationError, match="Refusing to deserialize disallowed builtin exception" + ): + BaseSerialization.deserialize(encoded) + + def test_base_exc_deserialization_roundtrips_builtin_exception(self): + """The builtin exceptions the encoder emits (KeyError / AttributeError) still deserialize.""" + from airflow.serialization.enums import DagAttributeTypes + + for exc_type in (KeyError, AttributeError): + encoded = BaseSerialization._encode( + BaseSerialization.serialize( + {"exc_cls_name": exc_type.__name__, "args": ["boom"], "kwargs": {}} + ), + type_=DagAttributeTypes.BASE_EXC_SER, + ) + result = BaseSerialization.deserialize(encoded) + assert isinstance(result, exc_type) + assert result.args == ("boom",) + def test_kubernetes_optional(): """Test that serialization module loads without kubernetes, but deserialization of PODs requires it""" From 894ea7c5fdabe43c4d8a56e05b42694eae481688 Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Tue, 4 Aug 2026 00:31:33 +0200 Subject: [PATCH 2/3] Resolve serialized exception names from loaded modules instead of a prebuilt map A prebuilt map keyed on each class's __module__ cannot follow the re-exports that import_string used to follow. These exceptions moved to airflow.sdk.exceptions in 3.2.0, so every blob written by 3.0/3.1 names airflow.exceptions. and would stop deserializing on upgrade. Reading the name out of an already-loaded module keeps that working, needs no cache to go stale when a provider or plugin registers its own subclass late, and still refuses to import anything the stored blob names. --- .../serialization/serialized_objects.py | 59 ++++----- .../serialization/test_dag_serialization.py | 124 ++++++++++++++---- 2 files changed, 119 insertions(+), 64 deletions(-) diff --git a/airflow-core/src/airflow/serialization/serialized_objects.py b/airflow-core/src/airflow/serialization/serialized_objects.py index 5aec9f666c038..3533d51b97377 100644 --- a/airflow-core/src/airflow/serialization/serialized_objects.py +++ b/airflow-core/src/airflow/serialization/serialized_objects.py @@ -29,7 +29,7 @@ import math import sys import weakref -from collections.abc import Collection, Iterable, Iterator, Mapping +from collections.abc import Collection, Iterable, Mapping from functools import cache, cached_property, lru_cache from inspect import signature from textwrap import dedent @@ -242,50 +242,39 @@ def _decode_priority_weight_strategy(var: str) -> PriorityWeightStrategy: return priority_weight_strategy_class() -# Builtin exceptions the serializer emits as ``BASE_EXC_SER``. Only these are ever -# serialized (see the encode side), so deserialization resolves the stored name against -# this fixed map instead of importing it -- ``builtins.eval`` / ``builtins.exec`` and any -# other name are rejected without importing anything. +# Builtin exceptions a ``BASE_EXC_SER`` node can be rebuilt into. The encode side matches +# ``KeyError`` / ``AttributeError`` *and their subclasses* while storing the concrete class name, +# so a user-defined subclass serializes to a name that is absent here and cannot be rebuilt -- +# which was equally true when the name was imported, since ``builtins`` does not hold it either. +# Resolving against this map keeps ``builtins.eval`` / ``builtins.exec`` out without importing. _DESERIALIZABLE_BUILTIN_EXCEPTIONS: dict[str, type[BaseException]] = { "KeyError": KeyError, "AttributeError": AttributeError, } -def _iter_subclasses(cls: type) -> Iterator[type]: - """Yield every (transitive) subclass of ``cls``.""" - for sub in cls.__subclasses__(): - yield sub - yield from _iter_subclasses(sub) - - -@cache -def _serializable_airflow_exceptions() -> dict[str, type[AirflowException]]: - """ - Map ``"." -> AirflowException subclass``, used to resolve ``AIRFLOW_EXC_SER`` nodes. - - Built once, from the in-memory ``AirflowException`` subclass tree (never from the - attacker-controlled stored name), and never rebuilt -- a name absent from it is rejected, not - imported. ``airflow.exceptions`` is imported by this module, so every built-in Airflow exception - is registered by the time this is first called; exceptions defined later are not added. - """ - return { - f"{cls.__module__}.{cls.__name__}": cls - for cls in (AirflowException, *_iter_subclasses(AirflowException)) - } - - def _resolve_airflow_exception(exc_cls_name: str) -> type[AirflowException]: """ Resolve a serialized ``AirflowException`` class name to the loaded class, without importing it. - The name is matched against the once-built ``AirflowException`` subclass map, so deserializing a - stored DAG never runs the top-level code of a module named in the blob. A name that is not a - registered ``AirflowException`` subclass -- e.g. an attacker's ``subprocess.check_output`` -- is - rejected rather than imported. + The module part is looked up in ``sys.modules`` and the class is read out of that module's + namespace, so a name in the stored blob can never cause an import: a module that is not already + loaded simply fails to resolve. The result must be an ``AirflowException`` subclass, so an + attacker's ``subprocess.check_output`` is rejected even when ``subprocess`` is loaded. + + The namespace is read directly rather than through ``getattr`` so that a module-level + ``__getattr__`` -- which Airflow uses for deprecation shims and lazy provider re-exports -- stays + out of the path, since those hooks do import on access. + + Resolving the name instead of matching it against a prebuilt map is also what keeps blobs + written by older versions readable: these exceptions moved to ``airflow.sdk.exceptions`` in + 3.2.0 and are re-exported from ``airflow.exceptions``, so a 3.0/3.1 blob naming the old module + still resolves, exactly as it did when the name was imported. """ - exc_cls = _serializable_airflow_exceptions().get(exc_cls_name) - if exc_cls is None: + module_name, _, attr_name = exc_cls_name.rpartition(".") + module = sys.modules.get(module_name) + exc_cls = vars(module).get(attr_name) if module is not None else None + if not (isinstance(exc_cls, type) and issubclass(exc_cls, AirflowException)): raise DeserializationError(f"Refusing to deserialize unknown exception class {exc_cls_name!r}") return exc_cls @@ -717,7 +706,7 @@ def deserialize(cls, encoded_var: Any) -> Any: builtin_exc_cls = _DESERIALIZABLE_BUILTIN_EXCEPTIONS.get(exc_cls_name) if builtin_exc_cls is None: raise DeserializationError( - f"Refusing to deserialize disallowed builtin exception {exc_cls_name!r}" + f"Refusing to deserialize unsupported builtin exception {exc_cls_name!r}" ) exc_cls = builtin_exc_cls return exc_cls(*args, **kwargs) diff --git a/airflow-core/tests/unit/serialization/test_dag_serialization.py b/airflow-core/tests/unit/serialization/test_dag_serialization.py index aeb8522a10934..f42292cf3d740 100644 --- a/airflow-core/tests/unit/serialization/test_dag_serialization.py +++ b/airflow-core/tests/unit/serialization/test_dag_serialization.py @@ -52,6 +52,7 @@ from airflow.dag_processing.dagbag import DagBag from airflow.exceptions import ( AirflowException, + DeserializationError, ParamValidationError, SerializationError, ) @@ -77,7 +78,7 @@ from airflow.serialization.definitions.param import SerializedParam from airflow.serialization.definitions.xcom_arg import SchedulerPlainXComArg from airflow.serialization.encoders import ensure_serialized_asset -from airflow.serialization.enums import Encoding +from airflow.serialization.enums import DagAttributeTypes, Encoding from airflow.serialization.json_schema import load_dag_schema_dict from airflow.serialization.serialized_objects import ( BaseSerialization, @@ -519,6 +520,14 @@ def timetable_plugin(monkeypatch: pytest.MonkeyPatch): ) +class PluginContributedError(AirflowException): + """Stands in for an AirflowException subclass contributed by a provider or a plugin. + + It lives outside ``airflow.exceptions`` on purpose: resolution must not be limited to the + exceptions Airflow itself ships. + """ + + class TestStringifiedDAGs: """Unit tests for stringified DAGs.""" @@ -2806,37 +2815,98 @@ def test_create_dagrun_accepts_partition_key_for_partitioned_at_runtime_dag(self dr = dag_maker.create_dagrun(partition_key="runtime-key") assert dr.partition_key == "runtime-key" - def test_airflow_exc_deserialization_rejects_unknown_class(self): - """An AIRFLOW_EXC_SER name that is not a loaded AirflowException subclass is rejected. + @pytest.mark.parametrize( + ("module_name", "attr_name"), + [ + pytest.param("subprocess", "check_output", id="callable_in_loaded_module"), + pytest.param("os", "system", id="another_callable"), + pytest.param("builtins", "eval", id="builtin_callable"), + pytest.param("airflow.exceptions", "NoSuchThing", id="missing_attr"), + ], + ) + def test_airflow_exc_deserialization_rejects_a_name_in_a_loaded_module(self, module_name, attr_name): + """Having the module loaded is not enough -- the name must be an AirflowException subclass. - The name is resolved against the in-memory subclass tree, so an attacker-controlled - ``subprocess.check_output`` is never imported. + The module is imported by the test first, so the rejection cannot be an artefact of the + module simply being absent. """ - from airflow.exceptions import DeserializationError - from airflow.serialization.enums import DagAttributeTypes - + importlib.import_module(module_name) encoded = BaseSerialization._encode( BaseSerialization.serialize( - {"exc_cls_name": "subprocess.check_output", "args": [], "kwargs": {}} + {"exc_cls_name": f"{module_name}.{attr_name}", "args": [], "kwargs": {}} ), type_=DagAttributeTypes.AIRFLOW_EXC_SER, ) with pytest.raises(DeserializationError, match="Refusing to deserialize unknown exception class"): BaseSerialization.deserialize(encoded) - def test_airflow_exc_deserialization_roundtrips_airflow_exception(self): - """A genuine AirflowException subclass round-trips via the registry, without importing.""" - from airflow.exceptions import AirflowException + @pytest.mark.parametrize( + "exc_cls_name", + [ + pytest.param("not.a.loaded.module.Thing", id="unloaded_module"), + pytest.param("NoModulePart", id="no_module_part"), + pytest.param("", id="empty"), + ], + ) + def test_airflow_exc_deserialization_rejects_an_unresolvable_name(self, exc_cls_name): + encoded = BaseSerialization._encode( + BaseSerialization.serialize({"exc_cls_name": exc_cls_name, "args": [], "kwargs": {}}), + type_=DagAttributeTypes.AIRFLOW_EXC_SER, + ) + with pytest.raises(DeserializationError, match="Refusing to deserialize unknown exception class"): + BaseSerialization.deserialize(encoded) + def test_airflow_exc_deserialization_does_not_import_the_named_module(self): + """Resolution reads ``sys.modules``; it never imports what the blob names.""" + module_name = "airflow_exc_module_that_must_not_be_imported" + encoded = BaseSerialization._encode( + BaseSerialization.serialize({"exc_cls_name": f"{module_name}.Boom", "args": [], "kwargs": {}}), + type_=DagAttributeTypes.AIRFLOW_EXC_SER, + ) + with mock.patch.object( + importlib, "import_module", side_effect=AssertionError("imported"), autospec=True + ): + with pytest.raises(DeserializationError): + BaseSerialization.deserialize(encoded) + assert module_name not in sys.modules + + def test_airflow_exc_deserialization_roundtrips_airflow_exception(self): + """A genuine AirflowException subclass round-trips.""" result = BaseSerialization.deserialize(BaseSerialization.serialize(AirflowException("boom"))) assert isinstance(result, AirflowException) assert result.args == ("boom",) + def test_airflow_exc_deserialization_resolves_a_subclass_outside_airflow(self): + """A subclass a provider or plugin contributes resolves as long as its module is loaded.""" + result = BaseSerialization.deserialize(BaseSerialization.serialize(PluginContributedError("boom"))) + assert isinstance(result, PluginContributedError) + assert result.args == ("boom",) + + @pytest.mark.parametrize( + "exc_cls_name", + [ + "airflow.exceptions.AirflowException", + "airflow.exceptions.AirflowNotFoundException", + "airflow.exceptions.ParamValidationError", + "airflow.sdk.exceptions.AirflowException", + ], + ) + def test_airflow_exc_deserialization_accepts_the_pre_3_2_module_spelling(self, exc_cls_name): + """A blob written before these exceptions moved to ``airflow.sdk.exceptions`` still reads. + + 3.0/3.1 stored ``airflow.exceptions.``; 3.2.0 moved the classes and left a re-export + behind. Both spellings have to resolve, or upgrading strands every stored blob that carries + an exception node. + """ + encoded = BaseSerialization._encode( + BaseSerialization.serialize({"exc_cls_name": exc_cls_name, "args": ["boom"], "kwargs": {}}), + type_=DagAttributeTypes.AIRFLOW_EXC_SER, + ) + result = BaseSerialization.deserialize(encoded) + assert isinstance(result, AirflowException) + def test_base_exc_deserialization_rejects_non_allowlisted_builtin(self): """A BASE_EXC_SER name outside the {KeyError, AttributeError} the encoder emits is rejected.""" - from airflow.exceptions import DeserializationError - from airflow.serialization.enums import DagAttributeTypes - # ``eval`` is the weaponisable case; ``ValueError`` is a harmless builtin the encoder # never emits as BASE_EXC_SER -- both must be rejected. for name in ("eval", "ValueError"): @@ -2845,24 +2915,20 @@ def test_base_exc_deserialization_rejects_non_allowlisted_builtin(self): type_=DagAttributeTypes.BASE_EXC_SER, ) with pytest.raises( - DeserializationError, match="Refusing to deserialize disallowed builtin exception" + DeserializationError, match="Refusing to deserialize unsupported builtin exception" ): BaseSerialization.deserialize(encoded) - def test_base_exc_deserialization_roundtrips_builtin_exception(self): - """The builtin exceptions the encoder emits (KeyError / AttributeError) still deserialize.""" - from airflow.serialization.enums import DagAttributeTypes + @pytest.mark.parametrize("exc_type", [KeyError, AttributeError]) + def test_base_exc_serialize_deserialize_round_trip(self, exc_type): + """Pins the allow-list to the encode branch, by going through ``serialize`` rather than + hand-building the node: if that branch ever accepts another builtin, this fails.""" + result = BaseSerialization.deserialize(BaseSerialization.serialize(exc_type("boom"))) - for exc_type in (KeyError, AttributeError): - encoded = BaseSerialization._encode( - BaseSerialization.serialize( - {"exc_cls_name": exc_type.__name__, "args": ["boom"], "kwargs": {}} - ), - type_=DagAttributeTypes.BASE_EXC_SER, - ) - result = BaseSerialization.deserialize(encoded) - assert isinstance(result, exc_type) - assert result.args == ("boom",) + assert isinstance(result, exc_type) + # The encode branch stores ``[var.args]``, so the args arrive nested. Pre-existing shape, + # asserted as it is rather than as it ought to be. + assert result.args == (("boom",),) def test_kubernetes_optional(): From 47bbe563e472eeacfe9c07c3b2ae83ba0d30932b Mon Sep 17 00:00:00 2001 From: Jarek Potiuk Date: Tue, 4 Aug 2026 12:45:31 +0200 Subject: [PATCH 3/3] Condense the docstrings around exception-name resolution Review feedback: the explanation was longer than the behaviour it describes. --- .../serialization/serialized_objects.py | 26 +++++-------------- .../serialization/test_dag_serialization.py | 9 ++----- 2 files changed, 9 insertions(+), 26 deletions(-) diff --git a/airflow-core/src/airflow/serialization/serialized_objects.py b/airflow-core/src/airflow/serialization/serialized_objects.py index 3533d51b97377..f42e51e28f7d0 100644 --- a/airflow-core/src/airflow/serialization/serialized_objects.py +++ b/airflow-core/src/airflow/serialization/serialized_objects.py @@ -242,11 +242,9 @@ def _decode_priority_weight_strategy(var: str) -> PriorityWeightStrategy: return priority_weight_strategy_class() -# Builtin exceptions a ``BASE_EXC_SER`` node can be rebuilt into. The encode side matches -# ``KeyError`` / ``AttributeError`` *and their subclasses* while storing the concrete class name, -# so a user-defined subclass serializes to a name that is absent here and cannot be rebuilt -- -# which was equally true when the name was imported, since ``builtins`` does not hold it either. -# Resolving against this map keeps ``builtins.eval`` / ``builtins.exec`` out without importing. +# Builtin exceptions a BASE_EXC_SER node can rebuild into. A user defined subclass +# (e.g. a custom KeyError) serializes to a name absent here and won't round-trip -- +# unchanged from before, since builtins never held it either. _DESERIALIZABLE_BUILTIN_EXCEPTIONS: dict[str, type[BaseException]] = { "KeyError": KeyError, "AttributeError": AttributeError, @@ -255,21 +253,11 @@ def _decode_priority_weight_strategy(var: str) -> PriorityWeightStrategy: def _resolve_airflow_exception(exc_cls_name: str) -> type[AirflowException]: """ - Resolve a serialized ``AirflowException`` class name to the loaded class, without importing it. + Resolve a stored ``AirflowException`` name without importing it. - The module part is looked up in ``sys.modules`` and the class is read out of that module's - namespace, so a name in the stored blob can never cause an import: a module that is not already - loaded simply fails to resolve. The result must be an ``AirflowException`` subclass, so an - attacker's ``subprocess.check_output`` is rejected even when ``subprocess`` is loaded. - - The namespace is read directly rather than through ``getattr`` so that a module-level - ``__getattr__`` -- which Airflow uses for deprecation shims and lazy provider re-exports -- stays - out of the path, since those hooks do import on access. - - Resolving the name instead of matching it against a prebuilt map is also what keeps blobs - written by older versions readable: these exceptions moved to ``airflow.sdk.exceptions`` in - 3.2.0 and are re-exported from ``airflow.exceptions``, so a 3.0/3.1 blob naming the old module - still resolves, exactly as it did when the name was imported. + Read via ``vars()`` rather than ``getattr`` -- a module's deprecation-shim + ``__getattr__`` can still import on access -- which also lets pre-3.2.0 blobs + naming the old ``airflow.exceptions`` path keep resolving. """ module_name, _, attr_name = exc_cls_name.rpartition(".") module = sys.modules.get(module_name) diff --git a/airflow-core/tests/unit/serialization/test_dag_serialization.py b/airflow-core/tests/unit/serialization/test_dag_serialization.py index f42292cf3d740..7852c25dc5ee8 100644 --- a/airflow-core/tests/unit/serialization/test_dag_serialization.py +++ b/airflow-core/tests/unit/serialization/test_dag_serialization.py @@ -521,11 +521,7 @@ def timetable_plugin(monkeypatch: pytest.MonkeyPatch): class PluginContributedError(AirflowException): - """Stands in for an AirflowException subclass contributed by a provider or a plugin. - - It lives outside ``airflow.exceptions`` on purpose: resolution must not be limited to the - exceptions Airflow itself ships. - """ + """Defined outside airflow.exceptions on purpose: resolution must not be limited to exceptions Airflow itself ships.""" class TestStringifiedDAGs: @@ -2871,13 +2867,12 @@ def test_airflow_exc_deserialization_does_not_import_the_named_module(self): assert module_name not in sys.modules def test_airflow_exc_deserialization_roundtrips_airflow_exception(self): - """A genuine AirflowException subclass round-trips.""" result = BaseSerialization.deserialize(BaseSerialization.serialize(AirflowException("boom"))) assert isinstance(result, AirflowException) assert result.args == ("boom",) def test_airflow_exc_deserialization_resolves_a_subclass_outside_airflow(self): - """A subclass a provider or plugin contributes resolves as long as its module is loaded.""" + """Resolves as long as its module is loaded, registration order doesn't matter.""" result = BaseSerialization.deserialize(BaseSerialization.serialize(PluginContributedError("boom"))) assert isinstance(result, PluginContributedError) assert result.args == ("boom",)