gh-148395: Fix a possible UAF in {LZMA,BZ2,_Zlib}Decompressor

Lib/test/test_bz2.py

@@ -13,7 +13,7 @@
 import shutil
 import subprocess
 import threading
-from test.support import import_helper
+from test.support import import_helper, script_helper
 from test.support import threading_helper
 from test.support.os_helper import unlink, FakePath
 from compression._common import _streams
@@ -936,6 +936,35 @@ def testPickle(self):
             with self.assertRaises(TypeError):
                 pickle.dumps(BZ2Decompressor(), proto)
 
+    def test_decompressor_reuse_after_tail_copy_memory_error(self):
+        import_helper.import_module("_testcapi")
+        code = """if 1:
+            import _testcapi
+            import bz2
+
+            data = bz2.compress(b"S" * (1 << 20))
+            for skip in range(1, 500):
+                d = bz2.BZ2Decompressor()
+                try:
+                    _testcapi.set_nomemory(skip, skip + 1)
+                    try:
+                        d.decompress(data, max_length=0)
+                    except MemoryError:
+                        pass
+                    else:
+                        continue
+                finally:
+                    _testcapi.remove_mem_hooks()
+                if d.needs_input:
+                    continue
+                try:
+                    d.decompress(b"B" * 64, max_length=0)
+                except Exception:
+                    pass
+                break
+        """
+        script_helper.assert_python_ok("-c", code)
+
     def testDecompressorChunksMaxsize(self):
         bzd = BZ2Decompressor()
         max_length = 100

Lib/test/test_lzma.py

@@ -8,7 +8,7 @@
 import unittest
 from compression._common import _streams
 
-from test.support import _4G, bigmemtest
+from test.support import _4G, bigmemtest, script_helper
 from test.support.import_helper import import_module
 from test.support.os_helper import (
     TESTFN, unlink, FakePath
@@ -383,6 +383,35 @@ def test_uninitialized_LZMADecompressor_crash(self):
         self.assertEqual(LZMADecompressor.__new__(LZMADecompressor).
                          decompress(bytes()), b'')
 
+    def test_decompressor_reuse_after_tail_copy_memory_error(self):
+        import_module("_testcapi")
+        code = """if 1:
+            import _testcapi
+            import lzma
+
+            data = lzma.compress(b"S" * (1 << 20))
+            for skip in range(1, 500):
+                d = lzma.LZMADecompressor()
+                try:
+                    _testcapi.set_nomemory(skip, skip + 1)
+                    try:
+                        d.decompress(data, max_length=0)
+                    except MemoryError:
+                        pass
+                    else:
+                        continue
+                finally:
+                    _testcapi.remove_mem_hooks()
+                if d.needs_input:
+                    continue
+                try:
+                    d.decompress(b"B" * 64, max_length=0)
+                except Exception:
+                    pass
+                break
+        """
+        script_helper.assert_python_ok("-c", code)
+
 
 class CompressDecompressFunctionTestCase(unittest.TestCase):
 

Lib/test/test_zlib.py

@@ -1,6 +1,6 @@
 import unittest
 from test import support
-from test.support import import_helper
+from test.support import import_helper, script_helper
 import binascii
 import copy
 import pickle
@@ -1207,6 +1207,35 @@ def test_failure(self):
         # Previously, a second call could crash due to internal inconsistency
         self.assertRaises(Exception, zlibd.decompress, self.BAD_DATA * 30)
 
+    def test_decompressor_reuse_after_tail_copy_memory_error(self):
+        import_helper.import_module("_testcapi")
+        code = """if 1:
+            import _testcapi
+            import zlib
+
+            data = zlib.compress(b"S" * (1 << 20))
+            for skip in range(1, 500):
+                d = zlib._ZlibDecompressor()
+                try:
+                    _testcapi.set_nomemory(skip, skip + 1)
+                    try:
+                        d.decompress(data, max_length=0)
+                    except MemoryError:
+                        pass
+                    else:
+                        continue
+                finally:
+                    _testcapi.remove_mem_hooks()
+                if d.needs_input:
+                    continue
+                try:
+                    d.decompress(b"B" * 64, max_length=0)
+                except Exception:
+                    pass
+                break
+        """
+        script_helper.assert_python_ok("-c", code)
+
     @support.refcount_test
     def test_refleaks_in___init__(self):
         gettotalrefcount = support.get_attribute(sys, 'gettotalrefcount')

Misc/NEWS.d/next/Security/2026-04-10-16-28-21.gh-issue-148395.kfzm0G.rst

@@ -0,0 +1,5 @@
+Fix a dangling input pointer in :class:`lzma.LZMADecompressor`,
+:class:`bz2.BZ2Decompressor`, and internal :class:`!zlib._ZlibDecompressor`
+when memory allocation fails with :exc:`MemoryError`, which could let a
+subsequent :meth:`!decompress` call read or write through a stale pointer to
+the already-released caller buffer.

Modules/_bz2module.c

@@ -569,6 +569,7 @@ decompress(BZ2Decompressor *d, char *data, size_t len, Py_ssize_t max_length)
     return result;
 
 error:
+    bzs->next_in = NULL;
     Py_XDECREF(result);
     return NULL;
 }

Modules/_lzmamodule.c

@@ -1100,6 +1100,7 @@ decompress(Decompressor *d, uint8_t *data, size_t len, Py_ssize_t max_length)
     return result;
 
 error:
+    lzs->next_in = NULL;
     Py_XDECREF(result);
     return NULL;
 }

Modules/zlibmodule.c

@@ -1669,6 +1669,7 @@ decompress(ZlibDecompressor *self, uint8_t *data,
     return result;
 
 error:
+    self->zst.next_in = NULL;
     Py_XDECREF(result);
     return NULL;
 }