core: treat malformed definition/data messages as corruption, not crashes

pyulog/__init__.py

@@ -1,3 +1,4 @@
 """ Wrapper to include the main library modules """
 from .core import ULog
+from .exceptions import ULogException, ULogIncompatibleFlagException
 from . import px4

pyulog/core.py

@@ -5,6 +5,7 @@
 import sys
 import contextlib
 import numpy as np
+from .exceptions import ULogException, ULogIncompatibleFlagException
 #pylint: disable=too-many-instance-attributes, unused-argument, missing-docstring
 #pylint: disable=protected-access, too-many-branches
 
@@ -101,6 +102,7 @@ def __init__(self, log_file, message_name_filter_list=None, disable_str_exceptio
         :param message_name_filter_list: list of strings, to only load messages
                with the given names. If None, load everything.
         :param disable_str_parser_exceptions: If True, ignore string parsing errors
+        :raises ULogException: if the file cannot be parsed (invalid header, incompatible flag)
         """
 
         self._debug = False
@@ -648,6 +650,9 @@ def _extract_type(field_str):
             else:
                 b_pos = type_str.find(']')
                 array_size = int(type_str[a_pos+1:b_pos])
+                # malformed/huge array size from an untrusted log -> range() DoS
+                if array_size < 0 or array_size > 0xFFFF:
+                    raise ValueError("invalid array size: %d" % array_size)
                 type_name = type_str[:a_pos]
             return type_name, array_size, name_str
 
@@ -873,7 +878,12 @@ def _load_file(self, log_file, message_name_filter_list, parse_header_only=False
                 print('This file has data appended')
             for offset in self._appended_offsets:
                 self._read_file_data(message_name_filter_list, read_until=offset)
-                self._file_handle.seek(offset)
+                try:
+                    self._file_handle.seek(offset)
+                except (OverflowError, ValueError, OSError):
+                    # bogus appended offset from a corrupt FLAG_BITS -> corruption
+                    self._file_corrupt = True
+                    break
 
         # read the whole file, or the rest if data appended
         self._read_file_data(message_name_filter_list)
@@ -884,9 +894,9 @@ def _load_file(self, log_file, message_name_filter_list, parse_header_only=False
     def _read_file_header(self):
         header_data = self._file_handle.read(16)
         if len(header_data) != 16:
-            raise TypeError("Invalid file format (Header too short)")
+            raise ULogException("Invalid file format (Header too short)")
         if header_data[:7] != self.HEADER_BYTES:
-            raise TypeError("Invalid file format (Failed to parse header)")
+            raise ULogException("Invalid file format (Failed to parse header)")
         self._file_version, = struct.unpack('B', header_data[7:8])
         if self._file_version > 1:
             print("Warning: unknown file version. Will attempt to read it anyway")
@@ -900,9 +910,9 @@ def _read_file_definitions(self):
             data = self._file_handle.read(3)
             if not data:
                 break
-            header.initialize(data)
-            data = self._file_handle.read(header.msg_size)
             try:
+                header.initialize(data)
+                data = self._file_handle.read(header.msg_size)
                 if header.msg_type == self.MSG_TYPE_INFO:
                     msg_info = self._MessageInfo(data, header)
                     self._msg_info_dict[msg_info.key] = msg_info.value
@@ -941,10 +951,10 @@ def _read_file_definitions(self):
                     unknown_incompat_flag_msg = \
                     "Unknown incompatible flag set: cannot parse the log"
                     if self._incompat_flags[0] & ~1:
-                        raise ValueError(unknown_incompat_flag_msg)
+                        raise ULogIncompatibleFlagException(unknown_incompat_flag_msg)
                     for i in range(1, 8):
                         if self._incompat_flags[i]:
-                            raise NotImplementedError(unknown_incompat_flag_msg)
+                            raise ULogIncompatibleFlagException(unknown_incompat_flag_msg)
 
                 else:
                     if self._debug:
@@ -958,7 +968,9 @@ def _read_file_definitions(self):
                         # skipping the message
                         self._file_handle.seek(-2-header.msg_size, 1)
 
-            except IndexError:
+            except UnicodeDecodeError:
+                raise # strict string parsing (disable_str_exceptions=False)
+            except (IndexError, ValueError, KeyError, OverflowError, struct.error):
                 if not self._file_corrupt:
                     print("File corruption detected while reading file definitions!")
                     self._file_corrupt = True
@@ -1118,7 +1130,9 @@ def _read_file_data(self, message_name_filter_list, read_until=None):
                             if self._has_sync:
                                 self._find_sync(header.msg_size)
 
-                except IndexError:
+                except UnicodeDecodeError:
+                    raise # strict string parsing (disable_str_exceptions=False)
+                except (IndexError, KeyError, ValueError, OverflowError):
                     if not self._file_corrupt:
                         print("File corruption detected while reading file data!")
                         self._file_corrupt = True

pyulog/exceptions.py

@@ -0,0 +1,15 @@
+""" Exceptions raised by pyulog """
+
+
+class ULogException(Exception):
+    """
+    Raised if a ULog file cannot be parsed at all, e.g. because of an invalid header.
+    Corrupt messages are skipped instead, and reported via ULog.file_corruption.
+    """
+
+
+class ULogIncompatibleFlagException(ULogException):
+    """
+    Raised if a ULog file has an unknown incompatible flag set, i.e. it uses a format
+    extension that this version of pyulog cannot parse.
+    """

test/test_ulog.py

@@ -6,6 +6,7 @@
 import inspect
 import unittest
 import tempfile
+import struct
 from io import BytesIO
 
 from ddt import ddt, data
@@ -89,4 +90,54 @@ def test_write_ulog_memory(self, base_name):
             else:
                 assert copied_value == original_value
 
+    @staticmethod
+    def _build_ulog(*messages):
+        '''
+        Build a minimal ULog file in memory from (msg_type, payload) tuples.
+        '''
+        content = pyulog.ULog.HEADER_BYTES + b'\x01' + struct.pack('<Q', 0)
+        for msg_type, payload in messages:
+            content += struct.pack('<HB', len(payload), ord(msg_type)) + payload
+        return BytesIO(content)
+
+    def test_invalid_header(self):
+        '''
+        Test that a file with an invalid header raises a ULogException.
+        '''
+        for content in [b'ULog', b'\x00' * 16]:
+            with self.assertRaises(pyulog.ULogException):
+                pyulog.ULog(BytesIO(content))
+
+    @data(b'\x02' + bytes(7), # unknown flag in the first byte
+          b'\x00\x01' + bytes(6)) # any flag in the other bytes
+    def test_unknown_incompat_flag(self, incompat_flags):
+        '''
+        Test that an unknown incompatible flag raises a ULogIncompatibleFlagException.
+        '''
+        flag_bits = bytes(8) + incompat_flags + bytes(3 * 8)
+        with self.assertRaises(pyulog.ULogIncompatibleFlagException):
+            pyulog.ULog(self._build_ulog(('B', flag_bits)))
+
+    def test_corrupt_definitions(self):
+        '''
+        Test that malformed definitions are flagged as file corruption instead of raising.
+        '''
+        invalid_array_size = b'foo:uint8_t[x] a;'
+        bogus_appended_offset = bytes(8) + b'\x01' + bytes(7) + struct.pack('<QQQ', 1 << 63, 0, 0)
+        for msg_type, payload in [('F', invalid_array_size), ('B', bogus_appended_offset)]:
+            ulog = pyulog.ULog(self._build_ulog((msg_type, payload)))
+            assert ulog.file_corruption
+
+    def test_strict_string_parsing(self):
+        '''
+        Test that string parsing errors still raise with disable_str_exceptions=False,
+        both for an info message (definitions) and a logged string (data section).
+        '''
+        info = bytes([10]) + b'char[2] ab' + b'\xff\xfe'
+        logged_string = b'6' + bytes(8) + b'\xff\xfe'
+        for msg_type, payload in [('I', info), ('L', logged_string)]:
+            with self.assertRaises(UnicodeDecodeError):
+                pyulog.ULog(self._build_ulog((msg_type, payload)), disable_str_exceptions=False)
+            assert not pyulog.ULog(self._build_ulog((msg_type, payload))).file_corruption
+
 # vim: set et fenc=utf-8 ft=python ff=unix sts=4 sw=4 ts=4