# -*- coding: utf-8 -*-
# mypy: disable-error-code=assignment
"""header schema for generically-parsed IPv6 extension headers
:mod:`pcapkit.protocols.internet.ipv6_ext.schema` holds the field callbacks
that size and select the IPv6 options [:rfc:`8200#section-4.2`] carried by
both the Hop-by-Hop Options header
(:mod:`~pcapkit.protocols.internet.hopopt.schema`) and the Destination Options
header (:mod:`~pcapkit.protocols.internet.ipv6_opts.schema`).
Each of those modules declares its own option schemas, so the helpers take
what differs between them as parameters: ``prefix`` is the name of the header,
which leads each error message (``HOPOPT`` or ``IPv6-Opts``), and the selectors
take the option schemas they choose from. Each module binds the parameters in
a function of the same name, which is what its fields use.
"""
from typing import TYPE_CHECKING
from pcapkit.const.ipv6.qs_function import QSFunction as Enum_QSFunction
from pcapkit.const.ipv6.smf_dpd_mode import SMFDPDMode as Enum_SMFDPDMode
from pcapkit.const.ipv6.tagger_id import TaggerID as Enum_TaggerID
from pcapkit.const.reg.transtype import TransType as Enum_TransType
from pcapkit.corekit.fields.ipaddress import IPv4AddressField, IPv6AddressField
from pcapkit.corekit.fields.misc import NoValueField, PayloadField, SchemaField
from pcapkit.corekit.fields.numbers import EnumField, UInt8Field
from pcapkit.corekit.fields.strings import BytesField
from pcapkit.protocols.schema import Schema, schema_final
from pcapkit.utilities.exceptions import FieldValueError, ProtocolError
__all__ = [
'IPv6_Ext', 'rpl_opt_sub_tlv_len', 'mpl_opt_seed_id_len', 'smf_i_dpd_id_len',
'smf_dpd_data_selector', 'smf_i_dpd_tid_selector', 'quick_start_data_selector',
'pad_opt_data_len', 'calipso_pad_len', 'mpl_opt_pad_len'
]
if TYPE_CHECKING:
from typing import Any, Optional, Type
from pcapkit.corekit.fields.field import FieldBase as Field
from pcapkit.protocols.protocol import ProtocolBase
from pcapkit.protocols.schema import EnumMeta
[docs]
@schema_final
class IPv6_Ext(Schema):
"""Header schema for a generically-parsed IPv6 extension header.
Only the two octets :rfc:`6564#section-4` guarantees are parsed at this
layer -- ``next`` and the raw ``Hdr Ext Len`` octet. Combining them into
an actual skip distance is per protocol (a constant for ``IPv6-Frag``,
4-octet units for ``AH``, 8-octet units for the rest), so that part is
done in :meth:`pcapkit.protocols.internet.ipv6_ext.protocol.IPv6_Ext.read`,
which knows which protocol this instance stands in for; this schema does
not.
"""
#: Next header.
next: 'Enum_TransType' = EnumField(length=1, namespace=Enum_TransType)
#: Raw ``Hdr Ext Len`` octet -- see the class docstring for why its
#: interpretation is not fixed here.
len: 'int' = UInt8Field()
#: Everything after the two fixed octets; opaque at this layer.
payload: 'bytes' = PayloadField()
if TYPE_CHECKING:
def __init__(self, next: 'Enum_TransType | int', len: 'int',
payload: 'bytes | ProtocolBase | Schema') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements
def _opt_data_len_cut(pkt: 'dict[str, Any]') -> 'bool':
"""Whether the data ends before an option's ``Opt Data Len``.
Args:
pkt: Option unpacked schema.
Returns:
Whether the field :meth:`Schema.unpack
<pcapkit.protocols.schema.Schema.unpack>` gives as the one the
data ends inside is ``len``, so that its value is the zero a capture cut
leaves rather than a length read (:issue:`2727`).
"""
record = pkt.get('__short_read__')
return record is not None and record[0] == 'len'
[docs]
def rpl_opt_sub_tlv_len(pkt: 'dict[str, Any]') -> 'int':
"""Return RPL option sub-TLV length.
Args:
pkt: RPL option unpacked schema.
Returns:
RPL option sub-TLV length, i.e. ``Opt Data Len`` less the four octets
of the fixed fields. An ``Opt Data Len`` below ``4`` gives ``0``, and
the reader then rejects the option.
"""
return max(pkt['len'] - 4, 0)
[docs]
def mpl_opt_seed_id_len(pkt: 'dict[str, Any]', *, prefix: 'str') -> 'int':
"""Return MPL Seed-ID length.
Args:
pkt: MPL option unpacked schema.
prefix: Name of the extension header, leading the error message.
Returns:
MPL Seed-ID length.
Raises:
FieldValueError: If ``flags.type`` is not a defined Seed-ID type.
"""
s_type = pkt['flags']['type']
if s_type == 0:
return 0
if s_type == 1:
return 2
if s_type == 2:
return 8
if s_type == 3:
return 16
raise FieldValueError(f'{prefix}: invalid MPL Seed-ID type: {s_type}')
[docs]
def smf_i_dpd_id_len(pkt: 'dict[str, Any]', *, prefix: 'str') -> 'int':
"""Return SMF I-DPD identifier length.
Args:
pkt: SMF identification-based DPD option unpacked schema.
prefix: Name of the extension header, leading the error message.
Returns:
SMF I-DPD identifier length, or zero if the data ends before
``Opt Data Len``, which leaves the option to the short read.
Raises:
ProtocolError: If ``Opt Data Len`` on the wire is too short to hold
the TaggerID it declares, which would otherwise underflow the
identifier length below zero.
.. versionchanged:: 1.6.0
A length the data ends before is left to the short read, and one too
short raises :exc:`~pcapkit.utilities.exceptions.ProtocolError` rather
than :exc:`~pcapkit.utilities.exceptions.FieldValueError`
(:issue:`2727`).
"""
length = pkt['len'] - (1 if pkt['info']['type'] == 0 else (pkt['info']['len'] + 2))
if length < 0:
if _opt_data_len_cut(pkt):
return 0
raise ProtocolError(f'{prefix}: invalid SMF I-DPD option length: {pkt["len"]}')
return length
[docs]
def smf_dpd_data_selector(pkt: 'dict[str, Any]', *, prefix: 'str',
base: 'EnumMeta') -> 'Field':
"""Selector function for the ``data`` field of an ``SMF_DPD`` option.
Args:
pkt: Packet data.
prefix: Name of the extension header, leading the error message.
base: The header's ``SMFDPDOption`` schema, whose registry maps the
DPD mode to the schema that reads the option.
Returns:
The field for the option data, chosen by the DPD mode.
* If ``mode`` is ``0``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField`
wrapped ``SMFIdentificationBasedDPDOption`` instance.
* If ``mode`` is ``1``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField`
wrapped ``SMFHashBasedDPDOption`` instance.
Raises:
ProtocolError: If ``Opt Data Len`` is ``0`` on the wire.
FieldValueError: If no schema is registered for the DPD mode.
Note:
The field is sized ``Opt Data Len + 2`` rather than ``Opt Data Len``.
``Opt Data Len`` counts only what follows the option header
[:rfc:`8200#section-4.2`], while both schemas this may return inherit
the option's ``type`` and ``len`` fields and so parse those two octets
themselves. Sizing the field at ``Opt Data Len`` would hand them an area two
octets short of the option they read, which
:class:`~pcapkit.corekit.fields.collections.OptionField` would then
mis-count against the option area.
Where the data ends before ``Opt Data Len``, the forward match reads
it as zero, and the option is the short read of the schema chosen by
the mode bit, likewise zero.
.. versionchanged:: 1.6.0
A length the data ends before is left to the short read, and a zero
on the wire raises :exc:`~pcapkit.utilities.exceptions.ProtocolError`
rather than :exc:`~pcapkit.utilities.exceptions.FieldValueError`
(:issue:`2727`).
"""
# NOTE: With ``Opt Data Len`` 0 the forward-matched mode bit belongs to what follows the
# option, unless the data ends before ``Opt Data Len``, which then reads as zero (#2727).
record = pkt.get('__short_read__')
cut = record is not None and record[0] == 'test' and record[1] < 2
if pkt['test']['len'] == 0 and not cut:
raise ProtocolError(f'{prefix}: invalid SMF DPD option length: 0')
mode = Enum_SMFDPDMode.get(pkt['test']['mode'])
schema: 'Optional[Type[Schema]]' = base.registry[mode]
if schema is None:
raise FieldValueError(f'{prefix}: invalid SMF DPD mode: {mode}')
return SchemaField(length=pkt['test']['len'] + 2, schema=schema)
[docs]
def smf_i_dpd_tid_selector(pkt: 'dict[str, Any]', *, prefix: 'str') -> 'Field':
"""Selector function for the ``tid`` field of an SMF I-DPD option.
Args:
pkt: Packet data.
prefix: Name of the extension header, leading the error message.
Returns:
The field for the TaggerId, chosen by its type.
* If ``tid_type`` is ``0`` (``NULL``), returns a
:class:`~pcapkit.corekit.fields.misc.NoValueField` instance.
* If ``tid_type`` is ``2`` (``IPv4``), returns an
:class:`~pcapkit.corekit.fields.ipaddress.IPv4AddressField` instance.
* If ``tid_type`` is ``3`` (``IPv6``), returns an
:class:`~pcapkit.corekit.fields.ipaddress.IPv6AddressField` instance.
* Otherwise, returns a :class:`~pcapkit.corekit.fields.strings.BytesField` instance.
Raises:
FieldValueError: If ``tid_len`` does not match the length of the
``NULL``, ``IPv4`` or ``IPv6`` TaggerID type.
"""
tid_type = Enum_TaggerID.get(pkt['info']['type'])
tid_len = pkt['info']['len']
# update type
pkt['info']['type'] = tid_type
if tid_type == Enum_TaggerID.NULL:
if tid_len != 0:
raise FieldValueError(f'{prefix}: invalid TaggerID length: {tid_len}')
return NoValueField()
if tid_type == Enum_TaggerID.IPv4:
if tid_len != 3:
raise FieldValueError(f'{prefix}: invalid TaggerID length: {tid_len}')
return IPv4AddressField()
if tid_type == Enum_TaggerID.IPv6:
if tid_len != 15:
raise FieldValueError(f'{prefix}: invalid TaggerID length: {tid_len}')
return IPv6AddressField()
return BytesField(length=tid_len + 1)
[docs]
def quick_start_data_selector(pkt: 'dict[str, Any]', *, base: 'EnumMeta',
unassigned: 'Type[Schema]') -> 'Field':
"""Selector function for the ``data`` field of a ``Quick_Start`` option.
Args:
pkt: Packet data.
base: The header's ``QuickStartOption`` schema, whose registry maps
the QS function to the schema that reads the option.
unassigned: The header's ``UnassignedOption`` schema.
Returns:
The field for the option data, chosen by the QS function.
* If ``func`` is ``0``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField`
wrapped ``QuickStartRequestOption`` instance.
* If ``func`` is ``8``, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField`
wrapped ``QuickStartReportOption`` instance.
* Otherwise, returns a :class:`~pcapkit.corekit.fields.misc.SchemaField`
wrapped ``unassigned`` instance, sized by the option's own
``Opt Data Len``.
"""
func = Enum_QSFunction.get(pkt['flags']['func'])
pkt['flags']['func'] = func
schema: 'Optional[Type[Schema]]' = base.registry[func]
if schema is None:
# An unassigned function has no defined layout, so the option is kept
# as raw data, sized by its own ``Opt Data Len``.
return SchemaField(length=pkt['flags']['len'] + 2, schema=unassigned)
# NOTE: The suboption schemas carry the option's ``type`` and ``len`` octets, so this is the
# whole option: two header octets plus the six data octets of :rfc:`4782#section-3.2`.
return SchemaField(length=8, schema=schema)
[docs]
def pad_opt_data_len(pkt: 'dict[str, Any]') -> 'int':
"""Return the length of the padding data of a padding option.
Args:
pkt: Padding option unpacked schema.
Returns:
Number of padding octets carried after the option header, i.e. the
value of the ``Opt Data Len`` field of a ``PadN`` option, and zero for
a ``Pad1`` option, which carries no such field.
Note:
A ``Pad1`` option is a single octet with neither an ``Opt Data Len``
field nor any option data (see :rfc:`8200#section-4.2`), which is why
the option's ``len`` field is declared as a
:class:`~pcapkit.corekit.fields.misc.ConditionalField` and is skipped
for it. A skipped conditional field is *recorded* in the packet data as
:data:`~pcapkit.corekit.sentinels.NO_VALUE`, rather than being left out
of it, so the test this function makes has to be on the **value** and
not on the presence of the key: ``pkt.get('len', 0)`` on its own hands
that :obj:`~pcapkit.corekit.sentinels.NoValueType` straight to
:class:`~pcapkit.corekit.fields.strings.PaddingField`, where it becomes
an unusable :mod:`struct` template and surfaces much later as an opaque
:exc:`struct.error`, which is how a ``Pad1`` option would fail to
parse.
"""
length = pkt.get('len', 0)
if not isinstance(length, int): # ``NO_VALUE`` (skipped) or :obj:`None` (unset)
return 0
return length
[docs]
def calipso_pad_len(pkt: 'dict[str, Any]', *, prefix: 'str') -> 'int':
"""Return CALIPSO option padding length.
Args:
pkt: CALIPSO option unpacked schema.
prefix: Name of the extension header, leading the error message.
Returns:
CALIPSO option padding length, or zero if the data ends before
``Opt Data Len``, which leaves the option to the short read.
Raises:
ProtocolError: If ``Opt Data Len`` on the wire is too short to hold
the fixed header and the compartment bitmap declared by
``cmpt_len``, which would otherwise underflow the padding length
below zero.
.. versionchanged:: 1.6.0
A length the data ends before is left to the short read, and one too
short raises :exc:`~pcapkit.utilities.exceptions.ProtocolError` rather
than :exc:`~pcapkit.utilities.exceptions.FieldValueError`
(:issue:`2727`).
"""
length = pkt['len'] - 8 - pkt['cmpt_len'] * 4
if length < 0:
if _opt_data_len_cut(pkt):
return 0
raise ProtocolError(f'{prefix}: invalid CALIPSO option length: {pkt["len"]}')
return length
[docs]
def mpl_opt_pad_len(pkt: 'dict[str, Any]', *, prefix: 'str') -> 'int':
"""Return MPL option padding length.
Args:
pkt: MPL option unpacked schema.
prefix: Name of the extension header, leading the error message.
Returns:
MPL option padding length, or zero if the data ends before
``Opt Data Len``, which leaves the option to the short read.
Raises:
ProtocolError: If ``Opt Data Len`` on the wire is too short to hold
the fixed header and the Seed-ID declared by ``flags.type``, which
would otherwise underflow the padding length below zero.
.. versionchanged:: 1.6.0
A length the data ends before is left to the short read, and one too
short raises :exc:`~pcapkit.utilities.exceptions.ProtocolError` rather
than :exc:`~pcapkit.utilities.exceptions.FieldValueError`
(:issue:`2727`).
"""
length = pkt['len'] - 2 - (0 if pkt['flags']['type'] == 0
else mpl_opt_seed_id_len(pkt, prefix=prefix))
if length < 0:
if _opt_data_len_cut(pkt):
return 0
raise ProtocolError(f'{prefix}: invalid MPL option length: {pkt["len"]}')
return length