Add file-aware decoding for READ/UPDATE/SEARCH RECORD

- Track the currently selected file (SELECT by FID/path) across the
  session and pass it to the decoder; reset on ATR/card change; seed via
  a bounded backward scan on paged views.
- Decode READ BINARY/RECORD, UPDATE BINARY/RECORD and (via GET RESPONSE)
  SEARCH RECORD bodies using a FILE_DECODERS registry: IMSI, ICCID,
  LI/PL, ADN-format (MSISDN/SDN/FDN/...), SMS records (SMSC + TPDU),
  UST/SST/IST service tables, PLMN lists (+access tech), EF_DIR,
  CBMI/R, ARR, PNN, ECC, SPN, LOCI, ACC, Phase, ISIM IMPI/DOMAIN/IMPU.
- File formats verified against TS 51.011 / TS 102 221 / TS 31.102
  (PLMNwAcT is 5-byte entries; EF_SMS = status + SMSC + TPDU; ISIM
  identities are 0x80 TLV).
- Offset != 0 READ BINARY/UPDATE BINARY is skipped (fragment reads).
- SEARCH RECORD results (record numbers) decoded in GET RESPONSE.
- GET RESPONSE summary now shows FCP file type/structure/size.
- New 'File data' section in the detail view.
This commit is contained in:
catarrh committed 2026-08-15 23:11:31 +03:00
1 parent 143e1854e2
commit 91cb4e6e02
4 files changed
+854 -3

No files matched your search

