Skip to content

Commit 384a4f6

Browse files
committed
fix(pcapng): default a made EPB/PB/ISB timestamp without an AttributeError (#2362)
Problem: _make_timestamp picked a defaulted timestamp's clock by reading self.nanosecond, which resolves its interface through the parsed data model. A block being made has no such model yet, so make() without an explicit timestamp raised AttributeError: '_info' for every block that carries one -- the Enhanced Packet Block, the obsolete Packet Block, and the Interface Statistics Block, through its own Timestamp field as well as its isb_starttime/isb_endtime options. Fix: read the resolution from _get_resolution(interface_id) instead -- the lookup the method already made for the scale; the offset keeps its own _get_offset(interface_id) -- no new fallback, matching every other make-path lookup in this file. Test: tests/protocols/misc/test_pcapng_epb_pb_default_timestamp_2362_unit.py covers all three block kinds defaulting their timestamp to the current time, scaling to the interface's declared resolution and offset, reading the correct interface rather than interface 0 in a multi-interface section, reading the nanosecond clock rather than the float one when resolution calls for it, and round-tripping. Each case raises AttributeError on the parent commit.
1 parent 7886d9f commit 384a4f6

2 files changed

Lines changed: 284 additions & 2 deletions

File tree

‎pcapkit/protocols/misc/pcapng.py‎

Lines changed: 18 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2067,10 +2067,26 @@ def _make_timestamp(self, timestamp: 'Optional[float | Decimal | dt_type | int]'
20672067
builds is defined as an offset from the UNIX epoch, and the read
20682068
side applies no timezone shift for it to mirror.
20692069
2070+
Which clock a defaulted ``timestamp`` is read to follows the
2071+
resolution of the interface ``interface_id`` names, not
2072+
:attr:`self.nanosecond <nanosecond>`. That property resolves its
2073+
interface through the parsed data model, which a block being made
2074+
has not got, so reading it here would make a ``make`` that names
2075+
no ``timestamp`` -- of an Enhanced Packet Block, an obsolete
2076+
Packet Block or an Interface Statistics Block -- raise
2077+
:exc:`AttributeError` on ``_info`` (:issue:`2362`).
2078+
2079+
.. versionchanged:: 1.5.0
2080+
2081+
A defaulted ``timestamp`` no longer reads
2082+
:attr:`self.nanosecond <nanosecond>`, so it no longer requires a
2083+
parsed data model (:issue:`2362`).
2084+
20702085
"""
20712086
with localcontext(prec=64):
2087+
resolution = self._get_resolution(interface_id)
20722088
if timestamp is None:
2073-
if self.nanosecond:
2089+
if resolution > RESOLUTION[False]:
20742090
timestamp = decimal.Decimal(time.time_ns()) / 1_000_000_000
20752091
else:
20762092
timestamp = decimal.Decimal(time.time())
@@ -2080,7 +2096,7 @@ def _make_timestamp(self, timestamp: 'Optional[float | Decimal | dt_type | int]'
20802096
timestamp = decimal.Decimal(timestamp)
20812097

20822098
ts_info = int(
2083-
(timestamp - self._get_offset(interface_id)) * self._get_resolution(interface_id)
2099+
(timestamp - self._get_offset(interface_id)) * resolution
20842100
)
20852101
return (ts_info >> 32) & 0xFFFF_FFFF, ts_info & 0xFFFF_FFFF
20862102

Lines changed: 266 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,266 @@
1+
# -*- coding: utf-8 -*-
2+
"""GitHub issue #2362: a made PCAP-NG block defaults its timestamp to the clock.
3+
4+
Without the fix, :meth:`PCAPNG._make_timestamp
5+
<pcapkit.protocols.misc.pcapng.PCAPNG._make_timestamp>` picks the clock a
6+
defaulted timestamp is read to by reading :attr:`PCAPNG.nanosecond
7+
<pcapkit.protocols.misc.pcapng.PCAPNG.nanosecond>`, which resolves its
8+
interface through the parsed data model a block being made has not got, so
9+
``make()`` without a ``timestamp`` raises
10+
``AttributeError: 'PCAPNG' object has no attribute '_info'`` for every block
11+
carrying one -- the Enhanced Packet Block, the obsolete Packet Block and the
12+
Interface Statistics Block, the last through its ``isb_starttime`` and
13+
``isb_endtime`` options as well as its own field. The resolution comes from the
14+
interface ID the maker is given, so a defaulted timestamp is the current time,
15+
read from the clock that interface's declared resolution calls for and scaled
16+
to its units.
17+
18+
Every case builds its own blocks in memory and reads no capture.
19+
20+
:mod:`pcapkit` is imported inside each test, after
21+
:func:`~tests._support.reimport_once_per_class`.
22+
23+
"""
24+
25+
import time
26+
import unittest
27+
import unittest.mock
28+
import warnings
29+
from typing import TYPE_CHECKING
30+
31+
from tests.protocols.misc.test_pcapng_roundtrip2_unit import PCAPNGTestCase
32+
33+
if TYPE_CHECKING:
34+
from typing import Any, Optional
35+
36+
#: Blocks whose maker defaults a timestamp, named rather than imported so that
37+
#: this module loads without :mod:`pcapkit`.
38+
TIMESTAMPED = ('Enhanced_Packet_Block', 'Packet_Block', 'Interface_Statistics_Block')
39+
40+
#: ``if_tsresol``'s own default, in units per second: a block whose interface
41+
#: declares none counts microseconds.
42+
DEFAULT_RESOLUTION = 1_000_000
43+
44+
#: ``if_tsresol`` of 9, in units per second: the resolution a block has to
45+
#: declare for its default to be read from :func:`time.time_ns`.
46+
NANOSECOND_RESOLUTION = 1_000_000_000
47+
48+
#: A :func:`time.time_ns` reading to pin, far enough from both the current time
49+
#: and :data:`PINNED_SECOND` that the field it reaches names the clock it came
50+
#: from.
51+
PINNED_NANOSECOND = 1_700_000_000_123_456_789
52+
53+
#: A :func:`time.time` reading to pin, exactly representable so that scaling it
54+
#: to :data:`DEFAULT_RESOLUTION` is exact.
55+
PINNED_SECOND = 1_600_000_000.5
56+
57+
#: The raw timestamp field a defaulted timestamp is written to at each
58+
#: resolution, with both clocks pinned and no ``if_tsoffset`` in play. The
59+
#: nanosecond entry is :data:`PINNED_NANOSECOND` written in its own units; the
60+
#: microsecond one is :data:`PINNED_SECOND` scaled, which is a value the
61+
#: nanosecond clock cannot produce and vice versa.
62+
PINNED_RAW = {
63+
NANOSECOND_RESOLUTION: PINNED_NANOSECOND,
64+
DEFAULT_RESOLUTION: int(PINNED_SECOND * DEFAULT_RESOLUTION),
65+
}
66+
67+
68+
class TestPCAPNGDefaultTimestamp(PCAPNGTestCase):
69+
"""Pin the timestamp a block is made with when the caller names none."""
70+
71+
def _timed_context(self, resolution: 'Optional[int]' = None,
72+
offset: 'Optional[int]' = None) -> 'Any':
73+
"""A one-interface context, optionally with ``if_tsresol``/``if_tsoffset``.
74+
75+
Its own, rather than :meth:`PCAPNGTestCase._context
76+
<tests.protocols.misc.test_pcapng_roundtrip2_unit.PCAPNGTestCase._context>`,
77+
because the interface needs options the base helper does not take.
78+
79+
"""
80+
from pcapkit.const.pcapng.block_type import BlockType
81+
from pcapkit.foundation.engines.pcapng import Context
82+
from pcapkit.protocols.misc.pcapng import PCAPNG
83+
84+
section = PCAPNG(num=0, sct=1, ctx=None, type=BlockType.Section_Header_Block, block={})
85+
context = Context(section.info) # type: Any
86+
self._add_interface(context, resolution=resolution, offset=offset)
87+
return context
88+
89+
def _add_interface(self, context: 'Any', resolution: 'Optional[int]' = None,
90+
offset: 'Optional[int]' = None) -> 'Any':
91+
"""Append one Ethernet interface to ``context``, and return its data model.
92+
93+
Its index in the section is the ``interface_id`` a block names to be
94+
read at this interface's ``if_tsresol`` and ``if_tsoffset``.
95+
96+
"""
97+
from pcapkit.const.pcapng.block_type import BlockType
98+
from pcapkit.const.pcapng.option_type import OptionType
99+
from pcapkit.const.reg.linktype import LinkType
100+
from pcapkit.protocols.misc.pcapng import PCAPNG
101+
102+
options = [] # type: list[Any]
103+
if resolution is not None:
104+
options.append((OptionType.if_tsresol, {'resolution': resolution}))
105+
if offset is not None:
106+
options.append((OptionType.if_tsoffset, {'offset': offset}))
107+
108+
interface = PCAPNG(num=len(context.interfaces) + 1, sct=1, ctx=context,
109+
type=BlockType.Interface_Description_Block,
110+
block={'linktype': LinkType.ETHERNET, 'snaplen': 0,
111+
'options': options})
112+
context.interfaces.append(interface.info)
113+
return interface.info
114+
115+
def assertIsNow(self, epoch: 'Any', before: 'float', after: 'float', # pylint: disable=invalid-name
116+
resolution: 'int' = DEFAULT_RESOLUTION) -> 'None':
117+
"""Assert ``epoch`` is a clock read taken between ``before`` and ``after``.
118+
119+
The field holds an integer count of ``resolution`` units, so the epoch
120+
it reads back as can sit up to one unit below ``before``, which is a
121+
full-precision :func:`time.time`.
122+
123+
"""
124+
self.assertGreaterEqual(float(epoch), before - 1 / resolution)
125+
self.assertLessEqual(float(epoch), after)
126+
127+
def _raw_timestamp(self, made: 'Any') -> 'int':
128+
"""The raw Timestamp (High)/(Low) pair of ``made``, in its own units."""
129+
block = made.schema.block
130+
return (block.timestamp_high << 32) | block.timestamp_low
131+
132+
def _make(self, name: 'str', context: 'Any', **block: 'Any') -> 'Any':
133+
"""Make the block named ``name`` over ``context``, with no ``timestamp``."""
134+
from pcapkit.const.pcapng.block_type import BlockType
135+
from pcapkit.protocols.misc.pcapng import PCAPNG
136+
137+
with warnings.catch_warnings():
138+
warnings.simplefilter('ignore') # the Packet Block is obsolete
139+
return PCAPNG(num=2, sct=1, ctx=context, type=BlockType.get(name), block=block)
140+
141+
def test_default_is_the_current_time(self) -> None:
142+
"""The issue's case: no ``timestamp`` makes one, and it is now."""
143+
context = self._timed_context()
144+
for name in TIMESTAMPED:
145+
with self.subTest(block=name):
146+
before = time.time()
147+
made = self._make(name, context, packet_data=b'A' * 4)
148+
after = time.time()
149+
self.assertIsNow(made.info.timestamp_epoch, before, after)
150+
151+
def test_default_survives_a_round_trip(self) -> None:
152+
"""A defaulted timestamp parses back to the value the maker reported."""
153+
from pcapkit.protocols.misc.pcapng import PCAPNG
154+
155+
context = self._timed_context()
156+
for name in TIMESTAMPED:
157+
with self.subTest(block=name):
158+
made = self._make(name, context, packet_data=b'A' * 4)
159+
with warnings.catch_warnings():
160+
warnings.simplefilter('ignore') # the Packet Block is obsolete
161+
parsed = PCAPNG(made.data, len(made.data), num=2, sct=1, ctx=context)
162+
self.assertEqual(parsed.info.timestamp_epoch, made.info.timestamp_epoch)
163+
self.assertRebuilds(made.data, context)
164+
165+
def test_default_is_read_at_the_interface_resolution(self) -> None:
166+
"""``if_tsresol`` of 9 scales the default to nanoseconds, not microseconds."""
167+
context = self._timed_context(resolution=NANOSECOND_RESOLUTION)
168+
for name in TIMESTAMPED:
169+
with self.subTest(block=name):
170+
before = time.time_ns()
171+
made = self._make(name, context, packet_data=b'A' * 4)
172+
after = time.time_ns()
173+
# The raw field counts nanoseconds, so it is of the order of
174+
# ``time.time_ns()``; read at microseconds it would be a
175+
# thousandth of that and fall below ``before``.
176+
raw = self._raw_timestamp(made)
177+
self.assertGreaterEqual(raw, before)
178+
self.assertLessEqual(raw, after)
179+
180+
def _make_on_pinned_clocks(self, name: 'str', context: 'Any', **block: 'Any') -> 'Any':
181+
"""Make the block named ``name`` with both clocks pinned.
182+
183+
:func:`time.time_ns` reads :data:`PINNED_NANOSECOND` and
184+
:func:`time.time` reads :data:`PINNED_SECOND`, so the field a default
185+
reaches names the clock it was read from. Left to the real clocks the
186+
two are each about the current time, and the field cannot tell them
187+
apart.
188+
189+
"""
190+
with unittest.mock.patch('time.time_ns', return_value=PINNED_NANOSECOND), \
191+
unittest.mock.patch('time.time', return_value=PINNED_SECOND):
192+
return self._make(name, context, **block)
193+
194+
def test_default_follows_the_interface_the_block_names(self) -> None:
195+
"""A block reads its own ``interface_id``'s resolution, not interface 0's.
196+
197+
The section carries two interfaces a thousandfold apart and every block
198+
names the second one, in both orders, with the clocks pinned: a lookup
199+
fixed at interface 0 therefore reads the wrong clock and scales by the
200+
wrong factor, whichever way round the section is written.
201+
202+
"""
203+
for first, second in ((None, NANOSECOND_RESOLUTION),
204+
(NANOSECOND_RESOLUTION, None)):
205+
resolution = second or DEFAULT_RESOLUTION
206+
context = self._timed_context(resolution=first)
207+
self._add_interface(context, resolution=second)
208+
for name in TIMESTAMPED:
209+
with self.subTest(block=name, resolution=resolution):
210+
made = self._make_on_pinned_clocks(name, context, interface_id=1,
211+
packet_data=b'A' * 4)
212+
self.assertEqual(made.info.interface_id, 1)
213+
self.assertEqual(self._raw_timestamp(made), PINNED_RAW[resolution])
214+
215+
def test_default_reads_the_clock_its_resolution_calls_for(self) -> None:
216+
"""A sub-microsecond resolution reads :func:`time.time_ns`, a coarser one :func:`time.time`.
217+
218+
``None`` is an interface that declares no ``if_tsresol`` at all, which
219+
counts :data:`DEFAULT_RESOLUTION` units and so reads the coarse clock.
220+
221+
"""
222+
for declared in (NANOSECOND_RESOLUTION, None):
223+
resolution = declared or DEFAULT_RESOLUTION
224+
context = self._timed_context(resolution=declared)
225+
for name in TIMESTAMPED:
226+
with self.subTest(block=name, resolution=resolution):
227+
made = self._make_on_pinned_clocks(name, context, packet_data=b'A' * 4)
228+
self.assertEqual(self._raw_timestamp(made), PINNED_RAW[resolution])
229+
230+
def test_default_is_shifted_by_the_interface_offset(self) -> None:
231+
"""``if_tsoffset`` shifts the written field, not the epoch reported."""
232+
context = self._timed_context(offset=1_000)
233+
for name in TIMESTAMPED:
234+
with self.subTest(block=name):
235+
before = time.time()
236+
made = self._make(name, context, packet_data=b'A' * 4)
237+
after = time.time()
238+
self.assertIsNow(made.info.timestamp_epoch, before, after)
239+
raw = self._raw_timestamp(made)
240+
self.assertAlmostEqual(raw / 1_000_000 + 1_000, before, delta=1)
241+
242+
def test_isb_start_and_end_times_default_too(self) -> None:
243+
"""``isb_starttime``/``isb_endtime`` reach the same maker, so they default too."""
244+
from pcapkit.const.pcapng.option_type import OptionType
245+
246+
context = self._timed_context()
247+
before = time.time()
248+
made = self._make('Interface_Statistics_Block', context,
249+
options=[(OptionType.isb_starttime, {}),
250+
(OptionType.isb_endtime, {})])
251+
after = time.time()
252+
for option in (OptionType.isb_starttime, OptionType.isb_endtime):
253+
with self.subTest(option=option):
254+
self.assertIsNow(made.info.options[option].timestamp_epoch, before, after)
255+
256+
def test_an_explicit_timestamp_is_unchanged(self) -> None:
257+
"""The fix touches only the default: an explicit ``timestamp`` still writes 0."""
258+
context = self._timed_context()
259+
for name in TIMESTAMPED:
260+
with self.subTest(block=name):
261+
made = self._make(name, context, packet_data=b'A' * 4, timestamp=0)
262+
self.assertEqual(made.info.timestamp_epoch, 0)
263+
264+
265+
if __name__ == '__main__':
266+
unittest.main()

0 commit comments

Comments
 (0)