Harden selection tracking against dropped/failed SELECTs

- Only update the selected-file state on a successful SELECT (SW1 in
  61/62/63/90/91/92/9E/9F); a failed SELECT (e.g. 6A82) leaves the
  current file unchanged.
- Track the file structure from the FCP (GET RESPONSE after SELECT).
- Skip file decoding for record ops (READ/UPDATE/SEARCH RECORD) when the
  selected file is transparent/DF, marking the result 'stale' instead of
  decoding garbage — this catches silently-dropped SELECTs.
This commit is contained in:
catarrh committed 2026-08-16 00:20:32 +03:00
1 parent 91cb4e6e02
commit 747974460e
4 files changed
+131 -5

No files matched your search

+4 -2
View File
@@ -634,7 +634,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.empty) { if (f.stale) {
rows.push(['File', title + ' — selection stale (record op on non-record file)']);
} else if (f.empty) {
rows.push(['File', title + ' — empty']); rows.push(['File', title + ' — empty']);
} else { } else {
rows.push(['File', title]); rows.push(['File', title]);
@@ -816,7 +818,7 @@ function showDetail(msg, entry) {
} }
const cla = decoded.cla; const cla = decoded.cla;
const claInfo = `${cla.interclass}, ch${cla.channel}, ${cla.secure_messaging}, chain: ${cla.chain}`; const claInfo = `${cla.interclass}, ch${cla.channel}, SM: ${cla.secure_messaging}, chain: ${cla.chain}`;
const catRow = decoded.cat_command const catRow = decoded.cat_command
? `<tr><td class="label">CAT</td><td class="name">${decoded.cat_command}</td></tr>` ? `<tr><td class="label">CAT</td><td class="name">${decoded.cat_command}</td></tr>`
+21 -2
View File
@@ -8,6 +8,11 @@ from contextlib import contextmanager
DEFAULT_DB_DIR = os.path.expanduser('~/.simtrace-analyser') DEFAULT_DB_DIR = os.path.expanduser('~/.simtrace-analyser')
DEFAULT_DB_PATH = os.path.join(DEFAULT_DB_DIR, 'sessions.db') DEFAULT_DB_PATH = os.path.join(DEFAULT_DB_DIR, 'sessions.db')
# SW1 values that mean a SELECT succeeded (normal, warnings, response pending).
# Everything else (0x64-0x6F: execution/checking/security errors) leaves the
# current file unchanged.
_SUCCESS_SW1 = {'61', '62', '63', '90', '91', '92', '9e', '9f'}
_SCHEMA = """ _SCHEMA = """
CREATE TABLE IF NOT EXISTS sessions ( CREATE TABLE IF NOT EXISTS sessions (
id INTEGER PRIMARY KEY AUTOINCREMENT, id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -289,13 +294,27 @@ class Database:
} }
def _selection_after(self, d, sel): def _selection_after(self, d, sel):
if d.get('ins_hex') == 'a4': ins_hex = d.get('ins_hex')
if ins_hex == 'a4':
from .decode import select_target_fid, KNOWN_FIDS from .decode import select_target_fid, KNOWN_FIDS
sw1 = (d.get('sw') or {}).get('sw1')
# Only a successful SELECT changes the current file. On an error
# status word (e.g. 6A82 file-not-found) the file is unchanged.
if sw1 not in _SUCCESS_SW1:
return sel
fid = select_target_fid(d) fid = select_target_fid(d)
if fid: if fid:
return {'fid': fid, 'name': KNOWN_FIDS.get(fid)} return {'fid': fid, 'name': KNOWN_FIDS.get(fid), 'structure': None}
if (d.get('p1') or {}).get('raw') in ('03', '04'): if (d.get('p1') or {}).get('raw') in ('03', '04'):
return None return None
elif ins_hex == 'c0' and d.get('response_for') == 'SELECT' and sel:
# GET RESPONSE carrying the FCP of the last SELECT → capture the
# file structure so we can sanity-check record operations.
fd = (d.get('response') or {}).get('file_descriptor')
if fd:
new = dict(sel)
new['structure'] = fd.get('structure')
return new
return sel return sel
def _decode_rows(self, rows, initial_prev=None): def _decode_rows(self, rows, initial_prev=None):
+63 -1
View File
@@ -115,7 +115,7 @@ def decode_cla(cla):
'hex': f'{cla:02x}', 'hex': f'{cla:02x}',
'interclass': interclass, 'interclass': interclass,
'channel': cla & 0x03, 'channel': cla & 0x03,
'secure_messaging': 'none' if (cla & 0x0c) == 0 else 'SM present', 'secure_messaging': 'none' if (cla & 0x0c) == 0 else 'present',
'chain': chain_names.get(chain_bits, f'unknown ({chain_bits})'), 'chain': chain_names.get(chain_bits, f'unknown ({chain_bits})'),
} }
return result return result
@@ -1961,6 +1961,31 @@ def _decode_li(raw, p1=None):
return {'languages': codes} return {'languages': codes}
def _looks_like_ber_tlv(data):
"""Heuristic: does *data* look like a BER-TLV rather than Annex-A text?
Annex-A UCS2 prefixes are 0x80/0x81/0x82, so those are excluded; other
0x83-0xBF leading bytes are treated as BER-TLV tags if their length
field is consistent with the data length.
"""
if len(data) < 2:
return False
if data[0] in (0x80, 0x81, 0x82):
return False
if not (0x80 <= data[0] <= 0xBF):
return False
length = data[1]
if length & 0x80:
num = length & 0x7F
if num == 0 or 2 + num > len(data):
return False
length = int.from_bytes(data[2:2 + num], 'big')
header = 2 + num
else:
header = 2
return header + length >= len(data) - 1
def _decode_adn(raw, p1=None): def _decode_adn(raw, p1=None):
"""ADN-format record: alpha + len_bcd + TON/NPI + number + CCP + ext1.""" """ADN-format record: alpha + len_bcd + TON/NPI + number + CCP + ext1."""
if not raw or all(b == 0xff for b in raw): if not raw or all(b == 0xff for b in raw):
@@ -1968,6 +1993,9 @@ def _decode_adn(raw, p1=None):
if len(raw) < 14: if len(raw) < 14:
return {'raw': raw.hex().upper()} return {'raw': raw.hex().upper()}
alpha = raw[:-14].rstrip(b'\xff') alpha = raw[:-14].rstrip(b'\xff')
# If the alpha is not text (e.g. BER-TLV), show raw hex instead.
if alpha and _looks_like_ber_tlv(alpha):
return {'raw': raw.hex().upper()}
name = _decode_annex_a(alpha) if alpha else '' name = _decode_annex_a(alpha) if alpha else ''
number = _decode_bcd_address(raw[-13:-2]) number = _decode_bcd_address(raw[-13:-2])
ccp = raw[-2] ccp = raw[-2]
@@ -2269,8 +2297,33 @@ def _decode_file_data(fid, raw, p1=None):
return out return out
_RECORD_STRUCTURES = ('linear fixed', 'cyclic')
def _is_record_file(sel):
"""True if *sel* is (or may be) a record-based EF.
A None structure means we don't know the file structure (e.g. the
FCP/GET RESPONSE was lost too), so we allow the decode as best effort.
"""
if not sel:
return True
structure = sel.get('structure')
if structure is None:
return True
return structure in _RECORD_STRUCTURES
def _stale_file_note(sel):
"""Mark a decode as skipped because the selection is inconsistent."""
fid = sel.get('fid') or ''
return {'fid': fid, 'ef': KNOWN_FIDS.get(fid, fid.upper()), 'stale': True}
def _file_summary(f): def _file_summary(f):
"""Compact one-line summary of a decoded file body.""" """Compact one-line summary of a decoded file body."""
if f.get('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('imsi'): if f.get('imsi'):
@@ -2535,6 +2588,9 @@ def decode_message(raw_data, prev=None):
if ins in (0xB0, 0xB2, 0xD6, 0xDC) and prev and prev.get('sel'): if ins in (0xB0, 0xB2, 0xD6, 0xDC) and prev and prev.get('sel'):
offset = (result.get('p1p2') or {}).get('value', 0) offset = (result.get('p1p2') or {}).get('value', 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']):
result['file'] = _stale_file_note(prev['sel'])
else:
file_dec = _decode_file_data(prev['sel']['fid'], body, p1=p1) file_dec = _decode_file_data(prev['sel']['fid'], body, p1=p1)
if file_dec: if file_dec:
result['file'] = file_dec result['file'] = file_dec
@@ -2551,10 +2607,16 @@ def decode_message(raw_data, prev=None):
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'): if prev_ins in (0xB0, 0xB2) and prev.get('sel') and prev.get('file_ok'):
if prev_ins == 0xB2 and not _is_record_file(prev['sel']):
result['file'] = _stale_file_note(prev['sel'])
else:
file_dec = _decode_file_data(prev['sel']['fid'], remaining[:cmd_body_len]) file_dec = _decode_file_data(prev['sel']['fid'], remaining[:cmd_body_len])
if file_dec: if file_dec:
result['file'] = file_dec result['file'] = file_dec
elif prev_ins == 0xA2 and prev.get('sel') and prev.get('file_ok'): elif prev_ins == 0xA2 and prev.get('sel') and prev.get('file_ok'):
if not _is_record_file(prev['sel']):
result['file'] = _stale_file_note(prev['sel'])
else:
# SEARCH RECORD → list of matching record numbers # SEARCH RECORD → list of matching record numbers
fid = prev['sel']['fid'] fid = prev['sel']['fid']
result['file'] = { result['file'] = {
+43
View File
@@ -869,6 +869,17 @@ class TestFileDecoders(unittest.TestCase):
self.assertEqual(f['name'], 'B HA 1 Sic') self.assertEqual(f['name'], 'B HA 1 Sic')
self.assertEqual(f['number'], '6082658001') self.assertEqual(f['number'], '6082658001')
def test_adn_ber_tlv_fallback(self):
# Non-ADN data (BER-TLV) must not be decoded as a name/number.
from simtrace2_pysniff.server.decode import _decode_file_data
raw = bytes.fromhex(
'A0348001078120E823FF53C3E271754A644ED63DEFCF24A916387E3C585F'
'0820CF3E27841852F7820400000000830400000000840102'
'FFFFFFFFFFFFFFFFFFFFFFFFFFFF')
f = _decode_file_data('6fc7', raw)
self.assertNotIn('name', f)
self.assertIn('raw', f)
def test_ust(self): def test_ust(self):
from simtrace2_pysniff.server.decode import _decode_file_data from simtrace2_pysniff.server.decode import _decode_file_data
f = _decode_file_data('6f38', bytes.fromhex('07')) f = _decode_file_data('6f38', bytes.fromhex('07'))
@@ -983,6 +994,38 @@ class TestSelectionTracking(unittest.TestCase):
msgs = db.get_messages(sid) msgs = db.get_messages(sid)
self.assertNotIn('file', msgs[2]['decoded']) self.assertNotIn('file', msgs[2]['decoded'])
def test_failed_select_keeps_selection(self):
# A SELECT that fails (6A82) must not change the current 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('a0a40804047fff6f079000'))
db.insert_message(sid, 0.1, 'tpdu', bytes.fromhex('a0a40804047fff6f136a82'))
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_record_op_on_transparent_is_stale(self):
# SELECT a transparent EF, then READ RECORD → selection is stale, so
# we must not decode garbage.
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('a0a40804047fff6f056123'))
db.insert_message(sid, 0.1, 'tpdu',
bytes.fromhex('00c000000a') + bytes.fromhex('62088202012183026f05') + bytes.fromhex('9000'))
db.insert_message(sid, 0.2, 'tpdu',
bytes.fromhex('00b2010408') + bytes.fromhex('800101a010a4068301019501') + bytes.fromhex('9000'))
msgs = db.get_messages(sid)
read = msgs[2]
self.assertTrue(read['decoded']['file']['stale'])
if __name__ == '__main__': if __name__ == '__main__':
unittest.main() unittest.main()