From 009e5f4e0f3f01d2abed0b04096a3fe3cfbb5540 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Thu, 8 Oct 2026 09:51:32 -0400 Subject: [PATCH] fix(mh): keep the reserved bits of 24 BitField flag namespaces on rebuild Ten messages (BRM, FBack, FBU, FB, HAck, HI, Heartbeat, LRA, SR, UPN), twelve options (DNS-UPDATE, DMNP, DynIP Multicast Selector, IPv4 Address Ack, IPv4 DHCP Support Mode, IPv4 HoA Reply/Request, IPv4 Traffic Offload Selector, MAG Multipath Binding, NAT Detection, Redirect, Transient Binding), the Multi-Prefix CGA extension and the per-session bit-rate QoS attribute left bits of their flags word unnamed, so they were packed back as zero. Each site now names a 'reserved' sub-field, read into the data model (flags_reserved on messages, reserved elsewhere) and written by make(), defaulting to 0. Closes #1418 Refs #1288 --- pcapkit/protocols/data/internet/mh.py | 96 +++++++-- pcapkit/protocols/internet/mh.py | 130 +++++++++++- pcapkit/protocols/schema/internet/mh.py | 45 ++++ .../internet/test_mh_flags_reserved_unit.py | 192 ++++++++++++++++++ tests/protocols/internet/test_mh_unit.py | 32 +-- 5 files changed, 446 insertions(+), 49 deletions(-) create mode 100644 tests/protocols/internet/test_mh_flags_reserved_unit.py diff --git a/pcapkit/protocols/data/internet/mh.py b/pcapkit/protocols/data/internet/mh.py index a36e950231..7a971715f4 100644 --- a/pcapkit/protocols/data/internet/mh.py +++ b/pcapkit/protocols/data/internet/mh.py @@ -357,6 +357,8 @@ class FastBindingUpdateMessage(MH): lla_compat: 'bool' #: Key management mobility capability flag. key_mngt: 'bool' + #: Reserved bits, i.e. the 12 bits after the ``K`` flag. + flags_reserved: 'int' #: Lifetime. :rfc:`5568#section-6.2.2` counts this in seconds rather than in #: the units of 4 seconds the binding messages use. lifetime: 'timedelta' @@ -366,6 +368,7 @@ class FastBindingUpdateMessage(MH): if TYPE_CHECKING: 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', + flags_reserved: 'int', lifetime: 'timedelta', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -381,6 +384,8 @@ class FastBindingAcknowledgmentMessage(MH): status: 'FastBindingAcknowledgmentStatus' #: Key management mobility capability flag. key_mngt: 'bool' + #: Reserved bits, i.e. the seven bits after the ``K`` flag. + flags_reserved: 'int' #: Sequence number. seq: 'int' #: Lifetime. :rfc:`5568#section-6.2.3` counts this in seconds rather than in @@ -391,7 +396,8 @@ class FastBindingAcknowledgmentMessage(MH): if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', - status: 'FastBindingAcknowledgmentStatus', key_mngt: 'bool', seq: 'int', + status: 'FastBindingAcknowledgmentStatus', key_mngt: 'bool', + flags_reserved: 'int', seq: 'int', lifetime: 'timedelta', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -434,6 +440,8 @@ class HandoverInitiateMessage(MH): proxy: 'bool' #: Forwarding flag. forward: 'bool' + #: Reserved bits, i.e. the four bits after the ``F`` flag. + flags_reserved: 'int' #: Code. code: 'Enum_HandoverInitiateStatus' #: Mobility options. @@ -442,6 +450,7 @@ class HandoverInitiateMessage(MH): if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', seq: 'int', assign: 'bool', buffer: 'bool', proxy: 'bool', forward: 'bool', + flags_reserved: 'int', code: 'Enum_HandoverInitiateStatus', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -458,6 +467,8 @@ class HandoverAcknowledgeMessage(MH): proxy: 'bool' #: Forwarding flag. forward: 'bool' + #: Reserved bits, i.e. the five bits after the ``F`` flag. + flags_reserved: 'int' #: Code. code: 'Enum_HandoverACKStatus' #: Mobility options. @@ -465,7 +476,7 @@ class HandoverAcknowledgeMessage(MH): if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', - seq: 'int', buffer: 'bool', proxy: 'bool', forward: 'bool', + seq: 'int', buffer: 'bool', proxy: 'bool', forward: 'bool', flags_reserved: 'int', code: 'Enum_HandoverACKStatus', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -479,6 +490,8 @@ class HeartbeatMessage(MH): unsolicited: 'bool' #: Response flag. Unset for a Heartbeat request, set for a response. response: 'bool' + #: Reserved bits, i.e. the 14 bits before the ``U`` flag. + flags_reserved: 'int' #: Sequence number. seq: 'int' #: Mobility options. @@ -486,7 +499,7 @@ class HeartbeatMessage(MH): if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', - unsolicited: 'bool', response: 'bool', seq: 'int', + unsolicited: 'bool', response: 'bool', flags_reserved: 'int', seq: 'int', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -530,6 +543,8 @@ class BindingRevocationMessage(MH): ipv4_hoa: 'bool' #: Global revocation flag. global_revocation: 'bool' + #: Reserved bits, i.e. the 13 bits after the ``G`` flag. + flags_reserved: 'int' #: Mobility options. options: 'OrderedMultiDict[Enum_Option, Option]' @@ -537,7 +552,7 @@ class BindingRevocationMessage(MH): 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', + proxy: 'bool', ipv4_hoa: 'bool', global_revocation: 'bool', flags_reserved: 'int', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -567,6 +582,8 @@ class LocalizedRoutingAcknowledgmentMessage(MH): seq: 'int' #: Unsolicited flag. unsolicited: 'bool' + #: Reserved bits, i.e. the seven bits after the ``U`` flag. + flags_reserved: 'int' #: Status. :rfc:`6705#section-10.2` defines its values inline with no IANA #: registry behind them, so this field carries the module-local #: :class:`~pcapkit.protocols.internet.mh.LocalizedRoutingStatus` rather than @@ -579,7 +596,7 @@ class LocalizedRoutingAcknowledgmentMessage(MH): if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', - seq: 'int', unsolicited: 'bool', status: 'LocalizedRoutingStatus', + seq: 'int', unsolicited: 'bool', flags_reserved: 'int', status: 'LocalizedRoutingStatus', lifetime: 'timedelta', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -596,13 +613,15 @@ class UpdateNotificationMessage(MH): ack: 'bool' #: Retransmission flag. retransmit: 'bool' + #: Reserved bits, i.e. the 14 bits after the ``D`` flag. + flags_reserved: 'int' #: Mobility options. options: 'OrderedMultiDict[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', seq: 'int', reason: 'Enum_UpdateNotificationReason', ack: 'bool', - retransmit: 'bool', + retransmit: 'bool', flags_reserved: 'int', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -642,6 +661,8 @@ class FlowBindingMessage(MH): #: Acknowledgement requested flag. Only meaningful in an indication; an #: acknowledgement reserves the whole octet this bit is taken from. ack: 'bool' + #: Reserved bits, i.e. the seven bits after the ``A`` flag. + flags_reserved: 'int' #: Mobility options. options: 'OrderedMultiDict[Enum_Option, Option]' @@ -649,7 +670,7 @@ class FlowBindingMessage(MH): 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', + ack: 'bool', flags_reserved: 'int', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -677,12 +698,14 @@ class SubscriptionResponseMessage(MH): #: Multicast information flag, set when the response carries active #: multicast subscription options. info: 'bool' + #: Reserved bits, i.e. the seven bits after the ``I`` flag. + flags_reserved: 'int' #: Mobility options. options: 'OrderedMultiDict[Enum_Option, Option]' if TYPE_CHECKING: def __init__(self, next: 'TransType', length: 'int', type: 'Packet', reserved: 'int', chksum: 'bytes', - seq: 'int', info: 'bool', + seq: 'int', info: 'bool', flags_reserved: 'int', options: 'OrderedMultiDict[Enum_Option, Option]') -> 'None': ... @@ -880,11 +903,14 @@ class MultiPrefixExtension(CGAExtension): #: P flag. Set if a public key is included in the Public Key field of #: the CGA Parameter Data Structure, reset otherwise. flag: 'bool' + #: Reserved bits, i.e. the 31 bits after the ``P`` flag. + reserved: 'int' #: Prefixes. prefixes: 'tuple[int, ...]' if TYPE_CHECKING: - def __init__(self, type: 'Enum_CGAExtension', length: 'int', flag: 'bool', prefixes: 'tuple[int, ...]') -> 'None': ... + def __init__(self, type: 'Enum_CGAExtension', length: 'int', flag: 'bool', reserved: 'int', + prefixes: 'tuple[int, ...]') -> 'None': ... @info_final @@ -1106,13 +1132,15 @@ class DNSUpdateOption(Option): #: Remove flag. Set when the mobile node asks the home agent to remove the #: DNS entry rather than to create or update it. remove: 'bool' + #: Reserved bits, i.e. the seven bits after the ``R`` flag. + reserved: 'int' #: Mobile node identity, in FQDN form. Kept as :obj:`bytes` because the RFC #: does not say which of the two FQDN encodings is meant. identity: 'bytes' if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', length: 'int', status: 'Enum_DNSStatusCode', - remove: 'bool', identity: 'bytes') -> 'None': ... + remove: 'bool', reserved: 'int', identity: 'bytes') -> 'None': ... @info_final @@ -1174,12 +1202,14 @@ class IPv4AddressAcknowledgementOption(Option): #: Allocated prefix length. Meaningful only on success, and required to be #: zero on failure. prefix_length: 'int' + #: Reserved bits, i.e. the two bits after the prefix length. + reserved: 'int' #: Assigned IPv4 home address. address: 'IPv4Address' if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', length: 'int', - status: 'Enum_DSMIPv6HomeAddress', prefix_length: 'int', + status: 'Enum_DSMIPv6HomeAddress', prefix_length: 'int', reserved: 'int', address: 'IPv4Address') -> 'None': ... @@ -1189,12 +1219,14 @@ class NATDetectionOption(Option): #: Force UDP encapsulation flag. force: 'bool' + #: Reserved bits, i.e. the 15 bits after the ``F`` flag. + reserved: 'int' #: Suggested NAT binding refresh interval, in seconds. ``0`` means the value #: is to be ignored, and all ones that no keep-alives are needed. refresh: 'timedelta' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Option', length: 'int', force: 'bool', + def __init__(self, type: 'Enum_Option', length: 'int', force: 'bool', reserved: 'int', refresh: 'timedelta') -> 'None': ... @@ -1250,12 +1282,14 @@ class IPv4HomeAddressRequestOption(Option): #: Prefix length of the requested home network. prefix_length: 'int' + #: Reserved bits, i.e. the ten bits after the prefix length. + reserved: 'int' #: Requested IPv4 home address. ``0.0.0.0`` asks the local mobility anchor to #: allocate one. address: 'IPv4Address' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Option', length: 'int', prefix_length: 'int', + def __init__(self, type: 'Enum_Option', length: 'int', prefix_length: 'int', reserved: 'int', address: 'IPv4Address') -> 'None': ... @@ -1268,12 +1302,14 @@ class IPv4HomeAddressReplyOption(Option): status: 'Enum_HomeAddressReply' #: Prefix length of the assigned home network. prefix_length: 'int' + #: Reserved bits, i.e. the two bits after the prefix length. + reserved: 'int' #: Assigned IPv4 home address. address: 'IPv4Address' if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', length: 'int', status: 'Enum_HomeAddressReply', - prefix_length: 'int', address: 'IPv4Address') -> 'None': ... + prefix_length: 'int', reserved: 'int', address: 'IPv4Address') -> 'None': ... @info_final @@ -1294,10 +1330,12 @@ class IPv4DHCPSupportModeOption(Option): #: DHCP support mode. Unset means the mobile access gateway acts as a DHCP #: relay, set that it acts as a DHCP server. mode: 'Enum_DHCPSupportMode' + #: Reserved bits, i.e. the 15 bits before the ``S`` flag. + reserved: 'int' if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', length: 'int', - mode: 'Enum_DHCPSupportMode') -> 'None': ... + mode: 'Enum_DHCPSupportMode', reserved: 'int') -> 'None': ... @info_final @@ -1360,12 +1398,14 @@ class TransientBindingOption(Option): #: Late path switch flag. late: 'bool' + #: Reserved bits, i.e. the seven bits before the ``L`` flag. + reserved: 'int' #: Maximum lifetime of the transient state. Counted on the wire in units of #: 100 milliseconds. lifetime: 'timedelta' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Option', length: 'int', late: 'bool', + def __init__(self, type: 'Enum_Option', length: 'int', late: 'bool', reserved: 'int', lifetime: 'timedelta') -> 'None': ... @@ -1496,10 +1536,12 @@ class RedirectOption(Option): ipv6: 'Optional[IPv6Address]' #: IPv4 address of the redirected-to local mobility anchor, if carried. ipv4: 'Optional[IPv4Address]' + #: Reserved bits, i.e. the 14 bits after the ``N`` flag. + reserved: 'int' if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', length: 'int', ipv6: 'Optional[IPv6Address]', - ipv4: 'Optional[IPv4Address]') -> 'None': ... + ipv4: 'Optional[IPv4Address]', reserved: 'int') -> 'None': ... @info_final @@ -1693,11 +1735,13 @@ class IPv4TrafficOffloadSelectorOption(Option): #: Offload mode flag. Unset offloads the flows matching the selector; set #: offloads everything except them. mode: 'bool' + #: Reserved bits, i.e. the 31 bits after the ``M`` flag. + reserved: 'int' #: Traffic selector sub-options. selector: 'OrderedMultiDict[Enum_FlowIDSuboption, FlowIdentificationSuboption]' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Option', length: 'int', mode: 'bool', + def __init__(self, type: 'Enum_Option', length: 'int', mode: 'bool', reserved: 'int', selector: 'OrderedMultiDict[Enum_FlowIDSuboption, FlowIdentificationSuboption]') -> 'None': ... @@ -1710,6 +1754,8 @@ class DynamicIPMulticastSelectorOption(Option): #: Selector mode flag. Set routes the group locally, unset via the home #: network. mode: 'bool' + #: Reserved bits, i.e. the seven bits after the ``M`` flag. + reserved: 'int' #: Number of multicast address records carried in :attr:`data`. records: 'int' #: Multicast address records. These are MLD or IGMP structures belonging to @@ -1718,7 +1764,7 @@ class DynamicIPMulticastSelectorOption(Option): data: 'bytes' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Option', length: 'int', protocol: 'int', mode: 'bool', + def __init__(self, type: 'Enum_Option', length: 'int', protocol: 'int', mode: 'bool', reserved: 'int', records: 'int', data: 'bytes') -> 'None': ... @@ -1728,13 +1774,15 @@ class DelegatedMNPOption(Option): #: IPv4 prefix flag, saying which family :attr:`prefix` is in. ipv4: 'bool' + #: Reserved bits, i.e. the seven bits after the ``V`` flag. + reserved: 'int' #: Prefix length. prefix_length: 'int' #: Delegated mobile network prefix, IPv4 or IPv6 as :attr:`ipv4` says. prefix: 'IPv4Address | IPv6Address' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Option', length: 'int', ipv4: 'bool', + def __init__(self, type: 'Enum_Option', length: 'int', ipv4: 'bool', reserved: 'int', prefix_length: 'int', prefix: 'IPv4Address | IPv6Address') -> 'None': ... @@ -1806,12 +1854,14 @@ class PerSessionBitRateAttribute(QoSAttribute): service: 'bool' #: Guaranteed-bit-rate exclusion flag. exclude: 'bool' + #: Reserved bits, i.e. the 14 bits after the ``E`` flag. + reserved: 'int' #: Bit rate, in **bits** per second. rate: 'int' if TYPE_CHECKING: def __init__(self, type: 'Enum_QoSAttribute', length: 'int', service: 'bool', - exclude: 'bool', rate: 'int') -> 'None': ... + exclude: 'bool', reserved: 'int', rate: 'int') -> 'None': ... @info_final @@ -2011,11 +2061,13 @@ class MAGMultipathBindingOption(Option): bulk: 'bool' #: Registration overwrite flag. overwrite: 'bool' + #: Reserved bits, i.e. the 22 bits after the ``O`` flag. + reserved: 'int' if TYPE_CHECKING: def __init__(self, type: 'Enum_Option', length: 'int', att: 'Enum_AccessType', label: 'int', bid: 'int', bulk: 'bool', - overwrite: 'bool') -> 'None': ... + overwrite: 'bool', reserved: 'int') -> 'None': ... @info_final diff --git a/pcapkit/protocols/internet/mh.py b/pcapkit/protocols/internet/mh.py index 90379becbd..5bf25b0b9a 100644 --- a/pcapkit/protocols/internet/mh.py +++ b/pcapkit/protocols/internet/mh.py @@ -2094,6 +2094,7 @@ def _read_msg_fbu(self, schema: 'Schema_FastBindingUpdateMessage', *, home=bool(schema.flags['H']), lla_compat=bool(schema.flags['L']), key_mngt=bool(schema.flags['K']), + flags_reserved=schema.flags['reserved'], lifetime=datetime.timedelta(seconds=schema.lifetime), options=self._read_mh_options(schema.options), ) @@ -2149,6 +2150,7 @@ def _read_msg_fback(self, schema: 'Schema_FastBindingAcknowledgmentMessage', *, chksum=header.chksum, status=FastBindingAcknowledgmentStatus(schema.status), key_mngt=bool(schema.flags['K']), + flags_reserved=schema.flags['reserved'], seq=schema.seq, lifetime=datetime.timedelta(seconds=schema.lifetime), options=self._read_mh_options(schema.options), @@ -2286,6 +2288,7 @@ def _read_msg_hi(self, schema: 'Schema_HandoverInitiateMessage', *, buffer=bool(schema.flags['U']), proxy=bool(schema.flags['P']), forward=bool(schema.flags['F']), + flags_reserved=schema.flags['reserved'], code=schema.code, options=self._read_mh_options(schema.options), ) @@ -2336,6 +2339,7 @@ def _read_msg_hack(self, schema: 'Schema_HandoverAcknowledgeMessage', *, buffer=bool(schema.flags['U']), proxy=bool(schema.flags['P']), forward=bool(schema.flags['F']), + flags_reserved=schema.flags['reserved'], code=schema.code, options=self._read_mh_options(schema.options), ) @@ -2440,6 +2444,7 @@ def _read_msg_hb(self, schema: 'Schema_HeartbeatMessage', *, chksum=header.chksum, unsolicited=bool(schema.flags['U']), response=bool(schema.flags['R']), + flags_reserved=schema.flags['reserved'], seq=schema.seq, options=self._read_mh_options(schema.options), ) @@ -2512,6 +2517,7 @@ def _read_msg_brm(self, schema: 'Schema_BindingRevocationMessage', *, proxy=bool(schema.flags['P']), ipv4_hoa=bool(schema.flags['V']), global_revocation=bool(schema.flags['G']), + flags_reserved=schema.flags['reserved'], options=self._read_mh_options(schema.options), ) return data @@ -2605,6 +2611,7 @@ def _read_msg_lra(self, schema: 'Schema_LocalizedRoutingAcknowledgmentMessage', chksum=header.chksum, seq=schema.seq, unsolicited=bool(schema.flags['U']), + flags_reserved=schema.flags['reserved'], status=LocalizedRoutingStatus(schema.status), lifetime=datetime.timedelta(seconds=schema.lifetime), options=self._read_mh_options(schema.options), @@ -2653,6 +2660,7 @@ def _read_msg_upn(self, schema: 'Schema_UpdateNotificationMessage', *, reason=schema.reason, ack=bool(schema.flags['A']), retransmit=bool(schema.flags['D']), + flags_reserved=schema.flags['reserved'], options=self._read_mh_options(schema.options), ) return data @@ -2759,6 +2767,7 @@ def _read_msg_fbm(self, schema: 'Schema_FlowBindingMessage', *, seq=schema.seq, code=schema.code, ack=bool(schema.flags['A']), + flags_reserved=schema.flags['reserved'], options=self._read_mh_options(schema.options), ) return data @@ -2840,6 +2849,7 @@ def _read_msg_sr(self, schema: 'Schema_SubscriptionResponseMessage', *, chksum=header.chksum, seq=schema.seq, info=bool(schema.flags['I']), + flags_reserved=schema.flags['reserved'], options=self._read_mh_options(schema.options), ) return data @@ -3740,6 +3750,7 @@ def _read_opt_dns(self, schema: 'Schema_DNSUpdateOption', *, length=self._mh_option_length(schema.length), status=schema.status, remove=bool(schema.flags['R']), + reserved=schema.flags['reserved'], identity=schema.identity, ) return data @@ -4210,6 +4221,7 @@ def _read_opt_ipv4_ack(self, schema: 'Schema_IPv4AddressAcknowledgementOption', length=self._mh_option_length(schema.length), status=schema.status, prefix_length=schema.flags['prefix_length'], + reserved=schema.flags['reserved'], address=schema.address, ) return data @@ -4251,6 +4263,7 @@ def _read_opt_nat(self, schema: 'Schema_NATDetectionOption', *, type=schema.type, length=self._mh_option_length(schema.length), force=bool(schema.flags['F']), + reserved=schema.flags['reserved'], refresh=datetime.timedelta(seconds=schema.refresh), ) return data @@ -4425,6 +4438,7 @@ def _read_opt_ipv4_hoa_req(self, schema: 'Schema_IPv4HomeAddressRequestOption', type=schema.type, length=self._mh_option_length(schema.length), prefix_length=schema.flags['prefix_length'], + reserved=schema.flags['reserved'], address=schema.address, ) return data @@ -4467,6 +4481,7 @@ def _read_opt_ipv4_hoa_rep(self, schema: 'Schema_IPv4HomeAddressReplyOption', *, length=self._mh_option_length(schema.length), status=schema.status, prefix_length=schema.flags['prefix_length'], + reserved=schema.flags['reserved'], address=schema.address, ) return data @@ -4540,6 +4555,7 @@ def _read_opt_ipv4_dhcp(self, schema: 'Schema_IPv4DHCPSupportModeOption', *, type=schema.type, length=self._mh_option_length(schema.length), mode=Enum_DHCPSupportMode(schema.flags['S']), + reserved=schema.flags['reserved'], ) return data @@ -4720,6 +4736,7 @@ def _read_opt_transient(self, schema: 'Schema_TransientBindingOption', *, type=schema.type, length=self._mh_option_length(schema.length), late=bool(schema.flags['L']), + reserved=schema.flags['reserved'], lifetime=datetime.timedelta(milliseconds=schema.lifetime * 100), ) return data @@ -5004,6 +5021,7 @@ def _read_opt_redirect(self, schema: 'Schema_RedirectOption', *, length=self._mh_option_length(schema.length), ipv6=schema.ipv6, ipv4=schema.ipv4, + reserved=schema.flags['reserved'], ) return data @@ -5382,6 +5400,7 @@ def _read_opt_offload(self, schema: 'Schema_IPv4TrafficOffloadSelectorOption', * type=schema.type, length=self._mh_option_length(schema.length), mode=bool(schema.flags['M']), + reserved=schema.flags['reserved'], selector=self._read_fid_suboptions(schema.selector), ) return data @@ -5437,6 +5456,7 @@ def _read_opt_mcast_sel(self, schema: 'Schema_DynamicIPMulticastSelectorOption', length=self._mh_option_length(schema.length), protocol=schema.protocol, mode=bool(schema.flags['M']), + reserved=schema.flags['reserved'], records=schema.records, data=schema.data, ) @@ -5489,6 +5509,7 @@ def _read_opt_dmnp(self, schema: 'Schema_DelegatedMNPOption', *, type=schema.type, length=self._mh_option_length(schema.length), ipv4=ipv4, + reserved=schema.flags['reserved'], prefix_length=schema.prefix_length, prefix=schema.prefix, ) @@ -5627,6 +5648,7 @@ def _read_qos_attributes( length=self._mh_option_length(session.length), service=bool(session.flags['S']), exclude=bool(session.flags['E']), + reserved=session.flags['reserved'], rate=session.rate, ) # type: Data_QoSAttribute elif code in (Enum_QoSAttribute.Per_MN_Agg_Max_DL_Bit_Rate, @@ -6014,6 +6036,7 @@ def _read_opt_mag_mp(self, schema: 'Schema_MAGMultipathBindingOption', *, bid=schema.bid, bulk=bulk, overwrite=overwrite, + reserved=schema.flags['reserved'], ) return data @@ -6468,6 +6491,7 @@ def _read_ext_multiprefix(self, schema: 'Schema_MultiPrefixExtension', *, type=schema.type, length=self._mh_extension_length(schema.length), flag=bool(schema.flags['P']), + reserved=schema.flags['reserved'], prefixes=tuple(schema.prefixes), ) return data @@ -6901,6 +6925,7 @@ def _make_msg_fbu(self, message: 'Optional[Data_FastBindingUpdateMessage]' = Non 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, + flags_reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_FastBindingUpdateMessage': """Make MH fast binding update (FBU) message type. @@ -6916,6 +6941,7 @@ def _make_msg_fbu(self, message: 'Optional[Data_FastBindingUpdateMessage]' = Non one counts seconds rather than units of 4 seconds [:rfc:`5568#section-6.2.2`], so the value is not scaled. options: Mobility options. + flags_reserved: Reserved bits, i.e. the 12 bits after the ``K`` flag. **kwargs: Arbitrary keyword arguments. Returns: @@ -6928,6 +6954,7 @@ def _make_msg_fbu(self, message: 'Optional[Data_FastBindingUpdateMessage]' = Non home = message.home lla_compat = message.lla_compat key_mngt = message.key_mngt + flags_reserved = message.flags_reserved lifetime_val = math.ceil(message.lifetime.total_seconds()) options = message.options else: @@ -6941,6 +6968,7 @@ def _make_msg_fbu(self, message: 'Optional[Data_FastBindingUpdateMessage]' = Non 'H': home, 'L': lla_compat, 'K': key_mngt, + 'reserved': flags_reserved, }, lifetime=lifetime_val, options=self._make_mh_options(options), @@ -6955,6 +6983,7 @@ def _make_msg_fback(self, message: 'Optional[Data_FastBindingAcknowledgmentMessa seq: 'int' = 0, lifetime: 'int | timedelta' = 4, # reasonable default value options: 'Optional[Option | list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes]]' = None, + flags_reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_FastBindingAcknowledgmentMessage': """Make MH fast binding acknowledgment (FBack) message type. @@ -6972,6 +7001,7 @@ def _make_msg_fback(self, message: 'Optional[Data_FastBindingAcknowledgmentMessa one counts seconds rather than units of 4 seconds [:rfc:`5568#section-6.2.3`], so the value is not scaled. options: Mobility options. + flags_reserved: Reserved bits, i.e. the seven bits after the ``K`` flag. **kwargs: Arbitrary keyword arguments. Returns: @@ -6981,6 +7011,7 @@ def _make_msg_fback(self, message: 'Optional[Data_FastBindingAcknowledgmentMessa if message is not None: status_val = message.status key_mngt = message.key_mngt + flags_reserved = message.flags_reserved seq = message.seq lifetime_val = math.ceil(message.lifetime.total_seconds()) options = message.options @@ -6994,6 +7025,7 @@ def _make_msg_fback(self, message: 'Optional[Data_FastBindingAcknowledgmentMessa status=status_val, flags={ 'K': key_mngt, + 'reserved': flags_reserved, }, seq=seq, lifetime=lifetime_val, @@ -7055,6 +7087,7 @@ def _make_msg_hi(self, message: 'Optional[Data_HandoverInitiateMessage]' = 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, + flags_reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_HandoverInitiateMessage': """Make MH handover initiate (HI) message type. @@ -7070,6 +7103,7 @@ def _make_msg_hi(self, message: 'Optional[Data_HandoverInitiateMessage]' = None, code_namespace: Code namespace. code_reversed: Reverse code namespace. options: Mobility options. + flags_reserved: Reserved bits, i.e. the four bits after the ``F`` flag. **kwargs: Arbitrary keyword arguments. Returns: @@ -7082,6 +7116,7 @@ def _make_msg_hi(self, message: 'Optional[Data_HandoverInitiateMessage]' = None, buffer = message.buffer proxy = message.proxy forward = message.forward + flags_reserved = message.flags_reserved code_val = message.code options = message.options else: @@ -7096,6 +7131,7 @@ def _make_msg_hi(self, message: 'Optional[Data_HandoverInitiateMessage]' = None, 'U': buffer, 'P': proxy, 'F': forward, + 'reserved': flags_reserved, }, code=code_val, options=self._make_mh_options(options), @@ -7111,6 +7147,7 @@ def _make_msg_hack(self, message: 'Optional[Data_HandoverAcknowledgeMessage]' = 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, + flags_reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_HandoverAcknowledgeMessage': """Make MH handover acknowledge (HAck) message type. @@ -7125,6 +7162,7 @@ def _make_msg_hack(self, message: 'Optional[Data_HandoverAcknowledgeMessage]' = code_namespace: Code namespace. code_reversed: Reverse code namespace. options: Mobility options. + flags_reserved: Reserved bits, i.e. the five bits after the ``F`` flag. **kwargs: Arbitrary keyword arguments. Returns: @@ -7136,6 +7174,7 @@ def _make_msg_hack(self, message: 'Optional[Data_HandoverAcknowledgeMessage]' = buffer = message.buffer proxy = message.proxy forward = message.forward + flags_reserved = message.flags_reserved code_val = message.code options = message.options else: @@ -7149,6 +7188,7 @@ def _make_msg_hack(self, message: 'Optional[Data_HandoverAcknowledgeMessage]' = 'U': buffer, 'P': proxy, 'F': forward, + 'reserved': flags_reserved, }, code=code_val, options=self._make_mh_options(options), @@ -7189,6 +7229,7 @@ def _make_msg_hb(self, message: 'Optional[Data_HeartbeatMessage]' = None, *, response: 'bool' = False, seq: 'int' = 0, options: 'Optional[Option | list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes]]' = None, + flags_reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_HeartbeatMessage': """Make MH heartbeat message type. @@ -7198,6 +7239,7 @@ def _make_msg_hb(self, message: 'Optional[Data_HeartbeatMessage]' = None, *, response: Response flag; unset makes this a heartbeat request. seq: Sequence number. options: Mobility options. + flags_reserved: Reserved bits, i.e. the 14 bits before the ``U`` flag. **kwargs: Arbitrary keyword arguments. Returns: @@ -7207,6 +7249,7 @@ def _make_msg_hb(self, message: 'Optional[Data_HeartbeatMessage]' = None, *, if message is not None: unsolicited = message.unsolicited response = message.response + flags_reserved = message.flags_reserved seq = message.seq options = message.options else: @@ -7216,6 +7259,7 @@ def _make_msg_hb(self, message: 'Optional[Data_HeartbeatMessage]' = None, *, flags={ 'U': unsolicited, 'R': response, + 'reserved': flags_reserved, }, seq=seq, options=self._make_mh_options(options), @@ -7235,6 +7279,7 @@ def _make_msg_brm(self, message: 'Optional[Data_BindingRevocationMessage]' = Non ipv4_hoa: 'bool' = False, global_revocation: 'bool' = False, options: 'Optional[Option | list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes]]' = None, + flags_reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_BindingRevocationMessage': """Make MH binding revocation (BRM) message type. @@ -7255,6 +7300,7 @@ def _make_msg_brm(self, message: 'Optional[Data_BindingRevocationMessage]' = Non ipv4_hoa: IPv4 home address binding only flag. global_revocation: Global revocation flag. options: Mobility options. + flags_reserved: Reserved bits, i.e. the 13 bits after the ``G`` flag. **kwargs: Arbitrary keyword arguments. Returns: @@ -7268,6 +7314,7 @@ def _make_msg_brm(self, message: 'Optional[Data_BindingRevocationMessage]' = Non proxy = message.proxy ipv4_hoa = message.ipv4_hoa global_revocation = message.global_revocation + flags_reserved = message.flags_reserved options = message.options else: br_type_val = self._make_index(br_type, br_type_default, namespace=br_type_namespace, # type: ignore[assignment] @@ -7284,6 +7331,7 @@ def _make_msg_brm(self, message: 'Optional[Data_BindingRevocationMessage]' = Non 'P': proxy, 'V': ipv4_hoa, 'G': global_revocation, + 'reserved': flags_reserved, }, options=self._make_mh_options(options), ) @@ -7332,6 +7380,7 @@ def _make_msg_lra(self, message: 'Optional[Data_LocalizedRoutingAcknowledgmentMe status_reversed: 'bool' = False, lifetime: 'int | timedelta' = 0, options: 'Optional[Option | list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes]]' = None, + flags_reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_LocalizedRoutingAcknowledgmentMessage': """Make MH localized routing acknowledgment (LRA) message type. @@ -7347,6 +7396,7 @@ def _make_msg_lra(self, message: 'Optional[Data_LocalizedRoutingAcknowledgmentMe lifetime: Lifetime, in seconds or as a :class:`~datetime.timedelta`. options: Mobility options. + flags_reserved: Reserved bits, i.e. the seven bits after the ``U`` flag. **kwargs: Arbitrary keyword arguments. Returns: @@ -7356,6 +7406,7 @@ def _make_msg_lra(self, message: 'Optional[Data_LocalizedRoutingAcknowledgmentMe if message is not None: seq = message.seq unsolicited = message.unsolicited + flags_reserved = message.flags_reserved status_val = message.status # type: int lifetime_val = math.ceil(message.lifetime.total_seconds()) options = message.options @@ -7369,6 +7420,7 @@ def _make_msg_lra(self, message: 'Optional[Data_LocalizedRoutingAcknowledgmentMe seq=seq, flags={ 'U': unsolicited, + 'reserved': flags_reserved, }, status=status_val, lifetime=lifetime_val, @@ -7384,6 +7436,7 @@ def _make_msg_upn(self, message: 'Optional[Data_UpdateNotificationMessage]' = No ack: 'bool' = False, retransmit: 'bool' = False, options: 'Optional[Option | list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes]]' = None, + flags_reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_UpdateNotificationMessage': """Make MH update notification (UPN) message type. @@ -7397,6 +7450,7 @@ def _make_msg_upn(self, message: 'Optional[Data_UpdateNotificationMessage]' = No ack: Acknowledgement requested flag. retransmit: Retransmission flag. options: Mobility options. + flags_reserved: Reserved bits, i.e. the 14 bits after the ``D`` flag. **kwargs: Arbitrary keyword arguments. Returns: @@ -7408,6 +7462,7 @@ def _make_msg_upn(self, message: 'Optional[Data_UpdateNotificationMessage]' = No reason_val = message.reason # type: Enum_UpdateNotificationReason ack = message.ack retransmit = message.retransmit + flags_reserved = message.flags_reserved options = message.options else: reason_val = self._make_index(reason, reason_default, namespace=reason_namespace, # type: ignore[assignment] @@ -7420,6 +7475,7 @@ def _make_msg_upn(self, message: 'Optional[Data_UpdateNotificationMessage]' = No flags={ 'A': ack, 'D': retransmit, + 'reserved': flags_reserved, }, options=self._make_mh_options(options), ) @@ -7475,6 +7531,7 @@ def _make_msg_fbm(self, message: 'Optional[Data_FlowBindingMessage]' = None, *, code_reversed: 'bool' = False, ack: 'bool' = False, options: 'Optional[Option | list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes]]' = None, + flags_reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_FlowBindingMessage': """Make MH flow binding (FB) message type. @@ -7493,6 +7550,7 @@ def _make_msg_fbm(self, message: 'Optional[Data_FlowBindingMessage]' = None, *, ack: Acknowledgement requested flag. Meaningful only in an indication. options: Mobility options. + flags_reserved: Reserved bits, i.e. the seven bits after the ``A`` flag. **kwargs: Arbitrary keyword arguments. Returns: @@ -7504,6 +7562,7 @@ def _make_msg_fbm(self, message: 'Optional[Data_FlowBindingMessage]' = None, *, seq = message.seq code_val = message.code # type: Enum_FlowBindingIndicationTrigger | Enum_FlowBindingACKStatus ack = message.ack + flags_reserved = message.flags_reserved options = message.options else: fb_type_val = self._make_index(fb_type, fb_type_default, namespace=fb_type_namespace, # type: ignore[assignment] @@ -7518,6 +7577,7 @@ def _make_msg_fbm(self, message: 'Optional[Data_FlowBindingMessage]' = None, *, code=code_val, flags={ 'A': ack, + 'reserved': flags_reserved, }, options=self._make_mh_options(options), ) @@ -7553,6 +7613,7 @@ def _make_msg_sr(self, message: 'Optional[Data_SubscriptionResponseMessage]' = N seq: 'int' = 0, info: 'bool' = False, options: 'Optional[Option | list[Schema_Option | tuple[Enum_Option, dict[str, Any]] | bytes]]' = None, + flags_reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_SubscriptionResponseMessage': """Make MH subscription response (SR) message type. @@ -7561,6 +7622,7 @@ def _make_msg_sr(self, message: 'Optional[Data_SubscriptionResponseMessage]' = N seq: Sequence number, echoed from the query. info: Multicast information flag. options: Mobility options. + flags_reserved: Reserved bits, i.e. the seven bits after the ``I`` flag. **kwargs: Arbitrary keyword arguments. Returns: @@ -7570,6 +7632,7 @@ def _make_msg_sr(self, message: 'Optional[Data_SubscriptionResponseMessage]' = N if message is not None: seq = message.seq info = message.info + flags_reserved = message.flags_reserved options = message.options else: options = options or [] @@ -7578,6 +7641,7 @@ def _make_msg_sr(self, message: 'Optional[Data_SubscriptionResponseMessage]' = N seq=seq, flags={ 'I': info, + 'reserved': flags_reserved, }, options=self._make_mh_options(options), ) @@ -8547,6 +8611,7 @@ def _make_opt_dns(self, type: 'Enum_Option', option: 'Optional[Data_DNSUpdateOpt status_reversed: 'bool' = False, remove: 'bool' = False, identity: 'bytes' = b'', + reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_DNSUpdateOption': """Make MH DNS-UPDATE-TYPE option. @@ -8559,6 +8624,7 @@ def _make_opt_dns(self, type: 'Enum_Option', option: 'Optional[Data_DNSUpdateOpt status_reversed: Reverse status namespace. remove: Remove flag. identity: Mobile node identity, in FQDN form. + reserved: Reserved bits, i.e. the seven bits after the ``R`` flag. **kwargs: Arbitrary keyword arguments. Returns: @@ -8568,6 +8634,7 @@ def _make_opt_dns(self, type: 'Enum_Option', option: 'Optional[Data_DNSUpdateOpt if option is not None: status_val = option.status # type: Enum_DNSStatusCode remove = option.remove + reserved = option.reserved identity = option.identity else: status_val = self._make_index(status, status_default, namespace=status_namespace, # type: ignore[assignment] @@ -8577,7 +8644,7 @@ def _make_opt_dns(self, type: 'Enum_Option', option: 'Optional[Data_DNSUpdateOpt type=type, length=2 + len(identity), status=status_val, - flags={'R': int(remove)}, + flags={'R': int(remove), 'reserved': reserved}, identity=identity, ) @@ -8906,6 +8973,7 @@ def _make_opt_ipv4_ack(self, type: 'Enum_Option', status_reversed: 'bool' = False, prefix_length: 'int' = 32, address: 'bytes | str | int | IPv4Address' = '0.0.0.0', + reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_IPv4AddressAcknowledgementOption': """Make MH IPv4 address acknowledgement option. @@ -8918,6 +8986,7 @@ def _make_opt_ipv4_ack(self, type: 'Enum_Option', status_reversed: Reverse status namespace. prefix_length: Allocated prefix length. address: Assigned IPv4 home address. + reserved: Reserved bits, i.e. the two bits after the prefix length. **kwargs: Arbitrary keyword arguments. Returns: @@ -8927,6 +8996,7 @@ def _make_opt_ipv4_ack(self, type: 'Enum_Option', if option is not None: status_val = option.status # type: Enum_DSMIPv6HomeAddress prefix_length = option.prefix_length + reserved = option.reserved address = option.address else: status_val = self._make_index(status, status_default, namespace=status_namespace, # type: ignore[assignment] @@ -8939,13 +9009,14 @@ def _make_opt_ipv4_ack(self, type: 'Enum_Option', type=type, length=6, status=status_val, - flags={'prefix_length': prefix_length}, + flags={'prefix_length': prefix_length, 'reserved': reserved}, address=address, ) def _make_opt_nat(self, type: 'Enum_Option', option: 'Optional[Data_NATDetectionOption]' = None, *, force: 'bool' = False, refresh: 'int | timedelta' = 0, + reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_NATDetectionOption': """Make MH NAT detection option. @@ -8955,6 +9026,7 @@ def _make_opt_nat(self, type: 'Enum_Option', option: 'Optional[Data_NATDetection force: Force UDP encapsulation flag. refresh: Suggested NAT binding refresh interval, in seconds or as a :class:`~datetime.timedelta`. + reserved: Reserved bits, i.e. the 15 bits after the ``F`` flag. **kwargs: Arbitrary keyword arguments. Returns: @@ -8963,6 +9035,7 @@ def _make_opt_nat(self, type: 'Enum_Option', option: 'Optional[Data_NATDetection """ if option is not None: force = option.force + reserved = option.reserved refresh_val = math.ceil(option.refresh.total_seconds()) else: refresh_val = refresh if isinstance(refresh, int) else math.ceil(refresh.total_seconds()) @@ -8970,7 +9043,7 @@ def _make_opt_nat(self, type: 'Enum_Option', option: 'Optional[Data_NATDetection return Schema_NATDetectionOption( type=type, length=6, - flags={'F': int(force)}, + flags={'F': int(force), 'reserved': reserved}, refresh=refresh_val, ) @@ -9102,6 +9175,7 @@ def _make_opt_ipv4_hoa_req(self, type: 'Enum_Option', option: 'Optional[Data_IPv4HomeAddressRequestOption]' = None, *, prefix_length: 'int' = 32, address: 'bytes | str | int | IPv4Address' = '0.0.0.0', + reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_IPv4HomeAddressRequestOption': """Make MH IPv4 home address request option. @@ -9110,6 +9184,7 @@ def _make_opt_ipv4_hoa_req(self, type: 'Enum_Option', option: Option data model. prefix_length: Prefix length of the requested home network. address: Requested IPv4 home address. + reserved: Reserved bits, i.e. the ten bits after the prefix length. **kwargs: Arbitrary keyword arguments. Returns: @@ -9118,6 +9193,7 @@ def _make_opt_ipv4_hoa_req(self, type: 'Enum_Option', """ if option is not None: prefix_length = option.prefix_length + reserved = option.reserved address = option.address if prefix_length > 32: @@ -9126,7 +9202,7 @@ def _make_opt_ipv4_hoa_req(self, type: 'Enum_Option', return Schema_IPv4HomeAddressRequestOption( type=type, length=6, - flags={'prefix_length': prefix_length}, + flags={'prefix_length': prefix_length, 'reserved': reserved}, address=address, ) @@ -9137,6 +9213,7 @@ def _make_opt_ipv4_hoa_rep(self, type: 'Enum_Option', option: 'Optional[Data_IPv status_reversed: 'bool' = False, prefix_length: 'int' = 32, address: 'bytes | str | int | IPv4Address' = '0.0.0.0', + reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_IPv4HomeAddressReplyOption': """Make MH IPv4 home address reply option. @@ -9149,6 +9226,7 @@ def _make_opt_ipv4_hoa_rep(self, type: 'Enum_Option', option: 'Optional[Data_IPv status_reversed: Reverse status namespace. prefix_length: Prefix length of the assigned home network. address: Assigned IPv4 home address. + reserved: Reserved bits, i.e. the two bits after the prefix length. **kwargs: Arbitrary keyword arguments. Returns: @@ -9158,6 +9236,7 @@ def _make_opt_ipv4_hoa_rep(self, type: 'Enum_Option', option: 'Optional[Data_IPv if option is not None: status_val = option.status # type: Enum_HomeAddressReply prefix_length = option.prefix_length + reserved = option.reserved address = option.address else: status_val = self._make_index(status, status_default, namespace=status_namespace, # type: ignore[assignment] @@ -9170,7 +9249,7 @@ def _make_opt_ipv4_hoa_rep(self, type: 'Enum_Option', option: 'Optional[Data_IPv type=type, length=6, status=status_val, - flags={'prefix_length': prefix_length}, + flags={'prefix_length': prefix_length, 'reserved': reserved}, address=address, ) @@ -9204,6 +9283,7 @@ def _make_opt_ipv4_dhcp(self, type: 'Enum_Option', option: 'Optional[Data_IPv4DH mode_default: 'Optional[int]' = None, mode_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long mode_reversed: 'bool' = False, + reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_IPv4DHCPSupportModeOption': """Make MH IPv4 DHCP support mode option. @@ -9214,6 +9294,7 @@ def _make_opt_ipv4_dhcp(self, type: 'Enum_Option', option: 'Optional[Data_IPv4DH mode_default: Default DHCP support mode. mode_namespace: DHCP support mode namespace. mode_reversed: Reverse DHCP support mode namespace. + reserved: Reserved bits, i.e. the 15 bits before the ``S`` flag. **kwargs: Arbitrary keyword arguments. Returns: @@ -9222,6 +9303,7 @@ def _make_opt_ipv4_dhcp(self, type: 'Enum_Option', option: 'Optional[Data_IPv4DH """ if option is not None: mode_val = int(option.mode) + reserved = option.reserved else: mode_val = self._make_index(mode, mode_default, namespace=mode_namespace, reversed=mode_reversed, pack=False) @@ -9232,7 +9314,7 @@ def _make_opt_ipv4_dhcp(self, type: 'Enum_Option', option: 'Optional[Data_IPv4DH return Schema_IPv4DHCPSupportModeOption( type=type, length=2, - flags={'S': mode_val}, + flags={'S': mode_val, 'reserved': reserved}, ) def _make_opt_cr(self, type: 'Enum_Option', option: 'Optional[Data_ContextRequestOption]' = None, *, @@ -9365,6 +9447,7 @@ def _make_opt_mn_lla_iid(self, type: 'Enum_Option', option: 'Optional[Data_MNLLA def _make_opt_transient(self, type: 'Enum_Option', option: 'Optional[Data_TransientBindingOption]' = None, *, late: 'bool' = False, lifetime: 'int | timedelta' = 0, + reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_TransientBindingOption': """Make MH transient binding option. @@ -9374,6 +9457,7 @@ def _make_opt_transient(self, type: 'Enum_Option', option: 'Optional[Data_Transi late: Late path switch flag. lifetime: Maximum lifetime of the transient state, in units of 100 milliseconds or as a :class:`~datetime.timedelta`. + reserved: Reserved bits, i.e. the seven bits before the ``L`` flag. **kwargs: Arbitrary keyword arguments. Returns: @@ -9382,6 +9466,7 @@ def _make_opt_transient(self, type: 'Enum_Option', option: 'Optional[Data_Transi """ if option is not None: late = option.late + reserved = option.reserved lifetime_val = math.ceil(option.lifetime.total_seconds() * 10) else: lifetime_val = lifetime if isinstance(lifetime, int) else math.ceil( @@ -9393,7 +9478,7 @@ def _make_opt_transient(self, type: 'Enum_Option', option: 'Optional[Data_Transi return Schema_TransientBindingOption( type=type, length=2, - flags={'L': int(late)}, + flags={'L': int(late), 'reserved': reserved}, lifetime=lifetime_val, ) @@ -9609,6 +9694,7 @@ def _make_opt_rc(self, type: 'Enum_Option', option: 'Optional[Data_RedirectCapab def _make_opt_redirect(self, type: 'Enum_Option', option: 'Optional[Data_RedirectOption]' = None, *, ipv6: 'Optional[bytes | str | int | IPv6Address]' = None, ipv4: 'Optional[bytes | str | int | IPv4Address]' = None, + reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_RedirectOption': """Make MH redirect mobility option. @@ -9617,6 +9703,7 @@ def _make_opt_redirect(self, type: 'Enum_Option', option: 'Optional[Data_Redirec option: Option data model. ipv6: IPv6 address of the redirected-to local mobility anchor. ipv4: IPv4 address of the redirected-to local mobility anchor. + reserved: Reserved bits, i.e. the 14 bits after the ``N`` flag. **kwargs: Arbitrary keyword arguments. Returns: @@ -9630,6 +9717,7 @@ def _make_opt_redirect(self, type: 'Enum_Option', option: 'Optional[Data_Redirec if option is not None: ipv6 = option.ipv6 ipv4 = option.ipv4 + reserved = option.reserved if (ipv6 is None) == (ipv4 is None): raise ProtocolError(f'{self.alias}: [OptNo {type}] exactly one of the IPv6 and ' @@ -9641,6 +9729,7 @@ def _make_opt_redirect(self, type: 'Enum_Option', option: 'Optional[Data_Redirec flags={ 'K': int(ipv6 is not None), 'N': int(ipv4 is not None), + 'reserved': reserved, }, ipv6=ipv6, ipv4=ipv4, @@ -9940,6 +10029,7 @@ def _make_opt_offload(self, type: 'Enum_Option', option: 'Optional[Data_IPv4TrafficOffloadSelectorOption]' = None, *, mode: 'bool' = False, selector: 'Optional[FlowIDSuboption | list[Schema_FlowIdentificationSuboption | tuple[Enum_FlowIDSuboption, dict[str, Any]] | bytes]]' = None, + reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_IPv4TrafficOffloadSelectorOption': """Make MH IPv4 traffic offload selector option. @@ -9948,6 +10038,7 @@ def _make_opt_offload(self, type: 'Enum_Option', option: Option data model. mode: Offload mode flag. selector: Traffic selector sub-options. + reserved: Reserved bits, i.e. the 31 bits after the ``M`` flag. **kwargs: Arbitrary keyword arguments. Returns: @@ -9956,6 +10047,7 @@ def _make_opt_offload(self, type: 'Enum_Option', """ if option is not None: mode = option.mode + reserved = option.reserved selector = option.selector else: selector = selector or [] @@ -9967,7 +10059,7 @@ def _make_opt_offload(self, type: 'Enum_Option', return Schema_IPv4TrafficOffloadSelectorOption( type=type, length=length, - flags={'M': int(mode)}, + flags={'M': int(mode), 'reserved': reserved}, selector=entries, ) @@ -9977,6 +10069,7 @@ def _make_opt_mcast_sel(self, type: 'Enum_Option', mode: 'bool' = False, records: 'int' = 0, data: 'bytes' = b'', + reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_DynamicIPMulticastSelectorOption': """Make MH dynamic IP multicast selector option. @@ -9987,6 +10080,7 @@ def _make_opt_mcast_sel(self, type: 'Enum_Option', mode: Selector mode flag. records: Number of multicast address records in ``data``. data: Multicast address records, opaque to this module. + reserved: Reserved bits, i.e. the seven bits after the ``M`` flag. **kwargs: Arbitrary keyword arguments. Returns: @@ -9996,6 +10090,7 @@ def _make_opt_mcast_sel(self, type: 'Enum_Option', if option is not None: protocol = option.protocol mode = option.mode + reserved = option.reserved records = option.records data = option.data @@ -10003,7 +10098,7 @@ def _make_opt_mcast_sel(self, type: 'Enum_Option', type=type, length=4 + len(data), protocol=protocol, - flags={'M': int(mode)}, + flags={'M': int(mode), 'reserved': reserved}, records=records, data=data, ) @@ -10011,6 +10106,7 @@ def _make_opt_mcast_sel(self, type: 'Enum_Option', def _make_opt_dmnp(self, type: 'Enum_Option', option: 'Optional[Data_DelegatedMNPOption]' = None, *, prefix_length: 'int' = 64, prefix: 'bytes | str | int | IPv4Address | IPv6Address' = '::', + reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_DelegatedMNPOption': """Make MH delegated mobile network prefix option. @@ -10019,6 +10115,7 @@ def _make_opt_dmnp(self, type: 'Enum_Option', option: 'Optional[Data_DelegatedMN option: Option data model. prefix_length: Prefix length. prefix: Delegated mobile network prefix. + reserved: Reserved bits, i.e. the seven bits after the ``V`` flag. **kwargs: Arbitrary keyword arguments. Returns: @@ -10031,6 +10128,7 @@ def _make_opt_dmnp(self, type: 'Enum_Option', option: 'Optional[Data_DelegatedMN """ if option is not None: + reserved = option.reserved prefix_length = option.prefix_length prefix = option.prefix @@ -10050,7 +10148,7 @@ def _make_opt_dmnp(self, type: 'Enum_Option', option: 'Optional[Data_DelegatedMN return Schema_DelegatedMNPOption( type=type, length=6 if ipv4 else 18, - flags={'V': int(ipv4)}, + flags={'V': int(ipv4), 'reserved': reserved}, prefix_length=prefix_length, prefix=addr, ) @@ -10172,13 +10270,15 @@ def _make_qos_attribute(self, code: 'Enum_QoSAttribute', if option is not None: session = cast('Data_PerSessionBitRateAttribute', option) service, exclude, rate = session.service, session.exclude, session.rate + reserved = session.reserved else: service = cast('bool', kwargs.get('service', False)) exclude = cast('bool', kwargs.get('exclude', False)) + reserved = cast('int', kwargs.get('reserved', 0)) rate = cast('int', kwargs.get('rate', 0)) return Schema_PerSessionBitRateAttribute( type=code, length=6, - flags={'S': int(service), 'E': int(exclude)}, rate=rate) + flags={'S': int(service), 'E': int(exclude), 'reserved': reserved}, rate=rate) if code in (Enum_QoSAttribute.Per_MN_Agg_Max_DL_Bit_Rate, Enum_QoSAttribute.Per_MN_Agg_Max_UL_Bit_Rate, @@ -10543,6 +10643,7 @@ def _make_opt_mag_mp(self, type: 'Enum_Option', option: 'Optional[Data_MAGMultip bid: 'int' = 1, bulk: 'bool' = False, overwrite: 'bool' = False, + reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_MAGMultipathBindingOption': """Make MH MAG multipath binding option. @@ -10557,6 +10658,7 @@ def _make_opt_mag_mp(self, type: 'Enum_Option', option: 'Optional[Data_MAGMultip bid: Binding identifier; ``0`` and ``255`` are reserved. bulk: Bulk re-registration flag. overwrite: Registration overwrite flag. + reserved: Reserved bits, i.e. the 22 bits after the ``O`` flag. **kwargs: Arbitrary keyword arguments. Returns: @@ -10573,6 +10675,7 @@ def _make_opt_mag_mp(self, type: 'Enum_Option', option: 'Optional[Data_MAGMultip bid = option.bid bulk = option.bulk overwrite = option.overwrite + reserved = option.reserved else: att_val = self._make_index(att, att_default, namespace=att_namespace, # type: ignore[assignment] reversed=att_reversed, pack=False) @@ -10590,6 +10693,7 @@ def _make_opt_mag_mp(self, type: 'Enum_Option', option: 'Optional[Data_MAGMultip flags={ 'B': int(bulk), 'O': int(overwrite), + 'reserved': reserved, }, ) @@ -10882,6 +10986,7 @@ def _make_ext_none(self, type: 'Enum_CGAExtension', option: 'Optional[Data_Unkno def _make_ext_multiprefix(self, type: 'Enum_CGAExtension', option: 'Optional[Data_MultiPrefixExtension]' = None, *, flag: 'bool' = False, prefixes: 'Optional[list[int]]' = None, + reserved: 'int' = 0, **kwargs: 'Any') -> 'Schema_MultiPrefixExtension': """Make CGA multi-prefix extension. @@ -10890,6 +10995,7 @@ def _make_ext_multiprefix(self, type: 'Enum_CGAExtension', option: 'Optional[Dat option: Extension data model. flag: Public key flag. prefixes: Prefixes. + reserved: Reserved bits, i.e. the 31 bits after the ``P`` flag. **kwargs: Arbitrary keyword arguments. Returns: @@ -10898,6 +11004,7 @@ def _make_ext_multiprefix(self, type: 'Enum_CGAExtension', option: 'Optional[Dat """ if option is not None: flag = option.flag + reserved = option.reserved # NOTE: ``list()`` rather than a cast: the data model stores the # prefixes as a :obj:`tuple`, which # :class:`~pcapkit.corekit.fields.collections.ListField` refuses to @@ -10919,6 +11026,7 @@ def _make_ext_multiprefix(self, type: 'Enum_CGAExtension', option: 'Optional[Dat length=4 + len(prefixes) * 8, flags={ 'P': int(flag), + 'reserved': reserved, }, prefixes=prefixes, ) diff --git a/pcapkit/protocols/schema/internet/mh.py b/pcapkit/protocols/schema/internet/mh.py index 80cf3664a2..a2a2fd3482 100644 --- a/pcapkit/protocols/schema/internet/mh.py +++ b/pcapkit/protocols/schema/internet/mh.py @@ -148,6 +148,7 @@ class MultiPrefixExtensionFlags(TypedDict): """Flags for :attr:`MultiPrefixExtension.flags`.""" P: int + reserved: int class BindingUpdateMessageFlags(TypedDict): """Flags for :attr:`BindingUpdateMessage.flags`.""" @@ -185,11 +186,13 @@ class FastBindingUpdateMessageFlags(TypedDict): H: 'int' L: 'int' K: 'int' + reserved: 'int' class FastBindingAcknowledgmentMessageFlags(TypedDict): """Flags for :attr:`FastBindingAcknowledgmentMessage.flags`.""" K: 'int' + reserved: 'int' class HandoverInitiateMessageFlags(TypedDict): """Flags for :attr:`HandoverInitiateMessage.flags`.""" @@ -198,6 +201,7 @@ class HandoverInitiateMessageFlags(TypedDict): U: 'int' P: 'int' F: 'int' + reserved: 'int' class HandoverAcknowledgeMessageFlags(TypedDict): """Flags for :attr:`HandoverAcknowledgeMessage.flags`.""" @@ -205,12 +209,14 @@ class HandoverAcknowledgeMessageFlags(TypedDict): U: 'int' P: 'int' F: 'int' + reserved: 'int' class HeartbeatMessageFlags(TypedDict): """Flags for :attr:`HeartbeatMessage.flags`.""" U: 'int' R: 'int' + reserved: 'int' class FixedPointTimestamp(TypedDict): """Fixed-point timestamp of :attr:`TimestampOption.timestamp`.""" @@ -224,32 +230,38 @@ class BindingRevocationMessageFlags(TypedDict): P: 'int' V: 'int' G: 'int' + reserved: 'int' class LocalizedRoutingAcknowledgmentMessageFlags(TypedDict): """Flags for :attr:`LocalizedRoutingAcknowledgmentMessage.flags`.""" U: 'int' + reserved: 'int' class FlowBindingMessageFlags(TypedDict): """Flags for :attr:`FlowBindingMessage.flags`.""" A: 'int' + reserved: 'int' class UpdateNotificationMessageFlags(TypedDict): """Flags for :attr:`UpdateNotificationMessage.flags`.""" A: 'int' D: 'int' + reserved: 'int' class SubscriptionResponseMessageFlags(TypedDict): """Flags for :attr:`SubscriptionResponseMessage.flags`.""" I: 'int' + reserved: 'int' class DNSUpdateOptionFlags(TypedDict): """Flags for :attr:`DNSUpdateOption.flags`.""" R: 'int' + reserved: 'int' class IPv4HomeAddressOptionFlags(TypedDict): """Prefix length and flags of :attr:`IPv4HomeAddressOption.flags`.""" @@ -265,22 +277,26 @@ class PrefixLengthOnly(TypedDict): :attr:`IPv4HomeAddressReplyOption.flags`.""" prefix_length: 'int' + reserved: 'int' class NATDetectionOptionFlags(TypedDict): """Flags for :attr:`NATDetectionOption.flags`.""" F: 'int' + reserved: 'int' class IPv4DHCPSupportModeOptionFlags(TypedDict): """Flags for :attr:`IPv4DHCPSupportModeOption.flags`.""" S: 'int' + reserved: 'int' class RedirectOptionFlags(TypedDict): """Flags for :attr:`RedirectOption.flags`.""" K: 'int' N: 'int' + reserved: 'int' class BindingIdentifierOptionFlags(TypedDict): """Flag and priority of :attr:`BindingIdentifierOption.flags`.""" @@ -292,6 +308,7 @@ class TransientBindingOptionFlags(TypedDict): """Flags for :attr:`TransientBindingOption.flags`.""" L: 'int' + reserved: 'int' class ModeFlagOnly(TypedDict): """A lone ``M`` mode flag, as in @@ -299,17 +316,20 @@ class ModeFlagOnly(TypedDict): :attr:`IPv4TrafficOffloadSelectorOption.flags`.""" M: 'int' + reserved: 'int' class DelegatedMNPOptionFlags(TypedDict): """Flags for :attr:`DelegatedMNPOption.flags`.""" V: 'int' + reserved: 'int' class MAGMultipathBindingOptionFlags(TypedDict): """Flags for :attr:`MAGMultipathBindingOption.flags`.""" B: 'int' O: 'int' + reserved: 'int' class ANINetworkIdentifierFlags(TypedDict): """Flags for :attr:`ANINetworkIdentifierSuboption.flags`.""" @@ -328,6 +348,7 @@ class SessionBitRateFlags(TypedDict): S: 'int' E: 'int' + reserved: 'int' class AllocationRetentionPriorityFields(TypedDict): """Priority fields of :attr:`AllocationRetentionPriorityAttribute.priority`.""" @@ -850,6 +871,7 @@ class MultiPrefixExtension(CGAExtension, code=Enum_CGAExtension.Multi_Prefix): #: Flags. flags: 'MultiPrefixExtensionFlags' = BitField(length=4, namespace={ 'P': (0, 1), + 'reserved': (1, 31), }) #: Prefixes. prefixes: 'list[int]' = ListField( @@ -1046,6 +1068,7 @@ class DNSUpdateOption(Option, code=Enum_Option.DNS_UPDATE_TYPE): #: Flags. flags: 'DNSUpdateOptionFlags' = BitField(length=1, namespace={ 'R': (0, 1), + 'reserved': (1, 7), }) #: Mobile node identity, in FQDN form. identity: 'bytes' = BytesField(length=lambda pkt: pkt['length'] - 2) @@ -1217,6 +1240,7 @@ class IPv4AddressAcknowledgementOption(Option, code=Enum_Option.IPv4_Address_Ack #: Prefix length, with 2 reserved bits below it. flags: 'PrefixLengthOnly' = BitField(length=1, namespace={ 'prefix_length': (0, 6), + 'reserved': (6, 2), }) #: IPv4 home address. address: 'IPv4Address' = IPv4AddressField() @@ -1234,6 +1258,7 @@ class NATDetectionOption(Option, code=Enum_Option.NAT_Detection): #: Flags. flags: 'NATDetectionOptionFlags' = BitField(length=2, namespace={ 'F': (0, 1), + 'reserved': (1, 15), }) #: Suggested NAT binding refresh time, in seconds. refresh: 'int' = UInt32Field() @@ -1326,6 +1351,7 @@ class IPv4HomeAddressRequestOption(Option, code=Enum_Option.IPv4_Home_Address_Re #: Prefix length, with 10 reserved bits below it. flags: 'PrefixLengthOnly' = BitField(length=2, namespace={ 'prefix_length': (0, 6), + 'reserved': (6, 10), }) #: Requested IPv4 home address. address: 'IPv4Address' = IPv4AddressField() @@ -1344,6 +1370,7 @@ class IPv4HomeAddressReplyOption(Option, code=Enum_Option.IPv4_Home_Address_Repl #: Prefix length, with 2 reserved bits below it. flags: 'PrefixLengthOnly' = BitField(length=1, namespace={ 'prefix_length': (0, 6), + 'reserved': (6, 2), }) #: Assigned IPv4 home address. address: 'IPv4Address' = IPv4AddressField() @@ -1375,6 +1402,7 @@ class IPv4DHCPSupportModeOption(Option, code=Enum_Option.IPv4_DHCP_Support_Mode) #: Flags. The ``S`` bit is the **last** bit of the option, not the first #: [:rfc:`5844#section-3.3.4`]. flags: 'IPv4DHCPSupportModeOptionFlags' = BitField(length=2, namespace={ + 'reserved': (0, 15), 'S': (15, 1), }) @@ -1462,6 +1490,7 @@ class TransientBindingOption(Option, code=Enum_Option.Transient_Binding): #: Flags. flags: 'TransientBindingOptionFlags' = BitField(length=1, namespace={ + 'reserved': (0, 7), 'L': (7, 1), }) #: Maximum lifetime of the transient state, in units of 100 milliseconds. @@ -1679,6 +1708,7 @@ class RedirectOption(Option, code=Enum_Option.Redirect_Mobility_Option): flags: 'RedirectOptionFlags' = BitField(length=2, namespace={ 'K': (0, 1), 'N': (1, 1), + 'reserved': (2, 14), }) #: IPv6 address of the redirected-to LMA, present when ``K`` is set. ipv6: 'Optional[IPv6Address]' = ConditionalField( @@ -1934,6 +1964,7 @@ class IPv4TrafficOffloadSelectorOption(Option, code=Enum_Option.IPv4_Traffic_Off #: Flags. flags: 'ModeFlagOnly' = BitField(length=4, namespace={ 'M': (0, 1), + 'reserved': (1, 31), }) #: Traffic selector, carried as the traffic selector sub-option of #: :rfc:`6089#section-4.2.1.4`. Optional in a proxy binding update and @@ -1970,6 +2001,7 @@ class DynamicIPMulticastSelectorOption(Option, code=Enum_Option.Dynamic_IP_Multi #: Flags. flags: 'ModeFlagOnly' = BitField(length=1, namespace={ 'M': (0, 1), + 'reserved': (1, 7), }) #: Number of multicast address records. records: 'int' = UInt16Field() @@ -1988,6 +2020,7 @@ class DelegatedMNPOption(Option, code=Enum_Option.Delegated_Mobile_Network_Prefi #: Flags. flags: 'DelegatedMNPOptionFlags' = BitField(length=1, namespace={ 'V': (0, 1), + 'reserved': (1, 7), }) #: Prefix length. prefix_length: 'int' = UInt8Field() @@ -2111,6 +2144,7 @@ class PerSessionBitRateAttribute( flags: 'SessionBitRateFlags' = BitField(length=2, namespace={ 'S': (0, 1), 'E': (1, 1), + 'reserved': (2, 14), }) #: Bit rate, in bits per second. rate: 'int' = UInt32Field() @@ -2378,6 +2412,7 @@ class MAGMultipathBindingOption(Option, code=Enum_Option.MAG_Multipath_Binding): flags: 'MAGMultipathBindingOptionFlags' = BitField(length=3, namespace={ 'B': (0, 1), 'O': (1, 1), + 'reserved': (2, 22), }) if TYPE_CHECKING: @@ -2731,6 +2766,7 @@ class FastBindingUpdateMessage(Packet, code=Enum_Packet.Fast_Binding_Update): 'H': (1, 1), 'L': (2, 1), 'K': (3, 1), + 'reserved': (4, 12), }) #: Lifetime, in seconds [:rfc:`5568#section-6.2.2`]. lifetime: 'int' = UInt16Field() @@ -2771,6 +2807,7 @@ class FastBindingAcknowledgmentMessage(Packet, code=Enum_Packet.Fast_Binding_Ack #: Flags. flags: 'FastBindingAcknowledgmentMessageFlags' = BitField(length=1, namespace={ 'K': (0, 1), + 'reserved': (1, 7), }) #: Sequence number. seq: 'int' = UInt16Field() @@ -2832,6 +2869,7 @@ class HandoverInitiateMessage(Packet, code=Enum_Packet.Handover_Initiate_Message 'U': (1, 1), 'P': (2, 1), 'F': (3, 1), + 'reserved': (4, 4), }) #: Code. code: 'Enum_HandoverInitiateStatus' = EnumField(length=1, namespace=Enum_HandoverInitiateStatus) @@ -2860,6 +2898,7 @@ class HandoverAcknowledgeMessage(Packet, code=Enum_Packet.Handover_Acknowledge_M 'U': (0, 1), 'P': (1, 1), 'F': (2, 1), + 'reserved': (3, 5), }) #: Code. code: 'Enum_HandoverACKStatus' = EnumField(length=1, namespace=Enum_HandoverACKStatus) @@ -2883,6 +2922,7 @@ class HeartbeatMessage(Packet, code=Enum_Packet.Heartbeat_Message): #: Flags. flags: 'HeartbeatMessageFlags' = BitField(length=2, namespace={ + 'reserved': (0, 14), 'U': (14, 1), 'R': (15, 1), }) @@ -2955,6 +2995,7 @@ class BindingRevocationMessage(Packet, code=Enum_Packet.Binding_Revocation_Messa 'P': (0, 1), 'V': (1, 1), 'G': (2, 1), + 'reserved': (3, 13), }) #: Mobility options. options: 'list[Option]' = OptionField( @@ -3020,6 +3061,7 @@ class LocalizedRoutingAcknowledgmentMessage( #: Flags. flags: 'LocalizedRoutingAcknowledgmentMessageFlags' = BitField(length=1, namespace={ 'U': (0, 1), + 'reserved': (1, 7), }) #: Status, c.f., #: :class:`pcapkit.protocols.internet.mh.LocalizedRoutingStatus`. @@ -3054,6 +3096,7 @@ class UpdateNotificationMessage(Packet, code=Enum_Packet.Update_Notification): flags: 'UpdateNotificationMessageFlags' = BitField(length=2, namespace={ 'A': (0, 1), 'D': (1, 1), + 'reserved': (2, 14), }) #: Mobility options. options: 'list[Option]' = OptionField( @@ -3123,6 +3166,7 @@ class FlowBindingMessage(Packet, code=Enum_Packet.Flow_Binding_Message): #: Flags. flags: 'FlowBindingMessageFlags' = BitField(length=1, namespace={ 'A': (0, 1), + 'reserved': (1, 7), }) #: Mobility options. options: 'list[Option]' = OptionField( @@ -3176,6 +3220,7 @@ class SubscriptionResponseMessage(Packet, code=Enum_Packet.Subscription_Response #: Flags. flags: 'SubscriptionResponseMessageFlags' = BitField(length=1, namespace={ 'I': (0, 1), + 'reserved': (1, 7), }) #: Mobility options. options: 'list[Option]' = OptionField( diff --git a/tests/protocols/internet/test_mh_flags_reserved_unit.py b/tests/protocols/internet/test_mh_flags_reserved_unit.py new file mode 100644 index 0000000000..4ef547afb4 --- /dev/null +++ b/tests/protocols/internet/test_mh_flags_reserved_unit.py @@ -0,0 +1,192 @@ +# -*- coding: utf-8 -*- +"""MH flag words keep their reserved bits on rebuild. + +GitHub issue #1418: 24 MH ``flags`` bit fields named only some of their bits, +and :class:`~pcapkit.corekit.fields.strings.BitField` packs an unnamed bit as +zero, so a message, option or sub-structure whose reserved bits were set came +back with them cleared. Each site now names a ``reserved`` sub-field, read into +the data model (``flags_reserved`` on a message, since :attr:`MH.reserved` is +the fixed-header octet, and ``reserved`` elsewhere) and written back by +``make()``, where it defaults to ``0``. + +Every frame sets every reserved bit of its site and nothing else unusual. Each +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 + +#: ``site -> (Mobility Header octets, width of the reserved sub-field)``. The +#: options ride in a Binding Refresh Request; next header is 59 throughout. +FRAMES = { + 'BindingRevocationMessage': ('3b0110000000010000001fff01020000', 13), + 'FastBindingAcknowledgmentMessage': ('3b0109000000007f0000000401020000', 7), + 'FastBindingUpdateMessage': ('3b01080000000000cfff000401020000', 12), + 'FlowBindingMessage': ('3b011500000000010000017f01020000', 7), + 'HandoverAcknowledgeMessage': ('3b010f00000000001f00010400000000', 5), + 'HandoverInitiateMessage': ('3b010e00000000000f00010400000000', 4), + 'HeartbeatMessage': ('3b010d000000fffc0000000001020000', 14), + 'LocalizedRoutingAcknowledgmentMessage': ('3b011200000000007f00000001020000', 7), + 'SubscriptionResponseMessage': ('3b0017000000007f', 7), + 'UpdateNotificationMessage': ('3b0113000000000000013fff01020000', 14), + 'DNSUpdateOption': ('3b010000000000001102007f01020000', 7), + 'DelegatedMNPOption': ('3b010000000000003706ff18c0000200', 7), + 'DynamicIPMulticastSelectorOption': ('3b0100000000000036048f7f00000000', 7), + 'IPv4AddressAcknowledgementOption': ('3b010000000000001e06008300000000', 2), + 'IPv4DHCPSupportModeOption': ('3b010000000000002702fffe01020000', 15), + 'IPv4HomeAddressReplyOption': ('3b010000000000002506008300000000', 2), + 'IPv4HomeAddressRequestOption': ('3b01000000000000240683ff00000000', 10), + 'IPv4TrafficOffloadSelectorOption': ('3b0100000000000035047fffffff0000', 31), + 'MAGMultipathBindingOption': ('3b010000000000003f060100013fffff', 22), + 'NATDetectionOption': ('3b010000000000001f067fff00000000', 15), + 'RedirectOption': ('3b010000000000002f067fffc0000201', 14), + 'TransientBindingOption': ('3b010000000000002b02fe0001020000', 7), + # CGA Parameters option carrying one CGA parameter with a Multi-Prefix extension + 'MultiPrefixExtension': ('3b060000000000000c2b086fca5e10b200c99c8ce00164277c08' + '0000000000000001013000' '0012000c7fffffff0000000000000001' + '010100', 31), + # Quality-of-Service option carrying one per-session bit-rate attribute + 'PerSessionBitRateAttribute': ('3b020000000000003a0e00000000000003063fff000003e8', 14), +} + +MESSAGES = { + 'BindingRevocationMessage': 'Binding_Revocation_Message', + 'FastBindingAcknowledgmentMessage': 'Fast_Binding_Acknowledgment', + 'FastBindingUpdateMessage': 'Fast_Binding_Update', + 'FlowBindingMessage': 'Flow_Binding_Message', + 'HandoverAcknowledgeMessage': 'Handover_Acknowledge_Message', + 'HandoverInitiateMessage': 'Handover_Initiate_Message', + 'HeartbeatMessage': 'Heartbeat_Message', + 'LocalizedRoutingAcknowledgmentMessage': 'Localized_Routing_Acknowledgment', + 'SubscriptionResponseMessage': 'Subscription_Response', + 'UpdateNotificationMessage': 'Update_Notification', +} + +#: ``site -> (option name, extra make() arguments)``. +OPTIONS = { + 'DNSUpdateOption': ('DNS_UPDATE_TYPE', {}), + 'DelegatedMNPOption': ('Delegated_Mobile_Network_Prefix', {'prefix': '192.0.2.0', 'prefix_length': 24}), + 'DynamicIPMulticastSelectorOption': ('Dynamic_IP_Multicast_Selector', {}), + 'IPv4AddressAcknowledgementOption': ('IPv4_Address_Acknowledgement', {}), + 'IPv4DHCPSupportModeOption': ('IPv4_DHCP_Support_Mode', {}), + 'IPv4HomeAddressReplyOption': ('IPv4_Home_Address_Reply', {}), + 'IPv4HomeAddressRequestOption': ('IPv4_Home_Address_Request', {}), + 'IPv4TrafficOffloadSelectorOption': ('IPv4_Traffic_Offload_Selector', {}), + 'MAGMultipathBindingOption': ('MAG_Multipath_Binding', {}), + 'NATDetectionOption': ('NAT_Detection', {}), + 'RedirectOption': ('Redirect_Mobility_Option', {'ipv4': '192.0.2.1'}), + 'TransientBindingOption': ('Transient_Binding', {}), +} + + +class TestMHFlagsReserved(unittest.TestCase): + """Pin the reserved bits of every MH ``flags`` bit field.""" + + 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) + + @staticmethod + def _site(info: 'object', site: 'str') -> 'tuple[object, str]': + """Return the data model holding the site's reserved bits and its name.""" + from pcapkit.const.mh.cga_extension import CGAExtension + from pcapkit.const.mh.option import Option + from pcapkit.const.mh.qos_attribute import QoSAttribute + + if site in MESSAGES: + return info, 'flags_reserved' + if site in OPTIONS: + return info.options[Option[OPTIONS[site][0]]], 'reserved' # type: ignore[attr-defined] + if site == 'MultiPrefixExtension': + param = info.options[Option.CGA_Parameters].parameters[0] # type: ignore[attr-defined] + return param.extensions[CGAExtension.Multi_Prefix], 'reserved' + qos = info.options[Option.Quality_of_Service] # type: ignore[attr-defined] + return qos.attributes[QoSAttribute.Per_Session_Agg_Max_DL_Bit_Rate], 'reserved' + + @staticmethod + def _make(site: 'str', **reserved: 'int') -> 'bytes': + """Build the site through ``make()``, passing ``reserved`` through.""" + from pcapkit.const.mh.cga_extension import CGAExtension + from pcapkit.const.mh.cga_type import CGAType + from pcapkit.const.mh.option import Option + from pcapkit.const.mh.packet import Packet + from pcapkit.const.mh.qos_attribute import QoSAttribute + from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.data.internet.mh import CGAParameter + from pcapkit.protocols.internet.mh import MH + + if site in MESSAGES: + data = {'flags_reserved': reserved['value']} if reserved else {} + return bytes(MH(next=TransType.IPv6_NoNxt, type=Packet[MESSAGES[site]], data=data)) + + args = {'reserved': reserved['value']} if reserved else {} + if site in OPTIONS: + name, extra = OPTIONS[site] + option = (Option[name], dict(extra, **args)) + elif site == 'MultiPrefixExtension': + param = CGAParameter(modifier=CGAType.Tag_086F_CA5E_10B2_00C9_9C8C_E001_6427_7C08, + prefix=1, collision_count=1, public_key=b'\x30\x00', + extensions=[(CGAExtension.Multi_Prefix, dict(args, prefixes=[1]))]) + option = (Option.CGA_Parameters, {'parameters': [param]}) + else: + option = (Option.Quality_of_Service, {'attributes': [ + (QoSAttribute.Per_Session_Agg_Max_DL_Bit_Rate, dict(args, rate=1000))]}) + return bytes(MH(next=TransType.IPv6_NoNxt, type=Packet.Binding_Refresh_Request, + data={'options': [option]})) + + def test_every_site_is_covered(self) -> None: + self.assertEqual(len(FRAMES), 24) + self.assertEqual(set(FRAMES), set(MESSAGES) | set(OPTIONS) + | {'MultiPrefixExtension', 'PerSessionBitRateAttribute'}) + + def test_reserved_bits_are_read(self) -> None: + for site, (frame, width) in FRAMES.items(): + with self.subTest(site=site): + model, attr = self._site(self._parse(bytes.fromhex(frame)).info, site) # type: ignore[attr-defined] + self.assertEqual(type(model).__name__, site) + self.assertEqual(getattr(model, attr), (1 << width) - 1) + + def test_from_data_rebuilds_byte_for_byte(self) -> None: + from pcapkit.protocols.internet.mh import MH + + for site, (frame, _) in FRAMES.items(): + with self.subTest(site=site): + 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_make_writes_the_reserved_bits(self) -> None: + for site, (frame, width) in FRAMES.items(): + with self.subTest(site=site): + built = self._make(site, value=(1 << width) - 1) + self.assertEqual(built.hex(), frame) + model, attr = self._site(self._parse(built).info, site) # type: ignore[attr-defined] + self.assertEqual(getattr(model, attr), (1 << width) - 1) + + def test_make_defaults_the_reserved_bits_to_zero(self) -> None: + for site in FRAMES: + with self.subTest(site=site): + model, attr = self._site(self._parse(self._make(site)).info, site) # type: ignore[attr-defined] + self.assertEqual(getattr(model, attr), 0) + + def test_make_rejects_a_value_wider_than_the_reserved_bits(self) -> None: + from pcapkit.utilities.exceptions import FieldValueError + + for site, (_, width) in FRAMES.items(): + with self.subTest(site=site): + with self.assertRaisesRegex(FieldValueError, r"subfield 'reserved' value \d+ needs"): + self._make(site, value=1 << width) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/internet/test_mh_unit.py b/tests/protocols/internet/test_mh_unit.py index dbae981fda..9edb916ee8 100644 --- a/tests/protocols/internet/test_mh_unit.py +++ b/tests/protocols/internet/test_mh_unit.py @@ -425,7 +425,7 @@ def test_mh_option_readers_cover_known_options_and_guards(self) -> None: ext = proto._read_cga_extensions([ schema.UnknownExtension(type=CGAExtension.Exp_FFFD, length=2, data=b'xx'), schema.MultiPrefixExtension(type=CGAExtension.Multi_Prefix, length=20, - flags={'P': 1}, prefixes=[1, 2]), + flags={'P': 1, 'reserved': 0}, prefixes=[1, 2]), ]) self.assertEqual(list(ext.keys()), [CGAExtension.Exp_FFFD, CGAExtension.Multi_Prefix]) self.assertEqual(proto._read_ext_none( @@ -434,7 +434,7 @@ def test_mh_option_readers_cover_known_options_and_guards(self) -> None: ).data, b'yy') self.assertTrue(proto._read_ext_multiprefix( schema.MultiPrefixExtension(type=CGAExtension.Multi_Prefix, length=20, - flags={'P': 1}, prefixes=[1]), + flags={'P': 1, 'reserved': 0}, prefixes=[1]), extensions=ext, ).flag) @@ -580,7 +580,7 @@ def test_mh_option_constructors_cover_known_options_and_dispatch(self) -> None: self.assertTrue(proto._make_ext_multiprefix( CGAExtension.Multi_Prefix, data.MultiPrefixExtension(type=CGAExtension.Multi_Prefix, - length=20, flag=True, prefixes=(3,)), + length=20, flag=True, reserved=0, prefixes=(3,)), ).flags['P']) ext_list, ext_len = proto._make_cga_extensions([ b'\xff\xfd\x00\x02xx', @@ -929,7 +929,7 @@ def header(type_: Packet) -> SimpleNamespace: # "the requested time in seconds", unlike the BU/BA lifetime's units # of 4 seconds, so it is read unscaled (c.f. #493). fbu = proto._read_msg_fbu(SimpleNamespace(seq=0x1234, - flags={'A': 1, 'H': 1, 'L': 0, 'K': 1}, + flags={'A': 1, 'H': 1, 'L': 0, 'K': 1, 'reserved': 0}, lifetime=10, options=[]), header=header(Packet.Fast_Binding_Update)) self.assertEqual(fbu.seq, 0x1234) @@ -944,7 +944,7 @@ def header(type_: Packet) -> SimpleNamespace: # which is *not* what StatusCode(1) means, hence the module-local enum. # The Lifetime here is likewise "the granted lifetime ... in seconds" # (c.f. #493), so it too is read unscaled. - fback = proto._read_msg_fback(SimpleNamespace(status=1, flags={'K': 1}, seq=0x1234, + fback = proto._read_msg_fback(SimpleNamespace(status=1, flags={'K': 1, 'reserved': 0}, seq=0x1234, lifetime=10, options=[]), header=header(Packet.Fast_Binding_Acknowledgment)) self.assertEqual(fback.status, 1) @@ -969,7 +969,7 @@ def header(type_: Packet) -> SimpleNamespace: # 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}, + flags={'S': 1, 'U': 0, 'P': 1, 'F': 0, 'reserved': 0}, code=HandoverInitiateStatus.FBU_whose_source_IP_address_is_not_PCoA, options=[], ), header=header(Packet.Handover_Initiate_Message)) @@ -983,7 +983,7 @@ def header(type_: Packet) -> SimpleNamespace: # 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}, + flags={'U': 1, 'P': 0, 'F': 1, 'reserved': 0}, code=HandoverACKStatus.Handover_Accepted_NCoA_assigned, options=[], ), header=header(Packet.Handover_Acknowledge_Message)) @@ -998,9 +998,9 @@ def header(type_: Packet) -> SimpleNamespace: lifetime=datetime.timedelta(seconds=40), options=[]) self.assertEqual(made_fbu.seq, 7) self.assertEqual(made_fbu.lifetime, 40) - self.assertEqual(made_fbu.flags, {'A': True, 'H': True, 'L': False, 'K': True}) + self.assertEqual(made_fbu.flags, {'A': True, 'H': True, 'L': False, 'K': True, 'reserved': 0}) self.assertEqual(proto._make_msg_fbu(SimpleNamespace( - seq=8, ack=False, home=True, lla_compat=True, key_mngt=False, + seq=8, ack=False, home=True, lla_compat=True, key_mngt=False, flags_reserved=0, lifetime=datetime.timedelta(seconds=8), options=[], )).seq, 8) @@ -1017,7 +1017,7 @@ def header(type_: Packet) -> SimpleNamespace: ).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=[], + key_mngt=False, flags_reserved=0, seq=10, lifetime=datetime.timedelta(seconds=8), options=[], )).seq, 10) self.assertEqual(proto._make_msg_fna(None, options=[]).options, ['made']) @@ -1031,9 +1031,9 @@ def header(type_: Packet) -> SimpleNamespace: 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(made_hi.flags, {'S': True, 'U': False, 'P': True, 'F': False, 'reserved': 0}) self.assertEqual(proto._make_msg_hi(SimpleNamespace( - seq=12, assign=False, buffer=True, proxy=False, forward=True, + seq=12, assign=False, buffer=True, proxy=False, forward=True, flags_reserved=0, code=HandoverInitiateStatus.All_available_context_transferred, options=[], )).seq, 12) @@ -1041,9 +1041,9 @@ def header(type_: Packet) -> SimpleNamespace: 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(made_hack.flags, {'U': True, 'P': False, 'F': True, 'reserved': 0}) self.assertEqual(proto._make_msg_hack(SimpleNamespace( - seq=14, buffer=False, proxy=True, forward=False, + seq=14, buffer=False, proxy=True, forward=False, flags_reserved=0, code=HandoverACKStatus.Handover_Accepted_use_PCoA, options=[], )).seq, 14) @@ -1080,12 +1080,12 @@ def header(type_: Packet) -> SimpleNamespace: # -- read side: wire lifetime=100 -- fbu = proto._read_msg_fbu(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, 'reserved': 0}, lifetime=100, options=[], ), header=header(Packet.Fast_Binding_Update)) self.assertEqual(fbu.lifetime, datetime.timedelta(seconds=100)) fback = proto._read_msg_fback(SimpleNamespace( - status=0, flags={'K': 0}, seq=1, lifetime=100, options=[], + status=0, flags={'K': 0, 'reserved': 0}, seq=1, lifetime=100, options=[], ), header=header(Packet.Fast_Binding_Acknowledgment)) self.assertEqual(fback.lifetime, datetime.timedelta(seconds=100))