diff --git a/python-stdlib/zipfile/README.md b/python-stdlib/zipfile/README.md new file mode 100644 index 000000000..3df2248de --- /dev/null +++ b/python-stdlib/zipfile/README.md @@ -0,0 +1,25 @@ +# zipfile + +This library implements part of Python's `zipfile` module, namely allowing you +to open archives and read data out of them - as long as files are not compressed +with an unsupported compressor or encrypted. + +One difference between CPython's `zipfile` implementation of `ZipFile` and this +module's is that the former does enumerate the files contained in an archive +when opening it, and this information is cached and still available even after +the archive has been closed via `ZipFile.close`. To save memory, this +implementation neither performs any automatic enumeration on open, nor caches +the files list. Attempting to interact with a `ZipFile` instance once it's been +closed will raise a `ValueError` exception. + +## Installation + +Use `mip` via `mpremote`: + +```bash +> mpremote mip install zipfile +``` + +See [Package +management](https://docs.micropython.org/en/latest/reference/packages.html) for +more details on using `mip` and `mpremote`. diff --git a/python-stdlib/zipfile/deflate.zip b/python-stdlib/zipfile/deflate.zip new file mode 100644 index 000000000..90d3e0f97 Binary files /dev/null and b/python-stdlib/zipfile/deflate.zip differ diff --git a/python-stdlib/zipfile/manifest.py b/python-stdlib/zipfile/manifest.py new file mode 100644 index 000000000..31f6f2dac --- /dev/null +++ b/python-stdlib/zipfile/manifest.py @@ -0,0 +1,2 @@ +metadata(version="0.1.0") +package("zipfile") diff --git a/python-stdlib/zipfile/null.zip b/python-stdlib/zipfile/null.zip new file mode 100644 index 000000000..235e25a75 Binary files /dev/null and b/python-stdlib/zipfile/null.zip differ diff --git a/python-stdlib/zipfile/stored.zip b/python-stdlib/zipfile/stored.zip new file mode 100644 index 000000000..69869c717 Binary files /dev/null and b/python-stdlib/zipfile/stored.zip differ diff --git a/python-stdlib/zipfile/test_zipfile.py b/python-stdlib/zipfile/test_zipfile.py new file mode 100644 index 000000000..c9d08d090 --- /dev/null +++ b/python-stdlib/zipfile/test_zipfile.py @@ -0,0 +1,205 @@ +import io +import sys +import unittest +import zipfile + +# TODO: Find out which archiver can create ZSTD blobs +null_zip_contents = ( + ("null_bzip2.bin", 12, 4096, 46, (2026, 9, 4, 10, 1, 28), 0xC71C0011), + ("null_deflate.bin", zipfile.ZIP_DEFLATED, 4096, 20, (2026, 9, 4, 10, 1, 32), 0xC71C0011), + ("null_lzma.bin", 14, 4096, 41, (2026, 9, 4, 10, 1, 32), 0xC71C0011), + ("null_stored.bin", zipfile.ZIP_STORED, 4096, 4096, (2026, 9, 4, 10, 1, 20), 0xC71C0011), +) + +deflate_zip_contents = ( + ("directory/", True, 0, 0, 0, None), + ("directory/hello_again.txt", False, 6656, 66, 0x946F8D8C, b"Hello again, MicroPython!\n"), + ("empty", False, 0, 0, 0, None), + ("hello.txt", False, 5120, 57, 0xA89FD9B3, b"Hello, MicroPython!\n"), +) + +stored_zip_contents = ( + ("directory/", True, 0, 0, 0, None), + ("directory/hello_again.txt", False, 6656, 6656, 0x946F8D8C, b"Hello again, MicroPython!\n"), + ("empty", False, 0, 0, 0, None), + ("hello.txt", False, 5120, 5120, 0xA89FD9B3, b"Hello, MicroPython!\n"), +) + + +class TestZipFile(unittest.TestCase): + def check_header(self, file, expected): + name, method, uncompressed_size, compressed_size, timestamp, crc32 = expected + self.assertEqual(name, file.filename) + self.assertEqual(method, file.compress_type) + self.assertEqual(uncompressed_size, file.file_size) + self.assertEqual(compressed_size, file.compress_size) + self.assertEqual(timestamp, file.date_time) + self.assertEqual(crc32, file.CRC) + + def test_null_iter(self): + zf = zipfile.ZipFile("null.zip") + for index, info in enumerate(zf.infolist()): + self.check_header(info, null_zip_contents[index]) + zf.close() + + def test_null_missing(self): + with zipfile.ZipFile("null.zip") as zf: + with self.assertRaises(KeyError): + zf.getinfo("missing.txt") + + def test_null_stored_read(self): + zf = zipfile.ZipFile("null.zip") + expected = bytes(4096) + for index, name in enumerate(zf.namelist()): + self.assertEqual(name, null_zip_contents[index][0]) + if null_zip_contents[index][1] != zipfile.ZIP_STORED: + continue + self.assertEqual(zf.read(name), expected) + zf.close() + + def test_null_deflated_read(self): + if sys.implementation.name == "micropython": + try: + import deflate + except ImportError: + self.skipTest("no deflate module") + zf = zipfile.ZipFile("null.zip") + expected = bytes(4096) + for index, name in enumerate(zf.namelist()): + self.assertEqual(name, null_zip_contents[index][0]) + if null_zip_contents[index][1] != zipfile.ZIP_DEFLATED: + continue + self.assertEqual(zf.read(name), expected) + zf.close() + + def test_null_unsupported_read(self): + if sys.implementation.name != "micropython": + self.skipTest("cannot guarantee lack of methods support") + zf = zipfile.ZipFile("null.zip") + for index, name in enumerate(zf.namelist()): + self.assertEqual(name, null_zip_contents[index][0]) + if null_zip_contents[index][1] not in (zipfile.ZIP_STORED, zipfile.ZIP_DEFLATED): + with self.assertRaises(NotImplementedError): + zf.read(name) + zf.close() + + def test_null_stored_open(self): + zf = zipfile.ZipFile("null.zip") + expected = bytes(4096) + for index, name in enumerate(zf.namelist()): + self.assertEqual(name, null_zip_contents[index][0]) + if null_zip_contents[index][1] != zipfile.ZIP_STORED: + continue + with zf.open(name) as f: + self.assertEqual(expected, f.read()) + zf.close() + + def test_null_deflated_open(self): + if sys.implementation.name == "micropython": + try: + import deflate + except ImportError: + self.skipTest("no deflate module") + zf = zipfile.ZipFile("null.zip") + expected = bytes(4096) + for index, name in enumerate(zf.namelist()): + self.assertEqual(name, null_zip_contents[index][0]) + if null_zip_contents[index][1] != zipfile.ZIP_DEFLATED: + continue + with zf.open(name) as f: + self.assertEqual(expected, f.read()) + zf.close() + + def test_null_unsupported_open(self): + if sys.implementation.name != "micropython": + self.skipTest("cannot guarantee lack of methods support") + zf = zipfile.ZipFile("null.zip") + for index, name in enumerate(zf.namelist()): + self.assertEqual(name, null_zip_contents[index][0]) + if null_zip_contents[index][1] not in (zipfile.ZIP_STORED, zipfile.ZIP_DEFLATED): + with self.assertRaises(NotImplementedError): + with zf.open(name): + self.fail() + zf.close() + + def test_null_getinfo(self): + zf = zipfile.ZipFile("null.zip") + for zc in null_zip_contents: + info = zf.getinfo(zc[0]) + self.check_header(info, zc) + zf.close() + + def test_context(self): + with zipfile.ZipFile("null.zip") as zf: + for index, info in enumerate(zf.infolist()): + self.check_header(info, null_zip_contents[index]) + + def test_close(self): + zf = zipfile.ZipFile("null.zip") + zf.close() + zf.close() # check this doesn't raise + + # CPython doesn't raise as the archive directory is cached and not + # flushed once `close` is called. `namelist` and `infolist` do not + # raise on CPython. + if sys.implementation.name == "micropython": + with self.assertRaises(ValueError): + zf.namelist() + with self.assertRaises(ValueError): + zf.infolist() + with self.assertRaises(ValueError): + zf.open(null_zip_contents[0][0]) + with self.assertRaises(ValueError): + with zf.open(null_zip_contents[0][0]): + self.fail() + with self.assertRaises(ValueError): + zf.read(null_zip_contents[0][0]) + + def check_contents(self, zf, found, expected, method): + self.assertEqual(found.is_dir(), expected[1]) + self.assertEqual(found.file_size, expected[2]) + self.assertEqual(found.compress_size, expected[3]) + self.assertEqual(found.CRC, expected[4]) + self.assertEqual(found.compress_type, zipfile.ZIP_STORED if expected[2] == 0 else method) + if payload := expected[5]: + data = payload * 256 + with zf.open(expected[0]) as f: + self.assertEqual(f.read(), data) + self.assertEqual(zf.read(expected[0]), data) + + def test_read_deflate(self): + if sys.implementation.name == "micropython": + try: + import deflate + except ImportError: + self.skipTest("no deflate module") + with zipfile.ZipFile("deflate.zip") as zf: + for file in deflate_zip_contents: + info = zf.getinfo(file[0]) + self.check_contents(zf, info, file, zipfile.ZIP_DEFLATED) + + def test_read_stored(self): + with zipfile.ZipFile("stored.zip") as zf: + for file in stored_zip_contents: + info = zf.getinfo(file[0]) + self.check_contents(zf, info, file, zipfile.ZIP_STORED) + + def test_is_zipfile(self): + for n in ("stored.zip", "null.zip", "deflate.zip"): + self.assertEqual(zipfile.is_zipfile(n), True) + with open(n, "rb") as f: + old_offset = f.tell() + self.assertEqual(zipfile.is_zipfile(f), True) + self.assertEqual(f.tell(), old_offset) + + self.assertEqual(zipfile.is_zipfile("test_zipfile.py"), False) + with open("test_zipfile.py", "rb") as f: + old_offset = f.tell() + self.assertEqual(zipfile.is_zipfile(f), False) + self.assertEqual(f.tell(), old_offset) + + self.assertEqual(zipfile.is_zipfile("missing.zip"), False) + self.assertEqual(zipfile.is_zipfile("zipfile"), False) + + with io.BytesIO() as empty: + self.assertEqual(zipfile.is_zipfile(empty), False) diff --git a/python-stdlib/zipfile/zipfile/__init__.py b/python-stdlib/zipfile/zipfile/__init__.py new file mode 100644 index 000000000..ddfc31c4d --- /dev/null +++ b/python-stdlib/zipfile/zipfile/__init__.py @@ -0,0 +1,299 @@ +import io +import struct + +try: + import deflate + + _has_deflate = True +except ImportError: + _has_deflate = False + + +class BadZipFile(Exception): + pass + + +_MAGIC = const(b"\x50\x4b\x03\x04") + + +ZIP_STORED = const(0) +ZIP_DEFLATED = const(8) + + +class _IOProxy(io.IOBase): + def __init__(self, inner, length): + self._i = inner + self._o = 0 # virtual offset + self._l = length # max virtual offset + self._b = inner.tell() # min physical offset + self._e = self._b + length # max physical offset + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_value, traceback): + self.close() + + def close(self): + if self._i is not None: + self._i = None + + def read(self, size=-1): + if self._i is None: + raise ValueError + physical_offset = self._o + self._b + if size < 0: + size = self._e - physical_offset + else: + size = min(self._e - physical_offset, size) + if size <= 0: + return b"" + old_offset = self._i.tell() + try: + self._i.seek(self._o + self._b) + data = self._i.read(size) + finally: + self._i.seek(old_offset) + self._o += len(data) + return data + + def readinto(self, buffer): + count = 0 + for index in range(len(buffer)): + byte = self.read(1) + if byte == b"": + break + buffer[index] = byte[0] + count += 1 + return count + + def tell(self): + if self._i is None: + raise ValueError + return self._o + + def seek(self, offset, whence=0): + if self._i is None: + raise ValueError + if whence == 0: + new_offset = offset + elif whence == 1: + new_offset = self._o + offset + elif whence == 2: + new_offset = self._l + offset + self._o = max(min(new_offset, self._l), 0) + return self._o + + +class ZipInfo: + def __init__(self, filename="NoName", date_time=(1980, 1, 1, 0, 0, 0)): + self.filename = filename + self.date_time = date_time + self.compress_type = None + self.comment = None + self.extra = None + self.create_system = None + self.create_version = None + self.extract_version = None + self.reserved = 0 + self.flag_bits = None + self.volume = None + self.internal_attr = None + self.external_attr = None + self.header_offset = 0 + self.CRC = 0 + self.compress_size = 0 + self.file_size = 0 + + @staticmethod + def from_file(*args, **kwargs): + raise NotImplementedError + + def is_dir(self): + return self.compress_size == 0 and self.file_size == 0 and self.filename.endswith("/") + + def __repr__(self): + if self.compress_type == ZIP_STORED: + compress_type = "stored" + elif self.compress_type == ZIP_DEFLATED: + compress_type = "deflated" + else: + compress_type = str(self.compress_type) + return "".format( + repr(self.filename), compress_type, self.file_size, self.compress_size + ) + + +def _iter_files(f): + if f is None: + raise ValueError + f.seek(0) + try: + while True: + # 4.3.7 "Local file header" + offset = f.tell() + ( + magic, + min_version, + create_sys, + flags, + method, + modification_time, + modification_date, + crc32, + compressed_size, + uncompressed_size, + name_length, + extra_length, + ) = struct.unpack_from("> 9) + 1980, + (modification_date >> 5) & 0x0F, + modification_date & 0x1F, + modification_time >> 11, + (modification_time >> 5) & 0x3F, + (modification_time & 0x1F) << 1, + ), + ) + info.compress_type = method + info.CRC = crc32 + info.flag_bits = flags + info.compress_size = compressed_size + info.file_size = uncompressed_size + info.header_offset = offset + info.extract_version = min_version + info.create_version = min_version + info.create_system = create_sys + if extra_length > 0: + info.extra = f.read(extra_length) + if len(info.extra) != extra_length: + raise BadZipFile + yield info + + # Skip encryption header + if info.flag_bits & 0b1: + f.seek(12, 1) + f.seek(info.compress_size, 1) + except: + pass + + +def is_zipfile(filename): + if isinstance(filename, str): + try: + with open(filename, "rb") as file: + return file.read(4) == _MAGIC + except: + return False + + offset = filename.tell() + try: + return filename.read(4) == _MAGIC + finally: + filename.seek(offset) + + +class ZipFile: + def __init__(self, file, mode="r", **kwargs): + if mode != "r": + raise NotImplementedError + if isinstance(file, str): + self.filename = file + self._f = open(self.filename, "rb") + else: + self._f = file + self.filename = "ZipFile" + self.debug = 0 # For CPython compatibility + self.comment = "" # For CPython compatibility + + # CPython would automatically load and cache the names list, so calling + # `getinfo`, `infolist`, and `namelist` would still return data even after + # `close` was called. + + def close(self): + if self._f: + self._f.close() + self._f = None + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_value, traceback): + self.close() + + def getinfo(self, name): + for file in _iter_files(self._f): + if file.filename == name: + return file + raise KeyError + + def infolist(self): + return list(_iter_files(self._f)) + + def namelist(self): + return [file.filename for file in self.infolist()] + + def open(self, name, mode="r", pwd=None, **kwargs): + if self._f is None: + raise ValueError + if mode != "r" or pwd is not None: + raise NotImplementedError + info = self.getinfo(name) + # bail out if any bits related to encryption, reserved flags, or + # enhanced compression are set, or if the compression method is + # unsupported. + if ( + info.flag_bits & 0b1111000001111001 != 0 + or (info.compress_type == ZIP_DEFLATED and not _has_deflate) + or (info.compress_type not in (ZIP_STORED, ZIP_DEFLATED)) + ): + raise NotImplementedError + io_proxy = _IOProxy(self._f, info.file_size) + if info.compress_type is ZIP_STORED: + return io_proxy + return deflate.DeflateIO(io_proxy, deflate.RAW, 15) + + def extract(self, *args, **kwargs): + raise NotImplementedError + + def extractall(self, *args, **kwargs): + raise NotImplementedError + + def printdir(self): + raise NotImplementedError + + def setpassword(self, *args, **kwargs): + raise NotImplementedError + + def read(self, name, pwd=None): + with self.open(name, pwd=pwd) as f: + return f.read() + + def testzip(self): + raise NotImplementedError + + def write(self, *args, **kwargs): + raise NotImplementedError + + def writestr(self, *args, **kwargs): + raise NotImplementedError + + def mkdir(self, *args, **kwargs): + raise NotImplementedError diff --git a/tools/ci.sh b/tools/ci.sh index 2fc6fe90c..64c7c9607 100755 --- a/tools/ci.sh +++ b/tools/ci.sh @@ -109,6 +109,7 @@ function ci_package_tests_run { python-stdlib/unittest/tests \ python-stdlib/unittest-discover/tests \ python-stdlib/uuid \ + python-stdlib/zipfile \ ; do (cd $path && "${MICROPYTHON}" -m unittest) if [ $? -ne 0 ]; then false; return; fi