From 183be95b914cb490876f04d5531135e3e3e20636 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Mon, 14 Sep 2026 14:49:09 -0400 Subject: [PATCH] mh: implement the FMIPv6 fast-handover messages and options First tranche of the MH work the Help Wanted page asks for, taking message dispatch from 8/24 to 14/24 and options from 17/71 to 20/71. Messages, per RFC 5568 unless noted: 8 Fast Binding Update, 9 Fast Binding Acknowledgment, 10 Fast Neighbor Advertisement (RFC 4068, deprecated by 5568), 11 Experimental (RFC 5096), 14 Handover Initiate and 15 Handover Acknowledge. Options: 18 Experimental, 21 Binding Authorization Data for FMIPv6, 34 Mobility Header IPv6 Address/Prefix. Each has a read handler, a make counterpart, a schema, a data model, dispatch registration and docs, and each is covered by a byte-level RFC conformance case and a byte-identical round trip. Four handover_* and handoff_* const enums that were generated but never imported are now consumed. Two value sets get enums local to this module rather than const plus a vendor crawler, because RFC 5568 defines them inline and IANA registers neither: FastBindingAcknowledgmentStatus (6.2.3) and IPv6AddressPrefixCode (6.4.2). The first cannot reuse pcapkit.const.mh.status_code.StatusCode - 1 and 131 mean different things in the two lists, so that would mislabel real packets. Both carry the const modules' _missing_ handling so an unassigned value parses instead of raising. Placement follows pcapkit/protocols/misc/pcapng.py, which already defines two enums locally. The schema fields stay UInt8Field: EnumField needs its namespace at class-body time and the schema module cannot import the protocol module, which imports it, so the enum is applied when the data model is built, as pcapng does. Not implemented, deliberately: the three remaining CGA extensions are RFC 4581 experimental code points with no defined layout, so the existing opaque handler is already the faithful reading. BADF's length field follows RFC 5568's literal wording, which excludes the SPI and contradicts RFC 6275's generic rule. Also fixes two typos in the class docstring ("og", "resgitered"). --- docs/source/pcapkit/protocols/internet/mh.rst | 120 ++ pcapkit/protocols/data/internet/mh.py | 181 ++- pcapkit/protocols/internet/mh.py | 1226 +++++++++++++++-- pcapkit/protocols/schema/internet/mh.py | 260 +++- tests/protocols/internet/test_mh_unit.py | 635 +++++++++ 5 files changed, 2293 insertions(+), 129 deletions(-) diff --git a/docs/source/pcapkit/protocols/internet/mh.rst b/docs/source/pcapkit/protocols/internet/mh.rst index 7c91e247b0..0d238e25c6 100644 --- a/docs/source/pcapkit/protocols/internet/mh.rst +++ b/docs/source/pcapkit/protocols/internet/mh.rst @@ -51,6 +51,12 @@ Octets Bits Name Description .. automethod:: _read_msg_bu .. automethod:: _read_msg_ba .. automethod:: _read_msg_be + .. automethod:: _read_msg_fbu + .. automethod:: _read_msg_fback + .. automethod:: _read_msg_fna + .. automethod:: _read_msg_emh + .. automethod:: _read_msg_hi + .. automethod:: _read_msg_hack .. automethod:: _make_msg_unknown .. automethod:: _make_msg_brr @@ -61,6 +67,12 @@ Octets Bits Name Description .. automethod:: _make_msg_bu .. automethod:: _make_msg_ba .. automethod:: _make_msg_be + .. automethod:: _make_msg_fbu + .. automethod:: _make_msg_fback + .. automethod:: _make_msg_fna + .. automethod:: _make_msg_emh + .. automethod:: _make_msg_hi + .. automethod:: _make_msg_hack .. automethod:: _read_mh_options .. automethod:: _read_opt_none @@ -81,6 +93,9 @@ Octets Bits Name Description .. automethod:: _read_opt_phkt .. automethod:: _read_opt_ct_init .. automethod:: _read_opt_ct + .. automethod:: _read_opt_exp + .. automethod:: _read_opt_badf + .. automethod:: _read_opt_ipv6_ap .. automethod:: _make_mh_options .. automethod:: _make_opt_none @@ -101,6 +116,9 @@ Octets Bits Name Description .. automethod:: _make_opt_phkt .. automethod:: _make_opt_ct_init .. automethod:: _make_opt_ct + .. automethod:: _make_opt_exp + .. automethod:: _make_opt_badf + .. automethod:: _make_opt_ipv6_ap .. automethod:: _read_cga_extensions .. automethod:: _read_ext_none @@ -137,6 +155,16 @@ Auxiliary Data Fraction of a second. +.. autoclass:: pcapkit.protocols.internet.mh.FastBindingAcknowledgmentStatus + :members: + :undoc-members: + :show-inheritance: + +.. autoclass:: pcapkit.protocols.internet.mh.IPv6AddressPrefixCode + :members: + :undoc-members: + :show-inheritance: + Header Schemas -------------- @@ -238,6 +266,18 @@ Header Schemas :members: :show-inheritance: +.. autoclass:: pcapkit.protocols.schema.internet.mh.ExperimentalMobilityOption + :members: + :show-inheritance: + +.. autoclass:: pcapkit.protocols.schema.internet.mh.BADFOption + :members: + :show-inheritance: + +.. autoclass:: pcapkit.protocols.schema.internet.mh.IPv6AddressPrefixOption + :members: + :show-inheritance: + .. autoclass:: pcapkit.protocols.schema.internet.mh.UnknownMessage :members: :show-inheritance: @@ -274,6 +314,30 @@ Header Schemas :members: :show-inheritance: +.. autoclass:: pcapkit.protocols.schema.internet.mh.FastBindingUpdateMessage + :members: + :show-inheritance: + +.. autoclass:: pcapkit.protocols.schema.internet.mh.FastBindingAcknowledgmentMessage + :members: + :show-inheritance: + +.. autoclass:: pcapkit.protocols.schema.internet.mh.FastNeighborAdvertisementMessage + :members: + :show-inheritance: + +.. autoclass:: pcapkit.protocols.schema.internet.mh.ExperimentalMessage + :members: + :show-inheritance: + +.. autoclass:: pcapkit.protocols.schema.internet.mh.HandoverInitiateMessage + :members: + :show-inheritance: + +.. autoclass:: pcapkit.protocols.schema.internet.mh.HandoverAcknowledgeMessage + :members: + :show-inheritance: + Type Stubs ~~~~~~~~~~ @@ -297,6 +361,26 @@ Type Stubs :undoc-members: :show-inheritance: +.. autoclass:: pcapkit.protocols.schema.internet.mh.FastBindingUpdateMessageFlags + :members: + :undoc-members: + :show-inheritance: + +.. autoclass:: pcapkit.protocols.schema.internet.mh.FastBindingAcknowledgmentMessageFlags + :members: + :undoc-members: + :show-inheritance: + +.. autoclass:: pcapkit.protocols.schema.internet.mh.HandoverInitiateMessageFlags + :members: + :undoc-members: + :show-inheritance: + +.. autoclass:: pcapkit.protocols.schema.internet.mh.HandoverAcknowledgeMessageFlags + :members: + :undoc-members: + :show-inheritance: + Auxiliary Functions ~~~~~~~~~~~~~~~~~~~ @@ -400,6 +484,18 @@ Data Models :members: :show-inheritance: +.. autoclass:: pcapkit.protocols.data.internet.mh.ExperimentalMobilityOption + :members: + :show-inheritance: + +.. autoclass:: pcapkit.protocols.data.internet.mh.BADFOption + :members: + :show-inheritance: + +.. autoclass:: pcapkit.protocols.data.internet.mh.IPv6AddressPrefixOption + :members: + :show-inheritance: + .. autoclass:: pcapkit.protocols.data.internet.mh.UnknownMessage :members: :show-inheritance: @@ -436,6 +532,30 @@ Data Models :members: :show-inheritance: +.. autoclass:: pcapkit.protocols.data.internet.mh.FastBindingUpdateMessage + :members: + :show-inheritance: + +.. autoclass:: pcapkit.protocols.data.internet.mh.FastBindingAcknowledgmentMessage + :members: + :show-inheritance: + +.. autoclass:: pcapkit.protocols.data.internet.mh.FastNeighborAdvertisementMessage + :members: + :show-inheritance: + +.. autoclass:: pcapkit.protocols.data.internet.mh.ExperimentalMessage + :members: + :show-inheritance: + +.. autoclass:: pcapkit.protocols.data.internet.mh.HandoverInitiateMessage + :members: + :show-inheritance: + +.. autoclass:: pcapkit.protocols.data.internet.mh.HandoverAcknowledgeMessage + :members: + :show-inheritance: + .. rubric:: Footnotes .. [*] https://en.wikipedia.org/wiki/Mobile_IP#Changes_in_IPv6_for_Mobile_IPv6 diff --git a/pcapkit/protocols/data/internet/mh.py b/pcapkit/protocols/data/internet/mh.py index 3a9fb0b136..af2a93f7ab 100644 --- a/pcapkit/protocols/data/internet/mh.py +++ b/pcapkit/protocols/data/internet/mh.py @@ -40,6 +40,8 @@ from pcapkit.const.mh.handover_ack_status import HandoverACKStatus as Enum_HandoverACKStatus from pcapkit.const.mh.handover_initiate_flag import \ HandoverInitiateFlag as Enum_HandoverInitiateFlag + from pcapkit.const.mh.handover_initiate_status import \ + HandoverInitiateStatus as Enum_HandoverInitiateStatus from pcapkit.const.mh.home_address_reply import HomeAddressReply as Enum_HomeAddressReply from pcapkit.const.mh.lla_code import LLACode as Enum_LLACode from pcapkit.const.mh.lma_mag_suboption import \ @@ -62,20 +64,23 @@ UpdateNotificationReason as Enum_UpdateNotificationReason from pcapkit.const.reg.transtype import TransType from pcapkit.corekit.multidict import OrderedMultiDict - from pcapkit.protocols.internet.mh import NTPTimestamp + from pcapkit.protocols.internet.mh import (FastBindingAcknowledgmentStatus, + IPv6AddressPrefixCode, NTPTimestamp) __all__ = [ 'MH', 'UnknownMessage', 'BindingRefreshRequestMessage', 'HomeTestInitMessage', 'CareofTestInitMessage', 'HomeTestMessage', 'CareofTestMessage', 'BindingUpdateMessage', 'BindingAcknowledgementMessage', - 'BindingErrorMessage', + 'BindingErrorMessage', 'FastBindingUpdateMessage', 'FastBindingAcknowledgmentMessage', + 'FastNeighborAdvertisementMessage', 'ExperimentalMessage', 'HandoverInitiateMessage', + 'HandoverAcknowledgeMessage', 'Option', 'UnassignedOption', 'PadOption', 'BindingRefreshAdviceOption', 'AlternateCareofAddressOption', 'NonceIndicesOption', 'AuthorizationDataOption', 'MobileNetworkPrefixOption', 'LinkLayerAddressOption', 'MNIDOption', 'AuthOption', 'MesgIDOption', 'CGAParametersRequestOption', 'CGAParametersOption', 'SignatureOption', 'PermanentHomeKeygenTokenOption', 'CareofTestInitOption', - 'CareofTestOption', + 'CareofTestOption', 'ExperimentalMobilityOption', 'BADFOption', 'IPv6AddressPrefixOption', 'CGAParameter', @@ -250,6 +255,131 @@ def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'by options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... +@info_final +class FastBindingUpdateMessage(MH): + """Data model for MH Fast Binding Update (FBU) message type.""" + + #: Sequence number. + seq: 'int' + #: Acknowledge flag. + ack: 'bool' + #: Home registration flag. + home: 'bool' + #: Link-local address compatibility flag. + lla_compat: 'bool' + #: Key management mobility capability flag. + key_mngt: 'bool' + #: Lifetime. + lifetime: 'timedelta' + #: Mobility options. + options: 'OrderedMultiDict[Enum_Option, Option]' + + if TYPE_CHECKING: + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + seq: 'int', ack: 'bool', home: 'bool', lla_compat: 'bool', key_mngt: 'bool', + lifetime: 'timedelta', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... + + +@info_final +class FastBindingAcknowledgmentMessage(MH): + """Data model for MH Fast Binding Acknowledgment (FBack) message type.""" + + #: Status. :rfc:`5568#section-6.2.3` defines the FBack status values inline + #: rather than drawing them from the IANA *Status Codes* registry, so this + #: field carries the module-local + #: :class:`~pcapkit.protocols.internet.mh.FastBindingAcknowledgmentStatus` + #: instead of :class:`~pcapkit.const.mh.status_code.StatusCode`. + status: 'FastBindingAcknowledgmentStatus' + #: Key management mobility capability flag. + key_mngt: 'bool' + #: Sequence number. + seq: 'int' + #: Lifetime. + lifetime: 'timedelta' + #: Mobility options. + options: 'OrderedMultiDict[Enum_Option, Option]' + + if TYPE_CHECKING: + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + status: 'FastBindingAcknowledgmentStatus', key_mngt: 'bool', seq: 'int', + lifetime: 'timedelta', + options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... + + +@info_final +class FastNeighborAdvertisementMessage(MH): + """Data model for MH Fast Neighbor Advertisement (FNA) message type.""" + + #: Mobility options. + options: 'OrderedMultiDict[Enum_Option, Option]' + + if TYPE_CHECKING: + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... + + +@info_final +class ExperimentalMessage(MH): + """Data model for MH Experimental Mobility Header message type.""" + + #: Experimental message data. + data: 'bytes' + + if TYPE_CHECKING: + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + data: 'bytes') -> 'None': ... + + +@info_final +class HandoverInitiateMessage(MH): + """Data model for MH Handover Initiate (HI) message type.""" + + #: Sequence number. + seq: 'int' + #: Assigned address configuration flag. + assign: 'bool' + #: Buffer flag. + buffer: 'bool' + #: Proxy flag. + proxy: 'bool' + #: Forwarding flag. + forward: 'bool' + #: Code. + code: 'Enum_HandoverInitiateStatus' + #: Mobility options. + options: 'OrderedMultiDict[Enum_Option, Option]' + + if TYPE_CHECKING: + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + seq: 'int', assign: 'bool', buffer: 'bool', proxy: 'bool', forward: 'bool', + code: 'Enum_HandoverInitiateStatus', + options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... + + +@info_final +class HandoverAcknowledgeMessage(MH): + """Data model for MH Handover Acknowledge (HAck) message type.""" + + #: Sequence number. + seq: 'int' + #: Buffer flag. + buffer: 'bool' + #: Proxy flag. + proxy: 'bool' + #: Forwarding flag. + forward: 'bool' + #: Code. + code: 'Enum_HandoverACKStatus' + #: Mobility options. + options: 'OrderedMultiDict[Enum_Option, Option]' + + if TYPE_CHECKING: + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + seq: 'int', buffer: 'bool', proxy: 'bool', forward: 'bool', + code: 'Enum_HandoverACKStatus', + options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... + + # TODO: Implement other message types. @@ -506,4 +636,49 @@ class CareofTestOption(Option): def __init__(self, type: 'Enum_Option', length: 'int', token: 'bytes') -> 'None': ... +@info_final +class ExperimentalMobilityOption(Option): + """Data model for MH Experimental Mobility option.""" + + #: Experimental data. + data: 'bytes' + + if TYPE_CHECKING: + def __init__(self, type: 'Enum_Option', length: 'int', data: 'bytes') -> 'None': ... + + +@info_final +class BADFOption(Option): + """Data model for MH Binding Authorization Data for FMIPv6 (BADF) option.""" + + #: Security parameter index. + spi: 'int' + #: Authenticator. + data: 'bytes' + + if TYPE_CHECKING: + def __init__(self, type: 'Enum_Option', length: 'int', spi: 'int', data: 'bytes') -> 'None': ... + + +@info_final +class IPv6AddressPrefixOption(Option): + """Data model for MH Mobility Header IPv6 Address/Prefix option.""" + + #: Option code, c.f., :rfc:`5568#section-6.4.2`. The codes are defined inline + #: by the RFC with no IANA registry behind them, so this field carries the + #: module-local + #: :class:`~pcapkit.protocols.internet.mh.IPv6AddressPrefixCode`. + code: 'IPv6AddressPrefixCode' + #: Prefix length. + prefix_length: 'int' + #: IPv6 address/prefix. Kept separate from :attr:`prefix_length` since option + #: codes ``1`` through ``3`` carry a full address whose host bits are + #: significant, so the pair cannot always be fused into a network. + address: 'IPv6Address' + + if TYPE_CHECKING: + def __init__(self, type: 'Enum_Option', length: 'int', code: 'IPv6AddressPrefixCode', + prefix_length: 'int', address: 'IPv6Address') -> 'None': ... + + # TODO: Implement other options. diff --git a/pcapkit/protocols/internet/mh.py b/pcapkit/protocols/internet/mh.py index 8e6b134368..d82ec48cc9 100644 --- a/pcapkit/protocols/internet/mh.py +++ b/pcapkit/protocols/internet/mh.py @@ -27,6 +27,8 @@ import math from typing import TYPE_CHECKING, cast, overload +from aenum import IntEnum, extend_enum + from pcapkit.const.mh.access_type import AccessType as Enum_AccessType from pcapkit.const.mh.ack_status_code import ACKStatusCode as Enum_ACKStatusCode from pcapkit.const.mh.ani_suboption import ANISuboption as Enum_ANISuboption @@ -54,6 +56,8 @@ from pcapkit.const.mh.handover_ack_status import HandoverACKStatus as Enum_HandoverACKStatus from pcapkit.const.mh.handover_initiate_flag import \ HandoverInitiateFlag as Enum_HandoverInitiateFlag +from pcapkit.const.mh.handover_initiate_status import \ + HandoverInitiateStatus as Enum_HandoverInitiateStatus from pcapkit.const.mh.home_address_reply import HomeAddressReply as Enum_HomeAddressReply from pcapkit.const.mh.lla_code import LLACode as Enum_LLACode from pcapkit.const.mh.lma_mag_suboption import \ @@ -80,6 +84,7 @@ from pcapkit.protocols.data.internet.mh import AuthOption as Data_AuthOption from pcapkit.protocols.data.internet.mh import \ AuthorizationDataOption as Data_AuthorizationDataOption +from pcapkit.protocols.data.internet.mh import BADFOption as Data_BADFOption from pcapkit.protocols.data.internet.mh import \ BindingAcknowledgementMessage as Data_BindingAcknowledgementMessage from pcapkit.protocols.data.internet.mh import BindingErrorMessage as Data_BindingErrorMessage @@ -97,8 +102,23 @@ from pcapkit.protocols.data.internet.mh import CGAParametersOption as Data_CGAParametersOption from pcapkit.protocols.data.internet.mh import \ CGAParametersRequestOption as Data_CGAParametersRequestOption +from pcapkit.protocols.data.internet.mh import ExperimentalMessage as Data_ExperimentalMessage +from pcapkit.protocols.data.internet.mh import \ + ExperimentalMobilityOption as Data_ExperimentalMobilityOption +from pcapkit.protocols.data.internet.mh import \ + FastBindingAcknowledgmentMessage as Data_FastBindingAcknowledgmentMessage +from pcapkit.protocols.data.internet.mh import \ + FastBindingUpdateMessage as Data_FastBindingUpdateMessage +from pcapkit.protocols.data.internet.mh import \ + FastNeighborAdvertisementMessage as Data_FastNeighborAdvertisementMessage +from pcapkit.protocols.data.internet.mh import \ + HandoverAcknowledgeMessage as Data_HandoverAcknowledgeMessage +from pcapkit.protocols.data.internet.mh import \ + HandoverInitiateMessage as Data_HandoverInitiateMessage from pcapkit.protocols.data.internet.mh import HomeTestInitMessage as Data_HomeTestInitMessage from pcapkit.protocols.data.internet.mh import HomeTestMessage as Data_HomeTestMessage +from pcapkit.protocols.data.internet.mh import \ + IPv6AddressPrefixOption as Data_IPv6AddressPrefixOption from pcapkit.protocols.data.internet.mh import LinkLayerAddressOption as Data_LinkLayerAddressOption from pcapkit.protocols.data.internet.mh import MesgIDOption as Data_MesgIDOption from pcapkit.protocols.data.internet.mh import MNIDOption as Data_MNIDOption @@ -120,6 +140,7 @@ from pcapkit.protocols.schema.internet.mh import AuthOption as Schema_AuthOption from pcapkit.protocols.schema.internet.mh import \ AuthorizationDataOption as Schema_AuthorizationDataOption +from pcapkit.protocols.schema.internet.mh import BADFOption as Schema_BADFOption from pcapkit.protocols.schema.internet.mh import \ BindingAcknowledgementMessage as Schema_BindingAcknowledgementMessage from pcapkit.protocols.schema.internet.mh import BindingErrorMessage as Schema_BindingErrorMessage @@ -138,8 +159,23 @@ from pcapkit.protocols.schema.internet.mh import CGAParametersOption as Schema_CGAParametersOption from pcapkit.protocols.schema.internet.mh import \ CGAParametersRequestOption as Schema_CGAParametersRequestOption +from pcapkit.protocols.schema.internet.mh import ExperimentalMessage as Schema_ExperimentalMessage +from pcapkit.protocols.schema.internet.mh import \ + ExperimentalMobilityOption as Schema_ExperimentalMobilityOption +from pcapkit.protocols.schema.internet.mh import \ + FastBindingAcknowledgmentMessage as Schema_FastBindingAcknowledgmentMessage +from pcapkit.protocols.schema.internet.mh import \ + FastBindingUpdateMessage as Schema_FastBindingUpdateMessage +from pcapkit.protocols.schema.internet.mh import \ + FastNeighborAdvertisementMessage as Schema_FastNeighborAdvertisementMessage +from pcapkit.protocols.schema.internet.mh import \ + HandoverAcknowledgeMessage as Schema_HandoverAcknowledgeMessage +from pcapkit.protocols.schema.internet.mh import \ + HandoverInitiateMessage as Schema_HandoverInitiateMessage from pcapkit.protocols.schema.internet.mh import HomeTestInitMessage as Schema_HomeTestInitMessage from pcapkit.protocols.schema.internet.mh import HomeTestMessage as Schema_HomeTestMessage +from pcapkit.protocols.schema.internet.mh import \ + IPv6AddressPrefixOption as Schema_IPv6AddressPrefixOption from pcapkit.protocols.schema.internet.mh import \ LinkLayerAddressOption as Schema_LinkLayerAddressOption from pcapkit.protocols.schema.internet.mh import MesgIDOption as Schema_MesgIDOption @@ -208,12 +244,138 @@ class NTPTimestamp(collections.namedtuple('NTPTimestamp', 'seconds fraction')): fraction: int +class FastBindingAcknowledgmentStatus(IntEnum): + """[FastBindingAcknowledgmentStatus] Fast Binding Acknowledgment Status Codes. + + Status values of the fast binding acknowledgment (FBack) message, c.f., + :rfc:`5568#section-6.2.3`. Values below ``128`` indicate that the fast + binding update was accepted by the receiving node, values of ``128`` and + above that it was rejected. + + Note: + :rfc:`5568#section-6.2.3` defines these values inline and IANA keeps no + registry of them, so the enumeration lives here rather than in + :mod:`pcapkit.const.mh`. It is also **not** interchangeable with the + registered :class:`~pcapkit.const.mh.status_code.StatusCode`, since two + of its values collide semantically: ``1`` means *NCoA is invalid* here + but *accepted but prefix discovery necessary* there, and ``131`` means + *incorrect interface identifier length* here but *home registration not + supported* there. + + """ + + #: Fast Binding Update accepted [:rfc:`5568#section-6.2.3`] + Fast_Binding_Update_accepted = 0 + + #: Fast Binding Update accepted but NCoA is invalid; use the NCoA supplied in + #: the "alternate" care-of address option [:rfc:`5568#section-6.2.3`] + Fast_Binding_Update_accepted_but_NCoA_is_invalid = 1 + + #: Reason unspecified [:rfc:`5568#section-6.2.3`] + Reason_unspecified = 128 + + #: Administratively prohibited [:rfc:`5568#section-6.2.3`] + Administratively_prohibited = 129 + + #: Insufficient resources [:rfc:`5568#section-6.2.3`] + Insufficient_resources = 130 + + #: Incorrect interface identifier length [:rfc:`5568#section-6.2.3`] + Incorrect_interface_identifier_length = 131 + + @staticmethod + def get(key: 'int | str', default: 'int' = -1) -> 'FastBindingAcknowledgmentStatus': + """Backport support for original codes. + + Args: + key: Key to get enum item. + default: Default value if not found. + + """ + if isinstance(key, int): + return FastBindingAcknowledgmentStatus(key) + if key not in FastBindingAcknowledgmentStatus._member_map_: # pylint: disable=no-member + extend_enum(FastBindingAcknowledgmentStatus, key, default) + return FastBindingAcknowledgmentStatus[key] # type: ignore[misc] + + @classmethod + def _missing_(cls, value: 'int') -> 'FastBindingAcknowledgmentStatus': + """Lookup function used when value is not found. + + Args: + value: Value to get enum item. + + """ + if not (isinstance(value, int) and 0 <= value <= 255): + raise ValueError('%r is not a valid %s' % (value, cls.__name__)) + #: Unassigned + return extend_enum(cls, 'Unassigned_%d' % value, value) + + +class IPv6AddressPrefixCode(IntEnum): + """[IPv6AddressPrefixCode] Mobility Header IPv6 Address/Prefix Option Codes. + + Option codes of the mobility header IPv6 address/prefix option, which + identify which address the option carries, c.f., + :rfc:`5568#section-6.4.2`. + + Note: + :rfc:`5568#section-6.4.2` defines these values inline and IANA keeps no + registry of them, so the enumeration lives here rather than in + :mod:`pcapkit.const.mh`. The identical code space of the neighbor + discovery IP address/prefix option of :rfc:`5568#section-6.4.1` is + likewise unregistered. + + """ + + #: Old Care-of Address [:rfc:`5568#section-6.4.2`] + Old_Care_of_Address = 1 + + #: New Care-of Address [:rfc:`5568#section-6.4.2`] + New_Care_of_Address = 2 + + #: NAR's IP address [:rfc:`5568#section-6.4.2`] + NAR_IP_address = 3 + + #: NAR's Prefix, sent in PrRtAdv; the prefix length field contains the number + #: of valid leading bits in the prefix [:rfc:`5568#section-6.4.2`] + NAR_Prefix = 4 + + @staticmethod + def get(key: 'int | str', default: 'int' = -1) -> 'IPv6AddressPrefixCode': + """Backport support for original codes. + + Args: + key: Key to get enum item. + default: Default value if not found. + + """ + if isinstance(key, int): + return IPv6AddressPrefixCode(key) + if key not in IPv6AddressPrefixCode._member_map_: # pylint: disable=no-member + extend_enum(IPv6AddressPrefixCode, key, default) + return IPv6AddressPrefixCode[key] # type: ignore[misc] + + @classmethod + def _missing_(cls, value: 'int') -> 'IPv6AddressPrefixCode': + """Lookup function used when value is not found. + + Args: + value: Value to get enum item. + + """ + if not (isinstance(value, int) and 0 <= value <= 255): + raise ValueError('%r is not a valid %s' % (value, cls.__name__)) + #: Unassigned + return extend_enum(cls, 'Unassigned_%d' % value, value) + + class MH(Internet[Data_MH, Schema_MH], schema=Schema_MH, data=Data_MH): """This class implements Mobility Header. - This class currently supports parsing og the following MH message types, - which are resgitered in the :attr:`self.__message__ ` + This class currently supports parsing of the following MH message types, + which are registered in the :attr:`self.__message__ ` attribute: .. list-table:: @@ -246,6 +408,24 @@ class MH(Internet[Data_MH, Schema_MH], * - :attr:`~pcapkit.const.mh.packet.Packet.Binding_Error` - :meth:`~pcapkit.protocols.internet.mh.MH._read_msg_be` - :meth:`~pcapkit.protocols.internet.mh.MH._make_msg_be` + * - :attr:`~pcapkit.const.mh.packet.Packet.Fast_Binding_Update` + - :meth:`~pcapkit.protocols.internet.mh.MH._read_msg_fbu` + - :meth:`~pcapkit.protocols.internet.mh.MH._make_msg_fbu` + * - :attr:`~pcapkit.const.mh.packet.Packet.Fast_Binding_Acknowledgment` + - :meth:`~pcapkit.protocols.internet.mh.MH._read_msg_fback` + - :meth:`~pcapkit.protocols.internet.mh.MH._make_msg_fback` + * - :attr:`~pcapkit.const.mh.packet.Packet.Fast_Neighbor_Advertisement` + - :meth:`~pcapkit.protocols.internet.mh.MH._read_msg_fna` + - :meth:`~pcapkit.protocols.internet.mh.MH._make_msg_fna` + * - :attr:`~pcapkit.const.mh.packet.Packet.Experimental_Mobility_Header` + - :meth:`~pcapkit.protocols.internet.mh.MH._read_msg_emh` + - :meth:`~pcapkit.protocols.internet.mh.MH._make_msg_emh` + * - :attr:`~pcapkit.const.mh.packet.Packet.Handover_Initiate_Message` + - :meth:`~pcapkit.protocols.internet.mh.MH._read_msg_hi` + - :meth:`~pcapkit.protocols.internet.mh.MH._make_msg_hi` + * - :attr:`~pcapkit.const.mh.packet.Packet.Handover_Acknowledge_Message` + - :meth:`~pcapkit.protocols.internet.mh.MH._read_msg_hack` + - :meth:`~pcapkit.protocols.internet.mh.MH._make_msg_hack` This class currently supports parsing the following MH options, which are registered in the :attr:`self.__option__ ` @@ -309,6 +489,15 @@ class MH(Internet[Data_MH, Schema_MH], * - :attr:`~pcapkit.const.mh.option.Option.Care_of_Test` - :meth:`~pcapkit.protocols.internet.mh.MH._read_opt_ct` - :meth:`~pcapkit.protocols.internet.mh.MH._make_opt_ct` + * - :attr:`~pcapkit.const.mh.option.Option.Experimental_Mobility_Option` + - :meth:`~pcapkit.protocols.internet.mh.MH._read_opt_exp` + - :meth:`~pcapkit.protocols.internet.mh.MH._make_opt_exp` + * - :attr:`~pcapkit.const.mh.option.Option.Binding_Authorization_Data_for_FMIPv6` + - :meth:`~pcapkit.protocols.internet.mh.MH._read_opt_badf` + - :meth:`~pcapkit.protocols.internet.mh.MH._make_opt_badf` + * - :attr:`~pcapkit.const.mh.option.Option.Mobility_Header_IPv6_Address_Prefix` + - :meth:`~pcapkit.protocols.internet.mh.MH._read_opt_ipv6_ap` + - :meth:`~pcapkit.protocols.internet.mh.MH._make_opt_ipv6_ap` This class currently supports parsing of the following MH CGA extensions, which are registered in the :attr:`self.__extension__ ` @@ -346,6 +535,12 @@ class MH(Internet[Data_MH, Schema_MH], Enum_Packet.Binding_Update: 'bu', Enum_Packet.Binding_Acknowledgement: 'ba', Enum_Packet.Binding_Error: 'be', + Enum_Packet.Fast_Binding_Update: 'fbu', + Enum_Packet.Fast_Binding_Acknowledgment: 'fback', + Enum_Packet.Fast_Neighbor_Advertisement: 'fna', + Enum_Packet.Experimental_Mobility_Header: 'emh', + Enum_Packet.Handover_Initiate_Message: 'hi', + Enum_Packet.Handover_Acknowledge_Message: 'hack', }, ) # type: DefaultDict[Enum_Packet | int, str | tuple[PacketParser, PacketConstructor]] @@ -374,6 +569,9 @@ class MH(Internet[Data_MH, Schema_MH], Enum_Option.Permanent_Home_Keygen_Token: 'phkt', Enum_Option.Care_of_Test_Init: 'ct_init', Enum_Option.Care_of_Test: 'ct', + Enum_Option.Experimental_Mobility_Option: 'exp', + Enum_Option.Binding_Authorization_Data_for_FMIPv6: 'badf', + Enum_Option.Mobility_Header_IPv6_Address_Prefix: 'ipv6_ap', }, ) # type: DefaultDict[Enum_Option | int, str | tuple[OptionParser, OptionConstructor]] @@ -1022,179 +1220,459 @@ def _read_msg_be(self, schema: 'Schema_BindingErrorMessage', *, ) return data - # TODO: Implement other message types. - - def _read_mh_options(self, options_schema: 'list[Schema_Option]') -> 'Option': - """Read MH options. + def _read_msg_fbu(self, schema: 'Schema_FastBindingUpdateMessage', *, + header: 'Schema_MH') -> 'Data_FastBindingUpdateMessage': + """Read MH fast binding update (FBU) message type. - Structure of MH option [:rfc:`6275`]: + Structure of MH Fast Binding Update Message [:rfc:`5568#section-6.2.2`]: .. 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 + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | Sequence # | +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ - | Option Type | Option Length | Option Data... + |A|H|L|K| Reserved | Lifetime | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | | + . . + . Mobility Options . + . . + | | +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + Note: + :rfc:`5568#section-6.2.2` states that the FBU is *identical* to the + Mobile IPv6 binding update (BU) message, so the lifetime is read in + units of 4 seconds exactly as in :meth:`_read_msg_bu`. + Args: - options_schema: Parsed MH options. + schema: Parsed message type schema. + header: Parsed MH header schema. Returns: - Parsed MH options data. + Parsed message type data. """ - options = OrderedMultiDict() # type: Option + data = Data_FastBindingUpdateMessage( + next=header.next, + length=(header.length + 1) * 8, + type=header.type, + chksum=header.chksum, + seq=schema.seq, + ack=bool(schema.flags['A']), + home=bool(schema.flags['H']), + lla_compat=bool(schema.flags['L']), + key_mngt=bool(schema.flags['K']), + lifetime=datetime.timedelta(seconds=schema.lifetime * 4), + options=self._read_mh_options(schema.options), + ) + return data - for schema in options_schema: - type = schema.type - name = self.__option__[type] + def _read_msg_fback(self, schema: 'Schema_FastBindingAcknowledgmentMessage', *, + header: 'Schema_MH') -> 'Data_FastBindingAcknowledgmentMessage': + """Read MH fast binding acknowledgment (FBack) message type. - if isinstance(name, str): - meth_name = f'_read_opt_{name}' - meth = cast('OptionParser', - getattr(self, meth_name, self._read_opt_none)) - else: - meth = name[0] - data = meth(schema, options=options) + Structure of MH Fast Binding Acknowledgment Message [:rfc:`5568#section-6.2.3`]: - # record option data - options.add(type, data) + .. code-block:: text - return options + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | Status |K| Reserved | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | Sequence # | Lifetime | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | | + . . + . Mobility Options . + . . + | | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ - def _read_opt_none(self, schema: 'Schema_UnassignedOption', *, - options: 'Option') -> 'Data_UnassignedOption': - """Read MH unassigned option. + Note: + :rfc:`5568#section-6.2.3` defines the FBack status values inline + instead of drawing them from the IANA *Status Codes* registry, whose + values ``1`` and ``131`` mean something else entirely. The status is + therefore reported as a + :class:`~pcapkit.protocols.internet.mh.FastBindingAcknowledgmentStatus`, + which is local to this module, rather than being mislabelled as a + :class:`~pcapkit.const.mh.status_code.StatusCode`. Args: - schema: Parsed option schema. - options: Parsed MH options. + schema: Parsed message type schema. + header: Parsed MH header schema. Returns: - Constructed option data. + Parsed message type data. """ - data = Data_UnassignedOption( - type=schema.type, - length=schema.length + 2, - data=schema.data, + data = Data_FastBindingAcknowledgmentMessage( + next=header.next, + length=(header.length + 1) * 8, + type=header.type, + chksum=header.chksum, + status=FastBindingAcknowledgmentStatus(schema.status), + key_mngt=bool(schema.flags['K']), + seq=schema.seq, + lifetime=datetime.timedelta(seconds=schema.lifetime * 4), + options=self._read_mh_options(schema.options), ) return data - def _read_opt_pad(self, schema: 'Schema_PadOption', *, - options: 'Option') -> 'Data_PadOption': - """Read MH padding option. - - Structure of MH padding option [:rfc:`6275`]: - - * ``Pad1`` option: - - .. code-block:: text + def _read_msg_fna(self, schema: 'Schema_FastNeighborAdvertisementMessage', *, + header: 'Schema_MH') -> 'Data_FastNeighborAdvertisementMessage': + """Read MH fast neighbor advertisement (FNA) message type. - 0 - 0 1 2 3 4 5 6 7 - +-+-+-+-+-+-+-+-+ - | Type = 0 | - +-+-+-+-+-+-+-+-+ + Structure of MH Fast Neighbor Advertisement Message [:rfc:`4068#section-6.3.3`]: - * ``PadN`` option: + .. code-block:: text - .. code-block:: text + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | Reserved | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + . . + . Mobility Options . + . . + | | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ - 0 1 - 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 - +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+- - - - - - - - - - | Type = 1 | Option Length | Option Data - +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+- - - - - - - - - + Note: + The FNA message is **deprecated** by :rfc:`5568#section-8`, which + replaces it with the unsolicited neighbor advertisement (UNA) of + :rfc:`4861`. Its wire format is therefore taken from :rfc:`4068`, + the specification :rfc:`5568` obsoletes, since a NAR is still + permitted to process the message as specified there. Args: - schema: Parsed option schema. - options: Parsed MH options. + schema: Parsed message type schema. + header: Parsed MH header schema. Returns: - Constructed option data. + Parsed message type data. """ - code, clen = schema.type, schema.length - - if code not in (Enum_Option.Pad1, Enum_Option.PadN): - raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid format') - if code == Enum_Option.Pad1 and clen != 0: - raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid format') - if code == Enum_Option.PadN and clen == 0: - raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid format') - - if code == Enum_Option.Pad1: - size = 1 - else: - size = clen + 2 - - data = Data_PadOption( - type=schema.type, - length=size, + data = Data_FastNeighborAdvertisementMessage( + next=header.next, + length=(header.length + 1) * 8, + type=header.type, + chksum=header.chksum, + options=self._read_mh_options(schema.options), ) return data - def _read_opt_bra(self, schema: 'Schema_BindingRefreshAdviceOption', *, - options: 'Option') -> 'Data_BindingRefreshAdviceOption': - """Read MH binding refresh advice option. + def _read_msg_emh(self, schema: 'Schema_ExperimentalMessage', *, + header: 'Schema_MH') -> 'Data_ExperimentalMessage': + """Read MH experimental mobility header message type. - Structure of MH Binding Refresh Advice option [:rfc:`6275`]: + Structure of MH Experimental Mobility Header Message [:rfc:`5096#section-3`]: .. 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 - +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ - | Type = 2 | Length = 2 | +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ - | Refresh Interval | - +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | Payload Proto | Header Len | MH Type | Reserved | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | Checksum | | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ | + | | + . . + . Message Data . + . . + | | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + + Note: + :rfc:`5096#section-3` defines no fields beyond those of the mobility + header itself, so the message data is carried opaquely -- its layout + is whatever the experiment in question defines. Two bytes of padding + are required when no data is present. Args: - schema: Parsed option schema. - options: Parsed MH options. + schema: Parsed message type schema. + header: Parsed MH header schema. Returns: - Constructed option data. + Parsed message type data. """ - if schema.length != 2: - raise ProtocolError(f'{self.alias}: [Opt {schema.type}] invalid format') - - data = Data_BindingRefreshAdviceOption( - type=schema.type, - length=schema.length + 2, - interval=schema.interval, + data = Data_ExperimentalMessage( + next=header.next, + length=(header.length + 1) * 8, + type=header.type, + chksum=header.chksum, + data=schema.data, ) return data - def _read_opt_aca(self, schema: 'Schema_AlternateCareofAddressOption', *, - options: 'Option') -> 'Data_AlternateCareofAddressOption': - """Read MH alternate care-of address option. + def _read_msg_hi(self, schema: 'Schema_HandoverInitiateMessage', *, + header: 'Schema_MH') -> 'Data_HandoverInitiateMessage': + """Read MH handover initiate (HI) message type. - Structure of MH Alternate Care-of Address option [:rfc:`6275`]: + Structure of MH Handover Initiate Message [:rfc:`5568#section-6.2.1.1`, + :rfc:`5949#section-6.1.1`]: .. 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 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ - | Type = 3 | Length = 16 | + | Sequence # | +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + |S|U|P|F| Resv | Code | | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ . | | - + + - | | - + Alternate Care-of Address + - | | - + + + . Mobility Options . + . . | | +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + Note: + :rfc:`5568` defines only the ``S`` and ``U`` flags; the ``P`` and + ``F`` flags, and code values ``2`` and ``3``, are added by + :rfc:`5949#section-6.1.1`. All four flags are registered in the IANA + *Handover Initiate Flags* registry, hence all four are parsed here. + Args: - schema: Parsed option schema. - options: Parsed MH options. + schema: Parsed message type schema. + header: Parsed MH header schema. + + Returns: + Parsed message type data. + + """ + data = Data_HandoverInitiateMessage( + next=header.next, + length=(header.length + 1) * 8, + type=header.type, + chksum=header.chksum, + seq=schema.seq, + assign=bool(schema.flags['S']), + buffer=bool(schema.flags['U']), + proxy=bool(schema.flags['P']), + forward=bool(schema.flags['F']), + code=schema.code, + options=self._read_mh_options(schema.options), + ) + return data + + def _read_msg_hack(self, schema: 'Schema_HandoverAcknowledgeMessage', *, + header: 'Schema_MH') -> 'Data_HandoverAcknowledgeMessage': + """Read MH handover acknowledge (HAck) message type. + + Structure of MH Handover Acknowledge Message [:rfc:`5568#section-6.2.1.2`, + :rfc:`5949#section-6.1.2`]: + + .. code-block:: text + + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | Sequence # | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + |U|P|F| Reserved| Code | | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ . + | | + . Mobility Options . + . . + | | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + + Note: + :rfc:`5568` reserves the whole first octet of the message data; the + ``U``, ``P`` and ``F`` flags carved out of it, and code values ``5`` + and ``6``, are added by :rfc:`5949#section-6.1.2`. All three flags + are registered in the IANA *Handover Acknowledge Flags* registry, + hence all three are parsed here. + + Args: + schema: Parsed message type schema. + header: Parsed MH header schema. + + Returns: + Parsed message type data. + + """ + data = Data_HandoverAcknowledgeMessage( + next=header.next, + length=(header.length + 1) * 8, + type=header.type, + chksum=header.chksum, + seq=schema.seq, + buffer=bool(schema.flags['U']), + proxy=bool(schema.flags['P']), + forward=bool(schema.flags['F']), + code=schema.code, + options=self._read_mh_options(schema.options), + ) + return data + + # TODO: Implement other message types. + + def _read_mh_options(self, options_schema: 'list[Schema_Option]') -> 'Option': + """Read MH options. + + Structure of MH option [:rfc:`6275`]: + + .. 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 + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | Option Type | Option Length | Option Data... + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + + Args: + options_schema: Parsed MH options. + + Returns: + Parsed MH options data. + + """ + options = OrderedMultiDict() # type: Option + + for schema in options_schema: + type = schema.type + name = self.__option__[type] + + if isinstance(name, str): + meth_name = f'_read_opt_{name}' + meth = cast('OptionParser', + getattr(self, meth_name, self._read_opt_none)) + else: + meth = name[0] + data = meth(schema, options=options) + + # record option data + options.add(type, data) + + return options + + def _read_opt_none(self, schema: 'Schema_UnassignedOption', *, + options: 'Option') -> 'Data_UnassignedOption': + """Read MH unassigned option. + + Args: + schema: Parsed option schema. + options: Parsed MH options. + + Returns: + Constructed option data. + + """ + data = Data_UnassignedOption( + type=schema.type, + length=schema.length + 2, + data=schema.data, + ) + return data + + def _read_opt_pad(self, schema: 'Schema_PadOption', *, + options: 'Option') -> 'Data_PadOption': + """Read MH padding option. + + Structure of MH padding option [:rfc:`6275`]: + + * ``Pad1`` option: + + .. code-block:: text + + 0 + 0 1 2 3 4 5 6 7 + +-+-+-+-+-+-+-+-+ + | Type = 0 | + +-+-+-+-+-+-+-+-+ + + * ``PadN`` option: + + .. code-block:: text + + 0 1 + 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+- - - - - - - - - + | Type = 1 | Option Length | Option Data + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+- - - - - - - - - + + Args: + schema: Parsed option schema. + options: Parsed MH options. + + Returns: + Constructed option data. + + """ + code, clen = schema.type, schema.length + + if code not in (Enum_Option.Pad1, Enum_Option.PadN): + raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid format') + if code == Enum_Option.Pad1 and clen != 0: + raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid format') + if code == Enum_Option.PadN and clen == 0: + raise ProtocolError(f'{self.alias}: [OptNo {code}] invalid format') + + if code == Enum_Option.Pad1: + size = 1 + else: + size = clen + 2 + + data = Data_PadOption( + type=schema.type, + length=size, + ) + return data + + def _read_opt_bra(self, schema: 'Schema_BindingRefreshAdviceOption', *, + options: 'Option') -> 'Data_BindingRefreshAdviceOption': + """Read MH binding refresh advice option. + + Structure of MH Binding Refresh Advice option [:rfc:`6275`]: + + .. 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 + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | Type = 2 | Length = 2 | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | Refresh Interval | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + + Args: + schema: Parsed option schema. + options: Parsed MH options. + + Returns: + Constructed option data. + + """ + if schema.length != 2: + raise ProtocolError(f'{self.alias}: [Opt {schema.type}] invalid format') + + data = Data_BindingRefreshAdviceOption( + type=schema.type, + length=schema.length + 2, + interval=schema.interval, + ) + return data + + def _read_opt_aca(self, schema: 'Schema_AlternateCareofAddressOption', *, + options: 'Option') -> 'Data_AlternateCareofAddressOption': + """Read MH alternate care-of address option. + + Structure of MH Alternate Care-of Address option [:rfc:`6275`]: + + .. 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 + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | Type = 3 | Length = 16 | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | | + + + + | | + + Alternate Care-of Address + + | | + + + + | | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + + Args: + schema: Parsed option schema. + options: Parsed MH options. Returns: Constructed option data. @@ -1636,33 +2114,170 @@ def _read_opt_ct_init(self, schema: 'Schema_CareofTestInitOption', *, Constructed option data. """ - if schema.length != 0: + if schema.length != 0: + raise ProtocolError(f'{self.alias}: [Opt {schema.type}] invalid format') + + data = Data_CareofTestInitOption( + type=schema.type, + length=schema.length + 2, + ) + return data + + def _read_opt_ct(self, schema: 'Schema_CareofTestOption', *, + options: 'Option') -> 'Data_CareofTestOption': + """Read MH Care-of Test option. + + Structure of MH Care-of Test option [:rfc:`4866`]: + + .. 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 + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | Option Type | Option Length | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | | + + Care-of Keygen Token + + | | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + + Args: + schema: Parsed option schema. + options: Parsed MH options. + + Returns: + Constructed option data. + + """ + if schema.length != 8: + raise ProtocolError(f'{self.alias}: [Opt {schema.type}] invalid format') + + data = Data_CareofTestOption( + type=schema.type, + length=schema.length + 2, + token=schema.token, + ) + return data + + def _read_opt_exp(self, schema: 'Schema_ExperimentalMobilityOption', *, + options: 'Option') -> 'Data_ExperimentalMobilityOption': + """Read MH experimental mobility option. + + Structure of MH Experimental Mobility option [:rfc:`5096#section-4`]: + + .. 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 + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | Type | Length | Data ..... + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + + Args: + schema: Parsed option schema. + options: Parsed MH options. + + Returns: + Constructed option data. + + """ + data = Data_ExperimentalMobilityOption( + type=schema.type, + length=schema.length + 2, + data=schema.data, + ) + return data + + def _read_opt_badf(self, schema: 'Schema_BADFOption', *, + options: 'Option') -> 'Data_BADFOption': + """Read MH binding authorization data for FMIPv6 (BADF) option. + + Structure of MH Binding Authorization Data for FMIPv6 option + [:rfc:`5568#section-6.4.5`]: + + .. 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 + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | Type = 21 | Option Length | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | SPI | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | | + + + + | Authenticator | + + + + | | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + + Note: + :rfc:`5568#section-6.4.5` defines the option length as *the length of + the Authenticator in bytes*, i.e. it does **not** cover the 4-byte + SPI, unlike every other mobility option, whose length covers all of + its option data. The wording is inherited from the :rfc:`6275` + binding authorization data option (type ``5``), which carries no SPI + and for which the two readings coincide. Because :rfc:`5568` also + requires this option to be the **last** mobility option present, the + discrepancy never has to be resolved in order to find the following + option, so the literal reading is used here and the reported + :attr:`~pcapkit.protocols.data.internet.mh.Option.length` accounts + for the extra 4 bytes. + + Args: + schema: Parsed option schema. + options: Parsed MH options. + + Returns: + Constructed option data. + + """ + if schema.length == 0: raise ProtocolError(f'{self.alias}: [Opt {schema.type}] invalid format') - data = Data_CareofTestInitOption( + data = Data_BADFOption( type=schema.type, - length=schema.length + 2, + length=schema.length + 6, # 2 bytes for type & length, 4 bytes for SPI + spi=schema.spi, + data=schema.data, ) return data - def _read_opt_ct(self, schema: 'Schema_CareofTestOption', *, - options: 'Option') -> 'Data_CareofTestOption': - """Read MH Care-of Test option. + def _read_opt_ipv6_ap(self, schema: 'Schema_IPv6AddressPrefixOption', *, + options: 'Option') -> 'Data_IPv6AddressPrefixOption': + """Read MH mobility header IPv6 address/prefix option. - Structure of MH Care-of Test option [:rfc:`4866`]: + Structure of MH Mobility Header IPv6 Address/Prefix option + [:rfc:`5568#section-6.4.2`]: .. 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 - +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ - | Option Type | Option Length | + +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + | Type | Length | Option-Code | Prefix Length | +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ | | - + Care-of Keygen Token + + + + + | | + + IPv6 Address/Prefix + + | | + + + | | +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + The option code identifies which address the option carries, c.f., + :class:`~pcapkit.protocols.internet.mh.IPv6AddressPrefixCode`. It, too, + is defined inline by :rfc:`5568#section-6.4.2` with no IANA registry + behind it, hence the enumeration is local to this module. + + Note: + :rfc:`5568` prints the type as ``17``, which is the *neighbor + discovery* option type of its sibling in + :rfc:`5568#section-6.4.1`. Errata ID 1816 (verified) corrects it to + the IANA-assigned mobility option type ``34``, which is what this + handler is registered against. + Args: schema: Parsed option schema. options: Parsed MH options. @@ -1671,13 +2286,17 @@ def _read_opt_ct(self, schema: 'Schema_CareofTestOption', *, Constructed option data. """ - if schema.length != 8: + if schema.length != 18: + raise ProtocolError(f'{self.alias}: [Opt {schema.type}] invalid format') + if schema.prefix_length > 128: raise ProtocolError(f'{self.alias}: [Opt {schema.type}] invalid format') - data = Data_CareofTestOption( + data = Data_IPv6AddressPrefixOption( type=schema.type, length=schema.length + 2, - token=schema.token, + code=IPv6AddressPrefixCode(schema.code), + prefix_length=schema.prefix_length, + address=schema.address, ) return data @@ -2102,6 +2721,261 @@ def _make_msg_be(self, message: 'Optional[Data_BindingErrorMessage]' = None, *, options=self._make_mh_options(options), ) + def _make_msg_fbu(self, message: 'Optional[Data_FastBindingUpdateMessage]' = None, *, + seq: 'int' = 0, + ack: 'bool' = True, # MUST be set, c.f., RFC 5568, section 6.2.2 + home: 'bool' = True, # MUST be set, c.f., RFC 5568, section 6.2.2 + lla_compat: 'bool' = False, + key_mngt: 'bool' = False, + lifetime: 'int | timedelta' = 4, # reasonable default value + options: 'Optional[Option | list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes]]' = None, + **kwargs: 'Any') -> 'Schema_FastBindingUpdateMessage': + """Make MH fast binding update (FBU) message type. + + Args: + message: Message data model. + seq: Sequence number. + ack: Acknowledgement flag. + home: Home registration flag. + lla_compat: LLA compatibility flag. + key_mngt: Key management mobility option flag. + lifetime: Lifetime in seconds or timedelta. + options: Mobility options. + **kwargs: Arbitrary keyword arguments. + + Returns: + Constructed message type. + + """ + if message is not None: + seq = message.seq + ack = message.ack + home = message.home + lla_compat = message.lla_compat + key_mngt = message.key_mngt + lifetime_val = math.ceil(message.lifetime.total_seconds()) + options = message.options + else: + lifetime_val = lifetime if isinstance(lifetime, int) else math.ceil(lifetime.total_seconds()) + options = options or [] + + return Schema_FastBindingUpdateMessage( + seq=seq, + flags={ + 'A': ack, + 'H': home, + 'L': lla_compat, + 'K': key_mngt, + }, + lifetime=math.ceil(lifetime_val / 4), + options=self._make_mh_options(options), + ) + + def _make_msg_fback(self, message: 'Optional[Data_FastBindingAcknowledgmentMessage]' = None, *, + status: 'FastBindingAcknowledgmentStatus | StdlibEnum | AenumEnum | str | int' = FastBindingAcknowledgmentStatus.Fast_Binding_Update_accepted, + status_default: 'Optional[int]' = None, + status_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long + status_reversed: 'bool' = False, + key_mngt: 'bool' = False, + seq: 'int' = 0, + lifetime: 'int | timedelta' = 4, # reasonable default value + options: 'Optional[Option | list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes]]' = None, + **kwargs: 'Any') -> 'Schema_FastBindingAcknowledgmentMessage': + """Make MH fast binding acknowledgment (FBack) message type. + + Args: + message: Message data model. + status: Status code, c.f., + :class:`~pcapkit.protocols.internet.mh.FastBindingAcknowledgmentStatus`. + status_default: Default status code. + status_namespace: Status code namespace. + status_reversed: Reverse status code namespace. + key_mngt: Key management mobility option flag. + seq: Sequence number. + lifetime: Lifetime in seconds or timedelta. + options: Mobility options. + **kwargs: Arbitrary keyword arguments. + + Returns: + Constructed message type. + + """ + if message is not None: + status_val = message.status + key_mngt = message.key_mngt + seq = message.seq + lifetime_val = math.ceil(message.lifetime.total_seconds()) + options = message.options + else: + status_val = self._make_index(status, status_default, namespace=status_namespace, # type: ignore[assignment] + reversed=status_reversed, pack=False) + lifetime_val = lifetime if isinstance(lifetime, int) else math.ceil(lifetime.total_seconds()) + options = options or [] + + return Schema_FastBindingAcknowledgmentMessage( + status=status_val, + flags={ + 'K': key_mngt, + }, + seq=seq, + lifetime=math.ceil(lifetime_val / 4), + options=self._make_mh_options(options), + ) + + def _make_msg_fna(self, message: 'Optional[Data_FastNeighborAdvertisementMessage]' = None, *, + options: 'Optional[Option | list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes]]' = None, + **kwargs: 'Any') -> 'Schema_FastNeighborAdvertisementMessage': + """Make MH fast neighbor advertisement (FNA) message type. + + Args: + message: Message data model. + options: Mobility options. + **kwargs: Arbitrary keyword arguments. + + Returns: + Constructed message type. + + """ + if message is not None: + options = message.options + else: + options = options or [] + + return Schema_FastNeighborAdvertisementMessage( + options=self._make_mh_options(options), + ) + + def _make_msg_emh(self, message: 'Optional[Data_ExperimentalMessage]' = None, *, + data: 'bytes' = b'\x00\x00', # 2 bytes of padding, c.f., RFC 5096, section 3 + **kwargs: 'Any') -> 'Schema_ExperimentalMessage': + """Make MH experimental mobility header message type. + + Args: + message: Message data model. + data: Experimental message data. + **kwargs: Arbitrary keyword arguments. + + Returns: + Constructed message type. + + """ + if message is not None: + data = message.data + + return Schema_ExperimentalMessage( + data=data, + ) + + def _make_msg_hi(self, message: 'Optional[Data_HandoverInitiateMessage]' = None, *, + seq: 'int' = 0, + assign: 'bool' = False, + buffer: 'bool' = False, + proxy: 'bool' = False, + forward: 'bool' = False, + code: 'Enum_HandoverInitiateStatus | StdlibEnum | AenumEnum | str | int' = Enum_HandoverInitiateStatus.FBU_with_the_PCoA_as_source_IP_address, + code_default: 'Optional[int]' = None, + code_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long + code_reversed: 'bool' = False, + options: 'Optional[Option | list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes]]' = None, + **kwargs: 'Any') -> 'Schema_HandoverInitiateMessage': + """Make MH handover initiate (HI) message type. + + Args: + message: Message data model. + seq: Sequence number. + assign: Assigned address configuration flag. + buffer: Buffer flag. + proxy: Proxy flag. + forward: Forwarding flag. + code: Code. + code_default: Default code. + code_namespace: Code namespace. + code_reversed: Reverse code namespace. + options: Mobility options. + **kwargs: Arbitrary keyword arguments. + + Returns: + Constructed message type. + + """ + if message is not None: + seq = message.seq + assign = message.assign + buffer = message.buffer + proxy = message.proxy + forward = message.forward + code_val = message.code + options = message.options + else: + code_val = self._make_index(code, code_default, namespace=code_namespace, # type: ignore[assignment] + reversed=code_reversed, pack=False) + options = options or [] + + return Schema_HandoverInitiateMessage( + seq=seq, + flags={ + 'S': assign, + 'U': buffer, + 'P': proxy, + 'F': forward, + }, + code=code_val, + options=self._make_mh_options(options), + ) + + def _make_msg_hack(self, message: 'Optional[Data_HandoverAcknowledgeMessage]' = None, *, + seq: 'int' = 0, + buffer: 'bool' = False, + proxy: 'bool' = False, + forward: 'bool' = False, + code: 'Enum_HandoverACKStatus | StdlibEnum | AenumEnum | str | int' = Enum_HandoverACKStatus.Handover_Accepted_with_NCoA_valid, + code_default: 'Optional[int]' = None, + code_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long + code_reversed: 'bool' = False, + options: 'Optional[Option | list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes]]' = None, + **kwargs: 'Any') -> 'Schema_HandoverAcknowledgeMessage': + """Make MH handover acknowledge (HAck) message type. + + Args: + message: Message data model. + seq: Sequence number. + buffer: Buffer flag. + proxy: Proxy flag. + forward: Forwarding flag. + code: Code. + code_default: Default code. + code_namespace: Code namespace. + code_reversed: Reverse code namespace. + options: Mobility options. + **kwargs: Arbitrary keyword arguments. + + Returns: + Constructed message type. + + """ + if message is not None: + seq = message.seq + buffer = message.buffer + proxy = message.proxy + forward = message.forward + code_val = message.code + options = message.options + else: + code_val = self._make_index(code, code_default, namespace=code_namespace, # type: ignore[assignment] + reversed=code_reversed, pack=False) + options = options or [] + + return Schema_HandoverAcknowledgeMessage( + seq=seq, + flags={ + 'U': buffer, + 'P': proxy, + 'F': forward, + }, + code=code_val, + options=self._make_mh_options(options), + ) + # TODO: Implement other message types. def _make_mh_options(self, options: 'Option | list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes]') -> 'list[Schema_Option | bytes]': @@ -2654,6 +3528,110 @@ def _make_opt_ct(self, type: 'Enum_Option', option: 'Optional[Data_CareofTestOpt token=token, ) + def _make_opt_exp(self, type: 'Enum_Option', option: 'Optional[Data_ExperimentalMobilityOption]' = None, *, + data: 'bytes' = b'', + **kwargs: 'Any') -> 'Schema_ExperimentalMobilityOption': + """Make MH experimental mobility option. + + Args: + type: Option type. + option: Option data model. + data: Experimental data. + **kwargs: Arbitrary keyword arguments. + + Returns: + Constructed option schema. + + """ + if option is not None: + data = option.data + + return Schema_ExperimentalMobilityOption( + type=type, + length=len(data), + data=data, + ) + + def _make_opt_badf(self, type: 'Enum_Option', option: 'Optional[Data_BADFOption]' = None, *, + spi: 'int' = 0, + data: 'bytes' = b'', + **kwargs: 'Any') -> 'Schema_BADFOption': + """Make MH binding authorization data for FMIPv6 (BADF) option. + + Args: + type: Option type. + option: Option data model. + spi: Security parameter index. + data: Authenticator. + **kwargs: Arbitrary keyword arguments. + + Returns: + Constructed option schema. + + Note: + The ``length`` field counts the authenticator only, excluding the + SPI, c.f., :meth:`_read_opt_badf`. + + """ + if option is not None: + spi = option.spi + data = option.data + + if not data: + raise ProtocolError(f'{self.alias}: [OptNo {type}] invalid format') + + return Schema_BADFOption( + type=type, + length=len(data), + spi=spi, + data=data, + ) + + def _make_opt_ipv6_ap(self, type: 'Enum_Option', option: 'Optional[Data_IPv6AddressPrefixOption]' = None, *, + code: 'IPv6AddressPrefixCode | StdlibEnum | AenumEnum | str | int' = IPv6AddressPrefixCode.New_Care_of_Address, + code_default: 'Optional[int]' = None, + code_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long + code_reversed: 'bool' = False, + prefix_length: 'int' = 128, + address: 'bytes | str | int | IPv6Address' = '::', + **kwargs: 'Any') -> 'Schema_IPv6AddressPrefixOption': + """Make MH mobility header IPv6 address/prefix option. + + Args: + type: Option type. + option: Option data model. + code: Option code, c.f., + :class:`~pcapkit.protocols.internet.mh.IPv6AddressPrefixCode`. + code_default: Default option code. + code_namespace: Option code namespace. + code_reversed: Reverse option code namespace. + prefix_length: Prefix length. + address: IPv6 address/prefix. + **kwargs: Arbitrary keyword arguments. + + Returns: + Constructed option schema. + + """ + if option is not None: + code_val = option.code + prefix_length = option.prefix_length + address = option.address + else: + code_val = self._make_index(code, code_default, namespace=code_namespace, # type: ignore[assignment] + reversed=code_reversed, pack=False) + + if prefix_length > 128: + raise ProtocolError(f'{self.alias}: [OptNo {type}] invalid prefix length: {prefix_length}') + + return Schema_IPv6AddressPrefixOption( + type=type, + length=18, + code=code_val, + prefix_length=prefix_length, + address=address, + ) + # TODO: Implement other options. def _make_cga_extensions(self, extensions: 'Extension | list[Schema_CGAExtension | tuple[Enum_CGAExtension, dict[str, Any]] | bytes]') -> 'tuple[list[Schema_CGAExtension | bytes], int]': diff --git a/pcapkit/protocols/schema/internet/mh.py b/pcapkit/protocols/schema/internet/mh.py index 38af924385..d798e2c8cd 100644 --- a/pcapkit/protocols/schema/internet/mh.py +++ b/pcapkit/protocols/schema/internet/mh.py @@ -33,6 +33,8 @@ from pcapkit.const.mh.handover_ack_status import HandoverACKStatus as Enum_HandoverACKStatus from pcapkit.const.mh.handover_initiate_flag import \ HandoverInitiateFlag as Enum_HandoverInitiateFlag +from pcapkit.const.mh.handover_initiate_status import \ + HandoverInitiateStatus as Enum_HandoverInitiateStatus from pcapkit.const.mh.home_address_reply import HomeAddressReply as Enum_HomeAddressReply from pcapkit.const.mh.lla_code import LLACode as Enum_LLACode from pcapkit.const.mh.lma_mag_suboption import \ @@ -68,14 +70,16 @@ 'Packet', 'UnknownMessage', 'BindingRefreshRequestMessage', 'HomeTestInitMessage', 'CareofTestInitMessage', 'HomeTestMessage', 'CareofTestMessage', 'BindingUpdateMessage', 'BindingAcknowledgementMessage', - 'BindingErrorMessage', + 'BindingErrorMessage', 'FastBindingUpdateMessage', 'FastBindingAcknowledgmentMessage', + 'FastNeighborAdvertisementMessage', 'ExperimentalMessage', 'HandoverInitiateMessage', + 'HandoverAcknowledgeMessage', 'Option', 'UnassignedOption', 'PadOption', 'BindingRefreshAdviceOption', 'AlternateCareofAddressOption', 'NonceIndicesOption', 'AuthorizationDataOption', 'MobileNetworkPrefixOption', 'LinkLayerAddressOption', 'MNIDOption', 'AuthOption', 'MesgIDOption', 'CGAParametersRequestOption', 'CGAParametersOption', 'SignatureOption', 'PermanentHomeKeygenTokenOption', 'CareofTestInitOption', - 'CareofTestOption', + 'CareofTestOption', 'ExperimentalMobilityOption', 'BADFOption', 'IPv6AddressPrefixOption', 'CGAParameter', @@ -118,6 +122,34 @@ class BindingAcknowledgementMessageFlags(TypedDict): K: 'int' + class FastBindingUpdateMessageFlags(TypedDict): + """Flags for :attr:`FastBindingUpdateMessage.flags`.""" + + A: 'int' + H: 'int' + L: 'int' + K: 'int' + + class FastBindingAcknowledgmentMessageFlags(TypedDict): + """Flags for :attr:`FastBindingAcknowledgmentMessage.flags`.""" + + K: 'int' + + class HandoverInitiateMessageFlags(TypedDict): + """Flags for :attr:`HandoverInitiateMessage.flags`.""" + + S: 'int' + U: 'int' + P: 'int' + F: 'int' + + class HandoverAcknowledgeMessageFlags(TypedDict): + """Flags for :attr:`HandoverAcknowledgeMessage.flags`.""" + + U: 'int' + P: 'int' + F: 'int' + def mh_data_selector(pkt: 'dict[str, Any]') -> 'Field': """Selector function for :attr:`MH.data` field. @@ -505,6 +537,72 @@ class CareofTestOption(Option, code=Enum_Option.Care_of_Test): def __init__(self, type: 'Enum_Option', length: 'int', token: 'bytes') -> 'None': ... +@schema_final +class ExperimentalMobilityOption(Option, code=Enum_Option.Experimental_Mobility_Option): + """Header schema for MH Experimental Mobility options.""" + + #: Experimental data. + data: 'bytes' = BytesField(length=lambda pkt: pkt['length']) + + if TYPE_CHECKING: + def __init__(self, type: 'Enum_Option', length: 'int', data: 'bytes') -> 'None': ... + + +@schema_final +class BADFOption(Option, code=Enum_Option.Binding_Authorization_Data_for_FMIPv6): + """Header schema for MH Binding Authorization Data for FMIPv6 (BADF) options. + + Note: + Per :rfc:`5568#section-6.4.5`, the ``length`` field of this option counts + the :attr:`data` (authenticator) bytes **only** -- it excludes the 4-byte + :attr:`spi` field, unlike every other mobility option, whose length covers + all option data. The option is therefore 6 bytes longer than its declared + length, which is why :rfc:`5568` also requires it to be the **last** + mobility option present. + + """ + + #: Security parameter index. ``0`` is reserved for authenticators computed + #: using SEND-based handover keys. + spi: 'int' = UInt32Field() + #: Authenticator. + data: 'bytes' = BytesField(length=lambda pkt: pkt['length']) + + if TYPE_CHECKING: + def __init__(self, type: 'Enum_Option', length: 'int', spi: 'int', data: 'bytes') -> 'None': ... + + +@schema_final +class IPv6AddressPrefixOption(Option, code=Enum_Option.Mobility_Header_IPv6_Address_Prefix): + """Header schema for MH Mobility Header IPv6 Address/Prefix options. + + Note: + The option codes of :rfc:`5568#section-6.4.2` are defined inline by the + RFC with no IANA registry behind them, so their enumeration lives in + :class:`pcapkit.protocols.internet.mh.IPv6AddressPrefixCode` rather than + in :mod:`pcapkit.const.mh`. Being module-local, it cannot be imported + here without a circular import, so :attr:`code` carries the raw wire + integer and the enumeration is applied when + :meth:`~pcapkit.protocols.internet.mh.MH._read_opt_ipv6_ap` builds the + data model -- the same split as + :class:`pcapkit.protocols.misc.pcapng.PacketDirection` and the + ``epb_flags`` option schema. + + """ + + #: Option code, c.f., + #: :class:`pcapkit.protocols.internet.mh.IPv6AddressPrefixCode`. + code: 'int' = UInt8Field() + #: Prefix length. + prefix_length: 'int' = UInt8Field() + #: IPv6 address/prefix. + address: 'IPv6Address' = IPv6AddressField() + + if TYPE_CHECKING: + def __init__(self, type: 'Enum_Option', length: 'int', code: 'int', prefix_length: 'int', + address: 'IPv6Address | int | bytes | str') -> 'None': ... + + # TODO: Implement other options. @@ -715,4 +813,162 @@ def __init__(self, status: 'Enum_BindingError', home: 'IPv6Address | str | int | options: 'list[Option | bytes]') -> 'None': ... +@schema_final +class FastBindingUpdateMessage(Packet, code=Enum_Packet.Fast_Binding_Update): + """Header schema for MH Fast Binding Update (FBU) messages.""" + + #: Sequence number. + seq: 'int' = UInt16Field() + #: Flags. + flags: 'FastBindingUpdateMessageFlags' = BitField(length=2, namespace={ + 'A': (0, 1), + 'H': (1, 1), + 'L': (2, 1), + 'K': (3, 1), + }) + #: Lifetime. One time unit is 4 seconds. + lifetime: 'int' = UInt16Field() + #: Mobility options. + options: 'list[Option]' = OptionField( + length=lambda pkt: pkt['__length__'], + base_schema=Option, + type_name='type', + registry=Option.registry, + eool=None, + ) + + if TYPE_CHECKING: + def __init__(self, seq: 'int', flags: 'FastBindingUpdateMessageFlags', + lifetime: 'int', options: 'list[Option | bytes]') -> 'None': ... + + +@schema_final +class FastBindingAcknowledgmentMessage(Packet, code=Enum_Packet.Fast_Binding_Acknowledgment): + """Header schema for MH Fast Binding Acknowledgment (FBack) messages. + + Note: + The status values of :rfc:`5568#section-6.2.3` are defined inline by the + RFC and are absent from the IANA *Status Codes* registry, so their + enumeration lives in + :class:`pcapkit.protocols.internet.mh.FastBindingAcknowledgmentStatus` + rather than in :mod:`pcapkit.const.mh`. Being module-local, it cannot be + imported here without a circular import, so :attr:`status` carries the + raw wire integer and the enumeration is applied when + :meth:`~pcapkit.protocols.internet.mh.MH._read_msg_fback` builds the data + model. + + """ + + #: Status, c.f., + #: :class:`pcapkit.protocols.internet.mh.FastBindingAcknowledgmentStatus`. + status: 'int' = UInt8Field() + #: Flags. + flags: 'FastBindingAcknowledgmentMessageFlags' = BitField(length=1, namespace={ + 'K': (0, 1), + }) + #: Sequence number. + seq: 'int' = UInt16Field() + #: Lifetime. One time unit is 4 seconds. + lifetime: 'int' = UInt16Field() + #: Mobility options. + options: 'list[Option]' = OptionField( + length=lambda pkt: pkt['__length__'], + base_schema=Option, + type_name='type', + registry=Option.registry, + eool=None, + ) + + if TYPE_CHECKING: + def __init__(self, status: 'int', flags: 'FastBindingAcknowledgmentMessageFlags', + seq: 'int', lifetime: 'int', options: 'list[Option | bytes]') -> 'None': ... + + +@schema_final +class FastNeighborAdvertisementMessage(Packet, code=Enum_Packet.Fast_Neighbor_Advertisement): + """Header schema for MH Fast Neighbor Advertisement (FNA) messages.""" + + #: Reserved. + reserved: 'bytes' = PaddingField(length=2) + #: Mobility options. + options: 'list[Option]' = OptionField( + length=lambda pkt: pkt['__length__'], + base_schema=Option, + type_name='type', + registry=Option.registry, + eool=None, + ) + + if TYPE_CHECKING: + def __init__(self, options: 'list[Option | bytes]') -> 'None': ... + + +@schema_final +class ExperimentalMessage(Packet, code=Enum_Packet.Experimental_Mobility_Header): + """Header schema for MH Experimental Mobility Header messages.""" + + #: Experimental message data. + data: 'bytes' = BytesField(length=lambda pkt: pkt['__length__']) + + if TYPE_CHECKING: + def __init__(self, data: 'bytes') -> 'None': ... + + +@schema_final +class HandoverInitiateMessage(Packet, code=Enum_Packet.Handover_Initiate_Message): + """Header schema for MH Handover Initiate (HI) messages.""" + + #: Sequence number. + seq: 'int' = UInt16Field() + #: Flags. + flags: 'HandoverInitiateMessageFlags' = BitField(length=1, namespace={ + 'S': (0, 1), + 'U': (1, 1), + 'P': (2, 1), + 'F': (3, 1), + }) + #: Code. + code: 'Enum_HandoverInitiateStatus' = EnumField(length=1, namespace=Enum_HandoverInitiateStatus) + #: Mobility options. + options: 'list[Option]' = OptionField( + length=lambda pkt: pkt['__length__'], + base_schema=Option, + type_name='type', + registry=Option.registry, + eool=None, + ) + + if TYPE_CHECKING: + def __init__(self, seq: 'int', flags: 'HandoverInitiateMessageFlags', + code: 'Enum_HandoverInitiateStatus', options: 'list[Option | bytes]') -> 'None': ... + + +@schema_final +class HandoverAcknowledgeMessage(Packet, code=Enum_Packet.Handover_Acknowledge_Message): + """Header schema for MH Handover Acknowledge (HAck) messages.""" + + #: Sequence number. + seq: 'int' = UInt16Field() + #: Flags. + flags: 'HandoverAcknowledgeMessageFlags' = BitField(length=1, namespace={ + 'U': (0, 1), + 'P': (1, 1), + 'F': (2, 1), + }) + #: Code. + code: 'Enum_HandoverACKStatus' = EnumField(length=1, namespace=Enum_HandoverACKStatus) + #: Mobility options. + options: 'list[Option]' = OptionField( + length=lambda pkt: pkt['__length__'], + base_schema=Option, + type_name='type', + registry=Option.registry, + eool=None, + ) + + if TYPE_CHECKING: + def __init__(self, seq: 'int', flags: 'HandoverAcknowledgeMessageFlags', + code: 'Enum_HandoverACKStatus', options: 'list[Option | bytes]') -> 'None': ... + + # TODO: Implement other message types. diff --git a/tests/protocols/internet/test_mh_unit.py b/tests/protocols/internet/test_mh_unit.py index 343becfef4..f8f424bbba 100644 --- a/tests/protocols/internet/test_mh_unit.py +++ b/tests/protocols/internet/test_mh_unit.py @@ -1,6 +1,7 @@ from __future__ import annotations import datetime +import io from ipaddress import ip_address, ip_network import importlib.util from types import SimpleNamespace @@ -799,6 +800,640 @@ def test_mh_schema_selectors_and_option_post_process_branches(self) -> None: tz=datetime.timezone.utc, )) + def test_mh_fmipv6_message_readers_and_constructors(self) -> None: + from pcapkit.const.mh.handover_ack_status import HandoverACKStatus + from pcapkit.const.mh.handover_initiate_status import HandoverInitiateStatus + from pcapkit.const.mh.packet import Packet + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.mh import MH, FastBindingAcknowledgmentStatus + + proto = object.__new__(MH) + proto._read_mh_options = mock.Mock(return_value='opts') + proto._make_mh_options = mock.Mock(return_value=['made']) + + def header(type_: Packet) -> SimpleNamespace: + return SimpleNamespace(next=TransType.UDP, length=5, type=type_, chksum=b'\x12\x34') + + # RFC 5568, section 6.2.2 -- FBU is identical to the RFC 6275 BU, so the + # lifetime is carried in units of 4 seconds. + fbu = proto._read_msg_fbu(SimpleNamespace(seq=0x1234, + flags={'A': 1, 'H': 1, 'L': 0, 'K': 1}, + lifetime=10, options=[]), + header=header(Packet.Fast_Binding_Update)) + self.assertEqual(fbu.seq, 0x1234) + self.assertTrue(fbu.ack) + self.assertTrue(fbu.home) + self.assertFalse(fbu.lla_compat) + self.assertTrue(fbu.key_mngt) + self.assertEqual(fbu.lifetime, datetime.timedelta(seconds=40)) + self.assertEqual(fbu.options, 'opts') + + # RFC 5568, section 6.2.3 -- status 1 is "FBU accepted but NCoA is invalid", + # which is *not* what StatusCode(1) means, hence the module-local enum. + fback = proto._read_msg_fback(SimpleNamespace(status=1, flags={'K': 1}, seq=0x1234, + lifetime=10, options=[]), + header=header(Packet.Fast_Binding_Acknowledgment)) + self.assertEqual(fback.status, 1) + self.assertIs(fback.status, + FastBindingAcknowledgmentStatus.Fast_Binding_Update_accepted_but_NCoA_is_invalid) + self.assertNotIsInstance(fback.status, type(HandoverACKStatus.Administratively_prohibited)) + self.assertTrue(fback.key_mngt) + self.assertEqual(fback.seq, 0x1234) + self.assertEqual(fback.lifetime, datetime.timedelta(seconds=40)) + + # RFC 4068, section 6.3.3 -- two reserved octets, then mobility options. + fna = proto._read_msg_fna(SimpleNamespace(options=[]), + header=header(Packet.Fast_Neighbor_Advertisement)) + self.assertEqual(fna.options, 'opts') + self.assertEqual(fna.type, Packet.Fast_Neighbor_Advertisement) + + # RFC 5096, section 3 -- opaque message data, no fields of its own. + emh = proto._read_msg_emh(SimpleNamespace(data=b'\xde\xad\xbe\xef'), + header=header(Packet.Experimental_Mobility_Header)) + self.assertEqual(emh.data, b'\xde\xad\xbe\xef') + + # RFC 5568, section 6.2.1.1 and RFC 5949, section 6.1.1. + hi = proto._read_msg_hi(SimpleNamespace( + seq=0x0102, + flags={'S': 1, 'U': 0, 'P': 1, 'F': 0}, + code=HandoverInitiateStatus.FBU_whose_source_IP_address_is_not_PCoA, + options=[], + ), header=header(Packet.Handover_Initiate_Message)) + self.assertEqual(hi.seq, 0x0102) + self.assertTrue(hi.assign) + self.assertFalse(hi.buffer) + self.assertTrue(hi.proxy) + self.assertFalse(hi.forward) + self.assertEqual(hi.code, HandoverInitiateStatus.FBU_whose_source_IP_address_is_not_PCoA) + + # RFC 5568, section 6.2.1.2 and RFC 5949, section 6.1.2. + hack = proto._read_msg_hack(SimpleNamespace( + seq=0x0102, + flags={'U': 1, 'P': 0, 'F': 1}, + code=HandoverACKStatus.Handover_Accepted_NCoA_assigned, + options=[], + ), header=header(Packet.Handover_Acknowledge_Message)) + self.assertEqual(hack.seq, 0x0102) + self.assertTrue(hack.buffer) + self.assertFalse(hack.proxy) + self.assertTrue(hack.forward) + self.assertEqual(hack.code, HandoverACKStatus.Handover_Accepted_NCoA_assigned) + + # constructors, both from keyword arguments and from a data model + made_fbu = proto._make_msg_fbu(None, seq=7, ack=True, home=True, key_mngt=True, + lifetime=datetime.timedelta(seconds=40), options=[]) + self.assertEqual(made_fbu.seq, 7) + self.assertEqual(made_fbu.lifetime, 10) + self.assertEqual(made_fbu.flags, {'A': True, 'H': True, 'L': False, 'K': True}) + self.assertEqual(proto._make_msg_fbu(SimpleNamespace( + seq=8, ack=False, home=True, lla_compat=True, key_mngt=False, + lifetime=datetime.timedelta(seconds=8), options=[], + )).seq, 8) + + made_fback = proto._make_msg_fback(None, status=131, key_mngt=True, seq=9, lifetime=40, + options=[]) + self.assertEqual(made_fback.status, 131) + self.assertEqual(made_fback.lifetime, 10) + self.assertEqual(proto._make_msg_fback( + None, status=FastBindingAcknowledgmentStatus.Insufficient_resources, options=[], + ).status, 130) + self.assertEqual(proto._make_msg_fback( + None, status='Administratively_prohibited', options=[], + status_namespace=FastBindingAcknowledgmentStatus, + ).status, 129) + self.assertEqual(proto._make_msg_fback(SimpleNamespace( + status=FastBindingAcknowledgmentStatus.Fast_Binding_Update_accepted_but_NCoA_is_invalid, + key_mngt=False, seq=10, lifetime=datetime.timedelta(seconds=8), options=[], + )).seq, 10) + + self.assertEqual(proto._make_msg_fna(None, options=[]).options, ['made']) + self.assertEqual(proto._make_msg_fna(SimpleNamespace(options=[])).options, ['made']) + + self.assertEqual(proto._make_msg_emh(None).data, b'\x00\x00') + self.assertEqual(proto._make_msg_emh(None, data=b'raw').data, b'raw') + self.assertEqual(proto._make_msg_emh(SimpleNamespace(data=b'model')).data, b'model') + + made_hi = proto._make_msg_hi(None, seq=11, assign=True, buffer=False, proxy=True, + forward=False, code=1, options=[]) + self.assertEqual(made_hi.seq, 11) + self.assertEqual(made_hi.code, HandoverInitiateStatus.FBU_whose_source_IP_address_is_not_PCoA) + self.assertEqual(made_hi.flags, {'S': True, 'U': False, 'P': True, 'F': False}) + self.assertEqual(proto._make_msg_hi(SimpleNamespace( + seq=12, assign=False, buffer=True, proxy=False, forward=True, + code=HandoverInitiateStatus.All_available_context_transferred, options=[], + )).seq, 12) + + made_hack = proto._make_msg_hack(None, seq=13, buffer=True, proxy=False, forward=True, + code=130, options=[]) + self.assertEqual(made_hack.seq, 13) + self.assertEqual(made_hack.code, HandoverACKStatus.Insufficient_resources) + self.assertEqual(made_hack.flags, {'U': True, 'P': False, 'F': True}) + self.assertEqual(proto._make_msg_hack(SimpleNamespace( + seq=14, buffer=False, proxy=True, forward=False, + code=HandoverACKStatus.Handover_Accepted_use_PCoA, options=[], + )).seq, 14) + + def test_mh_fmipv6_option_readers_constructors_and_guards(self) -> None: + from pcapkit.const.mh.option import Option + from pcapkit.corekit.multidict import OrderedMultiDict + from pcapkit.protocols.data.internet import mh as data + from pcapkit.protocols.internet.mh import MH, IPv6AddressPrefixCode + from pcapkit.protocols.schema.internet import mh as schema + from pcapkit.utilities.exceptions import ProtocolError + + proto = object.__new__(MH) + options = OrderedMultiDict() + + # RFC 5096, section 4 -- opaque experimental data. + exp = proto._read_opt_exp( + schema.ExperimentalMobilityOption(type=Option.Experimental_Mobility_Option, + length=6, data=b'\x01\x02\x03\x04\x05\x06'), + options=options, + ) + self.assertEqual(exp.data, b'\x01\x02\x03\x04\x05\x06') + self.assertEqual(exp.length, 8) + + # RFC 5568, section 6.4.5 -- the option length counts the authenticator only, + # so the reported total length is 6 bytes longer (2 header + 4 SPI). + badf = proto._read_opt_badf( + schema.BADFOption(type=Option.Binding_Authorization_Data_for_FMIPv6, + length=12, spi=0xdeadbeef, data=bytes(range(12))), + options=options, + ) + self.assertEqual(badf.spi, 0xdeadbeef) + self.assertEqual(badf.data, bytes(range(12))) + self.assertEqual(badf.length, 18) + + # RFC 5568, section 6.4.2 (type corrected to 34 by errata ID 1816). + apfx = proto._read_opt_ipv6_ap( + schema.IPv6AddressPrefixOption(type=Option.Mobility_Header_IPv6_Address_Prefix, + length=18, code=4, prefix_length=64, + address='2001:db8:2::'), + options=options, + ) + self.assertEqual(apfx.code, 4) + self.assertIs(apfx.code, IPv6AddressPrefixCode.NAR_Prefix) + self.assertEqual(apfx.prefix_length, 64) + self.assertEqual(str(apfx.address), '2001:db8:2::') + self.assertEqual(apfx.length, 20) + + for reader, bad_schema in [ + # an authenticator of zero bytes cannot authenticate anything + (proto._read_opt_badf, schema.BADFOption( + type=Option.Binding_Authorization_Data_for_FMIPv6, + length=0, spi=1, data=b'')), + # the address/prefix option is a fixed 18 bytes of option data + (proto._read_opt_ipv6_ap, schema.IPv6AddressPrefixOption( + type=Option.Mobility_Header_IPv6_Address_Prefix, + length=17, code=2, prefix_length=64, address='2001:db8:2::')), + # "The value ranges from 0 to 128", RFC 5568, section 6.4.2 + (proto._read_opt_ipv6_ap, schema.IPv6AddressPrefixOption( + type=Option.Mobility_Header_IPv6_Address_Prefix, + length=18, code=2, prefix_length=129, address='2001:db8:2::')), + ]: + with self.assertRaises(ProtocolError): + reader(bad_schema, options=options) + + self.assertEqual(proto._make_opt_exp(Option.Experimental_Mobility_Option, + data=b'\x01\x02').length, 2) + self.assertEqual(proto._make_opt_exp( + Option.Experimental_Mobility_Option, + data.ExperimentalMobilityOption(type=Option.Experimental_Mobility_Option, + length=4, data=b'\x03\x04'), + ).data, b'\x03\x04') + + made_badf = proto._make_opt_badf(Option.Binding_Authorization_Data_for_FMIPv6, + spi=0xdeadbeef, data=bytes(range(12))) + self.assertEqual(made_badf.length, 12) + self.assertEqual(made_badf.spi, 0xdeadbeef) + self.assertEqual(proto._make_opt_badf( + Option.Binding_Authorization_Data_for_FMIPv6, + data.BADFOption(type=Option.Binding_Authorization_Data_for_FMIPv6, + length=18, spi=7, data=bytes(range(12))), + ).spi, 7) + with self.assertRaises(ProtocolError): + proto._make_opt_badf(Option.Binding_Authorization_Data_for_FMIPv6, data=b'') + + made_apfx = proto._make_opt_ipv6_ap(Option.Mobility_Header_IPv6_Address_Prefix, + code=1, prefix_length=128, address='2001:db8:1::2') + self.assertEqual(made_apfx.length, 18) + self.assertEqual(made_apfx.code, 1) + self.assertEqual(proto._make_opt_ipv6_ap( + Option.Mobility_Header_IPv6_Address_Prefix, + code=IPv6AddressPrefixCode.NAR_IP_address, + ).code, 3) + self.assertEqual(proto._make_opt_ipv6_ap( + Option.Mobility_Header_IPv6_Address_Prefix, code='Old_Care_of_Address', + code_namespace=IPv6AddressPrefixCode, + ).code, 1) + self.assertEqual(proto._make_opt_ipv6_ap( + Option.Mobility_Header_IPv6_Address_Prefix, + data.IPv6AddressPrefixOption(type=Option.Mobility_Header_IPv6_Address_Prefix, + length=20, code=3, prefix_length=64, + address=ip_address('2001:db8:3::')), + ).code, 3) + with self.assertRaises(ProtocolError): + proto._make_opt_ipv6_ap(Option.Mobility_Header_IPv6_Address_Prefix, + prefix_length=129) + + def test_mh_fmipv6_wire_format_matches_rfc(self) -> None: + """Parse hand-written byte strings laid out straight from the RFC figures.""" + from pcapkit.const.mh.handover_ack_status import HandoverACKStatus + from pcapkit.const.mh.handover_initiate_status import HandoverInitiateStatus + from pcapkit.const.mh.lla_code import LLACode + from pcapkit.const.mh.option import Option + from pcapkit.const.mh.packet import Packet + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.mh import (MH, FastBindingAcknowledgmentStatus, + IPv6AddressPrefixCode) + + def parse(hexstr: str) -> object: + raw = bytes.fromhex(hexstr) + self.assertEqual(len(raw) % 8, 0, 'mobility header must be 8-octet aligned') + return MH(io.BytesIO(raw), len(raw), extension=True).info + + badf_type = Option.Binding_Authorization_Data_for_FMIPv6 + apfx_type = Option.Mobility_Header_IPv6_Address_Prefix + acoa_type = Option.Alternate_Care_of_Address + mhlla_type = Option.Mobility_Header_Link_Layer_Address_option + + with self.subTest('FBU, RFC 5568 section 6.2.2'): + fbu = parse( + '1105' '08' '00' '1234' # next, hdr len, type, resv, cksum + '1234' 'd000' '000a' # seq, A|H|L|K + resv, lifetime + '0310' '20010db8000100000000000000000002' # alternate care-of address + '150c' 'deadbeef' '000102030405060708090a0b' # BADF: SPI + authenticator + ) + self.assertEqual(fbu.next, TransType.UDP) + self.assertEqual(fbu.type, Packet.Fast_Binding_Update) + self.assertEqual(fbu.length, 48) + self.assertEqual(fbu.seq, 0x1234) + self.assertTrue(fbu.ack) + self.assertTrue(fbu.home) + self.assertFalse(fbu.lla_compat) + self.assertTrue(fbu.key_mngt) + self.assertEqual(fbu.lifetime, datetime.timedelta(seconds=40)) + self.assertEqual(str(fbu.options[acoa_type].address), '2001:db8:1::2') + self.assertEqual(fbu.options[badf_type].spi, 0xdeadbeef) + self.assertEqual(fbu.options[badf_type].data, bytes(range(12))) + self.assertEqual(fbu.options[badf_type].length, 18) + + with self.subTest('FBack, RFC 5568 section 6.2.3'): + fback = parse( + '1105' '09' '00' '1234' + '01' '80' '1234' '000a' # status, K + resv, seq, lifetime + '0310' '20010db8000100000000000000000002' + '150c' '00000000' '000102030405060708090a0b' # SPI 0: SEND-based handover key + ) + self.assertEqual(fback.type, Packet.Fast_Binding_Acknowledgment) + self.assertIs( + fback.status, + FastBindingAcknowledgmentStatus.Fast_Binding_Update_accepted_but_NCoA_is_invalid, + ) + self.assertTrue(fback.key_mngt) + self.assertEqual(fback.seq, 0x1234) + self.assertEqual(fback.lifetime, datetime.timedelta(seconds=40)) + self.assertEqual(fback.options[badf_type].spi, 0) + + with self.subTest('FNA, RFC 4068 section 6.3.3'): + fna = parse( + '1102' '0a' '00' '1234' + '0000' # reserved + '0707' '02' '001122334455' # MH-LLA, code 2 = LLA of the MN + '0105' '0000000000' # PadN + ) + self.assertEqual(fna.type, Packet.Fast_Neighbor_Advertisement) + self.assertEqual(fna.length, 24) + self.assertEqual(fna.options[mhlla_type].code, LLACode.MH) + self.assertEqual(fna.options[mhlla_type].lla, b'\x00\x11\x22\x33\x44\x55') + self.assertEqual(fna.options[Option.PadN].length, 7) + + with self.subTest('Experimental Mobility Header, RFC 5096 section 3'): + emh = parse('1101' '0b' '00' '1234' '00010203040506070809') + self.assertEqual(emh.type, Packet.Experimental_Mobility_Header) + self.assertEqual(emh.length, 16) + self.assertEqual(emh.data, bytes(range(10))) + + with self.subTest('HI, RFC 5568 section 6.2.1.1 / RFC 5949 section 6.1.1'): + hi = parse( + '1104' '0e' '00' '1234' + '0102' 'c0' '01' # seq, S|U|P|F + resv, code + '2212' '02' '40' '20010db8000200000000000000000000' # addr/prefix: new CoA, /64 + '0108' '0000000000000000' # PadN + ) + self.assertEqual(hi.type, Packet.Handover_Initiate_Message) + self.assertEqual(hi.length, 40) + self.assertEqual(hi.seq, 0x0102) + self.assertTrue(hi.assign) + self.assertTrue(hi.buffer) + self.assertFalse(hi.proxy) + self.assertFalse(hi.forward) + self.assertEqual(hi.code, HandoverInitiateStatus.FBU_whose_source_IP_address_is_not_PCoA) + self.assertIs(hi.options[apfx_type].code, IPv6AddressPrefixCode.New_Care_of_Address) + self.assertEqual(hi.options[apfx_type].prefix_length, 64) + self.assertEqual(str(hi.options[apfx_type].address), '2001:db8:2::') + self.assertEqual(hi.options[apfx_type].length, 20) + + with self.subTest('HAck, RFC 5568 section 6.2.1.2 / RFC 5949 section 6.1.2'): + hack = parse( + '1104' '0f' '00' '1234' + '0102' 'a0' '02' # seq, U|P|F + resv, code + '2212' '02' '80' '20010db8000200000000000000000005' + '0108' '0000000000000000' + ) + self.assertEqual(hack.type, Packet.Handover_Acknowledge_Message) + self.assertEqual(hack.seq, 0x0102) + self.assertTrue(hack.buffer) + self.assertFalse(hack.proxy) + self.assertTrue(hack.forward) + self.assertEqual(hack.code, HandoverACKStatus.Handover_Accepted_NCoA_assigned) + self.assertEqual(hack.options[apfx_type].prefix_length, 128) + self.assertEqual(str(hack.options[apfx_type].address), '2001:db8:2::5') + + with self.subTest('Experimental Mobility option, RFC 5096 section 4'): + brr = parse('1101' '00' '00' '1234' '0000' '1206' '010203040506') + exp = brr.options[Option.Experimental_Mobility_Option] + self.assertEqual(exp.data, b'\x01\x02\x03\x04\x05\x06') + self.assertEqual(exp.length, 8) + + def test_mh_fmipv6_round_trip_is_byte_identical(self) -> None: + """``make`` then ``read`` then ``make`` again must reproduce the same bytes.""" + from pcapkit.const.mh.handover_ack_status import HandoverACKStatus + from pcapkit.const.mh.handover_initiate_status import HandoverInitiateStatus + from pcapkit.const.mh.option import Option + from pcapkit.const.mh.packet import Packet + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.mh import (MH, FastBindingAcknowledgmentStatus, + IPv6AddressPrefixCode) + + badf_type = Option.Binding_Authorization_Data_for_FMIPv6 + apfx_type = Option.Mobility_Header_IPv6_Address_Prefix + acoa_type = Option.Alternate_Care_of_Address + mhlla_type = Option.Mobility_Header_Link_Layer_Address_option + exp_type = Option.Experimental_Mobility_Option + + # NOTE: every case below is chosen to be 8-octet aligned without a padding + # option, since ``_make_opt_pad`` round-trips a ``PadN`` data model two bytes + # too long -- a pre-existing defect of the RFC 6275 padding option, outside + # the FMIPv6 message and option types under test here. + cases = [ + ('FBU', Packet.Fast_Binding_Update, { + 'seq': 0x1234, 'ack': True, 'home': True, 'lla_compat': False, + 'key_mngt': True, 'lifetime': 40, + 'options': [(acoa_type, {'address': '2001:db8:1::2'}), + (badf_type, {'spi': 0xdeadbeef, 'data': bytes(range(12))})], + }, {'seq': 0x1234, 'ack': True, 'home': True, 'lla_compat': False, + 'key_mngt': True, 'lifetime': datetime.timedelta(seconds=40)}), + # the status goes in as an enum member and must come back as one + ('FBack', Packet.Fast_Binding_Acknowledgment, { + 'status': FastBindingAcknowledgmentStatus.Incorrect_interface_identifier_length, + 'key_mngt': True, 'seq': 0x1234, 'lifetime': 40, + 'options': [(acoa_type, {'address': '2001:db8:1::2'}), + (badf_type, {'spi': 0, 'data': bytes(range(12))})], + }, {'status': FastBindingAcknowledgmentStatus.Incorrect_interface_identifier_length, + 'key_mngt': True, 'seq': 0x1234, + 'lifetime': datetime.timedelta(seconds=40)}), + # ... and as a bare integer, for a value the RFC leaves unassigned + ('FBack, unassigned status', Packet.Fast_Binding_Acknowledgment, { + 'status': 77, 'key_mngt': False, 'seq': 1, 'lifetime': 4, + 'options': [(acoa_type, {'address': '2001:db8:1::3'}), + (badf_type, {'spi': 1, 'data': bytes(range(12))})], + }, {'status': 77, 'key_mngt': False, 'seq': 1, + 'lifetime': datetime.timedelta(seconds=4)}), + ('FNA', Packet.Fast_Neighbor_Advertisement, { + 'options': [(mhlla_type, {'address': b'\x00\x11\x22\x33\x44'})], + }, {}), + ('Experimental', Packet.Experimental_Mobility_Header, { + 'data': bytes(range(10)), + }, {'data': bytes(range(10))}), + ('HI', Packet.Handover_Initiate_Message, { + 'seq': 0x0102, 'assign': True, 'buffer': True, 'proxy': False, + 'forward': False, + 'code': HandoverInitiateStatus.FBU_whose_source_IP_address_is_not_PCoA, + 'options': [(apfx_type, {'code': IPv6AddressPrefixCode.New_Care_of_Address, + 'prefix_length': 64, + 'address': '2001:db8:2::'}), + (mhlla_type, {'address': b'\x00\x11\x22\x33\x44\x55\x66'})], + }, {'seq': 0x0102, 'assign': True, 'buffer': True, 'proxy': False, + 'forward': False, + 'code': HandoverInitiateStatus.FBU_whose_source_IP_address_is_not_PCoA}), + ('HAck', Packet.Handover_Acknowledge_Message, { + 'seq': 0x0102, 'buffer': True, 'proxy': False, 'forward': True, + 'code': HandoverACKStatus.Handover_Accepted_NCoA_assigned, + 'options': [(apfx_type, {'code': 2, 'prefix_length': 128, + 'address': '2001:db8:2::5'}), + (mhlla_type, {'address': b'\x00\x11\x22\x33\x44\x55\x66'})], + }, {'seq': 0x0102, 'buffer': True, 'proxy': False, 'forward': True, + 'code': HandoverACKStatus.Handover_Accepted_NCoA_assigned}), + ('Experimental option', Packet.Binding_Refresh_Request, { + 'options': [(exp_type, {'data': b'\x01\x02\x03\x04\x05\x06'})], + }, {}), + ] + + for name, packet_type, payload, expected in cases: + with self.subTest(name): + raw = bytes(MH(next=TransType.UDP, chksum=b'\x12\x34', + type=packet_type, data=payload)) + self.assertEqual(len(raw) % 8, 0) + + parsed = MH(io.BytesIO(raw), len(raw), extension=True).info + self.assertEqual(parsed.type, packet_type) + self.assertEqual(parsed.length, len(raw)) + for field, value in expected.items(): + self.assertEqual(getattr(parsed, field), value, field) + + rebuilt = bytes(MH(next=parsed.next, type=parsed.type, + chksum=parsed.chksum, data=parsed)) + self.assertEqual(rebuilt, raw) + + def test_mh_rfc5568_local_enums_cover_unregistered_value_sets(self) -> None: + """:rfc:`5568` defines two value sets inline, with no IANA registry. + + Both are therefore enumerated in :mod:`pcapkit.protocols.internet.mh` + itself rather than in :mod:`pcapkit.const.mh`, and both have to tolerate + the values the RFC leaves unassigned -- a capture in the wild carries + whatever it carries, so an unassigned code must parse rather than raise. + """ + from pcapkit.const.mh.option import Option + from pcapkit.const.mh.packet import Packet + from pcapkit.const.mh.status_code import StatusCode + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.mh import (MH, FastBindingAcknowledgmentStatus, + IPv6AddressPrefixCode) + + apfx_type = Option.Mobility_Header_IPv6_Address_Prefix + badf_type = Option.Binding_Authorization_Data_for_FMIPv6 + acoa_type = Option.Alternate_Care_of_Address + mhlla_type = Option.Mobility_Header_Link_Layer_Address_option + + # The membership assertions come first: ``_missing_`` extends the class in + # place, so any unassigned lookup below would otherwise show up here. + with self.subTest('FBack status members, RFC 5568 section 6.2.3'): + self.assertEqual( + {int(member): member.name for member in FastBindingAcknowledgmentStatus}, + {0: 'Fast_Binding_Update_accepted', + 1: 'Fast_Binding_Update_accepted_but_NCoA_is_invalid', + 128: 'Reason_unspecified', + 129: 'Administratively_prohibited', + 130: 'Insufficient_resources', + 131: 'Incorrect_interface_identifier_length'}, + ) + + with self.subTest('address/prefix option code members, RFC 5568 section 6.4.2'): + self.assertEqual( + {int(member): member.name for member in IPv6AddressPrefixCode}, + {1: 'Old_Care_of_Address', 2: 'New_Care_of_Address', + 3: 'NAR_IP_address', 4: 'NAR_Prefix'}, + ) + + with self.subTest('the collision that rules out reusing StatusCode'): + # RFC 6275 means something else by both of these, which is why the + # registry enumeration cannot stand in for the RFC 5568 one. + self.assertEqual(StatusCode.Accepted_but_prefix_discovery_necessary, 1) + self.assertEqual(StatusCode.Home_registration_not_supported, 131) + self.assertNotIsInstance(FastBindingAcknowledgmentStatus(1), StatusCode) + self.assertNotIsInstance(FastBindingAcknowledgmentStatus(131), StatusCode) + + def fback(status: int) -> object: + raw = bytes.fromhex( + '1105' '09' '00' '1234' + f'{status:02x}' '80' '1234' '000a' + '0310' '20010db8000100000000000000000002' + '150c' '00000000' '000102030405060708090a0b' + ) + return MH(io.BytesIO(raw), len(raw), extension=True).info + + def handover_initiate(code: int) -> object: + raw = bytes.fromhex( + '1104' '0e' '00' '1234' + '0102' 'c0' '01' + '2212' f'{code:02x}' '40' '20010db8000200000000000000000000' + '0108' '0000000000000000' + ) + return MH(io.BytesIO(raw), len(raw), extension=True).info + + with self.subTest('assigned values parse to the right member'): + for wire, member in [ + (0, FastBindingAcknowledgmentStatus.Fast_Binding_Update_accepted), + (1, FastBindingAcknowledgmentStatus.Fast_Binding_Update_accepted_but_NCoA_is_invalid), + (128, FastBindingAcknowledgmentStatus.Reason_unspecified), + (129, FastBindingAcknowledgmentStatus.Administratively_prohibited), + (130, FastBindingAcknowledgmentStatus.Insufficient_resources), + (131, FastBindingAcknowledgmentStatus.Incorrect_interface_identifier_length), + ]: + self.assertIs(fback(wire).status, member, wire) + + for wire, member in [ + (1, IPv6AddressPrefixCode.Old_Care_of_Address), + (2, IPv6AddressPrefixCode.New_Care_of_Address), + (3, IPv6AddressPrefixCode.NAR_IP_address), + (4, IPv6AddressPrefixCode.NAR_Prefix), + ]: + self.assertIs(handover_initiate(wire).options[apfx_type].code, member, wire) + + with self.subTest('unassigned values parse without raising'): + # 2..127 is the unassigned half of the "accepted" range, 132..255 of + # the "rejected" one; both must survive a parse. + for wire in (2, 77, 127, 132, 255): + status = fback(wire).status + self.assertIsInstance(status, FastBindingAcknowledgmentStatus) + self.assertEqual(status, wire) + self.assertEqual(status.name, 'Unassigned_%d' % wire) + + # RFC 5568 assigns 1 through 4 only, so 0 and 5..255 are unassigned. + for wire in (0, 5, 200, 255): + code = handover_initiate(wire).options[apfx_type].code + self.assertIsInstance(code, IPv6AddressPrefixCode) + self.assertEqual(code, wire) + self.assertEqual(code.name, 'Unassigned_%d' % wire) + + with self.subTest('out-of-octet values are still rejected'): + for enum_cls in (FastBindingAcknowledgmentStatus, IPv6AddressPrefixCode): + for value in (-1, 256): + with self.assertRaises(ValueError): + enum_cls(value) + + with self.subTest('get() backports string and integer lookups'): + self.assertIs(FastBindingAcknowledgmentStatus.get(130), + FastBindingAcknowledgmentStatus.Insufficient_resources) + self.assertIs(FastBindingAcknowledgmentStatus.get('Reason_unspecified'), + FastBindingAcknowledgmentStatus.Reason_unspecified) + self.assertEqual(FastBindingAcknowledgmentStatus.get('Vendor_specific', 200), 200) + self.assertIs(IPv6AddressPrefixCode.get(4), IPv6AddressPrefixCode.NAR_Prefix) + self.assertIs(IPv6AddressPrefixCode.get('New_Care_of_Address'), + IPv6AddressPrefixCode.New_Care_of_Address) + self.assertEqual(IPv6AddressPrefixCode.get('Vendor_specific', 201), 201) + + with self.subTest('make round trips both enums byte-identically'): + for status, code in [ + (FastBindingAcknowledgmentStatus.Insufficient_resources, + IPv6AddressPrefixCode.NAR_Prefix), + # unassigned on the wire, and handed in as a bare integer + (99, 250), + ]: + fback_raw = bytes(MH(next=TransType.UDP, chksum=b'\x12\x34', + type=Packet.Fast_Binding_Acknowledgment, + data={'status': status, 'key_mngt': True, 'seq': 0x1234, + 'lifetime': 40, + 'options': [ + (acoa_type, {'address': '2001:db8:1::2'}), + (badf_type, {'spi': 0, + 'data': bytes(range(12))})]})) + self.assertEqual(fback_raw[6], int(status)) + parsed = MH(io.BytesIO(fback_raw), len(fback_raw), extension=True).info + self.assertIsInstance(parsed.status, FastBindingAcknowledgmentStatus) + self.assertEqual(parsed.status, int(status)) + self.assertEqual(bytes(MH(next=parsed.next, type=parsed.type, + chksum=parsed.chksum, data=parsed)), fback_raw) + + hi_raw = bytes(MH(next=TransType.UDP, chksum=b'\x12\x34', + type=Packet.Handover_Initiate_Message, + data={'seq': 0x0102, 'assign': True, 'buffer': True, + 'proxy': False, 'forward': False, 'code': 1, + 'options': [ + (apfx_type, {'code': code, 'prefix_length': 64, + 'address': '2001:db8:2::'}), + (mhlla_type, + {'address': b'\x00\x11\x22\x33\x44\x55\x66'})]})) + parsed = MH(io.BytesIO(hi_raw), len(hi_raw), extension=True).info + self.assertIsInstance(parsed.options[apfx_type].code, IPv6AddressPrefixCode) + self.assertEqual(parsed.options[apfx_type].code, int(code)) + self.assertEqual(bytes(MH(next=parsed.next, type=parsed.type, + chksum=parsed.chksum, data=parsed)), hi_raw) + + def test_mh_message_flags_pack_each_bit_independently(self) -> None: + """Regression test: a cleared flag must not be emitted as a set bit. + + :class:`~pcapkit.corekit.fields.strings.BitField` used to seed its per-bit + buffer with NUL bytes and then truth-test it, so the ASCII ``b'0'`` written + for a cleared bit read back as set and every named bit came out as ``1`` + regardless of the value handed in. + """ + from pcapkit.const.mh.packet import Packet + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.internet.mh import MH + + # A|H|L|K, one flag set at a time, c.f. RFC 5568 section 6.2.2. + for flag, octets in [('ack', b'\x80\x00'), ('home', b'\x40\x00'), + ('lla_compat', b'\x20\x00'), ('key_mngt', b'\x10\x00')]: + with self.subTest(flag): + raw = bytes(MH(next=TransType.UDP, chksum=b'\x00\x00', + type=Packet.Fast_Binding_Update, + data={'seq': 0, 'ack': False, 'home': False, + 'lla_compat': False, 'key_mngt': False, + flag: True, 'lifetime': 4, 'options': []})) + self.assertEqual(raw[8:10], octets) + + # U|P|F of the HAck message, c.f. RFC 5949 section 6.1.2. + for flag, octet in [('buffer', 0x80), ('proxy', 0x40), ('forward', 0x20)]: + with self.subTest(flag): + raw = bytes(MH(next=TransType.UDP, chksum=b'\x00\x00', + type=Packet.Handover_Acknowledge_Message, + data={'seq': 0, 'buffer': False, 'proxy': False, + 'forward': False, flag: True, 'code': 0, + 'options': []})) + self.assertEqual(raw[8], octet) + if __name__ == '__main__': unittest.main()