From 5d91450b55ad90979aef55cdb4388ca65a685dd9 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: Wed, 19 Aug 2026 18:00:52 +0300 Subject: [PATCH] Track device connection and capture gaps; surface sniff error flags --- frontend/index.html | 94 ++++++++++++++++++++++++++-- simtrace2_pysniff/__main__.py | 4 +- simtrace2_pysniff/device.py | 39 +++++++++--- simtrace2_pysniff/server/capture.py | 13 ++++ simtrace2_pysniff/server/database.py | 21 ++++++- simtrace2_pysniff/server/decode.py | 52 ++++++++++++--- simtrace2_pysniff/server/server.py | 6 +- tests/test_capture.py | 30 +++++++++ tests/test_decode.py | 56 +++++++++++++++++ tests/test_pcap_import.py | 1 + 10 files changed, 287 insertions(+), 29 deletions(-) diff --git a/frontend/index.html b/frontend/index.html index daf1eef..712d1f1 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -76,6 +76,9 @@ body { font: 14px/1.5 system-ui, sans-serif; background: var(--bg); color: var(- .header-title a { color: var(--accent); text-decoration: none; } .header-title a:hover { text-decoration: underline; } .header-right { margin-left: auto; display: flex; align-items: center; gap: 12px; padding: 8px 16px; } +.capture-chip { font-size: 12px; font-weight: 600; padding: 4px 10px; border-radius: 12px; white-space: nowrap; } +.capture-chip.live { color: var(--green); background: var(--accent-bg); } +.capture-chip.warn { color: var(--warn); background: var(--warning-bg); } .header-right a { color: var(--link); text-decoration: none; font-size: 13px; } .header-right a:hover { color: var(--link-hover); } .header-right button { padding: 4px 10px; font-size: 14px; cursor: pointer; background: var(--bg-input); border: 1px solid var(--border); border-radius: 4px; color: var(--text); } @@ -115,6 +118,11 @@ body { font: 14px/1.5 system-ui, sans-serif; background: var(--bg); color: var(- .apdu-sw.sw-pending { color: var(--accent); } .apdu-sw.sw-warn { color: var(--warn); } .apdu-sw.sw-error { color: var(--error); } +.apdu-err { color: var(--error); white-space: nowrap; } +.apdu-gap { background: var(--warning-bg); border-bottom: 1px solid var(--warn); } +.apdu-gap .apdu-type { color: var(--warn); } +.apdu-gap .apdu-data { color: var(--warn); } +.device-warn { padding: 8px 12px; font-size: 12px; color: var(--warn); background: var(--warning-bg); border-bottom: 1px solid var(--warn); } /* APDU detail panel */ .apdu-detail { display: flex; flex-direction: column; max-height: 40%; overflow-y: auto; border-top: 2px solid var(--accent); background: var(--bg-panel); flex-shrink: 0; } @@ -232,6 +240,7 @@ body { font: 14px/1.5 system-ui, sans-serif; background: var(--bg); color: var(-
SIM-ME GSMTAP trace analyser
+ GitHub @@ -267,6 +276,7 @@ body { font: 14px/1.5 system-ui, sans-serif; background: var(--bg); color: var(- Not connected 0 messages
+ @@ -348,6 +358,7 @@ const STATE = { serverName: '', serverVersion: '', captureMode: '', + deviceConnected: null, }; const API_BASE = () => STATE.serverUrl; @@ -378,6 +389,7 @@ function ins_label(msg) { if (msg.type === 'pps') return 'PPS'; if (msg.type === 'change') return 'STATE CHANGE'; if (msg.type === 'fidi') return 'FIDI'; + if (msg.type === 'gap') return 'CAPTURE GAP'; const d = msg.data; if (d && d.length >= 2) { const ins = d.slice(2, 4); @@ -391,6 +403,7 @@ function _fallback_label(msg) { if (msg.type === 'pps') return 'PPS'; if (msg.type === 'change') return 'STATE CHANGE'; if (msg.type === 'fidi') return 'FIDI'; + if (msg.type === 'gap') return 'CAPTURE GAP'; const data = msg.data; if (data && data.length >= 2) { const ins = data.slice(2, 4); @@ -458,10 +471,16 @@ function data_text(m) { } function cmd_description(m) { + if (m.type === 'gap') return t('device disconnected — messages may be missing'); if (m.decoded && m.decoded.summary) return m.decoded.summary; return data_text(m); } +function error_names(m) { + const d = m.decoded; + return (d && Array.isArray(d.errors) && d.errors.length) ? d.errors : null; +} + function sw_text(m) { const d = m.decoded; if (!d || !d.sw) return ''; @@ -834,6 +853,7 @@ function atrPpsRows(decoded) { if (decoded.note) rows.push(['Note', decoded.note]); } rows.push(['Raw', (decoded.raw || '')]); + if (decoded.errors) rows.push(['Errors', decoded.errors.join(', ')]); return rows; } @@ -845,6 +865,12 @@ function showDetail(msg, entry) { const d = el('apdu-detail'); const decoded = msg.decoded; + if (msg.type === 'gap') { + d.innerHTML = `
Capture Gap${fmt_elapsed(msg.elapsed)}
+
⚠ ${t('The device disconnected and reconnected — some APDUs may be missing from this trace.')}
`; + return; + } + if (!decoded) { d.innerHTML = `
${ins_label(msg)}${fmt_elapsed(msg.elapsed)}
No decoded data available for this message type.
`; @@ -959,6 +985,7 @@ function showDetail(msg, entry) { ${responseRows(decoded.response)} ${fileRows(decoded.file)} ${sw} + ${decoded.errors ? `Errors${decoded.errors.join(', ')}` : ''} `; } @@ -1213,13 +1240,23 @@ function renderMessages(skipScroll) { } list.innerHTML = filtered.map(({ m, idx }) => { + if (m.type === 'gap') { + return `
+ ${fmt_elapsed(m.elapsed)} + ⚠ CAPTURE GAP + ${cmd_description(m)} +
`; + } const label = ins_label(m); const typeCls = m.type === 'atr' ? 'atr' : m.type === 'pps' ? 'pps' : ''; + const errs = error_names(m); + const errBadge = errs ? `⚠` : ''; return `
${fmt_elapsed(m.elapsed)} ${label} ${sw_text(m)} ${cmd_description(m)} + ${errBadge}
`; }).join(''); @@ -1246,17 +1283,48 @@ function updateStatus() { text.textContent = t('Not connected'); } count.textContent = t('{n} messages').replace('{n}', STATE.messages.length); + updateDeviceWarn(); + updateCaptureChip(); +} + +function updateCaptureChip() { + const chip = el('capture-chip'); + if (!chip) return; + const disconnected = STATE.captureMode === 'direct' && STATE.captureActive && STATE.deviceConnected === false; + if (disconnected) { + chip.style.display = ''; + chip.className = 'capture-chip warn'; + chip.textContent = '⚠ ' + t('device disconnected — reconnecting…'); + } else if (STATE.captureActive) { + chip.style.display = ''; + chip.className = 'capture-chip live'; + chip.textContent = '● ' + t('LIVE') + ' · ' + t('{n} messages').replace('{n}', STATE.messages.length); + } else { + chip.style.display = 'none'; + } +} + +function updateDeviceWarn() { + const warn = el('device-warn'); + if (!warn) return; + const disconnected = STATE.captureMode === 'direct' && STATE.captureActive && STATE.deviceConnected === false; + if (disconnected) { + warn.style.display = ''; + el('device-warn-text').textContent = t('device disconnected — reconnecting…'); + } else { + warn.style.display = 'none'; + } } // ──────────────────── Live Polling ──────────────────── let _lastAfter = 0; function startPolling() { - _lastAfter = 0; if (STATE.livePoller) return; STATE.livePoller = setInterval(async () => { try { const data = await API.getLatest(_lastAfter); + STATE.deviceConnected = data.device_connected; if (data.active) { STATE.captureActive = true; STATE.currentSessionId = data.session_id; @@ -1289,6 +1357,7 @@ async function startCapture() { STATE.captureActive = true; STATE.currentSessionId = data.session_id; STATE.messages = []; + _lastAfter = 0; buildTypeCounts(); buildTraceMap(); renderSidePanel(); @@ -1426,6 +1495,7 @@ async function importPcapFile(file) { } async function loadSession(id) { + if (STATE.captureActive && !confirm(t('Opening a session stops the active capture. Continue?'))) return; try { // Stop any active capture so it can't write into the session being viewed. await API.stopCapture(); @@ -1563,6 +1633,7 @@ function connectServer() { } function disconnectServer() { + if (STATE.captureActive && !confirm(t('Capture is still running on the server — it will continue after you disconnect. Disconnect?'))) return; if (_connTimer) { clearInterval(_connTimer); _connTimer = null; @@ -1585,10 +1656,6 @@ function disconnectServer() { btn.classList.add('disconnected'); updateCaptureButton(); stopPolling(); - if (STATE.captureActive) { - STATE.captureActive = false; - STATE.currentSessionId = null; - } renderTraceEmptyState(); } @@ -1599,6 +1666,7 @@ async function checkConnection() { STATE.captureMode = data.capture_mode || ''; STATE.serverName = data.server || 'simtrace-analyser-server'; STATE.serverVersion = data.version || ''; + STATE.deviceConnected = data.device_connected; el('conn-status').textContent = connectionStatusText(); el('conn-status').className = 'conn-status connected'; updateVersionWarn(data.version); @@ -1610,8 +1678,18 @@ async function checkConnection() { el('status-dot').className = 'status-dot ' + (data.capture_active ? 'live' : 'paused'); el('status-text').textContent = data.capture_active ? t('Capturing') : t('Connected'); if (data.capture_active) { + if (data.session_id && data.session_id !== STATE.currentSessionId) { + // A different capture session than the one we were viewing — + // reset the stream so we re-fetch it from the start. + STATE.currentSessionId = data.session_id; + STATE.messages = []; + _lastAfter = 0; + } STATE.captureActive = true; STATE.currentSessionId = data.session_id; + startPolling(); + } else { + STATE.captureActive = false; } updateCaptureButton(); renderTraceEmptyState(); @@ -1734,6 +1812,12 @@ const LANG_RU = { 'Import PCAP': 'Импорт PCAP', 'New capture': 'Новый захват', 'Server': 'Сервер', + 'device disconnected — reconnecting…': 'устройство отключено — переподключение…', + 'device disconnected — messages may be missing': 'устройство отключалось — сообщения могли быть потеряны', + 'The device disconnected and reconnected — some APDUs may be missing from this trace.': 'Устройство отключалось и переподключалось — часть APDU может отсутствовать в трейсе.', + 'LIVE': 'В ЭФИРЕ', + 'Opening a session stops the active capture. Continue?': 'Открытие сессии остановит активный захват. Продолжить?', + 'Capture is still running on the server — it will continue after you disconnect. Disconnect?': 'Захват всё ещё выполняется на сервере — он продолжится после отключения. Отключиться?', 'Same origin': 'Тот же источник', 'Custom URL': 'Свой URL', 'Connect': 'Подключить', diff --git a/simtrace2_pysniff/__main__.py b/simtrace2_pysniff/__main__.py index 90f7fba..31bd6dd 100644 --- a/simtrace2_pysniff/__main__.py +++ b/simtrace2_pysniff/__main__.py @@ -55,9 +55,6 @@ def build_parser(): p.add_argument('--vendor', type=lambda x: int(x, 16), default=0x1d50, metavar='HEX', help='USB vendor ID (default: 0x1d50)') - p.add_argument('--product', type=lambda x: int(x, 16), - default=None, metavar='HEX', - help='USB product ID (default: auto-detect)') rec = p.add_argument_group('recovery options') rec.add_argument('--no-reconnect', action='store_true', @@ -112,6 +109,7 @@ def main(): reconnect_delay_max=args.reconnect_delay_max, backoff_factor=args.backoff_factor, inactivity_timeout=args.inactivity_timeout, + vendor_id=args.vendor, ) exiting = False diff --git a/simtrace2_pysniff/device.py b/simtrace2_pysniff/device.py index 58f5cc4..2753342 100644 --- a/simtrace2_pysniff/device.py +++ b/simtrace2_pysniff/device.py @@ -45,13 +45,7 @@ def find_sniffer_device(vendor_id=USB_VENDOR_OPENMOKO): Returns a ``usb.core.Device`` on success, or ``None``. """ - devices = list(usb.core.find( - find_all=True, - idVendor=vendor_id, - custom_match=lambda d: ( - True - ), - )) + devices = list(usb.core.find(find_all=True, idVendor=vendor_id)) for dev in devices: for cfg in dev: @@ -71,7 +65,9 @@ def _get_ep_addrs(dev): intf.bInterfaceSubClass == SIMTRACE_SNIFFER_USB_SUBCLASS): for ep in intf: addr = ep.bEndpointAddress - if addr & 0x80: # IN endpoint + # BULK IN endpoint only — the interrupt IN endpoint carries + # cardem notifications, not sniff data. + if addr & 0x80 and ep.bmAttributes & 0x03 == 0x02: return addr return None @@ -100,16 +96,19 @@ class SniffSession: reconnect_delay_min=1.0, reconnect_delay_max=30.0, backoff_factor=1.5, - inactivity_timeout=0.0): + inactivity_timeout=0.0, + vendor_id=USB_VENDOR_OPENMOKO): self._reconnect = reconnect self._reconnect_delay_min = reconnect_delay_min self._reconnect_delay_max = reconnect_delay_max self._backoff_factor = backoff_factor self._inactivity_timeout = inactivity_timeout + self._vendor_id = vendor_id self._dev = None self._devh = None self._ep_in = None + self._connected = False self._last_msg_time = 0.0 self._disconnect_count = 0 @@ -121,7 +120,7 @@ class SniffSession: if self._devh is not None: return - dev = find_sniffer_device() + dev = find_sniffer_device(vendor_id=self._vendor_id) if dev is None: raise DeviceDisconnected('No SIMtrace2 sniffer device found') @@ -147,6 +146,7 @@ class SniffSession: self._dev = dev self._devh = devh self._ep_in = ep_in + self._connected = True self._last_msg_time = time.monotonic() self._disconnect_count = 0 @@ -167,6 +167,7 @@ class SniffSession: self._devh = None self._dev = None self._ep_in = None + self._connected = False def _backoff_wait(self): """Sleep with exponential backoff before reconnecting.""" @@ -236,20 +237,38 @@ class SniffSession: f'Inactivity timeout ({elapsed:.1f}s > ' f'{self._inactivity_timeout:.1f}s)') + @property + def connected(self): + """True while the USB sniffer device is open and claimed.""" + return self._connected + def iter_messages(self): """Generator of SniffMessage objects. Automatically reconnects on disconnect if ``reconnect=True``. Raises ``DeviceDisconnected`` if reconnect is disabled or exhausted. + + When an established connection is lost and a reconnect is pending, a + single synthetic ``SniffMessage(type='gap')`` is yielded so callers can + record the interruption (messages may have been missed). """ + from .protocol import SniffMessage + + was_connected = False while True: try: self._ensure_connected() + was_connected = True yield from self._read_loop() except DeviceDisconnected: self._close() if not self._reconnect: raise + if was_connected: + yield SniffMessage(type='gap', type_hex='0x00', data=b'', + flags=0, slot_nr=0, seq_nr=0, + timestamp=time.time()) + was_connected = False self._backoff_wait() def close(self): diff --git a/simtrace2_pysniff/server/capture.py b/simtrace2_pysniff/server/capture.py index f5b0efb..909381f 100644 --- a/simtrace2_pysniff/server/capture.py +++ b/simtrace2_pysniff/server/capture.py @@ -38,6 +38,11 @@ class DirectSniffer: ) self._running = False + @property + def connected(self): + """True while the USB sniffer device is open.""" + return self._session.connected + def start(self): self._running = True @@ -81,6 +86,14 @@ class CaptureManager: def latest_msg_id(self): return self._latest_msg_id + @property + def device_connected(self): + """True while the capture device is connected, or None when the + backend has no device concept (gsmtap).""" + if isinstance(self._backend, DirectSniffer): + return self._backend.connected + return None + def start_session(self): if self._session_id is not None: self.stop_session() diff --git a/simtrace2_pysniff/server/database.py b/simtrace2_pysniff/server/database.py index 8b10008..25a9ca5 100644 --- a/simtrace2_pysniff/server/database.py +++ b/simtrace2_pysniff/server/database.py @@ -257,11 +257,11 @@ class Database: from .decode import decode_sniff_msg rows = self._conn.execute( "SELECT data, type FROM messages WHERE session_id=? AND id < ? " - "AND type IN ('tpdu','atr','change') ORDER BY id DESC LIMIT 500", + "AND type IN ('tpdu','atr','change','gap') ORDER BY id DESC LIMIT 500", (session_id, before_id)).fetchall() states = {} for data, typ in reversed(rows): - if typ in ('atr', 'change'): + if typ in ('atr', 'change', 'gap'): states = {} continue channel = data[0] & 0x03 if data and len(data) >= 1 else 0 @@ -358,7 +358,22 @@ class Database: if msg['type'] == 'tpdu' and d and d.get('ins_hex'): st['sel'], st['df'], st['sfi_map'] = self._advance_state(d, st['sel'], st['df'], st['sfi_map']) st['prev'] = self._context_from_decoded(d) - elif msg['type'] in ('atr', 'change'): + elif msg['type'] == 'atr': states = {} + elif msg['type'] == 'gap': + # Device disconnected and reconnected — messages may have been + # missed, so the selection is uncertain. + states = {} + elif msg['type'] == 'change': + # Only card-level changes invalidate the selection. A + # waiting-time timeout interrupts a single TPDU but leaves the + # card selected, so the phone usually just retries. + from ..protocol import ( + CHANGE_FLAG_CARD_INSERT, CHANGE_FLAG_CARD_EJECT, + CHANGE_FLAG_RESET_ASSERT, CHANGE_FLAG_RESET_DEASSERT, + ) + if msg['flags'] & (CHANGE_FLAG_CARD_INSERT | CHANGE_FLAG_CARD_EJECT | + CHANGE_FLAG_RESET_ASSERT | CHANGE_FLAG_RESET_DEASSERT): + states = {} msgs.append(msg) return msgs diff --git a/simtrace2_pysniff/server/decode.py b/simtrace2_pysniff/server/decode.py index b0d2123..62433bd 100644 --- a/simtrace2_pysniff/server/decode.py +++ b/simtrace2_pysniff/server/decode.py @@ -3258,13 +3258,28 @@ CHANGE_FLAGS = { 1 << 4: 'Waiting time timeout', } +# Sniff data error flags (SNIFF_DATA_FLAG_ERROR_*, simtrace_prot.h). +DATA_FLAGS = { + 1 << 5: 'incomplete', + 1 << 6: 'malformed', + 1 << 7: 'checksum error', + 1 << 8: 'overrun', + 1 << 9: 'framing error', + 1 << 10: 'parity error', +} + + +def _flag_names(flags, mapping): + names = [] + for mask, name in sorted(mapping.items()): + if flags & mask: + names.append(name) + return names + def decode_change(flags): """Decode SIMtrace2 sniff_change flags into a list of human-readable names.""" - bits = [] - for mask, name in sorted(CHANGE_FLAGS.items()): - if flags & mask: - bits.append(name) + bits = _flag_names(flags, CHANGE_FLAGS) return { 'type': 'change', 'flags_hex': f'{flags:08x}', @@ -3272,6 +3287,14 @@ def decode_change(flags): } +def decode_data_flags(flags): + """Decode SIMtrace2 sniff_data error flags into a list of names (or None).""" + if not flags: + return None + names = _flag_names(flags, DATA_FLAGS) + return names or None + + def decode_fidi(raw_data): """Decode a SIMtrace2 sniff_fidi payload (1-byte Fi/Di).""" if len(raw_data) < 1: @@ -3514,15 +3537,30 @@ def decode_sniff_msg(raw_data, msg_type, flags=0, prev=None): resolving GET RESPONSE. """ if msg_type == 'tpdu' and raw_data: - return decode_message(raw_data, prev=prev) + result = decode_message(raw_data, prev=prev) + if result is not None: + errs = decode_data_flags(flags) + if errs: + result['errors'] = errs + return result if msg_type == 'change': return decode_change(flags) if msg_type == 'fidi' and raw_data: return decode_fidi(raw_data) if msg_type == 'atr': - return _decode_atr(raw_data) + result = _decode_atr(raw_data) + if result is not None: + errs = decode_data_flags(flags) + if errs: + result['errors'] = errs + return result if msg_type == 'pps': - return _decode_pps(raw_data) + result = _decode_pps(raw_data) + if result is not None: + errs = decode_data_flags(flags) + if errs: + result['errors'] = errs + return result return None diff --git a/simtrace2_pysniff/server/server.py b/simtrace2_pysniff/server/server.py index bfe4766..2c9d01f 100644 --- a/simtrace2_pysniff/server/server.py +++ b/simtrace2_pysniff/server/server.py @@ -202,6 +202,7 @@ class RequestHandler(BaseHTTPRequestHandler): 'session_id': capture.session_id if capture else None, 'mode': active_session['mode'] if active_session else None, 'messages_count': self.db.count_messages(capture.session_id) if capture and capture.session_id else 0, + 'device_connected': capture.device_connected if capture else None, }) def _handle_list_sessions(self): @@ -258,7 +259,8 @@ class RequestHandler(BaseHTTPRequestHandler): def _handle_capture_latest(self, params): after_id = int(params.get('after', [0])[0]) if not self.capture or not self.capture.active: - self._send_json({'messages': [], 'next_after': after_id, 'active': False}) + self._send_json({'messages': [], 'next_after': after_id, 'active': False, + 'device_connected': self.capture.device_connected if self.capture else None}) return msg_id = self.capture.latest_msg_id messages = self.db.get_messages_after(self.capture.session_id, after_id) @@ -267,12 +269,14 @@ class RequestHandler(BaseHTTPRequestHandler): 'next_after': msg_id, 'active': True, 'session_id': self.capture.session_id, + 'device_connected': self.capture.device_connected, }) def _handle_capture_status(self): self._send_json({ 'active': self.capture.active if self.capture else False, 'session_id': self.capture.session_id if self.capture else None, + 'device_connected': self.capture.device_connected if self.capture else None, }) def _handle_capture_start(self): diff --git a/tests/test_capture.py b/tests/test_capture.py index 66f86fb..4133d54 100644 --- a/tests/test_capture.py +++ b/tests/test_capture.py @@ -57,6 +57,36 @@ class TestCaptureManager(unittest.TestCase): sn._session = _DisconnectingSession() self.assertEqual(list(sn.iter_messages()), []) + def test_capture_gap_message_inserted(self): + # A 'gap' marker yielded by the backend is stored as a message. + mgr, db = self._manager(_FakeBackend([ + ('gap', b'', 0), + ('tpdu', b'\x80\xf2\x00\x00\x00', 0), + ])) + sid = mgr.start_session() + self.assertEqual(mgr.stop_session(), sid) + msgs = db.get_messages(sid) + self.assertEqual([m['type'] for m in msgs], ['gap', 'tpdu']) + + def test_direct_sniffer_connected(self): + from simtrace2_pysniff.server.capture import DirectSniffer + + class _FakeSession: + connected = True + + sn = DirectSniffer() + sn._session = _FakeSession() + self.assertTrue(sn.connected) + _FakeSession.connected = False + self.assertFalse(sn.connected) + + def test_device_connected_gsmtap_none(self): + from simtrace2_pysniff.server.capture import GsmtapListener + listener = GsmtapListener(bind_port=0) + self.addCleanup(listener.stop) + mgr, _ = self._manager(listener) + self.assertIsNone(mgr.device_connected) + if __name__ == '__main__': unittest.main() diff --git a/tests/test_decode.py b/tests/test_decode.py index 7c630f1..13c15d9 100644 --- a/tests/test_decode.py +++ b/tests/test_decode.py @@ -149,6 +149,19 @@ class TestChangeAndFidiDecoding(unittest.TestCase): self.assertEqual(r['fi_val'], 512) self.assertEqual(r['di_val'], 64) + def test_tpdu_error_flags_surfaced(self): + r = decode_sniff_msg(bytes.fromhex('00a40000023f009000'), 'tpdu', + flags=(1 << 7) | (1 << 6)) + self.assertEqual(r['errors'], ['malformed', 'checksum error']) + + def test_tpdu_no_error_flags(self): + r = decode_sniff_msg(bytes.fromhex('00a40000023f009000'), 'tpdu', flags=0) + self.assertNotIn('errors', r) + + def test_atr_error_flags_surfaced(self): + r = decode_sniff_msg(bytes.fromhex('3b00'), 'atr', flags=(1 << 5)) + self.assertEqual(r['errors'], ['incomplete']) + class TestTerminalResponse(unittest.TestCase): def test_tr_setup_event_list(self): @@ -1192,6 +1205,49 @@ class TestSelectionTracking(unittest.TestCase): msgs = db.get_messages(sid) self.assertNotIn('file', msgs[2]['decoded']) + def test_timeout_wt_keeps_selection(self): + # A waiting-time timeout interrupts one TPDU but the card stays + # selected, so the selection must not be reset. + import tempfile + from simtrace2_pysniff.server.database import Database + with tempfile.NamedTemporaryFile(suffix='.db') as tmp: + db = Database(tmp.name) + sid = db.create_session('capture') + db.insert_message(sid, 0.0, 'tpdu', bytes.fromhex('a0a40804047fff6f079000')) + db.insert_message(sid, 0.1, 'change', b'', 1 << 4) # TIMEOUT_WT + db.insert_message(sid, 0.2, 'tpdu', + bytes.fromhex('00b0000009') + bytes.fromhex('082905102143658709') + bytes.fromhex('9000')) + msgs = db.get_messages(sid) + self.assertEqual(msgs[2]['decoded']['file']['imsi'], '250011234567890') + + def test_reset_assert_resets_selection(self): + import tempfile + from simtrace2_pysniff.server.database import Database + with tempfile.NamedTemporaryFile(suffix='.db') as tmp: + db = Database(tmp.name) + sid = db.create_session('capture') + db.insert_message(sid, 0.0, 'tpdu', bytes.fromhex('a0a40804047fff6f079000')) + db.insert_message(sid, 0.1, 'change', b'', 1 << 2) # RESET_ASSERT + db.insert_message(sid, 0.2, 'tpdu', + bytes.fromhex('00b0000009') + bytes.fromhex('082905102143658709') + bytes.fromhex('9000')) + msgs = db.get_messages(sid) + self.assertNotIn('file', msgs[2]['decoded']) + + def test_gap_resets_selection(self): + # A capture gap (device disconnected/reconnected) means messages may + # have been missed, so the selection must be reset. + import tempfile + from simtrace2_pysniff.server.database import Database + with tempfile.NamedTemporaryFile(suffix='.db') as tmp: + db = Database(tmp.name) + sid = db.create_session('capture') + db.insert_message(sid, 0.0, 'tpdu', bytes.fromhex('a0a40804047fff6f079000')) + db.insert_message(sid, 0.1, 'gap', b'', 0) + db.insert_message(sid, 0.2, 'tpdu', + bytes.fromhex('00b0000009') + bytes.fromhex('082905102143658709') + bytes.fromhex('9000')) + msgs = db.get_messages(sid) + self.assertNotIn('file', msgs[2]['decoded']) + def test_failed_select_keeps_selection(self): # A SELECT that fails (6A82) must not change the current file. import tempfile diff --git a/tests/test_pcap_import.py b/tests/test_pcap_import.py index 8d4e7c4..1940938 100644 --- a/tests/test_pcap_import.py +++ b/tests/test_pcap_import.py @@ -208,6 +208,7 @@ class TestVersion(unittest.TestCase): class FakeCapture: active = False session_id = None + device_connected = None class Fake(RequestHandler): def __init__(self, db):