From 2465d9642d6bbbcefe66db42f3652fe184c05147 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Thu, 8 Oct 2026 18:42:18 -0400 Subject: [PATCH] fix(ipv6): dissect a jumbogram's payload up to its Jumbo Payload Length (#1434) - An IPv6 jumbogram (Payload Length 0 with a Hop-by-Hop Jumbo Payload option, RFC 2675) read no payload, so its next layer was NoPayload and the whole jumbo payload landed in info.trailer. The schema now splits payload and trailer at the Jumbo Payload Length, and the payload is dissected as the next header; only octets beyond it are trailer. - The option is honoured only in a Hop-by-Hop header immediately after the IPv6 header. Beside a non-zero Payload Length it is an RFC 2675 error: the Payload Length still governs, and a ProtocolWarning is emitted. - make() writes Payload Length 0 when the payload exceeds 65,535 octets. Rebuilds stay byte-exact; tests in test_ipv6_jumbogram_unit.py. --- pcapkit/protocols/data/internet/ipv6.py | 4 +- pcapkit/protocols/internet/ipv6.py | 34 ++++- pcapkit/protocols/schema/internet/ipv6.py | 72 +++++++++- .../internet/test_ipv6_jumbogram_unit.py | 134 ++++++++++++++++++ 4 files changed, 237 insertions(+), 7 deletions(-) create mode 100644 tests/protocols/internet/test_ipv6_jumbogram_unit.py diff --git a/pcapkit/protocols/data/internet/ipv6.py b/pcapkit/protocols/data/internet/ipv6.py index 93b2504c48..3c7062b9fe 100644 --- a/pcapkit/protocols/data/internet/ipv6.py +++ b/pcapkit/protocols/data/internet/ipv6.py @@ -60,8 +60,8 @@ class IPv6(Protocol): hdr_len: 'int' #: Raw payload length (excluding extension headers). raw_len: 'int' - #: Octets captured past the Payload Length, present only when there - #: are any. + #: Octets captured past the Payload Length -- or, for a jumbogram, + #: past the Jumbo Payload Length -- present only when there are any. trailer: 'bytes' #: Extension header chain, in wire order, as pairs of the parser #: class and its info. Kept apart from the per-header keys above, diff --git a/pcapkit/protocols/internet/ipv6.py b/pcapkit/protocols/internet/ipv6.py index 3717012e4b..7d6eeb073d 100644 --- a/pcapkit/protocols/internet/ipv6.py +++ b/pcapkit/protocols/internet/ipv6.py @@ -36,7 +36,10 @@ from pcapkit.protocols.data.internet.ipv6 import IPv6 as Data_IPv6 from pcapkit.protocols.internet.ip import IP from pcapkit.protocols.schema.internet.ipv6 import IPv6 as Schema_IPv6 +from pcapkit.protocols.schema.internet.ipv6 import jumbo_payload_length from pcapkit.utilities.decorators import beholder +from pcapkit.utilities.exceptions import stacklevel +from pcapkit.utilities.warnings import ProtocolWarning, warn if TYPE_CHECKING: from enum import IntEnum as StdlibEnum @@ -215,6 +218,23 @@ def read(self, length: 'Optional[int]' = None, *, 'trailer': schema.trailer, }) + # 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()) + # update packet info if __packet__ is None: __packet__ = {} @@ -223,7 +243,7 @@ def read(self, length: 'Optional[int]' = None, *, 'dst': ipv6.dst, }) - return self._decode_next_layer(ipv6, schema.next, ipv6.payload, packet=__packet__) # pylint: disable=no-member + return self._decode_next_layer(ipv6, schema.next, extent, packet=__packet__) # pylint: disable=no-member def make(self, traffic_class: 'int' = 0, @@ -252,7 +272,8 @@ def make(self, src: Source IP address. dst: Destination IP address. payload_length: Length of the payload, extension headers included; - computed from the payload when omitted. + computed from the payload when omitted, and zero (a jumbogram) + if that exceeds 65,535 octets. payload: Payload data. trailer: Octets after the packet, outside the Payload Length, such as link-layer padding; written as is. @@ -269,9 +290,14 @@ def make(self, # ``from_data`` rebuild of a capture truncated by snaplen keeps the # 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. + # 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`). if payload_length is None: payload_length = len(payload) + if payload_length > 0xFFFF: + payload_length = 0 return Schema_IPv6( hextet={ @@ -401,7 +427,7 @@ def _decode_next_layer(self, ipv6: 'Data_IPv6', proto: 'Optional[int]' = None, self._exthdr = OrderedMultiDict() # type: OrderedMultiDict[Enum_ExtensionHeader, ProtocolBase] # pylint: disable=attribute-defined-outside-init hdr_len = self.length # header length - raw_len = ipv6.payload # payload length + raw_len = ipv6.payload if length is None else length # payload length _protos = [] # ProtoChain buffer _exthdr = [] # (parser class, info) per extension header diff --git a/pcapkit/protocols/schema/internet/ipv6.py b/pcapkit/protocols/schema/internet/ipv6.py index e54e272514..75a2f8fedd 100644 --- a/pcapkit/protocols/schema/internet/ipv6.py +++ b/pcapkit/protocols/schema/internet/ipv6.py @@ -4,6 +4,7 @@ from typing import TYPE_CHECKING +from pcapkit.const.ipv6.option import Option as Enum_Option from pcapkit.const.reg.transtype import TransType as Enum_TransType from pcapkit.corekit.fields.ipaddress import IPv6AddressField from pcapkit.corekit.fields.misc import PayloadField @@ -16,6 +17,7 @@ if TYPE_CHECKING: from ipaddress import IPv6Address + from typing import Any, Optional from pcapkit.protocols.protocol import ProtocolBase @@ -63,7 +65,8 @@ class IPv6(Schema): dst: 'IPv6Address' = IPv6AddressField() #: Payload. payload: 'bytes' = PayloadField(length=lambda pkt: pkt['length']) - #: Octets captured past the Payload Length, such as Ethernet minimum-frame + #: Octets captured past the Payload Length (or a jumbogram's Jumbo Payload + #: Length, see :meth:`post_process`), such as Ethernet minimum-frame #: padding, c.f. :attr:`IPv4.trailer `. trailer: 'bytes' = PayloadField(default=b'') @@ -73,3 +76,70 @@ def __init__(self, hextet: 'IPv6Hextet', length: 'int', next: 'Enum_TransType', dst: 'IPv6Address | bytes | str | int', payload: 'bytes | ProtocolBase | Schema', trailer: 'bytes' = b'') -> None: ... + + def post_process(self, packet: 'dict[str, Any]') -> 'Schema': + """Revise ``schema`` data after unpacking process. + + A jumbogram [:rfc:`2675`] declares a Payload Length of zero and + carries its real length in a Jumbo Payload option instead, so the + :attr:`payload` field reads nothing and every octet lands in + :attr:`trailer`. When :func:`jumbo_payload_length` finds that option, + the two are split again at the Jumbo Payload Length, and only the + octets beyond it stay in :attr:`trailer`. The octets themselves are + untouched, so the packed header is unchanged. + + Args: + packet: Unpacked data. + + Returns: + Revised schema. + + """ + if self.length != 0 or self.__buffer__.get('payload'): + return self + + rest = self.__buffer__.get('trailer', b'') + jumbo = jumbo_payload_length(self.next, rest) + if jumbo is None: + return self + + self.payload = self.__buffer__['payload'] = rest[:jumbo] + self.trailer = self.__buffer__['trailer'] = rest[jumbo:] + return self + + +def jumbo_payload_length(next: 'int', data: 'bytes') -> 'Optional[int]': # pylint: disable=redefined-builtin + """Find the Jumbo Payload Length of an IPv6 packet [:rfc:`2675`]. + + The Jumbo Payload option is only honoured in a Hop-by-Hop Options header + that immediately follows the IPv6 header, as :rfc:`2675#section-2` and + :rfc:`8200#section-4.1` place it there; one anywhere else is left to the + extension header that carries it. + + Args: + next: Next Header of the IPv6 header. + data: Octets following the IPv6 header. + + Returns: + The Jumbo Payload Length, or :data:`None` if ``data`` does not start + with a Hop-by-Hop Options header carrying a well-formed (four-octet) + Jumbo Payload option. + + """ + if next != Enum_TransType.HOPOPT or len(data) < 2: + return None + + end = min((data[1] + 1) * 8, len(data)) + ptr = 2 + while ptr < end: + code = data[ptr] + if code == Enum_Option.Pad1: + ptr += 1 + continue + if ptr + 1 >= end: + break + size = data[ptr + 1] + if code == Enum_Option.Jumbo_Payload and size == 4 and ptr + 6 <= end: + return int.from_bytes(data[ptr + 2:ptr + 6], 'big', signed=False) + ptr += 2 + size + return None diff --git a/tests/protocols/internet/test_ipv6_jumbogram_unit.py b/tests/protocols/internet/test_ipv6_jumbogram_unit.py new file mode 100644 index 0000000000..1c8aab97b6 --- /dev/null +++ b/tests/protocols/internet/test_ipv6_jumbogram_unit.py @@ -0,0 +1,134 @@ +# -*- coding: utf-8 -*- +"""An IPv6 jumbogram's payload is dissected up to its Jumbo Payload Length. + +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. + +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 +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!!!' +#: Destination and source MAC addresses, then the IPv6 EtherType. +ETHER = bytes.fromhex('001122334455' '66778899aabb' '86dd') + + +def hbh(nxt: 'int', jumbo: 'int') -> 'bytes': + """A Hop-by-Hop Options header carrying only a Jumbo Payload option.""" + return bytes([nxt, 0, 0xC2, 4]) + struct.pack('>I', jumbo) + + +def ipv6(plen: 'int', nxt: 'int', body: 'bytes') -> 'bytes': + """An IPv6 header with the given Payload Length and Next Header.""" + 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) + + +class TestIPv6Jumbogram(unittest.TestCase): + """Pin how a jumbogram's payload extent is read.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def parse(self, data: 'bytes') -> 'object': + from pcapkit.protocols.internet.ipv6 import IPv6 + + proto = IPv6(data) + self.assertEqual(IPv6.from_data(proto.info).data, data) + return proto + + def test_payload_is_dissected(self) -> None: + proto = self.parse(JUMBO) + self.assertEqual(str(proto.protochain), 'IPv6:HOPOPT:UDP:Raw') + self.assertEqual(proto.info.payload, 0) + self.assertEqual(proto.info.raw_len, 24) + 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(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:]) + + 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) + self.assertNotIn('trailer', proto.info) + + 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(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 + 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 + + body = hbh(59, 8 + 70000) + bytes(70000) + proto = IPv6(next=0, payload=body, trailer=b'\xee\xee') + self.assertEqual(proto.data[4:6], b'\x00\x00') + self.assertEqual(proto.data[40:], body + b'\xee\xee') + + parsed = IPv6(proto.data) + self.assertEqual(parsed.info.raw_len, 70000) + self.assertEqual(parsed.info.trailer, b'\xee\xee') + self.assertEqual(IPv6.from_data(parsed.info).data, proto.data) + + +if __name__ == '__main__': + unittest.main()