Surface GSMTAP BAD_FCS desync flag end-to-end

sigrok-iso7816-stream sets GSMTAP_FLAG_BAD_FCS (0x01) in the GSMTAP
header `res` byte on desynced TPDUs (mis-framed exchanges kept for
forensics). Propagate that flag through the whole pipeline so such
packets are explicitly surfaced instead of decoding as garbage:

- gsmtap.py: GSMTAP_FLAG_BAD_FCS const; GsmtapReceiver.read_packet()
  now returns (sub_type, data, flags) from the res byte.
- server/capture.py: GsmtapListener forwards the flags to the DB.
- pcap.py: _parse_gsmtap_hdr/_scan_gsmtap/extract_gsmtap and
  parse_pcap/parse_pcapng now yield a (ts, type, payload, flags)
  tuple, so the flag survives a PCAP/PCAPNG round-trip.
- server/server.py: PCAP import stores the flags.
- server/decode.py: gsmtap_flag_names() maps bit 0x01 to
  'bad FCS (desynced)'; merged into decoded errors for tpdu/atr/pps
  and rst/vcc, so the frontend ⚠ badge + Errors row flag the entry.

User-visible: desynced TPDUs show a ⚠ "bad FCS (desynced)" marker
in the trace and detail views. Tests: +4 (decode BAD_FCS markers,
PCAP flag round-trip); existing tests updated for the new tuple
arities. Version 1.22.0, SW cache simtrace-analyser-v32.
This commit is contained in:
catarrh committed 2026-08-26 18:08:25 +03:00
1 parent 7eaf7e9b53
commit 66094a0114
13 files changed
+90 -35

No files matched your search

