From 5ba55ac3a9fa28987f94fda4291370e5300bbf84 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: Sat, 15 Aug 2026 00:55:07 +0300 Subject: [PATCH] Decode EVENT DOWNLOAD location/caller and CELL BROADCAST DOWNLOAD CB page - EVENT DOWNLOAD: decode location information (PLMN/LAC/Cell), caller number (MT call Address), transaction identifier, subaddress - CELL BROADCAST DOWNLOAD: decode CB page header (serial, message ID, DCS, page) with raw content hex - frontend: render Caller/Transaction/Subaddress and CB page fields Tests: 114. --- frontend/index.html | 11 ++++++++ simtrace2_pysniff/server/decode.py | 42 +++++++++++++++++++++++++++--- tests/test_decode.py | 30 +++++++++++++++++++++ 3 files changed, 79 insertions(+), 4 deletions(-) diff --git a/frontend/index.html b/frontend/index.html index 4ea2b3d..0825005 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -537,9 +537,20 @@ function cmdRows(cmd) { if (cmd.events) rows.push(['Event', cmd.events.join(', ')]); if (cmd.location_status) rows.push(['Location status', cmd.location_status]); if (cmd.location_info) rows.push(['Location info', cmd.location_info]); + if (cmd.caller) rows.push(['Caller', cmd.caller]); + if (cmd.transaction_id !== undefined) rows.push(['Transaction', cmd.transaction_id]); + if (cmd.subaddress) rows.push(['Subaddress', cmd.subaddress]); if (cmd.smsc) rows.push(['SMSC', cmd.smsc]); if (cmd.tp_da) rows.push(['TP-DA', cmd.tp_da]); if (cmd.device_ids) rows.push(['Device IDs', cmd.device_ids]); + if (cmd.cb_page) { + const cb = cmd.cb_page; + if (cb.serial) rows.push(['CB serial', cb.serial]); + if (cb.message_id) rows.push(['CB message', cb.message_id]); + if (cb.dcs) rows.push(['CB DCS', cb.dcs]); + if (cb.page) rows.push(['CB page', cb.page]); + if (cb.content) rows.push(['CB content', cb.content]); + } // Proactive command fields (FETCH) if (cmd.title) rows.push(['Title', cmd.title]); diff --git a/simtrace2_pysniff/server/decode.py b/simtrace2_pysniff/server/decode.py index 7aafbac..6367565 100644 --- a/simtrace2_pysniff/server/decode.py +++ b/simtrace2_pysniff/server/decode.py @@ -1152,6 +1152,32 @@ def _decode_ud(ud, udl, dcs, udhi, result): return result +def _decode_cb_page(value): + """Decode a CELL BROADCAST DOWNLOAD CB page (TS 31.111 §8.5 / TS 23.041). + + Layout: serial number (2), message identifier (2), DCS (1), page + parameter (1), content (up to 82 octets). Content is returned as raw + hex (no text decode). + """ + if len(value) < 6: + return {'content': value.hex().upper()} + serial = value[0:2].hex().upper() + message_id = value[2:4].hex().upper() + dcs = value[4] + page_byte = value[5] + total_pages = (page_byte >> 4) & 0x0F + page_num = page_byte & 0x0F + encoding, msg_class = _decode_dcs(dcs) + dcs_str = f'0x{dcs:02X} — {encoding}' + (f' (class {msg_class})' if msg_class else '') + return { + 'serial': serial, + 'message_id': '0x' + message_id, + 'dcs': dcs_str, + 'page': f'{page_num}/{total_pages}', + 'content': value[6:].hex().upper(), + } + + def _decode_envelope(body): """Decode an ENVELOPE command body (TS 102 223 / TS 31.111).""" tlvs = parse_tlv(body) @@ -1169,7 +1195,15 @@ def _decode_envelope(body): if v: result['location_status'] = LOCATION_STATUS.get(v[0], f'0x{v[0]:02X}') elif t == 0x13: # Location information - result['location_info'] = v.hex().upper() + li = _decode_local_info(PLI_LOCATION_INFO, v) + result['location_info'] = li['value'] if li else v.hex().upper() + elif t in (0x06, 0x86): # Address (caller's number, MT call) + result['caller'] = _decode_bcd_address(v) + elif t == 0x1C: # Transaction identifier (MT call) + if v: + result['transaction_id'] = v[0] + elif t in (0x08, 0x88): # Subaddress + result['subaddress'] = v.hex().upper() elif t == 0x02: # Device identities result['device_ids'] = v.hex().upper() elif tag == 0xD5: # MO SHORT MESSAGE CONTROL @@ -1191,12 +1225,12 @@ def _decode_envelope(body): result['tpdu'] = _decode_sm_tpdu(v) elif t == 0x02: result['device_ids'] = v.hex().upper() - elif tag == 0xD2: # CELL BROADCAST DOWNLOAD (structure only) + elif tag == 0xD2: # CELL BROADCAST DOWNLOAD for t, _l, v in inner: if t == 0x02: result['device_ids'] = v.hex().upper() - else: - result[f'tag_{t:02X}'] = v.hex().upper() + elif t in (0x0C, 0x8C): # CB page + result['cb_page'] = _decode_cb_page(v) else: # D3 / D4 / D7 — device identities only for t, _l, v in inner: if t == 0x02: diff --git a/tests/test_decode.py b/tests/test_decode.py index c5d1ad6..ee78113 100644 --- a/tests/test_decode.py +++ b/tests/test_decode.py @@ -339,6 +339,36 @@ class TestEventDownload(unittest.TestCase): self.assertEqual(r['cmd']['events'], ['Location status']) self.assertEqual(r['cmd']['location_status'], 'No service') + def test_location_status_with_info(self): + r = decode_message(bytes.fromhex( + '80C2000017D615190103020282811B0100130952F02026ADBC51A16F9000')) + cmd = r['cmd'] + self.assertEqual(cmd['location_status'], 'Normal service') + self.assertEqual(cmd['location_info'], + 'MCC 250 MNC 02 · LAC 0x26AD · Cell 0xBC51A16F') + + def test_mt_call(self): + r = decode_message(bytes.fromhex( + '80C2000014D612190100020283811C010586069121436587099000')) + cmd = r['cmd'] + self.assertEqual(cmd['events'], ['MT call']) + self.assertEqual(cmd['transaction_id'], 5) + self.assertEqual(cmd['caller'], '+1234567890') + + def test_cell_broadcast(self): + cbp = ('4C511002F211' + + '040000313A0103035E047FFB0F31303238322A333036343835363234' + + '0D' * 54) + r = decode_message(bytes.fromhex( + '80C2000060D25E020283810C58' + cbp + '9000')) + cmd = r['cmd'] + self.assertEqual(cmd['type'], 'CELL BROADCAST DOWNLOAD') + self.assertEqual(cmd['cb_page']['serial'], '4C51') + self.assertEqual(cmd['cb_page']['message_id'], '0x1002') + self.assertEqual(cmd['cb_page']['page'], '1/1') + self.assertEqual(cmd['cb_page']['content'], + '040000313A0103035E047FFB0F31303238322A333036343835363234' + '0D' * 54) + class TestSmTpdu(unittest.TestCase): def _bcd(self, number):