fix!: require unsafe loading for serialized Jinja filters

haystack/components/converters/output_adapter.py

@@ -182,9 +182,16 @@ def from_dict(cls, data: dict[str, Any]) -> "OutputAdapter":
                 "If you trust the source of this data, load it with Pipeline.load(..., unsafe=True)."
             )
 
+        custom_filters = init_params.get("custom_filters", {})
+        if custom_filters and not _is_unsafe_deserialization():
+            raise DeserializationError(
+                "Refusing to deserialize an OutputAdapter with custom filters while loading in safe mode. "
+                "Custom filters are arbitrary callables that can execute during pipeline loading. "
+                "If you trust the source of this data, load it with Pipeline.load(..., unsafe=True)."
+            )
+
         init_params["output_type"] = deserialize_type(init_params["output_type"])
 
-        custom_filters = init_params.get("custom_filters", {})
         if custom_filters:
             init_params["custom_filters"] = {
                 name: deserialize_callable(filter_func) if filter_func else None

haystack/components/routers/conditional_router.py

@@ -375,6 +375,14 @@ def from_dict(cls, data: dict[str, Any]) -> "ConditionalRouter":
                 "If you trust the source of this data, load it with Pipeline.load(..., unsafe=True)."
             )
 
+        custom_filters = init_params.get("custom_filters", {})
+        if custom_filters and not _is_unsafe_deserialization():
+            raise DeserializationError(
+                "Refusing to deserialize a ConditionalRouter with custom filters while loading in safe mode. "
+                "Custom filters are arbitrary callables that can execute during pipeline loading. "
+                "If you trust the source of this data, load it with Pipeline.load(..., unsafe=True)."
+            )
+
         routes = init_params.get("routes")
         for route in routes:
             # output_type needs to be deserialized from a string to a type
@@ -385,7 +393,6 @@ def from_dict(cls, data: dict[str, Any]) -> "ConditionalRouter":
 
         # Since the custom_filters are typed as optional in the init signature, we catch the
         # case where they are not present in the serialized data and set them to an empty dict.
-        custom_filters = init_params.get("custom_filters", {})
         if custom_filters is not None:
             for name, filter_func in custom_filters.items():
                 init_params["custom_filters"][name] = deserialize_callable(filter_func) if filter_func else None

releasenotes/notes/Require-unsafe-mode-for-serialized-Jinja-custom-filters-d8d1cb6cd3a36f95.yaml

@@ -0,0 +1,11 @@
+---
+upgrade:
+  - |
+    Serialized ``OutputAdapter`` and ``ConditionalRouter`` components containing Jinja
+    ``custom_filters`` must now be loaded with ``Pipeline.load(..., unsafe=True)`` (or the
+    equivalent ``Pipeline.loads`` / ``Pipeline.from_dict`` option).
+security:
+  - |
+    Prevent arbitrary callables registered as serialized Jinja custom filters from executing while
+    a pipeline is loaded in safe mode. Jinja can invoke filters with constant arguments during
+    template compilation, and its sandbox does not apply callable-safety checks to filters.

test/components/converters/test_output_adapter.py

@@ -122,7 +122,8 @@ def test_sede_with_custom_filters(self):
             template="{{ documents[0].content|custom_filter }}", output_type=str, custom_filters=custom_filters
         )
         adapter_dict = adapter.to_dict()
-        deserialized_adapter = OutputAdapter.from_dict(adapter_dict)
+        with _deserialization_context(unsafe=True):
+            deserialized_adapter = OutputAdapter.from_dict(adapter_dict)
 
         assert adapter.template == deserialized_adapter.template
         assert adapter.output_type == deserialized_adapter.output_type
@@ -139,7 +140,8 @@ def test_sede_with_multiple_custom_filters(self):
             template="{{ documents[0].content|custom_filter }}", output_type=str, custom_filters=custom_filters
         )
         adapter_dict = adapter.to_dict()
-        deserialized_adapter = OutputAdapter.from_dict(adapter_dict)
+        with _deserialization_context(unsafe=True):
+            deserialized_adapter = OutputAdapter.from_dict(adapter_dict)
 
         assert adapter.template == deserialized_adapter.template
         assert adapter.output_type == deserialized_adapter.output_type
