Increase testing and fix functions

Author: rhpvorderman

src/isal/igzip.py

@@ -38,7 +38,7 @@
 _COMPRESS_LEVEL_TRADEOFF = isal_zlib.ISAL_DEFAULT_COMPRESSION
 _COMPRESS_LEVEL_BEST = isal_zlib.ISAL_BEST_COMPRESSION
 
-FTEXT, FHCRC, FHEXTRA, FNAME, FCOMMENT = 1, 2, 4, 8, 16
+FTEXT, FHCRC, FEXTRA, FNAME, FCOMMENT = 1, 2, 4, 8, 16
 
 try:
     BadGzipFile = gzip.BadGzipFile  # type: ignore
@@ -259,32 +259,42 @@ def _gzip_header_end(data: bytes) -> int:
     :param data: Compressed data that starts with a gzip header.
     :return: The end of the header / start of the raw deflate block.
     """
+    eof_error = EOFError("Compressed file ended before the end-of-stream "
+                         "marker was reached")
     if len(data) < 10:
-        raise BadGzipFile("Gzip header should be 10 bytes or more")
+        raise eof_error
     # We are not interested in mtime, xfl and os flags.
     magic, method, flags = struct.unpack("<HBB", data[:4])
     if magic != 0x8b1f:
         raise BadGzipFile(f"Not a gzipped file ({repr(data[:2])})")
     if method != 8:
         raise BadGzipFile("Unknown compression method")
     pos = 10
-    if flags & FHEXTRA:
+    failure = False
+    if flags & FEXTRA:
         xlen = int.from_bytes(data[pos: pos + 2], "little", signed=False)
         pos += 2 + xlen
     if flags & FNAME:
-        fname_end = data.index(b"\x00", pos) + 1
+        fname_end = data.find(b"\x00", pos) + 1
+        # fname_end will be -1 + 1 when null byte not found.
+        if not fname_end:
+            raise eof_error
         pos = fname_end
     if flags & FCOMMENT:
-        fcomment_end = data.index(b"\x00", pos) + 1
+        fcomment_end = data.find(b"\x00", pos) + 1
+        if not fcomment_end:
+            raise eof_error
         pos = fcomment_end
     if flags & FHCRC:
-        pos += 2
         header_crc = int.from_bytes(data[pos: pos + 2], "little", signed=False)
         # CRC is stored as a 16-bit integer by taking last bits of crc32.
         crc = isal_zlib.crc32(data[:pos]) & 0xFFFF
         if header_crc != crc:
             raise BadGzipFile(f"Corrupted header. Checksums do not "
                               f"match: {crc} != {header_crc}")
+        pos += 2
+    if failure or pos > len(data):
+        raise eof_error
     return pos
 
 

tests/test_igzip.py

@@ -30,9 +30,10 @@
 import sys
 import tempfile
 import zlib
+from gzip import FCOMMENT, FEXTRA, FHCRC, FNAME, FTEXT  # type: ignore
 from pathlib import Path
 
-from isal import igzip
+from isal import igzip, isal_zlib
 
 import pytest
 
@@ -214,3 +215,60 @@ def test_decompress_unknown_compression_method():
 
 def test_decompress_empty():
     assert igzip.decompress(b"") == b""
+
+
+def headers():
+    magic = b"\x1f\x8b"
+    method = b"\x08"
+    mtime = b"\x00\x00\x00\x00"
+    xfl = b"\x00"
+    os = b"\xff"
+    common_hdr_start = magic + method
+    common_hdr_end = mtime + xfl + os
+    xtra = b"METADATA"
+    xlen = len(xtra)
+    fname = b"my_data.tar"
+    fcomment = b"I wrote this header with my bare hands"
+    yield (common_hdr_start + FEXTRA.to_bytes(1, "little") +
+           common_hdr_end + xlen.to_bytes(2, "little") + xtra)
+    yield (common_hdr_start + FNAME.to_bytes(1, "little") +
+           common_hdr_end + fname + b"\x00")
+    yield (common_hdr_start + FCOMMENT.to_bytes(1, "little") +
+           common_hdr_end + fcomment + b"\x00")
+    flag = FHCRC.to_bytes(1, "little")
+    header = common_hdr_start + flag + common_hdr_end
+    crc = zlib.crc32(header) & 0xFFFF
+    yield(header + crc.to_bytes(2, "little"))
+    flag_bits = FTEXT | FEXTRA | FNAME | FCOMMENT | FHCRC
+    flag = flag_bits.to_bytes(1, "little")
+    header = (common_hdr_start + flag + common_hdr_end +
+              xlen.to_bytes(2, "little") + xtra + fname + b"\x00" +
+              fcomment + b"\x00")
+    crc = zlib.crc32(header) & 0xFFFF
+    yield header + crc.to_bytes(2, "little")
+
+
+@pytest.mark.parametrize("header", list(headers()))
+def test_gzip_header_end(header):
+    assert igzip._gzip_header_end(header) == len(header)
+
+
+def test_header_too_short():
+    with pytest.raises(igzip.BadGzipFile):
+        gzip.decompress(b"00")
+
+
+def test_header_corrupt():
+    header = b"\x1f\x8b\x08\x02\x00\x00\x00\x00\x00\xff"
+    # Create corrupt checksum by using wrong seed.
+    crc = zlib.crc32(header, 50) & 0xFFFF
+    true_crc = zlib.crc32(header) & 0xFFFF
+    header += crc.to_bytes(2, "little")
+
+    data = isal_zlib.compress(b"", wbits=-15)
+    trailer = b"\x00" * 8
+    compressed = header + data + trailer
+    with pytest.raises(igzip.BadGzipFile) as error:
+        igzip.decompress(compressed)
+    error.match(f"Corrupted header. "
+                f"Checksums do not match: {true_crc} != {crc}")