From 497a3e95c275d1ecd0f518e2b1493d44e58a78b6 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Wed, 7 Oct 2026 17:21:23 -0400 Subject: [PATCH] fix(pcapng): blocks and options rebuild as parsed (#1272, #1273, #1277, #1278, #1281, #1286) - A Section Header Block keeps its parsed byte order; make() takes one. - epb_flags/pack_flags number bits from the least-significant bit, in the section's byte order; bits 9-23 are kept as `reserved`, and an undefined direction or reception raises ProtocolError. - NRB record names and ns_dnsname decode as UTF-8, and a name that is not UTF-8 keeps its octets through the split into names. - An unknown block's length counts the 12 framing octets and its padding. - epb_verdict may repeat. - The WireGuard key log error names the offending line. --- pcapkit/protocols/data/misc/pcapng.py | 12 +- pcapkit/protocols/misc/pcapng.py | 90 ++++++++-- pcapkit/protocols/schema/misc/pcapng.py | 158 ++++++++++++++---- .../misc/test_pcapng_roundtrip2_unit.py | 147 ++++++++++++++++ tests/protocols/misc/test_pcapng_unit.py | 10 +- 5 files changed, 364 insertions(+), 53 deletions(-) diff --git a/pcapkit/protocols/data/misc/pcapng.py b/pcapkit/protocols/data/misc/pcapng.py index 0df72152a0..5c90acc10b 100644 --- a/pcapkit/protocols/data/misc/pcapng.py +++ b/pcapkit/protocols/data/misc/pcapng.py @@ -353,6 +353,10 @@ 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). + reserved: 'int' #: Link-layer-dependent error - CRC error (bit 24). crc_error: 'bool' #: Link-layer-dependent error - packet too long error (bit 25). @@ -372,7 +376,7 @@ class EPB_FlagsOption(Option): if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', direction: 'PacketDirection', - reception: 'PacketReception', fcs_len: 'int', crc_error: 'bool', + reception: 'PacketReception', fcs_len: 'int', 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': ... @@ -861,6 +865,10 @@ 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). + reserved: 'int' #: Link-layer-dependent error - CRC error (bit 24). crc_error: 'bool' #: Link-layer-dependent error - packet too long error (bit 25). @@ -880,7 +888,7 @@ class PACK_FlagsOption(Option): if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', direction: 'PacketDirection', - reception: 'PacketReception', fcs_len: 'int', crc_error: 'bool', + reception: 'PacketReception', fcs_len: 'int', 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 1733b068bf..3e9f716045 100644 --- a/pcapkit/protocols/misc/pcapng.py +++ b/pcapkit/protocols/misc/pcapng.py @@ -262,6 +262,28 @@ def _utf8_length(text: 'str') -> 'int': return len(text.encode('utf-8')) +def _join_names(names: 'list[str]') -> 'str': + """Join NRB record names into zero-terminated name resolution data. + + Args: + names: Host names; a :class:`~pcapkit.corekit.fields.strings.DecodedString` + contributes the octets it was decoded from. + + Returns: + The name resolution data, as a + :class:`~pcapkit.corekit.fields.strings.DecodedString` when any name + carries octets that its text does not encode back to. + + """ + names = list(names) or [''] + text = ''.join(f'{name}\x00' for name in names) + raw = b''.join((name.raw if isinstance(name, DecodedString) else name.encode('utf-8')) + b'\x00' + for name in names) + if text.encode('utf-8') == raw: + return text + return DecodedString(text, raw) + + class PacketDirection(EnumLookup, enum.IntEnum): """Packet direction for ``epb_flags`` options. @@ -1020,6 +1042,10 @@ def pack(self, **kwargs: 'Any') -> 'bytes': if self._ctx is not None: self._byte = self._ctx.section.byteorder packet['byteorder'] = self._byte + elif self._type == Enum_BlockType.Section_Header_Block: + # NOTE: A section header starts its own section, so the byte order + # is the one :meth:`_make_block_shb` chose rather than the host's. + packet.setdefault('byteorder', self._byte) return self.__header__.pack(packet) def unpack(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_PCAPNG': @@ -1600,6 +1626,28 @@ def _make_data(cls, data: 'Data_PCAPNG') -> 'dict[str, Any]': # type: ignore[ov 'block': data, } + @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. @@ -2920,9 +2968,10 @@ def _read_option_epb_flags(self, schema: 'Schema_EPB_FlagsOption', *, option = Data_EPB_FlagsOption( type=schema.type, length=schema.length, - direction=PacketDirection(schema.flags['direction']), - reception=PacketReception(schema.flags['reception']), + direction=self._flags_enum(PacketDirection, schema.flags['direction'], schema.type), + reception=self._flags_enum(PacketReception, schema.flags['reception'], schema.type), fcs_len=schema.flags['fcs_len'], + reserved=schema.flags['reserved'], crc_error=bool(schema.flags['crc_error']), too_long=bool(schema.flags['too_long']), too_short=bool(schema.flags['too_short']), @@ -3378,9 +3427,10 @@ def _read_option_pack_flags(self, schema: 'Schema_PACK_FlagsOption', *, option = Data_PACK_FlagsOption( type=schema.type, length=schema.length, - direction=PacketDirection(schema.flags['direction']), - reception=PacketReception(schema.flags['reception']), + direction=self._flags_enum(PacketDirection, schema.flags['direction'], schema.type), + reception=self._flags_enum(PacketReception, schema.flags['reception'], schema.type), fcs_len=schema.flags['fcs_len'], + reserved=schema.flags['reserved'], crc_error=bool(schema.flags['crc_error']), too_long=bool(schema.flags['too_long']), too_short=bool(schema.flags['too_short']), @@ -3696,7 +3746,10 @@ def _make_block_unknown(self, block: 'Optional[Data_UnknownBlock]' = None, *, if block is not None: data = block.body - length = len(data) + # NOTE: The body is padded to 32 bits, and Block Total Length counts it + # together with the 12 octets of block type and both length fields. + data += bytes(-len(data) % 4) + length = len(data) + 12 return Schema_UnknownBlock( length=length, @@ -3710,6 +3763,7 @@ def _make_block_shb(self, block: 'Optional[Data_SectionHeaderBlock]' = None, *, minor_version: 'Optional[int]' = None, section_length: 'int' = -1, options: 'Optional[Option | list[Schema_Option | tuple[Enum_OptionType, dict[str, Any]] | bytes]]' = None, + byteorder: 'Literal["big", "little"]' = sys.byteorder, **kwargs: 'Any') -> 'Schema_SectionHeaderBlock': """Make PCAP-NG section header block (SHB). @@ -3720,13 +3774,16 @@ def _make_block_shb(self, block: 'Optional[Data_SectionHeaderBlock]' = None, *, minor_version: Minor version number. section_length: Section length. options: Block options. + byteorder: Byte order of the section; a parsed block keeps its own. **kwargs: Arbitrary keyword arguments. Returns: Constructed block schema. """ - self._byte = sys.byteorder + if block is not None: + byteorder = block.byteorder + self._byte = byteorder if block is not None: major_version = block.version.major @@ -4958,6 +5015,7 @@ 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, + reserved: 'int' = 0, crc_error: 'bool' = False, too_long: 'bool' = False, too_short: 'bool' = False, @@ -4981,6 +5039,7 @@ 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. crc_error: Whether CRC error occurred. too_long: Whether packet is too long. too_short: Whether packet is too short. @@ -5006,6 +5065,7 @@ 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 + reserved = option.reserved crc_error = option.crc_error too_long = option.too_long too_short = option.too_short @@ -5027,6 +5087,7 @@ 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, + 'reserved': reserved, 'crc_error': int(crc_error), 'too_long': int(too_long), 'too_short': int(too_short), @@ -5198,9 +5259,8 @@ def _make_option_epb_verdict(self, type: 'Enum_OptionType', option: 'Optional[Da if self._type != Enum_BlockType.Enhanced_Packet_Block: raise ProtocolError(f'PCAP-NG: [epb_verdict] option must be in Enhanced Packet Block, ' f'but found in {self._type} block.') - if self._opt[type] > 0: - raise ProtocolError(f'PCAP-NG: [epb_verdict] option must be only one, ' - f'but {self._opt[type] + 1} found.') + # NOTE: No only-one guard: the specification allows ``epb_verdict`` to + # appear multiple times, and the parser accepts every occurrence. if option is not None: verdict_val = option.verdict @@ -5540,6 +5600,7 @@ 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, + reserved: 'int' = 0, crc_error: 'bool' = False, too_long: 'bool' = False, too_short: 'bool' = False, @@ -5563,6 +5624,7 @@ 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. crc_error: Whether CRC error occurred. too_long: Whether packet is too long. too_short: Whether packet is too short. @@ -5588,6 +5650,7 @@ 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 + reserved = option.reserved crc_error = option.crc_error too_long = option.too_long too_short = option.too_short @@ -5609,6 +5672,7 @@ 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, + 'reserved': reserved, 'crc_error': int(crc_error), 'too_long': int(too_long), 'too_short': int(too_short), @@ -5809,11 +5873,11 @@ def _make_record_ipv4(self, type: 'Enum_RecordType', record: 'Optional[Data_IPv4 if names is None: names = [] - records = '\x00'.join(names) + '\x00' + records = _join_names(names) return Schema_IPv4Record( type=type, - length=4 + len(records), + length=4 + _utf8_length(records), ip=ip, resol=records, ) @@ -5841,11 +5905,11 @@ def _make_record_ipv6(self, type: 'Enum_RecordType', record: 'Optional[Data_IPv6 if names is None: names = [] - records = '\x00'.join(names) + '\x00' + records = _join_names(names) return Schema_IPv6Record( type=type, - length=16 + len(records), + length=16 + _utf8_length(records), ip=ip, resol=records, ) diff --git a/pcapkit/protocols/schema/misc/pcapng.py b/pcapkit/protocols/schema/misc/pcapng.py index 0e117bc98d..9a55f1c7fd 100644 --- a/pcapkit/protocols/schema/misc/pcapng.py +++ b/pcapkit/protocols/schema/misc/pcapng.py @@ -24,7 +24,8 @@ from pcapkit.corekit.fields.misc import ForwardMatchField, PayloadField, SchemaField, SwitchField from pcapkit.corekit.fields.numbers import (EnumField, Int32Field, Int64Field, NumberField, UInt8Field, UInt16Field, UInt32Field, UInt64Field) -from pcapkit.corekit.fields.strings import BitField, BytesField, PaddingField, StringField +from pcapkit.corekit.fields.strings import (BitField, BytesField, DecodedString, PaddingField, + StringField) from pcapkit.corekit.multidict import MultiDict, OrderedMultiDict from pcapkit.protocols.schema.schema import EnumSchema, Schema, schema_final from pcapkit.utilities.exceptions import FieldValueError, ProtocolError, stacklevel @@ -100,6 +101,9 @@ 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). + reserved: int #: Link-layer-dependent error - CRC error (bit 24). crc_error: int #: Link-layer-dependent error - packet too long error (bit 25). @@ -131,6 +135,9 @@ 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). + reserved: int #: Link-layer-dependent error - CRC error (bit 24). crc_error: int #: Link-layer-dependent error - packet too long error (bit 25). @@ -502,6 +509,87 @@ def dsb_secrets_selector(packet: 'dict[str, Any]') -> 'Field': return SchemaField(length=packet['secrets_length'], schema=schema) +class FlagsField(UInt32Field): + """32-bit flags word for protocol fields, with sub-fields numbered from the + least-significant bit, as the ``epb_flags`` and ``pack_flags`` options number + them. + + Unlike :class:`~pcapkit.corekit.fields.strings.BitField`, which numbers bits + from the most-significant bit of the raw octets and ignores byte order, the + word is read as an integer in the section's byte order first. + + Args: + namespace: Field namespace (a dict mapping field name to a tuple of + start bit, counted from the least-significant bit, and width). + callback: Callback function to be called upon + :meth:`self.__call__ `. + + """ + + def __init__(self, namespace: 'dict[str, tuple[int, int]]', + callback: 'Callable[[Self, dict[str, Any]], None]' = lambda *_: None) -> 'None': + super().__init__(callback=callback) + + self._flags = namespace + + def pre_process(self, value: 'dict[str, int]', packet: 'dict[str, Any]') -> 'int | bytes': # type: ignore[override] + """Process field value before construction (packing). + + Arguments: + value: Field value. + packet: Packet data. + + Returns: + Processed field value. + + Raises: + FieldValueError: If a sub-field value does not fit its bits. + + """ + word = 0 + for name, (start, size) in self._flags.items(): + part = value[name] + if not 0 <= part < 1 << size: + raise FieldValueError(f'{type(self).__name__}: subfield {name!r} value {part!r} ' + f'does not fit in {size} bit(s)') + word |= part << start + return super().pre_process(word, packet) + + def post_process(self, value: 'int | bytes', packet: 'dict[str, Any]') -> 'dict[str, int]': # type: ignore[override] + """Process field value after parsing (unpacked). + + Args: + value: Field value. + packet: Packet data. + + Returns: + Processed field value. + + """ + word = super().post_process(value, packet) + return {name: word >> start & ((1 << size) - 1) for name, (start, size) in self._flags.items()} + + +#: 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. +PACKET_FLAGS = { + 'direction': (0, 2), + 'reception': (2, 3), + 'fcs_len': (5, 4), + 'reserved': (9, 15), + 'crc_error': (24, 1), + 'too_long': (25, 1), + 'too_short': (26, 1), + 'gap_error': (27, 1), + 'unaligned_error': (28, 1), + 'delimiter_error': (29, 1), + 'preamble_error': (30, 1), + 'symbol_error': (31, 1), +} # type: dict[str, tuple[int, int]] + + class OptionEnumField(EnumField): """Enumerated value for protocol fields. @@ -1261,19 +1349,7 @@ class EPB_FlagsOption(_EPB_Option, code=Enum_OptionType.epb_flags): """Header schema for PCAP-NG ``epb_flags`` options.""" #: Flags. - flags: 'EPBFlags' = BitField(length=4, namespace={ - 'direction': (0, 2), - 'reception': (2, 3), - 'fcs_len': (5, 4), - 'crc_error': (24, 1), - 'too_long': (25, 1), - 'too_short': (26, 1), - 'gap_error': (27, 1), - 'unaligned_error': (28, 1), - 'delimiter_error': (29, 1), - 'preamble_error': (30, 1), - 'symbol_error': (31, 1), - }) + flags: 'EPBFlags' = FlagsField(namespace=PACKET_FLAGS, callback=byteorder_callback) #: Padding. padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) @@ -1424,6 +1500,29 @@ def __init__(self, length: 'int', original_len: 'int', length2: 'int') -> 'None': ... +def _split_names(resol: 'str') -> 'list[str]': + """Split NRB record name resolution data into its zero-terminated names. + + Args: + resol: Name resolution data, as a UTF-8 + :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 + :class:`~pcapkit.corekit.fields.strings.DecodedString` carrying its own + octets; splitting the decoded text would return plain strings and lose + them. + + """ + raw = resol.raw if isinstance(resol, DecodedString) else resol.encode('utf-8') + names = [] # type: list[str] + for octets in raw.rstrip(b'\x00').split(b'\x00'): + text = octets.decode('utf-8', 'replace') + names.append(text if text.encode('utf-8') == octets else DecodedString(text, octets)) + return names + + class NameResolutionRecord(EnumSchema[Enum_RecordType]): """Header schema for PCAP-NG NRB records.""" @@ -1463,7 +1562,7 @@ class IPv4Record(NameResolutionRecord, code=Enum_RecordType.nrb_record_ipv4): #: IPv4 address. ip: 'IPv4Address' = IPv4AddressField() #: Name resolution data. - resol: 'str' = StringField(length=bounded_option(lambda pkt: pkt['length'] - 4)) + resol: 'str' = StringField(length=bounded_option(lambda pkt: pkt['length'] - 4), encoding='utf-8') #: Padding. padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) @@ -1477,7 +1576,7 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': Revised schema. """ - self.names = self.resol.rstrip('\x00').split('\x00') + self.names = _split_names(self.resol) return self if TYPE_CHECKING: @@ -1494,7 +1593,7 @@ class IPv6Record(NameResolutionRecord, code=Enum_RecordType.nrb_record_ipv6): #: IPv6 address. ip: 'IPv6Address' = IPv6AddressField() #: Name resolution data. - resol: 'str' = StringField(length=bounded_option(lambda pkt: pkt['length'] - 16)) + resol: 'str' = StringField(length=bounded_option(lambda pkt: pkt['length'] - 16), encoding='utf-8') #: Padding. padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) @@ -1508,7 +1607,7 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': Revised schema. """ - self.names = self.resol.rstrip('\x00').split('\x00') + self.names = _split_names(self.resol) return self if TYPE_CHECKING: @@ -1532,7 +1631,7 @@ class NS_DNSNameOption(_NS_Option, code=Enum_OptionType.ns_dnsname): """Header schema for PCAP-NG ``ns_dnsname`` option.""" #: DNS name. - name: 'str' = StringField(length=bounded_option(lambda pkt: pkt['length'])) + name: 'str' = StringField(length=bounded_option(lambda pkt: pkt['length']), encoding='utf-8') #: Padding. padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) @@ -2083,9 +2182,10 @@ def post_process(self, packet: 'dict[str, Any]') -> 'Schema': if not line or line.startswith('#'): continue - label, op, secret = line.strip().split() - if op != '=': - raise FieldValueError('invalid WireGuard key log format: {line!r}') + fields = line.strip().split() + if len(fields) != 3 or fields[1] != '=': + raise FieldValueError(f'invalid WireGuard key log format: {line!r}') + label, _, secret = fields label_enum = WireGuardKeyLabel(label.upper()) entries.add(label_enum, base64.b64decode(secret)) @@ -2225,19 +2325,7 @@ class PACK_FlagsOption(_PACK_Option, code=Enum_OptionType.pack_flags): """Header schema for PCAP-NG ``pack_flags`` options.""" #: Flags. - flags: 'PACKFlags' = BitField(length=4, namespace={ - 'direction': (0, 2), - 'reception': (2, 3), - 'fcs_len': (5, 4), - 'crc_error': (24, 1), - 'too_long': (25, 1), - 'too_short': (26, 1), - 'gap_error': (27, 1), - 'unaligned_error': (28, 1), - 'delimiter_error': (29, 1), - 'preamble_error': (30, 1), - 'symbol_error': (31, 1), - }) + flags: 'PACKFlags' = FlagsField(namespace=PACKET_FLAGS, callback=byteorder_callback) #: Padding. padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) diff --git a/tests/protocols/misc/test_pcapng_roundtrip2_unit.py b/tests/protocols/misc/test_pcapng_roundtrip2_unit.py index 91740beb9d..ccfddde50b 100644 --- a/tests/protocols/misc/test_pcapng_roundtrip2_unit.py +++ b/tests/protocols/misc/test_pcapng_roundtrip2_unit.py @@ -149,5 +149,152 @@ def test_custom_option_shorter_than_pen_is_rejected(self) -> None: self._parse(octets) +#: A big-endian section header with no options -- the repro of #1272. +SHB_BE = H('0a0d0d0a 0000001c 1a2b3c4d 0001 0000 ffffffffffffffff 0000001c') + + +def _epb_with_options(options: 'bytes', order: 'str' = 'little') -> 'bytes': + """An EPB on interface 0 carrying ``abcd`` and ``options`` (``opt_endofopt`` appended).""" + def u32(value: 'int') -> 'bytes': + return value.to_bytes(4, order) # type: ignore[arg-type] + + body = u32(0) + u32(0) + u32(0) + u32(4) + u32(4) + b'abcd' + options + bytes(4) + length = len(body) + 12 + return u32(6) + u32(length) + body + u32(length) + + +class TestPCAPNGModelling(PCAPNGTestCase): + """Pin how PCAP-NG blocks and options are modelled.""" + + def _big_endian_context(self) -> 'Any': + """A one-section big-endian context with one Ethernet interface.""" + from pcapkit.const.pcapng.block_type import BlockType + from pcapkit.const.reg.linktype import LinkType + from pcapkit.foundation.engines.pcapng import Context + from pcapkit.protocols.misc.pcapng import PCAPNG + + context = Context(PCAPNG(SHB_BE, len(SHB_BE), num=0, sct=1, ctx=None).info) # type: Any + interface = PCAPNG(num=1, sct=1, ctx=context, type=BlockType.Interface_Description_Block, + block={'linktype': LinkType.ETHERNET, 'snaplen': 0x40000}) + context.interfaces.append(interface.info) + return context + + def test_big_endian_section_header_rebuilds(self) -> None: + """#1272: a big-endian SHB keeps its byte order, with or without a context.""" + from pcapkit.const.pcapng.block_type import BlockType + from pcapkit.protocols.misc.pcapng import PCAPNG + + self.assertEqual(self._parse(SHB_BE).info.byteorder, 'big') + self.assertRebuilds(SHB_BE) + built = PCAPNG(num=0, sct=1, ctx=None, type=BlockType.Section_Header_Block, + block={'byteorder': 'big'}) + self.assertEqual(built.data, SHB_BE) + + def test_flags_are_numbered_from_the_least_significant_bit(self) -> None: + """#1273: direction is bits 0-1 and the FCS length bits 5-8 of the word.""" + from pcapkit.const.pcapng.block_type import BlockType + from pcapkit.const.pcapng.option_type import OptionType + from pcapkit.protocols.misc.pcapng import PCAPNG, PacketDirection + + context = self._context() + for kwargs, word in (({'direction': PacketDirection.INBOUND}, '01000000'), + ({'fcs_len': 4}, '80000000')): + with self.subTest(**kwargs): + block = PCAPNG(num=2, sct=1, ctx=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) + + flags = self._parse(_epb_with_options(H('0200 0400 01000000'))).info.options[OptionType.epb_flags] + self.assertEqual(flags.direction, PacketDirection.INBOUND) + self.assertEqual(flags.fcs_len, 0) + + flags = self._parse(_epb_with_options(H('0200 0400 e5010000'))).info.options[OptionType.epb_flags] + self.assertEqual((flags.direction, flags.reception, flags.fcs_len), (1, 1, 15)) + + def test_flags_rebuild_in_either_byte_order(self) -> None: + """#1273: every bit of the word survives, little- and big-endian.""" + octets = _epb_with_options(H('0200 0400 e5fdffff')) + self.assertRebuilds(octets) + octets = _epb_with_options(H('0002 0004 fffffde5'), order='big') + self.assertRebuilds(octets, self._big_endian_context()) + + def test_pack_flags_are_numbered_from_the_least_significant_bit(self) -> None: + """#1273: ``pack_flags`` shares the numbering.""" + import warnings + + from pcapkit.const.pcapng.option_type import OptionType + from pcapkit.protocols.misc.pcapng import PacketDirection + + octets = H('02000000 30000000 0000 0000 00000000 00000000 04000000 04000000 61626364' + '0200 0400 02000000 00000000 30000000') + with warnings.catch_warnings(): + warnings.simplefilter('ignore') # the Packet Block is obsolete + flags = self._parse(octets).info.options[OptionType.pack_flags] + 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 + + with self.assertRaisesRegex(ProtocolError, r'invalid PacketDirection'): + self._parse(_epb_with_options(H('0200 0400 03000000'))) + + def test_nrb_names_decode_as_utf8(self) -> None: + """#1277: record names are UTF-8, not a guessed charset.""" + octets = H('04000000 2c000000 01001800 0a000001 612e6578616d706c6500 622e6578616d706c6500' + '00000000 2c000000') + records = self._parse(octets).info.records + self.assertEqual(records[1].records, ('a.example', 'b.example')) + self.assertRebuilds(octets) + + def test_non_utf8_nrb_name_rebuilds(self) -> None: + """#1277: a name that is not UTF-8 keeps its octets through the split.""" + octets = H('04000000 24000000 01000e00 0a000001 ff2e6578616d706c6500 0000' + '00000000 24000000') + self.assertRebuilds(octets) + + def test_ns_dnsname_decodes_as_utf8(self) -> None: + """#1277: ``ns_dnsname`` is UTF-8, never a guessed charset. + + ``caf\\xe9`` is Latin-1, not UTF-8: a guessing decoder reads it as + ``café``, a UTF-8 one as ``caf\\ufffd`` that still packs back exactly. + """ + from pcapkit.const.pcapng.option_type import OptionType + + for name, text in ((b'caf\xe9.example', 'caf\ufffd.example'), + ('mañana.example'.encode('utf-8'), 'mañana.example')): + with self.subTest(name=name): + option = H('0200') + len(name).to_bytes(2, 'little') + name + bytes(-len(name) % 4) + body = H('00000000') + option + H('00000000') + size = (len(body) + 12).to_bytes(4, 'little') + octets = H('04000000') + size + body + size + self.assertEqual(self._parse(octets).info.options[OptionType.ns_dnsname].name, text) + self.assertRebuilds(octets) + + def test_unknown_block_rebuilds(self) -> None: + """#1278: Block Total Length counts the 12 framing octets.""" + self.assertRebuilds(H('77070000 14000000 0102030405060708 14000000')) + + def test_repeated_epb_verdict_rebuilds(self) -> None: + """#1281: ``epb_verdict`` may appear more than once.""" + verdict = H('0700 0800 0001020304050607') + self.assertRebuilds(_epb_with_options(verdict + verdict)) + + def test_wireguard_error_names_the_line(self) -> None: + """#1286: the error message carries the offending line.""" + from pcapkit.utilities.exceptions import FieldValueError + + body = b'a b c\n\x00\x00' + octets = b''.join(( + (10).to_bytes(4, 'little'), (28).to_bytes(4, 'little'), + (0x57474b4c).to_bytes(4, 'little'), (6).to_bytes(4, 'little'), + body, (28).to_bytes(4, 'little'), + )) + with self.assertRaisesRegex(FieldValueError, r"format: 'a b c'"): + self._parse(octets) + + if __name__ == '__main__': unittest.main() diff --git a/tests/protocols/misc/test_pcapng_unit.py b/tests/protocols/misc/test_pcapng_unit.py index bd9112b9b8..fb84dc353e 100644 --- a/tests/protocols/misc/test_pcapng_unit.py +++ b/tests/protocols/misc/test_pcapng_unit.py @@ -922,6 +922,7 @@ 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 'crc_error': 1, 'too_long': 0, 'too_short': 1, @@ -1896,7 +1897,8 @@ def reset(block_type: BlockType, opt: collections.Counter | None = None) -> None short_address=0x12345678, ).addr_high, 0x1234) - self.assertEqual(pcapng._make_block_unknown(data=b'abc').body, b'abc') + # the body is padded to 32 bits (#1278) + self.assertEqual(pcapng._make_block_unknown(data=b'abc').body, b'abc\x00') shb = pcapng._make_block_shb(version=(1, 0), section_length=-1) self.assertEqual(shb.major, 1) idb = pcapng._make_block_idb(linktype=LinkType.ETHERNET, snaplen=65535) @@ -1927,7 +1929,7 @@ def reset(block_type: BlockType, opt: collections.Counter | None = None) -> None reset(BlockType.Reserved_0x00000000) unknown_data = DataUnknownBlock(type=BlockType.Reserved_0x00000000, length=3, body=b'raw') - self.assertEqual(pcapng._make_block_unknown(unknown_data).body, b'raw') + self.assertEqual(pcapng._make_block_unknown(unknown_data).body, b'raw\x00') # padded (#1278) reset(BlockType.Section_Header_Block) shb_data = DataSectionHeaderBlock( @@ -2286,6 +2288,7 @@ 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 crc_error=True, too_long=True, too_short=False, @@ -2326,7 +2329,7 @@ def assert_duplicate(method: str, code: OptionType, block: BlockType, **kwargs: ('_make_option_epb_flags', OptionType.epb_flags), ('_make_option_epb_dropcount', OptionType.epb_dropcount), ('_make_option_epb_packetid', OptionType.epb_packetid), - ('_make_option_epb_verdict', OptionType.epb_verdict), + # epb_verdict may repeat (#1281), so it has no duplicate guard ]: assert_duplicate(method, code, BlockType.Enhanced_Packet_Block) @@ -2373,6 +2376,7 @@ 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 crc_error=True, too_long=False, too_short=True,