new file mode 100644
@@ -0,0 +1,745 @@
+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)
@@ -6,6 +6,8 @@ LIC_FILES_CHKSUM = "file://LICENSE;md5=56e5e931172b6164b62dc7c4aba6c8cf"
SRC_URI[sha256sum] = "d7a10bc5ef5ab08322488bde8c726eeee5c8618723fdb399597ec58f3d82df81"
+SRC_URI += "file://CVE-2026-59939.patch"
+
inherit pypi python_setuptools_build_meta
RDEPENDS:${PN} += "\