Issue #1159051: GzipFile now raises EOFError when reading a corrupted file

with truncated header or footer.
Added tests for reading truncated gzip and bzip2 files.

2 files changed, 33 insertions, 0 deletions

diff --git a/Lib/test/test_bz2.py b/Lib/test/test_bz2.py
index 977d10b..fb104d7 100644
--- a/Lib/test/test_bz2.py
+++ b/Lib/test/test_bz2.py
@@ -295,6 +295,24 @@ class BZ2FileTest(BaseTest):
             self.assertRaises(ValueError, f.readline)
             self.assertRaises(ValueError, f.readlines)
 
+    def test_read_truncated(self):
+        # Drop the eos_magic field (6 bytes) and CRC (4 bytes).
+        truncated = self.DATA[:-10]
+        with open(self.filename, 'wb') as f:
+            f.write(truncated)
+        with BZ2File(self.filename) as f:
+            self.assertRaises(EOFError, f.read)
+        with BZ2File(self.filename) as f:
+            self.assertEqual(f.read(len(self.TEXT)), self.TEXT)
+            self.assertRaises(EOFError, f.read, 1)
+        # Incomplete 4-byte file header, and block header of at least 146 bits.
+        for i in range(22):
+            with open(self.filename, 'wb') as f:
+                f.write(truncated[:i])
+            with BZ2File(self.filename) as f:
+                self.assertRaises(EOFError, f.read, 1)
+
+
 class BZ2CompressorTest(BaseTest):
     def testCompress(self):
         # "Test BZ2Compressor.compress()/flush()"
diff --git a/Lib/test/test_gzip.py b/Lib/test/test_gzip.py
index ced226f..ba9d7da 100644
--- a/Lib/test/test_gzip.py
+++ b/Lib/test/test_gzip.py
@@ -365,6 +365,21 @@ class TestGzip(unittest.TestCase):
             datac = gzip.compress(data)
             self.assertEqual(gzip.decompress(datac), data)
 
+    def test_read_truncated(self):
+        data = data1*50
+        # Drop the CRC (4 bytes) and file size (4 bytes).
+        truncated = gzip.compress(data)[:-8]
+        with gzip.GzipFile(fileobj=io.BytesIO(truncated)) as f:
+            self.assertRaises(EOFError, f.read)
+        with gzip.GzipFile(fileobj=io.BytesIO(truncated)) as f:
+            self.assertEqual(f.read(len(data)), data)
+            self.assertRaises(EOFError, f.read, 1)
+        # Incomplete 10-byte header.
+        for i in range(2, 10):
+            with gzip.GzipFile(fileobj=io.BytesIO(truncated[:i])) as f:
+                self.assertRaises(EOFError, f.read, 1)
+
+
 def test_main(verbose=None):
     support.run_unittest(TestGzip)