# -*- coding: utf-8 -*-
"""IPv6-Frag - Fragment Header for IPv6
==========================================
.. module:: pcapkit.protocols.internet.ipv6_frag.protocol
:mod:`pcapkit.protocols.internet.ipv6_frag.protocol` contains
:class:`~pcapkit.protocols.internet.ipv6_frag.protocol.IPv6_Frag`
only, which implements the extractor for Fragment Header for
IPv6 (IPv6-Frag) [*]_, whose structure is described as
below:
======= ========= ========================= =======================
Octets Bits Name Description
======= ========= ========================= =======================
0 0 ``frag.next`` Next Header
1 8 ``frag.reserved_octet`` Reserved
2 16 ``frag.offset`` Fragment Offset
3 29 ``frag.reserved`` Reserved
3 31 ``frag.mf`` More Flag
4 32 ``frag.id`` Identification
======= ========= ========================= =======================
.. [*] https://en.wikipedia.org/wiki/IPv6_packet#Fragment
"""
from typing import TYPE_CHECKING, overload
from pcapkit.const.reg.transtype import TransType as Enum_TransType
from pcapkit.protocols.internet.ipv6_ext.protocol import IPv6_Ext
from pcapkit.protocols.internet.ipv6_frag.data import IPv6_Frag as Data_IPv6_Frag
from pcapkit.protocols.internet.ipv6_frag.schema import IPv6_Frag as Schema_IPv6_Frag
from pcapkit.protocols.misc.no_payload.protocol import NoPayload
from pcapkit.utilities.exceptions import UnsupportedCall
if TYPE_CHECKING:
from typing import IO, Any, NoReturn, Optional
from typing_extensions import Literal
from pcapkit.corekit.protochain import ProtoChain
from pcapkit.protocols.protocol import ProtocolBase
from pcapkit.protocols.schema import Schema
__all__ = ['IPv6_Frag']
[docs]
class IPv6_Frag(IPv6_Ext[Data_IPv6_Frag, Schema_IPv6_Frag],
schema=Schema_IPv6_Frag, data=Data_IPv6_Frag):
"""This class implements Fragment Header for IPv6.
.. versionchanged:: 1.5.0
A rebuild keeps the reserved bits (:issue:`1235`).
.. versionchanged:: 1.6.0
``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`).
:meth:`make` raises :exc:`~pcapkit.utilities.exceptions.ProtocolError`
for a ``reserved_octet`` that is not one octet, rather than padding or
cutting it (:issue:`2676`).
.. versionremoved:: 1.6.0
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`).
"""
##########################################################################
# Properties.
##########################################################################
@property
def name(self) -> 'Literal["Fragment Header for IPv6"]':
"""Name of current protocol."""
return 'Fragment Header for IPv6'
@property
def alias(self) -> 'Literal["IPv6-Frag"]':
"""Acronym of corresponding protocol."""
return 'IPv6-Frag'
@property
def length(self) -> 'Literal[8]':
"""Header length of current protocol."""
return 8
@property
def payload(self) -> 'ProtocolBase | NoReturn':
"""Payload of current instance.
Raises:
UnsupportedCall: if the protocol is used as an IPv6 extension header
"""
if self._extf:
raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute 'payload'")
return self._next
@property
def protocol(self) -> 'Optional[str] | NoReturn':
"""Name of next layer protocol (if any).
Raises:
UnsupportedCall: if the protocol is used as an IPv6 extension header
"""
if self._extf:
raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute 'protocol'")
return super().protocol
@property
def protochain(self) -> 'ProtoChain | NoReturn':
"""Protocol chain of current instance.
Raises:
UnsupportedCall: if the protocol is used as an IPv6 extension header
"""
if self._extf:
raise UnsupportedCall(f"'{self.__class__.__name__}' object has no attribute 'protochain'")
return super().protochain
##########################################################################
# Methods.
##########################################################################
[docs]
def read(self, length: 'Optional[int]' = None, *, extension: 'bool' = False, # pylint: disable=arguments-differ
**kwargs: 'Any') -> 'Data_IPv6_Frag': # pylint: disable=unused-argument
"""Read Fragment Header for IPv6.
Structure of IPv6-Frag header [:rfc:`8200`]:
.. code-block:: text
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Next Header | Reserved | Fragment Offset |Res|M|
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Identification |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Args:
length: Length of packet data.
extension: If the packet is used as an IPv6 extension header.
**kwargs: Arbitrary keyword arguments.
Returns:
Parsed packet data.
"""
if length is None:
length = len(self)
schema = self.__header__
# NOTE: Scaled from 8-octet units to octets, as ``IPv4.read`` does, so reassembly can
# index the datagram buffer with ``offset`` directly.
ipv6_frag = Data_IPv6_Frag(
next=schema.next,
reserved_octet=schema.reserved,
offset=int(schema.flags['offset']) * 8,
reserved=schema.flags['reserved'],
mf=bool(schema.flags['mf']),
id=schema.id,
)
if extension:
# NOTE: As an IPv6 extension header the upper layer belongs to the enclosing
# IPv6 header, so hold NoPayload here so that membership does not raise (#2487).
self._next = NoPayload() # pylint: disable=attribute-defined-outside-init
return ipv6_frag
return self._decode_next_layer(ipv6_frag, schema.next, length - self.length)
[docs]
def make(self,
next: 'Enum_TransType | int' = Enum_TransType.UDP,
reserved_octet: 'bytes' = b'\x00',
offset: 'int' = 0,
reserved: 'int' = 0,
mf: 'bool' = False,
id: 'int' = 0,
payload: 'bytes | ProtocolBase | Schema' = b'',
**kwargs: 'Any') -> 'Schema_IPv6_Frag':
"""Make (construct) packet data.
Args:
next: Next header.
reserved_octet: The reserved octet after the next header.
offset: Fragment offset, in on-wire 8-octet units (:rfc:`8200#section-4.5`).
Note that the data model's :attr:`offset
<pcapkit.protocols.internet.ipv6_frag.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.
**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, or ``reserved_octet``
is not one octet.
.. versionchanged:: 1.5.0
Takes ``reserved_octet`` ahead of ``offset`` (:issue:`1223`) and
``reserved`` ahead of ``mf`` (:issue:`1235`), which shifts every
later positional argument.
"""
next_val = self._make_index(next, Enum_TransType, bits=8, name='next header')
return Schema_IPv6_Frag(
next=next_val,
reserved=self._make_octets(reserved_octet, size=1, name='reserved octet'),
flags={
'offset': offset,
'reserved': reserved,
'mf': mf,
},
id=id,
payload=payload,
)
[docs]
@classmethod
def id(cls) -> 'tuple[Literal["IPv6_Frag"]]': # type: ignore[override]
"""Index ID of the protocol.
Returns:
Index ID of the protocol.
"""
return ('IPv6_Frag',)
##########################################################################
# Data models.
##########################################################################
@overload
def __post_init__(self, file: 'IO[bytes] | bytes', length: 'Optional[int]' = ..., *, # pylint: disable=arguments-differ
extension: 'bool' = ..., **kwargs: 'Any') -> 'None': ...
@overload
def __post_init__(self, **kwargs: 'Any') -> 'None': ... # pylint: disable=arguments-differ
[docs]
def __post_init__(self, file: 'Optional[IO[bytes] | bytes]' = None, length: 'Optional[int]' = None, *, # pylint: disable=arguments-differ
extension: 'bool' = False, **kwargs: 'Any') -> 'None':
"""Post initialisation hook.
Args:
file: Source packet stream.
length: Length of packet data.
extension: If the protocol is used as an IPv6 extension header.
**kwargs: Arbitrary keyword arguments.
See Also:
For construction arguments, please refer to :meth:`self.make <IPv6_Frag.make>`.
"""
#: bool: If the protocol is used as an IPv6 extension header.
self._extf = extension
# call super __post_init__
super().__post_init__(file, length, extension=extension, **kwargs) # type: ignore[arg-type]
def __length_hint__(self) -> 'Literal[8]':
"""Return an estimated length for the object."""
return 8
[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_Frag # type: ignore[return-value]
##########################################################################
# Utilities.
##########################################################################
[docs]
@classmethod
def _make_data(cls, data: 'Data_IPv6_Frag') -> '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: A capture cut before the octet read none of it (#1458).
reserved_octet = cls._fill_short_read(getattr(data, 'reserved_octet', b'\x00'),
size=1, model=data, field='reserved')
# NOTE: ``make`` takes ``offset`` in 8-octet units but ``data.offset`` is in octets (see
# :meth:`read`), so scale it down for an exact data-to-schema round trip.
return {
'next': data.next,
'reserved_octet': reserved_octet,
'offset': data.offset // 8,
'reserved': data.reserved,
'mf': data.mf,
'id': data.id,
'payload': cls._make_payload(data),
}