def __init__(self):

self.closed = False

def close(self):

self.closed = True

def __enter__(self):

return self

def __exit__(self, type, value, traceback):

def __enter__(self):

return self

def __exit__(self, type, value, traceback):

"""copy data from file-like object fsrc to file-like object fdst"""

if not length:

length = 1 * 1024

while True:

buf = fsrc.read(length)

if not buf:

break

def __init__(self, format):

self.format = format

def pack(self, *args):

return struct.pack(self.format, *args)

def unpack(self, buffer):

"""

# Remove Extra Fields with specified IDs.

_EXTRA_FIELD_STRUCT = Struct('<HH')

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

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:

pass

"""

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

endrec)

# Unable to find a valid end of central directory structure

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

# 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 ALTSEP and ALTSEP != "/" and ALTSEP in filename:

filename = filename.replace(ALTSEP, "/")

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

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. \n {field} not found.")

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:

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.

if isinstance(filename, PathLike):

filename = 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, 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 ALTSEP:

return self.filename.endswith(OS_PATH_SEP) or self.filename.endswith(ALTSEP)

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, compresslevel=5):

if compresslevel is None:

compresslevel = 5

self.compresslevel = compresslevel

def compress(self, data):

# compresslevel should be 0 through 9 for DEFLATED

# wbits for DeflateIO is 5 through 15 inclusive

wbits = 5+self.compresslevel

stream = io.BytesIO()

with deflate.DeflateIO(stream, deflate.RAW, wbits) as d:

d.write(data)

compressed = stream.getvalue()

return compressed

def flush(self):

def __init__(self):

self._decomp = None

self.unconsumed_tail = b''

self.eof = False

def decompress(self, data):

stream = io.BytesIO(data)

try:

with deflate.DeflateIO(stream, deflate.RAW, 15) as d:

decompressed = d.read()

except EOFError:

return b''

except OSError:

return b''

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:

pass

elif compression == ZIP_BZIP2:

raise NotImplementedError("bzip2 not supported")

elif compression == ZIP_LZMA:

raise NotImplementedError("lzma not supported")

else:

if compress_type == ZIP_DEFLATED:

if deflate:

return DeflateCompressor(compresslevel)

else:

# zlib/CPython

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:

if deflate:

return DeflateDecompressor()

else:

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

def seekable(self):

file_has_seekable = hasattr(self._file, 'seekable')

if file_has_seekable:

# CPython

return self._file.seekable()

else:

# seekable not implemented on MicroPython, try to infer

return hasattr(self._file, 'seek')

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

super().__init__()

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