Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 8 additions & 5 deletions pcapkit/protocols/data/internet/mh.py
Original file line number Diff line number Diff line change
Expand Up @@ -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]'
Expand All @@ -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]'
Expand Down Expand Up @@ -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
Expand Down
49 changes: 36 additions & 13 deletions pcapkit/protocols/internet/mh.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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
Expand Down Expand Up @@ -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.
Expand All @@ -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
Expand Down Expand Up @@ -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.
Expand All @@ -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

Expand Down Expand Up @@ -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.

Expand Down Expand Up @@ -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),
)

Expand All @@ -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.

Expand All @@ -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),
)

Expand Down Expand Up @@ -7471,27 +7490,31 @@ 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:
Constructed option schema.

"""
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, *,
Expand Down
6 changes: 3 additions & 3 deletions pcapkit/protocols/schema/internet/mh.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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(
Expand Down
105 changes: 95 additions & 10 deletions tests/protocols/internet/test_mh_unit.py
Original file line number Diff line number Diff line change
Expand Up @@ -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'),
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)

Expand Down Expand Up @@ -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=[]),
Expand All @@ -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))
Expand All @@ -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=[]),
Expand Down Expand Up @@ -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,
Expand All @@ -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)
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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)))
Expand All @@ -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'):
Expand Down
Loading