diff --git a/pcapkit/protocols/data/internet/mh.py b/pcapkit/protocols/data/internet/mh.py index af2bd7dbc..071b7f4b9 100644 --- a/pcapkit/protocols/data/internet/mh.py +++ b/pcapkit/protocols/data/internet/mh.py @@ -315,7 +315,8 @@ class FastBindingUpdateMessage(MH): lla_compat: 'bool' #: Key management mobility capability flag. key_mngt: 'bool' - #: Lifetime. + #: 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' #: Mobility options. options: 'OrderedMultiDict[Enum_Option, Option]' @@ -340,7 +341,8 @@ class FastBindingAcknowledgmentMessage(MH): key_mngt: 'bool' #: Sequence number. seq: 'int' - #: Lifetime. + #: Lifetime. :rfc:`5568#section-6.2.3` counts this in seconds rather than in + #: the units of 4 seconds the binding messages use. lifetime: 'timedelta' #: Mobility options. options: 'OrderedMultiDict[Enum_Option, Option]' @@ -677,11 +679,12 @@ def __init__(self, type: 'Enum_Option', length: 'int') -> 'None': ... class BindingRefreshAdviceOption(Option): """Data model for Binding Refresh Advice option.""" - #: Refresh interval. - interval: 'int' + #: Refresh interval before re-registration. Counted on the wire in units of + #: 4 seconds [:rfc:`6275#section-6.2.4`]. + interval: 'timedelta' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Option', length: 'int', interval: 'int') -> 'None': ... + def __init__(self, type: 'Enum_Option', length: 'int', interval: 'timedelta') -> 'None': ... @info_final diff --git a/pcapkit/protocols/internet/mh.py b/pcapkit/protocols/internet/mh.py index 7b6d588c8..dde59639f 100644 --- a/pcapkit/protocols/internet/mh.py +++ b/pcapkit/protocols/internet/mh.py @@ -1929,8 +1929,11 @@ def _read_msg_fbu(self, schema: 'Schema_FastBindingUpdateMessage', *, Note: :rfc:`5568#section-6.2.2` states that the FBU is *identical* to the - Mobile IPv6 binding update (BU) message, so the lifetime is read in - units of 4 seconds exactly as in :meth:`_read_msg_bu`. + Mobile IPv6 binding update (BU) message -- but that statement is + about **message layout**, not field units. The same section defines + the Lifetime as "the requested time in seconds", unlike the BU + lifetime's units of 4 seconds [:rfc:`6275#section-6.1.7`], so it is + not scaled on the way in. Args: schema: Parsed message type schema. @@ -1950,7 +1953,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']), - lifetime=datetime.timedelta(seconds=schema.lifetime * 4), + lifetime=datetime.timedelta(seconds=schema.lifetime), options=self._read_mh_options(schema.options), ) return data @@ -1984,6 +1987,11 @@ def _read_msg_fback(self, schema: 'Schema_FastBindingAcknowledgmentMessage', *, which is local to this module, rather than being mislabelled as a :class:`~pcapkit.const.mh.status_code.StatusCode`. + The Lifetime is likewise defined by :rfc:`5568#section-6.2.3` as + "the granted lifetime ... in seconds", unlike the BA lifetime's + units of 4 seconds [:rfc:`6275#section-6.1.7`], so it is not scaled + on the way in. + Args: schema: Parsed message type schema. header: Parsed MH header schema. @@ -2000,7 +2008,7 @@ def _read_msg_fback(self, schema: 'Schema_FastBindingAcknowledgmentMessage', *, status=FastBindingAcknowledgmentStatus(schema.status), key_mngt=bool(schema.flags['K']), seq=schema.seq, - lifetime=datetime.timedelta(seconds=schema.lifetime * 4), + lifetime=datetime.timedelta(seconds=schema.lifetime), options=self._read_mh_options(schema.options), ) return data @@ -2808,6 +2816,11 @@ def _read_opt_bra(self, schema: 'Schema_BindingRefreshAdviceOption', *, | Refresh Interval | +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ + Note: + The Refresh Interval is measured in units of four seconds + [:rfc:`6275#section-6.2.4`], the same unit the BU/BA lifetime uses, + so it is scaled here. + Args: schema: Parsed option schema. options: Parsed MH options. @@ -2822,7 +2835,7 @@ def _read_opt_bra(self, schema: 'Schema_BindingRefreshAdviceOption', *, data = Data_BindingRefreshAdviceOption( type=schema.type, length=schema.length + 2, - interval=schema.interval, + interval=datetime.timedelta(seconds=schema.interval * 4), ) return data @@ -6587,7 +6600,10 @@ def _make_msg_fbu(self, message: 'Optional[Data_FastBindingUpdateMessage]' = Non home: Home registration flag. lla_compat: LLA compatibility flag. key_mngt: Key management mobility option flag. - lifetime: Lifetime in seconds or timedelta. + lifetime: Lifetime, in seconds or as a + :class:`~datetime.timedelta`. Unlike the BU/BA lifetime, this + one counts seconds rather than units of 4 seconds + [:rfc:`5568#section-6.2.2`], so the value is not scaled. options: Mobility options. **kwargs: Arbitrary keyword arguments. @@ -6615,7 +6631,7 @@ def _make_msg_fbu(self, message: 'Optional[Data_FastBindingUpdateMessage]' = Non 'L': lla_compat, 'K': key_mngt, }, - lifetime=math.ceil(lifetime_val / 4), + lifetime=lifetime_val, options=self._make_mh_options(options), ) @@ -6640,7 +6656,10 @@ def _make_msg_fback(self, message: 'Optional[Data_FastBindingAcknowledgmentMessa status_reversed: Reverse status code namespace. key_mngt: Key management mobility option flag. seq: Sequence number. - lifetime: Lifetime in seconds or timedelta. + lifetime: Lifetime, in seconds or as a + :class:`~datetime.timedelta`. Unlike the BU/BA lifetime, this + one counts seconds rather than units of 4 seconds + [:rfc:`5568#section-6.2.3`], so the value is not scaled. options: Mobility options. **kwargs: Arbitrary keyword arguments. @@ -6666,7 +6685,7 @@ def _make_msg_fback(self, message: 'Optional[Data_FastBindingAcknowledgmentMessa 'K': key_mngt, }, seq=seq, - lifetime=math.ceil(lifetime_val / 4), + lifetime=lifetime_val, options=self._make_mh_options(options), ) @@ -7471,14 +7490,16 @@ def _make_opt_pad(self, type: 'Enum_Option', option: 'Optional[Data_PadOption]' ) def _make_opt_bra(self, type: 'Enum_Option', option: 'Optional[Data_BindingRefreshAdviceOption]' = None, *, - interval: 'int' = 0, + interval: 'int | timedelta' = 0, **kwargs: 'Any') -> 'Schema_BindingRefreshAdviceOption': """Make MH binding refresh advice option. Args: type: Option type. option: Option data model. - interval: Refresh interval. + interval: Refresh interval before re-registration, in units of 4 + seconds or as a :class:`~datetime.timedelta` + [:rfc:`6275#section-6.2.4`]. **kwargs: Arbitrary keyword arguments. Returns: @@ -7486,12 +7507,14 @@ def _make_opt_bra(self, type: 'Enum_Option', option: 'Optional[Data_BindingRefre """ if option is not None: - interval = option.interval + interval_val = math.ceil(option.interval.total_seconds() / 4) + else: + interval_val = self._seconds(interval, 4) return Schema_BindingRefreshAdviceOption( type=type, length=2, - interval=interval, + interval=interval_val, ) def _make_opt_aca(self, type: 'Enum_Option', option: 'Optional[Data_AlternateCareofAddressOption]' = None, *, diff --git a/pcapkit/protocols/schema/internet/mh.py b/pcapkit/protocols/schema/internet/mh.py index 08b7a6437..c08fff894 100644 --- a/pcapkit/protocols/schema/internet/mh.py +++ b/pcapkit/protocols/schema/internet/mh.py @@ -644,7 +644,7 @@ def __init__(self, type: 'Enum_Option', length: 'int') -> 'None': ... class BindingRefreshAdviceOption(Option, code=Enum_Option.Binding_Refresh_Advice): """Header schema for MH binding refresh advice options.""" - #: Refresh interval. + #: Refresh interval. One time unit is 4 seconds [:rfc:`6275#section-6.2.4`]. interval: 'int' = UInt16Field() if TYPE_CHECKING: @@ -2694,7 +2694,7 @@ class FastBindingUpdateMessage(Packet, code=Enum_Packet.Fast_Binding_Update): 'L': (2, 1), 'K': (3, 1), }) - #: Lifetime. One time unit is 4 seconds. + #: Lifetime, in seconds [:rfc:`5568#section-6.2.2`]. lifetime: 'int' = UInt16Field() #: Mobility options. options: 'list[Option]' = OptionField( @@ -2736,7 +2736,7 @@ class FastBindingAcknowledgmentMessage(Packet, code=Enum_Packet.Fast_Binding_Ack }) #: Sequence number. seq: 'int' = UInt16Field() - #: Lifetime. One time unit is 4 seconds. + #: Lifetime, in seconds [:rfc:`5568#section-6.2.3`]. lifetime: 'int' = UInt16Field() #: Mobility options. options: 'list[Option]' = OptionField( diff --git a/tests/protocols/internet/test_mh_unit.py b/tests/protocols/internet/test_mh_unit.py index 462c4a0ef..efc3bf0cb 100644 --- a/tests/protocols/internet/test_mh_unit.py +++ b/tests/protocols/internet/test_mh_unit.py @@ -322,11 +322,13 @@ def test_mh_option_readers_cover_known_options_and_guards(self) -> None: with self.assertRaises(ProtocolError): proto._read_opt_pad(bad, options=options) + # RFC 6275, section 6.2.4 -- the Refresh Interval is measured in units + # of four seconds, the same unit the BU/BA lifetime uses. self.assertEqual(proto._read_opt_bra( schema.BindingRefreshAdviceOption(type=Option.Binding_Refresh_Advice, length=2, interval=7), options=options, - ).interval, 7) + ).interval, datetime.timedelta(seconds=28)) self.assertEqual(str(proto._read_opt_aca( schema.AlternateCareofAddressOption(type=Option.Alternate_Care_of_Address, length=16, address='2001:db8::1'), @@ -480,6 +482,9 @@ def test_mh_option_constructors_cover_known_options_and_dispatch(self) -> None: self.assertEqual(warn.call_count, 2) self.assertEqual(proto._make_opt_bra(Option.Binding_Refresh_Advice, interval=9).interval, 9) + self.assertEqual(proto._make_opt_bra( + Option.Binding_Refresh_Advice, interval=datetime.timedelta(seconds=36), + ).interval, 9) self.assertEqual(str(proto._make_opt_aca(Option.Alternate_Care_of_Address, address='2001:db8::1').address), '2001:db8::1') self.assertEqual(proto._make_opt_ni(Option.Nonce_Indices, home=1, careof=2).careof, 2) @@ -544,7 +549,8 @@ def test_mh_option_constructors_cover_known_options_and_dispatch(self) -> None: option_dict = OrderedMultiDict([ (Option.Binding_Refresh_Advice, data.BindingRefreshAdviceOption(type=Option.Binding_Refresh_Advice, - length=4, interval=11)), + length=4, + interval=datetime.timedelta(seconds=44))), ]) self.assertEqual(proto._make_mh_options(option_dict)[0].interval, 11) @@ -896,8 +902,10 @@ def test_mh_fmipv6_message_readers_and_constructors(self) -> None: def header(type_: Packet) -> SimpleNamespace: return SimpleNamespace(next=TransType.UDP, length=5, type=type_, chksum=b'\x12\x34') - # RFC 5568, section 6.2.2 -- FBU is identical to the RFC 6275 BU, so the - # lifetime is carried in units of 4 seconds. + # 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 + # "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}, lifetime=10, options=[]), @@ -907,11 +915,13 @@ def header(type_: Packet) -> SimpleNamespace: self.assertTrue(fbu.home) self.assertFalse(fbu.lla_compat) self.assertTrue(fbu.key_mngt) - self.assertEqual(fbu.lifetime, datetime.timedelta(seconds=40)) + self.assertEqual(fbu.lifetime, datetime.timedelta(seconds=10)) self.assertEqual(fbu.options, 'opts') # RFC 5568, section 6.2.3 -- status 1 is "FBU accepted but NCoA is invalid", # which is *not* what StatusCode(1) means, hence the module-local enum. + # 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, lifetime=10, options=[]), header=header(Packet.Fast_Binding_Acknowledgment)) @@ -921,7 +931,7 @@ def header(type_: Packet) -> SimpleNamespace: self.assertNotIsInstance(fback.status, type(HandoverACKStatus.Administratively_prohibited)) self.assertTrue(fback.key_mngt) self.assertEqual(fback.seq, 0x1234) - self.assertEqual(fback.lifetime, datetime.timedelta(seconds=40)) + self.assertEqual(fback.lifetime, datetime.timedelta(seconds=10)) # RFC 4068, section 6.3.3 -- two reserved octets, then mobility options. fna = proto._read_msg_fna(SimpleNamespace(options=[]), @@ -965,7 +975,7 @@ def header(type_: Packet) -> SimpleNamespace: made_fbu = proto._make_msg_fbu(None, seq=7, ack=True, home=True, key_mngt=True, lifetime=datetime.timedelta(seconds=40), options=[]) self.assertEqual(made_fbu.seq, 7) - self.assertEqual(made_fbu.lifetime, 10) + self.assertEqual(made_fbu.lifetime, 40) self.assertEqual(made_fbu.flags, {'A': True, 'H': True, 'L': False, 'K': True}) self.assertEqual(proto._make_msg_fbu(SimpleNamespace( seq=8, ack=False, home=True, lla_compat=True, key_mngt=False, @@ -975,7 +985,7 @@ def header(type_: Packet) -> SimpleNamespace: made_fback = proto._make_msg_fback(None, status=131, key_mngt=True, seq=9, lifetime=40, options=[]) self.assertEqual(made_fback.status, 131) - self.assertEqual(made_fback.lifetime, 10) + self.assertEqual(made_fback.lifetime, 40) self.assertEqual(proto._make_msg_fback( None, status=FastBindingAcknowledgmentStatus.Insufficient_resources, options=[], ).status, 130) @@ -1015,6 +1025,79 @@ def header(type_: Packet) -> SimpleNamespace: code=HandoverACKStatus.Handover_Accepted_use_PCoA, options=[], )).seq, 14) + def test_mh_fbu_fback_lifetime_is_plain_seconds_unlike_bu_ba(self) -> None: + """FBU/FBack ``Lifetime`` is in seconds, not the BU/BA's 4-second units. + + :rfc:`5568#section-6.2.2` states that the FBU message is *identical* to + the Mobile IPv6 Binding Update (BU) -- but only in **layout**. The same + section, two paragraphs on, defines the FBU Lifetime as "the requested + time in seconds", and :rfc:`5568#section-6.2.3` likewise defines the + FBack Lifetime as "the granted lifetime ... in seconds". Neither + mentions a 4-second unit at all -- unlike :rfc:`6275#section-6.1.7`, + which spells it out for the BU/BA Lifetime it actually applies to + (c.f. #493). + + The BU/BA cases here are the control: they show the ``* 4`` / ``/ 4`` + scaling this module uses elsewhere is deliberately preserved, so the + FBU/FBack assertions below are evidence of the fix rather than of a + module-wide convention change. + """ + from pcapkit.const.mh.packet import Packet + from pcapkit.const.mh.status_code import StatusCode + from pcapkit.protocols.internet.mh import MH + + proto = object.__new__(MH) + proto._read_mh_options = mock.Mock(return_value='opts') + proto._make_mh_options = mock.Mock(return_value=['made']) + + def header(type_: Packet) -> SimpleNamespace: + return SimpleNamespace(next=None, length=2, type=type_, chksum=b'\x12\x34') + + accepted = StatusCode.Binding_Update_accepted_Proxy_Binding_Update_accepted + + # -- 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=[], + ), 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=[], + ), header=header(Packet.Fast_Binding_Acknowledgment)) + self.assertEqual(fback.lifetime, datetime.timedelta(seconds=100)) + + # 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=[], + ), 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=[], + ), header=header(Packet.Binding_Acknowledgement)) + self.assertEqual(ba.lifetime, datetime.timedelta(seconds=400)) + + # -- make side: caller lifetime=100 seconds -- + + made_fbu = proto._make_msg_fbu(None, seq=1, lifetime=datetime.timedelta(seconds=100), + options=[]) + self.assertEqual(made_fbu.lifetime, 100) + + made_fback = proto._make_msg_fback(None, status=0, seq=1, + lifetime=datetime.timedelta(seconds=100), + options=[]) + self.assertEqual(made_fback.lifetime, 100) + + # control: BU/BA still divide by 4 on make. + made_bu = proto._make_msg_bu(None, seq=1, lifetime=datetime.timedelta(seconds=400), + options=[]) + self.assertEqual(made_bu.lifetime, 100) + + made_ba = proto._make_msg_ba(None, status=accepted, seq=1, + lifetime=datetime.timedelta(seconds=400), options=[]) + self.assertEqual(made_ba.lifetime, 100) + def test_mh_fmipv6_option_readers_constructors_and_guards(self) -> None: from pcapkit.const.mh.option import Option from pcapkit.corekit.multidict import OrderedMultiDict @@ -1154,7 +1237,8 @@ def parse(hexstr: str) -> object: self.assertTrue(fbu.home) self.assertFalse(fbu.lla_compat) self.assertTrue(fbu.key_mngt) - self.assertEqual(fbu.lifetime, datetime.timedelta(seconds=40)) + # lifetime=0x000a=10 is read as 10 seconds, not units of 4 (c.f. #493) + self.assertEqual(fbu.lifetime, datetime.timedelta(seconds=10)) self.assertEqual(str(fbu.options[acoa_type].address), '2001:db8:1::2') self.assertEqual(fbu.options[badf_type].spi, 0xdeadbeef) self.assertEqual(fbu.options[badf_type].data, bytes(range(12))) @@ -1174,7 +1258,8 @@ def parse(hexstr: str) -> object: ) self.assertTrue(fback.key_mngt) self.assertEqual(fback.seq, 0x1234) - self.assertEqual(fback.lifetime, datetime.timedelta(seconds=40)) + # lifetime=0x000a=10 is read as 10 seconds, not units of 4 (c.f. #493) + self.assertEqual(fback.lifetime, datetime.timedelta(seconds=10)) self.assertEqual(fback.options[badf_type].spi, 0) with self.subTest('FNA, RFC 4068 section 6.3.3'):