"""

Read and write ZIP files.

XXX references to utf-8 need further investigation.

"""

import binascii

import importlib.util

import io

import os

import shutil

import stat

import struct

import sys

import threading

import time

try:

import zlib # We may need its compression method

crc32 = zlib.crc32

except ImportError:

zlib = None

crc32 = binascii.crc32

try:

import bz2 # We may need its compression method

except ImportError:

bz2 = None

try:

import lzma # We may need its compression method

except ImportError:

lzma = None

__all__ = ["BadZipFile", "BadZipfile", "error",

"ZIP_STORED", "ZIP_DEFLATED", "ZIP_BZIP2", "ZIP_LZMA",

"is_zipfile", "ZipInfo", "ZipFile", "PyZipFile", "LargeZipFile",

"Path"]

class BadZipFile(Exception):

pass

class LargeZipFile(Exception):

"""

Raised when writing a zipfile, the zipfile requires ZIP64 extensions

and those extensions are disabled.

"""

error = BadZipfile = BadZipFile # Pre-3.2 compatibility names

ZIP64_LIMIT = (1 << 31) - 1

ZIP_FILECOUNT_LIMIT = (1 << 16) - 1

ZIP_MAX_COMMENT = (1 << 16) - 1

# constants for Zip file compression methods

ZIP_STORED = 0

ZIP_DEFLATED = 8

ZIP_BZIP2 = 12

ZIP_LZMA = 14

# Other ZIP compression methods not supported

DEFAULT_VERSION = 20

ZIP64_VERSION = 45

BZIP2_VERSION = 46

LZMA_VERSION = 63

# we recognize (but not necessarily support) all features up to that version

MAX_EXTRACT_VERSION = 63

# Below are some formats and associated data for reading/writing headers using

# the struct module. The names and structures of headers/records are those used

# in the PKWARE description of the ZIP file format:

# http://www.pkware.com/documents/casestudies/APPNOTE.TXT

# (URL valid as of January 2008)

# The "end of central directory" structure, magic number, size, and indices

# (section V.I in the format document)

structEndArchive = b"<4s4H2LH"

stringEndArchive = b"PK\005\006"

sizeEndCentDir = struct.calcsize(structEndArchive)

_ECD_SIGNATURE = 0

_ECD_DISK_NUMBER = 1

_ECD_DISK_START = 2

_ECD_ENTRIES_THIS_DISK = 3

_ECD_ENTRIES_TOTAL = 4

_ECD_SIZE = 5

_ECD_OFFSET = 6

_ECD_COMMENT_SIZE = 7

# These last two indices are not part of the structure as defined in the

# spec, but they are used internally by this module as a convenience

_ECD_COMMENT = 8

_ECD_LOCATION = 9

# The "central directory" structure, magic number, size, and indices

# of entries in the structure (section V.F in the format document)

structCentralDir = "<4s4B4HL2L5H2L"

stringCentralDir = b"PK\001\002"

sizeCentralDir = struct.calcsize(structCentralDir)

# indexes of entries in the central directory structure

_CD_SIGNATURE = 0

_CD_CREATE_VERSION = 1

_CD_CREATE_SYSTEM = 2

_CD_EXTRACT_VERSION = 3

_CD_EXTRACT_SYSTEM = 4

_CD_FLAG_BITS = 5

_CD_COMPRESS_TYPE = 6

_CD_TIME = 7

_CD_DATE = 8

_CD_CRC = 9

_CD_COMPRESSED_SIZE = 10

_CD_UNCOMPRESSED_SIZE = 11

_CD_FILENAME_LENGTH = 12

_CD_EXTRA_FIELD_LENGTH = 13

_CD_COMMENT_LENGTH = 14

_CD_DISK_NUMBER_START = 15

_CD_INTERNAL_FILE_ATTRIBUTES = 16

_CD_EXTERNAL_FILE_ATTRIBUTES = 17

_CD_LOCAL_HEADER_OFFSET = 18

# General purpose bit flags

# Zip Appnote: 4.4.4 general purpose bit flag: (2 bytes)

