From 2581f94429524c630c9c518da6ad6b1dfd0752bb Mon Sep 17 00:00:00 2001 From: Alessandro Gatti Date: Fri, 4 Sep 2026 11:21:07 +0200 Subject: [PATCH] python-stdlib/zipfile: Add read-only zipfile module. This commit adds an implementation for the `zipfile` module that is able to open Zip archives, enumerate their contents, and read data out of the single files present in the archive. The `zipfile` implementation in this commit is relatively limited, but should be enough for most cases in which data is being read out of a Zip archive. Things like file extraction, or CRC checking are not implemented as they are probably better handled on a per-case basis on embedded targets anyway. Other features not supported are additional compression modes besides STORED and DEFLATED, encryption, large files, and files that contain an additional data descriptor with the proper file size and CRC. Most of the CPython API is implemented, except for `zipfile.Path`, `zipfile.PyZipFile` and other specific exceptions besides `zipfile.BadZipFile`. Signed-off-by: Alessandro Gatti --- python-stdlib/zipfile/README.md | 25 ++ python-stdlib/zipfile/deflate.zip | Bin 0 -> 691 bytes python-stdlib/zipfile/manifest.py | 2 + python-stdlib/zipfile/null.zip | Bin 0 -> 4789 bytes python-stdlib/zipfile/stored.zip | Bin 0 -> 12344 bytes python-stdlib/zipfile/test_zipfile.py | 205 +++++++++++++++ python-stdlib/zipfile/zipfile/__init__.py | 299 ++++++++++++++++++++++ tools/ci.sh | 1 + 8 files changed, 532 insertions(+) create mode 100644 python-stdlib/zipfile/README.md create mode 100644 python-stdlib/zipfile/deflate.zip create mode 100644 python-stdlib/zipfile/manifest.py create mode 100644 python-stdlib/zipfile/null.zip create mode 100644 python-stdlib/zipfile/stored.zip create mode 100644 python-stdlib/zipfile/test_zipfile.py create mode 100644 python-stdlib/zipfile/zipfile/__init__.py 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 0000000000000000000000000000000000000000..90d3e0f97f28bc1ae3b5bdba326e5ef8b62689e7 GIT binary patch literal 691 zcmWIWW@Zs#0D-13l~^zXN^k+`l+2>kM~Mu za`NL7(-SlE^hzp9-k#jZ*jL&y!1Y zUus+zPyc^E<8@Mx;W3G11d%p}5o9sQDGUtj{7@ap3KUDtEhwo(xUCH6hs`(VudoEE z2YQPWNP%4hcFT#4oCg$uZm|nisb%vNKEdZ0ARvF_*xS3#?@m<*tu36N|FgQL_({g~ zhjWV0Sw4plw;ziIcr!BDGvf*+6_8y(AkgsE5kw=xjEg}5B*4JPAi=Qj?ylutHg_4p z0!ReHY-}M5Gh3lyNuvf(7s70`a3fHuFz8gk!3O}IDV%}; literal 0 HcmV?d00001 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 0000000000000000000000000000000000000000..235e25a75b5a18d779fbd7dcc622b60b1d513f83 GIT binary patch literal 4789 zcmWIWW@gc2U|`^3&R($4Ir{ z$PNaJ#|#YY2UrdO6*Vv{U{F|T>o{ZE7546(N^>+kTo_~u8v=kPivXoL7!05$gLoj5 zfe2!9N@`k8Vo55UGQpmK3Fs1-37SB0kamz#u-kH~ zauY$?rI=V5V!;M8=E(Qg#JR^eygZmBfY0frx8NmWb1jKw? zgCxl2M*($#%|{wCLAHU-{`Yj;HsBgWLAGHI%m&zy3bOGFH*?LzZ9J~=6lCK+z>G(X ct{_{``a^Fbb_-b9Kmo=AgiXNklNAK<0QfkCK>z>% literal 0 HcmV?d00001 diff --git a/python-stdlib/zipfile/stored.zip b/python-stdlib/zipfile/stored.zip new file mode 100644 index 0000000000000000000000000000000000000000..69869c717d7dc10ad3b50e4f19f7e9351dec33ef GIT binary patch literal 12344 zcmWIWW@Zs#0D-13l~^zXN^k+`l+2>k=5!D!!kbo^j+3}tk*XmrGEbkuNks;&YYDg*(6hPRF&8nGaqi$MV-z`)2L!LaY{ zuH{}fcNxI~NCd=eT+7>GW-Bx-Y19DfLYR%V*qv0XFo#5ut%?V(tp^W~A{#qxMn;7b p0UuxvqaqvI1T+>tsETYf>s5<6*aL@^4HQ-^Kv)S3x|Mn$9spBVP9gvR literal 0 HcmV?d00001 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