Resolve SFI addressing for record commands; track selection per logical channel

This commit is contained in:
catarrh committed 2026-08-19 01:40:04 +03:00
1 parent 7b4f192a48
commit ee5cd8e1fe
4 files changed
+534 -80

No files matched your search

+27 -5
View File
@@ -700,7 +700,9 @@ function fileRows(f) {
if (!f) return ''; if (!f) return '';
const title = f.ef ? `${f.ef} (0x${(f.fid || '').toUpperCase()})` : (f.fid || '').toUpperCase(); const title = f.ef ? `${f.ef} (0x${(f.fid || '').toUpperCase()})` : (f.fid || '').toUpperCase();
const rows = []; 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)']); rows.push(['File', title + ' — selection stale (record op on non-record file)']);
} else if (f.empty) { } else if (f.empty) {
rows.push(['File', title + ' — 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.access_classes) rows.push(['Access classes', f.access_classes.join(', ')]);
if (f.phase) rows.push(['Phase', f.phase]); if (f.phase) rows.push(['Phase', f.phase]);
if (f.display_condition) rows.push(['Display condition', f.display_condition]); 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) 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.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.rules) rows.push(['Rules', f.rules.map(r => r.tag + ' ' + r.hex).join(', ')]);
if (f.raw) rows.push(['Data', f.raw]); if (f.raw) rows.push(['Data', f.raw]);
if (f.tpdu) { if (f.tpdu) {
@@ -883,8 +902,11 @@ function showDetail(msg, entry) {
} }
let p2 = ''; let p2 = '';
if (decoded.p2) { if (decoded.p2) {
const n = decoded.p2.name || (decoded.p2.bits ? decoded.p2.bits.join(', ') : ''); const parts = [];
p2 = `<tr><td class="label">P2</td><td class="value">0x${(decoded.p2.raw || '').toUpperCase()}${n ? ` — <span class="name">${n}</span>` : ''}</td></tr>`; 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 = `<tr><td class="label">P2</td><td class="value">0x${(decoded.p2.raw || '').toUpperCase()}${parts.length ? ` — <span class="name">${parts.join(', ')}</span>` : ''}</td></tr>`;
} }
let sw = ''; let sw = '';
@@ -1689,7 +1711,7 @@ const LANG_RU = {
'Click an APDU to see details': 'Нажмите на APDU для просмотра деталей', 'Click an APDU to see details': 'Нажмите на APDU для просмотра деталей',
'Command Stats': 'Статистика команд', 'Command Stats': 'Статистика команд',
'Stats': 'Статистика', 'Stats': 'Статистика',
'Map': 'Карта', 'Map': 'Обзор',
'Reads:': 'Чтение:', 'Reads:': 'Чтение:',
'Writes:': 'Запись:', 'Writes:': 'Запись:',
'Read/write:': 'Чтение/запись:', 'Read/write:': 'Чтение/запись:',
+73 -45
View File
@@ -171,20 +171,18 @@ class Database:
'SELECT id, session_id, elapsed, type, data, flags FROM messages ' 'SELECT id, session_id, elapsed, type, data, flags FROM messages '
'WHERE session_id=? ORDER BY elapsed, id LIMIT ? OFFSET ?', 'WHERE session_id=? ORDER BY elapsed, id LIMIT ? OFFSET ?',
(session_id, limit, offset)).fetchall() (session_id, limit, offset)).fetchall()
initial_prev = None initial_states = None
if offset > 0 and rows: if offset > 0 and rows:
initial_prev = self._last_tpdu_context(session_id, rows[0][0]) or {} initial_states = self._replay_context(session_id, rows[0][0])
initial_prev['sel'] = self._last_selection(session_id, rows[0][0]) return self._decode_rows(rows, initial_states=initial_states)
return self._decode_rows(rows, initial_prev=initial_prev)
def get_messages_after(self, session_id, after_id): def get_messages_after(self, session_id, after_id):
rows = self._conn.execute( rows = self._conn.execute(
'SELECT id, session_id, elapsed, type, data, flags FROM messages ' 'SELECT id, session_id, elapsed, type, data, flags FROM messages '
'WHERE session_id=? AND id > ? ORDER BY elapsed, id', 'WHERE session_id=? AND id > ? ORDER BY elapsed, id',
(session_id, after_id)).fetchall() (session_id, after_id)).fetchall()
initial_prev = self._last_tpdu_context(session_id, after_id + 1) or {} initial_states = self._replay_context(session_id, after_id + 1)
initial_prev['sel'] = self._last_selection(session_id, after_id + 1) return self._decode_rows(rows, initial_states=initial_states)
return self._decode_rows(rows, initial_prev=initial_prev)
def get_messages_raw(self, session_id): def get_messages_raw(self, session_id):
rows = self._conn.execute( rows = self._conn.execute(
@@ -242,41 +240,67 @@ class Database:
file=sys.stderr) file=sys.stderr)
return msg return msg
def _last_tpdu_context(self, session_id, before_id): def _new_channel_state(self):
"""Return previous-command context from the last TPDU with id < before_id.""" from .decode import sfi_table
row = self._conn.execute( return {'prev': None, 'sel': None, 'df': '3f00', 'sfi_map': dict(sfi_table('3f00'))}
"SELECT data FROM messages WHERE session_id=? AND id < ? AND type='tpdu' "
"ORDER BY id DESC LIMIT 1", (session_id, before_id)).fetchone() def _channel_state(self, states, channel):
if row is None: st = states.get(channel)
return None 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 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( rows = self._conn.execute(
"SELECT data, type FROM messages WHERE session_id=? AND id < ? " "SELECT data, type FROM messages WHERE session_id=? AND id < ? "
"AND type IN ('tpdu','atr','change') ORDER BY id DESC LIMIT 500", "AND type IN ('tpdu','atr','change') ORDER BY id DESC LIMIT 500",
(session_id, before_id)).fetchall() (session_id, before_id)).fetchall()
for data, typ in rows: states = {}
for data, typ in reversed(rows):
if typ in ('atr', 'change'): if typ in ('atr', 'change'):
return None states = {}
d = decode_sniff_msg(data, 'tpdu', 0) continue
if d and d.get('ins_hex') == 'a4': channel = data[0] & 0x03 if data and len(data) >= 1 else 0
fid = select_target_fid(d) st = self._channel_state(states, channel)
if fid: ctx = dict(st['prev']) if st['prev'] else {}
return {'fid': fid, 'name': KNOWN_FIDS.get(fid)} ctx['sel'] = st['sel']
if (d.get('p1') or {}).get('raw') in ('03', '04'): ctx['df'] = st['df']
return None ctx['sfi_map'] = st['sfi_map']
return None 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): def _context_from_decoded(self, d):
ins_hex = d.get('ins_hex') ins_hex = d.get('ins_hex')
@@ -317,20 +341,24 @@ class Database:
return new return new
return sel return sel
def _decode_rows(self, rows, initial_prev=None): def _decode_rows(self, rows, initial_states=None):
msgs = [] msgs = []
prev = initial_prev states = dict(initial_states or {})
sel = (initial_prev or {}).get('sel')
for row in rows: for row in rows:
ctx = dict(prev) if prev else {} channel = 0
ctx['sel'] = sel 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) msg = self._row_to_message(row, prev=ctx)
d = msg.get('decoded') d = msg.get('decoded')
if msg['type'] == 'tpdu' and d and d.get('ins_hex'): if msg['type'] == 'tpdu' and d and d.get('ins_hex'):
sel = self._selection_after(d, sel) st['sel'], st['df'], st['sfi_map'] = self._advance_state(d, st['sel'], st['df'], st['sfi_map'])
prev = self._context_from_decoded(d) st['prev'] = self._context_from_decoded(d)
elif msg['type'] in ('atr', 'change'): elif msg['type'] in ('atr', 'change'):
sel = None states = {}
prev = None
msgs.append(msg) msgs.append(msg)
return msgs return msgs
+251 -29
View File
@@ -133,6 +133,7 @@ KNOWN_FIDS = {
'2f00': 'EF_DIR', '2f00': 'EF_DIR',
'2f05': 'EF_PL', '2f05': 'EF_PL',
'2f06': 'EF_ARR', '2f06': 'EF_ARR',
'2f07': 'EF_ENV-CLASSES',
'2f08': 'EF_UMPC', '2f08': 'EF_UMPC',
'2fe2': 'EF_ICCID', '2fe2': 'EF_ICCID',
@@ -437,6 +438,91 @@ def select_target_fid(d):
return None 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 ──────────────────── # ──────────────────── Per-INS specifications ────────────────────
# TS 102 221 Table 9.3 — PIN mapping into key references (P2 of the PIN commands) # TS 102 221 Table 9.3 — PIN mapping into key references (P2 of the PIN commands)
@@ -482,12 +568,9 @@ APDU_SPEC = {
'p2': { 'p2': {
'label': 'Mode', 'label': 'Mode',
'bits': { 'bits': {
0x07: 'currently selected EF', 0x02: 'next record',
0x04: 'use record number from P1', 0x03: 'previous record',
0x02: 'previous record', 0x04: 'absolute mode (record number in P1)',
0x03: 'next record',
0x05: 'first record',
0x06: 'last record',
}, },
}, },
'body': None, 'body': None,
@@ -499,12 +582,9 @@ APDU_SPEC = {
'p2': { 'p2': {
'label': 'Mode', 'label': 'Mode',
'bits': { 'bits': {
0x07: 'currently selected EF', 0x02: 'next record',
0x04: 'use record number from P1', 0x03: 'previous record',
0x02: 'previous record', 0x04: 'absolute mode (record number in P1)',
0x03: 'next record',
0x05: 'first record',
0x06: 'last record',
}, },
}, },
'body': None, 'body': None,
@@ -520,12 +600,9 @@ APDU_SPEC = {
'p2': { 'p2': {
'label': 'Mode', 'label': 'Mode',
'bits': { 'bits': {
0x07: 'currently selected EF', 0x02: 'next record',
0x04: 'use record number from P1', 0x03: 'previous record',
0x02: 'previous record', 0x04: 'absolute mode (record number in P1)',
0x03: 'next record',
0x05: 'first record',
0x06: 'last record',
}, },
}, },
'body': {'label': 'Data'}, 'body': {'label': 'Data'},
@@ -1189,8 +1266,16 @@ def _decode_file_descriptor(value):
def _decode_fcp(data): 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 = {} result = {}
fid = None
sfi_seen = False
for tag, _length, value in parse_tlv(data): for tag, _length, value in parse_tlv(data):
if tag in (0x62, 0x64): # outer template — recurse if tag in (0x62, 0x64): # outer template — recurse
inner = _decode_fcp(value) inner = _decode_fcp(value)
@@ -1210,11 +1295,15 @@ def _decode_fcp(data):
elif tag == 0x81: elif tag == 0x81:
result['total_file_size'] = int.from_bytes(value, 'big') result['total_file_size'] = int.from_bytes(value, 'big')
elif tag == 0x88: 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: elif tag == 0x8A:
result['life_cycle'] = _LIFE_CYCLE.get(value[0] if value else 0, f'0x{(value[0] if value else 0):02X}') result['life_cycle'] = _LIFE_CYCLE.get(value[0] if value else 0, f'0x{(value[0] if value else 0):02X}')
elif tag == 0xAB: elif tag == 0xAB:
result['short_ef_id'] = value.hex().upper() 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 return result
@@ -1527,6 +1616,8 @@ def _decode_udh(udh):
iei = udh[i] iei = udh[i]
length = udh[i + 1] length = udh[i + 1]
data = udh[i + 2:i + 2 + length] data = udh[i + 2:i + 2 + length]
if len(data) < length:
break # truncated UDH
i += 2 + length i += 2 + length
el = {'iei': f'0x{iei:02X}', 'name': IEI_NAMES.get(iei), el = {'iei': f'0x{iei:02X}', 'name': IEI_NAMES.get(iei),
'length': length, 'hex': data.hex().upper()} 'length': length, 'hex': data.hex().upper()}
@@ -2501,7 +2592,8 @@ def _decode_spn(raw, p1=None):
def _decode_loci(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()} return {'raw': raw.hex().upper()}
tmsi = raw[0:4] tmsi = raw[0:4]
lai = raw[4:9] 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(), 'tmsi': None if all(b == 0xff for b in tmsi) else tmsi.hex().upper(),
'mcc': plmn.get('mcc'), 'mnc': plmn.get('mnc'), 'mcc': plmn.get('mcc'), 'mnc': plmn.get('mnc'),
'lac': '0x' + lai[3:5].hex().upper(), '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}')} 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): def _decode_nai(raw, p1=None):
"""ISIM identity files (EF_IMPI/DOMAIN/IMPU): BER-TLV tag 0x80 + ASCII value.""" """ISIM identity files (EF_IMPI/DOMAIN/IMPU): BER-TLV tag 0x80 + ASCII value."""
if not raw or all(b == 0xff for b in raw): 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()} 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): def _decode_hex(raw, p1=None):
if not raw or all(b == 0xff for b in raw): if not raw or all(b == 0xff for b in raw):
return {'empty': True} return {'empty': True}
@@ -2580,6 +2736,9 @@ FILE_DECODERS = {
'6fb7': _decode_ecc, '6fb7': _decode_ecc,
'6f46': _decode_spn, '6f46': _decode_spn,
'6f7e': _decode_loci, '6f7e': _decode_loci,
'6f73': _decode_psloci,
'6fe3': _decode_epsloci,
'6fe4': _decode_epsnsc,
'6f08': _decode_hex, '6f08': _decode_hex,
'6f09': _decode_hex, '6f09': _decode_hex,
'6f3e': _decode_hex, '6f3e': _decode_hex,
@@ -2644,6 +2803,8 @@ def _file_summary(f):
return 'record op on non-record file (selection stale?)' return 'record op on non-record file (selection stale?)'
if f.get('empty'): if f.get('empty'):
return 'empty' return 'empty'
if f.get('unknown'):
return 'unknown file'
if f.get('imsi'): if f.get('imsi'):
return f"IMSI {f['imsi']}" return f"IMSI {f['imsi']}"
if f.get('iccid'): if f.get('iccid'):
@@ -2673,8 +2834,21 @@ def _file_summary(f):
return f"SMS {s}".strip() return f"SMS {s}".strip()
if f.get('direction'): if f.get('direction'):
return f"SMS {f['direction']} — {f['status']}" 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'): 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: if f.get('access_classes') is not None:
return 'classes ' + ', '.join(str(c) for c in f['access_classes']) return 'classes ' + ', '.join(str(c) for c in f['access_classes'])
if f.get('phase'): if f.get('phase'):
@@ -2746,7 +2920,9 @@ def _build_summary(result):
p1txt = ', '.join(p1['bits']) p1txt = ', '.join(p1['bits'])
if p2: if p2:
p2txt = p2.get('name') 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']) p2txt = ', '.join(p2['bits'])
body = result.get('body') body = result.get('body')
@@ -2819,9 +2995,15 @@ def _build_summary(result):
file_dec = result.get('file') file_dec = result.get('file')
if file_dec: if file_dec:
txt = _file_summary(file_dec) if file_dec.get('unknown'):
if txt: parts.append('unknown file')
parts.append(f"{file_dec.get('ef', '')} {txt}".strip()) 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 return ', '.join(parts) if parts else None
@@ -2868,6 +3050,11 @@ def decode_message(raw_data, prev=None):
else: else:
result['p1'] = _decode_field(spec.get('p1', {}), p1) result['p1'] = _decode_field(spec.get('p1', {}), p1)
result['p2'] = _decode_field(spec.get('p2', {}), p2) 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 # Body: at least P3 bytes from byte 5; extra bytes may be SW
remaining = raw_data[5:] remaining = raw_data[5:]
@@ -2932,8 +3119,43 @@ def decode_message(raw_data, prev=None):
result['response'] = fcp result['response'] = fcp
# File data decode for READ/UPDATE using the current selection. # File data decode for READ/UPDATE using the current selection.
if ins in (0xB0, 0xB2, 0xD6, 0xDC) and prev and prev.get('sel'): p1p2 = result.get('p1p2') or {}
offset = (result.get('p1p2') or {}).get('offset', 0) 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) or offset == 0:
if ins in (0xB2, 0xDC) and not _is_record_file(prev['sel']): if ins in (0xB2, 0xDC) and not _is_record_file(prev['sel']):
result['file'] = _stale_file_note(prev['sel']) result['file'] = _stale_file_note(prev['sel'])
+183 -1
View File
@@ -247,6 +247,24 @@ class TestFcpResponse(unittest.TestCase):
self.assertEqual(fcp['file_size'], 9) self.assertEqual(fcp['file_size'], 9)
self.assertEqual(fcp['total_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): def test_fcp_summary_transparent(self):
# GET RESPONSE (FCP) for a transparent EF of 9 bytes # GET RESPONSE (FCP) for a transparent EF of 9 bytes
gr = bytes.fromhex('00c000000c') + bytes.fromhex('620c8202012183026f07800109810109') + bytes.fromhex('9000') gr = bytes.fromhex('00c000000c') + bytes.fromhex('620c8202012183026f07800109810109') + bytes.fromhex('9000')
@@ -866,7 +884,7 @@ class TestSummary(unittest.TestCase):
def test_read_record(self): def test_read_record(self):
r = decode_message(bytes.fromhex('a0b2010400')) 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): def test_verify_pin_no_leak(self):
# VERIFY PIN with P2 = 0x01 (PIN Appl 1): the PIN value must not leak. # 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']['da'], '999')
self.assertEqual(f['tpdu']['text'], 'Hi') 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): def test_read_binary_direct(self):
# SELECT EF_IMSI then READ BINARY → file decoded in body. # SELECT EF_IMSI then READ BINARY → file decoded in body.
r = decode_message(bytes.fromhex('00b0000009') + r = decode_message(bytes.fromhex('00b0000009') +
@@ -1155,6 +1213,130 @@ class TestSelectionTracking(unittest.TestCase):
read = msgs[2] read = msgs[2]
self.assertTrue(read['decoded']['file']['stale']) 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): class TestP1P2(unittest.TestCase):
def test_search_record_p2_modes(self): def test_search_record_p2_modes(self):