|
1 | 1 | import sys |
2 | 2 | import os |
3 | | -import py_compile |
4 | 3 | import marshal |
5 | 4 | import glob |
6 | 5 | import importlib |
@@ -301,52 +300,6 @@ def testBadMTime(self): |
301 | 300 | TESTMOD + pyc_ext: badtime_pyc} |
302 | 301 | self.doTest(".py", files, TESTMOD) |
303 | 302 |
|
304 | | - def testCorruptDeflateDataRaisesZipImportError(self): |
305 | | - # Regression test: a corrupt deflate stream inside a .pyc must |
306 | | - # raise ZipImportError, not leak zlib.error. |
307 | | - # See gh-158494. |
308 | | - if self.compression != ZIP_DEFLATED: |
309 | | - self.skipTest("requires ZIP_DEFLATED") |
310 | | - |
311 | | - modname = TESTMOD |
312 | | - |
313 | | - src = os.path.join(TEMP_DIR, modname + ".py") |
314 | | - self.addCleanup(os_helper.unlink, TEMP_ZIP) |
315 | | - self.addCleanup(os_helper.rmtree, TEMP_DIR) |
316 | | - |
317 | | - os.makedirs(TEMP_DIR, exist_ok=True) |
318 | | - with open(src, "w") as f: |
319 | | - f.write(test_src) |
320 | | - |
321 | | - # Build a real .pyc so its magic number matches this build. |
322 | | - pyc_src = os.path.join(TEMP_DIR, modname + ".pyc") |
323 | | - py_compile.compile(src, cfile=pyc_src, doraise=True) |
324 | | - |
325 | | - with ZipFile(TEMP_ZIP, "w", compression=ZIP_DEFLATED) as z: |
326 | | - z.write(src, modname + ".py") |
327 | | - z.write(pyc_src, modname + ".pyc") |
328 | | - |
329 | | - # Corrupt the compressed bytes of the .pyc entry in place. |
330 | | - with open(TEMP_ZIP, "r+b") as f: |
331 | | - data = f.read() |
332 | | - idx = data.find((modname + ".pyc").encode()) |
333 | | - self.assertNotEqual(idx, -1) |
334 | | - local_header = data.rfind(b"PK\x03\x04", 0, idx) |
335 | | - self.assertNotEqual(local_header, -1) |
336 | | - name_len, extra_len = struct.unpack_from( |
337 | | - "<HH", data, local_header + 26) |
338 | | - data_start = local_header + 30 + name_len + extra_len |
339 | | - f.seek(data_start) |
340 | | - f.write(b"\xff" * 8) |
341 | | - |
342 | | - zipimport._zip_directory_cache.clear() |
343 | | - sys.path.insert(0, TEMP_ZIP) |
344 | | - self.addCleanup(sys.modules.pop, modname, None) |
345 | | - importlib.invalidate_caches() |
346 | | - |
347 | | - with self.assertRaises(zipimport.ZipImportError): |
348 | | - importlib.import_module(modname) |
349 | | - |
350 | 303 | def test2038MTime(self): |
351 | 304 | # Make sure we can handle mtimes larger than what a 32-bit signed number |
352 | 305 | # can hold. |
@@ -1068,6 +1021,32 @@ def assertDataEntry(name): |
1068 | 1021 | class DeflateCompressedZipImportTestCase(UncompressedZipImportTestCase): |
1069 | 1022 | compression = ZIP_DEFLATED |
1070 | 1023 |
|
| 1024 | + def testCorruptDeflateDataRaisesZipImportError(self): |
| 1025 | + # gh-158494: a corrupt deflate stream in a .pyc must surface as |
| 1026 | + # ZipImportError, not zlib.error. |
| 1027 | + files = {TESTMOD + pyc_ext: test_pyc} |
| 1028 | + self.makeZip(files) |
| 1029 | + |
| 1030 | + # Corrupt the deflate stream of the .pyc entry in place. |
| 1031 | + with open(TEMP_ZIP, "r+b") as f: |
| 1032 | + data = f.read() |
| 1033 | + idx = data.find((TESTMOD + pyc_ext).encode()) |
| 1034 | + self.assertNotEqual(idx, -1) |
| 1035 | + lh = data.rfind(b"PK\x03\x04", 0, idx) |
| 1036 | + self.assertNotEqual(lh, -1) |
| 1037 | + name_len, extra_len = struct.unpack_from("<HH", data, lh + 26) |
| 1038 | + data_start = lh + 30 + name_len + extra_len |
| 1039 | + f.seek(data_start) |
| 1040 | + f.write(b"\xff" * 8) |
| 1041 | + |
| 1042 | + zipimport._zip_directory_cache.clear() |
| 1043 | + sys.path.insert(0, TEMP_ZIP) |
| 1044 | + self.addCleanup(sys.modules.pop, TESTMOD, None) |
| 1045 | + importlib.invalidate_caches() |
| 1046 | + |
| 1047 | + with self.assertRaises(zipimport.ZipImportError): |
| 1048 | + importlib.import_module(TESTMOD) |
| 1049 | + |
1071 | 1050 |
|
1072 | 1051 | @support.requires_zstd() |
1073 | 1052 | class ZStdCompressedZipImportTestCase(UncompressedZipImportTestCase): |
|
0 commit comments