amoghrajesh commented on code in PR #68511:
URL: https://github.com/apache/airflow/pull/68511#discussion_r3709922534
##########
airflow-core/src/airflow/serialization/serialized_objects.py:
##########
@@ -242,6 +242,43 @@ 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.
+_DESERIALIZABLE_BUILTIN_EXCEPTIONS: dict[str, type[BaseException]] = {
+ "KeyError": KeyError,
+ "AttributeError": AttributeError,
+}
+
+
+def _resolve_airflow_exception(exc_cls_name: str) -> type[AirflowException]:
+ """
+ Resolve a serialized ``AirflowException`` class name to the loaded class,
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.
+ """
Review Comment:
```suggestion
"""
Resolve a stored ``AirflowException`` name without importing it.
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.
"""
```
I think shorter but descriptive is better?
##########
airflow-core/src/airflow/serialization/serialized_objects.py:
##########
@@ -242,6 +242,43 @@ 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.
Review Comment:
```suggestion
# 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 wont
round-trip --
# unchanged from before, since builtins never held it either.
```
##########
airflow-core/tests/unit/serialization/test_dag_serialization.py:
##########
@@ -2806,6 +2815,121 @@ 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"
+ @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 module is imported by the test first, so the rejection cannot be
an artefact of the
+ module simply being absent.
+ """
+ importlib.import_module(module_name)
+ encoded = BaseSerialization._encode(
+ BaseSerialization.serialize(
+ {"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)
+
+ @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."""
Review Comment:
```suggestion
"""Resolves as long as its module is loaded, registration order
doesn't matter."""
```
##########
airflow-core/tests/unit/serialization/test_dag_serialization.py:
##########
@@ -2806,6 +2815,121 @@ 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"
+ @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 module is imported by the test first, so the rejection cannot be
an artefact of the
+ module simply being absent.
+ """
+ importlib.import_module(module_name)
+ encoded = BaseSerialization._encode(
+ BaseSerialization.serialize(
+ {"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)
+
+ @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."""
Review Comment:
```suggestion
```
##########
airflow-core/tests/unit/serialization/test_dag_serialization.py:
##########
@@ -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.
+ """
Review Comment:
```suggestion
"""Defined outside airflow.exceptions on purpose: resolution must not be
limited to exceptions Airflow itself ships."""
```
--
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]
