catarrh/simtrace-analyser
1
0 Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Initial commit: simtrace2-pysniff v0.1.0

Browse Source 
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).
This commit is contained in:
catarrh committed 2026-08-10 23:17:04 +03:00
commit 73b0277ef7
13 files changed
+1289
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.gitignore
+7
View File
@@ -0,0 +1,7 @@
__pycache__/
*.pyc
*.pyo
*.egg-info/
dist/
build/
.venv/
70-simtrace2-pysniff.rules
+12
View File
@@ -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"
README.md
+93
View File
@@ -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.
pyproject.toml
+20
View File
@@ -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*"]
simtrace2_pysniff/__init__.py
+21
View File
@@ -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__',
]
simtrace2_pysniff/__main__.py
+167
View File
@@ -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()
simtrace2_pysniff/device.py
+257
View File
@@ -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()
simtrace2_pysniff/dump.py
+106
View File
@@ -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,
}
simtrace2_pysniff/gsmtap.py
+61
View File
@@ -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()
simtrace2_pysniff/pcap.py
+63
View File
@@ -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 = '<IHHiIII'
_PCAP_GLOBAL_HDR_SIZE = struct.calcsize(_PCAP_GLOBAL_HDR_FMT)
_PCAP_PKT_HDR_FMT = '<IIII'
_PCAP_PKT_HDR_SIZE = struct.calcsize(_PCAP_PKT_HDR_FMT)
class PcapWriter:
"""Write sniffed messages as GSMTAP-encapsulated PCAP packets."""
def __init__(self, path):
self._f = open(path, 'wb')
self._write_global_header()
def _write_global_header(self):
hdr = struct.pack(
_PCAP_GLOBAL_HDR_FMT,
PCAP_MAGIC, # magic_number
2, # version_major
4, # version_minor
0, # thiszone (GMT)
0, # sigfigs
65535, # snaplen
LINKTYPE_GSMTAP, # network (LINKTYPE_GSMTAP)
)
self._f.write(hdr)
def write_packet(self, gsmtap_hdr, data, ts=None):
"""Write a single GSMTAP-encapsulated packet.
*gsmtap_hdr* is the 16-byte GSMTAP header (already packed).
*data* is the raw APDU/ATR bytes.
"""
if ts is None:
ts = time.time()
packet = gsmtap_hdr + data
ts_sec = int(ts)
ts_usec = int((ts - ts_sec) * 1_000_000)
pkt_hdr = struct.pack(
_PCAP_PKT_HDR_FMT,
ts_sec, # ts_sec
ts_usec, # ts_usec
len(packet), # incl_len
len(packet), # orig_len
)
self._f.write(pkt_hdr + packet)
def close(self):
self._f.close()
simtrace2_pysniff/protocol.py
+164
View File
@@ -0,0 +1,164 @@
"""Message parsing for SIMtrace2 USB sniffing protocol."""
import struct
from collections import namedtuple
MSG_HDR_FMT = '<BBBBHH'
MSG_HDR_SIZE = struct.calcsize(MSG_HDR_FMT) # 8 bytes
SIMTRACE_MSGC_SNIFF = 3
# --- Message types within SIMTRACE_MSGC_SNIFF ---
SIMTRACE_MSGT_SNIFF_CHANGE = 0
SIMTRACE_MSGT_SNIFF_FIDI = 1
SIMTRACE_MSGT_SNIFF_ATR = 2
SIMTRACE_MSGT_SNIFF_PPS = 3
SIMTRACE_MSGT_SNIFF_TPDU = 4
_TYPE_NAMES = {
SIMTRACE_MSGT_SNIFF_CHANGE: 'change',
SIMTRACE_MSGT_SNIFF_FIDI: 'fidi',
SIMTRACE_MSGT_SNIFF_ATR: 'atr',
SIMTRACE_MSGT_SNIFF_PPS: 'pps',
SIMTRACE_MSGT_SNIFF_TPDU: 'tpdu',
}
# --- Change flags ---
CHANGE_FLAG_CARD_INSERT = 1 << 0
CHANGE_FLAG_CARD_EJECT = 1 << 1
CHANGE_FLAG_RESET_ASSERT = 1 << 2
CHANGE_FLAG_RESET_DEASSERT = 1 << 3
CHANGE_FLAG_TIMEOUT_WT = 1 << 4
_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 flags ---
DATA_FLAG_ERROR_INCOMPLETE = 1 << 5
DATA_FLAG_ERROR_MALFORMED = 1 << 6
DATA_FLAG_ERROR_CHECKSUM = 1 << 7
DATA_FLAG_ERROR_OVERRUN = 1 << 8
DATA_FLAG_ERROR_FRAMING = 1 << 9
DATA_FLAG_ERROR_PARITY = 1 << 10
_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',
}
SniffMessage = namedtuple('SniffMessage', [
'type',
'type_hex',
'data',
'flags',
'slot_nr',
'seq_nr',
'timestamp',
])
def flags_to_names(value, mapping):
"""Convert a uint32 bitmask to a list of human-readable flag names."""
names = []
for bit, name in sorted(mapping.items()):
if value & bit:
names.append(name)
value &= ~bit
if value:
names.append(f'unknown(0x{value:08x})')
return names
def _parse_change(payload):
if len(payload) < 4:
return None
flags, = struct.unpack('<I', payload[:4])
consumed = 4
return consumed, flags, b''
_SNIFF_DATA_FMT = '<IH'
_SNIFF_DATA_HDR_SIZE = struct.calcsize(_SNIFF_DATA_FMT) # 6 bytes
def _parse_data(payload):
if len(payload) < _SNIFF_DATA_HDR_SIZE:
return None
flags, length = struct.unpack(_SNIFF_DATA_FMT, payload[:_SNIFF_DATA_HDR_SIZE])
total = _SNIFF_DATA_HDR_SIZE + length
if len(payload) < total:
return None
data = payload[_SNIFF_DATA_HDR_SIZE:total]
return total, flags, data
_PAYLOAD_PARSERS = {
SIMTRACE_MSGT_SNIFF_CHANGE: _parse_change,
SIMTRACE_MSGT_SNIFF_FIDI: _parse_change,
SIMTRACE_MSGT_SNIFF_ATR: _parse_data,
SIMTRACE_MSGT_SNIFF_PPS: _parse_data,
SIMTRACE_MSGT_SNIFF_TPDU: _parse_data,
}
def parse_message(buf, timestamp=None):
"""Try to extract one SniffMessage from the front of *buf*.
Returns ``(SniffMessage, consumed)`` on success, or ``(None, 0)`` on failure.
``consumed`` is the number of bytes to remove from the buffer.
"""
if len(buf) < MSG_HDR_SIZE:
return None, 0
msg_class, msg_type, seq_nr, slot_nr, _reserved, msg_len = \
struct.unpack(MSG_HDR_FMT, buf[:MSG_HDR_SIZE])
if msg_class != SIMTRACE_MSGC_SNIFF:
if msg_len < MSG_HDR_SIZE or msg_len > 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
sniff-start.sh
Executable
+83
View File
@@ -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[@]}"
tests/test_protocol.py
+235
View File
@@ -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('<H', buf, 6, 4) # set msg_len to 4 (< MSG_HDR_SIZE)
msg, consumed = parse_message(bytes(buf))
self.assertIsNone(msg)
self.assertEqual(consumed, MSG_HDR_SIZE)
def test_incomplete_message_needs_more_data(self):
buf = _build_hdr(SIMTRACE_MSGC_SNIFF, SIMTRACE_MSGT_SNIFF_ATR,
payload=b'\x01\x02\x03')
msg, consumed = parse_message(buf[:-2])
self.assertIsNone(msg)
self.assertEqual(consumed, 0)
def test_change_message_card_insert(self):
payload = struct.pack('<I', CHANGE_FLAG_CARD_INSERT)
buf = _build_hdr(SIMTRACE_MSGC_SNIFF, SIMTRACE_MSGT_SNIFF_CHANGE,
slot_nr=1, seq_nr=5, payload=payload)
msg, consumed = parse_message(buf)
self.assertEqual(consumed, len(buf))
self.assertEqual(msg.type, 'change')
self.assertEqual(msg.type_hex, '0x00')
self.assertEqual(msg.slot_nr, 1)
self.assertEqual(msg.seq_nr, 5)
self.assertEqual(msg.flags, CHANGE_FLAG_CARD_INSERT)
self.assertEqual(msg.data, b'')
def test_change_message_timeout(self):
payload = struct.pack('<I', CHANGE_FLAG_TIMEOUT_WT)
buf = _build_hdr(SIMTRACE_MSGC_SNIFF, SIMTRACE_MSGT_SNIFF_CHANGE,
payload=payload)
msg, consumed = parse_message(buf)
self.assertEqual(msg.type, 'change')
self.assertEqual(msg.flags, CHANGE_FLAG_TIMEOUT_WT)
def test_fidi_message(self):
payload = struct.pack('<I', 0x97) # Fi=9, Di=7
buf = _build_hdr(SIMTRACE_MSGC_SNIFF, SIMTRACE_MSGT_SNIFF_FIDI,
payload=payload)
msg, consumed = parse_message(buf)
self.assertEqual(msg.type, 'fidi')
self.assertEqual(len(msg.data), 0)
self.assertEqual(msg.flags, 0x97)
def test_atr_message(self):
atr = b'\x3b\x9e\x12\x34\x56\x78\x90'
payload = struct.pack('<IH', 0, len(atr)) + atr
buf = _build_hdr(SIMTRACE_MSGC_SNIFF, SIMTRACE_MSGT_SNIFF_ATR,
payload=payload)
msg, consumed = parse_message(buf)
self.assertEqual(consumed, len(buf))
self.assertEqual(msg.type, 'atr')
self.assertEqual(msg.data, atr)
self.assertEqual(msg.flags, 0)
def test_atr_message_with_flags(self):
atr = b'\x3b\x00'
flags = DATA_FLAG_ERROR_CHECKSUM | DATA_FLAG_ERROR_MALFORMED
payload = struct.pack('<IH', flags, len(atr)) + atr
buf = _build_hdr(SIMTRACE_MSGC_SNIFF, SIMTRACE_MSGT_SNIFF_ATR,
payload=payload)
msg, _ = parse_message(buf)
self.assertEqual(msg.flags, flags)
def test_pps_message(self):
pps = b'\xff\x10\x97\x7e'
payload = struct.pack('<IH', 0, len(pps)) + pps
buf = _build_hdr(SIMTRACE_MSGC_SNIFF, SIMTRACE_MSGT_SNIFF_PPS,
payload=payload)
msg, consumed = parse_message(buf)
self.assertEqual(msg.type, 'pps')
self.assertEqual(msg.data, pps)
def test_tpdu_message(self):
tpdu = b'\xa0\xa4\x00\x00\x02\x3f\x00'
payload = struct.pack('<IH', 0, len(tpdu)) + tpdu
buf = _build_hdr(SIMTRACE_MSGC_SNIFF, SIMTRACE_MSGT_SNIFF_TPDU,
payload=payload)
msg, _ = parse_message(buf)
self.assertEqual(msg.type, 'tpdu')
self.assertEqual(msg.data, tpdu)
def test_tpdu_flags_incomplete(self):
tpdu = b'\x00\xb0\x00\x00\x0a'
flags = DATA_FLAG_ERROR_INCOMPLETE
payload = struct.pack('<IH', flags, len(tpdu)) + tpdu
buf = _build_hdr(SIMTRACE_MSGC_SNIFF, SIMTRACE_MSGT_SNIFF_TPDU,
payload=payload)
msg, _ = parse_message(buf)
self.assertEqual(msg.flags, flags)
def test_timestamp_passed_through(self):
buf = _build_hdr(SIMTRACE_MSGC_SNIFF, SIMTRACE_MSGT_SNIFF_CHANGE,
payload=struct.pack('<I', 0))
msg, _ = parse_message(buf, timestamp=12345.678)
self.assertEqual(msg.timestamp, 12345.678)
def test_multiple_messages_in_buffer(self):
buf1 = _build_hdr(SIMTRACE_MSGC_SNIFF, SIMTRACE_MSGT_SNIFF_CHANGE,
payload=struct.pack('<I', 1))
atr_payload = struct.pack('<IH', 0, 2) + b'\x3b\x00'
buf2 = _build_hdr(SIMTRACE_MSGC_SNIFF, SIMTRACE_MSGT_SNIFF_ATR,
payload=atr_payload)
combined = buf1 + buf2
msg1, consumed1 = parse_message(combined)
self.assertIsNotNone(msg1)
self.assertEqual(msg1.type, 'change')
self.assertEqual(consumed1, len(buf1))
remaining = combined[consumed1:]
msg2, consumed2 = parse_message(remaining)
self.assertIsNotNone(msg2)
self.assertEqual(msg2.type, 'atr')
self.assertEqual(consumed2, len(buf2))
def test_trailing_garbage_before_valid_message(self):
msg1 = _build_hdr(0, 1, slot_nr=3, payload=b'card_emu_data!!!')
buf = msg1 + _build_hdr(SIMTRACE_MSGC_SNIFF, SIMTRACE_MSGT_SNIFF_CHANGE,
payload=struct.pack('<I', 1))
msg, consumed = parse_message(buf)
self.assertIsNone(msg)
self.assertEqual(consumed, len(msg1))
msg, consumed = parse_message(buf[len(msg1):])
self.assertIsNotNone(msg)
self.assertEqual(msg.type, 'change')
def test_non_sniff_msg_class_with_short_msg_len(self):
hdr = struct.pack(MSG_HDR_FMT, 0, 0, 0, 0, 0, 4)
msg, consumed = parse_message(hdr + b'\x00')
self.assertIsNone(msg)
self.assertEqual(consumed, MSG_HDR_SIZE)
def test_non_sniff_msg_class_known_len_skip(self):
payload = b'test_data_for_cardem'
buf = _build_hdr(1, 0, payload=payload)
msg, consumed = parse_message(buf)
self.assertIsNone(msg)
self.assertEqual(consumed, len(buf))
def test_zero_length_data(self):
payload = struct.pack('<IH', 0, 0)
buf = _build_hdr(SIMTRACE_MSGC_SNIFF, SIMTRACE_MSGT_SNIFF_ATR,
payload=payload)
msg, consumed = parse_message(buf)
self.assertEqual(msg.type, 'atr')
self.assertEqual(msg.data, b'')
self.assertEqual(msg.flags, 0)
def test_truncated_data_hdr(self):
buf = _build_hdr(SIMTRACE_MSGC_SNIFF, SIMTRACE_MSGT_SNIFF_TPDU,
payload=b'\x00\x00\x00\x00\x10')
msg, consumed = parse_message(buf)
self.assertIsNone(msg)
self.assertEqual(consumed, len(buf))
class TestFlagsToNames(unittest.TestCase):
def test_empty(self):
self.assertEqual(flags_to_names(0, {}), [])
def test_single(self):
names = flags_to_names(CHANGE_FLAG_CARD_INSERT, {
CHANGE_FLAG_CARD_INSERT: 'card inserted',
})
self.assertEqual(names, ['card inserted'])
def test_multiple(self):
names = flags_to_names(
CHANGE_FLAG_CARD_INSERT | CHANGE_FLAG_TIMEOUT_WT,
{
CHANGE_FLAG_CARD_INSERT: 'card inserted',
CHANGE_FLAG_TIMEOUT_WT: 'timeout',
}
)
self.assertEqual(names, ['card inserted', 'timeout'])
def test_unknown_bit_preserved(self):
names = flags_to_names(0xDEAD0000, {})
self.assertIn('unknown(0xdead0000)', names)
if __name__ == '__main__':
unittest.main()