+51
View File
@@ -630,6 +630,56 @@ function cmdRows(cmd) {
rows.map(([l, v]) => `<tr><td class="label">${l}</td><td class="value">${v}</td></tr>`).join(''); rows.map(([l, v]) => `<tr><td class="label">${l}</td><td class="value">${v}</td></tr>`).join('');
} }
function fileRows(f) {
if (!f) return '';
const title = f.ef ? `${f.ef} (0x${(f.fid || '').toUpperCase()})` : (f.fid || '').toUpperCase();
const rows = [];
if (f.empty) {
rows.push(['File', title + ' — empty']);
} else {
rows.push(['File', title]);
if (f.imsi) rows.push(['IMSI', f.imsi]);
if (f.iccid) rows.push(['ICCID', f.iccid]);
if (f.languages) rows.push(['Languages', f.languages.join(', ')]);
if (f.name !== undefined) rows.push(['Name', f.name]);
if (f.number) rows.push(['Number', f.number]);
if (f.text) rows.push(['Text', f.text]);
if (f.ccp !== undefined) rows.push(['CCP', '0x' + hx(f.ccp, 2)]);
if (f.ext1 !== undefined) rows.push(['Ext1', '0x' + hx(f.ext1, 2)]);
if (f.plmns) rows.push(['PLMNs', f.plmns.map(p => p.mcc + '/' + p.mnc + (p.access_tech ? ' (' + p.access_tech + ')' : '')).join(', ')]);
if (f.services) {
for (const s of f.services) rows.push(['Service ' + s.n, s.name || '']);
}
if (f.applications) {
for (const a of f.applications) rows.push(['App', (a.label ? a.label + ' · ' : '') + (a.aid || '')]);
}
if (f.direction) rows.push(['SMS', f.direction + ' — ' + f.status]);
if (f.message_ids) rows.push(['Message IDs', f.message_ids.join(', ')]);
if (f.record_numbers) rows.push(['Records found', f.record_numbers.join(', ')]);
if (f.range) rows.push(['Range', f.range[0] + '…' + f.range[1]]);
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.tmsi) rows.push(['TMSI', f.tmsi]);
if (f.location_update_status) rows.push(['Update status', f.location_update_status]);
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) {
const t = f.tpdu;
rows.push(['SMS TPDU', t.mti]);
if (t.oa) rows.push(['TP-OA', t.oa]);
if (t.da) rows.push(['TP-DA', t.da]);
if (t.text !== undefined) rows.push(['TP-UD', t.text]);
if (t.payload) rows.push(['TP-UD (payload)', t.payload]);
}
}
if (!rows.length) return '';
return `<tr><td class="label" colspan="2" style="padding-top:6px; border-top:1px solid var(--border);">File data</td></tr>` +
rows.map(([l, v]) => `<tr><td class="label">${l}</td><td class="value">${v}</td></tr>`).join('');
}
function ifaceValue(ifc) { function ifaceValue(ifc) {
const v = []; const v = [];
if (ifc.f !== undefined) v.push('F=' + ifc.f); if (ifc.f !== undefined) v.push('F=' + ifc.f);
@@ -795,6 +845,7 @@ function showDetail(msg, entry) {
${body} ${body}
${cmdRows(decoded.cmd)} ${cmdRows(decoded.cmd)}
${responseRows(decoded.response)} ${responseRows(decoded.response)}
${fileRows(decoded.file)}
${sw} ${sw}
</table> </table>
</div>`; </div>`;
+51 -3
View File
@@ -168,7 +168,8 @@ class Database:
(session_id, limit, offset)).fetchall() (session_id, limit, offset)).fetchall()
initial_prev = None initial_prev = None
if offset > 0 and rows: if offset > 0 and rows:
initial_prev = self._last_tpdu_context(session_id, rows[0][0]) 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) 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):
@@ -176,7 +177,8 @@ 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=? 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) 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) return self._decode_rows(rows, initial_prev=initial_prev)
def get_messages_raw(self, session_id): def get_messages_raw(self, session_id):
@@ -248,22 +250,68 @@ class Database:
return None return None
return self._context_from_decoded(d) 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:
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
def _context_from_decoded(self, d): def _context_from_decoded(self, d):
ins_hex = d.get('ins_hex') ins_hex = d.get('ins_hex')
ins = int(ins_hex, 16) if ins_hex else None ins = int(ins_hex, 16) if ins_hex else None
file_ok = False
if ins in (0xB2, 0xDC, 0xA2):
file_ok = True
elif ins in (0xB0, 0xD6):
file_ok = (d.get('p1p2') or {}).get('value', 0) == 0
return { return {
'ins': ins, 'ins': ins,
'ins_name': d.get('ins_name'), 'ins_name': d.get('ins_name'),
'sw1': (d.get('sw') or {}).get('sw1'), 'sw1': (d.get('sw') or {}).get('sw1'),
'file_ok': file_ok,
} }
def _selection_after(self, d, sel):
if d.get('ins_hex') == 'a4':
from .decode import select_target_fid, KNOWN_FIDS
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 sel
def _decode_rows(self, rows, initial_prev=None): def _decode_rows(self, rows, initial_prev=None):
msgs = [] msgs = []
prev = initial_prev prev = initial_prev
sel = (initial_prev or {}).get('sel')
for row in rows: for row in rows:
msg = self._row_to_message(row, prev=prev) ctx = dict(prev) if prev else {}
ctx['sel'] = sel
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)
prev = self._context_from_decoded(d) prev = self._context_from_decoded(d)
elif msg['type'] in ('atr', 'change'):
sel = None
prev = None
msgs.append(msg) msgs.append(msg)
return msgs return msgs
+588
View File
@@ -419,6 +419,24 @@ def _decode_select_path(body_hex):
return '/'.join(parts) return '/'.join(parts)
def select_target_fid(d):
"""Return the FID targeted by a decoded SELECT message, or None.
Used by the server's selection tracking to know which file is
currently selected (so READ/UPDATE bodies can be decoded).
"""
p1 = (d.get('p1') or {}).get('raw')
h = (d.get('body') or {}).get('hex') or ''
if p1 in ('00', '01'): # select by FID / child DF
return h if len(h) == 4 else None
if p1 in ('08', '09'): # select by path
if len(h) >= 4:
last = h[-4:]
return None if last == '7fff' else last
return None
return None
# ──────────────────── Per-INS specifications ──────────────────── # ──────────────────── Per-INS specifications ────────────────────
APDU_SPEC = { APDU_SPEC = {
@@ -1090,6 +1108,10 @@ def _decode_fcp(data):
result['file_id_name'] = KNOWN_FIDS[fid.lower()] result['file_id_name'] = KNOWN_FIDS[fid.lower()]
elif tag == 0x84: elif tag == 0x84:
result['df_name'] = value.hex().upper() result['df_name'] = value.hex().upper()
elif tag == 0x80:
result['file_size'] = int.from_bytes(value, 'big')
elif tag == 0x81:
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 result['sfi'] = f'0x{value[0]:02X}' if value else None
elif tag == 0x8A: elif tag == 0x8A:
@@ -1787,6 +1809,540 @@ def _decode_proactive(body):
# ──────────────────── Decode entry point ──────────────────── # ──────────────────── Decode entry point ────────────────────
# ──────────────────── File data decoders (READ/UPDATE) ────────────────────
# These decode the *body* of READ BINARY/RECORD and UPDATE BINARY/RECORD
# commands using the currently selected file (see select_target_fid and the
# server's selection tracking). Sourced from pySim (TS 51.011 SIM, TS 31.102
# USIM, TS 31.103 ISIM).
# USIM Service Table (TS 31.102) — service n ↔ bit (n-1)
UST_SERVICES = {
1: 'Local Phone Book', 2: 'Fixed Dialling Numbers (FDN)', 3: 'Extension 2',
4: 'Service Dialling Numbers (SDN)', 5: 'Extension3', 6: 'Barred Dialling Numbers (BDN)',
7: 'Extension4', 8: 'Outgoing Call Information (OCI and OCT)',
9: 'Incoming Call Information (ICI and ICT)', 10: 'Short Message Storage (SMS)',
11: 'Short Message Status Reports (SMSR)', 12: 'Short Message Service Parameters (SMSP)',
13: 'Advice of Charge (AoC)', 14: 'Capability Configuration Parameters 2 (CCP2)',
15: 'Cell Broadcast Message Identifier', 16: 'Cell Broadcast Message Identifier Ranges',
17: 'Group Identifier Level 1', 18: 'Group Identifier Level 2', 19: 'Service Provider Name',
20: 'User controlled PLMN selector with Access Technology', 21: 'MSISDN', 22: 'Image (IMG)',
23: 'Support of Localised Service Areas (SoLSA)', 24: 'Enhanced Multi-Level Precedence and Pre-emption Service',
25: 'Automatic Answer for eMLPP', 26: 'RFU', 27: 'GSM Access', 28: 'Data download via SMS-PP',
29: 'Data download via SMS-CB', 30: 'Call Control by USIM', 31: 'MO-SMS Control by USIM',
32: 'RUN AT COMMAND command', 33: 'shall be set to 1', 34: 'Enabled Services Table',
35: 'APN Control List (ACL)', 36: 'Depersonalisation Control Keys', 37: 'Co-operative Network List',
38: 'GSM security context', 39: 'CPBCCH Information', 40: 'Investigation Scan', 41: 'MexE',
42: 'Operator controlled PLMN selector with Access Technology', 43: 'HPLMN selector with Access Technology',
44: 'Extension 5', 45: 'PLMN Network Name', 46: 'Operator PLMN List', 47: 'Mailbox Dialling Numbers',
48: 'Message Waiting Indication Status', 49: 'Call Forwarding Indication Status', 50: 'Reserved and shall be ignored',
51: 'Service Provider Display Information', 52: 'Multimedia Messaging Service (MMS)', 53: 'Extension 8',
54: 'Call control on GPRS by USIM', 55: 'MMS User Connectivity Parameters',
56: "Network's indication of alerting in the MS (NIA)", 57: 'VGCS Group Identifier List (EFVGCS and EFVGCSS)',
58: 'VBS Group Identifier List (EFVBS and EFVBSS)', 59: 'Pseudonym',
60: 'User Controlled PLMN selector for I-WLAN access', 61: 'Operator Controlled PLMN selector for I-WLAN access',
62: 'User controlled WSID list', 63: 'Operator controlled WSID list', 64: 'VGCS security', 65: 'VBS security',
66: 'WLAN Reauthentication Identity', 67: 'Multimedia Messages Storage',
68: 'Generic Bootstrapping Architecture (GBA)', 69: 'MBMS security',
70: 'Data download via USSD and USSD application mode', 71: 'Equivalent HPLMN',
72: 'Additional TERMINAL PROFILE after UICC activation', 73: 'Equivalent HPLMN Presentation Indication',
74: 'Last RPLMN Selection Indication', 75: 'OMA BCAST Smart Card Profile',
76: 'GBA-based Local Key Establishment Mechanism', 77: 'Terminal Applications', 78: 'Service Provider Name Icon',
79: 'PLMN Network Name Icon', 80: 'Connectivity Parameters for USIM IP connections',
81: 'Home I-WLAN Specific Identifier List', 82: 'I-WLAN Equivalent HPLMN Presentation Indication',
83: 'I-WLAN HPLMN Priority Indication', 84: 'I-WLAN Last Registered PLMN', 85: 'EPS Mobility Management Information',
86: 'Allowed CSG Lists and corresponding indications', 87: 'Call control on EPS PDN connection by USIM',
88: 'HPLMN Direct Access', 89: 'eCall Data', 90: 'Operator CSG Lists and corresponding indications',
91: 'Support for SM-over-IP', 92: 'Support of CSG Display Control', 93: 'Communication Control for IMS by USIM',
94: 'Extended Terminal Applications', 95: 'Support of UICC access to IMS',
96: 'Non-Access Stratum configuration by USIM', 97: 'PWS configuration by USIM', 98: 'RFU',
99: 'URI support by UICC', 100: 'Extended EARFCN support', 101: 'ProSe', 102: 'USAT Application Pairing',
103: 'Media Type support', 104: 'IMS call disconnection cause', 105: 'URI support for MO SHORT MESSAGE CONTROL',
106: 'ePDG configuration Information support', 107: 'ePDG configuration Information configured', 108: 'ACDC support',
109: 'MCPTT', 110: 'ePDG configuration Information for Emergency Service support',
111: 'ePDG configuration Information for Emergency Service configured', 112: 'eCall Data over IMS',
113: 'URI support for SMS-PP DOWNLOAD as defined in 3GPP TS 31.111 [12]', 114: 'From Preferred',
115: 'IMS configuration data', 116: 'TV configuration', 117: '3GPP PS Data Off',
118: '3GPP PS Data Off Service List', 119: 'V2X', 120: 'XCAP Configuration Data',
121: 'EARFCN list for MTC/NB-IOT UEs', 122: '5GS Mobility Management Information', 123: '5G Security Parameters',
124: 'Subscription identifier privacy support', 125: 'SUCI calculation by the USIM',
126: 'UAC Access Identities support',
127: 'Expect control plane-based Steering of Roaming information during initial registration in VPLMN',
128: 'Call control on PDU Session by USIM', 129: '5GS Operator PLMN List',
130: 'Support for SUPI of type NSI or GLI or GCI', 131: '3GPP PS Data Off separate Home and Roaming lists',
132: 'Support for URSP by USIM', 133: '5G Security Parameters extended', 134: 'MuD and MiD configuration data',
135: 'Support for Trusted non-3GPP access networks by USIM',
136: 'Support for multiple records of NAS security context storage for multiple registration',
137: 'Pre-configured CAG information list', 138: 'SOR-CMCI storage in USIM', 139: '5G ProSe',
140: 'Storage of disaster roaming information in USIM', 141: 'Pre-configured eDRX parameters',
142: '5G NSWO support', 143: 'PWS configuration for SNPN in USIM',
144: 'Multiplier Coefficient for Higher Priority PLMN search via NG-RAN satellite access',
145: 'K_AUSF derivation configuration', 146: 'Network Identifier for SNPN (NID)',
}
# SIM Service Table (TS 51.011)
SST_SERVICES = {
1: 'CHV1 disable function', 2: 'Abbreviated Dialling Numbers (ADN)', 3: 'Fixed Dialling Numbers (FDN)',
4: 'Short Message Storage (SMS)', 5: 'Advice of Charge (AoC)', 6: 'Capability Configuration Parameters (CCP)',
7: 'PLMN selector', 8: 'RFU', 9: 'MSISDN', 10: 'Extension1', 11: 'Extension2', 12: 'SMS Parameters',
13: 'Last Number Dialled (LND)', 14: 'Cell Broadcast Message Identifier', 15: 'Group Identifier Level 1',
16: 'Group Identifier Level 2', 17: 'Service Provider Name', 18: 'Service Dialling Numbers (SDN)',
19: 'Extension3', 20: 'RFU', 21: 'VGCS Group Identifier List (EFVGCS and EFVGCSS)',
22: 'VBS Group Identifier List (EFVBS and EFVBSS)', 23: 'enhanced Multi-Level Precedence and Pre-emption Service',
24: 'Automatic Answer for eMLPP', 25: 'Data download via SMS-CB', 26: 'Data download via SMS-PP',
27: 'Menu selection', 28: 'Call control', 29: 'Proactive SIM', 30: 'Cell Broadcast Message Identifier Ranges',
31: 'Barred Dialling Numbers (BDN)', 32: 'Extension4', 33: 'De-personalization Control Keys',
34: 'Co-operative Network List', 35: 'Short Message Status Reports', 36: "Network's indication of alerting in the MS",
37: 'Mobile Originated Short Message control by SIM', 38: 'GPRS', 39: 'Image (IMG)',
40: 'SoLSA (Support of Local Service Area)', 41: 'USSD string data object supported in Call Control',
42: 'RUN AT COMMAND command', 43: 'User controlled PLMN Selector with Access Technology',
44: 'Operator controlled PLMN Selector with Access Technology', 45: 'HPLMN Selector with Access Technology',
46: 'CPBCCH Information', 47: 'Investigation Scan', 48: 'Extended Capability Configuration Parameters',
49: 'MExE', 50: 'Reserved and shall be ignored', 51: 'PLMN Network Name', 52: 'Operator PLMN List',
53: 'Mailbox Dialling Numbers', 54: 'Message Waiting Indication Status', 55: 'Call Forwarding Indication Status',
56: 'Service Provider Display Information', 57: 'Multimedia Messaging Service (MMS)', 58: 'Extension 8',
59: 'MMS User Connectivity Parameters',
}
# ISIM Service Table (TS 31.103)
IST_SERVICES = {
1: 'P-CSCF address', 2: 'Generic Bootstrapping Architecture (GBA)', 3: 'HTTP Digest',
4: 'GBA-based Local Key Establishment Mechanism', 5: 'Support of P-CSCF discovery for IMS Local Break Out',
6: 'Short Message Storage (SMS)', 7: 'Short Message Status Reports (SMSR)',
8: 'Support for SM-over-IP including data download via SMS-PP as defined in TS 31.111 [31]',
9: 'Communication Control for IMS by ISIM', 10: 'Support of UICC access to IMS', 11: 'URI support by UICC',
12: 'Media Type support', 13: 'IMS call disconnection cause', 14: 'URI support for MO SHORT MESSAGE CONTROL',
15: 'MCPTT', 16: 'URI support for SMS-PP DOWNLOAD as defined in 3GPP TS 31.111 [31]', 17: 'From Preferred',
18: 'IMS configuration data', 19: 'XCAP Configuration Data', 20: 'WebRTC URI',
21: 'MuD and MiD configuration data', 22: 'IMS Data Channel indication',
}
_PHASE_NAMES = {0x00: 'phase 1', 0x02: 'phase 2', 0x03: 'phase 2 and higher'}
def _decode_service_table(raw, table):
flags = []
for i, b in enumerate(raw):
for bit in range(8):
if b & (1 << bit):
n = i * 8 + bit + 1
flags.append({'n': n, 'name': table.get(n)})
return {'services': flags}
def _decode_imsi(raw, p1=None):
if not raw or all(b == 0xff for b in raw):
return {'empty': True}
if len(raw) <= 2: # 6F07 is also the ISIM Service Table (IST)
return _decode_service_table(raw, IST_SERVICES)
digits = _swap_nibbles(raw[1:].hex()).rstrip('fF')
return {'imsi': digits or None}
def _decode_iccid(raw, p1=None):
if not raw or all(b == 0xff for b in raw):
return {'empty': True}
digits = _swap_nibbles(raw.hex()).rstrip('fF')
return {'iccid': digits or None}
def _decode_li(raw, p1=None):
"""EF_LI / EF_PL — 2-letter ISO 639 language codes."""
codes = []
for i in range(0, len(raw) - 1, 2):
pair = raw[i:i + 2]
if pair == b'\xff\xff':
continue
try:
c = pair.decode('ascii')
except UnicodeDecodeError:
continue
if c.isalpha():
codes.append(c)
return {'languages': codes}
def _decode_adn(raw, p1=None):
"""ADN-format record: alpha + len_bcd + TON/NPI + number + CCP + ext1."""
if not raw or all(b == 0xff for b in raw):
return {'empty': True}
if len(raw) < 14:
return {'raw': raw.hex().upper()}
alpha = raw[:-14].rstrip(b'\xff')
name = _decode_annex_a(alpha) if alpha else ''
number = _decode_bcd_address(raw[-13:-2])
ccp = raw[-2]
ext1 = raw[-1]
out = {'name': name, 'number': number}
if ccp != 0xff:
out['ccp'] = ccp
if ext1 != 0xff:
out['ext1'] = ext1
return out
def _decode_sms_rec(raw, p1=None):
"""EF_SMS record (TS 51.011 §10.5.3): status byte + SMSC address + TPDU."""
if not raw or all(b == 0xff for b in raw):
return {'empty': True}
s = raw[0]
if s & 0x01 == 0x00:
direction, status = None, 'free space'
elif s & 0x07 == 0x01:
direction, status = 'MT', 'message read'
elif s & 0x07 == 0x03:
direction, status = 'MT', 'message to be read'
elif s & 0x07 == 0x07:
direction, status = 'MO', 'message to be sent'
elif s & 0x1f == 0x05:
direction, status = 'MO', 'sent (status not requested)'
elif s & 0x1f == 0x0d:
direction, status = 'MO', 'sent (status requested, not received)'
elif s & 0x1f == 0x15:
direction, status = 'MO', 'sent (status received, not stored)'
elif s & 0x1f == 0x1d:
direction, status = 'MO', 'sent (status received, stored)'
else:
direction, status = None, 'RFU'
out = {'direction': direction, 'status': status}
remainder = raw[1:]
# Skip the TS-Service-Centre-Address: 1 length octet + that many octets.
if remainder:
sc_len = remainder[0]
i = 1 + sc_len
tpdu_bytes = remainder[i:].rstrip(b'\xff')
if tpdu_bytes:
out['tpdu'] = _decode_sm_tpdu(tpdu_bytes)
return out
def _decode_plmn_list(raw, p1=None):
entries = []
for i in range(0, len(raw), 3):
chunk = raw[i:i + 3]
if len(chunk) < 3 or all(b == 0xff for b in chunk):
break
plmn = _decode_plmn(chunk)
if plmn:
entries.append(plmn)
return {'plmns': entries}
def _decode_plmn_wact(raw, p1=None):
"""PLMN selector with Access Technology (TS 51.011 §10.3.35): 5-byte entries.
Each entry: 3-byte PLMN + 2-byte access technology bitmask.
"""
entries = []
for i in range(0, len(raw) - 4, 5):
chunk = raw[i:i + 5]
if len(chunk) < 5:
break
plmn_bytes = chunk[:3]
if all(b == 0xff for b in plmn_bytes):
break
plmn = _decode_plmn(plmn_bytes)
if not plmn:
continue
act = (chunk[3] << 8) | chunk[4]
techs = _decode_plmn_act(act)
plmn['access_tech'] = ', '.join(techs) if techs else f'0x{act:04X}'
entries.append(plmn)
return {'plmns': entries}
def _decode_plmn_act(u16):
"""Decode the 2-byte access-technology field (TS 31.102 §4.2.5)."""
techs = set()
if u16 & 0x8000:
techs.add('UTRAN')
if u16 & 0x0800:
techs.add('NG-RAN')
if u16 & 0x0040:
techs.add('GSM COMPACT')
if u16 & 0x0020:
techs.add('cdma2000 HRPD')
if u16 & 0x0010:
techs.add('cdma2000 1xRTT')
e = u16 & 0x7000
if e in (0x4000, 0x7000):
techs.add('E-UTRAN WB-S1')
techs.add('E-UTRAN NB-S1')
elif e == 0x5000:
techs.add('E-UTRAN NB-S1')
elif e == 0x6000:
techs.add('E-UTRAN WB-S1')
g = u16 & 0x008C
if g in (0x0080, 0x008C):
techs.add('GSM')
techs.add('EC-GSM-IoT')
elif g == 0x0084:
techs.add('GSM')
elif g == 0x0086:
techs.add('EC-GSM-IoT')
return sorted(techs)
def _decode_dir(raw, p1=None):
"""EF_DIR record: application template(s) — AID (4F) + label (50)."""
apps = []
for tag, _length, value in parse_tlv(raw):
if tag != 0x61:
continue
app = {}
for t2, _l2, v2 in parse_tlv(value):
if t2 == 0x4F:
app['aid'] = v2.hex().upper()
elif t2 == 0x50:
label = _decode_annex_a(v2)
if label:
app['label'] = label
if app:
apps.append(app)
return {'applications': apps} if apps else {'raw': raw.hex().upper()}
def _decode_arr(raw, p1=None):
entries = []
for tag, _length, value in parse_tlv(raw):
entries.append({'tag': f'0x{tag:02X}', 'hex': value.hex().upper()})
return {'rules': entries} if entries else {'raw': raw.hex().upper()}
def _decode_pnn(raw, p1=None):
names = {}
for tag, _length, value in parse_tlv(raw):
if tag == 0x43:
names['full'] = _decode_annex_a(value)
elif tag == 0x45:
names['short'] = _decode_annex_a(value)
if not names:
txt = _decode_annex_a(raw.rstrip(b'\xff'))
if txt:
names['full'] = txt
return names or {'raw': raw.hex().upper()}
def _decode_cbmi(raw, p1=None):
ids = []
for i in range(0, len(raw) - 1, 2):
v = (raw[i] << 8) | raw[i + 1]
if v == 0xffff:
continue
ids.append(v)
return {'message_ids': ids}
def _decode_cbmir(raw, p1=None):
if len(raw) >= 4:
low = (raw[0] << 8) | raw[1]
high = (raw[2] << 8) | raw[3]
return {'range': [low, high]}
return {'raw': raw.hex().upper()}
def _decode_ecc(raw, p1=None):
txt = _decode_annex_a(raw.rstrip(b'\xff'))
return {'number': txt} if txt else {'raw': raw.hex().upper()}
def _decode_spn(raw, p1=None):
if len(raw) < 2:
return {'raw': raw.hex().upper()}
cond = raw[0]
text = _decode_annex_a(raw[1:].rstrip(b'\xff'))
out = {'display_condition': f'0x{cond:02X}'}
if text:
out['name'] = text
return out
def _decode_loci(raw, p1=None):
if len(raw) < 13:
return {'raw': raw.hex().upper()}
tmsi = raw[0:4]
lai = raw[4:9]
plmn = _decode_plmn(lai[:3]) or {}
return {
'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}',
}
def _decode_acc(raw, p1=None):
if len(raw) < 2:
return {'raw': raw.hex().upper()}
val = (raw[0] << 8) | raw[1]
classes = [i for i in range(16) if val & (0x8000 >> i)]
return {'access_classes': classes}
def _decode_phase(raw, p1=None):
v = raw[0] if raw else 0
return {'phase': _PHASE_NAMES.get(v, f'0x{v:02X}')}
def _decode_nai(raw, p1=None):
"""ISIM identity files (EF_IMPI/DOMAIN/IMPU): BER-TLV tag 0x80 + ASCII value."""
texts = []
for tag, _length, value in parse_tlv(raw):
if tag == 0x80:
txt = value.decode('utf-8', 'replace').rstrip('\xff')
if txt:
texts.append(txt)
if texts:
return {'text': texts[0] if len(texts) == 1 else ', '.join(texts)}
txt = raw.decode('ascii', 'replace').replace('\xff', '').strip()
return {'text': txt} if txt else {'raw': raw.hex().upper()}
def _decode_hex(raw, p1=None):
if not raw or all(b == 0xff for b in raw):
return {'empty': True}
return {'raw': raw.hex().upper()}
FILE_DECODERS = {
'2fe2': _decode_iccid,
'6f07': _decode_imsi,
'6f05': _decode_li,
'2f05': _decode_li,
'6f3c': _decode_sms_rec,
'6f38': lambda raw, p1=None: _decode_service_table(raw, UST_SERVICES),
'6f30': _decode_plmn_list,
'6f7b': _decode_plmn_list,
'6f31': _decode_plmn_list,
'6fd9': _decode_plmn_list,
'6f60': _decode_plmn_wact,
'6f61': _decode_plmn_wact,
'6f62': _decode_plmn_wact,
'2f00': _decode_dir,
'6f40': _decode_adn,
'6f49': _decode_adn,
'6f3b': _decode_adn,
'6f3a': _decode_adn,
'6f4d': _decode_adn,
'6fc7': _decode_adn,
'6f80': _decode_adn,
'6f81': _decode_adn,
'6f45': _decode_cbmi,
'6f50': _decode_cbmir,
'2f06': _decode_arr,
'6f06': _decode_arr,
'6fc5': _decode_pnn,
'6fb7': _decode_ecc,
'6f46': _decode_spn,
'6f7e': _decode_loci,
'6f08': _decode_hex,
'6f09': _decode_hex,
'6f3e': _decode_hex,
'6f3f': _decode_hex,
'6fae': _decode_phase,
'6f78': _decode_acc,
'6fad': _decode_hex,
'6f02': _decode_nai,
'6f03': _decode_nai,
'6f04': _decode_nai,
'6fcb': _decode_hex,
'6fc4': _decode_hex,
'6fca': _decode_hex,
'6fc9': _decode_hex,
}
def _decode_file_data(fid, raw, p1=None):
"""Decode file data using the FID's registered decoder."""
if not raw:
return None
fn = FILE_DECODERS.get(fid)
if not fn:
return None
try:
out = fn(raw, p1=p1)
except Exception:
return None
if not out:
return None
out['fid'] = fid
out['ef'] = KNOWN_FIDS.get(fid, fid.upper())
return out
def _file_summary(f):
"""Compact one-line summary of a decoded file body."""
if f.get('empty'):
return 'empty'
if f.get('imsi'):
return f"IMSI {f['imsi']}"
if f.get('iccid'):
return f"ICCID {f['iccid']}"
if f.get('languages'):
return 'languages ' + ' '.join(f['languages'])
if 'number' in f and 'name' in f:
return (f['name'] + ' ' + f['number']).strip()
if f.get('number'):
return f['number']
if f.get('text'):
return f['text']
if f.get('name'):
return f['name']
if f.get('plmns'):
return ', '.join(f"{p['mcc']}/{p['mnc']}" + (f" ({p['access_tech']})" if p.get('access_tech') else '')
for p in f['plmns'])
if f.get('services'):
return f"{len(f['services'])} services allocated"
if f.get('applications'):
return '; '.join((a.get('label') or a.get('aid') or '') for a in f['applications'])
if f.get('tpdu'):
t = f['tpdu']
s = t.get('mti', '')
if t.get('text'):
s += f" \u00ab{t['text']}\u00bb"
return f"SMS {s}".strip()
if f.get('direction'):
return f"SMS {f['direction']} — {f['status']}"
if f.get('mcc'):
return f"MCC {f['mcc']} MNC {f['mnc']} · LAC {f.get('lac', '')}"
if f.get('access_classes') is not None:
return 'classes ' + ', '.join(str(c) for c in f['access_classes'])
if f.get('phase'):
return f['phase']
if f.get('range'):
return f"range {f['range'][0]}-{f['range'][1]}"
if f.get('message_ids') is not None:
return f"{len(f['message_ids'])} message IDs"
if f.get('record_numbers') is not None:
nums = f['record_numbers']
shown = ', '.join(str(n) for n in nums[:10])
if len(nums) > 10:
shown += ', \u2026'
return f"{len(nums)} record(s): {shown}"
return None
def _fcp_summary(response_for, fd, response):
"""Build a summary for a GET RESPONSE (FCP) following SELECT."""
ft = fd.get('file_type') or ''
label = f"response for {response_for}, {ft}"
if ft == 'DF or ADF':
name = response.get('file_id_name')
if name:
label += f" ({name})"
else:
struct = fd.get('structure') or ''
if struct:
label += f", {struct}"
nrec = fd.get('num_records')
rlen = fd.get('record_length')
fsize = response.get('file_size') or response.get('total_file_size')
if nrec and rlen:
label += f", {nrec} rec \u00d7 {rlen} B"
elif fsize:
label += f", {fsize} B"
return label
def _build_summary(result): def _build_summary(result):
"""Build a concise human-readable description of a decoded command. """Build a concise human-readable description of a decoded command.
@@ -1868,11 +2424,22 @@ def _build_summary(result):
if result.get('response_to'): if result.get('response_to'):
parts.append('\u2192 ' + result['response_to']) parts.append('\u2192 ' + result['response_to'])
if result.get('response_for'): if result.get('response_for'):
response = result.get('response') or {}
fd = response.get('file_descriptor')
if fd:
parts.append(_fcp_summary(result['response_for'], fd, response))
else:
parts.append('response for ' + result['response_for']) parts.append('response for ' + result['response_for'])
response = result.get('response') response = result.get('response')
if response and response.get('name'): if response and response.get('name'):
parts.append(response['name']) parts.append(response['name'])
file_dec = result.get('file')
if file_dec:
txt = _file_summary(file_dec)
if txt:
parts.append(f"{file_dec.get('ef', '')} {txt}".strip())
return ', '.join(parts) if parts else None return ', '.join(parts) if parts else None
@@ -1964,6 +2531,14 @@ def decode_message(raw_data, prev=None):
elif ins == 0x12: # FETCH → proactive command body elif ins == 0x12: # FETCH → proactive command body
result['cmd'] = _decode_proactive(body) result['cmd'] = _decode_proactive(body)
# 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('value', 0)
if ins in (0xB2, 0xDC) or offset == 0:
file_dec = _decode_file_data(prev['sel']['fid'], body, p1=p1)
if file_dec:
result['file'] = file_dec
# SW # SW
if sw_bytes: if sw_bytes:
result['sw'] = decode_sw(sw_bytes) result['sw'] = decode_sw(sw_bytes)
@@ -1975,6 +2550,19 @@ def decode_message(raw_data, prev=None):
result['response_for'] = prev.get('ins_name') result['response_for'] = prev.get('ins_name')
prev_ins = prev.get('ins') prev_ins = prev.get('ins')
if prev_ins is not None and cmd_body_len > 0: if prev_ins is not None and cmd_body_len > 0:
if prev_ins in (0xB0, 0xB2) and prev.get('sel') and prev.get('file_ok'):
file_dec = _decode_file_data(prev['sel']['fid'], remaining[:cmd_body_len])
if file_dec:
result['file'] = file_dec
elif prev_ins == 0xA2 and prev.get('sel') and prev.get('file_ok'):
# SEARCH RECORD → list of matching record numbers
fid = prev['sel']['fid']
result['file'] = {
'fid': fid,
'ef': KNOWN_FIDS.get(fid, fid.upper()),
'record_numbers': list(remaining[:cmd_body_len]),
}
else:
response = _decode_response_for(prev_ins, remaining[:cmd_body_len]) response = _decode_response_for(prev_ins, remaining[:cmd_body_len])
if response: if response:
result['response'] = response result['response'] = response
+164
View File
@@ -234,6 +234,32 @@ class TestFcpResponse(unittest.TestCase):
self.assertEqual(fd['file_type'], 'Working EF') self.assertEqual(fd['file_type'], 'Working EF')
self.assertEqual(fd['structure'], 'linear fixed') self.assertEqual(fd['structure'], 'linear fixed')
def test_fcp_file_size(self):
from simtrace2_pysniff.server.decode import _decode_fcp
# FCP: file descriptor (transparent EF) + FID 6F07 + file size 9 + total 9
fcp = _decode_fcp(bytes.fromhex('8202012183026f07800109810109'))
self.assertEqual(fcp['file_descriptor']['file_type'], 'Working EF')
self.assertEqual(fcp['file_descriptor']['structure'], 'transparent')
self.assertEqual(fcp['file_id_name'], 'EF_IMSI')
self.assertEqual(fcp['file_size'], 9)
self.assertEqual(fcp['total_file_size'], 9)
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')
r = decode_message(gr, prev={'ins': 0xA4, 'ins_name': 'SELECT', 'sw1': '61'})
self.assertEqual(r['summary'], 'response for SELECT, Working EF, transparent, 9 B')
def test_fcp_summary_linear_fixed(self):
# linear fixed EF, 3 records x 64 bytes
gr = bytes.fromhex('00c000000e') + bytes.fromhex('620e8205022100400383026f07800109') + bytes.fromhex('9000')
r = decode_message(gr, prev={'ins': 0xA4, 'ins_name': 'SELECT', 'sw1': '61'})
self.assertEqual(r['summary'], 'response for SELECT, Working EF, linear fixed, 3 rec × 64 B')
def test_fcp_summary_df(self):
r = decode_message(self.GET_RESPONSE, prev={'ins': 0xA4, 'ins_name': 'SELECT', 'sw1': '61'})
self.assertEqual(r['summary'], 'response for SELECT, DF or ADF (MF)')
class TestAuthResponse(unittest.TestCase): class TestAuthResponse(unittest.TestCase):
def test_3g_success(self): def test_3g_success(self):
@@ -820,5 +846,143 @@ class TestSelectPath(unittest.TestCase):
self.assertEqual(r['body']['note'], 'current ADF/DF_5GS/EF_5GS3GPPLOCI') self.assertEqual(r['body']['note'], 'current ADF/DF_5GS/EF_5GS3GPPLOCI')
class TestFileDecoders(unittest.TestCase):
def test_imsi(self):
from simtrace2_pysniff.server.decode import _decode_file_data
f = _decode_file_data('6f07', bytes.fromhex('0f52001132547698f0'))
self.assertEqual(f['imsi'], '250011234567890')
def test_iccid(self):
from simtrace2_pysniff.server.decode import _decode_file_data
f = _decode_file_data('2fe2', bytes.fromhex('98891020000000460012'))
self.assertEqual(f['iccid'], '89980102000000640021')
def test_li(self):
from simtrace2_pysniff.server.decode import _decode_file_data
f = _decode_file_data('6f05', b'enru')
self.assertEqual(f['languages'], ['en', 'ru'])
def test_adn(self):
from simtrace2_pysniff.server.decode import _decode_file_data
f = _decode_file_data('6f3a', bytes.fromhex(
'42204841203120536963FFFFFFFFFFFF06810628560810FFFFFFFFFFFFFF'))
self.assertEqual(f['name'], 'B HA 1 Sic')
self.assertEqual(f['number'], '6082658001')
def test_ust(self):
from simtrace2_pysniff.server.decode import _decode_file_data
f = _decode_file_data('6f38', bytes.fromhex('07'))
self.assertEqual([s['n'] for s in f['services']], [1, 2, 3])
self.assertEqual(f['services'][0]['name'], 'Local Phone Book')
def test_plmn_list(self):
from simtrace2_pysniff.server.decode import _decode_file_data
f = _decode_file_data('6f30', bytes.fromhex('42f095'))
self.assertEqual(len(f['plmns']), 1)
self.assertIn('mcc', f['plmns'][0])
self.assertIn('mnc', f['plmns'][0])
def test_dir(self):
from simtrace2_pysniff.server.decode import _decode_file_data
f = _decode_file_data('2f00', bytes.fromhex(
'612b4f10a0000000871002fffff00189000001ff50074d656761466f6e'
'730ea00c80011781025f408203454150ffffffffffffffffffffffffffffffffffffff'))
self.assertEqual(f['applications'][0]['label'], 'MegaFon')
self.assertEqual(f['applications'][0]['aid'], 'A0000000871002FFFFF00189000001FF')
def test_plmn_wact(self):
from simtrace2_pysniff.server.decode import _decode_file_data
# TS 51.011 §10.3.35: 5-byte entries (3 PLMN + 2 access tech).
f = _decode_file_data('6f60', bytes.fromhex('52f020400052f0208000'))
self.assertEqual(f['plmns'][0]['mcc'], '250')
self.assertEqual(f['plmns'][0]['mnc'], '02')
self.assertEqual(f['plmns'][0]['access_tech'], 'E-UTRAN NB-S1, E-UTRAN WB-S1')
self.assertEqual(f['plmns'][1]['access_tech'], 'UTRAN')
def test_sms_record(self):
from simtrace2_pysniff.server.decode import _decode_file_data
# status 0x07 (MO, to be sent) + empty SMSC (len 0) + SMS-SUBMIT TPDU
rec = bytes.fromhex('0700') + bytes.fromhex('01ff038199f90004024869')
f = _decode_file_data('6f3c', rec)
self.assertEqual(f['direction'], 'MO')
self.assertEqual(f['status'], 'message to be sent')
self.assertEqual(f['tpdu']['mti'], 'SMS-SUBMIT')
self.assertEqual(f['tpdu']['da'], '999')
self.assertEqual(f['tpdu']['text'], 'Hi')
def test_read_binary_direct(self):
# SELECT EF_IMSI then READ BINARY → file decoded in body.
r = decode_message(bytes.fromhex('00b0000009') +
bytes.fromhex('0f52001132547698f0') + bytes.fromhex('9000'),
prev={'sel': {'fid': '6f07', 'name': 'EF_IMSI'}})
self.assertEqual(r['file']['imsi'], '250011234567890')
self.assertIn('IMSI 250011234567890', r['summary'])
def test_read_binary_offset_skip(self):
r = decode_message(bytes.fromhex('00b00100') +
bytes.fromhex('0f52001132547698f0') + bytes.fromhex('9000'),
prev={'sel': {'fid': '6f07'}})
self.assertNotIn('file', r)
def test_read_via_get_response(self):
iccid = bytes.fromhex('98891020000000460012')
gr = bytes.fromhex('00c00000') + bytes([len(iccid)]) + iccid + bytes.fromhex('9000')
r = decode_message(gr, prev={'ins': 0xB0, 'ins_name': 'READ BINARY', 'sw1': '61',
'sel': {'fid': '2fe2'}, 'file_ok': True})
self.assertEqual(r['file']['iccid'], '89980102000000640021')
def test_get_response_file_ok_false(self):
iccid = bytes.fromhex('98891020000000460012')
gr = bytes.fromhex('00c00000') + bytes([len(iccid)]) + iccid + bytes.fromhex('9000')
r = decode_message(gr, prev={'ins': 0xB0, 'ins_name': 'READ BINARY', 'sw1': '61',
'sel': {'fid': '2fe2'}, 'file_ok': False})
self.assertNotIn('file', r)
def test_search_record_numbers(self):
# SEARCH RECORD returns 61xx; GET RESPONSE carries the record numbers.
numbers = bytes(range(3, 21)) # 3..20
gr = bytes.fromhex('00c00000') + bytes([len(numbers)]) + numbers + bytes.fromhex('9000')
r = decode_message(gr, prev={'ins': 0xA2, 'ins_name': 'SEARCH RECORD', 'sw1': '61',
'sel': {'fid': '6f3a'}, 'file_ok': True})
self.assertEqual(r['file']['record_numbers'], list(range(3, 21)))
self.assertIn('18 record(s)', r['summary'])
def test_select_target_fid(self):
from simtrace2_pysniff.server.decode import select_target_fid
self.assertEqual(select_target_fid({'p1': {'raw': '00'}, 'body': {'hex': '6f07'}}), '6f07')
self.assertEqual(select_target_fid({'p1': {'raw': '08'}, 'body': {'hex': '7fff6f05'}}), '6f05')
self.assertIsNone(select_target_fid({'p1': {'raw': '04'}, 'body': {'hex': 'a0000000871002'}}))
self.assertIsNone(select_target_fid({'p1': {'raw': '08'}, 'body': {'hex': '7fff'}}))
class TestSelectionTracking(unittest.TestCase):
def test_select_then_read(self):
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('a0a40804047fff6f079000'))
db.insert_message(sid, 0.2, 'tpdu',
bytes.fromhex('00b0000009') + bytes.fromhex('0f52001132547698f0') + bytes.fromhex('9000'))
msgs = db.get_messages(sid)
read = msgs[2]
self.assertEqual(read['decoded']['file']['imsi'], '250011234567890')
def test_atr_resets_selection(self):
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('a0a40804047fff6f079000'))
db.insert_message(sid, 0.1, 'atr', b'\x3b\x00')
db.insert_message(sid, 0.2, 'tpdu',
bytes.fromhex('00b0000009') + bytes.fromhex('0f52001132547698f0') + bytes.fromhex('9000'))
msgs = db.get_messages(sid)
self.assertNotIn('file', msgs[2]['decoded'])
if __name__ == '__main__': if __name__ == '__main__':
unittest.main() unittest.main()