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()