From 747974460ef41ca551427271ef6fa3f7a5141c0a 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: Sun, 16 Aug 2026 00:20:32 +0300 Subject: [PATCH] Harden selection tracking against dropped/failed SELECTs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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. --- frontend/index.html | 6 +- simtrace2_pysniff/server/database.py | 23 ++++++- simtrace2_pysniff/server/decode.py | 90 +++++++++++++++++++++++----- tests/test_decode.py | 43 +++++++++++++ 4 files changed, 144 insertions(+), 18 deletions(-) diff --git a/frontend/index.html b/frontend/index.html index 37d2a48..677b96c 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -634,7 +634,9 @@ function fileRows(f) { if (!f) return ''; const title = f.ef ? `${f.ef} (0x${(f.fid || '').toUpperCase()})` : (f.fid || '').toUpperCase(); 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']); } else { rows.push(['File', title]); @@ -816,7 +818,7 @@ function showDetail(msg, entry) { } 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 ? `CAT${decoded.cat_command}` diff --git a/simtrace2_pysniff/server/database.py b/simtrace2_pysniff/server/database.py index d2bbe6d..ba98d70 100644 --- a/simtrace2_pysniff/server/database.py +++ b/simtrace2_pysniff/server/database.py @@ -8,6 +8,11 @@ from contextlib import contextmanager DEFAULT_DB_DIR = os.path.expanduser('~/.simtrace-analyser') 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 = """ CREATE TABLE IF NOT EXISTS sessions ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -289,13 +294,27 @@ class Database: } 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 + 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) 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'): 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 def _decode_rows(self, rows, initial_prev=None): diff --git a/simtrace2_pysniff/server/decode.py b/simtrace2_pysniff/server/decode.py index fd82a69..f22f10b 100644 --- a/simtrace2_pysniff/server/decode.py +++ b/simtrace2_pysniff/server/decode.py @@ -115,7 +115,7 @@ def decode_cla(cla): 'hex': f'{cla:02x}', 'interclass': interclass, '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})'), } return result @@ -1961,6 +1961,31 @@ def _decode_li(raw, p1=None): 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): """ADN-format record: alpha + len_bcd + TON/NPI + number + CCP + ext1.""" 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: return {'raw': raw.hex().upper()} 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 '' number = _decode_bcd_address(raw[-13:-2]) ccp = raw[-2] @@ -2269,8 +2297,33 @@ def _decode_file_data(fid, raw, p1=None): 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): """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'): return 'empty' if f.get('imsi'): @@ -2535,9 +2588,12 @@ def decode_message(raw_data, prev=None): 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 + 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) + if file_dec: + result['file'] = file_dec # SW if sw_bytes: @@ -2551,17 +2607,23 @@ def decode_message(raw_data, prev=None): prev_ins = prev.get('ins') 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 + 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]) + 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]), - } + if not _is_record_file(prev['sel']): + result['file'] = _stale_file_note(prev['sel']) + else: + # 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]) if response: diff --git a/tests/test_decode.py b/tests/test_decode.py index 1fdda05..8c2e5ed 100644 --- a/tests/test_decode.py +++ b/tests/test_decode.py @@ -869,6 +869,17 @@ class TestFileDecoders(unittest.TestCase): self.assertEqual(f['name'], 'B HA 1 Sic') 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): from simtrace2_pysniff.server.decode import _decode_file_data f = _decode_file_data('6f38', bytes.fromhex('07')) @@ -983,6 +994,38 @@ class TestSelectionTracking(unittest.TestCase): msgs = db.get_messages(sid) 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__': unittest.main()