fix(protocols): reject an IPv6 header whose Version is not 6 (#1680)

pcapkit/protocols/internet/ipv6.py

@@ -197,15 +197,21 @@ def read(self, length: 'Optional[int]' = None, *,
             Parsed packet data.
 
         Raises:
-            ProtocolError: If the packet is one of the errors
-                :rfc:`2675#section-3` has a receiver detect (see
+            ProtocolError: If the version is not 6, or if the packet is one of
+                the errors :rfc:`2675#section-3` has a receiver detect (see
                 :func:`~pcapkit.protocols.schema.internet.ipv6.jumbogram_error`).
 
         """
         if length is None:
             length = len(self)
         schema = self.__header__
 
+        # NOTE: This class reads IPv6 only, as :meth:`IPv4.read
+        # <pcapkit.protocols.internet.ipv4.IPv4.read>` reads IPv4 only; telling
+        # the two apart by the Version is left to the dispatchers (:issue:`1680`).
+        if schema.hextet['version'] != 6:
+            raise ProtocolError(f"[IPv6] invalid version: {schema.hextet['version']}")
+
         # NOTE: A receiver treats each of these as an error (:issue:`1687`), so
         # the frame stops here, c.f. :meth:`IPv4.read
         # <pcapkit.protocols.internet.ipv4.IPv4.read>`. A Fragment header beside

pcapkit/toolkit/pcap.py

@@ -164,8 +164,7 @@ def tcp_segment(frame: 'ProtocolBase') -> 'Optional[TCPSegment]':
         return None
 
     # A later fragment's payload starts mid-datagram, with no TCP header in it.
-    # NOTE: the layer's class says which header it is, not its Version field:
-    # IPv6 keeps whatever Version it reads, 4 included (#1666).
+    # NOTE: the layer's class says which header it is (#1666).
     if isinstance(ip, IPv4):
         offset = ip.info.offset
     else:

tests/protocols/internet/test_internet_edge_roundtrip_unit.py

@@ -65,10 +65,10 @@ def ipx(tlen: 'int | None' = None, tc: int = 0, ptype: int = 0, body: bytes = b'
 
 
 def ipv6(nxt: int, body: bytes, tc: int = 0, fl: int = 0, plen: 'int | None' = None,
-         hlim: int = 64) -> bytes:
+         hlim: int = 64, version: int = 6) -> bytes:
     """An IPv6 header from ``::1`` to ``::2``, then ``body``."""
     plen = len(body) if plen is None else plen
-    return (struct.pack('!IHBB', (6 << 28) | (tc << 20) | fl, plen, nxt, hlim)
+    return (struct.pack('!IHBB', (version << 28) | (tc << 20) | fl, plen, nxt, hlim)
             + bytes(15) + b'\x01' + bytes(15) + b'\x02' + body)
 
 
@@ -201,6 +201,7 @@ def hip_param(code: int, body: bytes) -> bytes:
     Case('ipv6/ah-then-frag', IPV6, ipv6(51, bytes([44, 1, 0, 0]) + bytes(8) + bytes([59, 0, 0, 1]) + bytes(4))),
     Case('ipv6/hopopt-then-trailer', IPV6, ipv6(0, opts(59, bytes(6)), plen=8) + b'\x00\x00'),
     Case('ipv6/esp', IPV6, ipv6(50, b'\x00\x00\x00\x01\x00\x00\x00\x02' + bytes(8))),
+    Case('ipv6/version-4', IPV6, ipv6(59, b'', version=4)),
     *(Case(label, cls, raw) for label, (cls, raw, _) in EXTENSIONS.items()),
     *(Case(f'{label}/extension', cls, raw, EXT) for label, (cls, raw, _) in EXTENSIONS.items()),
     *(Case(f'{label}/in-ipv6', IPV6, ipv6(nxt, raw)) for label, (cls, raw, nxt) in EXTENSIONS.items()),
@@ -248,6 +249,7 @@ def hip_param(code: int, body: bytes) -> bytes:
 REJECTED = {
     'ipv4/version-6': Reject('ProtocolError', '[IPv4] invalid version: 6'),
     'ipv4/ihl-4': Reject('ProtocolError', 'resolved to a negative length'),
+    'ipv6/version-4': Reject('ProtocolError', '[IPv6] invalid version: 4'),
     # a Jumbo Payload option beside a non-zero Payload Length, an RFC 2675 error (#1687)
     'hopopt/jumbo-payload-0/in-ipv6': Reject('ProtocolError', 'Jumbo Payload option with a non-zero Payload '
                                                               'Length 8'),

tests/protocols/internet/test_ipv6_version_1680_unit.py

@@ -0,0 +1,101 @@
+# -*- coding: utf-8 -*-
+"""IPv6 rejects a header whose Version is not 6. C.f. #1680.
+
+GitHub issue #1680: :meth:`IPv6.read <pcapkit.protocols.internet.ipv6.IPv6.read>`
+copied the Version nibble without checking it, so a header of any Version was
+read as IPv6, and :meth:`~pcapkit.protocols.internet.ipv6.IPv6.from_data`, which
+always writes 6, rebuilt it with another first octet. It now rejects any Version
+but 6, as :meth:`IPv4.read <pcapkit.protocols.internet.ipv4.IPv4.read>` rejects
+any but 4: each concrete class reads only its own protocol. Behind Ethernet or
+inside a tunnel the frame stops there, kept raw and rebuilt as captured.
+
+Every case builds its own octets in memory and reads no capture. Classes are
+imported inside each test, after :func:`~tests._support.reimport_once_per_class`,
+c.f. ``tests/protocols/link/test_ethernet_mac_roundtrip_unit.py``.
+
+"""
+
+import struct
+import unittest
+
+from tests._support import reimport_once_per_class
+
+#: The two IPv6 addresses, ``::1`` and ``::2``.
+ADDRS = bytes(15) + b'\x01' + bytes(15) + b'\x02'
+#: A UDP datagram to the discard port, carrying ``ab``.
+UDP = struct.pack('>HHHH', 40000, 9, 10, 0) + b'ab'
+#: Destination and source MAC addresses, then the IPv6 EtherType.
+ETHER = bytes.fromhex('001122334455' '66778899aabb' '86dd')
+#: Every Version but 6.
+VERSIONS = tuple(version for version in range(16) if version != 6)
+
+
+def ipv6(version: 'int', nxt: 'int' = 17, body: 'bytes' = UDP) -> 'bytes':
+    """An IPv6 header of ``version`` carrying ``body`` as ``nxt``."""
+    return struct.pack('>IHBB', version << 28, len(body), nxt, 64) + ADDRS + body
+
+
+def ipv4(body: 'bytes', proto: 'int') -> 'bytes':
+    """An IPv4 header, checksum zero, carrying ``body`` as ``proto``."""
+    return struct.pack('>BBHHHBBH4s4s', 0x45, 0, 20 + len(body), 1, 0x4000, 64, proto, 0,
+                       b'\x0a\x00\x00\x01', b'\x0a\x00\x00\x02') + body
+
+
+class TestIPv6Version(unittest.TestCase):
+    """Pin that only a Version of 6 is read as IPv6."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+
+    def test_ipv6_rejects_every_other_version(self) -> None:
+        from pcapkit.protocols.internet.ipv6 import IPv6
+        from pcapkit.utilities.exceptions import ProtocolError
+
+        for version in VERSIONS:
+            with self.subTest(version=version):
+                with self.assertRaises(ProtocolError) as caught:
+                    IPv6(ipv6(version))
+                self.assertEqual(str(caught.exception), f'[IPv6] invalid version: {version}')
+
+    def test_version_6_is_read(self) -> None:
+        from pcapkit.protocols.internet.ipv6 import IPv6
+
+        data = ipv6(6)
+        proto = IPv6(data)
+        self.assertEqual(proto.info.version, 6)
+        self.assertEqual(str(proto.protochain), 'IPv6:UDP:Raw')
+        self.assertEqual(IPv6.from_data(proto.info).data, data)
+
+    def test_behind_ethernet_the_frame_stops_at_ipv6(self) -> None:
+        # as at an IPv4 header of another version: kept raw, and rebuilt as captured
+        from pcapkit.protocols.link.ethernet import Ethernet
+
+        for version in VERSIONS:
+            for trailer in (b'', b'\xee' * 4):
+                with self.subTest(version=version, trailer=trailer):
+                    frame = ETHER + ipv6(version) + trailer
+                    proto = Ethernet(frame)
+                    self.assertEqual(str(proto.protochain), 'Ethernet:Internet_Protocol_version_6')
+                    self.assertNotIn('IPv6', proto)
+                    self.assertEqual(proto.info.packet, ipv6(version) + trailer)
+                    self.assertEqual(Ethernet.from_data(proto.info).data, frame)
+
+    def test_inside_a_tunnel_the_frame_stops_at_the_inner_header(self) -> None:
+        from pcapkit.protocols.internet.ipv4 import IPv4
+        from pcapkit.protocols.internet.ipv6 import IPv6
+
+        for version in VERSIONS:
+            for label, cls, frame, chain in (
+                ('IPv4', IPv4, ipv4(ipv6(version), 41), 'IPv4:IPv6'),
+                ('IPv6', IPv6, ipv6(6, 41, ipv6(version)), 'IPv6:IPv6'),
+            ):
+                with self.subTest(version=version, outer=label):
+                    proto = cls(frame)
+                    self.assertEqual(str(proto.protochain), chain)
+                    self.assertIsNot(type(proto.payload), IPv6)
+                    self.assertEqual(proto.payload.info.packet, ipv6(version))
+                    self.assertEqual(cls.from_data(proto.info).data, frame)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/toolkit/test_pcap_tcp_segment_ipv6_version_1666_unit.py

@@ -1,13 +1,17 @@
 # -*- coding: utf-8 -*-
-"""The default engine reads TCP the parser rejected under an IPv6 header whose Version is 4. C.f. #1666.
+"""The default engine reads past an IPv6 header whose Version is 4, and hands it nothing. C.f. #1666, #1680.
 
 :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.
+took the IPv4 branch of that fragment check by the header's Version field, and
+IPv6 then read any Version, 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 (#1666). The branch now goes by the layer's class. And
+IPv6 now rejects a Version other than 6 (#1680), so such a header is kept raw:
+its frame hands over no input, and every frame after it is still read.
+
+dpkt and scapy still read those headers as IPv6 and hand their segments over,
+which the best-effort ruling on third-party engines allows (#1687).
 
 Every capture is built here, so the module is in the unit tier.
 
@@ -33,7 +37,7 @@
 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.
+#: The third-party engines read here besides the default engine.
 ENGINES = ('dpkt', 'scapy')
 
 #: The endpoints of every IPv6 header built here, then of every IPv4 one.
@@ -82,6 +86,10 @@ def fragment(nxt: int) -> bytes:
     'IPv4 (control)': (IPV4, lambda s: eth(wire.ipv4(s, proto=6), 0x0800)),
 }
 
+#: The numbers of every frame of :func:`build`, then of the controls alone.
+FRAMES = list(range(1, len(CASES) + 1))
+CONTROLS = [number for number, name in enumerate(CASES, start=1) if name.endswith('(control)')]
+
 
 def build() -> 'list[bytes]':
     """The frames of :data:`CASES` in order, the ``n``-th carrying a segment from port ``1000 + n``."""
@@ -147,46 +155,54 @@ def spy_trace(this: 'Any', packet: 'Any') -> 'Any':
     return seen
 
 
+def check(test: unittest.TestCase, seen: 'dict[str, Any]', numbers: 'list[int]') -> None:
+    """Assert ``seen`` holds the TCP reassembly and trace inputs of the frames ``numbers``, and of no other."""
+    test.assertEqual(sorted(seen['TCP']), numbers)
+    test.assertEqual(sorted(seen['trace']), numbers)
+    test.assertEqual([index for _, index in seen['flows']], [(n,) for n in numbers])
+    for number, (name, ((src, dst), _)) in enumerate(CASES.items(), start=1):
+        if number not in numbers:
+            continue
+        with test.subTest(case=name):
+            segment = bad_tcp(1000 + number)
+            reasm, trace = seen['TCP'][number], seen['trace'][number]
+            test.assertEqual(reasm['bufid'], (src, 1000 + number, dst, 80))
+            test.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
+            test.assertEqual((reasm['header'], reasm['payload'], reasm['len']), (segment, b'', 0))
+            test.assertEqual((trace['src'], trace['srcport'], trace['dst'], trace['dstport']),
+                             (src, 1000 + number, dst, 80))
+            test.assertEqual((trace['seq'], trace['ack'], trace['header'], trace['payload']),
+                             (1000, 0, segment, b''))
+
+
 @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."""
+    """An IPv6 header of Version 4 is no IPv6 layer to the default engine, PCAP and PCAP-NG alike."""
 
     def setUp(self) -> None:
         reimport_once_per_class(self)
 
-    def test_the_default_engine_reads_the_segment(self) -> None:
+    def test_the_default_engine_reads_only_the_controls(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:
+            with self.subTest(container=container):
+                seen = read(self, 'default', frames, container)
+                # the controls come after every H4 frame, so each of those was read
+                check(self, seen, CONTROLS)
+                # nor does the H4 first fragment reach IPv6 reassembly
+                self.assertEqual(seen['IPv6'], {})
+                self.assertEqual(seen['datagrams']['IPv6'], [])
+
+    def test_dpkt_and_scapy_read_every_segment(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)
+                    check(self, read(self, engine, frames, container), FRAMES)
 
 
 if __name__ == '__main__':