From ee5cd8e1fe3c1d1d90528d62ce6b8335c04f4b2c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=90=D0=BD=D1=82=D0=BE=D0=BD=20=D0=A2=D1=80=D0=BE=D1=88?= =?UTF-8?q?=D0=B8=D0=BD?= Date: Wed, 19 Aug 2026 01:40:04 +0300 Subject: [PATCH] Resolve SFI addressing for record commands; track selection per logical channel --- frontend/index.html | 32 ++- simtrace2_pysniff/server/database.py | 118 ++++++----- simtrace2_pysniff/server/decode.py | 280 ++++++++++++++++++++++++--- tests/test_decode.py | 184 +++++++++++++++++- 4 files changed, 534 insertions(+), 80 deletions(-) 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 = `P20x${(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 = `P20x${(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):