feat: add HAYSTACK_UNSAFE_DESERIALIZATION env var

haystack/core/serialization_security.py

@@ -12,7 +12,14 @@
 - Process-wide programmatic API: :func:`allow_deserialization_module`
 - Environment variable: `HAYSTACK_DESERIALIZATION_ALLOWLIST="mypkg.*,otherpkg.*"`
 
-The two-mode loading API (`unsafe=True`) bypasses the allowlist entirely.
+The two-mode loading API (`unsafe=True`) bypasses the allowlist entirely. For deployments that only
+ever load fully trusted pipelines and cannot pass `unsafe=True` at every call site, the process-wide
+environment variable `HAYSTACK_UNSAFE_DESERIALIZATION=1` is equivalent to `unsafe=True` on every load:
+it disables *all* deserialization safety checks (the module allowlist, the builtin/import-primitive
+and control-plane denylists, the object-internals traversal guard, and the refusal to honor a
+component's own `unsafe: true` flag). Only enable it when every pipeline loaded by the process is
+trusted. Its value is read once, on the first deserialization in the process, and then frozen for the
+process lifetime, so nothing that runs later can turn the safety checks off (or back on).
 """
 
 import builtins
@@ -26,6 +33,7 @@
 from types import ModuleType
 from typing import TypeVar
 
+from haystack import logging
 from haystack.core.errors import DeserializationError
 
 # The default allowlist covers Haystack's own packages plus a small set of standard-library type modules
@@ -41,6 +49,12 @@
 )
 DESERIALIZATION_ALLOWLIST_ENV_VAR = "HAYSTACK_DESERIALIZATION_ALLOWLIST"
 
+# Process-wide "off switch": when set to a truthy value, deserialization behaves as if every load
+# were called with `unsafe=True` (see `_is_unsafe_deserialization`). Unlike the allowlist env var,
+# this disables *all* safety checks, so it must only be used when every pipeline the process loads
+# is trusted.
+UNSAFE_DESERIALIZATION_ENV_VAR = "HAYSTACK_UNSAFE_DESERIALIZATION"
+
 # `builtins` is on the default allowlist because deserialization legitimately needs builtin *types*
 # (e.g. `builtins.str`, used in serialized type annotations and as nested `{"type": ...}` class
 # references) and harmless builtin callables that Haystack's own serializer emits (e.g.
@@ -88,6 +102,8 @@
     {("haystack.utils.type_serialization", "thread_safe_import")}
 )
 
+logger = logging.getLogger(__name__)
+
 
 @dataclass(frozen=True)
 class _DeserializationContext:
@@ -105,16 +121,59 @@ def _get_context() -> _DeserializationContext:
     return ctx if ctx is not None else _DeserializationContext()
 
 
+# Snapshot of `UNSAFE_DESERIALIZATION_ENV_VAR`, read once and then frozen for the process lifetime
+# (`None` means "not read yet"). Freezing it is a security property, not an optimization: the first
+# read happens before any deserialized data can run, so a hostile pipeline cannot turn the safety
+# checks off *while it is being loaded*. Without it, any route from serialized data to `os.environ`
+# is a full bypass — e.g. an allowlisted module that binds `os.environ` at module scope, whose
+# `update` resolves because it is `collections.abc.MutableMapping.update` and `collections` is on
+# the default allowlist. The read is deliberately lazy rather than done at import time, so that a
+# `load_dotenv()` (or any other env setup) that runs before the first load is still honored.
+_unsafe_env_snapshot: bool | None = None
+
+
+def _unsafe_env_enabled() -> bool:
+    """
+    Return whether the process-wide unsafe-deserialization env var was set to a truthy value.
+
+    :data:`UNSAFE_DESERIALIZATION_ENV_VAR` is read on the first deserialization check in the process
+    and the result is then frozen for the process lifetime (see `_unsafe_env_snapshot`): later writes
+    to the variable are ignored, in either direction. A warning is logged when the snapshot is taken
+    and found active, since it turns off all deserialization safety for the whole process.
+    """
+    global _unsafe_env_snapshot
+    snapshot = _unsafe_env_snapshot
+    if snapshot is None:
+        # A benign race: concurrent first readers compute the same value from the same environment.
+        snapshot = os.environ.get(UNSAFE_DESERIALIZATION_ENV_VAR, "").strip().lower() in ("1", "true")
+        _unsafe_env_snapshot = snapshot
+        if snapshot:
+            logger.warning(
+                "{env} is set: pipeline deserialization safety is DISABLED process-wide "
+                "(equivalent to passing unsafe=True on every load). Only enable this if every "
+                "pipeline loaded by this process is fully trusted.",
+                env=UNSAFE_DESERIALIZATION_ENV_VAR,
+            )
+    return snapshot
+
+
 def _is_unsafe_deserialization() -> bool:
     """
-    Return whether the active deserialization context was entered with `unsafe=True`.
+    Return whether deserialization is running in unsafe mode.
+
+    This is the single source of truth consulted by every safety check in this module. It is `True`
+    when either the active deserialization context was entered with `unsafe=True`, or the process-wide
+    :data:`UNSAFE_DESERIALIZATION_ENV_VAR` env var was set when it was first read (which disables all
+    deserialization safety checks for the whole process — see :func:`_unsafe_env_enabled`).
 
     Components deserializing their own data (e.g. `OutputAdapter.from_dict`) use this to decide
     whether to honor an embedded `unsafe` flag: a serialized component may only disable its Jinja
     sandbox when the whole pipeline is being loaded in unsafe mode (`Pipeline.load(..., unsafe=True)`),
     never on its own from otherwise-untrusted data in default safe mode.
     """
-    return _get_context().unsafe
+    # `_unsafe_env_enabled()` first so the env snapshot is always taken on the earliest
+    # check in the process, even when that check happens inside an `unsafe=True` load.
+    return _unsafe_env_enabled() or _get_context().unsafe
 
 
 _F = TypeVar("_F", bound=Callable[..., object])
@@ -195,7 +254,7 @@ def _check_not_deserialization_internal(resolved: object, handle: str) -> None:
     :raises DeserializationError:
         If `resolved` is part of the deserialization control plane.
     """
-    if _get_context().unsafe:
+    if _is_unsafe_deserialization():
         return
     if _is_deserialization_internal(resolved):
         name = getattr(resolved, "__qualname__", None) or getattr(resolved, "__name__", None) or repr(resolved)
@@ -256,7 +315,7 @@ def _check_traversable_attribute(name: str, handle: str) -> None:
     :raises DeserializationError:
         If `name` names an object-internals attribute.
     """
-    if _get_context().unsafe:
+    if _is_unsafe_deserialization():
         return
     if name.startswith("__") or name in _UNSAFE_TRAVERSAL_ATTRS:
         raise DeserializationError(
@@ -315,7 +374,7 @@ def _patterns_from_env() -> list[str]:
 def _is_module_allowed(module_name: str) -> bool:
     """Return whether `module_name` is on the active deserialization allowlist."""
     ctx = _get_context()
-    if ctx.unsafe:
+    if _is_unsafe_deserialization():
         return True
     patterns: list[str] = []
     patterns.extend(DEFAULT_ALLOWED_MODULES)
@@ -367,7 +426,7 @@ def _check_resolved_module_allowed(resolved: object, declared_module: str | None
     :raises DeserializationError:
         If the resolved object's real module is not on the allowlist.
     """
-    if _get_context().unsafe:
+    if _is_unsafe_deserialization():
         return
     # Builtins are gated separately and authoritatively — by the identity denylist
     # (`_check_not_denied_builtin`) in the callable path and by the type requirement
@@ -424,7 +483,7 @@ def _check_not_denied_builtin(resolved: object, handle: str) -> None:
     :param handle:
         The original serialized handle, used only for the error message.
     """
-    if _get_context().unsafe:
+    if _is_unsafe_deserialization():
         return
     if _is_denied_builtin(resolved):
         name = getattr(resolved, "__name__", str(resolved))
@@ -452,7 +511,7 @@ def _check_not_denied_callable(resolved: object, handle: str) -> None:
     :param handle:
         The original serialized handle, used only for the error message.
     """
-    if _get_context().unsafe:
+    if _is_unsafe_deserialization():
         return
     ident = (getattr(resolved, "__module__", ""), getattr(resolved, "__qualname__", ""))
     if any(resolved is denied for denied in _DENIED_CALLABLE_OBJECTS) or ident in _DENIED_CALLABLE_QUALNAMES:
@@ -478,7 +537,7 @@ def _check_builtin_is_type(resolved: object, handle: str) -> None:
     :param handle:
         The original serialized handle, used only for the error message.
     """
-    if _get_context().unsafe:
+    if _is_unsafe_deserialization():
         return
     if not isinstance(resolved, type):
         raise DeserializationError(

releasenotes/notes/unsafe-deserialization-env-var-422b5427954fb4bf.yaml

@@ -0,0 +1,25 @@
+---
+features:
+  - |
+    Added the ``HAYSTACK_UNSAFE_DESERIALIZATION`` environment variable as a process-wide equivalent of
+    loading with ``unsafe=True``. When set to a truthy value (``1`` or ``true``), every
+    ``Pipeline.load`` / ``Pipeline.loads`` / ``Pipeline.from_dict`` call skips all deserialization safety
+    checks — the module allowlist, the builtin/import-primitive and control-plane denylists, the
+    object-internals traversal guard, and the refusal to honor a component's own ``unsafe: true`` flag.
+    This is intended for deployments that only ever load fully trusted pipelines and cannot pass
+    ``unsafe=True`` at every call site. A warning is logged the first time it takes effect. Only enable
+    it when every pipeline the process loads is trusted: a single untrusted pipeline then leads to
+    arbitrary code execution.
+
+    The switch is not limited to pipeline loading: it disables the same checks for every
+    deserialization path in the process, including ones that take no ``unsafe`` argument of their own
+    — ``Tool.from_dict``, ``State.from_dict`` (agent snapshot resume), and the ``ConditionalRouter``
+    and ``OutputAdapter`` Jinja sandbox flags. A deployment that loads trusted pipelines at startup
+    but accepts serialized tools or agent state at request time is therefore exposed on the request
+    path as well, not only at load time.
+
+    The variable is read once, on the first deserialization in the process, and the result is then
+    frozen for the process lifetime: writes to it afterwards are ignored, in either direction. Set it
+    before the first pipeline is loaded. Freezing keeps the safety mode of a process from changing
+    under a caller's feet and, because the first read happens before any deserialized data can run,
+    stops a hostile pipeline from switching the checks off while it is being loaded.

test/core/test_serialization_security.py

@@ -5,7 +5,9 @@
 import functools
 import io
 import json
+import logging
 import operator
+import os
 import subprocess
 import types
 from collections.abc import Callable
@@ -14,6 +16,7 @@
 import pytest
 
 from haystack import component as component_module
+from haystack.core import serialization_security as ss
 from haystack.core.errors import DeserializationError
 from haystack.core.pipeline import Pipeline
 from haystack.core.serialization import (
@@ -25,16 +28,18 @@
 from haystack.core.serialization_security import (
     _DENIED_BUILTIN_NAMES,
     DESERIALIZATION_ALLOWLIST_ENV_VAR,
+    UNSAFE_DESERIALIZATION_ENV_VAR,
     _check_module_allowed,
     _current_context,
     _deserialization_context,
     _DeserializationContext,
     _extra_allowed_modules,
     _is_module_allowed,
+    _is_unsafe_deserialization,
     _module_matches,
 )
 from haystack.marshal import YamlMarshaller
-from haystack.utils import deserialize_callable
+from haystack.utils import deserialize_callable, type_serialization
 from haystack.utils.type_serialization import deserialize_type
 
 
@@ -47,6 +52,11 @@ def _reset_allowlist_state(monkeypatch):
     "untrusted" really means untrusted.
     """
     monkeypatch.delenv(DESERIALIZATION_ALLOWLIST_ENV_VAR, raising=False)
+    monkeypatch.delenv(UNSAFE_DESERIALIZATION_ENV_VAR, raising=False)
+    # Clear the frozen env-var snapshot so each test starts like a fresh process. No `raising=False`
+    # here on purpose: if the attribute is ever renamed, this must fail loudly rather than silently
+    # become a no-op and leak one test's mode into the next.
+    monkeypatch.setattr(ss, "_unsafe_env_snapshot", None)
     snapshot = list(_extra_allowed_modules)
     _extra_allowed_modules.clear()
     token = _current_context.set(_DeserializationContext())
@@ -177,6 +187,118 @@ def test_env_var_ignores_empty_entries(self, monkeypatch):
         assert _is_module_allowed("mypkg.sub")
 
 
+class TestUnsafeDeserializationEnvVar:
+    """
+    `HAYSTACK_UNSAFE_DESERIALIZATION` is a process-wide off switch: when truthy it makes every load
+    behave as if `unsafe=True` was passed, disabling all deserialization safety checks. It is a
+    separate axis from the module allowlist (`HAYSTACK_DESERIALIZATION_ALLOWLIST`), which only widens
+    which modules may be imported.
+    """
+
+    def test_unset_keeps_safe_mode(self):
+        # Sanity: with the env var unset, the guards are active.
+        assert not _is_module_allowed("subprocess")
+        with pytest.raises(DeserializationError):
+            deserialize_callable("os.system")
+
+    @pytest.mark.parametrize("value", ["1", "true", "TRUE", "True"])
+    def test_truthy_values_disable_all_checks(self, monkeypatch, value):
+        monkeypatch.setenv(UNSAFE_DESERIALIZATION_ENV_VAR, value)
+        # Allowlist bypassed ...
+        assert _is_module_allowed("subprocess")
+        assert deserialize_callable("os.system") is __import__("os").system
+        # ... and so are the denylists / control-plane / traversal guards.
+        assert callable(deserialize_callable("builtins.eval"))
+        assert callable(deserialize_callable("haystack.core.serialization_security.allow_deserialization_module"))
+        assert callable(
+            deserialize_callable("haystack.core.serialization_security.allow_deserialization_module.__globals__.get")
+        )
+
+    @pytest.mark.parametrize("value", ["", "0", "false", "no", "off", "yes", "on", "nope"])
+    def test_falsey_values_keep_safe_mode(self, monkeypatch, value):
+        monkeypatch.setenv(UNSAFE_DESERIALIZATION_ENV_VAR, value)
+        with pytest.raises(DeserializationError):
+            deserialize_callable("os.system")
+
+    def test_allows_unsafe_output_adapter_component(self, monkeypatch):
+        # A serialized OutputAdapter with `unsafe: true` loads under the env var, exactly as it would
+        # under `Pipeline.loads(..., unsafe=True)`.
+        yaml = (
+            "components:\n"
+            "  adapter:\n"
+            "    type: haystack.components.converters.output_adapter.OutputAdapter\n"
+            "    init_parameters:\n"
+            '      template: "{{ documents[0] }}"\n'
+            "      output_type: str\n"
+            "      unsafe: true\n"
+            "connections: []\n"
+        )
+        with pytest.raises(DeserializationError):
+            Pipeline.loads(yaml)  # refused in safe mode
+        # The safe-mode load above froze the snapshot, so simulate a fresh process before enabling
+        # the env var — a mid-process flip is ignored by design (see the freezing tests below).
+        monkeypatch.setenv(UNSAFE_DESERIALIZATION_ENV_VAR, "1")
+        monkeypatch.setattr(ss, "_unsafe_env_snapshot", None)
+        Pipeline.loads(yaml)  # accepted with the env var set
+
+    def test_warns_once_when_active(self, monkeypatch, caplog):
+        monkeypatch.setenv(UNSAFE_DESERIALIZATION_ENV_VAR, "1")
+        with caplog.at_level(logging.WARNING):
+            assert ss._unsafe_env_enabled() is True
+            assert ss._unsafe_env_enabled() is True
+        warnings = [r for r in caplog.records if UNSAFE_DESERIALIZATION_ENV_VAR in r.getMessage()]
+        assert len(warnings) == 1, "the safety-disabled warning must be logged exactly once per process"
+        assert "DISABLED" in warnings[0].getMessage()
+
+    def test_value_is_frozen_after_the_first_check(self, monkeypatch):
+        # The first check (here: a safe-mode resolution) snapshots the env var ...
+        with pytest.raises(DeserializationError):
+            deserialize_callable("os.system")
+        # ... so setting it afterwards has no effect for the rest of the process.
+        monkeypatch.setenv(UNSAFE_DESERIALIZATION_ENV_VAR, "1")
+        assert not _is_module_allowed("subprocess")
+        with pytest.raises(DeserializationError):
+            deserialize_callable("os.system")
+
+    def test_freezing_applies_in_both_directions(self, monkeypatch):
+        # Symmetrically, unsetting it after the snapshot does not re-arm the checks: the switch is
+        # decided once, so the mode of a process cannot change under a caller's feet mid-run.
+        monkeypatch.setenv(UNSAFE_DESERIALIZATION_ENV_VAR, "1")
+        assert _is_module_allowed("subprocess")
+        monkeypatch.delenv(UNSAFE_DESERIALIZATION_ENV_VAR)
+        assert _is_module_allowed("subprocess")
+
+    def test_env_write_reachable_from_serialized_data_cannot_disable_safety(self, monkeypatch):
+        """
+        Regression test for the reason the snapshot is frozen.
+
+        `os.environ.update` is `collections.abc.MutableMapping.update`, and `collections` is on the
+        default allowlist — so if any allowlisted module binds `os.environ` at module scope, a
+        serialized handle can resolve that mutator in *safe* mode and call it (e.g. as an
+        `OutputAdapter` Jinja `custom_filters` entry, which runs while the component is being
+        constructed). Were the env var read fresh on every check, that would switch the ongoing load
+        into unsafe mode and hand the pipeline arbitrary code execution.
+        """
+        monkeypatch.setattr(type_serialization, "environ", os.environ, raising=False)
+        # Resolving the gadget is itself a deserialization check, so the snapshot is already frozen
+        # by the time the attacker gets to call it — as it would be in a real load.
+        update = deserialize_callable("haystack.utils.type_serialization.environ.update")
+
+        # Register the variable before the write: the write below goes straight to the real
+        # environment, so monkeypatch has to know the pre-attack state to undo it at teardown.
+        # (Registering afterwards would record the polluted value and leak it into other tests.)
+        monkeypatch.setenv(UNSAFE_DESERIALIZATION_ENV_VAR, "")
+        update({UNSAFE_DESERIALIZATION_ENV_VAR: "1"})
+        assert os.environ[UNSAFE_DESERIALIZATION_ENV_VAR] == "1"  # the write lands ...
+
+        assert not _is_unsafe_deserialization()  # ... and changes nothing
+        assert not _is_module_allowed("subprocess")
+        with pytest.raises(DeserializationError):
+            deserialize_callable("os.system")
+        with pytest.raises(DeserializationError):
+            deserialize_callable("builtins.eval")
+
+
 class TestCheckModuleAllowed:
     def test_passes_silently_for_allowed_module(self):
         _check_module_allowed("haystack.foo")