diff --git a/examples/generators/options.py b/examples/generators/options.py index a6bdc08e29..975efa30ce 100644 --- a/examples/generators/options.py +++ b/examples/generators/options.py @@ -27,7 +27,8 @@ than a packet assembled by :mod:`scapy` from its own protocol model. That is the point: the octets under test are the library's own output, so a capture is a record of what this version of ``pcapkit`` builds, and re-parsing it exercises -``_read_*`` against ``_make_*``. +``_read_*`` against ``_make_*``. The one frame with more in it is HOPOPT's Jumbo +Payload case, a genuine jumbogram -- see :func:`_jumbogram_datagram`. Only the codes that construct *successfully* can appear in a capture -- there are no octets for the ones that do not. A failing case in a captured family @@ -152,6 +153,18 @@ #: Destination address for every generated IPv6 envelope. DST_IP6 = '2001:db8::2' +#: Jumbo Payload Length of the ``hopopt-option/Jumbo_Payload`` case: the least +#: a jumbogram may carry, one more than the 16-bit Payload Length holds +#: (:rfc:`2675#section-2`). See :func:`_jumbogram_datagram`. +JUMBO_LEN = 0x10000 +#: Snapshot length of a capture whose frames all fit in it: scapy's default, +#: which keeps every capture but a jumbogram's as it always was. +SNAPLEN = 0xFFFF +#: Snapshot length of a capture holding a longer frame: libpcap's +#: ``MAXIMUM_SNAPLEN``. A libpcap reader cuts a record to the file's snaplen, so +#: a jumbogram under :data:`SNAPLEN` would reach it truncated. +JUMBO_SNAPLEN = 262144 + #: Seconds a single case may take before :func:`roundtrip` gives up on it and #: records ``'TIMEOUT'``, before :data:`TIMEOUT_SCALE_ENV` stretches it. A working #: case takes low single-digit milliseconds, so this is headroom for a stalled @@ -575,7 +588,8 @@ def _ipv6_option_overrides() -> 'dict[Any, dict[str, Any]]': from pcapkit.const.ipv6.option import Option as Enum_Option # HOPOPT and IPv6-Opts key on the same enumeration and behave identically - # on every code, so one table serves both. + # on every code, so one table serves both; ``_hopopt_overrides`` adds the + # one value that matters only in a Hop-by-Hop Options header. return { # The nonce is sized to its bit length rounded up to whole octets, at # least one, so the default ``nonce=0`` is a one-octet zero. A @@ -586,6 +600,17 @@ def _ipv6_option_overrides() -> 'dict[Any, dict[str, Any]]': } +def _hopopt_overrides() -> 'dict[Any, dict[str, Any]]': + from pcapkit.const.ipv6.option import Option as Enum_Option + + # A Jumbo Payload option in the Hop-by-Hop Options header after an IPv6 + # header makes the packet a jumbogram, and one whose Jumbo Payload Length is + # not above 65,535 is an error (RFC 2675 section 3) the default engine + # rejects (#1687). So the option carries a legal length, and ``_frame`` + # builds the jumbogram around it. In IPv6-Opts the option is just an option. + return {**_ipv6_option_overrides(), Enum_Option.Jumbo_Payload: {'len': JUMBO_LEN}} + + def _hopopt_base() -> 'dict[str, Any]': from pcapkit.const.reg.transtype import TransType return {'next': TransType.UDP, 'payload': b''} @@ -1588,7 +1613,7 @@ def _pcapng_secrets_build(code: 'Any', kwargs: 'dict[str, Any]') -> 'Any': Family('ipv4-option', _ipv4_registry, _ipv4_overrides, _ipv4_build, _ipv4_parse, _ipv4_extract, _ipv4_rebuild, capture='options-ipv4.pcap', envelope='ethernet-ipv4'), - Family('hopopt-option', _hopopt_registry, _ipv6_option_overrides, + Family('hopopt-option', _hopopt_registry, _hopopt_overrides, _hopopt_build, _hopopt_parse, _ipv6_extract, _hopopt_rebuild, capture='options-ipv6.pcap', envelope='ipv6', proto=0), Family('ipv6-opts-option', _ipv6_opts_registry, _ipv6_option_overrides, @@ -1880,6 +1905,54 @@ def outcomes(families: 'Optional[tuple[Family, ...]]' = None, yield roundtrip(case, deadline) +def _jumbogram_datagram(octets: 'bytes', jumbo: 'int') -> 'bytes': + """The UDP datagram that makes a jumbogram up to its Jumbo Payload Length. + + With a Payload Length of 0 it is the Jumbo Payload Length that says where + the packet ends, so the octets after the Hop-by-Hop Options header have to + fill it, or the frame is a jumbogram cut short. The header names UDP as its + next header, as :func:`_hopopt_base` has every HOPOPT case do, so they are a + datagram of zeros to the Discard port [:rfc:`863`], with the checksum + :rfc:`8200#section-8.1` has a receiver verify. Its length fits UDP's Length + field, which :rfc:`2675#section-4` therefore leaves as is. The checksum is + the one's complement sum of :rfc:`1071`, computed here rather than by + :mod:`scapy`, so that the unit-tier test can build the datagram on a leg + that installs no :mod:`scapy`. + + Args: + octets: The Hop-by-Hop Options header, as :func:`roundtrip` built it. + jumbo: Its Jumbo Payload Length. + + Returns: + The datagram, ``jumbo - len(octets)`` octets long. + + Raises: + ValueError: If the header does not name UDP, or the octets it leaves + are not the length of a UDP datagram. + + """ + import ipaddress + + size = jumbo - len(octets) + if octets[0] != 17 or not 8 <= size <= 0xFFFF: + raise ValueError(f'cannot fill a {jumbo}-octet jumbogram after a {len(octets)}-octet ' + f'header naming next header {octets[0]} with a UDP datagram') + + head = (50000).to_bytes(2, 'big') + (9).to_bytes(2, 'big') + size.to_bytes(2, 'big') + data = bytes(size - 8) + pseudo = (ipaddress.IPv6Address(SRC_IP6).packed + ipaddress.IPv6Address(DST_IP6).packed + + size.to_bytes(4, 'big') + bytes(3) + bytes([17])) + # The sum of the 16-bit words, an odd octet padded with a zero, with every + # carry folded back in; the checksum is its complement. + covered = pseudo + head + bytes(2) + data + bytes(size % 2) + total = sum(int.from_bytes(covered[i:i + 2], 'big') for i in range(0, len(covered), 2)) + while total >> 16: + total = (total & 0xFFFF) + (total >> 16) + # A computed checksum of zero goes out as all ones, since zero means none. + value = (~total & 0xFFFF) or 0xFFFF + return head + value.to_bytes(2, 'big') + data + + def _frame(family: 'Family', octets: 'bytes', index: 'int') -> 'Any': """Wrap constructed octets in the envelope that makes them decodable. @@ -1911,8 +1984,18 @@ def _frame(family: 'Family', octets: 'bytes', index: 'int') -> 'Any': frame = link / IP(src=SRC_IP, dst=DST_IP, proto=family.proto, id=index, ttl=64) / Raw(octets) elif family.envelope == 'ipv6': - frame = link / IPv6(src=SRC_IP6, dst=DST_IP6, nh=family.proto, - hlim=64) / Raw(octets) + from pcapkit.protocols.schema.internet.ipv6 import jumbo_payload_length + + # A Jumbo Payload option where it makes the packet a jumbogram gets a + # genuine one: a Payload Length of 0 (RFC 2675 section 3, #1687), and + # octets to make up the Jumbo Payload Length. + jumbo = None if family.proto is None else jumbo_payload_length(family.proto, octets) + if jumbo is None: + frame = link / IPv6(src=SRC_IP6, dst=DST_IP6, nh=family.proto, + hlim=64) / Raw(octets) + else: + frame = link / IPv6(src=SRC_IP6, dst=DST_IP6, nh=family.proto, hlim=64, + plen=0) / Raw(octets + _jumbogram_datagram(octets, jumbo)) elif family.envelope == 'tcp': frame = (link / IP(src=SRC_IP, dst=DST_IP, id=index, ttl=64) / TCP(sport=50000, dport=80, flags='A', seq=1, ack=1) @@ -1973,7 +2056,9 @@ def generate(dest: 'pathlib.Path | None' = None) -> 'list[pathlib.Path]': written = [] # type: list[pathlib.Path] for name in sorted(buckets): path = dest / name - wrpcap(str(path), buckets[name]) + frames = buckets[name] + snaplen = SNAPLEN if max(len(frame) for frame in frames) <= SNAPLEN else JUMBO_SNAPLEN + wrpcap(str(path), frames, snaplen=snaplen) written.append(path) total = sum(len(frames) for frames in buckets.values()) diff --git a/pcapkit/protocols/internet/ipv6.py b/pcapkit/protocols/internet/ipv6.py index 66b163dfe1..81ad84be1e 100644 --- a/pcapkit/protocols/internet/ipv6.py +++ b/pcapkit/protocols/internet/ipv6.py @@ -39,10 +39,9 @@ from pcapkit.protocols.internet.ip import IP from pcapkit.protocols.protocol import FRAME_LAYER_LIMIT, _frame_depth from pcapkit.protocols.schema.internet.ipv6 import IPv6 as Schema_IPv6 -from pcapkit.protocols.schema.internet.ipv6 import jumbo_payload_length +from pcapkit.protocols.schema.internet.ipv6 import jumbo_payload_length, jumbogram_error from pcapkit.utilities.decorators import beholder -from pcapkit.utilities.exceptions import stacklevel -from pcapkit.utilities.warnings import ProtocolWarning, warn +from pcapkit.utilities.exceptions import ProtocolError if TYPE_CHECKING: from enum import IntEnum as StdlibEnum @@ -197,11 +196,26 @@ def read(self, length: 'Optional[int]' = None, *, Returns: Parsed packet data. + Raises: + ProtocolError: 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: 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 + # the Jumbo Payload option is caught on the walk, see + # :meth:`_decode_next_layer`. + rest = schema.get_payload() + schema.trailer + error = jumbogram_error(schema.length, schema.next, rest) + if error is not None: + raise ProtocolError(f'{self.alias}: {error}') + ipv6 = Data_IPv6.from_dict({ 'version': schema.hextet['version'], 'class': schema.hextet['class'], @@ -224,19 +238,9 @@ def read(self, length: 'Optional[int]' = None, *, # NOTE: A jumbogram's extent is its Jumbo Payload Length, not the zero # Payload Length, c.f. :meth:`Schema_IPv6.post_process # ` - # (:issue:`1434`). :rfc:`2675#section-2` makes a Jumbo Payload option - # beside a non-zero Payload Length an error; such a packet is still - # kept as captured, with the Payload Length deciding its extent, since - # that is what the receiver had to go by. - extent = schema.length - jumbo = jumbo_payload_length(schema.next, schema.get_payload() + schema.trailer) - if jumbo is not None: - if schema.length == 0: - extent = jumbo - else: - warn(f'{self.alias}: Jumbo Payload option with non-zero Payload Length ' - f'{schema.length}; ignoring the Jumbo Payload Length {jumbo}', - ProtocolWarning, stacklevel=stacklevel()) + # (:issue:`1434`); the check above leaves a zero one beside it. + jumbo = jumbo_payload_length(schema.next, rest) + extent = schema.length if jumbo is None else jumbo # update packet info if __packet__ is None: @@ -276,7 +280,8 @@ def make(self, dst: Destination IP address. payload_length: Length of the payload, extension headers included; computed from the payload when omitted, and zero (a jumbogram) - if that exceeds 65,535 octets. + if that exceeds 65,535 octets or the payload starts with a + Hop-by-Hop Options header carrying a Jumbo Payload option. payload: Payload data. trailer: Octets after the packet, outside the Payload Length, such as link-layer padding; written as is. @@ -294,12 +299,13 @@ def make(self, # length the sender declared instead of the one that was recorded # (:issue:`1155`). Computed from the octets at hand otherwise, which is # what a plain ``IPv6(payload=...)`` wants -- or zero for a payload too - # long for the field, i.e. a jumbogram, whose length the caller carries - # in a Jumbo Payload option of ``payload``'s Hop-by-Hop Options header - # (:rfc:`2675#section-2`). + # long for the field, or one whose Hop-by-Hop Options header carries a + # Jumbo Payload option, i.e. a jumbogram, whose length the caller gives + # there (:rfc:`2675#section-2`); its Payload Length must be zero + # (:issue:`1687`). if payload_length is None: payload_length = len(payload) - if payload_length > 0xFFFF: + if payload_length > 0xFFFF or jumbo_payload_length(next_val, bytes(payload)) is not None: payload_length = 0 return Schema_IPv6( @@ -526,6 +532,11 @@ def _decode_next_layer(self, ipv6: 'Data_IPv6', proto: 'Optional[int]' = None, Returns: Current protocol with next layer extracted. + Raises: + ProtocolError: If the walk reaches a Fragment header in a packet + carrying a Jumbo Payload option, an error by + :rfc:`2675#section-3` (:issue:`1687`). + """ #: Extension headers. self._exthdr = OrderedMultiDict() # type: OrderedMultiDict[Enum_ExtensionHeader, ProtocolBase] # pylint: disable=attribute-defined-outside-init @@ -634,6 +645,12 @@ def _decode_next_layer(self, ipv6: 'Data_IPv6', proto: 'Optional[int]' = None, # reassembly, so ``fragment``, ``hdr_len`` and ``raw_len`` keep # that split however far the walk goes on. if split is None: + # NOTE: Nor may a jumbogram carry one (:issue:`1687`), so the + # frame stops here as for the errors :meth:`read` checks; + # past those, its Jumbo Payload option is in the payload. + error = jumbogram_error(ipv6.payload, ipv6.next, self.__header__.get_payload(), lambda: True) + if error is not None: + raise ProtocolError(f'{self.alias}: {error}') split = (hdr_len, raw_len) ipv6.__update__({ 'fragment': self._read_packet(header=hdr_len, payload=raw_len), diff --git a/pcapkit/protocols/schema/internet/ipv6.py b/pcapkit/protocols/schema/internet/ipv6.py index 75a2f8fedd..b2c11ffca4 100644 --- a/pcapkit/protocols/schema/internet/ipv6.py +++ b/pcapkit/protocols/schema/internet/ipv6.py @@ -17,7 +17,7 @@ if TYPE_CHECKING: from ipaddress import IPv6Address - from typing import Any, Optional + from typing import Any, Callable, Optional from pcapkit.protocols.protocol import ProtocolBase @@ -143,3 +143,44 @@ def jumbo_payload_length(next: 'int', data: 'bytes') -> 'Optional[int]': # pyli return int.from_bytes(data[ptr + 2:ptr + 6], 'big', signed=False) ptr += 2 + size return None + + +def jumbogram_error(length: 'int', next: 'int', data: 'bytes', # pylint: disable=redefined-builtin + fragment: 'Optional[Callable[[], bool]]' = None) -> 'Optional[str]': + """Name the error :rfc:`2675#section-3` has a receiver detect in an IPv6 packet, if any. + + The four errors that section lists for a node that understands the Jumbo + Payload option, each a packet the default engine rejects (:issue:`1687`): + + * a Payload Length of zero, a Next Header of Hop-by-Hop Options, and no + Jumbo Payload option; + * a non-zero Payload Length and a Jumbo Payload option; + * a Jumbo Payload option whose Jumbo Payload Length is below 65,536; + * a Jumbo Payload option and a Fragment header. + + The option counts only where :func:`jumbo_payload_length` finds it. + + Args: + length: Payload Length of the IPv6 header. + next: Next Header of the IPv6 header. + data: Octets following the IPv6 header, as captured. + fragment: Tells whether the extension header chain, as the default + engine walks it, has a Fragment header; called only for a packet + carrying the option, so a caller walks the chain only for those. + + Returns: + The error, or :data:`None` if the packet makes none of them. + + """ + jumbo = jumbo_payload_length(next, data) + if jumbo is None: + if not length and next == Enum_TransType.HOPOPT: + return 'Payload Length 0 and a Hop-by-Hop Options header with no Jumbo Payload option' + return None + if length: + return f'Jumbo Payload option with a non-zero Payload Length {length}' + if jumbo <= 0xFFFF: + return f'Jumbo Payload Length {jumbo}, not greater than 65,535' + if fragment is not None and fragment(): + return 'Jumbo Payload option with a Fragment header' + return None diff --git a/tests/protocols/internet/test_internet_edge_roundtrip_unit.py b/tests/protocols/internet/test_internet_edge_roundtrip_unit.py index 59ef863667..3abce4860c 100644 --- a/tests/protocols/internet/test_internet_edge_roundtrip_unit.py +++ b/tests/protocols/internet/test_internet_edge_roundtrip_unit.py @@ -248,6 +248,9 @@ 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'), + # 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'), } KNOWN_FAILURES = () # type: tuple[edge.Gap, ...] diff --git a/tests/protocols/internet/test_ipv6_jumbogram_unit.py b/tests/protocols/internet/test_ipv6_jumbogram_unit.py index 1c8aab97b6..ec4e3c043a 100644 --- a/tests/protocols/internet/test_ipv6_jumbogram_unit.py +++ b/tests/protocols/internet/test_ipv6_jumbogram_unit.py @@ -1,15 +1,25 @@ # -*- coding: utf-8 -*- -"""An IPv6 jumbogram's payload is dissected up to its Jumbo Payload Length. +"""An IPv6 jumbogram is dissected up to its Jumbo Payload Length, and one RFC 2675 calls an error not at all. GitHub issue #1434: a jumbogram [:rfc:`2675`] sets the Payload Length to zero and carries its real length in a Jumbo Payload option of the Hop-by-Hop Options header. :class:`~pcapkit.protocols.internet.ipv6.IPv6` read only the Payload Length, so its next layer was :class:`~pcapkit.protocols.misc.null.NoPayload` and the whole jumbo payload -ended up in ``info.trailer``. - -The Jumbo Payload Length now gives the payload extent, and only the octets -past it are ``trailer``. Every packet still rebuilds byte for byte. +ended up in ``info.trailer``. The Jumbo Payload Length now gives the payload +extent, and only the octets past it are ``trailer``. Every packet still +rebuilds byte for byte. + +GitHub issue #1687: :rfc:`2675#section-3` lists four errors a receiver must +detect -- a Payload Length of 0 with a Hop-by-Hop Options header and no Jumbo +Payload option, a non-zero Payload Length with the option, a Jumbo Payload +Length below 65,536, and the option with a Fragment header. The parser read +each as a jumbogram, or warned and went on; it now rejects each, as +:meth:`IPv4.read ` rejects a version +other than 4, so behind Ethernet the frame stops at the IPv6 layer, kept raw. +And ``IPv6.make`` with no ``payload_length`` writes a Payload Length of 0 for a +payload carrying a Jumbo Payload option, rather than its own length, so a +genuine Jumbo Payload Length builds a jumbogram the parser accepts. 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`, @@ -19,14 +29,15 @@ import struct import unittest -import warnings 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 of 24 octets. -UDP = struct.pack('>HHHH', 1234, 5678, 24, 0) + b'jumbo payload!!!' +#: A TCP header of 20 octets, no options. +TCP = struct.pack('>HHIIBBHHH', 1234, 5678, 1, 0, 0x50, 0x18, 0xFFFF, 0, 0) +#: A TCP segment too long for a Payload Length. +SEGMENT = TCP + b'j' * 70000 #: Destination and source MAC addresses, then the IPv6 EtherType. ETHER = bytes.fromhex('001122334455' '66778899aabb' '86dd') @@ -41,8 +52,35 @@ def ipv6(plen: 'int', nxt: 'int', body: 'bytes') -> 'bytes': return bytes.fromhex('60000000') + struct.pack('>HBB', plen, nxt, 64) + ADDRS + body -#: A well-formed jumbogram: Jumbo Payload Length covers HBH and UDP exactly. -JUMBO = ipv6(0, 0, hbh(17, 32) + UDP) +#: A jumbogram RFC 2675 accepts: its Jumbo Payload Length covers HBH and TCP exactly. +JUMBO = ipv6(0, 0, hbh(6, 8 + len(SEGMENT)) + SEGMENT) + +#: Each error :rfc:`2675#section-3` lists, by case: the packet, then what the +#: parser's :exc:`~pcapkit.utilities.exceptions.ProtocolError` says of it. +ERRORS = { + 'Payload Length 0, Hop-by-Hop Options header, no Jumbo Payload option': ( + ipv6(0, 0, bytes([6, 0, 1, 4, 0, 0, 0, 0]) + TCP), + 'Payload Length 0 and a Hop-by-Hop Options header with no Jumbo Payload option'), + 'Payload Length 0, Next Header 0, nothing after the header': ( + ipv6(0, 0, b''), 'Payload Length 0 and a Hop-by-Hop Options header with no Jumbo Payload option'), + 'Payload Length 28, Jumbo Payload Length 70000': ( + ipv6(8 + len(TCP), 0, hbh(6, 70000) + TCP), 'Jumbo Payload option with a non-zero Payload Length 28'), + 'Payload Length 4, Jumbo Payload option past it': ( + ipv6(4, 0, hbh(6, 70000) + TCP), 'Jumbo Payload option with a non-zero Payload Length 4'), + 'Jumbo Payload Length 28': ( + ipv6(0, 0, hbh(6, 8 + len(TCP)) + TCP), 'Jumbo Payload Length 28, not greater than 65,535'), + 'Jumbo Payload Length 65535': ( + ipv6(0, 0, hbh(6, 65535) + TCP + bytes(65535 - 8 - len(TCP))), + 'Jumbo Payload Length 65535, not greater than 65,535'), + 'Jumbo Payload Length 0': (ipv6(0, 0, hbh(59, 0)), 'Jumbo Payload Length 0, not greater than 65,535'), + 'Fragment header': ( + ipv6(0, 0, hbh(44, 8 + 8 + len(SEGMENT)) + bytes([6, 0, 0, 0, 0, 0, 0, 1]) + SEGMENT), + 'Jumbo Payload option with a Fragment header'), + 'Fragment header behind a Destination Options header': ( + ipv6(0, 0, hbh(60, 8 + 8 + 8 + len(SEGMENT)) + bytes([44, 0, 1, 4, 0, 0, 0, 0]) + + bytes([6, 0, 0, 1, 0, 0, 0, 1]) + SEGMENT), + 'Jumbo Payload option with a Fragment header'), +} class TestIPv6Jumbogram(unittest.TestCase): @@ -60,62 +98,48 @@ def parse(self, data: 'bytes') -> 'object': def test_payload_is_dissected(self) -> None: proto = self.parse(JUMBO) - self.assertEqual(str(proto.protochain), 'IPv6:HOPOPT:UDP:Raw') + self.assertEqual(str(proto.protochain), 'IPv6:HOPOPT:TCP:Raw') self.assertEqual(proto.info.payload, 0) - self.assertEqual(proto.info.raw_len, 24) + self.assertEqual(proto.info.raw_len, len(SEGMENT)) self.assertNotIn('trailer', proto.info) - self.assertEqual(proto.info.udp.len, 24) def test_octets_past_jumbo_length_are_trailer(self) -> None: proto = self.parse(JUMBO + b'\xee' * 6) - self.assertEqual(str(proto.protochain), 'IPv6:HOPOPT:UDP:Raw') + self.assertEqual(str(proto.protochain), 'IPv6:HOPOPT:TCP:Raw') self.assertEqual(proto.info.trailer, b'\xee' * 6) def test_jumbo_length_shorter_than_captured(self) -> None: - data = ipv6(0, 0, hbh(17, 16) + UDP) - proto = self.parse(data) - self.assertEqual(proto.info.raw_len, 8) - self.assertEqual(proto.info.trailer, UDP[8:]) + proto = self.parse(ipv6(0, 0, hbh(6, 8 + len(SEGMENT) - 16) + SEGMENT)) + self.assertEqual(proto.info.raw_len, len(SEGMENT) - 16) + self.assertEqual(proto.info.trailer, SEGMENT[-16:]) def test_jumbo_length_longer_than_captured(self) -> None: # a truncated capture keeps the declared length, c.f. #1155 - proto = self.parse(ipv6(0, 0, hbh(17, 100) + UDP)) - self.assertEqual(str(proto.protochain), 'IPv6:HOPOPT:UDP:Raw') - self.assertEqual(proto.info.raw_len, 92) + proto = self.parse(ipv6(0, 0, hbh(6, 8 + len(SEGMENT) + 100) + SEGMENT)) + self.assertEqual(str(proto.protochain), 'IPv6:HOPOPT:TCP:Raw') + self.assertEqual(proto.info.raw_len, len(SEGMENT) + 100) self.assertNotIn('trailer', proto.info) + def test_least_jumbo_payload_length(self) -> None: + # RFC 2675 section 2: greater than 65,535, so 65,536 is the least + segment = TCP + bytes(65536 - 8 - len(TCP)) + proto = self.parse(ipv6(0, 0, hbh(6, 65536) + segment)) + self.assertEqual(proto.info.raw_len, len(segment)) + def test_inside_ethernet(self) -> None: from pcapkit.protocols.link.ethernet import Ethernet frame = ETHER + JUMBO proto = Ethernet(frame) - self.assertEqual(str(proto.protochain), 'Ethernet:IPv6:HOPOPT:UDP:Raw') + self.assertEqual(str(proto.protochain), 'Ethernet:IPv6:HOPOPT:TCP:Raw') self.assertEqual(Ethernet.from_data(proto.info).data, frame) - def test_non_zero_payload_length_wins_and_warns(self) -> None: - from pcapkit.utilities.warnings import ProtocolWarning - - data = ipv6(32, 0, hbh(17, 16) + UDP) - with warnings.catch_warnings(record=True) as caught: - warnings.simplefilter('always') - proto = self.parse(data) - self.assertTrue(any(issubclass(w.category, ProtocolWarning) - and 'Jumbo Payload option with non-zero Payload Length 32' in str(w.message) - for w in caught)) - self.assertEqual(proto.info.raw_len, 24) - self.assertEqual(str(proto.protochain), 'IPv6:HOPOPT:UDP:Raw') - def test_jumbo_outside_leading_hop_by_hop_is_ignored(self) -> None: # a Hop-by-Hop header after a Destination Options header, RFC 8200 4.1 - body = bytes([0, 0, 1, 4, 0, 0, 0, 0]) + hbh(17, 40) + UDP + body = bytes([0, 0, 1, 4, 0, 0, 0, 0]) + hbh(6, 40) + TCP proto = self.parse(ipv6(0, 60, body)) self.assertEqual(proto.info.trailer, body) - def test_zero_payload_length_without_jumbo_is_unchanged(self) -> None: - body = bytes([17, 0, 1, 4, 0, 0, 0, 0]) + UDP - proto = self.parse(ipv6(0, 0, body)) - self.assertEqual(proto.info.trailer, body) - def test_make_writes_zero_payload_length_past_65535(self) -> None: from pcapkit.protocols.internet.ipv6 import IPv6 @@ -130,5 +154,109 @@ def test_make_writes_zero_payload_length_past_65535(self) -> None: self.assertEqual(IPv6.from_data(parsed.info).data, proto.data) +class TestRFC2675Errors(unittest.TestCase): + """Each error :rfc:`2675#section-3` lists is rejected, and the frame stops there (#1687).""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_ipv6_rejects_each_error(self) -> None: + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.utilities.exceptions import ProtocolError + + for label, (data, message) in ERRORS.items(): + with self.subTest(case=label): + with self.assertRaises(ProtocolError) as caught: + IPv6(data) + self.assertEqual(str(caught.exception), f'IPv6: {message}') + + 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 label, (data, _) in ERRORS.items(): + for trailer in (b'', b'\xee' * 4): + with self.subTest(case=label, trailer=trailer): + frame = ETHER + data + trailer + proto = Ethernet(frame) + self.assertEqual(str(proto.protochain), 'Ethernet:Internet_Protocol_version_6') + self.assertNotIn('IPv6', proto) + self.assertEqual(proto.info.packet, data + trailer) + self.assertEqual(Ethernet.from_data(proto.info).data, frame) + + +class TestMakeJumbogram(unittest.TestCase): + """``IPv6.make`` writes a Payload Length of 0 beside a Jumbo Payload option (#1687).""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_a_jumbo_payload_option_makes_payload_length_zero(self) -> None: + # a Jumbo Payload Length past the octets at hand, as in a capture cut short + from pcapkit.protocols.internet.hopopt import HOPOPT + from pcapkit.protocols.internet.ipv6 import IPv6 + + body = hbh(6, 70000) + TCP + for label, payload in (('bytes', body), ('protocol', HOPOPT(body)), ('schema', HOPOPT(body).schema)): + with self.subTest(payload=label): + proto = IPv6(next=0, src='::1', dst='::2', payload=payload) + self.assertEqual(proto.data, ipv6(0, 0, body)) + self.assertEqual(str(proto.protochain), 'IPv6:HOPOPT:TCP') + self.assertEqual((proto.info.payload, proto.info.raw_len), (0, 70000 - 8)) + self.assertEqual(IPv6.from_data(proto.info).data, proto.data) + self.assertEqual(IPv6(proto.data).data, proto.data) + + def test_explicitly_invalid_input_still_raises(self) -> None: + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.utilities.exceptions import ProtocolError + + for label, kwargs, message in ( + ('Jumbo Payload Length 28', {'payload': hbh(6, 8 + len(TCP)) + TCP}, + 'Jumbo Payload Length 28, not greater than 65,535'), + ('Jumbo Payload Length 65535', {'payload': hbh(6, 65535) + TCP}, + 'Jumbo Payload Length 65535, not greater than 65,535'), + ('a non-zero payload_length', {'payload': hbh(6, 70000) + TCP, 'payload_length': 8 + len(TCP)}, + 'Jumbo Payload option with a non-zero Payload Length 28'), + ('Fragment header', {'payload': hbh(44, 70000) + bytes([6, 0, 0, 0, 0, 0, 0, 1]) + TCP}, + 'Jumbo Payload option with a Fragment header'), + ): + with self.subTest(case=label): + with self.assertRaises(ProtocolError) as caught: + IPv6(next=0, **kwargs) + self.assertEqual(str(caught.exception), f'IPv6: {message}') + + def test_no_leading_jumbo_payload_option_keeps_the_payload_length(self) -> None: + # RFC 2675 section 2: only a Hop-by-Hop Options header right after the IPv6 header holds one + from pcapkit.protocols.internet.ipv6 import IPv6 + + for label, nxt, body in ( + ('Hop-by-Hop Options header, no Jumbo Payload option', 0, bytes([6, 0, 1, 4, 0, 0, 0, 0]) + TCP), + ('Jumbo Payload option behind a Destination Options header', 60, + bytes([0, 0, 1, 4, 0, 0, 0, 0]) + hbh(6, 70000) + TCP), + ): + with self.subTest(case=label): + proto = IPv6(next=nxt, src='::1', dst='::2', payload=body) + self.assertEqual(proto.data, ipv6(len(body), nxt, body)) + self.assertEqual(IPv6.from_data(proto.info).data, proto.data) + + def test_hopopt_alone_and_in_ipv4_are_unchanged(self) -> None: + # the Payload Length rule is IPv6's, so a short Jumbo Payload Length elsewhere is kept as given + from pcapkit.const.ipv6.option import Option + from pcapkit.protocols.internet.hopopt import HOPOPT + from pcapkit.protocols.internet.ipv4 import IPv4 + + hop = HOPOPT(next=6, options=[(Option.Jumbo_Payload, {'len': 8 + len(TCP)})], payload=TCP) + self.assertEqual(hop.data, hbh(6, 8 + len(TCP)) + TCP) + self.assertEqual(str(hop.protochain), 'HOPOPT:TCP') + self.assertEqual(HOPOPT.from_data(hop.info).data, hop.data) + + proto = IPv4(protocol=0, payload=hop) + self.assertEqual(proto.data[2:4], (20 + len(hop.data)).to_bytes(2, 'big')) + self.assertEqual(proto.data[20:], hop.data) + self.assertEqual(str(proto.protochain), 'IPv4:HOPOPT:TCP') + self.assertEqual(IPv4.from_data(proto.info).data, proto.data) + self.assertEqual(IPv4(proto.data).data, proto.data) + + if __name__ == '__main__': unittest.main() diff --git a/tests/protocols/test_option_coverage_runtime.py b/tests/protocols/test_option_coverage_runtime.py index 42a41a5c4d..bbb7d50909 100644 --- a/tests/protocols/test_option_coverage_runtime.py +++ b/tests/protocols/test_option_coverage_runtime.py @@ -132,6 +132,59 @@ def test_every_option_capture_extracts(self) -> None: 'examples/generators/make_samples.py to build them' ) + def test_the_jumbo_payload_frame_is_a_genuine_jumbogram(self) -> None: + """HOPOPT's Jumbo Payload frame is a jumbogram RFC 2675 accepts, read whole. + + The option behind a non-zero Payload Length is an error the default + engine rejects (#1687), and then the frame stops at IPv6 and the option + is never walked. So the generator makes this frame a jumbogram, and this + checks it is one: Payload Length 0, a Jumbo Payload Length above 65,535 + that the octets after the IPv6 header make up exactly, a record no reader + has to cut, and a decode that walks the option and goes on to the end. + + """ + import pcapkit + from pcapkit.const.ipv6.option import Option + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.schema.internet.ipv6 import jumbo_payload_length + + path = _capture('options-ipv6.pcap') + if path is None: + self.skipTest('options-ipv6.pcap not present') + with open(path, 'rb') as file: + ghdr = file.read(24) + # The global header's snaplen, in the byte order its magic number gives. + little = ghdr[:4] in (b'\xd4\xc3\xb2\xa1', b'\x4d\x3c\xb2\xa1') + snaplen = int.from_bytes(ghdr[16:20], 'little' if little else 'big') + + with time_limit(EXTRACT_TIMEOUT): + extractor = pcapkit.extract(fin=path, nofile=True, format='tree', + store=True, ip=True, tcp=True) + self.addCleanup(close_extractor, extractor) + + # Found from the octets after the Ethernet and IPv6 headers rather than + # from the decode, which is under test. + found = [] # type: list[tuple[Any, int]] + frames = list(extractor.frame) # type: list[Any] + for frame in frames: + jumbo = jumbo_payload_length(frame.info.packet[20], frame.info.packet[54:]) + if jumbo is not None: + found.append((frame, jumbo)) + self.assertEqual(len(found), 1, 'expected one frame carrying a Jumbo Payload option') + frame, jumbo = found[0] + packet = frame.info.packet + + self.assertEqual(packet[18:20], b'\x00\x00', 'Payload Length') + self.assertGreater(jumbo, 0xFFFF) + self.assertEqual(len(packet) - 14 - 40, jumbo) + self.assertEqual(frame.info.cap_len, frame.info.len) + self.assertLessEqual(frame.info.cap_len, snaplen) + + self.assertEqual(str(frame.protochain), 'Ethernet:IPv6:HOPOPT:UDP:Raw') + ipv6 = frame[IPv6].info + self.assertEqual(ipv6.hopopt.options[Option.Jumbo_Payload].jumbo_len, jumbo) + self.assertEqual(ipv6.hdr_len + ipv6.raw_len, 40 + jumbo) + def test_option_captures_dump_to_tree_and_json(self) -> None: """Each capture survives both dumpers, with reassembly turned on. diff --git a/tests/protocols/test_option_generator_jumbogram_1687_unit.py b/tests/protocols/test_option_generator_jumbogram_1687_unit.py new file mode 100644 index 0000000000..9c9a7c47b8 --- /dev/null +++ b/tests/protocols/test_option_generator_jumbogram_1687_unit.py @@ -0,0 +1,190 @@ +# -*- coding: utf-8 -*- +"""The generator's Jumbo Payload frame is a jumbogram RFC 2675 accepts. C.f. #1687. + +:file:`examples/generators/options.py` puts HOPOPT's Jumbo Payload case in an +IPv6 envelope in :file:`examples/captures/options-ipv6.pcap`. It used to carry +the option's default Jumbo Payload Length of 0 behind a non-zero Payload Length, +two of the errors :rfc:`2675#section-3` lists, and since #1687 the default +engine rejects such a packet: the frame stopped at IPv6 and its option was never +walked. The case now carries a legal length, and ``_frame`` builds a genuine +jumbogram around it, filled out by ``_jumbogram_datagram``. + +:data:`JUMBO_LEN` and the datagram's fields are spelled out here rather than +read from the generator, so a wrong generator constant fails as ``0 != 65536``. + +This module is unit tier: it builds its frames in memory from the generator's +own cases and reads no capture. The checks on ``_frame`` need scapy, which +builds the envelope, and skip without it. ``_jumbogram_datagram`` needs none, +and one check builds it with scapy unimportable, as on a unit-tier leg. + +""" + +from __future__ import annotations + +import importlib.util +import sys +import types +import unittest +import warnings +from typing import TYPE_CHECKING +from unittest import mock + +from tests._support import time_limit +from tests._tiers import ROOT + +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) +HAS_SCAPY = importlib.util.find_spec('scapy') is not None + +#: The Jumbo Payload Length the case has to carry: the least a jumbogram may, +#: one above what the 16-bit Payload Length holds (:rfc:`2675#section-2`). +JUMBO_LEN = 65536 +#: The envelope's IPv6 addresses, ``2001:db8::1`` and ``2001:db8::2``, as octets. +SRC6 = bytes.fromhex('20010db8000000000000000000000001') +DST6 = bytes.fromhex('20010db8000000000000000000000002') +#: Source and destination port of the datagram: an ephemeral port, and Discard. +PORTS = (50000, 9) + + +def _load_generator() -> 'types.ModuleType': + """Load :file:`examples/generators/options.py` by path, under a name of its own. + + Returns: + The generator module. + + Raises: + RuntimeError: If the module cannot be found or loaded. + + """ + path = ROOT / 'examples' / 'generators' / 'options.py' + spec = importlib.util.spec_from_file_location('pcapkit_samples_options_jumbogram', path) + if spec is None or spec.loader is None: # pragma: no cover + raise RuntimeError(f'cannot load the option generator from {path}') + + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def _ones_sum(data: 'bytes') -> 'int': + """The 16-bit one's complement sum of ``data`` [:rfc:`1071`], uncomplemented.""" + if len(data) % 2: + data += b'\x00' + total = sum(int.from_bytes(data[i:i + 2], 'big') for i in range(0, len(data), 2)) + while total >> 16: + total = (total & 0xFFFF) + (total >> 16) + return total + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class JumbogramCaseTests(unittest.TestCase): + """The ``hopopt-option/Jumbo_Payload`` case, and the datagram that fills it.""" + + options = None # type: Any + outcome = None # type: Any + octets = b'' + + @classmethod + def setUpClass(cls) -> None: + cls.options = _load_generator() + case, = [case for case in cls.options.cases((cls.options.FAMILY_MAP['hopopt-option'],)) + if case.name == 'Jumbo_Payload'] + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + cls.outcome = cls.options.roundtrip(case) + cls.octets = cls.outcome.octets or b'' + + def test_the_case_carries_a_legal_jumbo_payload_length(self) -> None: + """It round-trips, names UDP next, and its option says 65,536.""" + from pcapkit.protocols.schema.internet.ipv6 import jumbo_payload_length + + self.assertEqual(self.outcome.status, 'OK', self.outcome.detail) + self.assertEqual(self.octets[0], 17) + self.assertEqual(jumbo_payload_length(0, self.octets), JUMBO_LEN) + + def test_the_datagram_makes_up_the_jumbo_payload_length(self) -> None: + """Header plus datagram is the Jumbo Payload Length, with a valid UDP checksum.""" + datagram = self.options._jumbogram_datagram(self.octets, JUMBO_LEN) # pylint: disable=protected-access + + self.assertEqual(len(self.octets) + len(datagram), JUMBO_LEN) + self.assertEqual((int.from_bytes(datagram[0:2], 'big'), int.from_bytes(datagram[2:4], 'big')), PORTS) + self.assertEqual(int.from_bytes(datagram[4:6], 'big'), len(datagram)) + self.assertEqual(datagram[8:], bytes(len(datagram) - 8)) + self.assertNotEqual(datagram[6:8], b'\x00\x00') + # A receiver's check: the sum over the pseudo-header and the datagram, + # checksum included, is all ones (RFC 8200 section 8.1, RFC 1071). + pseudo = SRC6 + DST6 + len(datagram).to_bytes(4, 'big') + b'\x00\x00\x00\x11' + self.assertEqual(_ones_sum(pseudo + datagram), 0xFFFF) + + def test_the_datagram_needs_no_scapy(self) -> None: + """It is built with scapy unimportable, as a unit-tier leg installs none.""" + blocked = {name: None for name in list(sys.modules) if name.startswith('scapy.')} + blocked['scapy'] = None + with mock.patch.dict(sys.modules, blocked): + datagram = self.options._jumbogram_datagram(self.octets, JUMBO_LEN) # pylint: disable=protected-access + self.assertEqual(len(self.octets) + len(datagram), JUMBO_LEN) + + def test_a_jumbogram_it_cannot_fill_is_refused(self) -> None: + """No UDP next header, or no room for a datagram that fits its Length field.""" + build = self.options._jumbogram_datagram # pylint: disable=protected-access + size = len(self.octets) + for label, octets, jumbo in ( + ('next header TCP', b'\x06' + self.octets[1:], JUMBO_LEN), + ('no room for a UDP header', self.octets, size + 7), + ('more than a UDP Length holds', self.octets, size + 0x10000), + ): + with self.subTest(label): + with self.assertRaises(ValueError): + build(octets, jumbo) + + @unittest.skipUnless(HAS_SCAPY, 'scapy not installed') + def test_the_frame_is_a_jumbogram_read_whole(self) -> None: + """Payload Length 0, octets to the Jumbo Payload Length, decoded to the end.""" + from pcapkit.protocols.link.ethernet import Ethernet + + frame = bytes(self.options._frame(self.options.FAMILY_MAP['hopopt-option'], # pylint: disable=protected-access + self.octets, 12)) + self.assertEqual(frame[14 + 4:14 + 6], b'\x00\x00') + self.assertEqual(len(frame), 14 + 40 + JUMBO_LEN) + self.assertEqual(frame[14 + 40:14 + 40 + len(self.octets)], self.octets) + + with time_limit(): + proto = Ethernet(frame, len(frame)) + self.assertEqual(str(proto.protochain), 'Ethernet:IPv6:HOPOPT:UDP:Raw') + ipv6 = proto.payload.info # type: Any + self.assertEqual((ipv6.payload, ipv6.hdr_len + ipv6.raw_len), (0, 40 + JUMBO_LEN)) + + +@unittest.skipUnless(HAS_RUNTIME and HAS_SCAPY, 'runtime dependencies or scapy not installed') +class OtherIPv6EnvelopeTests(unittest.TestCase): + """Every other frame in an IPv6 envelope keeps the Payload Length of its octets.""" + + def test_only_the_jumbo_payload_frame_is_a_jumbogram(self) -> None: + options = _load_generator() + families = tuple(family for family in options.FAMILIES if family.envelope == 'ipv6') + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + outcomes = [outcome for outcome in options.outcomes(families) + if outcome.status in options.CAPTURABLE] + self.assertTrue(outcomes) + + jumbograms = [] + for outcome in outcomes: + family = options.FAMILY_MAP[outcome.case.family] + frame = bytes(options._frame(family, outcome.octets, 0)) # pylint: disable=protected-access + length = int.from_bytes(frame[14 + 4:14 + 6], 'big') + if length == 0: + jumbograms.append(outcome.case.label) + continue + with self.subTest(case=outcome.case.label): + self.assertEqual(length, len(outcome.octets)) + self.assertEqual(frame[14 + 40:], outcome.octets) + self.assertEqual(jumbograms, ['hopopt-option/Jumbo_Payload']) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/toolkit/test_dpkt_6in4_reassembly_and_mpls_isl_1637_1638_unit.py b/tests/toolkit/test_dpkt_6in4_reassembly_and_mpls_isl_1637_1638_unit.py index 9d05018bdd..d5897d24d3 100644 --- a/tests/toolkit/test_dpkt_6in4_reassembly_and_mpls_isl_1637_1638_unit.py +++ b/tests/toolkit/test_dpkt_6in4_reassembly_and_mpls_isl_1637_1638_unit.py @@ -146,10 +146,6 @@ def one(label: str, payloads: 'list[int]', carry: 'Any') -> None: lambda n: eth(v4(v4(first(n), 41), 4, length=40 + 48 + 6), trailer=TRAILER)) one('6in4 behind an AH, cut by the Total Length', [6], lambda n: eth(v4(ah(41) + first(n), 51, length=20 + 24 + 48 + 6), trailer=TRAILER)) - one('6in4 jumbogram, Total Length 0, with a trailer', [16], - lambda n: eth(v4(first(n, ext=jumbo(44, 8 + 24), nxt=0, plen=0), 41, length=0), trailer=TRAILER)) - one('6in4 jumbogram shorter than the packet', [4], - lambda n: eth(v4(first(n, ext=jumbo(44, 8 + 8 + 4), nxt=0, plen=0), 41))) # where the default engine dissects no Fragment header of the first IPv6 layer one('6in4, not fragmented', [], lambda n: eth(v4(ipv6(b'A' * 16, 253), 41))) one('ICMP quoting a 6in4 fragment', [], lambda n: eth(v4(icmp_unreach(v4(first(n), 41)), 1))) @@ -179,9 +175,6 @@ def payload_length() -> 'dict[str, tuple[list[int], list[bytes]]]': return { 'a zero Payload Length': ([], [eth(first(901, plen=0))]), 'a zero Payload Length, with a trailer': ([], [eth(first(902, plen=0), trailer=TRAILER)]), - 'a jumbogram, with a trailer': ([16], [eth(first(903, ext=jumbo(44, 8 + 24), nxt=0, plen=0), - trailer=TRAILER)]), - 'a jumbogram shorter than the packet': ([4], [eth(first(904, ext=jumbo(44, 8 + 8 + 4), nxt=0, plen=0))]), 'a jumbogram ending inside the Fragment header': ([], [eth(first(905, ext=jumbo(44, 8 + 4), nxt=0, plen=0))]), 'a Payload Length short of the packet (control)': ([6], [eth(first(906, plen=8 + 6), trailer=TRAILER)]), diff --git a/tests/toolkit/test_dpkt_ip_walk_1609_1626_1651_1654_unit.py b/tests/toolkit/test_dpkt_ip_walk_1609_1626_1651_1654_unit.py index a516eed7a7..804f3f08e7 100644 --- a/tests/toolkit/test_dpkt_ip_walk_1609_1626_1651_1654_unit.py +++ b/tests/toolkit/test_dpkt_ip_walk_1609_1626_1651_1654_unit.py @@ -139,11 +139,7 @@ def jumbogram(short: int = 0) -> bytes: (0, 0, 0, 0), [eth(ipv6(options(6) + segment(), 0, plen=0))]), 'Payload Length 0, 4in6, TCP': ((0, 0, 0, 0), [eth(ipv6(v4(segment(), 6), 4, plen=0))]), '6in6, inner Payload Length 0, TCP': ((0, 0, 0, 0), [eth(v6(ipv6(segment(), 6, plen=0), 41))]), - 'jumbogram, TCP (control)': ((1, 1, 0, 0), [eth(jumbogram())]), - 'jumbogram, TCP, trailer': ((1, 1, 0, 0), [eth(jumbogram()) + TRAILER]), - 'jumbogram, Jumbo Payload Length 2 short': ((1, 1, 0, 0), [eth(jumbogram(2))]), 'jumbogram, Jumbo Payload Length inside the TCP header': ((0, 0, 0, 0), [eth(jumbogram(12))]), - '6in4 jumbogram, Jumbo Payload Length 2 short': ((1, 1, 1, 0), [eth(v4(jumbogram(2), 41))]), } diff --git a/tests/toolkit/test_dpkt_ipv4_reassembly_input_1615_1616_unit.py b/tests/toolkit/test_dpkt_ipv4_reassembly_input_1615_1616_unit.py index 9808fca249..c125bc5d0e 100644 --- a/tests/toolkit/test_dpkt_ipv4_reassembly_input_1615_1616_unit.py +++ b/tests/toolkit/test_dpkt_ipv4_reassembly_input_1615_1616_unit.py @@ -102,7 +102,6 @@ def one(label: str, carry: 'Any', ident: int) -> None: one('4in6 cut by the Payload Length', lambda p: ethernet(ipv6(p, 4, plen=30)), 7) one('4in6 cut by the outer Payload Length of 6in6', lambda p: ethernet(ipv6(ipv6(p, 4, plen=200), 41, plen=70)), 8) - one('4in6 jumbogram', lambda p: ethernet(ipv6(jumbo(4, 8 + 30) + p, 0, plen=0), trailer=b'\xee' * 4), 9) # where the default engine dissects no IPv4 inside the IPv6 packet one('4in6 in a later IPv6 fragment', lambda p: ethernet(ipv6(options(44) + fragment(4, 8, False) + p, 0)), 10) one('4in6 behind 33 extension headers', lambda p: ethernet(ipv6(options(60) * 32 + options(4) + p, 60)), 11) @@ -211,7 +210,6 @@ def test_each_payload_length_above_bounds_the_input(self) -> None: ('4in6 with a trailer', [16, 8]), ('4in6 cut by the Payload Length', [10]), ('4in6 cut by the outer Payload Length of 6in6', [10]), - ('4in6 jumbogram', [10]), ('Total Length 0 in 4in6, with a trailer', [16]), ): for seen in (base, dpkt): diff --git a/tests/toolkit/test_pcap_jumbogram_rfc2675_1687_unit.py b/tests/toolkit/test_pcap_jumbogram_rfc2675_1687_unit.py new file mode 100644 index 0000000000..b08de3dca7 --- /dev/null +++ b/tests/toolkit/test_pcap_jumbogram_rfc2675_1687_unit.py @@ -0,0 +1,207 @@ +# -*- coding: utf-8 -*- +"""The default engine hands over nothing past an IPv6 packet RFC 2675 calls an error. C.f. #1687. + +:rfc:`2675#section-3` lists four errors a receiver must detect: a Payload Length +of 0 with a Hop-by-Hop Options header and no Jumbo Payload option, a non-zero +Payload Length with the option, a Jumbo Payload Length below 65,536, and the +option with a Fragment header. The default engine's IPv6 parser now rejects +each, so it dissects nothing past the IPv6 header: no TCP reassembly or flow +tracing input, and no IP reassembly input from the packet or what it carries. A +jumbogram RFC 2675 accepts is read as before, a capture cut short of its Jumbo +Payload Length included. + +Every capture is built here, so the module is in the unit tier. + +""" +from __future__ import annotations + +import importlib.util +import os +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 + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + +#: What each case's counts count, in order. +ASPECTS = ('TCP', 'trace', 'IPv4', 'IPv6') +#: Link types each case is captured on: Ethernet, and IPv6 with no link layer. +LINKTYPES = (1, 229) +#: Octets past the packet, e.g. an Ethernet FCS. +TRAILER = b'\xee' * 4 +#: A TCP segment too long for a Payload Length. +BIG = wire.tcp(b'j' * 70000, seq=1, ack=1, sport=1006) +#: Octets of the extension headers :func:`genuine` builds. +GENUINE_HEADERS = 32 * 2048 + + +def tcp(number: int) -> bytes: + """A TCP segment of 24 octets on a port pair of its own.""" + return wire.tcp(b'data', seq=1, ack=1, sport=1000 + number) + + +def ipv4(payload: bytes, *, ident: int) -> bytes: + """A first IPv4 fragment carrying ``payload``.""" + return wire.ipv4(payload, ident=ident, mf=True, proto=253) + + +def ipv6(payload: bytes, nxt: int, *, plen: 'int | None' = None) -> bytes: + """An IPv6 packet carrying ``payload``, its Payload Length as given.""" + plen = len(payload) if plen is None else plen + packet = wire.ipv6(b'', nxt=nxt) + return packet[:4] + plen.to_bytes(2, 'big') + packet[6:] + payload + + +def options(nxt: int) -> bytes: + """An 8-octet Hop-by-Hop Options header, one PadN option in it.""" + return bytes([nxt, 0, 1, 4, 0, 0, 0, 0]) + + +def fragment(nxt: int, mf: bool) -> bytes: + """An IPv6 Fragment header at offset 0.""" + return bytes([nxt, 0, 0, int(mf)]) + (7).to_bytes(4, 'big') + + +def jumbo(nxt: int, length: int) -> bytes: + """A Hop-by-Hop Options header carrying a Jumbo Payload option.""" + return bytes([nxt, 0, 0xC2, 4]) + length.to_bytes(4, 'big') + + +def padding(size: int) -> bytes: + """PadN options filling ``size`` octets, at least 2.""" + out = b'' + while size: + data = min(size - 2, 255) + if size - data - 2 == 1: # one octet left over holds no PadN + data -= 1 + out, size = out + bytes([1, data]) + bytes(data), size - data - 2 + return out + + +def genuine(nxt: int, length: int) -> bytes: + """The extension headers of a jumbogram RFC 2675 accepts, its Jumbo Payload Length ``length`` past them. + + A Hop-by-Hop Options header carrying the Jumbo Payload option, then 31 + Destination Options headers, the last naming ``nxt``: 2,048 octets each, so + :data:`GENUINE_HEADERS` (65,536) in all, within the default engine's 32 + (#1604), and the Jumbo Payload Length is above 65,535 however short + ``length`` is. + + """ + head = bytes([60, 255, 0xC2, 4]) + (GENUINE_HEADERS + length).to_bytes(4, 'big') + padding(2040) + return head + b''.join(bytes([nxt if index == 30 else 60, 255]) + padding(2046) for index in range(31)) + + +def cases() -> 'dict[str, tuple[tuple[int, ...], bytes, bytes]]': + """IPv6 packets by case: the default engine's input counts, the packet, and the octets captured after it.""" + first4 = ipv4(b'A' * 16, ident=5) + return { + # the errors, one per row of RFC 2675 section 3 + 'Payload Length 0, a Hop-by-Hop Options header, no Jumbo Payload option': ( + (0, 0, 0, 0), ipv6(options(6) + tcp(1), 0, plen=0), b''), + 'Payload Length 32, Jumbo Payload Length 70000': ((0, 0, 0, 0), ipv6(jumbo(6, 70000) + tcp(2), 0), b''), + 'Jumbo Payload Length 32': ((0, 0, 0, 0), ipv6(jumbo(6, 8 + 24) + tcp(3), 0, plen=0), TRAILER), + 'Jumbo Payload Length 65535': ( + (0, 0, 0, 0), ipv6(jumbo(6, 65535) + tcp(4) + bytes(65535 - 8 - 24), 0, plen=0), b''), + 'Jumbo Payload Length 44, 4in6': ((0, 0, 0, 0), ipv6(jumbo(4, 8 + 36) + first4, 0, plen=0), b''), + 'Fragment header, TCP': ( + (0, 0, 0, 0), ipv6(jumbo(44, 8 + 8 + len(BIG)) + fragment(6, False) + BIG, 0, plen=0), b''), + 'Fragment header, a first fragment': ( + (0, 0, 0, 0), ipv6(jumbo(44, 8 + 8 + 70000) + fragment(253, True) + b'A' * 70000, 0, plen=0), b''), + # jumbograms RFC 2675 accepts + 'jumbogram, TCP (control)': ((1, 1, 0, 0), ipv6(genuine(6, 24) + tcp(7), 0, plen=0), b''), + 'jumbogram, TCP, a trailer (control)': ((1, 1, 0, 0), ipv6(genuine(6, 24) + tcp(8), 0, plen=0), TRAILER), + 'jumbogram, Jumbo Payload Length 2 short of TCP (control)': ( + (1, 1, 0, 0), ipv6(genuine(6, 22) + tcp(9), 0, plen=0), b''), + 'jumbogram, 4in6 cut inside the IPv4 payload (control)': ( + (0, 0, 1, 0), ipv6(genuine(4, 30) + ipv4(b'A' * 16, ident=10), 0, plen=0), b''), + 'Jumbo Payload Length 65536 (control)': ( + (1, 1, 0, 0), ipv6(jumbo(6, 65536) + tcp(11) + bytes(65536 - 8 - 24), 0, plen=0), b''), + 'jumbogram cut short by the capture (control)': ((1, 1, 0, 0), ipv6(jumbo(6, 70000) + tcp(12), 0, plen=0), + b''), + } + + +def frames(linktype: int) -> 'tuple[list[bytes], dict[str, int]]': + """Each case's frame of ``linktype``, and its frame number.""" + built, numbers = [], {} # type: list[bytes], dict[str, int] + for label, (_, packet, trailer) in cases().items(): + built.append((wire.ethernet(packet, 0x86DD) if linktype == 1 else packet) + trailer) + numbers[label] = len(built) + return built, numbers + + +def read(test: unittest.TestCase, linktype: int) -> 'dict[str, set[int]]': + """Read every case's frame of ``linktype`` with IP and TCP reassembly and TCP flow tracing. + + Returns: + The frame numbers of every input the default engine hands IPv4, IPv6 + and TCP reassembly and TCP flow tracing. + + """ + 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': set(), 'IPv6': set(), 'TCP': set(), 'trace': set()} # type: dict[str, set[int]] + calls = {klass: klass.__call__ for klass in (IPv4, IPv6, TCP)} + trace = TraceFlowBase.__call__ + + def spy(klass: 'Any') -> 'Any': + def call(this: 'Any', packet: 'Any') -> 'Any': + seen[klass.__name__].add(packet.num) + return calls[klass](this, packet) + return call + + def spy_trace(this: 'Any', packet: 'Any') -> 'Any': + seen['trace'].add(packet.index) + return trace(this, packet) + + capture = wire.pcap([(1, number, octets) for number, octets in enumerate(frames(linktype)[0])], + linktype=linktype) + capture = capture[:16] + (262144).to_bytes(4, 'little') + capture[20:] # a snapshot length jumbograms fit + 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, 'jumbo.pcap') + with open(path, 'wb') as file: + file.write(capture) + warnings.simplefilter('ignore') + extractor = extract(fin=path, nofile=True, engine='default', reassembly=True, ipv4=True, ipv6=True, + tcp=True, trace=True, trace_fout=os.path.join(tmp, 'trace'), trace_format='json') + try: + test.assertEqual(extractor._exnam, 'default') # pylint: disable=protected-access + finally: + close_extractor(extractor) + return seen + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class DefaultEngineJumbogramTests(unittest.TestCase): + """The default engine's inputs stop at an RFC 2675 error, and not at a jumbogram it accepts.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_nothing_past_an_rfc_2675_error_is_an_input(self) -> None: + for linktype in LINKTYPES: + seen = read(self, linktype) + _, numbers = frames(linktype) + for label, (expected, _, _) in cases().items(): + with self.subTest(linktype=linktype, case=label): + self.assertEqual(tuple(int(numbers[label] in seen[aspect]) for aspect in ASPECTS), expected) + + +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 index 3d1a87140b..5fa56f5726 100644 --- 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 @@ -179,7 +179,6 @@ def test_each_payload_length_above_bounds_the_input(self) -> None: ('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]), - ('4in6 jumbogram', [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]), diff --git a/tests/toolkit/test_scapy_unit.py b/tests/toolkit/test_scapy_unit.py index 8e95a8ef00..ae608eb055 100644 --- a/tests/toolkit/test_scapy_unit.py +++ b/tests/toolkit/test_scapy_unit.py @@ -373,7 +373,6 @@ class ScapyTrailerTests(unittest.TestCase): 'pure-ack-padded': b'', 'ipv6-ext-header-trailer': b'six', 'ip-in-ip-trailers': b'in', - 'jumbogram-trailer': b'jmb', 'zero-total-length': b'tso-data', } @@ -405,7 +404,7 @@ def _ip_frames(self) -> 'list[bytes]': def _tcp_frames(self) -> 'list[bytes]': """One frame per :attr:`TCP_PAYLOADS` entry, in its order.""" from scapy.layers.inet import IP, TCP - from scapy.layers.inet6 import IPv6, IPv6ExtHdrDestOpt, IPv6ExtHdrHopByHop, Jumbo + from scapy.layers.inet6 import IPv6, IPv6ExtHdrDestOpt from scapy.packet import Raw ipv4 = {'src': '192.0.2.1', 'dst': '198.51.100.1'} @@ -426,10 +425,6 @@ def padded(frame: 'bytes') -> 'bytes': bytes(self._ether() / IPv6(**ipv6) / IPv6ExtHdrDestOpt() / TCP(sport=1, dport=2, flags='PA') / Raw(b'six')) + TRAILER, bytes(self._ether() / IP(**ipv4, proto=4) / Raw(inner)) + TRAILER, - # the Jumbo Payload Length counts the Hop-by-Hop Options header too - bytes(self._ether() / IPv6(**ipv6, plen=0, nh=0) / - IPv6ExtHdrHopByHop(nh=6, options=[Jumbo(jumboplen=8 + 20 + 3)]) / - TCP(sport=7, dport=8, flags='PA') / Raw(b'jmb')) + TRAILER, # a Total Length of 0 runs the datagram to the end of the frame tso[:16] + b'\x00\x00' + tso[18:], ]