diff --git a/README.md b/README.md index 2509135..38c8bfe 100644 --- a/README.md +++ b/README.md @@ -113,6 +113,13 @@ simtrace2-pysniff-server --capture direct simtrace2-pysniff-server --capture disabled ``` +The server logs its capture liveness to stderr: `Capture started` on +begin, a periodic `Capture alive: session=N messages=M bytes=B dropped=D` +heartbeat, and `Capture stopped` (or `ERROR: capture thread stopped on +error …` if the capture thread died unexpectedly). A mis-framed or +non-SIM GSMTAP datagram is dropped and counted (`dropped`) rather than +silently killing the capture. + The GSMTAP listener also accepts the **sigrok-iso7816-stream** decoder (https://github.com/anttro/sigrok_iso7816_stream) — an FX2-logic-analyzer passive SIM sniffer that emits the same GSMTAP-SIM format on UDP 4729. Its diff --git a/simtrace2_pysniff/server/capture.py b/simtrace2_pysniff/server/capture.py index a6e33ed..a8a98d3 100644 --- a/simtrace2_pysniff/server/capture.py +++ b/simtrace2_pysniff/server/capture.py @@ -1,7 +1,10 @@ """Capture backends: GSMTAP listener and direct SIMtrace2 sniffer.""" -import time +import struct +import sys import threading +import time +import traceback from ..gsmtap import (GsmtapReceiver, GSMTAP_SIM_ATR, GSMTAP_SIM_RST_EVENT, GSMTAP_SIM_VCC_EVENT) @@ -24,17 +27,33 @@ class GsmtapListener: def __init__(self, bind_port=4729): self._receiver = GsmtapReceiver(bind_port=bind_port) self._running = False + self._dropped = 0 def start(self): self._running = True + self._dropped = 0 def stop(self): self._running = False self._receiver.close() + @property + def dropped(self): + return self._dropped + def iter_messages(self): while self._running: - sub_type, data, flags = self._receiver.read_packet() + try: + sub_type, data, flags = self._receiver.read_packet() + except (ValueError, struct.error) as e: + # A malformed or non-SIM GSMTAP datagram (a different + # GSMTAP packet type, a truncated frame, …) must not kill + # the whole capture — drop it and keep listening. + self._dropped += 1 + if self._dropped <= 1 or self._dropped % 100 == 0: + print(f'GSMTAP: dropped packet ({self._dropped}): {e}', + file=sys.stderr) + continue if sub_type is None: continue yield gsmtap_msg_type(sub_type), data, flags @@ -75,13 +94,18 @@ class DirectSniffer: class CaptureManager: - def __init__(self, backend, db): + def __init__(self, backend, db, log_interval=60.0): self._backend = backend self._db = db self._session_id = None self._thread = None + self._log_thread = None + self._stop_event = threading.Event() self._start_time = 0.0 self._latest_msg_id = 0 + self._msg_count = 0 + self._byte_count = 0 + self._log_interval = log_interval @property def active(self): @@ -115,32 +139,80 @@ class CaptureManager: self._session_id = self._db.create_session(mode) self._start_time = time.monotonic() self._latest_msg_id = 0 + self._msg_count = 0 + self._byte_count = 0 self._backend.start() self._thread = threading.Thread(target=self._capture_loop, daemon=True) self._thread.start() + self._stop_event.clear() + print(f'Capture started (mode={mode}, session={self._session_id})', + file=sys.stderr) + self._log_thread = threading.Thread(target=self._log_loop, daemon=True) + self._log_thread.start() + return self._session_id def stop_session(self): if self._session_id is None: return None + self._stop_event.set() self._backend.stop() if self._thread is not None: self._thread.join(timeout=2.0) + if self._log_thread is not None: + self._log_thread.join(timeout=2.0) sid = self._session_id self._session_id = None self._start_time = 0.0 - if self._db.count_messages(sid) == 0: + empty = self._db.count_messages(sid) == 0 + if empty: self._db.delete_session(sid) + print(f'Capture stopped: empty session discarded (session={sid})', + file=sys.stderr) return None self._db.close_session(sid) + print(f'Capture stopped: session={sid} messages={self._msg_count} ' + f'bytes={self._byte_count}', file=sys.stderr) return sid def _capture_loop(self): - for msg_type, data, flags in self._backend.iter_messages(): + try: + for msg_type, data, flags in self._backend.iter_messages(): + if self._session_id is None: + break + try: + elapsed = round(time.monotonic() - self._start_time, 3) + mid = self._db.insert_message( + self._session_id, elapsed, msg_type, data, flags) + self._msg_count += 1 + self._byte_count += len(data) + self._latest_msg_id = mid + except Exception as e: + print(f'Capture: insert error, dropping msg: {e}', + file=sys.stderr) + continue + except Exception: + # The capture thread died unexpectedly (not a user stop). + sid = self._session_id + print(f'ERROR: capture thread stopped on error (session={sid}):', + file=sys.stderr) + traceback.print_exc() + self._stop_event.set() + if sid is not None: + try: + self._db.close_session(sid) + except Exception: + pass + self._session_id = None + return + + def _log_loop(self): + while not self._stop_event.wait(self._log_interval): if self._session_id is None: break - elapsed = round(time.monotonic() - self._start_time, 3) - mid = self._db.insert_message(self._session_id, elapsed, msg_type, data, flags) - self._latest_msg_id = mid + dropped = getattr(self._backend, 'dropped', 0) + print(f'Capture alive: session={self._session_id} ' + f'messages={self._msg_count} bytes={self._byte_count} ' + f'dropped={dropped}', file=sys.stderr) diff --git a/tests/test_capture.py b/tests/test_capture.py index aeccae0..b5b7bdc 100644 --- a/tests/test_capture.py +++ b/tests/test_capture.py @@ -1,7 +1,10 @@ """Tests for CaptureManager session lifecycle (empty-session discard).""" +import contextlib +import io import os import tempfile +import time import unittest from simtrace2_pysniff.server.capture import CaptureManager, gsmtap_msg_type @@ -99,5 +102,135 @@ class TestCaptureManager(unittest.TestCase): self.assertIsNone(mgr.device_connected) +class _FlakyReceiver: + """Mimics GsmtapReceiver.read_packet, emitting scripted behaviors.""" + + def __init__(self, listener, behaviors): + self._listener = listener + self._behaviors = list(behaviors) + self._i = 0 + + def read_packet(self): + b = self._behaviors[self._i] + self._i += 1 + if b == 'raise_non_sim': + raise ValueError('Not a GSMTAP SIM packet (type=99)') + if b == 'raise_short': + raise ValueError('GSMTAP packet too short: 3 bytes') + if b == 'none': + return None, None, 0 + # 'good': return a valid APDU and terminate the loop. + self._listener._running = False + return 0x00, b'\x80\xf2\x00\x00\x00', 0 + + def close(self): + pass + + +class _NullDb: + def create_session(self, mode): + return 1 + + def count_messages(self, sid): + return 0 + + def delete_session(self, sid): + pass + + def close_session(self, sid): + pass + + def insert_message(self, *a, **k): + raise RuntimeError('db down') + + +class _OneShotBackend: + def start(self): + pass + + def stop(self): + pass + + def iter_messages(self): + yield ('tpdu', b'\x80\xf2\x00\x00\x00', 0) + + +class _RaisingBackend: + def start(self): + pass + + def stop(self): + pass + + def iter_messages(self): + raise RuntimeError('boom') + yield # unreachable + + +class TestGsmtapListenerRobustness(unittest.TestCase): + def test_listener_survives_bad_packet(self): + from simtrace2_pysniff.server.capture import GsmtapListener + listener = GsmtapListener(bind_port=0) + listener._receiver.close() # drop the real socket before swapping + listener._receiver = _FlakyReceiver(listener, ['raise_non_sim', 'good']) + listener.start() + out = list(listener.iter_messages()) + listener.stop() + self.assertEqual(out, [('tpdu', b'\x80\xf2\x00\x00\x00', 0)]) + self.assertEqual(listener.dropped, 1) + + def test_listener_survives_short_packet(self): + from simtrace2_pysniff.server.capture import GsmtapListener + listener = GsmtapListener(bind_port=0) + listener._receiver.close() # drop the real socket before swapping + listener._receiver = _FlakyReceiver(listener, ['raise_short', 'good']) + listener.start() + out = list(listener.iter_messages()) + listener.stop() + self.assertEqual(out, [('tpdu', b'\x80\xf2\x00\x00\x00', 0)]) + self.assertEqual(listener.dropped, 1) + + +class TestCaptureLoopRobustness(unittest.TestCase): + def test_capture_loop_survives_insert_error(self): + mgr = CaptureManager(_OneShotBackend(), _NullDb()) + mgr._session_id = 1 + mgr._start_time = 0.0 + # Must not raise; counters stay untouched on a dropped insert. + mgr._capture_loop() + self.assertEqual(mgr._session_id, 1) + self.assertEqual(mgr._msg_count, 0) + + def test_capture_loop_logs_stop_on_error(self): + mgr = CaptureManager(_RaisingBackend(), _NullDb()) + mgr._session_id = 7 + mgr._start_time = 0.0 + buf = io.StringIO() + with contextlib.redirect_stderr(buf): + mgr._capture_loop() + self.assertIn('stopped on error', buf.getvalue()) + self.assertIsNone(mgr._session_id) + + def test_capture_heartbeat_and_stop_log(self): + tmp = tempfile.NamedTemporaryFile(suffix='.db', delete=False) + tmp.close() + db = Database(tmp.name) + mgr = CaptureManager(_FakeBackend([('tpdu', b'\x80\xf2\x00\x00\x00', 0)]), db) + mgr._log_interval = 0.05 + buf = io.StringIO() + with contextlib.redirect_stderr(buf): + mgr.start_session() + time.sleep(0.25) + mgr.stop_session() + log = buf.getvalue() + self.assertIn('Capture started', log) + self.assertIn('Capture alive', log) + self.assertIn('messages=1', log) + self.assertIn('bytes=5', log) + self.assertIn('Capture stopped', log) + db._conn.close() + os.remove(tmp.name) + + if __name__ == '__main__': unittest.main()