fix: guard RESP parser against deeply nested aggregate replies

The pure-Python RESP2 and RESP3 parsers recursively process nested
aggregate replies (arrays, sets, maps, push responses) without any
depth bound.  A malicious or misbehaving server can craft deeply
nested replies that exhaust the Python call stack, causing a
client-side RecursionError denial of service.

Add a _MAX_PARSE_DEPTH (256) guard to all four _read_response methods
(sync/async x RESP2/RESP3).  When the nesting depth exceeds the
limit, raise InvalidResponse with a clear message instead of letting
Python hit its recursion limit.

Fixes #4116

Author: michaelxer

redis/_parsers/resp2.py

@@ -6,6 +6,11 @@
 from .base import _AsyncRESPBase, _RESPBase
 from .socket import SERVER_CLOSED_CONNECTION_ERROR
 
+# Maximum nesting depth for aggregate RESP types (arrays, sets, maps).
+# A malicious or misbehaving server can craft deeply nested replies that
+# exhaust the Python call stack.  Guard against this by bounding recursion.
+_MAX_PARSE_DEPTH = 256
+
 
 class _RESP2Parser(_RESPBase):
     """RESP2 protocol implementation"""
@@ -27,8 +32,15 @@ def read_response(
             return result
 
     def _read_response(
-        self, disable_decoding=False, timeout: Union[float, object] = SENTINEL
+        self,
+        disable_decoding=False,
+        timeout: Union[float, object] = SENTINEL,
+        _depth: int = 0,
     ):
+        if _depth > _MAX_PARSE_DEPTH:
+            raise InvalidResponse(
+                f"Exceeded maximum response nesting depth ({_MAX_PARSE_DEPTH})"
+            )
         raw = self._buffer.readline(timeout=timeout)
         if not raw:
             raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
@@ -64,7 +76,11 @@ def _read_response(
             return None
         elif byte == b"*":
             response = [
-                self._read_response(disable_decoding=disable_decoding, timeout=timeout)
+                self._read_response(
+                    disable_decoding=disable_decoding,
+                    timeout=timeout,
+                    _depth=_depth + 1,
+                )
                 for i in range(int(response))
             ]
         else:
@@ -92,8 +108,14 @@ async def read_response(self, disable_decoding: bool = False):
         return response
 
     async def _read_response(
-        self, disable_decoding: bool = False
+        self,
+        disable_decoding: bool = False,
+        _depth: int = 0,
     ) -> Union[EncodableT, ResponseError, None]:
+        if _depth > _MAX_PARSE_DEPTH:
+            raise InvalidResponse(
+                f"Exceeded maximum response nesting depth ({_MAX_PARSE_DEPTH})"
+            )
         raw = await self._readline()
         response: Any
         byte, response = raw[:1], raw[1:]
@@ -128,7 +150,7 @@ async def _read_response(
             return None
         elif byte == b"*":
             response = [
-                (await self._read_response(disable_decoding))
+                (await self._read_response(disable_decoding, _depth=_depth + 1))
                 for _ in range(int(response))  # noqa
             ]
         else:

redis/_parsers/resp3.py

@@ -12,6 +12,11 @@
 )
 from .socket import SERVER_CLOSED_CONNECTION_ERROR
 
+# Maximum nesting depth for aggregate RESP types (arrays, sets, maps).
+# A malicious or misbehaving server can craft deeply nested replies that
+# exhaust the Python call stack.  Guard against this by bounding recursion.
+_MAX_PARSE_DEPTH = 256
+
 
 class _RESP3Parser(_RESPBase, PushNotificationsParser):
     """RESP3 protocol implementation"""
@@ -61,7 +66,12 @@ def _read_response(
         disable_decoding=False,
         push_request=False,
         timeout: Union[float, object] = SENTINEL,
+        _depth: int = 0,
     ):
+        if _depth > _MAX_PARSE_DEPTH:
+            raise InvalidResponse(
+                f"Exceeded maximum response nesting depth ({_MAX_PARSE_DEPTH})"
+            )
         raw = self._buffer.readline(timeout=timeout)
         if not raw:
             raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
@@ -107,15 +117,23 @@ def _read_response(
         # array response
         elif byte == b"*":
             response = [
-                self._read_response(disable_decoding=disable_decoding, timeout=timeout)
+                self._read_response(
+                    disable_decoding=disable_decoding,
+                    timeout=timeout,
+                    _depth=_depth + 1,
+                )
                 for _ in range(int(response))
             ]
         # set response
         elif byte == b"~":
             # redis can return unhashable types (like dict) in a set,
             # so we return sets as list, all the time, for predictability
             response = [
-                self._read_response(disable_decoding=disable_decoding, timeout=timeout)
+                self._read_response(
+                    disable_decoding=disable_decoding,
+                    timeout=timeout,
+                    _depth=_depth + 1,
+                )
                 for _ in range(int(response))
             ]
         # map response
@@ -126,12 +144,15 @@ def _read_response(
             resp_dict = {}
             for _ in range(int(response)):
                 key = self._read_response(
-                    disable_decoding=disable_decoding, timeout=timeout
+                    disable_decoding=disable_decoding,
+                    timeout=timeout,
+                    _depth=_depth + 1,
                 )
                 resp_dict[key] = self._read_response(
                     disable_decoding=disable_decoding,
                     push_request=push_request,
                     timeout=timeout,
+                    _depth=_depth + 1,
                 )
             response = resp_dict
         # push response
@@ -141,6 +162,7 @@ def _read_response(
                     disable_decoding=disable_decoding,
                     push_request=push_request,
                     timeout=timeout,
+                    _depth=_depth + 1,
                 )
                 for _ in range(int(response))
             ]
@@ -153,6 +175,7 @@ def _read_response(
             return self._read_response(
                 disable_decoding=disable_decoding,
                 push_request=push_request,
+                _depth=_depth + 1,
             )
         else:
             raise InvalidResponse(f"Protocol Error: {raw!r}")
@@ -190,8 +213,15 @@ async def read_response(
         return response
 
     async def _read_response(
-        self, disable_decoding: bool = False, push_request: bool = False
+        self,
+        disable_decoding: bool = False,
+        push_request: bool = False,
+        _depth: int = 0,
     ) -> Union[EncodableT, ResponseError, None]:
+        if _depth > _MAX_PARSE_DEPTH:
+            raise InvalidResponse(
+                f"Exceeded maximum response nesting depth ({_MAX_PARSE_DEPTH})"
+            )
         if not self._stream or not self.encoder:
             raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
         raw = await self._readline()
@@ -241,15 +271,23 @@ async def _read_response(
         # array response
         elif byte == b"*":
             response = [
-                (await self._read_response(disable_decoding=disable_decoding))
+                (
+                    await self._read_response(
+                        disable_decoding=disable_decoding, _depth=_depth + 1
+                    )
+                )
                 for _ in range(int(response))
             ]
         # set response
         elif byte == b"~":
             # redis can return unhashable types (like dict) in a set,
             # so we always convert to a list, to have predictable return types
             response = [
-                (await self._read_response(disable_decoding=disable_decoding))
+                (
+                    await self._read_response(
+                        disable_decoding=disable_decoding, _depth=_depth + 1
+                    )
+                )
                 for _ in range(int(response))
             ]
         # map response
@@ -259,25 +297,33 @@ async def _read_response(
             # became defined to be left-right in version 3.8
             resp_dict = {}
             for _ in range(int(response)):
-                key = await self._read_response(disable_decoding=disable_decoding)
+                key = await self._read_response(
+                    disable_decoding=disable_decoding, _depth=_depth + 1
+                )
                 resp_dict[key] = await self._read_response(
-                    disable_decoding=disable_decoding, push_request=push_request
+                    disable_decoding=disable_decoding,
+                    push_request=push_request,
+                    _depth=_depth + 1,
                 )
             response = resp_dict
         # push response
         elif byte == b">":
             response = [
                 (
                     await self._read_response(
-                        disable_decoding=disable_decoding, push_request=push_request
+                        disable_decoding=disable_decoding,
+                        push_request=push_request,
+                        _depth=_depth + 1,
                     )
                 )
                 for _ in range(int(response))
             ]
             response = await self.handle_push_response(response)
             if not push_request:
                 return await self._read_response(
-                    disable_decoding=disable_decoding, push_request=push_request
+                    disable_decoding=disable_decoding,
+                    push_request=push_request,
+                    _depth=_depth + 1,
                 )
             else:
                 return response

tests/test_parsers/test_resp_depth_guard.py

@@ -0,0 +1,126 @@
+"""Tests for the RESP parser recursion depth guard (issue #4116).
+
+A malicious or misbehaving server can send deeply nested aggregate replies
+that exhaust the Python call stack.  The parsers now bound recursion depth
+via ``_MAX_PARSE_DEPTH`` to prevent this.
+"""
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from redis._parsers.resp2 import _MAX_PARSE_DEPTH as _MAX_DEPTH_RESP2
+from redis._parsers.resp2 import _RESP2Parser
+from redis._parsers.resp3 import _MAX_PARSE_DEPTH as _MAX_DEPTH_RESP3
+from redis._parsers.resp3 import _RESP3Parser
+from redis._parsers.socket import SocketBuffer
+from redis.exceptions import InvalidResponse
+
+
+class _FakeSocket:
+    """Minimal socket stand-in that yields pre-built RESP bytes."""
+
+    def __init__(self, payload: bytes):
+        self._payload = payload
+
+    def recv(self, n: int) -> bytes:
+        if not self._payload:
+            return b""
+        chunk = self._payload[:n]
+        self._payload = self._payload[n:]
+        return chunk
+
+    def settimeout(self, _timeout: float) -> None:
+        pass
+
+
+def _make_buf(payload: bytes) -> SocketBuffer:
+    """Create a SocketBuffer backed by a fake socket with the given payload."""
+    return SocketBuffer(
+        _FakeSocket(payload), socket_read_size=65536, socket_timeout=5.0
+    )
+
+
+def _make_parser(cls, payload: bytes):
+    """Create a parser with buffer and encoder wired up."""
+    parser = cls(65536)
+    parser._buffer = _make_buf(payload)
+    # Wire up a mock encoder so decode doesn't fail
+    parser.encoder = MagicMock()
+    parser.encoder.decode = lambda x: x.decode("utf-8") if isinstance(x, bytes) else x
+    return parser
+
+
+def _make_nested_arrays(depth: int) -> bytes:
+    """Build ``depth`` nested single-element RESP arrays terminating in +OK."""
+    return (b"*1\r\n" * depth) + b"+OK\r\n"
+
+
+class TestRESP2DepthGuard:
+    """Verify _RESP2Parser rejects excessively nested aggregate replies."""
+
+    def test_depth_exceeded_raises(self):
+        """Exceeding _MAX_PARSE_DEPTH raises InvalidResponse."""
+        depth = _MAX_DEPTH_RESP2 + 5
+        parser = _make_parser(_RESP2Parser, _make_nested_arrays(depth))
+
+        with pytest.raises(InvalidResponse, match="nesting depth"):
+            parser._read_response()
+
+    def test_depth_exactly_at_limit_raises(self):
+        """Responses nested exactly at the limit still raise."""
+        depth = _MAX_DEPTH_RESP2 + 1
+        parser = _make_parser(_RESP2Parser, _make_nested_arrays(depth))
+
+        with pytest.raises(InvalidResponse, match="nesting depth"):
+            parser._read_response()
+
+    def test_shallow_nesting_accepted(self):
+        """A shallow nested response (within limit) parses without error."""
+        depth = 10
+        parser = _make_parser(_RESP2Parser, _make_nested_arrays(depth))
+
+        result = parser._read_response()
+        for _ in range(depth):
+            assert isinstance(result, list)
+            result = result[0]
+        assert result == "OK"
+
+    def test_depth_guard_message_includes_limit(self):
+        """The error message includes the configured depth limit."""
+        depth = _MAX_DEPTH_RESP2 + 1
+        parser = _make_parser(_RESP2Parser, _make_nested_arrays(depth))
+
+        with pytest.raises(InvalidResponse, match=str(_MAX_DEPTH_RESP2)):
+            parser._read_response()
+
+
+class TestRESP3DepthGuard:
+    """Verify _RESP3Parser rejects excessively nested aggregate replies."""
+
+    def test_depth_exceeded_raises(self):
+        """Exceeding _MAX_PARSE_DEPTH raises InvalidResponse."""
+        depth = _MAX_DEPTH_RESP3 + 5
+        parser = _make_parser(_RESP3Parser, _make_nested_arrays(depth))
+
+        with pytest.raises(InvalidResponse, match="nesting depth"):
+            parser._read_response()
+
+    def test_shallow_nesting_accepted(self):
+        """A shallow nested response (within limit) parses without error."""
+        depth = 10
+        parser = _make_parser(_RESP3Parser, _make_nested_arrays(depth))
+
+        result = parser._read_response()
+        for _ in range(depth):
+            assert isinstance(result, list)
+            result = result[0]
+        assert result == "OK"
+
+    def test_depth_guard_message_includes_limit(self):
+        """The error message includes the configured depth limit."""
+        depth = _MAX_DEPTH_RESP3 + 1
+        parser = _make_parser(_RESP3Parser, _make_nested_arrays(depth))
+
+        with pytest.raises(InvalidResponse, match=str(_MAX_DEPTH_RESP3)):
+            parser._read_response()