From 9a01482eccc895a5e49ae950181e34db974493f0 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Tue, 13 Oct 2026 03:05:09 -0400 Subject: [PATCH] fix(toolkit): dpkt reads IPv6 in IPv4, and nothing behind MPLS or Cisco ISL (#1637, #1638, #1646) - #1637: ipv6_reassembly took the network layer alone, so an IPv6 fragment carried in IPv4 gave no input. It now takes _first_ipv6, which walks IPv4 tunnels and the headers the default engine reads past, stops at a rejected IPv4 header, a later IPv4 fragment or any other protocol, and cuts the octets at each Total Length above. The IPv6 packet then ends at its Payload Length, or at the Jumbo Payload Length of a zero one; a zero one with no Jumbo option gives no input. The Fragment header must be among the first EXTENSION_HEADER_LIMIT extension headers and within FRAME_LAYER_LIMIT layers of the frame, as the default engine walks; _walk_extension_headers counts the headers in IPv4 it steps over. - #1638: _network took any ip/ip6 that dpkt's Ethernet named, including one behind MPLS, Cisco ISL or a legacy QinQ TPID, which the default engine reads as raw octets. _ethernet_accepted now refuses those, reading ISL from the captured octets once an inner 802.1Q tag has replaced it. - dpkt keeps only its second go's tags of a stack such as 802.1Q then 802.1ad, so _layer2bytes looked for the IP header too early and raised ProtocolError. _wire_hdr_len now adds the dropped tags, and _ethernet_accepted reads the TPIDs, off the captured octets. New tests/toolkit/test_dpkt_6in4_reassembly_and_mpls_isl_1637_1638_unit.py compares dpkt with the default engine over 97 cases; it fails without the fix. Closes #1637 Closes #1638 Closes #1646 --- pcapkit/toolkit/dpkt.py | 286 +++++++++- pcapkit/toolkit/pypcapfile.py | 20 +- ..._reassembly_and_mpls_isl_1637_1638_unit.py | 504 ++++++++++++++++++ 3 files changed, 792 insertions(+), 18 deletions(-) create mode 100644 tests/toolkit/test_dpkt_6in4_reassembly_and_mpls_isl_1637_1638_unit.py diff --git a/pcapkit/toolkit/dpkt.py b/pcapkit/toolkit/dpkt.py index 6e016913e5..a14279e065 100644 --- a/pcapkit/toolkit/dpkt.py +++ b/pcapkit/toolkit/dpkt.py @@ -23,8 +23,9 @@ from pcapkit.foundation.traceflow.data.data import FrameRecord from pcapkit.foundation.traceflow.data.tcp import Packet as TF_TCP_Packet from pcapkit.protocols.data.misc.pcap.frame import FrameInfo as Data_FrameInfo -from pcapkit.toolkit.pypcapfile import (IPV4_EXTENSION_HEADERS, TCP_MIN_HEADER_LEN, - _default_accepts, _past_extension_headers) +from pcapkit.toolkit.pypcapfile import (IPV4_EXTENSION_HEADERS, TCP_MIN_HEADER_LEN, VLAN_TAG_LEN, + VLAN_TAGS, _default_accepts, _past_extension_headers, + _walk_extension_headers) from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall if TYPE_CHECKING: @@ -33,6 +34,7 @@ from typing import Any, Optional from dpkt.dpkt import Packet + from dpkt.ethernet import Ethernet from dpkt.ip import IP from dpkt.ip6 import IP6, IP6FragmentHeader from dpkt.tcp import TCP @@ -80,11 +82,26 @@ #: IPv6 alike: IPv4 (protocol 4, :rfc:`2003`) and IPv6 (protocol 41, #: :rfc:`2473` and :rfc:`4213`). `DPKT`_ also decodes an IPv4 packet's #: protocol 0 as IPv4, its ``IP_PROTO_IP``, where the default engine reads an -#: IPv6 Hop-by-Hop Options header, so :func:`_tcp_carrier` does not enter it. +#: IPv6 Hop-by-Hop Options header, so neither :func:`_tcp_carrier` nor +#: :func:`_first_ipv6` enters it. #: #: .. _DPKT: https://dpkt.readthedocs.io IP_TUNNELS = frozenset((int(Enum_TransType.IPv4), int(Enum_TransType.IPv6))) +#: Destination address prefixes by which `DPKT`_ reads an Ethernet frame whose +#: EtherType field is an 802.3 length, at most 1500, as Cisco ISL; the default +#: engine reads one as raw octets (see :func:`_ethernet_accepted`). +#: +#: .. _DPKT: https://dpkt.readthedocs.io +ISL_ADDRESSES = (b'\x01\x00\x0c\x00\x00', b'\x03\x00\x0c\x00\x00') + +#: TPIDs of the VLAN tags `DPKT`_ steps over in an Ethernet II frame: 802.1Q, +#: 802.1ad and the legacy QinQ ``0x9100`` and ``0x9200``, i.e. +#: ``dpkt.ethernet._ETH_TYPES_QINQ`` (see :func:`_vlan_tpids`). +#: +#: .. _DPKT: https://dpkt.readthedocs.io +DPKT_VLAN_TAGS = frozenset((0x8100, 0x88A8, 0x9100, 0x9200)) + class RecordTimestamp: """A record's capture timestamp, with what a PCAP frame record needs of it. @@ -325,12 +342,49 @@ def _ipv6_ext_hdrs(ipv6: 'IP6') -> 'list[Any]': return list(getattr(ipv6, 'extension_hdrs', {}).values()) -def _wire_hdr_len(packet: 'Packet', payload: 'Packet') -> 'int': +def _vlan_tpids(wire: 'bytes | memoryview') -> 'list[int]': + """Read the TPIDs of the VLAN tags `DPKT`_ steps over in an Ethernet II frame, off its octets. + + Args: + wire: Octets of an Ethernet frame, from its destination address on. + + Returns: + The frame's EtherType, then each tag's but the last's, for as long as + each is one of the :data:`DPKT_VLAN_TAGS` and a whole tag follows it; + empty for a frame whose EtherType field is an 802.3 length. + + Note: + :class:`dpkt.ethernet.Ethernet` reads at most two tags at a time, and + reads on only while the tag it read names 802.1Q. Where it then has no + decoder for the EtherType, its constructor reads again from there, and + starts its :attr:`~dpkt.ethernet.Ethernet.vlan_tags` afresh: so it keeps + only the tags of its last read, one for 802.1Q then 802.1ad, two + 802.1ad, or three tags, and its re-serialisation is 4 octets short for + each tag it dropped. Where it decodes the payload, the tags it stepped + over are all those heading it, since none of the :data:`DPKT_VLAN_TAGS` + has a decoder, so they are read off the octets instead. + + .. _DPKT: https://dpkt.readthedocs.io + + """ + tpids = [] # type: list[int] + offset = 12 + while len(wire) >= offset + 2 + VLAN_TAG_LEN: + tpid = int.from_bytes(wire[offset:offset + 2], 'big') + if tpid not in DPKT_VLAN_TAGS: + break + tpids.append(tpid) + offset += VLAN_TAG_LEN + return tpids + + +def _wire_hdr_len(packet: 'Packet', payload: 'Packet', wire: 'Optional[bytes | memoryview]' = None) -> 'int': """Calculate the octets a layer occupies before its payload begins. Args: packet: DPKT packet, i.e. one layer of a frame. payload: That layer's payload, i.e. :attr:`packet.data `. + wire: The captured octets of ``packet``, from its first on, if any. Returns: The advance :meth:`~dpkt.dpkt.Packet.unpack` itself made to reach the @@ -349,14 +403,24 @@ def _wire_hdr_len(packet: 'Packet', payload: 'Packet') -> 'int': Every other layer's ``len()`` does bound its payload, so the difference of the two is used there. For IPv4 that difference is the Internet - Header Length, options included. + Header Length, options included. For an Ethernet II frame whose + ``wire`` is given, 4 octets are added for each VLAN tag + :class:`dpkt.ethernet.Ethernet` stepped over but dropped from its + :attr:`~dpkt.ethernet.Ethernet.vlan_tags`, which ``len()`` counts (see + :func:`_vlan_tpids`). """ + import dpkt # isort:skip + if hasattr(packet, 'extension_hdrs'): return packet.__hdr_len__ + sum( ext_hdr.length for ext_hdr in _ipv6_ext_hdrs(cast('IP6', packet)) ) - return len(packet) - len(payload) + hdr_len = len(packet) - len(payload) + if wire is not None and isinstance(packet, dpkt.ethernet.Ethernet) and int.from_bytes(wire[12:14], 'big') > 1500: + dropped = len(_vlan_tpids(wire)) - len(getattr(packet, 'vlan_tags', None) or ()) + hdr_len += VLAN_TAG_LEN * dropped + return hdr_len def _layer2bytes(packet: 'Packet', layer: 'Packet') -> 'tuple[bytes, bool]': @@ -390,11 +454,12 @@ def _layer2bytes(packet: 'Packet', layer: 'Packet') -> 'tuple[bytes, bool]': if buffer is not None: offset = 0 current = packet + view = memoryview(cast('bytes', buffer)) while current is not layer: payload = getattr(current, 'data', None) if payload is None or isinstance(payload, (bytes, bytearray)): break - offset += _wire_hdr_len(current, payload) + offset += _wire_hdr_len(current, payload, view[offset:]) current = payload else: wire = cast('bytes', buffer)[offset:] @@ -409,6 +474,54 @@ def _layer2bytes(packet: 'Packet', layer: 'Packet') -> 'tuple[bytes, bool]': return packet2bytes(layer), False +def _ethernet_accepted(packet: 'Ethernet') -> 'bool': + """Test if the default engine reads past a DPKT Ethernet frame's link layer as `DPKT`_ did. + + Args: + packet: DPKT Ethernet frame. + + Returns: + :data:`False` if `DPKT`_ reached its payload past an MPLS label stack, a + Cisco ISL header, or a VLAN tag whose TPID is neither of the + :data:`~pcapkit.toolkit.pypcapfile.VLAN_TAGS`; else :data:`True`. + + Note: + `DPKT`_ pops an MPLS label stack and guesses IPv4 or IPv6 from the + nibble after it, unwraps an ISL header, and steps over the legacy QinQ + tags ``0x9100`` and ``0x9200``, naming the IPv4 or IPv6 packet it finds + ``ip`` or ``ip6`` as for an untagged frame. The default engine reads + what follows any of them as raw octets (see :attr:`Link.__proto__ + `), an ISL frame's 802.3 + length field included (:issue:`1638`). A tag's TPID is the EtherType + before it: the frame's, then each tag's but the last's. + + `DPKT`_ records an ISL header as a tag, but unwraps it into the frame + itself, so a VLAN tag of the frame inside replaces it; the captured + octets attached by :func:`attach_buffer` still show it, by the + :data:`ISL_ADDRESSES` and length field `DPKT`_ recognises it by. Nor + does it keep every tag of a stack it reads in two goes, so the TPIDs + are read off those octets too (see :func:`_vlan_tpids`). + + .. _DPKT: https://dpkt.readthedocs.io + + """ + import dpkt # isort:skip + + if getattr(packet, 'mpls_labels', None): + return False + tags = list(getattr(packet, 'vlan_tags', None) or ()) + if any(isinstance(tag, dpkt.ethernet.VLANtagISL) for tag in tags): + return False + buffer = getattr(packet, BUFFER_ATTR, None) # type: Optional[bytes] + if buffer is None: + tpids = [packet.type, *(tag.type for tag in tags)][:len(tags)] + elif buffer[:5] in ISL_ADDRESSES and int.from_bytes(buffer[12:14], 'big') <= 1500: + return False + else: + tpids = _vlan_tpids(buffer) + return all(tpid in VLAN_TAGS for tpid in tpids) + + def _network(packet: 'Packet') -> 'Optional[IP | IP6]': """Fetch the network layer of a DPKT packet. @@ -419,7 +532,9 @@ def _network(packet: 'Packet') -> 'Optional[IP | IP6]': The packet itself if it is an IPv4 or IPv6 packet, as `DPKT`_ reads every frame of a raw-IP link type (:issue:`1548`); otherwise the IPv4 or IPv6 payload of its link layer, which `DPKT`_ names ``ip`` or ``ip6`` -- a - bare IP packet names neither -- or :data:`None` if it has none. + bare IP packet names neither -- or :data:`None` if it has none, or the + default engine dissects no payload of an Ethernet frame where `DPKT`_ + found one (see :func:`_ethernet_accepted`). .. _DPKT: https://dpkt.readthedocs.io @@ -428,11 +543,50 @@ def _network(packet: 'Packet') -> 'Optional[IP | IP6]': if isinstance(packet, (dpkt.ip.IP, dpkt.ip6.IP6)): return packet + if isinstance(packet, dpkt.ethernet.Ethernet) and not _ethernet_accepted(packet): + return None if hasattr(packet, 'ip'): return cast('IP', packet.ip) return getattr(packet, 'ip6', None) +def _network_depth(packet: 'Packet') -> 'int': + """Count the layers of a DPKT packet the default engine dissects up to its network layer. + + Args: + packet: DPKT packet, i.e. the outermost layer, whose network layer + :func:`_network` found. + + Returns: + The place of that layer in the frame, its first layer being ``1``, as + the default engine counts towards + :data:`~pcapkit.protocols.protocol.FRAME_LAYER_LIMIT`: ``1`` for the + packet of a raw-IP link type, which is the network layer itself; else + ``2``, and one more for each VLAN tag of an Ethernet frame, which the + default engine dissects as a layer of its own. + + Note: + The tags are read off the captured octets, as `DPKT`_ keeps only some + of a stack it reads in two goes (see :func:`_vlan_tpids`). Each is + 802.1Q or 802.1ad, or :func:`_ethernet_accepted` refuses the frame. + + .. _DPKT: https://dpkt.readthedocs.io + + """ + import dpkt # isort:skip + + if isinstance(packet, (dpkt.ip.IP, dpkt.ip6.IP6)): + return 1 + tags = 0 + if isinstance(packet, dpkt.ethernet.Ethernet): + buffer = getattr(packet, BUFFER_ATTR, None) # type: Optional[bytes] + if buffer is None: + tags = len(getattr(packet, 'vlan_tags', None) or ()) + else: + tags = len(_vlan_tpids(buffer)) + return 2 + tags + + def _ipv4_accepted(ipv4: 'IP') -> 'bool': """Test if the default engine reads a DPKT IPv4 packet as an IPv4 layer. @@ -778,6 +932,79 @@ def _first_ipv4(packet: 'Packet') -> 'tuple[Optional[IP], Optional[int]]': return ip, extent +def _first_ipv6(packet: 'Packet') -> 'tuple[Optional[IP6], Optional[tuple[bytes, bool]], int]': + """Fetch the first IPv6 layer of a DPKT packet, as the default engine dissects it. + + Args: + packet: DPKT packet, i.e. the outermost layer. + + Returns: + The network layer (see :func:`_network`) if it is IPv6, else the first + IPv6 packet its IPv4 packet carries (:rfc:`4213`) -- directly, behind + the headers the default engine reads past (see + :func:`~pcapkit.toolkit.pypcapfile._past_extension_headers`), or inside + more IPv4 (:rfc:`2003`). Beside it, :data:`None` for the former, whose + octets are those :func:`_layer2bytes` slices, and for the latter its + octets from its first on, as far as the Total Length of every IPv4 + packet carrying it goes, and whether they are the captured ones; then + its place in the frame (see :func:`_network_depth`), one more for each + IPv4 packet and each of those headers before it. Both are :data:`None`, + and the place ``0``, if there is no such packet, or the default engine + dissects nothing past an IPv4 packet on the way: one whose header it + rejects (see :func:`_ipv4_accepted`), a later fragment + (:rfc:`791#section-3.2`), or one carrying any other protocol, such as + the ICMP error that quotes an IPv6 packet. So too if the IPv6 packet is + past :data:`~pcapkit.protocols.protocol.FRAME_LAYER_LIMIT`, which the + default engine keeps raw (:issue:`1610`). + + Note: + The default engine takes ``frame['IPv6']``, the first IPv6 layer however + deep (:func:`pcapkit.toolkit.pcap.ipv6_reassembly`), and an IPv4 packet + hands its payload the octets its Total Length leaves -- a Total Length + of ``0``, which TCP segmentation offload leaves, the rest of what + carries it -- so they are cut here as it cuts them (:issue:`1637`). + `DPKT`_ decodes the IPv6 packet an IPv4 packet carries directly, but + reads protocol 0 as IPv4 and keeps the IPv6 extension headers as octets, + so one behind an AH or extension header is decoded from the octets + instead (see :func:`_decode_ip`), as in :func:`_tcp_carrier`. + + .. _DPKT: https://dpkt.readthedocs.io + + """ + import dpkt # isort:skip + + from pcapkit.protocols.protocol import FRAME_LAYER_LIMIT # isort:skip + + ip = _network(packet) + if ip is None: + return None, None, 0 + octets = None # type: Optional[tuple[bytes, bool]] # of ``ip``, once sliced + depth = _network_depth(packet) # of ``ip`` + while isinstance(ip, dpkt.ip.IP) and depth <= FRAME_LAYER_LIMIT: + if ip.p not in IP_TUNNELS and ip.p not in IPV4_EXTENSION_HEADERS: + return None, None, 0 + if not _ipv4_accepted(ip) or ip.offset: + return None, None, 0 + wire, captured = octets or _layer2bytes(packet, ip) + ihl = ip.hl * 4 + payload = wire[ihl:max(ip.len, ihl)] if ip.len else wire[ihl:] + if ip.p in IPV4_EXTENSION_HEADERS: + walked = _walk_extension_headers(ip.p, payload) + if walked is None or walked[0] not in IP_TUNNELS: + return None, None, 0 + proto, upper, hops = walked + if (inner := _decode_ip(proto, upper)) is None: + return None, None, 0 + ip, octets, depth = inner, (upper, captured), depth + 1 + hops + elif isinstance(ip.data, (dpkt.ip.IP, dpkt.ip6.IP6)): + ip, octets, depth = ip.data, (payload, captured), depth + 1 + else: + return None, None, 0 + if not hasattr(ip, 'extension_hdrs') or depth > FRAME_LAYER_LIMIT: + return None, None, 0 + return cast('IP6', ip), octets, depth + + def ipv4_reassembly(packet: 'Packet', timestamp: 'float', *, count: 'int' = -1) -> 'IP_Packet[IPv4Address] | None': """Make data for IPv4 reassembly. @@ -878,8 +1105,14 @@ def ipv6_reassembly(packet: 'Packet', timestamp: 'float', *, Data for IPv6 reassembly. * If the ``packet`` can be used for IPv6 reassembly. A packet can be reassembled - if it contains IPv6 layer (:class:`dpkt.ip6.IP6`) and IPv6 Fragment header - (:rfc:`2460#section-4.5`, i.e., :class:`dpkt.ip6.IP6FragmentHeader`). + if it contains IPv6 layer (:class:`dpkt.ip6.IP6`) -- the first of the frame, as + for the default engine, so one carried in IPv4 counts (see :func:`_first_ipv6`) + -- and IPv6 Fragment header (:rfc:`2460#section-4.5`, i.e., + :class:`dpkt.ip6.IP6FragmentHeader`) inside its Payload Length, and among its first + :data:`~pcapkit.protocols.internet.internet.EXTENSION_HEADER_LIMIT` extension + headers, the most the default engine walks (:issue:`1604`), and inside the first + :data:`~pcapkit.protocols.protocol.FRAME_LAYER_LIMIT` layers of the frame, past + which it dissects nothing (:issue:`1610`). * If the ``packet`` can be reassembled, then the :obj:`dict` mapping of data for IPv6 reassembly (:term:`reasm.ipv6.packet`) will be returned; otherwise, returns :data:`None`. @@ -892,22 +1125,41 @@ def ipv6_reassembly(packet: 'Packet', timestamp: 'float', *, :func:`attach_buffer` is what makes them the wire's again. The packet itself is left untouched either way. + The captured octets end at the Payload Length, as the default engine + cuts them, or at the Jumbo Payload Length (:rfc:`2675`) of a zero one. + A zero Payload Length with no Jumbo Payload option leaves the default + engine no extension header to read, and so no input; `DPKT`_ reads the + rest of the frame instead. + See Also: :class:`pcapkit.foundation.reassembly.ipv6.IPv6` + .. _DPKT: https://dpkt.readthedocs.io + """ - network = _network(packet) - ipv6 = cast('IP6', network) if hasattr(network, 'extension_hdrs') else None + from pcapkit.protocols.internet.internet import EXTENSION_HEADER_LIMIT # isort:skip + from pcapkit.protocols.protocol import FRAME_LAYER_LIMIT # isort:skip + from pcapkit.protocols.schema.internet.ipv6 import jumbo_payload_length # isort:skip + + ipv6, octets, depth = _first_ipv6(packet) if ipv6 is not None: ipv6_frag = ipv6.extension_hdrs.get(44) # type: Optional[IP6FragmentHeader] if ipv6_frag is None: # dismiss not fragmented packet return None + # the default engine walks no more extension headers than either bound + # leaves, each header one layer of the frame past the IPv6 header's + bound = min(EXTENSION_HEADER_LIMIT, FRAME_LAYER_LIMIT - depth) + if not any(ext_hdr is ipv6_frag for ext_hdr in _ipv6_ext_hdrs(ipv6)[:bound]): + return None hdr_len = ipv6_hdr_len(ipv6) - wire, captured = _layer2bytes(packet, ipv6) - # a zero Payload Length is a jumbogram or TCP segmentation offload, in - # which case the datagram runs to the end of the frame, as with DPKT - if captured and ipv6.plen: - wire = wire[:ipv6.__hdr_len__ + ipv6.plen] + wire, captured = octets or _layer2bytes(packet, ipv6) + if captured: + plen = ipv6.plen or jumbo_payload_length(ipv6.nxt, wire[ipv6.__hdr_len__:]) + if not plen: + return None + wire = wire[:ipv6.__hdr_len__ + plen] + if len(wire) < hdr_len + ipv6_frag.__hdr_len__: # Fragment header cut short + return None # payload following the IPv6 Fragment header payload = wire[hdr_len + ipv6_frag.__hdr_len__:] diff --git a/pcapkit/toolkit/pypcapfile.py b/pcapkit/toolkit/pypcapfile.py index 882fb84bab..4f1ff332fb 100644 --- a/pcapkit/toolkit/pypcapfile.py +++ b/pcapkit/toolkit/pypcapfile.py @@ -671,6 +671,24 @@ def _past_extension_headers(protocol: 'int', data: 'bytes') -> 'Optional[tuple[i IPv6, :func:`_upper_layer` steps over these as the extension headers they are there. + """ + walked = _walk_extension_headers(protocol, data) + return None if walked is None else walked[:2] + + +def _walk_extension_headers(protocol: 'int', data: 'bytes') -> '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. + + 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`. + """ depth = 0 while protocol in IPV4_EXTENSION_HEADERS: @@ -687,7 +705,7 @@ def _past_extension_headers(protocol: 'int', data: 'bytes') -> 'Optional[tuple[i if header is None: return None protocol, data = header[0], data[header[1]:] - return protocol, data + return protocol, data, depth def _upper_layer(version: 'int', data: 'bytes') -> 'Optional[tuple[int, bytes]]': diff --git a/tests/toolkit/test_dpkt_6in4_reassembly_and_mpls_isl_1637_1638_unit.py b/tests/toolkit/test_dpkt_6in4_reassembly_and_mpls_isl_1637_1638_unit.py new file mode 100644 index 0000000000..9d05018bdd --- /dev/null +++ b/tests/toolkit/test_dpkt_6in4_reassembly_and_mpls_isl_1637_1638_unit.py @@ -0,0 +1,504 @@ +# -*- coding: utf-8 -*- +"""dpkt reads IPv6 in IPv4, and nothing behind MPLS or Cisco ISL, as the default engine does. C.f. #1637, #1638. + +The default engine hands IPv6 reassembly the first IPv6 layer of a frame however +deep, so an IPv6 fragment carried in IPv4 (6in4, RFC 4213) is an input, with the +octets inside the Total Length of each IPv4 packet above it. dpkt took the +network layer only, so gave no input for 6in4, 6in4 inside 4in4 or 6in4 behind +an IPv4 AH (#1637). It now walks the IPv4 tunnels and the headers the default +engine reads past, and stops where that engine does: at an IPv4 header it +rejects, a later IPv4 fragment, or any other protocol, an ICMP error quoting the +packet included. The IPv6 packet ends at its Payload Length or, for a zero one, +at its Jumbo Payload Length, as the default engine reads it; a zero Payload +Length with no Jumbo Payload option leaves that engine no Fragment header to +read, carried in IPv4 or not. Nor does that engine walk more than 32 extension +headers (#1604), so a Fragment header past them is no input either, carried in +IPv4 or not (#1609), nor dissect a frame past its 64th layer (#1610), so neither +is one that deep in nested IPv4. + +dpkt's Ethernet decoder pops an MPLS label stack, unwraps a Cisco ISL header and +steps over the legacy QinQ tags ``0x9100`` and ``0x9200``, naming the IP packet +behind them ``ip`` or ``ip6``. The default engine reads all of them as raw octets, +so dpkt handed TCP reassembly, flow tracing and IP reassembly packets it never +reads, and an ISL frame carrying an 802.1Q tag ended the extraction with a +``ProtocolError`` (#1638). dpkt now takes an Ethernet frame's payload only past +the 802.1Q and 802.1ad tags the default engine dissects, and nothing else. + +That decoder also keeps only some tags of a stack it reads in two goes, e.g. +802.1Q then 802.1ad, or three tags, so the IP packet behind them was looked for +too early in the frame, and ``ProtocolError`` ended the extraction. The tags are +now read off the captured octets. + +Every capture is built here, so the module is in the unit tier. + +""" +from __future__ import annotations + +import importlib.util +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 wire +from tests.toolkit.test_dpkt_ipv4_reassembly_input_1615_1616_unit import ipv4, ipv6, jumbo +from tests.toolkit.test_tcp_carrier_walk_1606_1617_unit import ah, options, tcp, v4 + +if TYPE_CHECKING: + from typing import Any + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) +HAS_DPKT = importlib.util.find_spec('dpkt') is not None + +#: An Ethernet trailer, e.g. an FCS, which no IP packet reaches. +TRAILER = b'\xee' * 6 + + +def fragment(ident: int, *, offset: int = 0, mf: bool = True, nxt: int = 253) -> bytes: + """An IPv6 Fragment header; ``offset`` is in octets.""" + return bytes([nxt, 0]) + (offset | mf).to_bytes(2, 'big') + ident.to_bytes(4, 'big') + + +def first(ident: int, *, ext: bytes = b'', nxt: int = 44, plen: 'int | None' = None) -> bytes: + """The first fragment of a 24-octet datagram, behind ``ext``, its Payload Length as given.""" + return ipv6(ext + fragment(ident) + b'A' * 16, nxt, plen=plen) + + +def last(ident: int) -> bytes: + """The last fragment of that datagram.""" + return ipv6(fragment(ident, offset=16, mf=False) + b'B' * 8, 44) + + +def eth(packet: bytes, *, trailer: bytes = b'') -> bytes: + """An Ethernet frame carrying the IPv4 or IPv6 ``packet``, then ``trailer``.""" + return wire.ethernet(packet, 0x86DD if packet[0] >> 4 == 6 else 0x0800) + trailer + + +def tagged(tpids: 'list[int]', packet: bytes) -> bytes: + """An Ethernet frame whose VLAN tags have the TPIDs ``tpids``, then the IP ``packet``.""" + types = list(tpids) + [0x86DD if packet[0] >> 4 == 6 else 0x0800] + tags = b''.join((100 + i).to_bytes(2, 'big') + types[i + 1].to_bytes(2, 'big') for i in range(len(tpids))) + return wire.MAC_B + wire.MAC_A + types[0].to_bytes(2, 'big') + tags + packet + + +def mpls(packet: bytes, *, labels: int = 1, ethertype: int = 0x8847, tpid: 'int | None' = None) -> bytes: + """An Ethernet frame carrying ``packet`` behind ``labels`` MPLS labels, and an 802.1Q tag if ``tpid``.""" + stack = b''.join(((16 + i) << 12 | (i == labels - 1) << 8 | 64).to_bytes(4, 'big') for i in range(labels)) + tag = b'' if tpid is None else (100).to_bytes(2, 'big') + ethertype.to_bytes(2, 'big') + return wire.MAC_B + wire.MAC_A + (ethertype if tpid is None else tpid).to_bytes(2, 'big') + tag + stack + packet + + +def isl(frame: bytes) -> bytes: + """A Cisco ISL frame: its 26-octet header, the encapsulated ``frame`` and a 4-octet CRC.""" + body = b'\xaa\xaa\x03\x00\x00\x0c' + (1 << 1).to_bytes(2, 'big') + bytes(4) + frame + return b'\x01\x00\x0c\x00\x00\x00' + wire.MAC_A + len(body).to_bytes(2, 'big') + body + bytes(4) + + +def icmp_unreach(quote: bytes) -> bytes: + """An ICMP Destination Unreachable message quoting ``quote``.""" + return bytes([3, 1, 0, 0, 0, 0, 0, 0]) + quote + + +def gre(packet: bytes) -> bytes: + """A GRE header carrying the IPv6 ``packet``.""" + return b'\x00\x00\x86\xdd' + packet + + +def ipv6_in_ipv4() -> 'dict[str, tuple[list[int], list[bytes]]]': + """IPv6 fragments carried in IPv4, each case with the payload lengths the default engine hands over. + + Every case's datagram has an Identification of its own. + + """ + cases = {} # type: dict[str, tuple[list[int], list[bytes]]] + ident = iter(range(1, 1000)) + + def pair(label: str, carry: 'Any') -> None: + number = next(ident) + cases[label] = ([16, 8], [carry(first(number)), carry(last(number))]) + + def one(label: str, payloads: 'list[int]', carry: 'Any') -> None: + cases[label] = (payloads, [carry(next(ident))]) + + # #1637: the first IPv6 layer, in IPv4, in 4in4, and behind the headers the default engine reads past + one('IPv6 (control)', [16], lambda n: eth(first(n))) + pair('6in4', lambda p: eth(v4(p, 41))) + pair('6in4 inside 4in4', lambda p: eth(v4(v4(p, 41), 4))) + pair('6in4 behind an IPv4 AH', lambda p: eth(v4(ah(41) + p, 51))) + one('6in4 behind a Hop-by-Hop header in IPv4', [16], lambda n: eth(v4(options(41) + first(n), 0))) + one('6in4 behind an AH inside 4in4', [16], lambda n: eth(v4(v4(ah(41) + first(n), 51), 4))) + one('4in4 behind an AH, carrying 6in4', [16], lambda n: eth(v4(ah(4) + v4(first(n), 41), 51))) + one('6in4 in a first IPv4 fragment', [16], lambda n: eth(v4(first(n), 41, mf=True, ident=n))) + one('6in4 with IPv4 options', [16], + lambda n: eth(ipv4(first(n), ident=n, options=b'\x94\x04\x00\x00', proto=41))) + one('6in4, Hop-by-Hop then Fragment header', [16], lambda n: eth(v4(first(n, ext=options(44), nxt=0), 41))) + one('6in4 behind an 802.1Q tag', [16], lambda n: tagged([0x8100], v4(first(n), 41))) + # whose octets end at each Total Length above, or at the IPv6 packet's own length + one('6in4 with a trailer', [16], lambda n: eth(v4(first(n), 41), trailer=TRAILER)) + one('6in4, Total Length 0, with a trailer', [16], lambda n: eth(v4(first(n), 41, length=0), trailer=TRAILER)) + one('6in4 cut by the Total Length', [6], + lambda n: eth(v4(first(n), 41, length=20 + 48 + 6), trailer=TRAILER)) + one('6in4 cut by the outer Total Length of 4in4', [6], + lambda n: eth(v4(v4(first(n), 41), 4, length=40 + 48 + 6), trailer=TRAILER)) + one('6in4 behind an AH, cut by the Total Length', [6], + lambda n: eth(v4(ah(41) + first(n), 51, length=20 + 24 + 48 + 6), trailer=TRAILER)) + one('6in4 jumbogram, Total Length 0, with a trailer', [16], + lambda n: eth(v4(first(n, ext=jumbo(44, 8 + 24), nxt=0, plen=0), 41, length=0), trailer=TRAILER)) + one('6in4 jumbogram shorter than the packet', [4], + lambda n: eth(v4(first(n, ext=jumbo(44, 8 + 8 + 4), nxt=0, plen=0), 41))) + # where the default engine dissects no Fragment header of the first IPv6 layer + one('6in4, not fragmented', [], lambda n: eth(v4(ipv6(b'A' * 16, 253), 41))) + one('ICMP quoting a 6in4 fragment', [], lambda n: eth(v4(icmp_unreach(v4(first(n), 41)), 1))) + one('6in4 in a later IPv4 fragment', [], lambda n: eth(v4(first(n), 41, offset=8, ident=n))) + # dpkt decodes no later fragment's payload, but the IPv6 behind an AH or + # Hop-by-Hop header is decoded from the octets, so the fragment check is ours + one('6in4 behind an AH in a later IPv4 fragment', [], + lambda n: eth(v4(ah(41) + first(n), 51, offset=8, ident=n))) + one('6in4 behind a Hop-by-Hop header in a later IPv4 fragment', [], + lambda n: eth(v4(options(41) + first(n), 0, offset=8, ident=n))) + one('6in4 behind rejected IPv4 options', [], + lambda n: eth(ipv4(first(n), ident=n, options=b'\x02\x03\x05\xb4', proto=41))) + one('6in4, Total Length below the IHL', [], lambda n: eth(v4(first(n), 41, length=12))) + one('6in4, Total Length inside the IPv6 header', [], lambda n: eth(v4(first(n), 41, length=20 + 30))) + one('6in4 behind an AH too short', [], lambda n: eth(v4(ah(41, 0) + first(n), 51))) + one('6in4 with a zero Payload Length', [], lambda n: eth(v4(first(n, plen=0), 41), trailer=TRAILER)) + one('6in4 jumbogram ending inside the Fragment header', [], + lambda n: eth(v4(first(n, ext=jumbo(44, 8 + 4), nxt=0, plen=0), 41))) + one('4in6 carrying a 6in4 fragment', [], lambda n: eth(ipv6(v4(first(n), 41), 4))) + one('6in4 in GRE', [], lambda n: eth(v4(gre(first(n)), 47))) + one('6in4 behind MPLS', [], lambda n: mpls(v4(first(n), 41))) + return cases + + +def payload_length() -> 'dict[str, tuple[list[int], list[bytes]]]': + """IPv6 fragments in Ethernet with a zero Payload Length, each with the payload lengths handed over.""" + return { + 'a zero Payload Length': ([], [eth(first(901, plen=0))]), + 'a zero Payload Length, with a trailer': ([], [eth(first(902, plen=0), trailer=TRAILER)]), + 'a jumbogram, with a trailer': ([16], [eth(first(903, ext=jumbo(44, 8 + 24), nxt=0, plen=0), + trailer=TRAILER)]), + 'a jumbogram shorter than the packet': ([4], [eth(first(904, ext=jumbo(44, 8 + 8 + 4), nxt=0, plen=0))]), + 'a jumbogram ending inside the Fragment header': ([], [eth(first(905, ext=jumbo(44, 8 + 4), nxt=0, + plen=0))]), + 'a Payload Length short of the packet (control)': ([6], [eth(first(906, plen=8 + 6), trailer=TRAILER)]), + } + + +def chain_bound() -> 'dict[str, tuple[list[int], list[bytes]]]': + """IPv6 first fragments behind Destination Options headers, each with the payload lengths handed over. + + The default engine walks 32 extension headers at most, so a Fragment header + after 32 or more others is not among them. + + """ + cases = {} # type: dict[str, tuple[list[int], list[bytes]]] + ident = iter(range(501, 1000)) + for count, payloads in ((31, [16]), (32, []), (33, []), (40, [])): + ext = options(60) * (count - 1) + options(44) + label = f'{count} Destination Options headers' + (' (control)' if payloads else '') + cases[f'IPv6, {label}'] = (payloads, [eth(first(next(ident), ext=ext, nxt=60))]) + cases[f'6in4, {label}'] = (payloads, [eth(v4(first(next(ident), ext=ext, nxt=60), 41))]) + cases[f'6in4 inside 4in4, {label}'] = (payloads, [eth(v4(v4(first(next(ident), ext=ext, nxt=60), 41), 4))]) + return cases + + +def nested(packet: bytes, count: int, *, proto: int = 41) -> bytes: + """``packet`` carried as ``proto`` in the innermost of ``count`` nested IPv4 headers.""" + packet = v4(packet, proto) + for _ in range(count - 1): + packet = v4(packet, 4) + return packet + + +def layer_limit() -> 'dict[str, tuple[list[int], list[bytes]]]': + """IPv6 first fragments deep in nested IPv4 over Ethernet, each with the payload lengths handed over. + + The default engine dissects no layer past the 64th of a frame (#1610). The + first is Ethernet, then each VLAN tag, IPv4 header, header in IPv4 it reads + past, and the IPv6 header, then each extension header before the Fragment + header, which must be within the 64 too. + + """ + cases = {} # type: dict[str, tuple[list[int], list[bytes]]] + ident = iter(range(1001, 2000)) + + def one(label: str, payloads: 'list[int]', build: 'Any', tpids: 'tuple[int, ...]' = ()) -> None: + packet = build(next(ident)) + cases[label + (' (control)' if payloads else '')] = (payloads, [tagged(list(tpids), packet) if tpids + else eth(packet)]) + + for count, payloads in ((61, [16]), (62, [])): + one(f'6in4 in {count} IPv4', payloads, lambda n, c=count: nested(first(n), c)) + one(f'6in4 behind an AH in {count - 1} IPv4', payloads, + lambda n, c=count: nested(ah(41) + first(n), c - 1, proto=51)) + for count, payloads in ((60, [16]), (61, [])): + one(f'802.1Q, 6in4 in {count} IPv4', payloads, lambda n, c=count: nested(first(n), c), (0x8100,)) + for count, payloads in ((59, [16]), (60, [])): + # a stack dpkt reads in two goes, so keeps one tag of + one(f'802.1Q, 802.1ad, 6in4 in {count} IPv4', payloads, lambda n, c=count: nested(first(n), c), + (0x8100, 0x88A8)) + # where the extension headers before the Fragment header reach the 64th layer + # first, and where each header the default engine reads past in IPv4 counts + for count, chain, payloads in ((40, 21, [16]), (40, 22, []), (31, 30, [16]), (31, 31, [])): + ext = options(60) * (chain - 1) + options(44) + one(f'6in4 in {count} IPv4, {chain} Destination Options headers', payloads, + lambda n, c=count, e=ext: nested(first(n, ext=e, nxt=60), c)) + for hops, chain, payloads in ((10, 1, [16]), (10, 2, []), (11, 0, [16]), (12, 0, [])): + ext = options(60) * (chain - 1) + options(44) if chain else b'' + one(f'6in4 behind {hops} Hop-by-Hop headers in 50 IPv4, {chain} Destination Options headers', payloads, + lambda n, h=hops, c=chain, e=ext: nested(options(0) * (h - 1) + options(41) + + first(n, ext=e, nxt=60 if c else 44), 50, proto=0)) + return cases + + +def raw_layer_limit() -> 'dict[str, tuple[list[int], list[bytes]]]': + """IPv6 first fragments deep in nested IPv4 of a raw-IP link type, whose first layer is the IPv4 header.""" + return { + '6in4 in 62 IPv4 (control)': ([16], [nested(first(2001), 62)]), + '6in4 in 63 IPv4': ([], [nested(first(2002), 63)]), + '6in4 in 40 IPv4, 22 Destination Options headers (control)': + ([16], [nested(first(2003, ext=options(60) * 21 + options(44), nxt=60), 40)]), + '6in4 in 40 IPv4, 23 Destination Options headers': + ([], [nested(first(2004, ext=options(60) * 22 + options(44), nxt=60), 40)]), + } + + +#: VLAN tag stacks by name, each with whether the default engine reads past them. +#: dpkt reads each of them in two goes and keeps only the tags of the second. +STACKS = { + '802.1Q, 802.1ad': (True, [0x8100, 0x88A8]), + 'two 802.1ad': (True, [0x88A8, 0x88A8]), + 'three 802.1Q': (True, [0x8100] * 3), + '802.1ad, 802.1Q, 802.1Q': (True, [0x88A8, 0x8100, 0x8100]), + 'four 802.1Q': (True, [0x8100] * 4), + # where the tag dpkt drops is the one the default engine leaves raw + '802.1Q, legacy QinQ 0x9100': (False, [0x8100, 0x9100]), + '802.1ad, 802.1Q, legacy QinQ 0x9100': (False, [0x88A8, 0x8100, 0x9100]), +} + + +def stacked(tpids: 'list[int]') -> 'list[bytes]': + """Frames tagged ``tpids``: IPv4 and IPv6 TCP, then IPv4, IPv6 and 6in4 fragment pairs; then an untagged control.""" + return [ + tagged(tpids, v4(tcp(1), 6)), tagged(tpids, wire.ipv6(tcp(2), nxt=6)), + tagged(tpids, v4(b'A' * 16, 253, mf=True, ident=31)), tagged(tpids, v4(b'B' * 8, 253, offset=16, ident=31)), + tagged(tpids, first(32)), tagged(tpids, last(32)), + tagged(tpids, v4(first(33), 41)), tagged(tpids, v4(last(33), 41)), + eth(v4(tcp(3), 6)), + ] + + +def links() -> 'dict[str, tuple[tuple[int, int, int], list[bytes]]]': + """Frames by link-layer encapsulation, each with the TCP, IPv4 and IPv6 inputs the default engine hands over. + + Every TCP segment has a port of its own, and every fragmented datagram an + Identification of its own. + + """ + def each(build: 'Any', ident: int) -> 'list[bytes]': + return [build(v4(b'A' * 16, 253, mf=True, ident=ident)), build(v4(b'B' * 8, 253, offset=16, ident=ident))] + + def each6(build: 'Any', ident: int) -> 'list[bytes]': + return [build(first(ident)), build(last(ident))] + + return { + # an IPv4 packet that may be fragmented is an IPv4 reassembly input too + 'IPv4, TCP (control)': ((1, 1, 0), [eth(v4(tcp(1), 6))]), + '802.1Q, IPv4, TCP (control)': ((1, 1, 0), [tagged([0x8100], v4(tcp(2), 6))]), + '802.1ad, 802.1Q, IPv6, TCP (control)': ((1, 0, 0), [tagged([0x88A8, 0x8100], wire.ipv6(tcp(3), nxt=6))]), + 'MPLS, IPv4, TCP': ((0, 0, 0), [mpls(v4(tcp(4), 6))]), + 'MPLS, IPv6, TCP': ((0, 0, 0), [mpls(wire.ipv6(tcp(5), nxt=6))]), + 'MPLS multicast, IPv4, TCP': ((0, 0, 0), [mpls(v4(tcp(6), 6), ethertype=0x8848)]), + 'two MPLS labels, IPv4, TCP': ((0, 0, 0), [mpls(v4(tcp(7), 6), labels=2)]), + '802.1Q, MPLS, IPv4, TCP': ((0, 0, 0), [mpls(v4(tcp(8), 6), tpid=0x8100)]), + 'Cisco ISL, Ethernet, IPv4, TCP': ((0, 0, 0), [isl(eth(v4(tcp(9), 6)))]), + 'Cisco ISL, Ethernet, IPv6, TCP': ((0, 0, 0), [isl(eth(wire.ipv6(tcp(10), nxt=6)))]), + 'legacy QinQ 0x9100, IPv4, TCP': ((0, 0, 0), [tagged([0x9100], v4(tcp(11), 6))]), + 'legacy QinQ 0x9200, 802.1Q, IPv4, TCP': ((0, 0, 0), [tagged([0x9200, 0x8100], v4(tcp(12), 6))]), + 'LLC/SNAP, IPv4, TCP (control)': ((0, 0, 0), [ + wire.MAC_B + wire.MAC_A + (8 + 44).to_bytes(2, 'big') + b'\xaa\xaa\x03\x00\x00\x00\x08\x00' + + v4(tcp(13), 6)]), + 'MPLS, IPv4 fragments': ((0, 0, 0), each(mpls, 21)), + 'MPLS, IPv6 fragments': ((0, 0, 0), each6(mpls, 22)), + 'Cisco ISL, Ethernet, IPv4 fragments': ((0, 0, 0), each(lambda p: isl(eth(p)), 23)), + 'Cisco ISL, Ethernet, IPv6 fragments': ((0, 0, 0), each6(lambda p: isl(eth(p)), 24)), + 'legacy QinQ 0x9100, IPv4 fragments': ((0, 0, 0), each(lambda p: tagged([0x9100], p), 25)), + '802.1Q, IPv4 fragments (control)': ((0, 2, 0), each(lambda p: tagged([0x8100], p), 26)), + '802.1Q, IPv6 fragments (control)': ((0, 0, 2), each6(lambda p: tagged([0x8100], p), 27)), + } + + +def read(test: unittest.TestCase, engine: str, frames: 'list[bytes]', linktype: int = 1) -> 'dict[str, Any]': + """Read ``frames`` of ``linktype`` with ``engine``, with IP and TCP reassembly and TCP flow tracing. + + Returns: + Every input the engine hands IPv4, IPv6 and TCP reassembly and TCP flow + tracing, keyed by frame number -- every field, but a trace input's + ``frame`` -- the reassembled datagrams, and the traced flows. + + """ + from pcapkit import extract + from pcapkit.foundation.reassembly.ipv4 import IPv4 + from pcapkit.foundation.reassembly.ipv6 import IPv6 + from pcapkit.foundation.reassembly.tcp import TCP + from pcapkit.foundation.traceflow.traceflow import TraceFlowBase + + seen = {'IPv4': {}, 'IPv6': {}, 'TCP': {}, 'trace': {}} # type: dict[str, Any] + calls = {klass: klass.__call__ for klass in (IPv4, IPv6, TCP)} + trace = TraceFlowBase.__call__ + + 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} + + def spy(klass: 'Any') -> 'Any': + def call(this: 'Any', packet: 'Any') -> 'Any': + seen[klass.__name__][packet.num] = fields(packet) + return calls[klass](this, packet) + return call + + def spy_trace(this: 'Any', packet: 'Any') -> 'Any': + seen['trace'][packet.index] = fields(packet, ('frame',)) + return trace(this, packet) + + with tempfile.TemporaryDirectory() as tmp, warnings.catch_warnings(), \ + mock.patch.object(IPv4, '__call__', spy(IPv4)), mock.patch.object(IPv6, '__call__', spy(IPv6)), \ + mock.patch.object(TCP, '__call__', spy(TCP)), mock.patch.object(TraceFlowBase, '__call__', spy_trace): + path = os.path.join(tmp, 'encap.pcap') + with open(path, 'wb') as file: + file.write(wire.pcap([(1, number, octets) for number, octets in enumerate(frames)], linktype=linktype)) + warnings.simplefilter('ignore') + extractor = extract(fin=path, nofile=True, engine=engine, reassembly=True, ip=True, tcp=True, + trace=True, trace_fout=os.path.join(tmp, 'trace'), trace_format='json') + try: + if extractor._exnam != engine: # pylint: disable=protected-access + test.skipTest(f'{engine} did not run') + seen['datagrams'] = { + 'IPv4': [(datagram.completed, datagram.id, datagram.index, datagram.header, datagram.payload) + for datagram in extractor.reassembly.ipv4], + 'IPv6': [(datagram.completed, datagram.id, datagram.index, datagram.header, datagram.payload) + for datagram in extractor.reassembly.ipv6], + 'TCP': [datagram.id for datagram in extractor.reassembly.tcp], + } + seen['flows'] = [(flow.label, tuple(flow.index)) for flow in extractor.trace.tcp] + finally: + close_extractor(extractor) + return seen + + +def numbered(cases: 'dict[str, tuple[Any, list[bytes]]]') -> 'tuple[list[bytes], dict[str, list[int]]]': + """The frames of ``cases`` in order, and each case's one-based frame numbers.""" + frames, numbers = [], {} # type: list[bytes], dict[str, list[int]] + for label, (_, built) in cases.items(): + numbers[label] = list(range(len(frames) + 1, len(frames) + 1 + len(built))) + frames.extend(built) + return frames, numbers + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +@unittest.skipUnless(HAS_DPKT, 'dpkt not installed') +class DPKTEncapsulationTests(unittest.TestCase): + """dpkt's reassembly and trace inputs are the default engine's, field for field.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def agree_on_ipv6(self, cases: 'dict[str, tuple[list[int], list[bytes]]]', linktype: int = 1) -> None: + frames, numbers = numbered(cases) + base, dpkt = read(self, 'default', frames, linktype), read(self, 'dpkt', frames, linktype) + for label, (payloads, _) in cases.items(): + with self.subTest(case=label): + got = [base['IPv6'][number] for number in numbers[label] if number in base['IPv6']] + self.assertEqual([len(packet['payload']) for packet in got], payloads, 'the default engine') + self.assertEqual([dpkt['IPv6'].get(number) for number in numbers[label]], + [base['IPv6'].get(number) for number in numbers[label]]) + for aspect in ('IPv4', 'TCP', 'trace', 'datagrams', 'flows'): + self.assertEqual(dpkt[aspect], base[aspect], aspect) + + def test_ipv6_carried_in_ipv4_is_an_input(self) -> None: + # #1637 + self.agree_on_ipv6(ipv6_in_ipv4()) + + def test_a_zero_payload_length_gives_way_to_a_jumbo_payload_option(self) -> None: + # #1637: the rule the 6in4 rows need, which holds for IPv6 in Ethernet too + self.agree_on_ipv6(payload_length()) + + def test_a_fragment_header_past_the_chain_bound_is_no_input(self) -> None: + # #1604, #1609: dpkt walks the whole chain, the default engine 32 headers + self.agree_on_ipv6(chain_bound()) + + def test_a_fragment_header_past_the_frame_layer_limit_is_no_input(self) -> None: + # #1610: the default engine dissects 64 layers of a frame, dpkt any number + self.agree_on_ipv6(layer_limit()) + + def test_a_raw_ip_frame_counts_no_link_layer(self) -> None: + # #1610: LINKTYPE_RAW, so the outermost IPv4 header is the first layer + self.agree_on_ipv6(raw_layer_limit(), linktype=101) + + def test_a_stack_dpkt_reads_in_two_goes_ends_no_extraction(self) -> None: + # dpkt kept the second go's tags only, so the IP header was looked for 4 + # octets early per tag dropped, and ProtocolError ended the read + for label, (reads, tpids) in STACKS.items(): + with self.subTest(stack=label): + frames = stacked(tpids) + base, dpkt = read(self, 'default', frames), read(self, 'dpkt', frames) + self.assertEqual((sorted(base['TCP']), sorted(base['IPv6'])), + ([1, 2, 9], [5, 6, 7, 8]) if reads else ([9], []), 'the default engine') + for aspect in ('TCP', 'trace', 'IPv4', 'IPv6', 'datagrams', 'flows'): + self.assertEqual(dpkt[aspect], base[aspect], aspect) + + def test_nothing_is_read_behind_mpls_isl_or_legacy_qinq(self) -> None: + # #1638 + cases = links() + frames, numbers = numbered(cases) + base, dpkt = read(self, 'default', frames), read(self, 'dpkt', frames) + for label, (counts, _) in cases.items(): + with self.subTest(case=label): + self.assertEqual(tuple(sum(number in base[aspect] for number in numbers[label]) + for aspect in ('TCP', 'IPv4', 'IPv6')), counts, 'the default engine') + for aspect in ('TCP', 'trace', 'IPv4', 'IPv6'): + self.assertEqual([dpkt[aspect].get(number) for number in numbers[label]], + [base[aspect].get(number) for number in numbers[label]], aspect) + for aspect in ('datagrams', 'flows'): + self.assertEqual(dpkt[aspect], base[aspect], aspect) + + def test_an_isl_frame_carrying_an_8021q_tag_ends_no_extraction(self) -> None: + # #1638: dpkt overwrote the ISL tag with the 802.1Q one, so the IPv4 + # header was looked for 26 octets early and ProtocolError ended the read + frames = [isl(tagged([0x8100], v4(tcp(1), 6))), eth(v4(tcp(2), 6))] + base, dpkt = read(self, 'default', frames), read(self, 'dpkt', frames) + self.assertEqual((sorted(base['TCP']), sorted(base['trace'])), ([2], [2])) + for aspect in ('TCP', 'trace', 'IPv4', 'IPv6', 'datagrams', 'flows'): + self.assertEqual(dpkt[aspect], base[aspect], aspect) + + def test_a_packet_built_by_hand_is_read_too(self) -> None: + # with no record octets attached, DPKT's own fields tell what it read + # past, and the octets are its serialisation of the packet, which here + # is the octets it was parsed from + import dpkt + + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit import dpkt as toolkit + + for label, octets in (('MPLS', mpls(v4(tcp(1), 6))), ('Cisco ISL', isl(eth(v4(tcp(1), 6)))), + ('legacy QinQ', tagged([0x9100], v4(tcp(1), 6)))): + with self.subTest(case=label): + packet = dpkt.ethernet.Ethernet(octets) + self.assertEqual(type(packet.ip).__name__, 'IP') # what dpkt decodes + self.assertIsNone(toolkit.tcp_reassembly(packet, 1.0, count=1)) + self.assertIsNone(toolkit.tcp_traceflow(packet, 1.0, data_link=LinkType.ETHERNET, count=1)) + + tunnelled = eth(v4(first(1), 41), trailer=TRAILER) + data = toolkit.ipv6_reassembly(dpkt.ethernet.Ethernet(tunnelled), 1.0, count=1) + self.assertIsNotNone(data) + assert data is not None + self.assertEqual((data.ihl, bytes(data.header), bytes(data.payload)), + (40, tunnelled[34:74], b'A' * 16)) + + +if __name__ == '__main__': + unittest.main()