catarrh
903ead9791
A container's 127.0.0.1 is its own loopback, so a GSMTAP feed sent to the container's bridge address never reached the listener: it was hard-bound to 127.0.0.1. - `simtrace-analyser-server --gsmtap-bind ADDR` (default 127.0.0.1) is passed through GsmtapListener to GsmtapReceiver (whose bind_ip already existed), including the receiver re-created by a start() after close. The startup banner prints the effective bind address next to the port. - READMEs (EN/RU) document the option, naming the container case. - tests: the listener forwards the address on construction and on start(), binds 0.0.0.0, and keeps the loopback default. Verified end-to-end: with --gsmtap-bind 0.0.0.0 the socket listens on 0.0.0.0:14732 and an ATR + APDU sent to 127.0.0.1:14732 landed in the capture session (2 messages); without the flag it listens on 127.0.0.1. 557 Python + 80 Node tests.
287 lines
9.3 KiB
Python
287 lines
9.3 KiB
Python
"""Tests for CaptureManager session lifecycle (empty-session discard)."""
|
|
|
|
import contextlib
|
|
import io
|
|
import os
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
|
|
from simtrace_analyser.capture import CaptureManager, gsmtap_msg_type
|
|
from simtrace_analyser.database import Database
|
|
|
|
|
|
class TestGsmtapSubtypeMapping(unittest.TestCase):
|
|
def test_custom_line_event_subtypes(self):
|
|
self.assertEqual(gsmtap_msg_type(0x10), 'rst')
|
|
self.assertEqual(gsmtap_msg_type(0x11), 'vcc')
|
|
|
|
def test_standard_subtypes(self):
|
|
self.assertEqual(gsmtap_msg_type(0x00), 'tpdu')
|
|
self.assertEqual(gsmtap_msg_type(0x01), 'atr')
|
|
self.assertEqual(gsmtap_msg_type(0x02), 'pps') # PPS request
|
|
self.assertEqual(gsmtap_msg_type(0x03), 'pps') # PPS response
|
|
|
|
|
|
class _FakeBackend:
|
|
def __init__(self, messages=()):
|
|
self._messages = list(messages)
|
|
|
|
def start(self):
|
|
pass
|
|
|
|
def stop(self):
|
|
pass
|
|
|
|
def iter_messages(self):
|
|
yield from self._messages
|
|
|
|
|
|
class _DisconnectingSession:
|
|
def iter_messages(self):
|
|
from simtrace2_pysniff.device import DeviceDisconnected
|
|
raise DeviceDisconnected('disconnected')
|
|
yield
|
|
|
|
|
|
class TestCaptureManager(unittest.TestCase):
|
|
def _manager(self, backend):
|
|
tmp = tempfile.NamedTemporaryFile(suffix='.db', delete=False)
|
|
tmp.close()
|
|
db = Database(tmp.name)
|
|
self.addCleanup(lambda: (db._conn.close(), os.remove(tmp.name)))
|
|
return CaptureManager(backend, db), db
|
|
|
|
def test_empty_capture_discarded(self):
|
|
mgr, db = self._manager(_FakeBackend())
|
|
sid = mgr.start_session()
|
|
self.assertIsNotNone(sid)
|
|
self.assertIsNone(mgr.stop_session())
|
|
self.assertIsNone(db.get_session(sid))
|
|
|
|
def test_non_empty_capture_kept(self):
|
|
mgr, db = self._manager(_FakeBackend([('tpdu', b'\x80\xf2\x00\x00\x00', 0)]))
|
|
sid = mgr.start_session()
|
|
self.assertEqual(mgr.stop_session(), sid)
|
|
self.assertIsNotNone(db.get_session(sid))
|
|
self.assertEqual(db.count_messages(sid), 1)
|
|
|
|
def test_non_apdu_packets_are_skipped(self):
|
|
# A character-level or mis-sampled stream is not APDU-shaped and must
|
|
# not be stored as decodable exchanges; valid rows pass.
|
|
backend = _FakeBackend([
|
|
('tpdu', bytes.fromhex('ff10957aff10957a'), 0),
|
|
('tpdu', bytes.fromhex('2d6ccfb0ff'), 0),
|
|
('tpdu', bytes.fromhex('00a40004023f00'), 0),
|
|
])
|
|
mgr, db = self._manager(backend)
|
|
sid = mgr.start_session()
|
|
self.assertEqual(mgr.stop_session(), sid)
|
|
self.assertEqual(db.count_messages(sid), 1)
|
|
self.assertEqual(mgr.skipped, 2)
|
|
|
|
def test_direct_sniffer_device_disconnect(self):
|
|
from simtrace_analyser.capture import DirectSniffer
|
|
sn = DirectSniffer()
|
|
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 simtrace_analyser.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 simtrace_analyser.capture import GsmtapListener
|
|
listener = GsmtapListener(bind_port=0)
|
|
self.addCleanup(listener.stop)
|
|
mgr, _ = self._manager(listener)
|
|
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 simtrace_analyser.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 simtrace_analyser.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 TestGsmtapListenerBind(unittest.TestCase):
|
|
def test_bind_ip_reaches_the_receiver(self):
|
|
from simtrace_analyser import capture
|
|
calls = []
|
|
|
|
class _FakeReceiver:
|
|
def __init__(self, bind_ip='127.0.0.1', bind_port=4729):
|
|
calls.append((bind_ip, bind_port))
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
real = capture.GsmtapReceiver
|
|
capture.GsmtapReceiver = _FakeReceiver
|
|
try:
|
|
listener = capture.GsmtapListener(bind_ip='0.0.0.0', bind_port=4729)
|
|
listener._receiver._closed = True # force start() to re-create it
|
|
listener.start()
|
|
finally:
|
|
capture.GsmtapReceiver = real
|
|
self.assertEqual(calls, [('0.0.0.0', 4729), ('0.0.0.0', 4729)])
|
|
|
|
def test_all_interfaces_bind(self):
|
|
from simtrace_analyser.capture import GsmtapListener
|
|
listener = GsmtapListener(bind_ip='0.0.0.0', bind_port=0)
|
|
self.addCleanup(listener.stop)
|
|
self.assertEqual(listener._receiver._sock.getsockname()[0], '0.0.0.0')
|
|
|
|
def test_default_bind_is_loopback(self):
|
|
from simtrace_analyser.capture import GsmtapListener
|
|
listener = GsmtapListener(bind_port=0)
|
|
self.addCleanup(listener.stop)
|
|
self.assertEqual(listener._receiver._sock.getsockname()[0], '127.0.0.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()
|