# -*- coding: utf-8 -*-
"""IPv6 - Internet Protocol version 6
========================================
.. module:: pcapkit.protocols.internet.ipv6.protocol
:mod:`pcapkit.protocols.internet.ipv6.protocol` contains
:class:`~pcapkit.protocols.internet.ipv6.protocol.IPv6` only,
which implements the extractor for Internet Protocol
version 6 (IPv6) [*]_, whose structure is described
as below:
======= ========= ===================== =======================================
Octets Bits Name Description
======= ========= ===================== =======================================
0 0 ``ipv6.version`` Version (``6``)
0 4 ``ipv6.class`` Traffic Class
1 12 ``ipv6.label`` Flow Label
4 32 ``ipv6.payload`` Payload Length (header excluded)
6 48 ``ipv6.next`` Next Header
7 56 ``ipv6.limit`` Hop Limit
8 64 ``ipv6.src`` Source Address
24 192 ``ipv6.dst`` Destination Address
======= ========= ===================== =======================================
.. [*] https://en.wikipedia.org/wiki/IPv6_packet
"""
import contextlib
import ipaddress
from typing import TYPE_CHECKING, cast
from pcapkit.const.ipv6.extension_header import ExtensionHeader as Enum_ExtensionHeader
from pcapkit.const.reg.transtype import TransType as Enum_TransType
from pcapkit.corekit.decorators import beholder
from pcapkit.corekit.module import ModuleDescriptor
from pcapkit.corekit.multidict import OrderedMultiDict
from pcapkit.corekit.protochain import ProtoChain
from pcapkit.corekit.schema.jumbogram import jumbo_payload_length
from pcapkit.protocols.internet.ip.protocol import IP
from pcapkit.protocols.internet.ipv6.data import IPv6 as Data_IPv6
from pcapkit.protocols.internet.ipv6.schema import IP_MAX_LENGTH
from pcapkit.protocols.internet.ipv6.schema import IPv6 as Schema_IPv6
from pcapkit.protocols.internet.ipv6.schema import jumbogram_error
from pcapkit.utilities import config
from pcapkit.utilities.contextvars import frame_depth
from pcapkit.utilities.exceptions import ProtocolError
if TYPE_CHECKING:
from ipaddress import IPv6Address
from typing import IO, Any, ContextManager, Mapping, Optional, Type
from typing_extensions import Literal
from pcapkit.protocols.misc.checksum.ip_pseudo.protocol import IP_Pseudo
from pcapkit.protocols.protocol import ProtocolBase
from pcapkit.protocols.schema import Schema
__all__ = ['IPv6']
#: The pseudo-header :meth:`IPv6.make` computes a payload's checksum over, and the
#: check :meth:`IPv6.__post_init__` runs on the upper layer, resolved there rather
#: than imported, so their first import holds
#: :data:`~pcapkit.utilities.importlib.import_lock` (:issue:`2771`).
_IP_PSEUDO: 'ModuleDescriptor[IP_Pseudo]' = \
ModuleDescriptor('pcapkit.protocols.misc.checksum.ip_pseudo.protocol', 'IP_Pseudo')
_VERIFY_UPPER: 'ModuleDescriptor[Any]' = \
ModuleDescriptor('pcapkit.protocols.misc.checksum.ip_pseudo.protocol', 'verify_upper')
#: Length of the fixed IPv6 header, i.e. the offset of the first extension
#: header, :rfc:`8200#section-3`. The one home of this value (:issue:`1739`); it
#: is fixed by the RFC, so it is never configurable.
#:
#: .. versionadded:: 1.5.0
IPV6_HEADER_LEN = 40
#: Offset of the Next Header field within the fixed IPv6 header,
#: :rfc:`8200#section-3`. The one home of this value (:issue:`1739`).
#:
#: .. versionadded:: 1.5.0
IPV6_NEXT_HEADER_OFFSET = 6
[docs]
class IPv6(IP[Data_IPv6, Schema_IPv6],
schema=Schema_IPv6, data=Data_IPv6):
"""This class implements Internet Protocol version 6.
.. versionchanged:: 1.5.0
* A Shim6 header, which has no dedicated parser, and an extension
header other than ESP whose dedicated parser raises are parsed by
:class:`~pcapkit.protocols.internet.ipv6_ext.protocol.IPv6_Ext`, so the rest
of the chain is still walked (:issue:`891`).
* A rebuild from
:meth:`~pcapkit.protocols.internet.internet.Internet.from_data`
keeps the Payload Length a truncated capture declares
(:issue:`1155`), and the octets past it as
:attr:`~pcapkit.protocols.internet.ipv6.schema.IPv6.trailer`
(:issue:`1209`).
* Next Header codes 253 and 254 are parsed and rebuilt once
(:issue:`1231`), and a cut extension header chain rebuilds byte for
byte (:issue:`1471`, :issue:`1473`).
.. versionchanged:: 1.6.0
* An extension header whose dedicated parser raises is the upper
layer, parsed once as :class:`~pcapkit.protocols.misc.raw.protocol.Raw`, and
the chain walk stops there, as :rfc:`8200#section-4` has a node
discard such a packet (:issue:`2477`).
* ``next`` of :meth:`make` takes a member or an :class:`int` code,
resolved with the enumeration's ``get``; a name, a :obj:`bool`, or a
code that does not fit in one octet raises
:exc:`~pcapkit.utilities.exceptions.ProtocolError`, rather than
resolving against an enumeration inferred from the caller's frame or
writing a different value (:issue:`2594`).
.. versionremoved:: 1.6.0
* ``__generic_ext_codes__``, the extension headers whose dedicated
parser raising fell back to
:class:`~pcapkit.protocols.internet.ipv6_ext.protocol.IPv6_Ext`
(:issue:`2477`).
* The ``next_default``, ``next_namespace`` and ``next_reversed``
arguments of :meth:`make`; resolve a name with
:meth:`~pcapkit.protocols.protocol.Protocol.resolve_index` first
(:issue:`2594`).
"""
#: The ``Raw`` a header-walk stopped on, returned unparsed as the upper layer.
_walk_upper = None # type: Optional[ProtocolBase]
##########################################################################
# Properties.
##########################################################################
@property
def name(self) -> 'Literal["Internet Protocol version 6"]':
"""Name of corresponding protocol."""
return 'Internet Protocol version 6'
@property
def length(self) -> 'Literal[40]':
"""Header length of corresponding protocol."""
return IPV6_HEADER_LEN # type: ignore[return-value]
@property
def protocol(self) -> 'Enum_TransType':
"""Name of next layer protocol."""
return self._info.protocol
# source IP address
@property
def src(self) -> 'IPv6Address':
"""Source IP address."""
return self._info.src
# destination IP address
@property
def dst(self) -> 'IPv6Address':
"""Destination IP address."""
return self._info.dst
@property
def extension_headers(self) -> 'OrderedMultiDict[Enum_ExtensionHeader, ProtocolBase]':
"""IPv6 extension header records."""
return self._exthdr
##########################################################################
# Methods.
##########################################################################
[docs]
def read(self, length: 'Optional[int]' = None, *,
__packet__: 'Optional[dict[str, Any]]' = None, **kwargs: 'Any') -> 'Data_IPv6': # pylint: disable=unused-argument
"""Read Internet Protocol version 6 (IPv6).
Structure of IPv6 header [:rfc:`8200`]:
.. code-block:: text
0 1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|Version| Traffic Class | Flow Label |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Payload Length | Next Header | Hop Limit |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| |
+ +
| |
+ Source Address +
| |
+ +
| |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| |
+ +
| |
+ Destination Address +
| |
+ +
| |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Args:
length: Length of packet data.
__packet__: Optional packet data.
**kwargs: Arbitrary keyword arguments.
Returns:
Parsed packet data.
Raises:
ProtocolError: If the version is not 6, or if the packet is one of
the errors :rfc:`2675#section-3` has a receiver detect (see
:func:`~pcapkit.protocols.internet.ipv6.schema.jumbogram_error`).
"""
if length is None:
length = len(self)
schema = self.__header__
# NOTE: This class reads IPv6 only, as ``IPv4.read`` reads IPv4 only; telling the two
# apart by the Version is left to the dispatchers (#1680).
if schema.hextet['version'] != 6:
raise ProtocolError(f"[IPv6] invalid version: {schema.hextet['version']}")
# NOTE: A receiver treats each as an error (#1687), so the frame stops here, as in
# ``IPv4.read``; a Fragment header beside a Jumbo Payload is caught on the walk.
rest = schema.get_payload() + schema.trailer
error = jumbogram_error(schema.length, schema.next, rest)
if error is not None:
raise ProtocolError(f'{self.alias}: {error}')
ipv6 = Data_IPv6.from_dict({
'version': schema.hextet['version'],
'class': schema.hextet['class'],
'label': schema.hextet['label'],
'payload': schema.length,
'next': schema.next,
'limit': schema.limit,
'src': schema.src,
'dst': schema.dst,
}) # type: Data_IPv6
# NOTE: Octets captured past the Payload Length are kept as captured,
# see :meth:`IPv4.read <pcapkit.protocols.internet.ipv4.protocol.IPv4.read>` (#1209).
if schema.trailer:
ipv6.__update__({
'trailer': schema.trailer,
})
# NOTE: A jumbogram's extent is its Jumbo Payload Length, not the zero Payload Length, see
# ``Schema_IPv6.post_process`` (#1434); the check above leaves a zero one beside it.
jumbo = jumbo_payload_length(schema.next, rest)
extent = schema.length if jumbo is None else jumbo
# update packet info
if __packet__ is None:
__packet__ = {}
__packet__.update({
'src': ipv6.src,
'dst': ipv6.dst,
})
return self._decode_next_layer(ipv6, schema.next, extent, packet=__packet__) # pylint: disable=no-member
[docs]
def make(self,
traffic_class: 'int' = 0,
flow_label: 'int' = 0,
next: 'Enum_TransType | int' = Enum_TransType.UDP,
hop_limit: 'int' = 64, # reasonable default
src: 'IPv6Address | str | bytes | int' = '::1',
dst: 'IPv6Address | str | bytes | int' = '::',
payload_length: 'Optional[int]' = None,
payload: 'bytes | ProtocolBase | Schema' = b'',
trailer: 'bytes' = b'',
**kwargs: 'Any') -> 'Schema_IPv6':
"""Make (construct) packet data.
Args:
traffic_class: Traffic class.
flow_label: Flow label.
next: Next header.
hop_limit: Hop limit.
src: Source IP address.
dst: Destination IP address.
payload_length: Length of the payload, extension headers included;
computed from the payload when omitted, and zero (a jumbogram)
if that exceeds 65,535 octets or the payload starts with a
Hop-by-Hop Options header carrying a Jumbo Payload option.
payload: Payload data.
trailer: Octets after the packet, outside the Payload Length, such
as link-layer padding; written as is.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed packet data.
Raises:
ProtocolError: If ``next`` is a name, a :obj:`bool`, or a member
or code that does not fit in one octet.
.. versionchanged:: 1.5.0
Takes ``payload_length`` ahead of ``payload`` (:issue:`1155`), which
shifts a positional ``payload``.
"""
next_val = self._make_index(next, Enum_TransType, bits=8, name='next header')
# NOTE: Written as given, so a truncated capture keeps its declared length (#1155);
# otherwise computed, or zero for an oversize payload or a jumbogram (#1687).
if payload_length is None:
payload_length = len(payload)
if payload_length > IP_MAX_LENGTH or jumbo_payload_length(next_val, bytes(payload)) is not None:
payload_length = 0
schema = Schema_IPv6(
hextet={
'version': 6,
'class': traffic_class,
'label': flow_label,
},
length=payload_length,
next=next_val,
limit=hop_limit,
src=src,
dst=dst,
payload=payload,
trailer=trailer,
)
# NOTE: A payload that left its checksum out gets it over this header's
# pseudo-header; see pcapkit.protocols.misc.checksum.ip_pseudo.protocol (#2471).
if (config.CHECKSUM_COMPUTE and getattr(payload, '_checksum_pending', False)
and cast('ProtocolBase', payload).__index__() == next_val):
IP_Pseudo = _IP_PSEUDO.klass # pylint: disable=invalid-name
# NOTE: The payload follows this header directly, so no extension
# header moves an address; it puts its own length in the pseudo-header.
pseudo = IP_Pseudo(src=src, dst=dst, protocol=next_val, upper_length=0)
upper = cast('ProtocolBase', payload)
schema.payload = upper.from_data(upper.info, checksum=None, pseudo=pseudo).data
return schema
[docs]
def unpack(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_IPv6':
"""Unpack (parse) packet data.
The Version field is checked before the header schema is parsed, so
octets of another version are refused as such, rather than by whichever
field of this header they break first (:issue:`2574`).
Args:
length: Length of packet data.
**kwargs: Arbitrary keyword arguments.
Returns:
Parsed packet data.
Raises:
ProtocolError: If the Version field is not ``6``.
"""
if cast('Optional[Schema_IPv6]', self.__header__) is None and self._data:
version = self._data[0] >> 4
if version != 6:
raise ProtocolError(f'[IPv6] invalid version: {version}')
return super().unpack(length, **kwargs)
[docs]
@classmethod
def id(cls) -> 'tuple[Literal["IPv6"]]': # type: ignore[override]
"""Index ID of the protocol.
Returns:
Index ID of the protocol.
"""
return ('IPv6',)
##########################################################################
# Data models.
##########################################################################
def __length_hint__(self) -> 'Literal[40]':
"""Return an estimated length for the object."""
return IPV6_HEADER_LEN # type: ignore[return-value]
[docs]
@classmethod
def __index__(cls) -> 'Enum_TransType': # pylint: disable=invalid-index-returned
"""Numeral registry index of the protocol.
Returns:
Numeral registry index of the protocol in `IANA`_.
.. _IANA: https://www.iana.org/assignments/protocol-numbers/protocol-numbers.xhtml
"""
return Enum_TransType.IPv6 # type: ignore[return-value]
[docs]
def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None,
length: 'Optional[int]' = None, **kwargs: 'Any') -> 'None':
"""Post initialisation hook.
Once parsed, the checksum of the upper layer is verified over this
layer's pseudo-header if :data:`~pcapkit.utilities.config.CHECKSUM_VERIFY`
is on, see :func:`~pcapkit.protocols.misc.checksum.ip_pseudo.protocol.verify_upper`.
Args:
file: Source packet stream.
length: Length of packet data.
**kwargs: Arbitrary keyword arguments.
See Also:
For construction arguments, please refer to :meth:`make`.
"""
super().__post_init__(file, length, **kwargs)
if config.CHECKSUM_VERIFY:
_VERIFY_UPPER.klass(self)
##########################################################################
# Utilities.
##########################################################################
[docs]
@classmethod
def _make_data(cls, data: 'Data_IPv6') -> 'dict[str, Any]': # type: ignore[override]
"""Create key-value pairs from ``data`` for protocol construction.
Args:
data: protocol data
Returns:
Key-value pairs for protocol construction.
"""
# NOTE: Rebuilt from ``__exthdr__`` alone, never guessed from the keys of ``data``,
# whose order an ``Info`` does not promise and which differs on PyPy (#2003).
exthdr = data.get('__exthdr__') # type: Optional[tuple[tuple[Type[ProtocolBase], Any], ...]] # pylint: disable=line-too-long
chain = tuple(exthdr) if isinstance(exthdr, (tuple, list)) else ()
upper = cls._make_payload(data)
payload = upper # type: bytes | ProtocolBase
# NOTE: The extension headers sit between this header and the upper
# layer payload, so they are rebuilt in wire order ahead of it.
if chain:
payload = b''.join(proto.from_data(info).data for proto, info in chain) + upper.data
return {
'traffic_class': data['class'],
'flow_label': data.label,
'next': data.next,
'hop_limit': data.limit,
'src': data.src,
'dst': data.dst,
# NOTE: The parsed payload length, passed through rather than recomputed by ``make``,
# so a truncated capture rebuilds with the length that was on the wire (#1155).
'payload_length': data.payload,
'payload': payload,
# NOTE: And the octets captured past the Payload Length (#1209).
'trailer': data.get('trailer', b''),
}
[docs]
def _decode_next_layer(self, ipv6: 'Data_IPv6', proto: 'Optional[int]' = None, # type: ignore[override] # pylint: disable=arguments-differ,arguments-renamed
length: 'Optional[int]' = None, *, packet: 'Optional[dict[str, Any]]' = None) -> 'Data_IPv6': # pylint: disable=arguments-differ
"""Decode next layer extractor.
Arguments:
ipv6: info buffer
proto: next layer protocol index
length: valid (*not padding*) length
packet: packet info (passed from :meth:`self.unpack <pcapkit.protocols.protocol.Protocol.unpack>`)
Returns:
Current protocol with next layer extracted.
Raises:
ProtocolError: If the walk reaches a Fragment header in a packet
carrying a Jumbo Payload option, an error by
:rfc:`2675#section-3` (:issue:`1687`).
"""
#: Extension headers.
self._exthdr = OrderedMultiDict() # type: OrderedMultiDict[Enum_ExtensionHeader, ProtocolBase] # pylint: disable=attribute-defined-outside-init
hdr_len = self.length # header length
raw_len = ipv6.payload if length is None else length # payload length
_protos = [] # type: list[ProtocolBase] # ProtoChain buffer
_exthdr = [] # type: list[tuple[Type[ProtocolBase], Any]] # (parser class, info) per extension header
split = None # type: Optional[tuple[int, int]] # at first Fragment header
opaque = False # walk ended at a non-zero Fragment Offset
depth = frame_depth.get() # layers of the frame up to this one
# traverse if next header is an extension header
payload = self.__header__.get_payload()
while True:
try:
ex_proto = Enum_ExtensionHeader(proto)
except ValueError:
break
# NOTE: Past ``EXTENSION_HEADER_LIMIT`` headers, the rest is the upper layer, which
# ``_import_next_layer`` keeps raw with one warning (#1604).
if config.EXTENSION_HEADER_LIMIT != -1 and len(_protos) >= config.EXTENSION_HEADER_LIMIT:
break
# NOTE: Likewise past ``config.FRAME_LAYER_LIMIT`` layers of the frame (#1610), since
# each header walked is one of them.
if config.FRAME_LAYER_LIMIT != -1 and depth + len(_protos) >= config.FRAME_LAYER_LIMIT:
break
# # directly break when No Next Header occurs
# if proto.name == 'IPv6-NoNxt':
# proto = None
# break
# NOTE: The Payload Length left bounds this header, so with all of it present a longer
# header is malformed, not capture-cut, which its option area is told (#2694).
context = dict(packet or {}, __capture_cut__=len(payload) < raw_len)
# make protocol name
# NOTE: Counted after every header walked, as if each were dissected by the one before;
# ESP dissects its decrypted payload even when walked, and those layers count on.
token = frame_depth.set(depth + len(_protos))
try:
next_ = self._import_next_layer(proto, packet=context, version=6, extension=True,
payload=payload) # type: ignore[misc,call-arg]
finally:
frame_depth.reset(token)
info = next_.info
# A layer with no ``next`` field ends the walk, as RFC 8200 section 4
# has a node discard the packet on a malformed header (#2477).
# It is not recorded as an extension header.
# A Raw is reused as the upper layer, so the header is not parsed again.
if not hasattr(info, 'next'):
from pcapkit.protocols.misc.raw.protocol import Raw # isort: skip # pylint: disable=import-outside-toplevel
# Exact type on purpose: Raw subclasses exist, so isinstance would over-match.
if type(next_) is Raw: # pylint: disable=unidiomatic-typecheck
self._walk_upper = next_
break
# NOTE: Added through a ``MultiDict``, so a repeated header keeps its predecessor's
# info rather than overwriting it, and ``Info.to_dict`` writes every one (#1453).
name = self._exthdr_key(next_.alias)
ipv6.__update__(OrderedMultiDict([(name, info)]))
# record protocol name
# self._protos = ProtoChain(name, chain, alias)
_protos.append(next_)
_exthdr.append((type(next_), info))
# update header & payload length
hdr_len += next_.length # type: ignore[assignment]
raw_len -= next_.length
# keep record of extension headers
self._exthdr.add(ex_proto, next_)
# update payload for the next header -- either the next extension
# header, or the upper layer protocol should this be the last one;
# this must happen before any exit from the loop, since the payload
# is what gets handed to ``super()._decode_next_layer`` below
payload = payload[next_.length:]
proto = info.next
# keep original data after fragment header
if ex_proto == Enum_ExtensionHeader.IPv6_Frag:
# NOTE: The first Fragment header splits the packet for reassembly, so
# ``fragment``, ``hdr_len`` and ``raw_len`` keep that split as the walk goes on.
if split is None:
# NOTE: Nor may a jumbogram carry one (#1687): the frame stops as for the errors
# :meth:`read` checks; past those, its Jumbo Payload option is in the payload.
error = jumbogram_error(ipv6.payload, ipv6.next, self.__header__.get_payload(), lambda: True)
if error is not None:
raise ProtocolError(f'{self.alias}: {error}')
split = (hdr_len, raw_len)
ipv6.__update__({
'fragment': self._read_packet(header=hdr_len, payload=raw_len),
})
# NOTE: Only the first fragment goes on, as it carries every header through the
# upper layer (RFC 8200 Section 4.5) and is walked as if unfragmented (#1539).
if not self._is_first_fragment(info):
opaque = True
break
# record real header & payload length (headers exclude)
ipv6.__update__({
'hdr_len': hdr_len if split is None else split[0],
'raw_len': raw_len if split is None else split[1],
# update next header
'protocol': proto,
# extension header chain in wire order, with the parser of each,
# for ``from_data`` (#1473)
'__exthdr__': tuple(_exthdr),
})
# NOTE: A later fragment's slice is left ``Raw`` though ``protocol`` names the upper
# layer, so ``'UDP' in frame`` does not hold of it; reassembly uses ``fragment`` (#1545).
self._exthdr_depth = len(_protos) # the headers the upper layer follows
scope = contextlib.nullcontext() # type: ContextManager[None]
if config.CHECKSUM_VERIFY:
# NOTE: As in IPv4.read; an atomic fragment, offset zero and no more
# to come, is a whole packet (#2470).
from pcapkit.corekit.checksum.internet import fragment_payload # isort: skip # pylint: disable=import-outside-toplevel
frags = self._exthdr.getlist(Enum_ExtensionHeader.IPv6_Frag) # type: ignore[arg-type]
scope = fragment_payload(any(frag.info.get('offset') or frag.info.get('mf') for frag in frags))
token = frame_depth.set(depth + len(_protos))
try:
with scope:
return super()._decode_next_layer(ipv6, None if opaque else proto, raw_len, packet=packet,
ipv6_exthdr=ProtoChain.from_list(_protos), # type: ignore[arg-type]
payload=payload)
finally:
frame_depth.reset(token)
@staticmethod
def _exthdr_key(alias: 'str') -> 'str':
"""Key an extension header's info is written under in :class:`Data_IPv6`.
Args:
alias: :attr:`~pcapkit.protocols.protocol.Protocol.alias` of
the extension header's parser.
Returns:
The key; :meth:`str.lstrip` strips a character *set*, which every
parser's alias is chosen to suit.
"""
return alias.lstrip('IPv6-').lower()
@staticmethod
def _is_first_fragment(info: 'Mapping[str, Any]') -> 'bool':
"""Whether a Fragment header's info is that of a first fragment.
Args:
info: info of a Fragment header, as parsed or from :meth:`Info.to_dict
<pcapkit.corekit.infoclass.Info.to_dict>`
Returns:
:data:`True` at a Fragment Offset of zero.
"""
return info.get('offset') == 0
[docs]
def _read_ip_hextet(self) -> 'tuple[int, int, int]':
"""Read the first four octets of IPv6.
Returns:
Parsed data of those four octets: version number, traffic class
and flow label.
"""
_htet = self._read_fileng(4).hex()
_vers = int(_htet[0], base=16) # version number (6)
_tcls = int(_htet[1:3], base=16) # traffic class
_flow = int(_htet[3:], base=16) # flow label
return (_vers, _tcls, _flow)
def _read_ip_addr(self) -> 'IPv6Address':
"""Read IP address.
Returns:
Parsed IP address.
"""
return ipaddress.ip_address(self._read_fileng(16)) # type: ignore[return-value]
[docs]
@beholder # type: ignore[arg-type]
def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *, # pylint: disable=arguments-differ
packet: 'Optional[dict[str, Any]]' = None, version: 'Literal[4, 6]' = 4,
extension: 'bool' = False, payload: 'Optional[bytes]' = None) -> 'ProtocolBase':
"""Import next layer extractor.
Arguments:
proto: next layer protocol index
length: valid (*non-padding*) length
packet: packet info (passed from :meth:`self.unpack <pcapkit.protocols.protocol.Protocol.unpack>`)
version: IP protocol version
extension: if is extension header
payload: payload from packet. If not provided, will extract from
:meth:`self.__header__.get_payload <pcapkit.protocols.schema.Schema.get_payload>`
Returns:
Instance of next layer.
Notes:
A header whose dedicated parser raises reaches the
:func:`~pcapkit.corekit.decorators.beholder` decorating this
method, which substitutes plain
:class:`~pcapkit.protocols.misc.raw.protocol.Raw`. That carries no ``next``,
so :meth:`_decode_next_layer`'s walk stops there, as
:rfc:`8200#section-4` has a node discard such a packet
(:issue:`2477`). ``253`` and ``254`` have no dedicated parser and
reach plain :class:`~pcapkit.protocols.misc.raw.protocol.Raw` with no
exception involved, which stops the walk the same way. A ``Raw`` the
walk stopped on is then returned as the upper layer, not parsed
again.
:class:`~pcapkit.protocols.internet.esp.protocol.ESP`'s info always carries
a ``next``, :data:`None` unless its payload was decrypted, since
:rfc:`4303` encrypts the real value, so the walk ends after it.
The upper layer past a walk cut at :data:`EXTENSION_HEADER_LIMIT
<pcapkit.utilities.config.EXTENSION_HEADER_LIMIT>` is the
next extension header, and is kept raw with one
:exc:`~pcapkit.utilities.warnings.ProtocolWarning`, as in
:meth:`Internet._import_next_layer
<pcapkit.protocols.internet.internet.Internet._import_next_layer>`
(:issue:`1604`).
"""
upper, self._walk_upper = self._walk_upper, None
if upper is not None:
return upper
if payload is None:
file_ = self.__header__.get_payload()
else:
file_ = payload
if length is None:
length = len(file_)
protocol, extra = self._next_layer_class(proto, file_, length, extension)
return self._parse_next_layer(protocol, file_, length, version=version, extension=extension,
alias=proto, packet=packet, layer=self._exlayer,
protocol=self._exproto, __context__=self._exctx, **extra)