From fba8a2b5957c190b47c48fe9bd57d5256ca0805c Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Thu, 8 Oct 2026 14:41:05 -0400 Subject: [PATCH] fix(pcapng): keep out-of-range if_tzone as captured and catch every fromtimestamp failure (#1340) - An if_tzone of 24 hours or more, which no datetime.timezone can hold, is kept as captured as a timedelta with a ProtocolWarning instead of raising ValueError; the interface's timestamps render in UTC, and the option rebuilds byte-exactly. - _read_timestamp now falls back to the epoch with its existing warning on OverflowError and OSError as well as ValueError, so an extreme if_tsoffset no longer escapes from datetime.fromtimestamp. New tests/protocols/misc/test_pcapng_tzone_tsoffset_range_1340_unit.py. --- pcapkit/protocols/data/misc/pcapng.py | 10 +- pcapkit/protocols/misc/pcapng.py | 20 +- ...t_pcapng_tzone_tsoffset_range_1340_unit.py | 175 ++++++++++++++++++ 3 files changed, 199 insertions(+), 6 deletions(-) create mode 100644 tests/protocols/misc/test_pcapng_tzone_tsoffset_range_1340_unit.py diff --git a/pcapkit/protocols/data/misc/pcapng.py b/pcapkit/protocols/data/misc/pcapng.py index b9c20e280b..01f6323d7e 100644 --- a/pcapkit/protocols/data/misc/pcapng.py +++ b/pcapkit/protocols/data/misc/pcapng.py @@ -40,6 +40,7 @@ if TYPE_CHECKING: from datetime import datetime as dt_type + from datetime import timedelta as dt_timedelta from datetime import timezone as dt_timezone from decimal import Decimal from ipaddress import IPv4Address, IPv4Interface, IPv6Address, IPv6Interface @@ -240,11 +241,14 @@ def __init__(self, type: 'Enum_OptionType', length: 'int', resolution: 'int') -> class IF_TZoneOption(Option): """Data model for PCAP-NG file ``if_tzone`` options.""" - #: Time zone. - timezone: 'dt_timezone' + #: Time zone. An offset of 24 hours or more, which no + #: :class:`~datetime.timezone` can hold, is kept as captured as a + #: :class:`~datetime.timedelta`. + timezone: 'dt_timezone | dt_timedelta' if TYPE_CHECKING: - def __init__(self, type: 'Enum_OptionType', length: 'int', timezone: 'dt_timezone') -> None: ... + def __init__(self, type: 'Enum_OptionType', length: 'int', + timezone: 'dt_timezone | dt_timedelta') -> None: ... @info_final diff --git a/pcapkit/protocols/misc/pcapng.py b/pcapkit/protocols/misc/pcapng.py index 15a7a8a021..f6fca7bfb2 100644 --- a/pcapkit/protocols/misc/pcapng.py +++ b/pcapkit/protocols/misc/pcapng.py @@ -1682,7 +1682,9 @@ def _get_timezone(self, interface_id: 'int' = 0) -> 'timezone': options = self._get_interface(interface_id).options tzone = cast('Optional[Data_IF_TZoneOption]', options.get(Enum_OptionType.if_tzone)) - if tzone is None: + # an ``if_tzone`` kept as captured (a :class:`~datetime.timedelta` of 24 + # hours or more) names no zone a datetime can be rendered in + if tzone is None or not isinstance(tzone.timezone, datetime.timezone): return datetime.timezone.utc return tzone.timezone @@ -1742,7 +1744,9 @@ def _read_timestamp(self, timestamp_high: 'int', timestamp_low: 'int', *, # *rendered* in -- which is all a ``tzinfo`` can affect on an aware # datetime. It never moves the instant itself. ts_datetime = datetime.datetime.fromtimestamp(ts_ratio[0] / ts_ratio[1], tzone) - except ValueError: + # an epoch past what ``datetime`` or the platform's ``time_t`` holds, + # e.g. from an extreme ``if_tsoffset``, raises any of these three + except (ValueError, OverflowError, OSError): warn(f'PCAP-NG: [Block {self._type}] invalid timestamp: {timestamp_epoch}', ProtocolWarning, stacklevel=stacklevel()) ts_datetime = datetime.datetime.fromtimestamp(0, datetime.timezone.utc) @@ -3057,10 +3061,20 @@ def _read_option_if_tzone(self, schema: 'Schema_IF_TZoneOption', *, if schema.length != 4: raise ProtocolError(f'PCAP-NG: [if_tzone] invalid length (expected 4, got {schema.length})') + offset = datetime.timedelta(seconds=schema.tzone) + try: + timezone = datetime.timezone(offset) # type: datetime.timezone | datetime.timedelta + except ValueError: + # a :class:`~datetime.timezone` holds strictly less than 24 hours; + # the signed 32-bit field holds far more, so keep it as captured + warn(f'PCAP-NG: [if_tzone] offset of {schema.tzone} second(s) is not within 24 hours ' + f'of UTC; kept as captured', ProtocolWarning, stacklevel=stacklevel()) + timezone = offset + option = Data_IF_TZoneOption( type=schema.type, length=schema.length, - timezone=datetime.timezone(datetime.timedelta(seconds=schema.tzone)), + timezone=timezone, ) return option diff --git a/tests/protocols/misc/test_pcapng_tzone_tsoffset_range_1340_unit.py b/tests/protocols/misc/test_pcapng_tzone_tsoffset_range_1340_unit.py new file mode 100644 index 0000000000..684ffe5ebc --- /dev/null +++ b/tests/protocols/misc/test_pcapng_tzone_tsoffset_range_1340_unit.py @@ -0,0 +1,175 @@ +# -*- coding: utf-8 -*- +"""GitHub issue #1340: out-of-range ``if_tzone`` and ``if_tsoffset`` parse and rebuild. + +``if_tzone`` is a signed 32-bit count of seconds, but a +:class:`~datetime.timezone` holds strictly less than 24 hours, so an offset of +a day or more raised a bare :exc:`ValueError` out of the IDB reader. It is now +kept as captured, as a :class:`~datetime.timedelta`, with a +:class:`~pcapkit.utilities.warnings.ProtocolWarning`, and the block's +timestamps render in UTC. + +``if_tsoffset`` is a signed 64-bit count of seconds; an epoch past what +``datetime`` or the platform ``time_t`` holds raised :exc:`OverflowError` or +:exc:`OSError` from :meth:`datetime.datetime.fromtimestamp`, where only +:exc:`ValueError` fell back to the epoch with a warning. All three do now. + +Every block rebuilds byte-exactly from its data model. + +:mod:`pcapkit` is imported inside each test, after +:func:`~tests._support.reimport_once_per_class`. + +""" + +import datetime +import decimal +import struct +import unittest +import warnings +from typing import TYPE_CHECKING + +from tests.protocols.misc.test_pcapng_roundtrip2_unit import PCAPNGTestCase + +if TYPE_CHECKING: + from typing import Any + +#: ``datetime`` holds years 1 to 9999, so these are its first and last whole +#: seconds since the UNIX epoch. +DT_MIN = -62135596800 +DT_MAX = 253402300799 + + +def _idb(code: 'int', fmt: 'str', value: 'int') -> 'bytes': + """An Ethernet IDB carrying one fixed-size option and ``opt_endofopt``.""" + option = struct.pack(' 'bytes': + return _idb(10, ' 'bytes': + return _idb(14, ' 'bytes': + """A 36-octet Enhanced Packet Block on interface 0.""" + body = struct.pack(' 'tuple[Any, Any, list[str]]': + """Parse ``octets`` as interface 0 of a fresh context.""" + from pcapkit.protocols.misc.pcapng import PCAPNG + + context = self._context() + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + parsed = PCAPNG(octets, len(octets), num=1, sct=1, ctx=context) + context.interfaces[0] = parsed.info + return context, parsed, [str(item.message) for item in caught] + + def _packet(self, context: 'Any', octets: 'bytes') -> 'tuple[Any, list[str]]': + from pcapkit.protocols.misc.pcapng import PCAPNG + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + parsed = PCAPNG(octets, len(octets), num=2, sct=1, ctx=context) + # the four-octet payload is too short for Ethernet, which warns of its own + return parsed, [str(item.message) for item in caught if str(item.message).startswith('PCAP-NG')] + + def _assert_rebuilds(self, octets: 'bytes', context: 'Any', num: 'int') -> None: + from pcapkit.protocols.misc.pcapng import PCAPNG + + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + parsed = PCAPNG(octets, len(octets), num=num, sct=1, ctx=context) + rebuilt = PCAPNG.from_data(parsed.info, num=num, sct=1, ctx=context).data + self.assertEqual(rebuilt.hex(), octets.hex()) + + def test_tzone_out_of_range_is_kept(self) -> None: + from pcapkit.const.pcapng.option_type import OptionType + + for value in (-2**31, -86401, -86400, 86400, 86401, 2**31 - 1): + with self.subTest(tzone=value): + octets = _tzone(value) + context, parsed, messages = self._interface(octets) + kept = parsed.info.options[OptionType.if_tzone].timezone + self.assertEqual(kept, datetime.timedelta(seconds=value)) + self.assertNotIsInstance(kept, datetime.timezone) + self.assertIn(f'PCAP-NG: [if_tzone] offset of {value} second(s) is not within 24 hours ' + f'of UTC; kept as captured', messages) + self._assert_rebuilds(octets, self._context(), 1) + + # the block's packets render in UTC and keep the same instant + packet, messages = self._packet(context, _epb()) + self.assertEqual(packet.info.timestamp, + datetime.datetime(1970, 1, 1, 0, 0, 0, 1, datetime.timezone.utc)) + self.assertIs(packet.info.timestamp.tzinfo, datetime.timezone.utc) + self.assertEqual(packet.info.timestamp_epoch, decimal.Decimal('0.000001')) + self.assertEqual(messages, []) + self._assert_rebuilds(_epb(), context, 2) + + def test_tzone_in_range_is_a_timezone(self) -> None: + from pcapkit.const.pcapng.option_type import OptionType + + for value in (-86399, -3600, 0, 3600, 86399): + with self.subTest(tzone=value): + octets = _tzone(value) + context, parsed, messages = self._interface(octets) + zone = datetime.timezone(datetime.timedelta(seconds=value)) + self.assertEqual(parsed.info.options[OptionType.if_tzone].timezone, zone) + self.assertEqual(messages, []) + self._assert_rebuilds(octets, self._context(), 1) + packet, _ = self._packet(context, _epb()) + self.assertEqual(packet.info.timestamp.utcoffset(), datetime.timedelta(seconds=value)) + + def test_tsoffset_out_of_range_falls_back(self) -> None: + for value in (-2**63, -2**62, DT_MIN - 1, DT_MAX + 1, 2**62, 2**63 - 1): + with self.subTest(tsoffset=value): + octets = _tsoffset(value) + context, _, messages = self._interface(octets) + self.assertEqual(messages, []) + self._assert_rebuilds(octets, self._context(), 1) + + packet, messages = self._packet(context, _epb()) + epoch = decimal.Decimal(value) + decimal.Decimal('0.000001') + self.assertEqual(packet.info.timestamp_epoch, epoch) + self.assertEqual(packet.info.timestamp, datetime.datetime.fromtimestamp(0, datetime.timezone.utc)) + self.assertEqual(messages, [f'PCAP-NG: [Block 6] invalid timestamp: {epoch}']) + self._assert_rebuilds(_epb(), context, 2) + + def test_tsoffset_in_range_is_unchanged(self) -> None: + for value in (DT_MIN + 86400, -5, 0, 5, DT_MAX - 86400): + with self.subTest(tsoffset=value): + context, _, _ = self._interface(_tsoffset(value)) + packet, messages = self._packet(context, _epb(0, 0)) + self.assertEqual(messages, []) + self.assertEqual(packet.info.timestamp_epoch, decimal.Decimal(value)) + self.assertEqual(packet.info.timestamp, + datetime.datetime.fromtimestamp(0, datetime.timezone.utc) + + datetime.timedelta(seconds=value)) + self._assert_rebuilds(_epb(0, 0), context, 2) + + def test_largest_timestamp_with_extreme_tsoffset(self) -> None: + """The largest EPB timestamp, 2**64 - 1 microseconds, on top of either extreme offset.""" + for value in (-2**63, -1, 0, 2**63 - 1): + with self.subTest(tsoffset=value): + context, _, _ = self._interface(_tsoffset(value)) + octets = _epb(0xFFFFFFFF, 0xFFFFFFFF) + packet, messages = self._packet(context, octets) + epoch = decimal.Decimal(2**64 - 1) / 1_000_000 + value + self.assertEqual(packet.info.timestamp_epoch, epoch) + self.assertEqual(packet.info.timestamp, datetime.datetime.fromtimestamp(0, datetime.timezone.utc)) + self.assertEqual(messages, [f'PCAP-NG: [Block 6] invalid timestamp: {epoch}']) + self._assert_rebuilds(octets, context, 2) + + +if __name__ == '__main__': + unittest.main()