Keep GSMTAP capture alive + server-side liveness logging
The GSMTAP capture thread died silently whenever a single non-SIM or truncated GSMTAP datagram arrived (read_packet() raises ValueError, uncaught in the loop, killing the daemon thread while UDP kept flowing) — matching the "stops listening after a sniffer restart" symptom. Fixes: - GsmtapListener.iter_messages() now catches ValueError/struct.error from read_packet(), drops the offending datagram (counted via a new `dropped` counter) and keeps listening. - CaptureManager._capture_loop() guards the DB insert so a transient error can't kill the thread; a top-level guard logs "ERROR: capture thread stopped on error …" + traceback and clears the session so /api/status no longer reports a stuck "active". - Server now logs capture liveness to stderr: "Capture started", a periodic "Capture alive: session=N messages=M bytes=B dropped=D" heartbeat (default every 60 s, configurable via log_interval), and "Capture stopped" (or "empty session discarded"). Tests: +5 (listener survives non-SIM/short packet, capture loop survives insert error, logs stop-on-error, heartbeat+stop log). Stays on 1.22.0 (backend-only).
This commit is contained in:
1 parent
79acb3773f
commit
51e1b4ae08
3 files changed
+220
-8
No files matched your search
@@ -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
|
||||
|
||||
@@ -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)
|
||||
@@ -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()
|
||||
Reference in new issue
Block a user