Prevent issue in HTTPResponse().read() when decoded_content is True and then False
Provided it has initialized eligible decoder(decompressor) and did decode once
ADD
changelog/2800.bugfix.rst
+Prevented issue in HTTPResponse().read() when decoded_content is True and then False.
\ No newline at end of file

MODIFY
src/urllib3/response.py
src/urllib3/response.py
@@ -317,6 +317,7 @@ class BaseHTTPResponse(io.IOBase):
self.version = version
self.reason = reason
self.decode_content = decode_content
+ self._has_decoded_content = False
self._request_url: Optional[str] = request_url
self.retries = retries
@@ -436,11 +437,17 @@ class BaseHTTPResponse(io.IOBase):
Decode the data passed in and potentially flush the decoder.
"""
if not decode_content:
+ if self._has_decoded_content:
+ raise RuntimeError(
+ "Calling read(decode_content=False) is not supported after "
+ "read(decode_content=True) was called."
+ )
return data
try:
if self._decoder:
data = self._decoder.decompress(data)
+ self._has_decoded_content = True
except self.DECODER_ERROR_CLASSES as e:
content_encoding = self.headers.get("content-encoding", "").lower()
raise DecodeError(
@@ -891,6 +898,11 @@ class HTTPResponse(BaseHTTPResponse):
else:
# do not waste memory on buffer when not decoding
if not decode_content:
+ if self._has_decoded_content:
+ raise RuntimeError(
+ "Calling read(decode_content=False) is not supported after "
+ "read(decode_content=True) was called."
+ )
return data
decoded_data = self._decode(data, decode_content, flush_decoder)

MODIFY
test/test_response.py
test/test_response.py
@@ -565,6 +565,41 @@ class TestResponse:
with pytest.raises(ValueError, match="I/O operation on closed file.?"):
next(reader)
+ def test_read_with_illegal_mix_decode_toggle(self) -> None:
+ compress = zlib.compressobj(6, zlib.DEFLATED, -zlib.MAX_WBITS)
+ data = compress.compress(b"foo")
+ data += compress.flush()
+
+ fp = BytesIO(data)
+
+ resp = HTTPResponse(
+ fp, headers={"content-encoding": "deflate"}, preload_content=False
+ )
+
+ assert resp.read(1) == b"f"
+
+ with pytest.raises(
+ RuntimeError,
+ match=(
+ r"Calling read\(decode_content=False\) is not supported after "
+ r"read\(decode_content=True\) was called"
+ ),
+ ):
+ resp.read(1, decode_content=False)
+
+ def test_read_with_mix_decode_toggle(self) -> None:
+ compress = zlib.compressobj(6, zlib.DEFLATED, -zlib.MAX_WBITS)
+ data = compress.compress(b"foo")
+ data += compress.flush()
+
+ fp = BytesIO(data)
+
+ resp = HTTPResponse(
+ fp, headers={"content-encoding": "deflate"}, preload_content=False
+ )
+ resp.read(1, decode_content=False)
+ assert resp.read(1, decode_content=True) == b"o"
+
def test_streaming(self) -> None:
fp = BytesIO(b"foo")
resp = HTTPResponse(fp, preload_content=False)

Prevent issue in HTTPResponse().read() when decoded_content is True and then False
Provided it has initialized eligible decoder(decompressor) and did decode once