Restrict exception-node deserialization to BaseException subclasses (validate before import)

Stacked on apache#68528, which removed the dead AIP-44 BASE_TRIGGER path; the
remaining hardening here is for the exception nodes.

When a serialized DAG carries an exception node (AIRFLOW_EXC_SER /
BASE_EXC_SER), deserialization called import_string on the stored class
path and instantiated whatever it resolved to, without validating the
path or the resulting type.

Add _safe_import_for_deserialize, which validates the class path against a
trusted-namespace prefix list (airflow., plus builtins. for the
BASE_EXC_SER builtin case) before importing, and then confirms the
resolved object is a BaseException subclass. A path outside the trusted
namespaces is rejected without being imported.

Author: potiuk

airflow-core/src/airflow/serialization/serialized_objects.py

@@ -235,6 +235,33 @@ def _decode_priority_weight_strategy(var: str) -> PriorityWeightStrategy:
     return priority_weight_strategy_class()
 
 
+# Module prefixes whose code is trusted to import while deserializing a stored DAG.
+# The class path from the serialized blob is validated against these *before*
+# import_string runs, so a path outside the trusted namespaces is rejected without
+# importing it. Mirrors decode_timetable's prefix gate / the priority-weight registry.
+_TRUSTED_DESERIALIZE_PREFIXES = ("airflow.",)
+
+
+def _safe_import_for_deserialize(cls_name: str, base: type, *, allow_builtins: bool = False) -> type:
+    """
+    Resolve ``cls_name`` to a subclass of ``base``, validating the name before importing.
+
+    The check runs on the string before any import, so a class path outside the
+    trusted namespaces is never imported. A post-import ``issubclass`` check is
+    kept as a second gate.
+    """
+    module_path = cls_name.rpartition(".")[0]
+    trusted = cls_name.startswith(_TRUSTED_DESERIALIZE_PREFIXES) or (
+        allow_builtins and module_path == "builtins"
+    )
+    if not trusted:
+        raise ValueError(f"Refusing to deserialize disallowed class path {cls_name!r}")
+    cls = import_string(cls_name)
+    if not (isinstance(cls, type) and issubclass(cls, base)):
+        raise ValueError(f"{cls_name!r} is not a {base.__name__} subclass")
+    return cls
+
+
 def _encode_start_trigger_args(var: StartTriggerArgs) -> dict[str, Any]:
     """Encode a StartTriggerArgs."""
 
@@ -659,9 +686,11 @@ 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 = _safe_import_for_deserialize(exc_cls_name, BaseException)
             else:
-                exc_cls = import_string(f"builtins.{exc_cls_name}")
+                exc_cls = _safe_import_for_deserialize(
+                    f"builtins.{exc_cls_name}", BaseException, allow_builtins=True
+                )
             return exc_cls(*args, **kwargs)
         elif type_ == DAT.SET:
             return {cls.deserialize(v) for v in var}

airflow-core/tests/unit/serialization/test_dag_serialization.py

@@ -2786,6 +2786,42 @@ def test_create_dagrun_accepts_partition_key_for_partition_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_disallowed_class_path(self):
+        """An AIRFLOW_EXC_SER class path outside the trusted namespaces is rejected before import."""
+        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(ValueError, match="Refusing to deserialize disallowed class path"):
+            BaseSerialization.deserialize(encoded)
+
+    def test_base_exc_deserialization_rejects_non_exception(self):
+        """A builtins name that is not a BaseException subclass is rejected."""
+        from airflow.serialization.enums import DagAttributeTypes
+
+        encoded = BaseSerialization._encode(
+            BaseSerialization.serialize({"exc_cls_name": "eval", "args": ["1"], "kwargs": {}}),
+            type_=DagAttributeTypes.BASE_EXC_SER,
+        )
+        with pytest.raises(ValueError, match="not a BaseException subclass"):
+            BaseSerialization.deserialize(encoded)
+
+    def test_base_exc_deserialization_roundtrips_builtin_exception(self):
+        """A genuine builtin exception still deserializes."""
+        from airflow.serialization.enums import DagAttributeTypes
+
+        encoded = BaseSerialization._encode(
+            BaseSerialization.serialize({"exc_cls_name": "ValueError", "args": ["boom"], "kwargs": {}}),
+            type_=DagAttributeTypes.BASE_EXC_SER,
+        )
+        result = BaseSerialization.deserialize(encoded)
+        assert isinstance(result, ValueError)
+        assert result.args == ("boom",)
+
 
 def test_kubernetes_optional():
     """Test that serialization module loads without kubernetes, but deserialization of PODs requires it"""