httpcompression middleware: warn of unexpected and prevent manual unupported encodings

scrapy/downloadermiddlewares/httpcompression.py

@@ -1,12 +1,13 @@
 from __future__ import annotations
 
 import io
+import logging
 import zlib
 from typing import TYPE_CHECKING, List, Optional, Union
 
 from scrapy import Request, Spider
 from scrapy.crawler import Crawler
-from scrapy.exceptions import NotConfigured
+from scrapy.exceptions import NotConfigured, NotSupported
 from scrapy.http import Response, TextResponse
 from scrapy.responsetypes import responsetypes
 from scrapy.statscollectors import StatsCollector
@@ -17,6 +18,9 @@
     from typing_extensions import Self
 
 ACCEPTED_ENCODINGS: List[bytes] = [b"gzip", b"deflate"]
+logger = logging.getLogger(__name__)
+
+ACCEPTED_ENCODINGS = [b"gzip", b"deflate"]
 
 try:
     import brotli
@@ -49,6 +53,8 @@ def from_crawler(cls, crawler: Crawler) -> Self:
     def process_request(
         self, request: Request, spider: Spider
     ) -> Union[Request, Response, None]:
+        self._raise_unsupported_compressors(request)
+
         request.headers.setdefault("Accept-Encoding", b", ".join(ACCEPTED_ENCODINGS))
         return None
 
@@ -85,6 +91,19 @@ def process_response(
 
         return response
 
+    def _raise_unsupported_compressors(self, request: Request):
+        encodings = request.headers.getlist("Accept-Encoding")
+        unsupported = [key for key in encodings if key not in ACCEPTED_ENCODINGS]
+        if len(unsupported):
+            unsupported = [
+                unsupp for unsupp in unsupported if isinstance(unsupp, bytes)
+            ]
+            unsupported_msg = b", ".join(unsupported) if len(unsupported) else "-"
+            raise NotSupported(
+                f"Request is configured with Accept-Encoding header with unsupported encoding(s): "
+                f"{unsupported_msg}"
+            )
+
     def _decode(self, body: bytes, encoding: bytes) -> bytes:
         if encoding == b"gzip" or encoding == b"x-gzip":
             body = gunzip(body)
@@ -99,8 +118,15 @@ def _decode(self, body: bytes, encoding: bytes) -> bytes:
                 # http://www.port80software.com/200ok/archive/2005/10/31/868.aspx
                 # http://www.gzip.org/zlib/zlib_faq.html#faq38
                 body = zlib.decompress(body, -15)
-        if encoding == b"br" and b"br" in ACCEPTED_ENCODINGS:
-            body = brotli.decompress(body)
+        if encoding == b"br":
+            if b"br" in ACCEPTED_ENCODINGS:
+                body = brotli.decompress(body)
+            else:
+                body = bytes()
+                logger.warning(
+                    "Brotli encoding received. "
+                    "Cannot decompress the body as Brotli is not installed."
+                )
         if encoding == b"zstd" and b"zstd" in ACCEPTED_ENCODINGS:
             # Using its streaming API since its simple API could handle only cases
             # where there is content size data embedded in the frame

tests/test_downloadermiddleware_httpcompression.py

@@ -9,7 +9,7 @@
     ACCEPTED_ENCODINGS,
     HttpCompressionMiddleware,
 )
-from scrapy.exceptions import NotConfigured
+from scrapy.exceptions import NotConfigured, NotSupported
 from scrapy.http import HtmlResponse, Request, Response
 from scrapy.responsetypes import responsetypes
 from scrapy.spiders import Spider
@@ -102,6 +102,20 @@ def test_process_request(self):
             request.headers.get("Accept-Encoding"), b", ".join(ACCEPTED_ENCODINGS)
         )
 
+    def test_process_request_checks_encodings(self):
+        initial_encodings = ACCEPTED_ENCODINGS.copy()
+
+        ACCEPTED_ENCODINGS.append(b"bro")
+        request = Request("http://scrapytest.org", headers={"Accept-Encoding": b"bro"})
+        self.mw.process_request(request, self.spider)
+        """Expecting no Exception raised here as `bro` encoding is forced to be allowed."""
+
+        ACCEPTED_ENCODINGS.pop()
+        self.assertRaises(NotSupported, self.mw.process_request, request, self.spider)
+
+        """Checking that valid encondings are back"""
+        self.assertListEqual(ACCEPTED_ENCODINGS, initial_encodings)
+
     def test_process_response_gzip(self):
         response = self._getresponse("gzip")
         request = response.request