@@ -230,6 +232,28 @@ def test_from_dict_rejects_unsafe_in_safe_mode(self):
         with pytest.raises(DeserializationError, match="unsafe=True while loading in safe mode"):
             OutputAdapter.from_dict(data)
 
+    def test_from_dict_rejects_custom_filters_in_safe_mode(self):
+        adapter = OutputAdapter(
+            template="{{ value | custom_filter }}",
+            output_type=str,
+            custom_filters={"custom_filter": custom_filter_to_sede},
+        )
+
+        with pytest.raises(DeserializationError, match="custom filters while loading in safe mode"):
+            OutputAdapter.from_dict(adapter.to_dict())
+
+    def test_from_dict_allows_custom_filters_when_loading_unsafe(self):
+        adapter = OutputAdapter(
+            template="{{ value | custom_filter }}",
+            output_type=str,
+            custom_filters={"custom_filter": custom_filter_to_sede},
+        )
+
+        with _deserialization_context(unsafe=True):
+            deserialized_adapter = OutputAdapter.from_dict(adapter.to_dict())
+
+        assert deserialized_adapter.custom_filters == adapter.custom_filters
+
     def test_from_dict_allows_unsafe_when_loading_unsafe(self):
         # When the loader explicitly opts into unsafe mode, the embedded `unsafe=True` is honored.
         data = {

test/components/routers/test_conditional_router.py

@@ -396,7 +396,8 @@ def test_sede_with_custom_filter(self):
         result = router.run(**kwargs)
         assert result == {"test": 123}
         serialized_router = router.to_dict()
-        deserialized_router = ConditionalRouter.from_dict(serialized_router)
+        with _deserialization_context(unsafe=True):
+            deserialized_router = ConditionalRouter.from_dict(serialized_router)
         assert deserialized_router.custom_filters == router.custom_filters
         assert deserialized_router.custom_filters["custom_filter_to_sede"]("123-456-789") == 123
         assert result == deserialized_router.run(**kwargs)
@@ -443,6 +444,36 @@ def test_from_dict_allows_unsafe_when_loading_unsafe(self):
         assert router._unsafe
         assert isinstance(router._env, NativeEnvironment)
 
+    def test_from_dict_rejects_custom_filters_in_safe_mode(self):
+        routes: list[Route] = [
+            {
+                "condition": "{{ value | custom_filter_to_sede == 123 }}",
+                "output": "{{ value }}",
+                "output_type": str,
+                "output_name": "value",
+            }
+        ]
+        router = ConditionalRouter(routes, custom_filters={"custom_filter_to_sede": custom_filter_to_sede})
+
+        with pytest.raises(DeserializationError, match="custom filters while loading in safe mode"):
+            ConditionalRouter.from_dict(router.to_dict())
+
+    def test_from_dict_allows_custom_filters_when_loading_unsafe(self):
+        routes: list[Route] = [
+            {
+                "condition": "{{ value | custom_filter_to_sede == 123 }}",
+                "output": "{{ value }}",
+                "output_type": str,
+                "output_name": "value",
+            }
+        ]
+        router = ConditionalRouter(routes, custom_filters={"custom_filter_to_sede": custom_filter_to_sede})
+
+        with _deserialization_context(unsafe=True):
+            deserialized_router = ConditionalRouter.from_dict(router.to_dict())
+
+        assert deserialized_router.custom_filters == router.custom_filters
+
     def test_validate_output_type_without_unsafe(self):
         routes: list[Route] = [
             {
@@ -766,7 +797,8 @@ def test_sede_multiple_outputs(self):
         ]
 
         router = ConditionalRouter(routes, custom_filters={"get_area_code": custom_filter_to_sede})
-        reloaded_router = ConditionalRouter.from_dict(router.to_dict())
+        with _deserialization_context(unsafe=True):
+            reloaded_router = ConditionalRouter.from_dict(router.to_dict())
         assert reloaded_router.custom_filters == router.custom_filters
         assert reloaded_router.routes == router.routes
 

test/core/test_serialization_security.py

@@ -421,18 +421,20 @@ def _self_disable_yaml(self, component_type, extra_init):
         )
 
     def test_pipeline_loads_rejects_output_adapter_self_disable_vector(self):
-        # End-to-end reproduction of the reported RCE via OutputAdapter, in default safe mode.
+        # End-to-end reproduction of the reported RCE via OutputAdapter, in default safe mode. The
+        # broad serialized-filter gate rejects it before any control-plane callable is resolved.
         yaml = self._self_disable_yaml(
             "haystack.components.converters.output_adapter.OutputAdapter",
             "      template: \"{{ '*' | allow }}{{ 'os.system' | dc | mp(cmds) | list }}\"\n      output_type: str\n",
         )
         with mock.patch("os.system") as mocked_system:
-            with pytest.raises(DeserializationError, match="deserialization control plane"):
+            with pytest.raises(DeserializationError, match="custom filters while loading in safe mode"):
                 Pipeline.loads(yaml)
         assert not mocked_system.called
 
     def test_pipeline_loads_rejects_conditional_router_self_disable_vector(self):
-        # Same chain carried by ConditionalRouter; removing one component is not a mitigation.
+        # Same chain carried by ConditionalRouter; its serialized filters are rejected before the
+        # narrower control-plane callable checks need to run.
         extra = (
             "      routes:\n"
             "        - condition: \"{{ ['*'|allow, ('os.system'|dc|mp(cmds)|list)] and True }}\"\n"
@@ -442,7 +444,7 @@ def test_pipeline_loads_rejects_conditional_router_self_disable_vector(self):
         )
         yaml = self._self_disable_yaml("haystack.components.routers.conditional_router.ConditionalRouter", extra)
         with mock.patch("os.system") as mocked_system:
-            with pytest.raises(DeserializationError, match="deserialization control plane"):
+            with pytest.raises(DeserializationError, match="custom filters while loading in safe mode"):
                 Pipeline.loads(yaml)
         assert not mocked_system.called
 
@@ -505,7 +507,7 @@ def test_pipeline_loads_rejects_allowlist_poisoning_via_append(self):
             "      unsafe: false\n"
             "connections: []\n"
         )
