Changes to systemd journal parser to handle corrupt and binary data

plaso/parsers/systemd_journal.py

@@ -139,7 +139,13 @@ def _ParseDataObject(self, file_object, file_offset):
         data = file_object.read(data_size)
 
         if data_object.object_flags & self._OBJECT_COMPRESSED_FLAG_XZ:
-            data = lzma.decompress(data)
+            try:
+                data = lzma.decompress(data)
+            except (EOFError, lzma.LZMAError) as exception:
+                raise errors.ParseError(
+                    f"Unable to decompress XZ at offset: 0x{data_offset:08x} with "
+                    f"error: {exception!s}"
+                )
 
         elif data_object.object_flags & self._OBJECT_COMPRESSED_FLAG_LZ4:
             uncompressed_size_map = self._GetDataTypeMap("uint32le")
@@ -154,7 +160,15 @@ def _ParseDataObject(self, file_object, file_offset):
                     f"0x{data_offset:08x} with error: {exception!s}"
                 )
 
-            data = lz4_block.decompress(data[8:], uncompressed_size=uncompressed_size)
+            try:
+                data = lz4_block.decompress(
+                    data[8:], uncompressed_size=uncompressed_size
+                )
+            except (ValueError, lz4_block.LZ4BlockError) as exception:
+                raise errors.ParseError(
+                    f"Unable to decompress LZ4 at offset: 0x{data_offset:08x} with "
+                    f"error: {exception!s}"
+                )
 
         elif data_object.object_flags & self._OBJECT_COMPRESSED_FLAG_ZSTD:
             try:
@@ -272,12 +286,47 @@ def _ParseEntryObjectOffsets(self, file_object, file_offset):
 
         return entry_object_offsets
 
