simtrace-analyser/simtrace_analyser/capture.py

"""Capture backends: GSMTAP listener and direct SIMtrace2 sniffer."""

import struct
import sys
import threading
import time
import traceback

from simtrace2_pysniff.gsmtap import (GsmtapReceiver, GSMTAP_SIM_ATR,
                                      GSMTAP_SIM_PPS_REQ, GSMTAP_SIM_PPS_RSP,
                                      GSMTAP_SIM_RST_EVENT, GSMTAP_SIM_VCC_EVENT)

from .decode import looks_like_apdu


def _ts():
    """Local-time stamp for server-side capture log lines."""
    return time.strftime('%Y-%m-%d %H:%M:%S')


def _log(msg):
    print(f'[{_ts()}] {msg}', file=sys.stderr)


def gsmtap_msg_type(sub_type):
    """Map a GSMTAP-SIM sub_type to a stored message type.

    PPS request/response (0x02/0x03, standard GSMTAP) become 'pps'; custom
    sigrok-iso7816-stream line events (0x10/0x11) become 'rst'/'vcc';
    everything else defaults to 'tpdu'.
    """
    return {
        GSMTAP_SIM_ATR: 'atr',
        GSMTAP_SIM_PPS_REQ: 'pps',
        GSMTAP_SIM_PPS_RSP: 'pps',
        GSMTAP_SIM_RST_EVENT: 'rst',
        GSMTAP_SIM_VCC_EVENT: 'vcc',
    }.get(sub_type, 'tpdu')


class GsmtapListener:
    def __init__(self, bind_port=4729):
        self._bind_port = bind_port
        self._receiver = GsmtapReceiver(bind_port=bind_port)
        self._running = False
        self._dropped = 0

    def start(self):
        if getattr(self._receiver, '_closed', False):
            self._receiver = GsmtapReceiver(bind_port=self._bind_port)
        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:
            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:
                    _log(f'GSMTAP: dropped packet ({self._dropped}): {e}')
                continue
            if sub_type is None:
                continue
            yield gsmtap_msg_type(sub_type), data, flags


class DirectSniffer:
    def __init__(self):
        from simtrace2_pysniff.device import SniffSession
        self._session = SniffSession(
            reconnect=True,
            reconnect_delay_min=1.0,
            reconnect_delay_max=10.0,
            inactivity_timeout=0.0,
        )
        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

    def stop(self):
        self._running = False
        self._session.close()

    def iter_messages(self):
        from simtrace2_pysniff.device import DeviceDisconnected
        try:
            for msg in self._session.iter_messages():
                if not self._running:
                    break
                yield msg.type, msg.data, msg.flags
        except DeviceDisconnected:
            return


class CaptureManager:
    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._skipped = 0
        self._log_interval = log_interval

    @property
    def active(self):
        return self._session_id is not None

    @property
    def session_id(self):
        return self._session_id

    @property
    def start_time(self):
        return self._start_time

    @property
    def latest_msg_id(self):
        return self._latest_msg_id

    @property
    def skipped(self):
        """Packets dropped because they were not APDU-shaped."""
        return self._skipped

    @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()

        mode = 'gsmtap' if isinstance(self._backend, GsmtapListener) else 'direct'
        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._skipped = 0
        self._backend.start()

        self._thread = threading.Thread(target=self._capture_loop, daemon=True)
        self._thread.start()

        self._stop_event.clear()
        _log(f'Capture started (mode={mode}, session={self._session_id})')
        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
        empty = self._db.count_messages(sid) == 0
        if empty:
            self._db.delete_session(sid)
            _log(f'Capture stopped: empty session discarded (session={sid})')
            return None
        self._db.close_session(sid)
        _log(f'Capture stopped: session={sid} messages={self._msg_count} '
             f'bytes={self._byte_count} skipped={self._skipped}')
        return sid

    def _capture_loop(self):
        try:
            for msg_type, data, flags in self._backend.iter_messages():
                if self._session_id is None:
                    break
                if msg_type == 'tpdu' and not looks_like_apdu(data):
                    # Not APDU-shaped at all: a mis-sampled or character-level
                    # GSMTAP stream would only pollute the decode.
                    self._skipped += 1
                    if self._skipped <= 1 or self._skipped % 100 == 0:
                        _log(f'Capture: skipped packet {self._skipped} that is '
                             f'not APDU-shaped — check the sniffer stream')
                    continue
                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:
                    _log(f'Capture: insert error, dropping msg: {e}')
                    continue
        except Exception:
            # The capture thread died unexpectedly (not a user stop).
            sid = self._session_id
            _log(f'ERROR: capture thread stopped on error (session={sid}):')
            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
            dropped = getattr(self._backend, 'dropped', 0)
            _log(f'Capture alive: session={self._session_id} '
                 f'messages={self._msg_count} bytes={self._byte_count} '
                 f'dropped={dropped} skipped={self._skipped}')