fix(ipv6): dissect a jumbogram's payload up to its Jumbo Payload Length (#1434)

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,

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
+        # <pcapkit.protocols.schema.internet.ipv6.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
 

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 <pcapkit.protocols.schema.internet.ipv4.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

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()