diff --git a/src/backend/base/langflow/services/cache/service.py b/src/backend/base/langflow/services/cache/service.py index c0240b970d..cb93bfe6bd 100644 --- a/src/backend/base/langflow/services/cache/service.py +++ b/src/backend/base/langflow/services/cache/service.py @@ -1,5 +1,7 @@ import asyncio import atexit +import hashlib +import hmac import os import pickle import tempfile @@ -232,6 +234,9 @@ class RedisCache(ExternalAsyncBaseCacheService, Generic[LockType]): KEY_PREFIX = "langflow:cache:" + # Size of the HMAC-SHA256 tag prepended to every stored payload. + _HMAC_DIGEST_SIZE = hashlib.sha256().digest_size + def __init__(self, host="localhost", port=6379, db=0, url=None, expiration_time=60 * 60) -> None: """Initialize a new RedisCache instance. @@ -252,11 +257,45 @@ class RedisCache(ExternalAsyncBaseCacheService, Generic[LockType]): else: self._client = StrictRedis(host=host, port=port, db=db) self.expiration_time = expiration_time + self._signing_key: bytes | None = None def _key(self, key) -> str: """Return the namespaced Redis key.""" return f"{self.KEY_PREFIX}{key}" + def _get_signing_key(self) -> bytes: + """Derive the HMAC key for cache payload integrity from the server secret. + + Bound to the same ``SECRET_KEY`` used elsewhere, so no extra config is + required. Cached after first use (the secret does not change at runtime). + """ + if self._signing_key is None: + from langflow.services.deps import get_settings_service + + secret = get_settings_service().auth_settings.SECRET_KEY.get_secret_value() + self._signing_key = hashlib.sha256(b"langflow-redis-cache-hmac:" + secret.encode()).digest() + return self._signing_key + + def _integrity_tag(self, namespaced_key: str, payload: bytes) -> bytes: + """Compute the HMAC-SHA256 tag binding ``payload`` to ``namespaced_key``. + + The Redis key is mixed in as associated authenticated data so a tag is + only valid for the exact key the payload was written under. Without this + binding, a payload signed for one key verifies under any other key, + letting anyone with write access to the ``langflow:cache:`` namespace + relocate/replay a validly-signed entry across keys (cross-key + substitution → type confusion / stale-value injection) without ever + knowing the secret. The key is length-prefixed so the (key, payload) + framing is unambiguous and bytes cannot be shifted across the boundary + while keeping a valid tag. + """ + mac = hmac.new(self._get_signing_key(), digestmod=hashlib.sha256) + key_bytes = namespaced_key.encode("utf-8") + mac.update(len(key_bytes).to_bytes(8, "big")) + mac.update(key_bytes) + mac.update(payload) + return mac.digest() + async def is_connected(self) -> bool: """Check if the Redis client is connected.""" import redis @@ -273,14 +312,36 @@ class RedisCache(ExternalAsyncBaseCacheService, Generic[LockType]): async def get(self, key, lock=None): if key is None: return CACHE_MISS - value = await self._client.get(self._key(key)) - return dill.loads(value) if value else CACHE_MISS + namespaced_key = self._key(key) + value = await self._client.get(namespaced_key) + if not value: + return CACHE_MISS + # Integrity check before deserializing. The Redis datastore is an + # untrusted boundary (a co-tenant on a shared Redis, an exposed/un-ACL'd + # port, or anyone able to write under the langflow:cache: namespace could + # plant a payload). dill.loads() executes embedded reduce gadgets, so we + # only deserialize bytes carrying a valid HMAC produced with the server + # secret. Unsigned/tampered/legacy entries are treated as a miss and are + # never passed to dill.loads (CWE-502). + if len(value) < self._HMAC_DIGEST_SIZE: + return CACHE_MISS + tag, payload = value[: self._HMAC_DIGEST_SIZE], value[self._HMAC_DIGEST_SIZE :] + expected = self._integrity_tag(namespaced_key, payload) + if not hmac.compare_digest(tag, expected): + await logger.awarning("RedisCache: discarding cache entry with an invalid integrity tag") + return CACHE_MISS + return dill.loads(payload) @override async def set(self, key, value, lock=None) -> None: try: if pickled := dill.dumps(value, recurse=True): - result = await self._client.setex(self._key(key), self.expiration_time, pickled) + # Prefix an HMAC tag so get() can reject tampered/forged payloads + # before deserialization (see get()). The tag is bound to the + # namespaced key so it cannot be replayed under a different key. + namespaced_key = self._key(key) + tag = self._integrity_tag(namespaced_key, pickled) + result = await self._client.setex(namespaced_key, self.expiration_time, tag + pickled) if not result: msg = "RedisCache could not set the value." raise ValueError(msg) diff --git a/src/backend/tests/unit/services/cache/test_redis_cache.py b/src/backend/tests/unit/services/cache/test_redis_cache.py index ded23ecaa2..e934845949 100644 --- a/src/backend/tests/unit/services/cache/test_redis_cache.py +++ b/src/backend/tests/unit/services/cache/test_redis_cache.py @@ -86,3 +86,123 @@ class TestRedisCacheTeardown: await cache.teardown() mock_client.aclose.assert_called_once() + + +_MARKER_PATH_HOLDER: list[str] = [] + + +def _deser_side_effect(path: str) -> str: + """Module-level callable used as a pickle reduce gadget in the test below.""" + _MARKER_PATH_HOLDER.append(path) + return path + + +class _Gadget: + """A picklable object whose deserialization would run _deser_side_effect.""" + + def __init__(self, path: str) -> None: + self.path = path + + def __reduce__(self): + return (_deser_side_effect, (self.path,)) + + +@pytest.mark.asyncio +class TestRedisCacheDeserializationIntegrity: + """Regression (insecure dill.loads of untrusted Redis bytes).""" + + async def test_get_rejects_payload_without_valid_hmac(self): + """A payload lacking a valid HMAC tag must not be deserialized (no gadget run).""" + import dill + from lfx.services.cache.utils import CACHE_MISS + + _MARKER_PATH_HOLDER.clear() + with patch("redis.asyncio.StrictRedis") as mock_redis_class: + mock_client = AsyncMock() + mock_redis_class.return_value = mock_client + cache = RedisCache(host="localhost", port=6379, db=0, expiration_time=3600) + cache._signing_key = b"k" * 32 # inject a fixed key (no settings dependency) + + # Attacker-written value: a reduce gadget, prefixed with a WRONG 32-byte tag. + forged = b"\x00" * cache._HMAC_DIGEST_SIZE + dill.dumps(_Gadget("gadget-ran")) + mock_client.get.return_value = forged + + result = await cache.get("k") + + assert result is CACHE_MISS # rejected before dill.loads + assert _MARKER_PATH_HOLDER == [] # gadget never executed + + async def test_get_rejects_payload_shorter_than_tag(self): + """A value too short to even hold a tag is a miss (cheapest attacker write).""" + from lfx.services.cache.utils import CACHE_MISS + + with patch("redis.asyncio.StrictRedis") as mock_redis_class: + mock_client = AsyncMock() + mock_redis_class.return_value = mock_client + cache = RedisCache(host="localhost", port=6379, db=0, expiration_time=3600) + cache._signing_key = b"k" * 32 + + # Fewer bytes than the HMAC tag length: rejected before any slicing/HMAC. + mock_client.get.return_value = b"short" + + assert await cache.get("k") is CACHE_MISS + + async def test_set_get_roundtrip_with_signature(self): + """Values written by set() carry a valid tag and round-trip through get().""" + with patch("redis.asyncio.StrictRedis") as mock_redis_class: + mock_client = AsyncMock() + mock_redis_class.return_value = mock_client + cache = RedisCache(host="localhost", port=6379, db=0, expiration_time=3600) + cache._signing_key = b"k" * 32 + + store: dict[str, bytes] = {} + + async def fake_setex(key, _ttl, value): + store[key] = value + return True + + async def fake_get(key): + return store.get(key) + + mock_client.setex.side_effect = fake_setex + mock_client.get.side_effect = fake_get + + await cache.set("k", {"a": 1, "b": [2, 3]}) + assert await cache.get("k") == {"a": 1, "b": [2, 3]} + + async def test_get_rejects_payload_replayed_under_different_key(self): + """A validly-signed entry must not verify when relocated to another key. + + The integrity tag is bound to the namespaced Redis key, so copying a + legitimately-signed payload from key ``a`` into the slot for key ``b`` + (cross-key substitution) is rejected as a miss instead of deserialized. + """ + from lfx.services.cache.utils import CACHE_MISS + + with patch("redis.asyncio.StrictRedis") as mock_redis_class: + mock_client = AsyncMock() + mock_redis_class.return_value = mock_client + cache = RedisCache(host="localhost", port=6379, db=0, expiration_time=3600) + cache._signing_key = b"k" * 32 + + store: dict[str, bytes] = {} + + async def fake_setex(key, _ttl, value): + store[key] = value + return True + + async def fake_get(key): + return store.get(key) + + mock_client.setex.side_effect = fake_setex + mock_client.get.side_effect = fake_get + + # Write a real, validly-signed entry under key "a". + await cache.set("a", {"secret": "for-a"}) + assert await cache.get("a") == {"secret": "for-a"} + + # Attacker relocates a's signed bytes into b's namespaced slot. + store[cache._key("b")] = store[cache._key("a")] + + # The tag was bound to "a"'s key, so it fails verification under "b". + assert await cache.get("b") is CACHE_MISS