-    def _ParseJournalEntry(self, file_object, file_offset):
+    def _ParseKeyValuePair(self, parser_mediator, field_data):
+        """Parses a key value pair.
+
+        Args:
+          parser_mediator (ParserMediator): parser mediator.
+          field_data (bytes): journal field data, which is "key=value" where the
+              value can contain arbitrary binary (non-UTF-8) data.
+
+        Returns:
+          tuple[str, str]: key and value decoded as UTF-8, with invalid bytes
+              replaced rather than raising.
+        """
+        key_data, _, value_data = field_data.partition(b"=")
+
+        try:
+            key = key_data.decode("utf-8")
+        except UnicodeDecodeError:
+            parser_mediator.ProduceExtractionWarning(
+                "Unable to decode journal field key as UTF-8. Unsupported code points "
+                "are escaped."
+            )
+            key = key_data.decode("utf-8", errors="backslashreplace")
+
+        try:
+            value = value_data.decode("utf-8")
+        except UnicodeDecodeError:
+            parser_mediator.ProduceExtractionWarning(
+                f"Unable to decode journal field with key: {key:s} value as UTF-8. "
+                f"Unsupported code points are escaped."
+            )
+            value = value_data.decode("utf-8", errors="backslashreplace")
+
+        return key, value
+
+    def _ParseJournalEntry(self, parser_mediator, file_object, file_offset):
         """Parses a journal entry.
 
         This method will generate an event per ENTRY object.
 
         Args:
+          parser_mediator (ParserMediator): parser mediator.
           file_object (dfvfs.FileIO): a file-like object.
           file_offset (int): offset of the entry object relative to the start
               of the file-like object.
@@ -308,8 +357,8 @@ def _ParseJournalEntry(self, file_object, file_offset):
                 )
             except (ValueError, errors.ParseError) as exception:
                 raise errors.ParseError(
-                    f"Unable to parse entry item at offset: 0x{file_offset:08x} "
-                    f"with error: {exception!s}"
+                    f"Unable to parse entry item at offset: 0x{file_offset:08x} with "
+                    f"error: {exception!s}"
                 )
 
             file_offset += entry_item_data_size
@@ -321,9 +370,8 @@ def _ParseJournalEntry(self, file_object, file_offset):
                     f"{self._maximum_journal_file_offset:d})"
                 )
 
-            event_data = self._ParseDataObject(file_object, entry_item.object_offset)
-            event_string = event_data.decode("utf-8")
-            key, value = event_string.split("=", 1)
+            field_data = self._ParseDataObject(file_object, entry_item.object_offset)
+            key, value = self._ParseKeyValuePair(parser_mediator, field_data)
             fields[key] = value
 
         return fields
@@ -385,13 +433,15 @@ def ParseFileObject(self, parser_mediator, file_object):
                 continue
 
             try:
-                fields = self._ParseJournalEntry(file_object, entry_object_offset)
+                fields = self._ParseJournalEntry(
+                    parser_mediator, file_object, entry_object_offset
+                )
             except errors.ParseError as exception:
                 parser_mediator.ProduceExtractionWarning(
                     f"Unable to parse journal entry at offset: "
                     f"0x{entry_object_offset:08x} with error: {exception!s}"
                 )
-                return
+                continue
 
             date_time = dfdatetime_posix_time.PosixTimeInMicroseconds(
                 timestamp=fields["real_time"]

tests/parsers/systemd_journal.py

@@ -1,9 +1,11 @@
 #!/usr/bin/env python3
 """Tests for the Systemd Journal parser."""
 
+import io
 import unittest
 
 from plaso.containers import warnings
+from plaso.lib import errors
 from plaso.parsers import systemd_journal
 
 from tests.parsers import test_lib
@@ -12,13 +14,69 @@
 class SystemdJournalParserTest(test_lib.ParserTestCase):
     """Tests for the Systemd Journal parser."""
 
+    # pylint: disable=protected-access
+
+    def testParseDataObjectWithCorruptCompressedData(self):
+        """Tests _ParseDataObject raises ParseError on corrupt compressed data."""
+        parser = systemd_journal.SystemdJournalParser()
+
+        def _data_object(flags, payload):
+            # A 64-byte DATA object header (object_type=1, object_flags=flags,
+            # 6 reserved bytes, data_size, then 6 unused uint64 fields) followed
+            # by a corrupt compressed payload.
+            header = (
+                bytes([1, flags])
+                + b"\x00" * 6
+                + (64 + len(payload)).to_bytes(8, "little")
+                + b"\x00" * 48
+            )
+            return io.BytesIO(header + payload)
+
+        file_object = _data_object(parser._OBJECT_COMPRESSED_FLAG_XZ, b"corrupt-xz")
+        with self.assertRaises(errors.ParseError):
+            parser._ParseDataObject(file_object, 0)
+
+        file_object = _data_object(
+            parser._OBJECT_COMPRESSED_FLAG_LZ4,
+            b"\x10\x00\x00\x00\x00\x00\x00\x00corrupt-lz4",
+        )
+        with self.assertRaises(errors.ParseError):
+            parser._ParseDataObject(file_object, 0)
+
+    def testParseKeyValuePair(self):
+        """Tests the _ParseKeyValuePair function with text and binary values."""
+        parser = systemd_journal.SystemdJournalParser()
+        storage_writer = self._CreateStorageWriter()
+        parser_mediator = self._CreateParserMediator(storage_writer)
+
+        key, value = parser._ParseKeyValuePair(parser_mediator, b"MESSAGE=hello world")
+        self.assertEqual(key, "MESSAGE")
+        self.assertEqual(value, "hello world")
+
+        number_of_warnings = storage_writer.GetNumberOfAttributeContainers(
+            "extraction_warning"
+        )
+        self.assertEqual(number_of_warnings, 0)
+
+        # A key value pair containing invalid UTF-8 code points must decode with them
+        # escaped and produce an extraction warning.
+        key, value = parser._ParseKeyValuePair(
+            parser_mediator, b"MESSAGE=abc\xff\xfexyz"
+        )
+        self.assertEqual(key, "MESSAGE")
+        self.assertEqual(value, "abc\\xff\\xfexyz")
+
+        number_of_warnings = storage_writer.GetNumberOfAttributeContainers(
+            "extraction_warning"
+        )
+        self.assertEqual(number_of_warnings, 1)
+
     def testParse(self):
         """Tests the Parse function."""
         parser = systemd_journal.SystemdJournalParser()
         storage_writer = self._ParseFile(
             ["systemd", "journal", "system.journal"], parser
         )
-
         number_of_event_data = storage_writer.GetNumberOfAttributeContainers(
             "event_data"
         )
@@ -42,7 +100,6 @@ def testParse(self):
             "reporter": "systemd",
             "written_time": "2017-01-27T09:40:55.913258+00:00",
         }
-
         event_data = storage_writer.GetAttributeContainerByIndex("event_data", 0)
         self.CheckEventData(event_data, expected_event_values)
 
@@ -55,7 +112,6 @@ def testParse(self):
             "reporter": "root",
             "written_time": "2017-02-06T16:24:32.564585+00:00",
         }
-
         event_data = storage_writer.GetAttributeContainerByIndex("event_data", 2098)
         self.CheckEventData(event_data, expected_event_values)
 
@@ -65,7 +121,6 @@ def testParseLZ4(self):
         storage_writer = self._ParseFile(
             ["systemd", "journal", "system.journal.lz4"], parser
         )
-
         number_of_event_data = storage_writer.GetNumberOfAttributeContainers(
             "event_data"
         )
@@ -89,14 +144,13 @@ def testParseLZ4(self):
             "reporter": "systemd",
             "written_time": "2018-07-03T15:00:16.682340+00:00",
         }
-
         event_data = storage_writer.GetAttributeContainerByIndex("event_data", 0)
         self.CheckEventData(event_data, expected_event_values)
 
         # Test a LZ4 compressed data log entry.
-        # The text used in the test message was triplicated to make it long enough
-        # to trigger the LZ4 compression.
-        # Source: https://github.com/systemd/systemd/issues/6237
+        # The text used in the test message was trippled to make it long enough to
+        # trigger the LZ4 compression. Also see:
+        # https://github.com/systemd/systemd/issues/6237
         expected_body_parts = [" textual user names."]
         expected_body_parts.extend(
             (
@@ -117,7 +171,6 @@ def testParseLZ4(self):
             "reporter": "test",
             "written_time": "2018-07-03T15:19:04.667807+00:00",
         }
-
         event_data = storage_writer.GetAttributeContainerByIndex("event_data", 84)
         self.CheckEventData(event_data, expected_event_values)
 
@@ -127,7 +180,6 @@ def testParseCompactZSTD(self):
         storage_writer = self._ParseFile(
             ["systemd", "journal", "user-1000.journal"], parser
         )
-
         number_of_event_data = storage_writer.GetNumberOfAttributeContainers(
             "event_data"
         )
@@ -153,7 +205,6 @@ def testParseCompactZSTD(self):
             "reporter": "testapp",
             "written_time": "2023-09-26T07:42:46.445209+00:00",
         }
-
         event_data = storage_writer.GetAttributeContainerByIndex("event_data", 0)
         self.CheckEventData(event_data, expected_event_values)
 
@@ -169,7 +220,6 @@ def testParseDirty(self):
             ],
             parser,
         )
-
         number_of_event_data = storage_writer.GetNumberOfAttributeContainers(
             "event_data"
         )
@@ -178,7 +228,7 @@ def testParseDirty(self):
         number_of_warnings = storage_writer.GetNumberOfAttributeContainers(
             "extraction_warning"
         )
-        self.assertEqual(number_of_warnings, 1)
+        self.assertEqual(number_of_warnings, 46)
 
         number_of_warnings = storage_writer.GetNumberOfAttributeContainers(
             "recovery_warning"
@@ -195,14 +245,12 @@ def testParseDirty(self):
             "reporter": "systemd-journald",
             "written_time": "2016-10-24T13:20:01.063423+00:00",
         }
-
         event_data = storage_writer.GetAttributeContainerByIndex("event_data", 0)
         self.CheckEventData(event_data, expected_event_values)
 
         generator = storage_writer.GetAttributeContainers(
             warnings.ExtractionWarning.CONTAINER_TYPE
         )
-
         test_warnings = list(generator)
         test_warning = test_warnings[0]
         self.assertIsNotNone(test_warning)
@@ -213,6 +261,21 @@ def testParseDirty(self):
         )
         self.assertEqual(test_warning.message, expected_message)
 
+        # The 46 warnings are 1 corrupt entry pointer plus 45 unused/zeroed objects in
+        # the unclean tail.
+        unsupported_warnings = [
+            warning
+            for warning in test_warnings
+            if "Unsupported object type: 0" in warning.message
+        ]
+        self.assertEqual(len(unsupported_warnings), 45)
+
+        expected_message = (
+            "Unable to parse journal entry at offset: 0x004287a0 with error: "
+            "Unsupported object type: 0"
+        )
+        self.assertEqual(test_warnings[-1].message, expected_message)
+
 
 if __name__ == "__main__":
     unittest.main()