catarrh
76c5647dd4
- focusMessage()/scrollToType() highlighted the target and removed the class after 1500 ms, so a Files-row (or map/category) jump looked selected for a moment and then lost it; on top of that any re-render (the 500 ms live poll, a filter/search change) rebuilt the list and dropped the highlight — even for a clicked entry - the selection is now persistent app state (STATE.selectedIdx): the render marks the entry (gap rows included), focusMessage records the index before rendering, showDetail records it for plain clicks, scrollToType persists it too (clearing the previous highlight), and a new session/capture resets it so a stale index cannot mark an unrelated APDU - tests: a stub-driven focusMessage case (state + detail updated, no render and no timer), the filter-clearing case, and source pins for renderMessages, focusMessage, scrollToType, showDetail, loadSession and startCapture; 527 Python + 58 Node tests - README/README_RUS: the jump leaves the record selected; PWA_VERSION/ pyproject/version.py 2.6.2, SW cache simtrace-analyser-v76
313 lines
12 KiB
Python
313 lines
12 KiB
Python
"""Tests for PCAP/PCAPNG import parsing and the import endpoint."""
|
|
|
|
import os
|
|
import struct
|
|
import tempfile
|
|
import unittest
|
|
|
|
from simtrace2_pysniff.pcap import (
|
|
build_pcap, parse_pcap, parse_pcapng, wrap_gsmtap, extract_gsmtap,
|
|
LINKTYPE_ETHERNET,
|
|
)
|
|
from simtrace2_pysniff.gsmtap import (build_gsmtap_packet, GSMTAP_SIM_ATR,
|
|
GSMTAP_SIM_APDU, GSMTAP_SIM_RST_EVENT,
|
|
GSMTAP_SIM_VCC_EVENT)
|
|
|
|
|
|
def _gsmtap_hdr(sub_type):
|
|
return build_gsmtap_packet(sub_type, b'')[:16]
|
|
|
|
|
|
def _pcapng_block(block_type, body):
|
|
body = body + b'\x00' * ((-len(body)) % 4)
|
|
total = 8 + len(body) + 4
|
|
return struct.pack('>II', block_type, total) + body + struct.pack('>I', total)
|
|
|
|
|
|
def _build_pcapng(packets):
|
|
"""Build a minimal big-endian PCAPNG with one Ethernet interface.
|
|
|
|
*packets* is a list of (gsmtap_hdr, data, ts_hi, ts_lo) tuples.
|
|
"""
|
|
shb = _pcapng_block(0x0A0D0D0A, struct.pack('>IHHq', 0x1A2B3C4D, 1, 0, -1))
|
|
idb = _pcapng_block(0x00000001,
|
|
struct.pack('>HHI', LINKTYPE_ETHERNET, 0, 65535) +
|
|
struct.pack('>HH', 0, 0))
|
|
out = shb + idb
|
|
for gsmtap_hdr, data, ts_hi, ts_lo in packets:
|
|
frame = wrap_gsmtap(gsmtap_hdr + data)
|
|
epb = _pcapng_block(
|
|
0x00000006,
|
|
struct.pack('>IIIII', 0, ts_hi, ts_lo, len(frame), len(frame)) + frame)
|
|
out += epb
|
|
return out
|
|
|
|
|
|
class TestParsePcap(unittest.TestCase):
|
|
def test_roundtrip(self):
|
|
packets = [
|
|
(_gsmtap_hdr(GSMTAP_SIM_ATR), b'\x3b\x00', 1.0),
|
|
(_gsmtap_hdr(GSMTAP_SIM_APDU), b'\x80\xf2\x00\x00\x00', 2.5),
|
|
]
|
|
data = build_pcap(packets)
|
|
parsed = [(round(ts, 1), t, payload) for ts, t, payload, _f in parse_pcap(data)]
|
|
self.assertEqual(parsed, [
|
|
(1.0, 'atr', b'\x3b\x00'),
|
|
(2.5, 'tpdu', b'\x80\xf2\x00\x00\x00'),
|
|
])
|
|
|
|
def test_roundtrip_line_events(self):
|
|
# sigrok-iso7816-stream custom sub-types survive a PCAP round-trip.
|
|
packets = [
|
|
(_gsmtap_hdr(GSMTAP_SIM_RST_EVENT), b'\x01\x00\x00', 1.0),
|
|
(_gsmtap_hdr(GSMTAP_SIM_VCC_EVENT), b'\x00\x00\x00', 1.5),
|
|
]
|
|
data = build_pcap(packets)
|
|
parsed = [(round(ts, 1), t, payload) for ts, t, payload, _f in parse_pcap(data)]
|
|
self.assertEqual(parsed, [
|
|
(1.0, 'rst', b'\x01\x00\x00'),
|
|
(1.5, 'vcc', b'\x00\x00\x00'),
|
|
])
|
|
|
|
def test_no_gsmtap(self):
|
|
# A packet that is not GSMTAP-SIM yields nothing.
|
|
frame = wrap_gsmtap(b'\x03' + b'\x00' * 15) # type=3, not SIM
|
|
# build a fake pcap with an Ethernet frame that has no GSMTAP
|
|
packets = [(b'', b'\x00' * 64, 1.0)]
|
|
data = build_pcap(packets)
|
|
self.assertEqual(list(parse_pcap(data)), [])
|
|
|
|
def test_bad_fcs_flag_roundtrip(self):
|
|
# GSMTAP_FLAG_BAD_FCS (res byte 0x01) survives a PCAP round-trip.
|
|
from simtrace2_pysniff.gsmtap import GSMTAP_FLAG_BAD_FCS
|
|
hdr = bytearray(_gsmtap_hdr(GSMTAP_SIM_APDU))
|
|
hdr[15] |= GSMTAP_FLAG_BAD_FCS
|
|
packets = [(bytes(hdr), b'\x80\xf2\x00\x00', 1.0),
|
|
(_gsmtap_hdr(GSMTAP_SIM_APDU), b'\x80\xf2\x00\x00', 2.0)]
|
|
data = build_pcap(packets)
|
|
parsed = list(parse_pcap(data))
|
|
self.assertEqual(parsed[0], (1.0, 'tpdu', b'\x80\xf2\x00\x00', GSMTAP_FLAG_BAD_FCS))
|
|
self.assertEqual(parsed[1], (2.0, 'tpdu', b'\x80\xf2\x00\x00', 0))
|
|
|
|
|
|
class TestParsePcapng(unittest.TestCase):
|
|
def test_parse(self):
|
|
data = _build_pcapng([
|
|
(_gsmtap_hdr(GSMTAP_SIM_ATR), b'\x3b\x04', 0, 0),
|
|
(_gsmtap_hdr(GSMTAP_SIM_APDU), b'\x80\xf2\x00\x00\x00', 0, 1),
|
|
])
|
|
parsed = [(t, payload) for _ts, t, payload, _f in parse_pcapng(data)]
|
|
self.assertEqual(parsed, [
|
|
('atr', b'\x3b\x04'),
|
|
('tpdu', b'\x80\xf2\x00\x00\x00'),
|
|
])
|
|
|
|
def test_no_gsmtap(self):
|
|
# IDB + EPB carrying a GSMTAP header whose type != SIM → no messages.
|
|
shb = _pcapng_block(0x0A0D0D0A, struct.pack('>IHHq', 0x1A2B3C4D, 1, 0, -1))
|
|
idb = _pcapng_block(0x00000001,
|
|
struct.pack('>HHI', LINKTYPE_ETHERNET, 0, 65535) +
|
|
struct.pack('>HH', 0, 0))
|
|
non_sim = struct.pack('!BBBBHBBIBBBB', 2, 4, 7, 0, 0, 0, 0, 0, 0, 0, 0, 0) + b'\x00'
|
|
epb = _pcapng_block(0x00000006,
|
|
struct.pack('>IIIII', 0, 0, 0, len(non_sim), len(non_sim)) + non_sim)
|
|
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('<HH', 2, 2) + b'lo' + b'\x00\x00' +
|
|
struct.pack('<HH', 9, 1) + b'\x09' + b'\x00\x00\x00' +
|
|
struct.pack('<HH', 0, 0))
|
|
self.assertEqual(_parse_tsresol(opts, '<'), 1e-9)
|
|
|
|
def test_default_microseconds(self):
|
|
from simtrace2_pysniff.pcap import _parse_tsresol
|
|
self.assertEqual(_parse_tsresol(struct.pack('<HH', 0, 0), '<'), 1e-6)
|
|
|
|
def test_parse_pcapng_nanoseconds(self):
|
|
# IDB with if_tsresol=9 → EPB timestamps in nanoseconds → seconds.
|
|
shb = _pcapng_block(0x0A0D0D0A, 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 simtrace_analyser.server import RequestHandler
|
|
from simtrace_analyser.database import Database
|
|
|
|
class Fake(RequestHandler):
|
|
def __init__(self, body, db):
|
|
self.db = db
|
|
self._body = body
|
|
self._status = None
|
|
self._json = None
|
|
|
|
def _read_body(self):
|
|
return self._body
|
|
|
|
def _send_json(self, data, status=200):
|
|
self._status = status
|
|
self._json = data
|
|
|
|
def _send_error(self, status, message):
|
|
self._status = status
|
|
self._json = {'error': message}
|
|
|
|
tmp = tempfile.NamedTemporaryFile(suffix='.db', delete=False)
|
|
tmp.close()
|
|
db = Database(tmp.name)
|
|
self.addCleanup(lambda: (db._conn.close(), os.remove(tmp.name)))
|
|
return Fake(body, db), tmp.name
|
|
|
|
def test_import_pcap_creates_session(self):
|
|
packets = [
|
|
(_gsmtap_hdr(GSMTAP_SIM_ATR), b'\x3b\x00', 1.0),
|
|
(_gsmtap_hdr(GSMTAP_SIM_APDU), b'\x80\xf2\x00\x00\x00', 2.0),
|
|
]
|
|
h, _ = self._handler(build_pcap(packets))
|
|
h._handle_import_pcap({'name': ['mytrace']})
|
|
self.assertEqual(h._status, 200)
|
|
self.assertEqual(h._json['name'], 'mytrace')
|
|
self.assertEqual(h._json['message_count'], 2)
|
|
self.assertEqual(h.db.count_messages(h._json['session_id']), 2)
|
|
|
|
def test_import_no_gsmtap_400(self):
|
|
# Ethernet frame with no GSMTAP → no usable data.
|
|
h, _ = self._handler(build_pcap([(b'', b'\x00' * 64, 1.0)]))
|
|
h._handle_import_pcap({'name': ['empty']})
|
|
self.assertEqual(h._status, 400)
|
|
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):
|
|
from simtrace2_pysniff import __version__
|
|
self.assertEqual(__version__, '2.6.2')
|
|
|
|
def test_status_includes_version(self):
|
|
import os
|
|
from simtrace_analyser.server import RequestHandler
|
|
from simtrace_analyser.database import Database
|
|
from simtrace2_pysniff import __version__
|
|
|
|
class FakeCapture:
|
|
active = False
|
|
session_id = None
|
|
device_connected = None
|
|
|
|
class Fake(RequestHandler):
|
|
def __init__(self, db):
|
|
self.db = db
|
|
self.capture = FakeCapture()
|
|
self._json = None
|
|
|
|
def _send_json(self, data, status=200):
|
|
self._json = data
|
|
|
|
tmp = tempfile.NamedTemporaryFile(suffix='.db', delete=False)
|
|
tmp.close()
|
|
db = Database(tmp.name)
|
|
self.addCleanup(lambda: (db._conn.close(), os.remove(tmp.name)))
|
|
|
|
h = Fake(db)
|
|
h._handle_status()
|
|
self.assertEqual(h._json['version'], __version__)
|
|
|
|
|
|
class TestCaptureDisabled(unittest.TestCase):
|
|
def _handler(self, capture_mode):
|
|
import os
|
|
from simtrace_analyser.server import RequestHandler
|
|
from simtrace_analyser.database import Database
|
|
|
|
class Fake(RequestHandler):
|
|
def __init__(self, db, capture_mode):
|
|
self.db = db
|
|
self.capture = None
|
|
self.capture_mode = capture_mode
|
|
self._status = None
|
|
self._json = None
|
|
|
|
def _send_json(self, data, status=200):
|
|
self._status = status
|
|
self._json = data
|
|
|
|
def _send_error(self, status, message):
|
|
self._status = status
|
|
self._json = {'error': message}
|
|
|
|
tmp = tempfile.NamedTemporaryFile(suffix='.db', delete=False)
|
|
tmp.close()
|
|
db = Database(tmp.name)
|
|
self.addCleanup(lambda: (db._conn.close(), os.remove(tmp.name)))
|
|
return Fake(db, capture_mode)
|
|
|
|
def test_status_reports_disabled(self):
|
|
h = self._handler('disabled')
|
|
h._handle_status()
|
|
self.assertEqual(h._json['capture_mode'], 'disabled')
|
|
self.assertFalse(h._json['capture_active'])
|
|
self.assertIsNone(h._json['session_id'])
|
|
|
|
def test_capture_start_403(self):
|
|
h = self._handler('disabled')
|
|
h._handle_capture_start()
|
|
self.assertEqual(h._status, 403)
|
|
self.assertIn('Capture disabled', h._json['error'])
|
|
|
|
def test_capture_stop_403(self):
|
|
h = self._handler('disabled')
|
|
h._handle_capture_stop()
|
|
self.assertEqual(h._status, 403)
|
|
|
|
def test_status_reports_gsmtap(self):
|
|
h = self._handler('gsmtap')
|
|
h._handle_status()
|
|
self.assertEqual(h._json['capture_mode'], 'gsmtap')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|