diff --git a/pcapkit/corekit/fields/numbers.py b/pcapkit/corekit/fields/numbers.py index fa0add05df..fb18018a68 100644 --- a/pcapkit/corekit/fields/numbers.py +++ b/pcapkit/corekit/fields/numbers.py @@ -142,8 +142,19 @@ def pre_process(self, value: 'int', packet: 'dict[str, Any]') -> 'int | bytes': Returns: Processed field value. + Notes: + Masking against :attr:`self._bit_mask ` + truncates the value to the field's bit length, but it also turns a + negative value into its unsigned two's-complement pattern, which + neither :func:`struct.pack` nor :meth:`int.to_bytes` accepts for a + signed field. A signed field therefore maps the pattern back into + its signed range afterwards, so that e.g. a PCAP-NG section length + of ``-1`` (section length not specified) can be written out. + """ value = value & self._bit_mask + if self._signed and value > self._bit_mask >> 1: + value -= self._bit_mask + 1 if not self._need_process: return value diff --git a/pcapkit/foundation/engines/pcapng.py b/pcapkit/foundation/engines/pcapng.py index 67aba8d19a..8f2c12b81e 100644 --- a/pcapkit/foundation/engines/pcapng.py +++ b/pcapkit/foundation/engines/pcapng.py @@ -27,13 +27,11 @@ from pcapkit.protocols.data.misc.pcapng import CustomBlock as Data_CustomBlock from pcapkit.protocols.data.misc.pcapng import \ DecryptionSecretsBlock as Data_DecryptionSecretsBlock - from pcapkit.protocols.data.misc.pcapng import EnhancedPacketBlock as Data_EnhancedPacketBlock from pcapkit.protocols.data.misc.pcapng import \ InterfaceDescriptionBlock as Data_InterfaceDescriptionBlock from pcapkit.protocols.data.misc.pcapng import \ InterfaceStatisticsBlock as Data_InterfaceStatisticsBlock from pcapkit.protocols.data.misc.pcapng import NameResolutionBlock as Data_NameResolutionBlock - from pcapkit.protocols.data.misc.pcapng import PacketBlock as Data_PacketBlock from pcapkit.protocols.data.misc.pcapng import SectionHeaderBlock as Data_SectionHeaderBlock from pcapkit.protocols.data.misc.pcapng import \ SystemdJournalExportBlock as Data_SystemdJournalExportBlock @@ -214,9 +212,12 @@ def read_frame(self) -> 'P_PCAPNG': self._write_file(block.info, name=f'Decryption Secrets {len(self._ctx.secrets)}') elif block.info.type == Enum_BlockType.Interface_Statistics_Block: + # NOTE: The interface ID is bounds-checked while the block is + # parsed, by ``PCAPNG._get_interface``. Re-checking it here + # cannot fire: parsing an ISB resolves its interface to read the + # block's timestamp, so a block that reaches this branch has + # already been validated. isb_info = cast('Data_InterfaceStatisticsBlock', block.info) - if isb_info.interface_id >= len(self._ctx.interfaces): - raise FormatError(f'PCAP-NG: [ISB] invalid interface ID: {isb_info.interface_id}') self._ctx.statistics.append(isb_info) self._write_file(isb_info, name=f'Interface Statistics {len(self._ctx.statistics)}') @@ -227,9 +228,8 @@ def read_frame(self) -> 'P_PCAPNG': self._write_file(block.info, name=f'Custom {len(self._ctx.custom)}') elif block.info.type == Enum_BlockType.Enhanced_Packet_Block: - epb_info = cast('Data_EnhancedPacketBlock', block.info) - if epb_info.interface_id >= len(self._ctx.interfaces): - raise FormatError(f'PCAP-NG: [EPB] invalid interface ID: {epb_info.interface_id}') + # NOTE: as for the ISB above, the interface ID has already been + # bounds-checked by ``PCAPNG._get_interface`` while parsing. break elif block.info.type == Enum_BlockType.Simple_Packet_Block: @@ -242,10 +242,8 @@ def read_frame(self) -> 'P_PCAPNG': break elif block.info.type == Enum_BlockType.Packet_Block: - pack_info = cast('Data_PacketBlock', block.info) - if pack_info.interface_id >= len(self._ctx.interfaces): - raise FormatError(f'PCAP-NG: [Packet] invalid interface ID: {pack_info.interface_id}') - + # NOTE: as for the ISB above, the interface ID has already been + # bounds-checked by ``PCAPNG._get_interface`` while parsing. warn('PCAP-NG: [Packet] deprecated block type', DeprecatedFormatWarning, stacklevel=stacklevel()) break diff --git a/pcapkit/protocols/misc/pcapng.py b/pcapkit/protocols/misc/pcapng.py index efea7b9316..d5c4b8663d 100644 --- a/pcapkit/protocols/misc/pcapng.py +++ b/pcapkit/protocols/misc/pcapng.py @@ -175,7 +175,8 @@ from pcapkit.protocols.schema.misc.pcapng import ZigBeeNWKKey as Schema_ZigBeeNWKKey from pcapkit.protocols.schema.schema import Schema from pcapkit.utilities.compat import StrEnum, localcontext -from pcapkit.utilities.exceptions import ProtocolError, RegistryError, UnsupportedCall, stacklevel +from pcapkit.utilities.exceptions import (FormatError, ProtocolError, RegistryError, + UnsupportedCall, stacklevel) from pcapkit.utilities.warnings import (AttributeWarning, DeprecatedFormatWarning, ProtocolWarning, RegistryWarning, warn) @@ -532,6 +533,17 @@ class PCAPNG(Protocol[Data_PCAPNG, Schema_PCAPNG], Enum_BlockType.Simple_Packet_Block, Enum_BlockType.Packet_Block) + #: 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 + #: section's first interface, so it is listed here too. + INTERFACE_ID_BLOCK_TAGS = { + Enum_BlockType.Enhanced_Packet_Block: 'EPB', + Enum_BlockType.Simple_Packet_Block: 'SPB', + Enum_BlockType.Packet_Block: 'Packet', + Enum_BlockType.Interface_Statistics_Block: 'ISB', + } # type: dict[int, str] + ########################################################################## # Defaults. ########################################################################## @@ -1119,6 +1131,33 @@ def _get_local_timezone() -> 'timezone': return datetime.timezone.utc return cast('timezone', tzinfo) + def _get_interface(self, interface_id: 'int') -> 'Data_InterfaceDescriptionBlock': + """Interface description that ``interface_id`` names. + + Args: + interface_id: Interface ID that the current block associates with. + + Returns: + Interface Description Block (IDB) of the current section that + ``interface_id`` identifies. + + Raises: + FormatError: If the current section describes no such interface. + + Note: + The PCAP-NG specification requires a block's interface ID to name an + Interface Description Block of the block's own section, so an ID that + names none is a malformed file rather than a programming error -- + hence :exc:`~pcapkit.utilities.exceptions.FormatError` rather than + the bare :exc:`IndexError` that indexing the list would raise. + + """ + interfaces = self._ctx.interfaces + if not 0 <= interface_id < len(interfaces): + tag = self.INTERFACE_ID_BLOCK_TAGS.get(self._type, f'Block {self._type}') + raise FormatError(f'PCAP-NG: [{tag}] invalid interface ID: {interface_id}') + return interfaces[interface_id] + def _get_resolution(self, interface_id: 'int' = 0) -> 'int': """Timestamp resolution of the current block, in units per second. @@ -1135,7 +1174,7 @@ def _get_resolution(self, interface_id: 'int' = 0) -> 'int': AttributeWarning, stacklevel=stacklevel()) return 1_000_000 - options = self._ctx.interfaces[interface_id].options + options = self._get_interface(interface_id).options tsresol = cast('Optional[Data_IF_TSResolOption]', options.get(Enum_OptionType.if_tsresol)) if tsresol is None: @@ -1158,7 +1197,7 @@ def _get_offset(self, interface_id: 'int' = 0) -> 'int': AttributeWarning, stacklevel=stacklevel()) return 0 - options = self._ctx.interfaces[interface_id].options + options = self._get_interface(interface_id).options tsoffset = cast('Optional[Data_IF_TSOffsetOption]', options.get(Enum_OptionType.if_tsoffset)) if tsoffset is None: @@ -1181,7 +1220,7 @@ def _get_timezone(self, interface_id: 'int' = 0) -> 'timezone': AttributeWarning, stacklevel=stacklevel()) return self._get_local_timezone() - options = self._ctx.interfaces[interface_id].options + options = self._get_interface(interface_id).options tzone = cast('Optional[Data_IF_TZoneOption]', options.get(Enum_OptionType.if_tzone)) if tzone is None: @@ -1204,7 +1243,7 @@ def _get_linktype(self, interface_id: 'int' = 0) -> 'Enum_LinkType': """ if self._ctx is None or self._type not in self.PACKET_TYPES: raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute '_get_linktype'") - return self._ctx.interfaces[interface_id].linktype + return self._get_interface(interface_id).linktype def _read_timestamp(self, timestamp_high: 'int', timestamp_low: 'int', *, interface_id: 'int' = 0) -> 'tuple[dt_type, Decimal]': @@ -3338,7 +3377,7 @@ def _make_block_shb(self, block: 'Optional[Data_SectionHeaderBlock]' = None, *, minor_version = version[1] if options is not None: - options_value, total_length = self._make_pcapng_options(options, namespace='shb') + options_value, total_length = self._make_pcapng_options(options, namespace='opt') else: options_value, total_length = [], 0 @@ -3385,7 +3424,7 @@ def _make_block_idb(self, block: 'Optional[Data_InterfaceDescriptionBlock]' = No reversed=linktype_reversed, pack=False) if options is not None: - options_value, total_length = self._make_pcapng_options(options, namespace='shb') + options_value, total_length = self._make_pcapng_options(options, namespace='if') else: options_value, total_length = [], 0 @@ -3433,14 +3472,14 @@ def _make_block_epb(self, block: 'Optional[Data_EnhancedPacketBlock]' = None, *, if self._ctx is None: snaplen = 0xFFFF_FFFF_FFFF_FFFF else: - snaplen = self._ctx.interfaces[interface_id].snaplen + snaplen = self._get_interface(interface_id).snaplen captured_len = min(len(packet_data), snaplen) if original_len is None: original_len = len(packet_data) packet_len = math.ceil(len(packet_data) / 4) * 4 if options is not None: - options_value, total_length = self._make_pcapng_options(options, namespace='shb') + options_value, total_length = self._make_pcapng_options(options, namespace='epb') else: options_value, total_length = [], 0 @@ -3512,7 +3551,7 @@ def _make_block_nrb(self, block: 'Optional[Data_NameResolutionBlock]' = None, *, records_value, records_length = [], 0 if options is not None: - options_value, options_length = self._make_pcapng_options(options, namespace='shb') + options_value, options_length = self._make_pcapng_options(options, namespace='ns') else: options_value, options_length = [], 0 @@ -3549,7 +3588,7 @@ def _make_block_isb(self, block: 'Optional[Data_InterfaceStatisticsBlock]' = Non timestamp_high, timestamp_low = self._make_timestamp(timestamp, interface_id=interface_id) if options is not None: - options_value, total_length = self._make_pcapng_options(options, namespace='shb') + options_value, total_length = self._make_pcapng_options(options, namespace='isb') else: options_value, total_length = [], 0 @@ -3665,7 +3704,7 @@ def _make_block_dsb(self, block: 'Optional[Data_DecryptionSecretsBlock]' = None, secrets_length = len(secrets_data_val) if options is not None: - options_value, total_length = self._make_pcapng_options(options, namespace='shb') + options_value, total_length = self._make_pcapng_options(options, namespace='dsb') else: options_value, total_length = [], 0 @@ -3701,8 +3740,14 @@ def _make_block_cb(self, block: 'Optional[Data_CustomBlock]' = None, *, data = block.data options = cast('Option', getattr(block, 'options', None)) + # NOTE: Unlike every other block, a custom block has no ``OptionField`` + # to name an option registry: :class:`~pcapkit.protocols.schema.misc.pcapng.CustomBlock` + # spans the custom data, its padding and the options as one opaque blob, + # since only the owner of the private enterprise number knows where the + # custom data ends. Its options are therefore resolved in the generic + # ``opt_*`` space, which is the only one valid in every block. if options is not None: - options_value, _ = self._make_pcapng_options(options, namespace='shb') + options_value, _ = self._make_pcapng_options(options, namespace='opt') else: options_value, _ = [], 0 @@ -3765,14 +3810,14 @@ def _make_block_packet(self, block: 'Optional[Data_PacketBlock]' = None, *, if self._ctx is None: snaplen = 0xFFFF_FFFF_FFFF_FFFF else: - snaplen = self._ctx.interfaces[interface_id].snaplen + snaplen = self._get_interface(interface_id).snaplen captured_len = min(len(packet_data), snaplen) if original_len is None: original_len = len(packet_data) packet_len = math.ceil(len(packet_data) / 4) * 4 if options is not None: - options_value, total_length = self._make_pcapng_options(options, namespace='shb') + options_value, total_length = self._make_pcapng_options(options, namespace='pack') else: options_value, total_length = [], 0 diff --git a/pcapkit/protocols/schema/misc/pcapng.py b/pcapkit/protocols/schema/misc/pcapng.py index 2cdbffbd80..e3a0bedd64 100644 --- a/pcapkit/protocols/schema/misc/pcapng.py +++ b/pcapkit/protocols/schema/misc/pcapng.py @@ -149,6 +149,27 @@ class PACKFlags(TypedDict): symbol_error: int +def packet_byteorder(packet: 'dict[str, Any]') -> 'Literal["big", "little"]': + """Byte order declared for the section that ``packet`` belongs to. + + A nested schema is handed its parent's packet data under a ``__packet__`` + key (see :meth:`SchemaField.pack + `), so the section byte order + may live one level up. + + Args: + packet: Packet data. + + Returns: + Byte order of the enclosing section, falling back to the host byte + order when the packet data declares none. + + """ + if 'byteorder' not in packet and '__packet__' in packet: + return packet['__packet__'].get('byteorder', sys.byteorder) + return packet.get('byteorder', sys.byteorder) + + def byteorder_callback(field: 'NumberField', packet: 'dict[str, Any]') -> 'None': """Update byte order of PCAP-NG file. @@ -157,15 +178,20 @@ def byteorder_callback(field: 'NumberField', packet: 'dict[str, Any]') -> 'None' packet: Packet data. """ - if 'byteorder' not in packet and '__packet__' in packet: - field._byteorder = packet['__packet__'].get('byteorder', sys.byteorder) - else: - field._byteorder = packet.get('byteorder', sys.byteorder) + field._byteorder = packet_byteorder(packet) def shb_byteorder_callback(field: 'NumberField', packet: 'dict[str, Any]') -> 'None': """Update byte order of PCAP-NG file for SHB. + A Section Header Block declares the byte order of its own section through + its Byte-Order Magic, so it cannot take one from the enclosing packet data: + the first SHB of a file has no section context by construction, and a later + one would otherwise inherit the *previous* section's byte order. The magic + is therefore also written back as ``packet['byteorder']``, which is what the + SHB's own options -- read by :func:`byteorder_callback`, after this field -- + resolve their byte order from. + Args: field: Field instance. packet: Packet data. @@ -178,6 +204,7 @@ def shb_byteorder_callback(field: 'NumberField', packet: 'dict[str, Any]') -> 'N field._byteorder = 'little' else: raise ProtocolError(f'unknown byteorder magic: {magic:#x}') + packet['byteorder'] = field._byteorder def pcapng_block_selector(packet: 'dict[str, Any]') -> 'Field': @@ -532,8 +559,8 @@ class SectionHeaderBlock(BlockType, code=Enum_BlockType.Section_Header_Block): registry=Option.registry['opt'], eool=Enum_OptionType.opt_endofopt, ) - #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: pkt['__option_padding__']) + #: Padding, sized from the ``__option_padding__`` key that OptionField generates. + padding: 'bytes' = PaddingField(length=lambda pkt: pkt.get('__option_padding__', 0)) #: Block total length. length2: 'int' = UInt32Field(callback=shb_byteorder_callback) @@ -547,12 +574,17 @@ def pre_pack(self, packet: 'dict[str, Any]') -> 'None': This method is expected to directly modify any data stored in the ``packet`` and thus no return is required. + The Byte-Order Magic is not carried by any field of the schema -- + :attr:`magic` is the palindromic constant, identical in either byte + order -- so it is seeded here from the byte order the packet data + asks for, and from the host byte order when it asks for none. + """ if 'match' in packet: return packet['match'] = { - 'byteorder': 0x1A2B3C4D if sys.byteorder == 'big' else 0x4D3C2B1A, + 'byteorder': 0x1A2B3C4D if packet_byteorder(packet) == 'big' else 0x4D3C2B1A, } def post_process(self, packet: 'dict[str, Any]') -> 'SectionHeaderBlock': @@ -850,8 +882,8 @@ class InterfaceDescriptionBlock(BlockType, code=Enum_BlockType.Interface_Descrip registry=Option.registry['if'], eool=Enum_OptionType.opt_endofopt, ) - #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: pkt['__option_padding__']) + #: Padding, sized from the ``__option_padding__`` key that OptionField generates. + padding: 'bytes' = PaddingField(length=lambda pkt: pkt.get('__option_padding__', 0)) #: Block total length. length2: 'int' = UInt32Field(callback=byteorder_callback) @@ -987,14 +1019,18 @@ class EnhancedPacketBlock(BlockType, code=Enum_BlockType.Enhanced_Packet_Block): padding_data: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['captured_len'] % 4) % 4) #: Options. options: 'list[Option]' = OptionField( - length=lambda pkt: pkt['length'] - 32 - pkt['captured_len'] - len(pkt['padding_data']), + # NOTE: The padding is recomputed here rather than read back from + # ``padding_data``: a PaddingField is written straight into the schema + # buffer while packing and never lands in the packet data, so its name + # is not a key here on the packing path. + length=lambda pkt: pkt['length'] - 32 - pkt['captured_len'] - (4 - pkt['captured_len'] % 4) % 4, base_schema=_EPB_Option, type_name='type', registry=Option.registry['epb'], eool=Enum_OptionType.opt_endofopt, ) - #: Padding. - padding_opts: 'bytes' = PaddingField(length=lambda pkt: pkt['__option_padding__']) + #: Padding, sized from the ``__option_padding__`` key that OptionField generates. + padding_opts: 'bytes' = PaddingField(length=lambda pkt: pkt.get('__option_padding__', 0)) #: Block total length. length2: 'int' = UInt32Field(callback=byteorder_callback) @@ -1183,14 +1219,14 @@ class NameResolutionBlock(BlockType, code=Enum_BlockType.Name_Resolution_Block): ) #: Options. options: 'list[Option]' = OptionField( - length=lambda pkt: pkt['__option_padding__'], + length=lambda pkt: pkt.get('__option_padding__', 0), # key from OptionField base_schema=_NS_Option, type_name='type', registry=Option.registry['ns'], eool=Enum_OptionType.opt_endofopt, ) - #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: pkt['__option_padding__']) + #: Padding, sized from the ``__option_padding__`` key that OptionField generates. + padding: 'bytes' = PaddingField(length=lambda pkt: pkt.get('__option_padding__', 0)) #: Block total length. length2: 'int' = UInt32Field(callback=byteorder_callback) @@ -1340,8 +1376,8 @@ class InterfaceStatisticsBlock(BlockType, code=Enum_BlockType.Interface_Statisti registry=Option.registry['isb'], eool=Enum_OptionType.opt_endofopt, ) - #: Padding. - padding: 'bytes' = PaddingField(length=lambda pkt: pkt['__option_padding__']) + #: Padding, sized from the ``__option_padding__`` key that OptionField generates. + padding: 'bytes' = PaddingField(length=lambda pkt: pkt.get('__option_padding__', 0)) #: Block total length. length2: 'int' = UInt32Field(callback=byteorder_callback) @@ -1558,17 +1594,19 @@ class DecryptionSecretsBlock(BlockType, code=Enum_BlockType.Decryption_Secrets_B selector=dsb_secrets_selector, ) #: Padding. - padding_data: 'bytes' = BytesField(length=lambda pkt: (4 - pkt['secrets_length'] % 4) % 4) + padding_data: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['secrets_length'] % 4) % 4) #: Options. options: 'list[Option]' = OptionField( - length=lambda pkt: pkt['length'] - 20 - pkt['secrets_length'] - len(pkt['padding_data']), + # NOTE: see EnhancedPacketBlock.options on why the padding is recomputed + # here instead of being read back from ``padding_data``. + length=lambda pkt: pkt['length'] - 20 - pkt['secrets_length'] - (4 - pkt['secrets_length'] % 4) % 4, base_schema=_DSB_Option, type_name='type', registry=Option.registry['dsb'], eool=Enum_OptionType.opt_endofopt, ) - #: Padding. - padding_opts: 'bytes' = PaddingField(length=lambda pkt: pkt['__option_padding__']) + #: Padding, sized from the ``__option_padding__`` key that OptionField generates. + padding_opts: 'bytes' = PaddingField(length=lambda pkt: pkt.get('__option_padding__', 0)) #: Block total length. length2: 'int' = UInt32Field(callback=byteorder_callback) @@ -1677,17 +1715,19 @@ class PacketBlock(BlockType, code=Enum_BlockType.Packet_Block): #: Packet data. packet_data: 'bytes' = PayloadField(length=lambda pkt: pkt['captured_length']) #: Padding. - padding_data: 'bytes' = BytesField(length=lambda pkt: (4 - pkt['captured_length'] % 4) % 4) + padding_data: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['captured_length'] % 4) % 4) #: Options. options: 'list[Option]' = OptionField( - length=lambda pkt: pkt['length'] - 32 - pkt['captured_length'] - len(pkt['padding_data']), + # NOTE: see EnhancedPacketBlock.options on why the padding is recomputed + # here instead of being read back from ``padding_data``. + length=lambda pkt: pkt['length'] - 32 - pkt['captured_length'] - (4 - pkt['captured_length'] % 4) % 4, base_schema=_PACK_Option, type_name='type', registry=Option.registry['pack'], eool=Enum_OptionType.opt_endofopt, ) - #: Padding. - padding_opts: 'bytes' = PaddingField(length=lambda pkt: pkt['__option_padding__']) + #: Padding, sized from the ``__option_padding__`` key that OptionField generates. + padding_opts: 'bytes' = PaddingField(length=lambda pkt: pkt.get('__option_padding__', 0)) #: Block total length. length2: 'int' = UInt32Field(callback=byteorder_callback) diff --git a/pcapkit/protocols/schema/schema.py b/pcapkit/protocols/schema/schema.py index 2d897d64cd..4effad092b 100644 --- a/pcapkit/protocols/schema/schema.py +++ b/pcapkit/protocols/schema/schema.py @@ -486,7 +486,15 @@ class will consider negative value as a placeholder. packet = {} packet.update(self.__dict__) + # NOTE: ``pre_unpack`` is called here as well as from the unpacking path + # since a schema may seed ``packet`` keys that both paths need, as + # ``pcapkit.protocols.schema.internet.hip.EncryptedParameter`` does. + # ``pre_pack`` is what carries the packing-only preparation, such as the + # PCAP-NG section header block's byte order magic, which no field of the + # schema holds and which therefore cannot be recovered from + # ``self.__dict__`` above. self.pre_unpack(packet) + self.pre_pack(packet) if '__length__' not in packet: packet['__length__'] = -1 # reasonable default value diff --git a/tests/foundation/engines/test_pcapng_engine.py b/tests/foundation/engines/test_pcapng_engine.py index 6301417ebc..e9c4c6e0d8 100644 --- a/tests/foundation/engines/test_pcapng_engine.py +++ b/tests/foundation/engines/test_pcapng_engine.py @@ -94,6 +94,11 @@ def packet(self, interface_id: int = 0) -> PCAPNGWriter: len(ETHERNET_FRAME), len(ETHERNET_FRAME)) + self._pad(ETHERNET_FRAME)) + def interface_statistics(self, interface_id: int = 0) -> PCAPNGWriter: + """Interface Statistics Block, with no options.""" + return self._block(0x00000005, + struct.pack(f'{self._endian}III', interface_id, 0, 0)) + @unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') class PCAPNGEngineTests(unittest.TestCase): @@ -265,23 +270,24 @@ def test_read_frame_rejects_invalid_interface_contexts(self) -> None: from pcapkit.foundation.engines.pcapng import Context, PCAPNG from pcapkit.utilities.exceptions import FormatError - cases = [ - FakeBlock(self._info(BlockType.Interface_Statistics_Block, interface_id=0)), - FakeBlock(self._info(BlockType.Enhanced_Packet_Block, interface_id=0)), - FakeBlock(self._info(BlockType.Simple_Packet_Block)), - FakeBlock(self._info(BlockType.Packet_Block, interface_id=0)), - ] - - for block in cases: - with self.subTest(block=block.info.type): - extractor, _ = make_extractor(_flag_q=True, _flag_r=False, - _flag_t=False, _flag_d=False) - engine = PCAPNG(extractor) - engine._ctx = Context(self._section()) - engine._ctx_list = [engine._ctx] - with mock.patch('pcapkit.foundation.engines.pcapng.P_PCAPNG', side_effect=[block]): - with self.assertRaises(FormatError): - engine.read_frame() + # A Simple Packet Block carries no interface ID field, so a section that + # describes no interface is the only thing there is to reject about it, + # and ``read_frame`` is where that happens. The interface *ID* of an EPB, + # ISB or obsolete Packet Block is instead bounds-checked while the block + # is parsed -- see + # ``PCAPNGSectionShapeTests.test_out_of_range_interface_id_is_a_format_error`` + # for the equivalent end-to-end coverage of those three. + block = FakeBlock(self._info(BlockType.Simple_Packet_Block)) + + extractor, _ = make_extractor(_flag_q=True, _flag_r=False, + _flag_t=False, _flag_d=False) + engine = PCAPNG(extractor) + engine._ctx = Context(self._section()) + engine._ctx_list = [engine._ctx] + with mock.patch('pcapkit.foundation.engines.pcapng.P_PCAPNG', side_effect=[block]): + with self.assertRaises(FormatError) as context: + engine.read_frame() + self.assertIn('PCAP-NG: [SPB]', str(context.exception)) def test_check_packet_block_context_only_fires_for_packet_blocks(self) -> None: import io @@ -407,6 +413,60 @@ def test_packet_blocks_without_an_interface_description_are_format_errors(self) self.assertIn(f'PCAP-NG: [{tag}]', message) self.assertIn('interface description block', message) + def test_out_of_range_interface_id_is_a_format_error(self) -> None: + # Regression for #367: a section with at least one IDB and a block naming + # an interface it does not describe used to escape as a bare + # ``IndexError`` out of ``PCAPNG._get_timezone``, because the engine's + # bounds guards ran only after the block had already been parsed. + from pcapkit.utilities.exceptions import FormatError + + blocks = { + 'EPB': lambda writer, iface: writer.enhanced_packet(interface_id=iface), + 'Packet': lambda writer, iface: writer.packet(interface_id=iface), + 'ISB': lambda writer, iface: writer.interface_statistics(interface_id=iface), + } + + for byteorder in ('little', 'big'): + for tag, add_block in blocks.items(): + with self.subTest(byteorder=byteorder, tag=tag): + # two interfaces described, interface 7 named + capture = add_block(PCAPNGWriter(byteorder).section_header() + .interface_description() + .interface_description(), 7) + with self.assertRaises(FormatError) as context: + self._extract(capture) + + message = str(context.exception) + self.assertIn(f'PCAP-NG: [{tag}]', message) + self.assertIn('invalid interface ID: 7', message) + + def test_interface_statistics_without_an_interface_description_is_a_format_error(self) -> None: + # An ISB is not a packet block, so the pre-parse check for a section with + # no IDB does not cover it; the bounds check in the parse path does. + from pcapkit.utilities.exceptions import FormatError + + capture = PCAPNGWriter().section_header().interface_statistics(interface_id=0) + with self.assertRaises(FormatError) as context: + self._extract(capture) + + message = str(context.exception) + self.assertIn('PCAP-NG: [ISB]', message) + self.assertIn('invalid interface ID: 0', message) + + def test_in_range_interface_id_still_parses(self) -> None: + # The guard is a bound, not a ban: the highest valid ID must still work. + capture = (PCAPNGWriter().section_header() + .interface_description() + .interface_description() + .interface_statistics(interface_id=1) + .enhanced_packet(interface_id=1)) + extractor = self._extract(capture) + + self.assertEqual(len(extractor.frame), 1) + self.assertEqual(extractor.frame[0].info.interface_id, 1) + self.assertEqual(len(extractor.engine._ctx.statistics), 1) + self.assertEqual(extractor.engine._ctx.statistics[0].interface_id, 1) + if __name__ == '__main__': unittest.main() diff --git a/tests/protocols/misc/test_pcapng_unit.py b/tests/protocols/misc/test_pcapng_unit.py index 6ba4eddddc..b5c1c88e95 100644 --- a/tests/protocols/misc/test_pcapng_unit.py +++ b/tests/protocols/misc/test_pcapng_unit.py @@ -7,7 +7,9 @@ import decimal import io from ipaddress import ip_address, ip_interface +import os import struct +import sys import types import unittest from unittest import mock @@ -1510,7 +1512,7 @@ def reset(block_type: BlockType, opt: collections.Counter | None = None) -> None EndOfOption(type=OptionType.opt_endofopt, length=0), CommentOption(type=OptionType.opt_comment, length=5, comment='again'), b'\x88\x13\x03\x00raw\x00', - ], namespace='shb') + ], namespace='opt') self.assertEqual(opts[-1].type, OptionType.opt_endofopt) self.assertGreater(opt_len, 0) @@ -1522,7 +1524,7 @@ def reset(block_type: BlockType, opt: collections.Counter | None = None) -> None (OptionType.opt_endofopt, object()), ]) pcapng._opt = collections.Counter() - opts_from_data, data_opt_len = pcapng._make_pcapng_options(data_options, namespace='shb') + opts_from_data, data_opt_len = pcapng._make_pcapng_options(data_options, namespace='opt') self.assertEqual(opts_from_data[-1].type, OptionType.opt_endofopt) self.assertGreater(data_opt_len, 0) @@ -2082,28 +2084,28 @@ def custom_secrets_constructor(code, secrets=None, *, data=b's', **kwargs): b'\x00\x00\x00\x00', (option_code, {'data': b'list'}), (OptionType.opt_endofopt, {}), - ], namespace='shb') + ], namespace='opt') self.assertEqual(list_options[-1].type, OptionType.opt_endofopt) self.assertEqual(list_options[0].data, b'list') pcapng._opt = collections.Counter() no_end_options, _ = pcapng._make_pcapng_options([ (OptionType.opt_comment, {'comment': 'no-end'}), - ], namespace='shb') + ], namespace='opt') self.assertNotEqual(no_end_options[-1].type, OptionType.opt_endofopt) pcapng._opt = collections.Counter() dict_options, _ = pcapng._make_pcapng_options(OrderedMultiDict([ (option_code, DummyData(data=b'dict')), (OptionType.opt_endofopt, object()), - ]), namespace='shb') + ]), namespace='opt') self.assertEqual(dict_options[0].data, b'dict') self.assertEqual(dict_options[-1].type, OptionType.opt_endofopt) pcapng._opt = collections.Counter() dict_options_no_end, _ = pcapng._make_pcapng_options(OrderedMultiDict([ (OptionType.opt_comment, DummyData(comment='dict-no-end')), - ]), namespace='shb') + ]), namespace='opt') self.assertNotEqual(dict_options_no_end[-1].type, OptionType.opt_endofopt) records = pcapng._read_nrb_records([ @@ -2406,6 +2408,531 @@ def test_pcapng_read_block_packet_resolves_linktype_without_info(self) -> None: self.assertEqual(block.drop_count, 1) self.assertEqual(decoded, [(LinkType.ETHERNET, 4)]) + ########################################################################## + # Write path: packing block schemas (#366). + ########################################################################## + + @staticmethod + def _all_block_schemas(): + """Every PCAP-NG block schema, populated well enough to be packed. + + Returns a list of ``(label, block type, schema)`` triples covering all + eleven block schemas of + :mod:`pcapkit.protocols.schema.misc.pcapng`, each with the block total + length its own fields imply. + + """ + from pcapkit.const.pcapng.block_type import BlockType + from pcapkit.const.pcapng.secrets_type import SecretsType + from pcapkit.const.reg.linktype import LinkType + from pcapkit.protocols.schema.misc.pcapng import (CustomBlock, DecryptionSecretsBlock, + EnhancedPacketBlock, + InterfaceDescriptionBlock, + InterfaceStatisticsBlock, + NameResolutionBlock, PacketBlock, + SectionHeaderBlock, SimplePacketBlock, + SystemdJournalExportBlock, UnknownBlock) + + payload = b'\xde\xad\xbe\xef' + return [ + ('UnknownBlock', BlockType.get(0xFFFF_0000), + UnknownBlock(length=12, body=b'', length2=12)), + ('SectionHeaderBlock', BlockType.Section_Header_Block, + SectionHeaderBlock(length=28, magic=0x1A2B3C4D, major=1, minor=0, + section_length=-1, options=[], length2=28)), + ('InterfaceDescriptionBlock', BlockType.Interface_Description_Block, + InterfaceDescriptionBlock(length=20, linktype=LinkType.ETHERNET, snaplen=0, + options=[], length2=20)), + ('EnhancedPacketBlock', BlockType.Enhanced_Packet_Block, + EnhancedPacketBlock(length=36, interface_id=0, timestamp_high=0, timestamp_low=0, + captured_len=len(payload), original_len=len(payload), + packet_data=payload, options=[], length2=36)), + # captured_len=5 needs three genuine padding octets, unlike the + # 32-bit aligned case above + ('EnhancedPacketBlock, captured_len=5', BlockType.Enhanced_Packet_Block, + EnhancedPacketBlock(length=40, interface_id=0, timestamp_high=0, timestamp_low=0, + captured_len=5, original_len=5, packet_data=payload + b'\x00', + options=[], length2=40)), + ('SimplePacketBlock', BlockType.Simple_Packet_Block, + SimplePacketBlock(length=20, original_len=len(payload), packet_data=payload, + length2=20)), + ('NameResolutionBlock', BlockType.Name_Resolution_Block, + NameResolutionBlock(length=12, records=[], options=[], length2=12)), + ('InterfaceStatisticsBlock', BlockType.Interface_Statistics_Block, + InterfaceStatisticsBlock(length=24, interface_id=0, timestamp_high=0, + timestamp_low=0, options=[], length2=24)), + ('SystemdJournalExportBlock', BlockType.systemd_Journal_Export_Block, + SystemdJournalExportBlock(length=12, entry=b'', length2=12)), + ('DecryptionSecretsBlock', BlockType.Decryption_Secrets_Block, + DecryptionSecretsBlock(length=24, secrets_type=SecretsType.TLS_Key_Log, + secrets_length=len(payload), secrets_data=payload, + options=[], length2=24)), + ('CustomBlock', BlockType.Custom_Block_that_rewriters_can_copy_into_new_files, + CustomBlock(length=16, pen=0, data=b'', length2=16)), + ('PacketBlock', BlockType.Packet_Block, + PacketBlock(length=36, interface_id=0, drop_count=0, timestamp_high=0, + timestamp_low=0, captured_length=len(payload), + original_length=len(payload), packet_data=payload, options=[], + length2=36)), + ] + + def test_pcapng_every_block_schema_packs(self) -> None: + # Regression for #366: seven of the eleven block schemas could not be + # packed at all, from four independent causes -- a ``BytesField`` + # standing in for a ``PaddingField``, an option length callback reading + # a ``PaddingField``'s value, ``Schema.pre_pack`` never being called, + # and ``packet['__option_padding__']`` being subscripted on the packing + # path where only the unpacking path sets it. Since the SHB was among + # them, no valid PCAP-NG file could be written at all. + for label, _, schema in self._all_block_schemas(): + with self.subTest(schema=label): + packed = bytes(schema) + # every block schema packs the block without its 4-octet block + # type, so it is four short of the block total length it declares + self.assertEqual(len(packed) + 4, schema.length) + + def test_pcapng_padding_fields_are_padding_fields(self) -> None: + # Regression for #366, cause 1: ``PacketBlock.padding_data`` and + # ``DecryptionSecretsBlock.padding_data`` were declared ``BytesField``, + # which ``Schema.pack`` does not fill in, so packing them handed + # ``struct.pack`` the ``NoValue`` sentinel. + from pcapkit.corekit.fields.strings import PaddingField + from pcapkit.protocols.schema.misc.pcapng import (DecryptionSecretsBlock, + EnhancedPacketBlock, PacketBlock) + + for schema in (EnhancedPacketBlock, PacketBlock, DecryptionSecretsBlock): + with self.subTest(schema=schema.__name__): + self.assertIsInstance(schema.__fields__['padding_data'], PaddingField) + self.assertIsInstance(schema.__fields__['padding_opts'], PaddingField) + + def test_pcapng_option_length_callbacks_survive_missing_pack_keys(self) -> None: + # Regression for #366, causes 2 and 4: the option length callbacks read + # ``pkt['padding_data']`` and ``pkt['__option_padding__']``, neither of + # which is a key in the packet data while packing. + from pcapkit.protocols.schema.misc.pcapng import (DecryptionSecretsBlock, + EnhancedPacketBlock, + InterfaceDescriptionBlock, + InterfaceStatisticsBlock, + NameResolutionBlock, PacketBlock, + SectionHeaderBlock) + + cases = { + 'EnhancedPacketBlock': (EnhancedPacketBlock, {'length': 36, 'captured_len': 4}), + 'PacketBlock': (PacketBlock, {'length': 44, 'captured_length': 4}), + 'DecryptionSecretsBlock': (DecryptionSecretsBlock, {'length': 24, + 'secrets_length': 4}), + 'SectionHeaderBlock': (SectionHeaderBlock, {'length': 28}), + 'InterfaceDescriptionBlock': (InterfaceDescriptionBlock, {'length': 20}), + 'NameResolutionBlock': (NameResolutionBlock, {'length': 12}), + 'InterfaceStatisticsBlock': (InterfaceStatisticsBlock, {'length': 24}), + } + for label, (schema, packet) in cases.items(): + with self.subTest(schema=label): + for name, field in schema.__fields__.items(): + if field.__class__.__name__ not in ('OptionField', 'PaddingField'): + continue + with self.subTest(field=name): + # the callback must not raise on a packet dict that + # carries no padding field values and no + # ``__option_padding__`` key, i.e. the packing case + self.assertGreaterEqual(field(dict(packet)).length, 0) + + def test_pcapng_section_header_block_packs_byteorder_magic(self) -> None: + # Regression for #366, cause 3: nothing in the package called + # ``Schema.pre_pack``, so ``SectionHeaderBlock.pre_pack`` never got to + # seed ``packet['match']`` and packing an SHB raised ``KeyError``. + from pcapkit.protocols.schema.misc.pcapng import SectionHeaderBlock + + for byteorder, endian in (('big', '>'), ('little', '<')): + with self.subTest(byteorder=byteorder): + schema = SectionHeaderBlock(length=28, magic=0x1A2B3C4D, major=1, minor=0, + section_length=-1, options=[], length2=28) + packed = schema.pack({'byteorder': byteorder}) + + length, magic, major, minor, section_length, length2 = struct.unpack( + f'{endian}IIHHqI', packed) + self.assertEqual(length, 28) + self.assertEqual(length2, 28) + self.assertEqual(magic, 0x1A2B3C4D) + self.assertEqual((major, minor), (1, 0)) + # section length not specified, i.e. all ones on the wire + self.assertEqual(section_length, -1) + self.assertEqual(packed[12:20], b'\xff' * 8) + + def test_pcapng_section_header_block_round_trips_through_bytes(self) -> None: + # Regression for #366: the point of the write path is that what it + # writes can be read back. Build a whole capture out of ``bytes()`` on + # the block schemas, then dissect it with the public reader and check + # the fields survived. + import tempfile + + from pcapkit.const.pcapng.block_type import BlockType + from pcapkit.const.pcapng.option_type import OptionType + from pcapkit.const.reg.linktype import LinkType + from pcapkit.interface import extract + from pcapkit.protocols.schema.misc.pcapng import PCAPNG as Header + from pcapkit.protocols.schema.misc.pcapng import (CommentOption, EndOfOption, + EnhancedPacketBlock, + InterfaceDescriptionBlock, + InterfaceStatisticsBlock, + SectionHeaderBlock, SimplePacketBlock) + + frame = (b'\xff\xff\xff\xff\xff\xff\x00\x11\x22\x33\x44\x55\x08\x00' + b'\x45\x00\x00\x1c\x00\x01\x00\x00\x40\x11\x00\x00' + b'\x0a\x00\x00\x01\x0a\x00\x00\x02' + b'\x04\xd2\x16\x2e\x00\x08\x00\x00') # eth / ipv4 / udp, 42 octets + self.assertEqual(len(frame), 42) + + comment = CommentOption(type=OptionType.opt_comment, length=6, comment='hello!') + endofopt = EndOfOption(type=OptionType.opt_endofopt, length=0) + options = [comment, endofopt] + # 4 octets of option header plus a 32-bit aligned 6-octet comment, then + # the 4 octets of the end-of-option-list marker + options_length = 4 + 8 + 4 + + shb = SectionHeaderBlock(length=28 + options_length, magic=0x1A2B3C4D, major=1, + minor=0, section_length=-1, options=options, + length2=28 + options_length) + idb = InterfaceDescriptionBlock(length=20, linktype=LinkType.ETHERNET, snaplen=0xFFFF, + options=[], length2=20) + epb = EnhancedPacketBlock(length=32 + 44, interface_id=0, timestamp_high=0, + timestamp_low=1_000_000, captured_len=len(frame), + original_len=len(frame), packet_data=frame, options=[], + length2=32 + 44) + spb = SimplePacketBlock(length=16 + 44, original_len=len(frame), + packet_data=frame + bytes(-len(frame) % 4), length2=16 + 44) + isb = InterfaceStatisticsBlock(length=24, interface_id=0, timestamp_high=0, + timestamp_low=0, options=[], length2=24) + + capture = b''.join(bytes(Header(type=block_type, block=block)) for block_type, block in ( + (BlockType.Section_Header_Block, shb), + (BlockType.Interface_Description_Block, idb), + (BlockType.Enhanced_Packet_Block, epb), + (BlockType.Simple_Packet_Block, spb), + (BlockType.Interface_Statistics_Block, isb), + )) + + handle, path = tempfile.mkstemp(suffix='.pcapng') + try: + with os.fdopen(handle, 'wb') as file: + file.write(capture) + extractor = extract(fin=path, store=True, nofile=True) + finally: + os.unlink(path) + + # the two packet blocks are the frames; the SHB, IDB and ISB are context + self.assertEqual(len(extractor.frame), 2) + + section = extractor.engine._ctx_list[0].section + self.assertEqual(section.byteorder, sys.byteorder) + self.assertEqual(section.version.major, 1) + self.assertEqual(section.version.minor, 0) + self.assertEqual(section.section_length, -1) + self.assertEqual(section.options[OptionType.opt_comment].comment, 'hello!') + + interfaces = extractor.engine._ctx_list[0].interfaces + self.assertEqual(len(interfaces), 1) + self.assertEqual(interfaces[0].linktype, LinkType.ETHERNET) + self.assertEqual(interfaces[0].snaplen, 0xFFFF) + + epb_read = extractor.frame[0] + self.assertEqual(epb_read.info.type, BlockType.Enhanced_Packet_Block) + self.assertEqual(epb_read.info.interface_id, 0) + self.assertEqual(epb_read.info.captured_len, len(frame)) + self.assertEqual(epb_read.info.original_len, len(frame)) + self.assertEqual(str(epb_read.protochain), 'Ethernet:IPv4:UDP') + ipv4 = epb_read.info.ethernet.ipv4 + self.assertEqual((str(ipv4.src), str(ipv4.dst)), ('10.0.0.1', '10.0.0.2')) + self.assertEqual((int(ipv4.udp.srcport), int(ipv4.udp.dstport)), (1234, 5678)) + + spb_read = extractor.frame[1] + self.assertEqual(spb_read.info.type, BlockType.Simple_Packet_Block) + self.assertEqual(spb_read.info.original_len, len(frame)) + self.assertEqual(str(spb_read.protochain), 'Ethernet:IPv4:UDP') + + statistics = extractor.engine._ctx_list[0].statistics + self.assertEqual(len(statistics), 1) + self.assertEqual(statistics[0].interface_id, 0) + + ########################################################################## + # Signed numeric fields (#366, uncovered by the SHB repro). + ########################################################################## + + def test_signed_number_fields_pack_negative_values(self) -> None: + # Uncovered while fixing #366: ``NumberField.pre_process`` masks the + # value against the *unsigned* bit mask, which turns any negative value + # into a pattern ``struct.pack`` rejects for a signed template. No + # signed field in the library could write a negative value, which is + # what a PCAP-NG section length of -1 (not specified) needs. + from pcapkit.corekit.fields.numbers import (Int8Field, Int16Field, Int32Field, Int64Field, + UInt32Field) + + cases = { + Int8Field: ('b', 1), + Int16Field: ('h', 2), + Int32Field: ('i', 4), + Int64Field: ('q', 8), + } + for field_type, (code, size) in cases.items(): + for value in (-1, -8, 0, 7): + with self.subTest(field=field_type.__name__, value=value): + field = field_type()({}) + self.assertEqual(field.pack(value, {}), + struct.pack(f'>{code}', value)) + self.assertEqual(len(field.pack(value, {})), size) + + # unsigned fields keep truncating, i.e. -1 stays all ones + self.assertEqual(UInt32Field()({}).pack(-1, {}), b'\xff\xff\xff\xff') + + ########################################################################## + # Option namespaces per block (#365). + ########################################################################## + + def test_pcapng_make_block_uses_its_own_option_namespace(self) -> None: + # Regression for #365: all eight ``_make_block_*`` methods passed + # ``namespace='shb'``, which is not a namespace anywhere -- neither on + # ``OptionType`` nor in the schema option registry -- so every raw-bytes + # option was misclassified as ``shb_unknown_*`` and ``OptionType`` was + # permanently extended with a member per unrecognised code. + from pcapkit.const.pcapng.block_type import BlockType + from pcapkit.const.pcapng.option_type import OptionType + from pcapkit.const.pcapng.secrets_type import SecretsType + from pcapkit.const.reg.linktype import LinkType + from pcapkit.protocols.misc.pcapng import PCAPNG + from pcapkit.protocols.schema.misc.pcapng import Option + + # ``'shb'`` names no namespace on either side of the wire + namespaces = set(dict(OptionType.__members_ns__)) | set(dict(Option.registry)) + self.assertNotIn('shb', namespaces) + + recorded = [] + + pcapng = object.__new__(PCAPNG) + pcapng._byte = 'little' + pcapng._opt = collections.Counter() + pcapng._ctx = None + pcapng._make_pcapng_options = lambda options, namespace: ( + recorded.append(namespace) or ([], 0) + ) + + raw = [b'\x00\x00\x00\x00'] # opt_endofopt, valid in every namespace + calls = [ + ('shb', BlockType.Section_Header_Block, 'opt', + lambda: pcapng._make_block_shb(options=raw)), + ('idb', BlockType.Interface_Description_Block, 'if', + lambda: pcapng._make_block_idb(linktype=LinkType.ETHERNET, options=raw)), + ('epb', BlockType.Enhanced_Packet_Block, 'epb', + lambda: pcapng._make_block_epb(interface_id=0, timestamp=0, + packet_data=b'data', options=raw)), + ('nrb', BlockType.Name_Resolution_Block, 'ns', + lambda: pcapng._make_block_nrb(records=[], options=raw)), + ('isb', BlockType.Interface_Statistics_Block, 'isb', + lambda: pcapng._make_block_isb(interface_id=0, timestamp=0, options=raw)), + ('dsb', BlockType.Decryption_Secrets_Block, 'dsb', + lambda: pcapng._make_block_dsb(secrets_type=SecretsType.get(0xDEAD_BEEF), + secrets_data=b'data', options=raw)), + ('cb', BlockType.Custom_Block_that_rewriters_can_copy_into_new_files, 'opt', + lambda: pcapng._make_block_cb(pen=0, data=b'data', options=raw)), + ('packet', BlockType.Packet_Block, 'pack', + lambda: pcapng._make_block_packet(interface_id=0, timestamp=0, + packet_data=b'data', options=raw)), + ] + + for name, block_type, expected, call in calls: + with self.subTest(block=name): + recorded.clear() + pcapng._type = block_type + with mock.patch('pcapkit.protocols.misc.pcapng.warn'): + call() + self.assertEqual(recorded, [expected]) + # the namespace has to be a real one, otherwise every code in + # it resolves as unknown + self.assertIn(expected, namespaces) + + # ... and nothing minted a bogus ``shb_*`` member along the way + self.assertNotIn('shb', set(dict(OptionType.__members_ns__))) + self.assertEqual([name for name in OptionType.__members__ if name.startswith('shb_')], []) + + def test_pcapng_make_block_namespace_matches_read_side_registry(self) -> None: + # The read side is the authority on which registry a block's options + # come from: each block schema's ``OptionField`` names it explicitly. + # #365's fix has to agree with it, block for block. + from pcapkit.protocols.schema.misc.pcapng import (DecryptionSecretsBlock, + EnhancedPacketBlock, + InterfaceDescriptionBlock, + InterfaceStatisticsBlock, + NameResolutionBlock, Option, PacketBlock, + SectionHeaderBlock) + + expected = { + SectionHeaderBlock: 'opt', + InterfaceDescriptionBlock: 'if', + EnhancedPacketBlock: 'epb', + NameResolutionBlock: 'ns', + InterfaceStatisticsBlock: 'isb', + DecryptionSecretsBlock: 'dsb', + PacketBlock: 'pack', + } + for schema, namespace in expected.items(): + with self.subTest(schema=schema.__name__): + self.assertIs(schema.__fields__['options'].registry, + dict(Option.registry)[namespace]) + + ########################################################################## + # Interface ID bounds (#367). + ########################################################################## + + def test_pcapng_out_of_range_interface_id_is_a_format_error(self) -> None: + # Regression for #367: the engine's bounds guards ran after the block + # had been parsed, and the parse itself indexed the section's interface + # list, so an out-of-range interface ID escaped as a bare ``IndexError`` + # from ``_get_timezone`` instead of the intended ``FormatError``. + from pcapkit.const.pcapng.block_type import BlockType + from pcapkit.protocols.misc.pcapng import PCAPNG + from pcapkit.utilities.exceptions import FormatError + + tags = { + BlockType.Enhanced_Packet_Block: 'EPB', + BlockType.Simple_Packet_Block: 'SPB', + BlockType.Packet_Block: 'Packet', + BlockType.Interface_Statistics_Block: 'ISB', + } + getters = ('_get_resolution', '_get_offset', '_get_timezone', '_get_linktype') + + pcapng = object.__new__(PCAPNG) + # a section that describes two interfaces, as #367's capture does + pcapng._ctx = types.SimpleNamespace(interfaces=[object(), object()]) + + for block_type, tag in tags.items(): + pcapng._type = block_type + for getter in getters: + if getter == '_get_linktype' and block_type not in PCAPNG.PACKET_TYPES: + continue # documented to be unavailable off a packet block + for interface_id in (2, 7, -1): + with self.subTest(block=tag, getter=getter, interface_id=interface_id): + with self.assertRaises(FormatError) as context: + getattr(pcapng, getter)(interface_id) + message = str(context.exception) + self.assertIn(f'PCAP-NG: [{tag}]', message) + self.assertIn(f'invalid interface ID: {interface_id}', message) + + # an in-range ID still resolves, i.e. the guard is a bound and not a ban + pcapng._type = BlockType.Enhanced_Packet_Block + self.assertIs(pcapng._get_interface(1), pcapng._ctx.interfaces[1]) + + def test_pcapng_interface_id_bounds_reported_for_unlisted_blocks(self) -> None: + # A block that is not in ``INTERFACE_ID_BLOCK_TAGS`` still has to report + # a ``FormatError`` rather than an ``IndexError``, naming the block type. + from pcapkit.const.pcapng.block_type import BlockType + from pcapkit.protocols.misc.pcapng import PCAPNG + from pcapkit.utilities.exceptions import FormatError + + pcapng = object.__new__(PCAPNG) + pcapng._ctx = types.SimpleNamespace(interfaces=[]) + pcapng._type = BlockType.Name_Resolution_Block + + with self.assertRaises(FormatError) as context: + pcapng._get_interface(0) + self.assertIn('invalid interface ID: 0', str(context.exception)) + self.assertIn(str(BlockType.Name_Resolution_Block), str(context.exception)) + + ########################################################################## + # Section header block options and the section byte order (#368). + ########################################################################## + + @staticmethod + def _section_header_bytes(byteorder: str) -> bytes: + """A Section Header Block, options and all, in the given byte order. + + The buffer starts at the block total length, i.e. it is what a + :class:`~pcapkit.protocols.schema.misc.pcapng.SectionHeaderBlock` is + handed, without the leading 4-octet block type. + + """ + endian = '>' if byteorder == 'big' else '<' + + def option(code: int, value: bytes) -> bytes: + return (struct.pack(f'{endian}HH', code, len(value)) + + value + bytes(-len(value) % 4)) + + options = (option(2, b'Apple MBP') # shb_hardware + + option(3, b'OS-X 10.10.5') # shb_os + + option(4, b'pcap_writer.lua') # shb_userappl + + option(1, b'test001') # opt_comment + + option(0, b'')) # opt_endofopt + length = 28 + len(options) + return (struct.pack(f'{endian}IIHHq', length, 0x1A2B3C4D, 1, 0, -1) + + options + struct.pack(f'{endian}I', length)) + + def test_pcapng_section_header_options_use_the_section_byteorder(self) -> None: + # Regression for #368: an SHB's own options were read with the host byte + # order rather than the one its Byte-Order Magic declares, because + # ``packet['byteorder']`` is only seeded from an existing section + # context and the first SHB of a file has none by construction. In a + # big-endian section every option was replaced by one bogus + # ``opt_unknown`` whose byte-swapped length swallowed the option area. + from pcapkit.const.pcapng.option_type import OptionType + from pcapkit.protocols.schema.misc.pcapng import SectionHeaderBlock + + expected = [ + (2, 9, b'Apple MBP'), + (3, 12, b'OS-X 10.10.5'), + (4, 15, b'pcap_writer.lua'), + (OptionType.opt_comment, 7, 'test001'), + (OptionType.opt_endofopt, 0, None), + ] + + for byteorder in ('big', 'little'): + with self.subTest(byteorder=byteorder): + buffer = self._section_header_bytes(byteorder) + schema = SectionHeaderBlock.unpack(buffer, len(buffer), {}) + + self.assertEqual(schema.byteorder, byteorder) + self.assertEqual(schema.length, len(buffer) + 4) + self.assertEqual(schema.section_length, -1) + + parsed = [(option.type, option.length, + getattr(option, 'data', getattr(option, 'comment', None))) + for option in schema.options] + self.assertEqual(parsed, expected) + + def test_pcapng_section_header_options_ignore_the_previous_section(self) -> None: + # #368's multi-section case: for a second SHB ``self._ctx`` is not + # ``None``, it is the *previous* section's context, so seeding the byte + # order from it is wrong even when a context exists. An SHB has to read + # its own options through its own magic whatever the packet data says. + from pcapkit.protocols.schema.misc.pcapng import SectionHeaderBlock + + for byteorder in ('big', 'little'): + other = 'little' if byteorder == 'big' else 'big' + for declared in (byteorder, other): + with self.subTest(section=byteorder, declared=declared): + buffer = self._section_header_bytes(byteorder) + schema = SectionHeaderBlock.unpack(buffer, len(buffer), + {'byteorder': declared}) + + self.assertEqual(schema.byteorder, byteorder) + self.assertEqual([option.length for option in schema.options], + [9, 12, 15, 7, 0]) + + def test_pcapng_big_endian_sample_section_options_are_intact(self) -> None: + # The same defect, on the sample capture #368 reports it against. Skipped + # unless the samples have been generated, since they are not tracked. + from pcapkit.const.pcapng.option_type import OptionType + from pcapkit.interface import extract + + path = os.path.join('examples', 'captures', 'dhcp_big_endian.pcapng') + if not os.path.isfile(path): + self.skipTest('run examples/generators/make_samples.py first') + + extractor = extract(fin=path, store=True, nofile=True) + section = extractor.engine._ctx_list[0].section + + self.assertEqual(section.byteorder, 'big') + self.assertEqual([option.length for _, option in section.options.items(multi=True)], + [9, 12, 15, 7, 0]) + self.assertEqual(section.options[OptionType.opt_comment].comment, 'test001') + if __name__ == '__main__': unittest.main()