From 017c65de27b97b539cfb3dc6dc6ac157af2e606c 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: Mon, 17 Aug 2026 01:08:49 +0300 Subject: [PATCH] Add trace map view; decode PNN/SPN and 7-bit SMS TP-UD --- frontend/index.html | 170 ++++++++++++++++++++++++++++- simtrace2_pysniff/server/decode.py | 64 +++++++++-- tests/test_decode.py | 45 ++++++-- 3 files changed, 258 insertions(+), 21 deletions(-) diff --git a/frontend/index.html b/frontend/index.html index 5be24ca..44ee7ee 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -136,6 +136,11 @@ body { font: 14px/1.5 system-ui, sans-serif; background: var(--bg); color: var(- .side-item { display: flex; justify-content: space-between; padding: 3px 8px; cursor: pointer; border-radius: 4px; font-size: 13px; } .side-item:hover { background: var(--bg-input); } .side-item .count { color: var(--accent); font-weight: 600; } +.side-tabs { display: flex; margin-bottom: 8px; border-bottom: 1px solid var(--bg-input); } +.side-tab { flex: 1; padding: 6px 0; background: none; border: none; color: var(--text-muted); cursor: pointer; font-size: 12px; text-transform: uppercase; letter-spacing: 1px; border-bottom: 2px solid transparent; } +.side-tab.active { color: var(--accent); border-bottom-color: var(--accent); } +.map-item { padding: 3px 8px; cursor: pointer; border-radius: 4px; font-size: 13px; color: var(--text); word-break: break-word; } +.map-item:hover { background: var(--bg-input); } /* Status bar */ .status-bar { display: flex; padding: 8px 16px; background: var(--bg-panel); border-top: 1px solid var(--bg-input); font-size: 13px; gap: 16px; flex-shrink: 0; } @@ -250,8 +255,12 @@ body { font: 14px/1.5 system-ui, sans-serif; background: var(--bg); color: var(-
-

Command Stats

+
+ + +
—
+
@@ -332,6 +341,8 @@ const STATE = { livePoller: null, messages: [], typeCounts: [], + traceMap: [], + sideView: 'stats', pendingDeleteId: null, serverAvailable: false, serverName: '', @@ -695,6 +706,8 @@ function fileRows(f) { 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.full !== undefined) rows.push(['Full name', f.full]); + if (f.short !== undefined) rows.push(['Short name', f.short]); 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)]); @@ -916,6 +929,112 @@ function buildTypeCounts() { STATE.typeCounts = counts; } +function _ins_hex(m) { + if (m.type !== 'tpdu' || !m.decoded) return null; + return m.decoded.ins_hex; +} + +function map_file_label(m) { + const d = m.decoded; + if (!d) return null; + if (d.ins_hex === 'a4') { // SELECT + if (d.body && d.body.note) return d.body.note; + if (d.summary) return d.summary; + return null; + } + if (d.file) return d.file.ef || d.file.fid || null; + return null; +} + +const MAP_RW_INS = new Set(['a4', 'a2', 'b0', 'b1', 'b2', 'd6', 'dc']); + +function is_read_ins(ins) { return ins === 'a2' || ins === 'b0' || ins === 'b1' || ins === 'b2'; } +function is_write_ins(ins) { return ins === 'd6' || ins === 'dc'; } + +function map_other_label(m) { + const d = m.decoded; + if (d && d.ins_name === 'ENVELOPE' && d.summary) { + return 'ENVELOPE: ' + d.summary; + } + return stat_label(m); +} + +function buildTraceMap() { + const msgs = STATE.messages; + const groups = []; + let i = 0; + while (i < msgs.length) { + const ins = _ins_hex(msgs[i]); + if (ins === 'c0') { i++; continue; } // skip GET RESPONSE + if (ins && MAP_RW_INS.has(ins)) { + // Split the run into per-file records (delimited by SELECT). + const access = []; + let cur = null; + let j = i; + while (j < msgs.length) { + const jins = _ins_hex(msgs[j]); + if (jins === 'c0') { j++; continue; } // skip GET RESPONSE + if (!jins || !MAP_RW_INS.has(jins)) break; + if (jins === 'a4') { + if (cur) access.push(cur); + cur = { file: map_file_label(msgs[j]), hasRead: false, hasWrite: false, start: j, end: j }; + } else if (is_write_ins(jins)) { + if (!cur) cur = { file: map_file_label(msgs[j]), hasRead: false, hasWrite: false, start: j, end: j }; + cur.hasWrite = true; + cur.end = j; + } else { // SEARCH / READ + if (!cur) cur = { file: map_file_label(msgs[j]), hasRead: false, hasWrite: false, start: j, end: j }; + cur.hasRead = true; + cur.end = j; + } + j++; + } + if (cur) access.push(cur); + + let k = 0; + while (k < access.length) { + const rec = access[k]; + if (rec.hasRead && rec.hasWrite) { + groups.push({ label: t('Read/write:') + ' ' + (rec.file || '—'), start: rec.start, end: rec.end }); + k++; + } else if (rec.hasWrite) { + const files = []; + let end = rec.end; + let m = k; + while (m < access.length && access[m].hasWrite && !access[m].hasRead) { + if (access[m].file) files.push(access[m].file); + end = access[m].end; + m++; + } + groups.push({ label: t('Writes:') + ' ' + (files.join(', ') || '—'), start: rec.start, end }); + k = m; + } else { + const files = []; + let end = rec.end; + let m = k; + while (m < access.length && !access[m].hasWrite) { + if (access[m].file) files.push(access[m].file); + end = access[m].end; + m++; + } + groups.push({ label: t('Reads:') + ' ' + (files.join(', ') || '—'), start: rec.start, end }); + k = m; + } + } + i = j; + } else if (ins === 'f2') { + let j = i; + while (j < msgs.length && _ins_hex(msgs[j]) === 'f2') j++; + groups.push({ label: 'STATUS: ' + (j - i), start: i, end: j - 1 }); + i = j; + } else { + groups.push({ label: map_other_label(msgs[i]), start: i, end: i }); + i++; + } + } + STATE.traceMap = groups; +} + function renderSidePanel() { const entries = Object.entries(STATE.typeCounts).sort((a, b) => b[1] - a[1]); el('type-counts').innerHTML = entries.map(([k, v], i) => @@ -923,6 +1042,18 @@ function renderSidePanel() { ${k}${v}
` ).join('') || t('No commands'); + + const mapEl = el('type-map'); + mapEl.innerHTML = STATE.traceMap.map(g => + `
${g.label}
` + ).join('') || t('No commands'); + + const showMap = STATE.sideView === 'map'; + el('type-counts').style.display = showMap ? 'none' : ''; + mapEl.style.display = showMap ? '' : 'none'; + document.querySelectorAll('#side-panel .side-tab').forEach(tab => { + tab.classList.toggle('active', tab.dataset.view === STATE.sideView); + }); } function populateFilterOptions() { @@ -959,6 +1090,21 @@ function scrollToType(typeKey) { } } +function focusMessage(idx) { + const filterEl = el('filter-select'); + const searchEl = el('search-input'); + const needRender = filterEl.value !== '' || searchEl.value !== ''; + filterEl.value = ''; + searchEl.value = ''; + if (needRender) renderMessages(true); + const entry = el('apdu-list').querySelector(`.apdu-entry[data-idx="${idx}"]`); + if (!entry) return; + entry.scrollIntoView({ block: 'center', behavior: 'smooth' }); + entry.classList.add('selected'); + setTimeout(() => entry.classList.remove('selected'), 1500); + showDetail(STATE.messages[idx], entry); +} + // ──────────────────── Render Trace ──────────────────── function isOriginalCommand(i) { if (STATE.messages[i].type !== 'tpdu') return false; @@ -1001,7 +1147,7 @@ function renderTraceEmptyState() { updateStatus(); } -function renderMessages() { +function renderMessages(skipScroll) { const list = el('apdu-list'); const searchQ = el('search-input').value.toLowerCase(); const filterType = el('filter-select').value; @@ -1015,6 +1161,7 @@ function renderMessages() { } buildTypeCounts(); + buildTraceMap(); renderSidePanel(); populateFilterOptions(); @@ -1034,7 +1181,7 @@ function renderMessages() { `; }).join(''); - list.scrollTop = list.scrollHeight; + if (!skipScroll) list.scrollTop = list.scrollHeight; updateStatus(); } @@ -1101,6 +1248,7 @@ async function startCapture() { STATE.currentSessionId = data.session_id; STATE.messages = []; buildTypeCounts(); + buildTraceMap(); renderSidePanel(); renderMessages(); updateCaptureButton(); @@ -1527,6 +1675,11 @@ const LANG_RU = { 'Connect to server': 'Подключиться к серверу', 'Click an APDU to see details': 'Нажмите на APDU для просмотра деталей', 'Command Stats': 'Статистика команд', + 'Stats': 'Статистика', + 'Map': 'Карта', + 'Reads:': 'Чтение:', + 'Writes:': 'Запись:', + 'Read/write:': 'Чтение/запись:', 'Not connected': 'Не подключено', 'Live': 'Захват', 'Session': 'Сессия', @@ -1772,6 +1925,17 @@ function init() { $on('delete-confirm', 'click', confirmDelete); el('side-panel').addEventListener('click', e => { + const tab = e.target.closest('.side-tab'); + if (tab) { + STATE.sideView = tab.dataset.view; + renderSidePanel(); + return; + } + const mapItem = e.target.closest('.map-item'); + if (mapItem) { + focusMessage(parseInt(mapItem.dataset.start)); + return; + } const item = e.target.closest('.side-item'); if (!item) return; scrollToType(item.dataset.key); diff --git a/simtrace2_pysniff/server/decode.py b/simtrace2_pysniff/server/decode.py index 7f697e3..88c8def 100644 --- a/simtrace2_pysniff/server/decode.py +++ b/simtrace2_pysniff/server/decode.py @@ -997,7 +997,7 @@ def _decode_datetime(val): b = val[:7] if any((x >> 4) > 9 or (x & 0x0F) > 9 for x in b[:6]): return None - bcd = lambda x: (x >> 4) * 10 + (x & 0x0F) + bcd = lambda x: (x & 0x0F) * 10 + (x >> 4) yy, mm, dd, hh, mi, ss = (bcd(x) for x in b[:6]) if not (1 <= mm <= 12 and 1 <= dd <= 31 and hh <= 23 and mi <= 59 and ss <= 59): return None @@ -1547,10 +1547,12 @@ def _decode_gsm7_octets(data): def _decode_annex_a(raw): - """Decode a TS 102 221 Annex A text string (GsmOrUcs2). + """Decode a TS 102 221 Annex A / TS 24.008 text string (GsmOrUcs2). - Magic prefix 0x80/0x81/0x82 → UCS-2 variants; otherwise GSM 7-bit - default alphabet coded one octet per character. + Magic prefix 0x80/0x81/0x82 → UCS-2 variants (Annex A); 0x90 → UCS-2; + 0x83-0x87 → GSM 7-bit packed; 0x88-0x8F → UCS-2 with the high bit of + the first octet cleared; otherwise GSM 7-bit default alphabet, packed + when any octet has the high bit set, else one octet per character. """ if not raw: return '' @@ -1558,6 +1560,12 @@ def _decode_annex_a(raw): return '' if raw[0] == 0x80: return raw[1:].decode('utf_16_be', 'replace') + if raw[0] == 0x90: + data = raw[1:] + null = data.find(b'\x00\x00') + if null != -1: + data = data[:null] + return data.decode('utf-16-be', 'replace') if raw[0] == 0x81 and len(raw) >= 3: num_chars = raw[1] base_ptr = raw[2] << 7 @@ -1578,7 +1586,22 @@ def _decode_annex_a(raw): else: out.append(GSM7_ALPHABET[ch] if ch < len(GSM7_ALPHABET) else '?') return ''.join(out) - return _decode_gsm7_octets(raw) + if 0x83 <= raw[0] <= 0x87: + data = raw[1:].rstrip(b'\xff') + return _decode_gsm7(data, None) if data else '' + if raw[0] & 0x80: + ucs2 = bytearray(raw) + ucs2[0] &= 0x7F + while len(ucs2) >= 2 and ucs2[-1] == 0xFF and ucs2[-2] == 0xFF: + ucs2 = ucs2[:-2] + null = ucs2.find(b'\x00\x00') + if null != -1: + ucs2 = ucs2[:null] + return bytes(ucs2).decode('utf-16-be', 'replace') + data = raw.rstrip(b'\xff') + if all(b < 0x80 for b in data): + return _decode_gsm7_octets(data) + return _decode_gsm7(data, None) def _decode_dcs_text(raw): @@ -1834,6 +1857,7 @@ def _decode_ud(ud, udl, dcs, udhi, result): fill_bits = (udhl + 1) * 8 n = udl - ((fill_bits + 6) // 7) result['text'] = _decode_gsm7(body, n) + result['payload'] = body.hex().upper() elif encoding == 'UCS2': result['text'] = body[:udl].decode('utf-16-be', errors='replace') else: @@ -1844,6 +1868,7 @@ def _decode_ud(ud, udl, dcs, udhi, result): else: if encoding == 'GSM 7-bit': result['text'] = _decode_gsm7(ud_data, udl) + result['payload'] = ud_data.hex().upper() elif encoding == 'UCS2': result['text'] = ud_data[:udl].decode('utf-16-be', errors='replace') else: @@ -2334,17 +2359,34 @@ def _decode_arr(raw, p1=None): return {'rules': entries} if entries else {'raw': raw.hex().upper()} +def _decode_pnn_text(value): + """Decode a TS 24.008 §10.5.3.5a Network Name (DCS byte + text).""" + if len(value) < 2: + return '' + dcs = value[0] + coding = (dcs >> 4) & 0x07 + data = value[1:].rstrip(b'\xff') + if coding == 1: # UCS2 + null = data.find(b'\x00\x00') + if null != -1: + data = data[:null] + return data.decode('utf-16-be', 'replace') + if coding == 0: # GSM 7-bit default alphabet + if all(b < 0x80 for b in data): + return _decode_gsm7_octets(data) + return _decode_gsm7(data, None) + return '' + + def _decode_pnn(raw, p1=None): + if not raw or all(b == 0xff for b in raw): + return {'empty': True} names = {} for tag, _length, value in parse_tlv(raw): if tag == 0x43: - names['full'] = _decode_annex_a(value) + names['full'] = _decode_pnn_text(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 + names['short'] = _decode_pnn_text(value) return names or {'raw': raw.hex().upper()} diff --git a/tests/test_decode.py b/tests/test_decode.py index ff63885..046825f 100644 --- a/tests/test_decode.py +++ b/tests/test_decode.py @@ -95,7 +95,7 @@ class TestCatDecoding(unittest.TestCase): def test_envelope_sms_pp_download(self): r = decode_message(bytes.fromhex( '80C2000022D120020283810607919740430900F40B1104038154F50004' - '260812143000210248699000')) + '628021410300210248699000')) self.assertEqual(r['ins_name'], 'ENVELOPE') self.assertEqual(r['cat_command'], 'SMS-PP DOWNLOAD') cmd = r['cmd'] @@ -326,10 +326,10 @@ class TestTrAdditionalInfo(unittest.TestCase): def test_pli_datetime(self): from simtrace2_pysniff.server.decode import _decode_datetime self.assertEqual( - _decode_datetime(bytes.fromhex('26081214300021')), + _decode_datetime(bytes.fromhex('62802141030021')), '2026-08-12 14:30:00 (UTC+03:00)') self.assertEqual( - _decode_datetime(bytes.fromhex('260812143000FF')), + _decode_datetime(bytes.fromhex('628021410300FF')), '2026-08-12 14:30:00 (UTCunknown)') def test_datetime_invalid_bcd(self): @@ -338,7 +338,7 @@ class TestTrAdditionalInfo(unittest.TestCase): def test_datetime_invalid_semantics(self): from simtrace2_pysniff.server.decode import _decode_datetime - self.assertIsNone(_decode_datetime(bytes.fromhex('02209121654021'))) + self.assertIsNone(_decode_datetime(bytes.fromhex('62310100000000'))) def test_pli_language(self): from simtrace2_pysniff.server.decode import _decode_local_info, PLI_LANGUAGE @@ -532,7 +532,7 @@ class TestSmTpdu(unittest.TestCase): def test_deliver_scts(self): from simtrace2_pysniff.server.decode import _decode_sm_tpdu oa = b'\x91' + self._bcd('79031234567') - scts = bytes.fromhex('26081214300021') + scts = bytes.fromhex('62802141030021') tpdu = (b'\x00' + bytes([11]) + oa + b'\x00\x00' + scts + b'\x00') r = _decode_sm_tpdu(tpdu) self.assertEqual(r['scts'], '2026-08-12 14:30:00 (UTC+03:00)') @@ -609,7 +609,7 @@ class TestSmTpdu(unittest.TestCase): def test_submit_vp_absolute(self): from simtrace2_pysniff.server.decode import _decode_sm_tpdu da = b'\x81' + self._bcd('999') - scts = bytes.fromhex('26081214300021') + scts = bytes.fromhex('62802141030021') tpdu = b'\x11' + b'\x2A' + bytes([len(da)]) + da + b'\x00\x00' + scts + b'\x00' r = _decode_sm_tpdu(tpdu) self.assertEqual(r['vp'], '2026-08-12 14:30:00 (UTC+03:00)') @@ -747,6 +747,37 @@ class TestAnnexA(unittest.TestCase): # 'Н' = U+041D, base 0x08<<7 = 0x400, offset 0x1D self.assertEqual(_decode_annex_a(bytes.fromhex('8101089D')), 'Н') + def test_ucs2_variant_90(self): + from simtrace2_pysniff.server.decode import _decode_annex_a + self.assertEqual(_decode_annex_a(b'\x90' + 'Привет'.encode('utf_16_be')), 'Привет') + + def test_gsm7_packed(self): + from simtrace2_pysniff.server.decode import _decode_annex_a + self.assertEqual(_decode_annex_a(bytes.fromhex('41b6390c')), 'Alfa') + + +class TestPnnSpn(unittest.TestCase): + def test_pnn_full_name(self): + from simtrace2_pysniff.server.decode import _decode_pnn + r = _decode_pnn(bytes.fromhex('43058441b6390c' + 'ff' * 13)) + self.assertEqual(r['full'], 'Alfa') + + def test_pnn_ucs2(self): + from simtrace2_pysniff.server.decode import _decode_pnn + v = b'\x90' + 'Привет'.encode('utf-16-be') + r = _decode_pnn(b'\x43' + bytes([len(v)]) + v) + self.assertEqual(r['full'], 'Привет') + + def test_spn_ucs2_90(self): + from simtrace2_pysniff.server.decode import _decode_spn + r = _decode_spn(b'\x00' + b'\x90' + 'Привет'.encode('utf-16-be')) + self.assertEqual(r['name'], 'Привет') + + def test_spn_gsm7(self): + from simtrace2_pysniff.server.decode import _decode_spn + r = _decode_spn(b'\x00' + bytes.fromhex('41b6390c')) + self.assertEqual(r['name'], 'Alfa') + class TestDcsText(unittest.TestCase): def test_gsm7(self): @@ -849,7 +880,7 @@ class TestSummary(unittest.TestCase): def test_envelope_sms_pp(self): r = decode_message(bytes.fromhex( '80C2000022D120020283810607919740430900F40B1104038154F50004' - '260812143000210248699000')) + '628021410300210248699000')) self.assertEqual(r['summary'], 'SMSC +79043490004, SMS-DELIVER from 455 «Hi»') def test_terminal_response(self):