Script 'mail_helper' called by obssrc
Hello community,
here is the log from the commit of package python-httplib2 for openSUSE:Factory
checked in at 2026-07-07 21:01:45
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Comparing /work/SRC/openSUSE:Factory/python-httplib2 (Old)
and /work/SRC/openSUSE:Factory/.python-httplib2.new.1982 (New)
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Package is "python-httplib2"
Tue Jul 7 21:01:45 2026 rev:61 rq:1364066 version:0.32.0
Changes:
--------
--- /work/SRC/openSUSE:Factory/python-httplib2/python-httplib2.changes
2026-04-10 18:02:55.710400709 +0200
+++
/work/SRC/openSUSE:Factory/.python-httplib2.new.1982/python-httplib2.changes
2026-07-07 21:03:43.343342655 +0200
@@ -1,0 +2,8 @@
+Mon Jul 6 15:28:14 UTC 2026 - Dirk Müller <[email protected]>
+
+- update to 0.32.0:
+ * Python support 3.8+ only
+ * decompression limited by size and ratio
+ * decoder foundation to support more compression algorithms
+
+-------------------------------------------------------------------
Old:
----
httplib2-0.31.2.tar.gz
New:
----
httplib2-0.32.0.tar.gz
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Other differences:
------------------
++++++ python-httplib2.spec ++++++
--- /var/tmp/diff_new_pack.UINcYB/_old 2026-07-07 21:03:44.047367031 +0200
+++ /var/tmp/diff_new_pack.UINcYB/_new 2026-07-07 21:03:44.047367031 +0200
@@ -18,7 +18,7 @@
%{?sle15_python_module_pythons}
Name: python-httplib2
-Version: 0.31.2
+Version: 0.32.0
Release: 0
Summary: A Python HTTP client library
License: Apache-2.0 AND MIT AND (GPL-2.0-or-later OR MPL-1.1 OR
LGPL-2.1-or-later)
++++++ httplib2-0.31.2.tar.gz -> httplib2-0.32.0.tar.gz ++++++
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/httplib2-0.31.2/CHANGELOG
new/httplib2-0.32.0/CHANGELOG
--- old/httplib2-0.31.2/CHANGELOG 2026-01-23 12:00:25.000000000 +0100
+++ new/httplib2-0.32.0/CHANGELOG 2026-06-26 12:11:05.000000000 +0200
@@ -1,3 +1,11 @@
+0.32.0
+
+ Python support 3.8+ only
+
+ decompression limited by size and ratio
+
+ decoder foundation to support more compression algorithms
+
0.31.2
build(deps): pp.DelimitedList (camel case) only available in pyparsing>=3.1
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/httplib2-0.31.2/PKG-INFO new/httplib2-0.32.0/PKG-INFO
--- old/httplib2-0.31.2/PKG-INFO 2026-01-23 12:00:42.607651200 +0100
+++ new/httplib2-0.32.0/PKG-INFO 2026-06-26 12:11:20.274066200 +0200
@@ -1,6 +1,6 @@
Metadata-Version: 2.4
Name: httplib2
-Version: 0.31.2
+Version: 0.32.0
Summary: A comprehensive HTTP client library.
Home-page: https://github.com/httplib2/httplib2
Author: Joe Gregorio
@@ -15,7 +15,7 @@
Classifier: Programming Language :: Python :: 3
Classifier: Topic :: Internet :: WWW/HTTP
Classifier: Topic :: Software Development :: Libraries
-Requires-Python: >=3.6
+Requires-Python: >=3.8
License-File: LICENSE
Requires-Dist: pyparsing<4,>=3.1
Dynamic: author
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/httplib2-0.31.2/README.md
new/httplib2-0.32.0/README.md
--- old/httplib2-0.31.2/README.md 2026-01-23 12:00:25.000000000 +0100
+++ new/httplib2-0.32.0/README.md 2026-06-26 12:11:05.000000000 +0200
@@ -10,13 +10,13 @@
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 @@
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.
@@ -121,3 +121,26 @@
* 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 -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/httplib2-0.31.2/httplib2/__init__.py
new/httplib2-0.32.0/httplib2/__init__.py
--- old/httplib2-0.31.2/httplib2/__init__.py 2026-01-23 12:00:25.000000000
+0100
+++ new/httplib2-0.32.0/httplib2/__init__.py 2026-06-26 12:11:05.000000000
+0200
@@ -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 ([email protected])"
__copyright__ = "Copyright 2006, Joe Gregorio"
__contributors__ = [
@@ -16,7 +20,7 @@
"Lai Han",
]
__license__ = "MIT"
-__version__ = "0.31.2"
+__version__ = "0.32.0"
import base64
import calendar
@@ -385,26 +389,25 @@
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"]:
+ 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,
)
@@ -1235,6 +1238,10 @@
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
@@ -1261,6 +1268,11 @@
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
@@ -1309,6 +1321,22 @@
# 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.
@@ -1428,7 +1456,7 @@
content = response.read()
response = Response(response)
if method != "HEAD":
- content = _decompressContent(response, content)
+ content = _decompressContent(response, content,
self.limit_kwargs)
break
return (response, content)
@@ -1800,3 +1828,16 @@
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 -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/httplib2-0.31.2/httplib2/decode.py
new/httplib2-0.32.0/httplib2/decode.py
--- old/httplib2-0.31.2/httplib2/decode.py 1970-01-01 01:00:00.000000000
+0100
+++ new/httplib2-0.32.0/httplib2/decode.py 2026-06-26 12:11:05.000000000
+0200
@@ -0,0 +1,183 @@
+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) -> bytes:
+ ...
+
+ def flush(self) -> bytes:
+ ...
+
+ def consume_bytes(self, data: bytes, chunk_size: int = 64 << 10) -> bytes:
+ out = bytearray()
+ 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.
+
+ Note: zlib pushes all available decompressed data immediately upon
receiving
+ input. It never holds back output requiring `decode(b"")` to extract it.
+ Thus, `needs_input` naturally remains True.
+ """
+
+ __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.eof
+
+ def decode(self, b: bytes) -> bytes:
+ if self._decoder is None:
+ raise RuntimeError("used after flush()")
+ return self._decoder.decompress(b)
+
+ 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}"
+ )
+
+ @property
+ def needs_input(self) -> bool:
+ return self._decoder.needs_input
+
+ def decode(self, b: bytes) -> 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()
+ while True:
+ if not self._decoder.needs_input:
+ data = self._decoder.decode(b"")
+ if data:
+ output.extend(data)
+ self._output_length += len(data)
+ self._check_limits()
+ continue
+
+ if self._input_buffer:
+ chunk = bytes(self._input_buffer[: self._chunk_size])
+ del self._input_buffer[: self._chunk_size]
+
+ data = self._decoder.decode(chunk)
+ self._consumed_length += len(chunk)
+
+ if data:
+ output.extend(data)
+ self._output_length += len(data)
+ self._check_limits()
+
+ continue
+
+ # neither input nor decoder progress
+ break
+
+ return output
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/httplib2-0.31.2/httplib2.egg-info/PKG-INFO
new/httplib2-0.32.0/httplib2.egg-info/PKG-INFO
--- old/httplib2-0.31.2/httplib2.egg-info/PKG-INFO 2026-01-23
12:00:42.000000000 +0100
+++ new/httplib2-0.32.0/httplib2.egg-info/PKG-INFO 2026-06-26
12:11:20.000000000 +0200
@@ -1,6 +1,6 @@
Metadata-Version: 2.4
Name: httplib2
-Version: 0.31.2
+Version: 0.32.0
Summary: A comprehensive HTTP client library.
Home-page: https://github.com/httplib2/httplib2
Author: Joe Gregorio
@@ -15,7 +15,7 @@
Classifier: Programming Language :: Python :: 3
Classifier: Topic :: Internet :: WWW/HTTP
Classifier: Topic :: Software Development :: Libraries
-Requires-Python: >=3.6
+Requires-Python: >=3.8
License-File: LICENSE
Requires-Dist: pyparsing<4,>=3.1
Dynamic: author
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/httplib2-0.31.2/httplib2.egg-info/SOURCES.txt
new/httplib2-0.32.0/httplib2.egg-info/SOURCES.txt
--- old/httplib2-0.31.2/httplib2.egg-info/SOURCES.txt 2026-01-23
12:00:42.000000000 +0100
+++ new/httplib2-0.32.0/httplib2.egg-info/SOURCES.txt 2026-06-26
12:11:20.000000000 +0200
@@ -43,6 +43,7 @@
httplib2/auth.py
httplib2/cacerts.txt
httplib2/certs.py
+httplib2/decode.py
httplib2/error.py
httplib2/iri2uri.py
httplib2.egg-info/PKG-INFO
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/httplib2-0.31.2/setup.cfg
new/httplib2-0.32.0/setup.cfg
--- old/httplib2-0.31.2/setup.cfg 2026-01-23 12:00:42.607651200 +0100
+++ new/httplib2-0.32.0/setup.cfg 2026-06-26 12:11:20.274374200 +0200
@@ -2,7 +2,7 @@
[flake8]
exclude = *.egg*,.env,.git,.tox,_*,build*,dist*,venv*
-ignore = E261,E731,W503
+ignore = E203,E261,E731,W503
max-line-length = 121
[tool:pytest]
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/httplib2-0.31.2/setup.py new/httplib2-0.32.0/setup.py
--- old/httplib2-0.31.2/setup.py 2026-01-23 12:00:25.000000000 +0100
+++ new/httplib2-0.32.0/setup.py 2026-06-26 12:11:05.000000000 +0200
@@ -3,7 +3,7 @@
import setuptools.command.test
import sys
-VERSION = "0.31.2"
+VERSION = "0.32.0"
# `python setup.py test` uses existing Python environment, no virtualenv, no
pip.
@@ -86,7 +86,7 @@
package_data={"httplib2": ["*.txt"]},
install_requires=read_requirements("requirements.txt"),
tests_require=read_requirements("requirements-test.txt"),
- python_requires=">=3.6",
+ python_requires=">=3.8",
cmdclass={"test": TestCommand},
classifiers=[
"Development Status :: 4 - Beta",
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/httplib2-0.31.2/tests/__init__.py
new/httplib2-0.32.0/tests/__init__.py
--- old/httplib2-0.31.2/tests/__init__.py 2026-01-23 12:00:25.000000000
+0100
+++ new/httplib2-0.32.0/tests/__init__.py 2026-06-26 12:11:05.000000000
+0200
@@ -794,3 +794,11 @@
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 -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/httplib2-0.31.2/tests/test_encoding.py
new/httplib2-0.32.0/tests/test_encoding.py
--- old/httplib2-0.31.2/tests/test_encoding.py 2026-01-23 12:00:25.000000000
+0100
+++ new/httplib2-0.32.0/tests/test_encoding.py 2026-06-26 12:11:05.000000000
+0200
@@ -1,13 +1,14 @@
+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 +49,7 @@
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 +81,7 @@
# 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")
@@ -110,3 +107,157 @@
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
+
+
[email protected]("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 decoder._consumed_length == 512, "expected ratio error on first
chunk"
+
+
[email protected]("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=0,
+ hard_limit=len(original) - 1,
+ )
+ try:
+ decoder.consume_bytes(compressed)
+ assert False, "expected DecodeLimitError"
+ except DecodeLimitError:
+ pass
+
+
[email protected]("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)
