diff --git a/pcapkit/protocols/internet/ipv6.py b/pcapkit/protocols/internet/ipv6.py index 81ad84be1e..d60edbc84c 100644 --- a/pcapkit/protocols/internet/ipv6.py +++ b/pcapkit/protocols/internet/ipv6.py @@ -197,8 +197,8 @@ 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`). """ @@ -206,6 +206,12 @@ def read(self, length: 'Optional[int]' = None, *, length = len(self) schema = self.__header__ + # NOTE: This class reads IPv6 only, as :meth:`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 # `. A Fragment header beside diff --git a/pcapkit/toolkit/pcap.py b/pcapkit/toolkit/pcap.py index 54234c7d50..54b3900e30 100644 --- a/pcapkit/toolkit/pcap.py +++ b/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: diff --git a/tests/protocols/internet/test_internet_edge_roundtrip_unit.py b/tests/protocols/internet/test_internet_edge_roundtrip_unit.py index 3abce4860c..79e0dfdd72 100644 --- a/tests/protocols/internet/test_internet_edge_roundtrip_unit.py +++ b/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'), diff --git a/tests/protocols/internet/test_ipv6_version_1680_unit.py b/tests/protocols/internet/test_ipv6_version_1680_unit.py new file mode 100644 index 0000000000..41cf2305fc --- /dev/null +++ b/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 ` +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 ` 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() diff --git a/tests/toolkit/test_pcap_tcp_segment_ipv6_version_1666_unit.py b/tests/toolkit/test_pcap_tcp_segment_ipv6_version_1666_unit.py index 9faa422332..fd142121e5 100644 --- a/tests/toolkit/test_pcap_tcp_segment_ipv6_version_1666_unit.py +++ b/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__':