You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "amoghrajesh (via GitHub)" <gi...@apache.org> on 2026/08/03 05:17:12 UTC
Re: [PR] Restrict exception-node deserialization to known classes without importing the stored name [airflow]
amoghrajesh commented on code in PR #68511:
URL: https://github.com/apache/airflow/pull/68511#discussion_r3701474129
##########
airflow-core/src/airflow/serialization/serialized_objects.py:
##########
@@ -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 ``"<module>.<name>" -> 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))
+ }
Review Comment:
This never rebuilds the list, and it is quite common to have a plugin / provider subclass AirflowException lazily, this will not be able to deser them. This isn't just "blocked at launch" (per @uranusjr's comment above), it is blocked forever in that process due to the cache never invalidating.
Maybe we can have a two tier lookup? When a cache miss happens, clear `_serializable_airflow_exceptions.cache_clear()` and rebuild once before rejecting, so a class imported later in the lifecycle still resolves.
##########
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
Review Comment:
Top level import pls.
##########
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
Review Comment:
Top level import pls.
##########
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
Review Comment:
Top level import pls.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org