Introduce _validate_local_file_entry and refactor

Author: danny0838

Lib/zipfile/__init__.py

@@ -1500,7 +1500,9 @@ def _calc_initial_entry_offset(self, fp, data_offset):
             if self.debug > 2:
                 print('scanning file signatures before:', data_offset)
             for pos in self._iter_scan_signature(fp, stringFileHeader, 0, data_offset):
-                if self._starts_consecutive_file_entries(fp, pos, data_offset, checked_offsets):
+                if self.debug > 2:
+                    print('checking file signature at:', pos)
+                if self._validate_local_file_entry_sequence(fp, pos, data_offset, checked_offsets):
                     return data_offset - pos
         return 0
 
@@ -1529,12 +1531,12 @@ def _iter_scan_signature(self, fp, signature, start_offset, end_offset, chunk_si
             remainder = chunk[-(sig_len - 1):]
             pos += read_size
 
-    def _starts_consecutive_file_entries(self, fp, start_offset, end_offset, checked_offsets):
+    def _validate_local_file_entry_sequence(self, fp, start_offset, end_offset, checked_offsets):
         offset = start_offset
 
         while offset < end_offset:
             if self.debug > 2:
-                print('checking local file entry:', offset)
+                print('checking local file entry at:', offset)
 
             # Cache checked offsets to improve performance by failing
             # subsequent (possible) file entry offsets early. They are
@@ -1546,69 +1548,72 @@ def _starts_consecutive_file_entries(self, fp, start_offset, end_offset, checked
             else:
                 checked_offsets.add(offset)
 
-            fp.seek(offset)
-            try:
-                fheader = self._read_local_file_header(fp)
-            except BadZipFile:
+            entry_size = self._validate_local_file_entry(fp, offset, end_offset)
+            if entry_size is None:
                 return False
+            offset += entry_size
 
-            # Create a dummy ZipInfo to utilize parsing.
-            # Flush only the required information.
-            zinfo = ZipInfo()
-            zinfo.header_offset = offset
-            zinfo.flag_bits = fheader[_FH_GENERAL_PURPOSE_FLAG_BITS]
-            zinfo.compress_size = fheader[_FH_COMPRESSED_SIZE]
-            zinfo.file_size = fheader[_FH_UNCOMPRESSED_SIZE]
-            zinfo.CRC = fheader[_FH_CRC]
-
-            filename = fp.read(fheader[_FH_FILENAME_LENGTH])
-            zinfo.extra = fp.read(fheader[_FH_EXTRA_FIELD_LENGTH])
-            pos = fp.tell()
+        return offset == end_offset
 
-            if pos > end_offset:
-                return False
+    def _validate_local_file_entry(self, fp, offset, end_offset):
+        fp.seek(offset)
+        try:
+            fheader = self._read_local_file_header(fp)
+        except BadZipFile:
+            return None
 
-            try:
-                zinfo._decodeExtra(crc32(filename))  # parse zip64
-            except BadZipFile:
-                return False
+        # Create a dummy ZipInfo to utilize parsing.
+        # Flush only the required information.
+        zinfo = ZipInfo()
+        zinfo.header_offset = offset
+        zinfo.flag_bits = fheader[_FH_GENERAL_PURPOSE_FLAG_BITS]
+        zinfo.compress_size = fheader[_FH_COMPRESSED_SIZE]
+        zinfo.file_size = fheader[_FH_UNCOMPRESSED_SIZE]
+        zinfo.CRC = fheader[_FH_CRC]
 
-            data_descriptor_size = 0
+        filename = fp.read(fheader[_FH_FILENAME_LENGTH])
+        zinfo.extra = fp.read(fheader[_FH_EXTRA_FIELD_LENGTH])
+        pos = fp.tell()
 
-            if zinfo.flag_bits & _MASK_USE_DATA_DESCRIPTOR:
-                # According to the spec, these fields should be zero when data
-                # descriptor is used. Otherwise treat as a false positive on
-                # random bytes to return early, as scanning for data descriptor
-                # is rather intensive.
-                if not (zinfo.CRC == zinfo.compress_size == zinfo.file_size == 0):
-                    return False
+        if pos > end_offset:
+            return None
 
-                zip64 = (
-                    fheader[_FH_UNCOMPRESSED_SIZE] == 0xffffffff or
-                    fheader[_FH_COMPRESSED_SIZE] == 0xffffffff
-                )
+        try:
+            zinfo._decodeExtra(crc32(filename))  # parse zip64
+        except BadZipFile:
+            return None
 
-                dd = self._scan_data_descriptor(fp, pos, end_offset, zip64)
+        data_descriptor_size = 0
 
-                if dd is None:
-                    return False
+        if zinfo.flag_bits & _MASK_USE_DATA_DESCRIPTOR:
+            # According to the spec, these fields should be zero when data
+            # descriptor is used. Otherwise treat as a false positive on
+            # random bytes to return early, as scanning for data descriptor
+            # is rather intensive.
+            if not (zinfo.CRC == zinfo.compress_size == zinfo.file_size == 0):
+                return None
+
+            zip64 = (
+                fheader[_FH_UNCOMPRESSED_SIZE] == 0xffffffff or
+                fheader[_FH_COMPRESSED_SIZE] == 0xffffffff
+            )
 
-                crc, compress_size, file_size, data_descriptor_size = dd
-                zinfo.CRC = crc
-                zinfo.compress_size = compress_size
-                zinfo.file_size = file_size
+            dd = self._scan_data_descriptor(fp, pos, end_offset, zip64)
 
-            offset += (
-                sizeFileHeader +
-                fheader[_FH_FILENAME_LENGTH] + fheader[_FH_EXTRA_FIELD_LENGTH] +
-                zinfo.compress_size +
-                data_descriptor_size
-            )
+            if dd is None:
+                return None
 
-            if self.debug > 2:
-                print('next', offset)
+            crc, compress_size, file_size, data_descriptor_size = dd
+            zinfo.CRC = crc
+            zinfo.compress_size = compress_size
+            zinfo.file_size = file_size
 
-        return offset == end_offset
+        return (
+            sizeFileHeader +
+            fheader[_FH_FILENAME_LENGTH] + fheader[_FH_EXTRA_FIELD_LENGTH] +
+            zinfo.compress_size +
+            data_descriptor_size
+        )
 
     def _read_local_file_header(self, fp):
         fheader = fp.read(sizeFileHeader)