"""

# 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

try:

if _EndRecData(fp):

return True # file has correct magic number

except OSError:

pass

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

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, BadZipFile):

pass

"""

offset -= sizeEndCentDir64Locator

if offset < 0:

# 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

fpin.seek(offset)

data = fpin.read(sizeEndCentDir64Locator)

if len(data) != sizeEndCentDir64Locator:

raise OSError("Unknown I/O error")

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")

offset -= sizeEndCentDir64

if reloff > offset:

raise BadZipFile("Corrupt zip64 end of central directory locator")

# First, check the assumption that there is no prepended data.

fpin.seek(reloff)

extrasz = offset - reloff

data = fpin.read(sizeEndCentDir64)

if len(data) != sizeEndCentDir64:

raise OSError("Unknown I/O error")

if not data.startswith(stringEndArchive64) and reloff != offset:

# Since we already have seen the Zip64 EOCD Locator, it's

# possible we got here because there is prepended data.

# Assume no 'zip64 extensible data'

fpin.seek(offset)

extrasz = 0

data = fpin.read(sizeEndCentDir64)

if len(data) != sizeEndCentDir64:

raise OSError("Unknown I/O error")

if not data.startswith(stringEndArchive64):

raise BadZipFile("Zip64 end of central directory record not found")

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

dircount, dircount2, dirsize, diroffset = \

struct.unpack(structEndArchive64, data)

if (diroffset + dirsize != reloff or

sz + 12 != sizeEndCentDir64 + extrasz):

raise BadZipFile("Corrupt zip64 end of central directory record")

# 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

endrec[_ECD_LOCATION] = offset - extrasz

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

# 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()

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, filesize - 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 - (1 << 16) - sizeEndCentDir, 0)

fpin.seek(maxCommentStart, 0)

data = fpin.read()

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, endrec)

# Unable to find a valid end of central directory structure

"""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. 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, "/")

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."""

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 & 0x08:

# 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:

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)

if file_size > ZIP64_LIMIT or compress_size > ZIP64_LIMIT:

if not zip64:

raise LargeZipFile("Filesize would require ZIP64 extensions")

# File is larger than what fits into a 4 byte integer,

# fall back to the ZIP64 extension

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 | 0x800

def _decodeExtra(self):

# 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

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.

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."""

for j in range(8):

if crc & 1:

crc = (crc >> 1) ^ 0xEDB88320

else:

crc >>= 1

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)

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()

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

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',

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:

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:

_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:

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

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):

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

# 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._seekable = True

except AttributeError:

pass

self._decrypter = None

if pwd:

if zipinfo.flag_bits & 0x8:

# 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 < 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 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.

if self._expected_crc is None:

# No need to compute the CRC if we don't have a reference value

return

self._running_crc = crc32(newdata, self._running_crc)

# Check the CRC if we're at the end of the file

if self._eof and self._running_crc != self._expected_crc:

raise BadZipFile("Bad CRC-32 for file %r" % self.name)

def read1(self, n):

"""Read up to n bytes with at most one read() system call."""

if n is None or n < 0:

buf = self._readbuffer[self._offset:]

self._readbuffer = b''

self._offset = 0

while not self._eof:

data = self._read1(self.MAX_N)

if data:

buf += data

break

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

if n > 0:

while not self._eof:

data = self._read1(n)

if n < len(data):

self._readbuffer = data

self._offset = n

buf += data[:n]

break

if data:

buf += data

break

return buf

def _read1(self, n):

# Read up to n compressed bytes with at most one read() system call,