From 1c4621b503dfe9538fee9e008c872b7ff643d221 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Wed, 7 Oct 2026 14:25:38 -0400 Subject: [PATCH] fix(ipv6): IPv6 core headers rebuild byte-exactly (#1230, #1231, #1236, #1235) - IPv6_Ext: record the raw Hdr Ext Len octet, the declared next header and the header body (`len`, `declared_next`, `data`), and re-emit them, so a rebuild is byte-exact, an overrun included. `make()` gains `data=`. - IPv6: a layer with no `next` (253/254, or a beholder `Raw`) is no longer recorded in `__exthdr__`; its octets are decoded once, as the upper layer. - IPv6_Route: `make(data=)` unpacks the bytes into the routing type's schema, or raises ProtocolError when they do not parse. - IPv6_Frag `Res` bits and the RPL routing header's 20 reserved bits are named `reserved` and carried through info and `make()`. Closes #1230 Closes #1231 Closes #1236 Refs #1235 --- pcapkit/protocols/data/internet/ipv6_ext.py | 8 ++ pcapkit/protocols/data/internet/ipv6_frag.py | 5 +- pcapkit/protocols/data/internet/ipv6_route.py | 4 +- pcapkit/protocols/internet/ipv6.py | 52 ++++----- pcapkit/protocols/internet/ipv6_ext.py | 51 ++++---- pcapkit/protocols/internet/ipv6_frag.py | 5 + pcapkit/protocols/internet/ipv6_route.py | 19 ++- .../protocols/schema/internet/ipv6_frag.py | 3 + .../protocols/schema/internet/ipv6_route.py | 2 + .../internet/test_ipv6_core_roundtrip_unit.py | 109 ++++++++++++++++++ .../protocols/internet/test_ipv6_ext_unit.py | 12 +- .../internet/test_ipv6_extension_unit.py | 28 +++-- 12 files changed, 228 insertions(+), 70 deletions(-) create mode 100644 tests/protocols/internet/test_ipv6_core_roundtrip_unit.py diff --git a/pcapkit/protocols/data/internet/ipv6_ext.py b/pcapkit/protocols/data/internet/ipv6_ext.py index 303fa85d69..b0968b3520 100644 --- a/pcapkit/protocols/data/internet/ipv6_ext.py +++ b/pcapkit/protocols/data/internet/ipv6_ext.py @@ -37,8 +37,16 @@ class IPv6_Ext(Protocol): #: length would have overrun what remained and the walk stopped instead #: of trusting it. next: 'Optional[TransType]' + #: Next header as declared on the wire. Equal to :attr:`next` except on + #: an overrun, where :attr:`next` is :data:`None` but this keeps the + #: octet so that a rebuild reproduces it. + declared_next: 'TransType' + #: Raw ``Hdr Ext Len`` octet, as on the wire. + len: 'int' #: Length of this extension header, in octets, actually consumed. length: 'int' + #: Header-specific content after the two fixed octets, as consumed. + data: 'bytes' #: Original parsing error, if this instance was reached as a #: :func:`~pcapkit.utilities.decorators.beholder` fallback rather than by #: direct dispatch. diff --git a/pcapkit/protocols/data/internet/ipv6_frag.py b/pcapkit/protocols/data/internet/ipv6_frag.py index a76b07bec7..ceb625a9d8 100644 --- a/pcapkit/protocols/data/internet/ipv6_frag.py +++ b/pcapkit/protocols/data/internet/ipv6_frag.py @@ -20,10 +20,13 @@ class IPv6_Frag(Protocol): next: 'TransType' #: Fragment offset. offset: 'int' + #: Reserved bits (``Res``) between the offset and the M flag, carried + #: verbatim so that a rebuild reproduces them. + reserved: 'int' #: More flag. mf: 'bool' #: Identification. id: 'int' if TYPE_CHECKING: - def __init__(self, next: 'TransType', offset: 'int', mf: 'bool', id: 'int') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,redefined-builtin,multiple-statements + def __init__(self, next: 'TransType', offset: 'int', reserved: 'int', mf: 'bool', id: 'int') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,redefined-builtin,multiple-statements diff --git a/pcapkit/protocols/data/internet/ipv6_route.py b/pcapkit/protocols/data/internet/ipv6_route.py index 8ef4444f5d..274a63d6de 100644 --- a/pcapkit/protocols/data/internet/ipv6_route.py +++ b/pcapkit/protocols/data/internet/ipv6_route.py @@ -78,9 +78,11 @@ class RPL(IPv6_Route): cmpr_e: 'int' #: Pad. pad: 'int' + #: Reserved bits, carried verbatim so that a rebuild reproduces them. + reserved: 'int' #: Addresses. ip: 'tuple[IPv6Address | bytes, ...]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Routing', seg_left: 'int', - cmpr_i: 'int', cmpr_e: 'int', pad: 'int', ip: 'tuple[IPv6Address | bytes, ...]') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,super-init-not-called,redefined-builtin,line-too-long + cmpr_i: 'int', cmpr_e: 'int', pad: 'int', reserved: 'int', ip: 'tuple[IPv6Address | bytes, ...]') -> 'None': ... # pylint: disable=unused-argument,multiple-statements,super-init-not-called,redefined-builtin,line-too-long diff --git a/pcapkit/protocols/internet/ipv6.py b/pcapkit/protocols/internet/ipv6.py index ff0478f5cc..24fb58dad2 100644 --- a/pcapkit/protocols/internet/ipv6.py +++ b/pcapkit/protocols/internet/ipv6.py @@ -408,6 +408,30 @@ def _decode_next_layer(self, ipv6: 'Data_IPv6', proto: 'Optional[int]' = None, next_ = self._import_next_layer(proto, packet=packet, version=6, extension=True, payload=payload) # type: ignore[misc,call-arg,arg-type] info = next_.info + + # A layer with no ``next`` field cannot safely continue the walk. + # This is a *structural* check, not a fixed set of codes: every + # dedicated parser and + # :class:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext` carry + # ``next`` (possibly :data:`None`, on an overrun) -- Shim6 has no + # dedicated parser but is registered to ``IPv6_Ext`` directly -- + # while every other code with no dedicated parser -- ``253``, + # ``254``, or whatever IANA assigns next -- resolves to plain + # :class:`~pcapkit.protocols.misc.raw.Raw`, which does not (see + # :attr:`__generic_ext_codes__`). Reading ``info.next`` on that + # would raise ``AttributeError`` and let a further-out + # :func:`~pcapkit.utilities.decorators.beholder` degrade the + # *whole* packet. + # + # Such a layer is not recorded as an extension header: it consumed + # nothing (``Raw.length`` is 0), so recording it would emit its + # octets twice on rebuild, once from ``__exthdr__`` and once more as + # the upper layer. Stopping here with ``proto`` unchanged instead + # hands the remaining octets to ``super()._decode_next_layer`` + # below, which decodes them once, as the upper layer. + if not hasattr(info, 'next'): + break + name = next_.alias.lstrip('IPv6-').lower() ipv6.__update__({ name: info, @@ -431,34 +455,6 @@ def _decode_next_layer(self, ipv6: 'Data_IPv6', proto: 'Optional[int]' = None, # is what gets handed to ``super()._decode_next_layer`` below payload = payload[next_.length:] - # A layer with no ``next`` field cannot safely continue the walk. - # This is a *structural* check, not a fixed set of codes: every - # dedicated parser and - # :class:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext` carry - # ``next`` (possibly :data:`None`, on an overrun) -- Shim6 has no - # dedicated parser but is registered to ``IPv6_Ext`` directly -- - # while every other code with no dedicated parser -- ``253``, - # ``254``, or whatever IANA assigns next -- resolves to plain - # :class:`~pcapkit.protocols.misc.raw.Raw`, which does not (see - # :attr:`__generic_ext_codes__`). Reading - # ``info.next`` on that would raise ``AttributeError`` and let a - # further-out :func:`~pcapkit.utilities.decorators.beholder` degrade - # the *whole* packet. Stopping here instead keeps this layer's own - # fields (recorded above, in ``self._exthdr`` and in the packet - # dict) and reports no further next header, as on an overrun. - # - # This has to run -- and, on a hit, has to set ``proto`` -- - # *before* the fragment-header special case below: IPv6-Frag - # always carries a real ``next`` (this ``hasattr`` check never - # fires for it), and that ``next`` is the real - # transport layer's code, which the fragment branch's own - # ``break`` must leave in ``proto`` for the final - # ``super()._decode_next_layer`` call below the loop to dispatch - # to correctly. - if not hasattr(info, 'next'): - proto = None - break - proto = info.next # keep original data after fragment header diff --git a/pcapkit/protocols/internet/ipv6_ext.py b/pcapkit/protocols/internet/ipv6_ext.py index 44358f3d2b..bd133225b2 100644 --- a/pcapkit/protocols/internet/ipv6_ext.py +++ b/pcapkit/protocols/internet/ipv6_ext.py @@ -147,8 +147,9 @@ from pcapkit.const.reg.transtype import TransType as Enum_TransType from pcapkit.protocols.data.internet.ipv6_ext import IPv6_Ext as Data_IPv6_Ext from pcapkit.protocols.internet.internet import Internet -from pcapkit.protocols.protocol import _PT, _ST +from pcapkit.protocols.protocol import _PT, _ST, ProtocolBase from pcapkit.protocols.schema.internet.ipv6_ext import IPv6_Ext as Schema_IPv6_Ext +from pcapkit.protocols.schema.schema import Schema from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall, stacklevel from pcapkit.utilities.warnings import SchemaWarning, warn @@ -158,7 +159,6 @@ from typing_extensions import Literal from pcapkit.corekit.protochain import ProtoChain - from pcapkit.protocols.protocol import ProtocolBase class _NextHeaderData(Protocol): """The one field every IPv6 extension header's data model carries. @@ -442,20 +442,28 @@ def read(self, length: 'Optional[int]' = None, *, # pylint: disable=arguments-d ext_len = nominal next_header = schema.next + # NOTE: The schema's payload holds every octet after the two fixed + # ones; the first ``ext_len - 2`` of them are this header's own body. + rest = schema.get_payload() generic_ext = Data_IPv6_Ext( protocol=ext_code, next=next_header, + declared_next=schema.next, + len=schema.len, length=ext_len, + data=rest[:max(ext_len - 2, 0)], error=error, ) if extension: return cast('_PT', generic_ext) - return self._decode_next_layer(cast('_PT', generic_ext), next_header, length - ext_len) + return self._decode_next_layer(cast('_PT', generic_ext), next_header, length - ext_len, + payload=rest[max(ext_len - 2, 0):]) def make(self, *, next: 'Enum_TransType | int' = Enum_TransType.UDP, # pylint: disable=redefined-builtin len: 'int' = 0, # pylint: disable=redefined-builtin + data: 'bytes' = b'', payload: 'bytes | ProtocolBase | Any' = b'', **kwargs: 'Any') -> '_ST': """Make (construct) packet data. @@ -465,6 +473,8 @@ def make(self, *, len: Raw ``Hdr Ext Len`` octet to emit. The caller must size it, since this class cannot know at construction time which per-protocol rule (see the module docstring) it must satisfy. + data: Header-specific content after the two fixed octets, emitted + verbatim ahead of ``payload``. payload: Payload of current instance. **kwargs: Arbitrary keyword arguments. @@ -474,11 +484,11 @@ def make(self, *, Note: **Keyword-only**, for two reasons. - The three stay *declared*, unlike :meth:`read`'s own keywords, + The four stay *declared*, unlike :meth:`read`'s own keywords, because :func:`~pcapkit.protocols.protocol._check_construction_keywords` builds its allowlist from :func:`inspect.signature` of ``make``. Hiding them in ``**kwargs`` **breaks construction**: a plain - ``IPv6_Ext(next=..., len=..., payload=...)`` raises + ``IPv6_Ext(next=..., len=..., data=..., payload=...)`` raises ``UnsupportedCall: IPv6_Ext: unexpected keyword(s)``. The ``__keywords__`` escape hatch would cover that, but it is unioned down the MRO, so it would widen the allowlist, and weaken that @@ -500,6 +510,15 @@ def make(self, *, check. """ + # NOTE: The schema keeps everything after the two fixed octets as one + # opaque payload, so the body is packed in front of the upper layer. + if data: + if isinstance(payload, ProtocolBase): + payload = bytes(payload) + elif isinstance(payload, Schema): + payload = payload.pack() + payload = data + payload + return cast('_ST', Schema_IPv6_Ext( next=next, len=len, @@ -617,12 +636,10 @@ def __index__(cls) -> 'NoReturn': def _make_data(cls, data: 'Data_IPv6_Ext') -> 'dict[str, Any]': # type: ignore[override] """Create key-value pairs from ``data`` for protocol construction. - Inverts whichever per-protocol length rule :meth:`read` applied, - using ``data.protocol`` -- the extension header this instance stood - in for -- to pick the same rule back. Round-tripping a ``data.next`` - of :data:`None` (the overrun case) is not supported: there is no - octet value that both satisfies the rule and still fits, which is - exactly why :meth:`read` stopped rather than clipped. + Emits the ``Hdr Ext Len`` octet and the body as they were read, so + the rebuild is byte-exact whichever per-protocol length rule + :meth:`read` applied, and also for an overrun, where ``data.next`` is + :data:`None` and ``data.declared_next`` keeps the octet. Args: data: protocol data @@ -631,16 +648,10 @@ def _make_data(cls, data: 'Data_IPv6_Ext') -> 'dict[str, Any]': # type: ignore[ Key-value pairs for protocol construction. """ - if data.protocol == Enum_ExtensionHeader.IPv6_Frag: - len_octet = 0 # constant length; the octet itself is Reserved - elif data.protocol == Enum_ExtensionHeader.AH: - len_octet = data.length // 4 - 2 - else: - len_octet = data.length // 8 - 1 - return { - 'next': data.next, - 'len': len_octet, + 'next': data.declared_next, + 'len': data.len, + 'data': data.data, 'payload': cls._make_payload(data), } diff --git a/pcapkit/protocols/internet/ipv6_frag.py b/pcapkit/protocols/internet/ipv6_frag.py index 1af01bd09f..328d6d8c39 100644 --- a/pcapkit/protocols/internet/ipv6_frag.py +++ b/pcapkit/protocols/internet/ipv6_frag.py @@ -145,6 +145,7 @@ def read(self, length: 'Optional[int]' = None, *, extension: 'bool' = False, # ipv6_frag = Data_IPv6_Frag( next=schema.next, offset=int(schema.flags['offset']) * 8, + reserved=schema.flags['reserved'], mf=bool(schema.flags['mf']), id=schema.id, ) @@ -159,6 +160,7 @@ def make(self, next_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long next_reversed: 'bool' = False, offset: 'int' = 0, + reserved: 'int' = 0, mf: 'bool' = False, id: 'int' = 0, payload: 'bytes | ProtocolBase | Schema' = b'', @@ -174,6 +176,7 @@ def make(self, Note that :attr:`Data_IPv6_Frag.offset ` is in octets, so it must be divided by 8 before being passed here. + reserved: Reserved bits (``Res``) between the offset and the M flag. mf: More fragments flag. id: Identification. payload: Payload of current instance. @@ -190,6 +193,7 @@ def make(self, next=next_val, # type: ignore[arg-type] flags={ 'offset': offset, + 'reserved': reserved, 'mf': mf, }, id=id, @@ -264,6 +268,7 @@ def _make_data(cls, data: 'Data_IPv6_Frag') -> 'dict[str, Any]': # type: ignore return { 'next': data.next, 'offset': data.offset // 8, + 'reserved': data.reserved, 'mf': data.mf, 'id': data.id, 'payload': cls._make_payload(data), diff --git a/pcapkit/protocols/internet/ipv6_route.py b/pcapkit/protocols/internet/ipv6_route.py index d3ae36ad63..095be49ffa 100644 --- a/pcapkit/protocols/internet/ipv6_route.py +++ b/pcapkit/protocols/internet/ipv6_route.py @@ -38,6 +38,7 @@ from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext from pcapkit.protocols.schema.internet.ipv6_route import RPL as Schema_RPL from pcapkit.protocols.schema.internet.ipv6_route import IPv6_Route as Schema_IPv6_Route +from pcapkit.protocols.schema.internet.ipv6_route import RoutingType as Schema_RoutingType from pcapkit.protocols.schema.internet.ipv6_route import SourceRoute as Schema_SourceRoute from pcapkit.protocols.schema.internet.ipv6_route import Type2 as Schema_Type2 from pcapkit.protocols.schema.internet.ipv6_route import UnknownType as Schema_UnknownType @@ -58,7 +59,6 @@ from pcapkit.corekit.protochain import ProtoChain from pcapkit.protocols.protocol import ProtocolBase - from pcapkit.protocols.schema.internet.ipv6_route import RoutingType as Schema_RoutingType TypeParser = Callable[[Schema_RoutingType, NamedArg(Schema_IPv6_Route, 'header')], Data_IPv6_Route] TypeConstructor = Callable[[Enum_Routing, DefaultArg(Optional[Data_IPv6_Route]), @@ -299,7 +299,17 @@ def make(self, # ``_make_hdr_ext_len`` can express exactly, then derive # ``Hdr Ext Len`` from the aligned length with that same helper. length = self._make_hdr_ext_len(len(data)) - data_val = data.ljust(ipv6_route_data_length(length), b'\x00') # type: bytes | Schema_RoutingType + data_raw = data.ljust(ipv6_route_data_length(length), b'\x00') + + # NOTE: Unpacked into the routing type's own schema, since + # :meth:`read` hands the schema's ``data`` to the per-type reader, + # which reads fields rather than octets off it. + try: + data_val = Schema_RoutingType.registry[type_val].unpack( # pylint: disable=unsubscriptable-object + data_raw, len(data_raw), {}) # type: Schema_RoutingType + except Exception as exc: + raise ProtocolError(f'{self.alias}: [TypeNo {type_val}] routing data ' + f'{data_raw.hex()} does not parse as this type') from exc elif isinstance(data, (dict, Data_IPv6_Route)): name = self._lookup_registry(self.__routing__, type_val) if isinstance(name, str): @@ -644,6 +654,7 @@ def _read_data_type_rpl(self, schema: 'Schema_RPL', *, header: 'Schema_IPv6_Rout cmpr_i=cmpr_i, cmpr_e=cmpr_e, pad=pad_len, + reserved=schema.pad['reserved'], ip=tuple(schema.ip), ) return ipv6_route @@ -729,6 +740,7 @@ def _make_data_type_2(self, type: 'Enum_Routing', route: 'Optional[Data_Type2]' def _make_data_type_rpl(self, type: 'Enum_Routing', route: 'Optional[Data_RPL]' = None, *, dst: 'Optional[IPv6Address]' = None, ip: 'Optional[list[IPv6Address | str | bytes | int]]' = None, + reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_RPL': """Make IPv6-Route RPL Source data. @@ -737,6 +749,7 @@ def _make_data_type_rpl(self, type: 'Enum_Routing', route: 'Optional[Data_RPL]' route: route data dst: destination IPv6 address ip: list of IPv6 addresses + reserved: reserved bits after ``Pad`` **kwargs: arbitrary keyword arguments Returns: @@ -751,6 +764,7 @@ def _make_data_type_rpl(self, type: 'Enum_Routing', route: 'Optional[Data_RPL]' cmpr_i = route.cmpr_i cmpr_e = route.cmpr_e pad = route.pad + reserved = route.reserved ip_val = [ addr if isinstance(addr, bytes) else addr.packed for addr in route.ip ] @@ -817,6 +831,7 @@ def _make_data_type_rpl(self, type: 'Enum_Routing', route: 'Optional[Data_RPL]' }, pad={ 'pad_len': pad, + 'reserved': reserved, }, addresses=ip_val, ) diff --git a/pcapkit/protocols/schema/internet/ipv6_frag.py b/pcapkit/protocols/schema/internet/ipv6_frag.py index a5ec8ffe80..c1c7c790d5 100644 --- a/pcapkit/protocols/schema/internet/ipv6_frag.py +++ b/pcapkit/protocols/schema/internet/ipv6_frag.py @@ -22,6 +22,8 @@ class Flags(TypedDict): #: Fragment offset. offset: int + #: Reserved bits (``Res``). + reserved: int #: More fragments flag. mf: int @@ -37,6 +39,7 @@ class IPv6_Frag(Schema): #: Fragment offset and flags. flags: 'Flags' = BitField(length=2, namespace={ 'offset': (0, 13), + 'reserved': (13, 2), 'mf': (15, 1), }) #: Identification. diff --git a/pcapkit/protocols/schema/internet/ipv6_route.py b/pcapkit/protocols/schema/internet/ipv6_route.py index ad4020410f..118245fe04 100644 --- a/pcapkit/protocols/schema/internet/ipv6_route.py +++ b/pcapkit/protocols/schema/internet/ipv6_route.py @@ -42,6 +42,7 @@ class PadInfo(TypedDict): """Padding length and reserved.""" pad_len: int + reserved: int def ipv6_route_data_length(hdr_ext_len: 'int') -> 'int': @@ -205,6 +206,7 @@ class RPL(RoutingType, code=Enum_Routing.RPL_Source_Route_Header): #: Padding length and reserved. pad: 'PadInfo' = BitField(length=3, namespace={ 'pad_len': (0, 4), + 'reserved': (4, 20), }) #: Addresses. addresses: 'bytes' = ListField( diff --git a/tests/protocols/internet/test_ipv6_core_roundtrip_unit.py b/tests/protocols/internet/test_ipv6_core_roundtrip_unit.py new file mode 100644 index 0000000000..2077f7e3f6 --- /dev/null +++ b/tests/protocols/internet/test_ipv6_core_roundtrip_unit.py @@ -0,0 +1,109 @@ +# -*- coding: utf-8 -*- +"""IPv6 core headers rebuild byte for byte. + +* GitHub issue #1230: :class:`~pcapkit.protocols.internet.ipv6_ext.IPv6_Ext` + dropped the header body on rebuild, and wrote an overrun's ``Hdr Ext Len`` + as ``0xff``. +* GitHub issue #1231: next-header codes 253 and 254 were recorded as an + extension header *and* decoded as the upper layer, so the payload was + emitted twice. +* GitHub issue #1236: ``IPv6_Route.make(data=)`` raised for every + routing type. +* GitHub issue #1235 (part): the IPv6-Frag ``Res`` bits and the RPL routing + header's 20 reserved bits were written back as zero. + +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`. + +""" + +import io +import unittest +import warnings + +from tests._support import reimport_once_per_class + + +def ipv6(nxt: int, payload: bytes) -> bytes: + """Return an IPv6 header with next header ``nxt``, followed by ``payload``.""" + return (bytes.fromhex('60000000') + len(payload).to_bytes(2, 'big') + bytes([nxt, 64]) + + bytes(15) + b'\x01' + bytes(15) + b'\x02' + payload) + + +class TestIPv6CoreRoundTrip(unittest.TestCase): + """Pin the byte-exact rebuild of the IPv6 core headers.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_ipv6_ext_rebuilds_body(self) -> None: + from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext + + cases = { + 'len0': bytes.fromhex('3b00010203040506'), + 'len1': bytes.fromhex('3b01') + bytes(range(14)), + 'overrun': bytes.fromhex('3b00'), + } + for name, octets in cases.items(): + with self.subTest(case=name), warnings.catch_warnings(): + warnings.simplefilter('ignore') + parsed = IPv6_Ext(io.BytesIO(octets), len(octets), extension=True, alias=140) + self.assertEqual(IPv6_Ext.from_data(parsed.info).data, octets) + self.assertIsNone(parsed.info.next) + + def test_ipv6_ext_rebuilds_through_ipv6(self) -> None: + from pcapkit.protocols.internet.ipv6 import IPv6 + + frame = ipv6(140, bytes.fromhex('3b00010203040506')) + self.assertEqual(IPv6.from_data(IPv6(frame).info).data, frame) + + def test_private_codes_are_not_extension_headers(self) -> None: + from pcapkit.protocols.internet.ipv6 import IPv6 + + for code in (253, 254): + with self.subTest(code=code): + frame = ipv6(code, bytes.fromhex('3b00010203040506')) + parsed = IPv6(frame) + self.assertEqual(parsed.info.__exthdr__, ()) + self.assertEqual(len(parsed.protochain), 2) + self.assertEqual(IPv6.from_data(parsed.info).data, frame) + + def test_route_make_accepts_bytes_data(self) -> None: + from pcapkit.protocols.internet.ipv6_route import IPv6_Route + + for type_ in (0, 2, 3, 4, 5): + with self.subTest(type=type_): + built = IPv6_Route(next=59, type=type_, data=b'\x00' * 20) + octets = built.data + self.assertEqual(octets, bytes([59, 2, type_, 0]) + bytes(20)) + parsed = IPv6_Route(io.BytesIO(octets), len(octets), extension=True) + self.assertEqual(IPv6_Route.from_data(parsed.info).data, octets) + + def test_route_make_rejects_bytes_that_do_not_parse(self) -> None: + from pcapkit.protocols.internet.ipv6_route import IPv6_Route + from pcapkit.utilities.exceptions import ProtocolError + + # An RPL header whose Pad (15) exceeds the octets that follow it. + with self.assertRaises(ProtocolError), warnings.catch_warnings(): + warnings.simplefilter('ignore') + IPv6_Route(next=59, type=3, data=b'\x00\xf0\x00\x00') + + def test_frag_keeps_reserved_bits(self) -> None: + from pcapkit.protocols.internet.ipv6_frag import IPv6_Frag + + octets = bytes.fromhex('3b00000712345678') + parsed = IPv6_Frag(io.BytesIO(octets), len(octets), extension=True) + self.assertEqual(parsed.info.reserved, 3) + self.assertEqual(IPv6_Frag.from_data(parsed.info).data, octets) + + def test_rpl_route_keeps_reserved_bits(self) -> None: + from pcapkit.protocols.internet.ipv6_route import IPv6_Route + + octets = bytes.fromhex('3b020301000abcde') + bytes(16) + parsed = IPv6_Route(io.BytesIO(octets), len(octets), extension=True) + self.assertEqual(parsed.info.reserved, 0xabcde) + self.assertEqual(IPv6_Route.from_data(parsed.info).data, octets) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/internet/test_ipv6_ext_unit.py b/tests/protocols/internet/test_ipv6_ext_unit.py index e464ead184..b204877be6 100644 --- a/tests/protocols/internet/test_ipv6_ext_unit.py +++ b/tests/protocols/internet/test_ipv6_ext_unit.py @@ -375,14 +375,14 @@ def test_unimplemented_terminal_code_stops_the_walk_not_the_packet(self) -> None self.assertEqual(str(ipv6.src), '::1') self.assertEqual(str(ipv6.dst), '::1') - exthdrs = list(ipv6.extension_headers.items(multi=True)) - self.assertEqual(len(exthdrs), 1) - code, terminal = exthdrs[0] - self.assertEqual(code, ExtensionHeader.Use_for_experimentation_and_testing_253) - self.assertIsInstance(terminal, Raw) + # GitHub issue #1231: the ``Raw`` layer is not recorded as an + # extension header, since it consumed nothing; the octets are decoded + # once, as the upper layer, under the code that named them. + self.assertEqual(list(ipv6.extension_headers.items(multi=True)), []) + self.assertEqual(ipv6.info.protocol, ExtensionHeader.Use_for_experimentation_and_testing_253) self.assertIsInstance(ipv6.payload, Raw) - self.assertEqual(str(ipv6.protochain), 'IPv6:Raw:Raw') + self.assertEqual(str(ipv6.protochain), 'IPv6:Use_for_experimentation_and_testing_253') @unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') diff --git a/tests/protocols/internet/test_ipv6_extension_unit.py b/tests/protocols/internet/test_ipv6_extension_unit.py index 520a8601c5..5407ea64f6 100644 --- a/tests/protocols/internet/test_ipv6_extension_unit.py +++ b/tests/protocols/internet/test_ipv6_extension_unit.py @@ -165,6 +165,7 @@ def test_ipv6_frag_index_length_and_make_data(self) -> None: data = DummyDict( next=TransType.TCP, offset=16, + reserved=0, mf=True, id=99, __next_type__=None, @@ -246,7 +247,7 @@ def test_ipv6_frag_read_make_and_properties(self) -> None: proto = object.__new__(IPv6_Frag) proto.__header__ = Schema_IPv6_Frag( next=TransType.TCP, - flags={'offset': 12, 'mf': 1}, + flags={'offset': 12, 'reserved': 0, 'mf': 1}, id=0x12345678, payload=b'data', ) @@ -326,7 +327,7 @@ def test_ipv6_route_readers_and_constructors_cover_registered_types(self) -> Non proto._read_data_type_2(route_schema.Type2(ip='2001:db8::2'), header=header) rpl_schema = route_schema.RPL(cmpr={'cmpr_i': 0, 'cmpr_e': 0}, - pad={'pad_len': 0}, addresses=[]) + pad={'pad_len': 0, 'reserved': 0}, addresses=[]) object.__setattr__(rpl_schema, 'ip', (ip_address('2001:db8::3'),)) rpl_header = types.SimpleNamespace(next=TransType.TCP, length=16, type=Routing.RPL_Source_Route_Header, seg_left=1) @@ -373,7 +374,7 @@ def test_ipv6_route_readers_and_constructors_cover_registered_types(self) -> Non rpl_from_data = proto._make_data_type_rpl( Routing.RPL_Source_Route_Header, route_data.RPL(next=TransType.TCP, length=16, type=Routing.RPL_Source_Route_Header, - seg_left=1, cmpr_i=1, cmpr_e=2, pad=0, + seg_left=1, cmpr_i=1, cmpr_e=2, pad=0, reserved=0, ip=(ip_address('2001:db8::9'),)), ) self.assertEqual(rpl_from_data.cmpr['cmpr_i'], 1) @@ -489,7 +490,7 @@ def test_ipv6_route_read_data_type_errors_report_real_routing_type(self) -> None self.assertEqual(str(type2_ctx.exception), expected) rpl_schema = route_schema.RPL(cmpr={'cmpr_i': 0, 'cmpr_e': 0}, - pad={'pad_len': 0}, addresses=[]) + pad={'pad_len': 0, 'reserved': 0}, addresses=[]) with self.assertRaises(ProtocolError) as rpl_ctx: proto._read_data_type_rpl(rpl_schema, header=header) self.assertEqual(str(rpl_ctx.exception), expected) @@ -544,13 +545,16 @@ def test_ipv6_route_read_make_registry_and_property_edges(self) -> None: self.assertEqual(proto.read(length=208), 'decoded') decode.assert_called_once() - made_bytes = proto.make(type=Routing.Source_Route, data=b'abcd') + made_bytes = proto.make(type=Routing.Source_Route, data=bytes(4)) # Hdr Ext Len in 8-octet units (:rfc:`8200#section-4.4`): 4 octets of # raw data is exactly the 4-octet fixed part every routing type's # data starts with, so no additional 8-octet units are needed. See # #487 -- this used to assert ``1``, the pre-fix octet-count value. self.assertEqual(made_bytes.length, 0) - self.assertEqual(made_bytes.data, b'abcd') + # Raw bytes are unpacked into the routing type's own schema, which + # the per-type reader needs (#1236). + self.assertIsInstance(made_bytes.data, route_schema.SourceRoute) + self.assertEqual(bytes(made_bytes.data), bytes(4)) made_dict = proto.make(type=Routing.Source_Route, data={'ip': ['2001:db8::2']}) self.assertEqual(made_dict.type, Routing.Source_Route) made_model = proto.make( @@ -1532,19 +1536,19 @@ def test_ipv6_route_schema_selector_and_rpl_post_process_branches(self) -> None: first = ip_address('2001:db8::1') second = ip_address('2001:db8::2') - full = route_schema.RPL(cmpr={'cmpr_i': 0, 'cmpr_e': 0}, pad={'pad_len': 0}, + full = route_schema.RPL(cmpr={'cmpr_i': 0, 'cmpr_e': 0}, pad={'pad_len': 0, 'reserved': 0}, addresses=first.packed + second.packed) full.post_process({}) self.assertEqual([str(item) for item in full.ip], ['2001:db8::1', '2001:db8::2']) suffixes = first.packed[8:] + second.packed[8:] compressed = route_schema.RPL(cmpr={'cmpr_i': 8, 'cmpr_e': 8}, - pad={'pad_len': 0}, addresses=suffixes) + pad={'pad_len': 0, 'reserved': 0}, addresses=suffixes) compressed.post_process({}) self.assertEqual(compressed.ip, [first.packed[8:], second.packed[8:]]) with_dst = route_schema.RPL(cmpr={'cmpr_i': 8, 'cmpr_e': 8}, - pad={'pad_len': 0}, addresses=suffixes) + pad={'pad_len': 0, 'reserved': 0}, addresses=suffixes) with_dst.post_process({'dst': ip_address('2001:db8::ffff')}) self.assertEqual([str(item) for item in with_dst.ip], ['2001:db8::1', '2001:db8::2']) @@ -1572,7 +1576,7 @@ def test_ipv6_route_rpl_packs_a_multi_address_list(self) -> None: # Directly at the schema level: ``addresses`` is a ``list[bytes]``, # exactly as ``_make_data_type_rpl`` hands it to the constructor. rpl_schema = route_schema.RPL( - cmpr={'cmpr_i': 0, 'cmpr_e': 0}, pad={'pad_len': 0}, + cmpr={'cmpr_i': 0, 'cmpr_e': 0}, pad={'pad_len': 0, 'reserved': 0}, addresses=[first.packed, second.packed], ) packed = bytes(rpl_schema) @@ -1624,7 +1628,7 @@ def test_ipv6_route_rpl_fixed_area_is_four_octets(self) -> None: # The fixed area alone, with no addresses and no padding. bare = route_schema.RPL(cmpr={'cmpr_i': 0, 'cmpr_e': 0}, - pad={'pad_len': 0}, addresses=[]) + pad={'pad_len': 0, 'reserved': 0}, addresses=[]) self.assertEqual(len(bytes(bare)), 4) # CmprI is the high nibble of the first octet, CmprE the low one; Pad @@ -1632,7 +1636,7 @@ def test_ipv6_route_rpl_fixed_area_is_four_octets(self) -> None: # it. The three trailing octets are the ``padding`` field that # ``pad_len`` sizes, not part of the fixed area. nibbles = route_schema.RPL(cmpr={'cmpr_i': 1, 'cmpr_e': 2}, - pad={'pad_len': 3}, addresses=[]) + pad={'pad_len': 3, 'reserved': 0}, addresses=[]) self.assertEqual(bytes(nibbles)[:4], b'\x12\x30\x00\x00') proto = object.__new__(IPv6_Route)