diff --git a/pcapkit/toolkit/dpkt.py b/pcapkit/toolkit/dpkt.py index 6e016913e..a14279e06 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 882fb84ba..4f1ff332f 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 000000000..9d05018bd --- /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()