commit 73b0277ef7f1db786be836ffb9f6c44efbe21d95 Author: Антон Трошин Date: Mon Aug 10 23:17:04 2026 +0300 Initial commit: simtrace2-pysniff v0.1.0 Python-based SIMtrace2 sniffer — drop-in replacement for simtrace2-sniff. Single dependency (pyusb), auto-reconnect, GSMTAP/PCAP output, library API. Modules: protocol.py — SIMtrace2 USB sniffing protocol parser device.py — USB discovery + SniffSession with exponential backoff reconnect gsmtap.py — GSMTAP UDP sender for Wireshark pcap.py — PCAP file writer (LINKTYPE_GSMTAP) dump.py — hex dump formatters + file output __main__.py — CLI entry point Startup script sniff-start.sh handles udev permission checks and shows install instructions. udev rule uses TAG+=uaccess (systemd-logind ACL). diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..c67e6fc --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +__pycache__/ +*.pyc +*.pyo +*.egg-info/ +dist/ +build/ +.venv/ diff --git a/70-simtrace2-pysniff.rules b/70-simtrace2-pysniff.rules new file mode 100644 index 0000000..a301716 --- /dev/null +++ b/70-simtrace2-pysniff.rules @@ -0,0 +1,12 @@ +# udev rules for SIMtrace2 USB sniffing (simtrace2-pysniff) +# +# Install: +# sudo cp 70-simtrace2-pysniff.rules /etc/udev/rules.d/ +# sudo udevadm control --reload-rules +# sudo udevadm trigger + +ACTION!="add", GOTO="pysniff_end" +SUBSYSTEM!="usb", GOTO="pysniff_end" +ATTRS{idVendor}=="1d50", ATTRS{idProduct}=="60e3", TAG+="uaccess" +ATTRS{idVendor}=="1d50", ATTRS{idProduct}=="616e", TAG+="uaccess" +LABEL="pysniff_end" diff --git a/README.md b/README.md new file mode 100644 index 0000000..1d0137a --- /dev/null +++ b/README.md @@ -0,0 +1,93 @@ +# simtrace2-pysniff + +Python-based replacement for `simtrace2-sniff` — SIM card communication sniffer +for Osmocom SIMtrace2 hardware (firmware in trace mode). + +**Single dependency**: PyUSB (libusb wrapper). No libosmocore, no libosmosim. + +## Quick Start + +```sh +# Install + run +pip install pyusb +python -m simtrace2_pysniff +``` + +Or use the bundled startup script: + +```sh +./sniff-start.sh # hex dump to stdout +./sniff-start.sh --gsmtap 127.0.0.1:4729 # + Wireshark GSMTAP +./sniff-start.sh --pcap trace.pcap # save PCAP for later +./sniff-start.sh --inactivity-timeout 30 # reconnect on silence +``` + +The startup script performs a pre-flight USB permission check and prints +install instructions if the udev rule is missing (see below). + +## USB Device Access + +### Linux + +A udev rule is required for non-root USB access to the SIMtrace2 device. + +```sh +sudo cp 70-simtrace2-pysniff.rules /etc/udev/rules.d/ +sudo udevadm control --reload-rules +sudo udevadm trigger +``` + +Then reconnect the SIMtrace2 USB cable. The rule uses `TAG+="uaccess"` +(systemd-logind ACL) — no group membership or world-writable permissions needed. + +### Windows + +Windows does not ship a libusb-compatible driver for SIMtrace2. Use +[Zadig](https://zadig.akeo.ie/) to replace the default driver: + +1. Download and run **Zadig** +2. Plug in the SIMtrace2 device +3. Select it in the dropdown (VID `1D50`, PID `60E3` — "Osmocom SIMtrace 2") +4. Choose **WinUSB** (or libusbK) as the replacement driver +5. Click **Replace Driver** (one-time operation) + +After that, `python -m simtrace2_pysniff` works identically to Linux. + +## CLI Options + +``` +python -m simtrace2_pysniff [OPTIONS] + + --gsmtap HOST[:PORT] Send ATR/TPDU as GSMTAP over UDP (default port 4729) + --pcap FILE Write sniffed data to PCAP file for Wireshark + --output, -o FILE Write hex dump to file + --no-stdout Suppress hex dump to stdout + --vendor HEX USB vendor ID (default: 0x1d50) + --product HEX USB product ID (default: auto-detect) + +Recovery options: + --no-reconnect Exit on USB disconnect instead of reconnecting + --inactivity-timeout SEC Reconnect after N seconds of silence (default: disabled) +``` + +## Library Usage + +```python +from simtrace2_pysniff import SniffSession + +session = SniffSession(inactivity_timeout=30.0) +for msg in session.iter_messages(): + print(f"{msg.type}: {msg.data.hex()}") +``` + +## Output + +- **stdout**: `ATR: 3b 9e ...`, `TPDU: a0 a4 00 00 02 3f 00`, `Card state change: reset de-asserted` +- **GSMTAP**: ATR (subtype 1) and TPDU (subtype 2) over UDP to Wireshark +- **PCAP**: all messages as GSMTAP-encapsulated packets (LINKTYPE 155), openable in Wireshark + +## Recovery + +The tool survives hardware resets, cable disconnects, and firmware hangs +by default — it reconnects with exponential backoff (1s → 30s cap). +Use `--inactivity-timeout` to also trigger reconnect on silent firmware hangs. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..a3601dd --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,20 @@ +[build-system] +requires = ["setuptools>=64"] +build-backend = "setuptools.backends._legacy:_Backend" + +[project] +name = "simtrace2-pysniff" +version = "0.1.0" +description = "Python-based sniffing tool for SIMtrace2 hardware (SIM card communication sniffer)" +readme = {file = "README.md", content-type = "text/markdown"} +license = {text = "GPL-2.0-or-later"} +requires-python = ">=3.9" +dependencies = [ + "pyusb>=1.2", +] + +[project.scripts] +simtrace2-pysniff = "simtrace2_pysniff.__main__:main" + +[tool.setuptools.packages.find] +include = ["simtrace2_pysniff*"] diff --git a/simtrace2_pysniff/__init__.py b/simtrace2_pysniff/__init__.py new file mode 100644 index 0000000..7954fa8 --- /dev/null +++ b/simtrace2_pysniff/__init__.py @@ -0,0 +1,21 @@ +"""simtrace2-pysniff — Python replacement for simtrace2-sniff.""" + +from .device import SniffSession, DeviceDisconnected, find_sniffer_device +from .protocol import parse_message, SniffMessage +from .gsmtap import GsmtapSender +from .pcap import PcapWriter +from .dump import FileDumper, format_message + +__version__ = '0.1.0' +__all__ = [ + 'SniffSession', + 'DeviceDisconnected', + 'find_sniffer_device', + 'parse_message', + 'SniffMessage', + 'GsmtapSender', + 'PcapWriter', + 'FileDumper', + 'format_message', + '__version__', +] diff --git a/simtrace2_pysniff/__main__.py b/simtrace2_pysniff/__main__.py new file mode 100644 index 0000000..a71bc32 --- /dev/null +++ b/simtrace2_pysniff/__main__.py @@ -0,0 +1,167 @@ +"""simtrace2-pysniff — CLI entry point. + +Usage: python -m simtrace2_pysniff [options] +""" + +import argparse +import signal +import sys +import time + +from . import ( + SniffSession, + DeviceDisconnected, + GsmtapSender, + PcapWriter, + FileDumper, + format_message, + __version__, +) +from .gsmtap import build_gsmtap_packet, GSMTAP_SIM_ATR, GSMTAP_SIM_APDU + + +def parse_gsmtap_addr(value): + parts = value.rsplit(':', 1) + host = parts[0] + port = 4729 + if len(parts) == 2: + try: + port = int(parts[1]) + except ValueError: + raise argparse.ArgumentTypeError( + f'Invalid GSMTAP port: {parts[1]}') + return host, port + + +def build_parser(): + p = argparse.ArgumentParser( + prog='simtrace2-pysniff', + description='Python-based SIMtrace2 sniffer with GSMTAP/PCAP output', + ) + p.add_argument('--version', action='version', + version=f'simtrace2-pysniff {__version__}') + p.add_argument('--gsmtap', type=parse_gsmtap_addr, metavar='HOST[:PORT]', + help='Send ATR/TPDU as GSMTAP over UDP (default port 4729)') + p.add_argument('--pcap', metavar='FILE', + help='Write sniffed data to PCAP file for Wireshark') + p.add_argument('--output', '-o', metavar='FILE', + help='Write hex dump to file') + p.add_argument('--no-stdout', action='store_true', + help='Suppress hex dump to stdout') + 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', + help='Exit on USB disconnect instead of reconnecting') + rec.add_argument('--reconnect-delay-min', type=float, default=1.0, + metavar='SECONDS', + help='Initial reconnect delay (default: 1.0)') + rec.add_argument('--reconnect-delay-max', type=float, default=30.0, + metavar='SECONDS', + help='Maximum reconnect delay (default: 30.0)') + rec.add_argument('--backoff-factor', type=float, default=1.5, + metavar='N', + help='Exponential backoff multiplier (default: 1.5)') + rec.add_argument('--inactivity-timeout', type=float, default=0.0, + metavar='SECONDS', + help='Reconnect after N seconds of silence (default: disabled)') + return p + + +def print_banner(): + print('simtrace2-pysniff — Phone-SIM card communication sniffer\n' + f'Version {__version__} — Python-based replacement for simtrace2-sniff\n', + file=sys.stderr) + print('Entering main loop', file=sys.stderr) + + +def main(): + parser = build_parser() + args = parser.parse_args() + + print_banner() + + gsmtap = None + if args.gsmtap: + host, port = args.gsmtap + gsmtap = GsmtapSender(host, port) + print(f'GSMTAP output: {host}:{port}', file=sys.stderr) + + pcap = None + if args.pcap: + pcap = PcapWriter(args.pcap) + print(f'PCAP output: {args.pcap}', file=sys.stderr) + + file_dumper = None + if args.output: + file_dumper = FileDumper(args.output) + print(f'File output: {args.output}', file=sys.stderr) + + session = SniffSession( + reconnect=not args.no_reconnect, + reconnect_delay_min=args.reconnect_delay_min, + reconnect_delay_max=args.reconnect_delay_max, + backoff_factor=args.backoff_factor, + inactivity_timeout=args.inactivity_timeout, + ) + + exiting = False + + def on_signal(signum, frame): + nonlocal exiting + if exiting: + raise SystemExit(1) + exiting = True + print('\nShutting down...', file=sys.stderr) + session.close() + if gsmtap: + gsmtap.close() + if pcap: + pcap.close() + if file_dumper: + file_dumper.close() + sys.exit(0) + + signal.signal(signal.SIGINT, on_signal) + signal.signal(signal.SIGTERM, on_signal) + + try: + for msg in session.iter_messages(): + if not args.no_stdout: + print(format_message(msg)) + + if file_dumper is not None: + file_dumper.write(msg) + + if gsmtap is not None and msg.type in ('atr', 'tpdu'): + if msg.type == 'atr': + gsmtap.send_atr(msg.data, msg.slot_nr) + else: + gsmtap.send_apdu(msg.data, msg.slot_nr) + + if pcap is not None: + sub_type = GSMTAP_SIM_ATR if msg.type == 'atr' else GSMTAP_SIM_APDU + gsmtap_hdr = build_gsmtap_packet(sub_type, b'')[:16] + pcap.write_packet(gsmtap_hdr, msg.data, msg.timestamp) + + except DeviceDisconnected as e: + print(f'Device disconnected: {e}', file=sys.stderr) + except KeyboardInterrupt: + pass + finally: + session.close() + if gsmtap: + gsmtap.close() + if pcap: + pcap.close() + if file_dumper: + file_dumper.close() + + +if __name__ == '__main__': + main() diff --git a/simtrace2_pysniff/device.py b/simtrace2_pysniff/device.py new file mode 100644 index 0000000..58f5cc4 --- /dev/null +++ b/simtrace2_pysniff/device.py @@ -0,0 +1,257 @@ +"""USB device discovery and sniffing session with auto-reconnect.""" + +import os +import sys +import time +import usb.core +import usb.util + +USB_VENDOR_OPENMOKO = 0x1d50 +USB_CLASS_PROPRIETARY = 0xff +SIMTRACE_SNIFFER_USB_SUBCLASS = 1 + +FATAL_USB_ERRNOS = { + 2, # ENOENT + 5, # EIO (some platforms) + 19, # ENODEV (Linux) + 32, # EPIPE / ESHUTDOWN (some platforms) + 108, # ESHUTDOWN (Linux) +} + + +class DeviceDisconnected(Exception): + """Raised when the USB device disappears or becomes unresponsive.""" + pass + + +def _is_fatal_usb_error(err): + if isinstance(err, usb.core.USBError): + if hasattr(err, 'errno') and err.errno is not None: + return err.errno in FATAL_USB_ERRNOS + if err.strerror and ('device' in err.strerror.lower() or + 'disconnect' in err.strerror.lower()): + return True + return False + + +def _usb_error_str(err): + if isinstance(err, usb.core.USBError): + return f'USBError(errno={getattr(err, "errno", "N/A")}, strerror={err.strerror})' + return str(err) + + +def find_sniffer_device(vendor_id=USB_VENDOR_OPENMOKO): + """Find a SIMtrace2 device running sniffer firmware. + + 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 + ), + )) + + for dev in devices: + for cfg in dev: + for intf in cfg: + cls = intf.bInterfaceClass + sub = intf.bInterfaceSubClass + if cls == USB_CLASS_PROPRIETARY and sub == SIMTRACE_SNIFFER_USB_SUBCLASS: + return dev + return None + + +def _get_ep_addrs(dev): + """Return (ep_out, ep_in, ep_irq) addresses from the active config.""" + cfg = dev.get_active_configuration() + for intf in cfg: + if (intf.bInterfaceClass == USB_CLASS_PROPRIETARY and + intf.bInterfaceSubClass == SIMTRACE_SNIFFER_USB_SUBCLASS): + for ep in intf: + addr = ep.bEndpointAddress + if addr & 0x80: # IN endpoint + return addr + return None + + +class SniffSession: + """Manages a SIMtrace2 sniffer USB connection with auto-reconnect. + + Usage:: + + session = SniffSession( + inactivity_timeout=10.0, + reconnect_delay_min=1.0, + reconnect_delay_max=30.0, + ) + try: + for msg in session.iter_messages(): + print(f"{msg.type}: {msg.data.hex()}") + except KeyboardInterrupt: + pass + finally: + session.close() + """ + + def __init__(self, *, + reconnect=True, + reconnect_delay_min=1.0, + reconnect_delay_max=30.0, + backoff_factor=1.5, + inactivity_timeout=0.0): + 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._dev = None + self._devh = None + self._ep_in = None + self._last_msg_time = 0.0 + self._disconnect_count = 0 + + def _ensure_connected(self): + """Find, open, and claim the sniffer device. + + Raises DeviceDisconnected if no device found or open fails. + """ + if self._devh is not None: + return + + dev = find_sniffer_device() + if dev is None: + raise DeviceDisconnected('No SIMtrace2 sniffer device found') + + try: + if dev.is_kernel_driver_active(0): + dev.detach_kernel_driver(0) + except (usb.core.USBError, NotImplementedError): + pass + + try: + devh = dev + devh.set_configuration() + usb.util.claim_interface(devh, 0) + except usb.core.USBError as e: + raise DeviceDisconnected( + f'Failed to open/claim device: {_usb_error_str(e)}') from e + + ep_in = _get_ep_addrs(devh) + if ep_in is None: + usb.util.release_interface(devh, 0) + raise DeviceDisconnected('No BULK IN endpoint found on sniffer interface') + + self._dev = dev + self._devh = devh + self._ep_in = ep_in + self._last_msg_time = time.monotonic() + self._disconnect_count = 0 + + print(f'Connected to SIMtrace2 sniffer (EP IN 0x{ep_in:02x})', + file=sys.stderr) + + def _close(self): + """Release USB resources gracefully.""" + if self._devh is not None: + try: + usb.util.release_interface(self._devh, 0) + except Exception: + pass + try: + usb.util.dispose_resources(self._devh) + except Exception: + pass + self._devh = None + self._dev = None + self._ep_in = None + + def _backoff_wait(self): + """Sleep with exponential backoff before reconnecting.""" + delay = self._reconnect_delay_min * (self._backoff_factor ** self._disconnect_count) + delay = min(delay, self._reconnect_delay_max) + self._disconnect_count += 1 + + print(f'Disconnected — retrying in {delay:.1f}s ' + f'(attempt {self._disconnect_count})', + file=sys.stderr) + time.sleep(delay) + + def _read_loop(self): + """Inner read loop. Yields SniffMessages. Raises DeviceDisconnected on error.""" + from .protocol import parse_message + + buf = bytearray() + max_buf = 1024 * 1024 # 1 MB + + while True: + try: + chunk = self._devh.read( + self._ep_in, 65536, timeout=1000) + chunk = bytes(chunk) + except usb.core.USBTimeoutError: + self._check_inactivity() + continue + except usb.core.USBError as e: + if _is_fatal_usb_error(e): + raise DeviceDisconnected( + f'USB read error: {_usb_error_str(e)}') from e + print(f'Transient USB error: {_usb_error_str(e)}', + file=sys.stderr) + time.sleep(0.1) + continue + except Exception as e: + print(f'Unexpected read error: {e}', file=sys.stderr) + time.sleep(0.1) + continue + + if not chunk: + self._check_inactivity() + continue + + self._last_msg_time = time.monotonic() + buf.extend(chunk) + + if len(buf) > max_buf: + print(f'Buffer overflow ({len(buf)} > {max_buf}), ' + f'discarding oldest data', file=sys.stderr) + buf = buf[-max_buf // 2:] + + while True: + msg, consumed = parse_message(buf, timestamp=time.time()) + if msg is not None: + yield msg + if consumed <= 0 or consumed > len(buf): + break + buf = buf[consumed:] + + def _check_inactivity(self): + if self._inactivity_timeout <= 0: + return + elapsed = time.monotonic() - self._last_msg_time + if elapsed > self._inactivity_timeout: + raise DeviceDisconnected( + f'Inactivity timeout ({elapsed:.1f}s > ' + f'{self._inactivity_timeout:.1f}s)') + + def iter_messages(self): + """Generator of SniffMessage objects. + + Automatically reconnects on disconnect if ``reconnect=True``. + Raises ``DeviceDisconnected`` if reconnect is disabled or exhausted. + """ + while True: + try: + self._ensure_connected() + yield from self._read_loop() + except DeviceDisconnected: + self._close() + if not self._reconnect: + raise + self._backoff_wait() + + def close(self): + self._reconnect = False + self._close() diff --git a/simtrace2_pysniff/dump.py b/simtrace2_pysniff/dump.py new file mode 100644 index 0000000..d85df9f --- /dev/null +++ b/simtrace2_pysniff/dump.py @@ -0,0 +1,106 @@ +"""Hex dump formatters for simtrace2 sniffing output.""" + +import sys + + +def hexdump(data, sep=' '): + """Return hex string representation of bytes. + + >>> hexdump(b'\\x3b\\x9e\\x12\\x34') + '3b 9e 12 34' + """ + return data.hex(sep) + + +def _flag_str(flags, flag_names): + if not flags: + return '' + names = [] + for bit, name in sorted(flag_names.items()): + if flags & bit: + names.append(name) + flags &= ~bit + if flags: + names.append(f'unknown(0x{flags:08x})') + return ', '.join(names) + + +def format_message(msg, *, show_flags=True): + """Format a SniffMessage as a human-readable line. + + Returns a str suitable for stdout or file output. + """ + from .protocol import ( + CHANGE_FLAG_CARD_INSERT, CHANGE_FLAG_CARD_EJECT, + CHANGE_FLAG_RESET_ASSERT, CHANGE_FLAG_RESET_DEASSERT, + CHANGE_FLAG_TIMEOUT_WT, + DATA_FLAG_ERROR_INCOMPLETE, DATA_FLAG_ERROR_MALFORMED, + DATA_FLAG_ERROR_CHECKSUM, DATA_FLAG_ERROR_OVERRUN, + DATA_FLAG_ERROR_FRAMING, DATA_FLAG_ERROR_PARITY, + ) + + change_flag_names = { + CHANGE_FLAG_CARD_INSERT: 'card inserted', + CHANGE_FLAG_CARD_EJECT: 'card ejected', + CHANGE_FLAG_RESET_ASSERT: 'reset asserted', + CHANGE_FLAG_RESET_DEASSERT: 'reset de-asserted', + CHANGE_FLAG_TIMEOUT_WT: 'waiting time timeout', + } + + data_flag_names = { + DATA_FLAG_ERROR_INCOMPLETE: 'incomplete', + DATA_FLAG_ERROR_MALFORMED: 'malformed', + DATA_FLAG_ERROR_CHECKSUM: 'checksum error', + DATA_FLAG_ERROR_OVERRUN: 'overrun', + DATA_FLAG_ERROR_FRAMING: 'framing error', + DATA_FLAG_ERROR_PARITY: 'parity error', + } + + type_label = msg.type.upper() + + if msg.type == 'change': + flags_str = _flag_str(msg.flags, change_flag_names) + line = f'Card state change: {flags_str}' if flags_str else 'Card state change: no changes' + elif msg.type == 'fidi': + fidi = msg.data[0] if len(msg.data) >= 1 else 0 + fi = fidi >> 4 + di = fidi & 0x0f + line = f'Fi/Di switched to {_fi_table.get(fi, fi)}/{_di_table.get(di, di)}' + else: + flags_str = '' + if show_flags and msg.flags: + flags_str = f' ({_flag_str(msg.flags, data_flag_names)})' + line = f'{type_label}{flags_str}: {hexdump(msg.data)}' + + return line + + +class FileDumper: + """Write formatted sniff messages to a file.""" + + def __init__(self, path): + self._f = open(path, 'a', buffering=1) + + def write(self, msg): + line = format_message(msg) + self._f.write(line + '\n') + + def close(self): + self._f.close() + + +# ISO 7816-3:2006 Table 7 +_fi_table = { + 0: 372, 1: 372, 2: 558, 3: 744, + 4: 1116, 5: 1488, 6: 1860, 7: 0, + 8: 0, 9: 512, 10: 768, 11: 1024, + 12: 1536, 13: 2048, 14: 0, 15: 0, +} + +# ISO 7816-3:2006 Table 8 +_di_table = { + 0: 0, 1: 1, 2: 2, 3: 4, + 4: 8, 5: 16, 6: 32, 7: 64, + 8: 12, 9: 20, 10: 2, 11: 4, + 12: 8, 13: 16, 14: 32, 15: 64, +} diff --git a/simtrace2_pysniff/gsmtap.py b/simtrace2_pysniff/gsmtap.py new file mode 100644 index 0000000..df22c39 --- /dev/null +++ b/simtrace2_pysniff/gsmtap.py @@ -0,0 +1,61 @@ +"""GSMTAP UDP sender for Wireshark integration. + +GSMTAP header format (big-endian, 16 bytes). Based on libosmocore's gsmtap.h. +""" + +import struct +import socket + +GSMTAP_VERSION = 0x02 +GSMTAP_HDR_LEN = 4 # in 32-bit words (16 bytes) +GSMTAP_TYPE_SIM = 0x07 + +GSMTAP_SIM_ATR = 0x01 +GSMTAP_SIM_APDU = 0x02 + +GSMTAP_UDP_PORT = 4729 + +_GSMTAP_HDR_FMT = '!BBBBHBBIBBBB' # 16 bytes, big-endian +_GSMTAP_HDR_SIZE = struct.calcsize(_GSMTAP_HDR_FMT) + + +def build_gsmtap_packet(sub_type, data, slot_nr=0): + """Build a GSMTAP packet as bytes. + + Returns *gsmtap_hdr* + *data* as a single bytes object. + """ + hdr = struct.pack( + _GSMTAP_HDR_FMT, + GSMTAP_VERSION, # version + GSMTAP_HDR_LEN, # hdr_len (in 32-bit words) + GSMTAP_TYPE_SIM, # type + 0, # timeslot + 0, # arfcn + 0, # noise_db + 0, # signal_db + 0, # frame_number + sub_type, # sub_type + 0, # antenna_nr + slot_nr, # sub_slot + 0, # res + ) + return hdr + data + + +class GsmtapSender: + """Send ATR/APDU data as GSMTAP UDP packets.""" + + def __init__(self, host='127.0.0.1', port=GSMTAP_UDP_PORT): + self._addr = (host, port) + self._sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + + def send_atr(self, data, slot_nr=0): + packet = build_gsmtap_packet(GSMTAP_SIM_ATR, data, slot_nr) + self._sock.sendto(packet, self._addr) + + def send_apdu(self, data, slot_nr=0): + packet = build_gsmtap_packet(GSMTAP_SIM_APDU, data, slot_nr) + self._sock.sendto(packet, self._addr) + + def close(self): + self._sock.close() diff --git a/simtrace2_pysniff/pcap.py b/simtrace2_pysniff/pcap.py new file mode 100644 index 0000000..a86b260 --- /dev/null +++ b/simtrace2_pysniff/pcap.py @@ -0,0 +1,63 @@ +"""PCAP file writer for Wireshark analysis. + +Writes a standard PCAP file with LINKTYPE_GSMTAP (155, 0x009b). +Each packet is a GSMTAP header + raw APDU/ATR data. +""" + +import struct +import time + +PCAP_MAGIC = 0xa1b2c3d4 +LINKTYPE_GSMTAP = 155 # 0x009b + +_PCAP_GLOBAL_HDR_FMT = ' 65535: + consumed = MSG_HDR_SIZE + else: + consumed = msg_len + return None, consumed + + payload_size = msg_len - MSG_HDR_SIZE + if payload_size < 0: + return None, MSG_HDR_SIZE + + if len(buf) < msg_len: + return None, 0 + + payload = buf[MSG_HDR_SIZE:msg_len] + type_name = _TYPE_NAMES.get(msg_type, f'unknown(0x{msg_type:02x})') + + data = b'' + flags = 0 + + parser = _PAYLOAD_PARSERS.get(msg_type) + if parser is not None: + result = parser(payload) + if result is None: + return None, msg_len + consumed_payload, flags, data = result + else: + data = payload + consumed_payload = payload_size + if consumed_payload != payload_size: + return None, msg_len + + return SniffMessage( + type=type_name, + type_hex=f'0x{msg_type:02x}', + data=data, + flags=flags, + slot_nr=slot_nr, + seq_nr=seq_nr, + timestamp=timestamp, + ), msg_len diff --git a/sniff-start.sh b/sniff-start.sh new file mode 100755 index 0000000..9415e95 --- /dev/null +++ b/sniff-start.sh @@ -0,0 +1,83 @@ +#!/bin/bash +# simtrace2-pysniff startup script +# +# Usage: ./sniff-start.sh [--gsmtap HOST[:PORT]] [--pcap FILE] [--output FILE] [--inactivity-timeout SECONDS] +# +# Options set via environment variables act as defaults; CLI flags override them. + +set -euo pipefail +shopt -s nullglob + +PROJECT_DIR="$(cd "$(dirname "$0")" && pwd)" + +GSMTAP="${GSMTAP:-}" +PCAP="${PCAP:-}" +OUTPUT="${OUTPUT:-}" +INACTIVITY_TIMEOUT="${INACTIVITY_TIMEOUT:-0}" + +while [[ $# -gt 0 ]]; do + case "$1" in + --gsmtap) GSMTAP="$2"; shift 2 ;; + --pcap) PCAP="$2"; shift 2 ;; + --output|-o) OUTPUT="$2"; shift 2 ;; + --inactivity-timeout) INACTIVITY_TIMEOUT="$2"; shift 2 ;; + --keep-running|--no-reconnect) + EXTRA_ARGS+=("$1"); shift ;; + *) EXTRA_ARGS+=("$1"); shift ;; + esac +done + +check_permissions() { + local bad_perms=0 + + for devdir in /sys/bus/usb/devices/*/; do + [[ -f "${devdir}busnum" ]] || continue + [[ -f "${devdir}devnum" ]] || continue + [[ -f "${devdir}uevent" ]] || continue + + local vid + vid=$(grep -oP 'PRODUCT=\K[a-f0-9]+' "${devdir}uevent" 2>/dev/null || true) + [[ "$vid" == "1d50" ]] || continue + + local busnum devnum node + busnum=$(<"${devdir}busnum") + devnum=$(<"${devdir}devnum") + node=$(printf '/dev/bus/usb/%03d/%03d' "$busnum" "$devnum") + + if [[ ! -e "$node" ]]; then + continue + fi + + if [[ -w "$node" ]]; then + return 0 + fi + + bad_perms=1 + echo "Permission denied on $node" >&2 + done + + if [[ $bad_perms -eq 1 ]]; then + echo "" >&2 + echo "Install the udev rule (then reconnect the device):" >&2 + echo "──────────────────────────────────────────────────" >&2 + echo " sudo cp '$PROJECT_DIR/70-simtrace2-pysniff.rules' /etc/udev/rules.d/" >&2 + echo " sudo udevadm control --reload-rules" >&2 + echo " sudo udevadm trigger" >&2 + echo "──────────────────────────────────────────────────" >&2 + return 1 + fi + + return 0 +} + +check_permissions || exit 3 + +ARGS=() +[[ -n "$GSMTAP" ]] && ARGS+=(--gsmtap "$GSMTAP") +[[ -n "$PCAP" ]] && ARGS+=(--pcap "$PCAP") +[[ -n "$OUTPUT" ]] && ARGS+=(--output "$OUTPUT") +[[ "$INACTIVITY_TIMEOUT" != "0" ]] && ARGS+=(--inactivity-timeout "$INACTIVITY_TIMEOUT") +ARGS+=("${EXTRA_ARGS[@]}") + +echo "Starting simtrace2-pysniff..." >&2 +exec env PYTHONPATH="$PROJECT_DIR" python3 -m simtrace2_pysniff "${ARGS[@]}" diff --git a/tests/test_protocol.py b/tests/test_protocol.py new file mode 100644 index 0000000..ffdf177 --- /dev/null +++ b/tests/test_protocol.py @@ -0,0 +1,235 @@ +"""Tests for protocol.py message parsing.""" + +import struct +import unittest + +from simtrace2_pysniff.protocol import ( + parse_message, + MSG_HDR_FMT, + MSG_HDR_SIZE, + MSG_HDR_SIZE, + flags_to_names, + CHANGE_FLAG_CARD_INSERT, + CHANGE_FLAG_TIMEOUT_WT, + DATA_FLAG_ERROR_CHECKSUM, + DATA_FLAG_ERROR_MALFORMED, + DATA_FLAG_ERROR_INCOMPLETE, + SIMTRACE_MSGC_SNIFF, + SIMTRACE_MSGT_SNIFF_CHANGE, + SIMTRACE_MSGT_SNIFF_FIDI, + SIMTRACE_MSGT_SNIFF_ATR, + SIMTRACE_MSGT_SNIFF_PPS, + SIMTRACE_MSGT_SNIFF_TPDU, +) + + +def _build_hdr(msg_class, msg_type, slot_nr=0, seq_nr=0, payload=b''): + msg_len = MSG_HDR_SIZE + len(payload) + return struct.pack(MSG_HDR_FMT, msg_class, msg_type, seq_nr, slot_nr, 0, msg_len) + payload + + +class TestParseMessage(unittest.TestCase): + def test_empty_buffer(self): + msg, consumed = parse_message(b'') + self.assertIsNone(msg) + self.assertEqual(consumed, 0) + + def test_too_short_for_header(self): + msg, consumed = parse_message(b'\x00' * 5) + self.assertIsNone(msg) + self.assertEqual(consumed, 0) + + def test_non_sniff_class_skipped(self): + payload = b'junk_data' + buf = _build_hdr(0, 0, payload=payload) + msg, consumed = parse_message(buf) + self.assertIsNone(msg) + self.assertEqual(consumed, len(buf)) + + def test_msg_len_smaller_than_header_skip(self): + buf = _build_hdr(SIMTRACE_MSGC_SNIFF, SIMTRACE_MSGT_SNIFF_CHANGE) + buf = bytearray(buf) + struct.pack_into('