diff --git a/frontend/index.html b/frontend/index.html
index 794d4e7..85d8d40 100644
--- a/frontend/index.html
+++ b/frontend/index.html
@@ -700,7 +700,9 @@ function fileRows(f) {
if (!f) return '';
const title = f.ef ? `${f.ef} (0x${(f.fid || '').toUpperCase()})` : (f.fid || '').toUpperCase();
const rows = [];
- if (f.stale) {
+ if (f.unknown) {
+ rows.push(['File', `${f.ef || ('SFI ' + f.sfi)} — unresolved SFI reference`]);
+ } else if (f.stale) {
rows.push(['File', title + ' — selection stale (record op on non-record file)']);
} else if (f.empty) {
rows.push(['File', title + ' — empty']);
@@ -730,9 +732,26 @@ function fileRows(f) {
if (f.access_classes) rows.push(['Access classes', f.access_classes.join(', ')]);
if (f.phase) rows.push(['Phase', f.phase]);
if (f.display_condition) rows.push(['Display condition', f.display_condition]);
- if (f.mcc) rows.push(['Location', 'MCC ' + f.mcc + ' MNC ' + f.mnc + ' · LAC ' + f.lac]);
+ if (f.mcc) {
+ const parts = ['MCC ' + f.mcc + ' MNC ' + f.mnc];
+ if (f.lac) parts.push('LAC ' + f.lac);
+ if (f.tac) parts.push('TAC ' + f.tac);
+ if (f.rac) parts.push('RAC ' + f.rac);
+ rows.push(['Location', parts.join(' · ')]);
+ }
if (f.tmsi) rows.push(['TMSI', f.tmsi]);
+ if (f.tmsi_time) rows.push(['TMSI time', f.tmsi_time]);
if (f.location_update_status) rows.push(['Update status', f.location_update_status]);
+ if (f.guti) rows.push(['GUTI', f.guti]);
+ if (f.eps_update_status) rows.push(['EPS update status', f.eps_update_status]);
+ if (f.p_tmsi) rows.push(['P-TMSI', f.p_tmsi]);
+ if (f.p_tmsi_signature) rows.push(['P-TMSI signature', f.p_tmsi_signature]);
+ if (f.update_status) rows.push(['Update status', f.update_status]);
+ if (f.ksi_asme) rows.push(['KSI (ASME)', f.ksi_asme]);
+ if (f.k_asme) rows.push(['K_ASME', f.k_asme]);
+ if (f.uplink_nas_count !== undefined) rows.push(['UpLink NAS count', f.uplink_nas_count]);
+ if (f.downlink_nas_count !== undefined) rows.push(['DownLink NAS count', f.downlink_nas_count]);
+ if (f.nas_algorithms) rows.push(['NAS algorithms', f.nas_algorithms]);
if (f.rules) rows.push(['Rules', f.rules.map(r => r.tag + ' ' + r.hex).join(', ')]);
if (f.raw) rows.push(['Data', f.raw]);
if (f.tpdu) {
@@ -883,8 +902,11 @@ function showDetail(msg, entry) {
}
let p2 = '';
if (decoded.p2) {
- const n = decoded.p2.name || (decoded.p2.bits ? decoded.p2.bits.join(', ') : '');
- p2 = `| P2 | 0x${(decoded.p2.raw || '').toUpperCase()}${n ? ` — ${n}` : ''} |
`;
+ const parts = [];
+ if (decoded.p2.sfi) parts.push('SFI ' + decoded.p2.sfi);
+ const m = decoded.p2.name || (decoded.p2.bits ? decoded.p2.bits.join(', ') : '');
+ if (m) parts.push(m);
+ p2 = `| P2 | 0x${(decoded.p2.raw || '').toUpperCase()}${parts.length ? ` — ${parts.join(', ')}` : ''} |
`;
}
let sw = '';
@@ -1689,7 +1711,7 @@ const LANG_RU = {
'Click an APDU to see details': 'Нажмите на APDU для просмотра деталей',
'Command Stats': 'Статистика команд',
'Stats': 'Статистика',
- 'Map': 'Карта',
+ 'Map': 'Обзор',
'Reads:': 'Чтение:',
'Writes:': 'Запись:',
'Read/write:': 'Чтение/запись:',
diff --git a/simtrace2_pysniff/server/database.py b/simtrace2_pysniff/server/database.py
index ba98d70..8b10008 100644
--- a/simtrace2_pysniff/server/database.py
+++ b/simtrace2_pysniff/server/database.py
@@ -171,20 +171,18 @@ class Database:
'SELECT id, session_id, elapsed, type, data, flags FROM messages '
'WHERE session_id=? ORDER BY elapsed, id LIMIT ? OFFSET ?',
(session_id, limit, offset)).fetchall()
- initial_prev = None
+ initial_states = None
if offset > 0 and rows:
- initial_prev = self._last_tpdu_context(session_id, rows[0][0]) or {}
- initial_prev['sel'] = self._last_selection(session_id, rows[0][0])
- return self._decode_rows(rows, initial_prev=initial_prev)
+ initial_states = self._replay_context(session_id, rows[0][0])
+ return self._decode_rows(rows, initial_states=initial_states)
def get_messages_after(self, session_id, after_id):
rows = self._conn.execute(
'SELECT id, session_id, elapsed, type, data, flags FROM messages '
'WHERE session_id=? AND id > ? ORDER BY elapsed, id',
(session_id, after_id)).fetchall()
- initial_prev = self._last_tpdu_context(session_id, after_id + 1) or {}
- initial_prev['sel'] = self._last_selection(session_id, after_id + 1)
- return self._decode_rows(rows, initial_prev=initial_prev)
+ initial_states = self._replay_context(session_id, after_id + 1)
+ return self._decode_rows(rows, initial_states=initial_states)
def get_messages_raw(self, session_id):
rows = self._conn.execute(
@@ -242,41 +240,67 @@ class Database:
file=sys.stderr)
return msg
- def _last_tpdu_context(self, session_id, before_id):
- """Return previous-command context from the last TPDU with id < before_id."""
- row = self._conn.execute(
- "SELECT data FROM messages WHERE session_id=? AND id < ? AND type='tpdu' "
- "ORDER BY id DESC LIMIT 1", (session_id, before_id)).fetchone()
- if row is None:
- return None
+ def _new_channel_state(self):
+ from .decode import sfi_table
+ return {'prev': None, 'sel': None, 'df': '3f00', 'sfi_map': dict(sfi_table('3f00'))}
+
+ def _channel_state(self, states, channel):
+ st = states.get(channel)
+ if st is None:
+ st = self._new_channel_state()
+ states[channel] = st
+ return st
+
+ def _replay_context(self, session_id, before_id):
+ """Replay (bounded) messages before *before_id* to recover per-channel
+ selection, DF, and SFI map."""
from .decode import decode_sniff_msg
- d = decode_sniff_msg(row[0], 'tpdu', 0)
- if not d:
- return None
- return self._context_from_decoded(d)
-
- def _last_selection(self, session_id, before_id):
- """Replay (bounded, backwards) to find the current selection.
-
- Scans back from *before_id* to the last file-defining event:
- an ATR/card-change resets selection; a SELECT sets it.
- """
- from .decode import decode_sniff_msg, select_target_fid, KNOWN_FIDS
rows = self._conn.execute(
"SELECT data, type FROM messages WHERE session_id=? AND id < ? "
"AND type IN ('tpdu','atr','change') ORDER BY id DESC LIMIT 500",
(session_id, before_id)).fetchall()
- for data, typ in rows:
+ states = {}
+ for data, typ in reversed(rows):
if typ in ('atr', 'change'):
- return None
- d = decode_sniff_msg(data, 'tpdu', 0)
- if d and d.get('ins_hex') == 'a4':
- fid = select_target_fid(d)
- if fid:
- return {'fid': fid, 'name': KNOWN_FIDS.get(fid)}
- if (d.get('p1') or {}).get('raw') in ('03', '04'):
- return None
- return None
+ states = {}
+ continue
+ channel = data[0] & 0x03 if data and len(data) >= 1 else 0
+ st = self._channel_state(states, channel)
+ ctx = dict(st['prev']) if st['prev'] else {}
+ ctx['sel'] = st['sel']
+ ctx['df'] = st['df']
+ ctx['sfi_map'] = st['sfi_map']
+ d = decode_sniff_msg(data, 'tpdu', 0, prev=ctx)
+ if not d or not d.get('ins_hex'):
+ continue
+ st['sel'], st['df'], st['sfi_map'] = self._advance_state(d, st['sel'], st['df'], st['sfi_map'])
+ st['prev'] = self._context_from_decoded(d)
+ return states
+
+ def _advance_state(self, d, sel, df, sfi_map):
+ """Update (sel, df, sfi_map) for one decoded TPDU."""
+ from .decode import selected_df_fid, sfi_table, KNOWN_FIDS
+ new_df = selected_df_fid(d)
+ if new_df is not None and new_df != df:
+ # Only a successful SELECT changes the current DF.
+ if (d.get('sw') or {}).get('sw1') in _SUCCESS_SW1:
+ df = new_df
+ sfi_map = dict(sfi_table(new_df))
+ # Learn SFI from the FCP of the last SELECT (GET RESPONSE).
+ if d.get('ins_hex') == 'c0' and d.get('response_for') == 'SELECT' and sel:
+ sfi = (d.get('response') or {}).get('sfi')
+ if sfi is not None:
+ sfi_map = dict(sfi_map)
+ sfi_map[sfi] = sel['fid']
+ sel = self._selection_after(d, sel)
+ # A record/SEARCH command with a valid SFI sets the file as current EF
+ # (TS 102 221 §11.1.2).
+ sfi = (d.get('p2') or {}).get('sfi')
+ if d.get('ins_hex') in ('b2', 'dc', 'a2') and sfi:
+ fid = sfi_map.get(sfi)
+ if fid:
+ sel = {'fid': fid, 'name': KNOWN_FIDS.get(fid), 'structure': None}
+ return sel, df, sfi_map
def _context_from_decoded(self, d):
ins_hex = d.get('ins_hex')
@@ -317,20 +341,24 @@ class Database:
return new
return sel
- def _decode_rows(self, rows, initial_prev=None):
+ def _decode_rows(self, rows, initial_states=None):
msgs = []
- prev = initial_prev
- sel = (initial_prev or {}).get('sel')
+ states = dict(initial_states or {})
for row in rows:
- ctx = dict(prev) if prev else {}
- ctx['sel'] = sel
+ channel = 0
+ if row[3] == 'tpdu' and row[4] and len(row[4]) >= 5:
+ channel = row[4][0] & 0x03
+ st = self._channel_state(states, channel)
+ ctx = dict(st['prev']) if st['prev'] else {}
+ ctx['sel'] = st['sel']
+ ctx['df'] = st['df']
+ ctx['sfi_map'] = st['sfi_map']
msg = self._row_to_message(row, prev=ctx)
d = msg.get('decoded')
if msg['type'] == 'tpdu' and d and d.get('ins_hex'):
- sel = self._selection_after(d, sel)
- prev = self._context_from_decoded(d)
+ st['sel'], st['df'], st['sfi_map'] = self._advance_state(d, st['sel'], st['df'], st['sfi_map'])
+ st['prev'] = self._context_from_decoded(d)
elif msg['type'] in ('atr', 'change'):
- sel = None
- prev = None
+ states = {}
msgs.append(msg)
return msgs
diff --git a/simtrace2_pysniff/server/decode.py b/simtrace2_pysniff/server/decode.py
index 7cb74d9..0e5ce03 100644
--- a/simtrace2_pysniff/server/decode.py
+++ b/simtrace2_pysniff/server/decode.py
@@ -133,6 +133,7 @@ KNOWN_FIDS = {
'2f00': 'EF_DIR',
'2f05': 'EF_PL',
'2f06': 'EF_ARR',
+ '2f07': 'EF_ENV-CLASSES',
'2f08': 'EF_UMPC',
'2fe2': 'EF_ICCID',
@@ -437,6 +438,91 @@ def select_target_fid(d):
return None
+# Short File Identifier (SFI) → FID, scoped to the containing DF.
+# SFI is only unique within a DF (TS 102 221 §8.3). Sourced from the
+# normative SFI lists: TS 31.102 Annex H (USIM ADF, DF GSM-ACCESS) and
+# TS 102 221 Annex H (MF level).
+_SFI_BY_DF = {
+ '3f00': { # MF (TS 102 221 Annex H)
+ 0x02: '2fe2', # EF_ICCID
+ 0x05: '2f05', # EF_PL
+ 0x06: '2f06', # EF_ARR
+ 0x07: '2f07', # EF_ENV-CLASSES (TS 102 671)
+ 0x08: '2f08', # EF_UMPC
+ 0x1e: '2f00', # EF_DIR
+ },
+ '7fff': { # ADF_USIM (TS 31.102 Annex H H.1)
+ 0x01: '6fb7', # EF_ECC
+ 0x02: '6f05', # EF_LI
+ 0x03: '6fad', # EF_AD
+ 0x04: '6f38', # EF_UST
+ 0x05: '6f56', # EF_EST
+ 0x06: '6f78', # EF_ACC
+ 0x07: '6f07', # EF_IMSI
+ 0x08: '6f08', # EF_Keys
+ 0x09: '6f09', # EF_KeysPS
+ 0x0a: '6f60', # EF_PLMNwAcT
+ 0x0b: '6f7e', # EF_LOCI
+ 0x0c: '6f73', # EF_PSLOCI
+ 0x0d: '6f7b', # EF_FPLMN
+ 0x0e: '6f48', # EF_CBMID
+ 0x0f: '6f5b', # EF_START-HFN
+ 0x10: '6f5c', # EF_THRESHOLD
+ 0x11: '6f61', # EF_OPLMNwAcT
+ 0x12: '6f31', # EF_HPPLMN
+ 0x13: '6f62', # EF_HPLMNwAcT
+ 0x14: '6f80', # EF_ICI
+ 0x15: '6f81', # EF_OCI
+ 0x16: '6f4f', # EF_CCP2
+ 0x17: '6f06', # EF_ARR
+ 0x18: '6fe4', # EF_EPSNSC
+ 0x19: '6fc5', # EF_PNN
+ 0x1a: '6fc6', # EF_OPL
+ 0x1b: '6fcd', # EF_SPDI
+ 0x1c: '6f39', # EF_ACM
+ 0x1d: '6fd9', # EF_EHPLMN
+ 0x1e: '6fe3', # EF_EPSLOCI
+ },
+ '5f3b': { # DF_GSM_ACCESS (TS 31.102 Annex H H.2)
+ 0x01: '4f20', # EF_Kc
+ 0x02: '4f52', # EF_KcGPRS
+ },
+}
+
+
+def _is_df_fid(fid):
+ """True if *fid* (2-hex-byte string) denotes a DF/ADF (MF, 7Fxx, 5Fxx)."""
+ return fid == '3f00' or fid.startswith('7f') or fid.startswith('5f')
+
+
+def selected_df_fid(d):
+ """Return the current DF after a SELECT, or None if the SELECT does not
+ (re)enter a DF. Handles SELECT by FID (P1 00/01), by path (P1 08/09),
+ and by DF name/AID (P1 04)."""
+ if d.get('ins_hex') != 'a4':
+ return None
+ p1 = (d.get('p1') or {}).get('raw')
+ h = (d.get('body') or {}).get('hex') or ''
+ if p1 in ('00', '01'):
+ return h if len(h) == 4 and _is_df_fid(h) else None
+ if p1 in ('08', '09'):
+ comps = [h[i:i + 4] for i in range(0, len(h) - 3, 4)]
+ for comp in reversed(comps):
+ if _is_df_fid(comp):
+ return comp
+ return None
+ if p1 == '04': # select by DF name (AID) → map known application AIDs
+ if h.startswith('a0000000871002'): # USIM AID
+ return '7fff'
+ return None
+ return None
+
+
+def sfi_table(df):
+ """Return the SFI→FID table for a DF FID, or an empty dict."""
+ return _SFI_BY_DF.get(df or '', {})
+
+
# ──────────────────── Per-INS specifications ────────────────────
# TS 102 221 Table 9.3 — PIN mapping into key references (P2 of the PIN commands)
@@ -482,12 +568,9 @@ APDU_SPEC = {
'p2': {
'label': 'Mode',
'bits': {
- 0x07: 'currently selected EF',
- 0x04: 'use record number from P1',
- 0x02: 'previous record',
- 0x03: 'next record',
- 0x05: 'first record',
- 0x06: 'last record',
+ 0x02: 'next record',
+ 0x03: 'previous record',
+ 0x04: 'absolute mode (record number in P1)',
},
},
'body': None,
@@ -499,12 +582,9 @@ APDU_SPEC = {
'p2': {
'label': 'Mode',
'bits': {
- 0x07: 'currently selected EF',
- 0x04: 'use record number from P1',
- 0x02: 'previous record',
- 0x03: 'next record',
- 0x05: 'first record',
- 0x06: 'last record',
+ 0x02: 'next record',
+ 0x03: 'previous record',
+ 0x04: 'absolute mode (record number in P1)',
},
},
'body': None,
@@ -520,12 +600,9 @@ APDU_SPEC = {
'p2': {
'label': 'Mode',
'bits': {
- 0x07: 'currently selected EF',
- 0x04: 'use record number from P1',
- 0x02: 'previous record',
- 0x03: 'next record',
- 0x05: 'first record',
- 0x06: 'last record',
+ 0x02: 'next record',
+ 0x03: 'previous record',
+ 0x04: 'absolute mode (record number in P1)',
},
},
'body': {'label': 'Data'},
@@ -1189,8 +1266,16 @@ def _decode_file_descriptor(value):
def _decode_fcp(data):
- """Decode an FCP/FCI/FMD template (TS 102 221 §11.1.1.3)."""
+ """Decode an FCP/FCI/FMD template (TS 102 221 §11.1.1.3).
+
+ SFI (tag 0x88) is resolved per TS 102 221 §11.1.2:
+ - length 1 → SFI = value >> 3 (bits b8..b4);
+ - length 0 → file does not support SFI;
+ - absent → SFI = 5 LSBs of the FID (only if in range 1..30).
+ """
result = {}
+ fid = None
+ sfi_seen = False
for tag, _length, value in parse_tlv(data):
if tag in (0x62, 0x64): # outer template — recurse
inner = _decode_fcp(value)
@@ -1210,11 +1295,15 @@ def _decode_fcp(data):
elif tag == 0x81:
result['total_file_size'] = int.from_bytes(value, 'big')
elif tag == 0x88:
- result['sfi'] = f'0x{value[0]:02X}' if value else None
+ sfi_seen = True
+ result['sfi'] = value[0] >> 3 if value else None
elif tag == 0x8A:
result['life_cycle'] = _LIFE_CYCLE.get(value[0] if value else 0, f'0x{(value[0] if value else 0):02X}')
elif tag == 0xAB:
result['short_ef_id'] = value.hex().upper()
+ if not sfi_seen and fid:
+ sfi = int(fid, 16) & 0x1F
+ result['sfi'] = sfi if 1 <= sfi <= 30 else None
return result
@@ -1527,6 +1616,8 @@ def _decode_udh(udh):
iei = udh[i]
length = udh[i + 1]
data = udh[i + 2:i + 2 + length]
+ if len(data) < length:
+ break # truncated UDH
i += 2 + length
el = {'iei': f'0x{iei:02X}', 'name': IEI_NAMES.get(iei),
'length': length, 'hex': data.hex().upper()}
@@ -2501,7 +2592,8 @@ def _decode_spn(raw, p1=None):
def _decode_loci(raw, p1=None):
- if len(raw) < 13:
+ # EF_LOCI (TS 31.102 §4.2.16): TMSI 4 + LAI 5 + TMSI_TIME 1 + update status 1 = 11 bytes.
+ if len(raw) < 11:
return {'raw': raw.hex().upper()}
tmsi = raw[0:4]
lai = raw[4:9]
@@ -2510,7 +2602,8 @@ def _decode_loci(raw, p1=None):
'tmsi': None if all(b == 0xff for b in tmsi) else tmsi.hex().upper(),
'mcc': plmn.get('mcc'), 'mnc': plmn.get('mnc'),
'lac': '0x' + lai[3:5].hex().upper(),
- 'location_update_status': f'0x{raw[12]:02X}',
+ 'tmsi_time': f'0x{raw[9]:02X}',
+ 'location_update_status': f'0x{raw[10]:02X}',
}
@@ -2527,6 +2620,42 @@ def _decode_phase(raw, p1=None):
return {'phase': _PHASE_NAMES.get(v, f'0x{v:02X}')}
+def _decode_epsloci(raw, p1=None):
+ # EF_EPSLOCI (TS 31.102 §4.2.91): GUTI 12 + TAI 5 + update status 1 = 18 bytes.
+ if all(b == 0xff for b in raw):
+ return {'empty': True}
+ if len(raw) < 18:
+ return {'raw': raw.hex().upper()}
+ guti = raw[0:12]
+ tai = raw[12:17]
+ plmn = _decode_plmn(tai[:3]) or {}
+ return {
+ 'guti': None if all(b == 0xff for b in guti) else guti.hex().upper(),
+ 'mcc': plmn.get('mcc'), 'mnc': plmn.get('mnc'),
+ 'tac': '0x' + tai[3:5].hex().upper(),
+ 'eps_update_status': f'0x{raw[17]:02X}',
+ }
+
+
+def _decode_psloci(raw, p1=None):
+ # EF_PSLOCI (TS 31.102 §4.2.23): P-TMSI 4 + P-TMSI sig 3 + RAI 6 + status 1 = 14 bytes.
+ if all(b == 0xff for b in raw):
+ return {'empty': True}
+ if len(raw) < 14:
+ return {'raw': raw.hex().upper()}
+ ptmsi = raw[0:4]
+ rai = raw[7:13]
+ plmn = _decode_plmn(rai[:3]) or {}
+ return {
+ 'p_tmsi': None if all(b == 0xff for b in ptmsi) else ptmsi.hex().upper(),
+ 'p_tmsi_signature': raw[4:7].hex().upper(),
+ 'mcc': plmn.get('mcc'), 'mnc': plmn.get('mnc'),
+ 'lac': '0x' + rai[3:5].hex().upper(),
+ 'rac': '0x' + rai[5:6].hex().upper(),
+ 'update_status': f'0x{raw[13]:02X}',
+ }
+
+
def _decode_nai(raw, p1=None):
"""ISIM identity files (EF_IMPI/DOMAIN/IMPU): BER-TLV tag 0x80 + ASCII value."""
if not raw or all(b == 0xff for b in raw):
@@ -2543,6 +2672,33 @@ def _decode_nai(raw, p1=None):
return {'text': txt} if txt else {'raw': raw.hex().upper()}
+def _decode_epsnsc(raw, p1=None):
+ # EF_EPSNSC (TS 31.102 §4.2.92): record = A0 { 80 KSI_ASME, 81 K_ASME,
+ # 82 UpLink NAS COUNT, 83 DownLink NAS COUNT, 84 NAS algorithms }.
+ if not raw or all(b == 0xff for b in raw):
+ return {'empty': True}
+ inner = raw
+ for tag, _length, value in parse_tlv(raw):
+ if tag == 0xA0:
+ inner = value
+ break
+ out = {}
+ for tag, _length, value in parse_tlv(inner):
+ if tag == 0x80:
+ out['ksi_asme'] = value.hex().upper()
+ elif tag == 0x81:
+ out['k_asme'] = value.hex().upper()
+ elif tag == 0x82:
+ out['uplink_nas_count'] = int.from_bytes(value, 'big')
+ elif tag == 0x83:
+ out['downlink_nas_count'] = int.from_bytes(value, 'big')
+ elif tag == 0x84:
+ out['nas_algorithms'] = value.hex().upper()
+ if not out:
+ return {'raw': raw.hex().upper()}
+ return out
+
+
def _decode_hex(raw, p1=None):
if not raw or all(b == 0xff for b in raw):
return {'empty': True}
@@ -2580,6 +2736,9 @@ FILE_DECODERS = {
'6fb7': _decode_ecc,
'6f46': _decode_spn,
'6f7e': _decode_loci,
+ '6f73': _decode_psloci,
+ '6fe3': _decode_epsloci,
+ '6fe4': _decode_epsnsc,
'6f08': _decode_hex,
'6f09': _decode_hex,
'6f3e': _decode_hex,
@@ -2644,6 +2803,8 @@ def _file_summary(f):
return 'record op on non-record file (selection stale?)'
if f.get('empty'):
return 'empty'
+ if f.get('unknown'):
+ return 'unknown file'
if f.get('imsi'):
return f"IMSI {f['imsi']}"
if f.get('iccid'):
@@ -2673,8 +2834,21 @@ def _file_summary(f):
return f"SMS {s}".strip()
if f.get('direction'):
return f"SMS {f['direction']} — {f['status']}"
+ if f.get('guti'):
+ return f"GUTI {f['guti']} · TAC {f.get('tac', '')} · status {f.get('eps_update_status', '')}"
+ if f.get('p_tmsi'):
+ return f"P-TMSI {f['p_tmsi']} · status {f.get('update_status', '')}"
+ if 'ksi_asme' in f:
+ return f"KSI {f['ksi_asme']} · NAS counts {f.get('uplink_nas_count', '?')}/{f.get('downlink_nas_count', '?')}"
if f.get('mcc'):
- return f"MCC {f['mcc']} MNC {f['mnc']} · LAC {f.get('lac', '')}"
+ parts = [f"MCC {f['mcc']} MNC {f['mnc']}"]
+ if f.get('lac'):
+ parts.append(f"LAC {f['lac']}")
+ if f.get('tac'):
+ parts.append(f"TAC {f['tac']}")
+ if f.get('rac'):
+ parts.append(f"RAC {f['rac']}")
+ return ' · '.join(parts)
if f.get('access_classes') is not None:
return 'classes ' + ', '.join(str(c) for c in f['access_classes'])
if f.get('phase'):
@@ -2746,7 +2920,9 @@ def _build_summary(result):
p1txt = ', '.join(p1['bits'])
if p2:
p2txt = p2.get('name')
- if not p2txt and p2.get('bits'):
+ if p2.get('sfi'):
+ p2txt = f"SFI {p2['sfi']}"
+ elif not p2txt and p2.get('bits'):
p2txt = ', '.join(p2['bits'])
body = result.get('body')
@@ -2819,9 +2995,15 @@ def _build_summary(result):
file_dec = result.get('file')
if file_dec:
- txt = _file_summary(file_dec)
- if txt:
- parts.append(f"{file_dec.get('ef', '')} {txt}".strip())
+ if file_dec.get('unknown'):
+ parts.append('unknown file')
+ else:
+ ef = file_dec.get('ef') or file_dec.get('fid') or ''
+ txt = _file_summary(file_dec)
+ if txt:
+ parts.append(f"{ef} {txt}".strip())
+ elif ef:
+ parts.append(ef)
return ', '.join(parts) if parts else None
@@ -2868,6 +3050,11 @@ def decode_message(raw_data, prev=None):
else:
result['p1'] = _decode_field(spec.get('p1', {}), p1)
result['p2'] = _decode_field(spec.get('p2', {}), p2)
+ if ins in (0xB2, 0xDC, 0xA2):
+ # Record commands: P2 b8-b4 = SFI (TS 102 221 §11.1.5/6/7).
+ sfi = p2 >> 3
+ if sfi:
+ result['p2']['sfi'] = sfi
# Body: at least P3 bytes from byte 5; extra bytes may be SW
remaining = raw_data[5:]
@@ -2932,8 +3119,43 @@ def decode_message(raw_data, prev=None):
result['response'] = fcp
# File data decode for READ/UPDATE using the current selection.
- if ins in (0xB0, 0xB2, 0xD6, 0xDC) and prev and prev.get('sel'):
- offset = (result.get('p1p2') or {}).get('offset', 0)
+ p1p2 = result.get('p1p2') or {}
+ rec_sfi = (result.get('p2') or {}).get('sfi')
+ if ins in (0xB0, 0xD6) and prev and 'sfi' in p1p2:
+ # SFI referencing → resolve the target EF, not the last selection.
+ sfi = p1p2['sfi']
+ fid = (prev.get('sfi_map') or {}).get(sfi)
+ if fid:
+ file_dec = _decode_file_data(fid, body, p1=p1)
+ if file_dec:
+ result['file'] = file_dec
+ else:
+ result['file'] = {'fid': fid, 'ef': KNOWN_FIDS.get(fid, fid.upper()),
+ 'raw': body.hex().upper()}
+ else:
+ result['file'] = {'sfi': sfi, 'unknown': True, 'ef': f'SFI {sfi}'}
+ elif ins in (0xB2, 0xDC) and prev and rec_sfi:
+ # Record command with SFI (P2 b8-b4) → resolve the target EF.
+ fid = (prev.get('sfi_map') or {}).get(rec_sfi)
+ if fid:
+ file_dec = _decode_file_data(fid, body, p1=p1)
+ if file_dec:
+ result['file'] = file_dec
+ else:
+ result['file'] = {'fid': fid, 'ef': KNOWN_FIDS.get(fid, fid.upper()),
+ 'raw': body.hex().upper()}
+ else:
+ result['file'] = {'sfi': rec_sfi, 'unknown': True, 'ef': f'SFI {rec_sfi}'}
+ elif ins == 0xA2 and prev and rec_sfi:
+ # SEARCH RECORD with SFI → attach the target file name (the body
+ # is a search pattern, not file data).
+ fid = (prev.get('sfi_map') or {}).get(rec_sfi)
+ if fid:
+ result['file'] = {'fid': fid, 'ef': KNOWN_FIDS.get(fid, fid.upper())}
+ else:
+ result['file'] = {'sfi': rec_sfi, 'unknown': True, 'ef': f'SFI {rec_sfi}'}
+ elif ins in (0xB0, 0xB2, 0xD6, 0xDC) and prev and prev.get('sel'):
+ offset = p1p2.get('offset', 0)
if ins in (0xB2, 0xDC) or offset == 0:
if ins in (0xB2, 0xDC) and not _is_record_file(prev['sel']):
result['file'] = _stale_file_note(prev['sel'])
diff --git a/tests/test_decode.py b/tests/test_decode.py
index 8caee59..ebc35da 100644
--- a/tests/test_decode.py
+++ b/tests/test_decode.py
@@ -247,6 +247,24 @@ class TestFcpResponse(unittest.TestCase):
self.assertEqual(fcp['file_size'], 9)
self.assertEqual(fcp['total_file_size'], 9)
+ def test_fcp_sfi(self):
+ # FCP SFI tag 0x88 carries the SFI in bits b8..b4 (value >> 3).
+ from simtrace2_pysniff.server.decode import _decode_fcp
+ fcp = _decode_fcp(bytes.fromhex('8202412183026f7e880158'))
+ self.assertEqual(fcp['sfi'], 11)
+
+ def test_fcp_sfi_empty(self):
+ # FCP SFI tag 0x88 with length 0 → file does not support SFI.
+ from simtrace2_pysniff.server.decode import _decode_fcp
+ fcp = _decode_fcp(bytes.fromhex('82054221005a0d83026f068800'))
+ self.assertIsNone(fcp['sfi'])
+
+ def test_fcp_sfi_absent_uses_fid_lsbs(self):
+ # No SFI tag → SFI = 5 LSBs of the FID (TS 102 221 §11.1.2).
+ from simtrace2_pysniff.server.decode import _decode_fcp
+ fcp = _decode_fcp(bytes.fromhex('8202412183022f06'))
+ self.assertEqual(fcp['sfi'], 0x06)
+
def test_fcp_summary_transparent(self):
# GET RESPONSE (FCP) for a transparent EF of 9 bytes
gr = bytes.fromhex('00c000000c') + bytes.fromhex('620c8202012183026f07800109810109') + bytes.fromhex('9000')
@@ -866,7 +884,7 @@ class TestSummary(unittest.TestCase):
def test_read_record(self):
r = decode_message(bytes.fromhex('a0b2010400'))
- self.assertEqual(r['summary'], 'Record number: 1, use record number from P1')
+ self.assertEqual(r['summary'], 'Record number: 1, absolute mode (record number in P1)')
def test_verify_pin_no_leak(self):
# VERIFY PIN with P2 = 0x01 (PIN Appl 1): the PIN value must not leak.
@@ -1029,6 +1047,46 @@ class TestFileDecoders(unittest.TestCase):
self.assertEqual(f['tpdu']['da'], '999')
self.assertEqual(f['tpdu']['text'], 'Hi')
+ def test_loci(self):
+ from simtrace2_pysniff.server.decode import _decode_file_data
+ # EF_LOCI (11 bytes): TMSI 4 + LAI 5 + TMSI_TIME 1 + status 1.
+ f = _decode_file_data('6f7e', bytes.fromhex('1e731e5752f09969c50000'))
+ self.assertEqual(f['tmsi'], '1E731E57')
+ self.assertEqual(f['mcc'], '250')
+ self.assertEqual(f['mnc'], '99')
+ self.assertEqual(f['lac'], '0x69C5')
+ self.assertEqual(f['location_update_status'], '0x00')
+
+ def test_epsloci(self):
+ from simtrace2_pysniff.server.decode import _decode_file_data
+ # EF_EPSLOCI (18 bytes): GUTI 12 + TAI 5 + status 1.
+ f = _decode_file_data('6fe3', bytes.fromhex('0bf652f0998001b0fb91192652f0991d9900'))
+ self.assertEqual(f['guti'], '0BF652F0998001B0FB911926')
+ self.assertEqual(f['eps_update_status'], '0x00')
+
+ def test_psloci(self):
+ from simtrace2_pysniff.server.decode import _decode_file_data
+ # EF_PSLOCI (14 bytes): P-TMSI 4 + sig 3 + RAI 6 + status 1.
+ f = _decode_file_data('6f73', bytes.fromhex('1111111122222252f0991d990001'))
+ self.assertEqual(f['p_tmsi'], '11111111')
+ self.assertEqual(f['p_tmsi_signature'], '222222')
+ self.assertEqual(f['mcc'], '250')
+ self.assertEqual(f['mnc'], '99')
+ self.assertEqual(f['lac'], '0x1D99')
+ self.assertEqual(f['rac'], '0x00')
+ self.assertEqual(f['update_status'], '0x01')
+
+ def test_epsnsc(self):
+ from simtrace2_pysniff.server.decode import _decode_file_data
+ # EF_EPSNSC record: A0 { 80 KSI_ASME, 81 K_ASME, 82/83 NAS counts, 84 algos }.
+ f = _decode_file_data('6fe4', bytes.fromhex(
+ 'a0188001018104aabbccdd820400000003830400000004840102'))
+ self.assertEqual(f['ksi_asme'], '01')
+ self.assertEqual(f['k_asme'], 'AABBCCDD')
+ self.assertEqual(f['uplink_nas_count'], 3)
+ self.assertEqual(f['downlink_nas_count'], 4)
+ self.assertEqual(f['nas_algorithms'], '02')
+
def test_read_binary_direct(self):
# SELECT EF_IMSI then READ BINARY → file decoded in body.
r = decode_message(bytes.fromhex('00b0000009') +
@@ -1155,6 +1213,130 @@ class TestSelectionTracking(unittest.TestCase):
read = msgs[2]
self.assertTrue(read['decoded']['file']['stale'])
+ def test_sfi_update_resolves_eps_loci(self):
+ # SFI referencing must resolve the target EF (EF_EPSLOCI), not the
+ # last-selected file.
+ import tempfile
+ from simtrace2_pysniff.server.database import Database
+ with tempfile.NamedTemporaryFile(suffix='.db') as tmp:
+ db = Database(tmp.name)
+ sid = db.create_session('capture')
+ db.insert_message(sid, 0.0, 'atr', b'\x3b\x00')
+ db.insert_message(sid, 0.1, 'tpdu', bytes.fromhex('00a40804027fff612a'))
+ db.insert_message(sid, 0.2, 'tpdu',
+ bytes.fromhex('00d69e0012') +
+ bytes.fromhex('0bf652f0998001b0fb91192652f0991d9900') +
+ bytes.fromhex('9000'))
+ msgs = db.get_messages(sid)
+ up = msgs[2]
+ self.assertEqual(up['decoded']['file']['ef'], 'EF_EPSLOCI')
+ self.assertEqual(up['decoded']['file']['guti'], '0BF652F0998001B0FB911926')
+
+ def test_sfi_unknown_not_misattributed(self):
+ # An SFI not in the map must not fall back to the selected file.
+ import tempfile
+ from simtrace2_pysniff.server.database import Database
+ with tempfile.NamedTemporaryFile(suffix='.db') as tmp:
+ db = Database(tmp.name)
+ sid = db.create_session('capture')
+ db.insert_message(sid, 0.0, 'tpdu', bytes.fromhex('00a40804022fe2611e'))
+ db.insert_message(sid, 0.1, 'tpdu', bytes.fromhex('00d6890001079000'))
+ msgs = db.get_messages(sid)
+ up = msgs[1]
+ self.assertTrue(up['decoded']['file']['unknown'])
+ self.assertEqual(up['decoded']['file']['sfi'], 9)
+
+ def test_record_sfi_sets_current_ef(self):
+ # A record command with a valid SFI sets that file as the current EF,
+ # so a subsequent non-SFI record op targets it (TS 102 221 §11.1.2).
+ import tempfile
+ from simtrace2_pysniff.server.database import Database
+ with tempfile.NamedTemporaryFile(suffix='.db') as tmp:
+ db = Database(tmp.name)
+ sid = db.create_session('capture')
+ db.insert_message(sid, 0.0, 'atr', b'\x3b\x00')
+ db.insert_message(sid, 0.1, 'tpdu', bytes.fromhex('00a40804047fff6f7e611e'))
+ db.insert_message(sid, 0.2, 'tpdu', bytes.fromhex('00dc01c436') +
+ bytes.fromhex('a0348001018104aabbccdd820400000003830400000004840102') +
+ bytes.fromhex('9000'))
+ db.insert_message(sid, 0.3, 'tpdu', bytes.fromhex('00dc010436') +
+ bytes.fromhex('a0348001078104aabbccdd820400000004830400000003840102') +
+ bytes.fromhex('9000'))
+ msgs = db.get_messages(sid)
+ self.assertEqual(msgs[2]['decoded']['file']['ef'], 'EF_EPSNSC')
+ # 0.3 has P2=04 (no SFI) → must still target EF_EPSNSC.
+ self.assertEqual(msgs[3]['decoded']['file']['ef'], 'EF_EPSNSC')
+
+ def test_channel_selection_not_clobbered(self):
+ # A SELECT on channel 1 (CLA 01) must not change channel 0's selection.
+ import tempfile
+ from simtrace2_pysniff.server.database import Database
+ with tempfile.NamedTemporaryFile(suffix='.db') as tmp:
+ db = Database(tmp.name)
+ sid = db.create_session('capture')
+ db.insert_message(sid, 0.0, 'tpdu', bytes.fromhex('00a40804047fff6f409000'))
+ db.insert_message(sid, 0.1, 'tpdu', bytes.fromhex('01a40804047fff6f02611e'))
+ db.insert_message(sid, 0.2, 'tpdu',
+ bytes.fromhex('00b2010427') +
+ bytes.fromhex('8109089cbeb920bdbebcb5c0ffffffffffffffffffffffffff07919752738981f7ffffffffffff') +
+ bytes.fromhex('9000'))
+ msgs = db.get_messages(sid)
+ up = msgs[2]
+ # channel 0 still selects EF_MSISDN, not channel 1's EF_IMPI.
+ self.assertEqual(up['decoded']['file']['ef'], 'EF_MSISDN')
+
+
+class TestSfiResolution(unittest.TestCase):
+ def test_sfi_resolved(self):
+ r = decode_message(
+ bytes.fromhex('00d69e00120bf652f0998001b0fb91192652f0991d99009000'),
+ prev={'sfi_map': {30: '6fe3'}})
+ self.assertEqual(r['file']['ef'], 'EF_EPSLOCI')
+ self.assertEqual(r['file']['guti'], '0BF652F0998001B0FB911926')
+
+ def test_sfi_unknown(self):
+ r = decode_message(bytes.fromhex('00d6880001079000'), prev={'sfi_map': {}})
+ self.assertTrue(r['file']['unknown'])
+ self.assertEqual(r['file']['sfi'], 8)
+
+ def test_sfi_read_resolves(self):
+ # READ BINARY via SFI needs no prev['sel']; only the SFI map.
+ r = decode_message(bytes.fromhex('00b08b000b') +
+ bytes.fromhex('1e731e5752f09969c50000') + bytes.fromhex('9000'),
+ prev={'sfi_map': {11: '6f7e'}})
+ self.assertEqual(r['file']['ef'], 'EF_LOCI')
+ self.assertEqual(r['file']['tmsi'], '1E731E57')
+
+ def test_selected_df_fid(self):
+ from simtrace2_pysniff.server.decode import selected_df_fid, sfi_table
+ self.assertEqual(selected_df_fid({'ins_hex': 'a4', 'p1': {'raw': '08'},
+ 'body': {'hex': '3f007fff'}}), '7fff')
+ self.assertEqual(selected_df_fid({'ins_hex': 'a4', 'p1': {'raw': '00'},
+ 'body': {'hex': '7f20'}}), '7f20')
+ self.assertIsNone(selected_df_fid({'ins_hex': 'a4', 'p1': {'raw': '00'},
+ 'body': {'hex': '6f7e'}}))
+ self.assertIsNone(selected_df_fid({'ins_hex': 'b0'}))
+ self.assertEqual(sfi_table('7fff')[0x1e], '6fe3')
+ self.assertEqual(sfi_table('3f00')[0x1e], '2f00')
+ self.assertEqual(sfi_table('7fff')[0x17], '6f06')
+ self.assertEqual(sfi_table('7fff')[0x1a], '6fc6')
+ self.assertEqual(sfi_table('6f7e'), {})
+
+ def test_record_p2_sfi(self):
+ # READ RECORD P2 = 0xBC → SFI = 0xBC>>3 = 23, mode = 0x04.
+ r = decode_message(bytes.fromhex('00b204bc00'), prev={})
+ self.assertEqual(r['p2']['sfi'], 23)
+ self.assertIn('absolute mode', r['p2']['bits'][0])
+
+ def test_update_record_sfi_resolves(self):
+ # UPDATE RECORD P2 = 0xC4 → SFI 24 = EF_EPSNSC (record 54 B).
+ raw = bytes.fromhex('00dc01c436') + bytes.fromhex(
+ 'a0348001078120ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff'
+ '8204ffffffff8304ffffffff840100') + bytes.fromhex('9000')
+ r = decode_message(raw, prev={'sfi_map': {24: '6fe4'}})
+ self.assertEqual(r['file']['ef'], 'EF_EPSNSC')
+ self.assertEqual(r['file']['ksi_asme'], '07')
+
class TestP1P2(unittest.TestCase):
def test_search_record_p2_modes(self):