mirror of
https://github.com/langflow-ai/langflow.git
synced 2026-07-23 13:55:17 +08:00
fix(security): HMAC-sign RedisCache payloads before dill deserialization (#13698)
* fix(security): HMAC-sign RedisCache payloads before dill deserialization RedisCache.get() called dill.loads() directly on bytes read from Redis. dill executes embedded reduce gadgets, so anyone able to write under the langflow:cache: namespace (a co-tenant on a shared Redis, an exposed/un-ACL'd port, etc.) could plant a payload that runs arbitrary code in the Langflow process on the next get() (CWE-502, RCE). Cache values are now prefixed with an HMAC-SHA256 tag derived from the server SECRET_KEY. get() verifies the tag with hmac.compare_digest before deserializing; unsigned/tampered/legacy entries are treated as a cache miss and never passed to dill.loads(). Only active when LANGFLOW_CACHE_TYPE=redis. * [autofix.ci] apply automated fixes * [autofix.ci] apply automated fixes (attempt 2/3) * Update src/backend/base/langflow/services/cache/service.py Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com> * fix: redis payload rejection * test(cache): cover short-payload integrity short-circuit; align settings deps import Address review feedback on the RedisCache HMAC integrity fix: - Import get_settings_service from langflow.services.deps to match the rest of services/ and keep the non-optional SettingsService return type (the lfx variant is SettingsServiceProtocol | None and the next line dereferences .auth_settings without a guard). - Add test_get_rejects_payload_shorter_than_tag covering the len(value) < _HMAC_DIGEST_SIZE short-circuit in get() (a 1-byte write under the namespace is the cheapest attacker input and was previously uncovered). --------- Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com> Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>
This commit is contained in:
Eric Hare
@ -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)
|
||||
|
||||
@ -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
|
||||
|
||||
Reference in New Issue
Block a user