From c21b655fe155b8e2d7e9342a5bb7b33bcc91d29e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=90=D0=BD=D1=82=D0=BE=D0=BD=20=D0=A2=D1=80=D0=BE=D1=88?= =?UTF-8?q?=D0=B8=D0=BD?= Date: Sat, 15 Aug 2026 02:02:44 +0300 Subject: [PATCH] Fix pcapng timestamp resolution + imported-session duration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - pcap.py: _parse_tsresol now pads options to a 32-bit boundary ((-length) % 4) and uses 2** for the MSB-set (power-of-2) case, so nanosecond if_tsresol is read correctly (was defaulting to µs and scaling timestamps 1000x) - database/server: set imported session started/ended from the first/last packet timestamps (falling back to wall-clock for synthetic SPB) - frontend: formatDuration renders hours - tests: tsresol (ns after padded option, default µs, ns round-trip) and session-window/fallback (119 total) --- frontend/index.html | 4 +- simtrace2_pysniff/pcap.py | 4 +- simtrace2_pysniff/server/database.py | 11 ++++++ simtrace2_pysniff/server/server.py | 10 ++++- tests/test_pcap_import.py | 57 ++++++++++++++++++++++++++++ 5 files changed, 82 insertions(+), 4 deletions(-) diff --git a/frontend/index.html b/frontend/index.html index 0825005..132bd00 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -1031,7 +1031,9 @@ function formatDuration(start, end) { const s = Math.floor(ms / 1000); if (s < 60) return s + 's'; const m = Math.floor(s / 60); - return m + 'm ' + (s % 60) + 's'; + if (m < 60) return m + 'm ' + (s % 60) + 's'; + const h = Math.floor(m / 60); + return h + 'h ' + (m % 60) + 'm'; } function escapeHtml(str) { diff --git a/simtrace2_pysniff/pcap.py b/simtrace2_pysniff/pcap.py index 8578555..25ee06a 100644 --- a/simtrace2_pysniff/pcap.py +++ b/simtrace2_pysniff/pcap.py @@ -378,9 +378,9 @@ def _parse_tsresol(options, endian): if code == 9 and length == 1: b = options[i] if b & 0x80: - return 10 ** (-(b & 0x7F)) + return 2 ** (-(b & 0x7F)) return 10 ** (-b) - i += length + (length & 1) # options are padded to 32-bit + i += length + ((-length) % 4) # options are padded to a 32-bit boundary return 1e-6 # default: microseconds diff --git a/simtrace2_pysniff/server/database.py b/simtrace2_pysniff/server/database.py index 03d3ebd..1a6d6b8 100644 --- a/simtrace2_pysniff/server/database.py +++ b/simtrace2_pysniff/server/database.py @@ -37,6 +37,11 @@ def _iso_now(): f'{time.time() % 1 * 1000:03.0f}' +def _iso_from_ts(ts): + return time.strftime('%Y-%m-%dT%H:%M:%S.', time.localtime(ts)) + \ + f'{ts % 1 * 1000:03.0f}' + + class Database: def __init__(self, db_path=None): if db_path is None: @@ -64,6 +69,12 @@ class Database: (_iso_now(), session_id)) self._conn.commit() + def set_session_times_from_ts(self, session_id, first_ts, last_ts): + self._conn.execute( + 'UPDATE sessions SET started=?, ended=? WHERE id=?', + (_iso_from_ts(first_ts), _iso_from_ts(last_ts), session_id)) + self._conn.commit() + def rename_session(self, session_id, name): self._conn.execute( 'UPDATE sessions SET name=? WHERE id=?', diff --git a/simtrace2_pysniff/server/server.py b/simtrace2_pysniff/server/server.py index e8f04db..4a52ab7 100644 --- a/simtrace2_pysniff/server/server.py +++ b/simtrace2_pysniff/server/server.py @@ -322,12 +322,20 @@ class RequestHandler(BaseHTTPRequestHandler): self.db.rename_session(session_id, name) first_ts = packets[0][0] + last_ts = packets[-1][0] rows = [] for ts, msg_type, payload in packets: elapsed = round(ts - first_ts, 6) if ts is not None else 0.0 rows.append((max(0.0, elapsed), msg_type, payload, 0)) self.db.insert_messages(session_id, rows) - self.db.close_session(session_id) + # Use the packet timestamps as the session window when they are real + # (classic pcap / pcapng EPB). pcapng SPB-only files have synthetic + # 0,1,2,… timestamps, so fall back to the import wall-clock time. + _REAL_TS_MIN = 946684800 # 2000-01-01 + if first_ts is not None and first_ts > _REAL_TS_MIN: + self.db.set_session_times_from_ts(session_id, first_ts, last_ts) + else: + self.db.close_session(session_id) self._send_json({ 'session_id': session_id, diff --git a/tests/test_pcap_import.py b/tests/test_pcap_import.py index 6546e65..67b6ffa 100644 --- a/tests/test_pcap_import.py +++ b/tests/test_pcap_import.py @@ -87,6 +87,37 @@ class TestParsePcapng(unittest.TestCase): self.assertEqual(list(parse_pcapng(shb + idb + epb)), []) +class TestTsresol(unittest.TestCase): + def test_nanoseconds_after_padded_option(self): + from simtrace2_pysniff.pcap import _parse_tsresol + # if_name (code 2, len 2, "lo") then if_tsresol (code 9, len 1, 9), LE. + opts = (struct.pack('IHHq', 0x1A2B3C4D, 1, 0, -1)) + idb = _pcapng_block(0x00000001, + struct.pack('>HHI', LINKTYPE_ETHERNET, 0, 65535) + + struct.pack('>HH', 9, 1) + b'\x09' + b'\x00\x00\x00' + + struct.pack('>HH', 0, 0)) + # epoch 1600000000.5 s → 1600000000500000000 ns + raw = 1600000000500000000 + ts_hi, ts_lo = raw >> 32, raw & 0xFFFFFFFF + frame = wrap_gsmtap(_gsmtap_hdr(GSMTAP_SIM_ATR) + b'\x3b\x04') + epb = _pcapng_block(0x00000006, + struct.pack('>IIIII', 0, ts_hi, ts_lo, len(frame), len(frame)) + frame) + parsed = list(parse_pcapng(shb + idb + epb)) + self.assertEqual(len(parsed), 1) + self.assertAlmostEqual(parsed[0][0], 1600000000.5, places=6) + + class TestImportEndpoint(unittest.TestCase): def _handler(self, body): from simtrace2_pysniff.server.server import RequestHandler @@ -136,6 +167,32 @@ class TestImportEndpoint(unittest.TestCase): self.assertIn('No GSMTAP data', h._json['error']) self.assertEqual(h.db.count_sessions(), 0) + def test_import_session_window_from_timestamps(self): + from datetime import datetime + # Two packets one hour apart in epoch time. + t0 = 1600000000.0 + packets = [ + (_gsmtap_hdr(GSMTAP_SIM_ATR), b'\x3b\x00', t0), + (_gsmtap_hdr(GSMTAP_SIM_APDU), b'\x80\xf2\x00\x00\x00', t0 + 3600.0), + ] + h, _ = self._handler(build_pcap(packets)) + h._handle_import_pcap({'name': ['window']}) + session = h.db.get_session(h._json['session_id']) + started = datetime.fromisoformat(session['started']) + ended = datetime.fromisoformat(session['ended']) + self.assertEqual((ended - started).total_seconds(), 3600.0) + + def test_import_synthetic_ts_falls_back(self): + # SPB-style synthetic timestamps (0, 1, …) → keep wall-clock fallback. + packets = [ + (_gsmtap_hdr(GSMTAP_SIM_ATR), b'\x3b\x00', 0.0), + (_gsmtap_hdr(GSMTAP_SIM_APDU), b'\x80\xf2\x00\x00\x00', 1.0), + ] + h, _ = self._handler(build_pcap(packets)) + h._handle_import_pcap({'name': ['synth']}) + session = h.db.get_session(h._json['session_id']) + self.assertTrue(session['ended'] is not None) + class TestVersion(unittest.TestCase): def test_package_version(self):