working on MH protocol impl (message types)

JarryShaw · JarryShaw · commit 27467f1094d7 · 2023-05-23T21:13:21.000+08:00
diff --git a/pcapkit/protocols/data/internet/__init__.py b/pcapkit/protocols/data/internet/__init__.py
@@ -201,6 +201,8 @@
 from pcapkit.protocols.data.internet.mh import CareofTestInitMessage as MH_CareofTestInitMessage
 from pcapkit.protocols.data.internet.mh import HomeTestMessage as MH_HomeTestMessage
 from pcapkit.protocols.data.internet.mh import CareofTestMessage as MH_CareofTestMessage
+from pcapkit.protocols.data.internet.mh import BindingUpdateMessage as MH_BindingUpdateMessage
+from pcapkit.protocols.data.internet.mh import BindingAcknowledgementMessage as MH_BindingAcknowledgementMessage
 
 __all__ = [
     # Authentication Header
diff --git a/pcapkit/protocols/data/internet/mh.py b/pcapkit/protocols/data/internet/mh.py
@@ -2,12 +2,13 @@
 """data model for MH protocol"""
 
 from typing import TYPE_CHECKING
+from pcapkit.const.mh import status_code
 
 from pcapkit.corekit.infoclass import info_final
 from pcapkit.protocols.data.data import Data
 
 if TYPE_CHECKING:
-    from datetime import datetime as dt_type
+    from datetime import datetime as dt_type, timedelta
     from ipaddress import IPv6Address, IPv6Network
 
     from pcapkit.const.mh.access_type import AccessType as Enum_AccessType
@@ -60,11 +61,12 @@
     from pcapkit.const.reg.transtype import TransType
     from pcapkit.corekit.multidict import OrderedMultiDict
     from pcapkit.protocols.internet.mh import NTPTimestamp
+    from pcapkit.const.mh.binding_error import BindingError as Enum_BindingError
 
 __all__ = [
     'MH',
     'UnknownMessage', 'BindingRefreshRequestMessage', 'HomeTestInitMessage', 'CareofTestInitMessage',
-    'HomeTestMessage', 'CareofTestMessage',
+    'HomeTestMessage', 'CareofTestMessage', 'BindingUpdateMessage', 'BindingAcknowledgementMessage',
 
     'Option',
     'UnassignedOption', 'PadOption', 'BindRefreshAdviceOption', 'AlternateCareofAddressOption',
@@ -182,6 +184,50 @@ def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'by
                      options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ...
 
 
+@info_final
+class BindingUpdateMessage(MH):
+    """Data model for MH Binding Update message type."""
+
+    #: Sequence number.
+    seq: 'int'
+    #: Acknowledge flag.
+    ack: 'bool'
+    #: home registration flag.
+    home: 'bool'
+    #: Link-local address compability 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 BindingAcknowledgementMessage(MH):
+    """Data model for MH Binding Acknowledge (BA) message type."""
+
+    #: Status.
+    status: 'Enum_StatusCode'
+    #: 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: 'Enum_StatusCode', key_mngt: 'bool', seq: 'int', lifetime: 'timedelta',
+                     options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ...
 
 
 
diff --git a/pcapkit/protocols/internet/mh.py b/pcapkit/protocols/internet/mh.py
@@ -28,6 +28,7 @@
 from tokenize import Name
 from typing import TYPE_CHECKING, cast, overload
 
+from pcapkit.const.mh.binding_error import BindingError as Enum_BindingError
 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
@@ -146,9 +147,13 @@
 from pcapkit.protocols.data.internet.mh import HomeTestMessage as Data_HomeTestMessage
 from pcapkit.protocols.schema.internet.mh import CareofTestMessage as Schema_CareofTestMessage
 from pcapkit.protocols.data.internet.mh import CareofTestMessage as Data_CareofTestMessage
+from pcapkit.protocols.schema.internet.mh import BindingUpdateMessage as Schema_BindingUpdateMessage
+from pcapkit.protocols.data.internet.mh import BindingUpdateMessage as Data_BindingUpdateMessage
+from pcapkit.protocols.schema.internet.mh import BindingAcknowledgementMessage as Schema_BindingAcknowledgementMessage
+from pcapkit.protocols.data.internet.mh import BindingAcknowledgementMessage as Data_BindingAcknowledgementMessage
 
 if TYPE_CHECKING:
-    from datetime import datetime as dt_type
+    from datetime import timedelta, datetime as dt_type
     from enum import IntEnum as StdlibEnum
     from ipaddress import IPv6Address, IPv6Network
     from typing import IO, Any, Callable, DefaultDict, NoReturn, Optional, Type
@@ -207,6 +212,8 @@ class MH(Internet[Data_MH, Schema_MH],
             Enum_Packet.Care_of_Test_Init: 'coti',
             Enum_Packet.Home_Test: 'hot',
             Enum_Packet.Care_of_Test: 'cot',
+            Enum_Packet.Binding_Update: 'bu',
+            Enum_Packet.Binding_Acknowledgement: 'ba',
         },
     )  # type: DefaultDict[Enum_Packet | int, str | tuple[PacketParser, PacketConstructor]]
 
@@ -384,9 +391,9 @@ def make(self,
             Constructed packet data.
 
         """
-        next_val = self._make_index(next, next_default, namespace=next_namespace,  # type: ignore[call-overload]
+        next_val = self._make_index(next, next_default, namespace=next_namespace,
                                     reversed=next_reversed, pack=False)
-        type_val = self._make_index(type, type_default, namespace=type_namespace,  # type: ignore[call-overload]
+        type_val = self._make_index(type, type_default, namespace=type_namespace,
                                     reversed=type_reversed, pack=False)
 
         if isinstance(data, bytes):
@@ -716,6 +723,90 @@ def _read_msg_cot(self, schema: 'Schema_CareofTestMessage', *,
         )
         return data
 
+    def _read_msg_bu(self, schema: 'Schema_BindingUpdateMessage', *,
+                     header: 'Schema_MH') -> 'Data_BindingUpdateMessage':
+        """Read MH binding update (BU) message type.
+
+        Structure of MH Binding Update Message [:rfc:`6275`]:
+
+        .. code-block:: text
+
+                                           +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+                                           |          Sequence #           |
+           +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+           |A|H|L|K|        Reserved       |           Lifetime            |
+           +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+           |                                                               |
+           .                                                               .
+           .                        Mobility Options                       .
+           .                                                               .
+           |                                                               |
+           +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+
+        Args:
+            schema: Parsed message type schema.
+            header: Parsed MH header schema.
+
+        Returns:
+            Parsed message type data.
+
+        """
+        data = Data_BindingUpdateMessage(
+            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
+
+    def _read_msg_ba(self, schema: 'Schema_BindingAcknowledgementMessage', *,
+                     header: 'Schema_MH') -> 'Data_BindingAcknowledgementMessage':
+        """Read MH binding acknowledgement (BA) message type.
+
+        Structure of MH Binding Acknowledgement Message [:rfc:`6275`]:
+
+        .. code-block:: text
+
+                                           +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+                                           |    Status     |K|  Reserved   |
+           +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+           |           Sequence #          |           Lifetime            |
+           +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+           |                                                               |
+           .                                                               .
+           .                        Mobility Options                       .
+           .                                                               .
+           |                                                               |
+           +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+
+        Args:
+            schema: Parsed message type schema.
+            header: Parsed MH header schema.
+
+        Returns:
+            Parsed message type data.
+
+        """
+        data = Data_BindingAcknowledgementMessage(
+            next=header.next,
+            length=(header.length + 1) * 8,
+            type=header.type,
+            chksum=header.chksum,
+            status=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
+
 
 
 
@@ -1663,6 +1754,106 @@ def _make_msg_cot(self, message: 'Optional[Data_CareofTestMessage]', *,
             options=self._make_mh_options(options),
         )
 
+    def _make_msg_bu(self, message: 'Optional[Data_BindingUpdateMessage]', *,
+                     seq: 'int' = 0,
+                     ack: 'bool' = False,
+                     home: 'bool' = False,
+                     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_BindingUpdateMessage':
+        """Make MH care-of test (CoT) 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_BindingUpdateMessage(
+            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_ba(self, message: 'Optional[Data_BindingAcknowledgementMessage]', *,
+                     status: 'Enum_StatusCode | StdlibEnum | AenumEnum | str | int' = Enum_StatusCode.Binding_Update_accepted_Proxy_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_BindingAcknowledgementMessage':
+        """Make MH care-of test (CoT) message type.
+
+        Args:
+            message: Message data model.
+            status: Status code.
+            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_BindingAcknowledgementMessage(
+            status=status_val,
+            flags={
+                'K': key_mngt,
+            },
+            seq=seq,
+            lifetime=math.ceil(lifetime_val / 4),
+            options=self._make_mh_options(options),
+        )
+
 
 
 
@@ -1958,7 +2149,7 @@ def _make_opt_mn_id(self, type: 'Enum_Option', option: 'Optional[Data_MNIDOption
             subtype_val = option.subtype
             identifier = option.identifier
         else:
-            subtype_val = self._make_index(subtype, subtype_default, namespace=subtype_namespace,  # type: ignore[call-overload]
+            subtype_val = self._make_index(subtype, subtype_default, namespace=subtype_namespace,  # type: ignore[assignment]
                                            reversed=subtype_reversed, pack=False)
 
         if isinstance(identifier, ipaddress.IPv6Address):
@@ -2005,7 +2196,7 @@ def _make_opt_auth(self, type: 'Enum_Option', option: 'Optional[Data_AuthOption]
             spi = option.spi
             data = option.data
         else:
-            subtype_val = self._make_index(subtype, subtype_default, namespace=subtype_namespace,  # type: ignore[call-overload]
+            subtype_val = self._make_index(subtype, subtype_default, namespace=subtype_namespace,  # type: ignore[assignment]
                                            reversed=subtype_reversed, pack=False)
 
         if (len(data) + 6) % 4 != 0:
diff --git a/pcapkit/protocols/schema/internet/__init__.py b/pcapkit/protocols/schema/internet/__init__.py
@@ -200,6 +200,8 @@
 from pcapkit.protocols.schema.internet.mh import CareofTestInitMessage as MH_CareofTestInitMessage
 from pcapkit.protocols.schema.internet.mh import HomeTestMessage as MH_HomeTestMessage
 from pcapkit.protocols.schema.internet.mh import CareofTestMessage as MH_CareofTestMessage
+from pcapkit.protocols.schema.internet.mh import BindingUpdateMessage as MH_BindingUpdateMessage
+from pcapkit.protocols.schema.internet.mh import BindingAcknowledgementMessage as MH_BindingAcknowledgementMessage
 
 __all__ = [
     # Authentication Header
diff --git a/pcapkit/protocols/schema/internet/mh.py b/pcapkit/protocols/schema/internet/mh.py