+5 -1
View File
@@ -119,7 +119,11 @@ passive SIM sniffer that emits the same GSMTAP-SIM format on UDP 4729. Its
custom sub-types for RST/VCC line events (`0x10`/`0x11`, added for custom sub-types for RST/VCC line events (`0x10`/`0x11`, added for
compatibility with that project) are decoded as `RESET ASSERTED/DE-ASSERTED` compatibility with that project) are decoded as `RESET ASSERTED/DE-ASSERTED`
and `VCC ON/OFF` timeline markers, and a reset-assert or power-removal and `VCC ON/OFF` timeline markers, and a reset-assert or power-removal
correctly invalidates the file-selection tracking. correctly invalidates the file-selection tracking. GSMTAP packets carrying
the `GSMTAP_FLAG_BAD_FCS` desync flag (set by that decoder on mis-framed
exchanges) are surfaced as a flagged/desynced entry in the capture timeline,
so such artefacts are easy to spot when reviewing a capture rather than
silently altering the decode.
Server options: Server options:
+1 -1
View File
@@ -2511,7 +2511,7 @@ function onFilterChange() {
} }
// ──────────────────── Versioning ──────────────────── // ──────────────────── Versioning ────────────────────
const PWA_VERSION = '1.21.0'; const PWA_VERSION = '1.22.0';
const MIN_SERVER_VERSION = '1.0.0'; const MIN_SERVER_VERSION = '1.0.0';
function parseVersion(v) { function parseVersion(v) {
+1 -1
View File
@@ -1,4 +1,4 @@
const CACHE = 'simtrace-analyser-v31'; const CACHE = 'simtrace-analyser-v32';
const PRECACHE = [ const PRECACHE = [
'index.html', 'index.html',
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.backends._legacy:_Backend"
[project] [project]
name = "simtrace2-pysniff" name = "simtrace2-pysniff"
version = "1.21.0" version = "1.22.0"
description = "Python-based sniffing tool for SIMtrace2 hardware (SIM card communication sniffer)" description = "Python-based sniffing tool for SIMtrace2 hardware (SIM card communication sniffer)"
readme = {file = "README.md", content-type = "text/markdown"} readme = {file = "README.md", content-type = "text/markdown"}
license = {text = "GPL-2.0-or-later"} license = {text = "GPL-2.0-or-later"}
+1 -1
View File
@@ -6,7 +6,7 @@ from .gsmtap import GsmtapSender, GsmtapReceiver
from .pcap import PcapWriter from .pcap import PcapWriter
from .dump import FileDumper, format_message from .dump import FileDumper, format_message
__version__ = '1.21.0' __version__ = '1.22.0'
__all__ = [ __all__ = [
'SniffSession', 'SniffSession',
'DeviceDisconnected', 'DeviceDisconnected',
+11 -5
View File
@@ -19,6 +19,10 @@ GSMTAP_SIM_PPS_RSP = 0x03
GSMTAP_SIM_RST_EVENT = 0x10 GSMTAP_SIM_RST_EVENT = 0x10
GSMTAP_SIM_VCC_EVENT = 0x11 GSMTAP_SIM_VCC_EVENT = 0x11
# Per-packet flags carried in the GSMTAP header `res` byte (libosmocore
# gsmtap.h). sigrok-iso7816-stream sets BAD_FCS on a desynced TPDU.
GSMTAP_FLAG_BAD_FCS = 0x01
GSMTAP_UDP_PORT = 4729 GSMTAP_UDP_PORT = 4729
_GSMTAP_HDR_FMT = '!BBBBHBBIBBBB' # 16 bytes, big-endian _GSMTAP_HDR_FMT = '!BBBBHBBIBBBB' # 16 bytes, big-endian
@@ -79,27 +83,29 @@ class GsmtapReceiver:
def read_packet(self): def read_packet(self):
"""Block until a GSMTAP-SIM packet arrives. """Block until a GSMTAP-SIM packet arrives.
Returns ``(sub_type, data)`` on success or ``(None, None)`` Returns ``(sub_type, data, flags)`` on success or ``(None, None, 0)``
on timeout. *sub_type* is ``GSMTAP_SIM_ATR`` or ``GSMTAP_SIM_APDU``. on timeout. *sub_type* is ``GSMTAP_SIM_ATR``/``GSMTAP_SIM_APDU`` or
a custom sub-type; *flags* are the per-packet bits from the GSMTAP
header ``res`` byte (see ``GSMTAP_FLAG_*``).
Raises ``ValueError`` for non-SIM or unknown GSMTAP packets. Raises ``ValueError`` for non-SIM or unknown GSMTAP packets.
""" """
try: try:
packet, addr = self._sock.recvfrom(65536) packet, addr = self._sock.recvfrom(65536)
except (socket.timeout, OSError): except (socket.timeout, OSError):
return None, None return None, None, 0
if len(packet) < _GSMTAP_HDR_SIZE: if len(packet) < _GSMTAP_HDR_SIZE:
raise ValueError(f'GSMTAP packet too short: {len(packet)} bytes') raise ValueError(f'GSMTAP packet too short: {len(packet)} bytes')
(version, hdr_len, pkt_type, _timeslot, _arfcn, _noise, _signal, (version, hdr_len, pkt_type, _timeslot, _arfcn, _noise, _signal,
_frame, sub_type, _antenna, sub_slot, _res) = \ _frame, sub_type, _antenna, sub_slot, res) = \
struct.unpack(_GSMTAP_HDR_FMT, packet[:_GSMTAP_HDR_SIZE]) struct.unpack(_GSMTAP_HDR_FMT, packet[:_GSMTAP_HDR_SIZE])
if pkt_type != GSMTAP_TYPE_SIM: if pkt_type != GSMTAP_TYPE_SIM:
raise ValueError(f'Not a GSMTAP SIM packet (type={pkt_type})') raise ValueError(f'Not a GSMTAP SIM packet (type={pkt_type})')
data = packet[_GSMTAP_HDR_SIZE:] data = packet[_GSMTAP_HDR_SIZE:]
return sub_type, data return sub_type, data, res
def close(self): def close(self):
self._sock.close() self._sock.close()
+12 -12
View File
@@ -224,22 +224,22 @@ def _frame_to_udp_payload(frame, linktype):
def _parse_gsmtap_hdr(data): def _parse_gsmtap_hdr(data):
"""Parse a GSMTAP header; return (sub_type, payload) or None if not SIM.""" """Parse a GSMTAP header; return (sub_type, payload, flags) or None."""
if len(data) < _GSMTAP_HDR_SIZE: if len(data) < _GSMTAP_HDR_SIZE:
return None return None
(version, hdr_len, pkt_type, _ts, _arfcn, _noise, _signal, (version, hdr_len, pkt_type, _ts, _arfcn, _noise, _signal,
_frame, sub_type, _ant, _slot, _res) = struct.unpack( _frame, sub_type, _ant, _slot, res) = struct.unpack(
_GSMTAP_HDR_FMT, data[:_GSMTAP_HDR_SIZE]) _GSMTAP_HDR_FMT, data[:_GSMTAP_HDR_SIZE])
if version != 0x02 or pkt_type != GSMTAP_TYPE_SIM: if version != 0x02 or pkt_type != GSMTAP_TYPE_SIM:
return None return None
hdr_size = hdr_len * 4 hdr_size = hdr_len * 4
if hdr_size < _GSMTAP_HDR_SIZE or len(data) < hdr_size: if hdr_size < _GSMTAP_HDR_SIZE or len(data) < hdr_size:
return None return None
return sub_type, data[hdr_size:] return sub_type, data[hdr_size:], res
def _scan_gsmtap(frame): def _scan_gsmtap(frame):
"""Fallback: scan a frame for a GSMTAP-SIM header, return (sub_type, payload).""" """Fallback: scan a frame for a GSMTAP-SIM header, return (sub_type, payload, flags)."""
for off in range(0, len(frame) - _GSMTAP_HDR_SIZE + 1): for off in range(0, len(frame) - _GSMTAP_HDR_SIZE + 1):
parsed = _parse_gsmtap_hdr(frame[off:]) parsed = _parse_gsmtap_hdr(frame[off:])
if parsed is not None: if parsed is not None:
@@ -250,8 +250,8 @@ def _scan_gsmtap(frame):
def extract_gsmtap(frame, linktype): def extract_gsmtap(frame, linktype):
"""Extract a GSMTAP-SIM message from a link-layer frame. """Extract a GSMTAP-SIM message from a link-layer frame.
Returns ``(sub_type, payload)`` or ``None`` if the frame carries no Returns ``(sub_type, payload, flags)`` or ``None`` if the frame carries
GSMTAP-SIM data. no GSMTAP-SIM data.
""" """
udp = _frame_to_udp_payload(frame, linktype) udp = _frame_to_udp_payload(frame, linktype)
if udp: if udp:
@@ -304,10 +304,10 @@ def parse_pcap(data):
parsed = extract_gsmtap(frame, linktype) parsed = extract_gsmtap(frame, linktype)
if parsed is None: if parsed is None:
continue continue
sub_type, payload = parsed sub_type, payload, flags = parsed
msg_type = _gsmtap_to_msg(sub_type, payload) msg_type = _gsmtap_to_msg(sub_type, payload)
if msg_type is not None: if msg_type is not None:
yield ts, msg_type, payload yield ts, msg_type, payload, flags
def parse_pcapng(data): def parse_pcapng(data):
@@ -354,10 +354,10 @@ def parse_pcapng(data):
ts = _scale_ts(ts_hi, ts_lo, tsresols.get(iface)) ts = _scale_ts(ts_hi, ts_lo, tsresols.get(iface))
parsed = extract_gsmtap(frame, linktypes.get(iface)) parsed = extract_gsmtap(frame, linktypes.get(iface))
if parsed is not None: if parsed is not None:
sub_type, payload = parsed sub_type, payload, flags = parsed
msg_type = _gsmtap_to_msg(sub_type, payload) msg_type = _gsmtap_to_msg(sub_type, payload)
if msg_type is not None: if msg_type is not None:
yield ts, msg_type, payload yield ts, msg_type, payload, flags
elif block_type == _PNG_SPB and len(body) >= 4: elif block_type == _PNG_SPB and len(body) >= 4:
cap_len, = struct.unpack(endian + 'I', body[:4]) cap_len, = struct.unpack(endian + 'I', body[:4])
if cap_len <= len(body) - 4: if cap_len <= len(body) - 4:
@@ -365,10 +365,10 @@ def parse_pcapng(data):
# SPB has no timestamp and no interface; use file-order ts. # SPB has no timestamp and no interface; use file-order ts.
parsed = _scan_gsmtap(frame) parsed = _scan_gsmtap(frame)
if parsed is not None: if parsed is not None:
sub_type, payload = parsed sub_type, payload, flags = parsed
msg_type = _gsmtap_to_msg(sub_type, payload) msg_type = _gsmtap_to_msg(sub_type, payload)
if msg_type is not None: if msg_type is not None:
yield float(spb_ts), msg_type, payload yield float(spb_ts), msg_type, payload, flags
spb_ts += 1 spb_ts += 1
+2 -2
View File
@@ -34,10 +34,10 @@ class GsmtapListener:
def iter_messages(self): def iter_messages(self):
while self._running: while self._running:
sub_type, data = self._receiver.read_packet() sub_type, data, flags = self._receiver.read_packet()
if sub_type is None: if sub_type is None:
continue continue
yield gsmtap_msg_type(sub_type), data, 0 yield gsmtap_msg_type(sub_type), data, flags
class DirectSniffer: class DirectSniffer:
+21 -4
View File
@@ -4030,6 +4030,19 @@ def decode_data_flags(flags):
return names or None return names or None
def gsmtap_flag_names(flags):
"""Decode the GSMTAP header ``res`` per-packet flag bits into a list of
human-readable names (or None).
GSMTAP_FLAG_BAD_FCS is set by the sigrok-iso7816-stream decoder on a
desynced TPDU (mis-framed exchange / implausible status word).
"""
names = []
if flags & 0x01:
names.append('bad FCS (desynced)')
return names or None
def decode_fidi(raw_data): def decode_fidi(raw_data):
"""Decode a SIMtrace2 sniff_fidi payload (1-byte Fi/Di).""" """Decode a SIMtrace2 sniff_fidi payload (1-byte Fi/Di)."""
if len(raw_data) < 1: if len(raw_data) < 1:
@@ -4274,27 +4287,31 @@ def decode_sniff_msg(raw_data, msg_type, flags=0, prev=None):
if msg_type == 'tpdu' and raw_data: if msg_type == 'tpdu' and raw_data:
result = decode_message(raw_data, prev=prev) result = decode_message(raw_data, prev=prev)
if result is not None: if result is not None:
errs = decode_data_flags(flags) errs = (decode_data_flags(flags) or []) + (gsmtap_flag_names(flags) or [])
if errs: if errs:
result['errors'] = errs result['errors'] = errs
return result return result
if msg_type == 'change': if msg_type == 'change':
return decode_change(flags) return decode_change(flags)
if msg_type in ('rst', 'vcc'): if msg_type in ('rst', 'vcc'):
return decode_line_event(raw_data, msg_type) result = decode_line_event(raw_data, msg_type)
errs = gsmtap_flag_names(flags)
if errs:
result['errors'] = errs
return result
if msg_type == 'fidi' and raw_data: if msg_type == 'fidi' and raw_data:
return decode_fidi(raw_data) return decode_fidi(raw_data)
if msg_type == 'atr': if msg_type == 'atr':
result = _decode_atr(raw_data) result = _decode_atr(raw_data)
if result is not None: if result is not None:
errs = decode_data_flags(flags) errs = (decode_data_flags(flags) or []) + (gsmtap_flag_names(flags) or [])
if errs: if errs:
result['errors'] = errs result['errors'] = errs
return result return result
if msg_type == 'pps': if msg_type == 'pps':
result = _decode_pps(raw_data) result = _decode_pps(raw_data)
if result is not None: if result is not None:
errs = decode_data_flags(flags) errs = (decode_data_flags(flags) or []) + (gsmtap_flag_names(flags) or [])
if errs: if errs:
result['errors'] = errs result['errors'] = errs
return result return result
+2 -2
View File
@@ -331,9 +331,9 @@ class RequestHandler(BaseHTTPRequestHandler):
first_ts = packets[0][0] first_ts = packets[0][0]
last_ts = packets[-1][0] last_ts = packets[-1][0]
rows = [] rows = []
for ts, msg_type, payload in packets: for ts, msg_type, payload, flags in packets:
elapsed = round(ts - first_ts, 6) if ts is not None else 0.0 elapsed = round(ts - first_ts, 6) if ts is not None else 0.0
rows.append((max(0.0, elapsed), msg_type, payload, 0)) rows.append((max(0.0, elapsed), msg_type, payload, flags))
self.db.insert_messages(session_id, rows) self.db.insert_messages(session_id, rows)
# Use the packet timestamps as the session window when they are real # Use the packet timestamps as the session window when they are real
# (classic pcap / pcapng EPB). pcapng SPB-only files have synthetic # (classic pcap / pcapng EPB). pcapng SPB-only files have synthetic
+16
View File
@@ -2340,6 +2340,22 @@ class TestLineEvents(unittest.TestCase):
self.assertEqual(decode_sniff_msg(b'\x01\x00\x00', 'rst')['type'], 'rst') self.assertEqual(decode_sniff_msg(b'\x01\x00\x00', 'rst')['type'], 'rst')
self.assertEqual(decode_sniff_msg(b'\x00\x01\x00', 'vcc')['type'], 'vcc') self.assertEqual(decode_sniff_msg(b'\x00\x01\x00', 'vcc')['type'], 'vcc')
def test_bad_fcs_flag_marks_tpdu(self):
from simtrace2_pysniff.server.decode import decode_sniff_msg
r = decode_sniff_msg(b'\x80\xf2\x00\x00\x00', 'tpdu', flags=0x01)
self.assertIn('bad FCS (desynced)', r['errors'])
def test_bad_fcs_flag_marks_rst(self):
from simtrace2_pysniff.server.decode import decode_sniff_msg
r = decode_sniff_msg(b'\x01\x00\x00', 'rst', flags=0x01)
self.assertIn('bad FCS (desynced)', r['errors'])
self.assertEqual(r['event'], 'reset asserted')
def test_no_fcs_flag_no_error(self):
from simtrace2_pysniff.server.decode import decode_sniff_msg
r = decode_sniff_msg(b'\x01\x00\x00', 'rst', flags=0)
self.assertNotIn('errors', r)
def test_short_payload(self): def test_short_payload(self):
r = self._dec('rst', '01') r = self._dec('rst', '01')
self.assertEqual(r['type'], 'rst') self.assertEqual(r['type'], 'rst')
+1 -1
View File
@@ -9,7 +9,7 @@ class TestGsmtapReceiver(unittest.TestCase):
def test_read_after_close_returns_none(self): def test_read_after_close_returns_none(self):
r = GsmtapReceiver(bind_port=0) r = GsmtapReceiver(bind_port=0)
r.close() r.close()
self.assertEqual(r.read_packet(), (None, None)) self.assertEqual(r.read_packet(), (None, None, 0))
if __name__ == '__main__': if __name__ == '__main__':
+16 -4
View File
@@ -50,7 +50,7 @@ class TestParsePcap(unittest.TestCase):
(_gsmtap_hdr(GSMTAP_SIM_APDU), b'\x80\xf2\x00\x00\x00', 2.5), (_gsmtap_hdr(GSMTAP_SIM_APDU), b'\x80\xf2\x00\x00\x00', 2.5),
] ]
data = build_pcap(packets) data = build_pcap(packets)
parsed = [(round(ts, 1), t, payload) for ts, t, payload in parse_pcap(data)] parsed = [(round(ts, 1), t, payload) for ts, t, payload, _f in parse_pcap(data)]
self.assertEqual(parsed, [ self.assertEqual(parsed, [
(1.0, 'atr', b'\x3b\x00'), (1.0, 'atr', b'\x3b\x00'),
(2.5, 'tpdu', b'\x80\xf2\x00\x00\x00'), (2.5, 'tpdu', b'\x80\xf2\x00\x00\x00'),
@@ -63,7 +63,7 @@ class TestParsePcap(unittest.TestCase):
(_gsmtap_hdr(GSMTAP_SIM_VCC_EVENT), b'\x00\x00\x00', 1.5), (_gsmtap_hdr(GSMTAP_SIM_VCC_EVENT), b'\x00\x00\x00', 1.5),
] ]
data = build_pcap(packets) data = build_pcap(packets)
parsed = [(round(ts, 1), t, payload) for ts, t, payload in parse_pcap(data)] parsed = [(round(ts, 1), t, payload) for ts, t, payload, _f in parse_pcap(data)]
self.assertEqual(parsed, [ self.assertEqual(parsed, [
(1.0, 'rst', b'\x01\x00\x00'), (1.0, 'rst', b'\x01\x00\x00'),
(1.5, 'vcc', b'\x00\x00\x00'), (1.5, 'vcc', b'\x00\x00\x00'),
@@ -77,6 +77,18 @@ class TestParsePcap(unittest.TestCase):
data = build_pcap(packets) data = build_pcap(packets)
self.assertEqual(list(parse_pcap(data)), []) self.assertEqual(list(parse_pcap(data)), [])
def test_bad_fcs_flag_roundtrip(self):
# GSMTAP_FLAG_BAD_FCS (res byte 0x01) survives a PCAP round-trip.
from simtrace2_pysniff.gsmtap import GSMTAP_FLAG_BAD_FCS
hdr = bytearray(_gsmtap_hdr(GSMTAP_SIM_APDU))
hdr[15] |= GSMTAP_FLAG_BAD_FCS
packets = [(bytes(hdr), b'\x80\xf2\x00\x00', 1.0),
(_gsmtap_hdr(GSMTAP_SIM_APDU), b'\x80\xf2\x00\x00', 2.0)]
data = build_pcap(packets)
parsed = list(parse_pcap(data))
self.assertEqual(parsed[0], (1.0, 'tpdu', b'\x80\xf2\x00\x00', GSMTAP_FLAG_BAD_FCS))
self.assertEqual(parsed[1], (2.0, 'tpdu', b'\x80\xf2\x00\x00', 0))
class TestParsePcapng(unittest.TestCase): class TestParsePcapng(unittest.TestCase):
def test_parse(self): def test_parse(self):
@@ -84,7 +96,7 @@ class TestParsePcapng(unittest.TestCase):
(_gsmtap_hdr(GSMTAP_SIM_ATR), b'\x3b\x04', 0, 0), (_gsmtap_hdr(GSMTAP_SIM_ATR), b'\x3b\x04', 0, 0),
(_gsmtap_hdr(GSMTAP_SIM_APDU), b'\x80\xf2\x00\x00\x00', 0, 1), (_gsmtap_hdr(GSMTAP_SIM_APDU), b'\x80\xf2\x00\x00\x00', 0, 1),
]) ])
parsed = [(t, payload) for _ts, t, payload in parse_pcapng(data)] parsed = [(t, payload) for _ts, t, payload, _f in parse_pcapng(data)]
self.assertEqual(parsed, [ self.assertEqual(parsed, [
('atr', b'\x3b\x04'), ('atr', b'\x3b\x04'),
('tpdu', b'\x80\xf2\x00\x00\x00'), ('tpdu', b'\x80\xf2\x00\x00\x00'),
@@ -212,7 +224,7 @@ class TestImportEndpoint(unittest.TestCase):
class TestVersion(unittest.TestCase): class TestVersion(unittest.TestCase):
def test_package_version(self): def test_package_version(self):
from simtrace2_pysniff import __version__ from simtrace2_pysniff import __version__
self.assertEqual(__version__, '1.21.0') self.assertEqual(__version__, '1.22.0')
def test_status_includes_version(self): def test_status_includes_version(self):
import os import os