diff --git a/examples/generators/pcapng.py b/examples/generators/pcapng.py index 73c6055cc3..76a70c4f68 100644 --- a/examples/generators/pcapng.py +++ b/examples/generators/pcapng.py @@ -48,64 +48,91 @@ Block types deliberately left out --------------------------------- -Three constructs that :mod:`pcapkit.protocols.misc.pcapng` nominally supports -raise on spec-conformant input, so no fixture here contains them -- a fixture -that cannot be parsed at all is of no use to a regression test: - -* **Custom Block** (``0x00000BAD``/``0x40000BAD``) -- - ``pcapkit/protocols/schema/misc/pcapng.py:1591`` computes its padding as - ``(4 - pkt['data'] % 4) % 4`` where ``pkt['data']`` is :class:`bytes`, so any - custom block raises ``TypeError: not all arguments converted during bytes - formatting``. The ``len()`` call is missing. -* **Packet Block** (obsolete, ``0x00000002``) -- its ``interface_id`` and - ``drop_count`` are declared ``UInt32Field`` at lines 1657 and 1659, but the - format spec makes both 16-bit, and the ``options`` length at line 1674 - subtracts the 32-byte prefix that only the 16-bit layout produces. The field - widths and the arithmetic disagree, so neither layout parses. +Two block types that used to raise on spec-conformant input now parse, so they +are absent from these fixtures only because the fixtures have not been extended +to carry them -- not because pcapkit cannot read them: + +* **Custom Block** (``0x00000BAD``/``0x40000BAD``) -- **parses now.** Its + padding was computed as ``(4 - pkt['data'] % 4) % 4``, i.e. on the + :class:`bytes` object rather than on its length, so any custom block raised + ``TypeError: not all arguments converted during bytes formatting``. The + padding field is gone rather than repaired: the block carries no length for + its custom data, so ``data`` already spans the custom data, its padding *and* + the block's options, which is as much as a reader that does not own the + private enterprise number can tell apart (GitHub issue #341). +* **Packet Block** (obsolete, ``0x00000002``) -- **parses now.** Its + ``interface_id`` and ``drop_count`` were declared ``UInt32Field`` although + Appendix A of draft-ietf-opsawg-pcapng (Figure 19) packs both into one 32-bit + word -- which is the layout the ``options`` length's 32-octet overhead already + assumed. Both are ``UInt16Field`` now, and the reader resolves the link type + from the block's own ``interface_id`` rather than through the ``linktype`` + property, which needs a ``self._info`` that does not exist yet while the block + is still being read and so failed with ``AttributeError: 'PCAPNG' object has + no attribute '_info'`` (GitHub issue #345). The block MUST NOT appear in new + files, so a fixture carrying one would be documenting the past. + +Two further constructs used to be avoided rather than covered, and both are +**fixed** now, so the fixtures carry them as regression cover: + * **``if_IPv6addr``** -- ``IPv6InterfaceField.post_process`` in - ``pcapkit/corekit/fields/ipaddress.py`` reads the trailing prefix-length - octet as ``int(value[16:])``, i.e. it parses a raw byte as an ASCII decimal - string, so a ``/64`` prefix raises ``ValueError``. - -A fourth defect is worked around rather than avoided. A **Simple Packet Block** -in a section declaring more than one interface raises ``FormatError: PCAP-NG: -[SPB] invalid section with 2 interfaces`` at -``pcapkit/foundation/engines/pcapng.py:218``, which tests -``len(interfaces) != 1``. The format specification (section 4.4 of -draft-ietf-opsawg-pcapng) permits it -- "in a Section that has more than one -interface, only packets received or transmitted on the interface described by -the first Interface Description Block can be contained in a Simple Packet -Block" -- so the correct check is for *zero* interfaces, not for more than one. -``test.pcapng`` therefore carries its simple packet block in its -single-interface second section, which keeps the block type covered. - -Three further defects are exercised on purpose, because they warn rather than -raise, and a fixture that covers the code path is what will catch a future -crash there. All three are option-area sizing errors in -``pcapkit/protocols/schema/misc/pcapng.py``, and all three were confirmed by -bisecting one block type at a time: - -* **Interface Statistics Block** -- ``options`` is sized ``length - 20`` at line - 1335, but the block's fixed fields occupy 24 octets, so the option area - over-runs into the trailing block length. Unconditional: an explicit - ``opt_endofopt`` does not save it, because the field still consumes its - declared width. Symptom, on every capture containing one including the - downloaded ``many_interfaces.pcapng``: ``packet length < 0: -8`` and - ``[Block 5] block length mismatch: N != 0``. -* **Name Resolution Block options** -- ``options`` is sized - ``__option_padding__ - 4`` at line 1177, which over-runs whenever ``ns_*`` - options are present. Symptom: ``[Block 4] block length mismatch: 60 != 314``. -* **Name Resolution Block IPv6 records** -- ``resol`` is sized ``length - 4`` at - line 1102, copied from the IPv4 record where the address is four octets wide; - an IPv6 address is sixteen, so the name field over-reads by twelve and - swallows the record terminator and the block's options. Symptom: ``packet - length < 0: -29273``, and the block's ``ns_*`` options vanish. + ``pcapkit/corekit/fields/ipaddress.py`` read the trailing prefix-length octet + as ``int(value[16:])``, i.e. it parsed a raw byte as an ASCII decimal string, + so a ``/64`` prefix raised ``ValueError`` and only the ten lengths 48-57 + parsed at all -- each of them to the wrong value (GitHub issue #346). Every + interface built by :func:`_interface_profile` now carries the option, using + the ``/64`` address from section 4.2 of the format specification. +* **Simple Packet Block** in a multi-interface section -- the engine raised + ``FormatError: PCAP-NG: [SPB] invalid section with 2 interfaces`` for any + simple packet block in a section declaring more than one interface, which the + format specification (section 4.4 of draft-ietf-opsawg-pcapng) permits: it + says only that such a block then refers to the interface described by the + first Interface Description Block, not that it is invalid. The check is for + *zero* interfaces now (GitHub issue #347), so ``test.pcapng`` carries its + simple packet block in the two-interface first section, which is the case + that used to be rejected. + +Three further defects were exercised on purpose, because they warned rather than +raised, and a fixture that covers the code path is what will catch a future +crash there. All three were option-area sizing errors in +``pcapkit/protocols/schema/misc/pcapng.py``, all three were confirmed by +bisecting one block type at a time, and all three are now **fixed** -- so the +fixtures that reached them are regression cover rather than known-bad input: + +* **Interface Statistics Block** -- ``options`` was sized ``length - 20`` though + the block's fixed fields occupy 24 octets, so the option area over-ran into + the trailing block length. Unconditional: an explicit ``opt_endofopt`` did not + save it, because the field still consumed its declared width. Symptom, on + every capture containing one including the downloaded + ``many_interfaces.pcapng``: ``packet length < 0: -8`` and ``[Block 5] block + length mismatch: N != 0``. Now ``length - 24`` (GitHub issue #342). +* **Name Resolution Block options** -- ``options`` was sized + ``__option_padding__ - 4``, which over-ran whenever ``ns_*`` options were + present. Symptom: ``[Block 4] block length mismatch: 60 != 314``. Not a wrong + constant: ``Schema.unpack`` advances the file by each field's *declared* + length, so the octets the option field was trying to size had already been + consumed by ``records``. An ``OptionField`` now hands the remainder it did not + parse back to the file -- the rewind ``ForwardMatchField`` already did -- which + is what makes ``__option_padding__`` mean the same thing to the field that + reads it as to the field that reported it, and the NRB's ``options`` is sized + ``__option_padding__`` (GitHub issue #344). ``ns_dnsname`` also gained the + ``PaddingField`` its ``if_name`` and ``if_description`` siblings have, without + which an unaligned DNS name under-read by its own padding. +* **Name Resolution Block IPv6 records** -- ``resol`` was sized ``length - 4``, + copied from the IPv4 record where the address is four octets wide; an IPv6 + address is sixteen, so the name field over-read by twelve and swallowed the + record terminator and the block's options. Symptom: ``packet length < + 0: -29273``, and the block's ``ns_*`` options vanished. Now ``length - 16`` + (GitHub issue #343). + +``many_interfaces.pcapng`` and ``profile.pcapng`` now extract with neither a +block length mismatch nor a negative packet length; before these fixes each +reported one of each. Every fixture here was checked against an independent PCAP-NG implementation (scapy's ``PcapNgReader``) as well as against pcapkit, and against a structural walk asserting that each block's total length is 4-octet aligned, repeated identically at both ends, and that the block chain covers the file exactly. So -the complaints above are pcapkit's readings, not malformed fixtures. +the complaints above were pcapkit's readings, not malformed fixtures. One complaint is expected rather than a defect: ``test.pcapng`` carries a deliberately truncated packet (``captured_len`` below ``original_len``, to @@ -437,9 +464,9 @@ def _interface_profile(writer: '_Blocks', name: 'str', description: 'str', resolution: 'int' = 6) -> 'list[tuple[int, bytes]]': """The full set of ``if_*`` options pcapkit can parse for one interface. - ``if_IPv6addr`` is the one omission, and it is omitted because pcapkit - raises on it rather than because it does not belong here -- see the module - docstring. + ``if_IPv6addr`` used to be the one omission, because pcapkit raised on it + rather than because it did not belong here; it is included now that the + prefix-length octet is read correctly -- see the module docstring. Args: writer: Writer whose byte order the option values are packed in. @@ -456,6 +483,10 @@ def _interface_profile(writer: '_Blocks', name: 'str', description: 'str', (IF_NAME, name.encode('utf-8')), (IF_DESCRIPTION, description.encode('utf-8')), (IF_IPV4ADDR, octets + bytes([255, 255, 255, 0])), + # The address from section 4.2 of draft-ietf-opsawg-pcapng, whose /64 + # prefix is the case that used to raise: the trailing octet is 0x40, + # which read as an ASCII decimal string is the character ``@``. + (IF_IPV6ADDR, bytes.fromhex('20010db885a308d313198a2e03707344') + bytes([64])), (IF_MACADDR, mac), (IF_EUIADDR, mac[:3] + b'\xff\xfe' + mac[3:]), (IF_SPEED, writer.pack('Q', 1_000_000_000)), @@ -545,11 +576,12 @@ def build_test() -> 'bytes': Two sections, so the section-scoped interface table is exercised too: 1. an Ethernet and a raw-IPv4 interface, three packets between them -- - including one truncated -- then a name resolution block, an interface - statistics block, a decryption secrets block and a journal export block; + including one truncated -- a simple packet block, which belongs here + precisely because the section declares two interfaces, then a name + resolution block, an interface statistics block, a decryption secrets + block and a journal export block; 2. a second section header with its own interface, which must not inherit - anything from the first, one packet, and the simple packet block (see the - comment at the end of this function for why it lives here). + anything from the first, and one packet. """ writer = _Blocks('<') @@ -577,6 +609,12 @@ def build_test() -> 'bytes': blocks.append(writer.epb(0, _timestamp(2_000), packets[2][:32], original_len=len(packets[2]))) + # The simple packet block sits in this section, which declares two + # interfaces, because that is the shape the engine used to reject: the + # format spec says such a block refers to the first interface description + # block, not that it is invalid (GitHub issue #347). + blocks.append(writer.spb(packets[3])) + # Name resolution. The IPv6 record is included knowing pcapkit mis-sizes # it -- covering the path is what catches a future crash there. blocks.append(writer.nrb([ @@ -610,17 +648,6 @@ def build_test() -> 'bytes': ] + _timestamp_options(writer))) blocks.append(writer.epb(0, _timestamp(5_000), packets[0], [(OPT_COMMENT, b'first packet of the second section')])) - - # The simple packet block lives here, in the single-interface section, - # rather than in the first section where it would fit the narrative better. - # pcapkit's PCAP-NG engine raises ``FormatError: PCAP-NG: [SPB] invalid - # section with 2 interfaces`` at ``pcapkit/foundation/engines/pcapng.py:218`` - # for any simple packet block in a section declaring more than one - # interface, which the format spec permits: it says only that such a block - # then refers to the first interface description block, not that it is - # invalid. Putting it here keeps the block type covered without shipping a - # fixture the library cannot open. - blocks.append(writer.spb(packets[3])) return b''.join(blocks) diff --git a/pcapkit/corekit/fields/ipaddress.py b/pcapkit/corekit/fields/ipaddress.py index 08a1e3b80c..35b5a6c678 100644 --- a/pcapkit/corekit/fields/ipaddress.py +++ b/pcapkit/corekit/fields/ipaddress.py @@ -201,6 +201,11 @@ def post_process(self, value: 'bytes', packet: 'dict[str, Any]') -> 'IPv4Interfa Returns: Processed field value. + Notes: + The trailing four octets are a dotted netmask, as written by + :meth:`pre_process` -- not a prefix length as in + :meth:`IPv6InterfaceField.post_process`. + """ ip = ipaddress.IPv4Address(value[:4]) mask = ipaddress.IPv4Address(value[4:]) @@ -264,11 +269,23 @@ def post_process(self, value: 'bytes', packet: 'dict[str, Any]') -> 'IPv6Interfa Returns: Processed field value. + Raises: + FieldValueError: If the trailing octet is not a valid IPv6 prefix + length, i.e. greater than 128. + + Notes: + The trailing octet is the prefix length as a binary integer, as + written by :meth:`pre_process` -- not a dotted netmask as in + :meth:`IPv4InterfaceField.post_process`. + """ ip = ipaddress.IPv6Address(value[:16]) - mask = int(value[16:]) + prefixlen = value[16] - val = ipaddress.ip_interface(f'{ip}/{mask}') + if prefixlen > 128: + raise FieldValueError(f'invalid IPv6 prefix length: {prefixlen}') + + val = ipaddress.ip_interface(f'{ip}/{prefixlen}') if val.version != self.version: raise FieldValueError(f'IP version mismatch: {val.version} != {self.version}') return val diff --git a/pcapkit/foundation/engines/pcapng.py b/pcapkit/foundation/engines/pcapng.py index 913558a330..c6b09bf3a2 100644 --- a/pcapkit/foundation/engines/pcapng.py +++ b/pcapkit/foundation/engines/pcapng.py @@ -20,6 +20,8 @@ __all__ = ['PCAPNG'] if TYPE_CHECKING: + from typing import Optional + from pcapkit.foundation.extraction import Extractor from pcapkit.protocols.data.misc.pcapng import PCAPNG as Data_PCAPNG from pcapkit.protocols.data.misc.pcapng import CustomBlock as Data_CustomBlock @@ -96,6 +98,16 @@ class PCAPNG(Engine[P_PCAPNG]): b'\x0a\x0d\x0d\x0a', ) + #: Block types that carry a captured packet, mapped to the tag used when + #: reporting them. Parsing any of them resolves the interface the packet was + #: captured on, so the enclosing section must describe at least one + #: interface before such a block can be read at all. + PACKET_BLOCK_TYPES = { + Enum_BlockType.Simple_Packet_Block: 'SPB', + Enum_BlockType.Enhanced_Packet_Block: 'EPB', + Enum_BlockType.Packet_Block: 'Packet', + } # type: dict[int, str] + ########################################################################## # Defaults. ########################################################################## @@ -164,6 +176,11 @@ def read_frame(self) -> 'P_PCAPNG': ext = self._extractor while True: + # a packet block resolves the interface it was captured on while it + # is being parsed, so a section that has not described one yet has to + # be rejected before the block is read, not after + self._check_packet_block_context() + # read next block block = P_PCAPNG(ext._ifile, num=ext._frnum+1, sct=len(self._ctx_list), ctx=self._ctx, layer=ext._exlyr, protocol=ext._exptl, @@ -215,8 +232,12 @@ def read_frame(self) -> 'P_PCAPNG': break elif block.info.type == Enum_BlockType.Simple_Packet_Block: - if len(self._ctx.interfaces) != 1: - raise FormatError(f'PCAP-NG: [SPB] invalid section with {len(self._ctx.interfaces)} interfaces') + # an SPB has no interface ID field, so it refers to the interface + # described by the section's first IDB; a section with several + # interfaces is legal and merely means the other interfaces have + # to use an EPB instead (Section 4.4 of the PCAP-NG specification) + if not self._ctx.interfaces: + raise FormatError('PCAP-NG: [SPB] section has no interface description block') break elif block.info.type == Enum_BlockType.Packet_Block: @@ -294,6 +315,50 @@ def _write_file(self, block: 'Data_PCAPNG', *, name: 'str') -> 'None': ofile = ext._ofile ext._offmt = ofile.kind + def _peek_block_type(self) -> 'Optional[int]': + """Read the block type of the next block without consuming it. + + The block type is the first 32-bit field of every block, read in the + byte order declared by the enclosing section header block (SHB). + + Returns: + Block type of the next block, or :obj:`None` if the input file does + not have a whole block type field left to read. + + """ + ext = self._extractor + + buffer = ext._ifile.peek(4)[:4] + if len(buffer) < 4: + return None + return int.from_bytes(buffer, self._ctx.section.byteorder) + + def _check_packet_block_context(self) -> 'None': + """Reject a packet block in a section that describes no interface. + + Every packet block -- Simple Packet Block (SPB), Enhanced Packet Block + (EPB) and the obsolete Packet Block -- resolves the interface it was + captured on while it is being parsed, from the Interface Description + Blocks (IDB) of its section. A section that carries a packet block + before any IDB is therefore unparseable rather than merely invalid, so + the check has to run before the block is read. + + Raises: + FormatError: If the next block is a packet block while the current + section describes no interface. + + """ + if self._ctx.interfaces: + return + + block_type = self._peek_block_type() + if block_type is None: + return + + tag = self.PACKET_BLOCK_TYPES.get(block_type) + if tag is not None: + raise FormatError(f'PCAP-NG: [{tag}] section has no interface description block') + def _get_snaplen(self) -> 'int': """Get snapshot length from the current context. diff --git a/pcapkit/protocols/misc/pcapng.py b/pcapkit/protocols/misc/pcapng.py index b3a247604d..efea7b9316 100644 --- a/pcapkit/protocols/misc/pcapng.py +++ b/pcapkit/protocols/misc/pcapng.py @@ -1914,7 +1914,8 @@ def _read_block_packet(self, schema: 'Schema_PacketBlock', *, original_len=schema.original_length, options=self._read_pcapng_options(schema.options), ) - return self._decode_next_layer(data, self.linktype, schema.captured_length) # type: ignore[return-value] + return self._decode_next_layer(data, self._get_linktype(schema.interface_id), + schema.captured_length) # type: ignore[return-value] def _read_pcapng_options(self, options_schema: 'list[Schema_Option]') -> 'Option': """Read PCAP-NG options. @@ -3705,9 +3706,15 @@ def _make_block_cb(self, block: 'Optional[Data_CustomBlock]' = None, *, else: options_value, _ = [], 0 + # NOTE: ``-n % 4`` is the octet count needed to reach the next 32-bit + # boundary, and zero when already there. Integer arithmetic throughout, + # unlike the ``math.ceil(n / 4) * 4 - n`` idiom used elsewhere in this + # file, which routes a length through a float. cb_data = data.pack() if isinstance(data, Schema) else data + cb_data += bytes(-len(cb_data) % 4) for option in options_value: cb_data += option.pack() if isinstance(option, Schema) else option + cb_data += bytes(-len(cb_data) % 4) return Schema_CustomBlock( length=len(cb_data) + 16, diff --git a/pcapkit/protocols/schema/misc/pcapng.py b/pcapkit/protocols/schema/misc/pcapng.py index 48314a9ab8..2cdbffbd80 100644 --- a/pcapkit/protocols/schema/misc/pcapng.py +++ b/pcapkit/protocols/schema/misc/pcapng.py @@ -1094,12 +1094,12 @@ def __init__(self, type: 'Enum_RecordType', length: 'int', ip: 'IPv4Address | st @schema_final class IPv6Record(NameResolutionRecord, code=Enum_RecordType.nrb_record_ipv6): - """Header schema for PCAP-NG NRB ``nrb_record_ipv4`` records.""" + """Header schema for PCAP-NG NRB ``nrb_record_ipv6`` records.""" - #: IPv4 address. + #: IPv6 address. ip: 'IPv6Address' = IPv6AddressField() #: Name resolution data. - resol: 'str' = StringField(length=lambda pkt: pkt['length'] - 4) + resol: 'str' = StringField(length=lambda pkt: pkt['length'] - 16) #: Padding. padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) @@ -1138,6 +1138,8 @@ class NS_DNSNameOption(_NS_Option, code=Enum_OptionType.ns_dnsname): #: DNS name. name: 'str' = StringField(length=lambda pkt: pkt['length']) + #: Padding. + padding: 'bytes' = PaddingField(length=lambda pkt: (4 - pkt['length'] % 4) % 4) if TYPE_CHECKING: def __init__(self, type: 'Enum_OptionType', length: 'int', name: 'str') -> 'None': ... @@ -1181,7 +1183,7 @@ class NameResolutionBlock(BlockType, code=Enum_BlockType.Name_Resolution_Block): ) #: Options. options: 'list[Option]' = OptionField( - length=lambda pkt: pkt['__option_padding__'] - 4 if pkt['__option_padding__'] else 0, + length=lambda pkt: pkt['__option_padding__'], base_schema=_NS_Option, type_name='type', registry=Option.registry['ns'], @@ -1332,7 +1334,7 @@ class InterfaceStatisticsBlock(BlockType, code=Enum_BlockType.Interface_Statisti timestamp_low: 'int' = UInt32Field(callback=byteorder_callback) #: Options. options: 'list[Option]' = OptionField( - length=lambda pkt: pkt['length'] - 20, + length=lambda pkt: pkt['length'] - 24, base_schema=_ISB_Option, type_name='type', registry=Option.registry['isb'], @@ -1579,16 +1581,23 @@ def __init__(self, length: 'int', secrets_type: 'Enum_SecretsType', @schema_final class CustomBlock(BlockType, code=[Enum_BlockType.Custom_Block_that_rewriters_can_copy_into_new_files, Enum_BlockType.Custom_Block_that_rewriters_should_not_copy_into_new_files]): - """Header schema for PCAP-NG Custom Block (CB).""" + """Header schema for PCAP-NG Custom Block (CB). + + Note: + The block carries no length for its custom data, so where the custom + data ends and the block options begin is known only to the owner of + the private enterprise number. :attr:`data` therefore spans the whole + region between :attr:`pen` and the trailing block total length, i.e. + the custom data, its padding to a 32-bit boundary, and any options. + + """ #: Block total length. length: 'int' = UInt32Field(callback=byteorder_callback) #: Private enterprise number. pen: 'int' = UInt32Field(callback=byteorder_callback) - #: Custom data. + #: Custom data (incl. padding and options). data: 'bytes' = BytesField(length=lambda pkt: pkt['length'] - 16) - #: Padding. - padding: 'bytes' = BytesField(length=lambda pkt: (4 - pkt['data'] % 4) % 4) #: Block total length. length2: 'int' = UInt32Field(callback=byteorder_callback) @@ -1654,9 +1663,9 @@ class PacketBlock(BlockType, code=Enum_BlockType.Packet_Block): #: Block total length. length: 'int' = UInt32Field(callback=byteorder_callback) #: Interface ID. - interface_id: 'int' = UInt32Field(callback=byteorder_callback) + interface_id: 'int' = UInt16Field(callback=byteorder_callback) #: Drops count. - drop_count: 'int' = UInt32Field(callback=byteorder_callback, default=0xFFFF) + drop_count: 'int' = UInt16Field(callback=byteorder_callback, default=0xFFFF) #: Timestamp (high). timestamp_high: 'int' = UInt32Field(callback=byteorder_callback) #: Timestamp (low). diff --git a/pcapkit/protocols/schema/schema.py b/pcapkit/protocols/schema/schema.py index 20465633d7..2d897d64cd 100644 --- a/pcapkit/protocols/schema/schema.py +++ b/pcapkit/protocols/schema/schema.py @@ -580,11 +580,21 @@ def unpack(cls, data: 'bytes | IO[bytes]', of the remaining data, which is used to determine the length of the payload field. - And a ``__padding_length__`` key in the ``packet`` to record the - length of the padding field after an - :class:`~pcapkit.corekit.fields.collections.OptionField`, which - is used to potentially determine the length of the remaining - padding field data. + And an ``__option_padding__`` key in the ``packet`` to record how + much of an + :class:`~pcapkit.corekit.fields.collections.OptionField`'s declared + area it did not consume. What follows that area decides what the + remainder means: usually padding, to be skipped, but a schema may + equally read it as further options, as the PCAP-NG name resolution + block does with its records and options. + + An :class:`~pcapkit.corekit.fields.collections.OptionField` + declares the size of the whole area it may read, but stops at the + end-of-option-list marker and reports the unconsumed remainder + through ``__option_padding__``. Since that remainder has not been + parsed, we rewind ``data`` by it, so that the fields which size + themselves from ``__option_padding__`` read the remainder itself + rather than the same number of octets from beyond it. """ # force cast arg type since decorator changed their signatures @@ -640,6 +650,15 @@ def unpack(cls, data: 'bytes | IO[bytes]', if isinstance(field, ForwardMatchField): data.seek(-field.length, io.SEEK_CUR) + elif isinstance(field, OptionField) and field.option_padding > 0: + # the option list ended before the declared field length was + # exhausted; give the unconsumed remainder back to ``data`` + # so that the following fields can read it + data.seek(-field.option_padding, io.SEEK_CUR) + consumed = field.length - field.option_padding + + self.__buffer__[field.name] = byte[:consumed] + packet['__length__'] -= consumed else: packet['__length__'] -= field.length diff --git a/tests/corekit/test_fields_ipaddress.py b/tests/corekit/test_fields_ipaddress.py new file mode 100644 index 0000000000..cab607e131 --- /dev/null +++ b/tests/corekit/test_fields_ipaddress.py @@ -0,0 +1,156 @@ +from __future__ import annotations + +import importlib.util +import ipaddress +import unittest + +from tests._support import purge_modules + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + +#: Prefix lengths whose octet happens to be an ASCII digit, i.e. ``0x30``--``0x39``. +#: :meth:`IPv6InterfaceField.post_process` used to read the octet as an ASCII +#: decimal string, so these were the only prefix lengths that parsed at all -- +#: and every one of them decoded to ``prefixlen - 48``. Every other prefix length, +#: including all the common ones, raised :exc:`ValueError`. +ASCII_DIGIT_PREFIX_LENGTHS = tuple(range(48, 58)) + +#: Section 4.2 of the PCAP-NG specification: ``if_IPv6addr`` is 17 octets, of +#: which the first 16 are the address and the 17th is the prefix length, so +#: ``2001:0db8:85a3:08d3:1319:8a2e:0370:7344/64`` is written with a trailing ``40``. +SPEC_INTERFACE = '2001:0db8:85a3:08d3:1319:8a2e:0370:7344/64' +SPEC_ENCODING = bytes.fromhex('2001 0db8 85a3 08d3 1319 8a2e 0370 7344 40'.replace(' ', '')) + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class IPAddressFieldTests(unittest.TestCase): + def setUp(self) -> None: + purge_modules(['pcapkit']) + + def test_address_fields_round_trip_and_reject_the_other_version(self) -> None: + from pcapkit.corekit.fields.ipaddress import IPv4AddressField, IPv6AddressField + from pcapkit.utilities.exceptions import FieldValueError + + ipv4 = IPv4AddressField() + self.assertEqual(ipv4.length, 4) + self.assertEqual(ipv4.unpack(ipv4.pack(ipaddress.ip_address('192.0.2.1'), {}), {}), + ipaddress.ip_address('192.0.2.1')) + + ipv6 = IPv6AddressField() + self.assertEqual(ipv6.length, 16) + self.assertEqual(ipv6.unpack(ipv6.pack(ipaddress.ip_address('2001:db8::1'), {}), {}), + ipaddress.ip_address('2001:db8::1')) + + with self.assertRaises(FieldValueError): + ipv4.pre_process('2001:db8::1', {}) + with self.assertRaises(FieldValueError): + ipv6.pre_process('192.0.2.1', {}) + + def test_ipv6_interface_round_trips_every_prefix_length(self) -> None: + from pcapkit.corekit.fields.ipaddress import IPv6InterfaceField + + field = IPv6InterfaceField() + self.assertEqual(field.length, 17) + + for prefixlen in range(129): + interface = ipaddress.ip_interface(f'2001:db8:85a3:8d3:1319:8a2e:370:7344/{prefixlen}') + raw = field.pack(interface, {}) + + self.assertEqual(len(raw), 17, msg=f'/{prefixlen} packed to {len(raw)} octets') + self.assertEqual(raw[:16], interface.ip.packed, + msg=f'/{prefixlen} packed the wrong address') + self.assertEqual(raw[16], prefixlen, + msg=f'/{prefixlen} wrote prefix length octet {raw[16]:#04x}') + self.assertEqual(field.unpack(raw, {}), interface, + msg=f'/{prefixlen} did not survive the round trip') + + def test_ipv6_interface_prefix_length_octet_matches_specification(self) -> None: + from pcapkit.corekit.fields.ipaddress import IPv6InterfaceField + + field = IPv6InterfaceField() + interface = ipaddress.ip_interface(SPEC_INTERFACE) + + self.assertEqual(field.pack(interface, {}), SPEC_ENCODING) + self.assertEqual(field.unpack(SPEC_ENCODING, {}), interface) + self.assertEqual(field.unpack(SPEC_ENCODING, {}).network.prefixlen, 64) + + def test_ipv6_interface_does_not_read_prefix_length_octet_as_an_ascii_digit(self) -> None: + from pcapkit.corekit.fields.ipaddress import IPv6InterfaceField + + field = IPv6InterfaceField() + for prefixlen in ASCII_DIGIT_PREFIX_LENGTHS: + with self.subTest(prefixlen=prefixlen): + interface = ipaddress.ip_interface(f'2001:db8::1/{prefixlen}') + raw = field.pack(interface, {}) + + # the octet really is an ASCII digit, which is why these ten used + # to parse while every other prefix length raised + self.assertIn(raw[16:], [str(digit).encode() for digit in range(10)]) + + parsed = field.unpack(raw, {}) + self.assertEqual(parsed, interface) + self.assertEqual(parsed.network.prefixlen, prefixlen) + # the old ASCII reading decoded /48../57 as /0../9 + self.assertNotEqual(parsed.network.prefixlen, prefixlen - 48) + + def test_ipv6_interface_rejects_out_of_range_prefix_length(self) -> None: + from pcapkit.corekit.fields.ipaddress import IPv6InterfaceField + from pcapkit.utilities.exceptions import FieldValueError + + field = IPv6InterfaceField() + address = ipaddress.IPv6Address('2001:db8::1').packed + + for prefixlen in (129, 200, 255): + with self.subTest(prefixlen=prefixlen): + with self.assertRaises(FieldValueError) as context: + field.unpack(address + bytes([prefixlen]), {}) + self.assertIn(str(prefixlen), str(context.exception)) + + def test_ipv6_interface_rejects_the_other_version(self) -> None: + from pcapkit.corekit.fields.ipaddress import IPv6InterfaceField + from pcapkit.utilities.exceptions import FieldValueError + + with self.assertRaises(FieldValueError): + IPv6InterfaceField().pre_process('192.0.2.1/24', {}) + + def test_ipv4_interface_round_trips_every_prefix_length_as_a_netmask(self) -> None: + from pcapkit.corekit.fields.ipaddress import IPv4InterfaceField + + field = IPv4InterfaceField() + self.assertEqual(field.length, 8) + + for prefixlen in range(33): + interface = ipaddress.ip_interface(f'192.0.2.1/{prefixlen}') + raw = field.pack(interface, {}) + + self.assertEqual(len(raw), 8, msg=f'/{prefixlen} packed to {len(raw)} octets') + self.assertEqual(raw[:4], interface.ip.packed, + msg=f'/{prefixlen} packed the wrong address') + # the IPv4 option carries a dotted netmask, not a prefix length -- + # the two interface fields are deliberately not interchangeable + self.assertEqual(raw[4:], interface.netmask.packed, + msg=f'/{prefixlen} wrote {raw[4:].hex()} instead of a netmask') + self.assertEqual(field.unpack(raw, {}), interface, + msg=f'/{prefixlen} did not survive the round trip') + + def test_interface_field_encodings_are_not_interchangeable(self) -> None: + from pcapkit.corekit.fields.ipaddress import IPv4InterfaceField, IPv6InterfaceField + + ipv4 = IPv4InterfaceField().pack(ipaddress.ip_interface('192.0.2.1/24'), {}) + ipv6 = IPv6InterfaceField().pack(ipaddress.ip_interface('2001:db8::1/24'), {}) + + # /24 as four netmask octets on one side, as a single binary octet on the other + self.assertEqual(ipv4[4:], b'\xff\xff\xff\x00') + self.assertEqual(ipv6[16:], b'\x18') + + def test_ipv4_interface_rejects_the_other_version(self) -> None: + from pcapkit.corekit.fields.ipaddress import IPv4InterfaceField + from pcapkit.utilities.exceptions import FieldValueError + + with self.assertRaises(FieldValueError): + IPv4InterfaceField().pre_process('2001:db8::1/64', {}) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/foundation/engines/test_pcapng_engine.py b/tests/foundation/engines/test_pcapng_engine.py index 28e1c47163..6301417ebc 100644 --- a/tests/foundation/engines/test_pcapng_engine.py +++ b/tests/foundation/engines/test_pcapng_engine.py @@ -1,6 +1,9 @@ from __future__ import annotations import importlib.util +import os +import struct +import tempfile import unittest from unittest import mock @@ -10,6 +13,19 @@ RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) +#: Link layer type 1, i.e. ``LinkType.ETHERNET``. Spelled as a literal because the +#: captures below are assembled without importing :mod:`pcapkit`, which +#: :meth:`setUp` purges from :data:`sys.modules` before every test. +LINKTYPE_ETHERNET = 1 +#: Link layer type 101, i.e. ``LinkType.RAW``. Only used as a *second* interface, +#: to tell which interface a Simple Packet Block was decoded against. +LINKTYPE_RAW = 101 + +#: A minimal Ethernet frame carrying an (all-zero) ARP payload, used as the packet +#: data of every packet block below. Its only job is to decode differently under +#: :data:`LINKTYPE_ETHERNET` than under :data:`LINKTYPE_RAW`. +ETHERNET_FRAME = bytes.fromhex('ffffffffffff001122334455') + b'\x08\x06' + b'\x00' * 28 + class FakeBlock: def __init__(self, info: FakeInfo, *, nanosecond: bool = False) -> None: @@ -18,6 +34,67 @@ def __init__(self, info: FakeInfo, *, nanosecond: bool = False) -> None: self._ctx = None +class PCAPNGWriter: + """Assemble a PCAP-NG capture out of raw blocks. + + Only the blocks that Section 4.4 of the PCAP-NG specification is about are + supported, and none of them carry options -- the point is to build section + shapes that no sample capture provides, such as a Simple Packet Block in a + section with several interfaces, or a packet block in a section with no + Interface Description Block at all. + + """ + + def __init__(self, byteorder: str = 'little') -> None: + self._endian = '<' if byteorder == 'little' else '>' + self._data = b'' + + def __bytes__(self) -> bytes: + return self._data + + @staticmethod + def _pad(body: bytes) -> bytes: + """Pad a block body to a 32-bit boundary.""" + return body + b'\x00' * ((4 - len(body) % 4) % 4) + + def _block(self, block_type: int, body: bytes) -> PCAPNGWriter: + body = self._pad(body) + length = 12 + len(body) + self._data += (struct.pack(f'{self._endian}II', block_type, length) + body + + struct.pack(f'{self._endian}I', length)) + return self + + def section_header(self) -> PCAPNGWriter: + """Section Header Block, with an unspecified section length.""" + return self._block(0x0A0D0D0A, + struct.pack(f'{self._endian}IHHq', 0x1A2B3C4D, 1, 0, -1)) + + def interface_description(self, linktype: int = LINKTYPE_ETHERNET) -> PCAPNGWriter: + """Interface Description Block.""" + return self._block(0x00000001, + struct.pack(f'{self._endian}HHI', linktype, 0, 0x40000)) + + def simple_packet(self) -> PCAPNGWriter: + """Simple Packet Block, which has no interface ID field.""" + return self._block(0x00000003, + struct.pack(f'{self._endian}I', len(ETHERNET_FRAME)) + + self._pad(ETHERNET_FRAME)) + + def enhanced_packet(self, interface_id: int = 0) -> PCAPNGWriter: + """Enhanced Packet Block.""" + return self._block(0x00000006, + struct.pack(f'{self._endian}IIIII', interface_id, 0, 0, + len(ETHERNET_FRAME), len(ETHERNET_FRAME)) + + self._pad(ETHERNET_FRAME)) + + def packet(self, interface_id: int = 0) -> PCAPNGWriter: + """Obsolete Packet Block.""" + return self._block(0x00000002, + struct.pack(f'{self._endian}HHIIII', interface_id, 0, 0, 0, + len(ETHERNET_FRAME), len(ETHERNET_FRAME)) + + self._pad(ETHERNET_FRAME)) + + @unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') class PCAPNGEngineTests(unittest.TestCase): def setUp(self) -> None: @@ -26,6 +103,13 @@ def setUp(self) -> None: def _info(self, block_type, **kwargs) -> FakeInfo: return FakeInfo(type=block_type, **kwargs) + def _section(self, **kwargs) -> FakeInfo: + """Section header block info, which always declares a byte order.""" + from pcapkit.const.pcapng.block_type import BlockType + + kwargs.setdefault('byteorder', 'little') + return self._info(BlockType.Section_Header_Block, **kwargs) + def test_run_validates_section_header_and_writes_context(self) -> None: from pcapkit.const.pcapng.block_type import BlockType from pcapkit.foundation.engines.pcapng import PCAPNG @@ -33,7 +117,7 @@ def test_run_validates_section_header_and_writes_context(self) -> None: extractor, sink = make_extractor() engine = PCAPNG(extractor) - shb = FakeBlock(self._info(BlockType.Section_Header_Block, section='ok')) + shb = FakeBlock(self._section(section='ok')) with mock.patch('pcapkit.foundation.engines.pcapng.P_PCAPNG', side_effect=[shb]): engine.run() @@ -53,7 +137,7 @@ def test_write_file_and_snaplen_helpers_cover_modes(self) -> None: from pcapkit.const.pcapng.block_type import BlockType from pcapkit.foundation.engines.pcapng import Context, PCAPNG - block_info = self._info(BlockType.Section_Header_Block, section='ok') + block_info = self._section(section='ok') extractor, sink = make_extractor(_flag_f=True) engine = PCAPNG(extractor) engine._ctx = Context(block_info) @@ -76,11 +160,11 @@ def test_read_frame_walks_non_packet_blocks_then_enhanced_packet(self) -> None: extractor, sink = make_extractor() engine = PCAPNG(extractor) - engine._ctx = Context(self._info(BlockType.Section_Header_Block, section='initial')) + engine._ctx = Context(self._section(section='initial')) engine._ctx_list = [engine._ctx] blocks = [ - FakeBlock(self._info(BlockType.Section_Header_Block, section='new')), + FakeBlock(self._section(section='new')), FakeBlock(self._info(BlockType.Interface_Description_Block, snaplen=2048)), FakeBlock(self._info(BlockType.Name_Resolution_Block)), FakeBlock(self._info(BlockType.systemd_Journal_Export_Block)), @@ -128,7 +212,7 @@ def test_read_frame_supports_simple_and_deprecated_packet_blocks(self) -> None: extractor, _ = make_extractor(_flag_q=True, _flag_r=False, _flag_t=False, _flag_d=False) engine = PCAPNG(extractor) - engine._ctx = Context(self._info(BlockType.Section_Header_Block)) + engine._ctx = Context(self._section()) engine._ctx.interfaces.append(FakeInfo(snaplen=99)) engine._ctx_list = [engine._ctx] info = self._info(block_type, interface_id=0) @@ -143,7 +227,7 @@ def test_read_frame_covers_none_helper_results_and_disabled_protocol_flags(self) extractor, _ = make_extractor(_flag_q=True, _flag_d=False) engine = PCAPNG(extractor) - engine._ctx = Context(self._info(BlockType.Section_Header_Block)) + engine._ctx = Context(self._section()) engine._ctx.interfaces.append(FakeInfo(snaplen=99)) engine._ctx_list = [engine._ctx] block = FakeBlock(self._info(BlockType.Enhanced_Packet_Block, interface_id=0), @@ -163,7 +247,7 @@ def test_read_frame_covers_none_helper_results_and_disabled_protocol_flags(self) _flag_d=False, _ipv4=False, _ipv6=False, _tcp=False) engine = PCAPNG(no_protocols) - engine._ctx = Context(self._info(BlockType.Section_Header_Block)) + engine._ctx = Context(self._section()) engine._ctx.interfaces.append(FakeInfo(snaplen=99)) engine._ctx_list = [engine._ctx] block = FakeBlock(self._info(BlockType.Enhanced_Packet_Block, interface_id=0)) @@ -193,12 +277,136 @@ def test_read_frame_rejects_invalid_interface_contexts(self) -> None: extractor, _ = make_extractor(_flag_q=True, _flag_r=False, _flag_t=False, _flag_d=False) engine = PCAPNG(extractor) - engine._ctx = Context(self._info(BlockType.Section_Header_Block)) + engine._ctx = Context(self._section()) engine._ctx_list = [engine._ctx] with mock.patch('pcapkit.foundation.engines.pcapng.P_PCAPNG', side_effect=[block]): with self.assertRaises(FormatError): engine.read_frame() + def test_check_packet_block_context_only_fires_for_packet_blocks(self) -> None: + import io + + from pcapkit.foundation.engines.pcapng import Context, PCAPNG + from pcapkit.utilities.exceptions import FormatError + + def engine_for(payload: bytes, *, interfaces: int = 0, byteorder: str = 'little'): + extractor, _ = make_extractor(_ifile=io.BufferedReader(io.BytesIO(payload))) + engine = PCAPNG(extractor) + engine._ctx = Context(self._section(byteorder=byteorder)) + engine._ctx_list = [engine._ctx] + for _ in range(interfaces): + engine._ctx.interfaces.append(FakeInfo(snaplen=99, linktype=LINKTYPE_ETHERNET)) + return engine + + # a non-packet block in a section with no interface yet is perfectly normal: + # that is how every section starts, with its Interface Description Blocks + engine_for(struct.pack('' + engine = engine_for(struct.pack(f'{endian}I', block_type), + byteorder=byteorder) + self.assertEqual(engine._peek_block_type(), block_type) + with self.assertRaises(FormatError) as context: + engine._check_packet_block_context() + self.assertIn(f'[{tag}]', str(context.exception)) + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class PCAPNGSectionRuleTests(unittest.TestCase): + """End-to-end checks of the section rules of Section 4.4 of the PCAP-NG spec. + + These parse synthesised captures rather than sample files, because the section + shapes at issue -- a Simple Packet Block alongside several interfaces, and a + packet block with no interface described at all -- are not among the samples. + + """ + + def setUp(self) -> None: + purge_modules(['pcapkit']) + + def _extract(self, capture: PCAPNGWriter): + from pcapkit.interface import extract + + handle, path = tempfile.mkstemp(suffix='.pcapng') + try: + with os.fdopen(handle, 'wb') as file: + file.write(bytes(capture)) + return extract(fin=path, store=True, nofile=True) + finally: + os.unlink(path) + + def test_simple_packet_block_parses_in_a_multi_interface_section(self) -> None: + for byteorder in ('little', 'big'): + with self.subTest(byteorder=byteorder): + capture = (PCAPNGWriter(byteorder).section_header() + .interface_description() + .interface_description() + .simple_packet()) + extractor = self._extract(capture) + + self.assertEqual(len(extractor.frame), 1) + self.assertEqual(len(extractor.engine._ctx.interfaces), 2) + + def test_simple_packet_block_mixes_with_enhanced_packet_blocks(self) -> None: + # Section 4.4: packets on any interface other than the first have to use an EPB, + # which is what a real multi-interface section looks like + capture = (PCAPNGWriter().section_header() + .interface_description() + .interface_description() + .interface_description() + .simple_packet() + .enhanced_packet(interface_id=2)) + extractor = self._extract(capture) + + self.assertEqual(len(extractor.frame), 2) + self.assertEqual(len(extractor.engine._ctx.interfaces), 3) + + def test_simple_packet_block_decodes_against_the_first_interface(self) -> None: + # Section 4.4: an SPB has no interface ID field, so it refers to the interface + # described by the section's *first* IDB + ethernet_first = (PCAPNGWriter().section_header() + .interface_description(LINKTYPE_ETHERNET) + .interface_description(LINKTYPE_RAW) + .simple_packet()) + raw_first = (PCAPNGWriter().section_header() + .interface_description(LINKTYPE_RAW) + .interface_description(LINKTYPE_ETHERNET) + .simple_packet()) + + self.assertIn('Ethernet', str(self._extract(ethernet_first).frame[0].protochain)) + self.assertNotIn('Ethernet', str(self._extract(raw_first).frame[0].protochain)) + + def test_packet_blocks_without_an_interface_description_are_format_errors(self) -> None: + from pcapkit.utilities.exceptions import FormatError + + blocks = { + 'SPB': lambda writer: writer.simple_packet(), + 'EPB': lambda writer: writer.enhanced_packet(), + 'Packet': lambda writer: writer.packet(), + } + + for byteorder in ('little', 'big'): + for tag, add_block in blocks.items(): + with self.subTest(byteorder=byteorder, tag=tag): + capture = add_block(PCAPNGWriter(byteorder).section_header()) + with self.assertRaises(FormatError) as context: + self._extract(capture) + + message = str(context.exception) + self.assertIn(f'PCAP-NG: [{tag}]', message) + self.assertIn('interface description block', message) + if __name__ == '__main__': unittest.main() diff --git a/tests/foundation/engines/test_runtime_engines.py b/tests/foundation/engines/test_runtime_engines.py index 62e07b236a..37489321a5 100644 --- a/tests/foundation/engines/test_runtime_engines.py +++ b/tests/foundation/engines/test_runtime_engines.py @@ -33,7 +33,9 @@ def make_extractor(**overrides): reasm = types.SimpleNamespace(ipv4=mock.Mock(), ipv6=mock.Mock(), tcp=mock.Mock()) trace = types.SimpleNamespace(tcp=mock.Mock()) values = { - '_ifile': io.BytesIO(b'capture'), + # ``Extractor`` always hands the engines a buffered (peekable) reader, + # which the PCAP-NG engine relies on to look at the next block type + '_ifile': io.BufferedReader(io.BytesIO(b'capture')), '_ifnm': 'capture.pcap', '_ofile': sink, '_ofnm': 'out', diff --git a/tests/protocols/misc/test_pcapng_unit.py b/tests/protocols/misc/test_pcapng_unit.py index ec8d50cfcb..6ba4eddddc 100644 --- a/tests/protocols/misc/test_pcapng_unit.py +++ b/tests/protocols/misc/test_pcapng_unit.py @@ -7,6 +7,7 @@ import decimal import io from ipaddress import ip_address, ip_interface +import struct import types import unittest from unittest import mock @@ -26,6 +27,30 @@ def __update__(self, *values, **kwargs): self.update(kwargs) +def pad32(value: bytes) -> bytes: + """Pad ``value`` with zeroes up to the next 32-bit boundary.""" + return value + bytes(-len(value) % 4) + + +def tlv(code: int, value: bytes) -> bytes: + """Build a little-endian PCAP-NG option or NRB record, padded to 32 bits.""" + return struct.pack(' bytes: + """Wrap ``body`` in the two block total length fields of a PCAP-NG block. + + The returned buffer is what a block schema is handed by + :class:`~pcapkit.protocols.schema.misc.pcapng.PCAPNG`, i.e. the block + without its leading 4-octet block type, but *with* the block total length + at either end. The length itself counts the block type as well, so it is + the length of the returned buffer plus four. + + """ + length = len(body) + 12 + return struct.pack(' None: @@ -2179,6 +2204,208 @@ def custom_secrets_constructor(code, secrets=None, *, data=b's', **kwargs): DummyData(aps_key=b'\x04' * 16, pan_id=4, short_address=0x56789ABC), ).addr_low, 0x9ABC) + def test_pcapng_custom_block_data_spans_padding_and_options(self) -> None: + from pcapkit.protocols.misc.pcapng import PCAPNG + from pcapkit.protocols.schema.misc.pcapng import CustomBlock + + # A Custom Block carries no length for its custom data, so everything + # between the private enterprise number and the trailing block total + # length -- the custom data, its padding, and the block options -- is + # ``data``, and there is no separate padding field to size. + custom = pad32(b'hello') + options = tlv(0, b'') # opt_endofopt + raw = block_body(struct.pack(' None: + from pcapkit.const.pcapng.option_type import OptionType + from pcapkit.protocols.schema.misc.pcapng import InterfaceStatisticsBlock + + # The ISB's fixed fields occupy 24 octets: block type, block total + # length, interface ID, the 64-bit timestamp, and the trailing block + # total length. Sizing the option area as ``length - 20`` lets the + # options field read the trailing block total length as if it were an + # option, so the block is consumed 8 octets past its end. + options = tlv(2, struct.pack(' None: + from pcapkit.const.pcapng.option_type import OptionType + from pcapkit.protocols.schema.misc.pcapng import InterfaceStatisticsBlock + + # An option list that ends at ``opt_endofopt`` before the option area + # does leaves a remainder, reported through ``__option_padding__``. + # The padding field which sizes itself from it has to read *that* + # remainder, not the same number of octets from beyond the options. + options = tlv(2, struct.pack(' None: + from pcapkit.const.pcapng.record_type import RecordType + from pcapkit.protocols.schema.misc.pcapng import NameResolutionBlock + + # An ``nrb_record_ipv6`` record value is a 16-octet address followed by + # zero-terminated names, so the names span ``length - 16``. Sizing them + # as ``length - 4`` -- the IPv4 record's arithmetic -- over-reads by 12 + # octets and swallows whatever record follows. + v6_value = ip_address('2001:db8::1').packed + b'host6.example\x00' + v4_value = ip_address('192.0.2.1').packed + b'host4.example\x00' + records = tlv(2, v6_value) + tlv(1, v4_value) + tlv(0, b'') + raw = block_body(records) + + schema = NameResolutionBlock.unpack(raw, len(raw), {'byteorder': 'little'}) + + self.assertEqual(schema.length, schema.length2) + self.assertEqual([record.type for record in schema.records], + [RecordType.nrb_record_ipv6, + RecordType.nrb_record_ipv4, + RecordType.nrb_record_end]) + self.assertEqual(len(schema.records[0]), 4 + len(pad32(v6_value))) + self.assertEqual(schema.records[0].ip, ip_address('2001:db8::1')) + self.assertEqual(schema.records[0].names, ['host6.example']) + self.assertEqual(schema.records[1].ip, ip_address('192.0.2.1')) + self.assertEqual(schema.records[1].names, ['host4.example']) + self.assertEqual(schema.mapping.getlist(ip_address('2001:db8::1')), ['host6.example']) + self.assertEqual(len(schema), len(raw)) + + def test_pcapng_nrb_options_read_from_inside_the_block(self) -> None: + from pcapkit.const.pcapng.option_type import OptionType + from pcapkit.const.pcapng.record_type import RecordType + from pcapkit.protocols.schema.misc.pcapng import NameResolutionBlock + + # The NRB is the only block with two consecutive option fields: the + # record area's size is known only once its records have been parsed, + # so ``records`` is declared over the record *and* option areas and the + # options have to be read from the remainder it did not consume. + v4_value = ip_address('192.0.2.1').packed + b'host4.example\x00' + records = tlv(1, v4_value) + tlv(0, b'') + + for name in ('dns.example.', 'dns.example'): + with self.subTest(dnsname=name): + options = tlv(2, name.encode()) + tlv(0, b'') # ns_dnsname, opt_endofopt + raw = block_body(records + options) + + schema = NameResolutionBlock.unpack(raw, len(raw), {'byteorder': 'little'}) + + self.assertEqual(schema.length, schema.length2) + self.assertEqual(schema.length, len(raw) + 4) + self.assertEqual([record.type for record in schema.records], + [RecordType.nrb_record_ipv4, RecordType.nrb_record_end]) + self.assertEqual([option.type for option in schema.options], + [OptionType.ns_dnsname, OptionType.opt_endofopt]) + self.assertEqual(schema.options[0].name, name) + # the option is padded to a 32-bit boundary whether or not its + # value length happens to be a multiple of four + self.assertEqual(len(schema.options[0]), 4 + len(pad32(name.encode()))) + self.assertEqual(schema.padding, b'') + self.assertEqual(len(schema), len(raw)) + + def test_pcapng_obsolete_packet_block_uses_sixteen_bit_ids(self) -> None: + from pcapkit.const.pcapng.option_type import OptionType + from pcapkit.protocols.schema.misc.pcapng import PacketBlock + + # The obsolete Packet Block packs Interface ID and Drops Count into one + # 32-bit word, which is what the option area's ``length - 32`` overhead + # assumes; declaring them as 32-bit fields reads every later field from + # four octets too far in. + packet_data = bytes.fromhex('ffffffffffff001122334455') + b'\x08\x06' + bytes(28) + options = tlv(2, struct.pack(' None: + from pcapkit.const.pcapng.block_type import BlockType + from pcapkit.const.reg.linktype import LinkType + from pcapkit.protocols.misc.pcapng import PCAPNG + from pcapkit.protocols.schema.misc.pcapng import PacketBlock, PCAPNG as Header + + # ``_read_block_packet`` runs before ``self._info`` exists, so it has to + # resolve the link type from the schema's interface ID the way the EPB + # and SPB readers do, rather than through the ``linktype`` property. + pcapng = object.__new__(PCAPNG) + pcapng._sect = 1 + pcapng._fnum = 2 + pcapng._opt = collections.Counter() + pcapng._type = BlockType.Packet_Block + pcapng._ctx = types.SimpleNamespace( + interfaces=[types.SimpleNamespace(linktype=LinkType.ETHERNET)], + ) + pcapng._read_timestamp = lambda high, low, interface_id=0: ( + datetime.datetime.fromtimestamp(0, datetime.timezone.utc), + decimal.Decimal(0), + ) + decoded = [] + pcapng._decode_next_layer = lambda data, proto=None, length=None, packet=None: ( + decoded.append((proto, length)) or data + ) + self.assertFalse(hasattr(pcapng, '_info')) + + with mock.patch('pcapkit.protocols.misc.pcapng.warn'): + block = pcapng._read_block_packet( + PacketBlock(length=36, interface_id=0, drop_count=1, timestamp_high=0, + timestamp_low=0, captured_length=4, original_length=4, + packet_data=b'data', options=[], length2=36), + header=Header(type=BlockType.Packet_Block, block=b''), + ) + + self.assertEqual(block.drop_count, 1) + self.assertEqual(decoded, [(LinkType.ETHERNET, 4)]) + if __name__ == '__main__': unittest.main()