From 4e67948dcedfd0c1a3eec63680b4f611122087e7 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: Thu, 13 Aug 2026 09:00:14 +0300 Subject: [PATCH] Add CAT deep decode: TR local info (PLI) + proactive command bodies TERMINAL RESPONSE additional information: - PLI local-info decoders by tag: Location (PLMN/LAC/Cell), IMEI (BCD), date-time (TS 23.040 SCTS), language, access technology, timing advance, battery state, network search mode, NMR - Duration (POLL INTERVAL) and Item Identifier (SELECT ITEM) Proactive commands (FETCH, D0 TLV, tag&0x7F normalized): - Annex A text (0x80/0x81/0x82 UCS2 variants, else GSM-7 default) and DCS text decoders - SET UP MENU / SELECT ITEM title + items, SEND SHORT MESSAGE TPDU, SET UP EVENT LIST events, POLL INTERVAL duration, DISPLAY TEXT etc. Tests: 96 (TR PLI, proactive, Annex A variants, DCS text). --- simtrace2_pysniff/server/decode.py | 271 ++++++++++++++++++++++++++++- tests/test_decode.py | 104 +++++++++++ 2 files changed, 367 insertions(+), 8 deletions(-) diff --git a/simtrace2_pysniff/server/decode.py b/simtrace2_pysniff/server/decode.py index dab0136..55a78e4 100644 --- a/simtrace2_pysniff/server/decode.py +++ b/simtrace2_pysniff/server/decode.py @@ -600,17 +600,144 @@ TR_RESULTS = { } -def _decode_tr_result(body): - """Decode the Result TLV of a TERMINAL RESPONSE body.""" +def _swap_nibbles(h): + """Swap the nibbles of each hex byte in a hex string.""" + return ''.join(b[1] + b[0] for b in (h[i:i + 2] for i in range(0, len(h), 2))) + + +def _decode_plmn(b): + """Decode a 3-byte PLMN (TS 24.008) into {'mcc', 'mnc'}.""" + if len(b) < 3: + return None + h = b.hex() + mcc = (h[1] + h[0] + h[3]).upper().strip('F') or '0' + mnc = (h[5] + h[4] + h[2]).upper().strip('F') or '0' + return {'mcc': mcc, 'mnc': mnc} + + +def _decode_imei(val): + """Decode an 8-byte BCD IMEI (TS 24.008 Mobile Identity) into a string.""" + swapped = _swap_nibbles(val.hex()) + if not swapped: + return None + # First nibble is the odd/even indicator (0x8|…) with digit 1 following; + # drop it and any trailing 'F' padding. + digits = swapped[1:].rstrip('fF') + return digits or None + + +# TS 102 223 §6.8.7 Local information — COMPREHENSION-TLV tags (base form) +PLI_LOCATION_INFO = 0x13 +PLI_IMEI = 0x14 +PLI_NMR = 0x16 +PLI_DATETIME = 0x26 +PLI_LANGUAGE = 0x2D +PLI_ACCESS_TECH = 0x3F +PLI_TIMING_ADVANCE = 0x46 +PLI_BATTERY = 0x63 +PLI_SEARCH_MODE = 0x65 + +_ACCESS_TECH_NAMES = { + 0x00: 'GSM', 0x01: 'ANSI-136 (TIA/EIA-553)', 0x02: 'IS-136 (TIA/EIA-136)', + 0x03: 'UTRAN', 0x04: 'TETRA', 0x05: 'cdma2000 1xRTT', + 0x06: 'cdma2000 HRPD', 0x07: 'E-UTRAN', 0x08: 'eHRPD', 0x0A: 'NR', +} + +_BATTERY_NAMES = { + 0x00: 'very low', 0x01: 'low', 0x02: 'average', 0x03: 'good', 0x05: 'full', +} + +_ME_STATUS_NAMES = {0x00: 'idle', 0x01: 'not idle'} + +_SEARCH_MODE_NAMES = {0x00: 'manual', 0x01: 'automatic'} + + +def _decode_datetime(val): + """Decode TS 23.040 Service-Centre-Time-Stamp (7 bytes) into a string.""" + if len(val) < 7: + return None + b = val[:7] + bcd = lambda x: (x >> 4) * 10 + (x & 0x0F) + yy, mm, dd, hh, mi, ss = (bcd(x) for x in b[:6]) + tz_raw = b[6] + if (tz_raw & 0x0F) == 0x0F: + tz_str = 'unknown' + else: + tz_neg = bool(tz_raw & 0x08) + tz_q = (tz_raw & 0x07) * 10 + (tz_raw >> 4) + tz_str = f"{'-' if tz_neg else '+'}{tz_q // 4:02d}:{(tz_q % 4) * 15:02d}" + return f'20{yy:02d}-{mm:02d}-{dd:02d} {hh:02d}:{mi:02d}:{ss:02d} (UTC{tz_str})' + + +def _decode_local_info(tag, value): + """Decode a PLI local-information TLV value into a display dict.""" + if tag == PLI_LOCATION_INFO: + if len(value) >= 7: + plmn = _decode_plmn(value[:3]) or {} + lac = value[3:5].hex().upper() + cell = value[5:].hex().upper() + loc = f"MCC {plmn.get('mcc', '?')} MNC {plmn.get('mnc', '?')} · LAC 0x{lac} · Cell 0x{cell}" + return {'label': 'Location', 'value': loc} + elif tag == PLI_IMEI: + imei = _decode_imei(value) + if imei: + return {'label': 'IMEI', 'value': imei} + elif tag == PLI_DATETIME: + dt = _decode_datetime(value) + if dt: + return {'label': 'Date/time', 'value': dt} + elif tag == PLI_LANGUAGE: + lang = value.decode('ascii', 'replace').strip() or value.hex().upper() + return {'label': 'Language', 'value': lang} + elif tag == PLI_ACCESS_TECH: + names = [_ACCESS_TECH_NAMES.get(b, f'0x{b:02X}') for b in value] + return {'label': 'Access technology', 'value': ', '.join(names)} + elif tag == PLI_TIMING_ADVANCE: + if len(value) >= 2: + status = _ME_STATUS_NAMES.get(value[0], f'0x{value[0]:02X}') + return {'label': 'Timing advance', 'value': f'{status}, TA {value[1]}'} + elif tag == PLI_BATTERY: + if value: + return {'label': 'Battery', 'value': _BATTERY_NAMES.get(value[0], f'0x{value[0]:02X}')} + elif tag == PLI_SEARCH_MODE: + if value: + return {'label': 'Search mode', 'value': _SEARCH_MODE_NAMES.get(value[0], f'0x{value[0]:02X}')} + elif tag == PLI_NMR: + return {'label': 'Network measurement results', 'value': value.hex().upper()} + return None + + +def _decode_tr_response(body): + """Decode the TLVs of a TERMINAL RESPONSE body. + + Returns a dict with the Result (code/name), and any additional + information: Duration (0x04), Item Identifier (0x05) and the + PROVIDE LOCAL INFORMATION data objects. + """ + result = {} + local = {} for tag, _length, value in parse_tlv(body): if tag in (0x03, 0x83) and value: code = value[0] - return { - 'code': f'0x{code:02X}', - 'name': TR_RESULTS.get(code), - 'raw': value.hex().upper(), - } - return None + result['code'] = f'0x{code:02X}' + result['name'] = TR_RESULTS.get(code) + result['raw'] = value.hex().upper() + elif tag == 0x04 and len(value) >= 2: # Duration (POLL INTERVAL) + result['duration'] = (value[0] << 8) | value[1] + elif tag == 0x05 and value: # Item Identifier (SELECT ITEM) + result['item_identifier'] = value[0] + else: + info = _decode_local_info(tag, value) + if info: + local[info['label']] = info['value'] + if local: + result['local_info'] = local + return result or None + + +def _decode_tr_result(body): + """Back-compat wrapper returning the Result TLV of a TERMINAL RESPONSE.""" + return _decode_tr_response(body) # TS 102 221 Table 11.5 — file descriptor byte @@ -868,6 +995,69 @@ def _decode_dcs(dcs): return encoding, (dcs & 0x03) +def _decode_gsm7_octets(data): + """Decode 8-bit-per-octet GSM 7-bit default alphabet text (TS 102 221).""" + out = [] + i = 0 + while i < len(data): + b = data[i] + if b == 0x1B and i + 1 < len(data): + out.append(GSM7_EXTENSION.get(data[i + 1], ' ')) + i += 2 + else: + out.append(GSM7_ALPHABET[b] if b < len(GSM7_ALPHABET) else '?') + i += 1 + return ''.join(out) + + +def _decode_annex_a(raw): + """Decode a TS 102 221 Annex A text string (GsmOrUcs2). + + Magic prefix 0x80/0x81/0x82 → UCS-2 variants; otherwise GSM 7-bit + default alphabet coded one octet per character. + """ + if not raw: + return '' + if raw == b'\xff' * len(raw): + return '' + if raw[0] == 0x80: + return raw[1:].decode('utf_16_be', 'replace') + if raw[0] == 0x81 and len(raw) >= 3: + num_chars = raw[1] + base_ptr = raw[2] << 7 + out = [] + for ch in raw[3:3 + num_chars]: + if ch & 0x80: + out.append(chr((ch & 0x7F) + base_ptr)) + else: + out.append(GSM7_ALPHABET[ch] if ch < len(GSM7_ALPHABET) else '?') + return ''.join(out) + if raw[0] == 0x82 and len(raw) >= 4: + num_chars = raw[1] + base_ptr = (raw[2] << 8) | raw[3] + out = [] + for ch in raw[4:4 + num_chars]: + if ch & 0x80: + out.append(chr((ch & 0x7F) + base_ptr)) + else: + out.append(GSM7_ALPHABET[ch] if ch < len(GSM7_ALPHABET) else '?') + return ''.join(out) + return _decode_gsm7_octets(raw) + + +def _decode_dcs_text(raw): + """Decode a Text String (TS 102 223 §8.15): DCS byte + text.""" + if not raw or len(raw) < 2: + return raw.hex() if raw else '' + dcs = raw[0] + data = raw[1:] + if (dcs & 0x0C) == 0x08: + return data.decode('utf_16_be', 'replace') + if (dcs & 0x0C) == 0x04: + return data.decode('latin-1', 'replace') + return _decode_gsm7_octets(data) + + def _decode_sm_tpdu(data): """Decode a GSM 03.40 (TS 23.040) SM TPDU (SMS-DELIVER / SMS-SUBMIT).""" if not data: @@ -1009,6 +1199,69 @@ def _decode_envelope(body): return result +# ──────────────────── Proactive command-body decoder ──────────────────── + +# TS 102 223 §6.6 data-object tags (base value; comprehension bit masked off) +_P_CMD_DETAILS = 0x01 +_P_DEVICE_IDS = 0x02 +_P_DURATION = 0x04 +_P_ALPHA_ID = 0x05 +_P_ADDRESS = 0x06 +_P_SMS_TPDU = 0x0B +_P_TEXT_STRING = 0x0D +_P_ITEM = 0x0F +_P_RESPONSE_LEN = 0x11 +_P_EVENT_LIST = 0x19 + + +def _decode_proactive(body): + """Decode a FETCH proactive command body (D0 TLV) into a dict. + + Unwraps the D0 BER-TLV and decodes the command's data objects, + dispatching on the Type of Command from the Command Details (0x81). + """ + if not body: + return None + inner = None + for tag, _length, value in parse_tlv(body): + if tag == 0xD0: + inner = parse_tlv(value) + break + if inner is None: + return None + + cmd_type = None + result = {} + items = [] + for tag, _length, value in inner: + base = tag & 0x7F + if base == _P_CMD_DETAILS and len(value) >= 2: + cmd_type = value[1] + elif base == _P_ALPHA_ID and value: + result['title'] = _decode_annex_a(value) + elif base == _P_TEXT_STRING and value: + result['text'] = _decode_dcs_text(value) + elif base == _P_ITEM and len(value) >= 2: + items.append({'id': value[0], 'text': _decode_annex_a(value[1:])}) + elif base == _P_SMS_TPDU and value: + result['tpdu'] = _decode_sm_tpdu(value) + elif base == _P_DURATION and len(value) >= 2: + result['duration'] = (value[0] << 8) | value[1] + elif base == _P_RESPONSE_LEN and len(value) >= 2: + result['response_length'] = {'min': value[0], 'max': value[1]} + elif base == _P_EVENT_LIST and value: + result['events'] = [EVENT_TYPES.get(e, f'0x{e:02X}') for e in value] + elif base == _P_ADDRESS and value: + result['address'] = _decode_bcd_address(value) + + if cmd_type is None: + return None + result['type'] = CAT_COMMAND_TYPES.get(cmd_type, f'0x{cmd_type:02X}') + if items: + result['items'] = items + return result + + # ──────────────────── Decode entry point ──────────────────── def decode_message(raw_data, prev=None): @@ -1093,6 +1346,8 @@ def decode_message(raw_data, prev=None): result['cmd'] = _decode_auth_cmd(body, p2) elif ins == 0xC2: # ENVELOPE command body result['cmd'] = _decode_envelope(body) + elif ins == 0x12: # FETCH → proactive command body + result['cmd'] = _decode_proactive(body) # SW if sw_bytes: diff --git a/tests/test_decode.py b/tests/test_decode.py index b3f309d..27f85ab 100644 --- a/tests/test_decode.py +++ b/tests/test_decode.py @@ -249,6 +249,51 @@ class TestTrResult(unittest.TestCase): self.assertEqual(r['name'], 'Proactive UICC session terminated by the user') +class TestTrAdditionalInfo(unittest.TestCase): + def test_poll_interval_duration(self): + r = decode_message(bytes.fromhex( + '80140000108103010300020282810301000402011E9000')) + self.assertEqual(r['response_to'], 'POLL INTERVAL') + self.assertEqual(r['response']['duration'], 286) + + def test_pli_location_info(self): + r = decode_message(bytes.fromhex( + '8014000017810301260002028281030106130952F0991EC57A43C09F9000')) + self.assertEqual( + r['response']['local_info']['Location'], + 'MCC 250 MNC 99 · LAC 0x1EC5 · Cell 0x7A43C09F') + + def test_pli_imei(self): + r = decode_message(bytes.fromhex( + '801400001681030126010202828103010014088A763201652428089000')) + self.assertEqual(r['response']['local_info']['IMEI'], + '867231056428280') + + def test_pli_datetime(self): + from simtrace2_pysniff.server.decode import _decode_datetime + self.assertEqual( + _decode_datetime(bytes.fromhex('26081214300021')), + '2026-08-12 14:30:00 (UTC+03:00)') + self.assertEqual( + _decode_datetime(bytes.fromhex('260812143000FF')), + '2026-08-12 14:30:00 (UTCunknown)') + + def test_pli_language(self): + from simtrace2_pysniff.server.decode import _decode_local_info, PLI_LANGUAGE + self.assertEqual(_decode_local_info(PLI_LANGUAGE, b'en'), + {'label': 'Language', 'value': 'en'}) + + def test_pli_battery(self): + from simtrace2_pysniff.server.decode import _decode_local_info, PLI_BATTERY + self.assertEqual(_decode_local_info(PLI_BATTERY, b'\x05'), + {'label': 'Battery', 'value': 'full'}) + + def test_pli_access_tech(self): + from simtrace2_pysniff.server.decode import _decode_local_info, PLI_ACCESS_TECH + self.assertEqual(_decode_local_info(PLI_ACCESS_TECH, bytes([0x03, 0x07])), + {'label': 'Access technology', 'value': 'UTRAN, E-UTRAN'}) + + class TestAuthCommand(unittest.TestCase): def test_3g_rand_autn(self): r = decode_message(bytes.fromhex( @@ -347,5 +392,64 @@ class TestBcdAddress(unittest.TestCase): self.assertEqual(_decode_bcd_address(bytes.fromhex('9121436587F9')), '+123456789') +class TestProactiveDecode(unittest.TestCase): + def test_setup_menu(self): + r = decode_message(bytes.fromhex( + '8012000030D02E810301250082028182050B416C6661204D6F62696C' + '658F16808112089DB0C1C2C0BEB9BAB82F53657474696E67739000')) + cmd = r['cmd'] + self.assertEqual(cmd['type'], 'SET UP MENU') + self.assertEqual(cmd['title'], 'Alfa Mobile') + self.assertEqual(cmd['items'], [{'id': 0x80, 'text': 'Настройки/Settings'}]) + + def test_setup_event_list(self): + r = decode_message(bytes.fromhex( + '801200000FD00D810301050082028182990203129000')) + self.assertEqual(r['cmd']['events'], ['Location status', 'Network Rejection']) + + def test_poll_interval_duration(self): + r = decode_message(bytes.fromhex( + '801200000FD00D8103010300820281820402011E9000')) + self.assertEqual(r['cmd']['duration'], 286) + + def test_send_short_message_tpdu(self): + r = decode_message(bytes.fromhex( + '801200004AD04881030113008202818305000B3B11FF038199F90004A531' + '494D45492038363732333130353634323832383020494D53492032353031' + '3937373030313932373935204556454E5420319000')) + self.assertEqual(r['cmd']['type'], 'SEND SHORT MESSAGE') + self.assertEqual(r['cmd']['tpdu']['mti'], 'SMS-SUBMIT') + self.assertEqual(r['cmd']['tpdu']['da'], '999') + + +class TestAnnexA(unittest.TestCase): + def test_gsm_default(self): + from simtrace2_pysniff.server.decode import _decode_annex_a + self.assertEqual(_decode_annex_a(b'Alfa Mobile'), 'Alfa Mobile') + + def test_ucs2_variant1(self): + from simtrace2_pysniff.server.decode import _decode_annex_a + self.assertEqual(_decode_annex_a(b'\x80' + 'Привет'.encode('utf_16_be')), 'Привет') + + def test_ucs2_variant2_base_ptr(self): + from simtrace2_pysniff.server.decode import _decode_annex_a + # 'Н' = U+041D, base 0x08<<7 = 0x400, offset 0x1D + self.assertEqual(_decode_annex_a(bytes.fromhex('8101089D')), 'Н') + + +class TestDcsText(unittest.TestCase): + def test_gsm7(self): + from simtrace2_pysniff.server.decode import _decode_dcs_text + self.assertEqual(_decode_dcs_text(b'\x00Alfa'), 'Alfa') + + def test_ucs2(self): + from simtrace2_pysniff.server.decode import _decode_dcs_text + self.assertEqual(_decode_dcs_text(b'\x08' + 'OK'.encode('utf_16_be')), 'OK') + + def test_latin1(self): + from simtrace2_pysniff.server.decode import _decode_dcs_text + self.assertEqual(_decode_dcs_text(b'\x04caf\xe9'), 'café') + + if __name__ == '__main__': unittest.main()