fix(fetch-page): decode standard zlib-wrapped HTTP deflate (#711)

* fix(fetch-page): decode standard zlib-wrapped HTTP deflate

* docs(changelog): reference deflate fix PR 711

* chore: drop the CHANGELOG hunk, the file is frozen since #800

Release notes are generated from PR titles now.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

---------

Co-authored-by: Mads Lorentzen <madslorentzen17@gmail.com>
Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
authored and GitHub committed 2026-11-09 21:17:04 +01:00
1 parent 30791da9fd
commit ccc1c04765
2 files changed
+77

No files matched your search

+72
View File
@@ -0,0 +1,72 @@
"""Actual HTTP responses through the CLI entry point and urllib backend."""
import http.server
import contextlib
import io
from pathlib import Path
import sys
from unittest.mock import patch
from tempfile import TemporaryDirectory
import threading
import unittest
import zlib
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "tools"))
import fetch_page # noqa: E402
PAGE = b"<html><body><h1>Software Engineer</h1><p>Posting details.</p></body></html>"
class DeflateBackend(unittest.TestCase):
def test_standard_and_legacy_deflate_reach_saved_html_and_text(self):
requests = []
class Handler(http.server.BaseHTTPRequestHandler):
def do_GET(self):
if self.path == "/robots.txt":
body = b"User-agent: *\nAllow: /\n"
encoding = None
else:
requests.append((self.path, self.headers.get("Accept-Encoding")))
encoder = zlib.compressobj(wbits=-zlib.MAX_WBITS if self.path == "/raw" else zlib.MAX_WBITS)
body = encoder.compress(PAGE) + encoder.flush()
encoding = "deflate"
self.send_response(200)
if encoding:
self.send_header("Content-Encoding", encoding)
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, *args):
pass
server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
with TemporaryDirectory() as directory:
root = Path(directory)
for path in ("/standard", "/raw"):
with self.subTest(encoding=path):
output = root / (path[1:] + ".html")
stdout, stderr = io.StringIO(), io.StringIO()
# Select the existing urllib backend; robots.txt still
# goes through its real curl transport and parser.
with patch.object(fetch_page.shutil, "which", return_value=None), \
contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
code = fetch_page.main([
f"http://127.0.0.1:{server.server_port}{path}",
"--out", str(output), "--text",
])
self.assertEqual(code, 0, stdout.getvalue() + stderr.getvalue())
self.assertEqual(output.read_bytes(), PAGE)
self.assertIn("Software Engineer", stdout.getvalue())
self.assertIn(f"HTTP 200 size={len(PAGE)}", stdout.getvalue())
self.assertEqual(requests, [("/standard", "gzip, deflate"), ("/raw", "gzip, deflate")])
finally:
server.shutdown()
server.server_close()
thread.join()
if __name__ == "__main__":
unittest.main()
+5
View File
@@ -168,6 +168,11 @@ def fetch_with_urllib(url: str, out: Path, timeout: float) -> tuple[int, str | N
if encoding == "gzip":
body = gzip.decompress(body)
elif encoding == "deflate":
try:
# HTTP deflate carries a zlib wrapper (RFC 9110 section 8.4.1.2).
body = zlib.decompress(body)
except zlib.error:
# Keep compatibility with servers that send legacy raw DEFLATE.
body = zlib.decompress(body, -zlib.MAX_WBITS)
if len(body) > MAX_BYTES:
raise RuntimeError(f"response exceeds {MAX_BYTES} bytes after decompression")