-        with pytest.raises(DeserializationError, match="deserialization control plane"):
+        with pytest.raises(DeserializationError, match="custom filters while loading in safe mode"):
             Pipeline.loads(poison_yaml)
         assert _extra_allowed_modules == []
         with pytest.raises(DeserializationError):
@@ -586,8 +588,8 @@ def _nested_yaml(self):
 
     def test_pipeline_loads_rejects_nested_unsafe_load_run_vector(self):
         # End-to-end reproduction, in default safe mode. The top pipeline binds Pipeline.loads (L) and
-        # Pipeline.run (R) as custom_filters; the block fires while resolving L at load, before any
-        # template renders, so os.system is never reached and the allowlist is never poisoned.
+        # Pipeline.run (R) as custom_filters; the broad serialized-filter gate fires before resolving
+        # L, so os.system is never reached and the allowlist is never poisoned.
         top_yaml = (
             "components:\n"
             "  c:\n"
@@ -602,7 +604,7 @@ def test_pipeline_loads_rejects_nested_unsafe_load_run_vector(self):
             "connections: []\n"
         )
         with mock.patch("os.system") as mocked_system:
-            with pytest.raises(DeserializationError, match="deserialization control plane"):
+            with pytest.raises(DeserializationError, match="custom filters while loading in safe mode"):
                 Pipeline.loads(top_yaml)
         assert not mocked_system.called
         assert _extra_allowed_modules == []
@@ -760,7 +762,7 @@ def test_legitimate_attribute_walks_still_resolve(self):
 
     def test_pipeline_loads_rejects_globals_rewrite_vector(self):
         # End-to-end: rewriting `_extra_allowed_modules` via `__globals__.update` in default safe
-        # mode, without touching any blocked resolver or the protected state objects directly.
+        # mode. The serialized-filter gate rejects the handle before attribute traversal begins.
         poison_yaml = (
             "components:\n"
             "  adapter:\n"
@@ -773,7 +775,7 @@ def test_pipeline_loads_rejects_globals_rewrite_vector(self):
             "      unsafe: false\n"
             "connections: []\n"
         )
-        with pytest.raises(DeserializationError, match="internal attribute"):
+        with pytest.raises(DeserializationError, match="custom filters while loading in safe mode"):
             Pipeline.loads(poison_yaml)
         assert _extra_allowed_modules == []
         with pytest.raises(DeserializationError):