potiuk commented on code in PR #68511:
URL: https://github.com/apache/airflow/pull/68511#discussion_r3708132718
##########
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))
+ }
+
+
+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}")
Review Comment:
You are right, and this is the one that would have hurt. I confirmed it
locally: `AirflowException.__module__` is `airflow.sdk.exceptions`, while the
class is still present in `vars(airflow.exceptions)` via the re-export. So the
map only ever held the `airflow.sdk.exceptions.*` key, and every blob written
by 3.0/3.1 — which stores `airflow.exceptions.AirflowException`, because that
is what `__module__` was at the time — would have started raising on upgrade.
Rebuilding the cache never produces the old key, so the staleness fix in the
other thread does not touch this at all.
Switched to your `sys.modules` approach. The prebuilt map, the subclass walk
and the cache are all gone.
One change to the mechanics: it reads the class out of `vars(module)` rather
than with `getattr`. A module-level `__getattr__` (PEP 562) does run on
`getattr`, and Airflow uses those for deprecation shims and lazy provider
re-exports — some of them import on access, which would put an import back in
the path through a side door. A plain namespace lookup cannot trigger anything.
Tests now cover both spellings for four classes, a subclass defined outside
`airflow.*` (the provider/plugin case from the other thread), rejection of
`subprocess.check_output` / `os.system` / `builtins.eval` with the module
imported by the test first so the rejection cannot be an unloaded-module
artefact, and an assertion that resolution never imports the named module.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
##########
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
Review Comment:
Correct on both counts. `isinstance(var, (KeyError, AttributeError))`
matches subclasses and the branch stores `var.__class__.__name__`, so "only
these are ever serialized" was simply wrong — `MyKeyError` does get emitted,
and does not come back. As you say it did not come back before either, so
nothing regresses, but the comment and the message were describing a guarantee
that is not there.
Reworded both. The comment now says what the map actually is — the builtin
exceptions a node can be *rebuilt* into — and notes that a user-defined
subclass serializes to a name absent from it, exactly as it was absent from
`builtins` when the name was imported. The message is now "Refusing to
deserialize unsupported builtin exception", which is what it is; "disallowed"
read like a security decision about something that is really an unsupported
type.
Also took the point about the tests not pinning the map to the encode
branch. Added `test_base_exc_serialize_deserialize_round_trip`, which goes
through `BaseSerialization.serialize(KeyError("boom"))` rather than
hand-building the node, so if that tuple grows the test fails instead of
quietly passing.
Unrelated thing that surfaced while writing it: the encode branch stores
`"args": [var.args]`, so a round-tripped `KeyError("boom")` comes back as
`KeyError(("boom",))` — the args arrive nested one level deeper than they went
in. That is pre-existing on main and has nothing to do with this change, so I
have asserted the shape as it is rather than fixing it here.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
##########
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:
You were right that this was worse than "blocked at launch" — with the map
cached for the life of the process, a subclass registered later was blocked
permanently, not just early.
It is resolved, though not the way either of us proposed. The map is gone
entirely: the class is now looked up in `sys.modules` and read out of the
module's namespace, so there is nothing left to go stale and no cache to
invalidate. A provider or plugin that registers its own `AirflowException`
subclass resolves as soon as its module is loaded.
That route was picked because of the point raised in the other thread on
this file — keying on `cls.__module__` silently dropped the pre-3.2.0
`airflow.exceptions.*` spelling that older stored blobs carry, and rebuilding
the cache would not have brought that back. One change covers both.
`test_airflow_exc_deserialization_resolves_a_subclass_outside_airflow`
covers your case directly, using a subclass defined outside `airflow.*`.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
##########
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:
Moved to the top of the file — `DeserializationError` into the existing
`airflow.exceptions` block and `DagAttributeTypes` onto the existing
`airflow.serialization.enums` import.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
##########
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:
Moved to the top of the file — `DeserializationError` into the existing
`airflow.exceptions` block and `DagAttributeTypes` onto the existing
`airflow.serialization.enums` import.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
##########
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:
Moved to the top of the file — `DeserializationError` into the existing
`airflow.exceptions` block and `DagAttributeTypes` onto the existing
`airflow.serialization.enums` import.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
--
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]
