From 1c2d6af9e481c0963a1b2244d59584cc95a558dc 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 23:08:54 +0300 Subject: [PATCH] Decode OTA secured command/response packets in SMS TP-UD; collapse RFU P1/P2 --- frontend/index.html | 33 ++++- simtrace2_pysniff/server/decode.py | 199 ++++++++++++++++++++++++++--- tests/test_decode.py | 63 +++++++-- 3 files changed, 267 insertions(+), 28 deletions(-) diff --git a/frontend/index.html b/frontend/index.html index 1baa456..5be24ca 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -647,6 +647,33 @@ function cmdRows(cmd) { } if (t.text !== undefined) rows.push(['TP-UD', t.text]); if (t.payload) rows.push(['TP-UD (payload)', t.payload]); + + const sec = t.secured; + if (sec && !sec.raw) { + rows.push(['Secured packet', 'CPL ' + sec.cpl + ' · CHL ' + sec.chl]); + const s = sec.spi || {}; + rows.push(['SPI', '0x' + (s.hex || '') + ' — ' + (s.counter || '') + + (s.ciphering ? ', ciphering' : '') + ', ' + (s.rc_cc_ds || '') + ', ' + (s.por || '')]); + rows.push(['KIc', 'key ' + sec.kic.key + ' — ' + sec.kic.algo]); + rows.push(['KID', 'key ' + sec.kid.key + ' — ' + sec.kid.algo]); + rows.push(['TAR', '0x' + sec.tar]); + rows.push(['CNTR', '0x' + sec.cntr]); + rows.push(['PCNTR', sec.pcntr]); + if (sec.rc_cc_ds) rows.push(['RC/CC/DS', sec.rc_cc_ds]); + rows.push(['Secured data', sec.data]); + } + + const rp = t.response_packet; + if (rp && !rp.raw) { + rows.push(['Response packet', 'RPL ' + rp.rpl + ' · RHL ' + rp.rhl]); + rows.push(['TAR', '0x' + rp.tar]); + rows.push(['CNTR', '0x' + rp.cntr]); + rows.push(['PCNTR', rp.pcntr]); + const st = rp.status || {}; + rows.push(['Status', '0x' + st.code + (st.name ? ' — ' + st.name : '')]); + if (rp.rc_cc_ds) rows.push(['RC/CC/DS', rp.rc_cc_ds]); + rows.push(['Response data', rp.data]); + } } if (!rows.length) return ''; @@ -822,7 +849,11 @@ function showDetail(msg, entry) { let p1p2 = ''; if (decoded.p1p2) { - p1p2 = `P1-P20x${hx(decoded.p1p2.value, 4)} — ${decoded.p1p2.label}: ${decoded.p1p2.value}`; + if (decoded.p1p2.unused) { + p1p2 = `P1, P20x${hx(decoded.p1p2.value, 4)} (RFU)`; + } else { + p1p2 = `P1-P20x${hx(decoded.p1p2.value, 4)} — ${decoded.p1p2.label}: ${decoded.p1p2.value}`; + } } let p1 = ''; if (decoded.p1) { diff --git a/simtrace2_pysniff/server/decode.py b/simtrace2_pysniff/server/decode.py index 8c83a10..7f697e3 100644 --- a/simtrace2_pysniff/server/decode.py +++ b/simtrace2_pysniff/server/decode.py @@ -535,6 +535,7 @@ APDU_SPEC = { }, 0x21: { 'name': 'VERIFY', + 'p1p2': {'unused': True}, 'body': {'label': 'Data'}, }, 0x24: { @@ -579,6 +580,7 @@ APDU_SPEC = { }, 0x84: { 'name': 'GET CHALLENGE', + 'p1p2': {'unused': True}, 'body': None, }, 0x70: { @@ -589,34 +591,40 @@ APDU_SPEC = { }, 0xC0: { 'name': 'GET RESPONSE', + 'p1p2': {'unused': True}, 'body': None, }, 0xC2: { 'name': 'ENVELOPE', + 'p1p2': {'unused': True}, 'body': {'label': 'TLV data'}, }, 0x12: { 'name': 'FETCH', + 'p1p2': {'unused': True}, 'body': None, }, 0x14: { 'name': 'TERMINAL RESPONSE', + 'p1p2': {'unused': True}, 'body': {'label': 'TLV data'}, }, 0x32: { 'name': 'INCREASE', - 'p1p2': {'fmt': 'uint16be', 'label': 'Value'}, + 'p1p2': {'unused': True}, 'body': {'label': 'Data'}, }, 0x04: { 'name': 'DEACTIVATE FILE', - 'p1p2': {'fmt': 'uint16be', 'label': 'File ID'}, - 'body': None, + 'p1': {0x00: 'EF by file ID', 0x08: 'Path from MF', 0x09: 'Path from current DF'}, + 'p2': {0x00: 'No indication'}, + 'body': {'label': 'File ID/path'}, }, 0x44: { 'name': 'ACTIVATE FILE', - 'p1p2': {'fmt': 'uint16be', 'label': 'File ID'}, - 'body': None, + 'p1': {0x00: 'EF by file ID', 0x08: 'Path from MF', 0x09: 'Path from current DF'}, + 'p2': {0x00: 'No indication'}, + 'body': {'label': 'File ID/path'}, }, 0xF2: { 'name': 'STATUS', @@ -627,6 +635,7 @@ APDU_SPEC = { }, 0xE0: { 'name': 'CREATE FILE', + 'p1p2': {'unused': True}, 'body': {'label': 'TLV data'}, }, 0xE4: { @@ -637,6 +646,7 @@ APDU_SPEC = { }, 0xAA: { 'name': 'TERMINAL CAPABILITY', + 'p1p2': {'unused': True}, 'body': {'label': 'TLV data'}, }, 0x73: { @@ -649,6 +659,7 @@ APDU_SPEC = { }, 0x78: { 'name': 'GET IDENTITY', + 'p1p2': {'unused': True}, 'body': {'label': 'TLV data'}, }, 0xA2: { @@ -685,10 +696,12 @@ APDU_SPEC = { }, 0x10: { 'name': 'TERMINAL PROFILE', + 'p1p2': {'unused': True}, 'body': {'label': 'TLV data'}, }, 0x76: { 'name': 'SUSPEND UICC', + 'p1p2': {'unused': True}, 'body': None, }, 0xCA: { @@ -703,6 +716,7 @@ APDU_SPEC = { }, 0xE2: { 'name': 'STORE DATA', + 'p1p2': {'unused': True}, 'body': {'label': 'TLV data'}, }, } @@ -1444,6 +1458,8 @@ IEI_NAMES = { 0x24: 'National Language Single Shift', 0x25: 'National Language Locking Shift', 0x26: 'Filler', + 0x70: 'Command Packet Identifier (CPI)', + 0x71: 'Response Packet Identifier (RPI)', } _SPECIAL_SMS_TYPES = {0: 'voice', 1: 'fax', 2: 'email', 3: 'other'} @@ -1656,22 +1672,164 @@ def _decode_sm_tpdu(data): return result +# TS 102 225 §5.1.1 — SPI bit-field names +_OTA_COUNTER = {0: 'no counter', 1: 'counter (no replay/seq check)', + 2: 'counter must be higher', 3: 'counter must be one higher'} +_OTA_RC_CC_DS = {0: 'none', 1: 'RC', 2: 'CC', 3: 'DS'} +_OTA_POR = {0: 'no PoR', 1: 'PoR required', 2: 'PoR only on error'} + +# TS 102 225 §5.1.2 / §5.1.3 — KIc / KID algorithms (low nibble) +_KIC_ALGO = {0: 'implicit', 1: 'DES', 2: 'AES-CBC', 5: '3DES-CBC (2 keys)', 9: '3DES-CBC (3 keys)'} +_KID_CC_ALGO = {0: 'implicit', 1: 'DES', 5: '3DES-CBC', 2: 'AES-CMAC'} +_KID_RC_ALGO = {0: 'implicit', 1: 'CRC16', 5: 'CRC32', 3: 'proprietary'} + +# TS 102 225 §5.2 Table 5 + TS 31.115 §7 — Response status codes +RESPONSE_STATUS = { + 0x00: 'PoR OK', + 0x01: 'RC/CC/DS failed', + 0x02: 'CNTR low', + 0x03: 'CNTR high', + 0x04: 'CNTR blocked', + 0x05: 'Ciphering error', + 0x06: 'Unidentified security error', + 0x07: 'Insufficient memory', + 0x08: 'More time needed', + 0x09: 'TAR unknown', + 0x0A: 'Insufficient security level', + 0x0B: 'Actual response data to be sent using SMS-SUBMIT', + 0x0C: 'Actual response data to be sent using USSD', +} + + +def _udh_ieis(udh): + """Extract the numeric IEI values from a raw UDH byte string.""" + ieis = set() + i = 0 + while i + 2 <= len(udh): + ieis.add(udh[i]) + i += 2 + udh[i + 1] + return ieis + + +def _decode_secured_packet(body): + """Decode a secured Command Packet (TS 102 225 §5.1 / TS 31.115 §4.2). + + *body* is the TP-UD remainder after the UDH (CPI IEI 0x70). Returns a + dict of the decoded header fields; ``{'raw': hex}`` when unparseable. + """ + if not body or len(body) < 16: + return {'raw': body.hex().upper()} if body else {} + cpl = int.from_bytes(body[:2], 'big') + chl = body[2] + spi = body[3:5] + kic = body[5] + kid = body[6] + tar = body[7:10] + cntr = body[10:15] + pcntr = body[15] + spi1, spi2 = spi[0], spi[1] + rc_cc_ds_kind = spi1 & 0x03 + rc_cc_ds_len = max(0, chl - 13) # 13 = SPI+KIc+KID+TAR+CNTR+PCNTR + rc_cc_ds = body[16:16 + rc_cc_ds_len] + data = body[16 + rc_cc_ds_len:] + + result = { + 'cpl': cpl, + 'chl': chl, + 'spi': { + 'hex': spi.hex().upper(), + 'counter': _OTA_COUNTER.get((spi1 >> 4) & 0x03, 'reserved'), + 'ciphering': bool(spi1 & 0x08), + 'rc_cc_ds': _OTA_RC_CC_DS.get(rc_cc_ds_kind, 'reserved'), + 'por': _OTA_POR.get(spi2 & 0x03, 'reserved'), + 'por_rc_cc_ds': _OTA_RC_CC_DS.get((spi2 >> 2) & 0x03, 'reserved'), + 'por_ciphered': bool(spi2 & 0x20), + }, + 'kic': {'key': kic >> 4, 'algo': _KIC_ALGO.get(kic & 0x0F, 'reserved')}, + 'tar': tar.hex().upper(), + 'cntr': cntr.hex().upper(), + 'pcntr': pcntr, + 'data': data.hex().upper(), + } + if rc_cc_ds_kind == 2: # CC + result['kid'] = {'key': kid >> 4, 'algo': _KID_CC_ALGO.get(kid & 0x0F, 'reserved')} + elif rc_cc_ds_kind == 1: # RC + result['kid'] = {'key': kid >> 4, 'algo': _KID_RC_ALGO.get(kid & 0x0F, 'reserved')} + else: + result['kid'] = {'key': kid >> 4, 'algo': 'implicit'} + if rc_cc_ds: + result['rc_cc_ds'] = rc_cc_ds.hex().upper() + return result + + +def _decode_response_packet(body): + """Decode a Response Packet (TS 102 225 §5.2 / TS 31.115 §4.4). + + *body* is the TP-UD remainder after the UDH (RPI IEI 0x71). Returns a + dict of the decoded header fields; ``{'raw': hex}`` when unparseable. + """ + if not body or len(body) < 13: + return {'raw': body.hex().upper()} if body else {} + rpl = int.from_bytes(body[:2], 'big') + rhl = body[2] + tar = body[3:6] + cntr = body[6:11] + pcntr = body[11] + status = body[12] + rc_cc_ds_len = max(0, rhl - 10) # 10 = TAR+CNTR+PCNTR+Status + rc_cc_ds = body[13:13 + rc_cc_ds_len] + data = body[13 + rc_cc_ds_len:] + + result = { + 'rpl': rpl, + 'rhl': rhl, + 'tar': tar.hex().upper(), + 'cntr': cntr.hex().upper(), + 'pcntr': pcntr, + 'status': {'code': f'{status:02X}', 'name': RESPONSE_STATUS.get(status)}, + 'data': data.hex().upper(), + } + if rc_cc_ds: + result['rc_cc_ds'] = rc_cc_ds.hex().upper() + return result + + def _decode_ud(ud, udl, dcs, udhi, result): encoding, msg_class = _decode_dcs(dcs) result['encoding'] = encoding result['msg_class'] = msg_class pid = result.get('pid') - if pid == 0x7F: # SIM data download → secured packet (show hex) - result['pid_name'] = 'SIM data download (secured packet)' - result['payload'] = ud[:udl].hex().upper() + ud_data = ud[:udl] + + # Parse UDH first — the presence of a CPI/RPI IEI identifies a secured + # command/response packet (TS 31.115) regardless of the PID. + udh_ieis = set() + body = ud_data + if udhi and ud_data: + udhl = ud_data[0] + result['udhl'] = udhl + result['udh'] = _decode_udh(ud_data[1:1 + udhl]) + udh_ieis = _udh_ieis(ud_data[1:1 + udhl]) + body = ud_data[1 + udhl:] + + if pid == 0x7F or 0x70 in udh_ieis: # secured Command Packet + if pid == 0x7F: + result['pid_name'] = 'SIM data download (secured packet)' + secured = _decode_secured_packet(body) + result['secured'] = secured + if 'raw' in secured: + result['payload'] = secured['raw'] return result - if udhi and ud: - udhl = ud[0] - result['udhl'] = udhl - result['udh'] = _decode_udh(ud[1:1 + udhl]) - body = ud[1 + udhl:] + if 0x71 in udh_ieis: # secured Response Packet (PoR) + rp = _decode_response_packet(body) + result['response_packet'] = rp + if 'raw' in rp: + result['payload'] = rp['raw'] + return result + + if udhi and ud_data: if encoding == 'GSM 7-bit': fill_bits = (udhl + 1) * 8 n = udl - ((fill_bits + 6) // 7) @@ -1685,12 +1843,12 @@ def _decode_ud(ud, udl, dcs, udhi, result): result['text'] = text else: if encoding == 'GSM 7-bit': - result['text'] = _decode_gsm7(ud, udl) + result['text'] = _decode_gsm7(ud_data, udl) elif encoding == 'UCS2': - result['text'] = ud[:udl].decode('utf-16-be', errors='replace') + result['text'] = ud_data[:udl].decode('utf-16-be', errors='replace') else: - result['payload'] = ud[:udl].hex().upper() - text = _decode_8bit_text(ud[:udl]) + result['payload'] = ud_data[:udl].hex().upper() + text = _decode_8bit_text(ud_data[:udl]) if text is not None: result['text'] = text @@ -2451,7 +2609,7 @@ def _build_summary(result): p1 = result.get('p1') p2 = result.get('p2') p1txt = p2txt = None - if 'p1p2' in result: + if 'p1p2' in result and not result['p1p2'].get('unused'): p1txt = f"{result['p1p2']['label']}: 0x{result['p1p2']['value']:04X}" else: if p1: @@ -2571,7 +2729,10 @@ def decode_message(raw_data, prev=None): if spec: if 'p1p2' in spec: offset = (p1 << 8) | p2 - result['p1p2'] = {'label': spec['p1p2']['label'], 'value': offset} + if spec['p1p2'].get('unused'): + result['p1p2'] = {'unused': True, 'value': offset} + else: + result['p1p2'] = {'label': spec['p1p2']['label'], 'value': offset} else: result['p1'] = _decode_field(spec.get('p1', {}), p1) result['p2'] = _decode_field(spec.get('p2', {}), p2) diff --git a/tests/test_decode.py b/tests/test_decode.py index 8f465db..ff63885 100644 --- a/tests/test_decode.py +++ b/tests/test_decode.py @@ -487,11 +487,47 @@ class TestSmTpdu(unittest.TestCase): def test_sim_data_download(self): from simtrace2_pysniff.server.decode import _decode_sm_tpdu oa = b'\x91' + self._bcd('79031234567') - tpdu = (b'\x00' + bytes([11]) + oa + b'\x7f\x04' + b'\x00' * 7 + - b'\x10' + bytes(range(16))) + ud = bytes.fromhex('02700000200d000000000000020000000000000010100102a2090804420435044104420600') + tpdu = (b'\x44' + bytes([11]) + oa + b'\x7f\x04' + b'\x00' * 7 + + bytes([len(ud)]) + ud) r = _decode_sm_tpdu(tpdu) self.assertEqual(r['pid_name'], 'SIM data download (secured packet)') - self.assertEqual(r['payload'], '000102030405060708090A0B0C0D0E0F') + self.assertEqual(r['udh'][0]['name'], 'Command Packet Identifier (CPI)') + self.assertEqual(r['secured']['cpl'], 32) + self.assertEqual(r['secured']['chl'], 13) + self.assertEqual(r['secured']['spi']['counter'], 'no counter') + self.assertEqual(r['secured']['spi']['rc_cc_ds'], 'none') + self.assertEqual(r['secured']['tar'], '000002') + self.assertEqual(r['secured']['cntr'], '0000000000') + self.assertEqual(r['secured']['data'], '0010100102A2090804420435044104420600') + + def test_response_packet(self): + from simtrace2_pysniff.server.decode import _decode_response_packet + body = bytes.fromhex('001f0a00000000000000500000ab12800101230d08a0000001510000000f9a9000') + r = _decode_response_packet(body) + self.assertEqual(r['rpl'], 31) + self.assertEqual(r['rhl'], 10) + self.assertEqual(r['tar'], '000000') + self.assertEqual(r['cntr'], '0000000050') + self.assertEqual(r['pcntr'], 0) + self.assertEqual(r['status'], {'code': '00', 'name': 'PoR OK'}) + self.assertEqual(r['data'], 'AB12800101230D08A0000001510000000F9A9000') + + def test_secured_packet_fallback(self): + from simtrace2_pysniff.server.decode import _decode_secured_packet + self.assertEqual(_decode_secured_packet(bytes.fromhex('0001')), {'raw': '0001'}) + + def test_send_sm_response_packet(self): + from simtrace2_pysniff.server.decode import _decode_sm_tpdu + tpdu = bytes.fromhex('41000481112200f624027100001f0a00000000000000500000ab12800101230d08a0000001510000000f9a9000') + r = _decode_sm_tpdu(tpdu) + self.assertEqual(r['mti'], 'SMS-SUBMIT') + self.assertEqual(r['da'], '1122') + self.assertEqual(r['udh'][0]['name'], 'Response Packet Identifier (RPI)') + self.assertEqual(r['response_packet']['rpl'], 31) + self.assertEqual(r['response_packet']['cntr'], '0000000050') + self.assertEqual(r['response_packet']['status'], {'code': '00', 'name': 'PoR OK'}) + self.assertEqual(r['response_packet']['data'], 'AB12800101230D08A0000001510000000F9A9000') def test_deliver_scts(self): from simtrace2_pysniff.server.decode import _decode_sm_tpdu @@ -1093,13 +1129,24 @@ class TestP1P2(unittest.TestCase): self.assertEqual(r['p1p2'], {'label': 'Tag', 'value': 0x50}) def test_commands_always_have_p1p2(self): - # FETCH and GET RESPONSE have P1/P2 (raw '00') even without a spec. + # FETCH and GET RESPONSE have unused P1/P2 (always '00') — RFU. r = decode_message(bytes.fromhex('801200000fd00d8103010300820281820402011e9000')) - self.assertEqual(r['p1'], {'raw': '00'}) - self.assertEqual(r['p2'], {'raw': '00'}) + self.assertEqual(r['p1p2'], {'unused': True, 'value': 0}) r = decode_message(bytes.fromhex('00c0000000'), prev={'ins': 0xA4, 'ins_name': 'SELECT', 'sw1': '61'}) - self.assertEqual(r['p1'], {'raw': '00'}) - self.assertEqual(r['p2'], {'raw': '00'}) + self.assertEqual(r['p1p2'], {'unused': True, 'value': 0}) + + def test_increase_p1p2_unused(self): + r = decode_message(bytes.fromhex('80320000030100000000')) + self.assertEqual(r['p1p2'], {'unused': True, 'value': 0}) + + def test_deactivate_activate_file_p1(self): + r = decode_message(bytes.fromhex('00040000126f0a')) + self.assertEqual(r['p1']['name'], 'EF by file ID') + self.assertEqual(r['p2']['name'], 'No indication') + r = decode_message(bytes.fromhex('00040800123f002f00')) + self.assertEqual(r['p1']['name'], 'Path from MF') + r = decode_message(bytes.fromhex('00040900122f00')) + self.assertEqual(r['p1']['name'], 'Path from current DF') if __name__ == '__main__':