_MASK_ENCRYPTED = 1 << 0

# Bits 1 and 2 have different meanings depending on the compression used.

_MASK_COMPRESS_OPTION_1 = 1 << 1

# _MASK_COMPRESS_OPTION_2 = 1 << 2

# _MASK_USE_DATA_DESCRIPTOR: If set, crc-32, compressed size and uncompressed

# size are zero in the local header and the real values are written in the data

# descriptor immediately following the compressed data.

_MASK_USE_DATA_DESCRIPTOR = 1 << 3

# Bit 4: Reserved for use with compression method 8, for enhanced deflating.

# _MASK_RESERVED_BIT_4 = 1 << 4

_MASK_COMPRESSED_PATCH = 1 << 5

_MASK_STRONG_ENCRYPTION = 1 << 6

# _MASK_UNUSED_BIT_7 = 1 << 7

# _MASK_UNUSED_BIT_8 = 1 << 8

# _MASK_UNUSED_BIT_9 = 1 << 9

# _MASK_UNUSED_BIT_10 = 1 << 10

_MASK_UTF_FILENAME = 1 << 11

# Bit 12: Reserved by PKWARE for enhanced compression.

# _MASK_RESERVED_BIT_12 = 1 << 12

# _MASK_ENCRYPTED_CENTRAL_DIR = 1 << 13

# Bit 14, 15: Reserved by PKWARE

# _MASK_RESERVED_BIT_14 = 1 << 14

# _MASK_RESERVED_BIT_15 = 1 << 15

# The "local file header" structure, magic number, size, and indices

# (section V.A in the format document)

structFileHeader = "<4s2B4HL2L2H"

stringFileHeader = b"PK\003\004"

sizeFileHeader = struct.calcsize(structFileHeader)

_FH_SIGNATURE = 0

_FH_EXTRACT_VERSION = 1

_FH_EXTRACT_SYSTEM = 2

_FH_GENERAL_PURPOSE_FLAG_BITS = 3

_FH_COMPRESSION_METHOD = 4

_FH_LAST_MOD_TIME = 5

_FH_LAST_MOD_DATE = 6

_FH_CRC = 7

_FH_COMPRESSED_SIZE = 8

_FH_UNCOMPRESSED_SIZE = 9

_FH_FILENAME_LENGTH = 10

_FH_EXTRA_FIELD_LENGTH = 11

# The "Zip64 end of central directory locator" structure, magic number, and size

structEndArchive64Locator = "<4sLQL"

stringEndArchive64Locator = b"PK\x06\x07"

sizeEndCentDir64Locator = struct.calcsize(structEndArchive64Locator)

# The "Zip64 end of central directory" record, magic number, size, and indices

# (section V.G in the format document)

structEndArchive64 = "<4sQ2H2L4Q"

stringEndArchive64 = b"PK\x06\x06"

sizeEndCentDir64 = struct.calcsize(structEndArchive64)

_CD64_SIGNATURE = 0

_CD64_DIRECTORY_RECSIZE = 1

_CD64_CREATE_VERSION = 2

_CD64_EXTRACT_VERSION = 3

_CD64_DISK_NUMBER = 4

_CD64_DISK_NUMBER_START = 5

_CD64_NUMBER_ENTRIES_THIS_DISK = 6

_CD64_NUMBER_ENTRIES_TOTAL = 7

_CD64_DIRECTORY_SIZE = 8

_CD64_OFFSET_START_CENTDIR = 9

_DD_SIGNATURE = 0x08074b50

_EXTRA_FIELD_STRUCT = struct.Struct('<HH')

def _strip_extra(extra, xids):

# Remove Extra Fields with specified IDs.

unpack = _EXTRA_FIELD_STRUCT.unpack

modified = False

buffer = []

start = i = 0

while i + 4 <= len(extra):

xid, xlen = unpack(extra[i : i + 4])

j = i + 4 + xlen

if xid in xids:

if i != start:

buffer.append(extra[start : i])

start = j

modified = True

i = j

if not modified:

return extra

if start != len(extra):

buffer.append(extra[start:])

return b''.join(buffer)

