fix(toolkit): the default engine reads TCP under an IPv6 header of Version 4 (#1666)

pcapkit/toolkit/pcap.py

@@ -20,6 +20,7 @@
 from pcapkit.foundation.reassembly.data.tcp import Packet as TCP_Packet
 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.misc.null import NoPayload
 from pcapkit.protocols.misc.raw import Raw
 from pcapkit.protocols.protocol import ProtocolBase
@@ -32,7 +33,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.ipv4 import IPv4
     from pcapkit.protocols.internet.ipv6 import IPv6
     from pcapkit.protocols.internet.ipv6_frag import IPv6_Frag
     from pcapkit.protocols.misc.pcap import Frame
@@ -164,10 +164,12 @@ def tcp_segment(frame: 'ProtocolBase') -> 'Optional[TCPSegment]':
         return None
 
     # A later fragment's payload starts mid-datagram, with no TCP header in it.
-    if ip_info.version == 4:
-        offset = ip_info.offset
+    # NOTE: the layer's class says which header it is, not its Version field:
+    # IPv6 keeps whatever Version it reads, 4 included (#1666).
+    if isinstance(ip, IPv4):
+        offset = ip.info.offset
     else:
-        frag = cast('IPv6', ip).extension_headers.get(Enum_ExtensionHeader.IPv6_Frag)  # type: ignore[call-overload]
+        frag = ip.extension_headers.get(Enum_ExtensionHeader.IPv6_Frag)  # type: ignore[call-overload]
         offset = 0 if frag is None else cast('IPv6_Frag', frag).info.offset
     if offset:
         return None

tests/toolkit/test_pcap_tcp_segment_ipv6_version_1666_unit.py

@@ -0,0 +1,193 @@
+# -*- coding: utf-8 -*-
+"""The default engine reads TCP the parser rejected under an IPv6 header whose Version is 4. C.f. #1666.
+
+:func:`pcapkit.toolkit.pcap.tcp_segment` reads a TCP header the parser rejected
+from its fixed 20 octets, unless the datagram is a later fragment (#1518). It
+took the IPv4 branch of that fragment check by the header's Version field, but
+IPv6 keeps whatever Version it reads, so an IPv6 layer with Version 4 took it
+too and read an ``offset`` IPv6 does not have: the ``AttributeError`` ended the
+extraction, PCAP and PCAP-NG alike. The branch now goes by the layer's class,
+and those frames are read as dpkt and scapy read them.
+
+Every capture is built here, so the module is in the unit tier.
+
+"""
+from __future__ import annotations
+
+import importlib.util
+import ipaddress
+import os
+import struct
+import tempfile
+import unittest
+import warnings
+from typing import TYPE_CHECKING
+from unittest import mock
+
+from tests._support import close_extractor, reimport_once_per_class
+from tests.foundation import _roundtrip as wire
+
+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')
+
+#: The endpoints of every IPv6 header built here, then of every IPv4 one.
+IPV6 = (ipaddress.IPv6Address(wire.IPV6_A), ipaddress.IPv6Address(wire.IPV6_B))
+IPV4 = (ipaddress.IPv4Address(wire.IPV4_A), ipaddress.IPv4Address(wire.IPV4_B))
+
+
+def bad_tcp(port: int) -> bytes:
+    """A TCP segment from ``port`` to 80 whose Data Offset of 15 runs past its 25 octets, so TCP rejects it."""
+    return struct.pack('!HHIIBBHHH', port, 80, 1000, 0, 0xF0, 0x18, 65535, 0, 0) + b'hello'
+
+
+def h4(payload: bytes, nxt: int) -> bytes:
+    """An IPv6 header whose Version field is 4, carrying ``payload`` as ``nxt``."""
+    return struct.pack('!IHBB', 4 << 28, len(payload), nxt, 64) + wire.IPV6_A + wire.IPV6_B + payload
+
+
+def eth(packet: bytes, ethertype: int) -> bytes:
+    """An Ethernet frame carrying ``packet`` as ``ethertype``."""
+    return wire.ethernet(packet, ethertype)
+
+
+def dot1q(packet: bytes, ethertype: int) -> bytes:
+    """An Ethernet frame carrying ``packet`` as ``ethertype`` behind an 802.1Q tag."""
+    return wire.MAC_B + wire.MAC_A + b'\x81\x00' + (100).to_bytes(2, 'big') + ethertype.to_bytes(2, 'big') + packet
+
+
+def fragment(nxt: int) -> bytes:
+    """The Fragment header of a first fragment."""
+    return bytes([nxt, 0, 0, 1]) + (0x55).to_bytes(4, 'big')
+
+
+#: Frames carrying a TCP segment TCP rejects, by a name for each, with the
+#: endpoints the segment is keyed by. ``H4`` is an IPv6 header of Version 4.
+#: Each builder takes the segment it is to carry.
+CASES = {
+    'Ethernet, H4': (IPV6, lambda s: eth(h4(s, 6), 0x86DD)),
+    '802.1Q, H4': (IPV6, lambda s: dot1q(h4(s, 6), 0x86DD)),
+    'H4 in IPv4': (IPV6, lambda s: eth(wire.ipv4(h4(s, 6), proto=41), 0x0800)),
+    '802.1Q, H4 in IPv4': (IPV6, lambda s: dot1q(wire.ipv4(h4(s, 6), proto=41), 0x0800)),
+    'H4 in IPv6': (IPV6, lambda s: eth(wire.ipv6(h4(s, 6), nxt=41), 0x86DD)),
+    'H4 in H4': (IPV6, lambda s: eth(h4(h4(s, 6), 41), 0x86DD)),
+    'H4 first fragment': (IPV6, lambda s: eth(h4(fragment(6) + s, 44), 0x86DD)),
+    'H4, Hop-by-Hop header': (IPV6, lambda s: eth(h4(bytes([6, 0, 1, 4, 0, 0, 0, 0]) + s, 0), 0x86DD)),
+    'IPv6 (control)': (IPV6, lambda s: eth(wire.ipv6(s, nxt=6), 0x86DD)),
+    'IPv4 (control)': (IPV4, lambda s: eth(wire.ipv4(s, proto=6), 0x0800)),
+}
+
+
+def build() -> 'list[bytes]':
+    """The frames of :data:`CASES` in order, the ``n``-th carrying a segment from port ``1000 + n``."""
+    return [make(bad_tcp(1000 + number)) for number, (_, make) in enumerate(CASES.values(), start=1)]
+
+
+def read(test: unittest.TestCase, engine: str, frames: 'list[bytes]', container: str) -> 'dict[str, Any]':
+    """Read ``frames`` with ``engine`` from a ``container`` capture, with IP and TCP reassembly and flow tracing.
+
+    Returns:
+        Every input the engine hands IPv4, IPv6 and TCP reassembly and TCP flow
+        tracing, keyed by frame number -- every field, but a trace input's
+        ``frame`` -- the reassembled datagrams, and the traced flows.
+
+    """
+    from pcapkit import extract
+    from pcapkit.foundation.reassembly.ipv4 import IPv4
+    from pcapkit.foundation.reassembly.ipv6 import IPv6
+    from pcapkit.foundation.reassembly.tcp import TCP
+    from pcapkit.foundation.traceflow.traceflow import TraceFlowBase
+
+    seen = {'IPv4': {}, 'IPv6': {}, 'TCP': {}, 'trace': {}}  # type: dict[str, Any]
+    calls = {klass: klass.__call__ for klass in (IPv4, IPv6, TCP)}
+    trace = TraceFlowBase.__call__
+
+    def fields(packet: 'Any', skip: 'tuple[str, ...]' = ()) -> 'dict[str, Any]':
+        return {key: bytes(value) if isinstance(value, bytearray) else value
+                for key, value in packet.items() if key not in skip}
+
+    def spy(klass: 'Any') -> 'Callable[[Any, Any], Any]':
+        def call(this: 'Any', packet: 'Any') -> 'Any':
+            seen[klass.__name__][packet.num] = fields(packet)
+            return calls[klass](this, packet)
+        return call
+
+    def spy_trace(this: 'Any', packet: 'Any') -> 'Any':
+        seen['trace'][packet.index] = fields(packet, ('frame',))
+        return trace(this, packet)
+
+    write = wire.pcapng if container == 'pcapng' else wire.pcap
+    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, f'version.{container}')
+        with open(path, 'wb') as file:
+            file.write(write([(1, number, octets) for number, octets in enumerate(frames)]))
+        warnings.simplefilter('ignore')
+        extractor = extract(fin=path, nofile=True, engine=engine, reassembly=True, ip=True, tcp=True,
+                            trace=True, trace_fout=os.path.join(tmp, 'trace'), trace_format='json')
+        try:
+            if extractor._exnam != engine:  # pylint: disable=protected-access
+                test.skipTest(f'{engine} did not run')
+            seen['datagrams'] = {
+                'IPv4': [(datagram.completed, datagram.id, datagram.index, datagram.header, datagram.payload)
+                         for datagram in extractor.reassembly.ipv4],
+                'IPv6': [(datagram.completed, datagram.id, datagram.index, datagram.header, datagram.payload)
+                         for datagram in extractor.reassembly.ipv6],
+                'TCP': [datagram.id for datagram in extractor.reassembly.tcp],
+            }
+            seen['flows'] = [(flow.label, tuple(flow.index)) for flow in extractor.trace.tcp]
+        finally:
+            close_extractor(extractor)
+    return seen
+
+
+@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed')
+class IPv6VersionFourTests(unittest.TestCase):
+    """An IPv6 header of Version 4 takes the IPv6 branch of the fragment check, PCAP and PCAP-NG alike."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+
+    def test_the_default_engine_reads_the_segment(self) -> None:
+        frames = build()
+        for container in ('pcap', 'pcapng'):
+            seen = read(self, 'default', frames, container)
+            # every frame is read, those after an H4 one included
+            self.assertEqual(sorted(seen['TCP']), list(range(1, len(CASES) + 1)))
+            self.assertEqual(sorted(seen['trace']), list(range(1, len(CASES) + 1)))
+            self.assertEqual([index for _, index in seen['flows']], [(n,) for n in range(1, len(CASES) + 1)])
+            for number, (name, ((src, dst), _)) in enumerate(CASES.items(), start=1):
+                with self.subTest(container=container, case=name):
+                    segment = bad_tcp(1000 + number)
+                    reasm, trace = seen['TCP'][number], seen['trace'][number]
+                    self.assertEqual(reasm['bufid'], (src, 1000 + number, dst, 80))
+                    self.assertEqual((reasm['dsn'], reasm['ack'], reasm['syn'], reasm['fin'], reasm['rst']),
+                                     (1000, 0, False, False, False))
+                    # the Data Offset runs past the segment, so all of it is header
+                    self.assertEqual((reasm['header'], reasm['payload'], reasm['len']), (segment, b'', 0))
+                    self.assertEqual((trace['src'], trace['srcport'], trace['dst'], trace['dstport']),
+                                     (src, 1000 + number, dst, 80))
+                    self.assertEqual((trace['seq'], trace['ack'], trace['header'], trace['payload']),
+                                     (1000, 0, segment, b''))
+
+    def test_dpkt_and_scapy_hand_over_the_same_inputs(self) -> None:
+        frames = build()
+        for container in ('pcap', 'pcapng'):
+            base = read(self, 'default', frames, container)
+            for engine in ENGINES:
+                with self.subTest(container=container, engine=engine):
+                    if importlib.util.find_spec(engine) is None:
+                        self.skipTest(f'{engine} not installed')
+                    seen = read(self, engine, frames, container)
+                    for aspect in ('TCP', 'trace', 'IPv4', 'IPv6', 'datagrams', 'flows'):
+                        self.assertEqual(seen[aspect], base[aspect], aspect)
+
+
+if __name__ == '__main__':
+    unittest.main()