Raise a clear error on ABF files whose header reports an impossible signal size (integer overflow)

neo/rawio/axonrawio.py

@@ -46,6 +46,7 @@
 
 """
 
+import os
 import struct
 import datetime
 from io import open, BufferedReader
@@ -160,9 +161,17 @@ def _parse_header(self):
         self._t_starts = {}
         self._buffer_descriptions = {0: {}}
         self._stream_buffer_slice = {stream_id: None}
+        # Offsets and segment sizes come from header fields that can be corrupt
+        # (truncated file, header surgery). Do the arithmetic in Python ints so it
+        # cannot silently overflow as numpy int32 would, and validate the implied
+        # data extent against the file on disk so a bad header raises a clear error
+        # rather than returning garbage or negative signal sizes.
+        head_offset = int(head_offset)
+        file_size = os.path.getsize(self.filename)
+
         pos = 0
         for seg_index in range(nb_segment):
-            length = episode_array[seg_index]["len"]
+            length = int(episode_array[seg_index]["len"])
 
             if version < 2.0:
                 fSynchTimeUnit = info["fSynchTimeUnit"]
@@ -172,6 +181,11 @@ def _parse_header(self):
             if (fSynchTimeUnit != 0) and (mode == 1):
                 length /= fSynchTimeUnit
 
+            if length < 0:
+                raise NeoReadWriteError(
+                    f"Negative segment size ({length}) parsed from {self.filename}; the file header is corrupt."
+                )
+
             self._buffer_descriptions[0][seg_index] = {}
             self._buffer_descriptions[0][seg_index][buffer_id] = {
                 "type": "raw",
@@ -190,6 +204,13 @@ def _parse_header(self):
                 t_start = t_start * fSynchTimeUnit * 1e-6
             self._t_starts[seg_index] = t_start
 
+        implied_data_end = head_offset + pos * sig_dtype.itemsize
+        if implied_data_end > file_size:
+            raise NeoReadWriteError(
+                f"ABF header implies {pos} samples ending at byte {implied_data_end}, which exceeds the "
+                f"file size of {file_size} bytes for {self.filename}; the file header is corrupt or the file is truncated."
+            )
+
         # Create channel header
         if version < 2.0:
             channel_ids = [chan_num for chan_num in info["nADCSamplingSeq"] if chan_num >= 0]

neo/test/rawiotest/test_axonrawio.py

@@ -1,6 +1,7 @@
 import unittest
 
 from neo.rawio.axonrawio import AxonRawIO
+from neo.core import NeoReadWriteError
 
 from neo.test.rawiotest.common_rawio_test import BaseTestRawIO
 
@@ -28,6 +29,29 @@ def test_read_raw_protocol(self):
 
         reader.read_raw_protocol()
 
+    def test_integer_overflow_size_raises(self):
+        # An ABF header that claims more samples than the file can hold must raise a
+        # clear error instead of silently returning an overflowed signal size.
+        path = self.get_local_path("axon/intracellular_data/files_with_errors/integer_overflow_size.abf")
+        expected_error = (
+            "ABF header implies 3221225472 samples ending at byte 6442457600, which exceeds the "
+            f"file size of 8704 bytes for {path}; the file header is corrupt or the file is truncated."
+        )
+        reader = AxonRawIO(filename=path)
+        with self.assertRaises(NeoReadWriteError) as cm:
+            reader.parse_header()
+        self.assertEqual(str(cm.exception), expected_error)
+
+    def test_negative_segment_size_raises(self):
+        # An ABF header with a negative segment size must raise a clear error
+        # instead of silently returning a negative signal size.
+        path = self.get_local_path("axon/intracellular_data/files_with_errors/negative_synch_length.abf")
+        expected_error = f"Negative segment size (-1041598657) parsed from {path}; the file header is corrupt."
+        reader = AxonRawIO(filename=path)
+        with self.assertRaises(NeoReadWriteError) as cm:
+            reader.parse_header()
+        self.assertEqual(str(cm.exception), expected_error)
+
 
 if __name__ == "__main__":
     unittest.main()