def _check_zipfile(fp):

try:

if _EndRecData(fp):

return True # file has correct magic number

except OSError:

pass

return False

def is_zipfile(filename):

"""Quickly see if a file is a ZIP file by checking the magic number.

The filename argument may be a file or file-like object too.

"""

result = False

try:

if hasattr(filename, "read"):

result = _check_zipfile(fp=filename)

else:

with open(filename, "rb") as fp:

result = _check_zipfile(fp)

except OSError:

pass

return result

def _EndRecData64(fpin, offset, endrec):

"""

Read the ZIP64 end-of-archive records and use that to update endrec

"""

try:

fpin.seek(offset - sizeEndCentDir64Locator, 2)

except OSError:

# If the seek fails, the file is not large enough to contain a ZIP64

# end-of-archive record, so just return the end record we were given.

return endrec

data = fpin.read(sizeEndCentDir64Locator)

if len(data) != sizeEndCentDir64Locator:

return endrec

sig, diskno, reloff, disks = struct.unpack(structEndArchive64Locator, data)

if sig != stringEndArchive64Locator:

return endrec

if diskno != 0 or disks > 1:

raise BadZipFile("zipfiles that span multiple disks are not supported")

# Assume no 'zip64 extensible data'

fpin.seek(offset - sizeEndCentDir64Locator - sizeEndCentDir64, 2)

data = fpin.read(sizeEndCentDir64)

if len(data) != sizeEndCentDir64:

return endrec

sig, sz, create_version, read_version, disk_num, disk_dir, \

dircount, dircount2, dirsize, diroffset = \

struct.unpack(structEndArchive64, data)

if sig != stringEndArchive64:

return endrec

# Update the original endrec using data from the ZIP64 record

endrec[_ECD_SIGNATURE] = sig

endrec[_ECD_DISK_NUMBER] = disk_num

endrec[_ECD_DISK_START] = disk_dir

endrec[_ECD_ENTRIES_THIS_DISK] = dircount

endrec[_ECD_ENTRIES_TOTAL] = dircount2

endrec[_ECD_SIZE] = dirsize

endrec[_ECD_OFFSET] = diroffset

return endrec

def _EndRecData(fpin):

"""Return data from the "End of Central Directory" record, or None.

The data is a list of the nine items in the ZIP "End of central dir"

record followed by a tenth item, the file seek offset of this record."""

# Determine file size

fpin.seek(0, 2)

filesize = fpin.tell()

# Check to see if this is ZIP file with no archive comment (the

# "end of central directory" structure should be the last item in the

# file if this is the case).

try:

fpin.seek(-sizeEndCentDir, 2)

except OSError:

return None

data = fpin.read(sizeEndCentDir)

if (len(data) == sizeEndCentDir and

data[0:4] == stringEndArchive and

data[-2:] == b"\000\000"):

# the signature is correct and there's no comment, unpack structure

endrec = struct.unpack(structEndArchive, data)

endrec=list(endrec)

# Append a blank comment and record start offset

endrec.append(b"")

endrec.append(filesize - sizeEndCentDir)

# Try to read the "Zip64 end of central directory" structure

return _EndRecData64(fpin, -sizeEndCentDir, endrec)

# Either this is not a ZIP file, or it is a ZIP file with an archive

# comment. Search the end of the file for the "end of central directory"

# record signature. The comment is the last item in the ZIP file and may be

# up to 64K long. It is assumed that the "end of central directory" magic

# number does not appear in the comment.

maxCommentStart = max(filesize - ZIP_MAX_COMMENT - sizeEndCentDir, 0)

fpin.seek(maxCommentStart, 0)

data = fpin.read(ZIP_MAX_COMMENT + sizeEndCentDir)

start = data.rfind(stringEndArchive)

if start >= 0:

# found the magic number; attempt to unpack and interpret

recData = data[start:start+sizeEndCentDir]

if len(recData) != sizeEndCentDir:

# Zip file is corrupted.

return None

endrec = list(struct.unpack(structEndArchive, recData))

commentSize = endrec[_ECD_COMMENT_SIZE] #as claimed by the zip file

