diff --git a/Lib/idlelib/idle_test/template.py b/Lib/idlelib/idle_test/template.py index 0a4bd8b8e981fc7..7c3df6ba8fbce38 100644 --- a/Lib/idlelib/idle_test/template.py +++ b/Lib/idlelib/idle_test/template.py @@ -21,6 +21,7 @@ def tearDownClass(cls): cls.root.destroy() del cls.root + @unittest.skip('Dummy test') def test_init(self): self.assertTrue(True) diff --git a/Lib/idlelib/idle_test/test_autocomplete.py b/Lib/idlelib/idle_test/test_autocomplete.py index a811363c18d04e5..88af3efc35bbd18 100644 --- a/Lib/idlelib/idle_test/test_autocomplete.py +++ b/Lib/idlelib/idle_test/test_autocomplete.py @@ -230,16 +230,14 @@ def test_fetch_completions(self): # For file completion, a large list containing all files in the path, # and a small list containing files that do not start with '.'. acp = self.autocomplete - small, large = acp.fetch_completions( - '', ac.ATTRS) - if hasattr(__main__, '__file__') and __main__.__file__ != ac.__file__: - self.assertNotIn('AutoComplete', small) # See issue 36405. - # Test attributes - s, b = acp.fetch_completions('', ac.ATTRS) - self.assertLess(len(small), len(large)) - self.assertTrue(all(filter(lambda x: x.startswith('_'), s))) - self.assertTrue(any(filter(lambda x: x.startswith('_'), b))) + # Test current module (what='') attributes. + small, large = acp.fetch_completions('', ac.ATTRS) + if hasattr(__main__, '__file__') and __main__.__file__ != ac.__file__: + self.assertNotIn('AutoComplete', small) # See gh-80586. + self.assertLess(len(small), len(large)) # Not equal + self.assertFalse(any(a[:1] == '_' for a in small)) + self.assertTrue(any(a[:1] == '_' for a in large)) # Test smalll should respect to __all__. with patch.dict('__main__.__dict__', {'__all__': ['a', 'b']}): diff --git a/Lib/idlelib/idle_test/test_configdialog.py b/Lib/idlelib/idle_test/test_configdialog.py index 696a3b2f8f1bc23..f3e1c785a92674c 100644 --- a/Lib/idlelib/idle_test/test_configdialog.py +++ b/Lib/idlelib/idle_test/test_configdialog.py @@ -50,6 +50,7 @@ def tearDownModule(): root = dialog = None +@unittest.skip('Empty tests') class ConfigDialogTest(unittest.TestCase): def test_deactivate_current_config(self): diff --git a/Lib/idlelib/idle_test/test_editor.py b/Lib/idlelib/idle_test/test_editor.py index e28ee549f180aa0..1fcea4f1eb0d6a0 100644 --- a/Lib/idlelib/idle_test/test_editor.py +++ b/Lib/idlelib/idle_test/test_editor.py @@ -211,6 +211,7 @@ def test_searcher(self): self.assertEqual(actual_pair, expected_pair) +@unittest.skip('Empty test') class RMenuTest(unittest.TestCase): @classmethod diff --git a/Lib/test/test_zipfile/test_core.py b/Lib/test/test_zipfile/test_core.py index 1c6e3a9f0a9a2de..fdf2cd26f8c7c64 100644 --- a/Lib/test/test_zipfile/test_core.py +++ b/Lib/test/test_zipfile/test_core.py @@ -4874,6 +4874,48 @@ def tearDown(self): unlink(TESTFN2) +class AbstractBoundedDecompressTests: + # ZipExtFile._read1() bounds the output of each decompress() call so that a + # small member declaring a large uncompressed size cannot expand into one + # unbounded read. + def test_read1_output_is_bounded(self): + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w", compression=self.compression) as zf: + zf.writestr("big", b"\0" * (4 * 1024 * 1024)) + with zipfile.ZipFile(io.BytesIO(buf.getvalue())) as zf: + with zf.open("big") as f: + self.assertLessEqual(len(f._read1(100)), f.MIN_READ_SIZE) + + +class StoredBoundedDecompressTests(AbstractBoundedDecompressTests, + unittest.TestCase): + compression = zipfile.ZIP_STORED + + +@requires_zlib() +class DeflateBoundedDecompressTests(AbstractBoundedDecompressTests, + unittest.TestCase): + compression = zipfile.ZIP_DEFLATED + + +@requires_bz2() +class Bzip2BoundedDecompressTests(AbstractBoundedDecompressTests, + unittest.TestCase): + compression = zipfile.ZIP_BZIP2 + + +@requires_lzma() +class LzmaBoundedDecompressTests(AbstractBoundedDecompressTests, + unittest.TestCase): + compression = zipfile.ZIP_LZMA + + +@requires_zstd() +class ZstdBoundedDecompressTests(AbstractBoundedDecompressTests, + unittest.TestCase): + compression = zipfile.ZIP_ZSTANDARD + + class AbstractBadCrcTests: def test_testzip_with_bad_crc(self): """Tests that files with bad CRCs return their name from testzip.""" diff --git a/Lib/test/test_zipfile64.py b/Lib/test/test_zipfile64.py index 7d802d59849ce1f..e13f064f2ac4fd4 100644 --- a/Lib/test/test_zipfile64.py +++ b/Lib/test/test_zipfile64.py @@ -17,6 +17,7 @@ import sys import unittest.mock as mock +from contextlib import contextmanager from tempfile import TemporaryFile from test.support import os_helper @@ -91,176 +92,133 @@ def tearDown(self): os_helper.unlink(TESTFN2) -class TestRepack(unittest.TestCase): - def setUp(self): - # Create test data. - line_gen = ("Test of zipfile line %d." % i for i in range(1000000)) - self.data = '\n'.join(line_gen).encode('ascii') - - # It will contain enough copies of self.data to reach about 8 GiB. - self.datacount = 8*1024**3 // len(self.data) +class TestRepacker(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.largefilename = 'largefile.txt' - # memory usage should not exceed 10 MiB - self.allowed_memory = 10*1024**2 + line_gen = ("Test of zipfile line %d." % i for i in range(1000000)) + cls.chunk = '\n'.join(line_gen).encode('ascii') + + # It will contain enough copies of cls.chunk to reach about 4.1 GiB. + cls.chunkcount = int(4.1*1024**3 / len(cls.chunk)) + + cls.filename = 'file.txt' + cls.lorem = b'Sed ut perspiciatis unde omnis iste natus error sit voluptatem' + + # Memory usage should not exceed 10 MiB during repacking. + # This empirical threshold ensures that the internal processing + # like signature scanning, compressed block end tracing, and + # data copying are properly buffered without loading the entire + # large file into memory. + cls.allowed_memory = 10*1024**2 + + @contextmanager + def assert_memory_usage(self, threshold): + tracemalloc.start() + try: + yield + finally: + current, peak = tracemalloc.get_traced_memory() + tracemalloc.stop() + self.assertLess(peak, threshold) def _write_large_file(self, fh): next_time = time.monotonic() + _PRINT_WORKING_MSG_INTERVAL - for num in range(self.datacount): - fh.write(self.data) + for num in range(self.chunkcount): + fh.write(self.chunk) # Print still working message since this test can be really slow if next_time <= time.monotonic(): next_time = time.monotonic() + _PRINT_WORKING_MSG_INTERVAL print(( ' writing %d of %d, be patient...' % - (num, self.datacount)), file=sys.__stdout__) + (num, self.chunkcount)), file=sys.__stdout__) sys.__stdout__.flush() def test_strip_removed_large_file(self): """Should move the physical data of a file positioned after a large removed file without causing a memory issue.""" - # Try the temp file. If we do TESTFN2, then it hogs - # gigabytes of disk space for the duration of the test. with TemporaryFile() as f: - tracemalloc.start() - self._test_strip_removed_large_file(f) - self.assertFalse(f.closed) - current, peak = tracemalloc.get_traced_memory() - tracemalloc.stop() - self.assertLess(peak, self.allowed_memory) - - def _test_strip_removed_large_file(self, f): - file = 'file.txt' - file1 = 'largefile.txt' - data = b'Sed ut perspiciatis unde omnis iste natus error sit voluptatem' - with zipfile.ZipFile(f, 'w') as zh: - with zh.open(file1, 'w', force_zip64=True) as fh: - self._write_large_file(fh) - zh.writestr(file, data) - - with zipfile.ZipFile(f, 'a') as zh: - zh.remove(file1) - zh.repack() - self.assertIsNone(zh.testzip()) + with zipfile.ZipFile(f, 'w') as zh: + with zh.open(self.largefilename, 'w', force_zip64=True) as fh: + self._write_large_file(fh) + zh.writestr(self.filename, self.lorem) + + with self.assert_memory_usage(self.allowed_memory), \ + zipfile.ZipFile(f, 'a') as zh: + zh.remove(self.largefilename) + zh.repack() + self.assertIsNone(zh.testzip()) def test_strip_removed_file_before_large_file(self): """Should move the physical data of a large file positioned after a removed file without causing a memory issue.""" - # Try the temp file. If we do TESTFN2, then it hogs - # gigabytes of disk space for the duration of the test. with TemporaryFile() as f: - tracemalloc.start() - self._test_strip_removed_file_before_large_file(f) - self.assertFalse(f.closed) - current, peak = tracemalloc.get_traced_memory() - tracemalloc.stop() - self.assertLess(peak, self.allowed_memory) - - def _test_strip_removed_file_before_large_file(self, f): - file = 'file.txt' - file1 = 'largefile.txt' - data = b'Sed ut perspiciatis unde omnis iste natus error sit voluptatem' - with zipfile.ZipFile(f, 'w') as zh: - zh.writestr(file, data) - with zh.open(file1, 'w', force_zip64=True) as fh: - self._write_large_file(fh) - - with zipfile.ZipFile(f, 'a') as zh: - zh.remove(file) - zh.repack() - self.assertIsNone(zh.testzip()) + with zipfile.ZipFile(f, 'w') as zh: + zh.writestr(self.filename, self.lorem) + with zh.open(self.largefilename, 'w', force_zip64=True) as fh: + self._write_large_file(fh) + + with self.assert_memory_usage(self.allowed_memory), \ + zipfile.ZipFile(f, 'a') as zh: + zh.remove(self.filename) + zh.repack() + self.assertIsNone(zh.testzip()) def test_strip_removed_large_file_with_dd(self): """Should scan for the data descriptor of a removed large file without causing a memory issue.""" - # Try the temp file. If we do TESTFN2, then it hogs - # gigabytes of disk space for the duration of the test. with TemporaryFile() as f: - tracemalloc.start() - self._test_strip_removed_large_file_with_dd(f) - self.assertFalse(f.closed) - current, peak = tracemalloc.get_traced_memory() - tracemalloc.stop() - self.assertLess(peak, self.allowed_memory) - - def _test_strip_removed_large_file_with_dd(self, f): - file = 'file.txt' - file1 = 'largefile.txt' - data = b'Sed ut perspiciatis unde omnis iste natus error sit voluptatem' - with zipfile.ZipFile(Unseekable(f), 'w') as zh: - with zh.open(file1, 'w', force_zip64=True) as fh: - self._write_large_file(fh) - zh.writestr(file, data) - - with zipfile.ZipFile(f, 'a') as zh: - zh.remove(file1) - zh.repack() - self.assertIsNone(zh.testzip()) + with zipfile.ZipFile(Unseekable(f), 'w') as zh: + with zh.open(self.largefilename, 'w', force_zip64=True) as fh: + self._write_large_file(fh) + zh.writestr(self.filename, self.lorem) + + with self.assert_memory_usage(self.allowed_memory), \ + zipfile.ZipFile(f, 'a') as zh: + zh.remove(self.largefilename) + zh.repack() + self.assertIsNone(zh.testzip()) def test_strip_removed_large_file_with_dd_no_sig(self): - """Should scan for the data descriptor (without signature) of a removed - large file without causing a memory issue.""" + """Should scan for the unsigned data descriptor of a removed large file + without causing a memory issue.""" # Reduce data scale for this test, as it's especially slow... - self.datacount = 30*1024**2 // len(self.data) - self.allowed_memory = 200*1024 + self.chunkcount = int(30*1024**2 / len(self.chunk)) - # Try the temp file. If we do TESTFN2, then it hogs - # gigabytes of disk space for the duration of the test. with TemporaryFile() as f: - tracemalloc.start() - self._test_strip_removed_large_file_with_dd_no_sig(f) - self.assertFalse(f.closed) - current, peak = tracemalloc.get_traced_memory() - tracemalloc.stop() - self.assertLess(peak, self.allowed_memory) - - def _test_strip_removed_large_file_with_dd_no_sig(self, f): - file = 'file.txt' - file1 = 'largefile.txt' - data = b'Sed ut perspiciatis unde omnis iste natus error sit voluptatem' - with mock.patch('zipfile.struct.pack', side_effect=struct_pack_no_dd_sig): - with zipfile.ZipFile(Unseekable(f), 'w') as zh: - with zh.open(file1, 'w', force_zip64=True) as fh: + with mock.patch('zipfile.struct.pack', side_effect=struct_pack_no_dd_sig), \ + zipfile.ZipFile(Unseekable(f), 'w') as zh: + with zh.open(self.largefilename, 'w', force_zip64=True) as fh: self._write_large_file(fh) - zh.writestr(file, data) + zh.writestr(self.filename, self.lorem) - with zipfile.ZipFile(f, 'a') as zh: - zh.remove(file1) - # strict_descriptor=False to scan the unsigned data descriptor - # (scanning is disabled under the strict_descriptor=True default) - zh.repack(strict_descriptor=False) - self.assertIsNone(zh.testzip()) + with self.assert_memory_usage(self.allowed_memory), \ + zipfile.ZipFile(f, 'a') as zh: + zh.remove(self.largefilename) + # strict_descriptor=False to scan the unsigned data descriptor + # (scanning is disabled under the strict_descriptor=True default) + zh.repack(strict_descriptor=False) + self.assertIsNone(zh.testzip()) @requires_zlib() def test_strip_removed_large_file_with_dd_no_sig_by_decompression(self): - """Should scan for the data descriptor (without signature) of a removed - large file without causing a memory issue.""" - # Try the temp file. If we do TESTFN2, then it hogs - # gigabytes of disk space for the duration of the test. + """Should scan for the unsigned data descriptor (via tracing compressed + block end) of a removed large file without causing a memory issue.""" with TemporaryFile() as f: - tracemalloc.start() - self._test_strip_removed_large_file_with_dd_no_sig_by_decompression( - f, zipfile.ZIP_DEFLATED) - self.assertFalse(f.closed) - current, peak = tracemalloc.get_traced_memory() - tracemalloc.stop() - self.assertLess(peak, self.allowed_memory) - - def _test_strip_removed_large_file_with_dd_no_sig_by_decompression(self, f, method): - file = 'file.txt' - file1 = 'largefile.txt' - data = b'Sed ut perspiciatis unde omnis iste natus error sit voluptatem' - with mock.patch('zipfile.struct.pack', side_effect=struct_pack_no_dd_sig): - with zipfile.ZipFile(Unseekable(f), 'w', compression=method) as zh: - with zh.open(file1, 'w', force_zip64=True) as fh: + with mock.patch('zipfile.struct.pack', side_effect=struct_pack_no_dd_sig), \ + zipfile.ZipFile(Unseekable(f), 'w', compression=zipfile.ZIP_DEFLATED) as zh: + with zh.open(self.largefilename, 'w', force_zip64=True) as fh: self._write_large_file(fh) - zh.writestr(file, data) - - with zipfile.ZipFile(f, 'a') as zh: - zh.remove(file1) - # strict_descriptor=False to detect the unsigned data descriptor - # (scanning is disabled under the strict_descriptor=True default) - zh.repack(strict_descriptor=False) - self.assertIsNone(zh.testzip()) + zh.writestr(self.filename, self.lorem) + + with self.assert_memory_usage(self.allowed_memory), \ + zipfile.ZipFile(f, 'a') as zh: + zh.remove(self.largefilename) + # strict_descriptor=False to detect the unsigned data descriptor + # (scanning is disabled under the strict_descriptor=True default) + zh.repack(strict_descriptor=False) + self.assertIsNone(zh.testzip()) class OtherTests(unittest.TestCase): diff --git a/Lib/zipfile/__init__.py b/Lib/zipfile/__init__.py index 7a81aa8f44c8f4c..0accf324c90e3fd 100644 --- a/Lib/zipfile/__init__.py +++ b/Lib/zipfile/__init__.py @@ -801,7 +801,16 @@ def unused_data(self): except AttributeError: return b'' - def decompress(self, data): + @property + def _needs_input(self): + # While the LZMA properties header is still being buffered, more input + # is required; afterwards defer to the wrapped decompressor so a bounded + # decompress() call can be drained across reads. + if self._decomp is None: + return True + return self._decomp.needs_input + + def decompress(self, data, max_length=-1): if self._decomp is None: self._unconsumed += data if len(self._unconsumed) <= 4: @@ -817,7 +826,7 @@ def decompress(self, data): data = self._unconsumed[4 + psize:] del self._unconsumed - result = self._decomp.decompress(data) + result = self._decomp.decompress(data, max_length) self.eof = self._decomp.eof return result @@ -884,6 +893,13 @@ def _get_compressor(compress_type, compresslevel=None): return None +def _decompressor_needs_input(decompressor): + # bz2/zstd expose the stdlib decompressor's public needs_input; the LZMA + # wrapper keeps it private (_needs_input) to avoid adding public API. + needs_input = getattr(decompressor, "needs_input", None) + return decompressor._needs_input if needs_input is None else needs_input + + def _get_decompressor(compress_type): _check_compression(compress_type) if compress_type == ZIP_STORED: @@ -1186,8 +1202,15 @@ def _read1(self, n): data = self._decompressor.unconsumed_tail if n > len(data): data += self._read2(n - len(data)) - else: + elif self._compress_type == ZIP_STORED: data = self._read2(n) + else: + # bzip2/lzma/zstd: a bounded decompress() call may leave input + # buffered inside the decompressor; drain that before reading more. + if _decompressor_needs_input(self._decompressor): + data = self._read2(n) + else: + data = b'' if self._compress_type == ZIP_STORED: self._eof = self._compress_left <= 0 @@ -1200,8 +1223,13 @@ def _read1(self, n): if self._eof: data += self._decompressor.flush() else: - data = self._decompressor.decompress(data) - self._eof = self._decompressor.eof or self._compress_left <= 0 + # Bound the output of a single decompress() call (mirroring the + # DEFLATE path above) so that a small compressed member cannot + # expand into one unbounded read. + data = self._decompressor.decompress(data, max(n, self.MIN_READ_SIZE)) + self._eof = (self._decompressor.eof or + self._compress_left <= 0 and + _decompressor_needs_input(self._decompressor)) data = data[:self._left] self._left -= len(data) diff --git a/Misc/NEWS.d/next/Security/2026-08-18-13-54-05.gh-issue-156002.CcWXPP.rst b/Misc/NEWS.d/next/Security/2026-08-18-13-54-05.gh-issue-156002.CcWXPP.rst new file mode 100644 index 000000000000000..4e49ad5ce8fa00a --- /dev/null +++ b/Misc/NEWS.d/next/Security/2026-08-18-13-54-05.gh-issue-156002.CcWXPP.rst @@ -0,0 +1,4 @@ +Bound the amount of data :mod:`zipfile` decompresses per read for members +compressed with bzip2, LZMA, or Zstandard, matching the existing limit for +deflate. A small archive member could previously expand into an unbounded +allocation even when read in small chunks.