diff --git a/pcapkit/protocols/misc/pcapng.py b/pcapkit/protocols/misc/pcapng.py index ec0e24dfd9..15a7a8a021 100644 --- a/pcapkit/protocols/misc/pcapng.py +++ b/pcapkit/protocols/misc/pcapng.py @@ -665,6 +665,14 @@ class PCAPNG(ProtocolBase[Data_PCAPNG, Schema_PCAPNG], SHORT_PACKET_TYPES = (Enum_BlockType.Enhanced_Packet_Block, Enum_BlockType.Packet_Block) + #: Header blocks mapped to the octets their fixed fields span, so that a + #: shorter Block Total Length is kept as captured (see + #: :meth:`_read_block_short_header`). + SHORT_HEADER_TYPES = { + Enum_BlockType.Section_Header_Block: 28, + Enum_BlockType.Interface_Description_Block: 20, + } + #: Blocks that resolve an interface of their section, mapped to the tag used #: when reporting an interface ID that names no such interface. A Simple #: Packet Block carries no interface ID field and always refers to the @@ -1247,6 +1255,8 @@ def read(self, length: 'Optional[int]' = None, *, _read: 'bool' = True, name = self._lookup_registry(self.__block__, schema.type) if schema.type in self.SHORT_PACKET_TYPES and schema.block.length < 32: meth = cast('BlockParser', self._read_block_short) + elif schema.block.length < self.SHORT_HEADER_TYPES.get(schema.type, 0): + meth = cast('BlockParser', self._read_block_short_header) elif isinstance(name, str): meth_name = f'_read_block_{name}' meth = cast('BlockParser', @@ -2576,6 +2586,81 @@ def _read_block_short(self, schema: 'Schema_EnhancedPacketBlock | Schema_PacketB data.__update__([('block_raw', body)]) return self._decode_next_layer(data, self._get_linktype(interface_id), 0) # type: ignore[return-value] + def _read_block_short_header( + self, schema: 'Schema_SectionHeaderBlock | Schema_InterfaceDescriptionBlock | Schema_UnknownBlock', *, + header: 'Schema_PCAPNG', + ) -> 'Data_SectionHeaderBlock | Data_InterfaceDescriptionBlock': + """Read a PCAP-NG SHB or IDB shorter than its fixed fields. + + The fixed fields of a Section Header Block span 28 octets and those of + an Interface Description Block 20 (see :attr:`SHORT_HEADER_TYPES`), so a + smaller Block Total Length cannot hold them. + + Args: + schema: Parsed block schema, or on the construction path the + :class:`~pcapkit.protocols.schema.misc.pcapng.UnknownBlock` that + :meth:`_make_block_short` built. + header: Parsed PCAP-NG header schema. + + Returns: + Parsed packet data, with the fixed fields the block holds and zero + for those it does not, and no options. The block body is kept as + ``block_raw``, which the block's ``make`` writes back verbatim, so + the block rebuilds at the length it declared (:issue:`1422`). + + A section header keeps the byte order its Byte-Order Magic declares. + One of 12 octets has no room for the magic, and takes the byte order + in which its Block Total Length reads as 12 (see + :class:`~pcapkit.protocols.schema.misc.pcapng.SectionHeaderBlock`). + + """ + if isinstance(schema, Schema_UnknownBlock): + body = schema.body + else: + # the octets unpacked for the fixed fields, which start with the + # Block Total Length and run past the end of the block + octets = bytes(schema) + body = octets[4:schema.length - 8] + warn(f'PCAP-NG: [Block {header.type}] block length {schema.length} is below the ' + f'{self.SHORT_HEADER_TYPES[header.type]}-octet minimum; kept as captured', + ProtocolWarning, stacklevel=stacklevel()) + if octets[schema.length - 8:schema.length - 4] != octets[:4]: + warn(f'PCAP-NG: [Block {header.type}] block length mismatch: {schema.length} != ' + f'{octets[schema.length - 8:schema.length - 4].hex()}', ProtocolWarning, + stacklevel=stacklevel()) + + data: 'Data_SectionHeaderBlock | Data_InterfaceDescriptionBlock' + if header.type == Enum_BlockType.Section_Header_Block: + # NOTE: On the construction path :meth:`_make_block_shb` has already + # set the byte order from the data model. + if isinstance(schema, Schema_SectionHeaderBlock): + self._byte = schema.byteorder + endian = '<' if self._byte == 'little' else '>' + major, minor, section_length = struct.unpack(f'{endian}HHq', (body + bytes(16))[4:16]) + data = Data_SectionHeaderBlock( + type=header.type, + length=schema.length, + byteorder=self._byte, + version=VersionInfo( + major=major, + minor=minor, + ), + section_length=section_length, + options=self._read_pcapng_options([]), + ) + else: + endian = '<' if self._byte == 'little' else '>' + linktype, _, snaplen = struct.unpack(f'{endian}HHI', (body + bytes(8))[:8]) + data = Data_InterfaceDescriptionBlock( + type=header.type, + length=schema.length, + linktype=Enum_LinkType.get(linktype), + snaplen=snaplen, + options=self._read_pcapng_options([]), + ) + data.__update__([('block_raw', body)]) + return data + def _read_pcapng_options(self, options_schema: 'list[Schema_Option]') -> 'Option': """Read PCAP-NG options. @@ -4023,6 +4108,8 @@ def _make_block_shb(self, block: 'Optional[Data_SectionHeaderBlock]' = None, *, byteorder = block.byteorder self._byte = byteorder + if block is not None and hasattr(block, 'block_raw'): + return self._make_block_short(block) # type: ignore[return-value] if block is not None: major_version = block.version.major minor_version = block.version.minor @@ -4073,6 +4160,8 @@ def _make_block_idb(self, block: 'Optional[Data_InterfaceDescriptionBlock]' = No Constructed block schema. """ + if block is not None and hasattr(block, 'block_raw'): + return self._make_block_short(block) # type: ignore[return-value] if block is not None: linktype_val = block.linktype snaplen = block.snaplen @@ -4206,12 +4295,15 @@ def _make_block_spb(self, block: 'Optional[Data_SimplePacketBlock]' = None, *, length2=packet_len + 16, ) - def _make_block_short(self, block: 'Data_EnhancedPacketBlock | Data_PacketBlock') -> 'Schema_UnknownBlock': - """Make a PCAP-NG EPB or packet block (obsolete) kept as captured. + def _make_block_short( + self, block: 'Data_EnhancedPacketBlock | Data_PacketBlock | Data_SectionHeaderBlock | Data_InterfaceDescriptionBlock', # pylint: disable=line-too-long + ) -> 'Schema_UnknownBlock': + """Make a PCAP-NG EPB, packet block (obsolete), SHB or IDB kept as captured. Args: block: Block data model, whose ``block_raw`` holds the body of a - block shorter than its fixed fields (see :meth:`_read_block_short`). + block shorter than its fixed fields (see :meth:`_read_block_short` + and :meth:`_read_block_short_header`). Returns: Constructed block schema, the body written back verbatim. diff --git a/pcapkit/protocols/schema/misc/pcapng.py b/pcapkit/protocols/schema/misc/pcapng.py index e321e6f6a4..7de523d197 100644 --- a/pcapkit/protocols/schema/misc/pcapng.py +++ b/pcapkit/protocols/schema/misc/pcapng.py @@ -211,15 +211,21 @@ def shb_byteorder_callback(field: 'NumberField', packet: 'dict[str, Any]') -> 'N SHB's own options -- read by :func:`byteorder_callback`, after this field -- resolve their byte order from. + A Section Header Block of 12 octets has no room for the magic, and its + place holds the trailing Block Total Length instead. That length is 12 in + exactly one byte order, which is the one taken; :meth:`SectionHeaderBlock.post_process` + then rejects any block that matched this way with a length other than 12 + (:issue:`1422`). + Args: field: Field instance. packet: Packet data. """ magic = packet['match']['byteorder'] # type: int - if magic == 0x1A2B3C4D: + if magic in (0x1A2B3C4D, 0x0000000C): field._byteorder = 'big' - elif magic == 0x4D3C2B1A: + elif magic in (0x4D3C2B1A, 0x0C000000): field._byteorder = 'little' else: raise ProtocolError(f'unknown byteorder magic: {magic:#x}') @@ -1076,16 +1082,25 @@ def post_process(self, packet: 'dict[str, Any]') -> 'SectionHeaderBlock': Returns: Revised schema. + A Block Total Length below the 28 octets of the fixed fields reads the + trailing copy past the block, so the two are not compared here; + :meth:`PCAPNG._read_block_short_header + ` keeps + the block as captured and compares them itself (:issue:`1422`). A block + of 12 octets has no Byte-Order Magic and takes its byte order from its + Block Total Length (see :func:`shb_byteorder_callback`). + """ - self = cast('Self', super().post_process(packet)) + if self.length >= 28: + self = cast('Self', super().post_process(packet)) if self.section_length == 0xFFFF_FFFF_FFFF_FFFF: self.section_length = -1 magic = packet['match']['byteorder'] # type: int - if magic == 0x1A2B3C4D: + if magic == 0x1A2B3C4D or (magic == 0x0000000C and self.length == 12): self.byteorder = 'big' - elif magic == 0x4D3C2B1A: + elif magic == 0x4D3C2B1A or (magic == 0x0C000000 and self.length == 12): self.byteorder = 'little' else: raise ProtocolError(f'unknown byteorder magic: {magic:#x}') @@ -1365,6 +1380,26 @@ class InterfaceDescriptionBlock(BlockType, code=Enum_BlockType.Interface_Descrip #: Block total length. length2: 'int' = UInt32Field(callback=byteorder_callback) + def post_process(self, packet: 'dict[str, Any]') -> 'Schema': + """Revise ``schema`` data after unpacking process. + + Args: + packet: Unpacked data. + + Returns: + Revised schema. + + A Block Total Length below the 20 octets of the fixed fields reads the + trailing copy past the block, so the two are not compared here; + :meth:`PCAPNG._read_block_short_header + ` keeps + the block as captured and compares them itself (:issue:`1422`). + + """ + if self.length < 20: + return self + return super().post_process(packet) + if TYPE_CHECKING: def __init__(self, length: 'int', linktype: 'int', snaplen: 'int', options: 'list[Option | bytes] | bytes', length2: 'int') -> 'None': ... diff --git a/tests/protocols/misc/test_pcapng_short_shb_idb_1422_unit.py b/tests/protocols/misc/test_pcapng_short_shb_idb_1422_unit.py new file mode 100644 index 0000000000..966c273dc1 --- /dev/null +++ b/tests/protocols/misc/test_pcapng_short_shb_idb_1422_unit.py @@ -0,0 +1,187 @@ +# -*- coding: utf-8 -*- +"""GitHub issue #1422: a PCAP-NG SHB or IDB shorter than its fixed fields. + +A Section Header Block below 28 octets, or an Interface Description Block +below 20, read its fixed fields past the Block Total Length and rebuilt at the +full fixed size (SHB 16 or 20 to 28, IDB 12 or 16 to 20). The block body is +now kept as ``block_raw`` and written back at the declared length, as #1414 +does for EPB/PB. A 12-octet SHB has no Byte-Order Magic and takes the byte +order in which its Block Total Length reads as 12. + +:mod:`pcapkit` is imported inside each test, after +:func:`~tests._support.reimport_once_per_class`. + +""" + +import struct +import unittest +import warnings +from typing import TYPE_CHECKING + +from tests._support import reimport_once_per_class + +if TYPE_CHECKING: + from typing import Any + +#: Section and interface blocks below their fixed size, by issue length. +SHORT = ('shb-12', 'shb-16', 'shb-20', 'shb-24', 'idb-12', 'idb-16') +#: The same blocks at their full fixed size, already byte-exact before #1422. +FULL = ('shb-28', 'idb-20') + + +def _block(endian: 'str', type_: 'int', body: 'bytes') -> 'bytes': + """A block of ``type_`` around ``body`` in ``endian`` byte order.""" + length = len(body) + 12 + return struct.pack(f'{endian}II', type_, length) + body + struct.pack(f'{endian}I', length) + + +def _octets(name: 'str', endian: 'str') -> 'bytes': + """The block ``name`` (``shb-16``, ``idb-20``, ...) in ``endian`` byte order.""" + kind, size = name.split('-') + if kind == 'shb': + fixed = struct.pack(f'{endian}IHHq', 0x1A2B3C4D, 1, 0, -1) + return _block(endian, 0x0A0D0D0A, fixed[:int(size) - 12]) + fixed = struct.pack(f'{endian}HHI', 1, 0, 0x40000) + return _block(endian, 1, fixed[:int(size) - 12]) + + +class TestPCAPNGShortSHBIDB(unittest.TestCase): + """Pin that short SHBs and IDBs rebuild at the length they declare.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _kwargs(self, octets: 'bytes', endian: 'str') -> 'dict[str, Any]': + """Constructor arguments for ``octets``, an SHB or a block in a section.""" + from pcapkit.const.pcapng.block_type import BlockType + from pcapkit.foundation.engines.pcapng import Context + from pcapkit.protocols.misc.pcapng import PCAPNG + + if octets[:4] == b'\x0a\x0d\x0d\x0a': + return {'num': 0, 'sct': 1, 'ctx': None} + section = PCAPNG(num=0, sct=1, ctx=None, type=BlockType.Section_Header_Block, + block={'byteorder': 'little' if endian == '<' else 'big'}) + return {'num': 1, 'sct': 1, 'ctx': Context(section.info)} + + def _parse(self, octets: 'bytes', kwargs: 'dict[str, Any]') -> 'tuple[Any, list[str]]': + from pcapkit.protocols.misc.pcapng import PCAPNG + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + parsed = PCAPNG(octets, len(octets), **kwargs) + return parsed, [str(item.message) for item in caught] + + def _rebuild(self, info: 'Any', kwargs: 'dict[str, Any]') -> 'Any': + from pcapkit.protocols.misc.pcapng import PCAPNG + + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + return PCAPNG.from_data(info, **kwargs) + + def test_parse_rebuild_is_byte_exact(self) -> None: + """parse -> rebuild: every short and full block rebuilds at its own length.""" + for endian in '<>': + for name in SHORT + FULL: + with self.subTest(endian=endian, block=name): + octets = _octets(name, endian) + self.assertEqual(len(octets), int(name.split('-')[1])) + kwargs = self._kwargs(octets, endian) + parsed, messages = self._parse(octets, kwargs) + self.assertFalse(any('block length mismatch' in text for text in messages), messages) + self.assertEqual(self._rebuild(parsed.info, kwargs).data.hex(), octets.hex()) + + def test_short_block_is_kept_as_captured(self) -> None: + """A short block warns, keeps its body and reads no octet past its end.""" + for endian in '<>': + for name in SHORT: + with self.subTest(endian=endian, block=name): + octets = _octets(name, endian) + minimum = 28 if name.startswith('shb') else 20 + parsed, messages = self._parse(octets + b'\xff' * 16, self._kwargs(octets, endian)) + self.assertTrue(any(f'block length {len(octets)} is below the {minimum}-octet minimum' + in text for text in messages), messages) + info = parsed.info + self.assertEqual(info.length, len(octets)) + self.assertEqual(info.block_raw, octets[8:-4]) + self.assertEqual(len(info.options), 0) + + def test_full_block_has_no_block_raw(self) -> None: + """A full-size SHB or IDB parses its fields and keeps no raw body.""" + for endian in '<>': + for name in FULL: + with self.subTest(endian=endian, block=name): + octets = _octets(name, endian) + parsed, _ = self._parse(octets, self._kwargs(octets, endian)) + self.assertFalse(hasattr(parsed.info, 'block_raw')) + + def test_short_shb_keeps_byte_order_and_fields(self) -> None: + """The magic sets the byte order; fields past the block read as zero.""" + for endian, order in (('<', 'little'), ('>', 'big')): + for name, version, section_length in (('shb-12', (0, 0), 0), ('shb-16', (0, 0), 0), + ('shb-20', (1, 0), 0), + ('shb-24', (1, 0), 0xFFFF_FFFF if endian == '<' + else -0x1_0000_0000)): + with self.subTest(endian=endian, block=name): + octets = _octets(name, endian) + info = self._parse(octets, self._kwargs(octets, endian))[0].info + self.assertEqual(info.byteorder, order) + self.assertEqual((info.version.major, info.version.minor), version) + self.assertEqual(info.section_length, section_length) + + def test_short_idb_keeps_fields(self) -> None: + """A 16-octet IDB keeps its link type; the snaplen it lacks reads as zero.""" + from pcapkit.const.reg.linktype import LinkType + + for endian in '<>': + for name, linktype in (('idb-12', LinkType.NULL), ('idb-16', LinkType.ETHERNET)): + with self.subTest(endian=endian, block=name): + octets = _octets(name, endian) + info = self._parse(octets, self._kwargs(octets, endian))[0].info + self.assertEqual(info.linktype, linktype) + self.assertEqual(info.snaplen, 0) + + def test_make_parse_is_exact(self) -> None: + """make -> parse: the rebuilt block parses back to the same data model.""" + for endian in '<>': + for name in SHORT + FULL: + with self.subTest(endian=endian, block=name): + octets = _octets(name, endian) + kwargs = self._kwargs(octets, endian) + info = self._parse(octets, kwargs)[0].info + made = self._rebuild(info, kwargs) + reparsed = self._parse(made.data, kwargs)[0].info + for parsed in (made.info, reparsed): + self.assertEqual(parsed.length, info.length) + self.assertEqual(getattr(parsed, 'block_raw', None), getattr(info, 'block_raw', None)) + for field in ('byteorder', 'version', 'section_length', 'linktype', 'snaplen'): + self.assertEqual(getattr(parsed, field, None), getattr(info, field, None), field) + + def test_shb_without_magic_needs_a_length_of_12(self) -> None: + """A magic slot reading as length 12 is only a byte order when the block is 12 octets.""" + from pcapkit.utilities.exceptions import ProtocolError + + octets = struct.pack(' None: + """A stream opening with a 16-octet SHB and IDB still reads the EPB behind them.""" + from pcapkit.foundation.extraction import Extractor + from tests.protocols.misc.test_pcapng_unit import NamedBuffer + + good = _block('<', 6, struct.pack('