bpo-45509: Check gzip headers for corrupted fields

Lib/gzip.py

@@ -415,34 +415,75 @@ def _read_gzip_header(fp):
     Returns last mtime if header was present or None otherwise.
     '''
     magic = fp.read(2)
-    if magic == b'':
+    if not magic:
         return None
 
     if magic != b'\037\213':
         raise BadGzipFile('Not a gzipped file (%r)' % magic)
-
-    (method, flag, last_mtime) = struct.unpack("<BBIxx", _read_exact(fp, 8))
+    base_header = _read_exact(fp, 8)
+    (method, flag, last_mtime) = struct.unpack("<BBIxx", base_header)
     if method != 8:
         raise BadGzipFile('Unknown compression method')
 
+    # No flags. No need for further parsing. These headers are returned by
+    # gzip.compress or zlib.compress(..., wbits=31)
+    if not flag:
+        return last_mtime
+    # Most gzip files will have only FNAME set. For example: produced by gzip
+    # command line application or python's GzipFile.
+    if flag == FNAME:
+        while True:
+            s = fp.read(1)
+            if not s:
+                raise EOFError("Compressed file ended before the "
+                               "end-of-stream marker was reached")
+            if s == b'\000':
+                break
+        return last_mtime
+
+    # Processing for more complex flags.
+
+    # Save header parts for FHCRC checking
+    header_parts = [magic, base_header]
+
     if flag & FEXTRA:
-        # Read & discard the extra field, if present
-        extra_len, = struct.unpack("<H", _read_exact(fp, 2))
-        _read_exact(fp, extra_len)
+        # Read the extra field, if present, save the fields for FHCRC checking.
+        extra_len_bytes = _read_exact(fp, 2)
+        extra_len, = struct.unpack("<H", extra_len_bytes)
+        extra = _read_exact(fp, extra_len)
+        header_parts.extend([extra_len_bytes, extra])
+
     if flag & FNAME:
-        # Read and discard a null-terminated string containing the filename
+        # Read a null-terminated string containing the filename, save the name
+        # for FHCRC checking.
         while True:
             s = fp.read(1)
-            if not s or s==b'\000':
+            if not s:
+                raise EOFError("Compressed file ended before the "
+                               "end-of-stream marker was reached")
+            header_parts.append(s)
+            if s == b'\000':
                 break
     if flag & FCOMMENT:
-        # Read and discard a null-terminated string containing a comment
+        # Read a null-terminated string containing the comment, save the
+        # comment for FHCRC checking.
         while True:
             s = fp.read(1)
-            if not s or s==b'\000':
+            if not s:
+                raise EOFError("Compressed file ended before the "
+                               "end-of-stream marker was reached")
+            header_parts.append(s)
+            if s == b'\000':
                 break
+
     if flag & FHCRC:
-        _read_exact(fp, 2)     # Read & discard the 16-bit header CRC
+        # Read the 16-bit header CRC and check it against the header.
+        header_crc, = struct.unpack("<H", _read_exact(fp, 2))
+        header = b"".join(header_parts)
+        true_crc = zlib.crc32(header) & 0xFFFF
+        if header_crc != true_crc:
+            raise BadGzipFile(f"Corrupted gzip header. Checksums do not "
+                               f"match: {true_crc:04x} != {header_crc:04x}")
     return last_mtime
 
 

Lib/test/test_gzip.py

@@ -9,6 +9,7 @@
 import struct
 import sys
 import unittest
+import zlib
 from subprocess import PIPE, Popen
 from test.support import import_helper
 from test.support import os_helper
@@ -579,6 +580,35 @@ def test_decompress_missing_trailer(self):
         compressed_data = gzip.compress(data1)
         self.assertRaises(EOFError, gzip.decompress, compressed_data[:-8])
 
+    def test_truncated_header(self):
+        truncated_headers = [
+            b"\x1f\x8b\x08\x00\x00\x00\x00\x00\x00",             # Missing OS byte
+            b"\x1f\x8b\x08\x02\x00\x00\x00\x00\x00\xff",         # FHRC, but no checksum
+            b"\x1f\x8b\x08\x04\x00\x00\x00\x00\x00\xff",         # FEXTRA, but no xlen
+            b"\x1f\x8b\x08\x04\x00\x00\x00\x00\x00\xff\xaa\x00", # FEXTRA, xlen, but no data
+            b"\x1f\x8b\x08\x08\x00\x00\x00\x00\x00\xff",         # FNAME but no fname
+            b"\x1f\x8b\x08\x10\x00\x00\x00\x00\x00\xff",         # FCOMMENT, but no fcomment
+        ]
+        for header in truncated_headers:
+            with self.subTest(header=header):
+                with self.assertRaises(EOFError):
+                    gzip.decompress(header)
+
+    def test_corrupted_gzip_header(self):
+        header = (b"\x1f\x8b\x08\x1f\x00\x00\x00\x00\x00\xff"  # All flags set
+                  b"\x05\x00"  # Xlen = 5
+                  b"extra"
+                  b"name\x00"
+                  b"comment\x00")
+        true_crc = zlib.crc32(header) & 0xFFFF
+        corrupted_crc = true_crc ^ 0xFFFF
+        corrupted_header = header + corrupted_crc.to_bytes(2, "little")
+        with self.assertRaises(gzip.BadGzipFile) as err:
+            gzip.decompress(corrupted_header)
+        self.assertEqual(str(err.exception),
+                         f"Corrupted gzip header. Checksums do not "
+                         f"match: {true_crc:04x} != {corrupted_crc:04x}")
+
     def test_read_truncated(self):
         data = data1*50
         # Drop the CRC (4 bytes) and file size (4 bytes).

Misc/NEWS.d/next/Library/2021-10-18-13-46-55.bpo-45509.Upwb60.rst

@@ -0,0 +1 @@
+Gzip headers are now checked for corrupted NAME, COMMENT and HCRC fields.