diff --git a/docs/source/ext.rst b/docs/source/ext.rst index 534f3fb76..30c83a2f8 100644 --- a/docs/source/ext.rst +++ b/docs/source/ext.rst @@ -422,7 +422,10 @@ The available engines and the file formats they read: matters is that the gap is *announced* -- each engine warns, or raises, for the capability it cannot provide, rather than silently producing an empty result, save that `PyPCAPFile`_ leaves TCP behind a VLAN tag, in a tunnel or behind an - extension header (an IPv6 one, or AH in IPv4) out without a warning. See + extension header (an IPv6 one, or AH in IPv4) out without a warning, and the + `Scapy`_ engine reads IPv6 only as far as `Scapy`_ dissects it, which leaves + IPv6 behind some BSD loopback headers, and TCP and IPv6 fragments behind some + IPv6 extension headers, out without one. See :doc:`pcapkit/foundation/engines/index` for the full table. .. note:: diff --git a/docs/source/index.rst b/docs/source/index.rst index 9971b4338..6394480f4 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -150,7 +150,10 @@ so IPv6 reassembly is unavailable. Each gap is announced with a warning or an exception rather than silently returning nothing, save that :class:`~pcapkit.foundation.engines.pypcapfile.PyPCAPFile` leaves TCP behind a VLAN tag, in a tunnel or behind an extension header (an IPv6 one, or AH in IPv4) -out without one; :doc:`pcapkit/foundation/engines/index` tabulates them. +out without one, and :class:`~pcapkit.foundation.engines.scapy.Scapy` reads IPv6 +only as far as `Scapy`_ dissects it, which leaves IPv6 behind some BSD loopback +headers, and TCP and IPv6 fragments behind some IPv6 extension headers, out +without one; :doc:`pcapkit/foundation/engines/index` tabulates them. Every engine also answers a preflight check -- :meth:`~pcapkit.foundation.engines.engine.EngineBase.unsupported_reason`, which diff --git a/docs/source/pcapkit/foundation/engines/index.rst b/docs/source/pcapkit/foundation/engines/index.rst index 65477ce44..a673c3ada 100644 --- a/docs/source/pcapkit/foundation/engines/index.rst +++ b/docs/source/pcapkit/foundation/engines/index.rst @@ -101,6 +101,11 @@ their row says otherwise: | | PCAP savefiles only, otherwise | | | :exc:`~pcapkit.utilities.exceptions.FormatError` | +-----------------------------------------------------------------+---------------------------------------------------------------+ +| :class:`~pcapkit.foundation.engines.scapy.Scapy` | reads IPv6 only as far as `Scapy`_ dissects it, without a | +| | warning: behind a BSD loopback header only at the family | +| | Scapy binds it to on the host it runs on, and no TCP or | +| | Fragment header behind an AH, Mobility, HIP or Shim6 header | ++-----------------------------------------------------------------+---------------------------------------------------------------+ | :class:`~pcapkit.foundation.engines.dpkt.DPKT`, | no PCAP output, since each frame reaches the writer as a | | :class:`~pcapkit.foundation.engines.scapy.Scapy`, | plain :obj:`dict` -- ``json`` is written instead, with a | | :class:`~pcapkit.foundation.engines.pyshark.PyShark`, | :class:`~pcapkit.utilities.warnings.FormatWarning` | diff --git a/docs/source/pcapkit/protocols/internet/internet.rst b/docs/source/pcapkit/protocols/internet/internet.rst index fde58f976..0b5475b3f 100644 --- a/docs/source/pcapkit/protocols/internet/internet.rst +++ b/docs/source/pcapkit/protocols/internet/internet.rst @@ -37,9 +37,11 @@ A chain of IPv6 extension headers is dissected to at most a row, whether it follows IPv6 or another layer such as IPv4. What follows is kept as :class:`~pcapkit.protocols.misc.raw.Raw`, with one :exc:`~pcapkit.utilities.warnings.ProtocolWarning` (:issue:`1604`). The -``dpkt`` and ``scapy`` toolkits stop at the same depth. The ``pyshark`` engine -leaves the walk to its own parser, which does not stop there, so it may still -find the upper layer past it (:issue:`1609`). The ``pypcapfile`` toolkit walks -no extension header chain, so it finds no upper layer behind one. +``dpkt`` toolkit stops at the same depth. The ``pyshark`` engine leaves the walk +to its own parser, which does not stop there, so it may still find the upper +layer past it (:issue:`1609`). So does the ``scapy`` engine behind IPv6, where +Scapy dissects each extension header itself; behind IPv4, its toolkit steps over +them in the octets to find TCP, and stops at the same depth. The ``pypcapfile`` +toolkit walks no extension header chain, so it finds no upper layer behind one. .. autodata:: pcapkit.protocols.internet.internet.EXTENSION_HEADER_LIMIT diff --git a/docs/source/pcapkit/protocols/protocol.rst b/docs/source/pcapkit/protocols/protocol.rst index 844888870..b2e58af05 100644 --- a/docs/source/pcapkit/protocols/protocol.rst +++ b/docs/source/pcapkit/protocols/protocol.rst @@ -116,9 +116,10 @@ there back out, as they do one the TCP parser rejected (:issue:`1518`). A protocol constructed from its own octets is not counted itself, so the count starts at the first layer it dissects. One chain of extension headers is also bounded on its own, by :data:`~pcapkit.protocols.internet.internet.EXTENSION_HEADER_LIMIT`. -The ``pyshark`` and ``pypcapfile`` engines dissect with their own parsers, -which do not apply this bound (:issue:`1609`). The ``dpkt`` and ``scapy`` -toolkits count the layers they walk as the default engine does, and stop at it. +The ``pyshark``, ``scapy`` and ``pypcapfile`` engines dissect with their own +parsers, which do not apply the frame's layer bound (:issue:`1609`). The +``dpkt`` toolkit counts the layers it walks as the default engine does, and +stops at it. .. autodata:: pcapkit.protocols.protocol.FRAME_LAYER_LIMIT diff --git a/pcapkit/foundation/engines/scapy.py b/pcapkit/foundation/engines/scapy.py index 0cd46a6fb..fd61b4ea0 100644 --- a/pcapkit/foundation/engines/scapy.py +++ b/pcapkit/foundation/engines/scapy.py @@ -112,7 +112,7 @@ def _reader_type() -> 'Type[PcapReader]': to one table across Section Header Blocks, so a later section's interface 0 resolved to the first section's, with its link type and resolution (#1522). - And it dates two kinds of packet as the default engine does. A Simple Packet + And it dates two kinds of packet that `Scapy`_ misdates. A Simple Packet Block has no timestamp, and `Scapy`_ leaves its packet with the time it was built, so it is set to 0 (#1503). `Scapy`_ ignores an interface's ``if_tsoffset``, the seconds to add to each of its timestamps, so the @@ -210,8 +210,8 @@ def read_packet(self, *args: 'Any', **kwargs: 'Any') -> 'ScapyPacket': attach_resolution(packet, self._resolution) attach_linktype(packet, self._linktype) if not self._timed: - # no timestamp at all, which the default engine reads as 0, where - # Scapy leaves the time the packet was built (#1503) + # no timestamp at all, so 0, where Scapy leaves the time the + # packet was built (#1503) packet.time = EDecimal(0) elif self._offset: packet.time += self._offset # if_tsoffset (#1549) diff --git a/pcapkit/foundation/extraction.py b/pcapkit/foundation/extraction.py index ce764671f..a0cd5052b 100644 --- a/pcapkit/foundation/extraction.py +++ b/pcapkit/foundation/extraction.py @@ -1186,9 +1186,9 @@ def __init__(self, # NOTE: the third-party engines' adapters hand the output writer each # frame as a plain :obj:`dict` built by ``packet2dict``, which carries - # neither the frame's octets nor its record header, so there is nothing - # for the PCAP writer to write. The flow tracer has the same limitation, - # and is given the same replacement below. + # no record header, so there is nothing for the PCAP writer to write. + # The flow tracer has the same limitation where an engine's adapters + # build no frame record, and is given the same replacement below. if (self._exnam in ('dpkt', 'scapy', 'pyshark', 'pypcap', 'pcap_ct', 'pypcapfile') and format in ('pcap', 'cap')): warn(f"'Extractor(engine={self._exnam})' does not support 'format={format}'; " diff --git a/pcapkit/toolkit/scapy.py b/pcapkit/toolkit/scapy.py index 293f3cc45..19528b8c7 100644 --- a/pcapkit/toolkit/scapy.py +++ b/pcapkit/toolkit/scapy.py @@ -19,7 +19,7 @@ Several functions here import a `Scapy`_ layer module lazily, inside the function body -- :func:`ipv6_reassembly` needs - :class:`scapy.layers.inet6.IPv6`, for instance. Those imports are + :class:`scapy.layers.inet6.IPv6ExtHdrFragment`, for instance. Those imports are for the *classes* they name and nothing more. They must not be mistaken for how `Scapy`_'s layer registries get populated, even though they do populate them as a side effect, because by the time any of these functions runs the engine has @@ -37,10 +37,8 @@ """ import decimal import ipaddress -import struct -from typing import TYPE_CHECKING, NamedTuple, cast +from typing import TYPE_CHECKING, cast -from pcapkit.const.ipv6.extension_header import ExtensionHeader as Enum_ExtensionHeader from pcapkit.const.reg.linktype import LinkType as Enum_LinkType from pcapkit.const.reg.transtype import TransType as Enum_TransType from pcapkit.foundation.reassembly.data.ip import Packet as IP_Packet @@ -48,12 +46,10 @@ 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.protocols.internet.internet import EXTENSION_HEADER_LIMIT -from pcapkit.protocols.protocol import FRAME_LAYER_LIMIT from pcapkit.protocols.schema.internet.ipv6 import jumbo_payload_length from pcapkit.toolkit.pypcapfile import (IP_TUNNELS, IPV4_EXTENSION_HEADERS, IPV4_HEADER_LEN, - IPV6_EXTENSION_HEADERS, IPV6_HEADER_LEN, TCP_MIN_HEADER_LEN, - _default_accepts, _walk_extension_headers) + IPV6_HEADER_LEN, TCP_MIN_HEADER_LEN, _default_accepts, + _walk_extension_headers) from pcapkit.utilities.compat import ModuleNotFoundError # pylint: disable=redefined-builtin from pcapkit.utilities.exceptions import MissingKeyError, ModuleNotFound, stacklevel from pcapkit.utilities.warnings import ScapyWarning, warn @@ -67,7 +63,7 @@ if TYPE_CHECKING: from ipaddress import IPv4Address, IPv6Address - from typing import Any, Iterator, Optional + from typing import Any from scapy.layers.inet import IP, TCP from scapy.layers.inet6 import IPv6 @@ -87,12 +83,6 @@ #: ``LINKTYPE_*`` number, by :func:`attach_linktype`. LINKTYPE_ATTR = '__pcapkit_linktype__' -#: The BSD loopback address families the default engine dissects IP behind, -#: each mapped to the IP version it carries, as -#: :class:`~pcapkit.protocols.link.loopback.Loopback` reads them (:issue:`1574`): -#: ``AF_INET``, and the ``AF_INET6`` of each of the BSDs. -LOOPBACK_FAMILIES = {2: 4, 24: 6, 28: 6, 30: 6} - def attach_resolution(packet: 'Packet', resolution: 'int') -> 'None': """Stash the timestamp resolution of a packet's capture on the packet. @@ -211,8 +201,8 @@ def packet2frame(packet: 'Packet') -> 'FrameRecord': into (see :func:`_frame`). ``orig_len`` is ``packet.wirelen``, the record's original length. - The fraction in ``frame_info.ts_usec`` is in the capture's own resolution, - as the default engine's is. That resolution is the one + The fraction in ``frame_info.ts_usec`` is in the capture's own + resolution. That resolution is the one :class:`~pcapkit.foundation.engines.scapy.Scapy` attaches with :func:`attach_resolution`, never one guessed from the value: a nanosecond count that is a whole number of microseconds has the same @@ -258,11 +248,9 @@ def _frame(packet: 'Packet') -> 'bytes': Note: `Scapy`_ dissects a packet only from a non-empty string, so a zero-length frame comes back as the first layer its link type names, - every field left to its default, as one built by hand with no fields - would be. ``bytes()`` of it builds a header out of nothing -- 20 octets - at ``RAW`` and ``IPV4``, of an IPv4 layer whose ``ihl``, ``len`` and - ``chksum`` are :data:`None` until then -- where the record has none - (:issue:`1676`). + every field left to its default. ``bytes()`` of it builds a header out + of nothing -- 20 octets at ``RAW`` and ``IPV4``, 14 at ``ETHERNET`` -- + where the record has none (:issue:`1676`). """ original = getattr(packet, 'original', None) @@ -286,18 +274,19 @@ def _within_datagrams(layer: 'Packet') -> 'bytes': datagram's declared length as a :class:`~scapy.packet.Padding` layer -- an Ethernet frame's padding up to its 60-octet minimum, or a trailer such as an FCS -- and ``bytes()`` of *any* layer above that one ends - with it. Those octets are no part of the datagram, and the default - engine never hands them over (:issue:`1584`), so they are cut off here - by the default engine's rules: an IPv4 Total Length of zero, as TCP - segmentation offload leaves, runs the datagram to the end of the frame - (:issue:`1547`), and an IPv6 Payload Length of zero gives way to a - Jumbo Payload option (:issue:`1434`). The latter is not left to Scapy, - which reads that option's length in host byte order and adds the - Hop-by-Hop Options header to it a second time. - - Every IP layer above ``layer`` bounds it, not only the nearest, as each - bounds what the default engine hands the next: a datagram tunnelled in - IP may declare more than the one carrying it holds, or nothing at all. + with it. Those octets are no part of the datagram (:issue:`1584`), so + they are cut off here. An IPv4 Total Length of zero, as TCP + segmentation offload leaves, declares nothing, so the datagram runs to + the end of the frame (:issue:`1547`); an IPv6 Payload Length of zero + gives way to a Jumbo Payload option (:issue:`1434`), whose length is + read here rather than left to Scapy, which reads it in host byte order + and adds the Hop-by-Hop Options header to it a second time. + + Every IP layer above ``layer`` bounds it, not only the nearest: a + datagram tunnelled in IP may declare more than the one carrying it + holds, or nothing at all. An IPv4 layer built by hand has no ``ihl`` + until it is built, so one given a Total Length alone has its IHL read + off the octets it builds into (:issue:`1676`). """ from scapy.layers.inet import IP @@ -337,8 +326,7 @@ def _tcp_segment(tcp: 'TCP') -> 'tuple[bytes, bytes] | None': after it, up to the end of the datagrams that carry the segment, c.f. ``_within_datagrams()``; or :data:`None` if the datagrams hold fewer than the 20 octets of the fixed header, or the Data Offset counts fewer - (:rfc:`9293#section-3.1`), since the default engine then reads no - segment at all. + (:rfc:`9293#section-3.1`), as there is then no header to split off. Note: The Data Offset is read off the octets, which for a dissected layer @@ -354,458 +342,25 @@ def _tcp_segment(tcp: 'TCP') -> 'tuple[bytes, bytes] | None': return segment[:hdr_len], segment[hdr_len:] -def _ipv4_accepted(ipv4: 'IP') -> 'bool': - """Test if the default engine reads a Scapy IPv4 layer as an IPv4 layer. - - Args: - ipv4: IPv4 layer of a dissected Scapy packet. - - Returns: - :data:`False` if the default engine's IPv4 parser rejects its header -- - its version is not 4, its IHL is below 5 or runs past the captured - octets, or its options are ones the parser rejects (:issue:`1607`) -- - as the default engine then dissects nothing past it; else :data:`True`. - - Note: - `Scapy`_ dissects what an IPv4 EtherType or tunnel carries as IPv4 - whatever its version or IHL, and never checks its options, so the - header is asked of the parser itself (see - :func:`pcapkit.toolkit.pypcapfile._default_accepts`). Its octets - are those `Scapy`_ dissected the layer from, which ``bytes()`` gives - back as they were, as far as the IHL goes, and no fewer than the 20 of - the fixed header, as the default engine reads at least those. A header - of version 4 with an IHL of 5 is not parsed: `Scapy`_ dissects none - shorter than 20 octets, and the parser checks nothing else of one. - - """ - if ipv4.version == 4 and ipv4.ihl == 5: - return True - octets = bytes(ipv4) - return _default_accepts('IPv4', octets[:max((octets[0] & 0x0F) * 4, 20)]) - - -def _quoted(layer: 'Packet') -> 'bool': - """Test if a Scapy layer heads the datagram an ICMP or ICMPv6 error quotes. - - Args: - layer: A layer of a dissected Scapy packet. - - Returns: - :data:`True` if ``layer`` is a :class:`~scapy.layers.inet.IPerror` or - :class:`~scapy.layers.inet6.IPerror6`; else :data:`False`. - - Note: - The default engine has no ICMP or ICMPv6 parser, so it dissects nothing - an ICMP message carries. `Scapy`_ dissects the datagram an error quotes - (:rfc:`792`, :rfc:`4443#section-3`) as ``IPerror`` or ``IPerror6``, - subclasses of :class:`~scapy.layers.inet.IP` and - :class:`~scapy.layers.inet6.IPv6`, and those bind the full layers of an - IP-in-IP tunnel: ``IPerror`` with protocol 41 binds an ``IPv6``, and - ``IPerror6`` with next header 4 or 41 an ``IP`` or an ``IPv6``, each of - which carries a real ``TCP`` (:issue:`1627`). Everything from such a - layer on is out of the default engine's reach. - - """ - from scapy.layers.inet import IPerror - from scapy.layers.inet6 import IPerror6 - - return isinstance(layer, (IPerror, IPerror6)) - - -def _linktype(packet: 'Packet') -> 'Optional[int]': - """The data link type of the capture a Scapy packet was read from. - - Args: - packet: Scapy packet. - - Returns: - The ``LINKTYPE_*`` number :class:`~pcapkit.foundation.engines.scapy.Scapy` - attached with :func:`attach_linktype`. A packet built by hand carries - none, so its first layer stands in for it: ``ETHERNET`` for an Ethernet - II or 802.3 frame, ``LOOP`` and ``NULL`` for the two kinds of BSD - loopback header `Scapy`_ has, and ``IPV4`` or ``IPV6`` for an IP - packet; and :data:`None` for any other, which no link type the default - engine dissects selects. - - """ - linktype = getattr(packet, LINKTYPE_ATTR, None) - if linktype is not None: - return int(linktype) - - from scapy.layers.inet import IP - from scapy.layers.inet6 import IPv6 - from scapy.layers.l2 import Dot3, Ether, Loopback, LoopbackOpenBSD - - if isinstance(packet, (Ether, Dot3)): - return Enum_LinkType.ETHERNET - if isinstance(packet, LoopbackOpenBSD): - return Enum_LinkType.LOOP - if isinstance(packet, Loopback): - return Enum_LinkType.NULL - if isinstance(packet, IP): - return Enum_LinkType.IPV4 - if isinstance(packet, IPv6): - return Enum_LinkType.IPV6 - return None - - -def _network(packet: 'Packet') -> 'Optional[tuple[Packet, int]]': - """Fetch the layer of a Scapy packet the default engine dissects behind its link layer. - - Args: - packet: Scapy packet. - - Returns: - That layer and its place in the frame, the first layer being ``1`` (c.f. - :data:`~pcapkit.protocols.protocol.FRAME_LAYER_LIMIT`), by the - packet's link type (see :func:`_linktype`): the payload of an Ethernet - frame, which may be anything `Scapy`_ dissects there; the packet itself - for a raw IP link type; and for a BSD loopback header the IPv4 or IPv6 - packet its address family names, as the default engine reads it (see - :data:`LOOPBACK_FAMILIES`). :data:`None` for any other link type, whose - frames the default engine leaves raw (:issue:`1650`), a loopback - family it dissects nothing behind, or a zero-length frame, of which it - dissects nothing at all (:issue:`1676`). - - Note: - The default engine dissects the first layer of the ``NULL``, - ``ETHERNET``, ``LOOP``, ``RAW``, ``IPV4`` and ``IPV6`` link types alone - (:attr:`Frame.__proto__ `), - and ``RAW`` as IPv4 or IPv6 by its version nibble, as `Scapy`_ does. - `Scapy`_ dissects IP straight behind the headers of many others, such as - ``LINUX_SLL``, ``LINUX_SLL2``, ``PPP`` and the ``DLT_RAW`` of 12. - - `Scapy`_ hands a zero-length frame back as a layer of defaults, which - at ``RAW`` and ``IPV4`` is an IPv4 header with no IHL (see - :func:`_frame`), so it is told by the frame's octets, not its layers. - - A loopback header's family is read as - :func:`~pcapkit.protocols.link.loopback.family_byteorder` reads it -- - ``LOOP``'s in network byte order, ``NULL``'s in the byte order its - octets show -- and the packet behind it is dissected from the octets - wherever `Scapy`_ dissects otherwise. `Scapy`_ binds IPv6 to the - ``AF_INET6`` of the host it runs on, and IPv4 to family ``0`` too, so - what it dissects there depends on the host. - - """ - from scapy.layers.inet import IP - from scapy.layers.inet6 import IPv6 - - linktype = _linktype(packet) - if linktype not in (Enum_LinkType.NULL, Enum_LinkType.ETHERNET, Enum_LinkType.LOOP, - Enum_LinkType.RAW, Enum_LinkType.IPV4, Enum_LinkType.IPV6): - return None - octets = _frame(packet) - if not octets: - return None - if linktype == Enum_LinkType.ETHERNET: - return packet.payload, 2 - if linktype in (Enum_LinkType.RAW, Enum_LinkType.IPV4, Enum_LinkType.IPV6): - return packet, 1 - - from pcapkit.protocols.link.loopback import family_byteorder # isort:skip - - if len(octets) < 4: - return None - version = LOOPBACK_FAMILIES.get(int.from_bytes(octets[:4], family_byteorder(octets[:4], linktype))) - if version is None: - return None - layer = packet.payload # type: Packet - if not isinstance(layer, IP if version == 4 else IPv6): - data = octets[4:] - if len(data) < (IPV4_HEADER_LEN if version == 4 else IPV6_HEADER_LEN): - return None - layer = IP(data) if version == 4 else IPv6(data) - return layer, 2 - - -class _Chain(NamedTuple): - """The extension header chain of a Scapy IPv6 layer, as the default engine walks it.""" - - #: The octets of the IPv6 packet's payload, as far as its Payload Length and - #: every IP layer above it goes (see :func:`_within_datagrams`). - payload: 'bytes' - #: How many extension headers the default engine walks. - headers: 'int' - #: Where the first Fragment header it walks starts in :attr:`payload`, or - #: :data:`None` if it walks none. - fragment: 'Optional[int]' - #: The upper layer it dissects behind the chain: the `Scapy`_ layer, or one - #: dissected here from :attr:`payload`; or :data:`None` if it dissects no - #: TCP, IPv4 or IPv6 there. - upper: 'Optional[Packet]' - - -def _ipv6_chain(ipv6: 'IPv6', depth: 'int') -> '_Chain': - """Walk the extension header chain of a Scapy IPv6 layer as the default engine walks it. - - Args: - ipv6: IPv6 layer of a dissected Scapy packet. - depth: Its place in the frame (see :func:`_network`). - - Returns: - The chain, walked over the octets by the default engine's rules: the - :data:`~pcapkit.toolkit.pypcapfile.IPV6_EXTENSION_HEADERS`, each sized - by its own length field, up to - :data:`~pcapkit.protocols.internet.internet.EXTENSION_HEADER_LIMIT` of - them (:issue:`1604`) and as far as - :data:`~pcapkit.protocols.protocol.FRAME_LAYER_LIMIT`, each header being - one layer of the frame (:issue:`1610`), and no further than a later - fragment's Fragment header (:rfc:`8200#section-4.5`) or a header cut - short. The upper layer is a TCP, IPv4 or IPv6 one inside the frame - bound, dissected from the octets where `Scapy`_ dissected something - else there. - - Note: - `Scapy`_ dissects the Hop-by-Hop Options, Routing, Fragment and - Destination Options headers, and AH, but no TCP and no IPv4 behind an - AH, and IPv4 where the Next Header is ``0``; the Mobility header but - nothing behind it; and neither HIP nor Shim6 (:issue:`1626`). Nor does - it stop at either bound (:issue:`1609`). Its layers are kept as far as - they are the ones the walk steps over, so the upper layer is its own - wherever it dissected the chain as the default engine does. - - """ - from scapy.layers.inet import IP, TCP - from scapy.layers.inet6 import IPv6, _IPv6ExtHdr - from scapy.layers.ipsec import AH - - head = bytes(ipv6)[:IPV6_HEADER_LEN] - payload = _within_datagrams(ipv6.payload) - nxt = head[6] if len(head) == IPV6_HEADER_LEN else None - offset, headers, fragment = 0, 0, None # type: int, int, Optional[int] - bound = min(EXTENSION_HEADER_LIMIT, FRAME_LAYER_LIMIT - depth) - # the headers Scapy dissects, as AH or one of its ``_IPv6ExtHdr`` layers - dissects = (Enum_ExtensionHeader.HOPOPT, Enum_ExtensionHeader.IPv6_Route, Enum_ExtensionHeader.IPv6_Frag, - Enum_ExtensionHeader.IPv6_Opts, Enum_ExtensionHeader.AH) - layer = ipv6.payload # type: Optional[Packet] # Scapy's, while it walks the chain alike - while nxt in IPV6_EXTENSION_HEADERS: - if headers == bound or len(payload) < offset + 8: - return _Chain(payload, headers, fragment, None) - headers += 1 - if nxt in dissects and isinstance(layer, AH if nxt == Enum_ExtensionHeader.AH else _IPv6ExtHdr): - layer = layer.payload - else: - layer = None - if nxt == Enum_ExtensionHeader.IPv6_Frag: # fixed length, second octet reserved - if fragment is None: - fragment = offset - if struct.unpack_from('!H', payload, offset + 2)[0] >> 3: # a later fragment - return _Chain(payload, headers, fragment, None) - length = 8 - elif nxt == Enum_ExtensionHeader.AH: # length in 4-octet units, less two - length = (payload[offset + 1] + 2) * 4 - else: # length in 8-octet units, less one - length = (payload[offset + 1] + 1) * 8 - nxt, offset = payload[offset], offset + length - - upper = None # type: Optional[Packet] - data = payload[offset:] - if depth + headers + 1 > FRAME_LAYER_LIMIT or len(payload) < offset: - pass - elif nxt == Enum_TransType.TCP: - if isinstance(layer, TCP): - upper = layer - elif len(data) >= TCP_MIN_HEADER_LEN: - upper = TCP(data) - elif nxt in IP_TUNNELS: - version = IP_TUNNELS[nxt] - if isinstance(layer, IP if version == 4 else IPv6) and not _quoted(layer): - upper = layer - elif len(data) >= (IPV4_HEADER_LEN if version == 4 else IPV6_HEADER_LEN): - upper = IP(data) if version == 4 else IPv6(data) - return _Chain(payload, headers, fragment, upper) - - -class _Layer(NamedTuple): - """A layer of a Scapy packet that the default engine dissects too.""" - - #: The `Scapy`_ layer, or one dissected here from the octets. - layer: 'Packet' - #: Its place in the frame, the first layer being ``1`` (c.f. - #: :data:`~pcapkit.protocols.protocol.FRAME_LAYER_LIMIT`). - depth: 'int' - #: For a TCP layer the default engine reads behind an AH or IPv6 extension - #: header in IPv4, its header and payload as that engine splits them (see - #: :func:`_tcp_behind_extension_headers`); else :data:`None`. - segment: 'Optional[tuple[bytes, bytes]]' = None - - -def _dissected(packet: 'Packet') -> 'Iterator[_Layer]': - """Walk the layers of a Scapy packet that the default engine dissects too. - - Args: - packet: Scapy packet. - - Yields: - The layer the default engine dissects behind the link layer (see - :func:`_network`), and each one on from it that it dissects as well, - up to the first that carries nothing it dissects: anything but an - 802.1Q or 802.1ad tag, IPv4 or IPv6; an IPv4 layer that is a later - fragment, whose data is a slice of the datagram's payload with no - upper-layer header in it (:rfc:`791#section-3.2`, :issue:`1617`). It - stops short of an IPv4 layer whose header the default engine rejects - (see :func:`_ipv4_accepted`), of the datagram an ICMP or ICMPv6 error - quotes (see :func:`_quoted`), and of - :data:`~pcapkit.protocols.protocol.FRAME_LAYER_LIMIT`, past which the - default engine dissects nothing (:issue:`1609`). - - An IPv6 layer is followed by the upper layer behind its extension - headers (see :func:`_ipv6_chain`), which are not yielded. An IPv4 layer - whose protocol names an AH or IPv6 extension header carries what the - default engine reads past those headers, by its own rules and up to - :data:`~pcapkit.protocols.internet.internet.EXTENSION_HEADER_LIMIT` of - them in a row (see - :func:`~pcapkit.toolkit.pypcapfile._walk_extension_headers`): a tunnel, - or a TCP segment, each dissected from the octets (:issue:`1606`). - - Note: - The default engine dissects IP only where its dispatch tables send it: - a link type, an EtherType of the link layer or of a VLAN tag - (:attr:`Link.__proto__ `), - and a protocol number of IP or of an extension header - (:attr:`Internet.__proto__ - `). It leaves - raw the other layers `Scapy`_ dissects IP inside -- GRE, PPPoE, an - 802.3 frame's LLC/SNAP header, VXLAN, the PPP frame L2TPv2 carries, - and ICMP -- so it reads no IP, and no TCP, past one (:issue:`1636`). - - Inside IPv4, `Scapy`_ dissects an AH alone of those headers, and behind - one only IPv6 and, for a Next Header of ``0``, IPv4, so the default - engine's rules are applied to the octets instead. It also dissects IPv6 - and AH inside a later fragment, binding neither to a zero fragment - offset as it binds TCP. - - The payloads are walked, not ``getlayer()``, which searches the - packets a field holds as well, such as the IPv6 packet a Redirected - Header option carries (:rfc:`4861#section-4.6.3`). - - """ - from scapy.layers.inet import IP, TCP - from scapy.layers.inet6 import IPv6 - from scapy.layers.l2 import Dot1AD, Dot1Q - from scapy.packet import NoPayload - - if (network := _network(packet)) is None: - return - layer, depth = network - while depth <= FRAME_LAYER_LIMIT and not isinstance(layer, NoPayload) and not _quoted(layer): - if isinstance(layer, IP): - if not _ipv4_accepted(layer): - return - yield _Layer(layer, depth) - if layer.frag: - return - if layer.proto not in IPV4_EXTENSION_HEADERS: - layer, depth = layer.payload, depth + 1 - continue - if (walked := _walk_extension_headers(layer.proto, _within_datagrams(layer.payload))) is None: - return - proto, data, hops = walked - depth += hops + 1 - if depth > FRAME_LAYER_LIMIT: - return - if proto == Enum_TransType.TCP: - if (segment := _tcp_behind_extension_headers(data)) is not None: - yield _Layer(TCP(data), depth, segment) - return - if proto not in IP_TUNNELS: - return - version = IP_TUNNELS[proto] - if len(data) < (IPV4_HEADER_LEN if version == 4 else IPV6_HEADER_LEN): - return - layer = IP(data) if version == 4 else IPv6(data) - elif isinstance(layer, IPv6): - yield _Layer(layer, depth) - chain = _ipv6_chain(layer, depth) - if chain.upper is None: - return - layer, depth = chain.upper, depth + chain.headers + 1 - else: - yield _Layer(layer, depth) - if not isinstance(layer, (Dot1Q, Dot1AD)): - return - layer, depth = layer.payload, depth + 1 - - -def _first_ipv4(packet: 'Packet') -> 'IP | None': - """Fetch the first IPv4 layer of a Scapy packet, as the default engine dissects it. - - Args: - packet: Scapy packet. - - Returns: - The first :class:`~scapy.layers.inet.IP` layer of the packet the default - engine dissects too, however deep -- e.g. one carried in IPv6 - (:rfc:`2473`) -- or :data:`None` if there is none (see - :func:`_dissected`). - - Note: - The default engine takes ``frame['IPv4']`` - (:func:`pcapkit.toolkit.pcap.ipv4_reassembly`). ``packet['IP']`` finds - the same layer but for one inside an ICMPv6 error's quote, which it - reaches through the ``IPerror6`` there, or inside a layer the default - engine leaves raw, such as GRE (:issue:`1636`), or one past the frame's - layer bound (:issue:`1609`). - - """ - from scapy.layers.inet import IP - - for found in _dissected(packet): - if isinstance(found.layer, IP): - return found.layer - return None - - -def _first_ipv6(packet: 'Packet') -> 'Optional[_Layer]': - """Fetch the first IPv6 layer of a Scapy packet, as the default engine dissects it. - - Args: - packet: Scapy packet. - - Returns: - The first :class:`~scapy.layers.inet6.IPv6` layer of the packet the - default engine dissects too, however deep -- e.g. one carried in IPv4 - (:rfc:`4213`) -- with its place in the frame, or :data:`None` if there - is none (see :func:`_dissected`). - - Note: - The default engine takes ``frame['IPv6']`` - (:func:`pcapkit.toolkit.pcap.ipv6_reassembly`). ``packet['IPv6']`` - finds the same layer but for one inside an ICMP or ICMPv6 error's - quote, a later IPv4 fragment, or a layer the default engine leaves - raw, such as GRE (:issue:`1635`), or one past the frame's layer bound - (:issue:`1609`). - - """ - from scapy.layers.inet6 import IPv6 - - for found in _dissected(packet): - if isinstance(found.layer, IPv6): - return found - return None - - def _tcp_behind_extension_headers(segment: 'bytes') -> 'tuple[bytes, bytes] | None': - """Split a TCP segment the default engine reads behind an AH or IPv6 extension header in IPv4. + """Split a TCP segment carried behind an AH or IPv6 extension header in IPv4. Args: segment: Raw octets from the TCP header on, as far as the IPv4 layer - carrying it goes (see :func:`_dissected`). + carrying it goes (see :func:`_tcp_carrier`). Returns: The header, as many octets as the Data Offset counts, and the payload - after it; or :data:`None` if the default engine reads no TCP there: - fewer than :data:`TCP_MIN_HEADER_LEN - ` octets, a Data Offset - counting fewer or more than were captured, or options its TCP parser - rejects (see :func:`~pcapkit.toolkit.pypcapfile._default_accepts`). + after it; or :data:`None` if there are fewer than + :data:`TCP_MIN_HEADER_LEN ` + octets, the Data Offset counts fewer or more than were captured, or + pcapkit's own TCP parser rejects the options (see + :func:`~pcapkit.toolkit.pypcapfile._default_accepts`). Note: - Directly over IP, the default engine reads a TCP header its parser - rejects out of the raw octets left behind - (:func:`pcapkit.toolkit.pcap.tcp_segment`), but behind another header - it does not, so the parser's answer stands here. + `Scapy`_ dissects no TCP layer there (see :func:`_tcp_carrier`), so + the segment is read off the octets, and its options are left to + pcapkit's parser to check. """ hdr_len = (segment[12] >> 4) * 4 if len(segment) >= TCP_MIN_HEADER_LEN else 0 @@ -827,40 +382,70 @@ def _tcp_carrier(packet: 'Packet') -> 'tuple[IP | IPv6, TCP, bytes, bytes] | Non :class:`~scapy.layers.inet6.IPv6` layer above the packet's TCP layer, that layer, and the segment's header and payload (see :func:`_tcp_segment`); or :data:`None` if there is no such IP layer, or - the default engine reads no TCP there: it dissects no TCP layer in the - frame (see :func:`_dissected`) -- one behind a link type it leaves raw - (:issue:`1650`), behind a layer it leaves raw, such as GRE - (:issue:`1636`), an IPv4 layer whose header it rejects, or a later IPv4 - fragment (:issue:`1617`), in the datagram an ICMP or ICMPv6 error quotes - (see :func:`_quoted`, :issue:`1627`), or past the frame's layer bound or - an extension header chain's (:issue:`1609`). A TCP segment the default - engine reads behind an AH or IPv6 extension header in IPv4 is split by - :func:`_tcp_behind_extension_headers` instead (:issue:`1606`). + an IPv4 layer above it is a later fragment, whose data is a slice of + the datagram's payload with no upper-layer header in it + (:rfc:`791#section-3.2`, :issue:`1617`). + + An IPv4 layer whose protocol names an AH or IPv6 extension header + (:data:`~pcapkit.toolkit.pypcapfile.IPV4_EXTENSION_HEADERS`) has those + headers stepped over in its octets, up to + :data:`~pcapkit.protocols.internet.internet.EXTENSION_HEADER_LIMIT` of + them in a row (see + :func:`~pcapkit.toolkit.pypcapfile._walk_extension_headers`), and what + follows them is dissected from the octets: a tunnel, walked on as + above, or a TCP segment (see :func:`_tcp_behind_extension_headers`, + :issue:`1606`). In the datagram an ICMP error quotes, they are left as + `Scapy`_ dissects them. Note: The nearest is the innermost IP layer of an IP-in-IP tunnel (:rfc:`2003`, :rfc:`2473`, :rfc:`4213`), whose addresses are the TCP - endpoints', as for the default engine - (:func:`pcapkit.toolkit.pcap.tcp_segment`). ``packet['IP']`` finds the - outermost IPv4 layer instead, which is the tunnel's, or with IPv4 - carried in IPv6 the inner one (:issue:`1581`). The extension headers - between an IPv6 layer and its payload are not IP layers, so are - stepped over. + endpoints' (:issue:`1581`). ``packet['IP']`` finds the outermost IPv4 + layer instead, which is the tunnel's, or with IPv4 carried in IPv6 the + inner one. The extension header layers `Scapy`_ dissects between an + IPv6 layer and its payload are not IP layers, so are stepped over. + + Inside IPv4, `Scapy`_ dissects an AH alone of those headers, and behind + one only IPv6 and, for a Next Header of ``0``, IPv4, so the octets are + read instead. It also dissects IPv6 and AH inside a later fragment, + binding neither to a zero fragment offset as it binds TCP. """ - from scapy.layers.inet import IP, TCP + from scapy.layers.inet import IP, TCP, IPerror from scapy.layers.inet6 import IPv6 + from scapy.packet import NoPayload + tcp = cast('TCP | None', packet.getlayer('TCP')) carrier = None # type: IP | IPv6 | None - for found in _dissected(packet): - if isinstance(found.layer, (IP, IPv6)): - carrier = found.layer - elif isinstance(found.layer, TCP): - if carrier is None: + layer = packet # type: Packet + while layer is not tcp: + if isinstance(layer, NoPayload): + return None + if isinstance(layer, IP): + if layer.frag: return None - segment = _tcp_segment(found.layer) if found.segment is None else found.segment - return None if segment is None else (carrier, found.layer, *segment) - return None + carrier = layer + # the datagram an ICMP error quotes is left as Scapy dissects it + if layer.proto in IPV4_EXTENSION_HEADERS and not isinstance(layer, IPerror): + upper = _walk_extension_headers(layer.proto, _within_datagrams(layer.payload)) + if upper is not None and upper[0] == Enum_TransType.TCP: + segment = _tcp_behind_extension_headers(upper[1]) + return None if segment is None else (carrier, TCP(upper[1]), *segment) + if upper is None or upper[0] not in IP_TUNNELS: + return None + version, data = IP_TUNNELS[upper[0]], upper[1] + if len(data) < (IPV4_HEADER_LEN if version == 4 else IPV6_HEADER_LEN): + return None + layer = IP(data) if version == 4 else IPv6(data) + tcp = cast('TCP | None', layer.getlayer('TCP')) + continue + elif isinstance(layer, IPv6): + carrier = layer + layer = layer.payload + + if carrier is None or (segment := _tcp_segment(tcp)) is None: + return None + return carrier, tcp, *segment def ipv4_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv4Address] | None': @@ -874,13 +459,12 @@ def ipv4_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv4Ad Data for IPv4 reassembly. * If the ``packet`` can be used for IPv4 reassembly. A packet can be reassembled - if it contains IPv4 layer (:class:`scapy.layers.inet.IP`) -- the first of the - frame the default engine dissects, so one carried in IPv6 counts but none an - ICMP or ICMPv6 error quotes, a layer that engine leaves raw carries, such as - GRE, a link type it leaves raw carries, or one past the frame's layer bound (see - :func:`_first_ipv4`) -- whose header the default engine accepts (see - :func:`_ipv4_accepted`) and lies inside every IPv6 packet carrying it, and the - **DF** (:attr:`scapy.layers.inet.IP.flags.DF`) flag is :data:`False`. + if it contains IPv4 layer (:class:`scapy.layers.inet.IP`) whose header -- as + many octets as its IHL counts, and at least 20 -- lies inside the frame and + every IP datagram carrying it (see :func:`_within_datagrams`), and the **DF** + (:attr:`scapy.layers.inet.IP.flags.DF`) flag is :data:`False`. A zero-length + frame has none, though `Scapy`_ hands it back as an IPv4 layer of defaults at + the ``RAW`` and ``IPV4`` link types (see :func:`_frame`). * If the ``packet`` can be reassembled, then the :obj:`dict` mapping of data for IPv4 reassembly (:term:`reasm.ipv4.packet`) will be returned; otherwise, returns :data:`None`. @@ -888,23 +472,21 @@ def ipv4_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv4Ad :class:`pcapkit.foundation.reassembly.ipv4.IPv4` """ - if scapy is None: - raise ModuleNotFound("No module named 'scapy'", name='scapy') - - if (ipv4 := _first_ipv4(packet)) is not None: + if 'IP' in packet: + ipv4 = cast('IP', packet['IP']) if ipv4.flags.DF: # dismiss not fragmented packet return None + if not _frame(packet): # a zero-length frame (:issue:`1676`) + return None # NOTE: ``bytes(ipv4)`` ends with the frame's trailer, which Scapy files - # as padding at the end of the packet however deep the IPv4 layer, but - # the default engine hands it only the octets inside the Payload Length - # of each IPv6 packet above it, so those are all it keeps (:issue:`1623`) + # as padding at the end of the packet however deep the IPv4 layer, so + # only the octets inside every IP datagram above it are kept, the + # Payload Length of IPv6 carrying it included (:issue:`1623`). The IHL + # and Total Length are read off them, as Scapy leaves ``ihl`` and + # ``len`` None on a layer built by hand until it is built (:issue:`1676`). octets = _within_datagrams(ipv4) - # NOTE: the IHL and Total Length are read off the octets, as Scapy - # leaves ``ihl`` and ``len`` None on a layer built by hand until it is - # built (:issue:`1676`) ihl = (octets[0] & 0x0F) * 4 if octets else IPV4_HEADER_LEN - # the default engine dissects no IPv4 header those octets cut short - if len(octets) < max(ihl, IPV4_HEADER_LEN) or not _ipv4_accepted(ipv4): + if len(octets) < max(ihl, IPV4_HEADER_LEN): # a header those octets cut short return None data = IP_Packet( @@ -927,10 +509,10 @@ def ipv4_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv4Ad mf=bool(ipv4.flags.MF), # more fragment flag tl=int.from_bytes(octets[2:4], 'big'), # total length, header includes header=octets[:ihl], # raw bytes type header - # NOTE: this keeps the octets past the Total Length, as the default - # engine's adapter does, since the reassembly machinery cuts the - # payload at ``tl - ihl`` itself. :func:`ipv6_reassembly` derives - # its ``tl`` from the payload, so it has to cut them (:issue:`1584`). + # NOTE: this keeps the octets past the Total Length, since the + # reassembly machinery cuts the payload at ``tl - ihl`` itself. + # :func:`ipv6_reassembly` derives its ``tl`` from the payload, so it + # has to cut them (:issue:`1584`). payload=bytearray(octets[ihl:]), # raw bytearray type payload timestamp=float(packet.time), # capture timestamp ) @@ -949,13 +531,10 @@ def ipv6_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv6Ad Data for IPv6 reassembly. * If the ``packet`` can be used for IPv6 reassembly. A packet can be reassembled - if it contains IPv6 layer (:class:`scapy.layers.inet6.IPv6`) -- the first of the - frame the default engine dissects, so one carried in IPv4 counts but none an ICMP - or ICMPv6 error quotes, a later IPv4 fragment carries, a layer that engine leaves - raw carries, such as GRE, a link type it leaves raw carries, or one past the - frame's layer bound (see :func:`_first_ipv6`) -- and IPv6 Fragment header - (:rfc:`2460#section-4.5`) in that layer's own extension header chain, as far as - the default engine walks it (see :func:`_ipv6_chain`). + if it contains IPv6 layer (:class:`scapy.layers.inet6.IPv6`), in no later IPv4 + fragment, and IPv6 Fragment header (:rfc:`2460#section-4.5`, i.e., + :class:`scapy.layers.inet6.IPv6ExtHdrFragment`) among the extension headers + `Scapy`_ dissects in that layer's own chain. * 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`. @@ -965,27 +544,41 @@ def ipv6_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv6Ad See Also: :class:`pcapkit.foundation.reassembly.ipv6.IPv6` + Note: + A later IPv4 fragment's data is a slice of the datagram's payload, with + no upper-layer header in it (:rfc:`791#section-3.2`), but `Scapy`_ + dissects IPv6 there whatever the fragment offset. And the header and + addresses handed on are those of the IPv6 layer the Fragment header + follows, so one inside a packet that layer carries, such as the inner + packet of 6in6, is not taken. + """ if scapy is None: raise ModuleNotFound("No module named 'scapy'", name='scapy') + from scapy.layers.inet import IP + from scapy.layers.inet6 import IPv6ExtHdrFragment, _IPv6ExtHdr - if (found := _first_ipv6(packet)) is not None: - ipv6 = cast('IPv6', found.layer) - chain = _ipv6_chain(ipv6, found.depth) - if chain.fragment is None: + if 'IPv6' in packet: + ipv6 = cast('IPv6', packet['IPv6']) + above = ipv6.underlayer + while above is not None: + if isinstance(above, IP) and above.frag: # data of a later fragment + return None + above = above.underlayer + ipv6_frag = ipv6.payload + while isinstance(ipv6_frag, _IPv6ExtHdr) and not isinstance(ipv6_frag, IPv6ExtHdrFragment): + ipv6_frag = ipv6_frag.payload + if not isinstance(ipv6_frag, IPv6ExtHdrFragment): return None # dismiss not fragmented packet - # NOTE: The unfragmentable part is every octet before the Fragment - # header, which is the only part of the header the reassembled packet - # keeps (:rfc:`8200#section-4.5`). The walk read the Fragment header off - # the octets, as `Scapy`_ dissects none behind a Mobility, HIP or Shim6 - # header (:issue:`1626`). - hdr_len = IPV6_HEADER_LEN + chain.fragment - ipv6_frag = chain.payload[chain.fragment:chain.fragment + 8] - field = int.from_bytes(ipv6_frag[2:4], 'big') + # NOTE: ``len()`` of a Scapy layer spans that layer and everything after + # it, so the difference is the unfragmentable part -- every octet before + # the Fragment header, which is the only part of the header the + # reassembled packet keeps (:rfc:`8200#section-4.5`). + hdr_len = len(ipv6) - len(ipv6_frag) # the fragment's data stops at the Payload Length, so neither it nor # ``tl`` below counts the frame's trailer (:issue:`1584`) - payload = bytearray(chain.payload[chain.fragment + 8:]) + payload = bytearray(_within_datagrams(ipv6_frag.payload)) data = IP_Packet( bufid=( @@ -993,21 +586,22 @@ def ipv6_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv6Ad ipaddress.ip_address(ipv6.src)), # source IP address cast('IPv6Address', ipaddress.ip_address(ipv6.dst)), # destination IP address - int.from_bytes(ipv6_frag[4:8], 'big'), # identification - Enum_TransType.get(ipv6_frag[0]), # next header field in IPv6 Fragment Header + ipv6_frag.id, # identification + Enum_TransType.get(ipv6_frag.nh), # next header field in IPv6 Fragment Header ), num=count, # original packet range number - # NOTE: The Fragment Offset counts on-wire 8-octet units - # (:rfc:`8200#section-4.5`), but the reassembly machinery indexes + # NOTE: Scapy reports ``IPv6ExtHdrFragment.offset`` in on-wire 8-octet + # units (:rfc:`8200#section-4.5`), but the reassembly machinery indexes # the datagram buffer with ``fo``, so it must be scaled into octets. - fo=(field >> 3) * 8, # fragment offset + fo=ipv6_frag.offset * 8, # fragment offset ihl=hdr_len, # header length, only headers before IPv6-Frag - mf=bool(field & 1), # more fragment flag - # NOTE: the Fragment header is not counted, as the reassembly - # machinery writes the payload over the span ``tl - ihl``, so counting - # it would leave 8 octets of stray zeroes in every reassembled datagram. + mf=bool(ipv6_frag.m), # more fragment flag + # NOTE: ``len(ipv6)`` counts the Fragment header, so it overstates + # this by 8 -- and the reassembly machinery writes the payload over + # the span ``tl - ihl``, so using it would leave 8 octets of stray + # zeroes in every reassembled datagram. tl=hdr_len + len(payload), # total length, header includes - header=bytes(ipv6)[:IPV6_HEADER_LEN] + chain.payload[:chain.fragment], # header before IPv6-Frag + header=bytes(ipv6)[:hdr_len], # raw bytes type header before IPv6-Frag payload=payload, # raw bytearray type payload after IPv6-Frag timestamp=float(packet.time), # capture timestamp ) @@ -1028,13 +622,9 @@ def tcp_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'TCP_Packet | None * If the ``packet`` can be used for TCP reassembly. A packet can be reassembled if it contains TCP layer (:class:`scapy.layers.inet.TCP`) whose fixed header lies inside its datagram, and whose Data Offset counts at least that header, - behind no IPv4 header the default engine rejects, in no later IPv4 fragment, in - no datagram an ICMP or ICMPv6 error quotes, in no layer or link type the default - engine leaves raw, such as GRE or ``LINUX_SLL``, and inside the frame's layer - bound and an extension header chain's; or a TCP segment the default engine reads - behind an AH or IPv6 extension header in IPv4, or behind an IPv6 extension header - `Scapy`_ dissects nothing behind. It is keyed by the IP layer that carries it -- - in an IP-in-IP tunnel, the innermost (see :func:`_tcp_carrier`). + in no later IPv4 fragment; or a TCP segment behind an AH or IPv6 extension + header in IPv4, read off the octets. It is keyed by the IP layer that carries + it -- in an IP-in-IP tunnel, the innermost (see :func:`_tcp_carrier`). * If the ``packet`` can be reassembled, then the :obj:`dict` mapping of data for TCP reassembly (:term:`reasm.tcp.packet`) will be returned; otherwise, returns :data:`None`. diff --git a/tests/toolkit/test_dpkt_scapy_ipv4_header_checks_1607_unit.py b/tests/toolkit/test_dpkt_scapy_ipv4_header_checks_1607_unit.py index 6fe29272a..d14a66a75 100644 --- a/tests/toolkit/test_dpkt_scapy_ipv4_header_checks_1607_unit.py +++ b/tests/toolkit/test_dpkt_scapy_ipv4_header_checks_1607_unit.py @@ -8,9 +8,7 @@ so a bad one is still read. The helpers here build those headers and read a capture of them with any -engine; :mod:`tests.toolkit.test_dpkt_tcp_ns_flags_1618_unit` and -:mod:`tests.toolkit.test_scapy_ipv6_fragment_and_encap_walk_1635_1636_unit` -import them. +engine; :mod:`tests.toolkit.test_dpkt_tcp_ns_flags_1618_unit` imports them. Every capture is built here, so the module is in the unit tier. diff --git a/tests/toolkit/test_scapy_ipv6_fragment_and_encap_walk_1635_1636_unit.py b/tests/toolkit/test_scapy_ipv6_fragment_and_encap_walk_1635_1636_unit.py deleted file mode 100644 index 884d1f581..000000000 --- a/tests/toolkit/test_scapy_ipv6_fragment_and_encap_walk_1635_1636_unit.py +++ /dev/null @@ -1,408 +0,0 @@ -# -*- coding: utf-8 -*- -"""scapy reads IP and TCP only in the layers the default engine dissects. C.f. #1635, #1636. - -The default engine hands IPv6 reassembly the first IPv6 layer it dissects, with -the Fragment header of that layer's own extension header chain. `Scapy`_'s -``packet['IPv6']`` found the first IPv6 layer it dissected, which could lie in an -ICMP or ICMPv6 error's quote, a later IPv4 fragment, GRE or VXLAN, and -``IPv6ExtHdrFragment in ipv6`` then found a Fragment header anywhere below it, as -in the IPv6 packet a 6in6 tunnel carries (#1635). - -The default engine dissects IP only behind the link layer, a VLAN tag, IP or an -extension header, and leaves GRE, PPPoE, an 802.3 frame's LLC/SNAP header, VXLAN -and the PPP payload of L2TP raw. `Scapy`_ dissects IP and TCP inside all of them, -and its engine handed those to TCP reassembly and flow tracing, and to IPv4 and -IPv6 reassembly (#1636). - -The first layer is the one the link type selects, and a layer behind it carries -IP only if it is a VLAN tag, IP or an extension header. So at link types the -default engine leaves raw, where `Scapy`_ dissects two layers or more before IP --- HDLC-like framing then PPP, 802.11 then LLC -- scapy reads no IP either. Nor -does it at those where `Scapy`_ dissects IP straight behind the first layer, -such as LINUX_SLL, which -:mod:`tests.toolkit.test_scapy_linktype_and_layer_bounds_1609_1626_1650_unit` -covers with the frame and extension header bounds (#1650, #1609). - -Only scapy is compared with the default engine. - -Every capture is built here, so the module is in the unit tier. - -.. _Scapy: https://scapy.net - -""" -from __future__ import annotations - -import importlib.util -import os -import tempfile -import unittest -import warnings -from typing import TYPE_CHECKING, NamedTuple -from unittest import mock - -from tests._support import close_extractor, reimport_once_per_class -from tests.foundation import _roundtrip as wire -from tests.toolkit import test_dpkt_ipv4_reassembly_input_1615_1616_unit as ipv4_input -from tests.toolkit.test_dpkt_scapy_ipv4_header_checks_1607_unit import BAD_OPTIONS -from tests.toolkit.test_dpkt_scapy_ipv4_header_checks_1607_unit import ipv4 as ipv4_header -from tests.toolkit.test_scapy_ipv6_payload_and_icmp_quote_1623_1627_unit import ICMP6_PTB, ICMP_UNREACH -from tests.toolkit.test_tcp_carrier_walk_1606_1617_unit import ah, eth, options, tcp, v4, v6 - -if TYPE_CHECKING: - from typing import Any, Callable - -RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') -HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) -HAS_SCAPY = importlib.util.find_spec('scapy') is not None - -#: Ethernet II type of IPv4, IPv6, an 802.1Q and an 802.1ad tag, transparent -#: Ethernet bridging (in GRE), a PPPoE session and an 802.1ah I-TAG, and the PPP -#: protocol numbers of IPv4 and IPv6. -IPV4, IPV6, C_TAG, S_TAG, TEB, PPPOE, I_TAG = 0x0800, 0x86DD, 0x8100, 0x88A8, 0x6558, 0x8864, 0x88E7 -PPP_IPV4, PPP_IPV6 = 0x0021, 0x0057 - - -def ethertype(packet: bytes) -> int: - """The Ethernet II type of the IPv4 or IPv6 ``packet``.""" - return IPV6 if packet[0] >> 4 == 6 else IPV4 - - -def tagged(packet: bytes, *tpids: int) -> bytes: - """An Ethernet frame carrying the IP ``packet`` behind one VLAN tag (VID 100) per TPID in ``tpids``.""" - types = [*tpids, ethertype(packet)] - return wire.ethernet(b''.join(b'\x00\x64' + kind.to_bytes(2, 'big') for kind in types[1:]) + packet, types[0]) - - -def gre(packet: bytes, ptype: 'int | None' = None) -> bytes: - """A GRE header (RFC 2784) with no checksum, then ``packet`` of protocol type ``ptype``, by default its own.""" - return b'\x00\x00' + (ethertype(packet) if ptype is None else ptype).to_bytes(2, 'big') + packet - - -def vxlan(frame: bytes) -> bytes: - """UDP to port 4789 carrying a VXLAN header (RFC 7348), VNI 100, then the Ethernet ``frame``.""" - return wire.udp(b'\x08\x00\x00\x00' + (100).to_bytes(3, 'big') + b'\x00' + frame, dport=4789) - - -def session(packet: bytes) -> bytes: - """A PPPoE session header (RFC 2516), then a PPP frame carrying the IP ``packet``.""" - return ppp(packet, b'\x11\x00\x00\x01' + (2 + len(packet)).to_bytes(2, 'big')) - - -def pppoe(packet: bytes) -> bytes: - """An Ethernet frame of a PPPoE session (RFC 2516) whose PPP frame carries the IP ``packet``.""" - return wire.ethernet(session(packet), PPPOE) - - -def l2tp(packet: bytes) -> bytes: - """UDP between ports 1701 carrying an L2TPv2 data message (RFC 2661) whose PPP frame carries IPv4 ``packet``.""" - return wire.udp(b'\x00\x02\x00\x01\x00\x01' + b'\xff\x03' + PPP_IPV4.to_bytes(2, 'big') + packet, - sport=1701, dport=1701) - - -def snap(packet: bytes) -> bytes: - """An LLC/SNAP header (RFC 1042), then the IP ``packet``.""" - return b'\xaa\xaa\x03\x00\x00\x00' + ethertype(packet).to_bytes(2, 'big') + packet - - -def llc_snap(packet: bytes) -> bytes: - """An 802.3 frame whose LLC/SNAP header (RFC 1042) carries the IP ``packet``.""" - body = snap(packet) - return wire.MAC_B + wire.MAC_A + len(body).to_bytes(2, 'big') + body - - -def ppp(packet: bytes, prefix: bytes = b'\xff\x03') -> bytes: - """``prefix``, by default the HDLC-like framing of RFC 1662, then a PPP frame carrying the IP ``packet``.""" - return prefix + (PPP_IPV6 if packet[0] >> 4 == 6 else PPP_IPV4).to_bytes(2, 'big') + packet - - -def dot11(packet: bytes) -> bytes: - """An 802.11 data frame whose LLC/SNAP header carries the IP ``packet``.""" - return b'\x08\x00\x00\x00' + wire.MAC_B + wire.MAC_A + wire.MAC_A + b'\x00\x00' + snap(packet) - - -def radiotap(packet: bytes) -> bytes: - """A radiotap header with no fields, then an 802.11 frame carrying the IP ``packet``.""" - return b'\x00\x00\x08\x00' + b'\x00' * 4 + dot11(packet) - - -def ppi(packet: bytes) -> bytes: - """A PPI header with no fields, of link type 105, then an 802.11 frame carrying the IP ``packet``.""" - return b'\x00\x00\x08\x00' + (105).to_bytes(4, 'little') + dot11(packet) - - -def backbone(frame: bytes) -> bytes: - """An Ethernet frame whose 802.1ah I-TAG (I-SID 256) carries the customer Ethernet ``frame``.""" - return wire.ethernet(b'\x00\x00\x01\x00' + frame, I_TAG) - - -def fragment6(payload: bytes, *, ident: int, nxt: int = 253, offset: int = 0, mf: bool = False) -> bytes: - """An IPv6 Fragment header, then ``payload``; ``offset`` is in octets.""" - return bytes([nxt, 0]) + (offset | mf).to_bytes(2, 'big') + ident.to_bytes(4, 'big') + payload - - -#: Frames carrying TCP, each with whether the default engine reads the TCP, by a -#: name for it. Each builder takes the TCP segment it is to carry, given by its -#: frame number. -TCP_CASES = { - # #1636, the issue's rows first - 'gre': (False, lambda n: eth(v4(gre(v4(tcp(n), 6)), 47))), - 'vxlan': (False, lambda n: eth(v4(vxlan(eth(v4(tcp(n), 6))), 17))), - 'pppoe': (False, lambda n: pppoe(v4(tcp(n), 6))), - 'l2tp': (False, lambda n: eth(v4(l2tp(v4(tcp(n), 6)), 17))), - 'llc/snap': (False, lambda n: llc_snap(v4(tcp(n), 6))), - 'gre over ipv6': (False, lambda n: eth(v6(gre(v6(tcp(n), 6)), 47))), - 'gre carrying ethernet': (False, lambda n: eth(v4(gre(eth(v4(tcp(n), 6)), TEB), 47))), - 'gre in 6in4': (False, lambda n: eth(v4(v6(gre(v4(tcp(n), 6)), 47), 41))), - 'gre behind hop-by-hop': (False, lambda n: eth(v6(options(47) + gre(v4(tcp(n), 6)), 0))), - 'gre behind 802.1q': (False, lambda n: tagged(v4(gre(v4(tcp(n), 6)), 47), C_TAG)), - 'vxlan over ipv6': (False, lambda n: eth(v6(vxlan(eth(v6(tcp(n), 6))), 17))), - 'pppoe carrying ipv6': (False, lambda n: pppoe(v6(tcp(n), 6))), - 'llc/snap carrying ipv6': (False, lambda n: llc_snap(v6(tcp(n), 6))), - '802.1ah': (False, lambda n: backbone(eth(v4(tcp(n), 6)))), - # controls: what the default engine dissects, VLAN tags, IP-in-IP and extension headers among it - 'tcp': (True, lambda n: eth(v4(tcp(n), 6))), - '802.1q': (True, lambda n: tagged(v4(tcp(n), 6), C_TAG)), - '802.1ad then 802.1q': (True, lambda n: tagged(v6(tcp(n), 6), S_TAG, C_TAG)), - '6in4': (True, lambda n: eth(v4(v6(tcp(n), 6), 41))), - '4in6 behind hop-by-hop': (True, lambda n: eth(v6(options(4) + v4(tcp(n), 6), 0))), - '6in6 behind ah': (True, lambda n: eth(v6(ah(41) + v6(tcp(n), 6), 51))), - '6in6 in a first fragment': (True, lambda n: eth(v6(fragment6(v6(tcp(n), 6), ident=n, nxt=41), 44))), -} - - -class IPCase(NamedTuple): - """The two fragments of a 24-octet datagram, each in a frame of its own.""" - - #: ``'ipv4'`` or ``'ipv6'``, the datagram's version. - version: str - #: How many fragments the default engine hands its reassembly, 0 or 2. - inputs: int - #: The frame a fragment is carried in. - carry: 'Callable[[bytes], bytes]' - #: The IPv6 Fragment header's Next Header. - nxt: int = 253 - - -#: IP fragments in frames, by a name for each case. -IP_CASES = { - # #1635, the issue's rows first - '6in6, the inner packet fragmented': IPCase('ipv6', 0, lambda p: eth(v6(p, 41))), - 'icmpv6 packet too big quoting a fragment': IPCase('ipv6', 0, lambda p: eth(v6(ICMP6_PTB + p, 58))), - 'icmp quoting a 6in4 fragment': IPCase('ipv6', 0, lambda p: eth(v4(ICMP_UNREACH + v4(p, 41), 1))), - '4in6 carrying a 6in4 fragment': IPCase('ipv6', 0, lambda p: eth(v6(v4(p, 41), 4))), - '6in4 fragment in a later ipv4 fragment': IPCase('ipv6', 0, lambda p: eth(v4(p, 41, offset=16))), - 'ipv6 fragment in gre over ipv4': IPCase('ipv6', 0, lambda p: eth(v4(gre(p), 47))), - 'ipv6 fragment in vxlan over ipv4': IPCase('ipv6', 0, lambda p: eth(v4(vxlan(eth(p)), 17))), - '6in6 behind hop-by-hop, the inner packet fragmented': IPCase('ipv6', 0, lambda p: eth(v6(options(41) + p, 0))), - '6in6 behind ah, the inner packet fragmented': IPCase('ipv6', 0, lambda p: eth(v6(ah(41) + p, 51))), - '6in4 fragment behind a rejected ipv4 header': IPCase('ipv6', 0, lambda p: eth(ipv4_header( - p, options=BAD_OPTIONS['timestamp length 2'], proto=41))), - # #1636 - 'ipv6 fragments in gre over ipv4 (df)': IPCase('ipv6', 0, lambda p: eth(wire.ipv4(gre(p), df=True, proto=47))), - 'ipv6 fragments in pppoe': IPCase('ipv6', 0, pppoe), - 'ipv4 fragments in gre over ipv6': IPCase('ipv4', 0, lambda p: eth(v6(gre(p), 47))), - 'ipv4 fragments in vxlan over ipv6': IPCase('ipv4', 0, lambda p: eth(v6(vxlan(eth(p)), 17))), - 'ipv4 fragments in pppoe': IPCase('ipv4', 0, pppoe), - 'ipv4 fragments in llc/snap': IPCase('ipv4', 0, llc_snap), - 'ipv4 fragments in l2tp over ipv4 (df)': IPCase('ipv4', 0, lambda p: eth(wire.ipv4(l2tp(p), df=True, proto=17))), - # controls: the first IPv6 layer, its Fragment header behind another header, and VLAN tags - 'ipv6 fragments': IPCase('ipv6', 2, eth), - 'ipv6 fragments behind hop-by-hop': IPCase('ipv6', 2, lambda p: eth(v6(options(44) + p[40:], 0))), - '6in6, the outer packet fragmented': IPCase('ipv6', 2, eth, nxt=41), - '6in4 fragments': IPCase('ipv6', 2, lambda p: eth(v4(p, 41))), - '6in4 fragments behind ah': IPCase('ipv6', 2, lambda p: eth(v4(ah(41) + p, 51))), - '6in4 fragments behind 802.1ad then 802.1q': IPCase('ipv6', 2, lambda p: tagged(v4(p, 41), S_TAG, C_TAG)), - 'ipv4 fragments behind 802.1q': IPCase('ipv4', 2, lambda p: tagged(p, C_TAG)), - '4in6 fragments': IPCase('ipv4', 2, lambda p: eth(v6(p, 4))), -} - - -#: Link types the default engine leaves raw, at which `Scapy`_ dissects two -#: layers or more before IP, by a name for each: the link type, and the frame an -#: IP packet is carried in. -LINK_TYPES = { - 'ppp (9) with ff 03': (9, ppp), - 'ppp_hdlc (50)': (50, ppp), - 'ppp_ether (51)': (51, session), - 'ieee802_11 (105)': (105, dot11), - 'ieee802_11_radiotap (127)': (127, radiotap), - 'ppi (192)': (192, ppi), -} - - -def fragments(case: IPCase, ident: int) -> 'list[bytes]': - """The frames carrying ``case``'s two fragments, of an Identification of ``ident``.""" - if case.version == 'ipv4': - packets = [ipv4_input.ipv4(b'A' * 16, ident=ident, mf=True), - ipv4_input.ipv4(b'B' * 8, ident=ident, offset=16)] - else: - packets = [v6(fragment6(b'A' * 16, ident=ident, nxt=case.nxt, mf=True), 44), - v6(fragment6(b'B' * 8, ident=ident, nxt=case.nxt, offset=16), 44)] - return [case.carry(packet) for packet in packets] - - -def read(test: unittest.TestCase, engine: str, frames: 'list[bytes]', linktype: int = 1) -> 'dict[str, Any]': - """Read ``frames`` of ``linktype`` with ``engine``, with IPv4, IPv6 and TCP reassembly and TCP flow tracing. - - Returns: - Every input the engine hands each of the four, keyed by frame number -- - every field, but a trace input's ``frame`` -- the IPv4 and IPv6 - datagrams reassembled, the TCP datagrams' ``id``, and the traced flows' - labels and frames. - - """ - 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 = {'ipv4': IPv4.__call__, 'ipv6': IPv6.__call__, 'tcp': TCP.__call__} - 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(aspect: str) -> 'Callable[[Any, Any], Any]': - def call(this: 'Any', packet: 'Any') -> 'Any': - seen[aspect][packet.num] = fields(packet) - return calls[aspect](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, 'walk.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, ipv4=True, ipv6=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') - for version in ('ipv4', 'ipv6'): - seen[f'{version} datagrams'] = [ - (datagram.completed, datagram.id, datagram.index, datagram.header, datagram.payload) - for datagram in getattr(extractor.reassembly, version) - ] - seen['ids'] = [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 - - -@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') -class ScapyDissectedLayerWalkTests(unittest.TestCase): - """scapy hands on the IP and TCP the default engine dissects, frame for frame, and no other.""" - - @classmethod - def setUpClass(cls) -> None: - cls.frames = [make(number) for number, (_, make) in enumerate(TCP_CASES.values(), start=1)] - # one-based frame numbers of each case, and each IP case's frames - cls.numbers = {label: [number] for number, label in enumerate(TCP_CASES, start=1)} - cls.ip_frames = {} # type: dict[str, list[bytes]] - for ident, (label, case) in enumerate(IP_CASES.items(), start=1): - cls.ip_frames[label] = fragments(case, ident) - cls.numbers[label] = [len(cls.frames) + 1, len(cls.frames) + 2] - cls.frames.extend(cls.ip_frames[label]) - - def setUp(self) -> None: - reimport_once_per_class(self) - - def read(self, engine: str) -> 'dict[str, Any]': - """``engine``'s reading of every case, read once per class.""" - if engine == 'scapy' and not HAS_SCAPY: - self.skipTest('scapy not installed') - seen = type(self).__dict__.get('seen') - if seen is None: - seen = type(self).seen = {} - if engine not in seen: - seen[engine] = read(self, engine, self.frames) - return seen[engine] - - def disagree(self, aspect: str) -> 'list[str]': - """The cases whose ``aspect`` inputs scapy and the default engine do not agree on.""" - base, scapy = self.read('default'), self.read('scapy') - return [label for label, numbers in self.numbers.items() - if any(scapy[aspect].get(number) != base[aspect].get(number) for number in numbers)] - - def test_the_default_engine_reads_what_the_cases_say(self) -> None: - base = self.read('default') - reads = [self.numbers[label][0] for label, (expected, _) in TCP_CASES.items() if expected] - self.assertEqual((sorted(base['tcp']), sorted(base['trace'])), (reads, reads)) - self.assertEqual({label: sum(number in base[case.version] for number in self.numbers[label]) - for label, case in IP_CASES.items()}, - {label: case.inputs for label, case in IP_CASES.items()}) - # and the controls' datagrams are whole - for label, case in IP_CASES.items(): - if case.inputs: - with self.subTest(case=label): - self.assertIn(tuple(self.numbers[label]), - [datagram[2] for datagram in base[f'{case.version} datagrams'] if datagram[0]]) - - def test_ipv6_reassembly_takes_the_first_ipv6_layers_own_fragment_header(self) -> None: - # #1635 - self.assertEqual(self.disagree('ipv6'), []) - self.assertEqual(self.read('scapy')['ipv6 datagrams'], self.read('default')['ipv6 datagrams']) - - def test_no_tcp_is_read_inside_a_layer_the_default_engine_leaves_raw(self) -> None: - # #1636 - self.assertEqual((self.disagree('tcp'), self.disagree('trace')), ([], [])) - for aspect in ('ids', 'flows'): - self.assertEqual(self.read('scapy')[aspect], self.read('default')[aspect], aspect) - - def test_no_ipv4_is_read_inside_a_layer_the_default_engine_leaves_raw(self) -> None: - # #1636 - self.assertEqual(self.disagree('ipv4'), []) - self.assertEqual(self.read('scapy')['ipv4 datagrams'], self.read('default')['ipv4 datagrams']) - - @unittest.skipUnless(HAS_SCAPY, 'scapy not installed') - def test_no_ip_is_read_two_layers_into_a_link_type_the_default_engine_leaves_raw(self) -> None: - # #1636, at the first layer: TCP in IPv4 and in IPv6, then two IPv4 and two IPv6 fragments - for label, (linktype, carry) in LINK_TYPES.items(): - frames = [carry(v4(tcp(1), 6)), carry(v6(tcp(2), 6)), - *fragments(IPCase('ipv4', 0, carry), 3), *fragments(IPCase('ipv6', 0, carry), 4)] - with self.subTest(link_type=label): - base = read(self, 'default', frames, linktype) - self.assertEqual([base[aspect] for aspect in ('ipv4', 'ipv6', 'tcp', 'trace')], [{}] * 4) - self.assertEqual(read(self, 'scapy', frames, linktype), base) - - @unittest.skipUnless(HAS_SCAPY, 'scapy not installed') - def test_a_packet_built_by_hand_is_walked_too(self) -> None: - # ``scapy.all`` binds every layer here, as the engine has it do (see pcapkit.toolkit.scapy) - from scapy.all import Ether, IPerror, IPerror6 - - from pcapkit.toolkit import scapy as toolkit - from pcapkit.utilities.exceptions import ModuleNotFound - - got = {label: toolkit.tcp_reassembly(Ether(make(1))) is not None for label, (_, make) in TCP_CASES.items()} - self.assertEqual(got, {label: expected for label, (expected, _) in TCP_CASES.items()}) - got = {label: [getattr(toolkit, f'{case.version}_reassembly')(Ether(frame)) is not None - for frame in self.ip_frames[label]] for label, case in IP_CASES.items()} - self.assertEqual(got, {label: [bool(case.inputs)] * 2 for label, case in IP_CASES.items()}) - # nor is a quoted datagram read when it is handed over on its own - first4 = ipv4_input.ipv4(b'A' * 16, ident=1, mf=True) - first6 = v6(fragment6(b'A' * 16, ident=1, mf=True), 44) - self.assertEqual([toolkit.ipv4_reassembly(IPerror(first4)), toolkit.tcp_reassembly(IPerror(v4(tcp(1), 6))), - toolkit.ipv6_reassembly(IPerror6(first6))], [None] * 3) - self.assertIsNotNone(toolkit.ipv4_reassembly(Ether(eth(first4)))) - - # the input is the first IPv6 layer's, cut at its Fragment header - frame = self.ip_frames['6in6, the outer packet fragmented'][0] - data = toolkit.ipv6_reassembly(Ether(frame), count=1) - assert data is not None - self.assertEqual((data.ihl, bytes(data.header), bytes(data.payload)), (40, frame[14:54], b'A' * 16)) - with mock.patch.object(toolkit, 'scapy', None), self.assertRaises(ModuleNotFound): - toolkit.ipv6_reassembly(Ether(frame)) - - -if __name__ == '__main__': - unittest.main() diff --git a/tests/toolkit/test_scapy_ipv6_payload_and_icmp_quote_1623_1627_unit.py b/tests/toolkit/test_scapy_ipv6_payload_and_icmp_quote_1623_1627_unit.py deleted file mode 100644 index 5fa56f572..000000000 --- a/tests/toolkit/test_scapy_ipv6_payload_and_icmp_quote_1623_1627_unit.py +++ /dev/null @@ -1,266 +0,0 @@ -# -*- coding: utf-8 -*- -"""scapy reads IPv4 inside each IPv6 Payload Length, and nothing an ICMP error quotes. C.f. #1623, #1627. - -The default engine hands IPv4 reassembly the first IPv4 layer of a frame, with -the octets inside the Payload Length of each IPv6 packet above it. `Scapy`_ files -whatever follows a Payload Length as padding at the end of the packet, so -``bytes()`` of an IPv4 layer carried in IPv6 ran on into the frame's trailer, or -past a Payload Length shorter than the IPv4 packet (#1623). - -The default engine has no ICMP or ICMPv6 parser, so it reads nothing an ICMP -message carries. `Scapy`_ dissects the datagram an error quotes as ``IPerror`` or -``IPerror6``, which bind a full ``IPv6`` for 6in4 and 6in6, or ``IP`` for 4in6, -and so a real ``TCP``: that TCP went to TCP reassembly and flow tracing (#1627), -and that IPv4 to IPv4 reassembly. ``packet['IP']`` also reached the IPv4 inside -the IPv6 packet a Redirected Header option holds, in a field rather than a payload. - -dpkt is compared with the default engine as well. - -Every capture is built here, so the module is in the unit tier. - -.. _Scapy: https://scapy.net - -""" -from __future__ import annotations - -import importlib.util -import unittest -from typing import TYPE_CHECKING -from unittest import mock - -from tests._support import reimport_once_per_class -from tests.toolkit import test_dpkt_ipv4_reassembly_input_1615_1616_unit as ipv4_input -from tests.toolkit import test_tcp_carrier_walk_1606_1617_unit as carrier_walk -from tests.toolkit.test_tcp_carrier_walk_1606_1617_unit import eth, options, tcp, v4, v6 - -if TYPE_CHECKING: - from typing import Any, Callable - -RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') -HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) - -#: The engines compared with the default engine. -ENGINES = ('dpkt', 'scapy') - -#: ICMP Destination Unreachable and Time Exceeded, and an Echo Request, which quotes nothing. -ICMP_UNREACH = bytes([3, 3, 0, 0, 0, 0, 0, 0]) -ICMP_TIMEX = bytes([11, 0, 0, 0, 0, 0, 0, 0]) -ICMP_ECHO = bytes([8, 0, 0, 0, 0, 1, 0, 1]) -#: ICMPv6 Destination Unreachable, Packet Too Big, Time Exceeded and Parameter Problem. -ICMP6_UNREACH = bytes([1, 4, 0, 0, 0, 0, 0, 0]) -ICMP6_PTB = bytes([2, 0, 0, 0, 0, 0, 5, 0]) -ICMP6_TIMEX = bytes([3, 0, 0, 0, 0, 0, 0, 0]) -ICMP6_PARAM = bytes([4, 0, 0, 0, 0, 0, 0, 40]) - - -def has(engine: str) -> bool: - """Whether ``engine``'s third-party package is installed.""" - return importlib.util.find_spec(engine) is not None - - -def redirect(packet: bytes) -> bytes: - """An ICMPv6 Redirect whose Redirected Header option holds ``packet`` (RFC 4861, section 4.6.3).""" - held = packet + bytes(-(len(packet) + 8) % 8) - return bytes([137, 0, 0, 0]) + bytes(36) + bytes([4, (len(held) + 8) // 8]) + bytes(6) + held - - -def ipv4_cases() -> 'dict[str, list[bytes]]': - """#1615/#1616's frames by case, then those of #1623 and the ICMP quotes.""" - built = ipv4_input.cases() - ethernet, ipv6 = ipv4_input.ethernet, ipv4_input.ipv6 - - def first(ident: int) -> bytes: - return ipv4_input.ipv4(b'A' * 16, ident=ident, mf=True) - - def last(ident: int) -> bytes: - return ipv4_input.ipv4(b'B' * 8, ident=ident, offset=16) - - def pair(label: str, carry: 'Callable[[bytes], bytes]', ident: int) -> None: - built[label] = [carry(first(ident)), carry(last(ident))] - - # #1623: each Payload Length above bounds the IPv4 octets, the inner as well as the outer - built['4in6 cut by each Payload Length'] = [ethernet(ipv6(first(31), 4, plen=30)), - ethernet(ipv6(last(31), 4, plen=24))] - pair('4in6 in 6in6 with a trailer', lambda p: ethernet(ipv6(ipv6(p, 4), 41), trailer=b'\xee' * 6), 32) - built['4in6 in 6in6 cut by the inner Payload Length'] = [ethernet(ipv6(ipv6(first(33), 4, plen=30), 41))] - built['4in6 in 6in6 cut into the IPv4 header'] = [ethernet(ipv6(ipv6(first(34), 4), 41, plen=50))] - built['4in6 cut into the IPv4 options'] = [ethernet(ipv6(ipv4_input.ipv4( - b'A' * 16, ident=35, options=b'\x94\x04\x00\x00', mf=True), 4, plen=22), trailer=b'\x00' * 8)] - # the outer IPv4 of 4in6in4 is the first, its octets running on to the trailer - built['4in6in4 with a trailer'] = [ethernet(ipv4_input.ipv4(ipv6(first(36), 4), ident=37, proto=41), - trailer=b'\xee' * 6)] - # #1627: no IPv4 an ICMP message carries, quoted or held in an option - pair('4in6 quoted by ICMPv6', lambda p: ethernet(ipv6(ICMP6_UNREACH + ipv6(p, 4), 58)), 38) - pair('4in6 quoted by ICMPv6 behind Hop-by-Hop Options', - lambda p: ethernet(ipv6(ipv4_input.options(58) + ICMP6_UNREACH + ipv6(p, 4), 0)), 39) - pair('4in6 in a Redirected Header', lambda p: ethernet(ipv6(redirect(ipv6(p, 4)), 58)), 40) - return built - - -#: Frames carrying TCP inside an ICMP or ICMPv6 message, and controls, each with -#: whether the default engine reads the TCP, by a name for it. Each builder takes -#: the TCP segment it is to carry, given by its frame number. -QUOTES = { - # #1627: IPerror binds IPv6 for protocol 41, and IPerror6 binds IPv6 for 41 and IPv4 for 4 - 'icmp quoting 6in4': (False, lambda n: eth(v4(ICMP_UNREACH + v4(v6(tcp(n), 6), 41), 1))), - 'icmp time exceeded quoting 6in4': (False, lambda n: eth(v4(ICMP_TIMEX + v4(v6(tcp(n), 6), 41), 1))), - 'icmp quoting 4in6 in 6in4': (False, lambda n: eth(v4(ICMP_UNREACH + v4(v6(v4(tcp(n), 6), 4), 41), 1))), - 'icmp in 4in4 quoting 6in4': (False, lambda n: eth(v4(v4(ICMP_UNREACH + v4(v6(tcp(n), 6), 41), 1), 4))), - 'icmp in 4in6 quoting 6in4': (False, lambda n: eth(v6(v4(ICMP_UNREACH + v4(v6(tcp(n), 6), 41), 1), 4))), - 'icmpv6 quoting 6in6': (False, lambda n: eth(v6(ICMP6_UNREACH + v6(v6(tcp(n), 6), 41), 58))), - 'icmpv6 packet too big quoting 6in6': (False, lambda n: eth(v6(ICMP6_PTB + v6(v6(tcp(n), 6), 41), 58))), - 'icmpv6 time exceeded quoting 6in6': (False, lambda n: eth(v6(ICMP6_TIMEX + v6(v6(tcp(n), 6), 41), 58))), - 'icmpv6 parameter problem quoting 6in6': (False, lambda n: eth(v6(ICMP6_PARAM + v6(v6(tcp(n), 6), 41), 58))), - 'icmpv6 quoting 4in6': (False, lambda n: eth(v6(ICMP6_UNREACH + v6(v4(tcp(n), 6), 4), 58))), - 'icmpv6 behind hop-by-hop quoting 6in6': (False, lambda n: eth(v6(options(58) + ICMP6_UNREACH - + v6(v6(tcp(n), 6), 41), 0))), - 'icmpv6 in 6in4 quoting 6in6': (False, lambda n: eth(v4(v6(ICMP6_UNREACH + v6(v6(tcp(n), 6), 41), 58), 41))), - # controls: quotes Scapy dissects as TCPerror, a message quoting nothing, a packet - # Scapy holds in a field, and no ICMP at all - 'icmp quoting 4in4': (False, lambda n: eth(v4(ICMP_UNREACH + v4(v4(tcp(n), 6), 4), 1))), - 'icmp quoting tcp': (False, lambda n: eth(v4(ICMP_UNREACH + v4(tcp(n), 6), 1))), - 'icmpv6 quoting tcp': (False, lambda n: eth(v6(ICMP6_UNREACH + v6(tcp(n), 6), 58))), - 'icmp echo carrying 6in4': (False, lambda n: eth(v4(ICMP_ECHO + v4(v6(tcp(n), 6), 41), 1))), - 'redirected header holding 6in6': (False, lambda n: eth(v6(redirect(v6(v6(tcp(n), 6), 41)), 58))), - '6in4': (True, lambda n: eth(v4(v6(tcp(n), 6), 41))), - '6in6': (True, lambda n: eth(v6(v6(tcp(n), 6), 41))), - '4in6 with a trailer': (True, lambda n: eth(v6(v4(tcp(n), 6), 4)) + b'\xee' * 6), - 'tcp': (True, lambda n: eth(v4(tcp(n), 6))), -} - - -@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') -class ScapyIPv4ReassemblyInputTests(unittest.TestCase): - """scapy's IPv4 reassembly inputs are the default engine's, octet for octet.""" - - @classmethod - def setUpClass(cls) -> None: - cls.cases = ipv4_cases() - cls.frames = [frame for frames in cls.cases.values() for frame in frames] - # one-based frame numbers of each case - cls.numbers = {} # type: dict[str, list[int]] - number = 0 - for label, frames in cls.cases.items(): - cls.numbers[label] = list(range(number + 1, number + 1 + len(frames))) - number += len(frames) - - def setUp(self) -> None: - reimport_once_per_class(self) - - def read(self, engine: str) -> 'dict[str, Any]': - """``engine``'s reading of every case, read once per class.""" - if engine != 'default' and not has(engine): - self.skipTest(f'{engine} not installed') - seen = type(self).__dict__.get('seen') - if seen is None: - seen = type(self).seen = {} - if engine not in seen: - seen[engine] = ipv4_input.read(self, engine, self.frames) - return seen[engine] - - def inputs(self, seen: 'dict[str, Any]', label: str) -> 'list[dict[str, Any]]': - return [seen['inputs'][number] for number in self.numbers[label] if number in seen['inputs']] - - def test_every_input_agrees_with_the_default_engine(self) -> None: - base = self.read('default') - for engine in ENGINES: - with self.subTest(engine=engine): - seen = self.read(engine) - for label in self.cases: - with self.subTest(engine=engine, case=label): - self.assertEqual(self.inputs(seen, label), self.inputs(base, label)) - self.assertEqual(seen['datagrams'], base['datagrams']) - - def test_each_payload_length_above_bounds_the_input(self) -> None: - # #1623, the issue's two cases first - scapy = self.read('scapy') - for label, payloads in ( - ('4in6 with a trailer', [16, 8]), - ('4in6 cut by each Payload Length', [10, 4]), - ('4in6 cut by the Payload Length', [10]), - ('4in6 cut by the outer Payload Length of 6in6', [10]), - ('Total Length 0 in 4in6, with a trailer', [16]), - ('4in6 in 6in6 with a trailer', [16, 8]), - ('4in6 in 6in6 cut by the inner Payload Length', [10]), - ('4in6in4 with a trailer', [76 + 6]), - ): - with self.subTest(case=label): - self.assertEqual([len(packet['payload']) for packet in self.inputs(scapy, label)], payloads) - # so the datagrams cut short are left incomplete, and the others complete - complete = [datagram[2] for datagram in scapy['datagrams'] if datagram[0]] - self.assertNotIn(tuple(self.numbers['4in6 cut by each Payload Length']), complete) - self.assertIn(tuple(self.numbers['4in6 with a trailer']), complete) - - def test_no_input_where_the_default_engine_dissects_no_ipv4(self) -> None: - for label in ('4in6 jumbogram shorter than the IPv4 header', '4in6 in 6in6 cut into the IPv4 header', - '4in6 cut into the IPv4 options', '4in6 quoted by ICMPv6', - '4in6 quoted by ICMPv6 behind Hop-by-Hop Options', '4in6 in a Redirected Header'): - for engine in ('default', *ENGINES): - with self.subTest(case=label, engine=engine): - self.assertEqual(self.inputs(self.read(engine), label), []) - - def test_a_packet_built_by_hand_is_bounded_too(self) -> None: - # with no capture behind it, the octets are the ones Scapy dissected - if not has('scapy'): - self.skipTest('scapy not installed') - # ``scapy.all`` binds IPv6 and ICMPv6, as the engine has it do (see pcapkit.toolkit.scapy) - from scapy.all import Ether - - from pcapkit.toolkit import scapy as toolkit - from pcapkit.utilities.exceptions import ModuleNotFound - - frame = self.cases['4in6 with a trailer'][0] - data = toolkit.ipv4_reassembly(Ether(frame), count=1) - self.assertIsNotNone(data) - assert data is not None - self.assertEqual((bytes(data.header), bytes(data.payload)), (frame[54:74], b'A' * 16)) - self.assertIsNone(toolkit.ipv4_reassembly(Ether(self.cases['4in6 quoted by ICMPv6'][0]))) - with mock.patch.object(toolkit, 'scapy', None), self.assertRaises(ModuleNotFound): - toolkit.ipv4_reassembly(Ether(frame)) - - -@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') -class ScapyICMPQuoteTests(unittest.TestCase): - """scapy hands on no TCP inside the datagram an ICMP or ICMPv6 error quotes.""" - - def setUp(self) -> None: - reimport_once_per_class(self) - - def test_no_tcp_is_read_in_an_icmp_quote(self) -> None: - # #1627 - frames, reads = carrier_walk.build(QUOTES) - base = carrier_walk.read(self, 'default', frames) - # what the default engine reads, as the cases say - self.assertEqual(sorted(base['tcp']), reads) - self.assertEqual(sorted(base['trace']), reads) - names = dict(enumerate(QUOTES, start=1)) - for engine in ENGINES: - with self.subTest(engine=engine): - if not has(engine): - self.skipTest(f'{engine} not installed') - seen = carrier_walk.read(self, engine, frames) - for aspect in ('tcp', 'trace'): - self.assertEqual( - [names[number] for number in sorted(set(seen[aspect]) ^ set(base[aspect]))], [], - f'{aspect}: frames read by one of {engine} and the default engine alone') - self.assertEqual(seen[aspect], base[aspect], aspect) - for aspect in ('ids', 'flows'): - self.assertEqual(seen[aspect], base[aspect], aspect) - - def test_a_packet_built_by_hand_is_walked_too(self) -> None: - if not has('scapy'): - self.skipTest('scapy not installed') - # ``scapy.all`` binds IPv6 and ICMPv6, as the engine has it do (see pcapkit.toolkit.scapy) - from scapy.all import Ether - - from pcapkit.toolkit import scapy as toolkit - - for label, (reads, make) in QUOTES.items(): - with self.subTest(case=label): - packet = Ether(make(1)) - self.assertEqual(toolkit.tcp_reassembly(packet) is not None, reads) - self.assertEqual(toolkit.tcp_traceflow(packet) is not None, reads) - - -if __name__ == '__main__': - unittest.main() diff --git a/tests/toolkit/test_scapy_linktype_and_layer_bounds_1609_1626_1650_unit.py b/tests/toolkit/test_scapy_linktype_and_layer_bounds_1609_1626_1650_unit.py deleted file mode 100644 index e5194e85d..000000000 --- a/tests/toolkit/test_scapy_linktype_and_layer_bounds_1609_1626_1650_unit.py +++ /dev/null @@ -1,362 +0,0 @@ -# -*- coding: utf-8 -*- -"""scapy reads what the default engine dissects at every link type, and within its bounds. C.f. #1609, #1626, #1650. - -The default engine dissects the first layer of six link types alone, and keeps -the frames of any other raw. `Scapy`_ dissects IP and TCP straight behind the -headers of many of those, such as ``LINUX_SLL``, ``LINUX_SLL2``, ``PPP`` and the -``DLT_RAW`` of 12, and its engine handed them over. Behind a BSD loopback header -it read IPv6 at the ``AF_INET6`` of the host it ran on, and IPv4 at family 0 too, -where the default engine reads IPv4 at family 2 and IPv6 at 24, 28 and 30 on -every host (#1650). - -The default engine dissects no layer past the 64th of a frame, nor an -extension header past the 32nd in a row. `Scapy`_ stops at neither, and its -engine handed over the TCP, IPv4 and IPv6 fragments it found past them, behind -62 VLAN tags among them (#1609). - -`Scapy`_ dissects no TCP and no IPv4 behind an AH, and IPv4 where its Next Header -is 0; nothing behind a Mobility header; and neither HIP nor Shim6. The default -engine reads past each, so scapy now walks an IPv6 extension header chain over -the octets by the default engine's rules, as it walks those inside IPv4 (#1626). - -Every engine is compared frame for frame with the default engine, on every -input it hands TCP and IP reassembly and TCP flow tracing. Every capture is -built here, so the module is in the unit tier. - -.. _Scapy: https://scapy.net - -""" -from __future__ import annotations - -import importlib.util -import unittest -from typing import TYPE_CHECKING, NamedTuple -from unittest import mock - -from tests._support import reimport_once_per_class -from tests.foundation import _roundtrip as wire -from tests.toolkit import test_dpkt_ipv4_reassembly_input_1615_1616_unit as ipv4_input -from tests.toolkit.test_scapy_ipv6_fragment_and_encap_walk_1635_1636_unit import (C_TAG, fragment6, - read, tagged) -from tests.toolkit.test_tcp_carrier_walk_1606_1617_unit import ah, eth, options, tcp, v4, v6 - -if TYPE_CHECKING: - from typing import Any, Callable - -RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') -HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) -HAS_SCAPY = importlib.util.find_spec('scapy') is not None - -#: The inputs compared, as :func:`read` keys them. -ASPECTS = ('tcp', 'trace', 'ipv4', 'ipv6', 'ipv4 datagrams', 'ipv6 datagrams', 'ids', 'flows') - - -class Case(NamedTuple): - """Frames of one link type, and what the default engine reads in them.""" - - #: The capture's link type. - linktype: int - #: The frames. - frames: 'list[bytes]' - #: How many frames the default engine hands TCP, IPv4 and IPv6 reassembly. - reads: 'tuple[int, int, int]' - - -def version(packet: bytes) -> int: - return 6 if packet[0] >> 4 == 6 else 4 - - -def v4_fragments(ident: int) -> 'list[bytes]': - """The two fragments of a 24-octet IPv4 datagram.""" - return [ipv4_input.ipv4(b'A' * 16, ident=ident, mf=True), ipv4_input.ipv4(b'B' * 8, ident=ident, offset=16)] - - -def v6_fragments(ident: int, head: bytes = b'', code: int = 44) -> 'list[bytes]': - """The two fragments of a 24-octet IPv6 datagram, ``head`` (of code ``code``) before each Fragment header.""" - return [v6(head + fragment6(b'A' * 16, ident=ident, mf=True), code), - v6(head + fragment6(b'B' * 8, ident=ident, offset=16), code)] - - -def each(carry: 'Callable[[bytes], bytes]') -> 'list[bytes]': - """TCP in IPv4 and in IPv6, then two IPv4 and two IPv6 fragments, each carried by ``carry``.""" - return [carry(packet) for packet in (v4(tcp(1), 6), v6(tcp(2), 6), *v4_fragments(3), *v6_fragments(4))] - - -def ethertype(packet: bytes) -> bytes: - """The EtherType of the IP ``packet``, as octets.""" - return (0x86DD if version(packet) == 6 else 0x0800).to_bytes(2, 'big') - - -def sll(packet: bytes) -> bytes: - """A LINUX_SLL header (16 octets), then the IP ``packet``.""" - return b'\x00\x00\x00\x01\x00\x06' + wire.MAC_A + b'\x00\x00' + ethertype(packet) + packet - - -def sll2(packet: bytes) -> bytes: - """A LINUX_SLL2 header (20 octets), then the IP ``packet``.""" - return ethertype(packet) + b'\x00\x00\x00\x00\x00\x01\x00\x01\x00\x06' + wire.MAC_A + b'\x00\x00' + packet - - -def ppp(packet: bytes) -> bytes: - """A PPP frame with no HDLC-like framing, carrying the IP ``packet``.""" - return (0x0057 if version(packet) == 6 else 0x0021).to_bytes(2, 'big') + packet - - -def loopback(order: str, ipv4: int, ipv6: int) -> 'Callable[[bytes], bytes]': - """A BSD loopback header of family ``ipv4`` or ``ipv6``, written ``order``-endian, then the IP packet.""" - return lambda packet: (ipv6 if version(packet) == 6 else ipv4).to_bytes(4, order) + packet - - -def nest(packet: bytes, count: int, outer: int) -> bytes: - """``packet`` tunnelled in ``count`` IPv4 (``outer`` 4) or IPv6 (6) headers.""" - for _ in range(count): - proto = 41 if version(packet) == 6 else 4 - packet = v4(packet, proto) if outer == 4 else v6(packet, proto) - return packet - - -def vlan(packet: bytes, count: int) -> bytes: - """An Ethernet frame carrying the IP ``packet`` behind ``count`` 802.1Q tags.""" - return tagged(packet, *[C_TAG] * count) - - -def pairs(count: int, ipv6: bool) -> bytes: - """TCP behind ``count`` IPv4 packets, each with an AH, or IPv6 packets, each with a Hop-by-Hop header.""" - packet, nxt = tcp(1), 6 - for _ in range(count): - packet = v6(options(nxt) + packet, 0) if ipv6 else v4(ah(nxt) + packet, 51) - nxt = 41 if ipv6 else 4 - return packet - - -def chain(count: int, nxt: int, code: int = 60) -> bytes: - """``count`` 8-octet headers of type ``code``, each one PadN option, the last naming ``nxt``.""" - return b''.join(bytes([code if index < count - 1 else nxt, 0, 1, 4, 0, 0, 0, 0]) for index in range(count)) - - -def frag(nxt: int, mf: bool = False) -> bytes: - """A Fragment header of offset 0.""" - return bytes([nxt, 0, 0, int(mf)]) + (0xDEADBEEF).to_bytes(4, 'big') - - -#: The IPv6 extension headers of #1626, each a builder taking the Next Header it -#: names, and its own code. -HEADERS = { - 'fragment then ah': (lambda nxt: frag(51) + ah(nxt), 44), - 'ah': (ah, 51), - 'mobility': (lambda nxt: bytes([nxt, 0, 0, 0, 0, 0, 0, 0]), 135), - 'hip': (lambda nxt: bytes([nxt, 4, 1, 0x21, 0, 0, 0, 0]) + b'\x11' * 32, 139), - 'shim6': (lambda nxt: bytes([nxt, 0]) + bytes(6), 140), - 'hop-by-hop then mobility': (lambda nxt: options(135) + bytes([nxt, 0, 0, 0, 0, 0, 0, 0]), 0), - 'ah then hop-by-hop': (lambda nxt: ah(0) + options(nxt), 51), -} - - -def cases_1650() -> 'dict[str, Case]': - cases = { - 'linux_sll (113)': Case(113, each(sll), (0, 0, 0)), - 'linux_sll2 (276)': Case(276, each(sll2), (0, 0, 0)), - 'dlt_raw (12)': Case(12, each(lambda packet: packet), (0, 0, 0)), - 'ppp (9) without ff 03': Case(9, each(ppp), (0, 0, 0)), - # controls: the link types the default engine dissects - 'ethernet (1)': Case(1, each(eth), (2, 3, 2)), - 'raw (101)': Case(101, each(lambda packet: packet), (2, 3, 2)), - 'ipv4 (228)': Case(228, each(lambda packet: packet), (1, 3, 0)), - 'ipv6 (229)': Case(229, each(lambda packet: packet), (1, 0, 2)), - } - for linktype, name, order in ((0, 'null', 'little'), (0, 'null', 'big'), (108, 'loop', 'big')): - for family in (24, 28, 30): - cases[f'{name} ({linktype}) {order}-endian, ipv6 at {family}'] = Case( - linktype, each(loopback(order, 2, family)), (2, 3, 2)) - cases[f'{name} ({linktype}) {order}-endian, ipv6 at 10'] = Case( - linktype, each(loopback(order, 2, 10)), (1, 3, 0)) - cases[f'{name} ({linktype}) {order}-endian, family 0'] = Case( - linktype, each(loopback(order, 0, 0)), (0, 0, 0)) - cases['loop (108) little-endian'] = Case(108, each(loopback('little', 2, 24)), (0, 0, 0)) - cases['null (0) cut short'] = Case(0, [b'\x02\x00'], (0, 0, 0)) - cases['null (0) ipv4 cut short'] = Case(0, [loopback('little', 2, 24)(v4(tcp(1), 6))[:20]], (0, 0, 0)) - return cases - - -def cases_1609() -> 'dict[str, Case]': - # Ethernet is layer 1, and each VLAN tag, IP header and extension header one more - return { - 'tcp in 62 ipv4 (layer 64)': Case(1, [eth(nest(v4(tcp(1), 6), 61, 4))], (1, 1, 0)), - 'tcp in 63 ipv4 (layer 65)': Case(1, [eth(nest(v4(tcp(1), 6), 62, 4))], (0, 1, 0)), - 'tcp in 63 ipv6 (layer 65)': Case(1, [eth(nest(v6(tcp(1), 6), 62, 6))], (0, 0, 0)), - 'tcp behind 31 (ipv4, ah) pairs (layer 64)': Case(1, [eth(pairs(31, False))], (1, 1, 0)), - 'tcp behind 32 (ipv4, ah) pairs (layer 66)': Case(1, [eth(pairs(32, False))], (0, 1, 0)), - 'tcp behind 31 (ipv6, hop-by-hop) pairs (layer 64)': Case(1, [eth(pairs(31, True))], (1, 0, 0)), - 'tcp behind 32 (ipv6, hop-by-hop) pairs (layer 66)': Case(1, [eth(pairs(32, True))], (0, 0, 0)), - 'ipv4 fragments in 62 ipv6 (layer 64)': Case(1, [eth(nest(p, 62, 6)) for p in v4_fragments(1)], (0, 2, 0)), - 'ipv4 fragments in 63 ipv6 (layer 65)': Case(1, [eth(nest(p, 63, 6)) for p in v4_fragments(1)], (0, 0, 0)), - 'ipv6 fragments in 62 ipv4 (layer 65)': Case(1, [eth(nest(p, 62, 4)) for p in v6_fragments(1)], (0, 2, 0)), - 'tcp behind 61 vlan tags (layer 64)': Case(1, [vlan(v4(tcp(1), 6), 61)], (1, 1, 0)), - 'tcp behind 62 vlan tags (layer 65)': Case(1, [vlan(v4(tcp(1), 6), 62)], (0, 1, 0)), - 'ipv4 fragments behind 63 vlan tags (layer 65)': Case(1, [vlan(p, 63) for p in v4_fragments(1)], (0, 0, 0)), - 'ipv6 fragments behind 62 vlan tags (layer 65)': Case(1, [vlan(p, 62) for p in v6_fragments(1)], (0, 0, 0)), - 'tcp behind 32 extension headers': Case(1, [eth(v6(chain(32, 6) + tcp(1), 60))], (1, 0, 0)), - 'tcp behind 33 extension headers': Case(1, [eth(v6(chain(33, 6) + tcp(1), 60))], (0, 0, 0)), - 'tcp behind 33 hop-by-hop headers': Case(1, [eth(v6(chain(33, 6, 0) + tcp(1), 0))], (0, 0, 0)), - 'tcp behind 1000 extension headers': Case(1, [eth(v6(chain(1000, 6) + tcp(1), 60))], (0, 0, 0)), - 'fragment header after 32 extension headers': Case( - 1, [eth(p) for p in v6_fragments(1, chain(32, 44), 60)], (0, 0, 0)), - 'ipv4 fragments behind 33 extension headers': Case( - 1, [eth(v6(chain(33, 4) + p, 60)) for p in v4_fragments(1)], (0, 0, 0)), - # Ethernet, 50 IPv4 at 2 to 51, IPv6 at 52, then the chain: the frame bound cuts it - 'tcp behind 50 ipv4, ipv6 and 12 extension headers (layer 65)': Case( - 1, [eth(nest(v6(chain(12, 6) + tcp(1), 60), 50, 4))], (0, 1, 0)), - 'fragment header behind 50 ipv4, ipv6 and 12 extension headers (layer 65)': Case( - 1, [eth(nest(v6(chain(12, 44) + fragment6(b'A' * 16, ident=1, mf=True), 60), 50, 4))], (0, 1, 0)), - 'tcp behind 62 ipv4 and a hop-by-hop header (layer 65)': Case( - 1, [eth(nest(v4(options(6) + tcp(1), 0), 61, 4))], (0, 1, 0)), - 'tcp in 64 ipv4 at link type raw (layer 65)': Case(101, [nest(v4(tcp(1), 6), 63, 4)], (0, 1, 0)), - 'tcp in 63 ipv4 behind a loopback header (layer 65)': Case( - 0, [loopback('little', 2, 24)(nest(v4(tcp(1), 6), 62, 4))], (0, 1, 0)), - # controls, at the bound - 'tcp in 62 ipv6 (layer 64)': Case(1, [eth(nest(v6(tcp(1), 6), 61, 6))], (1, 0, 0)), - 'ipv6 fragments in 61 ipv4 (layer 64)': Case(1, [eth(nest(p, 61, 4)) for p in v6_fragments(1)], (0, 2, 2)), - 'ipv6 fragments behind 61 vlan tags (layer 64)': Case(1, [vlan(p, 61) for p in v6_fragments(1)], (0, 0, 2)), - 'fragment header after 31 extension headers': Case( - 1, [eth(p) for p in v6_fragments(1, chain(31, 44), 60)], (0, 0, 2)), - } - - -def cases_1626() -> 'dict[str, Case]': - cases = {} - for label, (head, code) in HEADERS.items(): - cases[f'ipv6, {label}, tcp'] = Case(1, [eth(v6(head(6) + tcp(1), code)), - eth(v6(head(6) + tcp(2, opts=b'\x02\x03\x05\xb4'), code))], - (2, 0, 2 if label == 'fragment then ah' else 0)) - cases[f'ipv6, {label}, ipv4 fragments'] = Case(1, [eth(v6(head(4) + p, code)) for p in v4_fragments(1)], - (0, 2, 2 if label == 'fragment then ah' else 0)) - cases[f'ipv6, {label}, 6in6'] = Case(1, [eth(v6(head(41) + v6(tcp(1), 6), code))], - (1, 0, 1 if label == 'fragment then ah' else 0)) - if label != 'fragment then ah': - cases[f'ipv6, {label}, then ipv6 fragments'] = Case(1, [eth(p) for p in v6_fragments(1, head(44), code)], - (0, 0, 2)) - # behind an extension header in IPv4, the IPv6 packet IPv6 reassembly takes - cases['6in4 fragments behind hop-by-hop in ipv4'] = Case(1, [eth(v4(options(41) + p, 0)) for p in v6_fragments(1)], - (0, 2, 2)) - cases['6in4 fragments behind mobility in ipv4'] = Case( - 1, [eth(v4(bytes([41, 0, 0, 0, 0, 0, 0, 0]) + p, 135)) for p in v6_fragments(1)], (0, 2, 2)) - # controls: what Scapy already dissects as the default engine does, and what neither reads past - cases['ipv6, fragment then destination options, tcp'] = Case(1, [eth(v6(frag(60) + options(6) + tcp(1), 44))], - (1, 0, 1)) - cases['ipv6, two fragment headers, tcp'] = Case(1, [eth(v6(frag(44, mf=True) + frag(6) + tcp(1), 44))], (1, 0, 1)) - cases['ipv6, mobility, tcp cut short'] = Case(1, [eth(v6(HEADERS['mobility'][0](6) + tcp(1)[:12], 135))], - (0, 0, 0)) - cases['ipv6, esp'] = Case(1, [eth(v6(b'\x00\x00\x01\x00\x00\x00\x00\x01' + bytes(8) + tcp(1), 50))], (0, 0, 0)) - cases['ipv6, 253'] = Case(1, [eth(v6(bytes([6, 0, 0, 0, 0, 0, 0, 0]) + tcp(1), 253))], (0, 0, 0)) - cases['ipv6, payload length cuts the chain'] = Case( - 1, [eth(v6(options(6) + tcp(1), 60)[:4] + (4).to_bytes(2, 'big') + v6(options(6) + tcp(1), 60)[6:])], (0, 0, 0)) - return cases - - -@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') -@unittest.skipUnless(HAS_SCAPY, 'scapy not installed') -class ScapyLinkTypeAndLayerBoundTests(unittest.TestCase): - """scapy hands on the TCP and IP the default engine dissects, frame for frame, and no other.""" - - def setUp(self) -> None: - reimport_once_per_class(self) - - def agree(self, cases: 'dict[str, Case]') -> None: - reads, disagree = {}, [] - for label, case in cases.items(): - base = read(self, 'default', case.frames, case.linktype) - reads[label] = tuple(len(base[aspect]) for aspect in ('tcp', 'ipv4', 'ipv6')) - self.assertEqual(sorted(base['tcp']), sorted(base['trace']), label) - seen = read(self, 'scapy', case.frames, case.linktype) - if any(seen[aspect] != base[aspect] for aspect in ASPECTS): - disagree.append(label) - # what the default engine reads, as the cases say, then that scapy reads the same - self.assertEqual(reads, {label: case.reads for label, case in cases.items()}) - self.assertEqual(disagree, []) - - def test_no_ip_is_read_at_a_link_type_the_default_engine_leaves_raw(self) -> None: - # #1650 - self.agree(cases_1650()) - - def test_nothing_is_read_past_the_frame_or_extension_header_bound(self) -> None: - # #1609 - self.agree(cases_1609()) - - def test_tcp_and_ip_are_read_behind_every_extension_header_the_default_engine_reads_past(self) -> None: - # #1626 - self.agree(cases_1626()) - - def test_the_loopback_families_do_not_depend_on_the_host(self) -> None: - # #1650: Scapy binds IPv6 to the host's own AF_INET6 (10 here), and IPv4 - # to family 0 too; bound as on a BSD instead, it dissects otherwise, but - # scapy reads the same as the default engine either way - from scapy.layers.inet import IP - from scapy.layers.inet6 import IPv6 - from scapy.layers.l2 import Loopback - - bsd = [({'type': 2}, IP), ({'type': 24}, IPv6), ({'type': 28}, IPv6)] - for linktype, order in ((0, 'little'), (0, 'big'), (108, 'big')): - for ipv4, ipv6 in ((2, 10), (2, 24), (2, 28), (2, 30), (0, 0)): - frames = each(loopback(order, ipv4, ipv6)) - with self.subTest(linktype=linktype, order=order, families=(ipv4, ipv6)): - base = read(self, 'default', frames, linktype) - here = read(self, 'scapy', frames, linktype) - with mock.patch.object(Loopback, 'payload_guess', bsd): - there = read(self, 'scapy', frames, linktype) - self.assertEqual([(here[aspect], there[aspect]) for aspect in ASPECTS], - [(base[aspect], base[aspect]) for aspect in ASPECTS]) - - def test_a_packet_built_by_hand_is_read_by_its_first_layer(self) -> None: - # ``scapy.all`` binds every layer here, as the engine has it do (see pcapkit.toolkit.scapy) - import scapy.all # noqa: F401 # pylint: disable=unused-import - from scapy.layers.inet import IP, TCP - from scapy.layers.inet6 import IPv6 - from scapy.layers.l2 import CookedLinux, Dot3, Ether, Loopback, LoopbackOpenBSD - from scapy.packet import Raw - - from pcapkit.toolkit import scapy as toolkit - - segment = v4(tcp(1), 6) - - def reads(packet: 'Any') -> bool: - return toolkit.tcp_reassembly(packet) is not None - - self.assertEqual({ - 'ethernet': reads(Ether(eth(segment))), - 'ipv4': reads(IP(segment)), - 'ipv6': reads(IPv6(v6(tcp(1), 6))), - 'null, family 24': reads(Loopback(loopback('little', 2, 24)(v6(tcp(1), 6)))), - 'null, family 10': reads(Loopback(loopback('little', 2, 10)(v6(tcp(1), 6)))), - 'null, family 0': reads(Loopback(loopback('little', 0, 0)(segment))), - 'loop, family 2': reads(LoopbackOpenBSD(loopback('big', 2, 24)(segment))), - 'loop, little-endian': reads(LoopbackOpenBSD(loopback('little', 2, 24)(segment))), - 'linux_sll': reads(CookedLinux(sll(segment))), - 'tcp with no ip': reads(Ether() / TCP()), - '802.3': reads(Dot3() / Raw(segment)), - }, { - 'ethernet': True, 'ipv4': True, 'ipv6': True, 'null, family 24': True, 'null, family 10': False, - 'null, family 0': False, 'loop, family 2': True, 'loop, little-endian': False, 'linux_sll': False, - 'tcp with no ip': False, '802.3': False, - }) - # the link type the engine attaches stands over the first layer - packet = Ether(eth(segment)) - toolkit.attach_linktype(packet, 113) - self.assertEqual([toolkit.tcp_reassembly(packet), toolkit.ipv4_reassembly(packet)], [None, None]) - - def test_the_fragment_header_behind_a_header_scapy_reads_nothing_past_is_handed_over(self) -> None: - # #1626: the input is the octets either side of the Fragment header, as the default engine cuts them - import scapy.all # noqa: F401 # pylint: disable=unused-import - from scapy.layers.l2 import Ether - - from pcapkit.toolkit import scapy as toolkit - - mobility = bytes([44, 0, 0, 0, 0, 0, 0, 0]) - frame = eth(v6(mobility + fragment6(b'B' * 8, ident=7, nxt=6, offset=16), 135)) + b'\x00' * 6 - data = toolkit.ipv6_reassembly(Ether(frame), count=1) - assert data is not None - self.assertEqual((data.bufid[2:], data.fo, data.mf, data.ihl, data.tl, bytes(data.header), bytes(data.payload)), - ((7, 6), 16, False, 48, 56, frame[14:62], b'B' * 8)) - - -if __name__ == '__main__': - unittest.main() diff --git a/tests/toolkit/test_scapy_unit.py b/tests/toolkit/test_scapy_unit.py index ae608eb05..40d6403d4 100644 --- a/tests/toolkit/test_scapy_unit.py +++ b/tests/toolkit/test_scapy_unit.py @@ -23,17 +23,32 @@ DATAGRAM = bytes(range(80)) -def write_pcap(frames: 'list[bytes]') -> 'str': - """Write ``frames`` to a temporary Ethernet PCAP file and return its path.""" +def write_pcap(frames: 'list[bytes]', linktype: 'int' = 1) -> 'str': + """Write ``frames`` to a temporary PCAP file, Ethernet unless ``linktype`` says otherwise, and return its path.""" fd, path = tempfile.mkstemp(suffix='.pcap') with os.fdopen(fd, 'wb') as file: - file.write(struct.pack(' None: @@ -362,7 +377,7 @@ class ScapyTrailerTests(unittest.TestCase): ``Padding`` layer, and ``bytes()`` of every layer above it ends with them, so an IPv6 fragment's data and a TCP segment's payload both carried the frame's trailer. Every frame here is written to a capture and read back through the - engine, and the default engine reads the same capture for comparison. + engine. """ @@ -429,16 +444,8 @@ def padded(frame: 'bytes') -> 'bytes': tso[:16] + b'\x00\x00' + tso[18:], ] - def _extract(self, frames: 'list[bytes]', engine: 'str' = 'scapy', **kwargs: 'object'): - import pcapkit - - path = write_pcap(frames) - self.addCleanup(os.remove, path) - with warnings.catch_warnings(): - warnings.simplefilter('ignore') - extractor = pcapkit.extract(fin=path, nofile=True, engine=engine, **kwargs) - self.addCleanup(close_extractor, extractor) - return extractor + def _extract(self, frames: 'list[bytes]', **kwargs: 'object'): + return extract(self, frames, **kwargs) def test_ipv6_fragment_data_stops_at_the_payload_length(self) -> None: from pcapkit.toolkit import scapy as toolkit @@ -485,88 +492,33 @@ def test_tcp_payload_stops_at_the_datagram_end(self) -> None: self.assertEqual(flow.header, header) def test_tcp_reassembly_is_the_stream_sent(self) -> None: - def streams(engine: 'str') -> 'dict[tuple[str, int], bytes]': - extractor = self._extract(self._tcp_frames(), engine=engine, store=False, - reassembly=True, tcp=True) - return {(str(datagram.id.src[0]), datagram.id.src[1]): datagram.payload - for datagram in extractor.reassembly.tcp} - - scapy = streams('scapy') + extractor = self._extract(self._tcp_frames(), store=False, reassembly=True, tcp=True) + scapy = {(str(datagram.id.src[0]), datagram.id.src[1]): datagram.payload + for datagram in extractor.reassembly.tcp} # ``b'X\x00\x00\x00\x00\x00\xef'`` before the fix: the first segment's # padding took the place of the second's data, and its trailer followed self.assertEqual(scapy.get(('192.0.2.1', 1)), b'XYZ') self.assertEqual(scapy.get(('2001:db8::1', 1)), b'six') - self.assertEqual(scapy, streams('default')) - - def test_every_adapter_agrees_with_the_default_engine(self) -> None: - from pcapkit.const.reg.linktype import LinkType - from pcapkit.toolkit import pcap as toolkit_pcap - from pcapkit.toolkit import scapy as toolkit - - frames = self._ip_frames() + self._tcp_frames() - packets = self._extract(frames, store=True).frame - defaults = self._extract(frames, engine='default', store=True).frame - self.assertEqual(len(packets), len(defaults)) - - def fields(data: 'object', names: 'tuple[str, ...]') -> 'object': - if data is None: - return None - return {name: bytes(value) if isinstance(value, (bytes, bytearray)) else value - for name, value in ((name, getattr(data, name)) for name in names)} - - ip_fields = ('fo', 'ihl', 'mf', 'tl', 'header', 'payload') - tcp_fields = ('dsn', 'ack', 'header', 'payload', 'first', 'last', 'len') - trace_fields = ('seq', 'ack', 'header', 'payload') - compared = 0 - for index, (packet, default) in enumerate(zip(packets, defaults)): - for adapter, expected, actual in ( - ('ipv4', fields(toolkit_pcap.ipv4_reassembly(default), ip_fields), - fields(toolkit.ipv4_reassembly(packet), ip_fields)), - ('ipv6', fields(toolkit_pcap.ipv6_reassembly(default), ip_fields), - fields(toolkit.ipv6_reassembly(packet), ip_fields)), - ('tcp', fields(toolkit_pcap.tcp_reassembly(default), tcp_fields), - fields(toolkit.tcp_reassembly(packet), tcp_fields)), - ('trace', fields(toolkit_pcap.tcp_traceflow(default, data_link=LinkType.ETHERNET), - trace_fields), - fields(toolkit.tcp_traceflow(packet), trace_fields)), - ): - compared += expected is not None - with self.subTest(frame=index, adapter=adapter): - self.assertEqual(actual, expected) - # every IPv4 frame here leaves DF clear, so each of the 7 makes an IPv4 - # record; then the 2 IPv6 fragments, and a TCP and a trace record for - # each TCP frame - self.assertEqual(compared, 7 + 2 + 2 * len(self.TCP_PAYLOADS)) def _assert_tcp_records(self, frames: 'list[bytes]', expected: 'list[bytes | None]') -> None: - """Each frame gives ``expected`` TCP payload, or no record, as on the default engine.""" - from pcapkit.const.reg.linktype import LinkType - from pcapkit.toolkit import pcap as toolkit_pcap + """Each frame gives ``expected`` TCP payload, or no record.""" from pcapkit.toolkit import scapy as toolkit - def octets(record: 'object') -> 'object': - return None if record is None else (bytes(record.header), bytes(record.payload)) - packets = self._extract(frames, store=True).frame - defaults = self._extract(frames, engine='default', store=True).frame self.assertEqual(len(packets), len(expected)) - for index, (packet, default, payload) in enumerate(zip(packets, defaults, expected)): + for index, (packet, payload) in enumerate(zip(packets, expected)): with self.subTest(frame=index): segment = toolkit.tcp_reassembly(packet) flow = toolkit.tcp_traceflow(packet) self.assertEqual(None if segment is None else bytes(segment.payload), payload) self.assertEqual(None if flow is None else bytes(flow.payload), payload) - self.assertEqual(octets(segment), octets(toolkit_pcap.tcp_reassembly(default))) - self.assertEqual(octets(flow), octets(toolkit_pcap.tcp_traceflow( - default, data_link=LinkType.ETHERNET))) def test_a_data_offset_below_the_fixed_header_gives_no_segment(self) -> None: from scapy.layers.inet import IP, TCP from scapy.packet import Raw - # a header shorter than 20 octets is malformed (:rfc:`9293#section-3.1`), - # and the default engine reads no segment; the payload here had started - # with the last of the header's octets + # a header shorter than 20 octets is malformed (:rfc:`9293#section-3.1`); + # the payload here had started with the last of the header's octets self._assert_tcp_records([ bytes(self._ether() / IP(src='192.0.2.1', dst='198.51.100.1') / TCP(sport=1, dport=2, flags='PA', dataofs=dataofs) / Raw(b'abcdef')) + TRAILER @@ -623,5 +575,158 @@ def test_data_with_no_length_to_stop_at_is_taken_whole(self) -> None: self.assertEqual(bytes(fragment.payload), b'hand') +@unittest.skipUnless(HAS_RUNTIME and HAS_SCAPY, 'runtime dependencies not installed') +class ScapyFrameGuardTests(unittest.TestCase): + """Frames whose Scapy layers claim more than their octets hold (GitHub issue #1708). + + Each guard here first came with a fix that #1708 reverted, as that fix's + aim was to read frames as the default engine does. They are kept because + they are the adapters' own job, whatever any other engine reads: to end no + extraction, and to hand on only the frame's own octets, as the datagram + the adapter reports them for. + + """ + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _ether(self): + from scapy.layers.l2 import Ether + + return Ether(src='aa:aa:aa:aa:aa:aa', dst='bb:bb:bb:bb:bb:bb') + + def test_a_zero_length_frame_ends_no_extraction(self) -> None: + # #1676: Scapy hands a zero-length frame back as ``IP()`` at RAW (101) + # and IPV4 (228), whose ``ihl`` is None, and IPv4 reassembly raised + # TypeError on it, which ended the extraction + from pcapkit.toolkit import scapy as toolkit + + frame = bytes.fromhex('4500001c00010000400600000a0000010a000002') + bytes(8) + for linktype in (101, 228): + with self.subTest(linktype=linktype): + extractor = extract(self, [b'', frame], linktype, store=True, reassembly=True, ipv4=True) + self.assertEqual(len(extractor.frame), 2) + self.assertIsNone(toolkit.ipv4_reassembly(extractor.frame[0])) + self.assertEqual([tuple(datagram.index) for datagram in extractor.reassembly.ipv4], [(2,)]) + + def test_a_zero_length_frame_carries_no_octets(self) -> None: + # #1676: ``bytes()`` of the layer of defaults Scapy hands a zero-length + # frame back as builds a header out of nothing, which ``packet2dict`` + # reported as the frame's octets + from pcapkit.toolkit import scapy as toolkit + + for linktype in (0, 1, 101, 108, 228, 229): + with self.subTest(linktype=linktype): + packet, = extract(self, [b''], linktype, store=True).frame + self.assertEqual(toolkit.packet2dict(packet)['packet'], b'') + record = toolkit.packet2frame(packet) + self.assertEqual((bytes(record.packet), record.frame_info.incl_len), (b'', 0)) + + def test_ipv4_in_ipv6_ends_at_the_payload_length(self) -> None: + # #1623: Scapy files what follows an IPv6 Payload Length as padding at the + # end of the packet, so ``bytes()`` of the IPv4 layer inside ended with + # the frame's trailer, and IPv4 reassembly took it for the datagram's data + from scapy.layers.inet import IP + from scapy.layers.inet6 import IPv6 + from scapy.packet import Raw + + from pcapkit.toolkit import scapy as toolkit + + ip = {'src': '192.0.2.1', 'dst': '198.51.100.1', 'proto': 17} + ipv6 = {'src': '2001:db8::1', 'dst': '2001:db8::2', 'nh': 4} + tso = bytearray(bytes(IP(**ip, id=3) / Raw(b'A' * 16))) + tso[2:4] = b'\x00\x00' # a Total Length of 0, as TCP segmentation offload leaves it + # an IPv4 header of 24 octets, its options cut short by a Payload Length of 22 + fixed = bytes(IP(**ip, id=4, flags='MF', len=40))[:20] + options = bytes([0x46]) + fixed[1:] + b'\x01' * 4 + b'B' * 16 + frames = [bytes(self._ether() / IPv6(**ipv6) / Raw(bytes(tso))) + TRAILER, + bytes(self._ether() / IPv6(**ipv6, plen=22) / Raw(options)) + TRAILER] + + extractor = extract(self, frames, store=True, reassembly=True, ipv4=True) + whole, cut = extractor.frame + data = toolkit.ipv4_reassembly(whole) + assert data is not None + self.assertEqual((bytes(data.header), bytes(data.payload)), (bytes(tso[:20]), b'A' * 16)) + self.assertIsNone(toolkit.ipv4_reassembly(cut)) + datagram, = extractor.reassembly.ipv4 + # only ``Completion.COMPLETE`` is truthy + self.assertTrue(datagram.completed) + self.assertEqual(datagram.payload, b'A' * 16) + + def test_a_packet_built_by_hand_is_read_as_the_octets_it_builds(self) -> None: + # #1676: Scapy leaves ``ihl``, ``len`` and ``dataofs`` None on a layer + # built by hand until it is built; IPv4 and TCP reassembly and TCP flow + # tracing raised TypeError on them, and the frame record that flow + # tracing hands on carried none of the packet's octets + from scapy.layers.inet import IP, TCP + from scapy.layers.inet6 import IPv6 + from scapy.packet import Raw + + from pcapkit.toolkit import scapy as toolkit + + ip = {'src': '192.0.2.1', 'dst': '198.51.100.1'} + fragment = self._ether() / IP(**ip, id=5, flags='MF') / Raw(b'x' * 8) + self.assertIsNone(fragment[IP].ihl) + data = toolkit.ipv4_reassembly(fragment) + assert data is not None + self.assertEqual((data.ihl, data.tl, bytes(data.header), bytes(data.payload)), + (20, 28, bytes(fragment)[14:34], b'x' * 8)) + + segment = TCP(sport=1, dport=2) / Raw(b'data') + for label, packet, start in ( + ('ipv4', self._ether() / IP(**ip, flags='DF') / segment, 34), + ('ipv6', self._ether() / IPv6(src='2001:db8::1', dst='2001:db8::2') / segment, 54), + # given a Total Length but no IHL + ('ipv4 of a total length', self._ether() / IP(**ip, flags='DF', len=44) / segment, 34), + ): + with self.subTest(packet=label): + octets = bytes(packet) + record = toolkit.tcp_reassembly(packet) + assert record is not None + self.assertEqual((bytes(record.header), bytes(record.payload)), (octets[start:start + 20], b'data')) + flow = toolkit.tcp_traceflow(packet) + assert flow is not None + self.assertEqual((bytes(flow.header), bytes(flow.payload)), (octets[start:start + 20], b'data')) + self.assertEqual((bytes(flow.frame.packet), flow.frame.frame_info.incl_len), (octets, len(octets))) + + def test_ipv6_reassembly_takes_only_its_own_fragment_header(self) -> None: + # The Fragment header of a packet the IPv6 layer carries, here the inner + # packet of 6in6, had been handed on with the outer packet's addresses, + # and both IPv6 headers as the part before the Fragment header + from scapy.layers.inet import IP + from scapy.layers.inet6 import IPv6, IPv6ExtHdrFragment + from scapy.layers.l2 import Ether + from scapy.packet import Raw + + from pcapkit.toolkit import scapy as toolkit + + inner = IPv6(src='fd00::1', dst='fd00::2') / IPv6ExtHdrFragment(nh=17, id=7, m=1) / Raw(b'B' * 16) + tunnelled = Ether(bytes(self._ether() / IPv6(src='2001:db8::1', dst='2001:db8::2') / inner)) + self.assertIsNone(toolkit.ipv6_reassembly(tunnelled)) + # carried in IPv4, the inner packet is the first IPv6 layer, and the Fragment header its own + data = toolkit.ipv6_reassembly(Ether(bytes(self._ether() / IP(src='192.0.2.1', dst='198.51.100.1') / inner))) + assert data is not None + self.assertEqual((str(data.bufid[0]), data.bufid[2], bytes(data.header)), ('fd00::1', 7, bytes(inner)[:40])) + + def test_no_ipv6_fragment_is_read_in_a_later_ipv4_fragment(self) -> None: + # A later IPv4 fragment's data is a slice of the datagram's payload, with + # no upper-layer header in it (:rfc:`791#section-3.2`), but Scapy dissects + # IPv6 in it whatever the fragment offset + from scapy.layers.inet import IP + from scapy.layers.inet6 import IPv6, IPv6ExtHdrFragment + from scapy.layers.l2 import Ether + from scapy.packet import Raw + + from pcapkit.toolkit import scapy as toolkit + + ip = {'src': '192.0.2.1', 'dst': '198.51.100.1', 'proto': 41} + inner = IPv6(src='fd00::1', dst='fd00::2') / IPv6ExtHdrFragment(nh=17, id=7, m=1) / Raw(b'B' * 16) + later = Ether(bytes(self._ether() / IP(**ip, frag=5) / inner)) + self.assertIsInstance(later[IP].payload, IPv6) + self.assertIsNone(toolkit.ipv6_reassembly(later)) + # the first fragment's data does start with the IPv6 header + self.assertIsNotNone(toolkit.ipv6_reassembly(Ether(bytes(self._ether() / IP(**ip, flags='MF') / inner)))) + + if __name__ == '__main__': unittest.main() diff --git a/tests/toolkit/test_scapy_zero_length_frames_1676_unit.py b/tests/toolkit/test_scapy_zero_length_frames_1676_unit.py deleted file mode 100644 index 431505166..000000000 --- a/tests/toolkit/test_scapy_zero_length_frames_1676_unit.py +++ /dev/null @@ -1,308 +0,0 @@ -# -*- coding: utf-8 -*- -"""scapy reads nothing in a zero-length frame, and a packet built by hand as the frame it builds. C.f. #1676. - -`Scapy`_ dissects a packet only from a non-empty string, so a zero-length frame -comes back as the first layer its link type names with every field left to its -default -- at ``RAW`` and ``IPV4`` an IPv4 header whose ``ihl`` and ``len`` are -:data:`None` -- and ``bytes()`` of it builds a header out of nothing. IPv4 -reassembly raised ``TypeError`` on the IHL, which ended the extraction, where -the default engine and dpkt read nothing in the frame and go on. ``packet2dict`` -reported the octets of the header it built (#1676). - -A packet built by hand is a layer of defaults too: `Scapy`_ leaves its ``ihl``, -``len`` and a TCP layer's ``dataofs`` :data:`None` until it is built, so IPv4 and -TCP reassembly and TCP flow tracing raised on one, and the frame record flow -tracing hands on carried none of its octets. It is now read as the frame it -builds, at the link type its first layer stands for. - -Each engine is compared with the default engine, on every input it hands TCP -and IP reassembly and TCP flow tracing. Every capture is built here, so the -module is in the unit tier. - -.. _Scapy: https://scapy.net - -""" -from __future__ import annotations - -import importlib.util -import os -import tempfile -import unittest -import warnings -from typing import TYPE_CHECKING, NamedTuple - -from tests._support import close_extractor, reimport_once_per_class -from tests.foundation import _roundtrip as wire -from tests.toolkit import test_dpkt_ipv4_reassembly_input_1615_1616_unit as ipv4_input -from tests.toolkit.test_scapy_ipv6_fragment_and_encap_walk_1635_1636_unit import fragment6, read -from tests.toolkit.test_tcp_carrier_walk_1606_1617_unit import eth, has, tcp, v4, v6 - -if TYPE_CHECKING: - from typing import Any, Callable - -RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') -HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) -HAS_SCAPY = importlib.util.find_spec('scapy') is not None - -#: The inputs compared, as :func:`read` keys them. -ASPECTS = ('tcp', 'trace', 'ipv4', 'ipv6', 'ipv4 datagrams', 'ipv6 datagrams', 'ids', 'flows') - -#: The engines compared with the default engine. -ENGINES = ('scapy', 'dpkt') - -#: TCP in IPv4, then the two fragments of a 24-octet IPv4 datagram. -V4 = [v4(tcp(1), 6), ipv4_input.ipv4(b'A' * 16, ident=3, mf=True), ipv4_input.ipv4(b'B' * 8, ident=3, offset=16)] -#: The two fragments of a 24-octet IPv6 datagram, then TCP in IPv6. -V6 = [v6(fragment6(b'A' * 16, ident=4, mf=True), 44), v6(fragment6(b'B' * 8, ident=4, offset=16), 44), v6(tcp(2), 6)] - - -def loopback(order: str) -> 'Callable[[bytes], bytes]': - """A BSD loopback header, ``order``-endian, of family 2 or 24 by the IP ``packet``'s version, then the packet.""" - return lambda packet: (24 if packet[0] >> 4 == 6 else 2).to_bytes(4, order) + packet - - -class LinkType(NamedTuple): - """A link type the default engine dissects, and the frames it carries IP in.""" - - #: The capture's link type. - linktype: int - #: The frame an IP packet is carried in. - carry: 'Callable[[bytes], bytes]' - #: The IP packets that follow the frame under test. - packets: 'list[bytes]' - #: How many frames the default engine hands TCP, IPv4 and IPv6 reassembly. - reads: 'tuple[int, int, int]' - - def frames(self, first: bytes) -> 'list[bytes]': - """``first``, then a frame carrying each of :attr:`packets`.""" - return [first, *map(self.carry, self.packets)] - - -LINK_TYPES = { - 'raw (101)': LinkType(101, lambda packet: packet, V4 + V6, (2, 3, 2)), - 'ipv4 (228)': LinkType(228, lambda packet: packet, V4, (1, 3, 0)), - 'ipv6 (229)': LinkType(229, lambda packet: packet, V6, (1, 0, 2)), - 'null (0), little-endian': LinkType(0, loopback('little'), V4 + V6, (2, 3, 2)), - 'null (0), big-endian': LinkType(0, loopback('big'), V4 + V6, (2, 3, 2)), - 'loop (108)': LinkType(108, loopback('big'), V4 + V6, (2, 3, 2)), - 'ethernet (1)': LinkType(1, eth, V4 + V6, (2, 3, 2)), -} - -#: Frames cut short, each before the frames of a link type in -#: :data:`LINK_TYPES`, with what the default engine reads in the cut frame, -#: TCP, IPv4 and IPv6. Controls: `Scapy`_ dissects none of them as a layer of -#: defaults, so scapy read them as the default engine does already. -CUT_SHORT = { - 'raw (101), ipv4 header cut at 19': ('raw (101)', V4[0][:19], (0, 0, 0)), - 'raw (101), ipv6 header cut at 39': ('raw (101)', V6[2][:39], (0, 0, 0)), - 'raw (101), ipv4 tcp cut at 10': ('raw (101)', V4[0][:30], (0, 1, 0)), - 'ipv4 (228), one octet': ('ipv4 (228)', V4[0][:1], (0, 0, 0)), - 'ipv6 (229), one octet': ('ipv6 (229)', V6[2][:1], (0, 0, 0)), - 'null (0), 3 octets': ('null (0), little-endian', b'\x02\x00\x00', (0, 0, 0)), - 'null (0), the header alone': ('null (0), little-endian', b'\x02\x00\x00\x00', (0, 0, 0)), - 'loop (108), the header alone': ('loop (108)', b'\x00\x00\x00\x18', (0, 0, 0)), - 'ethernet (1), one octet': ('ethernet (1)', b'\x02', (0, 0, 0)), - 'ethernet (1), the header alone': ('ethernet (1)', wire.ethernet(b'', 0x0800), (0, 0, 0)), - 'ethernet (1), an ipv4 fragment with no data': ('ethernet (1)', eth(ipv4_input.ipv4(b'', ident=9, mf=True)), - (0, 1, 0)), -} - - -def fields(packet: 'Any', skip: 'tuple[str, ...]' = ()) -> 'Any': - """``packet``'s fields as :func:`read` records them, or :data:`None`.""" - if packet is None: - return None - return {key: bytes(value) if isinstance(value, bytearray) else value - for key, value in packet.items() if key not in skip} - - -@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') -@unittest.skipUnless(HAS_SCAPY, 'scapy not installed') -class ScapyZeroLengthFrameTests(unittest.TestCase): - """scapy reads nothing in a zero-length frame, and reads on past it.""" - - def setUp(self) -> None: - reimport_once_per_class(self) - - def test_the_issue_capture_is_read_to_the_end(self) -> None: - from pcapkit import extract - - frames = [(0, 0, b''), (0, 0, bytes.fromhex('4500001c00010000400600000a0000010a000002') + bytes(8))] - read_ = {} - with tempfile.TemporaryDirectory() as tmp, warnings.catch_warnings(): - warnings.simplefilter('ignore') - # the issue's PCAP, and the same records as a PCAP-NG's EPBs - for savefile in ('pcap', 'pcapng'): - path = os.path.join(tmp, f'raw0.{savefile}') - with open(path, 'wb') as file: - file.write(getattr(wire, savefile)(frames, linktype=101)) - for engine in ('default', *(engine for engine in ENGINES if has(engine))): - extractor = extract(fin=path, engine=engine, nofile=True, reassembly=True, ipv4=True) - try: - read_[savefile, engine] = len(extractor.frame) - finally: - close_extractor(extractor) - self.assertEqual(read_, dict.fromkeys(read_, 2)) - - def test_no_input_is_read_in_a_zero_length_frame_at_any_link_type(self) -> None: - cases = {f'{label}, zero-length': (label, b'', (0, 0, 0)) for label in LINK_TYPES} - cases.update(CUT_SHORT) - reads, disagree = {}, [] - for label, (name, first, cut) in cases.items(): - link = LINK_TYPES[name] - frames = link.frames(first) - base = read(self, 'default', frames, link.linktype) - reads[label] = tuple(len(base[aspect]) for aspect in ('tcp', 'ipv4', 'ipv6')) - for engine in ENGINES: - if not has(engine): - continue - try: - seen = read(self, engine, frames, link.linktype) - except Exception as exc: # pylint: disable=broad-except - disagree.append((label, engine, f'{type(exc).__name__}: {exc}')) - continue - differ = [aspect for aspect in ASPECTS if seen[aspect] != base[aspect]] - if differ: - disagree.append((label, engine, differ)) - # what the default engine reads -- the frames after the first, and what - # it reads in that one -- then that every other engine reads the same - self.assertEqual(reads, {label: tuple(map(sum, zip(LINK_TYPES[name].reads, cut))) - for label, (name, _, cut) in cases.items()}) - self.assertEqual(disagree, []) - - def test_a_zero_length_frame_carries_no_octets(self) -> None: - from pcapkit import extract - from pcapkit.toolkit.scapy import packet2dict, packet2frame - - seen, expected = {}, {} - with tempfile.TemporaryDirectory() as tmp, warnings.catch_warnings(): - warnings.simplefilter('ignore') - for label, link in LINK_TYPES.items(): - path = os.path.join(tmp, 'zero.pcap') - with open(path, 'wb') as file: - file.write(wire.pcap([(1, 0, b'')], linktype=link.linktype)) - extractor = extract(fin=path, engine='default', nofile=True) - try: - octets = bytes(extractor.frame[0].info.packet) - expected[label] = (octets, len(octets), octets) - finally: - close_extractor(extractor) - extractor = extract(fin=path, engine='scapy', nofile=True) - try: - if extractor._exnam != 'scapy': # pylint: disable=protected-access - self.skipTest('scapy did not run') - packet = extractor.frame[0] - record = packet2frame(packet) - seen[label] = (bytes(record.packet), record.frame_info.incl_len, packet2dict(packet)['packet']) - finally: - close_extractor(extractor) - self.assertEqual(seen, expected) - self.assertEqual(set(seen.values()), {(b'', 0, b'')}) - - -@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') -@unittest.skipUnless(HAS_SCAPY, 'scapy not installed') -class ScapyHandBuiltPacketTests(unittest.TestCase): - """A packet built by hand is read as the frame it builds, at the link type its first layer stands for.""" - - def setUp(self) -> None: - reimport_once_per_class(self) - - @staticmethod - def cases() -> 'dict[str, Any]': - # ``scapy.all`` binds every layer here, as the engine has it do (see pcapkit.toolkit.scapy) - from scapy.all import IP, TCP, Ether, IPv6, IPv6ExtHdrFragment, IPv6ExtHdrHopByHop, Raw - - ether = {'src': '02:00:00:00:00:01', 'dst': '02:00:00:00:00:02'} - ip = {'src': '192.0.2.1', 'dst': '198.51.100.1'} - ip6 = {'src': '2001:db8::1', 'dst': '2001:db8::2'} - segment = TCP(sport=1, dport=2) / Raw(b'data') - return { - # IPv4 reassembly raised on the IHL - 'ipv4 with no fields': IP(), - 'ipv4 fragment': Ether(**ether) / IP(**ip, id=5, flags='MF') / Raw(b'x' * 8), - 'ipv4 fragment in ipv6': Ether(**ether) / IPv6(**ip6) / IP(**ip, id=5, flags='MF') / Raw(b'x' * 8), - # TCP reassembly and flow tracing raised on the Data Offset - 'tcp in ipv4': Ether(**ether) / IP(**ip, flags='DF') / segment, - 'tcp in ipv6': Ether(**ether) / IPv6(**ip6) / segment, - 'tcp in ipv6 behind hop-by-hop': Ether(**ether) / IPv6(**ip6) / IPv6ExtHdrHopByHop() / segment, - 'tcp in 4in4': Ether(**ether) / IP(**ip, flags='DF') / IP(**ip, flags='DF') / segment, - # ... and on the IHL of an IPv4 layer given a Total Length alone - 'tcp in ipv4 of a total length': Ether(**ether) / IP(**ip, flags='DF', len=44) / segment, - # read off the octets already, but traced with a frame record of none - 'tcp in ipv4 behind hop-by-hop': Ether(**ether) / IP(**ip, flags='DF', proto=0) - / IPv6ExtHdrHopByHop(nh=6) / segment, - # controls: read off the octets already - 'ipv6 fragment': Ether(**ether) / IPv6(**ip6) / IPv6ExtHdrFragment(id=1, m=1) / Raw(b'hand'), - 'ipv6 fragment in ipv4': Ether(**ether) / IP(**ip, flags='DF') / IPv6(**ip6) - / IPv6ExtHdrFragment(id=1, m=1) / Raw(b'hand'), - } - - def test_a_packet_built_by_hand_is_read_as_the_frame_it_builds(self) -> None: - from scapy.layers.l2 import Ether - - from pcapkit.toolkit import scapy as toolkit - - seen, expected = {}, {} - for label, packet in self.cases().items(): - packet.time = 1 # as :func:`read` dates the frame - frame = bytes(packet) - ethernet = isinstance(packet, Ether) - base = read(self, 'default', [frame], 1 if ethernet else 228) - # an IP-rooted packet names no link type to trace it at (see below) - aspects = ('ipv4', 'ipv6', 'tcp', 'trace') if ethernet else ('ipv4', 'ipv6', 'tcp') - expected[label] = {aspect: base[aspect].get(1) for aspect in aspects} - if expected[label].get('trace') is not None: - # the frame record carries the octets the input was read from - expected[label]['frame'] = (frame, len(frame)) - got = {} # type: dict[str, Any] - for aspect in aspects: - adapter = toolkit.tcp_traceflow if aspect == 'trace' else getattr(toolkit, f'{aspect}_reassembly') - try: - with warnings.catch_warnings(): - # as :func:`read` has the default engine do - warnings.simplefilter('ignore') - result = adapter(packet, count=1) - except Exception as exc: # pylint: disable=broad-except - got[aspect] = f'{type(exc).__name__}: {exc}' - continue - got[aspect] = fields(result, ('frame',) if aspect == 'trace' else ()) - if aspect == 'trace' and result is not None: - got['frame'] = (bytes(result.frame.packet), result.frame.frame_info.incl_len) - seen[label] = got - self.assertEqual(seen, expected) - # what the default engine reads in each - self.assertEqual({label: tuple(aspect for aspect, value in inputs.items() if value is not None) - for label, inputs in expected.items()}, { - 'ipv4 with no fields': ('ipv4',), - 'ipv4 fragment': ('ipv4',), - 'ipv4 fragment in ipv6': ('ipv4',), - 'tcp in ipv4': ('tcp', 'trace', 'frame'), - 'tcp in ipv6': ('tcp', 'trace', 'frame'), - 'tcp in ipv6 behind hop-by-hop': ('tcp', 'trace', 'frame'), - 'tcp in 4in4': ('tcp', 'trace', 'frame'), - 'tcp in ipv4 of a total length': ('tcp', 'trace', 'frame'), - 'tcp in ipv4 behind hop-by-hop': ('tcp', 'trace', 'frame'), - 'ipv6 fragment': ('ipv6',), - 'ipv6 fragment in ipv4': ('ipv6',), - }) - - def test_an_ip_rooted_packet_built_by_hand_is_traced_by_its_name(self) -> None: - # with no link type attached, the first layer's name is looked up, and - # ``IP`` names none, so it raises as documented (#775), not on the Data - # Offset Scapy left unset - from scapy.all import IP, TCP, Raw - - from pcapkit.toolkit.scapy import tcp_reassembly, tcp_traceflow - from pcapkit.utilities.exceptions import MissingKeyError - - packet = IP(src='192.0.2.1', dst='198.51.100.1') / TCP(sport=1, dport=2) / Raw(b'data') - data = tcp_reassembly(packet, count=1) - assert data is not None - self.assertEqual((bytes(data.header), bytes(data.payload)), (bytes(packet)[20:40], b'data')) - with self.assertRaises(MissingKeyError): - tcp_traceflow(packet, count=1) - - -if __name__ == '__main__': - unittest.main()