comment = data[start+sizeEndCentDir:start+sizeEndCentDir+commentSize]

endrec.append(comment)

endrec.append(maxCommentStart + start)

# Try to read the "Zip64 end of central directory" structure

return _EndRecData64(fpin, maxCommentStart + start - filesize,

endrec)

# Unable to find a valid end of central directory structure

return None

def _sanitize_filename(filename):

"""Terminate the file name at the first null byte and

ensure paths always use forward slashes as the directory separator."""

# Terminate the file name at the first null byte. Null bytes in file

# names are used as tricks by viruses in archives.

null_byte = filename.find(chr(0))

if null_byte >= 0:

filename = filename[0:null_byte]

# This is used to ensure paths in generated ZIP files always use

# forward slashes as the directory separator, as required by the

# ZIP format specification.

if os.sep != "/" and os.sep in filename:

filename = filename.replace(os.sep, "/")

if os.altsep and os.altsep != "/" and os.altsep in filename:

filename = filename.replace(os.altsep, "/")

return filename

class ZipInfo (object):

"""Class with attributes describing each file in the ZIP archive."""

__slots__ = (

'orig_filename',

'filename',

'date_time',

'compress_type',

'_compresslevel',

'comment',

'extra',

'create_system',

'create_version',

'extract_version',

'reserved',

'flag_bits',

'volume',

'internal_attr',

'external_attr',

'header_offset',

'CRC',

'compress_size',

'file_size',

'_raw_time',

'_end_offset',

)

def __init__(self, filename="NoName", date_time=(1980,1,1,0,0,0)):

self.orig_filename = filename # Original file name in archive

# Terminate the file name at the first null byte and

# ensure paths always use forward slashes as the directory separator.

filename = _sanitize_filename(filename)

self.filename = filename # Normalized file name

self.date_time = date_time # year, month, day, hour, min, sec

if date_time[0] < 1980:

raise ValueError('ZIP does not support timestamps before 1980')

# Standard values:

self.compress_type = ZIP_STORED # Type of compression for the file

self._compresslevel = None # Level for the compressor

self.comment = b"" # Comment for each file

self.extra = b"" # ZIP extra data

if sys.platform == 'win32':

self.create_system = 0 # System which created ZIP archive

else:

# Assume everything else is unix-y

self.create_system = 3 # System which created ZIP archive

self.create_version = DEFAULT_VERSION # Version which created ZIP archive

self.extract_version = DEFAULT_VERSION # Version needed to extract archive

self.reserved = 0 # Must be zero

self.flag_bits = 0 # ZIP flag bits

self.volume = 0 # Volume number of file header

self.internal_attr = 0 # Internal attributes

self.external_attr = 0 # External file attributes

self.compress_size = 0 # Size of the compressed file

self.file_size = 0 # Size of the uncompressed file

self._end_offset = None # Start of the next local header or central directory

# Other attributes are set by class ZipFile:

# header_offset Byte offset to the file header

# CRC CRC-32 of the uncompressed file

def __repr__(self):

result = ['<%s filename=%r' % (self.__class__.__name__, self.filename)]

if self.compress_type != ZIP_STORED:

result.append(' compress_type=%s' %

compressor_names.get(self.compress_type,

self.compress_type))

hi = self.external_attr >> 16

lo = self.external_attr & 0xFFFF

if hi:

result.append(' filemode=%r' % stat.filemode(hi))

if lo:

result.append(' external_attr=%#x' % lo)

isdir = self.is_dir()

if not isdir or self.file_size:

result.append(' file_size=%r' % self.file_size)

if ((not isdir or self.compress_size) and

(self.compress_type != ZIP_STORED or

self.file_size != self.compress_size)):

result.append(' compress_size=%r' % self.compress_size)

result.append('>')

return ''.join(result)

def FileHeader(self, zip64=None):

"""Return the per-file header as a bytes object.

When the optional zip64 arg is None rather than a bool, we will

decide based upon the file_size and compress_size, if known,

False otherwise.

"""

dt = self.date_time

