diff --git a/pcapkit/protocols/data/misc/pcapng.py b/pcapkit/protocols/data/misc/pcapng.py index 5c90acc10b..b9c20e280b 100644 --- a/pcapkit/protocols/data/misc/pcapng.py +++ b/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': ... diff --git a/pcapkit/protocols/misc/pcapng.py b/pcapkit/protocols/misc/pcapng.py index ff47268f01..7bd027d667 100644 --- a/pcapkit/protocols/misc/pcapng.py +++ b/pcapkit/protocols/misc/pcapng.py @@ -178,8 +178,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 (FormatError, ProtocolError, RegistryError, StreamEOFError, - UnsupportedCall, stacklevel) +from pcapkit.utilities.exceptions import (EnumValueError, FormatError, ProtocolError, RegistryError, + StreamEOFError, UnsupportedCall, stacklevel) from pcapkit.utilities.warnings import (AttributeWarning, DeprecatedFormatWarning, ProtocolWarning, RegistryWarning, warn) @@ -191,7 +191,8 @@ from decimal import Decimal from enum import IntEnum as StdlibEnum from ipaddress import IPv4Address, IPv4Interface, IPv6Address, IPv6Interface - from typing import IO, Any, Callable, Counter, DefaultDict, Optional, Tuple, Type, Union + from typing import (IO, Any, Callable, Counter, DefaultDict, Optional, Tuple, Type, TypeVar, + Union) from aenum import IntEnum as AenumEnum from mypy_extensions import DefaultArg, KwArg, NamedArg @@ -221,6 +222,8 @@ SecretsConstructor = Callable[[Enum_SecretsType, DefaultArg(Optional[Data_DSBSecrets]), KwArg(Any)], Schema_DSBSecrets] + _FlagsEnum = TypeVar('_FlagsEnum', 'PacketDirection', 'PacketReception') + # check Python version py37 = ((version_info := sys.version_info).major >= 3 and version_info.minor >= 7) py38 = ((version_info := sys.version_info).major >= 3 and version_info.minor >= 8) @@ -284,12 +287,31 @@ def _join_names(names: 'list[str]') -> 'str': return DecodedString(text, raw) +def _unassigned_member(enum_cls: 'Type[_FlagsEnum]', value: 'int') -> '_FlagsEnum': + """Build an unassigned ``epb_flags`` sub-field member, absent from the lookup tables. + + Args: + enum_cls: :class:`PacketDirection` or :class:`PacketReception`. + value: The member's value. + + Returns: + The unassigned member. + + """ + obj = int.__new__(enum_cls, value) + obj._name_ = 'Unassigned' # pylint: disable=protected-access + obj._value_ = value # pylint: disable=protected-access + return obj + + class PacketDirection(EnumLookup, enum.IntEnum): """Packet direction for ``epb_flags`` options. Shares the :class:`~pcapkit.corekit.enum.EnumLookup` lookup contract that - every non-registry enumeration follows (GitHub issue :issue:`877`); it - defines neither ``get`` nor ``_missing_`` of its own. + every non-registry enumeration follows (GitHub issue :issue:`877`). Its + ``_missing_`` resolves the one value the specification leaves undefined, + ``0b11``, to an unassigned member, so a capture carrying it still parses + and rebuilds (:issue:`1382`). """ @@ -300,13 +322,33 @@ class PacketDirection(EnumLookup, enum.IntEnum): #: Outbound packet. OUTBOUND = 0b10 + @classmethod + def _missing_(cls, value: 'object') -> 'PacketDirection': + """Lookup function used when value is not found. + + Args: + value: Value to get enum item. + + Returns: + An unassigned member, absent from the lookup tables. + + Raises: + EnumValueError: If ``value`` does not fit the 2-bit sub-field. + + """ + if not (isinstance(value, int) and 0b00 <= value <= 0b11): + raise EnumValueError(f'{value!r} is not a valid {cls.__name__}') + return _unassigned_member(cls, value) + class PacketReception(EnumLookup, enum.IntEnum): """Reception type for ``epb_flags`` options. Shares the :class:`~pcapkit.corekit.enum.EnumLookup` lookup contract that - every non-registry enumeration follows (GitHub issue :issue:`877`); it - defines neither ``get`` nor ``_missing_`` of its own. + every non-registry enumeration follows (GitHub issue :issue:`877`). Its + ``_missing_`` resolves the values the specification leaves undefined, + ``0b101`` to ``0b111``, to an unassigned member, so a capture carrying one + still parses and rebuilds (:issue:`1382`). """ @@ -321,6 +363,24 @@ class PacketReception(EnumLookup, enum.IntEnum): #: Promiscuous. PROMISCUOUS = 0b100 + @classmethod + def _missing_(cls, value: 'object') -> 'PacketReception': + """Lookup function used when value is not found. + + Args: + value: Value to get enum item. + + Returns: + An unassigned member, absent from the lookup tables. + + Raises: + EnumValueError: If ``value`` does not fit the 3-bit sub-field. + + """ + if not (isinstance(value, int) and 0b000 <= value <= 0b111): + raise EnumValueError(f'{value!r} is not a valid {cls.__name__}') + return _unassigned_member(cls, value) + # NOTE: RFC 9850 :rfc:`9850#section-4.2` makes the TLS key labels an IANA # registry ("TLS SSLKEYLOGFILE Labels", Specification Required) rather than a @@ -961,8 +1021,18 @@ def register_block(cls, code: 'Enum_BlockType', meth: 'str | tuple[BlockParser, code: PCAP-NG block type code. meth: Method name or callable to parse and/or construct the block. + Warns: + RegistryWarning: If a different entry is already registered for + ``code``; it is overwritten. Handing back the entry already + stored, or one equal to it, is a silent no-op. The entry + ``code`` shipped with is accepted like any other, so an + override can be undone (:issue:`1362`, :issue:`1363`). + """ - if code in cls.__block__: + incumbent = cls.__block__.get(code) + if incumbent is not None and incumbent == meth: + return + if incumbent is not None: warn(f'PCAP-NG: [Type {code}] block already registered', RegistryWarning) cls.__block__[code] = meth @@ -974,9 +1044,19 @@ def register_option(cls, code: 'Enum_OptionType', meth: 'str | tuple[OptionParse code: PCAP-NG option type code. meth: Method name or callable to parse and/or construct the option. + Warns: + RegistryWarning: If a different entry is already registered for + ``code``; it is overwritten. Handing back the entry already + stored, or one equal to it, is a silent no-op. The entry + ``code`` shipped with is accepted like any other, so an + override can be undone (:issue:`1362`, :issue:`1363`). + """ code_key = _option_key(code) - if code_key in cls.__option__: + incumbent = cls.__option__.get(code_key) + if incumbent is not None and incumbent == meth: + return + if incumbent is not None: warn(f'PCAP-NG: [Option {code}] option already registered', RegistryWarning) cls.__option__[code_key] = meth @@ -988,8 +1068,18 @@ def register_record(cls, code: 'Enum_RecordType', meth: 'str | tuple[RecordParse code: PCAP-NG name resolution record type code. meth: Method name or callable to parse and/or construct the name resolution record. + Warns: + RegistryWarning: If a different entry is already registered for + ``code``; it is overwritten. Handing back the entry already + stored, or one equal to it, is a silent no-op. The entry + ``code`` shipped with is accepted like any other, so an + override can be undone (:issue:`1362`, :issue:`1363`). + """ - if code in cls.__record__: + incumbent = cls.__record__.get(code) + if incumbent is not None and incumbent == meth: + return + if incumbent is not None: warn(f'PCAP-NG: [Type {code}] name resolution record already registered', RegistryWarning) cls.__record__[code] = meth @@ -1001,8 +1091,18 @@ def register_secrets(cls, code: 'Enum_SecretsType', meth: 'str | tuple[SecretsPa code: PCAP-NG decryption secrets type code. meth: Method name or callable to parse and/or construct the decryption secrets. + Warns: + RegistryWarning: If a different entry is already registered for + ``code``; it is overwritten. Handing back the entry already + stored, or one equal to it, is a silent no-op. The entry + ``code`` shipped with is accepted like any other, so an + override can be undone (:issue:`1362`, :issue:`1363`). + """ - if code in cls.__secrets__: + incumbent = cls.__secrets__.get(code) + if incumbent is not None and incumbent == meth: + return + if incumbent is not None: warn(f'PCAP-NG: [Secrets {code}] decryption secrets already registered', RegistryWarning) cls.__secrets__[code] = meth @@ -1652,28 +1752,6 @@ def _block_area(block: 'Data_PCAPNG', name: 'str') -> 'Any': """ return getattr(block, f'{name}_raw', getattr(block, name)) - @staticmethod - def _flags_enum(enum: 'Type[PacketDirection] | Type[PacketReception]', value: 'int', - option: 'Enum_OptionType') -> 'Any': - """Look a flags word sub-field up in its enumeration. - - Args: - enum: :class:`PacketDirection` or :class:`PacketReception`. - value: Sub-field value. - option: Option type, for the error message. - - Returns: - The enumeration member. - - Raises: - ProtocolError: If the specification assigns ``value`` no meaning. - - """ - try: - return enum(value) - except ValueError: - raise ProtocolError(f'PCAP-NG: [{option.name}] invalid {enum.__name__}: {value:#b}') from None - @classmethod def _make_packet_data(cls, block: 'Data_PCAPNG') -> 'bytes | ProtocolBase': """Create the packet data of an EPB, SPB or PB from its data model. @@ -2025,6 +2103,13 @@ def _read_block_epb(self, schema: 'Schema_EnhancedPacketBlock', *, original_len=schema.original_len, options=self._read_pcapng_options(schema.options), ) + + # NOTE: A captured length running past the block is kept as declared, + # and the packet data is the octets the block holds (:issue:`1405`). + if schema.captured_len > len(schema.packet_data): + warn(f'PCAP-NG: [Block {header.type}] captured length {schema.captured_len} runs past ' + f'the block, which holds {len(schema.packet_data)} octet(s); kept as captured', + ProtocolWarning, stacklevel=stacklevel()) return self._decode_next_layer(data, self._get_linktype(schema.interface_id), schema.captured_len) # type: ignore[return-value] @@ -2069,6 +2154,15 @@ def _read_block_spb(self, schema: 'Schema_SimplePacketBlock', *, captured_len=len(schema.packet_data), original_len=schema.original_len, ) + + # NOTE: A block holding more packet data than the snaplen of interface 0 + # allows keeps the excess (:issue:`1384`); the specification says the + # snaplen determines the length, so the excess is reported. + snaplen = self._ctx.interfaces[0].snaplen if self._ctx is not None and self._ctx.interfaces else 0 + if snaplen and data.captured_len > snaplen: + warn(f'PCAP-NG: [Block {header.type}] packet data of {data.captured_len} octet(s) ' + f'exceeds the snaplen of {snaplen}; kept as captured', + ProtocolWarning, stacklevel=stacklevel()) return self._decode_next_layer(data, self._get_linktype(0), data.captured_len) # type: ignore[return-value] def _read_block_nrb(self, schema: 'Schema_NameResolutionBlock', *, @@ -2201,11 +2295,19 @@ def _read_block_systemd(self, schema: 'Schema_SystemdJournalExportBlock', *, Parsed packet data. """ + entries = schema.data + # NOTE: A malformed entry is kept as the octets captured, as one + # :obj:`bytes` item, and parses as no entries. The octets are kept on + # the block as ``entry_raw``, which the block's ``make`` writes back + # verbatim (:issue:`1406`). + kept = len(entries) == 1 and isinstance(entries[0], bytes) data = Data_SystemdJournalExportBlock( type=header.type, length=schema.length, - data=tuple(schema.data), + data=() if kept else tuple(entries), # type: ignore[arg-type] ) + if kept: + data.__update__([('entry_raw', entries[0])]) return data def _read_block_dsb(self, schema: 'Schema_DecryptionSecretsBlock', *, @@ -2375,6 +2477,13 @@ def _read_block_packet(self, schema: 'Schema_PacketBlock', *, original_len=schema.original_length, options=self._read_pcapng_options(schema.options), ) + + # NOTE: A captured length running past the block is kept as declared, + # and the packet data is the octets the block holds (:issue:`1405`). + if schema.captured_length > len(schema.packet_data): + warn(f'PCAP-NG: [Block {header.type}] captured length {schema.captured_length} runs past ' + f'the block, which holds {len(schema.packet_data)} octet(s); kept as captured', + ProtocolWarning, stacklevel=stacklevel()) return self._decode_next_layer(data, self._get_linktype(schema.interface_id), schema.captured_length) # type: ignore[return-value] @@ -2998,9 +3107,12 @@ def _read_option_epb_flags(self, schema: 'Schema_EPB_FlagsOption', *, option = Data_EPB_FlagsOption( type=schema.type, length=schema.length, - direction=self._flags_enum(PacketDirection, schema.flags['direction'], schema.type), - reception=self._flags_enum(PacketReception, schema.flags['reception'], schema.type), + direction=PacketDirection(schema.flags['direction']), + reception=PacketReception(schema.flags['reception']), fcs_len=schema.flags['fcs_len'], + checksum_not_ready=bool(schema.flags['checksum_not_ready']), + checksum_valid=bool(schema.flags['checksum_valid']), + tcp_segmentation_offloaded=bool(schema.flags['tcp_segmentation_offloaded']), reserved=schema.flags['reserved'], crc_error=bool(schema.flags['crc_error']), too_long=bool(schema.flags['too_long']), @@ -3457,9 +3569,12 @@ def _read_option_pack_flags(self, schema: 'Schema_PACK_FlagsOption', *, option = Data_PACK_FlagsOption( type=schema.type, length=schema.length, - direction=self._flags_enum(PacketDirection, schema.flags['direction'], schema.type), - reception=self._flags_enum(PacketReception, schema.flags['reception'], schema.type), + direction=PacketDirection(schema.flags['direction']), + reception=PacketReception(schema.flags['reception']), fcs_len=schema.flags['fcs_len'], + checksum_not_ready=bool(schema.flags['checksum_not_ready']), + checksum_valid=bool(schema.flags['checksum_valid']), + tcp_segmentation_offloaded=bool(schema.flags['tcp_segmentation_offloaded']), reserved=schema.flags['reserved'], crc_error=bool(schema.flags['crc_error']), too_long=bool(schema.flags['too_long']), @@ -3975,17 +4090,20 @@ def _make_block_spb(self, block: 'Optional[Data_SimplePacketBlock]' = None, *, if block is not None: original_len = block.original_len packet_data = self._make_packet_data(block) - - 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 + # NOTE: The data model's ``captured_len`` already bounds the data, + # including a block that kept octets past the snaplen (:issue:`1384`). + captured_len = block.captured_len else: - snaplen = self._ctx.interfaces[0].snaplen or 0xFFFF_FFFF_FFFF_FFFF - captured_len = min(original_len, snaplen) + 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 @@ -4093,7 +4211,8 @@ def _make_block_systemd(self, block: 'Optional[Data_SystemdJournalExportBlock]' self._byte = sys.byteorder if block is not None: - entries = cast('list[OrderedMultiDict[str, str | bytes]]', block.data) + entries = cast('list[OrderedMultiDict[str, str | bytes]] | bytes', + getattr(block, 'entry_raw', block.data)) if entries is None: data = b'' @@ -5053,6 +5172,9 @@ def _make_option_epb_flags(self, type: 'Enum_OptionType', option: 'Optional[Data reception_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long reception_reversed: 'bool' = False, fcs_len: 'int' = 0, + checksum_not_ready: 'bool' = False, + checksum_valid: 'bool' = False, + tcp_segmentation_offloaded: 'bool' = False, reserved: 'int' = 0, crc_error: 'bool' = False, too_long: 'bool' = False, @@ -5077,7 +5199,10 @@ def _make_option_epb_flags(self, type: 'Enum_OptionType', option: 'Optional[Data reception_namespace: Namespace of packet reception. reception_reversed: Whether to reverse packet reception namespace. fcs_len: Length of FCS field, in bytes. - reserved: Bits 9 to 23 of the flags word, verbatim. + checksum_not_ready: Whether the checksum is not ready. + checksum_valid: Whether the checksum is valid. + tcp_segmentation_offloaded: Whether TCP segmentation was offloaded. + reserved: Bits 12 to 23 of the flags word, verbatim. crc_error: Whether CRC error occurred. too_long: Whether packet is too long. too_short: Whether packet is too short. @@ -5103,6 +5228,9 @@ def _make_option_epb_flags(self, type: 'Enum_OptionType', option: 'Optional[Data direction_val = option.direction reception_val = option.reception fcs_len = option.fcs_len + checksum_not_ready = option.checksum_not_ready + checksum_valid = option.checksum_valid + tcp_segmentation_offloaded = option.tcp_segmentation_offloaded reserved = option.reserved crc_error = option.crc_error too_long = option.too_long @@ -5125,6 +5253,9 @@ def _make_option_epb_flags(self, type: 'Enum_OptionType', option: 'Optional[Data 'direction': getattr(direction_val, 'value', direction_val), 'reception': getattr(reception_val, 'value', reception_val), 'fcs_len': fcs_len, + 'checksum_not_ready': int(checksum_not_ready), + 'checksum_valid': int(checksum_valid), + 'tcp_segmentation_offloaded': int(tcp_segmentation_offloaded), 'reserved': reserved, 'crc_error': int(crc_error), 'too_long': int(too_long), @@ -5638,6 +5769,9 @@ def _make_option_pack_flags(self, type: 'Enum_OptionType', option: 'Optional[Dat reception_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long reception_reversed: 'bool' = False, fcs_len: 'int' = 0, + checksum_not_ready: 'bool' = False, + checksum_valid: 'bool' = False, + tcp_segmentation_offloaded: 'bool' = False, reserved: 'int' = 0, crc_error: 'bool' = False, too_long: 'bool' = False, @@ -5662,7 +5796,10 @@ def _make_option_pack_flags(self, type: 'Enum_OptionType', option: 'Optional[Dat reception_namespace: Namespace of packet reception. reception_reversed: Whether to reverse packet reception namespace. fcs_len: Length of FCS field, in bytes. - reserved: Bits 9 to 23 of the flags word, verbatim. + checksum_not_ready: Whether the checksum is not ready. + checksum_valid: Whether the checksum is valid. + tcp_segmentation_offloaded: Whether TCP segmentation was offloaded. + reserved: Bits 12 to 23 of the flags word, verbatim. crc_error: Whether CRC error occurred. too_long: Whether packet is too long. too_short: Whether packet is too short. @@ -5688,6 +5825,9 @@ def _make_option_pack_flags(self, type: 'Enum_OptionType', option: 'Optional[Dat direction_val = option.direction reception_val = option.reception fcs_len = option.fcs_len + checksum_not_ready = option.checksum_not_ready + checksum_valid = option.checksum_valid + tcp_segmentation_offloaded = option.tcp_segmentation_offloaded reserved = option.reserved crc_error = option.crc_error too_long = option.too_long @@ -5710,6 +5850,9 @@ def _make_option_pack_flags(self, type: 'Enum_OptionType', option: 'Optional[Dat 'direction': getattr(direction_val, 'value', direction_val), 'reception': getattr(reception_val, 'value', reception_val), 'fcs_len': fcs_len, + 'checksum_not_ready': int(checksum_not_ready), + 'checksum_valid': int(checksum_valid), + 'tcp_segmentation_offloaded': int(tcp_segmentation_offloaded), 'reserved': reserved, 'crc_error': int(crc_error), 'too_long': int(too_long), diff --git a/pcapkit/protocols/schema/misc/pcapng.py b/pcapkit/protocols/schema/misc/pcapng.py index 4e1a25f817..f0ba086ddd 100644 --- a/pcapkit/protocols/schema/misc/pcapng.py +++ b/pcapkit/protocols/schema/misc/pcapng.py @@ -101,8 +101,14 @@ class EPBFlags(TypedDict): #: Block, and is used with those link layers (e.g. PPP) where the length of #: the FCS can change during time. fcs_len: int - #: Bits 9 to 23, kept verbatim (checksum and segmentation offload - #: flags, reserved bits and unnamed link-layer-dependent errors). + #: Checksum not ready (bit 9). + checksum_not_ready: int + #: Checksum valid (bit 10). + checksum_valid: int + #: TCP segmentation offloaded (bit 11). + tcp_segmentation_offloaded: int + #: Bits 12 to 23, kept verbatim (reserved bits and unnamed + #: link-layer-dependent errors). reserved: int #: Link-layer-dependent error - CRC error (bit 24). crc_error: int @@ -135,8 +141,14 @@ class PACKFlags(TypedDict): #: Block, and is used with those link layers (e.g. PPP) where the length of #: the FCS can change during time. fcs_len: int - #: Bits 9 to 23, kept verbatim (checksum and segmentation offload - #: flags, reserved bits and unnamed link-layer-dependent errors). + #: Checksum not ready (bit 9). + checksum_not_ready: int + #: Checksum valid (bit 10). + checksum_valid: int + #: TCP segmentation offloaded (bit 11). + tcp_segmentation_offloaded: int + #: Bits 12 to 23, kept verbatim (reserved bits and unnamed + #: link-layer-dependent errors). reserved: int #: Link-layer-dependent error - CRC error (bit 24). crc_error: int @@ -443,6 +455,30 @@ def callback(pkt: 'dict[str, Any]') -> 'int': return callback +def captured_area(name: 'str') -> 'Callable[[dict[str, Any]], int]': + """Bound a packet block's Packet Data to the octets the block holds. + + Args: + name: Name of the block's captured length field, ``captured_len`` or + ``captured_length``. + + Returns: + A callback returning the captured length, never past the block's + Block Total Length less its 32 framing octets, and never below zero. + + A captured length running past the block would otherwise read the trailing + Block Total Length, and whatever follows the block, as packet data, so the + rebuild could not be byte-exact. Bounded, the block keeps its declared + captured length and exactly the octets it holds, and the padding and option + area are sized from those octets rather than from the declared length + (:issue:`1405`). + + """ + def callback(pkt: 'dict[str, Any]') -> 'int': + return max(0, min(pkt[name], pkt['length'] - 32)) + return callback + + def pcapng_block_selector(packet: 'dict[str, Any]') -> 'Field': """Selector function for :attr:`PCAPNG.block` field. @@ -568,14 +604,16 @@ def post_process(self, value: 'int | bytes', packet: 'dict[str, Any]') -> 'dict[ #: Sub-fields of the ``epb_flags`` and ``pack_flags`` words. ``reserved`` keeps -#: bits 9 to 23 verbatim -- the checksum and segmentation offload bits 9-11, the -#: reserved bits 12-15 and the link-layer-dependent error bits 16-23 that the -#: data model does not name. +#: bits 12 to 23 verbatim -- the reserved bits 12-15 and the link-layer-dependent +#: error bits 16-23 that the data model does not name. PACKET_FLAGS = { 'direction': (0, 2), 'reception': (2, 3), 'fcs_len': (5, 4), - 'reserved': (9, 15), + 'checksum_not_ready': (9, 1), + 'checksum_valid': (10, 1), + 'tcp_segmentation_offloaded': (11, 1), + 'reserved': (12, 12), 'crc_error': (24, 1), 'too_long': (25, 1), 'too_short': (26, 1), @@ -1441,18 +1479,18 @@ class EnhancedPacketBlock(BlockType, code=Enum_BlockType.Enhanced_Packet_Block): captured_len: 'int' = UInt32Field(callback=byteorder_callback) #: Original packet length. original_len: 'int' = UInt32Field(callback=byteorder_callback) - #: Packet data. - packet_data: 'bytes' = PayloadField(length=lambda pkt: pkt['captured_len']) + #: Packet data, bounded by the block (see :func:`captured_area`). + packet_data: 'bytes' = PayloadField(length=captured_area('captured_len')) #: Padding. - padding_data: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['captured_len'] % 4) % 4) + padding_data: 'bytes' = PaddingField(length=lambda pkt: (4 - captured_area('captured_len')(pkt) % 4) % 4) #: Options. options: 'list[Option]' = OptionAreaField( # 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=bounded_area(lambda pkt: pkt['length'] - 32 - pkt['captured_len'] - - (4 - pkt['captured_len'] % 4) % 4), + length=bounded_area(lambda pkt: pkt['length'] - 32 + - (captured_area('captured_len')(pkt) + 3) // 4 * 4), base_schema=_EPB_Option, type_name='type', registry=Option.registry['epb'], @@ -1471,6 +1509,29 @@ def __init__(self, length: 'int', interface_id: 'int', timestamp_high: 'int', options: 'list[Option | bytes] | bytes', length2: 'int') -> 'None': ... +def _spb_data_length(packet: 'dict[str, Any]') -> 'int': + """Length of a Simple Packet Block's packet data. + + Args: + packet: Unpacked data so far, with ``length``, ``original_len`` and, + when interface 0 sets one, ``snaplen``. + + Returns: + ``min(original_len, snaplen)``, bounded by the octets the block holds + less its 16 framing octets. A block that holds more than the snaplen + and its padding keeps the excess as packet data instead, so no octet + is dropped and the trailing Block Total Length is read where it sits + (:issue:`1384`). + + """ + area = max(0, packet['length'] - 16) + size = min(packet['original_len'], area) + snaplen = packet.get('snaplen') + if snaplen is not None and snaplen < size and area <= (snaplen + 3) // 4 * 4: + size = snaplen + return size + + @schema_final class SimplePacketBlock(BlockType, code=Enum_BlockType.Simple_Packet_Block): """Header schema for PCAP-NG Simple Packet Block (SPB).""" @@ -1481,11 +1542,9 @@ 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, ``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))) + #: Packet data, ``min(original_len, snaplen)`` octets, bounded by what the + #: block holds; see :func:`_spb_data_length`. + packet_data: 'bytes' = PayloadField(length=_spb_data_length) #: Padding. padding: 'bytes' = PaddingField(length=lambda pkt: (4 - len(pkt['packet_data']) % 4) % 4) #: Block total length. @@ -1505,16 +1564,25 @@ def _split_names(resol: 'str') -> 'list[str]': :class:`~pcapkit.corekit.fields.strings.StringField` decodes it. Returns: - The names. They are split from the *octets* the data was decoded from, - so that a name that is not valid UTF-8 comes back as a + The names, one per terminator, so an empty name is kept as ``''`` and + joining them back yields the data exactly (:issue:`1383`). They are + split from the *octets* the data was decoded from, so that a name that + is not valid UTF-8 comes back as a :class:`~pcapkit.corekit.fields.strings.DecodedString` carrying its own octets; splitting the decoded text would return plain strings and lose them. + Raises: + ProtocolError: If the data is empty or does not end in a zero + terminator. The record area is then kept as the octets captured + (see :class:`OptionAreaField`). + """ raw = resol.raw if isinstance(resol, DecodedString) else resol.encode('utf-8') + if not raw.endswith(b'\x00'): + raise ProtocolError(f'PCAP-NG: [NRB] name resolution data is not zero-terminated: {raw!r}') names = [] # type: list[str] - for octets in raw.rstrip(b'\x00').split(b'\x00'): + for octets in raw[:-1].split(b'\x00'): text = octets.decode('utf-8', 'replace') names.append(text if text.encode('utf-8') == octets else DecodedString(text, octets)) return names @@ -1890,9 +1958,14 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Self': :exc:`OverflowError` (``cannot fit 'int' into an index-sized integer``), and below that it silently returns whatever is there, so the same malformed prefix would be either fatal or invisible - depending only on its magnitude. It is clamped to the octets the - entry has left and reported, which is :func:`nonnegative`'s rule at - the other end of the same range. + depending only on its magnitude. A length past the octets the entry + has left raises :exc:`~pcapkit.utilities.exceptions.ProtocolError` + instead, and the whole entry is then kept as the octets captured, as + the one :obj:`bytes` item of :attr:`data`, which the block's ``make`` + writes back verbatim. Clamping the field would rebuild a different + length prefix and a terminator that was never there (:issue:`1406`, + the #1325 ruling that :class:`OptionAreaField` applies to option + areas). Field names, keys and values are decoded with ``errors='replace'`` rather than strictly. A non-UTF-8 octet in any of the three would @@ -1960,6 +2033,25 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Self': """ self = cast('Self', super().post_process(packet)) + try: + self.data = self._read_entries() + except ProtocolError as error: + warn(f'{error}; journal entry of {len(self.entry)} octet(s) kept as captured', + ProtocolWarning, stacklevel=stacklevel()) + self.data = [self.entry] + return self + + def _read_entries(self) -> 'list[OrderedMultiDict[str, str | bytes]]': + """Walk :attr:`entry` into its journal entries, as :meth:`post_process` describes. + + Returns: + The journal entries. + + Raises: + ProtocolError: If a binary field declares a length past the octets + left in the entry. + + """ data = [] # type: list[OrderedMultiDict[str, str | bytes]] total = len(self.entry) entry_data = io.BytesIO(self.entry) @@ -2037,39 +2129,30 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Self': length = struct.unpack(' available - if clamped: - warn(f'PCAP-NG: [systemd Journal Export] binary field {line!r} ' - f'declares {length} octet(s) with {available} left in its ' - f'entry; reading {available}', SchemaWarning, - stacklevel=stacklevel()) - length = available + if length > available: + raise ProtocolError(f'PCAP-NG: [systemd Journal Export] binary field {line!r} ' + f'declares {length} octet(s) with {available} left in its entry') entry.add(self._decode_text(line), entry_data.read(length)) - if not clamped: - # the one octet the format puts here to terminate the - # field; a value already clamped to the entry's own end - # left nothing behind to check, and was reported above. - # the reader's position is well defined either way -- - # exactly length + 1 octets past where the field name - # started -- so a bad octet here ends only this - # entry's field collection: it does not abort the - # walk, which keeps looking for the next entry's - # separator from here. - terminator = entry_data.read(1) - if terminator != b'\n': - warn(f'PCAP-NG: [systemd Journal Export] binary field ' - f'{line!r} is not followed by the newline that ' - f'terminates it; ending the entry', SchemaWarning, - stacklevel=stacklevel()) - break + # the one octet the format puts here to terminate the + # field. the reader's position is well defined -- exactly + # length + 1 octets past where the field name started -- + # so a bad octet here ends only this entry's field + # collection: it does not abort the walk, which keeps + # looking for the next entry's separator from here. + terminator = entry_data.read(1) + if terminator != b'\n': + warn(f'PCAP-NG: [systemd Journal Export] binary field ' + f'{line!r} is not followed by the newline that ' + f'terminates it; ending the entry', SchemaWarning, + stacklevel=stacklevel()) + break data.append(entry) if entry_data.tell() >= total and not separator: break - self.data = data - return self + return data @staticmethod def _decode_text(octets: 'bytes') -> 'str': @@ -2093,8 +2176,9 @@ def _decode_text(octets: 'bytes') -> 'str': return octets.decode('utf-8', errors='replace') if TYPE_CHECKING: - #: Journal entry (decoded). - data: 'list[OrderedMultiDict[str, str | bytes]]' + #: Journal entry (decoded), or the entry as captured, as one + #: :obj:`bytes` item, if it is malformed. + data: 'list[OrderedMultiDict[str, str | bytes] | bytes]' def __init__(self, length: 'int', entry: 'bytes', length2: 'int') -> 'None': ... @@ -2365,16 +2449,16 @@ class PacketBlock(BlockType, code=Enum_BlockType.Packet_Block): captured_length: 'int' = UInt32Field(callback=byteorder_callback) #: Original packet length. original_length: 'int' = UInt32Field(callback=byteorder_callback) - #: Packet data. - packet_data: 'bytes' = PayloadField(length=lambda pkt: pkt['captured_length']) + #: Packet data, bounded by the block (see :func:`captured_area`). + packet_data: 'bytes' = PayloadField(length=captured_area('captured_length')) #: Padding. - padding_data: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['captured_length'] % 4) % 4) + padding_data: 'bytes' = PaddingField(length=lambda pkt: (4 - captured_area('captured_length')(pkt) % 4) % 4) #: Options. options: 'list[Option]' = OptionAreaField( # NOTE: see EnhancedPacketBlock.options on why the padding is recomputed # here instead of being read back from ``padding_data``. - length=bounded_area(lambda pkt: pkt['length'] - 32 - pkt['captured_length'] - - (4 - pkt['captured_length'] % 4) % 4), + length=bounded_area(lambda pkt: pkt['length'] - 32 + - (captured_area('captured_length')(pkt) + 3) // 4 * 4), base_schema=_PACK_Option, type_name='type', registry=Option.registry['pack'], diff --git a/tests/protocols/misc/test_pcapng_captured_len_overrun_1405_unit.py b/tests/protocols/misc/test_pcapng_captured_len_overrun_1405_unit.py new file mode 100644 index 0000000000..7f50d5b32b --- /dev/null +++ b/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(' 'bytes': + """An obsolete Packet Block on interface 0 declaring ``captured_len`` over ``data``.""" + body = struct.pack(' '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() diff --git a/tests/protocols/misc/test_pcapng_flags_undefined_1382_unit.py b/tests/protocols/misc/test_pcapng_flags_undefined_1382_unit.py new file mode 100644 index 0000000000..b0da80cdf1 --- /dev/null +++ b/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() diff --git a/tests/protocols/misc/test_pcapng_journal_binary_overrun_1406_unit.py b/tests/protocols/misc/test_pcapng_journal_binary_overrun_1406_unit.py new file mode 100644 index 0000000000..145d916c0d --- /dev/null +++ b/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(' None: + from pcapkit.utilities.warnings import ProtocolWarning + + for entry in (b'BIN\n' + struct.pack(' None: + for entry, field, value in ((b'BIN\n' + struct.pack(' 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(' 'bytes': + """A Name Resolution Block holding one IPv4 (or IPv6) record with ``names`` as its data.""" + address = bytes(range(16)) if ipv6 else bytes([10, 0, 0, 1]) + record = struct.pack(' None: + for ipv6 in (False, True): + for data, names in ((b'a\x00\x00', ('a', '')), + (b'a\x00\x00\x00', ('a', '', '')), + (b'\x00\x00', ('', '')), + (b'a\x00\x00b\x00', ('a', '', 'b')), + (b'a\x00b\x00', ('a', 'b')), + (b'\x00', ('',))): + with self.subTest(data=data, ipv6=ipv6): + octets = _nrb(data, ipv6) + records = self._parse(octets).info.records + self.assertEqual(records[2 if ipv6 else 1].records, names) + self.assertRebuilds(octets) + + def test_unterminated_data_is_kept_as_captured(self) -> None: + from pcapkit.utilities.warnings import ProtocolWarning + + for ipv6 in (False, True): + for data in (b'a', b'a\x00b', b''): + with self.subTest(data=data, ipv6=ipv6): + octets = _nrb(data, ipv6) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + info = self._parse(octets).info + self.assertTrue(any(issubclass(item.category, ProtocolWarning) for item in caught)) + self.assertEqual(len(info.records), 0) + self.assertEqual(info.records_raw, octets[8:-4]) # the whole record area + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + self.assertRebuilds(octets) + + def test_make_writes_empty_names(self) -> None: + """``make`` -> parse -> pack keeps a trailing empty name.""" + from pcapkit.const.pcapng.block_type import BlockType + from pcapkit.const.pcapng.record_type import RecordType + from pcapkit.protocols.misc.pcapng import PCAPNG + + context = self._context() + block = PCAPNG(num=2, sct=1, ctx=context, type=BlockType.Name_Resolution_Block, + block={'records': [(RecordType.nrb_record_ipv4, + {'ip': '10.0.0.1', 'names': ['a', '']})]}) + self.assertIn(b'\x0a\x00\x00\x01a\x00\x00', block.data) + self.assertEqual(self._parse(block.data, context).info.records[RecordType.nrb_record_ipv4].records, ('a', '')) + self.assertRebuilds(block.data, context) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/misc/test_pcapng_registrar_identity_unit.py b/tests/protocols/misc/test_pcapng_registrar_identity_unit.py new file mode 100644 index 0000000000..7c97232b45 --- /dev/null +++ b/tests/protocols/misc/test_pcapng_registrar_identity_unit.py @@ -0,0 +1,115 @@ +# -*- coding: utf-8 -*- +"""GitHub issue #1401: the PCAP-NG parser registrars warn only on a real displacement. + +``register_block``, ``register_option``, ``register_record`` and +``register_secrets`` tested only whether ``code`` was present, so handing back +the entry already stored warned that it was being overwritten. They now apply +the #1400 guard: an equal entry is a silent no-op, and the entry a code shipped +with is accepted back after an override, which undoes it. + +Every registry is process-wide, so each case runs under +:func:`unittest.mock.patch.dict`. Classes are imported inside each test, after +:func:`~tests._support.reimport_once_per_class`. + +""" + +import importlib +import unittest +import warnings +from unittest import mock + +from tests._support import reimport_once_per_class + +#: ``(registrar, registry)`` for each PCAP-NG parser registrar. +SITES = ( + ('register_block', '__block__'), + ('register_option', '__option__'), + ('register_record', '__record__'), + ('register_secrets', '__secrets__'), +) + + +def _parse(*args: object, **kwargs: object) -> None: + """Stand-in parser.""" + + +def _make(*args: object, **kwargs: object) -> None: + """Stand-in constructor.""" + + +class TestPCAPNGRegistrarIdentity(unittest.TestCase): + """Pin when a PCAP-NG parser registrar warns.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _each_site(self, check): # type: ignore[no-untyped-def] + """Run ``check(register, registry, code_of)`` for every site, each in its own ``subTest``. + + ``code_of`` maps a registry key back to the code the registrar takes: + the option registry is keyed by :func:`_option_key`, not by the code. + + """ + module = importlib.import_module('pcapkit.protocols.misc.pcapng') + option_type = importlib.import_module('pcapkit.const.pcapng.option_type').OptionType + option_codes = {module._option_key(member): member for member in option_type} + for meth, attr in SITES: + registry = module.PCAPNG.__dict__[attr] + code_of = option_codes.__getitem__ if attr == '__option__' else (lambda key: key) + with self.subTest(site=meth), mock.patch.dict(registry): + check(getattr(module.PCAPNG, meth), registry, code_of) + + def _count(self, call, *args) -> int: # type: ignore[no-untyped-def] + """Run ``call(*args)`` and return how many registry warnings it raised.""" + from pcapkit.utilities.warnings import RegistryWarning + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + call(*args) + return sum(issubclass(item.category, RegistryWarning) for item in caught) + + def test_reregistering_the_stored_entry_is_silent(self) -> None: + def check(register, registry, code_of): # type: ignore[no-untyped-def] + for key, value in list(registry.items()): + self.assertEqual(self._count(register, code_of(key), value), 0, key) + self.assertIs(registry[key], value) + # An equal method name built at run time is the same entry, and + # the stored object is kept. + self.assertEqual(self._count(register, code_of(key), ''.join(list(value))), 0, key) + self.assertIs(registry[key], value) + self._each_site(check) + + def test_reregistering_a_parser_pair_is_silent(self) -> None: + def check(register, registry, code_of): # type: ignore[no-untyped-def] + key = next(iter(registry)) + pair = (_parse, _make) + self.assertEqual(self._count(register, code_of(key), pair), 1) + self.assertIs(registry[key], pair) + self.assertEqual(self._count(register, code_of(key), pair), 0) + self.assertEqual(self._count(register, code_of(key), (_parse, _make)), 0) + self.assertIs(registry[key], pair) + self._each_site(check) + + def test_restoring_the_shipped_entry_undoes_an_override(self) -> None: + def check(register, registry, code_of): # type: ignore[no-untyped-def] + key, shipped = next(iter(registry.items())) + self.assertEqual(self._count(register, code_of(key), 'w1401_override'), 1) + self.assertEqual(registry[key], 'w1401_override') + # Putting the shipped entry back displaces the override, so it warns. + self.assertEqual(self._count(register, code_of(key), shipped), 1) + self.assertIs(registry[key], shipped) + self.assertEqual(self._count(register, code_of(key), shipped), 0) + self._each_site(check) + + def test_an_unregistered_code_is_silent_and_not_defaulted(self) -> None: + def check(register, registry, code_of): # type: ignore[no-untyped-def] + key = next(iter(registry)) + registry.pop(key) + # The lookup must not trip the ``defaultdict`` factory. + self.assertEqual(self._count(register, code_of(key), 'w1401_new'), 0) + self.assertEqual(registry[key], 'w1401_new') + self._each_site(check) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/misc/test_pcapng_roundtrip2_unit.py b/tests/protocols/misc/test_pcapng_roundtrip2_unit.py index 8e9d10c7d4..1cf4a766e9 100644 --- a/tests/protocols/misc/test_pcapng_roundtrip2_unit.py +++ b/tests/protocols/misc/test_pcapng_roundtrip2_unit.py @@ -243,12 +243,13 @@ def test_pack_flags_are_numbered_from_the_least_significant_bit(self) -> None: self.assertEqual(flags.direction, PacketDirection.OUTBOUND) self.assertRebuilds(octets) - def test_undefined_direction_is_rejected(self) -> None: - """#1273: direction ``0b11`` raises an in-library error.""" - from pcapkit.utilities.exceptions import ProtocolError + def test_undefined_direction_is_kept(self) -> None: + """#1382: direction ``0b11`` is kept as an unassigned member and rebuilds.""" + from pcapkit.const.pcapng.option_type import OptionType - with self.assertRaisesRegex(ProtocolError, r'invalid PacketDirection'): - self._parse(_epb_with_options(H('0200 0400 03000000'))) + octets = _epb_with_options(H('0200 0400 03000000')) + self.assertEqual(self._parse(octets).info.options[OptionType.epb_flags].direction, 0b11) + self.assertRebuilds(octets) def test_nrb_names_decode_as_utf8(self) -> None: """#1277: record names are UTF-8, not a guessed charset.""" diff --git a/tests/protocols/misc/test_pcapng_spb_snaplen_1384_unit.py b/tests/protocols/misc/test_pcapng_spb_snaplen_1384_unit.py new file mode 100644 index 0000000000..8f34a2e33f --- /dev/null +++ b/tests/protocols/misc/test_pcapng_spb_snaplen_1384_unit.py @@ -0,0 +1,79 @@ +# -*- coding: utf-8 -*- +"""GitHub issue #1384: a Simple Packet Block longer than the snaplen keeps its excess. + +An SPB holding 16 data octets under an interface snaplen of 8 parsed as 8 +octets: the other 8 were dropped, the trailing Block Total Length was read from +the packet data, and the rebuild was 8 octets shorter. The parse now keeps every +octet the block holds past the snaplen and its padding, warns that the data +exceeds the snaplen, and rebuilds byte-exactly. A block sized for the snaplen +still reads ``min(original_len, snaplen)`` octets. + +: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 _spb(data: 'bytes', original_len: 'int | None' = None) -> 'bytes': + """A Simple Packet Block carrying ``data``, zero-padded to 32 bits.""" + body = struct.pack(' 'tuple[object, list[str]]': + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + parsed = self._parse(octets, self._context(snaplen=snaplen)) + return parsed, [str(item.message) for item in caught] + + def test_excess_past_the_snaplen_is_kept(self) -> None: + for data, original_len in ((b'A' * 16, None), (b'A' * 16, 20), (b'A' * 12, 16)): + with self.subTest(size=len(data), original_len=original_len): + octets = _spb(data, original_len) + parsed, messages = self._parse_quiet(octets, snaplen=8) + self.assertEqual(parsed.info.captured_len, len(data)) # type: ignore[attr-defined] + self.assertEqual(parsed.packet.payload, data) # type: ignore[attr-defined] + self.assertTrue(any('exceeds the snaplen of 8' 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, self._context(snaplen=8)) + + def test_block_sized_for_the_snaplen_is_unchanged(self) -> None: + for snaplen, data, original_len in ((8, b'A' * 8, 16), (6, b'A' * 6, 10), (0, b'A' * 16, None)): + with self.subTest(snaplen=snaplen, original_len=original_len): + octets = _spb(data, original_len) + parsed, messages = self._parse_quiet(octets, snaplen=snaplen) + self.assertEqual(parsed.info.captured_len, len(data)) # type: ignore[attr-defined] + self.assertFalse(any('exceeds the snaplen' in text for text in messages), messages) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + self.assertRebuilds(octets, self._context(snaplen=snaplen)) + + def test_make_still_clips_to_the_snaplen(self) -> None: + """``make`` -> parse -> pack: a fresh block holds the snaplen's worth.""" + 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') + block = PCAPNG(num=2, sct=1, ctx=context, type=BlockType.Simple_Packet_Block, + block={'packet_data': b'A' * 16}) + self.assertEqual(len(block.data), 24) + self.assertRebuilds(block.data, context) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/misc/test_pcapng_unit.py b/tests/protocols/misc/test_pcapng_unit.py index 23da0866d7..3ad181178d 100644 --- a/tests/protocols/misc/test_pcapng_unit.py +++ b/tests/protocols/misc/test_pcapng_unit.py @@ -108,16 +108,16 @@ def test_pcapng_registry_overwrites_warn(self) -> None: ) try: with mock.patch('pcapkit.protocols.misc.pcapng.warn') as warn: - PCAPNG.register_block(BlockType.Section_Header_Block, 'shb') - PCAPNG.register_option(OptionType.opt_endofopt, 'endofopt') - PCAPNG.register_record(RecordType.nrb_record_end, 'end') - PCAPNG.register_secrets(SecretsType.TLS_Key_Log, 'tls') + PCAPNG.register_block(BlockType.Section_Header_Block, 'shb_w1401') + PCAPNG.register_option(OptionType.opt_endofopt, 'endofopt_w1401') + PCAPNG.register_record(RecordType.nrb_record_end, 'end_w1401') + PCAPNG.register_secrets(SecretsType.TLS_Key_Log, 'tls_w1401') self.assertEqual(warn.call_count, 4) - self.assertEqual(block_map[BlockType.Section_Header_Block], 'shb') - self.assertEqual(option_map[_option_key(OptionType.opt_endofopt)], 'endofopt') - self.assertEqual(record_map[RecordType.nrb_record_end], 'end') - self.assertEqual(secrets_map[SecretsType.TLS_Key_Log], 'tls') + self.assertEqual(block_map[BlockType.Section_Header_Block], 'shb_w1401') + self.assertEqual(option_map[_option_key(OptionType.opt_endofopt)], 'endofopt_w1401') + self.assertEqual(record_map[RecordType.nrb_record_end], 'end_w1401') + self.assertEqual(secrets_map[SecretsType.TLS_Key_Log], 'tls_w1401') finally: block_map[BlockType.Section_Header_Block] = originals[0] option_map[_option_key(OptionType.opt_endofopt)] = originals[1] @@ -134,7 +134,7 @@ def test_pcapng_record_registry_warning_names_the_name_resolution_record(self) - original = record_map[RecordType.nrb_record_end] try: with mock.patch('pcapkit.protocols.misc.pcapng.warn') as warn: - PCAPNG.register_record(RecordType.nrb_record_end, 'end') + PCAPNG.register_record(RecordType.nrb_record_end, 'end_w1401') finally: record_map[RecordType.nrb_record_end] = original @@ -922,7 +922,10 @@ def with_length(schema: object, length: int) -> object: 'direction': PacketDirection.INBOUND.value, 'reception': PacketReception.UNICAST.value, 'fcs_len': 4, - 'reserved': 0, # bits 9-23, kept since #1273 + 'checksum_not_ready': 0, + 'checksum_valid': 0, + 'tcp_segmentation_offloaded': 0, + 'reserved': 0, # bits 12-23, kept since #1273 'crc_error': 1, 'too_long': 0, 'too_short': 1, @@ -1627,6 +1630,7 @@ def test_pcapng_block_readers_cover_container_blocks(self) -> None: pcapng._sect = 2 pcapng._fnum = 3 pcapng._opt = collections.Counter() + pcapng._ctx = None # the SPB reader looks up interface 0's snaplen pcapng._get_linktype = lambda interface_id=0: LinkType.ETHERNET pcapng._read_timestamp = lambda high, low, interface_id=0: ( datetime.datetime.fromtimestamp(high + low, datetime.timezone.utc), @@ -2288,7 +2292,10 @@ def assert_duplicate(method: str, code: OptionType, block: BlockType, **kwargs: direction=PacketDirection.INBOUND, reception=PacketReception.MULTICAST, fcs_len=3, - reserved=0, # bits 9-23, kept since #1273 + checksum_not_ready=False, + checksum_valid=False, + tcp_segmentation_offloaded=False, + reserved=0, # bits 12-23, kept since #1273 crc_error=True, too_long=True, too_short=False, @@ -2376,7 +2383,10 @@ def assert_duplicate(method: str, code: OptionType, block: BlockType, **kwargs: direction=PacketDirection.OUTBOUND, reception=PacketReception.BROADCAST, fcs_len=1, - reserved=0, # bits 9-23, kept since #1273 + checksum_not_ready=False, + checksum_valid=False, + tcp_segmentation_offloaded=False, + reserved=0, # bits 12-23, kept since #1273 crc_error=True, too_long=False, too_short=True, @@ -4311,7 +4321,10 @@ def test_an_epb_declaring_captured_len_past_its_block_does_not_reach_struct(self self.assertEqual(schema.captured_len, 0xFFFFFF) self.assertEqual(len(schema.options), 0) self.assertEqual(schema.padding_opts, b'') - self.assertTrue(caught) + # #1405: the packet data is bounded by the block, so nothing goes + # negative and the trailing Block Total Length is read where it sits. + self.assertEqual(schema.packet_data, bytes(8)) + self.assertEqual(schema.length2, len(raw) + 4) # the length counts the block type def test_a_block_total_length_below_its_own_floor_does_not_reach_struct(self) -> None: """The other shape: a declared length under the block's fixed fields. @@ -4437,32 +4450,33 @@ def test_a_well_formed_journal_binary_field_still_reads_its_value(self) -> None: self.assertEqual([entry for entry in caught if entry.category.__name__ == 'SchemaWarning'], []) - def test_a_journal_binary_field_declaring_more_than_its_entry_is_clamped(self) -> None: + def test_a_journal_binary_field_declaring_more_than_its_entry_is_kept_as_captured(self) -> None: """A 64-bit length was either fatal or invisible, by magnitude alone. At ``2**63`` and above :meth:`io.BytesIO.read` refuses the length outright with a bare :exc:`OverflowError`; below that it silently returned whatever happened to be there. Both are the same malformed prefix, so both get the - same answer: clamp to the octets the entry has left, and say so. + same answer: the whole entry is kept as captured, parses as no entries, + and the extraction says so and carries on (#1406). """ - from pcapkit.utilities.warnings import SchemaWarning + from pcapkit.utilities.warnings import ProtocolWarning, SchemaWarning # ``b'BINARY\n' + 8 octets + b'abc\n'`` is 19 octets, so the block pads it # with one NUL and five octets follow the length prefix for declared in (6, 1 << 10, 1 << 30, 1 << 62, 1 << 63, 2 ** 64 - 1): - with self.subTest(declared=declared, expect='clamped'): + with self.subTest(declared=declared, expect='kept'): entries, caught = self._extract_journal( b'BINARY\n' + struct.pack(' No per-entry loop on a bad terminator -- so #722's per-entry check silently widened into a per-block one, and a well-formed entry behind a corrupted one was dropped entirely, which - :meth:`test_a_journal_binary_field_declaring_more_than_its_entry_is_clamped` + :meth:`test_a_journal_binary_field_declaring_more_than_its_entry_is_kept_as_captured` above cannot show because its bad terminator always lands at the block's own end. There is nothing between the corrupted octet and ``GOOD=1`` here -- not even the format's own separator -- so recovery