Fix dropped change flags + FIDI parsing

- capture.py: DirectSniffer/GsmtapListener yield (type, data, flags),
  CaptureManager passes flags to insert_message (change flags were
  previously dropped, leaving every change row with flags=0)
- protocol.py: add _parse_fidi (1-byte value) — FIDI was mis-mapped to
  _parse_change (4-byte flags), so FIDI messages were silently dropped
- database.py: pass flags into decode_sniff_msg
- decode.py: decode_change(flags) reads flags directly; decode_sniff_msg
  gains flags parameter; change messages no longer require non-empty data
- tests: fix FIDI test, add change/fidi decode tests (43 total)
This commit is contained in:
catarrh committed 2026-08-12 19:25:14 +03:00
1 parent 4097bb864e
commit 2a58fec0a3
7 files changed
+52 -20

No files matched your search

+4 -4
View File
@@ -24,7 +24,7 @@ class GsmtapListener:
if sub_type is None:
continue
msg_type = 'atr' if sub_type == GSMTAP_SIM_ATR else 'tpdu'
yield msg_type, data
yield msg_type, data, 0
class DirectSniffer:
@@ -49,7 +49,7 @@ class DirectSniffer:
for msg in self._session.iter_messages():
if not self._running:
break
yield msg.type, msg.data
yield msg.type, msg.data, msg.flags
class CaptureManager:
@@ -103,9 +103,9 @@ class CaptureManager:
return sid
def _capture_loop(self):
for msg_type, data in self._backend.iter_messages():
for msg_type, data, flags in self._backend.iter_messages():
if self._session_id is None:
break
elapsed = round(time.monotonic() - self._start_time, 3)
mid = self._db.insert_message(self._session_id, elapsed, msg_type, data)
mid = self._db.insert_message(self._session_id, elapsed, msg_type, data, flags)
self._latest_msg_id = mid
+4 -3
View File
@@ -168,18 +168,19 @@ class Database:
def _row_to_message(self, row):
data_blob = row[4]
flags = row[5]
msg = {
'id': row[0],
'session_id': row[1],
'elapsed': row[2],
'type': row[3],
'data': data_blob.hex(),
'flags': row[5],
'flags': flags,
}
if row[3] in ('tpdu', 'change', 'fidi', 'atr', 'pps') and data_blob:
if row[3] in ('tpdu', 'change', 'fidi', 'atr', 'pps'):
try:
from .decode import decode_sniff_msg
msg['decoded'] = decode_sniff_msg(data_blob, row[3])
msg['decoded'] = decode_sniff_msg(data_blob, row[3], flags)
except Exception as e:
import sys
print(f'Decode error for msg {row[0]} ({row[3]}, {len(data_blob)} bytes): {e}',
+5 -9
View File
@@ -590,12 +590,8 @@ CHANGE_FLAGS = {
}
def decode_change(raw_data):
"""Decode a SIMtrace2 sniff_change payload (4-byte flags)."""
if len(raw_data) < 4:
return None
import struct
flags = struct.unpack('<I', raw_data[:4])[0]
def decode_change(flags):
"""Decode SIMtrace2 sniff_change flags into a list of human-readable names."""
bits = []
for mask, name in sorted(CHANGE_FLAGS.items()):
if flags & mask:
@@ -628,7 +624,7 @@ def decode_fidi(raw_data):
}
def decode_sniff_msg(raw_data, msg_type):
def decode_sniff_msg(raw_data, msg_type, flags=0):
"""Decode a raw sniff message of the given type.
Returns a structured dict with at least a 'type' key, or None if
@@ -636,8 +632,8 @@ def decode_sniff_msg(raw_data, msg_type):
"""
if msg_type == 'tpdu' and raw_data:
return decode_message(raw_data)
if msg_type == 'change' and raw_data:
return decode_change(raw_data)
if msg_type == 'change':
return decode_change(flags)
if msg_type == 'fidi' and raw_data:
return decode_fidi(raw_data)
if msg_type == 'atr':