Skip to content

fix(ipv6): IPv6 core headers rebuild byte-exactly (#1230, #1231, #1236, #1235) #1307

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JarryShaw merged 1 commit into from
Oct 7, 2026
Merged

fix(ipv6): IPv6 core headers rebuild byte-exactly (#1230, #1231, #1236, #1235) #1307

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions pcapkit/protocols/data/internet/ipv6_ext.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
5 changes: 4 additions & 1 deletion pcapkit/protocols/data/internet/ipv6_frag.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
4 changes: 3 additions & 1 deletion pcapkit/protocols/data/internet/ipv6_route.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
52 changes: 24 additions & 28 deletions pcapkit/protocols/internet/ipv6.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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
Expand Down
51 changes: 31 additions & 20 deletions pcapkit/protocols/internet/ipv6_ext.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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.
Expand Down Expand Up @@ -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.
Expand All @@ -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.

Expand All @@ -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
Expand All @@ -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,
Expand Down Expand Up @@ -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
Expand All @@ -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),
}

Expand Down
5 changes: 5 additions & 0 deletions pcapkit/protocols/internet/ipv6_frag.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
)
Expand All @@ -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'',
Expand All @@ -174,6 +176,7 @@ def make(self,
Note that :attr:`Data_IPv6_Frag.offset
<pcapkit.protocols.data.internet.ipv6_frag.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.
Expand All @@ -190,6 +193,7 @@ def make(self,
next=next_val, # type: ignore[arg-type]
flags={
'offset': offset,
'reserved': reserved,
'mf': mf,
},
id=id,
Expand Down Expand Up @@ -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),
Expand Down
19 changes: 17 additions & 2 deletions pcapkit/protocols/internet/ipv6_route.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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]),
Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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.

Expand All @@ -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:
Expand All @@ -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
]
Expand Down Expand Up @@ -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,
)
3 changes: 3 additions & 0 deletions pcapkit/protocols/schema/internet/ipv6_frag.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,8 @@ class Flags(TypedDict):

#: Fragment offset.
offset: int
#: Reserved bits (``Res``).
reserved: int
#: More fragments flag.
mf: int

Expand All @@ -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.
Expand Down
2 changes: 2 additions & 0 deletions pcapkit/protocols/schema/internet/ipv6_route.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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':
Expand Down Expand Up @@ -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(
Expand Down
Loading
Loading