diff --git a/pcapkit/protocols/misc/pcap/frame.py b/pcapkit/protocols/misc/pcap/frame.py index 962ed341a0..9d976669cb 100644 --- a/pcapkit/protocols/misc/pcap/frame.py +++ b/pcapkit/protocols/misc/pcap/frame.py @@ -367,6 +367,11 @@ def make(self, incl_len = min(len(packet), self._ghdr.snaplen) if orig_len is None: orig_len = len(packet) + # NOTE: The record holds exactly ``incl_len`` octets of packet data, so a + # packet longer than that -- e.g. than the snaplen -- is cut to it rather + # than written out whole behind a clipped length. + if len(packet) > incl_len: + packet = bytes(packet)[:incl_len] return Schema_Frame( ts_sec=ts_sec, diff --git a/pcapkit/protocols/misc/pcapng.py b/pcapkit/protocols/misc/pcapng.py index d0a057e764..1733b068bf 100644 --- a/pcapkit/protocols/misc/pcapng.py +++ b/pcapkit/protocols/misc/pcapng.py @@ -38,6 +38,7 @@ from pcapkit.const.pcapng.verdict_type import VerdictType as Enum_VerdictType from pcapkit.const.reg.linktype import LinkType as Enum_LinkType from pcapkit.corekit.enum import EnumLookup +from pcapkit.corekit.fields.strings import DecodedString from pcapkit.corekit.module import ModuleDescriptor from pcapkit.corekit.multidict import OrderedMultiDict from pcapkit.corekit.version import VersionInfo @@ -245,6 +246,22 @@ def _option_key(code: 'Enum_OptionType') -> 'Union[Enum_OptionType, Tuple[str, i return code +def _utf8_length(text: 'str') -> 'int': + """Return the number of octets a UTF-8 :class:`~pcapkit.corekit.fields.strings.StringField` packs ``text`` as. + + Args: + text: String value; a :class:`~pcapkit.corekit.fields.strings.DecodedString` + packs as the octets it was decoded from. + + Returns: + Length of the encoded value, in octets. + + """ + if isinstance(text, DecodedString): + return len(text.raw) + return len(text.encode('utf-8')) + + class PacketDirection(EnumLookup, enum.IntEnum): """Packet direction for ``epb_flags`` options. @@ -1033,6 +1050,12 @@ def unpack(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_PCAPN if self._ctx is not None: self._byte = self._ctx.section.byteorder packet['byteorder'] = self._byte + # NOTE: A Simple Packet Block's data is ``min(original_len, + # snaplen)`` octets, with the snaplen of interface 0. + # A snaplen of zero means no limit. + interfaces = getattr(self._ctx, 'interfaces', None) + if interfaces and interfaces[0].snaplen: + packet.setdefault('snaplen', interfaces[0].snaplen) self.__header__ = cast('Schema_PCAPNG', self.__schema__.unpack(self._file, length, packet)) # type: ignore[call-arg,misc] data = self.read(length, **kwargs) @@ -2422,7 +2445,14 @@ def _read_option_custom(self, schema: 'Schema_CustomOption', *, Returns: Constructed option data. + Raises: + ProtocolError: If the option is shorter than its 4-octet private + enterprise number, which Option Length includes. + """ + if schema.length < 4: + raise ProtocolError(f'PCAP-NG: [opt_custom] invalid length (expected at least 4, got {schema.length})') + option = Data_CustomOption( type=schema.type, length=schema.length, @@ -3806,10 +3836,16 @@ def _make_block_epb(self, block: 'Optional[Data_EnhancedPacketBlock]' = None, *, if self._ctx is None: snaplen = 0xFFFF_FFFF_FFFF_FFFF else: - snaplen = self._get_interface(interface_id).snaplen + # a snaplen of zero means no limit + snaplen = self._get_interface(interface_id).snaplen or 0xFFFF_FFFF_FFFF_FFFF captured_len = min(len(packet_data), snaplen) if original_len is None: original_len = len(packet_data) + # NOTE: Packet Data holds exactly Captured Packet Length octets, so a + # packet longer than that -- e.g. than the interface's snaplen -- is cut + # to it rather than written out whole behind a clipped length. + if len(packet_data) > captured_len: + packet_data = bytes(packet_data)[:captured_len] packet_len = math.ceil(len(packet_data) / 4) * 4 if options is not None: @@ -3851,6 +3887,16 @@ def _make_block_spb(self, block: 'Optional[Data_SimplePacketBlock]' = None, *, if original_len is None: original_len = len(packet_data) + # NOTE: An SPB always belongs to interface 0, and its Packet Data is + # ``min(original_len, snaplen)`` octets long -- the length the parser + # reads back -- so a longer packet is cut to it. + if self._ctx is None or not self._ctx.interfaces: + snaplen = 0xFFFF_FFFF_FFFF_FFFF + else: + snaplen = self._ctx.interfaces[0].snaplen or 0xFFFF_FFFF_FFFF_FFFF + captured_len = min(original_len, snaplen) + if len(packet_data) > captured_len: + packet_data = bytes(packet_data)[:captured_len] packet_len = math.ceil(len(packet_data) / 4) * 4 return Schema_SimplePacketBlock( @@ -3980,6 +4026,9 @@ def _make_block_systemd(self, block: 'Optional[Data_SystemdJournalExportBlock]' temp.append(b''.join(tmp_buf)) data = b'\n'.join(temp) + # NOTE: The entry MUST be zero-padded to 32 bits, and the Block Total + # Length counts that padding, as the ``entry`` field reads it back. + data += bytes(-len(data) % 4) return Schema_SystemdJournalExportBlock( length=len(data) + 12, @@ -4038,7 +4087,10 @@ def _make_block_dsb(self, block: 'Optional[Data_DecryptionSecretsBlock]' = None, secrets_data_val = secrets_data else: raise ProtocolError(f'PCAP-NG: [DSB] secrets {secrets_type_val} invalid format') + # NOTE: Secrets Length excludes the padding to the next 32-bit boundary, + # which ``padding_data`` writes and the block length has to include. secrets_length = len(secrets_data_val) + secrets_padded = secrets_length + -secrets_length % 4 if options is not None: options_value, total_length = self._make_pcapng_options(options, namespace='dsb') @@ -4046,12 +4098,12 @@ def _make_block_dsb(self, block: 'Optional[Data_DecryptionSecretsBlock]' = None, options_value, total_length = [], 0 return Schema_DecryptionSecretsBlock( - length=total_length + secrets_length + 20, + length=total_length + secrets_padded + 20, secrets_type=secrets_type_val, secrets_length=secrets_length, secrets_data=secrets_data_val, options=options_value, - length2=total_length + secrets_length + 20, + length2=total_length + secrets_padded + 20, ) def _make_block_cb(self, block: 'Optional[Data_CustomBlock]' = None, *, @@ -4148,10 +4200,16 @@ def _make_block_packet(self, block: 'Optional[Data_PacketBlock]' = None, *, if self._ctx is None: snaplen = 0xFFFF_FFFF_FFFF_FFFF else: - snaplen = self._get_interface(interface_id).snaplen + # a snaplen of zero means no limit + snaplen = self._get_interface(interface_id).snaplen or 0xFFFF_FFFF_FFFF_FFFF captured_len = min(len(packet_data), snaplen) if original_len is None: original_len = len(packet_data) + # NOTE: Packet Data holds exactly Captured Packet Length octets, so a + # packet longer than that -- e.g. than the interface's snaplen -- is cut + # to it rather than written out whole behind a clipped length. + if len(packet_data) > captured_len: + packet_data = bytes(packet_data)[:captured_len] packet_len = math.ceil(len(packet_data) / 4) * 4 if options is not None: @@ -4328,7 +4386,7 @@ def _make_option_comment(self, type: 'Enum_OptionType', option: 'Optional[Data_C return Schema_CommentOption( type=type, - length=len(comment), + length=_utf8_length(comment), comment=comment, ) @@ -4387,7 +4445,7 @@ def _make_option_if_name(self, type: 'Enum_OptionType', option: 'Optional[Data_I return Schema_IF_NameOption( type=type, - length=len(name), + length=_utf8_length(name), name=name, ) @@ -4418,7 +4476,7 @@ def _make_option_if_description(self, type: 'Enum_OptionType', option: 'Optional return Schema_IF_DescriptionOption( type=type, - length=len(description), + length=_utf8_length(description), description=description, ) @@ -4731,7 +4789,7 @@ def _make_option_if_os(self, type: 'Enum_OptionType', option: 'Optional[Data_IF_ return Schema_IF_OSOption( type=type, - length=len(os), + length=_utf8_length(os), os=os, ) @@ -4824,7 +4882,7 @@ def _make_option_if_hardware(self, type: 'Enum_OptionType', option: 'Optional[Da return Schema_IF_HardwareOption( type=type, - length=len(hardware), + length=_utf8_length(hardware), hardware=hardware, ) @@ -5185,7 +5243,7 @@ def _make_option_ns_dnsname(self, type: 'Enum_OptionType', option: 'Optional[Dat return Schema_NS_DNSNameOption( type=type, - length=len(name), + length=_utf8_length(name), name=name, ) diff --git a/pcapkit/protocols/schema/misc/pcapng.py b/pcapkit/protocols/schema/misc/pcapng.py index 6c8a211fa6..0e117bc98d 100644 --- a/pcapkit/protocols/schema/misc/pcapng.py +++ b/pcapkit/protocols/schema/misc/pcapng.py @@ -1408,9 +1408,11 @@ class SimplePacketBlock(BlockType, code=Enum_BlockType.Simple_Packet_Block): length: 'int' = UInt32Field(callback=byteorder_callback) #: Original packet length. original_len: 'int' = UInt32Field(callback=byteorder_callback) - #: Packet data. - packet_data: 'bytes' = PayloadField(length=lambda pkt: min(pkt.get('snaplen', 0xFFFFFFFFFFFFFFFF), - pkt['original_len'])) + #: Packet data, ``min(original_len, snaplen)`` octets; a packet cut by the + #: snaplen is bounded by what the block holds, less its 16 framing octets. + packet_data: 'bytes' = PayloadField(length=lambda pkt: max(0, min(pkt.get('snaplen', 0xFFFFFFFFFFFFFFFF), + pkt['original_len'], + pkt['length'] - 16))) #: Padding. padding: 'bytes' = PaddingField(length=lambda pkt: (4 - len(pkt['packet_data']) % 4) % 4) #: Block total length. @@ -2105,8 +2107,10 @@ class ZigBeeNWKKey(DSBSecrets, code=Enum_SecretsType.ZigBee_NWK_Key): key: 'bytes' = BytesField(length=16) #: ZigBee PANID. panid: 'int' = UInt16Field(byteorder='little') - #: Padding. - padding: 'bytes' = BytesField(length=2) + + # NOTE: The two zero octets after the PAN ID are the Decryption Secrets + # Block's own padding, which Secrets Length does not count, so they belong + # to :attr:`DecryptionSecretsBlock.padding_data` rather than to this schema. if TYPE_CHECKING: def __init__(self, key: 'bytes', panid: 'int') -> 'None': ... @@ -2124,8 +2128,9 @@ class ZigBeeAPSKey(DSBSecrets, code=Enum_SecretsType.ZigBee_APS_Key): addr_low: 'int' = UInt16Field(byteorder='little') #: High node short address. addr_high: 'int' = UInt16Field(byteorder='little') - #: Padding. - padding: 'bytes' = BytesField(length=2) + + # NOTE: As for :class:`ZigBeeNWKKey`, the trailing two zero octets are the + # block's padding, not part of the secrets. if TYPE_CHECKING: def __init__(self, key: 'bytes', panid: 'int', addr_low: 'int', addr_high: 'int') -> 'None': ... diff --git a/tests/protocols/misc/test_pcapng_roundtrip2_unit.py b/tests/protocols/misc/test_pcapng_roundtrip2_unit.py new file mode 100644 index 0000000000..91740beb9d --- /dev/null +++ b/tests/protocols/misc/test_pcapng_roundtrip2_unit.py @@ -0,0 +1,153 @@ +# -*- coding: utf-8 -*- +"""PCAP-NG blocks keep their lengths and padding when built and rebuilt. + +* #1276 -- string option lengths counted characters, not octets. +* #1279 -- the Decryption Secrets Block never padded its secrets data. +* #1280 -- the systemd Journal Export Block was never padded. +* #1282 -- the ZigBee NWK/APS secrets counted the block padding in Secrets Length. +* #1283 -- a packet longer than the snaplen was written whole behind a clipped + captured length (PCAP-NG and PCAP). +* #1284 -- a custom option shorter than its 4-octet PEN read the PEN from the + padding. + +Every case builds its own octets in memory and reads no capture. + +:mod:`pcapkit` is imported inside each test, after +:func:`~tests._support.reimport_once_per_class`, not at module load, so that a +module that purged it earlier does not leave these tests holding stale classes. + +""" + +import unittest +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class + +if TYPE_CHECKING: + from typing import Any + +H = bytes.fromhex + + +class PCAPNGTestCase(unittest.TestCase): + """Helpers that parse and rebuild one PCAP-NG block.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _context(self, snaplen: 'int' = 0x40000) -> 'Any': + """A one-section little-endian context with one Ethernet interface.""" + from pcapkit.const.pcapng.block_type import BlockType + from pcapkit.const.reg.linktype import LinkType + from pcapkit.foundation.engines.pcapng import Context + from pcapkit.protocols.misc.pcapng import PCAPNG + + section = PCAPNG(num=0, sct=1, ctx=None, type=BlockType.Section_Header_Block, block={}) + context = Context(section.info) # type: Any + interface = PCAPNG(num=1, sct=1, ctx=context, type=BlockType.Interface_Description_Block, + block={'linktype': LinkType.ETHERNET, 'snaplen': snaplen}) + context.interfaces.append(interface.info) + return context + + def _parse(self, octets: 'bytes', context: 'Any' = None) -> 'Any': + """Parse ``octets`` as one block; a section header needs no context.""" + from pcapkit.protocols.misc.pcapng import PCAPNG + + if octets[:4] == b'\x0a\x0d\x0d\x0a': + return PCAPNG(octets, len(octets), num=0, sct=1, ctx=None) + return PCAPNG(octets, len(octets), num=2, sct=1, ctx=context or self._context()) + + def _rebuild(self, octets: 'bytes', context: 'Any' = None) -> 'bytes': + """Parse ``octets`` as one block and rebuild it from its data model.""" + from pcapkit.protocols.misc.pcapng import PCAPNG + + if octets[:4] == b'\x0a\x0d\x0d\x0a': + parsed = PCAPNG(octets, len(octets), num=0, sct=1, ctx=None) + return PCAPNG.from_data(parsed.info, num=0, sct=1, ctx=None).data + context = context or self._context() + parsed = PCAPNG(octets, len(octets), num=2, sct=1, ctx=context) + return PCAPNG.from_data(parsed.info, num=2, sct=1, ctx=context).data + + def assertRebuilds(self, octets: 'bytes', context: 'Any' = None) -> None: # pylint: disable=invalid-name + """Assert that ``octets`` rebuild byte-exactly from their data model.""" + self.assertEqual(self._rebuild(octets, context).hex(), octets.hex()) + + +class TestPCAPNGLengths(PCAPNGTestCase): + """Pin lengths and padding of rebuilt PCAP-NG blocks.""" + + def test_string_option_length_counts_octets(self) -> None: + """#1276: ``opt_comment`` of ``é`` is two octets long, not one.""" + octets = H('0a0d0d0a 24000000 4d3c2b1a 0100 0000 ffffffffffffffff' + '01000200 c3a90000 24000000') + self.assertEqual(self._parse(octets).info.options[1].comment, 'é') + self.assertRebuilds(octets) + + def test_dsb_pads_odd_secrets(self) -> None: + """#1279: three octets of secrets get one pad octet; Secrets Length stays 3.""" + self.assertRebuilds(H('0a000000 18000000 78563412 03000000 01020300 18000000')) + + def test_systemd_entry_is_padded(self) -> None: + """#1280: a 14-octet journal entry is padded to 16.""" + self.assertRebuilds(H('09000000 1c000000 4d4553534147453d68656c6c6f0a0000 1c000000')) + + def test_zigbee_secrets_length_excludes_padding(self) -> None: + """#1282: Secrets Length is 18 for an NWK key and 22 for an APS key.""" + key = bytes(range(16)) + nwk = H('0a000000 28000000 4b574e5a 12000000') + key + H('3412 0000 28000000') + aps = H('0a000000 2c000000 5350415a 16000000') + key + H('3412 7856 bc9a 0000 2c000000') + self.assertRebuilds(nwk) + self.assertRebuilds(aps) + + def test_packet_blocks_clip_to_snaplen(self) -> None: + """#1283: EPB, PB and SPB built over a snaplen of 8 hold 8 data octets.""" + import warnings + + from pcapkit.const.pcapng.block_type import BlockType + from pcapkit.protocols.misc.pcapng import PCAPNG + + context = self._context(snaplen=8) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') # the Packet Block is obsolete + for type_, size in ((BlockType.Enhanced_Packet_Block, 40), + (BlockType.Packet_Block, 40), + (BlockType.Simple_Packet_Block, 24)): + with self.subTest(type=type_): + block = PCAPNG(num=2, sct=1, ctx=context, type=type_, + block={'packet_data': b'A' * 16, 'timestamp': 0}) + self.assertEqual(len(block.data), size) + self.assertEqual(block.info.original_len, 16) + self.assertEqual(block.info.captured_len, 8) + self.assertRebuilds(block.data, context) + + def test_snaplen_zero_means_no_limit(self) -> None: + """#1283: a snaplen of zero clips nothing.""" + from pcapkit.const.pcapng.block_type import BlockType + from pcapkit.protocols.misc.pcapng import PCAPNG + + block = PCAPNG(num=2, sct=1, ctx=self._context(snaplen=0), type=BlockType.Enhanced_Packet_Block, + block={'packet_data': b'A' * 16, 'timestamp': 0}) + self.assertEqual(block.info.captured_len, 16) + + def test_pcap_frame_clips_to_snaplen(self) -> None: + """#1283: a PCAP record over a snaplen of 8 is 16 + 8 octets.""" + from pcapkit.protocols.misc.pcap.frame import Frame + from pcapkit.protocols.misc.pcap.header import Header + + frame = Frame(num=1, header=Header(snaplen=8).info, packet=b'A' * 16) + self.assertEqual(len(frame.data), 24) + self.assertEqual(frame.info.cap_len, 8) + self.assertEqual(frame.info.len, 16) + + def test_custom_option_shorter_than_pen_is_rejected(self) -> None: + """#1284: Option Length 3 cannot hold the 4-octet PEN.""" + from pcapkit.utilities.exceptions import ProtocolError + + octets = H('0a0d0d0a 24000000 4d3c2b1a 0100 0000 ffffffffffffffff' + 'ad0b0300 61626300 24000000') + with self.assertRaisesRegex(ProtocolError, r'opt_custom\] invalid length'): + self._parse(octets) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/misc/test_pcapng_unit.py b/tests/protocols/misc/test_pcapng_unit.py index 14ef160670..bd9112b9b8 100644 --- a/tests/protocols/misc/test_pcapng_unit.py +++ b/tests/protocols/misc/test_pcapng_unit.py @@ -2203,7 +2203,8 @@ def test_pcapng_remaining_constructor_branches_and_custom_dispatch(self) -> None self.assertEqual(pcapng._make_block_epb(packet_data=b'abcdef').captured_len, 6) self.assertEqual(pcapng._make_block_nrb().records, []) self.assertEqual(pcapng._make_block_systemd().entry, b'') - self.assertEqual(pcapng._make_block_systemd(entries=b'raw').entry, b'raw') + # the entry is zero-padded to 32 bits (#1280) + self.assertEqual(pcapng._make_block_systemd(entries=b'raw').entry, b'raw\x00') self.assertEqual(pcapng._make_block_packet(packet_data=b'abcdef').captured_length, 6) def assert_wrong_block(method: str, code: OptionType, correct: BlockType, **kwargs: object) -> None: diff --git a/tests/protocols/test_option_roundtrip_unit.py b/tests/protocols/test_option_roundtrip_unit.py index 845fd7372d..bba9af1e0c 100644 --- a/tests/protocols/test_option_roundtrip_unit.py +++ b/tests/protocols/test_option_roundtrip_unit.py @@ -401,25 +401,10 @@ class Gap(NamedTuple): # path set. And an EPB, SPB or PB rebuilt from its data model lost its # packet data. All of them round-trip now. - # The two key-log secrets used to fail for #1271, which is fixed: lines - # ended with ``os.sep``, and the body carried ``datetime.now()``. That - # unmasked #1279: ``_make_block_dsb`` never pads the secrets data, so the - # newline-terminated logs this suite builds give a Block Total Length that - # is not a multiple of four. The logs' header comment embeds - # ``pcapkit.__version__``, so the length varies by release and the - # fragment names only the block and the check. - # - # A release whose version string happens to make the log a multiple of four - # octets long turns these 'OK' without #1279 being fixed. If they start - # passing, check the log length before deleting them. - **{ - f'pcapng-secrets/{name}': Gap( - 'CONSTRUCT', ('PCAP-NG: [Block 10]', 'invalid length'), - 'pcapkit/protocols/misc/pcapng.py _make_block_dsb -- the block ' - 'length omits the pad that DecryptionSecretsBlock.padding_data in ' - 'pcapkit/protocols/schema/misc/pcapng.py reads (#1279)') - for name in ('TLS_Key_Log', 'WireGuard_Key_Log') - }, + # The two key-log secrets failed for #1271 (``os.sep`` line endings and a + # ``datetime.now()`` body), then for #1279: ``_make_block_dsb`` never padded + # the secrets data, so a log whose length is not a multiple of four gave a + # Block Total Length that is not either. Both are fixed. }