meta-openembedded/meta-python/recipes-devtools/python/python3-httplib2/CVE-2026-59939.patch

From 0000000000000000000000000000000000000000 Mon Sep 17 00:00:00 2001
From: Sergey Shepelev <temotor@gmail.com>
Date: Wed, 24 Jun 2026 12:34:50 +0300
Subject: [PATCH] decompression limited by size and ratio; require python 3.8+

CVE: CVE-2026-59939
Upstream-Status: Backport [https://github.com/httplib2/httplib2/commit/87581ad6cf752fe3da2090c59058261d2d00a427]

Backport Changes:
- Place httplib2 sources under python3/httplib2 to match the
  Scarthgap 0.22.0 source layout.
- Cap each zlib decompression call with max_length derived from the
  remaining hard and ratio allowance, requeue unconsumed_tail, and
  enforce both limits immediately after each returned output chunk.
- Reject empty gzip and deflate bodies through the existing
  FailedToDecompressContent contract.
- Add constrained-memory and empty encoded-body regression tests.
- Keep Scarthgap's Python version metadata; the upstream Python 3.8
  minimum is unrelated to the security fix.
- Omit .github/workflows updates because those CI files are absent
  from the extracted source and are not part of the security fix.

(cherry picked from commit 87581ad6cf752fe3da2090c59058261d2d00a427)
Signed-off-by: Darsh Kelaiya <dkelaiya@cisco.com>
---
 README.md                    |  37 ++++++--
 python3/httplib2/__init__.py |  71 +++++++++++---
 python3/httplib2/decode.py   | 209 ++++++++++++++++++++++++++++++++++++++++
 setup.cfg                    |   2 +-
 tests/__init__.py            |   8 ++
 tests/test_encoding.py       | 221 +++++++++++++++++++++++++++++++++++++++++--
 6 files changed, 517 insertions(+), 31 deletions(-)

diff --git a/README.md b/README.md
index 6193699..99a0d3b 100644
--- a/README.md
+++ b/README.md
@@ -10,13 +10,13 @@ If you want to help this project by bug report or code change, [contribution gui
 
 HTTPS support is only available if the socket module was
 compiled with SSL support.
-    
+
 ### Keep-Alive
 
 Supports HTTP 1.1 Keep-Alive, keeping the socket open and
 performing multiple requests over the same connection if
 possible.
-    
+
 ### Authentication
 
 The following three types of HTTP Authentication are
@@ -31,26 +31,26 @@ supported. These can be used over both HTTP and HTTPS.
 The module can optionally operate with a private cache that
 understands the Cache-Control: header and uses both the ETag
 and Last-Modified cache validators.
-    
+
 ### All Methods
 
 The module can handle any HTTP request method, not just GET
 and POST.
-    
+
 ### Redirects
 
 Automatically follows 3XX redirects on GETs.
-    
+
 ### Compression
 
 Handles both 'deflate' and 'gzip' types of compression.
-    
+
 ### Lost update support
 
 Automatically adds back ETags into PUT requests to resources
 we have already cached. This implements Section 3.2 of
 Detecting the Lost Update Problem Using Unreserved Checkout.
-    
+
 ### Unit Tested
 
 A large and growing set of unit tests.
@@ -113,3 +113,26 @@ More example usage can be found at:
 
  * https://github.com/httplib2/httplib2/wiki/Examples
  * https://github.com/httplib2/httplib2/wiki/Examples-Python3
+
+
+### Decompression Limits
+
+To mitigate denial-of-service risks from maliciously crafted compressed responses, the library enforces configurable limits during decompression. Limits are checked in fixed order: **hard limit** → **safe limit** → **ratio**.
+
+- **hard limit**
+  Absolute maximum decompressed output size (bytes). Exceeding it raises `DecodeLimitError`. Default: `10 GiB`.
+- **safe limit**
+  Output size below which the ratio check is skipped (avoids false positives on small payloads). Default: `10 MiB`.
+- **ratio**
+  Maximum allowed inflation factor (`output_bytes ÷ consumed_input_bytes`). Once output exceeds `safe_limit`, the ratio is enforced. Default: `100`.
+- **chunk size**
+  Internal processing chunk size in bytes (affects granularity of limit checks). Default: `65536` (64 KiB).
+
+Configuration priority (highest first):
+1. `Http()` constructor arguments: `decode_limit_hard`, `decode_limit_safe`, `decode_limit_ratio`, `decode_limit_chunk`
+2. Environment variables: `httplib2_decode_limit_hard`, `httplib2_decode_limit_safe`, `httplib2_decode_limit_ratio`, `httplib2_decode_limit_chunk` (case-insensitive, uppercase also accepted)
+3. Library defaults (listed above)
+
+Example:
+```python
+h = Http(decode_limit_hard=50_000_000, decode_limit_ratio=50)
diff --git a/python3/httplib2/__init__.py b/python3/httplib2/__init__.py
index 723a63c..ca2868a 100644
--- a/python3/httplib2/__init__.py
+++ b/python3/httplib2/__init__.py
@@ -1,6 +1,10 @@
 # -*- coding: utf-8 -*-
 """Small, fast HTTP client library for Python."""
 
+import functools
+
+from httplib2.decode import ZlibDecoder, DecoderProtocol, LimitDecoder, DeflateDecoder
+
 __author__ = "Joe Gregorio (joe@bitworking.org)"
 __copyright__ = "Copyright 2006, Joe Gregorio"
 __contributors__ = [
@@ -386,26 +390,27 @@ def _entry_disposition(response_headers, request_headers):
     return retval
 
 
-def _decompressContent(response, new_content):
+def _decompressContent(response, new_content, limit_kwargs):
     content = new_content
+    encoding_header = "content-encoding"
+    encoding = response.get(encoding_header, None)
+    limit_wrap = functools.partial(LimitDecoder, **limit_kwargs)
     try:
-        encoding = response.get("content-encoding", None)
-        if encoding in ["gzip", "deflate"]:
-            if encoding == "gzip":
-                content = gzip.GzipFile(fileobj=io.BytesIO(new_content)).read()
-            if encoding == "deflate":
-                try:
-                    content = zlib.decompress(content, zlib.MAX_WBITS)
-                except (IOError, zlib.error):
-                    content = zlib.decompress(content, -zlib.MAX_WBITS)
+        if encoding in ["gzip", "deflate", "zlib"]:
+            if not new_content:
+                raise zlib.error("empty compressed response")
+            try:
+                content = limit_wrap(ZlibDecoder()).consume_bytes(new_content, 0)
+            except (IOError, zlib.error):
+                content = limit_wrap(DeflateDecoder()).consume_bytes(new_content, 0)
             response["content-length"] = str(len(content))
             # Record the historical presence of the encoding in a way the won't interfere.
-            response["-content-encoding"] = response["content-encoding"]
-            del response["content-encoding"]
+            response["-content-encoding"] = response.pop(encoding_header)
     except (IOError, zlib.error):
         content = ""
         raise FailedToDecompressContent(
-            _("Content purported to be compressed with %s but failed to decompress.") % response.get("content-encoding"),
+            _("Content purported to be compressed with %s but failed to decompress.")
+            % encoding,
             response,
             content,
         )
@@ -1232,6 +1237,10 @@ class Http(object):
         disable_ssl_certificate_validation=False,
         tls_maximum_version=None,
         tls_minimum_version=None,
+        decode_limit_hard=None,
+        decode_limit_safe=None,
+        decode_limit_ratio=None,
+        decode_limit_chunk=None,
     ):
         """If 'cache' is a string then it is used as a directory name for
         a disk cache. Otherwise it must be an object that supports the
@@ -1258,6 +1267,11 @@ class Http(object):
 
         tls_maximum_version / tls_minimum_version require Python 3.7+ /
         OpenSSL 1.1.0g+. A value of "TLSv1_3" requires OpenSSL 1.1.1+.
+
+        `decode_limit_{hard,safe,ratio,chunk}` options configure `httplib2.decode.LimitDecoder` in attempt order:
+          - Http() argument - top priority
+          - environment httplib2_decode_limit_{hard,safe,ratio,chunk}
+          - LimitDecoder defaults - least priority
         """
         self.proxy_info = proxy_info
         self.ca_certs = ca_certs
@@ -1306,6 +1320,22 @@ class Http(object):
         # Keep Authorization: headers on a redirect.
         self.forward_authorization_headers = False
 
+        limit_kwargs = dict(
+            hard_limit=try_value_or_env(
+                int, decode_limit_hard, "httplib2_decode_limit_hard"
+            ),
+            safe_limit=try_value_or_env(
+                int, decode_limit_safe, "httplib2_decode_limit_safe"
+            ),
+            ratio=try_value_or_env(
+                float, decode_limit_ratio, "httplib2_decode_limit_ratio"
+            ),
+            chunk_size=try_value_or_env(
+                int, decode_limit_chunk, "httplib2_decode_limit_chunk"
+            ),
+        )
+        self.limit_kwargs = {k: v for k, v in limit_kwargs.items() if v is not None}
+
     def close(self):
         """Close persistent connections, clear sensitive data.
         Not thread-safe, requires external synchronization against concurrent requests.
@@ -1425,7 +1455,7 @@ class Http(object):
                     content = response.read()
                 response = Response(response)
                 if method != "HEAD":
-                    content = _decompressContent(response, content)
+                    content = _decompressContent(response, content, self.limit_kwargs)
 
             break
         return (response, content)
@@ -1797,3 +1827,16 @@ class Response(dict):
             return self
         else:
             raise AttributeError(name)
+
+
+def try_value_or_env(to, value, env_key, default=None):
+    candidates = (value, os.environ.get(env_key), os.environ.get(env_key.upper()))
+    # same as `to(x1) or to(x2) or to(x3)` except accepting falsey values like 0
+    for x in candidates:
+        if x is None:
+            continue
+        try:
+            return to(x)
+        except ValueError:
+            pass
+    return default
diff --git a/python3/httplib2/decode.py b/python3/httplib2/decode.py
new file mode 100644
index 0000000..588cf42
--- /dev/null
+++ b/python3/httplib2/decode.py
@@ -0,0 +1,209 @@
+from typing import Protocol
+import zlib
+
+
+class DecodeRatioError(Exception):
+    """Output-to-input amplification ratio exceeded the configured limit."""
+
+
+class DecodeLimitError(Exception):
+    """Total output length exceeded the hard limit."""
+
+
+class DecoderProtocol(Protocol):
+    @property
+    def needs_input(self) -> bool:
+        ...
+
+    def decode(self, b: bytes, max_length: int = 0) -> bytes:
+        ...
+
+    @property
+    def unconsumed_tail(self) -> bytes:
+        ...
+
+    def flush(self) -> bytes:
+        ...
+
+    def consume_bytes(self, data: bytes, chunk_size: int = 64 << 10) -> bytes:
+        out = bytearray()
+        if not data:
+            return self.flush()
+        if chunk_size == 0:
+            chunk_size = len(data)
+        for i in range(0, len(data), chunk_size):
+            chunk = data[i : i + chunk_size]
+            out.extend(self.decode(chunk))
+        out.extend(self.flush())
+        return bytes(out)
+
+
+class ZlibDecoder(DecoderProtocol):
+    """
+    Thin wrapper around zlib.Decompressor conforming to the Decoder interface.
+
+    When max_length caps output, zlib exposes input that still needs processing
+    through unconsumed_tail.
+    """
+
+    __slots__ = ("_decoder",)
+
+    WBITS_DEFLATE = -15
+    WBITS_ZLIB = 15
+    WBITS_GZIP = 15 | 16
+    WBITS_AUTO_GZIP_ZLIB = 15 | 32  # but not deflate
+
+    def __init__(self, wbits: int = WBITS_AUTO_GZIP_ZLIB):
+        self._decoder: zlib._Decompress | None = zlib.decompressobj(wbits)
+
+    @property
+    def needs_input(self) -> bool:
+        if self._decoder is None:
+            raise RuntimeError("used after flush()")
+        return not self._decoder.unconsumed_tail and not self._decoder.eof
+
+    def decode(self, b: bytes, max_length: int = 0) -> bytes:
+        if self._decoder is None:
+            raise RuntimeError("used after flush()")
+        return self._decoder.decompress(b, max_length)
+
+    @property
+    def unconsumed_tail(self) -> bytes:
+        if self._decoder is None:
+            raise RuntimeError("used after flush()")
+        return self._decoder.unconsumed_tail
+
+    def flush(self) -> bytes:
+        if self._decoder is None:
+            raise RuntimeError("used after flush()")
+        result = self._decoder.flush()
+        self._decoder = None
+        return result
+
+
+def DeflateDecoder() -> ZlibDecoder:
+    return ZlibDecoder(ZlibDecoder.WBITS_DEFLATE)
+
+
+class LimitDecoder(DecoderProtocol):
+    __slots__ = (
+        "_decoder",
+        "_ratio",
+        "_chunk_size",
+        "_safe_limit",
+        "_hard_limit",
+        "_consumed_length",
+        "_output_length",
+        "_input_buffer",
+        "_flushed",
+    )
+
+    def __init__(
+        self,
+        decoder: DecoderProtocol,
+        ratio: float = 100,
+        chunk_size: int = 64 << 10,
+        safe_limit: int = 10 << 20,
+        hard_limit: int = 10 << 30,
+    ) -> None:
+        if ratio < 0:
+            raise ValueError(f"LimitDecoder() ratio={ratio} expected >= 0")
+        if chunk_size < 0:
+            raise ValueError(f"LimitDecoder() chunk_size={chunk_size} expected >= 0")
+        if safe_limit < 0:
+            raise ValueError(f"LimitDecoder() safe_limit={safe_limit} expected >= 0")
+        if hard_limit < 0:
+            raise ValueError(f"LimitDecoder() safe_limit={safe_limit} expected >= 0")
+
+        self._decoder: DecoderProtocol = decoder
+        self._ratio: float = ratio
+        self._chunk_size: int = chunk_size
+        self._safe_limit: int = safe_limit
+        self._hard_limit: int = hard_limit
+        self._consumed_length: int = 0
+        self._output_length: int = 0
+        self._input_buffer: bytearray = bytearray()
+        self._flushed: bool = False
+
+    def _check_limits(self) -> None:
+        if (self._hard_limit > 0) and (self._output_length > self._hard_limit):
+            raise DecodeLimitError(f"Output length {self._output_length} exceeds hard limit {self._hard_limit}")
+        if (self._safe_limit > 0) and (self._output_length < self._safe_limit):
+            return
+        if (self._ratio > 0) and (self._output_length > self._consumed_length * self._ratio):
+            actual_ratio = self._output_length / self._consumed_length if self._consumed_length > 0 else float("inf")
+            raise DecodeRatioError(
+                f"Amplification ratio {actual_ratio:.1f} ({self._output_length}/{self._consumed_length})"
+                f" exceeds limit {self._ratio}"
+            )
+
+    def _decode_max_length(self, input_length: int) -> int:
+        limits = []
+        if self._hard_limit > 0:
+            limits.append(self._hard_limit)
+        if self._ratio > 0:
+            ratio_limit = int(
+                (self._consumed_length + input_length) * self._ratio
+            )
+            if self._safe_limit > 0:
+                ratio_limit = max(ratio_limit, self._safe_limit - 1)
+            limits.append(ratio_limit)
+        if not limits:
+            return 0
+        return max(1, min(limits) - self._output_length + 1)
+
+    @property
+    def needs_input(self) -> bool:
+        return self._decoder.needs_input
+
+    def decode(self, b: bytes, max_length: int = 0) -> bytes:
+        if self._flushed:
+            raise RuntimeError("decode() called after flush()")
+        output = self._pump(b)
+        return bytes(output)
+
+    def flush(self) -> bytes:
+        if self._flushed:
+            raise RuntimeError("flush() called more than once")
+        self._flushed = True
+
+        output = self._pump(b"")
+
+        data = self._decoder.flush()
+        output.extend(data)
+        self._output_length += len(data)
+        self._check_limits()
+
+        return bytes(output)
+
+    def _pump(self, b: bytes) -> bytearray:
+        self._input_buffer.extend(b)
+
+        output = bytearray()
+        drain = False
+        while True:
+            if self._input_buffer:
+                chunk = bytes(self._input_buffer[: self._chunk_size])
+                del self._input_buffer[: self._chunk_size]
+            elif drain:
+                chunk = b""
+            else:
+                break
+
+            max_length = self._decode_max_length(len(chunk))
+            data = self._decoder.decode(chunk, max_length)
+            tail = self._decoder.unconsumed_tail
+            consumed = len(chunk) - len(tail)
+            self._consumed_length += consumed
+            if tail:
+                self._input_buffer[:0] = tail
+
+            output.extend(data)
+            self._output_length += len(data)
+            self._check_limits()
+
+            drain = bool(max_length and len(data) == max_length)
+            if not data and not consumed and not tail:
+                break
+
+        return output
diff --git a/setup.cfg b/setup.cfg
index fa392d1..93641b0 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -2,7 +2,7 @@
 
 [flake8]
 exclude = *.egg*,.env,.git,.tox,_*,build*,dist*,venv*,python2/,python3/
-ignore = E261,E731,W503
+ignore = E203,E261,E731,W503
 max-line-length = 121
 
 [tool:pytest]
diff --git a/tests/__init__.py b/tests/__init__.py
index b5c76a2..d0f4538 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -818,3 +818,11 @@ def rebuild_uri(old, scheme=_missing, netloc=_missing, host=_missing, port=_miss
         path = u.path
     new = (scheme, netloc, path) + u[3:]
     return urllib.parse.urlunsplit(new)
+
+
+# remove when python version is raised to 3.9+
+try:
+    randbytes = random.randbytes
+except AttributeError:
+    def randbytes(n):
+        return random.getrandbits(n * 8).to_bytes(n, "little")
diff --git a/tests/test_encoding.py b/tests/test_encoding.py
index 1d9770a..7431e24 100644
--- a/tests/test_encoding.py
+++ b/tests/test_encoding.py
@@ -1,13 +1,17 @@
+import subprocess
+import sys
+
+import pytest
+
 import httplib2
+from httplib2.decode import LimitDecoder, ZlibDecoder, DecodeLimitError, DecodeRatioError
 import tests
 
 
 def test_gzip_head():
     # Test that we don't try to decompress a HEAD response
     http = httplib2.Http()
-    response = tests.http_response_bytes(
-        headers={"content-encoding": "gzip", "content-length": 42}
-    )
+    response = tests.http_response_bytes(headers={"content-encoding": "gzip", "content-length": 42})
     with tests.server_const_bytes(response) as uri:
         response, content = http.request(uri, "HEAD")
         assert response.status == 200
@@ -48,9 +52,7 @@ def test_gzip_malformed_response():
     http = httplib2.Http()
     # Test that we raise a good exception when the gzip fails
     http.force_exception_to_status_code = False
-    response = tests.http_response_bytes(
-        headers={"content-encoding": "gzip"}, body=b"obviously not compressed"
-    )
+    response = tests.http_response_bytes(headers={"content-encoding": "gzip"}, body=b"obviously not compressed")
     with tests.server_const_bytes(response, request_count=2) as uri:
         with tests.assert_raises(httplib2.FailedToDecompressContent):
             http.request(uri, "GET")
@@ -82,9 +84,7 @@ def test_deflate_malformed_response():
     # Test that we raise a good exception when the deflate fails
     http = httplib2.Http()
     http.force_exception_to_status_code = False
-    response = tests.http_response_bytes(
-        headers={"content-encoding": "deflate"}, body=b"obviously not compressed"
-    )
+    response = tests.http_response_bytes(headers={"content-encoding": "deflate"}, body=b"obviously not compressed")
     with tests.server_const_bytes(response, request_count=2) as uri:
         with tests.assert_raises(httplib2.FailedToDecompressContent):
             http.request(uri, "GET")
@@ -97,6 +97,18 @@ def test_deflate_malformed_response():
         assert response.reason.startswith("Content purported")
 
 
+@pytest.mark.parametrize("encoding", ("gzip", "deflate"))
+def test_empty_encoded_response(encoding):
+    http = httplib2.Http()
+    http.force_exception_to_status_code = False
+    response = tests.http_response_bytes(
+        headers={"content-encoding": encoding}, body=b""
+    )
+    with tests.server_const_bytes(response) as uri:
+        with tests.assert_raises(httplib2.FailedToDecompressContent):
+            http.request(uri, "GET")
+
+
 def test_zlib_get():
     # Test that we support zlib compression
     http = httplib2.Http()
@@ -110,3 +122,194 @@ def test_zlib_get():
         assert "content-encoding" not in response
         assert int(response["content-length"]) == len(b"properly compressed")
         assert content == b"properly compressed"
+
+
+def test_gzip_excess_ratio():
+    http = httplib2.Http()
+    original = b"\x00" * (50 << 20)  # 50 MiB to ~50 KiB
+    response = tests.http_response_bytes(
+        headers={"content-encoding": "gzip"},
+        body=tests.gzip_compress(original),
+    )
+    with tests.server_const_bytes(response) as uri:
+        try:
+            http.request(uri, "GET")
+            assert False, "expected DecodeRatioError"
+        except DecodeRatioError:
+            pass
+
+
+@pytest.mark.parametrize("safe_limit", (0, 1000, 20000))
+def test_limitdecoder_normal_decompression_no_limits(safe_limit):
+    """Standard decompression of random data should pass with any safe_limit"""
+    original = tests.randbytes(10 << 10)
+    compressed = tests.zlib_compress(original)
+
+    decoder = LimitDecoder(
+        ZlibDecoder(),
+        ratio=10,
+        safe_limit=safe_limit,
+        hard_limit=len(original) + 1,
+    )
+    result = decoder.consume_bytes(compressed)
+    assert result == original
+    assert decoder._output_length == len(original)
+
+
+def test_limitdecoder_normal_rechunking():
+    """Passing a massive single chunk should be re-chunked internally without error"""
+    original = b"\x00" * (10 << 20)
+    compressed = tests.zlib_compress(original)
+    assert len(compressed) > 2000
+
+    decoder = LimitDecoder(
+        ZlibDecoder(),
+        ratio=2000,
+        chunk_size=512,
+        safe_limit=0,
+        hard_limit=len(original) + 1,
+    )
+    result = decoder.consume_bytes(compressed, chunk_size=0)
+    assert result == original
+    assert decoder._consumed_length == len(compressed)
+
+
+def test_limitdecoder_amplification_ratio_exceeded():
+    """High ratio should trigger DecodeRatioError above safe_limit"""
+    original = b"\x00" * (1 << 20)
+    compressed = tests.zlib_compress(original)
+
+    decoder = LimitDecoder(
+        ZlibDecoder(),
+        ratio=10,
+        chunk_size=512,
+        safe_limit=0,
+        hard_limit=len(original) + 1,
+    )
+    try:
+        decoder.consume_bytes(compressed, chunk_size=0)
+        assert False, "expected DecodeRatioError"
+    except DecodeRatioError:
+        pass
+    assert 0 < decoder._consumed_length <= 512
+
+
+@pytest.mark.parametrize("ratio", (0, 10, 1000))
+def test_limitdecoder_hard_limit_exceeded(ratio):
+    """Output exceeding hard_limit must trigger DecodeLimitError regardless of ratio"""
+    original = b"\x00" * (10 << 10)
+    compressed = tests.zlib_compress(original)
+
+    decoder = LimitDecoder(
+        ZlibDecoder(),
+        ratio=ratio,
+        safe_limit=len(original) + 1,
+        hard_limit=len(original) - 1,
+    )
+    try:
+        decoder.consume_bytes(compressed)
+        assert False, "expected DecodeLimitError"
+    except DecodeLimitError:
+        pass
+
+
+def test_limitdecoder_bounds_allocation_before_hard_limit():
+    code = r"""
+import os
+import resource
+import zlib
+
+from httplib2.decode import DecodeLimitError, DecodeRatioError, LimitDecoder
+from httplib2.decode import ZlibDecoder
+
+compressor = zlib.compressobj(wbits=31)
+compressed = b"".join(
+    compressor.compress(b"\x00" * (1 << 20)) for _ in range(50)
+)
+compressed += compressor.flush()
+page_size = os.sysconf("SC_PAGE_SIZE")
+with open("/proc/self/statm", "r") as statm:
+    current_vms = int(statm.read().split()[0]) * page_size
+_, hard = resource.getrlimit(resource.RLIMIT_AS)
+resource.setrlimit(resource.RLIMIT_AS, (current_vms + (30 << 20), hard))
+decoder = LimitDecoder(
+    ZlibDecoder(), ratio=100, safe_limit=0, hard_limit=1 << 20
+)
+try:
+    decoder.consume_bytes(compressed, chunk_size=0)
+except (DecodeLimitError, DecodeRatioError):
+    pass
+except MemoryError:
+    raise AssertionError("decompressor allocated past the configured limit")
+else:
+    raise AssertionError("expected a decompression limit error")
+"""
+    result = subprocess.run(
+        [sys.executable, "-c", code], capture_output=True, text=True
+    )
+    assert result.returncode == 0, result.stderr
+
+
+@pytest.mark.parametrize("ratio", (0, 10, 1000))
+def test_limitdecoder_safe_limit_bypass(ratio):
+    """Any ratio allowed if total output < safe_limit"""
+    original = b"\x00" * (10 << 10)
+    compressed = tests.zlib_compress(original)
+
+    decoder = LimitDecoder(
+        ZlibDecoder(),
+        ratio=ratio,
+        safe_limit=len(original) + 1,
+        hard_limit=len(original) + 1,
+    )
+    result = decoder.consume_bytes(compressed)
+    assert result == original
+    assert decoder._output_length == len(original)
+
+
+def test_limitdecoder_single_byte_feeding():
+    """Feeding compressed data 1 byte at a time should still decode correctly"""
+    original = tests.randbytes(10 << 10)
+    compressed = tests.zlib_compress(original)
+
+    decoder = LimitDecoder(
+        ZlibDecoder(),
+        ratio=10,
+        safe_limit=5 << 10,
+        hard_limit=len(original) + 1,
+    )
+    result = decoder.consume_bytes(compressed, chunk_size=1)
+    assert result == original
+
+
+def test_limitdecoder_invalid_argument():
+    checks = (
+        ("ratio", dict(ratio=-1)),
+        ("chunk_size", dict(chunk_size=-1)),
+        ("safe_limit", dict(safe_limit=-1)),
+        ("hard_limit", dict(hard_limit=-1)),
+    )
+    for name, check in checks:
+        zd = ZlibDecoder()
+        try:
+            LimitDecoder(zd, **check)
+            assert False, f"check={name} expected ValueError"
+        except ValueError as e:
+            assert "expected >= 0" in str(e).lower(), str(e)
+
+
+def test_zlibdecoder_invalid_after_flush():
+    checks = (
+        ("needs_input", lambda d: d.needs_input),
+        ("decode", lambda d: d.decode(b"")),
+        ("flush", lambda d: d.flush()),
+    )
+    for name, check in checks:
+        d = ZlibDecoder()
+        d.decode(tests.zlib_compress(b""))
+        d.flush()
+        try:
+            check(d)
+            assert False, f"check={name} expected RuntimeError"
+        except RuntimeError as e:
+            assert "used after flush" in str(e).lower(), str(e)