Fixed #34170 -- Implemented Heal The Breach (HTB) in GzipMiddleware.

tests/middleware/tests.py

@@ -3,6 +3,7 @@ import random
 import re
 import struct
 from io import BytesIO
+from unittest import mock
 from urllib.parse import quote
 
 from django.conf import settings
@@ -978,12 +979,47 @@ class GZipMiddlewareTest(SimpleTestCase):
         ConditionalGetMiddleware from recognizing conditional matches
         on gzipped content).
         """
-        r1 = GZipMiddleware(self.get_response)(self.req)
-        r2 = GZipMiddleware(self.get_response)(self.req)
+
+        class DeterministicGZipMiddleware(GZipMiddleware):
+            max_random_bytes = 0
+
+        r1 = DeterministicGZipMiddleware(self.get_response)(self.req)
+        r2 = DeterministicGZipMiddleware(self.get_response)(self.req)
         self.assertEqual(r1.content, r2.content)
         self.assertEqual(self.get_mtime(r1.content), 0)
         self.assertEqual(self.get_mtime(r2.content), 0)
 
+    def test_random_bytes(self):
+        """A random number of bytes is added to mitigate the BREACH attack."""
+        with mock.patch(
+            "django.utils.text.secrets.randbelow", autospec=True, return_value=3
+        ):
+            r = GZipMiddleware(self.get_response)(self.req)
+        # The fourth byte of a gzip stream contains flags.
+        self.assertEqual(r.content[3], gzip.FNAME)
+        # A 3 byte filename "aaa" and a null byte are added.
+        self.assertEqual(r.content[10:14], b"aaa\x00")
+        self.assertEqual(self.decompress(r.content), self.compressible_string)
+
+    def test_random_bytes_streaming_response(self):
+        """A random number of bytes is added to mitigate the BREACH attack."""
+
+        def get_stream_response(request):
+            resp = StreamingHttpResponse(self.sequence)
+            resp["Content-Type"] = "text/html; charset=UTF-8"
+            return resp
+
+        with mock.patch(
+            "django.utils.text.secrets.randbelow", autospec=True, return_value=3
+        ):
+            r = GZipMiddleware(get_stream_response)(self.req)
+            content = b"".join(r)
+        # The fourth byte of a gzip stream contains flags.
+        self.assertEqual(content[3], gzip.FNAME)
+        # A 3 byte filename "aaa" and a null byte are added.
+        self.assertEqual(content[10:14], b"aaa\x00")
+        self.assertEqual(self.decompress(content), b"".join(self.sequence))
+
 
 class ETagGZipMiddlewareTest(SimpleTestCase):
     """