dosdate = (dt[0] - 1980) << 9 | dt[1] << 5 | dt[2]

dostime = dt[3] << 11 | dt[4] << 5 | (dt[5] // 2)

if self.flag_bits & _MASK_USE_DATA_DESCRIPTOR:

# Set these to zero because we write them after the file data

CRC = compress_size = file_size = 0

else:

CRC = self.CRC

compress_size = self.compress_size

file_size = self.file_size

extra = self.extra

min_version = 0

if zip64 is None:

# We always explicitly pass zip64 within this module.... This

# remains for anyone using ZipInfo.FileHeader as a public API.

zip64 = file_size > ZIP64_LIMIT or compress_size > ZIP64_LIMIT

if zip64:

fmt = '<HHQQ'

extra = extra + struct.pack(fmt,

1, struct.calcsize(fmt)-4, file_size, compress_size)

file_size = 0xffffffff

compress_size = 0xffffffff

min_version = ZIP64_VERSION

if self.compress_type == ZIP_BZIP2:

min_version = max(BZIP2_VERSION, min_version)

elif self.compress_type == ZIP_LZMA:

min_version = max(LZMA_VERSION, min_version)

self.extract_version = max(min_version, self.extract_version)

self.create_version = max(min_version, self.create_version)

filename, flag_bits = self._encodeFilenameFlags()

header = struct.pack(structFileHeader, stringFileHeader,

self.extract_version, self.reserved, flag_bits,

self.compress_type, dostime, dosdate, CRC,

compress_size, file_size,

len(filename), len(extra))

return header + filename + extra

def _encodeFilenameFlags(self):

try:

return self.filename.encode('ascii'), self.flag_bits

except UnicodeEncodeError:

return self.filename.encode('utf-8'), self.flag_bits | _MASK_UTF_FILENAME

def _decodeExtra(self, filename_crc):

# Try to decode the extra field.

extra = self.extra

unpack = struct.unpack

while len(extra) >= 4:

tp, ln = unpack('<HH', extra[:4])

if ln+4 > len(extra):

raise BadZipFile("Corrupt extra field %04x (size=%d)" % (tp, ln))

if tp == 0x0001:

data = extra[4:ln+4]

# ZIP64 extension (large files and/or large archives)

try:

if self.file_size in (0xFFFF_FFFF_FFFF_FFFF, 0xFFFF_FFFF):

field = "File size"

self.file_size, = unpack('<Q', data[:8])

data = data[8:]

if self.compress_size == 0xFFFF_FFFF:

field = "Compress size"

self.compress_size, = unpack('<Q', data[:8])

data = data[8:]

if self.header_offset == 0xFFFF_FFFF:

field = "Header offset"

self.header_offset, = unpack('<Q', data[:8])

except struct.error:

raise BadZipFile(f"Corrupt zip64 extra field. "

f"{field} not found.") from None

elif tp == 0x7075:

data = extra[4:ln+4]

# Unicode Path Extra Field

try:

up_version, up_name_crc = unpack('<BL', data[:5])

if up_version == 1 and up_name_crc == filename_crc:

up_unicode_name = data[5:].decode('utf-8')

if up_unicode_name:

self.filename = _sanitize_filename(up_unicode_name)

else:

import warnings

warnings.warn("Empty unicode path extra field (0x7075)", stacklevel=2)

except struct.error as e:

raise BadZipFile("Corrupt unicode path extra field (0x7075)") from e

except UnicodeDecodeError as e:

raise BadZipFile('Corrupt unicode path extra field (0x7075): invalid utf-8 bytes') from e

extra = extra[ln+4:]

@classmethod

def from_file(cls, filename, arcname=None, *, strict_timestamps=True):

"""Construct an appropriate ZipInfo for a file on the filesystem.

filename should be the path to a file or directory on the filesystem.

arcname is the name which it will have within the archive (by default,

this will be the same as filename, but without a drive letter and with

leading path separators removed).

"""

if isinstance(filename, os.PathLike):

filename = os.fspath(filename)

st = os.stat(filename)

isdir = stat.S_ISDIR(st.st_mode)

mtime = time.localtime(st.st_mtime)

date_time = mtime[0:6]

if not strict_timestamps and date_time[0] < 1980:

date_time = (1980, 1, 1, 0, 0, 0)

elif not strict_timestamps and date_time[0] > 2107:

date_time = (2107, 12, 31, 23, 59, 59)

# Create ZipInfo instance to store file information

if arcname is None:

arcname = filename

arcname = os.path.normpath(os.path.splitdrive(arcname)[1])

while arcname[0] in (os.sep, os.altsep):

arcname = arcname[1:]

if isdir:

arcname += '/'

zinfo = cls(arcname, date_time)

zinfo.external_attr = (st.st_mode & 0xFFFF) << 16 # Unix attributes

if isdir:

zinfo.file_size = 0

zinfo.external_attr |= 0x10 # MS-DOS directory flag

else:

zinfo.file_size = st.st_size

return zinfo

def is_dir(self):

"""Return True if this archive member is a directory."""

if self.filename.endswith('/'):

return True

# The ZIP format specification requires to use forward slashes

# as the directory separator, but in practice some ZIP files

# created on Windows can use backward slashes. For compatibility

# with the extraction code which already handles this:

if os.path.altsep:

return self.filename.endswith((os.path.sep, os.path.altsep))

return False

# ZIP encryption uses the CRC32 one-byte primitive for scrambling some

# internal keys. We noticed that a direct implementation is faster than

# relying on binascii.crc32().

_crctable = None

def _gen_crc(crc):

for j in range(8):

if crc & 1:

crc = (crc >> 1) ^ 0xEDB88320

else:

crc >>= 1

return crc

# ZIP supports a password-based form of encryption. Even though known

# plaintext attacks have been found against it, it is still useful

# to be able to get data out of such a file.

#

# Usage:

# zd = _ZipDecrypter(mypwd)

# plain_bytes = zd(cypher_bytes)

def _ZipDecrypter(pwd):

key0 = 305419896

key1 = 591751049

key2 = 878082192

global _crctable

if _crctable is None:

_crctable = list(map(_gen_crc, range(256)))

crctable = _crctable

def crc32(ch, crc):

"""Compute the CRC32 primitive on one byte."""

return (crc >> 8) ^ crctable[(crc ^ ch) & 0xFF]

def update_keys(c):

nonlocal key0, key1, key2

key0 = crc32(c, key0)

key1 = (key1 + (key0 & 0xFF)) & 0xFFFFFFFF

key1 = (key1 * 134775813 + 1) & 0xFFFFFFFF

key2 = crc32(key1 >> 24, key2)

for p in pwd:

update_keys(p)

def decrypter(data):

"""Decrypt a bytes object."""

result = bytearray()

append = result.append

for c in data:

k = key2 | 2

c ^= ((k * (k^1)) >> 8) & 0xFF

update_keys(c)

append(c)

return bytes(result)

return decrypter

class LZMACompressor:

def __init__(self):

self._comp = None

def _init(self):

props = lzma._encode_filter_properties({'id': lzma.FILTER_LZMA1})

self._comp = lzma.LZMACompressor(lzma.FORMAT_RAW, filters=[

lzma._decode_filter_properties(lzma.FILTER_LZMA1, props)

])

return struct.pack('<BBH', 9, 4, len(props)) + props

def compress(self, data):

if self._comp is None:

return self._init() + self._comp.compress(data)

return self._comp.compress(data)

def flush(self):

if self._comp is None:

return self._init() + self._comp.flush()

return self._comp.flush()

class LZMADecompressor:

def __init__(self):

self._decomp = None

self._unconsumed = b''

self.eof = False

def decompress(self, data):

if self._decomp is None:

self._unconsumed += data

if len(self._unconsumed) <= 4:

return b''

psize, = struct.unpack('<H', self._unconsumed[2:4])

if len(self._unconsumed) <= 4 + psize:

return b''

self._decomp = lzma.LZMADecompressor(lzma.FORMAT_RAW, filters=[

lzma._decode_filter_properties(lzma.FILTER_LZMA1,

self._unconsumed[4:4 + psize])

])

data = self._unconsumed[4 + psize:]

del self._unconsumed

result = self._decomp.decompress(data)

self.eof = self._decomp.eof

return result

compressor_names = {

0: 'store',

1: 'shrink',

2: 'reduce',

3: 'reduce',

4: 'reduce',

5: 'reduce',

6: 'implode',

7: 'tokenize',

8: 'deflate',

9: 'deflate64',

10: 'implode',

12: 'bzip2',

14: 'lzma',

18: 'terse',

19: 'lz77',

97: 'wavpack',

98: 'ppmd',

}

def _check_compression(compression):

if compression == ZIP_STORED:

pass

elif compression == ZIP_DEFLATED:

if not zlib:

raise RuntimeError(

"Compression requires the (missing) zlib module")

elif compression == ZIP_BZIP2:

if not bz2:

raise RuntimeError(

"Compression requires the (missing) bz2 module")

elif compression == ZIP_LZMA:

if not lzma:

raise RuntimeError(

"Compression requires the (missing) lzma module")

else:

raise NotImplementedError("That compression method is not supported")

def _get_compressor(compress_type, compresslevel=None):

if compress_type == ZIP_DEFLATED:

if compresslevel is not None:

return zlib.compressobj(compresslevel, zlib.DEFLATED, -15)

return zlib.compressobj(zlib.Z_DEFAULT_COMPRESSION, zlib.DEFLATED, -15)

elif compress_type == ZIP_BZIP2:

if compresslevel is not None:

return bz2.BZ2Compressor(compresslevel)

return bz2.BZ2Compressor()

# compresslevel is ignored for ZIP_LZMA

elif compress_type == ZIP_LZMA:

return LZMACompressor()

else:

return None

def _get_decompressor(compress_type):

_check_compression(compress_type)

if compress_type == ZIP_STORED:

return None

elif compress_type == ZIP_DEFLATED:

return zlib.decompressobj(-15)

elif compress_type == ZIP_BZIP2:

return bz2.BZ2Decompressor()

elif compress_type == ZIP_LZMA:

return LZMADecompressor()

else:

descr = compressor_names.get(compress_type)

if descr:

raise NotImplementedError("compression type %d (%s)" % (compress_type, descr))

else:

raise NotImplementedError("compression type %d" % (compress_type,))

class _SharedFile:

def __init__(self, file, pos, close, lock, writing):

self._file = file

self._pos = pos

self._close = close

self._lock = lock

self._writing = writing

self.seekable = file.seekable

def tell(self):

return self._pos

def seek(self, offset, whence=0):

with self._lock:

if self._writing():

raise ValueError("Can't reposition in the ZIP file while "

"there is an open writing handle on it. "

"Close the writing handle before trying to read.")

self._file.seek(offset, whence)

self._pos = self._file.tell()

return self._pos

def read(self, n=-1):

with self._lock:

if self._writing():

raise ValueError("Can't read from the ZIP file while there "

"is an open writing handle on it. "

"Close the writing handle before trying to read.")

self._file.seek(self._pos)

data = self._file.read(n)

self._pos = self._file.tell()

return data

def close(self):

if self._file is not None:

fileobj = self._file

self._file = None

self._close(fileobj)

# Provide the tell method for unseekable stream

class _Tellable:

def __init__(self, fp):

self.fp = fp

self.offset = 0

def write(self, data):

n = self.fp.write(data)

self.offset += n

return n

def tell(self):

return self.offset

def flush(self):

self.fp.flush()

def close(self):

self.fp.close()

class ZipExtFile(io.BufferedIOBase):

"""File-like object for reading an archive member.

Is returned by ZipFile.open().

"""

# Max size supported by decompressor.

MAX_N = 1 << 31 - 1

# Read from compressed files in 4k blocks.

MIN_READ_SIZE = 4096

# Chunk size to read during seek

MAX_SEEK_READ = 1 << 24

def __init__(self, fileobj, mode, zipinfo, pwd=None,

close_fileobj=False):

self._fileobj = fileobj

self._pwd = pwd

self._close_fileobj = close_fileobj

self._compress_type = zipinfo.compress_type

self._compress_left = zipinfo.compress_size

self._left = zipinfo.file_size

self._decompressor = _get_decompressor(self._compress_type)

self._eof = False

self._readbuffer = b''

self._offset = 0

self.newlines = None

self.mode = mode

self.name = zipinfo.filename

if hasattr(zipinfo, 'CRC'):

self._expected_crc = zipinfo.CRC

self._running_crc = crc32(b'')

else:

self._expected_crc = None

self._seekable = False

try:

if fileobj.seekable():

self._orig_compress_start = fileobj.tell()

self._orig_compress_size = zipinfo.compress_size

self._orig_file_size = zipinfo.file_size

self._orig_start_crc = self._running_crc

self._orig_crc = self._expected_crc

self._seekable = True

except AttributeError:

pass

self._decrypter = None

if pwd:

if zipinfo.flag_bits & _MASK_USE_DATA_DESCRIPTOR:

# compare against the file type from extended local headers

check_byte = (zipinfo._raw_time >> 8) & 0xff

else:

# compare against the CRC otherwise

check_byte = (zipinfo.CRC >> 24) & 0xff

h = self._init_decrypter()

if h != check_byte:

raise RuntimeError("Bad password for file %r" % zipinfo.orig_filename)

def _init_decrypter(self):

self._decrypter = _ZipDecrypter(self._pwd)

# The first 12 bytes in the cypher stream is an encryption header

# used to strengthen the algorithm. The first 11 bytes are

# completely random, while the 12th contains the MSB of the CRC,

# or the MSB of the file time depending on the header type

# and is used to check the correctness of the password.

header = self._fileobj.read(12)

self._compress_left -= 12

return self._decrypter(header)[11]

def __repr__(self):

result = ['<%s.%s' % (self.__class__.__module__,

self.__class__.__qualname__)]

if not self.closed:

result.append(' name=%r mode=%r' % (self.name, self.mode))

if self._compress_type != ZIP_STORED:

result.append(' compress_type=%s' %

compressor_names.get(self._compress_type,

self._compress_type))

else:

result.append(' [closed]')

result.append('>')

return ''.join(result)

def readline(self, limit=-1):

"""Read and return a line from the stream.

If limit is specified, at most limit bytes will be read.

"""

if limit < 0:

# Shortcut common case - newline found in buffer.

i = self._readbuffer.find(b'\n', self._offset) + 1

if i > 0:

line = self._readbuffer[self._offset: i]

self._offset = i

return line

return io.BufferedIOBase.readline(self, limit)

def peek(self, n=1):

"""Returns buffered bytes without advancing the position."""

if n > len(self._readbuffer) - self._offset:

chunk = self.read(n)

if len(chunk) > self._offset:

self._readbuffer = chunk + self._readbuffer[self._offset:]

self._offset = 0

else:

self._offset -= len(chunk)

# Return up to 512 bytes to reduce allocation overhead for tight loops.

return self._readbuffer[self._offset: self._offset + 512]

def readable(self):

if self.closed:

raise ValueError("I/O operation on closed file.")

return True

def read(self, n=-1):

"""Read and return up to n bytes.

If the argument is omitted, None, or negative, data is read and returned until EOF is reached.

"""

if self.closed:

raise ValueError("read from closed file.")

if n is None or n < 0:

buf = self._readbuffer[self._offset:]

self._readbuffer = b''

self._offset = 0

while not self._eof:

buf += self._read1(self.MAX_N)

return buf

end = n + self._offset

if end < len(self._readbuffer):

buf = self._readbuffer[self._offset:end]

self._offset = end

return buf

n = end - len(self._readbuffer)

buf = self._readbuffer[self._offset:]

self._readbuffer = b''

self._offset = 0

while n > 0 and not self._eof:

data = self._read1(n)

if n < len(data):

self._readbuffer = data

self._offset = n

buf += data[:n]

break

buf += data

n -= len(data)

return buf

def _update_crc(self, newdata):

# Update the CRC using the given data.