Decode OTA secured command/response packets in SMS TP-UD; collapse RFU P1/P2

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,8 +849,12 @@ function showDetail(msg, entry) {
 
     let p1p2 = '';
     if (decoded.p1p2) {
+        if (decoded.p1p2.unused) {
+            p1p2 = `<tr><td class="label">P1, P2</td><td class="value">0x${hx(decoded.p1p2.value, 4)} (RFU)</td></tr>`;
+        } else {
             p1p2 = `<tr><td class="label">P1-P2</td><td class="value">0x${hx(decoded.p1p2.value, 4)} — ${decoded.p1p2.label}: ${decoded.p1p2.value}</td></tr>`;
         }
+    }
     let p1 = '';
     if (decoded.p1) {
         const n = decoded.p1.name || (decoded.p1.label && decoded.p1.value != null ? `${decoded.p1.label}: ${decoded.p1.value}` : '');

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'},
+        'p1': {0x00: 'EF by file ID', 0x08: 'Path from MF', 0x09: 'Path from current DF'},
-        'body': None,
+        'p2': {0x00: 'No indication'},
+        'body': {'label': 'File ID/path'},
     },
     0x44: {
         'name': 'ACTIVATE FILE',
-        'p1p2': {'fmt': 'uint16be', 'label': 'File ID'},
+        'p1': {0x00: 'EF by file ID', 0x08: 'Path from MF', 0x09: 'Path from current DF'},
-        'body': None,
+        '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)
+    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)'
-        result['payload'] = ud[:udl].hex().upper()
+        secured = _decode_secured_packet(body)
+        result['secured'] = secured
+        if 'raw' in secured:
+            result['payload'] = secured['raw']
         return result
 
-    if udhi and ud:
+    if 0x71 in udh_ieis:  # secured Response Packet (PoR)
-        udhl = ud[0]
+        rp = _decode_response_packet(body)
-        result['udhl'] = udhl
+        result['response_packet'] = rp
-        result['udh'] = _decode_udh(ud[1:1 + udhl])
+        if 'raw' in rp:
-        body = ud[1 + udhl:]
+            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()
+            result['payload'] = ud_data[:udl].hex().upper()
-            text = _decode_8bit_text(ud[:udl])
+            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,6 +2729,9 @@ def decode_message(raw_data, prev=None):
     if spec:
         if 'p1p2' in spec:
             offset = (p1 << 8) | p2
+            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)

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 +
+        ud = bytes.fromhex('02700000200d000000000000020000000000000010100102a2090804420435044104420600')
-                b'\x10' + bytes(range(16)))
+        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['p1p2'], {'unused': True, 'value': 0})
-        self.assertEqual(r['p2'], {'raw': '00'})
         r = decode_message(bytes.fromhex('00c0000000'), prev={'ins': 0xA4, 'ins_name': 'SELECT', 'sw1': '61'})
-        self.assertEqual(r['p1'], {'raw': '00'})
+        self.assertEqual(r['p1p2'], {'unused': True, 'value': 0})
-        self.assertEqual(r['p2'], {'raw': '00'})
+
+    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__':