# -*- coding: utf-8 -*-
# mypy: disable-error-code=assignment
"""header schema for IPv6 destination options"""
import collections
from typing import TYPE_CHECKING
from pcapkit.const.ipv6.option import Option as Enum_Option
from pcapkit.const.ipv6.qs_function import QSFunction as Enum_QSFunction
from pcapkit.const.ipv6.router_alert import RouterAlert as Enum_RouterAlert
from pcapkit.const.ipv6.seed_id import SeedID as Enum_SeedID
from pcapkit.const.ipv6.smf_dpd_mode import SMFDPDMode as Enum_SMFDPDMode
from pcapkit.const.reg.transtype import TransType as Enum_TransType
from pcapkit.corekit.fields.collections import OptionField
from pcapkit.corekit.fields.ipaddress import IPv6AddressField
from pcapkit.corekit.fields.misc import (ConditionalField, ForwardMatchField, PayloadField,
SwitchField)
from pcapkit.corekit.fields.numbers import (EnumField, NumberField, UInt8Field, UInt16Field,
UInt32Field)
from pcapkit.corekit.fields.strings import BitField, BytesField, PaddingField
from pcapkit.corekit.sentinels import NO_VALUE
from pcapkit.protocols.schema.internet import ipv6_option
from pcapkit.protocols.schema.internet.ipv6_option import pad_opt_data_len, rpl_opt_sub_tlv_len
from pcapkit.protocols.schema.schema import EnumSchema, Schema, schema_final
from pcapkit.utilities.logging import SPHINX_TYPE_CHECKING
__all__ = [
'IPv6_Opts',
'SMFDPDOption', 'QuickStartOption',
'UnassignedOption', 'PadOption', 'TunnelEncapsulationLimitOption',
'RouterAlertOption', 'CALIPSOOption', 'SMFIdentificationBasedDPDOption',
'SMFHashBasedDPDOption', 'PDMOption', 'QuickStartRequestOption',
'QuickStartReportOption', 'RPLOption', 'MPLOption', 'ILNPOption',
'LineIdentificationOption', 'JumboPayloadOption', 'HomeAddressOption',
'IPDFFOption',
]
if TYPE_CHECKING:
from ipaddress import IPv4Address, IPv6Address
from typing import Any, DefaultDict, Optional, Type
from pcapkit.corekit.fields.field import FieldBase as Field
from pcapkit.protocols.protocol import ProtocolBase
if SPHINX_TYPE_CHECKING: # pragma: no cover
from typing_extensions import TypedDict
[docs]
class TaggerIDInfo(TypedDict):
"""TaggerID information."""
#: SMF mode.
mode: int
#: TaggerID type.
type: int
#: TaggerID length.
len: int
[docs]
class QuickStartFlags(TypedDict):
"""Quick-Start flags."""
#: QS function.
func: int
#: Rate request/report.
rate: int
[docs]
class RPLFlags(TypedDict):
"""RPL flags."""
#: Down flag.
down: int
#: Rank error flag.
rank_err: int
#: Forwarding error flag.
fwd_err: int
#: Reserved flag bits.
reserved: int
[docs]
class MPLFlags(TypedDict):
"""MPL flags."""
#: Seed-ID type. Identifies the length of the
#: Seed-ID.
type: int
#: Max flag. ``1`` indicates that the value in the
#: sequence field is known to be the largest sequence
#: number that was received from the MPL Seed.
max: int
#: Verification flag. ``0`` indicates that the MPL Option
#: conforms to this specification.
drop: int
#: Reserved bits (``rsv``).
reserved: int
[docs]
class DFFFlags(TypedDict):
"""``IP_DFF`` flags."""
#: Version.
ver: int
#: Duplicate flag.
dup: int
#: Return flag.
ret: int
#: Reserved bits.
reserved: int
[docs]
class SMFDPDTestFlag(TypedDict):
"""``SMF_DPD`` test flag."""
#: Length.
len: int
#: DPD mode.
mode: int
[docs]
class QSTestFlags(TypedDict):
"""Quick start test flag."""
#: Option length.
len: int
#: QS function.
func: int
[docs]
class QSNonce(TypedDict):
"""Quick start nonce."""
#: Nonce.
nonce: int
#: Reserved bits (``R``).
reserved: int
# NOTE: The option helpers are shared with HOPOPT and live in
# :mod:`pcapkit.protocols.schema.internet.ipv6_option`. The functions below bind
# what differs between the two headers: the ``IPv6-Opts`` error prefix, and the
# option schemas of this module.
[docs]
def mpl_opt_seed_id_len(pkt: 'dict[str, Any]') -> 'int':
"""Return MPL Seed-ID length.
See :func:`pcapkit.protocols.schema.internet.ipv6_option.mpl_opt_seed_id_len`;
errors are reported as ``IPv6-Opts``.
"""
return ipv6_option.mpl_opt_seed_id_len(pkt, prefix='IPv6-Opts')
[docs]
def smf_i_dpd_id_len(pkt: 'dict[str, Any]') -> 'int':
"""Return SMF I-DPD identifier length.
See :func:`pcapkit.protocols.schema.internet.ipv6_option.smf_i_dpd_id_len`;
errors are reported as ``IPv6-Opts``.
"""
return ipv6_option.smf_i_dpd_id_len(pkt, prefix='IPv6-Opts')
[docs]
def smf_dpd_data_selector(pkt: 'dict[str, Any]') -> 'Field':
"""Selector function for :attr:`_SMFDPDOption.data` field.
See :func:`pcapkit.protocols.schema.internet.ipv6_option.smf_dpd_data_selector`,
choosing among the schemas registered with :class:`SMFDPDOption`; errors
are reported as ``IPv6-Opts``.
"""
return ipv6_option.smf_dpd_data_selector(pkt, prefix='IPv6-Opts', base=SMFDPDOption)
[docs]
def smf_i_dpd_tid_selector(pkt: 'dict[str, Any]') -> 'Field':
"""Selector function for :attr:`SMFIdentificationBasedDPDOption.tid` field.
See :func:`pcapkit.protocols.schema.internet.ipv6_option.smf_i_dpd_tid_selector`;
errors are reported as ``IPv6-Opts``.
"""
return ipv6_option.smf_i_dpd_tid_selector(pkt, prefix='IPv6-Opts')
[docs]
def quick_start_data_selector(pkt: 'dict[str, Any]') -> 'Field':
"""Selector function for :attr:`_QuickStartOption.data` field.
See :func:`pcapkit.protocols.schema.internet.ipv6_option.quick_start_data_selector`,
choosing among the schemas registered with :class:`QuickStartOption`, and
:class:`UnassignedOption` for an unassigned QS function.
"""
return ipv6_option.quick_start_data_selector(pkt, base=QuickStartOption,
unassigned=UnassignedOption)
[docs]
def calipso_pad_len(pkt: 'dict[str, Any]') -> 'int':
"""Return CALIPSO option padding length.
See :func:`pcapkit.protocols.schema.internet.ipv6_option.calipso_pad_len`;
errors are reported as ``IPv6-Opts``.
"""
return ipv6_option.calipso_pad_len(pkt, prefix='IPv6-Opts')
[docs]
def mpl_opt_pad_len(pkt: 'dict[str, Any]') -> 'int':
"""Return MPL option padding length.
See :func:`pcapkit.protocols.schema.internet.ipv6_option.mpl_opt_pad_len`;
errors are reported as ``IPv6-Opts``.
"""
return ipv6_option.mpl_opt_pad_len(pkt, prefix='IPv6-Opts')
[docs]
class Option(EnumSchema[Enum_Option]):
"""Header schema for IPv6-Opts options."""
__default__ = lambda: UnassignedOption
#: Option type.
type: 'Enum_Option' = EnumField(length=1, namespace=Enum_Option)
#: Option length (conditional in case of ``Pad1`` option).
len: 'int' = ConditionalField(
UInt8Field(default=0),
lambda pkt: pkt['type'] != Enum_Option.Pad1,
)
[docs]
def post_process(self, packet: 'dict[str, Any]') -> 'Schema':
"""Revise ``schema`` data after unpacking process.
Args:
packet: Unpacked data.
Returns:
Revised schema.
"""
# for Pad1 option, length is always 0
if self.type == Enum_Option.Pad1:
self.len = 0
return self
[docs]
@schema_final
class UnassignedOption(Option):
"""Header schema for IPv6-Opts unassigned options."""
#: Option data.
data: 'bytes' = BytesField(length=lambda pkt: pkt['len'])
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Option', len: 'int', data: 'bytes') -> 'None': ...
[docs]
@schema_final
class PadOption(Option, code=[Enum_Option.Pad1,
Enum_Option.PadN]):
"""Header schema for IPv6-Opts padding options.
The two padding options do **not** share a wire shape: a ``Pad1`` option is
a lone type octet, whereas a ``PadN`` option is a type octet, an
``Opt Data Len`` octet, and that many padding octets [:rfc:`8200#section-4.2`].
Both the ``Opt Data Len`` octet (:attr:`Option.len`) and the padding data
(:attr:`self.pad <PadOption.pad>`) are therefore sized from the option type,
so that a ``Pad1`` option consumes exactly one octet.
"""
#: Padding.
pad: 'bytes' = PaddingField(length=pad_opt_data_len)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Option', len: 'int', pad: 'bytes') -> 'None': ...
[docs]
@schema_final
class TunnelEncapsulationLimitOption(Option, code=Enum_Option.Tunnel_Encapsulation_Limit):
"""Header schema for IPv6-Opts tunnel encapsulation limit options."""
#: Tunnel encapsulation limit.
limit: 'int' = UInt8Field()
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Option', len: 'int', limit: 'int') -> 'None': ...
[docs]
@schema_final
class RouterAlertOption(Option, code=Enum_Option.Router_Alert):
"""Header schema for IPv6-Opts router alert options."""
#: Router alert.
alert: 'Enum_RouterAlert' = EnumField(length=2, namespace=Enum_RouterAlert)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Option', len: 'int', alert: 'Enum_RouterAlert') -> 'None': ...
[docs]
@schema_final
class CALIPSOOption(Option, code=Enum_Option.CALIPSO):
"""Header schema for IPv6-Opts common architecture label IPv6 security options."""
#: CALIPSO domain of interpretation.
domain: 'int' = UInt32Field()
#: Compartment length.
cmpt_len: 'int' = UInt8Field()
#: Sens level.
level: 'int' = UInt8Field()
#: Checksum (CRC-16).
checksum: 'bytes' = BytesField(length=2)
#: Compartment bitmap.
bitmap: 'bytes' = ConditionalField(
BytesField(length=lambda pkt: pkt['cmpt_len'] * 4),
lambda pkt: pkt['cmpt_len'] > 0,
)
#: Octets that ``Opt Data Len`` covers past the compartment bitmap, kept as
#: read rather than packed as zeros.
pad: 'bytes' = BytesField(length=calipso_pad_len, default=b'')
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Option', len: 'int', domain: 'int', cmpt_len: 'int',
level: 'int', checksum: 'bytes', bitmap: 'Optional[bytes]', pad: 'bytes' = ...) -> 'None': ...
[docs]
@schema_final
class _SMFDPDOption(Schema):
"""Header schema for IPv6-Opts SMF DPD options with generic representation.
The ``test`` field forward-matches the first three octets of the option --
``Option Type``, ``Opt Data Len`` and the octet carrying the DPD mode bit --
without consuming them, so that :func:`smf_dpd_data_selector` can size and
choose the schema which then reads the option properly. Its ``namespace``
offsets are therefore bit offsets into the *option*, not into any one field
of it: ``Opt Data Len`` is octet 1, i.e. bits 8 to 15, and the mode bit is
the first bit of octet 2 [:rfc:`6621#section-8.1`].
"""
#: SMF DPD mode.
test: 'SMFDPDTestFlag' = ForwardMatchField(BitField(length=3, namespace={
'len': (8, 8),
'mode': (16, 1),
}))
#: SMF DPD data.
data: 'SMFIdentificationBasedDPDOption | SMFHashBasedDPDOption' = SwitchField(
selector=smf_dpd_data_selector,
)
[docs]
def post_process(self, packet: 'dict[str, Any]') -> 'SMFDPDOption':
"""Revise ``schema`` data after unpacking process.
Args:
packet: Unpacked data.
Returns:
Revised schema.
"""
ret = self.data
return ret
# register ``_SMFDPDOption`` as ``SMF_DPD`` option
Option.register(Enum_Option.SMF_DPD, _SMFDPDOption)
[docs]
class SMFDPDOption(Option, EnumSchema[Enum_SMFDPDMode]):
"""Header schema for IPv6-Opts simplified multicast forwarding duplicate packet
detection (``SMF_DPD``) options."""
__enum__: 'DefaultDict[Enum_SMFDPDMode, Type[SMFDPDOption]]' = collections.defaultdict(lambda: None) # type: ignore[arg-type,return-value]
if TYPE_CHECKING:
mode: 'Enum_SMFDPDMode'
[docs]
@schema_final
class SMFIdentificationBasedDPDOption(SMFDPDOption, code=Enum_SMFDPDMode.I_DPD):
"""Header schema for IPv6-Opts SMF identification-based DPD options.
.. versionremoved:: 1.5.0
``test``, a field the HOPOPT schema lacks, so that the same octets
parse alike under both headers (:issue:`441`).
"""
#: TaggerID information.
info: 'TaggerIDInfo' = BitField(length=1, namespace={
'mode': (0, 1),
'type': (1, 3),
'len': (4, 4),
})
#: TaggerID.
tid: 'bytes | IPv4Address | IPv6Address' = ConditionalField(
SwitchField(selector=smf_i_dpd_tid_selector),
lambda pkt: pkt['info']['type'] != 0,
)
#: Identifier.
id: 'bytes' = BytesField(length=smf_i_dpd_id_len)
[docs]
def post_process(self, packet: 'dict[str, Any]') -> 'SMFIdentificationBasedDPDOption':
"""Revise ``schema`` data after unpacking process.
Args:
packet: Unpacked data.
Returns:
Revised schema.
"""
ret = super().post_process(packet) # type: SMFIdentificationBasedDPDOption
ret.mode = Enum_SMFDPDMode.I_DPD
return ret
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Option', len: 'int', info: 'TaggerIDInfo',
tid: 'Optional[bytes]', id: 'bytes') -> 'None': ...
[docs]
@schema_final
class SMFHashBasedDPDOption(SMFDPDOption, code=Enum_SMFDPDMode.H_DPD):
"""Header schema for IPv6-Opts SMF hash-based DPD options."""
#: Hash assist value (HAV).
hav: 'bytes' = BytesField(length=lambda pkt: pkt['len'])
[docs]
def post_process(self, packet: 'dict[str, Any]') -> 'SMFIdentificationBasedDPDOption':
"""Revise ``schema`` data after unpacking process.
Args:
packet: Unpacked data.
Returns:
Revised schema.
"""
ret = super().post_process(packet) # type: SMFIdentificationBasedDPDOption
ret.mode = Enum_SMFDPDMode.H_DPD
return ret
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Option', len: 'int', hav: 'bytes') -> 'None': ...
[docs]
@schema_final
class PDMOption(Option, code=Enum_Option.PDM):
"""Header schema for IPv6-Opts performance and diagnostic metrics (PDM) options."""
#: Scale delta time last received (DTLR).
scaledtlr: 'int' = UInt8Field()
#: Scale delta time last sent (DTLS).
scaledtls: 'int' = UInt8Field()
#: Packet sequence number (PSN) this packet.
psntp: 'int' = UInt16Field()
#: Packet sequence number (PSN) last received.
psnlr: 'int' = UInt16Field()
#: Delta time last received (DTLR).
deltatlr: 'int' = UInt16Field()
#: Delta time last sent (DTLS).
deltatls: 'int' = UInt16Field()
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Option', len: 'int', scaledtlr: 'int', scaledtls: 'int',
psntp: 'int', psnlr: 'int', deltatlr: 'int', deltatls: 'int') -> 'None': ...
[docs]
@schema_final
class _QuickStartOption(Schema):
"""Header schema for IPv6-Opts quick start options in generic representation."""
#: Flags.
flags: 'QSTestFlags' = ForwardMatchField(BitField(length=3, namespace={
'len': (8, 8),
'func': (16, 4),
}))
#: QS data.
data: 'QuickStartRequestOption | QuickStartReportOption' = SwitchField(
selector=quick_start_data_selector,
)
[docs]
def post_process(self, packet: 'dict[str, Any]') -> 'QuickStartOption':
"""Revise ``schema`` data after unpacking process.
Args:
packet: Unpacked data.
Returns:
Revised schema.
"""
ret = self.data
if isinstance(ret, QuickStartOption):
ret.func = Enum_QSFunction.get(self.flags['func'])
return ret
# register ``_QuickStartOption`` as ``Quick_Start`` option
Option.register(Enum_Option.Quick_Start, _QuickStartOption)
[docs]
class QuickStartOption(Option, EnumSchema[Enum_QSFunction]):
"""Header schema for IPv6-Opts quick start options."""
# NOTE: This declaration scopes the QS-function registry to this class. Without
# it the subclasses below register into the *parent* ``Option`` registry, whose
# keys are IPv6-Opts option *types* -- and since ``Quick_Start_Request`` is 0
# and ``Report_of_Approved_Rate`` is 8, they would clobber option types 0
# (``Pad1``) and 8 (``SMF_DPD``). ``hopopt.py`` carries the same declaration.
__enum__: 'DefaultDict[Enum_QSFunction, Type[QuickStartOption]]' = collections.defaultdict(lambda: None) # type: ignore[arg-type,return-value]
#: Flags.
flags: 'QuickStartFlags' = BitField(length=1, namespace={
'func': (0, 4),
'rate': (4, 4),
})
if TYPE_CHECKING:
func: 'Enum_QSFunction'
[docs]
@schema_final
class QuickStartRequestOption(QuickStartOption, code=Enum_QSFunction.Quick_Start_Request):
"""Header schema for IPv6-Opts quick start request options."""
#: QS time-to-live (TTL).
ttl: 'int' = UInt8Field()
#: QS nonce.
nonce: 'QSNonce' = BitField(length=4, namespace={
'nonce': (0, 30),
'reserved': (30, 2),
})
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Option', len: 'int', flags: 'QuickStartFlags',
ttl: 'int', nonce: 'QSNonce') -> 'None': ...
[docs]
@schema_final
class QuickStartReportOption(QuickStartOption, code=Enum_QSFunction.Report_of_Approved_Rate):
"""Header schema for IPv6-Opts quick start report of approved rate options."""
#: Reserved.
reserved: 'bytes' = PaddingField(length=1)
#: QS nonce.
nonce: 'QSNonce' = BitField(length=4, namespace={
'nonce': (0, 30),
'reserved': (30, 2),
})
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Option', len: 'int', flags: 'QuickStartFlags',
reserved: 'bytes', nonce: 'QSNonce') -> 'None': ...
[docs]
@schema_final
class RPLOption(Option, code=[Enum_Option.RPL_Option_0x23,
Enum_Option.RPL_Option_0x63]):
"""Header schema for IPv6-Opts routing protocol for low-power and lossy networks (RPL) options."""
#: Flags.
flags: 'RPLFlags' = BitField(length=1, namespace={
'down': (0, 1),
'rank_err': (1, 1),
'fwd_err': (2, 1),
'reserved': (3, 5),
})
#: RPL instance ID.
id: 'int' = UInt8Field()
#: Sender rank.
rank: 'int' = UInt16Field()
#: Sub-TLVs, kept as raw octets.
sub_tlvs: 'bytes' = BytesField(length=rpl_opt_sub_tlv_len)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Option', len: 'int', flags: 'RPLFlags', id: 'int',
rank: 'int', sub_tlvs: 'bytes') -> 'None': ...
[docs]
@schema_final
class MPLOption(Option, code=Enum_Option.MPL_Option):
"""Header schema for IPv6-Opts multicast protocol for low-power and lossy networks (MPL) options."""
#: Flags.
flags: 'MPLFlags' = BitField(length=1, namespace={
'type': (0, 2),
'max': (2, 1),
'drop': (3, 1),
'reserved': (4, 4),
})
#: MPL sequence number.
seq: 'int' = UInt8Field()
#: MPL Seed-ID.
seed: 'int | IPv6Address' = ConditionalField(
NumberField(length=mpl_opt_seed_id_len, signed=False),
lambda pkt: pkt['flags']['type'] != Enum_SeedID.IPV6_SOURCE_ADDRESS,
)
#: Reserved data (padding).
pad: 'bytes' = PaddingField(length=mpl_opt_pad_len)
[docs]
def post_process(self, packet: 'dict[str, Any]') -> 'Schema':
"""Revise ``schema`` data after unpacking process.
Args:
packet: Unpacked data.
Returns:
Revised schema.
"""
if self.flags['type'] == Enum_SeedID.IPV6_SOURCE_ADDRESS:
self.seed = packet.get('src', NO_VALUE)
return self
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Option', len: 'int', flags: 'MPLFlags', seq: 'int',
seed: 'Optional[int | IPv6Address]') -> 'None': ...
[docs]
@schema_final
class ILNPOption(Option, code=Enum_Option.ILNP_Nonce):
"""Header schema for IPv6-Opts identifier-locator network protocol (ILNP) options."""
#: Nonce value.
nonce: 'int' = NumberField(length=lambda pkt: pkt['len'], signed=False)
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Option', len: 'int', nonce: 'int') -> 'None': ...
[docs]
@schema_final
class LineIdentificationOption(Option, code=Enum_Option.Line_Identification_Option):
"""Header schema for IPv6-Opts line-identification options."""
#: Line ID length.
id_len: 'int' = UInt8Field()
#: Line ID.
id: 'bytes' = BytesField(length=lambda pkt: pkt['id_len'])
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Option', len: 'int', id_len: 'int', id: 'bytes') -> 'None': ...
[docs]
@schema_final
class JumboPayloadOption(Option, code=Enum_Option.Jumbo_Payload):
"""Header schema for IPv6-Opts jumbo payload options."""
#: Jumbo payload length.
jumbo_len: 'int' = UInt32Field()
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Option', len: 'int', jumbo_len: 'int') -> 'None': ...
[docs]
@schema_final
class HomeAddressOption(Option, code=Enum_Option.Home_Address):
"""Header schema for IPv6-Opts home address options."""
#: Home address.
addr: 'IPv6Address' = IPv6AddressField()
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Option', len: 'int', addr: 'IPv6Address | int | str | bytes') -> 'None': ...
[docs]
@schema_final
class IPDFFOption(Option, code=Enum_Option.IP_DFF):
"""Header schema for IPv6-Opts depth-first forwarding (``IP_DFF``) options."""
#: Flags.
flags: 'DFFFlags' = BitField(length=1, namespace={
'ver': (0, 2),
'dup': (2, 1),
'ret': (3, 1),
'reserved': (4, 4),
})
#: Sequence number.
seq: 'int' = UInt16Field()
if TYPE_CHECKING:
def __init__(self, type: 'Enum_Option', len: 'int', flags: 'DFFFlags', seq: 'int') -> 'None': ...
[docs]
@schema_final
class IPv6_Opts(Schema):
"""Header schema for IPv6-Opts packet."""
#: Next header.
next: 'Enum_TransType' = EnumField(length=1, namespace=Enum_TransType)
#: Header length.
len: 'int' = UInt8Field()
#: Options.
options: 'list[Option]' = OptionField(
length=lambda pkt: pkt['len'] * 8 + 6,
base_schema=Option,
type_name='type',
registry=Option.registry,
)
#: Payload.
payload: 'bytes' = PayloadField()
if TYPE_CHECKING:
def __init__(self, next: 'Enum_TransType', len: 'int',
options: 'bytes | list[bytes | Option]',
payload: 'bytes | ProtocolBase | Schema') -> 'None': ...