From 0783176aa592ff639760b77feaf9bd150c6d47af Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Tue, 13 Oct 2026 15:04:58 -0400 Subject: [PATCH 1/2] fix(toolkit): pypcapfile reads IP over loopback and raw-IP link types, up to the frame layer bound (#1609, #1667) - pypcapfile decodes Ethernet only, so the adapters handed no TCP, trace or IPv4 input over on NULL, LOOP, RAW, IPV4 and IPV6. They now read the IP packet there themselves (READ_LINKTYPES): the loopback family by Loopback's table and family_byteorder, RAW by its version nibble. TCP over IPv6 stays declined with the per-capture warning; 12 and LINUX_SLL stay unread, and only those still get the engine's "unrecognised link layer protocol" warning. - The VLAN, tunnel and extension-header walks ran past FRAME_LAYER_LIMIT. Each VLAN tag, IP header, AH and IPv6 extension header now counts one layer, as in the default engine, and nothing past layer 64 is read or warned of. _upper_layer, which pyshark imports, is unchanged. - packet2dict reported an unread frame's hex text (raw_len 94 for 47 octets); it now reports the octets. New tests fail on main and pass here on 3.11 and 3.14. Closes #1667 Part of #1609 --- pcapkit/foundation/engines/pypcapfile.py | 34 +- pcapkit/toolkit/pypcapfile.py | 382 ++++++++++++++---- .../engines/test_pypcapfile_engine.py | 16 + ...nd_layer_bound_agreement_1609_1667_unit.py | 309 ++++++++++++++ ...st_pypcapfile_ext_chain_limit_1604_unit.py | 8 +- .../test_pypcapfile_header_checks_unit.py | 4 +- ...inktypes_and_layer_bound_1609_1667_unit.py | 345 ++++++++++++++++ tests/toolkit/test_pypcapfile_unit.py | 13 +- 8 files changed, 1011 insertions(+), 100 deletions(-) create mode 100644 tests/foundation/engines/test_pypcapfile_linktype_and_layer_bound_agreement_1609_1667_unit.py create mode 100644 tests/toolkit/test_pypcapfile_linktypes_and_layer_bound_1609_1667_unit.py diff --git a/pcapkit/foundation/engines/pypcapfile.py b/pcapkit/foundation/engines/pypcapfile.py index 6bb97b2f87..24808c608e 100644 --- a/pcapkit/foundation/engines/pypcapfile.py +++ b/pcapkit/foundation/engines/pypcapfile.py @@ -47,10 +47,15 @@ class PyPCAPFile(EngineBase['PCAPFilePacket']): objects, as `PyPCAPFile`_ 0.12.0 drops records the ``default`` engine reads -- see :meth:`_read_records`, * disables IPv6 reassembly, warning as it does so, while leaving IPv4 and - TCP reassembly and TCP flow tracing in place, and + TCP reassembly and TCP flow tracing in place, * leaves TCP over IPv6 out of TCP reassembly and flow tracing, with one :exc:`~pcapkit.utilities.warnings.AttributeWarning` per capture -- see - :mod:`pcapkit.toolkit.pypcapfile`. + :mod:`pcapkit.toolkit.pypcapfile` -- and + * leaves the frames of a link type `PyPCAPFile`_ has no decoder for + undecoded, from which :mod:`pcapkit.toolkit.pypcapfile` reads the IP + packet of the BSD loopback and raw-IP ones itself, as the ``default`` + engine dissects it (see + :data:`~pcapkit.toolkit.pypcapfile.READ_LINKTYPES`). The engine stops decoding at the network layer rather than descending into the transport layer. `PyPCAPFile`_ decoders *replace* the payload bytes of @@ -184,8 +189,10 @@ def run(self) -> 'None': support such operations. * if IPv6 reassembly is enabled, as :mod:`pcapfile` has no IPv6 decoder. - * if :mod:`pcapfile` has no decoder for the savefile's link layer - type, in which case frames are left undecoded. + * if the savefile's link layer type is left unread: neither + :mod:`pcapfile` has a decoder for it nor + :mod:`pcapkit.toolkit.pypcapfile` reads it (see + :meth:`_get_decoder`). Raises: FormatError: If the file format is not supported, i.e., not a PCAP @@ -302,20 +309,27 @@ def _get_decoder(self, linktype: 'int') -> 'Optional[Callable[..., Any]]': The decoder class, or :data:`None` when :mod:`pcapfile` has none. Warns: - AttributeWarning: If no decoder is available, as frames will then be - left hexlified -- see :meth:`run` -- and no reassembly or flow - tracing is possible. + AttributeWarning: If no decoder is available and the link layer type + is none of the :data:`~pcapkit.toolkit.pypcapfile.READ_LINKTYPES`, + as frames will then be left hexlified -- see :meth:`run` -- and + no reassembly or flow tracing is possible. The frames of those + link types are left undecoded too, but reassembly and flow + tracing read the IP packet in them, as the ``default`` engine + does, so nothing is lost there to warn of. """ + from pcapkit.toolkit.pypcapfile import READ_LINKTYPES # isort:skip + try: decoder = self._expkg.linklayer.clookup(linktype) except IndexError: # malformed entry in ``pcapfile.linklayer.__LL_TYPES__`` decoder = None if not callable(decoder): - warn(f'unrecognised link layer protocol: {self._dlink!r}; frames will be left ' - 'undecoded and all analysis functions ignored', AttributeWarning, - stacklevel=stacklevel()) + if linktype not in READ_LINKTYPES: + warn(f'unrecognised link layer protocol: {self._dlink!r}; frames will be left ' + 'undecoded and all analysis functions ignored', AttributeWarning, + stacklevel=stacklevel()) return None return decoder diff --git a/pcapkit/toolkit/pypcapfile.py b/pcapkit/toolkit/pypcapfile.py index 4f1ff332fb..3bca95d9ca 100644 --- a/pcapkit/toolkit/pypcapfile.py +++ b/pcapkit/toolkit/pypcapfile.py @@ -42,6 +42,17 @@ :func:`_ipv4_in_ipv6`), and TCP behind AH or an IPv6 extension header inside IPv4 is read past them (see :func:`_transport`). + `PyPCAPFile`_ decodes no link layer but Ethernet. The default engine also + dissects the IP packet of the BSD loopback and raw-IP link types, so the + adapters read it there themselves (see :data:`READ_LINKTYPES`, + :issue:`1667`), and leave the rest of the link types unread, as it does. + + Nor does it bound how deep it reads. The default engine dissects at most + :data:`~pcapkit.protocols.protocol.FRAME_LAYER_LIMIT` layers of a frame, + each VLAN tag, IP header, AH and IPv6 extension header one of them, so the + adapters read nothing past that either (see :func:`_walk_upper_layer`, + :issue:`1609`). + `PyPCAPFile`_ checks neither that an IPv4 header was captured whole nor its options, and the default engine rejects a header that fails either check, so such a header is left out of reassembly and flow tracing alike (see @@ -75,6 +86,8 @@ from pcapkit.foundation.traceflow.data.tcp import Packet as TF_TCP_Packet from pcapkit.protocols.internet.internet import EXTENSION_HEADER_LIMIT from pcapkit.protocols.internet.ipv4 import IPv4 as Protocol_IPv4 +from pcapkit.protocols.link.loopback import family_byteorder +from pcapkit.protocols.protocol import FRAME_LAYER_LIMIT from pcapkit.protocols.schema.internet.ipv6 import jumbo_payload_length from pcapkit.protocols.transport.tcp import TCP as Protocol_TCP from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall, stacklevel @@ -146,6 +159,27 @@ #: then the EtherType of what follows the tag. VLAN_TAG_LEN = 4 +#: Link types `PyPCAPFile`_ has no decoder for, whose IP packet the default +#: engine dissects all the same (see :attr:`Frame.__proto__ +#: `), and which the +#: adapters therefore read themselves (see :func:`_link_network`, +#: :issue:`1667`): BSD loopback, ``NULL`` and ``LOOP``, and raw IP, ``RAW``, +#: ``IPV4`` and ``IPV6``. Every other link type but ``ETHERNET``, which +#: `PyPCAPFile`_ decodes, is left unread, as the default engine leaves it raw. +READ_LINKTYPES = frozenset(( + Enum_LinkType.NULL, Enum_LinkType.LOOP, Enum_LinkType.RAW, + Enum_LinkType.IPV4, Enum_LinkType.IPV6, +)) + +#: Length of a BSD loopback header: the address family. +LOOPBACK_HEADER_LEN = 4 + +#: Address families of the IP packets the default engine dissects behind a BSD +#: loopback header, each mapped to its IP version (see :attr:`Loopback.__proto__ +#: `): ``AF_INET``, and the +#: ``AF_INET6`` of each of the BSDs. +LOOPBACK_IP_VERSIONS = {2: 4, 24: 6, 28: 6, 30: 6} # type: dict[int, int] + #: IP protocol numbers of the IP-in-IP tunnels the default engine dissects #: (see :attr:`Internet.__proto__ #: `), each mapped to the @@ -383,17 +417,21 @@ def _is_raw(layer: 'Any') -> 'bool': return isinstance(layer, (bytes, bytearray, memoryview)) -def _untag(link: 'Any') -> 'Optional[tuple[int, bytes]]': +def _untag(link: 'Any') -> 'Optional[tuple[int, bytes, int]]': """Read past the VLAN tags of an Ethernet frame whose payload is undecoded. Args: link: Decoded PyPCAPFile link layer. Returns: - The EtherType after the last of the tags in :data:`VLAN_TAGS` and the - raw octets it heads -- for an untagged frame, the frame's own EtherType - and payload. :data:`None` if `PyPCAPFile`_ decoded the payload, or a tag - is cut short. + The EtherType after the last of the tags in :data:`VLAN_TAGS`, the raw + octets it heads, and their place in the frame (c.f. + :data:`~pcapkit.protocols.protocol.FRAME_LAYER_LIMIT`), the Ethernet + header being the first layer and each tag one more -- for an untagged + frame, the frame's own EtherType and payload, and ``2``. :data:`None` if + `PyPCAPFile`_ decoded the payload, a tag is cut short, or what follows + the tags is past that bound, where the default engine dissects nothing + (:issue:`1609`). Note: `PyPCAPFile`_ leaves a payload it has no decoder for hex-encoded (see @@ -407,12 +445,12 @@ def _untag(link: 'Any') -> 'Optional[tuple[int, bytes]]': if ethertype is None or not _is_raw(payload): return None - data = _maybe_unhex(bytes(payload)) + data, depth = _maybe_unhex(bytes(payload)), 2 while ethertype in VLAN_TAGS: - if len(data) < VLAN_TAG_LEN: + if depth == FRAME_LAYER_LIMIT or len(data) < VLAN_TAG_LEN: return None - ethertype, data = int.from_bytes(data[2:4], 'big'), data[VLAN_TAG_LEN:] - return ethertype, data + ethertype, data, depth = int.from_bytes(data[2:4], 'big'), data[VLAN_TAG_LEN:], depth + 1 + return ethertype, data, depth @functools.lru_cache(maxsize=2) @@ -478,19 +516,87 @@ def _ipv4_accepted(ipv4: 'IP') -> 'bool': return _default_accepts('IPv4', ipv4_header(ipv4)) +def _link_network(packet: 'Packet') -> 'Optional[tuple[int, bytes, int]]': + """Find the IP packet the default engine dissects in a frame `PyPCAPFile`_ has no decoder for. + + Args: + packet: PyPCAPFile packet, whose frame `PyPCAPFile`_ left undecoded. + + Returns: + The packet's IP version, its octets on to the end of the frame, and its + place in the frame (c.f. + :data:`~pcapkit.protocols.protocol.FRAME_LAYER_LIMIT`), by the + capture's link type, if it is one of the :data:`READ_LINKTYPES`: ``2`` + behind a BSD loopback header whose address family is one of the + :data:`LOOPBACK_IP_VERSIONS`, read in the byte order + :func:`~pcapkit.protocols.link.loopback.family_byteorder` gives -- + ``LOOP``'s in network byte order, ``NULL``'s in the order its octets + show; and ``1`` for an ``IPV4`` or ``IPV6`` frame, or a ``RAW`` one of + the IP version its first octet gives, 4 or 6. :data:`None` for any other + link type, family or version, a frame cut short of its loopback header, + and an empty one, where the default engine dissects no IP either. + + Note: + The frame is hex-encoded, as the engine reads it (see + :meth:`PyPCAPFile._read_records + `), and + the link type is the savefile header's, which the packet points at. + + .. _PyPCAPFile: https://github.com/kisom/pypcapfile + + """ + linktype = packet.header[0].ll_type + if linktype not in READ_LINKTYPES: + return None + octets = _maybe_unhex(bytes(packet.packet)) + if linktype in (Enum_LinkType.NULL, Enum_LinkType.LOOP): + if len(octets) < LOOPBACK_HEADER_LEN: + return None + family = octets[:LOOPBACK_HEADER_LEN] + version = LOOPBACK_IP_VERSIONS.get(int.from_bytes(family, family_byteorder(family, linktype))) + return None if version is None else (version, octets[LOOPBACK_HEADER_LEN:], 2) + if linktype == Enum_LinkType.RAW: + version = octets[0] >> 4 if octets else 0 + if version not in (4, 6): + return None + else: + version = 4 if linktype == Enum_LinkType.IPV4 else 6 + return version, octets, 1 + + def _network(packet: 'Packet', count: 'int' = -1) -> 'Optional[IP]': - """Fetch the decoded IPv4 layer of a PyPCAPFile packet, if any. + """Fetch the outermost IPv4 layer of a PyPCAPFile packet, if any. + + Args: + packet: PyPCAPFile packet. + count: Packet index, for the warning :func:`_network_layer` gives. + + Returns: + The IPv4 layer :func:`_network_layer` fetches, or :data:`None`. + + """ + layer = _network_layer(packet, count) + return None if layer is None else layer[0] + + +def _network_layer(packet: 'Packet', count: 'int' = -1) -> 'Optional[tuple[IP, int]]': + """Fetch the outermost IPv4 layer of a PyPCAPFile packet, and its place in the frame. Args: packet: PyPCAPFile packet. count: Packet index, for the warning below. Returns: - The decoded :class:`~pcapfile.protocols.network.ip.IP` layer, or - :data:`None` when the frame was not decoded that far -- which is the - case for every non-IPv4 frame, `PyPCAPFile`_ having no other network - layer decoder -- or the default engine rejects its header (see - :func:`_ipv4_accepted`). + The decoded :class:`~pcapfile.protocols.network.ip.IP` layer and its + place in the frame (c.f. + :data:`~pcapkit.protocols.protocol.FRAME_LAYER_LIMIT`), or :data:`None` + when the frame carries no IPv4 right after its link layer -- + `PyPCAPFile`_ having no other network layer decoder -- or the default + engine rejects its header (see :func:`_ipv4_accepted`). Behind an + Ethernet header that is ``2``, and one more for each VLAN tag (see + :func:`_untag`). A frame of a link type `PyPCAPFile`_ has no decoder for + is read by :func:`_link_network`, and its IPv4 packet decoded by + :func:`_decode_ipv4`. Warns: AttributeWarning: If an IPv4 packet behind VLAN tags will not decode, @@ -508,18 +614,23 @@ def _network(packet: 'Packet', count: 'int' = -1) -> 'Optional[IP]': untagged frame, so the two decode alike; one that will not decode is left out with a warning, as an untagged one is. The engine warns of an untagged frame as it reads it, and this of a tagged one as reassembly or - flow tracing does. + flow tracing does. Over the other link types the engine decodes nothing, + so has nothing to warn of. .. _PyPCAPFile: https://github.com/kisom/pypcapfile """ link = packet.packet if _is_raw(link): - return None + found = _link_network(packet) + if found is None or found[0] != 4: + return None + decoded = _decode_ipv4(found[1]) + return None if decoded is None else (decoded, found[2]) payload = getattr(link, 'payload', None) if type(payload).__name__ == 'IP': - return payload if _ipv4_accepted(payload) else None + return (payload, 2) if _ipv4_accepted(payload) else None if getattr(link, 'type', None) not in VLAN_TAGS: return None @@ -542,7 +653,33 @@ def _network(packet: 'Packet', count: 'int' = -1) -> 'Optional[IP]': warn(f'Frame {count}: {Enum_LinkType.ETHERNET!r} decoding failed ({error!r}); ' 'frame left undecoded', AttributeWarning, stacklevel=stacklevel()) return None - return ipv4 if _ipv4_accepted(ipv4) else None + return (ipv4, untagged[2]) if _ipv4_accepted(ipv4) else None + + +def _outer_ipv6(packet: 'Packet') -> 'Optional[tuple[bytes, int]]': + """Fetch the outermost IPv6 packet of a PyPCAPFile frame, which `PyPCAPFile`_ cannot decode. + + Args: + packet: PyPCAPFile packet. + + Returns: + The octets of the IPv6 packet right after the frame's link layer, on to + the end of the frame, and its place in the frame: behind an Ethernet + header and any VLAN tags (see :func:`_untag`), or in a frame of a link + type `PyPCAPFile`_ has no decoder for (see :func:`_link_network`). + :data:`None` if there is none. + + .. _PyPCAPFile: https://github.com/kisom/pypcapfile + + """ + link = packet.packet + if _is_raw(link): + found = _link_network(packet) + return None if found is None or found[0] != 6 else (found[1], found[2]) + untagged = _untag(link) + if untagged is None or untagged[0] != Enum_EtherType.Internet_Protocol_version_6: + return None + return untagged[1], untagged[2] def _ipv4_in_ipv6(packet: 'Packet') -> 'Optional[IP]': @@ -556,9 +693,11 @@ def _ipv4_in_ipv6(packet: 'Packet') -> 'Optional[IP]': extension headers, or inside more IPv6 (6in6) -- decoded as :func:`_network` decodes one behind VLAN tags, or :data:`None` if there is none, it will not decode, or the default engine rejects its header - (see :func:`_ipv4_accepted`). The frame may be tagged (see - :func:`_untag`), and each IPv6 header is stepped over as - :func:`_upper_layer` does. + (see :func:`_ipv4_accepted`). The outer IPv6 packet is the one + :func:`_outer_ipv6` fetches, and each IPv6 header is stepped over as + :func:`_walk_upper_layer` does, so no IPv4 packet past + :data:`~pcapkit.protocols.protocol.FRAME_LAYER_LIMIT` layers of the + frame is found (:issue:`1609`). Note: `PyPCAPFile`_ has no IPv6 decoder, so leaves the IPv4 inside undecoded. @@ -571,25 +710,25 @@ def _ipv4_in_ipv6(packet: 'Packet') -> 'Optional[IP]': .. _PyPCAPFile: https://github.com/kisom/pypcapfile """ - link = packet.packet - untagged = None if _is_raw(link) else _untag(link) - if untagged is None or untagged[0] != Enum_EtherType.Internet_Protocol_version_6: + outer = _outer_ipv6(packet) + if outer is None: return None - upper = _upper_layer(6, untagged[1]) + upper = _walk_upper_layer(6, *outer) while upper is not None and upper[0] == Enum_TransType.IPv6: - upper = _upper_layer(6, upper[1]) + upper = _walk_upper_layer(6, upper[1], upper[2]) if upper is None or upper[0] != Enum_TransType.IPv4: return None return _decode_ipv4(upper[1]) def _decode_ipv4(data: 'bytes') -> 'Optional[IP]': - """Decode an IPv4 packet that `PyPCAPFile`_ left undecoded inside another IP packet. + """Decode an IPv4 packet that `PyPCAPFile`_ left undecoded. Args: data: Raw octets from the IPv4 header on, as far as the IP packet - carrying it goes (see :func:`_upper_layer`). + carrying it goes (see :func:`_upper_layer`), or the frame of a link + type `PyPCAPFile`_ has no decoder for (see :func:`_link_network`). Returns: The packet, decoded as :func:`_network` decodes one behind VLAN tags, or @@ -676,23 +815,30 @@ def _past_extension_headers(protocol: 'int', data: 'bytes') -> 'Optional[tuple[i return None if walked is None else walked[:2] -def _walk_extension_headers(protocol: 'int', data: 'bytes') -> 'Optional[tuple[int, bytes, int]]': +def _walk_extension_headers(protocol: 'int', data: 'bytes', + limit: 'int' = EXTENSION_HEADER_LIMIT) -> 'Optional[tuple[int, bytes, int]]': """Step over the headers :func:`_past_extension_headers` steps over, and count them. Args: protocol: The IPv4 packet's protocol number. data: Raw octets of its payload. + limit: How many of them in a row the default engine dissects at most, + by default + :data:`~pcapkit.protocols.internet.internet.EXTENSION_HEADER_LIMIT`; + fewer where the frame has fewer layers left (see + :func:`_walk_ipv4_payload`). Returns: What :func:`_past_extension_headers` returns, then how many headers it stepped over, each of them one layer of the frame to the default engine (c.f. :data:`~pcapkit.protocols.protocol.FRAME_LAYER_LIMIT`); or - :data:`None` where it returns :data:`None`. + :data:`None` where it returns :data:`None`, or a header past ``limit`` + follows, which the default engine leaves undissected with the rest. """ depth = 0 while protocol in IPV4_EXTENSION_HEADERS: - if depth == EXTENSION_HEADER_LIMIT: + if depth >= limit: return None depth += 1 if protocol == Enum_TransType.AH: @@ -747,7 +893,37 @@ def _upper_layer(version: 'int', data: 'bytes') -> 'Optional[tuple[int, bytes]]' layer, tunnelled or not, it still reads out of the raw octets left behind (:func:`pcapkit.toolkit.pcap.tcp_segment`). + The header is read as the first layer of its frame, where + :data:`~pcapkit.protocols.protocol.FRAME_LAYER_LIMIT` leaves room for + more headers than either chain bound allows, so that bound never applies + here; :func:`_walk_upper_layer` reads one deeper in a frame. + """ + walked = _walk_upper_layer(version, data, 1) + return None if walked is None else walked[:2] + + +def _walk_upper_layer(version: 'int', data: 'bytes', depth: 'int') -> 'Optional[tuple[int, bytes, int]]': + """Step over an IP header to the upper-layer protocol it carries, and place that in the frame. + + Args: + version: IP version of the header ``data`` starts with, ``4`` or ``6``. + data: Raw octets of the IP packet. + depth: The IP header's place in its frame, the first layer being ``1`` + (c.f. :data:`~pcapkit.protocols.protocol.FRAME_LAYER_LIMIT`), at + most that bound. + + Returns: + What :func:`_upper_layer` returns, by its rules, then the upper layer's + place in the frame: one past the IP header and each header stepped over + after it, every one of them a layer of its own to the default engine. Or + :data:`None` where :func:`_upper_layer` returns :data:`None`, or the + upper layer is past + :data:`~pcapkit.protocols.protocol.FRAME_LAYER_LIMIT`, where the default + engine dissects nothing (:issue:`1609`); no header past it is read. + + """ + room = FRAME_LAYER_LIMIT - depth # layers the frame has left after the IP header if version == 4: if len(data) < IPV4_HEADER_LEN or data[0] >> 4 != 4: return None @@ -757,7 +933,7 @@ def _upper_layer(version: 'int', data: 'bytes') -> 'Optional[tuple[int, bytes]]' if struct.unpack_from('!H', data, 6)[0] & 0x1FFF: # a later fragment return None total = struct.unpack_from('!H', data, 2)[0] - return _past_extension_headers(data[9], data[ihl:max(total, ihl)] if total else data[ihl:]) + return _walk_ipv4_payload(data[9], data[ihl:max(total, ihl)] if total else data[ihl:], depth) if len(data) < IPV6_HEADER_LEN: return None @@ -765,11 +941,12 @@ def _upper_layer(version: 'int', data: 'bytes') -> 'Optional[tuple[int, bytes]]' if not extent: extent = jumbo_payload_length(data[6], data[IPV6_HEADER_LEN:]) or 0 data = data[:IPV6_HEADER_LEN + extent] - nxt, offset, depth = data[6], IPV6_HEADER_LEN, 0 + nxt, offset, headers = data[6], IPV6_HEADER_LEN, 0 + bound = min(EXTENSION_HEADER_LIMIT, room) while nxt in IPV6_EXTENSION_HEADERS: - if depth == EXTENSION_HEADER_LIMIT or len(data) < offset + 8: + if headers >= bound or len(data) < offset + 8: return None - depth += 1 + headers += 1 if nxt == Enum_ExtensionHeader.IPv6_Frag: # fixed length, second octet reserved if struct.unpack_from('!H', data, offset + 2)[0] >> 3: # a later fragment return None @@ -779,31 +956,58 @@ def _upper_layer(version: 'int', data: 'bytes') -> 'Optional[tuple[int, bytes]]' else: # length in 8-octet units, less one length = (data[offset + 1] + 1) * 8 nxt, offset = data[offset], offset + length - if len(data) < offset: + if len(data) < offset or headers >= room: + return None + return nxt, data[offset:], depth + headers + 1 + + +def _walk_ipv4_payload(protocol: 'int', data: 'bytes', depth: 'int') -> 'Optional[tuple[int, bytes, int]]': + """Step over the headers at the head of an IPv4 payload, and place what follows in the frame. + + Args: + protocol: The IPv4 packet's protocol number. + data: Raw octets of its payload. + depth: The IPv4 header's place in its frame, as for + :func:`_walk_upper_layer`. + + Returns: + What :func:`_past_extension_headers` returns, then the place in the + frame of the layer it names: one past the IPv4 header and each header + stepped over. :data:`None` where :func:`_past_extension_headers` returns + :data:`None`, or that layer or a header before it is past + :data:`~pcapkit.protocols.protocol.FRAME_LAYER_LIMIT` (:issue:`1609`). + + """ + room = FRAME_LAYER_LIMIT - depth # layers the frame has left after the IPv4 header + walked = _walk_extension_headers(protocol, data, min(EXTENSION_HEADER_LIMIT, room)) + if walked is None or walked[2] >= room: return None - return nxt, data[offset:] + return walked[0], walked[1], depth + walked[2] + 1 -def _innermost(packet: 'Packet', count: 'int' = -1) -> 'tuple[Optional[IP], Optional[tuple[int, bytes]]]': +def _innermost(packet: 'Packet', count: 'int' = -1) -> 'tuple[Optional[IP], Optional[tuple[int, bytes]], int]': """Fetch the innermost IP layer of a PyPCAPFile frame, the one carrying its TCP. Args: packet: PyPCAPFile packet. - count: Packet index, for the warning :func:`_network` gives. + count: Packet index, for the warning :func:`_network_layer` gives. Returns: The innermost IPv4 packet and :data:`None`, if the innermost IP layer is IPv4; :data:`None` and the upper-layer protocol and octets of the - innermost IPv6 packet, as :func:`_upper_layer` gives them, if it is - IPv6; otherwise neither. - - The outermost IP layer is the IPv4 packet :func:`_network` fetches, or - an IPv6 packet, behind VLAN tags (see :func:`_untag`) or not. Each IPv4 - or IPv6 packet one of them carries in the :data:`IP_TUNNELS`, directly - or behind the headers :func:`_upper_layer` steps over, is the next, as - deep as they go. An IPv4 packet in a tunnel is decoded by - :func:`_decode_ipv4`, and one it does not decode ends the walk with - neither, as the default engine then dissects no TCP past it. + innermost IPv6 packet, as :func:`_walk_upper_layer` gives them, if it is + IPv6; otherwise neither. Then the innermost IP layer's place in the + frame (c.f. :data:`~pcapkit.protocols.protocol.FRAME_LAYER_LIMIT`), for + :func:`_transport`; ``0`` if there is none. + + The outermost IP layer is the IPv4 packet :func:`_network_layer` + fetches, or the IPv6 packet :func:`_outer_ipv6` does. Each IPv4 or IPv6 + packet one of them carries in the :data:`IP_TUNNELS`, directly or behind + the headers :func:`_walk_upper_layer` steps over, is the next, as deep as + they go and no deeper than the frame's layer bound (:issue:`1609`). An + IPv4 packet in a tunnel is decoded by :func:`_decode_ipv4`, and one it + does not decode ends the walk with neither, as the default engine then + dissects no TCP past it. Note: The default engine keys a TCP segment by the IP layer nearest above it @@ -812,30 +1016,31 @@ def _innermost(packet: 'Packet', count: 'int' = -1) -> 'tuple[Optional[IP], Opti (:issue:`1581`). """ - ipv4 = _network(packet, count) - if ipv4 is not None: + ipv4 = None # type: Optional[IP] + layer = _network_layer(packet, count) + if layer is not None: + outer4, depth = layer # a later fragment has no upper-layer header, see ``_upper_layer`` - if (ipv4.off or (ipv4.p not in IP_TUNNELS and ipv4.p not in IPV4_EXTENSION_HEADERS) - or not _is_raw(ipv4.payload)): - return ipv4, None - upper = _past_extension_headers(ipv4.p, _ipv4_payload(ipv4)) # type: Optional[tuple[int, bytes]] + if (outer4.off or (outer4.p not in IP_TUNNELS and outer4.p not in IPV4_EXTENSION_HEADERS) + or not _is_raw(outer4.payload)): + return outer4, None, depth + ipv4, upper = outer4, _walk_ipv4_payload(outer4.p, _ipv4_payload(outer4), depth) else: - link = packet.packet - untagged = None if _is_raw(link) else _untag(link) - if untagged is None or untagged[0] != Enum_EtherType.Internet_Protocol_version_6: - return None, None - upper = _upper_layer(6, untagged[1]) + outer6 = _outer_ipv6(packet) + if outer6 is None: + return None, None, 0 + depth, upper = outer6[1], _walk_upper_layer(6, *outer6) while upper is not None and upper[0] in IP_TUNNELS: - version, data = IP_TUNNELS[upper[0]], upper[1] - upper = _upper_layer(version, data) + version, data, depth = IP_TUNNELS[upper[0]], upper[1], upper[2] + upper = _walk_upper_layer(version, data, depth) if version == 6: ipv4 = None elif (ipv4 := _decode_ipv4(data)) is None: - return None, None + return None, None, depth if ipv4 is not None: - return ipv4, None - return None, upper + return ipv4, None, depth + return None, None if upper is None else upper[:2], depth def _left_out(upper: 'Optional[tuple[int, bytes]]') -> 'Optional[str]': @@ -959,11 +1164,13 @@ def _ipv4_payload(ipv4: 'IP') -> 'bytes': return payload[:max(ipv4.len - ipv4.hl * 4, 0)] -def _transport(ipv4: 'IP') -> 'Optional[bytes]': +def _transport(ipv4: 'IP', depth: 'int' = 1) -> 'Optional[bytes]': """Fetch the verbatim TCP segment carried by a PyPCAPFile IPv4 packet. Args: ipv4: PyPCAPFile IPv4 packet. + depth: Its place in the frame (see :func:`_innermost`), by default the + first layer, as for a packet read on its own. Returns: The raw TCP segment, as far as the Total Length (see @@ -974,7 +1181,10 @@ def _transport(ipv4: 'IP') -> 'Optional[bytes]': carries none: from its offset on, its data is a slice of the datagram's payload, with no upper-layer header in it (:rfc:`791#section-3.2`, :issue:`1576`). Nor does a segment too short for a header, or whose Data - Offset is below 5 words (:issue:`1590`). + Offset is below 5 words (:issue:`1590`), or one past + :data:`~pcapkit.protocols.protocol.FRAME_LAYER_LIMIT` layers of the + frame, counting each header stepped over (see + :func:`_walk_ipv4_payload`, :issue:`1609`). Behind those, the default engine reads TCP only as its TCP parser does, so the whole of the header its Data Offset gives must be there, and any @@ -989,7 +1199,7 @@ def _transport(ipv4: 'IP') -> 'Optional[bytes]': if not _is_raw(ipv4.payload): return None - upper = _past_extension_headers(ipv4.p, _ipv4_payload(ipv4)) + upper = _walk_ipv4_payload(ipv4.p, _ipv4_payload(ipv4), depth) if upper is None or upper[0] != Enum_TransType.TCP: return None segment = upper[1] @@ -1134,12 +1344,24 @@ def packet2dict(packet: 'Packet', *, data_link: 'Enum_LinkType') -> 'dict[str, A parsed as an IPv4 address, or a decoded Ethernet layer's ``src``/``dst`` cannot be parsed as a MAC address. + Note: + A frame `PyPCAPFile`_ left undecoded is hex-encoded, as the engine reads + it (see :meth:`PyPCAPFile._read_records + `), so + it is un-hexed (see :func:`_maybe_unhex`) and reported as its octets, + not as twice as many characters of hex text (:issue:`1667`). + + .. _PyPCAPFile: https://github.com/kisom/pypcapfile + """ + frame = packet.packet + if _is_raw(frame): + frame = _maybe_unhex(bytes(frame)) return { 'timestamp': packet2timestamp(packet), 'capture_len': packet.capture_len, 'packet_len': packet.packet_len, - data_link.name: _layer2dict(packet.packet), + data_link.name: _layer2dict(frame), } @@ -1155,10 +1377,12 @@ def ipv4_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv4Ad * If the ``packet`` can be used for IPv4 reassembly. A packet can be reassembled if it contains an IPv4 layer (:class:`pcapfile.protocols.network.ip.IP`), behind - 802.1Q or 802.1ad VLAN tags or not, and the **DF** flag is :data:`False`. The + 802.1Q or 802.1ad VLAN tags or not, or in a frame of one of the + :data:`READ_LINKTYPES`, and the **DF** flag is :data:`False`. The layer is the first IPv4 packet of the frame, as for the default engine: one carried in IPv6 counts (see :func:`_ipv4_in_ipv6`), and one whose header the - default engine rejects does not (see :func:`_ipv4_accepted`). + default engine rejects, or past its frame layer bound, does not (see + :func:`_ipv4_accepted` and :func:`_network_layer`). * If the ``packet`` can be reassembled, then the :obj:`dict` mapping of data for IPv4 reassembly (:term:`reasm.ipv4.packet`) will be returned; otherwise, returns :data:`None`. @@ -1227,7 +1451,8 @@ def tcp_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'TCP_Packet | None * If the ``packet`` can be used for TCP reassembly. A packet can be reassembled if it contains an IPv4 layer carrying a TCP segment, directly or behind AH or - IPv6 extension headers, and behind 802.1Q or 802.1ad VLAN tags or not (see + IPv6 extension headers, and behind 802.1Q or 802.1ad VLAN tags or not, or in a + frame of one of the :data:`READ_LINKTYPES`, within the frame layer bound (see :func:`_transport`). In an IP-in-IP tunnel the IPv4 layer is the innermost IP layer, whose addresses key the segment (see :func:`_innermost`); TCP over IPv6, tunnelled or not, cannot be, as `PyPCAPFile`_ has no IPv6 decoder. @@ -1251,8 +1476,8 @@ def tcp_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'TCP_Packet | None .. _PyPCAPFile: https://github.com/kisom/pypcapfile """ - ipv4, upper = _innermost(packet, count) - if ipv4 is None or (segment := _transport(ipv4)) is None: + ipv4, upper, depth = _innermost(packet, count) + if ipv4 is None or (segment := _transport(ipv4, depth)) is None: _decline_tcp(packet, upper, count) return None @@ -1300,7 +1525,8 @@ def tcp_traceflow(packet: 'Packet', *, data_link: 'Enum_LinkType', * If the ``packet`` can be used for TCP flow tracing. A packet can be traced if it contains an IPv4 layer carrying a TCP segment, directly or behind AH or - IPv6 extension headers, and behind 802.1Q or 802.1ad VLAN tags or not (see + IPv6 extension headers, and behind 802.1Q or 802.1ad VLAN tags or not, or in a + frame of one of the :data:`READ_LINKTYPES`, within the frame layer bound (see :func:`_transport`). In an IP-in-IP tunnel the IPv4 layer is the innermost IP layer, whose addresses key the segment (see :func:`_innermost`); TCP over IPv6, tunnelled or not, cannot be, as `PyPCAPFile`_ has no IPv6 decoder. @@ -1324,8 +1550,8 @@ def tcp_traceflow(packet: 'Packet', *, data_link: 'Enum_LinkType', .. _PyPCAPFile: https://github.com/kisom/pypcapfile """ - ipv4, upper = _innermost(packet, count) - if ipv4 is None or (segment := _transport(ipv4)) is None: + ipv4, upper, depth = _innermost(packet, count) + if ipv4 is None or (segment := _transport(ipv4, depth)) is None: _decline_tcp(packet, upper, count) return None diff --git a/tests/foundation/engines/test_pypcapfile_engine.py b/tests/foundation/engines/test_pypcapfile_engine.py index 152ca6bca7..e4b9535f74 100644 --- a/tests/foundation/engines/test_pypcapfile_engine.py +++ b/tests/foundation/engines/test_pypcapfile_engine.py @@ -301,6 +301,22 @@ def test_run_warns_and_gives_up_decoding_for_an_unknown_link_layer(self) -> None self.assertIn('unrecognised link layer protocol', warn.call_args.args[0]) self.assertIs(warn.call_args.args[1], AttributeWarning) + def test_run_warns_only_of_a_link_layer_left_unread(self) -> None: + # C.f. #1667: pcapfile decodes none of NULL, LOOP, RAW, IPV4 and IPV6, + # but the toolkit reads the IP packet in their frames, as the default + # engine does, so nothing is lost to warn of. 12 and LINUX_SLL stay unread. + for linktype, unread in ((0, False), (108, False), (101, False), (228, False), (229, False), + (12, True), (113, True), (1, True)): + with self.subTest(linktype=linktype): + extractor, _ = self.make_extractor() + engine = self.engine(extractor, savefile=FakeSaveFile(ll_type=linktype), decoder=None) + with mock.patch('pcapkit.foundation.engines.pypcapfile.warn') as warn: + engine.run() + self.assertIsNone(engine._declf) + messages = [call.args[0] for call in warn.call_args_list] + self.assertEqual(any('unrecognised link layer protocol' in message for message in messages), + unread, messages) + def test_run_survives_the_malformed_upstream_link_layer_table(self) -> None: # ``pcapfile.linklayer.__LL_TYPES__`` carries a three-element entry for # LINKTYPE_IEEE802_11_RADIOTAP, so ``clookup`` raises IndexError for it. diff --git a/tests/foundation/engines/test_pypcapfile_linktype_and_layer_bound_agreement_1609_1667_unit.py b/tests/foundation/engines/test_pypcapfile_linktype_and_layer_bound_agreement_1609_1667_unit.py new file mode 100644 index 0000000000..25474acf3b --- /dev/null +++ b/tests/foundation/engines/test_pypcapfile_linktype_and_layer_bound_agreement_1609_1667_unit.py @@ -0,0 +1,309 @@ +# -*- coding: utf-8 -*- +"""``pypcapfile`` hands over what ``default`` does, by link type and to the frame layer bound. C.f. #1609, #1667. + +The ``default`` engine dissects the IP packet of the ``NULL``, ``LOOP``, +``RAW``, ``IPV4`` and ``IPV6`` link types, which `pypcapfile`_ 0.12.0 has no +decoder for (#1667), and at most +:data:`~pcapkit.protocols.protocol.FRAME_LAYER_LIMIT` layers of a frame, each +VLAN tag, IP header, AH and IPv6 extension header one of them (#1609). Both +engines read one capture per case, built in a temporary directory, with IPv4 +and TCP reassembly and flow tracing on, and ``pypcapfile``'s ``tcp``, ``trace`` +and ``ipv4`` inputs are compared with ``default``'s field for field, as in +:mod:`tests.foundation.engines.test_ext_chain_limit_agreement_1604_unit`. + +``pypcapfile`` leaves TCP over IPv6 out, so its ``tcp`` and ``trace`` inputs are +compared over IPv4 only, and ``declined`` checks that it warns of leaving some +out exactly where ``default`` finds any. ``unread`` checks that the engine warns +of a link type left unread exactly where ``default`` dissects nothing of it. It +imports on Python 3.11 only, and is skipped elsewhere. + +.. _pypcapfile: https://github.com/kisom/pypcapfile + +""" + +from __future__ import annotations + +import importlib.util +import ipaddress +import os +import tempfile +import unittest +import warnings +from typing import TYPE_CHECKING +from unittest import mock + +from tests._support import close_extractor, reimport_once_per_class +from tests.foundation import _roundtrip as harness +from tests.foundation._roundtrip import Outcome + +if TYPE_CHECKING: + from typing import Any, Optional + +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None + for name in ('tbtrim', 'aenum', 'chardet', 'dictdumper')) + +#: ``LINKTYPE_*`` numbers, by name. +LINKTYPES = {'NULL': 0, 'ETHERNET': 1, 'LOOP': 108, 'RAW': 101, 'IPV4': 228, 'IPV6': 229, + 'DLT12': 12, 'LINUX_SLL': 113} +#: Link types the ``default`` engine dissects nothing of. +UNREAD = ('DLT12', 'LINUX_SLL') + +TCP = harness.tcp(b'payload', seq=1, syn=True) +TCP4 = harness.ipv4(TCP, proto=6) +TCP6 = harness.ipv6(TCP, nxt=6) +UDP = harness.udp(b'u' * 24) +FRAG4 = (harness.ipv4(UDP[:16], proto=17, mf=True), harness.ipv4(UDP[16:], proto=17, offset=16)) +FRAG6 = (harness.ipv6_fragment(UDP[:16], mf=True), harness.ipv6_fragment(UDP[16:], offset=16)) +#: Payloads, each the IP version it starts with and the packets of its frames. +PAYLOADS = { + 'ipv4-tcp': (4, [TCP4]), + 'ipv6-tcp': (6, [TCP6]), + 'ipv4-frags': (4, list(FRAG4)), + 'ipv6-frags': (6, list(FRAG6)), + '4in6-tcp': (6, [harness.ipv6(TCP4, nxt=4)]), + '4in6-frags': (6, [harness.ipv6(fragment, nxt=4) for fragment in FRAG4]), +} + + +def link(name: 'str', packet: 'bytes', version: 'int') -> 'bytes': + """``packet`` as a frame of the link type ``name``.""" + if name == 'ETHERNET': + return harness.ethernet(packet, 0x0800 if version == 4 else 0x86DD) + if name in ('NULL', 'LOOP'): + family = 2 if version == 4 else 24 + return family.to_bytes(4, 'little' if name == 'NULL' else 'big') + packet + if name == 'LINUX_SLL': + return (b'\x00\x00\x00\x01\x00\x06' + b'\x02' * 8 + + (0x0800 if version == 4 else 0x86DD).to_bytes(2, 'big') + packet) + return packet + + +def nest4(inner: 'bytes', count: 'int', proto: 'int') -> 'bytes': + """``count`` IPv4 headers, the innermost carrying ``inner`` as ``proto``.""" + packet = harness.ipv4(inner, proto=proto) + for _ in range(count - 1): + packet = harness.ipv4(packet, proto=4) + return packet + + +def nest6(inner: 'bytes', count: 'int', nxt: 'int') -> 'bytes': + """``count`` IPv6 headers, the innermost carrying ``inner`` as ``nxt``.""" + packet = harness.ipv6(inner, nxt=nxt) + for _ in range(count - 1): + packet = harness.ipv6(packet, nxt=41) + return packet + + +def pairs(count: 'int', version: 'int') -> 'bytes': + """``count`` pairs of an IPv4 header and an AH, or of an IPv6 header and a Hop-by-Hop Options header, then TCP.""" + packet = TCP + for index in range(count): + nxt = 6 if index == 0 else (4 if version == 4 else 41) + if version == 4: + ah = bytes([nxt, 4, 0, 0]) + (0x100).to_bytes(4, 'big') + (1).to_bytes(4, 'big') + b'\xaa' * 12 + packet = harness.ipv4(ah + packet, proto=51) + else: + packet = harness.ipv6(bytes([nxt, 0, 1, 4, 0, 0, 0, 0]) + packet, nxt=0) + return packet + + +def chain(count: 'int', last: 'int') -> 'bytes': + """``count`` Destination Options headers, the last naming ``last``.""" + return b''.join(bytes([60 if index < count - 1 else last, 0, 1, 4, 0, 0, 0, 0]) for index in range(count)) + + +def tagged(packet: 'bytes', ethertype: 'int', count: 'int', tpid: 'int') -> 'bytes': + """An Ethernet frame carrying ``packet`` behind ``count`` VLAN tags of ``tpid``.""" + tags = b''.join((100 + index).to_bytes(2, 'big') + (tpid if index < count - 1 else ethertype).to_bytes(2, 'big') + for index in range(count)) + return harness.MAC_B + harness.MAC_A + tpid.to_bytes(2, 'big') + tags + packet + + +def cases() -> 'dict[str, tuple[int, list[bytes]]]': + """Every case, as the link type of its capture and its frames.""" + out = {} # type: dict[str, tuple[int, list[bytes]]] + for name, linktype in LINKTYPES.items(): + for kind, (version, packets) in PAYLOADS.items(): + if (name, version) not in (('IPV4', 6), ('IPV6', 4)): + out[f'{name}-{kind}'] = (linktype, [link(name, packet, version) for packet in packets]) + out['NULL-big-endian-family'] = (0, [(2).to_bytes(4, 'big') + TCP4]) + out['LOOP-little-endian-family'] = (108, [(2).to_bytes(4, 'little') + TCP4]) + out['NULL-af28-ipv6-tcp'] = (0, [(28).to_bytes(4, 'little') + TCP6]) + out['RAW-version-5'] = (101, [b'\x55' + TCP4[1:]]) + out['RAW-6in4-tcp'] = (101, [harness.ipv4(TCP6, proto=41)]) + out['RAW-tcp-behind-ah'] = (101, [pairs(1, 4)]) + for count in (62, 63): + out[f'ETHERNET-tcp-in-{count}-ipv4'] = (1, [harness.ethernet(nest4(TCP, count, 6), 0x0800)]) + out[f'ETHERNET-ipv4-frags-in-{count}-ipv6'] = ( + 1, [harness.ethernet(nest6(fragment, count, 4), 0x86DD) for fragment in FRAG4]) + out[f'NULL-tcp-in-{count}-ipv4'] = (0, [(2).to_bytes(4, 'little') + nest4(TCP, count, 6)]) + out[f'IPV6-4in6-tcp-in-{count}-ipv6'] = (229, [nest6(TCP4, count, 4)]) + for count in (63, 64): + out[f'RAW-tcp-in-{count}-ipv4'] = (101, [nest4(TCP, count, 6)]) + out[f'RAW-ipv4-frags-in-{count}-ipv6'] = (101, [nest6(fragment, count, 4) for fragment in FRAG4]) + for count in (61, 62): + out[f'ETHERNET-4in6-tcp-in-{count}-ipv6'] = (1, [harness.ethernet(nest6(TCP4, count, 4), 0x86DD)]) + out[f'ETHERNET-ipv6-tcp-behind-{count}-tags'] = (1, [tagged(TCP6, 0x86DD, count, 0x8100)]) + for count in (31, 32): + out[f'ETHERNET-tcp-behind-{count}-ipv4-ah'] = (1, [harness.ethernet(pairs(count, 4), 0x0800)]) + out[f'ETHERNET-tcp-behind-{count}-ipv6-hbh'] = (1, [harness.ethernet(pairs(count, 6), 0x86DD)]) + out[f'ETHERNET-tcp-in-31-ipv4-behind-{count}-headers'] = ( + 1, [harness.ethernet(nest4(chain(count, 6) + TCP, 31, 60), 0x0800)]) + for count in (61, 62, 63): + for tpid in (0x8100, 0x88A8): + out[f'ETHERNET-tcp-behind-{count}-tags-{tpid:04x}'] = (1, [tagged(TCP4, 0x0800, count, tpid)]) + out[f'ETHERNET-ipv4-frags-behind-{count}-tags-{tpid:04x}'] = ( + 1, [tagged(fragment, 0x0800, count, tpid) for fragment in FRAG4]) + return out + + +#: Every case, built once. +CASES = cases() + + +def _fields(packet: 'Any', skip: 'tuple[str, ...]' = ()) -> 'dict[str, Any]': + return {key: bytes(value) if isinstance(value, bytearray) else value + for key, value in packet.items() if key not in skip} + + +class _Run: + """One engine's reading of one capture.""" + + def __init__(self, engine: 'str', path: 'str') -> None: + from pcapkit import extract + from pcapkit.foundation.reassembly.reassembly import ReassemblyBase + from pcapkit.foundation.traceflow.traceflow import TraceFlowBase + from pcapkit.utilities.warnings import AttributeWarning, EngineWarning + + self.error = self.fallback = None # type: Optional[str] + #: What the ``pypcapfile`` toolkit warned it leaves out, if anything. + self.declined = None # type: Optional[str] + #: Whether the ``pypcapfile`` engine warned it leaves the link type unread. + self.unread = False + self.inputs = {'ipv4': {}, 'tcp': {}, 'trace': {}} # type: dict[str, dict[int, list[dict[str, Any]]]] + inputs = self.inputs + reassemble, trace = ReassemblyBase.__call__, TraceFlowBase.__call__ + + def spy_reassembly(this: 'Any', packet: 'Any') -> 'None': + inputs[type(this).__name__.lower()].setdefault(packet.num, []).append(_fields(packet)) + return reassemble(this, packet) + + def spy_trace(this: 'Any', packet: 'Any') -> 'None': + inputs['trace'].setdefault(packet.index, []).append(_fields(packet, ('frame',))) + return trace(this, packet) + + with warnings.catch_warnings(), tempfile.TemporaryDirectory() as tracedir, \ + mock.patch.object(ReassemblyBase, '__call__', spy_reassembly), \ + mock.patch.object(TraceFlowBase, '__call__', spy_trace), \ + mock.patch('pcapkit.foundation.extraction.warn') as warn, \ + mock.patch('pcapkit.foundation.engines.pypcapfile.warn') as engine_warn, \ + mock.patch('pcapkit.toolkit.pypcapfile.warn') as toolkit_warn: + warnings.simplefilter('ignore') + try: + extractor = extract(fin=path, nofile=True, engine=engine, reassembly=True, ipv4=True, + tcp=True, trace=True, trace_fout=tracedir, trace_format='json') + except Exception as exc: # pylint: disable=broad-except + self.error = f'{type(exc).__name__}: {exc}' + return + close_extractor(extractor) + fallen = [str(call.args[0]) for call in warn.call_args_list + if len(call.args) > 1 and call.args[1] is EngineWarning] + if fallen or extractor._exnam != engine: # pylint: disable=protected-access + self.fallback = '; '.join(fallen) or f'ran as {extractor._exnam}' # pylint: disable=protected-access + return + self.declined = '; '.join(str(call.args[0]) for call in toolkit_warn.call_args_list + if len(call.args) > 1 and call.args[1] is AttributeWarning + and 'is left out of TCP reassembly' in str(call.args[0])) or None + self.unread = any('unrecognised link layer protocol' in str(call.args[0]) + for call in engine_warn.call_args_list) + + +_RUNS = {} # type: dict[tuple[str, str], _Run] + + +def _over_ipv4(record: 'dict[str, Any]') -> 'bool': + address = record['bufid'][0] if 'bufid' in record else record['src'] + return isinstance(address, ipaddress.IPv4Address) + + +def _difference(got: 'dict[int, Any]', want: 'dict[int, Any]') -> 'Optional[str]': + for number in sorted(set(got) | set(want)): + if number not in got: + return f'frame {number}: missing, default has {len(want[number])} input(s)' + if number not in want: + return f'frame {number}: extra, default has none' + if got[number] != want[number]: + return f'frame {number}: {got[number]!r:.120} != {want[number]!r:.120}' + return None + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class TestPyPCAPFileLinkTypeAndLayerBoundAgreement(harness.RoundTripBase): + """``pypcapfile`` hands over what ``default`` does over every case, or is a recorded gap.""" + + STATUSES = ('ERROR', 'INPUTS', 'DECLINED', 'UNREAD') + KNOWN_FAILURES = () + + @classmethod + def setUpClass(cls) -> None: + super().setUpClass() + cls.tmp = tempfile.TemporaryDirectory() # pylint: disable=consider-using-with + for name, (linktype, frames) in CASES.items(): + with open(cls.path(name), 'wb') as file: + file.write(harness.pcap([(1, index, frame) for index, frame in enumerate(frames)], + linktype=linktype)) + + @classmethod + def tearDownClass(cls) -> None: + _RUNS.clear() + cls.tmp.cleanup() + super().tearDownClass() + + @classmethod + def path(cls, name: 'str') -> 'str': + return os.path.join(cls.tmp.name, f'{name}.pcap') + + def setUp(self) -> None: + super().setUp() + reimport_once_per_class(self) + + def labels(self) -> 'list[str]': + return [f'{name}/{aspect}' for name in CASES + for aspect in ('tcp', 'trace', 'ipv4', 'declined', 'unread')] + + def run_of(self, engine: 'str', name: 'str') -> '_Run': + if (engine, name) not in _RUNS: + _RUNS[engine, name] = _Run(engine, self.path(name)) + return _RUNS[engine, name] + + def outcome(self, label: 'str') -> 'Outcome': + name, aspect = label.split('/') + run = self.run_of('pypcapfile', name) + if run.fallback is not None: + return Outcome('SKIP', f'pypcapfile did not run: {run.fallback}') + if run.error is not None: + return Outcome('ERROR', run.error) + base = self.run_of('default', name) + if base.error is not None or base.fallback is not None: + return Outcome('ERROR', f'default: {base.error or base.fallback}') + + if aspect == 'unread': + if run.unread != name.startswith(UNREAD): + return Outcome('UNREAD', f'warned of the link type left unread: {run.unread}') + return Outcome('OK') + over_ipv6 = any(not _over_ipv4(records[0]) for records in base.inputs['tcp'].values()) + if aspect == 'declined': + if (run.declined is not None) != over_ipv6: + return Outcome('DECLINED', f'warned {run.declined!r}, but default finds TCP over IPv6: ' + f'{over_ipv6}') + return Outcome('OK') + + mine, theirs = run.inputs[aspect], base.inputs[aspect] + if aspect != 'ipv4': + theirs = {number: records for number, records in theirs.items() if _over_ipv4(records[0])} + diff = _difference(mine, theirs) + return Outcome('INPUTS', diff) if diff else Outcome('OK') + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/toolkit/test_pypcapfile_ext_chain_limit_1604_unit.py b/tests/toolkit/test_pypcapfile_ext_chain_limit_1604_unit.py index c30c519cfc..2e3495936a 100644 --- a/tests/toolkit/test_pypcapfile_ext_chain_limit_1604_unit.py +++ b/tests/toolkit/test_pypcapfile_ext_chain_limit_1604_unit.py @@ -103,10 +103,10 @@ def test_a_tunnel_behind_the_chain_is_entered_only_within_the_limit(self) -> Non mock.patch.dict('sys.modules', base.fake_ip_decoder()) as modules: IP = modules['pcapfile.protocols.network.ip'].IP packet = base.make_packet(base.tunnel(chain(count, 4) + tcp4, 60)) - ipv4, upper = _innermost(packet) + ipv4, upper, depth = _innermost(packet) self.assertIsNone(upper) self.assertEqual(IP.calls, [(tcp4, 0)] if entered else []) - self.assertEqual(_transport(ipv4), base.TCP_SEGMENT if entered else None) + self.assertEqual(_transport(ipv4, depth), base.TCP_SEGMENT if entered else None) if not entered: # no TCP, and none left out either self.assertEqual(self.decline(packet), []) @@ -130,8 +130,8 @@ def test_a_new_chain_inside_a_tunnel_is_counted_afresh(self) -> None: from pcapkit.toolkit.pypcapfile import _innermost, _transport inner = wire.ipv4(chain(count, 6) + tcp4, proto=60) - ipv4, _ = _innermost(base.make_packet(base.tunnel(chain(20, 4) + inner, 60))) - self.assertEqual(_transport(ipv4), tcp4 if found else None) + ipv4, _, depth = _innermost(base.make_packet(base.tunnel(chain(20, 4) + inner, 60))) + self.assertEqual(_transport(ipv4, depth), tcp4 if found else None) def test_ipv4_in_ipv6_is_found_only_within_the_limit(self) -> None: from pcapkit.toolkit.pypcapfile import _ipv4_in_ipv6 diff --git a/tests/toolkit/test_pypcapfile_header_checks_unit.py b/tests/toolkit/test_pypcapfile_header_checks_unit.py index 839e38c1bb..71345829ff 100644 --- a/tests/toolkit/test_pypcapfile_header_checks_unit.py +++ b/tests/toolkit/test_pypcapfile_header_checks_unit.py @@ -254,10 +254,10 @@ def test_tunnelled_tcp_behind_ah_is_found(self) -> None: ): with self.subTest(case=label), mock.patch.dict('sys.modules', base.fake_ip_decoder()) as modules: IP = modules['pcapfile.protocols.network.ip'].IP - ipv4, _ = _innermost(base.make_packet(frame)) + ipv4, _, depth = _innermost(base.make_packet(frame)) self.assertIsInstance(ipv4, IP) self.assertEqual(IP.calls, [(inner, 0)]) - self.assertEqual(_transport(ipv4), base.TCP_SEGMENT) + self.assertEqual(_transport(ipv4, depth), base.TCP_SEGMENT) if __name__ == '__main__': diff --git a/tests/toolkit/test_pypcapfile_linktypes_and_layer_bound_1609_1667_unit.py b/tests/toolkit/test_pypcapfile_linktypes_and_layer_bound_1609_1667_unit.py new file mode 100644 index 0000000000..f7ebaac68d --- /dev/null +++ b/tests/toolkit/test_pypcapfile_linktypes_and_layer_bound_1609_1667_unit.py @@ -0,0 +1,345 @@ +""":mod:`pcapkit.toolkit.pypcapfile` reads the link types, and stops at the frame depth, the default engine does. + +GitHub issue #1667: `pypcapfile`_ 0.12.0 decodes Ethernet alone, so the engine +left every frame of the ``NULL``, ``LOOP``, ``RAW``, ``IPV4`` and ``IPV6`` link +types undecoded, and the adapters handed nothing to TCP reassembly, flow tracing +or IPv4 reassembly, where the default engine reads the IP packet in them. They +now read it themselves, and :func:`~pcapkit.toolkit.pypcapfile.packet2dict` +reports an unread frame's octets rather than their hex text. + +GitHub issue #1609: the default engine dissects at most +:data:`~pcapkit.protocols.protocol.FRAME_LAYER_LIMIT` layers of a frame, each +VLAN tag, IP header, AH and IPv6 extension header one of them. The adapters' +walks ran past it, and now stop there too. + +These use stand-ins for `pypcapfile`_'s IP and TCP decoders, so run without it; +the two engines are compared over real captures in +:mod:`tests.foundation.engines.test_pypcapfile_linktype_and_layer_bound_agreement_1609_1667_unit`. + +.. _pypcapfile: https://github.com/kisom/pypcapfile + +""" +from __future__ import annotations + +import binascii +import ipaddress +import unittest +from unittest import mock + +from tests._support import reimport_once_per_class +from tests.foundation import _roundtrip as wire +from tests.toolkit import test_pypcapfile_tcp_last_unit as tcp_last +from tests.toolkit import test_pypcapfile_unit as base + +#: ``LINKTYPE_*`` numbers. +NULL, ETHERNET, RAW, LOOP, IPV4, IPV6, LINUX_SLL = 0, 1, 101, 108, 228, 229, 113 + +#: An IPv4 packet carrying :data:`~tests.toolkit.test_pypcapfile_unit.TCP_SEGMENT`, +#: 47 octets long, and an IPv6 one. +TCP4 = wire.ipv4(base.TCP_SEGMENT, proto=6) +TCP6 = wire.ipv6(base.TCP_SEGMENT, nxt=6) +#: A first IPv4 fragment of a UDP datagram. +FRAGMENT = base.udp_fragment() +#: The addresses :mod:`tests.foundation._roundtrip` gives a packet. +SRC, DST = ipaddress.IPv4Address(wire.IPV4_A), ipaddress.IPv4Address(wire.IPV4_B) + + +def stand_ins() -> dict: + """Stand-in :mod:`pcapfile` modules, for :data:`sys.modules`: its IP and its TCP decoder.""" + modules = base.fake_ip_decoder() + for name, module in tcp_last.fake_pcapfile().items(): + modules.setdefault(name, module) + return modules + + +def unread(frame: bytes, linktype: int): + """A frame of a capture of ``linktype``, as the engine hands one over that `pypcapfile`_ cannot decode.""" + return base.make_packet(binascii.hexlify(frame), linktype=linktype) + + +def loopback(packet: bytes, family: int, byteorder: str) -> bytes: + """``packet`` behind a BSD loopback header of address family ``family``.""" + return family.to_bytes(4, byteorder) + packet + + +def nest4(inner: bytes, count: int, proto: int) -> bytes: + """``count`` IPv4 headers, the innermost carrying ``inner`` as protocol ``proto``, each outer one IPv4.""" + packet = wire.ipv4(inner, proto=proto) + for _ in range(count - 1): + packet = wire.ipv4(packet, proto=4) + return packet + + +def nest6(inner: bytes, count: int, nxt: int) -> bytes: + """``count`` IPv6 headers, the innermost carrying ``inner`` as ``nxt``, each outer one IPv6.""" + packet = wire.ipv6(inner, nxt=nxt) + for _ in range(count - 1): + packet = wire.ipv6(packet, nxt=41) + return packet + + +def ethernet(packet: bytes): + """``packet``, an IPv4 packet with no options, behind Ethernet, as `pypcapfile`_ decodes it.""" + return base.make_packet(base.tunnel(packet[20:], packet[9])) + + +def ethernet6(packet: bytes): + """``packet``, an IPv6 packet, behind Ethernet, as `pypcapfile`_ leaves it.""" + return base.make_packet(base.ipv6_frame(packet)) + + +def ah_pairs(count: int) -> bytes: + """``count`` pairs of an IPv4 header and an AH, then the TCP segment.""" + packet = base.TCP_SEGMENT + for index in range(count): + packet = wire.ipv4(base.ah(6 if index == 0 else 4) + packet, proto=51) + return packet + + +def hbh_pairs(count: int) -> bytes: + """``count`` pairs of an IPv6 header and a Hop-by-Hop Options header, then the TCP segment.""" + packet = base.TCP_SEGMENT + for index in range(count): + packet = wire.ipv6(base.options_header(6 if index == 0 else 41) + packet, nxt=0) + return packet + + +def tags(packet: bytes, ethertype: int, count: int, tpid: int): + """``packet`` behind ``count`` VLAN tags of ``tpid``, as `pypcapfile`_ leaves it.""" + return base.make_packet(base.tagged_frame(packet, ethertype, *(tpid,) * count)) + + +def chain(count: int, last: int) -> bytes: + """``count`` Destination Options headers, the last naming ``last``.""" + return b''.join(base.options_header(60 if index < count - 1 else last) for index in range(count)) + + +@unittest.skipUnless(base.HAS_RUNTIME, 'runtime dependencies not installed') +class PyPCAPFileLinkTypeTests(unittest.TestCase): + """C.f. #1667.""" + + decline = base.PyPCAPFileTCPOverIPv6Tests.decline + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_tcp_is_read_over_each_link_type_the_default_engine_dissects(self) -> None: + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit.pypcapfile import tcp_reassembly, tcp_traceflow + + tcp4in6 = wire.ipv6(TCP4, nxt=4) + for label, linktype, frame, read in ( + ('NULL, little-endian', NULL, loopback(TCP4, 2, 'little'), True), + ('NULL, big-endian', NULL, loopback(TCP4, 2, 'big'), True), + ('NULL, 4in6 over AF_INET6 28', NULL, loopback(tcp4in6, 28, 'little'), True), + ('NULL, 4in6 over AF_INET6 30', NULL, loopback(tcp4in6, 30, 'little'), True), + ('LOOP', LOOP, loopback(TCP4, 2, 'big'), True), + ('LOOP, 4in6', LOOP, loopback(tcp4in6, 24, 'big'), True), + ('RAW', RAW, TCP4, True), + ('RAW, 4in4', RAW, wire.ipv4(TCP4, proto=4), True), + ('RAW, 4in6', RAW, tcp4in6, True), + ('RAW, behind AH', RAW, wire.ipv4(base.ah(6) + base.TCP_SEGMENT, proto=51), True), + ('IPV4', IPV4, TCP4, True), + ('IPV6, 4in6', IPV6, tcp4in6, True), + # where the default engine dissects no IP either + ('LOOP, family written little-endian', LOOP, loopback(TCP4, 2, 'little'), False), + ('NULL, family 7', NULL, loopback(TCP4, 7, 'little'), False), + ('NULL, cut short of its header', NULL, b'\x02\x00\x00', False), + ('RAW, version 5', RAW, b'\x55' + TCP4[1:], False), + ('RAW, empty', RAW, b'', False), + ('IPV4, an IPv6 packet', IPV4, TCP6, False), + ('12', 12, TCP4, False), + ('LINUX_SLL', LINUX_SLL, TCP4, False), + ): + with self.subTest(case=label), mock.patch.dict('sys.modules', stand_ins()): + packet = unread(frame, linktype) + data = tcp_reassembly(packet, count=1) + trace = tcp_traceflow(packet, data_link=LinkType.get(linktype), count=1) + if not read: + self.assertIsNone(data) + self.assertIsNone(trace) + self.assertEqual(self.decline(packet), []) + continue + self.assertIsNotNone(data) + self.assertEqual(data.bufid, (SRC, 40000, DST, 9)) + self.assertEqual((data.header, data.payload), (base.TCP_SEGMENT[:20], bytearray(b'payload'))) + self.assertEqual((trace.src, trace.dst, trace.srcport, trace.dstport), (SRC, DST, 40000, 9)) + self.assertEqual(trace.header, base.TCP_SEGMENT[:20]) + + def test_ipv4_fragments_are_read_over_each_link_type_the_default_engine_dissects(self) -> None: + from pcapkit.const.reg.transtype import TransType + from pcapkit.toolkit.pypcapfile import ipv4_reassembly + + in6 = wire.ipv6(FRAGMENT, nxt=4) + for label, linktype, frame, read in ( + ('NULL', NULL, loopback(FRAGMENT, 2, 'little'), True), + ('NULL, 4in6', NULL, loopback(in6, 24, 'little'), True), + ('LOOP', LOOP, loopback(FRAGMENT, 2, 'big'), True), + ('LOOP, 4in6', LOOP, loopback(in6, 30, 'big'), True), + ('RAW', RAW, FRAGMENT, True), + ('RAW, 4in6', RAW, in6, True), + ('IPV4', IPV4, FRAGMENT, True), + ('IPV6, 4in6', IPV6, in6, True), + ('IPV6, an IPv4 packet', IPV6, FRAGMENT, False), + ('RAW, cut short of the header', RAW, FRAGMENT[:15], False), + ('12', 12, FRAGMENT, False), + ('LINUX_SLL', LINUX_SLL, FRAGMENT, False), + ): + with self.subTest(case=label), mock.patch.dict('sys.modules', stand_ins()): + data = ipv4_reassembly(unread(frame, linktype), count=3) + if not read: + self.assertIsNone(data) + continue + self.assertIsNotNone(data) + self.assertEqual(data.bufid, (SRC, DST, 0x1234, TransType.UDP)) + self.assertEqual((data.num, data.fo, data.mf, data.tl), (3, 0, True, len(FRAGMENT))) + self.assertEqual((data.header, data.payload), (FRAGMENT[:20], bytearray(FRAGMENT[20:]))) + + def test_tcp_over_ipv6_is_left_out_with_a_warning_over_each_link_type(self) -> None: + for label, linktype, frame in ( + ('NULL, AF_INET6 24', NULL, loopback(TCP6, 24, 'little')), + ('NULL, AF_INET6 28', NULL, loopback(TCP6, 28, 'little')), + ('NULL, AF_INET6 30, big-endian', NULL, loopback(TCP6, 30, 'big')), + ('LOOP', LOOP, loopback(TCP6, 24, 'big')), + ('RAW', RAW, TCP6), + ('RAW, 6in4', RAW, wire.ipv4(TCP6, proto=41)), + ('IPV6', IPV6, TCP6), + ): + with self.subTest(case=label), mock.patch.dict('sys.modules', stand_ins()): + messages = self.decline(unread(frame, linktype)) + self.assertEqual(len(messages), 1, messages) + self.assertTrue(messages[0].startswith('Frame 1: TCP over IPv6 is left out'), messages) + + def test_packet2dict_reports_an_unread_frames_octets(self) -> None: + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit.pypcapfile import packet2dict + + self.assertEqual(len(TCP4), 47) + for linktype in (RAW, LINUX_SLL, ETHERNET): + with self.subTest(linktype=linktype): + link = LinkType.get(linktype) + info = packet2dict(unread(TCP4, linktype), data_link=link) + self.assertEqual(info[link.name], {'raw_len': 47, 'raw': TCP4}) + + +@unittest.skipUnless(base.HAS_RUNTIME, 'runtime dependencies not installed') +class PyPCAPFileFrameLayerBoundTests(unittest.TestCase): + """C.f. #1609.""" + + decline = base.PyPCAPFileTCPOverIPv6Tests.decline + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_tcp_is_read_only_within_the_frame_layer_bound(self) -> None: + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit.pypcapfile import tcp_reassembly, tcp_traceflow + + # each with the layer the TCP header is at, Ethernet or the loopback + # header being the first + cases = [ + ('in 62 IPv4', 64, ethernet(nest4(base.TCP_SEGMENT, 62, 6))), + ('in 63 IPv4', 65, ethernet(nest4(base.TCP_SEGMENT, 63, 6))), + ('in 100 IPv4', 102, ethernet(nest4(base.TCP_SEGMENT, 100, 6))), + ('behind 31 (IPv4, AH) pairs', 64, ethernet(ah_pairs(31))), + ('behind 32 (IPv4, AH) pairs', 66, ethernet(ah_pairs(32))), + ('4in6 in 61 IPv6', 64, ethernet6(nest6(TCP4, 61, 4))), + ('4in6 in 62 IPv6', 65, ethernet6(nest6(TCP4, 62, 4))), + ('in 31 IPv4 behind 31 headers', 64, ethernet(nest4(chain(31, 6) + base.TCP_SEGMENT, 31, 60))), + ('in 31 IPv4 behind 32 headers', 65, ethernet(nest4(chain(32, 6) + base.TCP_SEGMENT, 31, 60))), + ('in 40 IPv4 behind 23 headers', 65, ethernet(nest4(chain(23, 6) + base.TCP_SEGMENT, 40, 60))), + ('RAW, in 63 IPv4', 64, unread(nest4(base.TCP_SEGMENT, 63, 6), RAW)), + ('RAW, in 64 IPv4', 65, unread(nest4(base.TCP_SEGMENT, 64, 6), RAW)), + ('NULL, in 62 IPv4', 64, unread(loopback(nest4(base.TCP_SEGMENT, 62, 6), 2, 'little'), NULL)), + ('NULL, in 63 IPv4', 65, unread(loopback(nest4(base.TCP_SEGMENT, 63, 6), 2, 'little'), NULL)), + ('IPV6, 4in6 in 62 IPv6', 64, unread(nest6(TCP4, 62, 4), IPV6)), + ('IPV6, 4in6 in 63 IPv6', 65, unread(nest6(TCP4, 63, 4), IPV6)), + ] + for count, tpid in ((61, base.C_TAG), (62, base.C_TAG), (61, base.S_TAG), (62, base.S_TAG), + (63, base.C_TAG)): + cases.append((f'behind {count} tags of {tpid:#06x}', count + 3, tags(TCP4, 0x0800, count, tpid))) + for label, layer, packet in cases: + with self.subTest(case=label), mock.patch.dict('sys.modules', stand_ins()): + data = tcp_reassembly(packet, count=1) + trace = tcp_traceflow(packet, data_link=LinkType.ETHERNET, count=1) + if layer > 64: + self.assertIsNone(data) + self.assertIsNone(trace) + self.assertEqual(self.decline(packet), []) + continue + self.assertIsNotNone(data) + self.assertIsNotNone(trace) + self.assertEqual(data.header, base.TCP_SEGMENT[:20]) + + def test_ipv4_fragments_are_read_only_within_the_frame_layer_bound(self) -> None: + from pcapkit.toolkit.pypcapfile import ipv4_reassembly + + # each with the layer the IPv4 header is at + cases = [ + ('in 62 IPv6', 64, ethernet6(nest6(FRAGMENT, 62, 4))), + ('in 63 IPv6', 65, ethernet6(nest6(FRAGMENT, 63, 4))), + ('RAW, in 63 IPv6', 64, unread(nest6(FRAGMENT, 63, 4), RAW)), + ('RAW, in 64 IPv6', 65, unread(nest6(FRAGMENT, 64, 4), RAW)), + ('LOOP, in 62 IPv6', 64, unread(loopback(nest6(FRAGMENT, 62, 4), 24, 'big'), LOOP)), + ('LOOP, in 63 IPv6', 65, unread(loopback(nest6(FRAGMENT, 63, 4), 24, 'big'), LOOP)), + ('in 61 IPv6 behind a header', 64, ethernet6(nest6(chain(1, 4) + FRAGMENT, 61, 60))), + ('in 61 IPv6 behind 2 headers', 65, ethernet6(nest6(chain(2, 4) + FRAGMENT, 61, 60))), + ] + for count, tpid in ((62, base.C_TAG), (63, base.C_TAG), (62, base.S_TAG), (63, base.S_TAG)): + cases.append((f'behind {count} tags of {tpid:#06x}', count + 2, tags(FRAGMENT, 0x0800, count, tpid))) + for label, layer, packet in cases: + with self.subTest(case=label), mock.patch.dict('sys.modules', stand_ins()): + data = ipv4_reassembly(packet, count=1) + if layer > 64: + self.assertIsNone(data) + else: + self.assertIsNotNone(data) + self.assertEqual(data.header, FRAGMENT[:20]) + + def test_tcp_over_ipv6_warns_only_within_the_frame_layer_bound(self) -> None: + # The default engine finds no TCP past the bound, so none is left out there. + cases = [ + ('behind 31 (IPv6, Hop-by-Hop) pairs', 64, ethernet6(hbh_pairs(31))), + ('behind 32 (IPv6, Hop-by-Hop) pairs', 66, ethernet6(hbh_pairs(32))), + ('behind 61 tags', 64, tags(TCP6, 0x86DD, 61, base.C_TAG)), + ('behind 62 tags', 65, tags(TCP6, 0x86DD, 62, base.C_TAG)), + ('in 40 IPv6 behind 22 headers', 64, ethernet6(nest6(chain(22, 6) + base.TCP_SEGMENT, 40, 60))), + ('in 40 IPv6 behind 23 headers', 65, ethernet6(nest6(chain(23, 6) + base.TCP_SEGMENT, 40, 60))), + ('6in4 in 61 IPv4', 64, ethernet(nest4(TCP6, 61, 41))), + ('6in4 in 62 IPv4', 65, ethernet(nest4(TCP6, 62, 41))), + ('RAW, behind 31 (IPv6, Hop-by-Hop) pairs', 63, unread(hbh_pairs(31), RAW)), + ('RAW, in 63 IPv6', 64, unread(nest6(base.TCP_SEGMENT, 63, 6), RAW)), + ('RAW, in 64 IPv6', 65, unread(nest6(base.TCP_SEGMENT, 64, 6), RAW)), + ] + for label, layer, packet in cases: + with self.subTest(case=label), mock.patch.dict('sys.modules', stand_ins()): + messages = self.decline(packet) + self.assertEqual(len(messages), int(layer <= 64), messages) + + def test_the_walks_read_nothing_past_the_bound(self) -> None: + from pcapkit.toolkit import pypcapfile + + with self.subTest(case='1000 VLAN tags'): + self.assertIsNone(pypcapfile._untag(tags(TCP4, 0x0800, 1000, base.C_TAG).packet)) + self.assertEqual(pypcapfile._untag(tags(TCP4, 0x0800, 62, base.C_TAG).packet)[2], 64) + with self.subTest(case='a chain past the bound'), mock.patch.dict('sys.modules', stand_ins()): + # the innermost IPv4 header is at layer 41, so 23 headers fit behind it + packet = ethernet(nest4(chain(32, 6) + base.TCP_SEGMENT, 40, 60)) + with mock.patch.object(pypcapfile, '_default_extension_header', + wraps=pypcapfile._default_extension_header) as parse: + self.assertIsNone(pypcapfile.tcp_reassembly(packet, count=1)) + # no header past the bound is asked of the default engine's parser + self.assertEqual(len({call.args for call in parse.call_args_list}), 23) + with self.subTest(case='the helpers other toolkits import'): + # They read a header as the first layer of its frame, as before; + # the same header deeper in a frame is read past the bound. + inner = wire.ipv4(chain(32, 6) + base.TCP_SEGMENT, proto=60) + self.assertEqual(pypcapfile._upper_layer(4, inner), (6, base.TCP_SEGMENT)) + self.assertEqual(pypcapfile._past_extension_headers(60, inner[20:]), (6, base.TCP_SEGMENT)) + self.assertEqual(pypcapfile._walk_upper_layer(4, inner, 1), (6, base.TCP_SEGMENT, 34)) + self.assertIsNone(pypcapfile._walk_upper_layer(4, inner, 41)) + self.assertIsNone(pypcapfile._walk_ipv4_payload(60, inner[20:], 41)) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/toolkit/test_pypcapfile_unit.py b/tests/toolkit/test_pypcapfile_unit.py index fc9949f42c..b49334cb60 100644 --- a/tests/toolkit/test_pypcapfile_unit.py +++ b/tests/toolkit/test_pypcapfile_unit.py @@ -129,10 +129,11 @@ def __init__(self, payload) -> None: def make_packet(layer, *, timestamp: int = 1511106545, timestamp_us: int = 471719, - ns_resolution: bool = False, capture_len: int = 86, packet_len: int = 86): - """Build a stand-in for :class:`pcapfile.structs.pcap_packet`.""" + ns_resolution: bool = False, capture_len: int = 86, packet_len: int = 86, + linktype: int = 1): + """Build a stand-in for :class:`pcapfile.structs.pcap_packet`, of a capture of ``linktype``.""" return types.SimpleNamespace( - header=[types.SimpleNamespace(ns_resolution=ns_resolution)], + header=[types.SimpleNamespace(ns_resolution=ns_resolution, ll_type=linktype)], timestamp=timestamp, timestamp_us=timestamp_us, capture_len=capture_len, @@ -790,11 +791,11 @@ def test_tunnelled_tcp_over_ipv4_is_found_in_the_innermost_ipv4(self) -> None: ): with self.subTest(case=label), mock.patch.dict('sys.modules', fake_ip_decoder()) as modules: IP = modules['pcapfile.protocols.network.ip'].IP - ipv4, upper = _innermost(make_packet(frame)) + ipv4, upper, depth = _innermost(make_packet(frame)) self.assertIsInstance(ipv4, IP) self.assertIsNone(upper) self.assertEqual(IP.calls[-1], (inner, 0)) - self.assertEqual(_transport(ipv4), segment) + self.assertEqual(_transport(ipv4, depth), segment) def test_no_warning_where_the_default_engine_finds_no_tcp(self) -> None: from unittest import mock @@ -964,7 +965,7 @@ def spy_init(this, *args, **kwargs): packets = [make_packet(ipv6_frame(cut)) for _ in range(10)] # a TCP header cut short in a tunnel for packet in packets[1:]: packet.header = packets[0].header # one capture, which warns once - with mock.patch.object(toolkit, '_upper_layer', wraps=toolkit._upper_layer) as walk, \ + with mock.patch.object(toolkit, '_walk_upper_layer', wraps=toolkit._walk_upper_layer) as walk, \ mock.patch.object(ProtocolBase, '__init__', spy_init), \ warnings.catch_warnings(record=True) as caught: warnings.simplefilter('always') From 53f1c156457eda8d9e048bf0ab2b748d4b94af21 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Tue, 13 Oct 2026 16:06:44 -0400 Subject: [PATCH 2/2] docs(toolkit): pypcapfile reads loopback and raw IP, and stops at the frame layer bound (#1609, #1667) - protocol.rst listed pypcapfile with pyshark as not applying FRAME_LAYER_LIMIT. It now stands with dpkt and scapy; pyshark's sentence is otherwise unchanged. - The toolkit note names the link types the adapters read and the layer bound, and documents READ_LINKTYPES. The engine note says the toolkit reads the IP packet of loopback and raw-IP frames. - The previous commit's message says 12 and LINUX_SLL are the only link types still warned of. Every link type but Ethernet and the five READ_LINKTYPES still is; the engine test now checks PPP (9), 105 and 147 as well. The pypcapfile tests pass on 3.11 and the touched tests on 3.14; the engine test still fails on main on both. Closes #1667 Part of #1609 --- .../source/pcapkit/foundation/engines/3rdparty.rst | 4 +++- docs/source/pcapkit/protocols/protocol.rst | 6 +++--- docs/source/pcapkit/toolkit/3rdparty.rst | 14 ++++++++++++++ tests/foundation/engines/test_pypcapfile_engine.py | 6 ++++-- 4 files changed, 24 insertions(+), 6 deletions(-) diff --git a/docs/source/pcapkit/foundation/engines/3rdparty.rst b/docs/source/pcapkit/foundation/engines/3rdparty.rst index 0eb686f10b..95e6eee903 100644 --- a/docs/source/pcapkit/foundation/engines/3rdparty.rst +++ b/docs/source/pcapkit/foundation/engines/3rdparty.rst @@ -399,7 +399,9 @@ support, as is used by :class:`pcapkit.foundation.extraction.Extractor`. TCP and UDP and nothing else. IPv6 reassembly is therefore unavailable and is disabled -- with an :class:`~pcapkit.utilities.warnings.AttributeWarning` -- when requested; IPv4 and TCP reassembly and TCP flow tracing remain - available. PCAP-NG is rejected with a + available. The IP packet of a BSD loopback or raw-IP frame, which it does + not decode, is read by :mod:`pcapkit.toolkit.pypcapfile` instead + (:issue:`1667`). PCAP-NG is rejected with a :exc:`~pcapkit.utilities.exceptions.FormatError`. `PyPCAPFile`_ 0.12.0 drops every record of a big-endian savefile, every record from a zero-length one on, and a truncated final record, so the engine reads the records itself, diff --git a/docs/source/pcapkit/protocols/protocol.rst b/docs/source/pcapkit/protocols/protocol.rst index 8448888702..fe0f3b7b63 100644 --- a/docs/source/pcapkit/protocols/protocol.rst +++ b/docs/source/pcapkit/protocols/protocol.rst @@ -116,9 +116,9 @@ there back out, as they do one the TCP parser rejected (:issue:`1518`). A protocol constructed from its own octets is not counted itself, so the count starts at the first layer it dissects. One chain of extension headers is also bounded on its own, by :data:`~pcapkit.protocols.internet.internet.EXTENSION_HEADER_LIMIT`. -The ``pyshark`` and ``pypcapfile`` engines dissect with their own parsers, -which do not apply this bound (:issue:`1609`). The ``dpkt`` and ``scapy`` -toolkits count the layers they walk as the default engine does, and stop at it. +The ``pyshark`` engine dissects with its own parser, which does not apply this +bound (:issue:`1609`). The ``dpkt``, ``scapy`` and ``pypcapfile`` toolkits +count the layers they walk as the default engine does, and stop at it. .. autodata:: pcapkit.protocols.protocol.FRAME_LAYER_LIMIT diff --git a/docs/source/pcapkit/toolkit/3rdparty.rst b/docs/source/pcapkit/toolkit/3rdparty.rst index eea167ab8b..cdf164b77c 100644 --- a/docs/source/pcapkit/toolkit/3rdparty.rst +++ b/docs/source/pcapkit/toolkit/3rdparty.rst @@ -193,6 +193,17 @@ cannot be used for it. :func:`~pcapkit.toolkit.pypcapfile.tcp_traceflow`, with an :class:`~pcapkit.utilities.warnings.AttributeWarning` once per capture. + Nor does it decode a link layer but Ethernet. The adapters read the IP packet + of a BSD loopback (``NULL``, ``LOOP``) or raw-IP (``RAW``, ``IPV4``, + ``IPV6``) frame themselves, as the default engine dissects it (see + :data:`~pcapkit.toolkit.pypcapfile.READ_LINKTYPES`, :issue:`1667`). Every + other link type is left unread, as the default engine leaves it, and the + engine warns of it with an :class:`~pcapkit.utilities.warnings.AttributeWarning`. + Nor do the adapters read deeper than the default engine dissects: each VLAN + tag, IP header, AH and IPv6 extension header is one of the + :data:`~pcapkit.protocols.protocol.FRAME_LAYER_LIMIT` layers of a frame, and + nothing past the last of them is read (:issue:`1609`). + .. autofunction:: pcapkit.toolkit.pypcapfile.ipv4_reassembly .. autofunction:: pcapkit.toolkit.pypcapfile.ipv6_reassembly @@ -215,6 +226,9 @@ Auxiliary Functions Internal Definitions -------------------- +.. autodata:: pcapkit.toolkit.pypcapfile.READ_LINKTYPES + :no-value: + .. autodata:: pcapkit.toolkit.pypcapfile.TCP_MIN_HEADER_LEN .. autodata:: pcapkit.toolkit.pypcapfile.IPV4_FLAG_DF diff --git a/tests/foundation/engines/test_pypcapfile_engine.py b/tests/foundation/engines/test_pypcapfile_engine.py index e4b9535f74..5126a22a0a 100644 --- a/tests/foundation/engines/test_pypcapfile_engine.py +++ b/tests/foundation/engines/test_pypcapfile_engine.py @@ -304,9 +304,11 @@ def test_run_warns_and_gives_up_decoding_for_an_unknown_link_layer(self) -> None def test_run_warns_only_of_a_link_layer_left_unread(self) -> None: # C.f. #1667: pcapfile decodes none of NULL, LOOP, RAW, IPV4 and IPV6, # but the toolkit reads the IP packet in their frames, as the default - # engine does, so nothing is lost to warn of. 12 and LINUX_SLL stay unread. + # engine does, so nothing is lost to warn of. Every other link type it + # has no decoder for, e.g. PPP (9), 12, 105, LINUX_SLL and 147, stays unread. for linktype, unread in ((0, False), (108, False), (101, False), (228, False), (229, False), - (12, True), (113, True), (1, True)): + (9, True), (12, True), (105, True), (113, True), (147, True), + (1, True)): with self.subTest(linktype=linktype): extractor, _ = self.make_extractor() engine = self.engine(extractor, savefile=FakeSaveFile(ll_type=linktype), decoder=None)