From 7796d3179b71536dd1d2ca7fdbc1255bdb8cfb52 Mon Sep 17 00:00:00 2001 From: "Jason R. Coombs" Date: Sat, 26 Nov 2022 09:44:13 -0500 Subject: gh-98098: Create packages from zipfile and test_zipfile (gh-98103) * gh-98098: Move zipfile into a package. * Moved test_zipfile to a package * Extracted module for test_path. * Add blurb * Add jaraco as owner of zipfile.Path. * Synchronize with minor changes found at jaraco/zipp@d9e7f4352d. --- .github/CODEOWNERS | 3 + Lib/test/test_zipfile.py | 3430 -------------------- Lib/test/test_zipfile/__init__.py | 5 + Lib/test/test_zipfile/test_core.py | 3014 +++++++++++++++++ Lib/test/test_zipfile/test_path.py | 423 +++ Lib/zipfile.py | 2566 --------------- Lib/zipfile/__init__.py | 2193 +++++++++++++ Lib/zipfile/__main__.py | 77 + Lib/zipfile/_path.py | 315 ++ .../2022-10-08-15-41-00.gh-issue-98098.DugpWi.rst | 2 + 10 files changed, 6032 insertions(+), 5996 deletions(-) delete mode 100644 Lib/test/test_zipfile.py create mode 100644 Lib/test/test_zipfile/__init__.py create mode 100644 Lib/test/test_zipfile/test_core.py create mode 100644 Lib/test/test_zipfile/test_path.py delete mode 100644 Lib/zipfile.py create mode 100644 Lib/zipfile/__init__.py create mode 100644 Lib/zipfile/__main__.py create mode 100644 Lib/zipfile/_path.py create mode 100644 Misc/NEWS.d/next/Library/2022-10-08-15-41-00.gh-issue-98098.DugpWi.rst diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS index 5f6d862..5d30c09 100644 --- a/.github/CODEOWNERS +++ b/.github/CODEOWNERS @@ -154,3 +154,6 @@ Lib/ast.py @isidentical # pathlib **/*pathlib* @brettcannon + +# zipfile.Path +**/*zipfile/*_path.py @jaraco diff --git a/Lib/test/test_zipfile.py b/Lib/test/test_zipfile.py deleted file mode 100644 index 6f6f4bc..0000000 --- a/Lib/test/test_zipfile.py +++ /dev/null @@ -1,3430 +0,0 @@ -import array -import contextlib -import importlib.util -import io -import itertools -import os -import pathlib -import posixpath -import string -import struct -import subprocess -import sys -import time -import unittest -import unittest.mock as mock -import zipfile -import functools - - -from tempfile import TemporaryFile -from random import randint, random, randbytes - -from test.support import script_helper -from test.support import ( - findfile, requires_zlib, requires_bz2, requires_lzma, - captured_stdout, captured_stderr, requires_subprocess -) -from test.support.os_helper import ( - TESTFN, unlink, rmtree, temp_dir, temp_cwd, fd_count -) - - -TESTFN2 = TESTFN + "2" -TESTFNDIR = TESTFN + "d" -FIXEDTEST_SIZE = 1000 -DATAFILES_DIR = 'zipfile_datafiles' - -SMALL_TEST_DATA = [('_ziptest1', '1q2w3e4r5t'), - ('ziptest2dir/_ziptest2', 'qawsedrftg'), - ('ziptest2dir/ziptest3dir/_ziptest3', 'azsxdcfvgb'), - ('ziptest2dir/ziptest3dir/ziptest4dir/_ziptest3', '6y7u8i9o0p')] - -def get_files(test): - yield TESTFN2 - with TemporaryFile() as f: - yield f - test.assertFalse(f.closed) - with io.BytesIO() as f: - yield f - test.assertFalse(f.closed) - -class AbstractTestsWithSourceFile: - @classmethod - def setUpClass(cls): - cls.line_gen = [bytes("Zipfile test line %d. random float: %f\n" % - (i, random()), "ascii") - for i in range(FIXEDTEST_SIZE)] - cls.data = b''.join(cls.line_gen) - - def setUp(self): - # Make a source file with some lines - with open(TESTFN, "wb") as fp: - fp.write(self.data) - - def make_test_archive(self, f, compression, compresslevel=None): - kwargs = {'compression': compression, 'compresslevel': compresslevel} - # Create the ZIP archive - with zipfile.ZipFile(f, "w", **kwargs) as zipfp: - zipfp.write(TESTFN, "another.name") - zipfp.write(TESTFN, TESTFN) - zipfp.writestr("strfile", self.data) - with zipfp.open('written-open-w', mode='w') as f: - for line in self.line_gen: - f.write(line) - - def zip_test(self, f, compression, compresslevel=None): - self.make_test_archive(f, compression, compresslevel) - - # Read the ZIP archive - with zipfile.ZipFile(f, "r", compression) as zipfp: - self.assertEqual(zipfp.read(TESTFN), self.data) - self.assertEqual(zipfp.read("another.name"), self.data) - self.assertEqual(zipfp.read("strfile"), self.data) - - # Print the ZIP directory - fp = io.StringIO() - zipfp.printdir(file=fp) - directory = fp.getvalue() - lines = directory.splitlines() - self.assertEqual(len(lines), 5) # Number of files + header - - self.assertIn('File Name', lines[0]) - self.assertIn('Modified', lines[0]) - self.assertIn('Size', lines[0]) - - fn, date, time_, size = lines[1].split() - self.assertEqual(fn, 'another.name') - self.assertTrue(time.strptime(date, '%Y-%m-%d')) - self.assertTrue(time.strptime(time_, '%H:%M:%S')) - self.assertEqual(size, str(len(self.data))) - - # Check the namelist - names = zipfp.namelist() - self.assertEqual(len(names), 4) - self.assertIn(TESTFN, names) - self.assertIn("another.name", names) - self.assertIn("strfile", names) - self.assertIn("written-open-w", names) - - # Check infolist - infos = zipfp.infolist() - names = [i.filename for i in infos] - self.assertEqual(len(names), 4) - self.assertIn(TESTFN, names) - self.assertIn("another.name", names) - self.assertIn("strfile", names) - self.assertIn("written-open-w", names) - for i in infos: - self.assertEqual(i.file_size, len(self.data)) - - # check getinfo - for nm in (TESTFN, "another.name", "strfile", "written-open-w"): - info = zipfp.getinfo(nm) - self.assertEqual(info.filename, nm) - self.assertEqual(info.file_size, len(self.data)) - - # Check that testzip thinks the archive is ok - # (it returns None if all contents could be read properly) - self.assertIsNone(zipfp.testzip()) - - def test_basic(self): - for f in get_files(self): - self.zip_test(f, self.compression) - - def zip_open_test(self, f, compression): - self.make_test_archive(f, compression) - - # Read the ZIP archive - with zipfile.ZipFile(f, "r", compression) as zipfp: - zipdata1 = [] - with zipfp.open(TESTFN) as zipopen1: - while True: - read_data = zipopen1.read(256) - if not read_data: - break - zipdata1.append(read_data) - - zipdata2 = [] - with zipfp.open("another.name") as zipopen2: - while True: - read_data = zipopen2.read(256) - if not read_data: - break - zipdata2.append(read_data) - - self.assertEqual(b''.join(zipdata1), self.data) - self.assertEqual(b''.join(zipdata2), self.data) - - def test_open(self): - for f in get_files(self): - self.zip_open_test(f, self.compression) - - def test_open_with_pathlike(self): - path = pathlib.Path(TESTFN2) - self.zip_open_test(path, self.compression) - with zipfile.ZipFile(path, "r", self.compression) as zipfp: - self.assertIsInstance(zipfp.filename, str) - - def zip_random_open_test(self, f, compression): - self.make_test_archive(f, compression) - - # Read the ZIP archive - with zipfile.ZipFile(f, "r", compression) as zipfp: - zipdata1 = [] - with zipfp.open(TESTFN) as zipopen1: - while True: - read_data = zipopen1.read(randint(1, 1024)) - if not read_data: - break - zipdata1.append(read_data) - - self.assertEqual(b''.join(zipdata1), self.data) - - def test_random_open(self): - for f in get_files(self): - self.zip_random_open_test(f, self.compression) - - def zip_read1_test(self, f, compression): - self.make_test_archive(f, compression) - - # Read the ZIP archive - with zipfile.ZipFile(f, "r") as zipfp, \ - zipfp.open(TESTFN) as zipopen: - zipdata = [] - while True: - read_data = zipopen.read1(-1) - if not read_data: - break - zipdata.append(read_data) - - self.assertEqual(b''.join(zipdata), self.data) - - def test_read1(self): - for f in get_files(self): - self.zip_read1_test(f, self.compression) - - def zip_read1_10_test(self, f, compression): - self.make_test_archive(f, compression) - - # Read the ZIP archive - with zipfile.ZipFile(f, "r") as zipfp, \ - zipfp.open(TESTFN) as zipopen: - zipdata = [] - while True: - read_data = zipopen.read1(10) - self.assertLessEqual(len(read_data), 10) - if not read_data: - break - zipdata.append(read_data) - - self.assertEqual(b''.join(zipdata), self.data) - - def test_read1_10(self): - for f in get_files(self): - self.zip_read1_10_test(f, self.compression) - - def zip_readline_read_test(self, f, compression): - self.make_test_archive(f, compression) - - # Read the ZIP archive - with zipfile.ZipFile(f, "r") as zipfp, \ - zipfp.open(TESTFN) as zipopen: - data = b'' - while True: - read = zipopen.readline() - if not read: - break - data += read - - read = zipopen.read(100) - if not read: - break - data += read - - self.assertEqual(data, self.data) - - def test_readline_read(self): - # Issue #7610: calls to readline() interleaved with calls to read(). - for f in get_files(self): - self.zip_readline_read_test(f, self.compression) - - def zip_readline_test(self, f, compression): - self.make_test_archive(f, compression) - - # Read the ZIP archive - with zipfile.ZipFile(f, "r") as zipfp: - with zipfp.open(TESTFN) as zipopen: - for line in self.line_gen: - linedata = zipopen.readline() - self.assertEqual(linedata, line) - - def test_readline(self): - for f in get_files(self): - self.zip_readline_test(f, self.compression) - - def zip_readlines_test(self, f, compression): - self.make_test_archive(f, compression) - - # Read the ZIP archive - with zipfile.ZipFile(f, "r") as zipfp: - with zipfp.open(TESTFN) as zipopen: - ziplines = zipopen.readlines() - for line, zipline in zip(self.line_gen, ziplines): - self.assertEqual(zipline, line) - - def test_readlines(self): - for f in get_files(self): - self.zip_readlines_test(f, self.compression) - - def zip_iterlines_test(self, f, compression): - self.make_test_archive(f, compression) - - # Read the ZIP archive - with zipfile.ZipFile(f, "r") as zipfp: - with zipfp.open(TESTFN) as zipopen: - for line, zipline in zip(self.line_gen, zipopen): - self.assertEqual(zipline, line) - - def test_iterlines(self): - for f in get_files(self): - self.zip_iterlines_test(f, self.compression) - - def test_low_compression(self): - """Check for cases where compressed data is larger than original.""" - # Create the ZIP archive - with zipfile.ZipFile(TESTFN2, "w", self.compression) as zipfp: - zipfp.writestr("strfile", '12') - - # Get an open object for strfile - with zipfile.ZipFile(TESTFN2, "r", self.compression) as zipfp: - with zipfp.open("strfile") as openobj: - self.assertEqual(openobj.read(1), b'1') - self.assertEqual(openobj.read(1), b'2') - - def test_writestr_compression(self): - zipfp = zipfile.ZipFile(TESTFN2, "w") - zipfp.writestr("b.txt", "hello world", compress_type=self.compression) - info = zipfp.getinfo('b.txt') - self.assertEqual(info.compress_type, self.compression) - - def test_writestr_compresslevel(self): - zipfp = zipfile.ZipFile(TESTFN2, "w", compresslevel=1) - zipfp.writestr("a.txt", "hello world", compress_type=self.compression) - zipfp.writestr("b.txt", "hello world", compress_type=self.compression, - compresslevel=2) - - # Compression level follows the constructor. - a_info = zipfp.getinfo('a.txt') - self.assertEqual(a_info.compress_type, self.compression) - self.assertEqual(a_info._compresslevel, 1) - - # Compression level is overridden. - b_info = zipfp.getinfo('b.txt') - self.assertEqual(b_info.compress_type, self.compression) - self.assertEqual(b_info._compresslevel, 2) - - def test_read_return_size(self): - # Issue #9837: ZipExtFile.read() shouldn't return more bytes - # than requested. - for test_size in (1, 4095, 4096, 4097, 16384): - file_size = test_size + 1 - junk = randbytes(file_size) - with zipfile.ZipFile(io.BytesIO(), "w", self.compression) as zipf: - zipf.writestr('foo', junk) - with zipf.open('foo', 'r') as fp: - buf = fp.read(test_size) - self.assertEqual(len(buf), test_size) - - def test_truncated_zipfile(self): - fp = io.BytesIO() - with zipfile.ZipFile(fp, mode='w') as zipf: - zipf.writestr('strfile', self.data, compress_type=self.compression) - end_offset = fp.tell() - zipfiledata = fp.getvalue() - - fp = io.BytesIO(zipfiledata) - with zipfile.ZipFile(fp) as zipf: - with zipf.open('strfile') as zipopen: - fp.truncate(end_offset - 20) - with self.assertRaises(EOFError): - zipopen.read() - - fp = io.BytesIO(zipfiledata) - with zipfile.ZipFile(fp) as zipf: - with zipf.open('strfile') as zipopen: - fp.truncate(end_offset - 20) - with self.assertRaises(EOFError): - while zipopen.read(100): - pass - - fp = io.BytesIO(zipfiledata) - with zipfile.ZipFile(fp) as zipf: - with zipf.open('strfile') as zipopen: - fp.truncate(end_offset - 20) - with self.assertRaises(EOFError): - while zipopen.read1(100): - pass - - def test_repr(self): - fname = 'file.name' - for f in get_files(self): - with zipfile.ZipFile(f, 'w', self.compression) as zipfp: - zipfp.write(TESTFN, fname) - r = repr(zipfp) - self.assertIn("mode='w'", r) - - with zipfile.ZipFile(f, 'r') as zipfp: - r = repr(zipfp) - if isinstance(f, str): - self.assertIn('filename=%r' % f, r) - else: - self.assertIn('file=%r' % f, r) - self.assertIn("mode='r'", r) - r = repr(zipfp.getinfo(fname)) - self.assertIn('filename=%r' % fname, r) - self.assertIn('filemode=', r) - self.assertIn('file_size=', r) - if self.compression != zipfile.ZIP_STORED: - self.assertIn('compress_type=', r) - self.assertIn('compress_size=', r) - with zipfp.open(fname) as zipopen: - r = repr(zipopen) - self.assertIn('name=%r' % fname, r) - self.assertIn("mode='r'", r) - if self.compression != zipfile.ZIP_STORED: - self.assertIn('compress_type=', r) - self.assertIn('[closed]', repr(zipopen)) - self.assertIn('[closed]', repr(zipfp)) - - def test_compresslevel_basic(self): - for f in get_files(self): - self.zip_test(f, self.compression, compresslevel=9) - - def test_per_file_compresslevel(self): - """Check that files within a Zip archive can have different - compression levels.""" - with zipfile.ZipFile(TESTFN2, "w", compresslevel=1) as zipfp: - zipfp.write(TESTFN, 'compress_1') - zipfp.write(TESTFN, 'compress_9', compresslevel=9) - one_info = zipfp.getinfo('compress_1') - nine_info = zipfp.getinfo('compress_9') - self.assertEqual(one_info._compresslevel, 1) - self.assertEqual(nine_info._compresslevel, 9) - - def test_writing_errors(self): - class BrokenFile(io.BytesIO): - def write(self, data): - nonlocal count - if count is not None: - if count == stop: - raise OSError - count += 1 - super().write(data) - - stop = 0 - while True: - testfile = BrokenFile() - count = None - with zipfile.ZipFile(testfile, 'w', self.compression) as zipfp: - with zipfp.open('file1', 'w') as f: - f.write(b'data1') - count = 0 - try: - with zipfp.open('file2', 'w') as f: - f.write(b'data2') - except OSError: - stop += 1 - else: - break - finally: - count = None - with zipfile.ZipFile(io.BytesIO(testfile.getvalue())) as zipfp: - self.assertEqual(zipfp.namelist(), ['file1']) - self.assertEqual(zipfp.read('file1'), b'data1') - - with zipfile.ZipFile(io.BytesIO(testfile.getvalue())) as zipfp: - self.assertEqual(zipfp.namelist(), ['file1', 'file2']) - self.assertEqual(zipfp.read('file1'), b'data1') - self.assertEqual(zipfp.read('file2'), b'data2') - - - def tearDown(self): - unlink(TESTFN) - unlink(TESTFN2) - - -class StoredTestsWithSourceFile(AbstractTestsWithSourceFile, - unittest.TestCase): - compression = zipfile.ZIP_STORED - test_low_compression = None - - def zip_test_writestr_permissions(self, f, compression): - # Make sure that writestr and open(... mode='w') create files with - # mode 0600, when they are passed a name rather than a ZipInfo - # instance. - - self.make_test_archive(f, compression) - with zipfile.ZipFile(f, "r") as zipfp: - zinfo = zipfp.getinfo('strfile') - self.assertEqual(zinfo.external_attr, 0o600 << 16) - - zinfo2 = zipfp.getinfo('written-open-w') - self.assertEqual(zinfo2.external_attr, 0o600 << 16) - - def test_writestr_permissions(self): - for f in get_files(self): - self.zip_test_writestr_permissions(f, zipfile.ZIP_STORED) - - def test_absolute_arcnames(self): - with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp: - zipfp.write(TESTFN, "/absolute") - - with zipfile.ZipFile(TESTFN2, "r", zipfile.ZIP_STORED) as zipfp: - self.assertEqual(zipfp.namelist(), ["absolute"]) - - def test_append_to_zip_file(self): - """Test appending to an existing zipfile.""" - with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp: - zipfp.write(TESTFN, TESTFN) - - with zipfile.ZipFile(TESTFN2, "a", zipfile.ZIP_STORED) as zipfp: - zipfp.writestr("strfile", self.data) - self.assertEqual(zipfp.namelist(), [TESTFN, "strfile"]) - - def test_append_to_non_zip_file(self): - """Test appending to an existing file that is not a zipfile.""" - # NOTE: this test fails if len(d) < 22 because of the first - # line "fpin.seek(-22, 2)" in _EndRecData - data = b'I am not a ZipFile!'*10 - with open(TESTFN2, 'wb') as f: - f.write(data) - - with zipfile.ZipFile(TESTFN2, "a", zipfile.ZIP_STORED) as zipfp: - zipfp.write(TESTFN, TESTFN) - - with open(TESTFN2, 'rb') as f: - f.seek(len(data)) - with zipfile.ZipFile(f, "r") as zipfp: - self.assertEqual(zipfp.namelist(), [TESTFN]) - self.assertEqual(zipfp.read(TESTFN), self.data) - with open(TESTFN2, 'rb') as f: - self.assertEqual(f.read(len(data)), data) - zipfiledata = f.read() - with io.BytesIO(zipfiledata) as bio, zipfile.ZipFile(bio) as zipfp: - self.assertEqual(zipfp.namelist(), [TESTFN]) - self.assertEqual(zipfp.read(TESTFN), self.data) - - def test_read_concatenated_zip_file(self): - with io.BytesIO() as bio: - with zipfile.ZipFile(bio, 'w', zipfile.ZIP_STORED) as zipfp: - zipfp.write(TESTFN, TESTFN) - zipfiledata = bio.getvalue() - data = b'I am not a ZipFile!'*10 - with open(TESTFN2, 'wb') as f: - f.write(data) - f.write(zipfiledata) - - with zipfile.ZipFile(TESTFN2) as zipfp: - self.assertEqual(zipfp.namelist(), [TESTFN]) - self.assertEqual(zipfp.read(TESTFN), self.data) - - def test_append_to_concatenated_zip_file(self): - with io.BytesIO() as bio: - with zipfile.ZipFile(bio, 'w', zipfile.ZIP_STORED) as zipfp: - zipfp.write(TESTFN, TESTFN) - zipfiledata = bio.getvalue() - data = b'I am not a ZipFile!'*1000000 - with open(TESTFN2, 'wb') as f: - f.write(data) - f.write(zipfiledata) - - with zipfile.ZipFile(TESTFN2, 'a') as zipfp: - self.assertEqual(zipfp.namelist(), [TESTFN]) - zipfp.writestr('strfile', self.data) - - with open(TESTFN2, 'rb') as f: - self.assertEqual(f.read(len(data)), data) - zipfiledata = f.read() - with io.BytesIO(zipfiledata) as bio, zipfile.ZipFile(bio) as zipfp: - self.assertEqual(zipfp.namelist(), [TESTFN, 'strfile']) - self.assertEqual(zipfp.read(TESTFN), self.data) - self.assertEqual(zipfp.read('strfile'), self.data) - - def test_ignores_newline_at_end(self): - with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp: - zipfp.write(TESTFN, TESTFN) - with open(TESTFN2, 'a', encoding='utf-8') as f: - f.write("\r\n\00\00\00") - with zipfile.ZipFile(TESTFN2, "r") as zipfp: - self.assertIsInstance(zipfp, zipfile.ZipFile) - - def test_ignores_stuff_appended_past_comments(self): - with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp: - zipfp.comment = b"this is a comment" - zipfp.write(TESTFN, TESTFN) - with open(TESTFN2, 'a', encoding='utf-8') as f: - f.write("abcdef\r\n") - with zipfile.ZipFile(TESTFN2, "r") as zipfp: - self.assertIsInstance(zipfp, zipfile.ZipFile) - self.assertEqual(zipfp.comment, b"this is a comment") - - def test_write_default_name(self): - """Check that calling ZipFile.write without arcname specified - produces the expected result.""" - with zipfile.ZipFile(TESTFN2, "w") as zipfp: - zipfp.write(TESTFN) - with open(TESTFN, "rb") as f: - self.assertEqual(zipfp.read(TESTFN), f.read()) - - def test_io_on_closed_zipextfile(self): - fname = "somefile.txt" - with zipfile.ZipFile(TESTFN2, mode="w") as zipfp: - zipfp.writestr(fname, "bogus") - - with zipfile.ZipFile(TESTFN2, mode="r") as zipfp: - with zipfp.open(fname) as fid: - fid.close() - self.assertRaises(ValueError, fid.read) - self.assertRaises(ValueError, fid.seek, 0) - self.assertRaises(ValueError, fid.tell) - self.assertRaises(ValueError, fid.readable) - self.assertRaises(ValueError, fid.seekable) - - def test_write_to_readonly(self): - """Check that trying to call write() on a readonly ZipFile object - raises a ValueError.""" - with zipfile.ZipFile(TESTFN2, mode="w") as zipfp: - zipfp.writestr("somefile.txt", "bogus") - - with zipfile.ZipFile(TESTFN2, mode="r") as zipfp: - self.assertRaises(ValueError, zipfp.write, TESTFN) - - with zipfile.ZipFile(TESTFN2, mode="r") as zipfp: - with self.assertRaises(ValueError): - zipfp.open(TESTFN, mode='w') - - def test_add_file_before_1980(self): - # Set atime and mtime to 1970-01-01 - os.utime(TESTFN, (0, 0)) - with zipfile.ZipFile(TESTFN2, "w") as zipfp: - self.assertRaises(ValueError, zipfp.write, TESTFN) - - with zipfile.ZipFile(TESTFN2, "w", strict_timestamps=False) as zipfp: - zipfp.write(TESTFN) - zinfo = zipfp.getinfo(TESTFN) - self.assertEqual(zinfo.date_time, (1980, 1, 1, 0, 0, 0)) - - def test_add_file_after_2107(self): - # Set atime and mtime to 2108-12-30 - ts = 4386268800 - try: - time.localtime(ts) - except OverflowError: - self.skipTest(f'time.localtime({ts}) raises OverflowError') - try: - os.utime(TESTFN, (ts, ts)) - except OverflowError: - self.skipTest('Host fs cannot set timestamp to required value.') - - mtime_ns = os.stat(TESTFN).st_mtime_ns - if mtime_ns != (4386268800 * 10**9): - # XFS filesystem is limited to 32-bit timestamp, but the syscall - # didn't fail. Moreover, there is a VFS bug which returns - # a cached timestamp which is different than the value on disk. - # - # Test st_mtime_ns rather than st_mtime to avoid rounding issues. - # - # https://bugzilla.redhat.com/show_bug.cgi?id=1795576 - # https://bugs.python.org/issue39460#msg360952 - self.skipTest(f"Linux VFS/XFS kernel bug detected: {mtime_ns=}") - - with zipfile.ZipFile(TESTFN2, "w") as zipfp: - self.assertRaises(struct.error, zipfp.write, TESTFN) - - with zipfile.ZipFile(TESTFN2, "w", strict_timestamps=False) as zipfp: - zipfp.write(TESTFN) - zinfo = zipfp.getinfo(TESTFN) - self.assertEqual(zinfo.date_time, (2107, 12, 31, 23, 59, 59)) - - -@requires_zlib() -class DeflateTestsWithSourceFile(AbstractTestsWithSourceFile, - unittest.TestCase): - compression = zipfile.ZIP_DEFLATED - - def test_per_file_compression(self): - """Check that files within a Zip archive can have different - compression options.""" - with zipfile.ZipFile(TESTFN2, "w") as zipfp: - zipfp.write(TESTFN, 'storeme', zipfile.ZIP_STORED) - zipfp.write(TESTFN, 'deflateme', zipfile.ZIP_DEFLATED) - sinfo = zipfp.getinfo('storeme') - dinfo = zipfp.getinfo('deflateme') - self.assertEqual(sinfo.compress_type, zipfile.ZIP_STORED) - self.assertEqual(dinfo.compress_type, zipfile.ZIP_DEFLATED) - -@requires_bz2() -class Bzip2TestsWithSourceFile(AbstractTestsWithSourceFile, - unittest.TestCase): - compression = zipfile.ZIP_BZIP2 - -@requires_lzma() -class LzmaTestsWithSourceFile(AbstractTestsWithSourceFile, - unittest.TestCase): - compression = zipfile.ZIP_LZMA - - -class AbstractTestZip64InSmallFiles: - # These tests test the ZIP64 functionality without using large files, - # see test_zipfile64 for proper tests. - - @classmethod - def setUpClass(cls): - line_gen = (bytes("Test of zipfile line %d." % i, "ascii") - for i in range(0, FIXEDTEST_SIZE)) - cls.data = b'\n'.join(line_gen) - - def setUp(self): - self._limit = zipfile.ZIP64_LIMIT - self._filecount_limit = zipfile.ZIP_FILECOUNT_LIMIT - zipfile.ZIP64_LIMIT = 1000 - zipfile.ZIP_FILECOUNT_LIMIT = 9 - - # Make a source file with some lines - with open(TESTFN, "wb") as fp: - fp.write(self.data) - - def zip_test(self, f, compression): - # Create the ZIP archive - with zipfile.ZipFile(f, "w", compression, allowZip64=True) as zipfp: - zipfp.write(TESTFN, "another.name") - zipfp.write(TESTFN, TESTFN) - zipfp.writestr("strfile", self.data) - - # Read the ZIP archive - with zipfile.ZipFile(f, "r", compression) as zipfp: - self.assertEqual(zipfp.read(TESTFN), self.data) - self.assertEqual(zipfp.read("another.name"), self.data) - self.assertEqual(zipfp.read("strfile"), self.data) - - # Print the ZIP directory - fp = io.StringIO() - zipfp.printdir(fp) - - directory = fp.getvalue() - lines = directory.splitlines() - self.assertEqual(len(lines), 4) # Number of files + header - - self.assertIn('File Name', lines[0]) - self.assertIn('Modified', lines[0]) - self.assertIn('Size', lines[0]) - - fn, date, time_, size = lines[1].split() - self.assertEqual(fn, 'another.name') - self.assertTrue(time.strptime(date, '%Y-%m-%d')) - self.assertTrue(time.strptime(time_, '%H:%M:%S')) - self.assertEqual(size, str(len(self.data))) - - # Check the namelist - names = zipfp.namelist() - self.assertEqual(len(names), 3) - self.assertIn(TESTFN, names) - self.assertIn("another.name", names) - self.assertIn("strfile", names) - - # Check infolist - infos = zipfp.infolist() - names = [i.filename for i in infos] - self.assertEqual(len(names), 3) - self.assertIn(TESTFN, names) - self.assertIn("another.name", names) - self.assertIn("strfile", names) - for i in infos: - self.assertEqual(i.file_size, len(self.data)) - - # check getinfo - for nm in (TESTFN, "another.name", "strfile"): - info = zipfp.getinfo(nm) - self.assertEqual(info.filename, nm) - self.assertEqual(info.file_size, len(self.data)) - - # Check that testzip thinks the archive is valid - self.assertIsNone(zipfp.testzip()) - - def test_basic(self): - for f in get_files(self): - self.zip_test(f, self.compression) - - def test_too_many_files(self): - # This test checks that more than 64k files can be added to an archive, - # and that the resulting archive can be read properly by ZipFile - zipf = zipfile.ZipFile(TESTFN, "w", self.compression, - allowZip64=True) - zipf.debug = 100 - numfiles = 15 - for i in range(numfiles): - zipf.writestr("foo%08d" % i, "%d" % (i**3 % 57)) - self.assertEqual(len(zipf.namelist()), numfiles) - zipf.close() - - zipf2 = zipfile.ZipFile(TESTFN, "r", self.compression) - self.assertEqual(len(zipf2.namelist()), numfiles) - for i in range(numfiles): - content = zipf2.read("foo%08d" % i).decode('ascii') - self.assertEqual(content, "%d" % (i**3 % 57)) - zipf2.close() - - def test_too_many_files_append(self): - zipf = zipfile.ZipFile(TESTFN, "w", self.compression, - allowZip64=False) - zipf.debug = 100 - numfiles = 9 - for i in range(numfiles): - zipf.writestr("foo%08d" % i, "%d" % (i**3 % 57)) - self.assertEqual(len(zipf.namelist()), numfiles) - with self.assertRaises(zipfile.LargeZipFile): - zipf.writestr("foo%08d" % numfiles, b'') - self.assertEqual(len(zipf.namelist()), numfiles) - zipf.close() - - zipf = zipfile.ZipFile(TESTFN, "a", self.compression, - allowZip64=False) - zipf.debug = 100 - self.assertEqual(len(zipf.namelist()), numfiles) - with self.assertRaises(zipfile.LargeZipFile): - zipf.writestr("foo%08d" % numfiles, b'') - self.assertEqual(len(zipf.namelist()), numfiles) - zipf.close() - - zipf = zipfile.ZipFile(TESTFN, "a", self.compression, - allowZip64=True) - zipf.debug = 100 - self.assertEqual(len(zipf.namelist()), numfiles) - numfiles2 = 15 - for i in range(numfiles, numfiles2): - zipf.writestr("foo%08d" % i, "%d" % (i**3 % 57)) - self.assertEqual(len(zipf.namelist()), numfiles2) - zipf.close() - - zipf2 = zipfile.ZipFile(TESTFN, "r", self.compression) - self.assertEqual(len(zipf2.namelist()), numfiles2) - for i in range(numfiles2): - content = zipf2.read("foo%08d" % i).decode('ascii') - self.assertEqual(content, "%d" % (i**3 % 57)) - zipf2.close() - - def tearDown(self): - zipfile.ZIP64_LIMIT = self._limit - zipfile.ZIP_FILECOUNT_LIMIT = self._filecount_limit - unlink(TESTFN) - unlink(TESTFN2) - - -class StoredTestZip64InSmallFiles(AbstractTestZip64InSmallFiles, - unittest.TestCase): - compression = zipfile.ZIP_STORED - - def large_file_exception_test(self, f, compression): - with zipfile.ZipFile(f, "w", compression, allowZip64=False) as zipfp: - self.assertRaises(zipfile.LargeZipFile, - zipfp.write, TESTFN, "another.name") - - def large_file_exception_test2(self, f, compression): - with zipfile.ZipFile(f, "w", compression, allowZip64=False) as zipfp: - self.assertRaises(zipfile.LargeZipFile, - zipfp.writestr, "another.name", self.data) - - def test_large_file_exception(self): - for f in get_files(self): - self.large_file_exception_test(f, zipfile.ZIP_STORED) - self.large_file_exception_test2(f, zipfile.ZIP_STORED) - - def test_absolute_arcnames(self): - with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED, - allowZip64=True) as zipfp: - zipfp.write(TESTFN, "/absolute") - - with zipfile.ZipFile(TESTFN2, "r", zipfile.ZIP_STORED) as zipfp: - self.assertEqual(zipfp.namelist(), ["absolute"]) - - def test_append(self): - # Test that appending to the Zip64 archive doesn't change - # extra fields of existing entries. - with zipfile.ZipFile(TESTFN2, "w", allowZip64=True) as zipfp: - zipfp.writestr("strfile", self.data) - with zipfile.ZipFile(TESTFN2, "r", allowZip64=True) as zipfp: - zinfo = zipfp.getinfo("strfile") - extra = zinfo.extra - with zipfile.ZipFile(TESTFN2, "a", allowZip64=True) as zipfp: - zipfp.writestr("strfile2", self.data) - with zipfile.ZipFile(TESTFN2, "r", allowZip64=True) as zipfp: - zinfo = zipfp.getinfo("strfile") - self.assertEqual(zinfo.extra, extra) - - def make_zip64_file( - self, file_size_64_set=False, file_size_extra=False, - compress_size_64_set=False, compress_size_extra=False, - header_offset_64_set=False, header_offset_extra=False, - ): - """Generate bytes sequence for a zip with (incomplete) zip64 data. - - The actual values (not the zip 64 0xffffffff values) stored in the file - are: - file_size: 8 - compress_size: 8 - header_offset: 0 - """ - actual_size = 8 - actual_header_offset = 0 - local_zip64_fields = [] - central_zip64_fields = [] - - file_size = actual_size - if file_size_64_set: - file_size = 0xffffffff - if file_size_extra: - local_zip64_fields.append(actual_size) - central_zip64_fields.append(actual_size) - file_size = struct.pack("e|f"g?h*i', ','), r'a\b,c_d_e_f_g_h_i') - self.assertEqual(san('../../foo../../ba..r', '/'), r'foo/ba..r') - self.assertEqual(san(' / /foo / /ba r', '/'), r'foo/ba r') - self.assertEqual(san(' . /. /foo ./ . /. ./ba .r', '/'), r'foo/ba .r') - - def test_extract_hackers_arcnames_common_cases(self): - common_hacknames = [ - ('../foo/bar', 'foo/bar'), - ('foo/../bar', 'foo/bar'), - ('foo/../../bar', 'foo/bar'), - ('foo/bar/..', 'foo/bar'), - ('./../foo/bar', 'foo/bar'), - ('/foo/bar', 'foo/bar'), - ('/foo/../bar', 'foo/bar'), - ('/foo/../../bar', 'foo/bar'), - ] - self._test_extract_hackers_arcnames(common_hacknames) - - @unittest.skipIf(os.path.sep != '\\', 'Requires \\ as path separator.') - def test_extract_hackers_arcnames_windows_only(self): - """Test combination of path fixing and windows name sanitization.""" - windows_hacknames = [ - (r'..\foo\bar', 'foo/bar'), - (r'..\/foo\/bar', 'foo/bar'), - (r'foo/\..\/bar', 'foo/bar'), - (r'foo\/../\bar', 'foo/bar'), - (r'C:foo/bar', 'foo/bar'), - (r'C:/foo/bar', 'foo/bar'), - (r'C://foo/bar', 'foo/bar'), - (r'C:\foo\bar', 'foo/bar'), - (r'//conky/mountpoint/foo/bar', 'foo/bar'), - (r'\\conky\mountpoint\foo\bar', 'foo/bar'), - (r'///conky/mountpoint/foo/bar', 'conky/mountpoint/foo/bar'), - (r'\\\conky\mountpoint\foo\bar', 'conky/mountpoint/foo/bar'), - (r'//conky//mountpoint/foo/bar', 'conky/mountpoint/foo/bar'), - (r'\\conky\\mountpoint\foo\bar', 'conky/mountpoint/foo/bar'), - (r'//?/C:/foo/bar', 'foo/bar'), - (r'\\?\C:\foo\bar', 'foo/bar'), - (r'C:/../C:/foo/bar', 'C_/foo/bar'), - (r'a:b\ce|f"g?h*i', 'b/c_d_e_f_g_h_i'), - ('../../foo../../ba..r', 'foo/ba..r'), - ] - self._test_extract_hackers_arcnames(windows_hacknames) - - @unittest.skipIf(os.path.sep != '/', r'Requires / as path separator.') - def test_extract_hackers_arcnames_posix_only(self): - posix_hacknames = [ - ('//foo/bar', 'foo/bar'), - ('../../foo../../ba..r', 'foo../ba..r'), - (r'foo/..\bar', r'foo/..\bar'), - ] - self._test_extract_hackers_arcnames(posix_hacknames) - - def _test_extract_hackers_arcnames(self, hacknames): - for arcname, fixedname in hacknames: - content = b'foobar' + arcname.encode() - with zipfile.ZipFile(TESTFN2, 'w', zipfile.ZIP_STORED) as zipfp: - zinfo = zipfile.ZipInfo() - # preserve backslashes - zinfo.filename = arcname - zinfo.external_attr = 0o600 << 16 - zipfp.writestr(zinfo, content) - - arcname = arcname.replace(os.sep, "/") - targetpath = os.path.join('target', 'subdir', 'subsub') - correctfile = os.path.join(targetpath, *fixedname.split('/')) - - with zipfile.ZipFile(TESTFN2, 'r') as zipfp: - writtenfile = zipfp.extract(arcname, targetpath) - self.assertEqual(writtenfile, correctfile, - msg='extract %r: %r != %r' % - (arcname, writtenfile, correctfile)) - self.check_file(correctfile, content) - rmtree('target') - - with zipfile.ZipFile(TESTFN2, 'r') as zipfp: - zipfp.extractall(targetpath) - self.check_file(correctfile, content) - rmtree('target') - - correctfile = os.path.join(os.getcwd(), *fixedname.split('/')) - - with zipfile.ZipFile(TESTFN2, 'r') as zipfp: - writtenfile = zipfp.extract(arcname) - self.assertEqual(writtenfile, correctfile, - msg="extract %r" % arcname) - self.check_file(correctfile, content) - rmtree(fixedname.split('/')[0]) - - with zipfile.ZipFile(TESTFN2, 'r') as zipfp: - zipfp.extractall() - self.check_file(correctfile, content) - rmtree(fixedname.split('/')[0]) - - unlink(TESTFN2) - - -class OtherTests(unittest.TestCase): - def test_open_via_zip_info(self): - # Create the ZIP archive - with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp: - zipfp.writestr("name", "foo") - with self.assertWarns(UserWarning): - zipfp.writestr("name", "bar") - self.assertEqual(zipfp.namelist(), ["name"] * 2) - - with zipfile.ZipFile(TESTFN2, "r") as zipfp: - infos = zipfp.infolist() - data = b"" - for info in infos: - with zipfp.open(info) as zipopen: - data += zipopen.read() - self.assertIn(data, {b"foobar", b"barfoo"}) - data = b"" - for info in infos: - data += zipfp.read(info) - self.assertIn(data, {b"foobar", b"barfoo"}) - - def test_writestr_extended_local_header_issue1202(self): - with zipfile.ZipFile(TESTFN2, 'w') as orig_zip: - for data in 'abcdefghijklmnop': - zinfo = zipfile.ZipInfo(data) - zinfo.flag_bits |= zipfile._MASK_USE_DATA_DESCRIPTOR # Include an extended local header. - orig_zip.writestr(zinfo, data) - - def test_close(self): - """Check that the zipfile is closed after the 'with' block.""" - with zipfile.ZipFile(TESTFN2, "w") as zipfp: - for fpath, fdata in SMALL_TEST_DATA: - zipfp.writestr(fpath, fdata) - self.assertIsNotNone(zipfp.fp, 'zipfp is not open') - self.assertIsNone(zipfp.fp, 'zipfp is not closed') - - with zipfile.ZipFile(TESTFN2, "r") as zipfp: - self.assertIsNotNone(zipfp.fp, 'zipfp is not open') - self.assertIsNone(zipfp.fp, 'zipfp is not closed') - - def test_close_on_exception(self): - """Check that the zipfile is closed if an exception is raised in the - 'with' block.""" - with zipfile.ZipFile(TESTFN2, "w") as zipfp: - for fpath, fdata in SMALL_TEST_DATA: - zipfp.writestr(fpath, fdata) - - try: - with zipfile.ZipFile(TESTFN2, "r") as zipfp2: - raise zipfile.BadZipFile() - except zipfile.BadZipFile: - self.assertIsNone(zipfp2.fp, 'zipfp is not closed') - - def test_unsupported_version(self): - # File has an extract_version of 120 - data = (b'PK\x03\x04x\x00\x00\x00\x00\x00!p\xa1@\x00\x00\x00\x00\x00\x00' - b'\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00xPK\x01\x02x\x03x\x00\x00\x00\x00' - b'\x00!p\xa1@\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00' - b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x80\x01\x00\x00\x00\x00xPK\x05\x06' - b'\x00\x00\x00\x00\x01\x00\x01\x00/\x00\x00\x00\x1f\x00\x00\x00\x00\x00') - - self.assertRaises(NotImplementedError, zipfile.ZipFile, - io.BytesIO(data), 'r') - - @requires_zlib() - def test_read_unicode_filenames(self): - # bug #10801 - fname = findfile('zip_cp437_header.zip') - with zipfile.ZipFile(fname) as zipfp: - for name in zipfp.namelist(): - zipfp.open(name).close() - - def test_write_unicode_filenames(self): - with zipfile.ZipFile(TESTFN, "w") as zf: - zf.writestr("foo.txt", "Test for unicode filename") - zf.writestr("\xf6.txt", "Test for unicode filename") - self.assertIsInstance(zf.infolist()[0].filename, str) - - with zipfile.ZipFile(TESTFN, "r") as zf: - self.assertEqual(zf.filelist[0].filename, "foo.txt") - self.assertEqual(zf.filelist[1].filename, "\xf6.txt") - - def test_read_after_write_unicode_filenames(self): - with zipfile.ZipFile(TESTFN2, 'w') as zipfp: - zipfp.writestr('приклад', b'sample') - self.assertEqual(zipfp.read('приклад'), b'sample') - - def test_exclusive_create_zip_file(self): - """Test exclusive creating a new zipfile.""" - unlink(TESTFN2) - filename = 'testfile.txt' - content = b'hello, world. this is some content.' - with zipfile.ZipFile(TESTFN2, "x", zipfile.ZIP_STORED) as zipfp: - zipfp.writestr(filename, content) - with self.assertRaises(FileExistsError): - zipfile.ZipFile(TESTFN2, "x", zipfile.ZIP_STORED) - with zipfile.ZipFile(TESTFN2, "r") as zipfp: - self.assertEqual(zipfp.namelist(), [filename]) - self.assertEqual(zipfp.read(filename), content) - - def test_create_non_existent_file_for_append(self): - if os.path.exists(TESTFN): - os.unlink(TESTFN) - - filename = 'testfile.txt' - content = b'hello, world. this is some content.' - - try: - with zipfile.ZipFile(TESTFN, 'a') as zf: - zf.writestr(filename, content) - except OSError: - self.fail('Could not append data to a non-existent zip file.') - - self.assertTrue(os.path.exists(TESTFN)) - - with zipfile.ZipFile(TESTFN, 'r') as zf: - self.assertEqual(zf.read(filename), content) - - def test_close_erroneous_file(self): - # This test checks that the ZipFile constructor closes the file object - # it opens if there's an error in the file. If it doesn't, the - # traceback holds a reference to the ZipFile object and, indirectly, - # the file object. - # On Windows, this causes the os.unlink() call to fail because the - # underlying file is still open. This is SF bug #412214. - # - with open(TESTFN, "w", encoding="utf-8") as fp: - fp.write("this is not a legal zip file\n") - try: - zf = zipfile.ZipFile(TESTFN) - except zipfile.BadZipFile: - pass - - def test_is_zip_erroneous_file(self): - """Check that is_zipfile() correctly identifies non-zip files.""" - # - passing a filename - with open(TESTFN, "w", encoding='utf-8') as fp: - fp.write("this is not a legal zip file\n") - self.assertFalse(zipfile.is_zipfile(TESTFN)) - # - passing a path-like object - self.assertFalse(zipfile.is_zipfile(pathlib.Path(TESTFN))) - # - passing a file object - with open(TESTFN, "rb") as fp: - self.assertFalse(zipfile.is_zipfile(fp)) - # - passing a file-like object - fp = io.BytesIO() - fp.write(b"this is not a legal zip file\n") - self.assertFalse(zipfile.is_zipfile(fp)) - fp.seek(0, 0) - self.assertFalse(zipfile.is_zipfile(fp)) - - def test_damaged_zipfile(self): - """Check that zipfiles with missing bytes at the end raise BadZipFile.""" - # - Create a valid zip file - fp = io.BytesIO() - with zipfile.ZipFile(fp, mode="w") as zipf: - zipf.writestr("foo.txt", b"O, for a Muse of Fire!") - zipfiledata = fp.getvalue() - - # - Now create copies of it missing the last N bytes and make sure - # a BadZipFile exception is raised when we try to open it - for N in range(len(zipfiledata)): - fp = io.BytesIO(zipfiledata[:N]) - self.assertRaises(zipfile.BadZipFile, zipfile.ZipFile, fp) - - def test_is_zip_valid_file(self): - """Check that is_zipfile() correctly identifies zip files.""" - # - passing a filename - with zipfile.ZipFile(TESTFN, mode="w") as zipf: - zipf.writestr("foo.txt", b"O, for a Muse of Fire!") - - self.assertTrue(zipfile.is_zipfile(TESTFN)) - # - passing a file object - with open(TESTFN, "rb") as fp: - self.assertTrue(zipfile.is_zipfile(fp)) - fp.seek(0, 0) - zip_contents = fp.read() - # - passing a file-like object - fp = io.BytesIO() - fp.write(zip_contents) - self.assertTrue(zipfile.is_zipfile(fp)) - fp.seek(0, 0) - self.assertTrue(zipfile.is_zipfile(fp)) - - def test_non_existent_file_raises_OSError(self): - # make sure we don't raise an AttributeError when a partially-constructed - # ZipFile instance is finalized; this tests for regression on SF tracker - # bug #403871. - - # The bug we're testing for caused an AttributeError to be raised - # when a ZipFile instance was created for a file that did not - # exist; the .fp member was not initialized but was needed by the - # __del__() method. Since the AttributeError is in the __del__(), - # it is ignored, but the user should be sufficiently annoyed by - # the message on the output that regression will be noticed - # quickly. - self.assertRaises(OSError, zipfile.ZipFile, TESTFN) - - def test_empty_file_raises_BadZipFile(self): - f = open(TESTFN, 'w', encoding='utf-8') - f.close() - self.assertRaises(zipfile.BadZipFile, zipfile.ZipFile, TESTFN) - - with open(TESTFN, 'w', encoding='utf-8') as fp: - fp.write("short file") - self.assertRaises(zipfile.BadZipFile, zipfile.ZipFile, TESTFN) - - def test_negative_central_directory_offset_raises_BadZipFile(self): - # Zip file containing an empty EOCD record - buffer = bytearray(b'PK\x05\x06' + b'\0'*18) - - # Set the size of the central directory bytes to become 1, - # causing the central directory offset to become negative - for dirsize in 1, 2**32-1: - buffer[12:16] = struct.pack(' os.path.getsize(TESTFN)) - with zipfile.ZipFile(TESTFN,mode="r") as zipf: - self.assertEqual(zipf.comment, b"shorter comment") - - def test_unicode_comment(self): - with zipfile.ZipFile(TESTFN, "w", zipfile.ZIP_STORED) as zipf: - zipf.writestr("foo.txt", "O, for a Muse of Fire!") - with self.assertRaises(TypeError): - zipf.comment = "this is an error" - - def test_change_comment_in_empty_archive(self): - with zipfile.ZipFile(TESTFN, "a", zipfile.ZIP_STORED) as zipf: - self.assertFalse(zipf.filelist) - zipf.comment = b"this is a comment" - with zipfile.ZipFile(TESTFN, "r") as zipf: - self.assertEqual(zipf.comment, b"this is a comment") - - def test_change_comment_in_nonempty_archive(self): - with zipfile.ZipFile(TESTFN, "w", zipfile.ZIP_STORED) as zipf: - zipf.writestr("foo.txt", "O, for a Muse of Fire!") - with zipfile.ZipFile(TESTFN, "a", zipfile.ZIP_STORED) as zipf: - self.assertTrue(zipf.filelist) - zipf.comment = b"this is a comment" - with zipfile.ZipFile(TESTFN, "r") as zipf: - self.assertEqual(zipf.comment, b"this is a comment") - - def test_empty_zipfile(self): - # Check that creating a file in 'w' or 'a' mode and closing without - # adding any files to the archives creates a valid empty ZIP file - zipf = zipfile.ZipFile(TESTFN, mode="w") - zipf.close() - try: - zipf = zipfile.ZipFile(TESTFN, mode="r") - except zipfile.BadZipFile: - self.fail("Unable to create empty ZIP file in 'w' mode") - - zipf = zipfile.ZipFile(TESTFN, mode="a") - zipf.close() - try: - zipf = zipfile.ZipFile(TESTFN, mode="r") - except: - self.fail("Unable to create empty ZIP file in 'a' mode") - - def test_open_empty_file(self): - # Issue 1710703: Check that opening a file with less than 22 bytes - # raises a BadZipFile exception (rather than the previously unhelpful - # OSError) - f = open(TESTFN, 'w', encoding='utf-8') - f.close() - self.assertRaises(zipfile.BadZipFile, zipfile.ZipFile, TESTFN, 'r') - - def test_create_zipinfo_before_1980(self): - self.assertRaises(ValueError, - zipfile.ZipInfo, 'seventies', (1979, 1, 1, 0, 0, 0)) - - def test_create_empty_zipinfo_repr(self): - """Before bpo-26185, repr() on empty ZipInfo object was failing.""" - zi = zipfile.ZipInfo(filename="empty") - self.assertEqual(repr(zi), "") - - def test_create_empty_zipinfo_default_attributes(self): - """Ensure all required attributes are set.""" - zi = zipfile.ZipInfo() - self.assertEqual(zi.orig_filename, "NoName") - self.assertEqual(zi.filename, "NoName") - self.assertEqual(zi.date_time, (1980, 1, 1, 0, 0, 0)) - self.assertEqual(zi.compress_type, zipfile.ZIP_STORED) - self.assertEqual(zi.comment, b"") - self.assertEqual(zi.extra, b"") - self.assertIn(zi.create_system, (0, 3)) - self.assertEqual(zi.create_version, zipfile.DEFAULT_VERSION) - self.assertEqual(zi.extract_version, zipfile.DEFAULT_VERSION) - self.assertEqual(zi.reserved, 0) - self.assertEqual(zi.flag_bits, 0) - self.assertEqual(zi.volume, 0) - self.assertEqual(zi.internal_attr, 0) - self.assertEqual(zi.external_attr, 0) - - # Before bpo-26185, both were missing - self.assertEqual(zi.file_size, 0) - self.assertEqual(zi.compress_size, 0) - - def test_zipfile_with_short_extra_field(self): - """If an extra field in the header is less than 4 bytes, skip it.""" - zipdata = ( - b'PK\x03\x04\x14\x00\x00\x00\x00\x00\x93\x9b\xad@\x8b\x9e' - b'\xd9\xd3\x01\x00\x00\x00\x01\x00\x00\x00\x03\x00\x03\x00ab' - b'c\x00\x00\x00APK\x01\x02\x14\x03\x14\x00\x00\x00\x00' - b'\x00\x93\x9b\xad@\x8b\x9e\xd9\xd3\x01\x00\x00\x00\x01\x00\x00' - b'\x00\x03\x00\x02\x00\x00\x00\x00\x00\x00\x00\x00\x00\xa4\x81\x00' - b'\x00\x00\x00abc\x00\x00PK\x05\x06\x00\x00\x00\x00' - b'\x01\x00\x01\x003\x00\x00\x00%\x00\x00\x00\x00\x00' - ) - with zipfile.ZipFile(io.BytesIO(zipdata), 'r') as zipf: - # testzip returns the name of the first corrupt file, or None - self.assertIsNone(zipf.testzip()) - - def test_open_conflicting_handles(self): - # It's only possible to open one writable file handle at a time - msg1 = b"It's fun to charter an accountant!" - msg2 = b"And sail the wide accountant sea" - msg3 = b"To find, explore the funds offshore" - with zipfile.ZipFile(TESTFN2, 'w', zipfile.ZIP_STORED) as zipf: - with zipf.open('foo', mode='w') as w2: - w2.write(msg1) - with zipf.open('bar', mode='w') as w1: - with self.assertRaises(ValueError): - zipf.open('handle', mode='w') - with self.assertRaises(ValueError): - zipf.open('foo', mode='r') - with self.assertRaises(ValueError): - zipf.writestr('str', 'abcde') - with self.assertRaises(ValueError): - zipf.write(__file__, 'file') - with self.assertRaises(ValueError): - zipf.close() - w1.write(msg2) - with zipf.open('baz', mode='w') as w2: - w2.write(msg3) - - with zipfile.ZipFile(TESTFN2, 'r') as zipf: - self.assertEqual(zipf.read('foo'), msg1) - self.assertEqual(zipf.read('bar'), msg2) - self.assertEqual(zipf.read('baz'), msg3) - self.assertEqual(zipf.namelist(), ['foo', 'bar', 'baz']) - - def test_seek_tell(self): - # Test seek functionality - txt = b"Where's Bruce?" - bloc = txt.find(b"Bruce") - # Check seek on a file - with zipfile.ZipFile(TESTFN, "w") as zipf: - zipf.writestr("foo.txt", txt) - with zipfile.ZipFile(TESTFN, "r") as zipf: - with zipf.open("foo.txt", "r") as fp: - fp.seek(bloc, os.SEEK_SET) - self.assertEqual(fp.tell(), bloc) - fp.seek(-bloc, os.SEEK_CUR) - self.assertEqual(fp.tell(), 0) - fp.seek(bloc, os.SEEK_CUR) - self.assertEqual(fp.tell(), bloc) - self.assertEqual(fp.read(5), txt[bloc:bloc+5]) - self.assertEqual(fp.tell(), bloc + 5) - fp.seek(0, os.SEEK_END) - self.assertEqual(fp.tell(), len(txt)) - fp.seek(0, os.SEEK_SET) - self.assertEqual(fp.tell(), 0) - # Check seek on memory file - data = io.BytesIO() - with zipfile.ZipFile(data, mode="w") as zipf: - zipf.writestr("foo.txt", txt) - with zipfile.ZipFile(data, mode="r") as zipf: - with zipf.open("foo.txt", "r") as fp: - fp.seek(bloc, os.SEEK_SET) - self.assertEqual(fp.tell(), bloc) - fp.seek(-bloc, os.SEEK_CUR) - self.assertEqual(fp.tell(), 0) - fp.seek(bloc, os.SEEK_CUR) - self.assertEqual(fp.tell(), bloc) - self.assertEqual(fp.read(5), txt[bloc:bloc+5]) - self.assertEqual(fp.tell(), bloc + 5) - fp.seek(0, os.SEEK_END) - self.assertEqual(fp.tell(), len(txt)) - fp.seek(0, os.SEEK_SET) - self.assertEqual(fp.tell(), 0) - - @requires_bz2() - def test_decompress_without_3rd_party_library(self): - data = b'PK\x05\x06\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00' - zip_file = io.BytesIO(data) - with zipfile.ZipFile(zip_file, 'w', compression=zipfile.ZIP_BZIP2) as zf: - zf.writestr('a.txt', b'a') - with mock.patch('zipfile.bz2', None): - with zipfile.ZipFile(zip_file) as zf: - self.assertRaises(RuntimeError, zf.extract, 'a.txt') - - def tearDown(self): - unlink(TESTFN) - unlink(TESTFN2) - - -class AbstractBadCrcTests: - def test_testzip_with_bad_crc(self): - """Tests that files with bad CRCs return their name from testzip.""" - zipdata = self.zip_with_bad_crc - - with zipfile.ZipFile(io.BytesIO(zipdata), mode="r") as zipf: - # testzip returns the name of the first corrupt file, or None - self.assertEqual('afile', zipf.testzip()) - - def test_read_with_bad_crc(self): - """Tests that files with bad CRCs raise a BadZipFile exception when read.""" - zipdata = self.zip_with_bad_crc - - # Using ZipFile.read() - with zipfile.ZipFile(io.BytesIO(zipdata), mode="r") as zipf: - self.assertRaises(zipfile.BadZipFile, zipf.read, 'afile') - - # Using ZipExtFile.read() - with zipfile.ZipFile(io.BytesIO(zipdata), mode="r") as zipf: - with zipf.open('afile', 'r') as corrupt_file: - self.assertRaises(zipfile.BadZipFile, corrupt_file.read) - - # Same with small reads (in order to exercise the buffering logic) - with zipfile.ZipFile(io.BytesIO(zipdata), mode="r") as zipf: - with zipf.open('afile', 'r') as corrupt_file: - corrupt_file.MIN_READ_SIZE = 2 - with self.assertRaises(zipfile.BadZipFile): - while corrupt_file.read(2): - pass - - -class StoredBadCrcTests(AbstractBadCrcTests, unittest.TestCase): - compression = zipfile.ZIP_STORED - zip_with_bad_crc = ( - b'PK\003\004\024\0\0\0\0\0 \213\212;:r' - b'\253\377\f\0\0\0\f\0\0\0\005\0\0\000af' - b'ilehello,AworldP' - b'K\001\002\024\003\024\0\0\0\0\0 \213\212;:' - b'r\253\377\f\0\0\0\f\0\0\0\005\0\0\0\0' - b'\0\0\0\0\0\0\0\200\001\0\0\0\000afi' - b'lePK\005\006\0\0\0\0\001\0\001\0003\000' - b'\0\0/\0\0\0\0\0') - -@requires_zlib() -class DeflateBadCrcTests(AbstractBadCrcTests, unittest.TestCase): - compression = zipfile.ZIP_DEFLATED - zip_with_bad_crc = ( - b'PK\x03\x04\x14\x00\x00\x00\x08\x00n}\x0c=FA' - b'KE\x10\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00af' - b'ile\xcbH\xcd\xc9\xc9W(\xcf/\xcaI\xc9\xa0' - b'=\x13\x00PK\x01\x02\x14\x03\x14\x00\x00\x00\x08\x00n' - b'}\x0c=FAKE\x10\x00\x00\x00n\x00\x00\x00\x05' - b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x80\x01\x00\x00\x00' - b'\x00afilePK\x05\x06\x00\x00\x00\x00\x01\x00' - b'\x01\x003\x00\x00\x003\x00\x00\x00\x00\x00') - -@requires_bz2() -class Bzip2BadCrcTests(AbstractBadCrcTests, unittest.TestCase): - compression = zipfile.ZIP_BZIP2 - zip_with_bad_crc = ( - b'PK\x03\x04\x14\x03\x00\x00\x0c\x00nu\x0c=FA' - b'KE8\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00af' - b'ileBZh91AY&SY\xd4\xa8\xca' - b'\x7f\x00\x00\x0f\x11\x80@\x00\x06D\x90\x80 \x00 \xa5' - b'P\xd9!\x03\x03\x13\x13\x13\x89\xa9\xa9\xc2u5:\x9f' - b'\x8b\xb9"\x9c(HjTe?\x80PK\x01\x02\x14' - b'\x03\x14\x03\x00\x00\x0c\x00nu\x0c=FAKE8' - b'\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00\x00\x00\x00\x00\x00' - b'\x00 \x80\x80\x81\x00\x00\x00\x00afilePK' - b'\x05\x06\x00\x00\x00\x00\x01\x00\x01\x003\x00\x00\x00[\x00' - b'\x00\x00\x00\x00') - -@requires_lzma() -class LzmaBadCrcTests(AbstractBadCrcTests, unittest.TestCase): - compression = zipfile.ZIP_LZMA - zip_with_bad_crc = ( - b'PK\x03\x04\x14\x03\x00\x00\x0e\x00nu\x0c=FA' - b'KE\x1b\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00af' - b'ile\t\x04\x05\x00]\x00\x00\x00\x04\x004\x19I' - b'\xee\x8d\xe9\x17\x89:3`\tq!.8\x00PK' - b'\x01\x02\x14\x03\x14\x03\x00\x00\x0e\x00nu\x0c=FA' - b'KE\x1b\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00\x00\x00' - b'\x00\x00\x00\x00 \x80\x80\x81\x00\x00\x00\x00afil' - b'ePK\x05\x06\x00\x00\x00\x00\x01\x00\x01\x003\x00\x00' - b'\x00>\x00\x00\x00\x00\x00') - - -class DecryptionTests(unittest.TestCase): - """Check that ZIP decryption works. Since the library does not - support encryption at the moment, we use a pre-generated encrypted - ZIP file.""" - - data = ( - b'PK\x03\x04\x14\x00\x01\x00\x00\x00n\x92i.#y\xef?&\x00\x00\x00\x1a\x00' - b'\x00\x00\x08\x00\x00\x00test.txt\xfa\x10\xa0gly|\xfa-\xc5\xc0=\xf9y' - b'\x18\xe0\xa8r\xb3Z}Lg\xbc\xae\xf9|\x9b\x19\xe4\x8b\xba\xbb)\x8c\xb0\xdbl' - b'PK\x01\x02\x14\x00\x14\x00\x01\x00\x00\x00n\x92i.#y\xef?&\x00\x00\x00' - b'\x1a\x00\x00\x00\x08\x00\x00\x00\x00\x00\x00\x00\x01\x00 \x00\xb6\x81' - b'\x00\x00\x00\x00test.txtPK\x05\x06\x00\x00\x00\x00\x01\x00\x01\x006\x00' - b'\x00\x00L\x00\x00\x00\x00\x00' ) - data2 = ( - b'PK\x03\x04\x14\x00\t\x00\x08\x00\xcf}38xu\xaa\xb2\x14\x00\x00\x00\x00\x02' - b'\x00\x00\x04\x00\x15\x00zeroUT\t\x00\x03\xd6\x8b\x92G\xda\x8b\x92GUx\x04' - b'\x00\xe8\x03\xe8\x03\xc7e|f"g?h*i', ','), r'a\b,c_d_e_f_g_h_i') + self.assertEqual(san('../../foo../../ba..r', '/'), r'foo/ba..r') + self.assertEqual(san(' / /foo / /ba r', '/'), r'foo/ba r') + self.assertEqual(san(' . /. /foo ./ . /. ./ba .r', '/'), r'foo/ba .r') + + def test_extract_hackers_arcnames_common_cases(self): + common_hacknames = [ + ('../foo/bar', 'foo/bar'), + ('foo/../bar', 'foo/bar'), + ('foo/../../bar', 'foo/bar'), + ('foo/bar/..', 'foo/bar'), + ('./../foo/bar', 'foo/bar'), + ('/foo/bar', 'foo/bar'), + ('/foo/../bar', 'foo/bar'), + ('/foo/../../bar', 'foo/bar'), + ] + self._test_extract_hackers_arcnames(common_hacknames) + + @unittest.skipIf(os.path.sep != '\\', 'Requires \\ as path separator.') + def test_extract_hackers_arcnames_windows_only(self): + """Test combination of path fixing and windows name sanitization.""" + windows_hacknames = [ + (r'..\foo\bar', 'foo/bar'), + (r'..\/foo\/bar', 'foo/bar'), + (r'foo/\..\/bar', 'foo/bar'), + (r'foo\/../\bar', 'foo/bar'), + (r'C:foo/bar', 'foo/bar'), + (r'C:/foo/bar', 'foo/bar'), + (r'C://foo/bar', 'foo/bar'), + (r'C:\foo\bar', 'foo/bar'), + (r'//conky/mountpoint/foo/bar', 'foo/bar'), + (r'\\conky\mountpoint\foo\bar', 'foo/bar'), + (r'///conky/mountpoint/foo/bar', 'conky/mountpoint/foo/bar'), + (r'\\\conky\mountpoint\foo\bar', 'conky/mountpoint/foo/bar'), + (r'//conky//mountpoint/foo/bar', 'conky/mountpoint/foo/bar'), + (r'\\conky\\mountpoint\foo\bar', 'conky/mountpoint/foo/bar'), + (r'//?/C:/foo/bar', 'foo/bar'), + (r'\\?\C:\foo\bar', 'foo/bar'), + (r'C:/../C:/foo/bar', 'C_/foo/bar'), + (r'a:b\ce|f"g?h*i', 'b/c_d_e_f_g_h_i'), + ('../../foo../../ba..r', 'foo/ba..r'), + ] + self._test_extract_hackers_arcnames(windows_hacknames) + + @unittest.skipIf(os.path.sep != '/', r'Requires / as path separator.') + def test_extract_hackers_arcnames_posix_only(self): + posix_hacknames = [ + ('//foo/bar', 'foo/bar'), + ('../../foo../../ba..r', 'foo../ba..r'), + (r'foo/..\bar', r'foo/..\bar'), + ] + self._test_extract_hackers_arcnames(posix_hacknames) + + def _test_extract_hackers_arcnames(self, hacknames): + for arcname, fixedname in hacknames: + content = b'foobar' + arcname.encode() + with zipfile.ZipFile(TESTFN2, 'w', zipfile.ZIP_STORED) as zipfp: + zinfo = zipfile.ZipInfo() + # preserve backslashes + zinfo.filename = arcname + zinfo.external_attr = 0o600 << 16 + zipfp.writestr(zinfo, content) + + arcname = arcname.replace(os.sep, "/") + targetpath = os.path.join('target', 'subdir', 'subsub') + correctfile = os.path.join(targetpath, *fixedname.split('/')) + + with zipfile.ZipFile(TESTFN2, 'r') as zipfp: + writtenfile = zipfp.extract(arcname, targetpath) + self.assertEqual(writtenfile, correctfile, + msg='extract %r: %r != %r' % + (arcname, writtenfile, correctfile)) + self.check_file(correctfile, content) + rmtree('target') + + with zipfile.ZipFile(TESTFN2, 'r') as zipfp: + zipfp.extractall(targetpath) + self.check_file(correctfile, content) + rmtree('target') + + correctfile = os.path.join(os.getcwd(), *fixedname.split('/')) + + with zipfile.ZipFile(TESTFN2, 'r') as zipfp: + writtenfile = zipfp.extract(arcname) + self.assertEqual(writtenfile, correctfile, + msg="extract %r" % arcname) + self.check_file(correctfile, content) + rmtree(fixedname.split('/')[0]) + + with zipfile.ZipFile(TESTFN2, 'r') as zipfp: + zipfp.extractall() + self.check_file(correctfile, content) + rmtree(fixedname.split('/')[0]) + + unlink(TESTFN2) + + +class OtherTests(unittest.TestCase): + def test_open_via_zip_info(self): + # Create the ZIP archive + with zipfile.ZipFile(TESTFN2, "w", zipfile.ZIP_STORED) as zipfp: + zipfp.writestr("name", "foo") + with self.assertWarns(UserWarning): + zipfp.writestr("name", "bar") + self.assertEqual(zipfp.namelist(), ["name"] * 2) + + with zipfile.ZipFile(TESTFN2, "r") as zipfp: + infos = zipfp.infolist() + data = b"" + for info in infos: + with zipfp.open(info) as zipopen: + data += zipopen.read() + self.assertIn(data, {b"foobar", b"barfoo"}) + data = b"" + for info in infos: + data += zipfp.read(info) + self.assertIn(data, {b"foobar", b"barfoo"}) + + def test_writestr_extended_local_header_issue1202(self): + with zipfile.ZipFile(TESTFN2, 'w') as orig_zip: + for data in 'abcdefghijklmnop': + zinfo = zipfile.ZipInfo(data) + zinfo.flag_bits |= zipfile._MASK_USE_DATA_DESCRIPTOR # Include an extended local header. + orig_zip.writestr(zinfo, data) + + def test_close(self): + """Check that the zipfile is closed after the 'with' block.""" + with zipfile.ZipFile(TESTFN2, "w") as zipfp: + for fpath, fdata in SMALL_TEST_DATA: + zipfp.writestr(fpath, fdata) + self.assertIsNotNone(zipfp.fp, 'zipfp is not open') + self.assertIsNone(zipfp.fp, 'zipfp is not closed') + + with zipfile.ZipFile(TESTFN2, "r") as zipfp: + self.assertIsNotNone(zipfp.fp, 'zipfp is not open') + self.assertIsNone(zipfp.fp, 'zipfp is not closed') + + def test_close_on_exception(self): + """Check that the zipfile is closed if an exception is raised in the + 'with' block.""" + with zipfile.ZipFile(TESTFN2, "w") as zipfp: + for fpath, fdata in SMALL_TEST_DATA: + zipfp.writestr(fpath, fdata) + + try: + with zipfile.ZipFile(TESTFN2, "r") as zipfp2: + raise zipfile.BadZipFile() + except zipfile.BadZipFile: + self.assertIsNone(zipfp2.fp, 'zipfp is not closed') + + def test_unsupported_version(self): + # File has an extract_version of 120 + data = (b'PK\x03\x04x\x00\x00\x00\x00\x00!p\xa1@\x00\x00\x00\x00\x00\x00' + b'\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00xPK\x01\x02x\x03x\x00\x00\x00\x00' + b'\x00!p\xa1@\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00' + b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x80\x01\x00\x00\x00\x00xPK\x05\x06' + b'\x00\x00\x00\x00\x01\x00\x01\x00/\x00\x00\x00\x1f\x00\x00\x00\x00\x00') + + self.assertRaises(NotImplementedError, zipfile.ZipFile, + io.BytesIO(data), 'r') + + @requires_zlib() + def test_read_unicode_filenames(self): + # bug #10801 + fname = findfile('zip_cp437_header.zip') + with zipfile.ZipFile(fname) as zipfp: + for name in zipfp.namelist(): + zipfp.open(name).close() + + def test_write_unicode_filenames(self): + with zipfile.ZipFile(TESTFN, "w") as zf: + zf.writestr("foo.txt", "Test for unicode filename") + zf.writestr("\xf6.txt", "Test for unicode filename") + self.assertIsInstance(zf.infolist()[0].filename, str) + + with zipfile.ZipFile(TESTFN, "r") as zf: + self.assertEqual(zf.filelist[0].filename, "foo.txt") + self.assertEqual(zf.filelist[1].filename, "\xf6.txt") + + def test_read_after_write_unicode_filenames(self): + with zipfile.ZipFile(TESTFN2, 'w') as zipfp: + zipfp.writestr('приклад', b'sample') + self.assertEqual(zipfp.read('приклад'), b'sample') + + def test_exclusive_create_zip_file(self): + """Test exclusive creating a new zipfile.""" + unlink(TESTFN2) + filename = 'testfile.txt' + content = b'hello, world. this is some content.' + with zipfile.ZipFile(TESTFN2, "x", zipfile.ZIP_STORED) as zipfp: + zipfp.writestr(filename, content) + with self.assertRaises(FileExistsError): + zipfile.ZipFile(TESTFN2, "x", zipfile.ZIP_STORED) + with zipfile.ZipFile(TESTFN2, "r") as zipfp: + self.assertEqual(zipfp.namelist(), [filename]) + self.assertEqual(zipfp.read(filename), content) + + def test_create_non_existent_file_for_append(self): + if os.path.exists(TESTFN): + os.unlink(TESTFN) + + filename = 'testfile.txt' + content = b'hello, world. this is some content.' + + try: + with zipfile.ZipFile(TESTFN, 'a') as zf: + zf.writestr(filename, content) + except OSError: + self.fail('Could not append data to a non-existent zip file.') + + self.assertTrue(os.path.exists(TESTFN)) + + with zipfile.ZipFile(TESTFN, 'r') as zf: + self.assertEqual(zf.read(filename), content) + + def test_close_erroneous_file(self): + # This test checks that the ZipFile constructor closes the file object + # it opens if there's an error in the file. If it doesn't, the + # traceback holds a reference to the ZipFile object and, indirectly, + # the file object. + # On Windows, this causes the os.unlink() call to fail because the + # underlying file is still open. This is SF bug #412214. + # + with open(TESTFN, "w", encoding="utf-8") as fp: + fp.write("this is not a legal zip file\n") + try: + zf = zipfile.ZipFile(TESTFN) + except zipfile.BadZipFile: + pass + + def test_is_zip_erroneous_file(self): + """Check that is_zipfile() correctly identifies non-zip files.""" + # - passing a filename + with open(TESTFN, "w", encoding='utf-8') as fp: + fp.write("this is not a legal zip file\n") + self.assertFalse(zipfile.is_zipfile(TESTFN)) + # - passing a path-like object + self.assertFalse(zipfile.is_zipfile(pathlib.Path(TESTFN))) + # - passing a file object + with open(TESTFN, "rb") as fp: + self.assertFalse(zipfile.is_zipfile(fp)) + # - passing a file-like object + fp = io.BytesIO() + fp.write(b"this is not a legal zip file\n") + self.assertFalse(zipfile.is_zipfile(fp)) + fp.seek(0, 0) + self.assertFalse(zipfile.is_zipfile(fp)) + + def test_damaged_zipfile(self): + """Check that zipfiles with missing bytes at the end raise BadZipFile.""" + # - Create a valid zip file + fp = io.BytesIO() + with zipfile.ZipFile(fp, mode="w") as zipf: + zipf.writestr("foo.txt", b"O, for a Muse of Fire!") + zipfiledata = fp.getvalue() + + # - Now create copies of it missing the last N bytes and make sure + # a BadZipFile exception is raised when we try to open it + for N in range(len(zipfiledata)): + fp = io.BytesIO(zipfiledata[:N]) + self.assertRaises(zipfile.BadZipFile, zipfile.ZipFile, fp) + + def test_is_zip_valid_file(self): + """Check that is_zipfile() correctly identifies zip files.""" + # - passing a filename + with zipfile.ZipFile(TESTFN, mode="w") as zipf: + zipf.writestr("foo.txt", b"O, for a Muse of Fire!") + + self.assertTrue(zipfile.is_zipfile(TESTFN)) + # - passing a file object + with open(TESTFN, "rb") as fp: + self.assertTrue(zipfile.is_zipfile(fp)) + fp.seek(0, 0) + zip_contents = fp.read() + # - passing a file-like object + fp = io.BytesIO() + fp.write(zip_contents) + self.assertTrue(zipfile.is_zipfile(fp)) + fp.seek(0, 0) + self.assertTrue(zipfile.is_zipfile(fp)) + + def test_non_existent_file_raises_OSError(self): + # make sure we don't raise an AttributeError when a partially-constructed + # ZipFile instance is finalized; this tests for regression on SF tracker + # bug #403871. + + # The bug we're testing for caused an AttributeError to be raised + # when a ZipFile instance was created for a file that did not + # exist; the .fp member was not initialized but was needed by the + # __del__() method. Since the AttributeError is in the __del__(), + # it is ignored, but the user should be sufficiently annoyed by + # the message on the output that regression will be noticed + # quickly. + self.assertRaises(OSError, zipfile.ZipFile, TESTFN) + + def test_empty_file_raises_BadZipFile(self): + f = open(TESTFN, 'w', encoding='utf-8') + f.close() + self.assertRaises(zipfile.BadZipFile, zipfile.ZipFile, TESTFN) + + with open(TESTFN, 'w', encoding='utf-8') as fp: + fp.write("short file") + self.assertRaises(zipfile.BadZipFile, zipfile.ZipFile, TESTFN) + + def test_negative_central_directory_offset_raises_BadZipFile(self): + # Zip file containing an empty EOCD record + buffer = bytearray(b'PK\x05\x06' + b'\0'*18) + + # Set the size of the central directory bytes to become 1, + # causing the central directory offset to become negative + for dirsize in 1, 2**32-1: + buffer[12:16] = struct.pack(' os.path.getsize(TESTFN)) + with zipfile.ZipFile(TESTFN,mode="r") as zipf: + self.assertEqual(zipf.comment, b"shorter comment") + + def test_unicode_comment(self): + with zipfile.ZipFile(TESTFN, "w", zipfile.ZIP_STORED) as zipf: + zipf.writestr("foo.txt", "O, for a Muse of Fire!") + with self.assertRaises(TypeError): + zipf.comment = "this is an error" + + def test_change_comment_in_empty_archive(self): + with zipfile.ZipFile(TESTFN, "a", zipfile.ZIP_STORED) as zipf: + self.assertFalse(zipf.filelist) + zipf.comment = b"this is a comment" + with zipfile.ZipFile(TESTFN, "r") as zipf: + self.assertEqual(zipf.comment, b"this is a comment") + + def test_change_comment_in_nonempty_archive(self): + with zipfile.ZipFile(TESTFN, "w", zipfile.ZIP_STORED) as zipf: + zipf.writestr("foo.txt", "O, for a Muse of Fire!") + with zipfile.ZipFile(TESTFN, "a", zipfile.ZIP_STORED) as zipf: + self.assertTrue(zipf.filelist) + zipf.comment = b"this is a comment" + with zipfile.ZipFile(TESTFN, "r") as zipf: + self.assertEqual(zipf.comment, b"this is a comment") + + def test_empty_zipfile(self): + # Check that creating a file in 'w' or 'a' mode and closing without + # adding any files to the archives creates a valid empty ZIP file + zipf = zipfile.ZipFile(TESTFN, mode="w") + zipf.close() + try: + zipf = zipfile.ZipFile(TESTFN, mode="r") + except zipfile.BadZipFile: + self.fail("Unable to create empty ZIP file in 'w' mode") + + zipf = zipfile.ZipFile(TESTFN, mode="a") + zipf.close() + try: + zipf = zipfile.ZipFile(TESTFN, mode="r") + except: + self.fail("Unable to create empty ZIP file in 'a' mode") + + def test_open_empty_file(self): + # Issue 1710703: Check that opening a file with less than 22 bytes + # raises a BadZipFile exception (rather than the previously unhelpful + # OSError) + f = open(TESTFN, 'w', encoding='utf-8') + f.close() + self.assertRaises(zipfile.BadZipFile, zipfile.ZipFile, TESTFN, 'r') + + def test_create_zipinfo_before_1980(self): + self.assertRaises(ValueError, + zipfile.ZipInfo, 'seventies', (1979, 1, 1, 0, 0, 0)) + + def test_create_empty_zipinfo_repr(self): + """Before bpo-26185, repr() on empty ZipInfo object was failing.""" + zi = zipfile.ZipInfo(filename="empty") + self.assertEqual(repr(zi), "") + + def test_create_empty_zipinfo_default_attributes(self): + """Ensure all required attributes are set.""" + zi = zipfile.ZipInfo() + self.assertEqual(zi.orig_filename, "NoName") + self.assertEqual(zi.filename, "NoName") + self.assertEqual(zi.date_time, (1980, 1, 1, 0, 0, 0)) + self.assertEqual(zi.compress_type, zipfile.ZIP_STORED) + self.assertEqual(zi.comment, b"") + self.assertEqual(zi.extra, b"") + self.assertIn(zi.create_system, (0, 3)) + self.assertEqual(zi.create_version, zipfile.DEFAULT_VERSION) + self.assertEqual(zi.extract_version, zipfile.DEFAULT_VERSION) + self.assertEqual(zi.reserved, 0) + self.assertEqual(zi.flag_bits, 0) + self.assertEqual(zi.volume, 0) + self.assertEqual(zi.internal_attr, 0) + self.assertEqual(zi.external_attr, 0) + + # Before bpo-26185, both were missing + self.assertEqual(zi.file_size, 0) + self.assertEqual(zi.compress_size, 0) + + def test_zipfile_with_short_extra_field(self): + """If an extra field in the header is less than 4 bytes, skip it.""" + zipdata = ( + b'PK\x03\x04\x14\x00\x00\x00\x00\x00\x93\x9b\xad@\x8b\x9e' + b'\xd9\xd3\x01\x00\x00\x00\x01\x00\x00\x00\x03\x00\x03\x00ab' + b'c\x00\x00\x00APK\x01\x02\x14\x03\x14\x00\x00\x00\x00' + b'\x00\x93\x9b\xad@\x8b\x9e\xd9\xd3\x01\x00\x00\x00\x01\x00\x00' + b'\x00\x03\x00\x02\x00\x00\x00\x00\x00\x00\x00\x00\x00\xa4\x81\x00' + b'\x00\x00\x00abc\x00\x00PK\x05\x06\x00\x00\x00\x00' + b'\x01\x00\x01\x003\x00\x00\x00%\x00\x00\x00\x00\x00' + ) + with zipfile.ZipFile(io.BytesIO(zipdata), 'r') as zipf: + # testzip returns the name of the first corrupt file, or None + self.assertIsNone(zipf.testzip()) + + def test_open_conflicting_handles(self): + # It's only possible to open one writable file handle at a time + msg1 = b"It's fun to charter an accountant!" + msg2 = b"And sail the wide accountant sea" + msg3 = b"To find, explore the funds offshore" + with zipfile.ZipFile(TESTFN2, 'w', zipfile.ZIP_STORED) as zipf: + with zipf.open('foo', mode='w') as w2: + w2.write(msg1) + with zipf.open('bar', mode='w') as w1: + with self.assertRaises(ValueError): + zipf.open('handle', mode='w') + with self.assertRaises(ValueError): + zipf.open('foo', mode='r') + with self.assertRaises(ValueError): + zipf.writestr('str', 'abcde') + with self.assertRaises(ValueError): + zipf.write(__file__, 'file') + with self.assertRaises(ValueError): + zipf.close() + w1.write(msg2) + with zipf.open('baz', mode='w') as w2: + w2.write(msg3) + + with zipfile.ZipFile(TESTFN2, 'r') as zipf: + self.assertEqual(zipf.read('foo'), msg1) + self.assertEqual(zipf.read('bar'), msg2) + self.assertEqual(zipf.read('baz'), msg3) + self.assertEqual(zipf.namelist(), ['foo', 'bar', 'baz']) + + def test_seek_tell(self): + # Test seek functionality + txt = b"Where's Bruce?" + bloc = txt.find(b"Bruce") + # Check seek on a file + with zipfile.ZipFile(TESTFN, "w") as zipf: + zipf.writestr("foo.txt", txt) + with zipfile.ZipFile(TESTFN, "r") as zipf: + with zipf.open("foo.txt", "r") as fp: + fp.seek(bloc, os.SEEK_SET) + self.assertEqual(fp.tell(), bloc) + fp.seek(-bloc, os.SEEK_CUR) + self.assertEqual(fp.tell(), 0) + fp.seek(bloc, os.SEEK_CUR) + self.assertEqual(fp.tell(), bloc) + self.assertEqual(fp.read(5), txt[bloc:bloc+5]) + self.assertEqual(fp.tell(), bloc + 5) + fp.seek(0, os.SEEK_END) + self.assertEqual(fp.tell(), len(txt)) + fp.seek(0, os.SEEK_SET) + self.assertEqual(fp.tell(), 0) + # Check seek on memory file + data = io.BytesIO() + with zipfile.ZipFile(data, mode="w") as zipf: + zipf.writestr("foo.txt", txt) + with zipfile.ZipFile(data, mode="r") as zipf: + with zipf.open("foo.txt", "r") as fp: + fp.seek(bloc, os.SEEK_SET) + self.assertEqual(fp.tell(), bloc) + fp.seek(-bloc, os.SEEK_CUR) + self.assertEqual(fp.tell(), 0) + fp.seek(bloc, os.SEEK_CUR) + self.assertEqual(fp.tell(), bloc) + self.assertEqual(fp.read(5), txt[bloc:bloc+5]) + self.assertEqual(fp.tell(), bloc + 5) + fp.seek(0, os.SEEK_END) + self.assertEqual(fp.tell(), len(txt)) + fp.seek(0, os.SEEK_SET) + self.assertEqual(fp.tell(), 0) + + @requires_bz2() + def test_decompress_without_3rd_party_library(self): + data = b'PK\x05\x06\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00' + zip_file = io.BytesIO(data) + with zipfile.ZipFile(zip_file, 'w', compression=zipfile.ZIP_BZIP2) as zf: + zf.writestr('a.txt', b'a') + with mock.patch('zipfile.bz2', None): + with zipfile.ZipFile(zip_file) as zf: + self.assertRaises(RuntimeError, zf.extract, 'a.txt') + + def tearDown(self): + unlink(TESTFN) + unlink(TESTFN2) + + +class AbstractBadCrcTests: + def test_testzip_with_bad_crc(self): + """Tests that files with bad CRCs return their name from testzip.""" + zipdata = self.zip_with_bad_crc + + with zipfile.ZipFile(io.BytesIO(zipdata), mode="r") as zipf: + # testzip returns the name of the first corrupt file, or None + self.assertEqual('afile', zipf.testzip()) + + def test_read_with_bad_crc(self): + """Tests that files with bad CRCs raise a BadZipFile exception when read.""" + zipdata = self.zip_with_bad_crc + + # Using ZipFile.read() + with zipfile.ZipFile(io.BytesIO(zipdata), mode="r") as zipf: + self.assertRaises(zipfile.BadZipFile, zipf.read, 'afile') + + # Using ZipExtFile.read() + with zipfile.ZipFile(io.BytesIO(zipdata), mode="r") as zipf: + with zipf.open('afile', 'r') as corrupt_file: + self.assertRaises(zipfile.BadZipFile, corrupt_file.read) + + # Same with small reads (in order to exercise the buffering logic) + with zipfile.ZipFile(io.BytesIO(zipdata), mode="r") as zipf: + with zipf.open('afile', 'r') as corrupt_file: + corrupt_file.MIN_READ_SIZE = 2 + with self.assertRaises(zipfile.BadZipFile): + while corrupt_file.read(2): + pass + + +class StoredBadCrcTests(AbstractBadCrcTests, unittest.TestCase): + compression = zipfile.ZIP_STORED + zip_with_bad_crc = ( + b'PK\003\004\024\0\0\0\0\0 \213\212;:r' + b'\253\377\f\0\0\0\f\0\0\0\005\0\0\000af' + b'ilehello,AworldP' + b'K\001\002\024\003\024\0\0\0\0\0 \213\212;:' + b'r\253\377\f\0\0\0\f\0\0\0\005\0\0\0\0' + b'\0\0\0\0\0\0\0\200\001\0\0\0\000afi' + b'lePK\005\006\0\0\0\0\001\0\001\0003\000' + b'\0\0/\0\0\0\0\0') + +@requires_zlib() +class DeflateBadCrcTests(AbstractBadCrcTests, unittest.TestCase): + compression = zipfile.ZIP_DEFLATED + zip_with_bad_crc = ( + b'PK\x03\x04\x14\x00\x00\x00\x08\x00n}\x0c=FA' + b'KE\x10\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00af' + b'ile\xcbH\xcd\xc9\xc9W(\xcf/\xcaI\xc9\xa0' + b'=\x13\x00PK\x01\x02\x14\x03\x14\x00\x00\x00\x08\x00n' + b'}\x0c=FAKE\x10\x00\x00\x00n\x00\x00\x00\x05' + b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x80\x01\x00\x00\x00' + b'\x00afilePK\x05\x06\x00\x00\x00\x00\x01\x00' + b'\x01\x003\x00\x00\x003\x00\x00\x00\x00\x00') + +@requires_bz2() +class Bzip2BadCrcTests(AbstractBadCrcTests, unittest.TestCase): + compression = zipfile.ZIP_BZIP2 + zip_with_bad_crc = ( + b'PK\x03\x04\x14\x03\x00\x00\x0c\x00nu\x0c=FA' + b'KE8\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00af' + b'ileBZh91AY&SY\xd4\xa8\xca' + b'\x7f\x00\x00\x0f\x11\x80@\x00\x06D\x90\x80 \x00 \xa5' + b'P\xd9!\x03\x03\x13\x13\x13\x89\xa9\xa9\xc2u5:\x9f' + b'\x8b\xb9"\x9c(HjTe?\x80PK\x01\x02\x14' + b'\x03\x14\x03\x00\x00\x0c\x00nu\x0c=FAKE8' + b'\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00\x00\x00\x00\x00\x00' + b'\x00 \x80\x80\x81\x00\x00\x00\x00afilePK' + b'\x05\x06\x00\x00\x00\x00\x01\x00\x01\x003\x00\x00\x00[\x00' + b'\x00\x00\x00\x00') + +@requires_lzma() +class LzmaBadCrcTests(AbstractBadCrcTests, unittest.TestCase): + compression = zipfile.ZIP_LZMA + zip_with_bad_crc = ( + b'PK\x03\x04\x14\x03\x00\x00\x0e\x00nu\x0c=FA' + b'KE\x1b\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00af' + b'ile\t\x04\x05\x00]\x00\x00\x00\x04\x004\x19I' + b'\xee\x8d\xe9\x17\x89:3`\tq!.8\x00PK' + b'\x01\x02\x14\x03\x14\x03\x00\x00\x0e\x00nu\x0c=FA' + b'KE\x1b\x00\x00\x00n\x00\x00\x00\x05\x00\x00\x00\x00\x00' + b'\x00\x00\x00\x00 \x80\x80\x81\x00\x00\x00\x00afil' + b'ePK\x05\x06\x00\x00\x00\x00\x01\x00\x01\x003\x00\x00' + b'\x00>\x00\x00\x00\x00\x00') + + +class DecryptionTests(unittest.TestCase): + """Check that ZIP decryption works. Since the library does not + support encryption at the moment, we use a pre-generated encrypted + ZIP file.""" + + data = ( + b'PK\x03\x04\x14\x00\x01\x00\x00\x00n\x92i.#y\xef?&\x00\x00\x00\x1a\x00' + b'\x00\x00\x08\x00\x00\x00test.txt\xfa\x10\xa0gly|\xfa-\xc5\xc0=\xf9y' + b'\x18\xe0\xa8r\xb3Z}Lg\xbc\xae\xf9|\x9b\x19\xe4\x8b\xba\xbb)\x8c\xb0\xdbl' + b'PK\x01\x02\x14\x00\x14\x00\x01\x00\x00\x00n\x92i.#y\xef?&\x00\x00\x00' + b'\x1a\x00\x00\x00\x08\x00\x00\x00\x00\x00\x00\x00\x01\x00 \x00\xb6\x81' + b'\x00\x00\x00\x00test.txtPK\x05\x06\x00\x00\x00\x00\x01\x00\x01\x006\x00' + b'\x00\x00L\x00\x00\x00\x00\x00' ) + data2 = ( + b'PK\x03\x04\x14\x00\t\x00\x08\x00\xcf}38xu\xaa\xb2\x14\x00\x00\x00\x00\x02' + b'\x00\x00\x04\x00\x15\x00zeroUT\t\x00\x03\xd6\x8b\x92G\xda\x8b\x92GUx\x04' + b'\x00\xe8\x03\xe8\x03\xc7 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() - 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 - return None - - -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', - ) - - 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 - # 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: - zip64 = file_size > ZIP64_LIMIT or compress_size > ZIP64_LIMIT - if zip64: - fmt = ' 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 | _MASK_UTF_FILENAME - - def _decodeExtra(self): - # Try to decode the extra field. - extra = self.extra - unpack = struct.unpack - while len(extra) >= 4: - tp, ln = unpack(' 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(' 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.""" - return self.filename.endswith('/') - - -# 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('> 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. - 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, - # decrypt and decompress them. - if self._eof or n <= 0: - return b'' - - # Read from file. - if self._compress_type == ZIP_DEFLATED: - ## Handle unconsumed data. - data = self._decompressor.unconsumed_tail - if n > len(data): - data += self._read2(n - len(data)) - else: - data = self._read2(n) - - if self._compress_type == ZIP_STORED: - self._eof = self._compress_left <= 0 - elif self._compress_type == ZIP_DEFLATED: - n = max(n, self.MIN_READ_SIZE) - data = self._decompressor.decompress(data, n) - self._eof = (self._decompressor.eof or - self._compress_left <= 0 and - not self._decompressor.unconsumed_tail) - if self._eof: - data += self._decompressor.flush() - else: - data = self._decompressor.decompress(data) - self._eof = self._decompressor.eof or self._compress_left <= 0 - - data = data[:self._left] - self._left -= len(data) - if self._left <= 0: - self._eof = True - self._update_crc(data) - return data - - def _read2(self, n): - if self._compress_left <= 0: - return b'' - - n = max(n, self.MIN_READ_SIZE) - n = min(n, self._compress_left) - - data = self._fileobj.read(n) - self._compress_left -= len(data) - if not data: - raise EOFError - - if self._decrypter is not None: - data = self._decrypter(data) - return data - - def close(self): - try: - if self._close_fileobj: - self._fileobj.close() - finally: - super().close() - - def seekable(self): - if self.closed: - raise ValueError("I/O operation on closed file.") - return self._seekable - - def seek(self, offset, whence=os.SEEK_SET): - if self.closed: - raise ValueError("seek on closed file.") - if not self._seekable: - raise io.UnsupportedOperation("underlying stream is not seekable") - curr_pos = self.tell() - if whence == os.SEEK_SET: - new_pos = offset - elif whence == os.SEEK_CUR: - new_pos = curr_pos + offset - elif whence == os.SEEK_END: - new_pos = self._orig_file_size + offset - else: - raise ValueError("whence must be os.SEEK_SET (0), " - "os.SEEK_CUR (1), or os.SEEK_END (2)") - - if new_pos > self._orig_file_size: - new_pos = self._orig_file_size - - if new_pos < 0: - new_pos = 0 - - read_offset = new_pos - curr_pos - buff_offset = read_offset + self._offset - - # Fast seek uncompressed unencrypted file - if self._compress_type == ZIP_STORED and self._decrypter is None and read_offset > 0: - # disable CRC checking after first seeking - it would be invalid - self._expected_crc = None - # seek actual file taking already buffered data into account - read_offset -= len(self._readbuffer) - self._offset - self._fileobj.seek(read_offset, os.SEEK_CUR) - self._left -= read_offset - read_offset = 0 - # flush read buffer - self._readbuffer = b'' - self._offset = 0 - elif buff_offset >= 0 and buff_offset < len(self._readbuffer): - # Just move the _offset index if the new position is in the _readbuffer - self._offset = buff_offset - read_offset = 0 - elif read_offset < 0: - # Position is before the current position. Reset the ZipExtFile - self._fileobj.seek(self._orig_compress_start) - self._running_crc = self._orig_start_crc - self._expected_crc = self._orig_crc - self._compress_left = self._orig_compress_size - self._left = self._orig_file_size - self._readbuffer = b'' - self._offset = 0 - self._decompressor = _get_decompressor(self._compress_type) - self._eof = False - read_offset = new_pos - if self._decrypter is not None: - self._init_decrypter() - - while read_offset > 0: - read_len = min(self.MAX_SEEK_READ, read_offset) - self.read(read_len) - read_offset -= read_len - - return self.tell() - - def tell(self): - if self.closed: - raise ValueError("tell on closed file.") - if not self._seekable: - raise io.UnsupportedOperation("underlying stream is not seekable") - filepos = self._orig_file_size - self._left - len(self._readbuffer) + self._offset - return filepos - - -class _ZipWriteFile(io.BufferedIOBase): - def __init__(self, zf, zinfo, zip64): - self._zinfo = zinfo - self._zip64 = zip64 - self._zipfile = zf - self._compressor = _get_compressor(zinfo.compress_type, - zinfo._compresslevel) - self._file_size = 0 - self._compress_size = 0 - self._crc = 0 - - @property - def _fileobj(self): - return self._zipfile.fp - - def writable(self): - return True - - def write(self, data): - if self.closed: - raise ValueError('I/O operation on closed file.') - - # Accept any data that supports the buffer protocol - if isinstance(data, (bytes, bytearray)): - nbytes = len(data) - else: - data = memoryview(data) - nbytes = data.nbytes - self._file_size += nbytes - - self._crc = crc32(data, self._crc) - if self._compressor: - data = self._compressor.compress(data) - self._compress_size += len(data) - self._fileobj.write(data) - return nbytes - - def close(self): - if self.closed: - return - try: - super().close() - # Flush any data from the compressor, and update header info - if self._compressor: - buf = self._compressor.flush() - self._compress_size += len(buf) - self._fileobj.write(buf) - self._zinfo.compress_size = self._compress_size - else: - self._zinfo.compress_size = self._file_size - self._zinfo.CRC = self._crc - self._zinfo.file_size = self._file_size - - # Write updated header info - if self._zinfo.flag_bits & _MASK_USE_DATA_DESCRIPTOR: - # Write CRC and file sizes after the file data - fmt = ' ZIP64_LIMIT: - raise RuntimeError( - 'File size unexpectedly exceeded ZIP64 limit') - if self._compress_size > ZIP64_LIMIT: - raise RuntimeError( - 'Compressed size unexpectedly exceeded ZIP64 limit') - # Seek backwards and write file header (which will now include - # correct CRC and file sizes) - - # Preserve current position in file - self._zipfile.start_dir = self._fileobj.tell() - self._fileobj.seek(self._zinfo.header_offset) - self._fileobj.write(self._zinfo.FileHeader(self._zip64)) - self._fileobj.seek(self._zipfile.start_dir) - - # Successfully written: Add file to our caches - self._zipfile.filelist.append(self._zinfo) - self._zipfile.NameToInfo[self._zinfo.filename] = self._zinfo - finally: - self._zipfile._writing = False - - - -class ZipFile: - """ Class with methods to open, read, write, close, list zip files. - - z = ZipFile(file, mode="r", compression=ZIP_STORED, allowZip64=True, - compresslevel=None) - - file: Either the path to the file, or a file-like object. - If it is a path, the file will be opened and closed by ZipFile. - mode: The mode can be either read 'r', write 'w', exclusive create 'x', - or append 'a'. - compression: ZIP_STORED (no compression), ZIP_DEFLATED (requires zlib), - ZIP_BZIP2 (requires bz2) or ZIP_LZMA (requires lzma). - allowZip64: if True ZipFile will create files with ZIP64 extensions when - needed, otherwise it will raise an exception when this would - be necessary. - compresslevel: None (default for the given compression type) or an integer - specifying the level to pass to the compressor. - When using ZIP_STORED or ZIP_LZMA this keyword has no effect. - When using ZIP_DEFLATED integers 0 through 9 are accepted. - When using ZIP_BZIP2 integers 1 through 9 are accepted. - - """ - - fp = None # Set here since __del__ checks it - _windows_illegal_name_trans_table = None - - def __init__(self, file, mode="r", compression=ZIP_STORED, allowZip64=True, - compresslevel=None, *, strict_timestamps=True, metadata_encoding=None): - """Open the ZIP file with mode read 'r', write 'w', exclusive create 'x', - or append 'a'.""" - if mode not in ('r', 'w', 'x', 'a'): - raise ValueError("ZipFile requires mode 'r', 'w', 'x', or 'a'") - - _check_compression(compression) - - self._allowZip64 = allowZip64 - self._didModify = False - self.debug = 0 # Level of printing: 0 through 3 - self.NameToInfo = {} # Find file info given name - self.filelist = [] # List of ZipInfo instances for archive - self.compression = compression # Method of compression - self.compresslevel = compresslevel - self.mode = mode - self.pwd = None - self._comment = b'' - self._strict_timestamps = strict_timestamps - self.metadata_encoding = metadata_encoding - - # Check that we don't try to write with nonconforming codecs - if self.metadata_encoding and mode != 'r': - raise ValueError( - "metadata_encoding is only supported for reading files") - - # Check if we were passed a file-like object - if isinstance(file, os.PathLike): - file = os.fspath(file) - if isinstance(file, str): - # No, it's a filename - self._filePassed = 0 - self.filename = file - modeDict = {'r' : 'rb', 'w': 'w+b', 'x': 'x+b', 'a' : 'r+b', - 'r+b': 'w+b', 'w+b': 'wb', 'x+b': 'xb'} - filemode = modeDict[mode] - while True: - try: - self.fp = io.open(file, filemode) - except OSError: - if filemode in modeDict: - filemode = modeDict[filemode] - continue - raise - break - else: - self._filePassed = 1 - self.fp = file - self.filename = getattr(file, 'name', None) - self._fileRefCnt = 1 - self._lock = threading.RLock() - self._seekable = True - self._writing = False - - try: - if mode == 'r': - self._RealGetContents() - elif mode in ('w', 'x'): - # set the modified flag so central directory gets written - # even if no files are added to the archive - self._didModify = True - try: - self.start_dir = self.fp.tell() - except (AttributeError, OSError): - self.fp = _Tellable(self.fp) - self.start_dir = 0 - self._seekable = False - else: - # Some file-like objects can provide tell() but not seek() - try: - self.fp.seek(self.start_dir) - except (AttributeError, OSError): - self._seekable = False - elif mode == 'a': - try: - # See if file is a zip file - self._RealGetContents() - # seek to start of directory and overwrite - self.fp.seek(self.start_dir) - except BadZipFile: - # file is not a zip file, just append - self.fp.seek(0, 2) - - # set the modified flag so central directory gets written - # even if no files are added to the archive - self._didModify = True - self.start_dir = self.fp.tell() - else: - raise ValueError("Mode must be 'r', 'w', 'x', or 'a'") - except: - fp = self.fp - self.fp = None - self._fpclose(fp) - raise - - def __enter__(self): - return self - - def __exit__(self, type, value, traceback): - self.close() - - def __repr__(self): - result = ['<%s.%s' % (self.__class__.__module__, - self.__class__.__qualname__)] - if self.fp is not None: - if self._filePassed: - result.append(' file=%r' % self.fp) - elif self.filename is not None: - result.append(' filename=%r' % self.filename) - result.append(' mode=%r' % self.mode) - else: - result.append(' [closed]') - result.append('>') - return ''.join(result) - - def _RealGetContents(self): - """Read in the table of contents for the ZIP file.""" - fp = self.fp - try: - endrec = _EndRecData(fp) - except OSError: - raise BadZipFile("File is not a zip file") - if not endrec: - raise BadZipFile("File is not a zip file") - if self.debug > 1: - print(endrec) - size_cd = endrec[_ECD_SIZE] # bytes in central directory - offset_cd = endrec[_ECD_OFFSET] # offset of central directory - self._comment = endrec[_ECD_COMMENT] # archive comment - - # "concat" is zero, unless zip was concatenated to another file - concat = endrec[_ECD_LOCATION] - size_cd - offset_cd - if endrec[_ECD_SIGNATURE] == stringEndArchive64: - # If Zip64 extension structures are present, account for them - concat -= (sizeEndCentDir64 + sizeEndCentDir64Locator) - - if self.debug > 2: - inferred = concat + offset_cd - print("given, inferred, offset", offset_cd, inferred, concat) - # self.start_dir: Position of start of central directory - self.start_dir = offset_cd + concat - if self.start_dir < 0: - raise BadZipFile("Bad offset for central directory") - fp.seek(self.start_dir, 0) - data = fp.read(size_cd) - fp = io.BytesIO(data) - total = 0 - while total < size_cd: - centdir = fp.read(sizeCentralDir) - if len(centdir) != sizeCentralDir: - raise BadZipFile("Truncated central directory") - centdir = struct.unpack(structCentralDir, centdir) - if centdir[_CD_SIGNATURE] != stringCentralDir: - raise BadZipFile("Bad magic number for central directory") - if self.debug > 2: - print(centdir) - filename = fp.read(centdir[_CD_FILENAME_LENGTH]) - flags = centdir[_CD_FLAG_BITS] - if flags & _MASK_UTF_FILENAME: - # UTF-8 file names extension - filename = filename.decode('utf-8') - else: - # Historical ZIP filename encoding - filename = filename.decode(self.metadata_encoding or 'cp437') - # Create ZipInfo instance to store file information - x = ZipInfo(filename) - x.extra = fp.read(centdir[_CD_EXTRA_FIELD_LENGTH]) - x.comment = fp.read(centdir[_CD_COMMENT_LENGTH]) - x.header_offset = centdir[_CD_LOCAL_HEADER_OFFSET] - (x.create_version, x.create_system, x.extract_version, x.reserved, - x.flag_bits, x.compress_type, t, d, - x.CRC, x.compress_size, x.file_size) = centdir[1:12] - if x.extract_version > MAX_EXTRACT_VERSION: - raise NotImplementedError("zip file version %.1f" % - (x.extract_version / 10)) - x.volume, x.internal_attr, x.external_attr = centdir[15:18] - # Convert date/time code to (year, month, day, hour, min, sec) - x._raw_time = t - x.date_time = ( (d>>9)+1980, (d>>5)&0xF, d&0x1F, - t>>11, (t>>5)&0x3F, (t&0x1F) * 2 ) - - x._decodeExtra() - x.header_offset = x.header_offset + concat - self.filelist.append(x) - self.NameToInfo[x.filename] = x - - # update total bytes read from central directory - total = (total + sizeCentralDir + centdir[_CD_FILENAME_LENGTH] - + centdir[_CD_EXTRA_FIELD_LENGTH] - + centdir[_CD_COMMENT_LENGTH]) - - if self.debug > 2: - print("total", total) - - - def namelist(self): - """Return a list of file names in the archive.""" - return [data.filename for data in self.filelist] - - def infolist(self): - """Return a list of class ZipInfo instances for files in the - archive.""" - return self.filelist - - def printdir(self, file=None): - """Print a table of contents for the zip file.""" - print("%-46s %19s %12s" % ("File Name", "Modified ", "Size"), - file=file) - for zinfo in self.filelist: - date = "%d-%02d-%02d %02d:%02d:%02d" % zinfo.date_time[:6] - print("%-46s %s %12d" % (zinfo.filename, date, zinfo.file_size), - file=file) - - def testzip(self): - """Read all the files and check the CRC. - - Return None if all files could be read successfully, or the name - of the offending file otherwise.""" - chunk_size = 2 ** 20 - for zinfo in self.filelist: - try: - # Read by chunks, to avoid an OverflowError or a - # MemoryError with very large embedded files. - with self.open(zinfo.filename, "r") as f: - while f.read(chunk_size): # Check CRC-32 - pass - except BadZipFile: - return zinfo.filename - - def getinfo(self, name): - """Return the instance of ZipInfo given 'name'.""" - info = self.NameToInfo.get(name) - if info is None: - raise KeyError( - 'There is no item named %r in the archive' % name) - - return info - - def setpassword(self, pwd): - """Set default password for encrypted files.""" - if pwd and not isinstance(pwd, bytes): - raise TypeError("pwd: expected bytes, got %s" % type(pwd).__name__) - if pwd: - self.pwd = pwd - else: - self.pwd = None - - @property - def comment(self): - """The comment text associated with the ZIP file.""" - return self._comment - - @comment.setter - def comment(self, comment): - if not isinstance(comment, bytes): - raise TypeError("comment: expected bytes, got %s" % type(comment).__name__) - # check for valid comment length - if len(comment) > ZIP_MAX_COMMENT: - import warnings - warnings.warn('Archive comment is too long; truncating to %d bytes' - % ZIP_MAX_COMMENT, stacklevel=2) - comment = comment[:ZIP_MAX_COMMENT] - self._comment = comment - self._didModify = True - - def read(self, name, pwd=None): - """Return file bytes for name.""" - with self.open(name, "r", pwd) as fp: - return fp.read() - - def open(self, name, mode="r", pwd=None, *, force_zip64=False): - """Return file-like object for 'name'. - - name is a string for the file name within the ZIP file, or a ZipInfo - object. - - mode should be 'r' to read a file already in the ZIP file, or 'w' to - write to a file newly added to the archive. - - pwd is the password to decrypt files (only used for reading). - - When writing, if the file size is not known in advance but may exceed - 2 GiB, pass force_zip64 to use the ZIP64 format, which can handle large - files. If the size is known in advance, it is best to pass a ZipInfo - instance for name, with zinfo.file_size set. - """ - if mode not in {"r", "w"}: - raise ValueError('open() requires mode "r" or "w"') - if pwd and (mode == "w"): - raise ValueError("pwd is only supported for reading files") - if not self.fp: - raise ValueError( - "Attempt to use ZIP archive that was already closed") - - # Make sure we have an info object - if isinstance(name, ZipInfo): - # 'name' is already an info object - zinfo = name - elif mode == 'w': - zinfo = ZipInfo(name) - zinfo.compress_type = self.compression - zinfo._compresslevel = self.compresslevel - else: - # Get info object for name - zinfo = self.getinfo(name) - - if mode == 'w': - return self._open_to_write(zinfo, force_zip64=force_zip64) - - 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.") - - # Open for reading: - self._fileRefCnt += 1 - zef_file = _SharedFile(self.fp, zinfo.header_offset, - self._fpclose, self._lock, lambda: self._writing) - try: - # Skip the file header: - fheader = zef_file.read(sizeFileHeader) - if len(fheader) != sizeFileHeader: - raise BadZipFile("Truncated file header") - fheader = struct.unpack(structFileHeader, fheader) - if fheader[_FH_SIGNATURE] != stringFileHeader: - raise BadZipFile("Bad magic number for file header") - - fname = zef_file.read(fheader[_FH_FILENAME_LENGTH]) - if fheader[_FH_EXTRA_FIELD_LENGTH]: - zef_file.seek(fheader[_FH_EXTRA_FIELD_LENGTH], whence=1) - - if zinfo.flag_bits & _MASK_COMPRESSED_PATCH: - # Zip 2.7: compressed patched data - raise NotImplementedError("compressed patched data (flag bit 5)") - - if zinfo.flag_bits & _MASK_STRONG_ENCRYPTION: - # strong encryption - raise NotImplementedError("strong encryption (flag bit 6)") - - if fheader[_FH_GENERAL_PURPOSE_FLAG_BITS] & _MASK_UTF_FILENAME: - # UTF-8 filename - fname_str = fname.decode("utf-8") - else: - fname_str = fname.decode(self.metadata_encoding or "cp437") - - if fname_str != zinfo.orig_filename: - raise BadZipFile( - 'File name in directory %r and header %r differ.' - % (zinfo.orig_filename, fname)) - - # check for encrypted flag & handle password - is_encrypted = zinfo.flag_bits & _MASK_ENCRYPTED - if is_encrypted: - if not pwd: - pwd = self.pwd - if pwd and not isinstance(pwd, bytes): - raise TypeError("pwd: expected bytes, got %s" % type(pwd).__name__) - if not pwd: - raise RuntimeError("File %r is encrypted, password " - "required for extraction" % name) - else: - pwd = None - - return ZipExtFile(zef_file, mode, zinfo, pwd, True) - except: - zef_file.close() - raise - - def _open_to_write(self, zinfo, force_zip64=False): - if force_zip64 and not self._allowZip64: - raise ValueError( - "force_zip64 is True, but allowZip64 was False when opening " - "the ZIP file." - ) - if self._writing: - raise ValueError("Can't write to the ZIP file while there is " - "another write handle open on it. " - "Close the first handle before opening another.") - - # Size and CRC are overwritten with correct data after processing the file - zinfo.compress_size = 0 - zinfo.CRC = 0 - - zinfo.flag_bits = 0x00 - if zinfo.compress_type == ZIP_LZMA: - # Compressed data includes an end-of-stream (EOS) marker - zinfo.flag_bits |= _MASK_COMPRESS_OPTION_1 - if not self._seekable: - zinfo.flag_bits |= _MASK_USE_DATA_DESCRIPTOR - - if not zinfo.external_attr: - zinfo.external_attr = 0o600 << 16 # permissions: ?rw------- - - # Compressed size can be larger than uncompressed size - zip64 = self._allowZip64 and \ - (force_zip64 or zinfo.file_size * 1.05 > ZIP64_LIMIT) - - if self._seekable: - self.fp.seek(self.start_dir) - zinfo.header_offset = self.fp.tell() - - self._writecheck(zinfo) - self._didModify = True - - self.fp.write(zinfo.FileHeader(zip64)) - - self._writing = True - return _ZipWriteFile(self, zinfo, zip64) - - def extract(self, member, path=None, pwd=None): - """Extract a member from the archive to the current working directory, - using its full name. Its file information is extracted as accurately - as possible. `member' may be a filename or a ZipInfo object. You can - specify a different directory using `path'. - """ - if path is None: - path = os.getcwd() - else: - path = os.fspath(path) - - return self._extract_member(member, path, pwd) - - def extractall(self, path=None, members=None, pwd=None): - """Extract all members from the archive to the current working - directory. `path' specifies a different directory to extract to. - `members' is optional and must be a subset of the list returned - by namelist(). - """ - if members is None: - members = self.namelist() - - if path is None: - path = os.getcwd() - else: - path = os.fspath(path) - - for zipinfo in members: - self._extract_member(zipinfo, path, pwd) - - @classmethod - def _sanitize_windows_name(cls, arcname, pathsep): - """Replace bad characters and remove trailing dots from parts.""" - table = cls._windows_illegal_name_trans_table - if not table: - illegal = ':<>|"?*' - table = str.maketrans(illegal, '_' * len(illegal)) - cls._windows_illegal_name_trans_table = table - arcname = arcname.translate(table) - # remove trailing dots and spaces - arcname = (x.rstrip(' .') for x in arcname.split(pathsep)) - # rejoin, removing empty parts. - arcname = pathsep.join(x for x in arcname if x) - return arcname - - def _extract_member(self, member, targetpath, pwd): - """Extract the ZipInfo object 'member' to a physical - file on the path targetpath. - """ - if not isinstance(member, ZipInfo): - member = self.getinfo(member) - - # build the destination pathname, replacing - # forward slashes to platform specific separators. - arcname = member.filename.replace('/', os.path.sep) - - if os.path.altsep: - arcname = arcname.replace(os.path.altsep, os.path.sep) - # interpret absolute pathname as relative, remove drive letter or - # UNC path, redundant separators, "." and ".." components. - arcname = os.path.splitdrive(arcname)[1] - invalid_path_parts = ('', os.path.curdir, os.path.pardir) - arcname = os.path.sep.join(x for x in arcname.split(os.path.sep) - if x not in invalid_path_parts) - if os.path.sep == '\\': - # filter illegal characters on Windows - arcname = self._sanitize_windows_name(arcname, os.path.sep) - - if not arcname: - raise ValueError("Empty filename.") - - targetpath = os.path.join(targetpath, arcname) - targetpath = os.path.normpath(targetpath) - - # Create all upper directories if necessary. - upperdirs = os.path.dirname(targetpath) - if upperdirs and not os.path.exists(upperdirs): - os.makedirs(upperdirs) - - if member.is_dir(): - if not os.path.isdir(targetpath): - os.mkdir(targetpath) - return targetpath - - with self.open(member, pwd=pwd) as source, \ - open(targetpath, "wb") as target: - shutil.copyfileobj(source, target) - - return targetpath - - def _writecheck(self, zinfo): - """Check for errors before writing a file to the archive.""" - if zinfo.filename in self.NameToInfo: - import warnings - warnings.warn('Duplicate name: %r' % zinfo.filename, stacklevel=3) - if self.mode not in ('w', 'x', 'a'): - raise ValueError("write() requires mode 'w', 'x', or 'a'") - if not self.fp: - raise ValueError( - "Attempt to write ZIP archive that was already closed") - _check_compression(zinfo.compress_type) - if not self._allowZip64: - requires_zip64 = None - if len(self.filelist) >= ZIP_FILECOUNT_LIMIT: - requires_zip64 = "Files count" - elif zinfo.file_size > ZIP64_LIMIT: - requires_zip64 = "Filesize" - elif zinfo.header_offset > ZIP64_LIMIT: - requires_zip64 = "Zipfile size" - if requires_zip64: - raise LargeZipFile(requires_zip64 + - " would require ZIP64 extensions") - - def write(self, filename, arcname=None, - compress_type=None, compresslevel=None): - """Put the bytes from filename into the archive under the name - arcname.""" - if not self.fp: - raise ValueError( - "Attempt to write to ZIP archive that was already closed") - if self._writing: - raise ValueError( - "Can't write to ZIP archive while an open writing handle exists" - ) - - zinfo = ZipInfo.from_file(filename, arcname, - strict_timestamps=self._strict_timestamps) - - if zinfo.is_dir(): - zinfo.compress_size = 0 - zinfo.CRC = 0 - self.mkdir(zinfo) - else: - if compress_type is not None: - zinfo.compress_type = compress_type - else: - zinfo.compress_type = self.compression - - if compresslevel is not None: - zinfo._compresslevel = compresslevel - else: - zinfo._compresslevel = self.compresslevel - - with open(filename, "rb") as src, self.open(zinfo, 'w') as dest: - shutil.copyfileobj(src, dest, 1024*8) - - def writestr(self, zinfo_or_arcname, data, - compress_type=None, compresslevel=None): - """Write a file into the archive. The contents is 'data', which - may be either a 'str' or a 'bytes' instance; if it is a 'str', - it is encoded as UTF-8 first. - 'zinfo_or_arcname' is either a ZipInfo instance or - the name of the file in the archive.""" - if isinstance(data, str): - data = data.encode("utf-8") - if not isinstance(zinfo_or_arcname, ZipInfo): - zinfo = ZipInfo(filename=zinfo_or_arcname, - date_time=time.localtime(time.time())[:6]) - zinfo.compress_type = self.compression - zinfo._compresslevel = self.compresslevel - if zinfo.filename.endswith('/'): - zinfo.external_attr = 0o40775 << 16 # drwxrwxr-x - zinfo.external_attr |= 0x10 # MS-DOS directory flag - else: - zinfo.external_attr = 0o600 << 16 # ?rw------- - else: - zinfo = zinfo_or_arcname - - if not self.fp: - raise ValueError( - "Attempt to write to ZIP archive that was already closed") - if self._writing: - raise ValueError( - "Can't write to ZIP archive while an open writing handle exists." - ) - - if compress_type is not None: - zinfo.compress_type = compress_type - - if compresslevel is not None: - zinfo._compresslevel = compresslevel - - zinfo.file_size = len(data) # Uncompressed size - with self._lock: - with self.open(zinfo, mode='w') as dest: - dest.write(data) - - def mkdir(self, zinfo_or_directory_name, mode=511): - """Creates a directory inside the zip archive.""" - if isinstance(zinfo_or_directory_name, ZipInfo): - zinfo = zinfo_or_directory_name - if not zinfo.is_dir(): - raise ValueError("The given ZipInfo does not describe a directory") - elif isinstance(zinfo_or_directory_name, str): - directory_name = zinfo_or_directory_name - if not directory_name.endswith("/"): - directory_name += "/" - zinfo = ZipInfo(directory_name) - zinfo.compress_size = 0 - zinfo.CRC = 0 - zinfo.external_attr = ((0o40000 | mode) & 0xFFFF) << 16 - zinfo.file_size = 0 - zinfo.external_attr |= 0x10 - else: - raise TypeError("Expected type str or ZipInfo") - - with self._lock: - if self._seekable: - self.fp.seek(self.start_dir) - zinfo.header_offset = self.fp.tell() # Start of header bytes - if zinfo.compress_type == ZIP_LZMA: - # Compressed data includes an end-of-stream (EOS) marker - zinfo.flag_bits |= _MASK_COMPRESS_OPTION_1 - - self._writecheck(zinfo) - self._didModify = True - - self.filelist.append(zinfo) - self.NameToInfo[zinfo.filename] = zinfo - self.fp.write(zinfo.FileHeader(False)) - self.start_dir = self.fp.tell() - - def __del__(self): - """Call the "close()" method in case the user forgot.""" - self.close() - - def close(self): - """Close the file, and for mode 'w', 'x' and 'a' write the ending - records.""" - if self.fp is None: - return - - if self._writing: - raise ValueError("Can't close the ZIP file while there is " - "an open writing handle on it. " - "Close the writing handle before closing the zip.") - - try: - if self.mode in ('w', 'x', 'a') and self._didModify: # write ending records - with self._lock: - if self._seekable: - self.fp.seek(self.start_dir) - self._write_end_record() - finally: - fp = self.fp - self.fp = None - self._fpclose(fp) - - def _write_end_record(self): - for zinfo in self.filelist: # write central directory - dt = zinfo.date_time - dosdate = (dt[0] - 1980) << 9 | dt[1] << 5 | dt[2] - dostime = dt[3] << 11 | dt[4] << 5 | (dt[5] // 2) - extra = [] - if zinfo.file_size > ZIP64_LIMIT \ - or zinfo.compress_size > ZIP64_LIMIT: - extra.append(zinfo.file_size) - extra.append(zinfo.compress_size) - file_size = 0xffffffff - compress_size = 0xffffffff - else: - file_size = zinfo.file_size - compress_size = zinfo.compress_size - - if zinfo.header_offset > ZIP64_LIMIT: - extra.append(zinfo.header_offset) - header_offset = 0xffffffff - else: - header_offset = zinfo.header_offset - - extra_data = zinfo.extra - min_version = 0 - if extra: - # Append a ZIP64 field to the extra's - extra_data = _strip_extra(extra_data, (1,)) - extra_data = struct.pack( - ' ZIP_FILECOUNT_LIMIT: - requires_zip64 = "Files count" - elif centDirOffset > ZIP64_LIMIT: - requires_zip64 = "Central directory offset" - elif centDirSize > ZIP64_LIMIT: - requires_zip64 = "Central directory size" - if requires_zip64: - # Need to write the ZIP64 end-of-archive records - if not self._allowZip64: - raise LargeZipFile(requires_zip64 + - " would require ZIP64 extensions") - zip64endrec = struct.pack( - structEndArchive64, stringEndArchive64, - 44, 45, 45, 0, 0, centDirCount, centDirCount, - centDirSize, centDirOffset) - self.fp.write(zip64endrec) - - zip64locrec = struct.pack( - structEndArchive64Locator, - stringEndArchive64Locator, 0, pos2, 1) - self.fp.write(zip64locrec) - centDirCount = min(centDirCount, 0xFFFF) - centDirSize = min(centDirSize, 0xFFFFFFFF) - centDirOffset = min(centDirOffset, 0xFFFFFFFF) - - endrec = struct.pack(structEndArchive, stringEndArchive, - 0, 0, centDirCount, centDirCount, - centDirSize, centDirOffset, len(self._comment)) - self.fp.write(endrec) - self.fp.write(self._comment) - if self.mode == "a": - self.fp.truncate() - self.fp.flush() - - def _fpclose(self, fp): - assert self._fileRefCnt > 0 - self._fileRefCnt -= 1 - if not self._fileRefCnt and not self._filePassed: - fp.close() - - -class PyZipFile(ZipFile): - """Class to create ZIP archives with Python library files and packages.""" - - def __init__(self, file, mode="r", compression=ZIP_STORED, - allowZip64=True, optimize=-1): - ZipFile.__init__(self, file, mode=mode, compression=compression, - allowZip64=allowZip64) - self._optimize = optimize - - def writepy(self, pathname, basename="", filterfunc=None): - """Add all files from "pathname" to the ZIP archive. - - If pathname is a package directory, search the directory and - all package subdirectories recursively for all *.py and enter - the modules into the archive. If pathname is a plain - directory, listdir *.py and enter all modules. Else, pathname - must be a Python *.py file and the module will be put into the - archive. Added modules are always module.pyc. - This method will compile the module.py into module.pyc if - necessary. - If filterfunc(pathname) is given, it is called with every argument. - When it is False, the file or directory is skipped. - """ - pathname = os.fspath(pathname) - if filterfunc and not filterfunc(pathname): - if self.debug: - label = 'path' if os.path.isdir(pathname) else 'file' - print('%s %r skipped by filterfunc' % (label, pathname)) - return - dir, name = os.path.split(pathname) - if os.path.isdir(pathname): - initname = os.path.join(pathname, "__init__.py") - if os.path.isfile(initname): - # This is a package directory, add it - if basename: - basename = "%s/%s" % (basename, name) - else: - basename = name - if self.debug: - print("Adding package in", pathname, "as", basename) - fname, arcname = self._get_codename(initname[0:-3], basename) - if self.debug: - print("Adding", arcname) - self.write(fname, arcname) - dirlist = sorted(os.listdir(pathname)) - dirlist.remove("__init__.py") - # Add all *.py files and package subdirectories - for filename in dirlist: - path = os.path.join(pathname, filename) - root, ext = os.path.splitext(filename) - if os.path.isdir(path): - if os.path.isfile(os.path.join(path, "__init__.py")): - # This is a package directory, add it - self.writepy(path, basename, - filterfunc=filterfunc) # Recursive call - elif ext == ".py": - if filterfunc and not filterfunc(path): - if self.debug: - print('file %r skipped by filterfunc' % path) - continue - fname, arcname = self._get_codename(path[0:-3], - basename) - if self.debug: - print("Adding", arcname) - self.write(fname, arcname) - else: - # This is NOT a package directory, add its files at top level - if self.debug: - print("Adding files from directory", pathname) - for filename in sorted(os.listdir(pathname)): - path = os.path.join(pathname, filename) - root, ext = os.path.splitext(filename) - if ext == ".py": - if filterfunc and not filterfunc(path): - if self.debug: - print('file %r skipped by filterfunc' % path) - continue - fname, arcname = self._get_codename(path[0:-3], - basename) - if self.debug: - print("Adding", arcname) - self.write(fname, arcname) - else: - if pathname[-3:] != ".py": - raise RuntimeError( - 'Files added with writepy() must end with ".py"') - fname, arcname = self._get_codename(pathname[0:-3], basename) - if self.debug: - print("Adding file", arcname) - self.write(fname, arcname) - - def _get_codename(self, pathname, basename): - """Return (filename, archivename) for the path. - - Given a module name path, return the correct file path and - archive name, compiling if necessary. For example, given - /python/lib/string, return (/python/lib/string.pyc, string). - """ - def _compile(file, optimize=-1): - import py_compile - if self.debug: - print("Compiling", file) - try: - py_compile.compile(file, doraise=True, optimize=optimize) - except py_compile.PyCompileError as err: - print(err.msg) - return False - return True - - file_py = pathname + ".py" - file_pyc = pathname + ".pyc" - pycache_opt0 = importlib.util.cache_from_source(file_py, optimization='') - pycache_opt1 = importlib.util.cache_from_source(file_py, optimization=1) - pycache_opt2 = importlib.util.cache_from_source(file_py, optimization=2) - if self._optimize == -1: - # legacy mode: use whatever file is present - if (os.path.isfile(file_pyc) and - os.stat(file_pyc).st_mtime >= os.stat(file_py).st_mtime): - # Use .pyc file. - arcname = fname = file_pyc - elif (os.path.isfile(pycache_opt0) and - os.stat(pycache_opt0).st_mtime >= os.stat(file_py).st_mtime): - # Use the __pycache__/*.pyc file, but write it to the legacy pyc - # file name in the archive. - fname = pycache_opt0 - arcname = file_pyc - elif (os.path.isfile(pycache_opt1) and - os.stat(pycache_opt1).st_mtime >= os.stat(file_py).st_mtime): - # Use the __pycache__/*.pyc file, but write it to the legacy pyc - # file name in the archive. - fname = pycache_opt1 - arcname = file_pyc - elif (os.path.isfile(pycache_opt2) and - os.stat(pycache_opt2).st_mtime >= os.stat(file_py).st_mtime): - # Use the __pycache__/*.pyc file, but write it to the legacy pyc - # file name in the archive. - fname = pycache_opt2 - arcname = file_pyc - else: - # Compile py into PEP 3147 pyc file. - if _compile(file_py): - if sys.flags.optimize == 0: - fname = pycache_opt0 - elif sys.flags.optimize == 1: - fname = pycache_opt1 - else: - fname = pycache_opt2 - arcname = file_pyc - else: - fname = arcname = file_py - else: - # new mode: use given optimization level - if self._optimize == 0: - fname = pycache_opt0 - arcname = file_pyc - else: - arcname = file_pyc - if self._optimize == 1: - fname = pycache_opt1 - elif self._optimize == 2: - fname = pycache_opt2 - else: - msg = "invalid value for 'optimize': {!r}".format(self._optimize) - raise ValueError(msg) - if not (os.path.isfile(fname) and - os.stat(fname).st_mtime >= os.stat(file_py).st_mtime): - if not _compile(file_py, optimize=self._optimize): - fname = arcname = file_py - archivename = os.path.split(arcname)[1] - if basename: - archivename = "%s/%s" % (basename, archivename) - return (fname, archivename) - - -def _parents(path): - """ - Given a path with elements separated by - posixpath.sep, generate all parents of that path. - - >>> list(_parents('b/d')) - ['b'] - >>> list(_parents('/b/d/')) - ['/b'] - >>> list(_parents('b/d/f/')) - ['b/d', 'b'] - >>> list(_parents('b')) - [] - >>> list(_parents('')) - [] - """ - return itertools.islice(_ancestry(path), 1, None) - - -def _ancestry(path): - """ - Given a path with elements separated by - posixpath.sep, generate all elements of that path - - >>> list(_ancestry('b/d')) - ['b/d', 'b'] - >>> list(_ancestry('/b/d/')) - ['/b/d', '/b'] - >>> list(_ancestry('b/d/f/')) - ['b/d/f', 'b/d', 'b'] - >>> list(_ancestry('b')) - ['b'] - >>> list(_ancestry('')) - [] - """ - path = path.rstrip(posixpath.sep) - while path and path != posixpath.sep: - yield path - path, tail = posixpath.split(path) - - -_dedupe = dict.fromkeys -"""Deduplicate an iterable in original order""" - - -def _difference(minuend, subtrahend): - """ - Return items in minuend not in subtrahend, retaining order - with O(1) lookup. - """ - return itertools.filterfalse(set(subtrahend).__contains__, minuend) - - -class CompleteDirs(ZipFile): - """ - A ZipFile subclass that ensures that implied directories - are always included in the namelist. - """ - - @staticmethod - def _implied_dirs(names): - parents = itertools.chain.from_iterable(map(_parents, names)) - as_dirs = (p + posixpath.sep for p in parents) - return _dedupe(_difference(as_dirs, names)) - - def namelist(self): - names = super(CompleteDirs, self).namelist() - return names + list(self._implied_dirs(names)) - - def _name_set(self): - return set(self.namelist()) - - def resolve_dir(self, name): - """ - If the name represents a directory, return that name - as a directory (with the trailing slash). - """ - names = self._name_set() - dirname = name + '/' - dir_match = name not in names and dirname in names - return dirname if dir_match else name - - @classmethod - def make(cls, source): - """ - Given a source (filename or zipfile), return an - appropriate CompleteDirs subclass. - """ - if isinstance(source, CompleteDirs): - return source - - if not isinstance(source, ZipFile): - return cls(source) - - # Only allow for FastLookup when supplied zipfile is read-only - if 'r' not in source.mode: - cls = CompleteDirs - - source.__class__ = cls - return source - - -class FastLookup(CompleteDirs): - """ - ZipFile subclass to ensure implicit - dirs exist and are resolved rapidly. - """ - - def namelist(self): - with contextlib.suppress(AttributeError): - return self.__names - self.__names = super(FastLookup, self).namelist() - return self.__names - - def _name_set(self): - with contextlib.suppress(AttributeError): - return self.__lookup - self.__lookup = super(FastLookup, self)._name_set() - return self.__lookup - - -class Path: - """ - A pathlib-compatible interface for zip files. - - Consider a zip file with this structure:: - - . - ├── a.txt - └── b - ├── c.txt - └── d - └── e.txt - - >>> data = io.BytesIO() - >>> zf = ZipFile(data, 'w') - >>> zf.writestr('a.txt', 'content of a') - >>> zf.writestr('b/c.txt', 'content of c') - >>> zf.writestr('b/d/e.txt', 'content of e') - >>> zf.filename = 'mem/abcde.zip' - - Path accepts the zipfile object itself or a filename - - >>> root = Path(zf) - - From there, several path operations are available. - - Directory iteration (including the zip file itself): - - >>> a, b = root.iterdir() - >>> a - Path('mem/abcde.zip', 'a.txt') - >>> b - Path('mem/abcde.zip', 'b/') - - name property: - - >>> b.name - 'b' - - join with divide operator: - - >>> c = b / 'c.txt' - >>> c - Path('mem/abcde.zip', 'b/c.txt') - >>> c.name - 'c.txt' - - Read text: - - >>> c.read_text() - 'content of c' - - existence: - - >>> c.exists() - True - >>> (b / 'missing.txt').exists() - False - - Coercion to string: - - >>> import os - >>> str(c).replace(os.sep, posixpath.sep) - 'mem/abcde.zip/b/c.txt' - - At the root, ``name``, ``filename``, and ``parent`` - resolve to the zipfile. Note these attributes are not - valid and will raise a ``ValueError`` if the zipfile - has no filename. - - >>> root.name - 'abcde.zip' - >>> str(root.filename).replace(os.sep, posixpath.sep) - 'mem/abcde.zip' - >>> str(root.parent) - 'mem' - """ - - __repr = "{self.__class__.__name__}({self.root.filename!r}, {self.at!r})" - - def __init__(self, root, at=""): - """ - Construct a Path from a ZipFile or filename. - - Note: When the source is an existing ZipFile object, - its type (__class__) will be mutated to a - specialized type. If the caller wishes to retain the - original type, the caller should either create a - separate ZipFile object or pass a filename. - """ - self.root = FastLookup.make(root) - self.at = at - - def open(self, mode='r', *args, pwd=None, **kwargs): - """ - Open this entry as text or binary following the semantics - of ``pathlib.Path.open()`` by passing arguments through - to io.TextIOWrapper(). - """ - if self.is_dir(): - raise IsADirectoryError(self) - zip_mode = mode[0] - if not self.exists() and zip_mode == 'r': - raise FileNotFoundError(self) - stream = self.root.open(self.at, zip_mode, pwd=pwd) - if 'b' in mode: - if args or kwargs: - raise ValueError("encoding args invalid for binary operation") - return stream - else: - kwargs["encoding"] = io.text_encoding(kwargs.get("encoding")) - return io.TextIOWrapper(stream, *args, **kwargs) - - @property - def name(self): - return pathlib.Path(self.at).name or self.filename.name - - @property - def suffix(self): - return pathlib.Path(self.at).suffix or self.filename.suffix - - @property - def suffixes(self): - return pathlib.Path(self.at).suffixes or self.filename.suffixes - - @property - def stem(self): - return pathlib.Path(self.at).stem or self.filename.stem - - @property - def filename(self): - return pathlib.Path(self.root.filename).joinpath(self.at) - - def read_text(self, *args, **kwargs): - kwargs["encoding"] = io.text_encoding(kwargs.get("encoding")) - with self.open('r', *args, **kwargs) as strm: - return strm.read() - - def read_bytes(self): - with self.open('rb') as strm: - return strm.read() - - def _is_child(self, path): - return posixpath.dirname(path.at.rstrip("/")) == self.at.rstrip("/") - - def _next(self, at): - return self.__class__(self.root, at) - - def is_dir(self): - return not self.at or self.at.endswith("/") - - def is_file(self): - return self.exists() and not self.is_dir() - - def exists(self): - return self.at in self.root._name_set() - - def iterdir(self): - if not self.is_dir(): - raise ValueError("Can't listdir a file") - subs = map(self._next, self.root.namelist()) - return filter(self._is_child, subs) - - def __str__(self): - return posixpath.join(self.root.filename, self.at) - - def __repr__(self): - return self.__repr.format(self=self) - - def joinpath(self, *other): - next = posixpath.join(self.at, *other) - return self._next(self.root.resolve_dir(next)) - - __truediv__ = joinpath - - @property - def parent(self): - if not self.at: - return self.filename.parent - parent_at = posixpath.dirname(self.at.rstrip('/')) - if parent_at: - parent_at += '/' - return self._next(parent_at) - - -def main(args=None): - import argparse - - description = 'A simple command-line interface for zipfile module.' - parser = argparse.ArgumentParser(description=description) - group = parser.add_mutually_exclusive_group(required=True) - group.add_argument('-l', '--list', metavar='', - help='Show listing of a zipfile') - group.add_argument('-e', '--extract', nargs=2, - metavar=('', ''), - help='Extract zipfile into target dir') - group.add_argument('-c', '--create', nargs='+', - metavar=('', ''), - help='Create zipfile from sources') - group.add_argument('-t', '--test', metavar='', - help='Test if a zipfile is valid') - parser.add_argument('--metadata-encoding', metavar='', - help='Specify encoding of member names for -l, -e and -t') - args = parser.parse_args(args) - - encoding = args.metadata_encoding - - if args.test is not None: - src = args.test - with ZipFile(src, 'r', metadata_encoding=encoding) as zf: - badfile = zf.testzip() - if badfile: - print("The following enclosed file is corrupted: {!r}".format(badfile)) - print("Done testing") - - elif args.list is not None: - src = args.list - with ZipFile(src, 'r', metadata_encoding=encoding) as zf: - zf.printdir() - - elif args.extract is not None: - src, curdir = args.extract - with ZipFile(src, 'r', metadata_encoding=encoding) as zf: - zf.extractall(curdir) - - elif args.create is not None: - if encoding: - print("Non-conforming encodings not supported with -c.", - file=sys.stderr) - sys.exit(1) - - zip_name = args.create.pop(0) - files = args.create - - def addToZip(zf, path, zippath): - if os.path.isfile(path): - zf.write(path, zippath, ZIP_DEFLATED) - elif os.path.isdir(path): - if zippath: - zf.write(path, zippath) - for nm in sorted(os.listdir(path)): - addToZip(zf, - os.path.join(path, nm), os.path.join(zippath, nm)) - # else: ignore - - with ZipFile(zip_name, 'w') as zf: - for path in files: - zippath = os.path.basename(path) - if not zippath: - zippath = os.path.basename(os.path.dirname(path)) - if zippath in ('', os.curdir, os.pardir): - zippath = '' - addToZip(zf, path, zippath) - - -if __name__ == "__main__": - main() diff --git a/Lib/zipfile/__init__.py b/Lib/zipfile/__init__.py new file mode 100644 index 0000000..8f83426 --- /dev/null +++ b/Lib/zipfile/__init__.py @@ -0,0 +1,2193 @@ +""" +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(' 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() + 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 + return None + + +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', + ) + + 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 + # 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: + zip64 = file_size > ZIP64_LIMIT or compress_size > ZIP64_LIMIT + if zip64: + fmt = ' 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 | _MASK_UTF_FILENAME + + def _decodeExtra(self): + # Try to decode the extra field. + extra = self.extra + unpack = struct.unpack + while len(extra) >= 4: + tp, ln = unpack(' 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(' 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.""" + return self.filename.endswith('/') + + +# 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('> 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. + 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, + # decrypt and decompress them. + if self._eof or n <= 0: + return b'' + + # Read from file. + if self._compress_type == ZIP_DEFLATED: + ## Handle unconsumed data. + data = self._decompressor.unconsumed_tail + if n > len(data): + data += self._read2(n - len(data)) + else: + data = self._read2(n) + + if self._compress_type == ZIP_STORED: + self._eof = self._compress_left <= 0 + elif self._compress_type == ZIP_DEFLATED: + n = max(n, self.MIN_READ_SIZE) + data = self._decompressor.decompress(data, n) + self._eof = (self._decompressor.eof or + self._compress_left <= 0 and + not self._decompressor.unconsumed_tail) + if self._eof: + data += self._decompressor.flush() + else: + data = self._decompressor.decompress(data) + self._eof = self._decompressor.eof or self._compress_left <= 0 + + data = data[:self._left] + self._left -= len(data) + if self._left <= 0: + self._eof = True + self._update_crc(data) + return data + + def _read2(self, n): + if self._compress_left <= 0: + return b'' + + n = max(n, self.MIN_READ_SIZE) + n = min(n, self._compress_left) + + data = self._fileobj.read(n) + self._compress_left -= len(data) + if not data: + raise EOFError + + if self._decrypter is not None: + data = self._decrypter(data) + return data + + def close(self): + try: + if self._close_fileobj: + self._fileobj.close() + finally: + super().close() + + def seekable(self): + if self.closed: + raise ValueError("I/O operation on closed file.") + return self._seekable + + def seek(self, offset, whence=os.SEEK_SET): + if self.closed: + raise ValueError("seek on closed file.") + if not self._seekable: + raise io.UnsupportedOperation("underlying stream is not seekable") + curr_pos = self.tell() + if whence == os.SEEK_SET: + new_pos = offset + elif whence == os.SEEK_CUR: + new_pos = curr_pos + offset + elif whence == os.SEEK_END: + new_pos = self._orig_file_size + offset + else: + raise ValueError("whence must be os.SEEK_SET (0), " + "os.SEEK_CUR (1), or os.SEEK_END (2)") + + if new_pos > self._orig_file_size: + new_pos = self._orig_file_size + + if new_pos < 0: + new_pos = 0 + + read_offset = new_pos - curr_pos + buff_offset = read_offset + self._offset + + # Fast seek uncompressed unencrypted file + if self._compress_type == ZIP_STORED and self._decrypter is None and read_offset > 0: + # disable CRC checking after first seeking - it would be invalid + self._expected_crc = None + # seek actual file taking already buffered data into account + read_offset -= len(self._readbuffer) - self._offset + self._fileobj.seek(read_offset, os.SEEK_CUR) + self._left -= read_offset + read_offset = 0 + # flush read buffer + self._readbuffer = b'' + self._offset = 0 + elif buff_offset >= 0 and buff_offset < len(self._readbuffer): + # Just move the _offset index if the new position is in the _readbuffer + self._offset = buff_offset + read_offset = 0 + elif read_offset < 0: + # Position is before the current position. Reset the ZipExtFile + self._fileobj.seek(self._orig_compress_start) + self._running_crc = self._orig_start_crc + self._expected_crc = self._orig_crc + self._compress_left = self._orig_compress_size + self._left = self._orig_file_size + self._readbuffer = b'' + self._offset = 0 + self._decompressor = _get_decompressor(self._compress_type) + self._eof = False + read_offset = new_pos + if self._decrypter is not None: + self._init_decrypter() + + while read_offset > 0: + read_len = min(self.MAX_SEEK_READ, read_offset) + self.read(read_len) + read_offset -= read_len + + return self.tell() + + def tell(self): + if self.closed: + raise ValueError("tell on closed file.") + if not self._seekable: + raise io.UnsupportedOperation("underlying stream is not seekable") + filepos = self._orig_file_size - self._left - len(self._readbuffer) + self._offset + return filepos + + +class _ZipWriteFile(io.BufferedIOBase): + def __init__(self, zf, zinfo, zip64): + self._zinfo = zinfo + self._zip64 = zip64 + self._zipfile = zf + self._compressor = _get_compressor(zinfo.compress_type, + zinfo._compresslevel) + self._file_size = 0 + self._compress_size = 0 + self._crc = 0 + + @property + def _fileobj(self): + return self._zipfile.fp + + def writable(self): + return True + + def write(self, data): + if self.closed: + raise ValueError('I/O operation on closed file.') + + # Accept any data that supports the buffer protocol + if isinstance(data, (bytes, bytearray)): + nbytes = len(data) + else: + data = memoryview(data) + nbytes = data.nbytes + self._file_size += nbytes + + self._crc = crc32(data, self._crc) + if self._compressor: + data = self._compressor.compress(data) + self._compress_size += len(data) + self._fileobj.write(data) + return nbytes + + def close(self): + if self.closed: + return + try: + super().close() + # Flush any data from the compressor, and update header info + if self._compressor: + buf = self._compressor.flush() + self._compress_size += len(buf) + self._fileobj.write(buf) + self._zinfo.compress_size = self._compress_size + else: + self._zinfo.compress_size = self._file_size + self._zinfo.CRC = self._crc + self._zinfo.file_size = self._file_size + + # Write updated header info + if self._zinfo.flag_bits & _MASK_USE_DATA_DESCRIPTOR: + # Write CRC and file sizes after the file data + fmt = ' ZIP64_LIMIT: + raise RuntimeError( + 'File size unexpectedly exceeded ZIP64 limit') + if self._compress_size > ZIP64_LIMIT: + raise RuntimeError( + 'Compressed size unexpectedly exceeded ZIP64 limit') + # Seek backwards and write file header (which will now include + # correct CRC and file sizes) + + # Preserve current position in file + self._zipfile.start_dir = self._fileobj.tell() + self._fileobj.seek(self._zinfo.header_offset) + self._fileobj.write(self._zinfo.FileHeader(self._zip64)) + self._fileobj.seek(self._zipfile.start_dir) + + # Successfully written: Add file to our caches + self._zipfile.filelist.append(self._zinfo) + self._zipfile.NameToInfo[self._zinfo.filename] = self._zinfo + finally: + self._zipfile._writing = False + + + +class ZipFile: + """ Class with methods to open, read, write, close, list zip files. + + z = ZipFile(file, mode="r", compression=ZIP_STORED, allowZip64=True, + compresslevel=None) + + file: Either the path to the file, or a file-like object. + If it is a path, the file will be opened and closed by ZipFile. + mode: The mode can be either read 'r', write 'w', exclusive create 'x', + or append 'a'. + compression: ZIP_STORED (no compression), ZIP_DEFLATED (requires zlib), + ZIP_BZIP2 (requires bz2) or ZIP_LZMA (requires lzma). + allowZip64: if True ZipFile will create files with ZIP64 extensions when + needed, otherwise it will raise an exception when this would + be necessary. + compresslevel: None (default for the given compression type) or an integer + specifying the level to pass to the compressor. + When using ZIP_STORED or ZIP_LZMA this keyword has no effect. + When using ZIP_DEFLATED integers 0 through 9 are accepted. + When using ZIP_BZIP2 integers 1 through 9 are accepted. + + """ + + fp = None # Set here since __del__ checks it + _windows_illegal_name_trans_table = None + + def __init__(self, file, mode="r", compression=ZIP_STORED, allowZip64=True, + compresslevel=None, *, strict_timestamps=True, metadata_encoding=None): + """Open the ZIP file with mode read 'r', write 'w', exclusive create 'x', + or append 'a'.""" + if mode not in ('r', 'w', 'x', 'a'): + raise ValueError("ZipFile requires mode 'r', 'w', 'x', or 'a'") + + _check_compression(compression) + + self._allowZip64 = allowZip64 + self._didModify = False + self.debug = 0 # Level of printing: 0 through 3 + self.NameToInfo = {} # Find file info given name + self.filelist = [] # List of ZipInfo instances for archive + self.compression = compression # Method of compression + self.compresslevel = compresslevel + self.mode = mode + self.pwd = None + self._comment = b'' + self._strict_timestamps = strict_timestamps + self.metadata_encoding = metadata_encoding + + # Check that we don't try to write with nonconforming codecs + if self.metadata_encoding and mode != 'r': + raise ValueError( + "metadata_encoding is only supported for reading files") + + # Check if we were passed a file-like object + if isinstance(file, os.PathLike): + file = os.fspath(file) + if isinstance(file, str): + # No, it's a filename + self._filePassed = 0 + self.filename = file + modeDict = {'r' : 'rb', 'w': 'w+b', 'x': 'x+b', 'a' : 'r+b', + 'r+b': 'w+b', 'w+b': 'wb', 'x+b': 'xb'} + filemode = modeDict[mode] + while True: + try: + self.fp = io.open(file, filemode) + except OSError: + if filemode in modeDict: + filemode = modeDict[filemode] + continue + raise + break + else: + self._filePassed = 1 + self.fp = file + self.filename = getattr(file, 'name', None) + self._fileRefCnt = 1 + self._lock = threading.RLock() + self._seekable = True + self._writing = False + + try: + if mode == 'r': + self._RealGetContents() + elif mode in ('w', 'x'): + # set the modified flag so central directory gets written + # even if no files are added to the archive + self._didModify = True + try: + self.start_dir = self.fp.tell() + except (AttributeError, OSError): + self.fp = _Tellable(self.fp) + self.start_dir = 0 + self._seekable = False + else: + # Some file-like objects can provide tell() but not seek() + try: + self.fp.seek(self.start_dir) + except (AttributeError, OSError): + self._seekable = False + elif mode == 'a': + try: + # See if file is a zip file + self._RealGetContents() + # seek to start of directory and overwrite + self.fp.seek(self.start_dir) + except BadZipFile: + # file is not a zip file, just append + self.fp.seek(0, 2) + + # set the modified flag so central directory gets written + # even if no files are added to the archive + self._didModify = True + self.start_dir = self.fp.tell() + else: + raise ValueError("Mode must be 'r', 'w', 'x', or 'a'") + except: + fp = self.fp + self.fp = None + self._fpclose(fp) + raise + + def __enter__(self): + return self + + def __exit__(self, type, value, traceback): + self.close() + + def __repr__(self): + result = ['<%s.%s' % (self.__class__.__module__, + self.__class__.__qualname__)] + if self.fp is not None: + if self._filePassed: + result.append(' file=%r' % self.fp) + elif self.filename is not None: + result.append(' filename=%r' % self.filename) + result.append(' mode=%r' % self.mode) + else: + result.append(' [closed]') + result.append('>') + return ''.join(result) + + def _RealGetContents(self): + """Read in the table of contents for the ZIP file.""" + fp = self.fp + try: + endrec = _EndRecData(fp) + except OSError: + raise BadZipFile("File is not a zip file") + if not endrec: + raise BadZipFile("File is not a zip file") + if self.debug > 1: + print(endrec) + size_cd = endrec[_ECD_SIZE] # bytes in central directory + offset_cd = endrec[_ECD_OFFSET] # offset of central directory + self._comment = endrec[_ECD_COMMENT] # archive comment + + # "concat" is zero, unless zip was concatenated to another file + concat = endrec[_ECD_LOCATION] - size_cd - offset_cd + if endrec[_ECD_SIGNATURE] == stringEndArchive64: + # If Zip64 extension structures are present, account for them + concat -= (sizeEndCentDir64 + sizeEndCentDir64Locator) + + if self.debug > 2: + inferred = concat + offset_cd + print("given, inferred, offset", offset_cd, inferred, concat) + # self.start_dir: Position of start of central directory + self.start_dir = offset_cd + concat + if self.start_dir < 0: + raise BadZipFile("Bad offset for central directory") + fp.seek(self.start_dir, 0) + data = fp.read(size_cd) + fp = io.BytesIO(data) + total = 0 + while total < size_cd: + centdir = fp.read(sizeCentralDir) + if len(centdir) != sizeCentralDir: + raise BadZipFile("Truncated central directory") + centdir = struct.unpack(structCentralDir, centdir) + if centdir[_CD_SIGNATURE] != stringCentralDir: + raise BadZipFile("Bad magic number for central directory") + if self.debug > 2: + print(centdir) + filename = fp.read(centdir[_CD_FILENAME_LENGTH]) + flags = centdir[_CD_FLAG_BITS] + if flags & _MASK_UTF_FILENAME: + # UTF-8 file names extension + filename = filename.decode('utf-8') + else: + # Historical ZIP filename encoding + filename = filename.decode(self.metadata_encoding or 'cp437') + # Create ZipInfo instance to store file information + x = ZipInfo(filename) + x.extra = fp.read(centdir[_CD_EXTRA_FIELD_LENGTH]) + x.comment = fp.read(centdir[_CD_COMMENT_LENGTH]) + x.header_offset = centdir[_CD_LOCAL_HEADER_OFFSET] + (x.create_version, x.create_system, x.extract_version, x.reserved, + x.flag_bits, x.compress_type, t, d, + x.CRC, x.compress_size, x.file_size) = centdir[1:12] + if x.extract_version > MAX_EXTRACT_VERSION: + raise NotImplementedError("zip file version %.1f" % + (x.extract_version / 10)) + x.volume, x.internal_attr, x.external_attr = centdir[15:18] + # Convert date/time code to (year, month, day, hour, min, sec) + x._raw_time = t + x.date_time = ( (d>>9)+1980, (d>>5)&0xF, d&0x1F, + t>>11, (t>>5)&0x3F, (t&0x1F) * 2 ) + + x._decodeExtra() + x.header_offset = x.header_offset + concat + self.filelist.append(x) + self.NameToInfo[x.filename] = x + + # update total bytes read from central directory + total = (total + sizeCentralDir + centdir[_CD_FILENAME_LENGTH] + + centdir[_CD_EXTRA_FIELD_LENGTH] + + centdir[_CD_COMMENT_LENGTH]) + + if self.debug > 2: + print("total", total) + + + def namelist(self): + """Return a list of file names in the archive.""" + return [data.filename for data in self.filelist] + + def infolist(self): + """Return a list of class ZipInfo instances for files in the + archive.""" + return self.filelist + + def printdir(self, file=None): + """Print a table of contents for the zip file.""" + print("%-46s %19s %12s" % ("File Name", "Modified ", "Size"), + file=file) + for zinfo in self.filelist: + date = "%d-%02d-%02d %02d:%02d:%02d" % zinfo.date_time[:6] + print("%-46s %s %12d" % (zinfo.filename, date, zinfo.file_size), + file=file) + + def testzip(self): + """Read all the files and check the CRC. + + Return None if all files could be read successfully, or the name + of the offending file otherwise.""" + chunk_size = 2 ** 20 + for zinfo in self.filelist: + try: + # Read by chunks, to avoid an OverflowError or a + # MemoryError with very large embedded files. + with self.open(zinfo.filename, "r") as f: + while f.read(chunk_size): # Check CRC-32 + pass + except BadZipFile: + return zinfo.filename + + def getinfo(self, name): + """Return the instance of ZipInfo given 'name'.""" + info = self.NameToInfo.get(name) + if info is None: + raise KeyError( + 'There is no item named %r in the archive' % name) + + return info + + def setpassword(self, pwd): + """Set default password for encrypted files.""" + if pwd and not isinstance(pwd, bytes): + raise TypeError("pwd: expected bytes, got %s" % type(pwd).__name__) + if pwd: + self.pwd = pwd + else: + self.pwd = None + + @property + def comment(self): + """The comment text associated with the ZIP file.""" + return self._comment + + @comment.setter + def comment(self, comment): + if not isinstance(comment, bytes): + raise TypeError("comment: expected bytes, got %s" % type(comment).__name__) + # check for valid comment length + if len(comment) > ZIP_MAX_COMMENT: + import warnings + warnings.warn('Archive comment is too long; truncating to %d bytes' + % ZIP_MAX_COMMENT, stacklevel=2) + comment = comment[:ZIP_MAX_COMMENT] + self._comment = comment + self._didModify = True + + def read(self, name, pwd=None): + """Return file bytes for name.""" + with self.open(name, "r", pwd) as fp: + return fp.read() + + def open(self, name, mode="r", pwd=None, *, force_zip64=False): + """Return file-like object for 'name'. + + name is a string for the file name within the ZIP file, or a ZipInfo + object. + + mode should be 'r' to read a file already in the ZIP file, or 'w' to + write to a file newly added to the archive. + + pwd is the password to decrypt files (only used for reading). + + When writing, if the file size is not known in advance but may exceed + 2 GiB, pass force_zip64 to use the ZIP64 format, which can handle large + files. If the size is known in advance, it is best to pass a ZipInfo + instance for name, with zinfo.file_size set. + """ + if mode not in {"r", "w"}: + raise ValueError('open() requires mode "r" or "w"') + if pwd and (mode == "w"): + raise ValueError("pwd is only supported for reading files") + if not self.fp: + raise ValueError( + "Attempt to use ZIP archive that was already closed") + + # Make sure we have an info object + if isinstance(name, ZipInfo): + # 'name' is already an info object + zinfo = name + elif mode == 'w': + zinfo = ZipInfo(name) + zinfo.compress_type = self.compression + zinfo._compresslevel = self.compresslevel + else: + # Get info object for name + zinfo = self.getinfo(name) + + if mode == 'w': + return self._open_to_write(zinfo, force_zip64=force_zip64) + + 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.") + + # Open for reading: + self._fileRefCnt += 1 + zef_file = _SharedFile(self.fp, zinfo.header_offset, + self._fpclose, self._lock, lambda: self._writing) + try: + # Skip the file header: + fheader = zef_file.read(sizeFileHeader) + if len(fheader) != sizeFileHeader: + raise BadZipFile("Truncated file header") + fheader = struct.unpack(structFileHeader, fheader) + if fheader[_FH_SIGNATURE] != stringFileHeader: + raise BadZipFile("Bad magic number for file header") + + fname = zef_file.read(fheader[_FH_FILENAME_LENGTH]) + if fheader[_FH_EXTRA_FIELD_LENGTH]: + zef_file.seek(fheader[_FH_EXTRA_FIELD_LENGTH], whence=1) + + if zinfo.flag_bits & _MASK_COMPRESSED_PATCH: + # Zip 2.7: compressed patched data + raise NotImplementedError("compressed patched data (flag bit 5)") + + if zinfo.flag_bits & _MASK_STRONG_ENCRYPTION: + # strong encryption + raise NotImplementedError("strong encryption (flag bit 6)") + + if fheader[_FH_GENERAL_PURPOSE_FLAG_BITS] & _MASK_UTF_FILENAME: + # UTF-8 filename + fname_str = fname.decode("utf-8") + else: + fname_str = fname.decode(self.metadata_encoding or "cp437") + + if fname_str != zinfo.orig_filename: + raise BadZipFile( + 'File name in directory %r and header %r differ.' + % (zinfo.orig_filename, fname)) + + # check for encrypted flag & handle password + is_encrypted = zinfo.flag_bits & _MASK_ENCRYPTED + if is_encrypted: + if not pwd: + pwd = self.pwd + if pwd and not isinstance(pwd, bytes): + raise TypeError("pwd: expected bytes, got %s" % type(pwd).__name__) + if not pwd: + raise RuntimeError("File %r is encrypted, password " + "required for extraction" % name) + else: + pwd = None + + return ZipExtFile(zef_file, mode, zinfo, pwd, True) + except: + zef_file.close() + raise + + def _open_to_write(self, zinfo, force_zip64=False): + if force_zip64 and not self._allowZip64: + raise ValueError( + "force_zip64 is True, but allowZip64 was False when opening " + "the ZIP file." + ) + if self._writing: + raise ValueError("Can't write to the ZIP file while there is " + "another write handle open on it. " + "Close the first handle before opening another.") + + # Size and CRC are overwritten with correct data after processing the file + zinfo.compress_size = 0 + zinfo.CRC = 0 + + zinfo.flag_bits = 0x00 + if zinfo.compress_type == ZIP_LZMA: + # Compressed data includes an end-of-stream (EOS) marker + zinfo.flag_bits |= _MASK_COMPRESS_OPTION_1 + if not self._seekable: + zinfo.flag_bits |= _MASK_USE_DATA_DESCRIPTOR + + if not zinfo.external_attr: + zinfo.external_attr = 0o600 << 16 # permissions: ?rw------- + + # Compressed size can be larger than uncompressed size + zip64 = self._allowZip64 and \ + (force_zip64 or zinfo.file_size * 1.05 > ZIP64_LIMIT) + + if self._seekable: + self.fp.seek(self.start_dir) + zinfo.header_offset = self.fp.tell() + + self._writecheck(zinfo) + self._didModify = True + + self.fp.write(zinfo.FileHeader(zip64)) + + self._writing = True + return _ZipWriteFile(self, zinfo, zip64) + + def extract(self, member, path=None, pwd=None): + """Extract a member from the archive to the current working directory, + using its full name. Its file information is extracted as accurately + as possible. `member' may be a filename or a ZipInfo object. You can + specify a different directory using `path'. + """ + if path is None: + path = os.getcwd() + else: + path = os.fspath(path) + + return self._extract_member(member, path, pwd) + + def extractall(self, path=None, members=None, pwd=None): + """Extract all members from the archive to the current working + directory. `path' specifies a different directory to extract to. + `members' is optional and must be a subset of the list returned + by namelist(). + """ + if members is None: + members = self.namelist() + + if path is None: + path = os.getcwd() + else: + path = os.fspath(path) + + for zipinfo in members: + self._extract_member(zipinfo, path, pwd) + + @classmethod + def _sanitize_windows_name(cls, arcname, pathsep): + """Replace bad characters and remove trailing dots from parts.""" + table = cls._windows_illegal_name_trans_table + if not table: + illegal = ':<>|"?*' + table = str.maketrans(illegal, '_' * len(illegal)) + cls._windows_illegal_name_trans_table = table + arcname = arcname.translate(table) + # remove trailing dots and spaces + arcname = (x.rstrip(' .') for x in arcname.split(pathsep)) + # rejoin, removing empty parts. + arcname = pathsep.join(x for x in arcname if x) + return arcname + + def _extract_member(self, member, targetpath, pwd): + """Extract the ZipInfo object 'member' to a physical + file on the path targetpath. + """ + if not isinstance(member, ZipInfo): + member = self.getinfo(member) + + # build the destination pathname, replacing + # forward slashes to platform specific separators. + arcname = member.filename.replace('/', os.path.sep) + + if os.path.altsep: + arcname = arcname.replace(os.path.altsep, os.path.sep) + # interpret absolute pathname as relative, remove drive letter or + # UNC path, redundant separators, "." and ".." components. + arcname = os.path.splitdrive(arcname)[1] + invalid_path_parts = ('', os.path.curdir, os.path.pardir) + arcname = os.path.sep.join(x for x in arcname.split(os.path.sep) + if x not in invalid_path_parts) + if os.path.sep == '\\': + # filter illegal characters on Windows + arcname = self._sanitize_windows_name(arcname, os.path.sep) + + if not arcname: + raise ValueError("Empty filename.") + + targetpath = os.path.join(targetpath, arcname) + targetpath = os.path.normpath(targetpath) + + # Create all upper directories if necessary. + upperdirs = os.path.dirname(targetpath) + if upperdirs and not os.path.exists(upperdirs): + os.makedirs(upperdirs) + + if member.is_dir(): + if not os.path.isdir(targetpath): + os.mkdir(targetpath) + return targetpath + + with self.open(member, pwd=pwd) as source, \ + open(targetpath, "wb") as target: + shutil.copyfileobj(source, target) + + return targetpath + + def _writecheck(self, zinfo): + """Check for errors before writing a file to the archive.""" + if zinfo.filename in self.NameToInfo: + import warnings + warnings.warn('Duplicate name: %r' % zinfo.filename, stacklevel=3) + if self.mode not in ('w', 'x', 'a'): + raise ValueError("write() requires mode 'w', 'x', or 'a'") + if not self.fp: + raise ValueError( + "Attempt to write ZIP archive that was already closed") + _check_compression(zinfo.compress_type) + if not self._allowZip64: + requires_zip64 = None + if len(self.filelist) >= ZIP_FILECOUNT_LIMIT: + requires_zip64 = "Files count" + elif zinfo.file_size > ZIP64_LIMIT: + requires_zip64 = "Filesize" + elif zinfo.header_offset > ZIP64_LIMIT: + requires_zip64 = "Zipfile size" + if requires_zip64: + raise LargeZipFile(requires_zip64 + + " would require ZIP64 extensions") + + def write(self, filename, arcname=None, + compress_type=None, compresslevel=None): + """Put the bytes from filename into the archive under the name + arcname.""" + if not self.fp: + raise ValueError( + "Attempt to write to ZIP archive that was already closed") + if self._writing: + raise ValueError( + "Can't write to ZIP archive while an open writing handle exists" + ) + + zinfo = ZipInfo.from_file(filename, arcname, + strict_timestamps=self._strict_timestamps) + + if zinfo.is_dir(): + zinfo.compress_size = 0 + zinfo.CRC = 0 + self.mkdir(zinfo) + else: + if compress_type is not None: + zinfo.compress_type = compress_type + else: + zinfo.compress_type = self.compression + + if compresslevel is not None: + zinfo._compresslevel = compresslevel + else: + zinfo._compresslevel = self.compresslevel + + with open(filename, "rb") as src, self.open(zinfo, 'w') as dest: + shutil.copyfileobj(src, dest, 1024*8) + + def writestr(self, zinfo_or_arcname, data, + compress_type=None, compresslevel=None): + """Write a file into the archive. The contents is 'data', which + may be either a 'str' or a 'bytes' instance; if it is a 'str', + it is encoded as UTF-8 first. + 'zinfo_or_arcname' is either a ZipInfo instance or + the name of the file in the archive.""" + if isinstance(data, str): + data = data.encode("utf-8") + if not isinstance(zinfo_or_arcname, ZipInfo): + zinfo = ZipInfo(filename=zinfo_or_arcname, + date_time=time.localtime(time.time())[:6]) + zinfo.compress_type = self.compression + zinfo._compresslevel = self.compresslevel + if zinfo.filename.endswith('/'): + zinfo.external_attr = 0o40775 << 16 # drwxrwxr-x + zinfo.external_attr |= 0x10 # MS-DOS directory flag + else: + zinfo.external_attr = 0o600 << 16 # ?rw------- + else: + zinfo = zinfo_or_arcname + + if not self.fp: + raise ValueError( + "Attempt to write to ZIP archive that was already closed") + if self._writing: + raise ValueError( + "Can't write to ZIP archive while an open writing handle exists." + ) + + if compress_type is not None: + zinfo.compress_type = compress_type + + if compresslevel is not None: + zinfo._compresslevel = compresslevel + + zinfo.file_size = len(data) # Uncompressed size + with self._lock: + with self.open(zinfo, mode='w') as dest: + dest.write(data) + + def mkdir(self, zinfo_or_directory_name, mode=511): + """Creates a directory inside the zip archive.""" + if isinstance(zinfo_or_directory_name, ZipInfo): + zinfo = zinfo_or_directory_name + if not zinfo.is_dir(): + raise ValueError("The given ZipInfo does not describe a directory") + elif isinstance(zinfo_or_directory_name, str): + directory_name = zinfo_or_directory_name + if not directory_name.endswith("/"): + directory_name += "/" + zinfo = ZipInfo(directory_name) + zinfo.compress_size = 0 + zinfo.CRC = 0 + zinfo.external_attr = ((0o40000 | mode) & 0xFFFF) << 16 + zinfo.file_size = 0 + zinfo.external_attr |= 0x10 + else: + raise TypeError("Expected type str or ZipInfo") + + with self._lock: + if self._seekable: + self.fp.seek(self.start_dir) + zinfo.header_offset = self.fp.tell() # Start of header bytes + if zinfo.compress_type == ZIP_LZMA: + # Compressed data includes an end-of-stream (EOS) marker + zinfo.flag_bits |= _MASK_COMPRESS_OPTION_1 + + self._writecheck(zinfo) + self._didModify = True + + self.filelist.append(zinfo) + self.NameToInfo[zinfo.filename] = zinfo + self.fp.write(zinfo.FileHeader(False)) + self.start_dir = self.fp.tell() + + def __del__(self): + """Call the "close()" method in case the user forgot.""" + self.close() + + def close(self): + """Close the file, and for mode 'w', 'x' and 'a' write the ending + records.""" + if self.fp is None: + return + + if self._writing: + raise ValueError("Can't close the ZIP file while there is " + "an open writing handle on it. " + "Close the writing handle before closing the zip.") + + try: + if self.mode in ('w', 'x', 'a') and self._didModify: # write ending records + with self._lock: + if self._seekable: + self.fp.seek(self.start_dir) + self._write_end_record() + finally: + fp = self.fp + self.fp = None + self._fpclose(fp) + + def _write_end_record(self): + for zinfo in self.filelist: # write central directory + dt = zinfo.date_time + dosdate = (dt[0] - 1980) << 9 | dt[1] << 5 | dt[2] + dostime = dt[3] << 11 | dt[4] << 5 | (dt[5] // 2) + extra = [] + if zinfo.file_size > ZIP64_LIMIT \ + or zinfo.compress_size > ZIP64_LIMIT: + extra.append(zinfo.file_size) + extra.append(zinfo.compress_size) + file_size = 0xffffffff + compress_size = 0xffffffff + else: + file_size = zinfo.file_size + compress_size = zinfo.compress_size + + if zinfo.header_offset > ZIP64_LIMIT: + extra.append(zinfo.header_offset) + header_offset = 0xffffffff + else: + header_offset = zinfo.header_offset + + extra_data = zinfo.extra + min_version = 0 + if extra: + # Append a ZIP64 field to the extra's + extra_data = _strip_extra(extra_data, (1,)) + extra_data = struct.pack( + ' ZIP_FILECOUNT_LIMIT: + requires_zip64 = "Files count" + elif centDirOffset > ZIP64_LIMIT: + requires_zip64 = "Central directory offset" + elif centDirSize > ZIP64_LIMIT: + requires_zip64 = "Central directory size" + if requires_zip64: + # Need to write the ZIP64 end-of-archive records + if not self._allowZip64: + raise LargeZipFile(requires_zip64 + + " would require ZIP64 extensions") + zip64endrec = struct.pack( + structEndArchive64, stringEndArchive64, + 44, 45, 45, 0, 0, centDirCount, centDirCount, + centDirSize, centDirOffset) + self.fp.write(zip64endrec) + + zip64locrec = struct.pack( + structEndArchive64Locator, + stringEndArchive64Locator, 0, pos2, 1) + self.fp.write(zip64locrec) + centDirCount = min(centDirCount, 0xFFFF) + centDirSize = min(centDirSize, 0xFFFFFFFF) + centDirOffset = min(centDirOffset, 0xFFFFFFFF) + + endrec = struct.pack(structEndArchive, stringEndArchive, + 0, 0, centDirCount, centDirCount, + centDirSize, centDirOffset, len(self._comment)) + self.fp.write(endrec) + self.fp.write(self._comment) + if self.mode == "a": + self.fp.truncate() + self.fp.flush() + + def _fpclose(self, fp): + assert self._fileRefCnt > 0 + self._fileRefCnt -= 1 + if not self._fileRefCnt and not self._filePassed: + fp.close() + + +class PyZipFile(ZipFile): + """Class to create ZIP archives with Python library files and packages.""" + + def __init__(self, file, mode="r", compression=ZIP_STORED, + allowZip64=True, optimize=-1): + ZipFile.__init__(self, file, mode=mode, compression=compression, + allowZip64=allowZip64) + self._optimize = optimize + + def writepy(self, pathname, basename="", filterfunc=None): + """Add all files from "pathname" to the ZIP archive. + + If pathname is a package directory, search the directory and + all package subdirectories recursively for all *.py and enter + the modules into the archive. If pathname is a plain + directory, listdir *.py and enter all modules. Else, pathname + must be a Python *.py file and the module will be put into the + archive. Added modules are always module.pyc. + This method will compile the module.py into module.pyc if + necessary. + If filterfunc(pathname) is given, it is called with every argument. + When it is False, the file or directory is skipped. + """ + pathname = os.fspath(pathname) + if filterfunc and not filterfunc(pathname): + if self.debug: + label = 'path' if os.path.isdir(pathname) else 'file' + print('%s %r skipped by filterfunc' % (label, pathname)) + return + dir, name = os.path.split(pathname) + if os.path.isdir(pathname): + initname = os.path.join(pathname, "__init__.py") + if os.path.isfile(initname): + # This is a package directory, add it + if basename: + basename = "%s/%s" % (basename, name) + else: + basename = name + if self.debug: + print("Adding package in", pathname, "as", basename) + fname, arcname = self._get_codename(initname[0:-3], basename) + if self.debug: + print("Adding", arcname) + self.write(fname, arcname) + dirlist = sorted(os.listdir(pathname)) + dirlist.remove("__init__.py") + # Add all *.py files and package subdirectories + for filename in dirlist: + path = os.path.join(pathname, filename) + root, ext = os.path.splitext(filename) + if os.path.isdir(path): + if os.path.isfile(os.path.join(path, "__init__.py")): + # This is a package directory, add it + self.writepy(path, basename, + filterfunc=filterfunc) # Recursive call + elif ext == ".py": + if filterfunc and not filterfunc(path): + if self.debug: + print('file %r skipped by filterfunc' % path) + continue + fname, arcname = self._get_codename(path[0:-3], + basename) + if self.debug: + print("Adding", arcname) + self.write(fname, arcname) + else: + # This is NOT a package directory, add its files at top level + if self.debug: + print("Adding files from directory", pathname) + for filename in sorted(os.listdir(pathname)): + path = os.path.join(pathname, filename) + root, ext = os.path.splitext(filename) + if ext == ".py": + if filterfunc and not filterfunc(path): + if self.debug: + print('file %r skipped by filterfunc' % path) + continue + fname, arcname = self._get_codename(path[0:-3], + basename) + if self.debug: + print("Adding", arcname) + self.write(fname, arcname) + else: + if pathname[-3:] != ".py": + raise RuntimeError( + 'Files added with writepy() must end with ".py"') + fname, arcname = self._get_codename(pathname[0:-3], basename) + if self.debug: + print("Adding file", arcname) + self.write(fname, arcname) + + def _get_codename(self, pathname, basename): + """Return (filename, archivename) for the path. + + Given a module name path, return the correct file path and + archive name, compiling if necessary. For example, given + /python/lib/string, return (/python/lib/string.pyc, string). + """ + def _compile(file, optimize=-1): + import py_compile + if self.debug: + print("Compiling", file) + try: + py_compile.compile(file, doraise=True, optimize=optimize) + except py_compile.PyCompileError as err: + print(err.msg) + return False + return True + + file_py = pathname + ".py" + file_pyc = pathname + ".pyc" + pycache_opt0 = importlib.util.cache_from_source(file_py, optimization='') + pycache_opt1 = importlib.util.cache_from_source(file_py, optimization=1) + pycache_opt2 = importlib.util.cache_from_source(file_py, optimization=2) + if self._optimize == -1: + # legacy mode: use whatever file is present + if (os.path.isfile(file_pyc) and + os.stat(file_pyc).st_mtime >= os.stat(file_py).st_mtime): + # Use .pyc file. + arcname = fname = file_pyc + elif (os.path.isfile(pycache_opt0) and + os.stat(pycache_opt0).st_mtime >= os.stat(file_py).st_mtime): + # Use the __pycache__/*.pyc file, but write it to the legacy pyc + # file name in the archive. + fname = pycache_opt0 + arcname = file_pyc + elif (os.path.isfile(pycache_opt1) and + os.stat(pycache_opt1).st_mtime >= os.stat(file_py).st_mtime): + # Use the __pycache__/*.pyc file, but write it to the legacy pyc + # file name in the archive. + fname = pycache_opt1 + arcname = file_pyc + elif (os.path.isfile(pycache_opt2) and + os.stat(pycache_opt2).st_mtime >= os.stat(file_py).st_mtime): + # Use the __pycache__/*.pyc file, but write it to the legacy pyc + # file name in the archive. + fname = pycache_opt2 + arcname = file_pyc + else: + # Compile py into PEP 3147 pyc file. + if _compile(file_py): + if sys.flags.optimize == 0: + fname = pycache_opt0 + elif sys.flags.optimize == 1: + fname = pycache_opt1 + else: + fname = pycache_opt2 + arcname = file_pyc + else: + fname = arcname = file_py + else: + # new mode: use given optimization level + if self._optimize == 0: + fname = pycache_opt0 + arcname = file_pyc + else: + arcname = file_pyc + if self._optimize == 1: + fname = pycache_opt1 + elif self._optimize == 2: + fname = pycache_opt2 + else: + msg = "invalid value for 'optimize': {!r}".format(self._optimize) + raise ValueError(msg) + if not (os.path.isfile(fname) and + os.stat(fname).st_mtime >= os.stat(file_py).st_mtime): + if not _compile(file_py, optimize=self._optimize): + fname = arcname = file_py + archivename = os.path.split(arcname)[1] + if basename: + archivename = "%s/%s" % (basename, archivename) + return (fname, archivename) + + +from ._path import ( # noqa: E402 + Path, + + # used privately for tests + CompleteDirs, # noqa: F401 +) + +# used privately for tests +from .__main__ import main # noqa: F401, E402 diff --git a/Lib/zipfile/__main__.py b/Lib/zipfile/__main__.py new file mode 100644 index 0000000..a9e5fb1 --- /dev/null +++ b/Lib/zipfile/__main__.py @@ -0,0 +1,77 @@ +import sys +import os +from . import ZipFile, ZIP_DEFLATED + + +def main(args=None): + import argparse + + description = 'A simple command-line interface for zipfile module.' + parser = argparse.ArgumentParser(description=description) + group = parser.add_mutually_exclusive_group(required=True) + group.add_argument('-l', '--list', metavar='', + help='Show listing of a zipfile') + group.add_argument('-e', '--extract', nargs=2, + metavar=('', ''), + help='Extract zipfile into target dir') + group.add_argument('-c', '--create', nargs='+', + metavar=('', ''), + help='Create zipfile from sources') + group.add_argument('-t', '--test', metavar='', + help='Test if a zipfile is valid') + parser.add_argument('--metadata-encoding', metavar='', + help='Specify encoding of member names for -l, -e and -t') + args = parser.parse_args(args) + + encoding = args.metadata_encoding + + if args.test is not None: + src = args.test + with ZipFile(src, 'r', metadata_encoding=encoding) as zf: + badfile = zf.testzip() + if badfile: + print("The following enclosed file is corrupted: {!r}".format(badfile)) + print("Done testing") + + elif args.list is not None: + src = args.list + with ZipFile(src, 'r', metadata_encoding=encoding) as zf: + zf.printdir() + + elif args.extract is not None: + src, curdir = args.extract + with ZipFile(src, 'r', metadata_encoding=encoding) as zf: + zf.extractall(curdir) + + elif args.create is not None: + if encoding: + print("Non-conforming encodings not supported with -c.", + file=sys.stderr) + sys.exit(1) + + zip_name = args.create.pop(0) + files = args.create + + def addToZip(zf, path, zippath): + if os.path.isfile(path): + zf.write(path, zippath, ZIP_DEFLATED) + elif os.path.isdir(path): + if zippath: + zf.write(path, zippath) + for nm in sorted(os.listdir(path)): + addToZip(zf, + os.path.join(path, nm), os.path.join(zippath, nm)) + # else: ignore + + with ZipFile(zip_name, 'w') as zf: + for path in files: + zippath = os.path.basename(path) + if not zippath: + zippath = os.path.basename(os.path.dirname(path)) + if zippath in ('', os.curdir, os.pardir): + zippath = '' + addToZip(zf, path, zippath) + + +if __name__ == "__main__": + main() diff --git a/Lib/zipfile/_path.py b/Lib/zipfile/_path.py new file mode 100644 index 0000000..67ef07a --- /dev/null +++ b/Lib/zipfile/_path.py @@ -0,0 +1,315 @@ +import io +import posixpath +import zipfile +import itertools +import contextlib +import pathlib + + +__all__ = ['Path'] + + +def _parents(path): + """ + Given a path with elements separated by + posixpath.sep, generate all parents of that path. + + >>> list(_parents('b/d')) + ['b'] + >>> list(_parents('/b/d/')) + ['/b'] + >>> list(_parents('b/d/f/')) + ['b/d', 'b'] + >>> list(_parents('b')) + [] + >>> list(_parents('')) + [] + """ + return itertools.islice(_ancestry(path), 1, None) + + +def _ancestry(path): + """ + Given a path with elements separated by + posixpath.sep, generate all elements of that path + + >>> list(_ancestry('b/d')) + ['b/d', 'b'] + >>> list(_ancestry('/b/d/')) + ['/b/d', '/b'] + >>> list(_ancestry('b/d/f/')) + ['b/d/f', 'b/d', 'b'] + >>> list(_ancestry('b')) + ['b'] + >>> list(_ancestry('')) + [] + """ + path = path.rstrip(posixpath.sep) + while path and path != posixpath.sep: + yield path + path, tail = posixpath.split(path) + + +_dedupe = dict.fromkeys +"""Deduplicate an iterable in original order""" + + +def _difference(minuend, subtrahend): + """ + Return items in minuend not in subtrahend, retaining order + with O(1) lookup. + """ + return itertools.filterfalse(set(subtrahend).__contains__, minuend) + + +class CompleteDirs(zipfile.ZipFile): + """ + A ZipFile subclass that ensures that implied directories + are always included in the namelist. + """ + + @staticmethod + def _implied_dirs(names): + parents = itertools.chain.from_iterable(map(_parents, names)) + as_dirs = (p + posixpath.sep for p in parents) + return _dedupe(_difference(as_dirs, names)) + + def namelist(self): + names = super(CompleteDirs, self).namelist() + return names + list(self._implied_dirs(names)) + + def _name_set(self): + return set(self.namelist()) + + def resolve_dir(self, name): + """ + If the name represents a directory, return that name + as a directory (with the trailing slash). + """ + names = self._name_set() + dirname = name + '/' + dir_match = name not in names and dirname in names + return dirname if dir_match else name + + @classmethod + def make(cls, source): + """ + Given a source (filename or zipfile), return an + appropriate CompleteDirs subclass. + """ + if isinstance(source, CompleteDirs): + return source + + if not isinstance(source, zipfile.ZipFile): + return cls(source) + + # Only allow for FastLookup when supplied zipfile is read-only + if 'r' not in source.mode: + cls = CompleteDirs + + source.__class__ = cls + return source + + +class FastLookup(CompleteDirs): + """ + ZipFile subclass to ensure implicit + dirs exist and are resolved rapidly. + """ + + def namelist(self): + with contextlib.suppress(AttributeError): + return self.__names + self.__names = super(FastLookup, self).namelist() + return self.__names + + def _name_set(self): + with contextlib.suppress(AttributeError): + return self.__lookup + self.__lookup = super(FastLookup, self)._name_set() + return self.__lookup + + +class Path: + """ + A pathlib-compatible interface for zip files. + + Consider a zip file with this structure:: + + . + ├── a.txt + └── b + ├── c.txt + └── d + └── e.txt + + >>> data = io.BytesIO() + >>> zf = ZipFile(data, 'w') + >>> zf.writestr('a.txt', 'content of a') + >>> zf.writestr('b/c.txt', 'content of c') + >>> zf.writestr('b/d/e.txt', 'content of e') + >>> zf.filename = 'mem/abcde.zip' + + Path accepts the zipfile object itself or a filename + + >>> root = Path(zf) + + From there, several path operations are available. + + Directory iteration (including the zip file itself): + + >>> a, b = root.iterdir() + >>> a + Path('mem/abcde.zip', 'a.txt') + >>> b + Path('mem/abcde.zip', 'b/') + + name property: + + >>> b.name + 'b' + + join with divide operator: + + >>> c = b / 'c.txt' + >>> c + Path('mem/abcde.zip', 'b/c.txt') + >>> c.name + 'c.txt' + + Read text: + + >>> c.read_text() + 'content of c' + + existence: + + >>> c.exists() + True + >>> (b / 'missing.txt').exists() + False + + Coercion to string: + + >>> import os + >>> str(c).replace(os.sep, posixpath.sep) + 'mem/abcde.zip/b/c.txt' + + At the root, ``name``, ``filename``, and ``parent`` + resolve to the zipfile. Note these attributes are not + valid and will raise a ``ValueError`` if the zipfile + has no filename. + + >>> root.name + 'abcde.zip' + >>> str(root.filename).replace(os.sep, posixpath.sep) + 'mem/abcde.zip' + >>> str(root.parent) + 'mem' + """ + + __repr = "{self.__class__.__name__}({self.root.filename!r}, {self.at!r})" + + def __init__(self, root, at=""): + """ + Construct a Path from a ZipFile or filename. + + Note: When the source is an existing ZipFile object, + its type (__class__) will be mutated to a + specialized type. If the caller wishes to retain the + original type, the caller should either create a + separate ZipFile object or pass a filename. + """ + self.root = FastLookup.make(root) + self.at = at + + def open(self, mode='r', *args, pwd=None, **kwargs): + """ + Open this entry as text or binary following the semantics + of ``pathlib.Path.open()`` by passing arguments through + to io.TextIOWrapper(). + """ + if self.is_dir(): + raise IsADirectoryError(self) + zip_mode = mode[0] + if not self.exists() and zip_mode == 'r': + raise FileNotFoundError(self) + stream = self.root.open(self.at, zip_mode, pwd=pwd) + if 'b' in mode: + if args or kwargs: + raise ValueError("encoding args invalid for binary operation") + return stream + else: + kwargs["encoding"] = io.text_encoding(kwargs.get("encoding")) + return io.TextIOWrapper(stream, *args, **kwargs) + + @property + def name(self): + return pathlib.Path(self.at).name or self.filename.name + + @property + def suffix(self): + return pathlib.Path(self.at).suffix or self.filename.suffix + + @property + def suffixes(self): + return pathlib.Path(self.at).suffixes or self.filename.suffixes + + @property + def stem(self): + return pathlib.Path(self.at).stem or self.filename.stem + + @property + def filename(self): + return pathlib.Path(self.root.filename).joinpath(self.at) + + def read_text(self, *args, **kwargs): + kwargs["encoding"] = io.text_encoding(kwargs.get("encoding")) + with self.open('r', *args, **kwargs) as strm: + return strm.read() + + def read_bytes(self): + with self.open('rb') as strm: + return strm.read() + + def _is_child(self, path): + return posixpath.dirname(path.at.rstrip("/")) == self.at.rstrip("/") + + def _next(self, at): + return self.__class__(self.root, at) + + def is_dir(self): + return not self.at or self.at.endswith("/") + + def is_file(self): + return self.exists() and not self.is_dir() + + def exists(self): + return self.at in self.root._name_set() + + def iterdir(self): + if not self.is_dir(): + raise ValueError("Can't listdir a file") + subs = map(self._next, self.root.namelist()) + return filter(self._is_child, subs) + + def __str__(self): + return posixpath.join(self.root.filename, self.at) + + def __repr__(self): + return self.__repr.format(self=self) + + def joinpath(self, *other): + next = posixpath.join(self.at, *other) + return self._next(self.root.resolve_dir(next)) + + __truediv__ = joinpath + + @property + def parent(self): + if not self.at: + return self.filename.parent + parent_at = posixpath.dirname(self.at.rstrip('/')) + if parent_at: + parent_at += '/' + return self._next(parent_at) diff --git a/Misc/NEWS.d/next/Library/2022-10-08-15-41-00.gh-issue-98098.DugpWi.rst b/Misc/NEWS.d/next/Library/2022-10-08-15-41-00.gh-issue-98098.DugpWi.rst new file mode 100644 index 0000000..202275e --- /dev/null +++ b/Misc/NEWS.d/next/Library/2022-10-08-15-41-00.gh-issue-98098.DugpWi.rst @@ -0,0 +1,2 @@ +Created packages from zipfile and test_zipfile modules, separating +``zipfile.Path`` functionality. -- cgit v0.12