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.
This commit is contained in:
catarrh committed 2026-08-15 00:55:07 +03:00
1 parent d32da2b375
commit 5ba55ac3a9
3 files changed
+79 -4

No files matched your search

+11
View File
@@ -537,9 +537,20 @@ function cmdRows(cmd) {
if (cmd.events) rows.push(['Event', cmd.events.join(', ')]); if (cmd.events) rows.push(['Event', cmd.events.join(', ')]);
if (cmd.location_status) rows.push(['Location status', cmd.location_status]); if (cmd.location_status) rows.push(['Location status', cmd.location_status]);
if (cmd.location_info) rows.push(['Location info', cmd.location_info]); 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.smsc) rows.push(['SMSC', cmd.smsc]);
if (cmd.tp_da) rows.push(['TP-DA', cmd.tp_da]); if (cmd.tp_da) rows.push(['TP-DA', cmd.tp_da]);
if (cmd.device_ids) rows.push(['Device IDs', cmd.device_ids]); 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) // Proactive command fields (FETCH)
if (cmd.title) rows.push(['Title', cmd.title]); if (cmd.title) rows.push(['Title', cmd.title]);
+38 -4
View File
@@ -1152,6 +1152,32 @@ def _decode_ud(ud, udl, dcs, udhi, result):
return 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): def _decode_envelope(body):
"""Decode an ENVELOPE command body (TS 102 223 / TS 31.111).""" """Decode an ENVELOPE command body (TS 102 223 / TS 31.111)."""
tlvs = parse_tlv(body) tlvs = parse_tlv(body)
@@ -1169,7 +1195,15 @@ def _decode_envelope(body):
if v: if v:
result['location_status'] = LOCATION_STATUS.get(v[0], f'0x{v[0]:02X}') result['location_status'] = LOCATION_STATUS.get(v[0], f'0x{v[0]:02X}')
elif t == 0x13: # Location information 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 elif t == 0x02: # Device identities
result['device_ids'] = v.hex().upper() result['device_ids'] = v.hex().upper()
elif tag == 0xD5: # MO SHORT MESSAGE CONTROL elif tag == 0xD5: # MO SHORT MESSAGE CONTROL
@@ -1191,12 +1225,12 @@ def _decode_envelope(body):
result['tpdu'] = _decode_sm_tpdu(v) result['tpdu'] = _decode_sm_tpdu(v)
elif t == 0x02: elif t == 0x02:
result['device_ids'] = v.hex().upper() 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: for t, _l, v in inner:
if t == 0x02: if t == 0x02:
result['device_ids'] = v.hex().upper() result['device_ids'] = v.hex().upper()
else: elif t in (0x0C, 0x8C): # CB page
result[f'tag_{t:02X}'] = v.hex().upper() result['cb_page'] = _decode_cb_page(v)
else: # D3 / D4 / D7 — device identities only else: # D3 / D4 / D7 — device identities only
for t, _l, v in inner: for t, _l, v in inner:
if t == 0x02: if t == 0x02:
+30
View File
@@ -339,6 +339,36 @@ class TestEventDownload(unittest.TestCase):
self.assertEqual(r['cmd']['events'], ['Location status']) self.assertEqual(r['cmd']['events'], ['Location status'])
self.assertEqual(r['cmd']['location_status'], 'No service') 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): class TestSmTpdu(unittest.TestCase):
def _bcd(self, number): def _bcd(self, number):