fix(pcapng)!: PCAP-NG round-trip follow-ups (#1382, #1383, #1384, #1401, #1405, #1406)

pcapkit/protocols/data/misc/pcapng.py

@@ -353,9 +353,14 @@ class EPB_FlagsOption(Option):
     reception: 'PacketReception'
     #: FCS length.
     fcs_len: 'int'
-    #: Bits 9 to 23 of the flags word, kept verbatim (checksum and
-    #: segmentation offload flags, reserved bits and unnamed
-    #: link-layer-dependent errors).
+    #: Checksum not ready (bit 9).
+    checksum_not_ready: 'bool'
+    #: Checksum valid (bit 10).
+    checksum_valid: 'bool'
+    #: TCP segmentation offloaded (bit 11).
+    tcp_segmentation_offloaded: 'bool'
+    #: Bits 12 to 23 of the flags word, kept verbatim (reserved bits and
+    #: unnamed link-layer-dependent errors).
     reserved: 'int'
     #: Link-layer-dependent error - CRC error (bit 24).
     crc_error: 'bool'
@@ -376,7 +381,8 @@ class EPB_FlagsOption(Option):
 
     if TYPE_CHECKING:
         def __init__(self, type: 'Enum_OptionType', length: 'int', direction: 'PacketDirection',
-                     reception: 'PacketReception', fcs_len: 'int', reserved: 'int', crc_error: 'bool',
+                     reception: 'PacketReception', fcs_len: 'int', checksum_not_ready: 'bool',
+                     checksum_valid: 'bool', tcp_segmentation_offloaded: 'bool', reserved: 'int', crc_error: 'bool',
                      too_long: 'bool', too_short: 'bool', gap_error: 'bool', unaligned_error: 'bool',
                      delimiter_error: 'bool', preamble_error: 'bool', symbol_error: 'bool') -> 'None': ...
 
@@ -865,9 +871,14 @@ class PACK_FlagsOption(Option):
     reception: 'PacketReception'
     #: FCS length.
     fcs_len: 'int'
-    #: Bits 9 to 23 of the flags word, kept verbatim (checksum and
-    #: segmentation offload flags, reserved bits and unnamed
-    #: link-layer-dependent errors).
+    #: Checksum not ready (bit 9).
+    checksum_not_ready: 'bool'
+    #: Checksum valid (bit 10).
+    checksum_valid: 'bool'
+    #: TCP segmentation offloaded (bit 11).
+    tcp_segmentation_offloaded: 'bool'
+    #: Bits 12 to 23 of the flags word, kept verbatim (reserved bits and
+    #: unnamed link-layer-dependent errors).
     reserved: 'int'
     #: Link-layer-dependent error - CRC error (bit 24).
     crc_error: 'bool'
@@ -888,7 +899,8 @@ class PACK_FlagsOption(Option):
 
     if TYPE_CHECKING:
         def __init__(self, type: 'Enum_OptionType', length: 'int', direction: 'PacketDirection',
-                     reception: 'PacketReception', fcs_len: 'int', reserved: 'int', crc_error: 'bool',
+                     reception: 'PacketReception', fcs_len: 'int', checksum_not_ready: 'bool',
+                     checksum_valid: 'bool', tcp_segmentation_offloaded: 'bool', reserved: 'int', crc_error: 'bool',
                      too_long: 'bool', too_short: 'bool', gap_error: 'bool', unaligned_error: 'bool',
                      delimiter_error: 'bool', preamble_error: 'bool', symbol_error: 'bool') -> 'None': ...
 

tests/protocols/misc/test_pcapng_captured_len_overrun_1405_unit.py

@@ -0,0 +1,94 @@
+# -*- coding: utf-8 -*-
+"""GitHub issue #1405: a captured length running past the block rebuilds byte-exactly.
+
+An Enhanced Packet Block whose ``captured_len`` runs past the block read that
+many octets as packet data: the trailing Block Total Length and whatever
+followed went with them, so a 36-octet block rebuilt as 40 or 41. The packet
+data is now bounded by the block, the padding and option area are sized from
+the octets read, and the declared ``captured_len`` is kept, so the block rebuilds
+as captured. The obsolete Packet Block has the same shape and the same fix.
+
+:mod:`pcapkit` is imported inside each test, after
+:func:`~tests._support.reimport_once_per_class`.
+
+"""
+
+import struct
+import unittest
+import warnings
+
+from tests.protocols.misc.test_pcapng_roundtrip2_unit import PCAPNGTestCase
+
+
+def _epb(captured_len: 'int', data: 'bytes', options: 'bytes' = b'') -> 'bytes':
+    """An Enhanced Packet Block on interface 0 declaring ``captured_len`` over ``data``."""
+    body = struct.pack('<IIIII', 0, 0, 0, captured_len, captured_len) + data + options
+    length = len(body) + 12
+    return struct.pack('<II', 6, length) + body + struct.pack('<I', length)
+
+
+def _pb(captured_len: 'int', data: 'bytes') -> 'bytes':
+    """An obsolete Packet Block on interface 0 declaring ``captured_len`` over ``data``."""
+    body = struct.pack('<HHIIII', 0, 0, 0, 0, captured_len, captured_len) + data
+    length = len(body) + 12
+    return struct.pack('<II', 2, length) + body + struct.pack('<I', length)
+
+
+class TestPCAPNGCapturedLenOverrun(PCAPNGTestCase):
+    """Pin how a captured length past the block parses and rebuilds."""
+
+    def _parse_noisy(self, octets: 'bytes') -> 'tuple[object, list[str]]':
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter('always')
+            parsed = self._parse(octets)
+        return parsed, [str(item.message) for item in caught]
+
+    def test_epb_overrun_rebuilds(self) -> None:
+        for captured_len in (5, 6, 8, 100, 0xFFFFFF):
+            with self.subTest(captured_len=captured_len):
+                octets = _epb(captured_len, b'abcd')
+                self.assertEqual(len(octets), 36)
+                parsed, messages = self._parse_noisy(octets)
+                self.assertEqual(parsed.info.captured_len, captured_len)  # type: ignore[attr-defined]
+                self.assertEqual(parsed.packet.payload, b'abcd')  # type: ignore[attr-defined]
+                self.assertTrue(any('runs past the block' in text for text in messages), messages)
+                self.assertFalse(any('block length mismatch' in text for text in messages), messages)
+                with warnings.catch_warnings():
+                    warnings.simplefilter('ignore')
+                    self.assertRebuilds(octets)
+
+    def test_overrun_reads_option_octets_as_data(self) -> None:
+        """The octets after the declared data are the block's, so they stay in it."""
+        octets = _epb(100, b'abcd', b'\x01\x00\x03\x00abc\x00')
+        with warnings.catch_warnings():
+            warnings.simplefilter('ignore')
+            parsed = self._parse(octets)
+            self.assertEqual(parsed.packet.payload, b'abcd\x01\x00\x03\x00abc\x00')
+            self.assertRebuilds(octets)
+
+    def test_pb_overrun_rebuilds(self) -> None:
+        for captured_len in (5, 100):
+            with self.subTest(captured_len=captured_len):
+                octets = _pb(captured_len, b'abcd')
+                with warnings.catch_warnings():
+                    warnings.simplefilter('ignore')  # the Packet Block is obsolete
+                    self.assertEqual(self._parse(octets).info.captured_len, captured_len)
+                    self.assertRebuilds(octets)
+
+    def test_well_formed_blocks_are_unchanged(self) -> None:
+        from pcapkit.const.pcapng.option_type import OptionType
+
+        for captured_len, data in ((4, b'abcd'), (5, b'abcde\x00\x00\x00'), (0, b'')):
+            with self.subTest(captured_len=captured_len):
+                octets = _epb(captured_len, data, b'\x01\x00\x03\x00abc\x00\x00\x00\x00\x00')
+                parsed, messages = self._parse_noisy(octets)
+                self.assertEqual(parsed.packet.payload, data[:captured_len])  # type: ignore[attr-defined]
+                self.assertEqual(parsed.info.options[OptionType.opt_comment].comment, 'abc')  # type: ignore[attr-defined]
+                self.assertFalse(any('runs past the block' in text for text in messages), messages)
+                with warnings.catch_warnings():
+                    warnings.simplefilter('ignore')
+                    self.assertRebuilds(octets)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/protocols/misc/test_pcapng_flags_undefined_1382_unit.py

@@ -0,0 +1,120 @@
+# -*- coding: utf-8 -*-
+"""GitHub issue #1382: undefined ``epb_flags`` sub-field values round-trip.
+
+A direction of ``0b11`` or a reception type of ``0b101`` to ``0b111`` raised
+:exc:`~pcapkit.utilities.exceptions.ProtocolError`, so a capture carrying one
+could be neither parsed nor rebuilt. Both now resolve to an ``Unassigned``
+member that keeps the wire value. Bits 9 to 11, which pcapng-06 Table 5 defines
+(checksum not ready, checksum valid, TCP segmentation offloaded), are named
+instead of sitting in ``reserved``.
+
+:mod:`pcapkit` is imported inside each test, after
+:func:`~tests._support.reimport_once_per_class`.
+
+"""
+
+import unittest
+import warnings
+
+from tests.protocols.misc.test_pcapng_roundtrip2_unit import H, PCAPNGTestCase, _epb_with_options
+
+
+def _pb_with_flags(word: 'bytes') -> 'bytes':
+    """An obsolete Packet Block on interface 0 carrying ``abcd`` and a ``pack_flags`` word."""
+    return H('02000000 30000000 0000 0000 00000000 00000000 04000000 04000000 61626364'
+             '0200 0400') + word + H('00000000 30000000')
+
+
+class TestPCAPNGFlagsUndefined(PCAPNGTestCase):
+    """Pin how undefined and newly named flags bits parse and rebuild."""
+
+    def _flags(self, word: 'str') -> 'tuple[bytes, object]':
+        from pcapkit.const.pcapng.option_type import OptionType
+
+        octets = _epb_with_options(H('0200 0400' + word))
+        return octets, self._parse(octets).info.options[OptionType.epb_flags]
+
+    def test_undefined_direction_parses_and_rebuilds(self) -> None:
+        octets, flags = self._flags('03000000')
+        self.assertEqual(int(flags.direction), 0b11)  # type: ignore[attr-defined]
+        self.assertEqual(flags.direction.name, 'Unassigned')  # type: ignore[attr-defined]
+        self.assertRebuilds(octets)
+
+    def test_undefined_reception_parses_and_rebuilds(self) -> None:
+        for value in (0b101, 0b110, 0b111):
+            with self.subTest(reception=value):
+                octets, flags = self._flags(f'{value << 2:02x}000000')
+                self.assertEqual(int(flags.reception), value)  # type: ignore[attr-defined]
+                self.assertEqual(flags.reception.name, 'Unassigned')  # type: ignore[attr-defined]
+                self.assertRebuilds(octets)
+
+    def test_every_word_rebuilds(self) -> None:
+        """Every direction and reception value, with every other bit set."""
+        for low in range(0x20):
+            with self.subTest(low=low):
+                octets, _ = self._flags(f'{0xffffffe0 | low:08x}')
+                self.assertRebuilds(octets)
+
+    def test_pack_flags_undefined_values_rebuild(self) -> None:
+        from pcapkit.const.pcapng.option_type import OptionType
+
+        octets = _pb_with_flags(H('1f000000'))
+        with warnings.catch_warnings():
+            warnings.simplefilter('ignore')  # the Packet Block is obsolete
+            flags = self._parse(octets).info.options[OptionType.pack_flags]
+            self.assertEqual((int(flags.direction), int(flags.reception)), (0b11, 0b111))
+            self.assertRebuilds(octets)
+
+    def test_make_accepts_undefined_values(self) -> None:
+        """``make`` -> parse -> pack keeps an undefined value."""
+        from pcapkit.const.pcapng.block_type import BlockType
+        from pcapkit.const.pcapng.option_type import OptionType
+        from pcapkit.protocols.misc.pcapng import PCAPNG
+
+        context = self._context()
+        block = PCAPNG(num=2, sct=1, ctx=context, type=BlockType.Enhanced_Packet_Block,
+                       block={'packet_data': b'abcd', 'timestamp': 0,
+                              'options': [(OptionType.epb_flags, {'direction': 0b11, 'reception': 0b110})]})
+        self.assertIn(H('0200 0400 1b000000'), block.data)
+        self.assertRebuilds(block.data, context)
+
+    def test_offload_bits_are_named(self) -> None:
+        _, flags = self._flags('000e0000')
+        self.assertEqual((flags.checksum_not_ready, flags.checksum_valid,  # type: ignore[attr-defined]
+                          flags.tcp_segmentation_offloaded, flags.reserved),  # type: ignore[attr-defined]
+                         (True, True, True, 0))
+        _, flags = self._flags('00f0ff00')
+        self.assertEqual((flags.checksum_not_ready, flags.checksum_valid,  # type: ignore[attr-defined]
+                          flags.tcp_segmentation_offloaded, flags.reserved),  # type: ignore[attr-defined]
+                         (False, False, False, 0xfff))
+
+    def test_make_sets_offload_bits(self) -> None:
+        from pcapkit.const.pcapng.block_type import BlockType
+        from pcapkit.const.pcapng.option_type import OptionType
+        from pcapkit.protocols.misc.pcapng import PCAPNG
+
+        for kwargs, word in (({'checksum_not_ready': True}, '00020000'),
+                             ({'checksum_valid': True}, '00040000'),
+                             ({'tcp_segmentation_offloaded': True}, '00080000'),
+                             ({'reserved': 1}, '00100000')):
+            with self.subTest(**kwargs):
+                block = PCAPNG(num=2, sct=1, ctx=self._context(), type=BlockType.Enhanced_Packet_Block,
+                               block={'packet_data': b'abcd', 'timestamp': 0,
+                                      'options': [(OptionType.epb_flags, kwargs)]})
+                self.assertIn(H('0200 0400' + word), block.data)
+
+    def test_out_of_range_value_still_raises(self) -> None:
+        from pcapkit.protocols.misc.pcapng import PacketDirection, PacketReception
+        from pcapkit.utilities.exceptions import EnumValueError
+
+        with self.assertRaises(EnumValueError):
+            PacketDirection(4)
+        with self.assertRaises(EnumValueError):
+            PacketReception(8)
+        # The unassigned member stays out of the lookup tables.
+        PacketDirection(3)
+        self.assertNotIn(3, PacketDirection._value2member_map_)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/protocols/misc/test_pcapng_journal_binary_overrun_1406_unit.py

@@ -0,0 +1,85 @@
+# -*- coding: utf-8 -*-
+"""GitHub issue #1406: a journal binary field past its entry keeps the entry as captured.
+
+A systemd Journal Export binary field declaring more octets than its entry
+holds was clamped. The rebuild then differed: the length prefix rewritten to the
+clamped size, and a terminating newline appended that was never there. Under the
+owner's ruling (option 1, strict, like #1325) the overrun raises, and the whole
+entry is kept as the octets captured: it parses as no entries, is exposed as
+``entry_raw``, and the block's ``make`` writes it back verbatim.
+
+:mod:`pcapkit` is imported inside each test, after
+:func:`~tests._support.reimport_once_per_class`.
+
+"""
+
+import struct
+import unittest
+import warnings
+
+from tests.protocols.misc.test_pcapng_roundtrip2_unit import PCAPNGTestCase
+
+
+def _journal(entry: 'bytes') -> 'bytes':
+    """A systemd Journal Export Block holding ``entry``, zero-padded to 32 bits."""
+    entry += bytes(-len(entry) % 4)
+    length = len(entry) + 12
+    return struct.pack('<II', 9, length) + entry + struct.pack('<I', length)
+
+
+class TestPCAPNGJournalBinaryOverrun(PCAPNGTestCase):
+    """Pin how an overrunning journal binary field parses and rebuilds."""
+
+    def test_overrun_keeps_the_entry_and_rebuilds(self) -> None:
+        from pcapkit.utilities.warnings import ProtocolWarning
+
+        for entry in (b'BIN\n' + struct.pack('<Q', 100) + b'abc',
+                      b'MESSAGE=hi\nBIN\n' + struct.pack('<Q', 100) + b'abc',
+                      b'A=1\n\nBIN\n' + struct.pack('<Q', 2 ** 64 - 1) + b'xy',
+                      b'BIN\n' + struct.pack('<Q', 1 << 63) + b'abc\n'):
+            with self.subTest(entry=entry):
+                octets = _journal(entry)
+                with warnings.catch_warnings(record=True) as caught:
+                    warnings.simplefilter('always')
+                    info = self._parse(octets).info
+                self.assertTrue(any(item.category is ProtocolWarning
+                                    and 'kept as captured' in str(item.message) for item in caught))
+                self.assertEqual(info.data, ())
+                self.assertEqual(info.entry_raw, octets[8:-4])
+                with warnings.catch_warnings():
+                    warnings.simplefilter('ignore')
+                    self.assertRebuilds(octets)
+
+    def test_well_formed_entries_still_parse(self) -> None:
+        for entry, field, value in ((b'BIN\n' + struct.pack('<Q', 3) + b'abc\n', 'BIN', b'abc'),
+                                    (b'MESSAGE=hello\n', 'MESSAGE', 'hello')):
+            with self.subTest(entry=entry):
+                octets = _journal(entry)
+                info = self._parse(octets).info
+                self.assertEqual(info.data[0][field], value)
+                self.assertFalse(hasattr(info, 'entry_raw'))
+                self.assertRebuilds(octets)
+
+    def test_extraction_continues_past_the_kept_block(self) -> None:
+        """The block after the malformed journal entry is still extracted."""
+        from pcapkit.foundation.extraction import Extractor
+        from pcapkit.protocols.misc.pcapng import PCAPNG
+        from tests.protocols.misc.test_pcapng_unit import NamedBuffer
+
+        shb = struct.pack('<IIIHHqI', 0x0A0D0D0A, 28, 0x1A2B3C4D, 1, 0, -1, 28)
+        idb = struct.pack('<IIHHII', 0x00000001, 20, 1, 0, 0, 20)
+        bad = _journal(b'BIN\n' + struct.pack('<Q', 100) + b'abc')
+        good = _journal(b'MESSAGE=hello\n')
+        with warnings.catch_warnings():
+            warnings.simplefilter('ignore')
+            extractor = Extractor(NamedBuffer(shb + idb + bad + good), nofile=True, store=True)
+            journals = extractor.engine._ctx_list[0].journals
+            self.assertEqual(len(journals), 2)
+            self.assertEqual(journals[0].data, ())
+            self.assertEqual(journals[1].data[0]['MESSAGE'], 'hello')
+            rebuilt = PCAPNG.from_data(journals[0], num=0, sct=1, ctx=None).data
+        self.assertEqual(rebuilt, bad)
+
+
+if __name__ == '__main__':
+    unittest.main()