diff --git a/pcapkit/protocols/data/internet/mh.py b/pcapkit/protocols/data/internet/mh.py index 82f10e2512..8cdd45009d 100644 --- a/pcapkit/protocols/data/internet/mh.py +++ b/pcapkit/protocols/data/internet/mh.py @@ -10,7 +10,7 @@ if TYPE_CHECKING: from datetime import datetime as dt_type from datetime import timedelta - from ipaddress import IPv4Address, IPv6Address, IPv6Network + from ipaddress import IPv4Address, IPv6Address from typing import Optional from pcapkit.const.mh.access_type import AccessType as Enum_AccessType @@ -932,15 +932,28 @@ class MobileNetworkPrefixOption(Option): New required field ``reserved`` (:issue:`1437`). + .. versionchanged:: 1.6.0 + + New required field ``prefix_length``, and :attr:`prefix` is the + :class:`~ipaddress.IPv6Address` captured rather than an + :class:`~ipaddress.IPv6Network`, which cannot hold a prefix with bits set + past its length or a length past 128 (:issue:`2702`). + """ #: The reserved octet, as captured. reserved: 'bytes' - #: Mobile Network Prefix. - prefix: 'IPv6Network' + #: Prefix length, as captured. + prefix_length: 'int' + #: Mobile Network Prefix. Kept apart from :attr:`prefix_length` rather than + #: fused into an :class:`~ipaddress.IPv6Network`, so that bits past + #: :attr:`prefix_length`, which :rfc:`3963#section-6.2` reads as given, survive a + #: round trip. Same reasoning as :class:`HomeNetworkPrefixOption`. + prefix: 'IPv6Address' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Option', length: 'int', reserved: 'bytes', prefix: 'IPv6Network') -> 'None': ... + def __init__(self, type: 'Enum_Option', length: 'int', reserved: 'bytes', prefix_length: 'int', + prefix: 'IPv6Address') -> 'None': ... @info_final @@ -1310,17 +1323,26 @@ def __init__(self, seconds: 'int', fraction: 'int') -> 'None': ... @info_final class TimestampOption(Option): - """Data model for MH Timestamp option.""" + """Data model for MH Timestamp option. - #: Timestamp. - timestamp: 'dt_type' + .. versionchanged:: 1.6.0 + + :attr:`timestamp` is :data:`None` for a time past the year 9999, which + :class:`~datetime.datetime` cannot hold, rather than the read raising + :exc:`ValueError` (:issue:`2702`). + + """ + + #: Timestamp, or :data:`None` past the year 9999, which the 48-bit second + #: count reaches and :class:`~datetime.datetime` cannot hold. + timestamp: 'Optional[dt_type]' #: Timestamp in the fixed-point format of :rfc:`5213#section-8.8`, kept #: alongside :attr:`timestamp` so that the 1/65536-second resolution of the #: wire format survives a round trip. pmip_timestamp: 'PMIPv6Timestamp' if TYPE_CHECKING: - def __init__(self, type: 'Enum_Option', length: 'int', timestamp: 'dt_type', + def __init__(self, type: 'Enum_Option', length: 'int', timestamp: 'Optional[dt_type]', pmip_timestamp: 'PMIPv6Timestamp') -> 'None': ... diff --git a/pcapkit/protocols/internet/mh.py b/pcapkit/protocols/internet/mh.py index fe7a5f7f0a..d439bba93f 100644 --- a/pcapkit/protocols/internet/mh.py +++ b/pcapkit/protocols/internet/mh.py @@ -3350,14 +3350,16 @@ def _read_opt_mnp(self, schema: 'Schema_MobileNetworkPrefixOption', *, if schema.length != 18: raise ProtocolError(f'{self.alias}: [Opt {schema.type}] invalid format') - prefix = cast('IPv6Network', - ipaddress.ip_network((schema.prefix, schema.prefix_length))) - + # NOTE: The prefix and its length stay apart, as captured: an IPv6Network + # holds neither bits set past the length nor a length past 128. RFC 3963 + # section 6.2 reads both fields as given, and answers a prefix the home + # agent cannot use with a status rather than discarding the option (#2702). data = Data_MobileNetworkPrefixOption( type=schema.type, length=self._mh_option_length(schema.length), reserved=schema.reserved, - prefix=prefix, + prefix_length=schema.prefix_length, + prefix=schema.prefix, ) return data @@ -4020,9 +4022,10 @@ def _read_opt_hnp(self, schema: 'Schema_HomeNetworkPrefixOption', *, """ if schema.length != 18: raise ProtocolError(f'{self.alias}: [Opt {schema.type}] invalid format') - if schema.prefix_length > 128: - raise ProtocolError(f'{self.alias}: [Opt {schema.type}] invalid format') + # NOTE: A prefix length past 128 is kept, as for option 6: RFC 5213 + # section 5.3.2 answers a prefix the local mobility anchor cannot + # authorise with a status rather than discarding the option (#2702). data = Data_HomeNetworkPrefixOption( type=schema.type, length=self._mh_option_length(schema.length), @@ -4237,12 +4240,23 @@ def _read_opt_timestamp(self, schema: 'Schema_TimestampOption', *, seconds = schema.timestamp['seconds'] fraction = schema.timestamp['fraction'] + try: + # NOTE: Not the stdlib's ``fromtimestamp``, which takes no time + # after 3000 on Windows (#2099). The whole seconds go in as an int + # and the fraction as a timedelta, which rounds it to microseconds + # half to even as one float did, but loses no precision to the + # seconds, so the last second of 9999 is not carried past it. + timestamp = (fromtimestamp(seconds, datetime.timezone.utc) + + datetime.timedelta(seconds=fraction / 65536)) # type: Optional[dt_type] + except ValueError: + # past the year 9999, which the 48-bit count reaches and datetime + # cannot hold, so pmip_timestamp alone keeps it (#2702) + timestamp = None + data = Data_TimestampOption( type=schema.type, length=self._mh_option_length(schema.length), - # not the stdlib's ``fromtimestamp``, which takes no time after 3000 - # on Windows (#2099) - timestamp=fromtimestamp(seconds + fraction / 65536, datetime.timezone.utc), + timestamp=timestamp, pmip_timestamp=Data_PMIPv6Timestamp(seconds=seconds, fraction=fraction), ) return data @@ -8258,7 +8272,8 @@ def _make_opt_bad(self, type: 'Enum_Option', option: 'Optional[Data_Authorizatio ) def _make_opt_mnp(self, type: 'Enum_Option', option: 'Optional[Data_MobileNetworkPrefixOption]' = None, *, - prefix: 'bytes | str | IPv6Network' = '::/0', + prefix: 'bytes | str | int | IPv6Address | IPv6Network' = '::/0', + prefix_length: 'Optional[int]' = None, reserved: 'bytes' = b'\x00', **kwargs: 'Any') -> 'Schema_MobileNetworkPrefixOption': """Make MH mobile network prefix option. @@ -8266,7 +8281,10 @@ def _make_opt_mnp(self, type: 'Enum_Option', option: 'Optional[Data_MobileNetwor Args: type: Option type. option: Option data model. - prefix: Mobile network prefix. + prefix: Mobile network prefix, an IPv6 address or network, whose + bits past ``prefix_length`` are written as given. + prefix_length: Prefix length; defaults to that of a network or of + text ending in ``/``, and to 128 for an address. reserved: The reserved octet as captured, truncated or zero-filled to 1 octet. **kwargs: Arbitrary keyword arguments. @@ -8274,16 +8292,42 @@ def _make_opt_mnp(self, type: 'Enum_Option', option: 'Optional[Data_MobileNetwor Returns: Constructed option schema. + Raises: + ProtocolError: If ``prefix`` is not an IPv6 address or network, or + ``prefix_length`` is negative, a :obj:`bool`, or does not fit in + one octet. + + Note: + The prefix length is checked against its 8-bit field alone, as + :rfc:`3963#section-6.2` has a home agent answer a prefix it cannot + use with a status rather than discard the option, so whatever the + reader keeps is written back. + + .. versionchanged:: 1.6.0 + + New argument ``prefix_length``. ``prefix`` keeps the bits set past + its length, rather than raising :exc:`ValueError`, and a prefix + length past 128 is written (:issue:`2702`). + """ if option is not None: reserved = getattr(option, 'reserved', b'\x00') - prefix = option.prefix - - prefix_val = ipaddress.ip_network(prefix) - if prefix_val.version != 6: - raise ProtocolError(f'{self.alias}: [OptNo {type}] invalid mobile network prefix: {prefix!r}') - prefix_length = prefix_val.prefixlen - prefix_addr = prefix_val.network_address + prefix_length = option.prefix_length + prefix_addr = option.prefix + else: + try: + # NOTE: An interface, unlike a network, keeps the bits past its + # length, and takes an address, a network or text alike. A bool + # is refused first, as it would pass for the address ``::1``. + prefix_val = None if isinstance(prefix, bool) else ipaddress.IPv6Interface(prefix) + except ValueError: + prefix_val = None + if prefix_val is None: + raise ProtocolError(f'{self.alias}: [OptNo {type}] invalid mobile network prefix: {prefix!r}') + if prefix_length is None: + prefix_length = prefix_val.network.prefixlen + prefix_addr = prefix_val.ip + prefix_length = self._make_uint(prefix_length, bits=8, name='prefix length') return Schema_MobileNetworkPrefixOption( type=type, @@ -9088,7 +9132,18 @@ def _make_opt_hnp(self, type: 'Enum_Option', option: 'Optional[Data_HomeNetworkP Raises: ProtocolError: If ``prefix_length`` is negative, a :obj:`bool`, - or longer than the address. + or does not fit in one octet. + + Note: + The prefix length is checked against its 8-bit field alone, as + :rfc:`5213#section-5.3.2` has a local mobility anchor answer a + prefix it cannot authorise with a status rather than discard the + option, so whatever the reader keeps is written back. + + .. versionchanged:: 1.6.0 + + A prefix length longer than the address is written, rather than + refused, as the reader keeps one (:issue:`2702`). """ if option is not None: @@ -9097,8 +9152,6 @@ def _make_opt_hnp(self, type: 'Enum_Option', option: 'Optional[Data_HomeNetworkP prefix = option.prefix prefix_length = self._make_uint(prefix_length, bits=8, name='prefix length') - if prefix_length > 128: - raise ProtocolError(f'{self.alias}: [OptNo {type}] invalid prefix length: {prefix_length}') return Schema_HomeNetworkPrefixOption( type=type, diff --git a/tests/protocols/internet/test_internet_maker_ranges_2614_unit.py b/tests/protocols/internet/test_internet_maker_ranges_2614_unit.py index ed3e135f7f..d22374131a 100644 --- a/tests/protocols/internet/test_internet_maker_ranges_2614_unit.py +++ b/tests/protocols/internet/test_internet_maker_ranges_2614_unit.py @@ -447,6 +447,7 @@ class MHMakerTests(_Case): 'bra': ('Binding_Refresh_Advice', {}, {'interval': (0xFFFF, 'refresh interval')}), 'ni': ('Nonce_Indices', {}, {'home': (0xFFFF, 'home nonce index'), 'careof': (0xFFFF, 'care-of nonce index')}), + 'mnp': ('Mobile_Network_Prefix_Option', {}, {'prefix_length': (0xFF, 'prefix length')}), 'auth': ('AUTH_OPTION_TYPE', {'data': b'\x00\x00'}, {'spi': (0xFFFFFFFF, 'SPI')}), 'badf': ('Binding_Authorization_Data_for_FMIPv6', {'data': b'\x00'}, {'spi': (0xFFFFFFFF, 'SPI')}), @@ -454,7 +455,7 @@ class MHMakerTests(_Case): 'dns': ('DNS_UPDATE_TYPE', {}, {'reserved': (0x7F, 'reserved bits')}), 'vendor': ('Vendor_Specific_Mobility_Option', {}, {'vendor': (0xFFFFFFFF, 'vendor ID'), 'subtype': (0xFF, 'vendor subtype')}), - 'hnp': ('Home_Network_Prefix_Option', {}, {'prefix_length': (128, 'prefix length')}), + 'hnp': ('Home_Network_Prefix_Option', {}, {'prefix_length': (0xFF, 'prefix length')}), 'hi': ('Handoff_Indicator_Option', {}, {'reserved': (0xFF, 'reserved octet')}), 'timestamp': ('Timestamp_Option', {}, {'seconds': ((1 << 48) - 1, 'timestamp seconds'), 'fraction': (0xFFFF, 'timestamp fraction')}), @@ -562,11 +563,6 @@ class MHMakerTests(_Case): 'maximum retransmissions'), ) - UNPARSED_MAXIMA = { - '_make_opt_timestamp.seconds': 'MH._read_opt_timestamp turns the seconds into a ' - 'datetime, which cannot hold the 48-bit maximum', - } - #: Integer arguments outside the sweep, and why. EXEMPT = { # The CGA extension makers are reached only through a CGA parameter diff --git a/tests/protocols/internet/test_mh_option_mutation_2684_unit.py b/tests/protocols/internet/test_mh_option_mutation_2684_unit.py index 340ba7f7d5..d28676b41b 100644 --- a/tests/protocols/internet/test_mh_option_mutation_2684_unit.py +++ b/tests/protocols/internet/test_mh_option_mutation_2684_unit.py @@ -31,13 +31,9 @@ LONG_KEY = b'\x30\x81\x9f' + bytes(range(159)) #: Options a one-octet change makes the reader raise an error that is not -#: pcapkit's, by the reader: its data model has no form for the octets, a prefix -#: with host bits set or a time past datetime.MAXYEAR. Inside IPv6 the header is -#: then Raw, so the round trip holds. Found by this sweep, not fixed here. -KNOWN_CRASHES = { - 'Mobile_Network_Prefix_Option': '_read_opt_mnp', - 'Timestamp_Option': '_read_opt_timestamp', -} +#: pcapkit's, by the reader whose data model has no form for the octets. Inside +#: IPv6 the header is then Raw, so the round trip holds. +KNOWN_CRASHES = {} # type: dict[str, str] def _ipv6(body: bytes) -> bytes: diff --git a/tests/protocols/internet/test_mh_unit.py b/tests/protocols/internet/test_mh_unit.py index 9c9a2b5e47..7822e7bc45 100644 --- a/tests/protocols/internet/test_mh_unit.py +++ b/tests/protocols/internet/test_mh_unit.py @@ -2,7 +2,7 @@ import datetime import io -from ipaddress import ip_address, ip_network +from ipaddress import ip_address import importlib.util from types import SimpleNamespace import unittest @@ -353,12 +353,13 @@ def test_mh_option_readers_cover_known_options_and_guards(self) -> None: schema.AuthorizationDataOption(type=Option.Authorization_Data, length=8, data=b'12345678'), options=options, ).data, b'12345678') - self.assertEqual(str(proto._read_opt_mnp( + mnp = proto._read_opt_mnp( schema.MobileNetworkPrefixOption(type=Option.Mobile_Network_Prefix_Option, length=18, prefix_length=64, prefix='2001:db8::'), options=options, - ).prefix), '2001:db8::/64') + ) + self.assertEqual((mnp.prefix_length, str(mnp.prefix)), (64, '2001:db8::')) self.assertEqual(proto._read_opt_lla( schema.LinkLayerAddressOption(type=Option.Mobility_Header_Link_Layer_Address_option, length=7, code=LLACode.MN, lla=b'abcdef'), @@ -789,8 +790,8 @@ def make_extension(code: CGAExtension, extension: data.UnknownExtension | None = self.assertEqual(proto._make_opt_mnp( Option.Mobile_Network_Prefix_Option, data.MobileNetworkPrefixOption(type=Option.Mobile_Network_Prefix_Option, - length=18, reserved=b'\x00', - prefix=ip_network('2001:db8:1::/64')), + length=18, reserved=b'\x00', prefix_length=64, + prefix=ip_address('2001:db8:1::')), ).prefix_length, 64) self.assertEqual(proto._make_opt_lla( Option.Mobility_Header_Link_Layer_Address_option, diff --git a/tests/protocols/internet/test_mh_unrepresentable_values_2702_unit.py b/tests/protocols/internet/test_mh_unrepresentable_values_2702_unit.py new file mode 100644 index 0000000000..2a5a850984 --- /dev/null +++ b/tests/protocols/internet/test_mh_unrepresentable_values_2702_unit.py @@ -0,0 +1,192 @@ +# -*- coding: utf-8 -*- +"""Mobility options 6 and 27 parse values an IPv6Network and a datetime cannot hold. + +GitHub issue #2702. The Mobile Network Prefix option fused its prefix and length +into an :class:`~ipaddress.IPv6Network`, which raised a bare :exc:`ValueError` on +bits set past the length or a length past 128, and the Timestamp option turned a +48-bit second count past the year 9999 into a :class:`~datetime.datetime`, which +raised one too. Neither RFC has the receiver discard such a value, so both parse: +the prefix and its length are kept as captured, and the timestamp is +:data:`None` past 9999, with ``pmip_timestamp`` holding it. The Home Network +Prefix option, option 6's twin, refused a length past 128 on both paths, and +keeps one in the same way. + +Each case is parsed alone and inside IPv6, rebuilt from its info and from +``info.to_dict()``, and built with ``make()`` from keywords, against the octets +:rfc:`3963#section-4.3`, :rfc:`5213#section-8.3` and :rfc:`5213#section-8.8` lay +out. + +Every case builds its own octets in memory and reads no capture. + +""" + +import datetime +import ipaddress +import struct +import unittest +import warnings + +from tests._support import reimport_once_per_class +from tests.utilities._harness import capture + +#: Test tier (#1958): hand-picked round-trip cases, not a sweep. +__tier__ = 'ALL' + +UTC = datetime.timezone.utc +#: The last whole second since the UNIX epoch that :class:`~datetime.datetime` +#: holds, 31 December 9999 23:59:59 UTC. +LAST_SECOND = 253_402_300_799 + + +def _mnp(prefix_length: int, prefix: str, code: int = 6) -> bytes: + """A Mobile Network Prefix option, :rfc:`3963#section-4.3`, or ``code``'s. + + Type 6, length 18, a reserved octet, the prefix length, then the 16-octet + prefix. The Home Network Prefix option, :rfc:`5213#section-8.3`, is the same + with type 22. + """ + return bytes([code, 18, 0, prefix_length]) + ipaddress.IPv6Address(prefix).packed + + +def _timestamp(seconds: int, fraction: int) -> bytes: + """A Timestamp option, :rfc:`5213#section-8.8`. + + Type 27, length 8, the seconds in 48 bits and the 1/65536ths in 16. + """ + return bytes([27, 8]) + seconds.to_bytes(6, 'big') + fraction.to_bytes(2, 'big') + + +def _brr(option: bytes) -> bytes: + """A Binding Refresh Request holding ``option``, then a PadN to a multiple of eight.""" + pad = -len(option) % 8 # 4 for options 6 and 22 and 6 for option 27, never a Pad1 + body = option + bytes([1, pad - 2]) + bytes(pad - 2) + return bytes([59, (8 + len(body)) // 8 - 1, 0, 0, 0, 0, 0, 0]) + body + + +def _ipv6(body: bytes) -> bytes: + return (bytes([0x60, 0, 0, 0]) + struct.pack('!HBB', len(body), 135, 64) + + bytes(15) + b'\x01' + bytes(15) + b'\x02' + body) + + +class TestMHUnrepresentableValues(unittest.TestCase): + """Pin options 6 and 27 parsing such values quietly and rebuilding them byte for byte.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + catcher = warnings.catch_warnings() + catcher.__enter__() + self.addCleanup(catcher.__exit__, None, None, None) + warnings.simplefilter('ignore') + + def _option(self, raw: bytes, code: 'object') -> 'object': + """Parse ``raw`` alone and inside IPv6, check both rebuild, and return the option.""" + from pcapkit.protocols.internet.ipv6 import IPv6 + from pcapkit.protocols.internet.mh import MH + from pcapkit.utilities import exceptions + from pcapkit.utilities.logging import logger + + mh = MH(raw, len(raw)) + self.assertEqual(MH.from_data(mh.info).data, raw) + self.assertEqual(MH.from_data(mh.info.to_dict()).data, raw) + + packet = _ipv6(raw) + with capture(logger) as package, capture(exceptions.logger) as module: + ipv6 = IPv6(packet, len(packet)) + self.assertEqual(package.messages + module.messages, []) + self.assertEqual(IPv6.from_data(ipv6.info).data, packet) + self.assertEqual(IPv6.from_data(ipv6.info.to_dict()).data, packet) + (inner,) = [ext for ext in ipv6.extension_headers.values() if isinstance(ext, MH)] + self.assertEqual(inner.info.options[code], mh.info.options[code]) + return mh.info.options[code] + + @staticmethod + def _make(code: 'object', kwargs: 'dict') -> bytes: + from pcapkit.const.mh.packet import Packet + from pcapkit.protocols.internet.mh import MH + + return MH(type=Packet.Binding_Refresh_Request, next=59, chksum=b'\x00\x00', + data={'options': [(code, kwargs)]}).data + + def test_mobile_network_prefix_keeps_its_octets(self) -> None: + from pcapkit.const.mh.option import Option + + code = Option.Mobile_Network_Prefix_Option + # (prefix length, prefix): bits set past the length, at the edges of the + # 8-bit field and the 128-bit address, which RFC 3963 section 6.2 reads + # as given + for prefix_length, prefix in ((64, '2001:db8::1'), (0, 'ffff::'), (127, '2001:db8::1'), + (128, '2001:db8::1'), (129, '2001:db8::'), (255, '2001:db8::1')): + raw = _brr(_mnp(prefix_length, prefix)) + with self.subTest(prefix_length=prefix_length, prefix=prefix): + option = self._option(raw, code) + self.assertEqual(option.reserved, b'\x00') + self.assertEqual(option.prefix_length, prefix_length) + self.assertEqual(option.prefix, ipaddress.IPv6Address(prefix)) + self.assertEqual(self._make(code, {'prefix': prefix, 'prefix_length': prefix_length}), raw) + + def test_mobile_network_prefix_make_takes_a_network(self) -> None: + from pcapkit.const.mh.option import Option + from pcapkit.utilities.exceptions import ProtocolError + + code = Option.Mobile_Network_Prefix_Option + # text ending in / gives the length and keeps the bits past it + for kwargs, prefix_length, prefix in ( + ({'prefix': '2001:db8::1/64'}, 64, '2001:db8::1'), + ({'prefix': ipaddress.IPv6Network('2001:db8::/48')}, 48, '2001:db8::'), + ({'prefix': ipaddress.IPv6Address('2001:db8::1')}, 128, '2001:db8::1'), + ({'prefix': '2001:db8::1/64', 'prefix_length': 200}, 200, '2001:db8::1'), + ): + with self.subTest(kwargs=kwargs): + self.assertEqual(self._make(code, kwargs), _brr(_mnp(prefix_length, prefix))) + for kwargs, message in (({'prefix': '192.0.2.0/24'}, 'invalid mobile network prefix'), + ({'prefix': True}, 'invalid mobile network prefix'), + ({'prefix': '::/129'}, 'invalid mobile network prefix'), + ({'prefix': '::', 'prefix_length': 256}, 'invalid prefix length'), + ({'prefix': '::', 'prefix_length': True}, 'invalid prefix length')): + with self.subTest(kwargs=kwargs): + with self.assertRaisesRegex(ProtocolError, message): + self._make(code, kwargs) + + def test_home_network_prefix_keeps_a_length_past_128(self) -> None: + from pcapkit.const.mh.option import Option + from pcapkit.utilities.exceptions import ProtocolError + + code = Option.Home_Network_Prefix_Option + # RFC 5213 section 5.3.2 answers a prefix it cannot authorise with a + # status, so a length past the address is kept up to the 8-bit field + for prefix_length, prefix in ((128, '2001:db8::1'), (129, '2001:db8::'), (255, '2001:db8::1')): + raw = _brr(_mnp(prefix_length, prefix, code=22)) + with self.subTest(prefix_length=prefix_length, prefix=prefix): + option = self._option(raw, code) + self.assertEqual(option.reserved, b'\x00') + self.assertEqual(option.prefix_length, prefix_length) + self.assertEqual(option.prefix, ipaddress.IPv6Address(prefix)) + self.assertEqual(self._make(code, {'prefix': prefix, 'prefix_length': prefix_length}), raw) + with self.assertRaisesRegex(ProtocolError, 'invalid prefix length'): + self._make(code, {'prefix': '::', 'prefix_length': 256}) + + def test_timestamp_past_9999_is_kept(self) -> None: + from pcapkit.const.mh.option import Option + + code = Option.Timestamp_Option + # (seconds, fraction, the datetime RFC 5213 section 8.8 names, or None + # past what datetime holds); a 1/65536th is 15.2587890625 microseconds + for seconds, fraction, expected in ( + (0, 0, datetime.datetime(1970, 1, 1, tzinfo=UTC)), + (LAST_SECOND, 0, datetime.datetime(9999, 12, 31, 23, 59, 59, tzinfo=UTC)), + (LAST_SECOND, 1, datetime.datetime(9999, 12, 31, 23, 59, 59, 15, tzinfo=UTC)), + (LAST_SECOND, 0xFFFF, datetime.datetime(9999, 12, 31, 23, 59, 59, 999985, tzinfo=UTC)), + (LAST_SECOND + 1, 0, None), + ((1 << 48) - 1, 0xFFFF, None), + ): + raw = _brr(_timestamp(seconds, fraction)) + with self.subTest(seconds=seconds, fraction=fraction): + option = self._option(raw, code) + self.assertEqual(option.timestamp, expected) + self.assertEqual((option.pmip_timestamp.seconds, option.pmip_timestamp.fraction), + (seconds, fraction)) + self.assertEqual(self._make(code, {'seconds': seconds, 'fraction': fraction}), raw) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/misc/test_fromtimestamp_windows_range_2099_unit.py b/tests/protocols/misc/test_fromtimestamp_windows_range_2099_unit.py index 309eef2164..de3c6bf09c 100644 --- a/tests/protocols/misc/test_fromtimestamp_windows_range_2099_unit.py +++ b/tests/protocols/misc/test_fromtimestamp_windows_range_2099_unit.py @@ -250,10 +250,16 @@ def test_timestamp_option_after_3000(self) -> None: rebuilt = proto._make_opt_timestamp(Option.Timestamp_Option, data) self.assertEqual(rebuilt.pack(), schema.pack()) - # past year 9999 it raises what it raises on Linux + # past year 9999 it reads as it reads on Linux: no datetime, and the + # count kept in pmip_timestamp, which rebuilds it (#2702) schema = proto._make_opt_timestamp(Option.Timestamp_Option, seconds=2**48 - 1, fraction=0) - with windows_gmtime(), self.assertRaises(ValueError): - proto._read_opt_timestamp(schema, options=None) # type: ignore[arg-type] + linux = proto._read_opt_timestamp(schema, options=None) # type: ignore[arg-type] + with windows_gmtime(): + data = proto._read_opt_timestamp(schema, options=None) # type: ignore[arg-type] + self.assertEqual(data.to_dict(), linux.to_dict()) + self.assertIsNone(data.timestamp) + self.assertEqual((data.pmip_timestamp.seconds, data.pmip_timestamp.fraction), (2**48 - 1, 0)) + self.assertEqual(proto._make_opt_timestamp(Option.Timestamp_Option, data).pack(), schema.pack()) if __name__ == '__main__':