From d2841387ca95a53a41a47eef68381d3e88fa0852 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Mon, 26 Oct 2026 16:32:35 -0400 Subject: [PATCH] fix(toolkit): reassemble the first fragmented IP layer, so tunnelled fragments reassemble (#2307) Problem: the IPv4 and IPv6 reassembly adapters read the frame's outermost IPv4 or IPv6 layer, so in an IP-in-IP tunnel a fragmented inner datagram was reassembled whole under the tunnel's header, or dropped (tunnel DF set, or no Fragment header in the tunnel's IPv6), and never reassembled itself. Fix: take the first layer of that IP version, from the outside in, that is a fragment (IPv4 MF or a non-zero offset; an IPv6 Fragment header), else the outermost as before. Applied to the default engine's PCAP and PCAP-NG adapters, and to scapy and dpkt, which picked the outermost too. Test: tests/toolkit/test_tunnelled_ip_reassembly_2307_unit.py reassembles a fragmented inner datagram (4in4 with DF clear and set, 6in6) and a fragmented tunnel carrying a whole one, per engine; the inner cases fail on main. The scapy 6in6 pin now expects the inner packet reassembled on its own header. --- docs/source/pcapkit/toolkit/3rdparty.rst | 6 + docs/source/pcapkit/toolkit/builtin.rst | 2 + pcapkit/toolkit/dpkt.py | 63 +++++- pcapkit/toolkit/pcap.py | 51 ++++- pcapkit/toolkit/pcapng.py | 19 +- pcapkit/toolkit/scapy.py | 85 ++++++-- .../test_rfc_constants_one_home_1739_unit.py | 4 + tests/toolkit/test_scapy_unit.py | 15 +- .../test_tunnelled_ip_reassembly_2307_unit.py | 204 ++++++++++++++++++ 9 files changed, 401 insertions(+), 48 deletions(-) create mode 100644 tests/toolkit/test_tunnelled_ip_reassembly_2307_unit.py diff --git a/docs/source/pcapkit/toolkit/3rdparty.rst b/docs/source/pcapkit/toolkit/3rdparty.rst index d4ef051373..797ca9ce04 100644 --- a/docs/source/pcapkit/toolkit/3rdparty.rst +++ b/docs/source/pcapkit/toolkit/3rdparty.rst @@ -46,6 +46,10 @@ Auxiliary Functions .. autofunction:: pcapkit.toolkit.scapy._within_datagrams +.. autofunction:: pcapkit.toolkit.scapy._fragment + +.. autofunction:: pcapkit.toolkit.scapy._fragment_header + .. autofunction:: pcapkit.toolkit.scapy._tcp_carrier .. autofunction:: pcapkit.toolkit.scapy._tcp_segment @@ -91,6 +95,8 @@ Auxiliary Functions .. autofunction:: pcapkit.toolkit.dpkt._capture_time +.. autofunction:: pcapkit.toolkit.dpkt._fragment + Auxiliary Data -------------- diff --git a/docs/source/pcapkit/toolkit/builtin.rst b/docs/source/pcapkit/toolkit/builtin.rst index 57e3520eb5..086262ace0 100644 --- a/docs/source/pcapkit/toolkit/builtin.rst +++ b/docs/source/pcapkit/toolkit/builtin.rst @@ -27,6 +27,8 @@ Auxiliary Functions .. autofunction:: pcapkit.toolkit.pcap.tcp_segment +.. autofunction:: pcapkit.toolkit.pcap._fragment + .. autoclass:: pcapkit.toolkit.pcap.TCPSegment :members: :show-inheritance: diff --git a/pcapkit/toolkit/dpkt.py b/pcapkit/toolkit/dpkt.py index 3a14808f37..272f8adf47 100644 --- a/pcapkit/toolkit/dpkt.py +++ b/pcapkit/toolkit/dpkt.py @@ -529,13 +529,56 @@ def wrapper(packet: 'Packet') -> 'dict[str, Any]': } +def _fragment(packet: 'Packet', version: 'int') -> 'Optional[IP | IP6]': + """Fetch the IPv4 or IPv6 layer of a DPKT packet that IP reassembly reads. + + Args: + packet: DPKT packet, i.e. the outermost layer. + version: ``4`` or ``6``, the IP version of the layer to fetch. + + Returns: + The first layer of that version, from the network layer in, that is a + fragment -- for IPv4, one with **MF** set or a non-zero fragment + offset, for IPv6, one with an IPv6 Fragment header; with none, the + network layer if it is of that version; otherwise :data:`None`. + + Note: + As with :func:`pcapkit.toolkit.pcap._fragment`, either datagram of an + IP-in-IP tunnel may be the fragmented one, and each is reassembled on + its own header (:issue:`2307`). An IPv6 layer is the one with extension + headers. The walk goes on past an IP layer only where its payload + protocol is IPv4 or IPv6, i.e. a tunnel, so a datagram an ICMP error + quotes, which `DPKT`_ decodes as an IP layer below the ICMP one, is not + walked to. + + """ + import dpkt # isort:skip + + network = _network(packet) + layer = network # type: Any + while isinstance(layer, (dpkt.ip.IP, dpkt.ip6.IP6)): + if isinstance(layer, dpkt.ip6.IP6): + if version == 6 and layer.extension_hdrs.get(44) is not None: + return layer + elif version == 4 and (layer.mf or layer.offset): + return layer + # ``IP6.p`` is the protocol after the extension headers + if getattr(layer, 'p', None) not in (Enum_TransType.IPv4, Enum_TransType.IPv6): + break + layer = layer.data + if network is None or hasattr(network, 'extension_hdrs') is not (version == 6): + return None + return network + + def ipv4_reassembly(packet: 'Packet', timestamp: 'float', *, count: 'int' = -1) -> 'IP_Packet[IPv4Address] | None': """Make data for IPv4 reassembly. - A packet can be reassembled if its network layer is IPv4 (:class:`dpkt.ip.IP`) -- - one carried in IPv6 is not read -- whose header was captured whole, and the **DF** - (:attr:`dpkt.ip.IP.df`) flag is :data:`False`. + A packet can be reassembled if it has an IPv4 layer (:class:`dpkt.ip.IP`) whose + header was captured whole, and the **DF** (:attr:`dpkt.ip.IP.df`) flag is + :data:`False`. The layer is the first that is a fragment, or with none, the network + layer if it is IPv4 (see :func:`_fragment`). Args: packet: DPKT packet. @@ -581,9 +624,7 @@ def ipv4_reassembly(packet: 'Packet', timestamp: 'float', *, with ``packet`` alone fails. """ - network = _network(packet) - # an IPv6 layer is the one with extension headers, as in ``_wire_hdr_len`` - ipv4 = None if hasattr(network, 'extension_hdrs') else network + ipv4 = cast('Optional[IP]', _fragment(packet, 4)) if ipv4 is not None: if ipv4.df: # dismiss not fragmented packet return None @@ -628,9 +669,10 @@ def ipv6_reassembly(packet: 'Packet', timestamp: 'float', *, count: 'int' = -1) -> 'IP_Packet[IPv6Address] | None': """Make data for IPv6 reassembly. - A packet can be reassembled if its network layer is IPv6 (:class:`dpkt.ip6.IP6`) -- - one carried in IPv4 is not read -- and has an IPv6 Fragment header - (:rfc:`2460#section-4.5`, i.e., :class:`dpkt.ip6.IP6FragmentHeader`). + A packet can be reassembled if it has an IPv6 layer (:class:`dpkt.ip6.IP6`) with an + IPv6 Fragment header (:rfc:`2460#section-4.5`, i.e., + :class:`dpkt.ip6.IP6FragmentHeader`). The layer is the first that has one (see + :func:`_fragment`). Args: packet: DPKT packet. @@ -663,8 +705,7 @@ def ipv6_reassembly(packet: 'Packet', timestamp: 'float', *, with ``packet`` alone fails. """ - network = _network(packet) - ipv6 = cast('IP6', network) if hasattr(network, 'extension_hdrs') else None + ipv6 = cast('Optional[IP6]', _fragment(packet, 6)) if ipv6 is not None: # dpkt keeps each extension header under its protocol number, so 44 is a # Fragment header; the dict's value type cannot say so diff --git a/pcapkit/toolkit/pcap.py b/pcapkit/toolkit/pcap.py index e11ada291c..a43fcd5633 100644 --- a/pcapkit/toolkit/pcap.py +++ b/pcapkit/toolkit/pcap.py @@ -21,6 +21,7 @@ from pcapkit.foundation.traceflow.data.tcp import Packet as TF_TCP_Packet from pcapkit.protocols.internet.ip import IP as Protocol_IP from pcapkit.protocols.internet.ipv4 import IPv4 +from pcapkit.protocols.internet.ipv6 import IPv6 from pcapkit.protocols.misc.null import NoPayload from pcapkit.protocols.misc.raw import Raw from pcapkit.protocols.protocol import ProtocolBase @@ -34,7 +35,6 @@ from pcapkit.const.reg.linktype import LinkType from pcapkit.protocols.data.internet.ipv4 import IPv4 as Data_IPv4 from pcapkit.protocols.data.internet.ipv6 import IPv6 as Data_IPv6 - from pcapkit.protocols.internet.ipv6 import IPv6 from pcapkit.protocols.internet.ipv6_frag import IPv6_Frag from pcapkit.protocols.misc.pcap import Frame @@ -193,13 +193,51 @@ def tcp_segment(frame: 'ProtocolBase') -> 'Optional[TCPSegment]': ) +def _fragment(frame: 'ProtocolBase', version: 'type[IPv4 | IPv6]') -> 'Optional[IPv4 | IPv6]': + """Fetch the IPv4 or IPv6 layer of a frame that IP reassembly reads. + + Args: + frame: PCAP frame or PCAP-NG block. + version: :class:`~pcapkit.protocols.internet.ipv4.IPv4` or + :class:`~pcapkit.protocols.internet.ipv6.IPv6`, the layer to fetch. + + Returns: + The first ``version`` layer, from the outside in, that is a fragment -- + for IPv4, one with **MF** set or a non-zero Fragment Offset, for IPv6, + one with an IPv6 Fragment header; with none, the outermost ``version`` + layer; with no ``version`` layer, :data:`None`. + + Note: + In an IP-in-IP tunnel (:rfc:`2003`, :rfc:`2473`, :rfc:`4213`) either + datagram may be the fragmented one: the inner, each fragment of it + carried whole by the tunnel, or the tunnel's own, whose first fragment + alone carries the inner header. Each is reassembled on its own header + (:issue:`2307`). A frame of no tunnel has one layer to choose from. + + """ + if version.__name__ not in frame: + return None + outermost = cast('IPv4 | IPv6', frame[version.__name__]) + layer = outermost # type: ProtocolBase + while isinstance(layer, ProtocolBase) and not isinstance(layer, NoPayload): + if isinstance(layer, IPv4) and version is IPv4: + if layer.info.flags.mf or layer.info.offset: + return layer + elif isinstance(layer, IPv6) and version is IPv6: + if Enum_ExtensionHeader.IPv6_Frag in layer.extension_headers: + return layer + layer = layer.payload + return outermost + + def ipv4_reassembly(frame: 'Frame') -> 'IP_Packet[IPv4Address] | None': """Make data for IPv4 reassembly. A frame can be reassembled if it contains an IPv4 layer (:class:`~pcapkit.protocols.internet.ipv4.IPv4`) and the **DF** (:attr:`IPv4.flags.df `) flag is - :data:`False`. + :data:`False`. In an IP-in-IP tunnel, the layer is the first that is a fragment + (see :func:`_fragment`). Args: frame: PCAP frame. @@ -212,8 +250,7 @@ def ipv4_reassembly(frame: 'Frame') -> 'IP_Packet[IPv4Address] | None': :class:`pcapkit.foundation.reassembly.ipv4.IPv4` """ - if 'IPv4' in frame: - ipv4 = cast('IPv4', frame['IPv4']) + if (ipv4 := cast('Optional[IPv4]', _fragment(frame, IPv4))) is not None: ipv4_info = ipv4.info if ipv4_info.flags.df: # dismiss not fragmented frame return None @@ -244,7 +281,8 @@ def ipv6_reassembly(frame: 'Frame') -> 'IP_Packet[IPv6Address] | None': A frame can be reassembled if it contains an IPv6 layer (:class:`~pcapkit.protocols.internet.ipv6.IPv6`) and an IPv6 Fragment header (:rfc:`2460#section-4.5`, i.e., - :class:`~pcapkit.protocols.internet.ipv6_frag.IPv6_Frag`). + :class:`~pcapkit.protocols.internet.ipv6_frag.IPv6_Frag`). In an IP-in-IP tunnel, + the layer is the first that has one (see :func:`_fragment`). Args: frame: PCAP frame. @@ -257,8 +295,7 @@ def ipv6_reassembly(frame: 'Frame') -> 'IP_Packet[IPv6Address] | None': :class:`pcapkit.foundation.reassembly.ipv6.IPv6` """ - if 'IPv6' in frame: - ipv6 = cast('IPv6', frame['IPv6']) + if (ipv6 := cast('Optional[IPv6]', _fragment(frame, IPv6))) is not None: ipv6_info = ipv6.info if (ipv6_frag := ipv6.extension_headers.get( # type: ignore[call-overload] Enum_ExtensionHeader.IPv6_Frag diff --git a/pcapkit/toolkit/pcapng.py b/pcapkit/toolkit/pcapng.py index 2e6eff13a1..3523f535b0 100644 --- a/pcapkit/toolkit/pcapng.py +++ b/pcapkit/toolkit/pcapng.py @@ -18,14 +18,15 @@ from pcapkit.foundation.traceflow.data.tcp import Packet as TF_TCP_Packet from pcapkit.protocols.data.misc.pcap.frame import Frame as Data_Frame from pcapkit.protocols.data.misc.pcap.frame import FrameInfo as Data_FrameInfo +from pcapkit.protocols.internet.ipv4 import IPv4 +from pcapkit.protocols.internet.ipv6 import IPv6 from pcapkit.protocols.schema.misc.pcap.header import RESOLUTION -from pcapkit.toolkit.pcap import tcp_segment +from pcapkit.toolkit.pcap import _fragment, tcp_segment if TYPE_CHECKING: from ipaddress import IPv4Address, IPv6Address + from typing import Optional - from pcapkit.protocols.internet.ipv4 import IPv4 - from pcapkit.protocols.internet.ipv6 import IPv6 from pcapkit.protocols.internet.ipv6_frag import IPv6_Frag from pcapkit.protocols.misc.pcapng import PCAPNG, Packet @@ -39,7 +40,8 @@ def ipv4_reassembly(frame: 'PCAPNG') -> 'IP_Packet[IPv4Address] | None': A frame can be reassembled if it contains an IPv4 layer (:class:`~pcapkit.protocols.internet.ipv4.IPv4`) and the **DF** (:attr:`IPv4.flags.df `) flag is - :data:`False`. + :data:`False`. In an IP-in-IP tunnel, the layer is the first that is a fragment + (see :func:`pcapkit.toolkit.pcap._fragment`). Args: frame: PCAP-NG frame. @@ -52,8 +54,7 @@ def ipv4_reassembly(frame: 'PCAPNG') -> 'IP_Packet[IPv4Address] | None': :class:`pcapkit.foundation.reassembly.ipv4.IPv4` """ - if 'IPv4' in frame: - ipv4 = cast('IPv4', frame['IPv4']) + if (ipv4 := cast('Optional[IPv4]', _fragment(frame, IPv4))) is not None: ipv4_info = ipv4.info if ipv4_info.flags.df: # dismiss not fragmented frame return None @@ -85,7 +86,8 @@ def ipv6_reassembly(frame: 'PCAPNG') -> 'IP_Packet[IPv6Address] | None': A frame can be reassembled if it contains an IPv6 layer (:class:`~pcapkit.protocols.internet.ipv6.IPv6`) and an IPv6 Fragment header (:rfc:`2460#section-4.5`, i.e., - :class:`~pcapkit.protocols.internet.ipv6_frag.IPv6_Frag`). + :class:`~pcapkit.protocols.internet.ipv6_frag.IPv6_Frag`). In an IP-in-IP tunnel, + the layer is the first that has one (see :func:`pcapkit.toolkit.pcap._fragment`). Args: frame: PCAP-NG frame. @@ -98,8 +100,7 @@ def ipv6_reassembly(frame: 'PCAPNG') -> 'IP_Packet[IPv6Address] | None': :class:`pcapkit.foundation.reassembly.ipv6.IPv6` """ - if 'IPv6' in frame: - ipv6 = cast('IPv6', frame['IPv6']) + if (ipv6 := cast('Optional[IPv6]', _fragment(frame, IPv6))) is not None: ipv6_info = ipv6.info if (ipv6_frag := ipv6.extension_headers.get( # type: ignore[call-overload] Enum_ExtensionHeader.IPv6_Frag diff --git a/pcapkit/toolkit/scapy.py b/pcapkit/toolkit/scapy.py index 40a592cf40..4367ff6c58 100644 --- a/pcapkit/toolkit/scapy.py +++ b/pcapkit/toolkit/scapy.py @@ -67,7 +67,7 @@ from typing import Any from scapy.layers.inet import IP, TCP - from scapy.layers.inet6 import IPv6 + from scapy.layers.inet6 import IPv6, IPv6ExtHdrFragment from scapy.packet import Packet __all__ = [ @@ -449,6 +449,68 @@ def _tcp_carrier(packet: 'Packet') -> 'tuple[IP | IPv6, TCP, bytes, bytes] | Non return carrier, tcp, *segment +def _fragment_header(ipv6: 'IPv6') -> 'IPv6ExtHdrFragment | None': + """Fetch the IPv6 Fragment header of an IPv6 layer, if any. + + Args: + ipv6: Scapy IPv6 layer. + + Returns: + The IPv6 Fragment header (:class:`scapy.layers.inet6.IPv6ExtHdrFragment`) + among the extension headers `Scapy`_ dissects in that layer's own + chain, or :data:`None` if there is none. + + """ + from scapy.layers.inet6 import IPv6ExtHdrFragment, _IPv6ExtHdr + + header = ipv6.payload + while isinstance(header, _IPv6ExtHdr) and not isinstance(header, IPv6ExtHdrFragment): + header = header.payload + return header if isinstance(header, IPv6ExtHdrFragment) else None + + +def _fragment(packet: 'Packet', name: 'str') -> 'IP | IPv6 | None': + """Fetch the IPv4 or IPv6 layer of a packet that IP reassembly reads. + + Args: + packet: Scapy packet. + name: ``'IP'`` or ``'IPv6'``, the layer to fetch. + + Returns: + The first ``name`` layer, from the outside in, that is a fragment -- + for IPv4, one with **MF** set or a non-zero fragment offset, for IPv6, + one with an IPv6 Fragment header (see :func:`_fragment_header`); with + none, ``packet[name]``, the outermost; with no ``name`` layer, + :data:`None`. + + Note: + As with :func:`pcapkit.toolkit.pcap._fragment`, either datagram of an + IP-in-IP tunnel may be the fragmented one, and each is reassembled on + its own header (:issue:`2307`). The datagram an ICMP error quotes is + of another class, so is not ``packet[name]`` and is not walked to. + + """ + from scapy.layers.inet import IP + from scapy.layers.inet6 import IPv6 + from scapy.packet import NoPayload + + if name not in packet: + return None + outermost = cast('IP | IPv6', packet[name]) + kind = IP if name == 'IP' else IPv6 + layer = outermost # type: Packet + while not isinstance(layer, NoPayload): + # an exact match, as ``packet[name]`` is, so a quoted datagram is not one + if type(layer) is kind: # pylint: disable=unidiomatic-typecheck + if kind is IP: + if layer.flags.MF or layer.frag: + return layer + elif _fragment_header(cast('IPv6', layer)) is not None: + return layer + layer = layer.payload + return outermost + + def ipv4_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv4Address] | None': """Make data for IPv4 reassembly. @@ -458,7 +520,8 @@ def ipv4_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv4Ad :func:`_within_datagrams`), and the **DF** (``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`). + :func:`_frame`). In an IP-in-IP tunnel, the layer is the first that is a + fragment (see :func:`_fragment`). Args: packet: Scapy packet. @@ -472,8 +535,7 @@ def ipv4_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv4Ad :class:`pcapkit.foundation.reassembly.ipv4.IPv4` """ - if 'IP' in packet: - ipv4 = cast('IP', packet['IP']) + if (ipv4 := cast('IP | None', _fragment(packet, 'IP'))) is not None: if ipv4.flags.DF: # dismiss not fragmented packet return None if not _frame(packet): # a zero-length frame (#1676) @@ -527,7 +589,8 @@ def ipv6_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv6Ad (:class:`scapy.layers.inet6.IPv6`), in no later IPv4 fragment, and an 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. + dissects in that layer's own chain. In an IP-in-IP tunnel, the layer is the first + that has one (see :func:`_fragment`). Args: packet: Scapy packet. @@ -548,26 +611,20 @@ def ipv6_reassembly(packet: 'Packet', *, count: 'int' = -1) -> 'IP_Packet[IPv6Ad 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. + follows. """ 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 'IPv6' in packet: - ipv6 = cast('IPv6', packet['IPv6']) + if (ipv6 := cast('IPv6 | None', _fragment(packet, 'IPv6'))) is not None: 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): + if (ipv6_frag := _fragment_header(ipv6)) is None: return None # dismiss not fragmented packet # NOTE: ``len()`` of a Scapy layer spans that layer and everything after diff --git a/tests/project/test_rfc_constants_one_home_1739_unit.py b/tests/project/test_rfc_constants_one_home_1739_unit.py index b735d25565..5dfa4bd9dd 100644 --- a/tests/project/test_rfc_constants_one_home_1739_unit.py +++ b/tests/project/test_rfc_constants_one_home_1739_unit.py @@ -324,6 +324,10 @@ 'the 8-octet value of an if_tsoffset option', ('pcapkit/foundation/reassembly/tcp.py', '1 << 24'): 'a bit width, not a PCAP global header', + ('pcapkit/toolkit/dpkt.py', '_fragment(6)'): + 'the IP version number', + ('pcapkit/toolkit/dpkt.py', 'version == 6'): + 'the IP version number', ('pcapkit/toolkit/pcap.py', 'octets[8:12]'): "a TCP header's Acknowledgment Number, octets 8 to 12", ('pcapkit/toolkit/pypcapfile.py', 'len(raw) == 6'): diff --git a/tests/toolkit/test_scapy_unit.py b/tests/toolkit/test_scapy_unit.py index 4b97da2f26..ebed3d4b11 100644 --- a/tests/toolkit/test_scapy_unit.py +++ b/tests/toolkit/test_scapy_unit.py @@ -697,7 +697,8 @@ def test_a_packet_built_by_hand_is_read_as_the_octets_it_builds(self) -> None: 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, is handed on without the fix with the outer packet's - # addresses, and both IPv6 headers as the part before the Fragment header + # addresses, and both IPv6 headers as the part before the Fragment header; + # the inner packet is the one reassembled, on its own header (#2307) from scapy.layers.inet import IP from scapy.layers.inet6 import IPv6, IPv6ExtHdrFragment from scapy.layers.l2 import Ether @@ -706,12 +707,12 @@ def test_ipv6_reassembly_takes_only_its_own_fragment_header(self) -> None: 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])) + for carrier in (IPv6(src='2001:db8::1', dst='2001:db8::2'), IP(src='192.0.2.1', dst='198.51.100.1')): + with self.subTest(carrier=carrier.name): + data = toolkit.ipv6_reassembly(Ether(bytes(self._ether() / carrier / 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 diff --git a/tests/toolkit/test_tunnelled_ip_reassembly_2307_unit.py b/tests/toolkit/test_tunnelled_ip_reassembly_2307_unit.py new file mode 100644 index 0000000000..3245b8d38f --- /dev/null +++ b/tests/toolkit/test_tunnelled_ip_reassembly_2307_unit.py @@ -0,0 +1,204 @@ +# -*- coding: utf-8 -*- +"""IP reassembly in an IP-in-IP tunnel reads the layer that is a fragment. C.f. #2307. + +The IPv4 and IPv6 reassembly adapters read the frame's *outermost* IPv4 or +IPv6 layer. In a tunnel whose inner datagram was fragmented, each fragment was +reassembled whole under the tunnel's header, or, with the tunnel's DF set, or +with no Fragment header in the tunnel's IPv6, dropped; the inner datagram was +never reassembled. Each adapter takes the first layer of its IP version, from +the outside in, that is a fragment -- for IPv4, one with MF set or a non-zero +Fragment Offset, for IPv6, one with a Fragment header -- and with none, the +outermost as before. So a tunnel that is itself fragmented, its first fragment +alone carrying the inner header, is reassembled on the tunnel's header still. + +Every capture is built here, so the module is in the unit tier. The +``pypcapfile`` engine decodes no tunnel, so has no counterpart. + +""" +from __future__ import annotations + +import importlib.util +import ipaddress +import os +import tempfile +import unittest +import warnings +from typing import TYPE_CHECKING + +from tests._support import close_extractor, reimport_once_per_class +from tests.foundation import _roundtrip as wire + +if TYPE_CHECKING: + from typing import Any + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + +#: The tunnel's endpoints, by IP version. +OUTER = {4: ('192.0.2.1', '198.51.100.2'), 6: ('2001:db8:ffff::1', '2001:db8:ffff::2')} +#: The endpoints of the datagram carried through it, by IP version. +INNER = {4: ('10.0.0.1', '10.0.0.2'), 6: ('fd00::1', 'fd00::2')} +#: The IP protocol number that carries each IP version in a tunnel. +TUNNEL = {4: 4, 6: 41} +#: The UDP datagram the inner IP datagram carries, 40 octets long. +DATAGRAM = wire.udp(bytes(range(32))) +#: Identification of the inner datagram, and of the tunnel's datagram. +INNER_ID, OUTER_ID = 7, 100 +#: Where each datagram is cut into two fragments, in octets. +CUT = 16 + + +def addressed(packet: bytes, src: str, dst: str) -> bytes: + """``packet``, an IPv4 or IPv6 one, from ``src`` to ``dst``.""" + octets = bytearray(packet) + if octets[0] >> 4 == 4: + octets[12:20] = ipaddress.ip_address(src).packed + ipaddress.ip_address(dst).packed + octets[10:12] = b'\x00\x00' + octets[10:12] = wire.checksum(bytes(octets[:20])).to_bytes(2, 'big') + else: + octets[8:40] = ipaddress.ip_address(src).packed + ipaddress.ip_address(dst).packed + return bytes(octets) + + +def whole(version: int, payload: bytes, ends: 'tuple[str, str]', proto: int, *, + ident: int = 0, df: bool = False) -> bytes: + """An unfragmented IPv4 or IPv6 packet carrying ``payload`` as ``proto``.""" + if version == 4: + return addressed(wire.ipv4(payload, ident=ident, df=df, proto=proto), *ends) + return addressed(wire.ipv6(payload, nxt=proto), *ends) + + +def fragments(version: int, payload: bytes, ends: 'tuple[str, str]', proto: int, + ident: int) -> 'list[bytes]': + """The two fragments of an IPv4 or IPv6 datagram carrying ``payload`` as ``proto``.""" + pieces = ((0, payload[:CUT], True), (CUT, payload[CUT:], False)) + if version == 4: + return [addressed(wire.ipv4(piece, ident=ident, offset=offset, mf=mf, proto=proto), *ends) + for offset, piece, mf in pieces] + return [addressed(wire.ipv6_fragment(piece, ident=ident, offset=offset, mf=mf, nxt=proto), *ends) + for offset, piece, mf in pieces] + + +def frame(packet: bytes) -> bytes: + """An Ethernet frame carrying the IPv4 or IPv6 ``packet``.""" + return wire.ethernet(packet, 0x0800 if packet[0] >> 4 == 4 else 0x86DD) + + +def inner_fragmented(version: int, *, df: bool = False) -> 'list[bytes]': + """A whole tunnel packet per fragment of the inner datagram, one tunnel id each. + + ``df`` sets the DF flag of an IPv4 tunnel's header. + + """ + return [frame(whole(version, piece, OUTER[version], TUNNEL[version], ident=OUTER_ID + index, df=df)) + for index, piece in enumerate(fragments(version, DATAGRAM, INNER[version], 17, INNER_ID))] + + +def outer_fragmented(version: int) -> 'tuple[list[bytes], bytes]': + """The fragments of a tunnel packet carrying the whole inner datagram, and that datagram.""" + inner = whole(version, DATAGRAM, INNER[version], 17, ident=INNER_ID) + return [frame(piece) for piece in fragments(version, inner, OUTER[version], + TUNNEL[version], OUTER_ID)], inner + + +def has(engine: str) -> bool: + """Whether ``engine``'s third-party package is installed.""" + return engine == 'default' or importlib.util.find_spec(engine) is not None + + +def reassemble(test: unittest.TestCase, engine: str, version: int, frames: 'list[bytes]', *, + pcapng: bool = False) -> 'list[Any]': + """The datagrams ``engine`` reassembles out of ``frames`` with IPv4 or IPv6 reassembly.""" + from pcapkit import extract + + records = [(1, number, octets) for number, octets in enumerate(frames)] + with tempfile.TemporaryDirectory() as tmp, warnings.catch_warnings(): + path = os.path.join(tmp, 'tunnel.pcapng' if pcapng else 'tunnel.pcap') + with open(path, 'wb') as file: + file.write(wire.pcapng(records) if pcapng else wire.pcap(records)) + warnings.simplefilter('ignore') + extractor = extract(fin=path, nofile=True, engine=engine, reassembly=True, + ipv4=version == 4, ipv6=version == 6) + try: + if extractor._exnam != engine: # pylint: disable=protected-access + test.skipTest(f'{engine} did not run') + return list(extractor.reassembly.ipv4 if version == 4 else extractor.reassembly.ipv6) + finally: + close_extractor(extractor) + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class TunnelledIPReassemblyTests(unittest.TestCase): + """The default engine, scapy and dpkt reassemble the IP layer that is a fragment.""" + + ENGINES = ('default', 'scapy', 'dpkt') + + def setUp(self) -> None: + reimport_once_per_class(self) + + def cases(self) -> 'Any': + """Each engine, and for the default engine each format, as a sub-test.""" + for engine in self.ENGINES: + for pcapng in ((False, True) if engine == 'default' else (False,)): + yield engine, pcapng + + def test_fragmented_inner_datagram_is_reassembled(self) -> None: + from pcapkit.foundation.reassembly.data.data import Completion + + for engine, pcapng in self.cases(): + for version, df in ((4, False), (4, True), (6, False)): + with self.subTest(engine=engine, pcapng=pcapng, version=version, df=df): + if not has(engine): + self.skipTest(f'{engine} not installed') + frames = inner_fragmented(version, df=df) + datagram, = reassemble(self, engine, version, frames, pcapng=pcapng) + self.assertIs(datagram.completed, Completion.COMPLETE) + self.assertEqual((str(datagram.id.src), str(datagram.id.dst), datagram.id.id), + (*INNER[version], INNER_ID)) + self.assertEqual(datagram.index, (1, 2)) + self.assertEqual(datagram.payload, DATAGRAM) + + def test_fragmented_tunnel_is_reassembled_on_its_own_header(self) -> None: + from pcapkit.foundation.reassembly.data.data import Completion + + for engine, pcapng in self.cases(): + for version in (4, 6): + with self.subTest(engine=engine, pcapng=pcapng, version=version): + if not has(engine): + self.skipTest(f'{engine} not installed') + frames, inner = outer_fragmented(version) + datagram, = reassemble(self, engine, version, frames, pcapng=pcapng) + self.assertIs(datagram.completed, Completion.COMPLETE) + self.assertEqual((str(datagram.id.src), str(datagram.id.dst), datagram.id.id), + (*OUTER[version], OUTER_ID)) + self.assertEqual(datagram.index, (1, 2)) + self.assertEqual(datagram.payload, inner) + + def test_a_datagram_an_icmp_error_quotes_is_not_walked_to(self) -> None: + # A quoted datagram is no tunnel, though DPKT decodes one as an IP layer + # below the ICMP one: its fragment is never the one reassembled + quoted = {4: addressed(wire.ipv4(DATAGRAM[:8], ident=INNER_ID, mf=True, proto=17), *INNER[4]), + 6: addressed(wire.ipv6_fragment(DATAGRAM[:8], ident=INNER_ID, mf=True, nxt=17), *INNER[6])} + # Destination Unreachable: ICMP port unreachable, ICMPv6 port unreachable + errors = {4: bytes([3, 3, 0, 0]) + bytes(4) + quoted[4], 6: bytes([1, 4, 0, 0]) + bytes(4) + quoted[6]} + for engine, pcapng in self.cases(): + for version in (4, 6): + with self.subTest(engine=engine, pcapng=pcapng, version=version): + if not has(engine): + self.skipTest(f'{engine} not installed') + error = whole(version, errors[version], OUTER[version], 1 if version == 4 else 58, + ident=OUTER_ID) + datagrams = reassemble(self, engine, version, [frame(error)], pcapng=pcapng) + if version == 6: + # the outer IPv6 packet has no Fragment header + self.assertEqual(datagrams, []) + continue + # the outer IPv4 packet, read whole as for any frame of no fragment + datagram, = datagrams + self.assertEqual((str(datagram.id.src), str(datagram.id.dst), datagram.id.id), + (*OUTER[4], OUTER_ID)) + self.assertEqual(datagram.payload, errors[4]) + + +if __name__ == '__main__': + unittest.main()