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