From f81c3927ab29c642477ba0ab2d963afb40347159 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Wed, 7 Oct 2026 15:18:31 -0400 Subject: [PATCH] fix(mh): keep zero-length PadN, reserved fields and BU/BA flag bits on rebuild - PadN with Option Length 0 is a valid two-octet pad (RFC 6275 6.2.3); it is no longer rejected on read or rewritten as Pad1 on make (#1324). - The fixed-header reserved octet, the BRR reserved field and the Handoff Indicator reserved octet are read as integers and written back (new data fields MH.reserved, BRR.msg_reserved, HI.reserved). - BU and BA model every flag IANA assigns (M R P F T B S D / R P T B S D) and keep the remaining reserved bits (flags_reserved). - IPv4 Home Address and Network-Identifier ANI sub-option keep their reserved bits (reserved). Closes #1324 Refs #1329 --- pcapkit/protocols/data/internet/mh.py | 110 +++++++++---- pcapkit/protocols/internet/mh.py | 152 +++++++++++++++++- pcapkit/protocols/schema/internet/mh.py | 51 +++++- .../internet/test_mh_roundtrip_unit.py | 148 +++++++++++++++++ tests/protocols/internet/test_mh_unit.py | 52 ++++-- 5 files changed, 455 insertions(+), 58 deletions(-) create mode 100644 tests/protocols/internet/test_mh_roundtrip_unit.py diff --git a/pcapkit/protocols/data/internet/mh.py b/pcapkit/protocols/data/internet/mh.py index cf44ac641f..a36e950231 100644 --- a/pcapkit/protocols/data/internet/mh.py +++ b/pcapkit/protocols/data/internet/mh.py @@ -145,6 +145,8 @@ class MH(Protocol): length: 'int' #: Mobility header type. type: 'Packet' + #: Reserved octet after the MH type. + reserved: 'int' #: Checksum. chksum: 'bytes' @@ -157,19 +159,22 @@ class UnknownMessage(MH): data: 'bytes' if TYPE_CHECKING: - def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', data: 'bytes') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,redefined-builtin,line-too-long + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', data: 'bytes') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,redefined-builtin,line-too-long @info_final class BindingRefreshRequestMessage(MH): """Data model for MH Binding Refresh Request (BRR) message type.""" + #: Reserved field of the message body, before the mobility options. Named + #: apart from :attr:`MH.reserved`, the reserved octet of the fixed header. + msg_reserved: 'int' #: 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': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,redefined-builtin,line-too-long + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', + msg_reserved: 'int', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,redefined-builtin,line-too-long @info_final @@ -182,7 +187,7 @@ class HomeTestInitMessage(MH): options: 'OrderedMultiDict[Enum_Option, Option]' if TYPE_CHECKING: - def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', cookie: 'bytes', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -196,7 +201,7 @@ class CareofTestInitMessage(MH): options: 'OrderedMultiDict[Enum_Option, Option]' if TYPE_CHECKING: - def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', cookie: 'bytes', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -214,7 +219,7 @@ class HomeTestMessage(MH): options: 'OrderedMultiDict[Enum_Option, Option]' if TYPE_CHECKING: - def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', nonce_index: 'int', cookie: 'bytes', token: 'bytes', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -233,7 +238,7 @@ class CareofTestMessage(MH): options: 'OrderedMultiDict[Enum_Option, Option]' if TYPE_CHECKING: - def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', nonce_index: 'int', cookie: 'bytes', token: 'bytes', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -252,15 +257,36 @@ class BindingUpdateMessage(MH): lla_compat: 'bool' #: Key management mobility capability flag. key_mngt: 'bool' + #: MAP registration flag (``M``) [:rfc:`4140`]. + map_reg: 'bool' + #: Mobile router flag (``R``) [:rfc:`3963`]. + mobile_router: 'bool' + #: Proxy registration flag (``P``) [:rfc:`5213`]. + proxy_reg: 'bool' + #: Forcing UDP encapsulation flag (``F``) [:rfc:`5555`]. + udp_encap: 'bool' + #: TLV-header format flag (``T``) [:rfc:`5845`]. + tlv_format: 'bool' + #: Bulk binding update flag (``B``) [:rfc:`6602`]. + bulk_binding: 'bool' + #: Multicast signaling flag (``S``) [:rfc:`7161`]. + multicast: 'bool' + #: DMM flag (``D``) [:rfc:`8885`]. + dmm: 'bool' + #: Reserved bits, i.e. the low four bits of the flags word. + flags_reserved: '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', + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', seq: 'int', ack: 'bool', home: 'bool', lla_compat: 'bool', key_mngt: 'bool', - lifetime: 'timedelta', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... + map_reg: 'bool', mobile_router: 'bool', proxy_reg: 'bool', udp_encap: 'bool', + tlv_format: 'bool', bulk_binding: 'bool', multicast: 'bool', dmm: 'bool', + flags_reserved: 'int', lifetime: 'timedelta', + options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @info_final @@ -271,6 +297,20 @@ class BindingAcknowledgementMessage(MH): status: 'Enum_StatusCode' #: Key management mobility capability flag. key_mngt: 'bool' + #: Mobile router flag (``R``) [:rfc:`3963`]. + mobile_router: 'bool' + #: Proxy registration flag (``P``) [:rfc:`5213`]. + proxy_reg: 'bool' + #: TLV-header format flag (``T``) [:rfc:`5845`]. + tlv_format: 'bool' + #: Bulk binding update flag (``B``) [:rfc:`6602`]. + bulk_binding: 'bool' + #: Multicast signaling flag (``S``) [:rfc:`7161`]. + multicast: 'bool' + #: DMM flag (``D``) [:rfc:`8885`]. + dmm: 'bool' + #: Reserved bit, i.e. the lowest bit of the flags octet. + flags_reserved: 'int' #: Sequence number. seq: 'int' #: Lifetime. @@ -279,8 +319,10 @@ class BindingAcknowledgementMessage(MH): 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', + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', + status: 'Enum_StatusCode', key_mngt: 'bool', mobile_router: 'bool', + proxy_reg: 'bool', tlv_format: 'bool', bulk_binding: 'bool', multicast: 'bool', + dmm: 'bool', flags_reserved: 'int', seq: 'int', lifetime: 'timedelta', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -296,7 +338,7 @@ class BindingErrorMessage(MH): options: 'OrderedMultiDict[Enum_Option, Option]' if TYPE_CHECKING: - def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', status: 'Enum_BindingError', home: 'IPv6Address', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -322,7 +364,7 @@ class FastBindingUpdateMessage(MH): options: 'OrderedMultiDict[Enum_Option, Option]' if TYPE_CHECKING: - def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', seq: 'int', ack: 'bool', home: 'bool', lla_compat: 'bool', key_mngt: 'bool', lifetime: 'timedelta', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -348,7 +390,7 @@ class FastBindingAcknowledgmentMessage(MH): options: 'OrderedMultiDict[Enum_Option, Option]' if TYPE_CHECKING: - def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', status: 'FastBindingAcknowledgmentStatus', key_mngt: 'bool', seq: 'int', lifetime: 'timedelta', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -362,7 +404,7 @@ class FastNeighborAdvertisementMessage(MH): options: 'OrderedMultiDict[Enum_Option, Option]' if TYPE_CHECKING: - def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -374,7 +416,7 @@ class ExperimentalMessage(MH): data: 'bytes' if TYPE_CHECKING: - def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', data: 'bytes') -> 'None': ... @@ -398,7 +440,7 @@ class HandoverInitiateMessage(MH): options: 'OrderedMultiDict[Enum_Option, Option]' if TYPE_CHECKING: - def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', seq: 'int', assign: 'bool', buffer: 'bool', proxy: 'bool', forward: 'bool', code: 'Enum_HandoverInitiateStatus', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -422,7 +464,7 @@ class HandoverAcknowledgeMessage(MH): options: 'OrderedMultiDict[Enum_Option, Option]' if TYPE_CHECKING: - def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', seq: 'int', buffer: 'bool', proxy: 'bool', forward: 'bool', code: 'Enum_HandoverACKStatus', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -443,7 +485,7 @@ class HeartbeatMessage(MH): options: 'OrderedMultiDict[Enum_Option, Option]' if TYPE_CHECKING: - def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', unsolicited: 'bool', response: 'bool', seq: 'int', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -459,7 +501,7 @@ class HomeAgentSwitchMessage(MH): options: 'OrderedMultiDict[Enum_Option, Option]' if TYPE_CHECKING: - def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', addresses: 'tuple[IPv6Address, ...]', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -492,7 +534,7 @@ class BindingRevocationMessage(MH): options: 'OrderedMultiDict[Enum_Option, Option]' if TYPE_CHECKING: - def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', br_type: 'Enum_BindingRevocation', code: 'Enum_RevocationTrigger | Enum_RevocationStatusCode', seq: 'int', proxy: 'bool', ipv4_hoa: 'bool', global_revocation: 'bool', @@ -512,7 +554,7 @@ class LocalizedRoutingInitiationMessage(MH): options: 'OrderedMultiDict[Enum_Option, Option]' if TYPE_CHECKING: - def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', seq: 'int', lifetime: 'timedelta', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -536,7 +578,7 @@ class LocalizedRoutingAcknowledgmentMessage(MH): options: 'OrderedMultiDict[Enum_Option, Option]' if TYPE_CHECKING: - def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', seq: 'int', unsolicited: 'bool', status: 'LocalizedRoutingStatus', lifetime: 'timedelta', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -558,7 +600,7 @@ class UpdateNotificationMessage(MH): options: 'OrderedMultiDict[Enum_Option, Option]' if TYPE_CHECKING: - def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', seq: 'int', reason: 'Enum_UpdateNotificationReason', ack: 'bool', retransmit: 'bool', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -576,7 +618,7 @@ class UpdateNotificationAcknowledgementMessage(MH): options: 'OrderedMultiDict[Enum_Option, Option]' if TYPE_CHECKING: - def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', seq: 'int', status: 'Enum_UpdateNotificationACKStatus', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -604,7 +646,7 @@ class FlowBindingMessage(MH): options: 'OrderedMultiDict[Enum_Option, Option]' if TYPE_CHECKING: - def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', fb_type: 'Enum_FlowBindingType', seq: 'int', code: 'Enum_FlowBindingIndicationTrigger | Enum_FlowBindingACKStatus', ack: 'bool', @@ -621,7 +663,7 @@ class SubscriptionQueryMessage(MH): options: 'OrderedMultiDict[Enum_Option, Option]' if TYPE_CHECKING: - def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', seq: 'int', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -639,7 +681,7 @@ class SubscriptionResponseMessage(MH): options: 'OrderedMultiDict[Enum_Option, Option]' if TYPE_CHECKING: - def __init__(self, next: 'TransType', length: 'int', type: 'Packet', chksum: 'bytes', + def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', seq: 'int', info: 'bool', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -985,11 +1027,13 @@ def __init__(self, type: 'Enum_Option', length: 'int', prefix_length: 'int', class HandoffIndicatorOption(Option): """Data model for MH Handoff Indicator option.""" + #: Reserved octet before the handoff indicator. + reserved: 'int' #: Handoff indicator. hi: 'Enum_HandoffType' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Option', length: 'int', hi: 'Enum_HandoffType') -> 'None': ... + def __init__(self, type: 'Enum_Option', length: 'int', reserved: 'int', hi: 'Enum_HandoffType') -> 'None': ... @info_final @@ -1111,10 +1155,12 @@ class IPv4HomeAddressOption(Option): address: 'IPv4Address' #: Mobile network prefix request flag. request_prefix: 'bool' + #: Reserved bits, i.e. the nine bits after the ``P`` flag. + reserved: 'int' if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', length: 'int', prefix_length: 'int', - address: 'IPv4Address', request_prefix: 'bool') -> 'None': ... + address: 'IPv4Address', request_prefix: 'bool', reserved: 'int') -> 'None': ... @info_final @@ -1542,13 +1588,15 @@ class ANINetworkIdentifierSuboption(ANISuboption): #: UTF-8 encoding flag for :attr:`net_name`. utf8: 'bool' + #: Reserved bits, i.e. the seven bits after the ``E`` flag. + reserved: 'int' #: Network name, e.g. an SSID or a PLMN identifier. net_name: 'bytes' #: Access-point name. ap_name: 'bytes' if TYPE_CHECKING: - def __init__(self, type: 'Enum_ANISuboption', length: 'int', utf8: 'bool', + def __init__(self, type: 'Enum_ANISuboption', length: 'int', utf8: 'bool', reserved: 'int', net_name: 'bytes', ap_name: 'bytes') -> 'None': ... diff --git a/pcapkit/protocols/internet/mh.py b/pcapkit/protocols/internet/mh.py index cd9e7da453..ec868ba929 100644 --- a/pcapkit/protocols/internet/mh.py +++ b/pcapkit/protocols/internet/mh.py @@ -1444,6 +1444,7 @@ def make(self, type_default: 'Optional[int]' = None, type_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long type_reversed: 'bool' = False, + reserved: 'int' = 0, chksum: 'bytes' = b'', data: 'bytes | Data_MH | Schema_Packet | dict[str, Any]' = b'\x00\x00', # minimum length payload: 'ProtocolBase | Schema | bytes' = b'', @@ -1459,6 +1460,7 @@ def make(self, type_default: Default value for Mobility Header type field. type_namespace: Namespace of Mobility Header type field. type_reversed: Whether the bits of Mobility Header type field is reversed. + reserved: Reserved octet after the MH type. chksum: Checksum. data: Message data. payload: Payload of next layer protocol. @@ -1504,6 +1506,7 @@ def make(self, next=next_val, length=(len(data_val) + 6) // 8 - 1, type=type_val, + reserved=reserved, chksum=chksum, data=data_val, payload=payload, @@ -1618,6 +1621,7 @@ def _make_data(cls, data: 'Data_MH') -> 'dict[str, Any]': # type: ignore[overri return { 'next': data.next, 'type': data.type, + 'reserved': data.reserved, 'chksum': data.chksum, 'data': data, 'payload': cls._make_payload(data), @@ -1639,6 +1643,7 @@ def _read_msg_unknown(self, schema: 'Schema_UnknownMessage', *, next=header.next, length=self._mh_message_length(header.length), type=header.type, + reserved=header.reserved, chksum=header.chksum, data=schema.data, ) @@ -1674,7 +1679,9 @@ def _read_msg_brr(self, schema: 'Schema_BindingRefreshRequestMessage', *, next=header.next, length=self._mh_message_length(header.length), type=header.type, + reserved=header.reserved, chksum=header.chksum, + msg_reserved=schema.reserved, options=self._read_mh_options(schema.options) ) return data @@ -1713,6 +1720,7 @@ def _read_msg_hoti(self, schema: 'Schema_HomeTestInitMessage', *, next=header.next, length=self._mh_message_length(header.length), type=header.type, + reserved=header.reserved, chksum=header.chksum, cookie=schema.cookie, options=self._read_mh_options(schema.options) @@ -1753,6 +1761,7 @@ def _read_msg_coti(self, schema: 'Schema_CareofTestInitMessage', *, next=header.next, length=self._mh_message_length(header.length), type=header.type, + reserved=header.reserved, chksum=header.chksum, cookie=schema.cookie, options=self._read_mh_options(schema.options) @@ -1797,6 +1806,7 @@ def _read_msg_hot(self, schema: 'Schema_HomeTestMessage', *, next=header.next, length=self._mh_message_length(header.length), type=header.type, + reserved=header.reserved, chksum=header.chksum, nonce_index=schema.nonce_index, cookie=schema.cookie, @@ -1843,6 +1853,7 @@ def _read_msg_cot(self, schema: 'Schema_CareofTestMessage', *, next=header.next, length=self._mh_message_length(header.length), type=header.type, + reserved=header.reserved, chksum=header.chksum, nonce_index=schema.nonce_index, cookie=schema.cookie, @@ -1871,6 +1882,10 @@ def _read_msg_bu(self, schema: 'Schema_BindingUpdateMessage', *, | | +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + Later RFCs assign eight more of the flag bits, ``M``, ``R``, ``P``, + ``F``, ``T``, ``B``, ``S`` and ``D`` in that order (IANA *Binding + Update Flags*); the low four bits remain reserved. + Args: schema: Parsed message type schema. header: Parsed MH header schema. @@ -1883,12 +1898,22 @@ def _read_msg_bu(self, schema: 'Schema_BindingUpdateMessage', *, next=header.next, length=self._mh_message_length(header.length), type=header.type, + reserved=header.reserved, 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']), + map_reg=bool(schema.flags['M']), + mobile_router=bool(schema.flags['R']), + proxy_reg=bool(schema.flags['P']), + udp_encap=bool(schema.flags['F']), + tlv_format=bool(schema.flags['T']), + bulk_binding=bool(schema.flags['B']), + multicast=bool(schema.flags['S']), + dmm=bool(schema.flags['D']), + flags_reserved=schema.flags['reserved'], lifetime=datetime.timedelta(seconds=schema.lifetime * 4), options=self._read_mh_options(schema.options), ) @@ -1914,6 +1939,10 @@ def _read_msg_ba(self, schema: 'Schema_BindingAcknowledgementMessage', *, | | +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + Later RFCs assign six more of the flag bits, ``R``, ``P``, ``T``, + ``B``, ``S`` and ``D`` in that order (IANA *Binding Acknowledgment + Flags*); the lowest bit remains reserved. + Args: schema: Parsed message type schema. header: Parsed MH header schema. @@ -1926,9 +1955,17 @@ def _read_msg_ba(self, schema: 'Schema_BindingAcknowledgementMessage', *, next=header.next, length=self._mh_message_length(header.length), type=header.type, + reserved=header.reserved, chksum=header.chksum, status=schema.status, key_mngt=bool(schema.flags['K']), + mobile_router=bool(schema.flags['R']), + proxy_reg=bool(schema.flags['P']), + tlv_format=bool(schema.flags['T']), + bulk_binding=bool(schema.flags['B']), + multicast=bool(schema.flags['S']), + dmm=bool(schema.flags['D']), + flags_reserved=schema.flags['reserved'], seq=schema.seq, lifetime=datetime.timedelta(seconds=schema.lifetime * 4), options=self._read_mh_options(schema.options), @@ -1972,6 +2009,7 @@ def _read_msg_be(self, schema: 'Schema_BindingErrorMessage', *, next=header.next, length=self._mh_message_length(header.length), type=header.type, + reserved=header.reserved, chksum=header.chksum, status=schema.status, home=schema.home, @@ -2019,6 +2057,7 @@ def _read_msg_fbu(self, schema: 'Schema_FastBindingUpdateMessage', *, next=header.next, length=self._mh_message_length(header.length), type=header.type, + reserved=header.reserved, chksum=header.chksum, seq=schema.seq, ack=bool(schema.flags['A']), @@ -2076,6 +2115,7 @@ def _read_msg_fback(self, schema: 'Schema_FastBindingAcknowledgmentMessage', *, next=header.next, length=self._mh_message_length(header.length), type=header.type, + reserved=header.reserved, chksum=header.chksum, status=FastBindingAcknowledgmentStatus(schema.status), key_mngt=bool(schema.flags['K']), @@ -2121,6 +2161,7 @@ def _read_msg_fna(self, schema: 'Schema_FastNeighborAdvertisementMessage', *, next=header.next, length=self._mh_message_length(header.length), type=header.type, + reserved=header.reserved, chksum=header.chksum, options=self._read_mh_options(schema.options), ) @@ -2164,6 +2205,7 @@ def _read_msg_emh(self, schema: 'Schema_ExperimentalMessage', *, next=header.next, length=self._mh_message_length(header.length), type=header.type, + reserved=header.reserved, chksum=header.chksum, data=schema.data, ) @@ -2207,6 +2249,7 @@ def _read_msg_hi(self, schema: 'Schema_HandoverInitiateMessage', *, next=header.next, length=self._mh_message_length(header.length), type=header.type, + reserved=header.reserved, chksum=header.chksum, seq=schema.seq, assign=bool(schema.flags['S']), @@ -2257,6 +2300,7 @@ def _read_msg_hack(self, schema: 'Schema_HandoverAcknowledgeMessage', *, next=header.next, length=self._mh_message_length(header.length), type=header.type, + reserved=header.reserved, chksum=header.chksum, seq=schema.seq, buffer=bool(schema.flags['U']), @@ -2317,6 +2361,7 @@ def _read_msg_has(self, schema: 'Schema_HomeAgentSwitchMessage', *, next=header.next, length=self._mh_message_length(header.length), type=header.type, + reserved=header.reserved, chksum=header.chksum, addresses=tuple(schema.addresses), options=self._read_mh_options(schema.options), @@ -2361,6 +2406,7 @@ def _read_msg_hb(self, schema: 'Schema_HeartbeatMessage', *, next=header.next, length=self._mh_message_length(header.length), type=header.type, + reserved=header.reserved, chksum=header.chksum, unsolicited=bool(schema.flags['U']), response=bool(schema.flags['R']), @@ -2428,6 +2474,7 @@ def _read_msg_brm(self, schema: 'Schema_BindingRevocationMessage', *, next=header.next, length=self._mh_message_length(header.length), type=header.type, + reserved=header.reserved, chksum=header.chksum, br_type=schema.br_type, code=schema.code, @@ -2477,6 +2524,7 @@ def _read_msg_lri(self, schema: 'Schema_LocalizedRoutingInitiationMessage', *, next=header.next, length=self._mh_message_length(header.length), type=header.type, + reserved=header.reserved, chksum=header.chksum, seq=schema.seq, lifetime=datetime.timedelta(seconds=schema.lifetime), @@ -2523,6 +2571,7 @@ def _read_msg_lra(self, schema: 'Schema_LocalizedRoutingAcknowledgmentMessage', next=header.next, length=self._mh_message_length(header.length), type=header.type, + reserved=header.reserved, chksum=header.chksum, seq=schema.seq, unsolicited=bool(schema.flags['U']), @@ -2568,6 +2617,7 @@ def _read_msg_upn(self, schema: 'Schema_UpdateNotificationMessage', *, next=header.next, length=self._mh_message_length(header.length), type=header.type, + reserved=header.reserved, chksum=header.chksum, seq=schema.seq, reason=schema.reason, @@ -2610,6 +2660,7 @@ def _read_msg_upa(self, schema: 'Schema_UpdateNotificationAcknowledgementMessage next=header.next, length=self._mh_message_length(header.length), type=header.type, + reserved=header.reserved, chksum=header.chksum, seq=schema.seq, status=schema.status, @@ -2672,6 +2723,7 @@ def _read_msg_fbm(self, schema: 'Schema_FlowBindingMessage', *, next=header.next, length=self._mh_message_length(header.length), type=header.type, + reserved=header.reserved, chksum=header.chksum, fb_type=schema.fb_type, seq=schema.seq, @@ -2716,6 +2768,7 @@ def _read_msg_sq(self, schema: 'Schema_SubscriptionQueryMessage', *, next=header.next, length=self._mh_message_length(header.length), type=header.type, + reserved=header.reserved, chksum=header.chksum, seq=schema.seq, options=self._read_mh_options(schema.options), @@ -2753,6 +2806,7 @@ def _read_msg_sr(self, schema: 'Schema_SubscriptionResponseMessage', *, next=header.next, length=self._mh_message_length(header.length), type=header.type, + reserved=header.reserved, chksum=header.chksum, seq=schema.seq, info=bool(schema.flags['I']), @@ -2899,7 +2953,9 @@ def _read_opt_pad(self, schema: 'Schema_PadOption', *, which sizes both the length octet and the padding data from the option type [:rfc:`6275#section-6.2.5`]; ``clen`` is therefore always ``0`` here for a parsed ``Pad1``, and the check below only - guards a schema built by hand. + guards a schema built by hand. A ``PadN`` whose ``Option Length`` + is ``0`` is a valid two-octet padding option + [:rfc:`6275#section-6.2.3`]. """ code, clen = schema.type, schema.length @@ -2908,8 +2964,6 @@ def _read_opt_pad(self, schema: 'Schema_PadOption', *, 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 @@ -3814,6 +3868,7 @@ def _read_opt_hi(self, schema: 'Schema_HandoffIndicatorOption', *, data = Data_HandoffIndicatorOption( type=schema.type, length=self._mh_option_length(schema.length), + reserved=schema.reserved, hi=schema.hi, ) return data @@ -4078,6 +4133,7 @@ def _read_opt_ipv4_hoa(self, schema: 'Schema_IPv4HomeAddressOption', *, prefix_length=prefix_length, address=schema.address, request_prefix=bool(schema.flags['P']), + reserved=schema.flags['reserved'], ) return data @@ -5132,6 +5188,7 @@ def _read_ani_suboptions( type=code, length=self._mh_option_length(net.length), utf8=bool(net.flags['E']), + reserved=net.flags['reserved'], net_name=net.net_name, ap_name=net.ap_name, ) # type: Data_ANISuboption @@ -6451,12 +6508,14 @@ def _make_msg_unknown(self, message: 'Optional[Data_UnknownMessage]' = None, *, ) def _make_msg_brr(self, message: 'Optional[Data_BindingRefreshRequestMessage]' = None, *, + msg_reserved: 'int' = 0, options: 'Optional[Option | list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes]]' = None, **kwargs: 'Any') -> 'Schema_BindingRefreshRequestMessage': """Make MH binding refresh request (BRR) message type. Args: message: Message data model. + msg_reserved: Reserved field before the mobility options. options: Mobility options. **kwargs: Arbitrary keyword arguments. @@ -6465,11 +6524,13 @@ def _make_msg_brr(self, message: 'Optional[Data_BindingRefreshRequestMessage]' = """ if message is not None: + msg_reserved = message.msg_reserved options = message.options else: options = options or [] return Schema_BindingRefreshRequestMessage( + reserved=msg_reserved, options=self._make_mh_options(options), ) @@ -6603,6 +6664,15 @@ def _make_msg_bu(self, message: 'Optional[Data_BindingUpdateMessage]' = None, *, home: 'bool' = False, lla_compat: 'bool' = False, key_mngt: 'bool' = False, + map_reg: 'bool' = False, + mobile_router: 'bool' = False, + proxy_reg: 'bool' = False, + udp_encap: 'bool' = False, + tlv_format: 'bool' = False, + bulk_binding: 'bool' = False, + multicast: 'bool' = False, + dmm: 'bool' = False, + flags_reserved: '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_BindingUpdateMessage': @@ -6615,6 +6685,16 @@ def _make_msg_bu(self, message: 'Optional[Data_BindingUpdateMessage]' = None, *, home: Home registration flag. lla_compat: LLA compatibility flag. key_mngt: Key management mobility option flag. + map_reg: MAP registration flag (``M``). + mobile_router: Mobile router flag (``R``). + proxy_reg: Proxy registration flag (``P``). + udp_encap: Forcing UDP encapsulation flag (``F``). + tlv_format: TLV-header format flag (``T``). + bulk_binding: Bulk binding update flag (``B``). + multicast: Multicast signaling flag (``S``). + dmm: DMM flag (``D``). + flags_reserved: Reserved bits, i.e. the low four bits of the flags + word. lifetime: Lifetime in seconds or timedelta. options: Mobility options. **kwargs: Arbitrary keyword arguments. @@ -6629,6 +6709,15 @@ def _make_msg_bu(self, message: 'Optional[Data_BindingUpdateMessage]' = None, *, home = message.home lla_compat = message.lla_compat key_mngt = message.key_mngt + map_reg = message.map_reg + mobile_router = message.mobile_router + proxy_reg = message.proxy_reg + udp_encap = message.udp_encap + tlv_format = message.tlv_format + bulk_binding = message.bulk_binding + multicast = message.multicast + dmm = message.dmm + flags_reserved = message.flags_reserved lifetime_val = math.ceil(message.lifetime.total_seconds()) options = message.options else: @@ -6642,6 +6731,15 @@ def _make_msg_bu(self, message: 'Optional[Data_BindingUpdateMessage]' = None, *, 'H': home, 'L': lla_compat, 'K': key_mngt, + 'M': map_reg, + 'R': mobile_router, + 'P': proxy_reg, + 'F': udp_encap, + 'T': tlv_format, + 'B': bulk_binding, + 'S': multicast, + 'D': dmm, + 'reserved': flags_reserved, }, lifetime=math.ceil(lifetime_val / 4), options=self._make_mh_options(options), @@ -6653,6 +6751,13 @@ def _make_msg_ba(self, message: 'Optional[Data_BindingAcknowledgementMessage]' = 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, + mobile_router: 'bool' = False, + proxy_reg: 'bool' = False, + tlv_format: 'bool' = False, + bulk_binding: 'bool' = False, + multicast: 'bool' = False, + dmm: 'bool' = False, + flags_reserved: 'int' = 0, seq: 'int' = 0, lifetime: 'int | timedelta' = 4, # reasonable default value options: 'Optional[Option | list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes]]' = None, @@ -6666,6 +6771,14 @@ def _make_msg_ba(self, message: 'Optional[Data_BindingAcknowledgementMessage]' = status_namespace: Status code namespace. status_reversed: Reverse status code namespace. key_mngt: Key management mobility option flag. + mobile_router: Mobile router flag (``R``). + proxy_reg: Proxy registration flag (``P``). + tlv_format: TLV-header format flag (``T``). + bulk_binding: Bulk binding update flag (``B``). + multicast: Multicast signaling flag (``S``). + dmm: DMM flag (``D``). + flags_reserved: Reserved bit, i.e. the lowest bit of the flags + octet. seq: Sequence number. lifetime: Lifetime in seconds or timedelta. options: Mobility options. @@ -6678,6 +6791,13 @@ def _make_msg_ba(self, message: 'Optional[Data_BindingAcknowledgementMessage]' = if message is not None: status_val = message.status key_mngt = message.key_mngt + mobile_router = message.mobile_router + proxy_reg = message.proxy_reg + tlv_format = message.tlv_format + bulk_binding = message.bulk_binding + multicast = message.multicast + dmm = message.dmm + flags_reserved = message.flags_reserved seq = message.seq lifetime_val = math.ceil(message.lifetime.total_seconds()) options = message.options @@ -6691,6 +6811,13 @@ def _make_msg_ba(self, message: 'Optional[Data_BindingAcknowledgementMessage]' = status=status_val, flags={ 'K': key_mngt, + 'R': mobile_router, + 'P': proxy_reg, + 'T': tlv_format, + 'B': bulk_binding, + 'S': multicast, + 'D': dmm, + 'reserved': flags_reserved, }, seq=seq, lifetime=math.ceil(lifetime_val / 4), @@ -7625,6 +7752,9 @@ def _make_opt_pad(self, type: 'Enum_Option', option: 'Optional[Data_PadOption]' for a ``Pad1``. Copying one into the other unconverted would make a re-made parsed ``PadN`` come back two octets too long. + A ``PadN`` with ``length`` ``0`` is a valid two-octet option + [:rfc:`6275#section-6.2.3`] and is kept as one. + """ if option is not None: length = 0 if option.type == Enum_Option.Pad1 else option.length - 2 @@ -7633,10 +7763,6 @@ def _make_opt_pad(self, type: 'Enum_Option', option: 'Optional[Data_PadOption]' # raise ProtocolError(f'{self.alias}: [OptNo {type}] invalid format') warn(f'{self.alias}: [OptNo {type}] invalid format', ProtocolWarning) type = Enum_Option.PadN # type: ignore[assignment] - if type == Enum_Option.PadN and length == 0: - # raise ProtocolError(f'{self.alias}: [OptNo {type}] invalid format') - warn(f'{self.alias}: [OptNo {type}] invalid format', ProtocolWarning) - type = Enum_Option.Pad1 # type: ignore[assignment] return Schema_PadOption( type=type, @@ -8522,6 +8648,7 @@ def _make_opt_hi(self, type: 'Enum_Option', option: 'Optional[Data_HandoffIndica hi_default: 'Optional[int]' = None, hi_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long hi_reversed: 'bool' = False, + reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_HandoffIndicatorOption': """Make MH handoff indicator option. @@ -8532,6 +8659,7 @@ def _make_opt_hi(self, type: 'Enum_Option', option: 'Optional[Data_HandoffIndica hi_default: Default handoff indicator. hi_namespace: Handoff indicator namespace. hi_reversed: Reverse handoff indicator namespace. + reserved: Reserved octet before the handoff indicator. **kwargs: Arbitrary keyword arguments. Returns: @@ -8540,6 +8668,7 @@ def _make_opt_hi(self, type: 'Enum_Option', option: 'Optional[Data_HandoffIndica """ if option is not None: hi_val = option.hi # type: Enum_HandoffType + reserved = option.reserved else: hi_val = self._make_index(hi, hi_default, namespace=hi_namespace, # type: ignore[assignment] reversed=hi_reversed, pack=False) @@ -8547,6 +8676,7 @@ def _make_opt_hi(self, type: 'Enum_Option', option: 'Optional[Data_HandoffIndica return Schema_HandoffIndicatorOption( type=type, length=2, + reserved=reserved, hi=hi_val, ) @@ -8701,6 +8831,7 @@ def _make_opt_ipv4_hoa(self, type: 'Enum_Option', option: 'Optional[Data_IPv4Hom prefix_length: 'int' = 32, address: 'bytes | str | int | IPv4Address' = '0.0.0.0', request_prefix: 'bool' = False, + reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_IPv4HomeAddressOption': """Make MH IPv4 home address option. @@ -8710,6 +8841,7 @@ def _make_opt_ipv4_hoa(self, type: 'Enum_Option', option: 'Optional[Data_IPv4Hom prefix_length: Prefix length; ``32`` for a single address. address: IPv4 home address. request_prefix: Mobile network prefix request flag. + reserved: Reserved bits, i.e. the nine bits after the ``P`` flag. **kwargs: Arbitrary keyword arguments. Returns: @@ -8720,6 +8852,7 @@ def _make_opt_ipv4_hoa(self, type: 'Enum_Option', option: 'Optional[Data_IPv4Hom prefix_length = option.prefix_length address = option.address request_prefix = option.request_prefix + reserved = option.reserved if prefix_length == 0 or prefix_length > 32: raise ProtocolError(f'{self.alias}: [OptNo {type}] invalid prefix length: {prefix_length}') @@ -8730,6 +8863,7 @@ def _make_opt_ipv4_hoa(self, type: 'Enum_Option', option: 'Optional[Data_IPv4Hom flags={ 'prefix_length': prefix_length, 'P': int(request_prefix), + 'reserved': reserved, }, address=address, ) @@ -9669,13 +9803,15 @@ def _make_ani_suboption(self, code: 'Enum_ANISuboption', if option is not None: net = cast('Data_ANINetworkIdentifierSuboption', option) utf8, net_name, ap_name = net.utf8, net.net_name, net.ap_name + reserved = net.reserved else: utf8 = cast('bool', kwargs.get('utf8', False)) + reserved = cast('int', kwargs.get('reserved', 0)) net_name = cast('bytes', kwargs.get('net_name', b'')) ap_name = cast('bytes', kwargs.get('ap_name', b'')) return Schema_ANINetworkIdentifierSuboption( type=code, length=3 + len(net_name) + len(ap_name), - flags={'E': int(utf8)}, net_name_len=len(net_name), net_name=net_name, + flags={'E': int(utf8), 'reserved': reserved}, net_name_len=len(net_name), net_name=net_name, ap_name_len=len(ap_name), ap_name=ap_name) if code == Enum_ANISuboption.Geo_Location: diff --git a/pcapkit/protocols/schema/internet/mh.py b/pcapkit/protocols/schema/internet/mh.py index deab01f395..80cf3664a2 100644 --- a/pcapkit/protocols/schema/internet/mh.py +++ b/pcapkit/protocols/schema/internet/mh.py @@ -156,11 +156,27 @@ class BindingUpdateMessageFlags(TypedDict): H: 'int' L: 'int' K: 'int' + M: 'int' + R: 'int' + P: 'int' + F: 'int' + T: 'int' + B: 'int' + S: 'int' + D: 'int' + reserved: 'int' class BindingAcknowledgementMessageFlags(TypedDict): """Flags for :attr:`BindingAcknowledgementMessage.flags`.""" K: 'int' + R: 'int' + P: 'int' + T: 'int' + B: 'int' + S: 'int' + D: 'int' + reserved: 'int' class FastBindingUpdateMessageFlags(TypedDict): """Flags for :attr:`FastBindingUpdateMessage.flags`.""" @@ -240,6 +256,7 @@ class IPv4HomeAddressOptionFlags(TypedDict): prefix_length: 'int' P: 'int' + reserved: 'int' class PrefixLengthOnly(TypedDict): """Prefix length packed alone in its own word, as in @@ -298,6 +315,7 @@ class ANINetworkIdentifierFlags(TypedDict): """Flags for :attr:`ANINetworkIdentifierSuboption.flags`.""" E: 'int' + reserved: 'int' class GeoLocationDegrees(TypedDict): """Latitude and longitude of :attr:`ANIGeoLocationSuboption.location`.""" @@ -562,7 +580,7 @@ class MH(Schema): #: MH type. type: 'Enum_Packet' = EnumField(length=1, namespace=Enum_Packet) #: Reserved. - reserved: 'bytes' = PaddingField(length=1) + reserved: 'int' = UInt8Field(default=0) #: Checksum. chksum: 'bytes' = BytesField(length=2) #: Message data. @@ -572,7 +590,8 @@ class MH(Schema): if TYPE_CHECKING: def __init__(self, next: 'Enum_TransType | int', length: 'int', type: 'Enum_Packet | int', - chksum: 'bytes', data: 'Packet | bytes', payload: 'bytes | ProtocolBase | Schema') -> 'None': ... + reserved: 'int', chksum: 'bytes', data: 'Packet | bytes', + payload: 'bytes | ProtocolBase | Schema') -> 'None': ... class Option(EnumSchema[Enum_Option]): @@ -1084,12 +1103,12 @@ class HandoffIndicatorOption(Option, code=Enum_Option.Handoff_Indicator_Option): """Header schema for MH Handoff Indicator options.""" #: Reserved. - reserved: 'bytes' = PaddingField(length=1) + reserved: 'int' = UInt8Field(default=0) #: Handoff indicator. hi: 'Enum_HandoffType' = EnumField(length=1, namespace=Enum_HandoffType) if TYPE_CHECKING: - def __init__(self, type: 'Enum_Option', length: 'int', hi: 'Enum_HandoffType') -> 'None': ... + def __init__(self, type: 'Enum_Option', length: 'int', reserved: 'int', hi: 'Enum_HandoffType') -> 'None': ... @schema_final @@ -1179,6 +1198,7 @@ class IPv4HomeAddressOption(Option, code=Enum_Option.IPv4_Home_Address): flags: 'IPv4HomeAddressOptionFlags' = BitField(length=2, namespace={ 'prefix_length': (0, 6), 'P': (6, 1), + 'reserved': (7, 9), }) #: IPv4 home address. address: 'IPv4Address' = IPv4AddressField() @@ -1771,6 +1791,7 @@ class ANINetworkIdentifierSuboption(ANISuboption, code=Enum_ANISuboption.Network #: Flags. flags: 'ANINetworkIdentifierFlags' = BitField(length=1, namespace={ 'E': (0, 1), + 'reserved': (1, 7), }) #: Network name length. net_name_len: 'int' = UInt8Field() @@ -2495,7 +2516,7 @@ class BindingRefreshRequestMessage(Packet, code=Enum_Packet.Binding_Refresh_Requ """Header schema for MH Binding Refresh Request (BRR) message.""" #: Reserved. - reserved: 'bytes' = PaddingField(length=2) + reserved: 'int' = UInt16Field(default=0) #: Mobility options. options: 'list[Option]' = OptionField( length=lambda pkt: pkt['__length__'], @@ -2506,7 +2527,7 @@ class BindingRefreshRequestMessage(Packet, code=Enum_Packet.Binding_Refresh_Requ ) if TYPE_CHECKING: - def __init__(self, options: 'list[Option | bytes]') -> 'None': ... + def __init__(self, reserved: 'int', options: 'list[Option | bytes]') -> 'None': ... @schema_final @@ -2611,6 +2632,17 @@ class BindingUpdateMessage(Packet, code=Enum_Packet.Binding_Update): 'H': (1, 1), 'L': (2, 1), 'K': (3, 1), + 'M': (4, 1), + 'R': (5, 1), + 'P': (6, 1), + 'F': (7, 1), + 'T': (8, 1), + 'B': (9, 1), + # NOTE: RFC 7161 draws ``S`` at bit 7, colliding with RFC 5555's ``F``; + # this follows IANA and RFC 8885, which place it at bit 10 (0x0020). + 'S': (10, 1), + 'D': (11, 1), + 'reserved': (12, 4), }) #: Lifetime. One time unit is 4 seconds. lifetime: 'int' = UInt16Field() @@ -2637,6 +2669,13 @@ class BindingAcknowledgementMessage(Packet, code=Enum_Packet.Binding_Acknowledge #: Flags. flags: 'BindingAcknowledgementMessageFlags' = BitField(length=1, namespace={ 'K': (0, 1), + 'R': (1, 1), + 'P': (2, 1), + 'T': (3, 1), + 'B': (4, 1), + 'S': (5, 1), + 'D': (6, 1), + 'reserved': (7, 1), }) #: Sequence number. seq: 'int' = UInt16Field() diff --git a/tests/protocols/internet/test_mh_roundtrip_unit.py b/tests/protocols/internet/test_mh_roundtrip_unit.py new file mode 100644 index 0000000000..dbde5fe1a9 --- /dev/null +++ b/tests/protocols/internet/test_mh_roundtrip_unit.py @@ -0,0 +1,148 @@ +# -*- coding: utf-8 -*- +"""MH zero-length ``PadN``, reserved fields and flag bits rebuild byte for byte. + +GitHub issue #1324: a ``PadN`` whose ``Option Length`` is ``0`` -- two octets +of padding, which :rfc:`6275#section-6.2.3` allows -- was rejected on read and +rewritten as a ``Pad1`` on make. + +GitHub issue #1329: the reserved octet of the fixed header, the Binding Refresh +Request reserved field, the Handoff Indicator reserved octet and the reserved +bits of the IPv4 Home Address option and the Network-Identifier ANI sub-option +were written back as zero, and so were the Binding Update and Binding +Acknowledgement flag bits other than ``A``/``H``/``L``/``K`` and ``K``, which +later RFCs assign (IANA *Binding Update Flags* and *Binding Acknowledgment +Flags*). + +Non-zero ``PadN`` contents are #1223 and are not covered here. + +Every case builds its own octets in memory and reads no capture. :class:`MH` is +imported inside each test, after :func:`~tests._support.reimport_once_per_class`. + +""" + +import unittest + +from tests._support import reimport_once_per_class + +#: Mobility Header octets, parsed as an IPv6 extension header (next header 59). +FRAMES = { + # BRR carrying a PadN of Option Length 0, then a PadN of Option Length 4 + 'padn-zero': '3b01000000000000' '0100' '010400000000', + 'header-reserved': '3b0000aa00000000', + 'brr-reserved': '3b0000000000aabb', + # BU: seq 0, every flag bit set, lifetime 1, PadN of Option Length 2 + 'bu-flags': '3b0105000000' '0000' 'ffff' '0001' '01020000', + # BA: status 0, every flag bit set, seq 0, lifetime 1, PadN of Option Length 2 + 'ba-flags': '3b0106000000' '00ff' '0000' '0001' '01020000', + # BRR carrying a Handoff Indicator with reserved 0xaa, then a PadN + 'hi-reserved': '3b01000000000000' '1702aa01' '01020000', + # BRR carrying an IPv4 Home Address: prefix 32, P set, reserved 0x155 + 'ipv4-hoa-reserved': '3b01000000000000' '1d068355c0000201', + # BRR carrying an ANI option whose Network-Identifier flags octet is 0xff + 'ani-reserved': '3b02000000000000' '3407' '0105ff0161' '0162' '01050000000000', +} + + +class TestMHRoundTrip(unittest.TestCase): + """Pin the wire form of MH padding, reserved fields and flag bits.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + @staticmethod + def _parse(octets: 'bytes') -> 'object': + import io + + from pcapkit.protocols.internet.mh import MH + + return MH(io.BytesIO(octets), len(octets), extension=True) + + def test_from_data_rebuilds_byte_for_byte(self) -> None: + from pcapkit.protocols.internet.mh import MH + + for name, frame in FRAMES.items(): + with self.subTest(frame=name): + octets = bytes.fromhex(frame) + rebuilt = MH.from_data(self._parse(octets).info) # type: ignore[attr-defined] + self.assertEqual(bytes(rebuilt).hex(), octets.hex()) + + def test_zero_length_padn_is_read_as_a_two_octet_padn(self) -> None: + from pcapkit.const.mh.option import Option + + options = self._parse(bytes.fromhex(FRAMES['padn-zero'])).info.options # type: ignore[attr-defined] + pads = options.getlist(Option.PadN) + self.assertEqual([pad.length for pad in pads], [2, 6]) + + def test_make_keeps_a_zero_length_padn(self) -> None: + import warnings + + from pcapkit.const.mh.option import Option + from pcapkit.protocols.internet.mh import MH + + proto = object.__new__(MH) + with warnings.catch_warnings(): + warnings.simplefilter('error') + pad = proto._make_opt_pad(Option.PadN, length=0) + self.assertEqual(pad.type, Option.PadN) + self.assertEqual(bytes(pad), b'\x01\x00') + + def test_reserved_fields_are_read(self) -> None: + from pcapkit.const.mh.ani_suboption import ANISuboption + from pcapkit.const.mh.option import Option + + self.assertEqual(self._parse(bytes.fromhex(FRAMES['header-reserved'])).info.reserved, # type: ignore[attr-defined] + 0xaa) + self.assertEqual(self._parse(bytes.fromhex(FRAMES['brr-reserved'])).info.msg_reserved, # type: ignore[attr-defined] + 0xaabb) + + hi = self._parse(bytes.fromhex(FRAMES['hi-reserved'])).info.options[Option.Handoff_Indicator_Option] # type: ignore[attr-defined] + self.assertEqual(hi.reserved, 0xaa) + + hoa = self._parse(bytes.fromhex(FRAMES['ipv4-hoa-reserved'])).info.options[Option.IPv4_Home_Address] # type: ignore[attr-defined] + self.assertEqual((hoa.prefix_length, hoa.request_prefix, hoa.reserved), (32, True, 0x155)) + + ani = self._parse(bytes.fromhex(FRAMES['ani-reserved'])).info.options[Option.Access_Network_Identifier] # type: ignore[attr-defined] + net = ani.suboptions[ANISuboption.Network_Identifier] + self.assertEqual((net.utf8, net.reserved), (True, 0x7f)) + + def test_binding_update_and_acknowledgement_flags_are_modelled(self) -> None: + bu = self._parse(bytes.fromhex(FRAMES['bu-flags'])).info # type: ignore[attr-defined] + for name in ('ack', 'home', 'lla_compat', 'key_mngt', 'map_reg', 'mobile_router', + 'proxy_reg', 'udp_encap', 'tlv_format', 'bulk_binding', 'multicast', 'dmm'): + with self.subTest(message='BU', flag=name): + self.assertIs(getattr(bu, name), True) + self.assertEqual(bu.flags_reserved, 0xf) + + ba = self._parse(bytes.fromhex(FRAMES['ba-flags'])).info # type: ignore[attr-defined] + for name in ('key_mngt', 'mobile_router', 'proxy_reg', 'tlv_format', 'bulk_binding', + 'multicast', 'dmm'): + with self.subTest(message='BA', flag=name): + self.assertIs(getattr(ba, name), True) + self.assertEqual(ba.flags_reserved, 1) + + def test_fresh_make_places_each_flag_on_its_iana_bit(self) -> None: + from pcapkit.const.mh.binding_ack_flag import BindingACKFlag + from pcapkit.const.mh.binding_update_flag import BindingUpdateFlag + from pcapkit.protocols.internet.mh import MH + + proto = object.__new__(MH) + bu_flags = {'map_reg': BindingUpdateFlag.M, 'mobile_router': BindingUpdateFlag.R, + 'proxy_reg': BindingUpdateFlag.P, 'udp_encap': BindingUpdateFlag.F, + 'tlv_format': BindingUpdateFlag.T, 'bulk_binding': BindingUpdateFlag.B, + 'multicast': BindingUpdateFlag.S, 'dmm': BindingUpdateFlag.D} + for name, bit in bu_flags.items(): + with self.subTest(message='BU', flag=name): + packed = bytes(proto._make_msg_bu(None, options=[], **{name: True})) + self.assertEqual(int.from_bytes(packed[2:4], 'big'), int(bit)) + + ba_flags = {'mobile_router': BindingACKFlag.R, 'proxy_reg': BindingACKFlag.P, + 'tlv_format': BindingACKFlag.T, 'bulk_binding': BindingACKFlag.B, + 'multicast': BindingACKFlag.S, 'dmm': BindingACKFlag.D} + for name, bit in ba_flags.items(): + with self.subTest(message='BA', flag=name): + packed = bytes(proto._make_msg_ba(None, options=[], **{name: True})) + self.assertEqual(packed[1], int(bit)) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/internet/test_mh_unit.py b/tests/protocols/internet/test_mh_unit.py index 443df73584..885748a790 100644 --- a/tests/protocols/internet/test_mh_unit.py +++ b/tests/protocols/internet/test_mh_unit.py @@ -31,6 +31,7 @@ def test_mh_index_length_and_make_data(self) -> None: data = DummyDict( next=TransType.UDP, type=Packet.Binding_Refresh_Request, + reserved=0x5a, chksum=b'\x12\x34', __next_type__=None, ) @@ -41,6 +42,7 @@ def test_mh_index_length_and_make_data(self) -> None: values = MH._make_data(data) self.assertEqual(values['next'], TransType.UDP) self.assertEqual(values['type'], Packet.Binding_Refresh_Request) + self.assertEqual(values['reserved'], 0x5a) self.assertEqual(values['chksum'], b'\x12\x34') self.assertIs(values['data'], data) self.assertIn('payload', values) @@ -210,11 +212,11 @@ def test_mh_message_readers_and_constructors_cover_known_types(self) -> None: proto._make_mh_options = mock.Mock(return_value=['made']) def header(type_: Packet) -> SimpleNamespace: - return SimpleNamespace(next=TransType.UDP, length=2, type=type_, chksum=b'\x12\x34') + return SimpleNamespace(next=TransType.UDP, length=2, type=type_, reserved=0, chksum=b'\x12\x34') unknown = proto._read_msg_unknown(SimpleNamespace(data=b'raw'), header=header(Packet.get(250))) self.assertEqual(unknown.data, b'raw') - self.assertEqual(proto._read_msg_brr(SimpleNamespace(options=[]), + self.assertEqual(proto._read_msg_brr(SimpleNamespace(reserved=0, options=[]), header=header(Packet.Binding_Refresh_Request)).options, 'opts') self.assertEqual(proto._read_msg_hoti(SimpleNamespace(cookie=b'12345678', options=[]), header=header(Packet.Home_Test_Init)).cookie, b'12345678') @@ -228,13 +230,17 @@ def header(type_: Packet) -> SimpleNamespace: token=b'ABCDEFGH', options=[]), header=header(Packet.Care_of_Test)) self.assertEqual(cot.nonce_index, 2) - bu = proto._read_msg_bu(SimpleNamespace(seq=3, flags={'A': 1, 'H': 1, 'L': 0, 'K': 1}, + bu = proto._read_msg_bu(SimpleNamespace(seq=3, flags={'A': 1, 'H': 1, 'L': 0, 'K': 1, 'M': 0, + 'R': 0, 'P': 0, 'F': 0, 'T': 0, 'B': 0, + 'S': 0, 'D': 0, 'reserved': 0}, lifetime=4, options=[]), header=header(Packet.Binding_Update)) self.assertTrue(bu.ack) self.assertEqual(bu.lifetime, datetime.timedelta(seconds=16)) ba = proto._read_msg_ba(SimpleNamespace(status=StatusCode.Reason_unspecified, - flags={'K': 1}, seq=4, lifetime=5, + flags={'K': 1, 'R': 0, 'P': 0, 'T': 0, 'B': 0, + 'S': 0, 'D': 0, 'reserved': 0}, + seq=4, lifetime=5, options=[]), header=header(Packet.Binding_Acknowledgement)) self.assertTrue(ba.key_mngt) @@ -248,7 +254,7 @@ def header(type_: Packet) -> SimpleNamespace: self.assertEqual(proto._make_msg_unknown(None, data=b'xx').data, b'xx') self.assertEqual(proto._make_msg_unknown(SimpleNamespace(data=b'yy')).data, b'yy') self.assertEqual(proto._make_msg_brr(None, options=[]).options, ['made']) - self.assertEqual(proto._make_msg_brr(SimpleNamespace(options=[])).options, ['made']) + self.assertEqual(proto._make_msg_brr(SimpleNamespace(msg_reserved=0, options=[])).options, ['made']) self.assertEqual(proto._make_msg_hoti(None, cookie=b'12345678', options=[]).cookie, b'12345678') self.assertEqual(proto._make_msg_hoti(SimpleNamespace(cookie=b'abcdefgh', options=[])).cookie, b'abcdefgh') self.assertEqual(proto._make_msg_coti(None, cookie=b'ABCDEFGH', options=[]).cookie, b'ABCDEFGH') @@ -266,13 +272,21 @@ def header(type_: Packet) -> SimpleNamespace: options=[]).lifetime, 3) self.assertEqual(proto._make_msg_bu(SimpleNamespace(seq=12, ack=False, home=True, lla_compat=True, key_mngt=False, + map_reg=False, mobile_router=False, + proxy_reg=False, udp_encap=False, + tlv_format=False, bulk_binding=False, + multicast=False, dmm=False, + flags_reserved=0, lifetime=datetime.timedelta(seconds=8), options=[])).seq, 12) self.assertEqual(proto._make_msg_ba(None, status=StatusCode.Reason_unspecified, key_mngt=True, seq=13, lifetime=8, options=[]).seq, 13) self.assertEqual(proto._make_msg_ba(SimpleNamespace(status=StatusCode.Reason_unspecified, - key_mngt=False, seq=14, + key_mngt=False, mobile_router=False, + proxy_reg=False, tlv_format=False, + bulk_binding=False, multicast=False, + dmm=False, flags_reserved=0, seq=14, lifetime=datetime.timedelta(seconds=8), options=[])).seq, 14) self.assertEqual(str(proto._make_msg_be(None, status=BindingError.Unrecognized_MH_Type_value, @@ -317,7 +331,6 @@ def test_mh_option_readers_cover_known_options_and_guards(self) -> None: for bad in [ schema.PadOption(type=Option.Binding_Refresh_Advice, length=0), schema.PadOption(type=Option.Pad1, length=1), - schema.PadOption(type=Option.PadN, length=0), ]: with self.assertRaises(ProtocolError): proto._read_opt_pad(bad, options=options) @@ -478,8 +491,8 @@ def test_mh_option_constructors_cover_known_options_and_dispatch(self) -> None: data=b'yy')).data, b'yy') with mock.patch('pcapkit.protocols.internet.mh.warn') as warn: self.assertEqual(proto._make_opt_pad(Option.Pad1, length=2).type, Option.PadN) - self.assertEqual(proto._make_opt_pad(Option.PadN, length=0).type, Option.Pad1) - self.assertEqual(warn.call_count, 2) + self.assertEqual(proto._make_opt_pad(Option.PadN, length=0).type, Option.PadN) + self.assertEqual(warn.call_count, 1) self.assertEqual(proto._make_opt_bra(Option.Binding_Refresh_Advice, interval=9).interval, 9) self.assertEqual(proto._make_opt_bra( @@ -673,6 +686,7 @@ def make_extension(code: CGAExtension, extension: data.UnknownExtension | None = next=TransType.UDP, length=2, type=custom_packet, + reserved=0, chksum=b'', data=b'model', ) @@ -827,9 +841,17 @@ def make_extension(code: CGAExtension, extension: data.UnknownExtension | None = next=TransType.UDP, length=12, type=Packet.Binding_Acknowledgement, + reserved=0, chksum=b'', status=StatusCode.Reason_unspecified, key_mngt=True, + mobile_router=False, + proxy_reg=False, + tlv_format=False, + bulk_binding=False, + multicast=False, + dmm=False, + flags_reserved=0, seq=1, lifetime=datetime.timedelta(seconds=8), options=OrderedMultiDict(), @@ -900,7 +922,7 @@ def test_mh_fmipv6_message_readers_and_constructors(self) -> None: 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') + return SimpleNamespace(next=TransType.UDP, length=5, type=type_, reserved=0, chksum=b'\x12\x34') # RFC 5568, section 6.2.2 -- the FBU message *layout* is identical to # the RFC 6275 BU, but its Lifetime field is separately defined as @@ -1051,7 +1073,7 @@ def test_mh_fbu_fback_lifetime_is_plain_seconds_unlike_bu_ba(self) -> None: proto._make_mh_options = mock.Mock(return_value=['made']) def header(type_: Packet) -> SimpleNamespace: - return SimpleNamespace(next=None, length=2, type=type_, chksum=b'\x12\x34') + return SimpleNamespace(next=None, length=2, type=type_, reserved=0, chksum=b'\x12\x34') accepted = StatusCode.Binding_Update_accepted_Proxy_Binding_Update_accepted @@ -1069,12 +1091,16 @@ def header(type_: Packet) -> SimpleNamespace: # control: BU/BA still scale by 4 on read. bu = proto._read_msg_bu(SimpleNamespace( - seq=1, flags={'A': 1, 'H': 1, 'L': 0, 'K': 0}, lifetime=100, options=[], + seq=1, flags={'A': 1, 'H': 1, 'L': 0, 'K': 0, 'M': 0, 'R': 0, 'P': 0, 'F': 0, + 'T': 0, 'B': 0, 'S': 0, 'D': 0, 'reserved': 0}, + lifetime=100, options=[], ), header=header(Packet.Binding_Update)) self.assertEqual(bu.lifetime, datetime.timedelta(seconds=400)) ba = proto._read_msg_ba(SimpleNamespace( - status=accepted, flags={'K': 0}, seq=1, lifetime=100, options=[], + status=accepted, flags={'K': 0, 'R': 0, 'P': 0, 'T': 0, 'B': 0, 'S': 0, 'D': 0, + 'reserved': 0}, + seq=1, lifetime=100, options=[], ), header=header(Packet.Binding_Acknowledgement)) self.assertEqual(ba.lifetime, datetime.timedelta(seconds=400))