From f30fe41d75711f7ebb43b4d2785d05639d3ad3a5 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: Wed, 12 Aug 2026 20:45:06 +0300 Subject: [PATCH] Add per-command response decoders (FCP, AUTH, TERMINAL RESPONSE) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit decode.py: - _decode_fcp: FCP/FCI/FMD template — full bit-level file descriptor (TS 102 221 Table 11.5: file type, structure, shareable, data coding, record length, num records), file ID (+ KNOWN_FIDS name), DF name, SFI, life cycle - _decode_auth: GSM SRES+Kc, or 3G tag DB (RES/CK/IK/KC) / DC (AUTS) - _decode_tr_result: TERMINAL RESPONSE Result TLV (code + name) - _decode_response_for(ins, data): GET RESPONSE decodes its body using the previous command's INS (SELECT→FCP, AUTHENTICATE→auth) - decode_message wires response decoding for TR and GET RESPONSE database.py: prev context now carries 'ins' (INS byte) via _context_from_decoded, used by GET RESPONSE response decoding tests: FCP, AUTH (GSM/3G), TR result, GET RESPONSE wiring (57 total) --- simtrace2_pysniff/server/database.py | 16 ++- simtrace2_pysniff/server/decode.py | 180 +++++++++++++++++++++++++++ tests/test_decode.py | 70 +++++++++++ 3 files changed, 261 insertions(+), 5 deletions(-) diff --git a/simtrace2_pysniff/server/database.py b/simtrace2_pysniff/server/database.py index 3ba5af0..af9a330 100644 --- a/simtrace2_pysniff/server/database.py +++ b/simtrace2_pysniff/server/database.py @@ -202,7 +202,16 @@ class Database: d = decode_sniff_msg(row[0], 'tpdu', 0) if not d: return None - return {'ins_name': d.get('ins_name'), 'sw1': (d.get('sw') or {}).get('sw1')} + return self._context_from_decoded(d) + + def _context_from_decoded(self, d): + ins_hex = d.get('ins_hex') + ins = int(ins_hex, 16) if ins_hex else None + return { + 'ins': ins, + 'ins_name': d.get('ins_name'), + 'sw1': (d.get('sw') or {}).get('sw1'), + } def _decode_rows(self, rows, initial_prev=None): msgs = [] @@ -211,9 +220,6 @@ class Database: msg = self._row_to_message(row, prev=prev) d = msg.get('decoded') if msg['type'] == 'tpdu' and d and d.get('ins_hex'): - prev = { - 'ins_name': d.get('ins_name'), - 'sw1': (d.get('sw') or {}).get('sw1'), - } + prev = self._context_from_decoded(d) msgs.append(msg) return msgs diff --git a/simtrace2_pysniff/server/decode.py b/simtrace2_pysniff/server/decode.py index 8346b8f..e3df5a9 100644 --- a/simtrace2_pysniff/server/decode.py +++ b/simtrace2_pysniff/server/decode.py @@ -513,6 +513,180 @@ def decode_tr_command(body): return None +# ──────────────────── Response decoders ──────────────────── + +# TS 102 223 Result (TERMINAL RESPONSE) — general result codes +TR_RESULTS = { + 0x00: 'Command performed successfully', + 0x01: 'Command performed with partial comprehension', + 0x02: 'Command performed, with missing information', + 0x03: 'REFRESH performed with additional EFs read', + 0x04: 'Command performed successfully, but requested icon could not be displayed', + 0x05: 'Command performed, but modified by call control by NAA', + 0x06: 'Command performed successfully, limited service', + 0x07: 'Command performed with modification', + 0x08: 'REFRESH performed but indicated NAA was not active', + 0x09: 'Command performed successfully, tone not played', + 0x0A: 'Proactive UICC session terminated by the user', + 0x0B: 'Backward move in the proactive UICC session requested by the user', + 0x0C: 'No response from user', + 0x0D: 'Help information required by the user', + 0x0E: 'USSD/SS transaction terminated by the user', + 0x10: 'Terminal currently unable to process command (screen busy)', + 0x11: 'Terminal currently unable to process command (busy on call)', + 0x12: 'Terminal currently unable to process command (USSD/SS ongoing)', + 0x13: 'Terminal currently unable to process command (no service)', + 0x14: 'Terminal currently unable to process command (access control class bar)', + 0x15: 'Terminal currently unable to process command (radio resource unavailable)', + 0x20: 'Network currently unable to process command', + 0x30: 'Beyond terminal capability', + 0x31: 'Command type not understood by terminal', + 0x32: 'Command data not understood by terminal', + 0x33: 'Command number not known by terminal', + 0x34: 'SS Return Error', + 0x35: 'SMS RP-ERROR', + 0x36: 'Error, required values are missing', + 0x37: 'USSD Return Error', + 0x38: 'MultipleCard commands error', + 0x39: 'Interaction with CC by NAA, wrong procedure', + 0x3C: 'Access Technology unable to process command', + 0x3D: 'Frames error', + 0x3E: 'MMS Error', +} + + +def _decode_tr_result(body): + """Decode the Result TLV of a TERMINAL RESPONSE body.""" + 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 + + +# TS 102 221 Table 11.5 — file descriptor byte +_FILE_TYPES = { + 0b000: 'Working EF', + 0b001: 'Internal EF', + 0b111: 'DF or ADF', +} +_EF_STRUCTURES = { + 0b000: 'no information', + 0b001: 'transparent', + 0b010: 'linear fixed', + 0b110: 'cyclic', +} +_LIFE_CYCLE = { + 0x00: 'no information given', + 0x01: 'creation state', + 0x03: 'initialization state', + 0x05: 'operational state (activated)', + 0x07: 'operational state (deactivated)', + 0x0C: 'termination state', + 0x0D: 'termination state (permanently)', +} + + +def _decode_file_descriptor(value): + """Decode a file descriptor byte string (TS 102 221 Table 11.5).""" + if not value: + return {} + b = value[0] + ft = (b >> 3) & 0x07 + result = { + 'shareable': 'shareable' if (b & 0x40) else 'not shareable', + 'file_type': _FILE_TYPES.get(ft, 'RFU'), + } + if ft in (0b000, 0b001): # EF + result['structure'] = _EF_STRUCTURES.get(b & 0x07, 'RFU') + if len(value) >= 2: + result['data_coding'] = f'0x{value[1]:02X}' + if len(value) >= 4: + result['record_length'] = int.from_bytes(value[2:4], 'big') + if len(value) >= 5: + result['num_records'] = value[4] + elif len(value) >= 2: + result['num_dfs_efs'] = value[1] + return result + + +def _decode_fcp(data): + """Decode an FCP/FCI/FMD template (TS 102 221 §11.1.1.3).""" + result = {} + for tag, _length, value in parse_tlv(data): + if tag in (0x62, 0x64): # outer template — recurse + inner = _decode_fcp(value) + inner['template'] = 'FCP' if tag == 0x62 else 'FMD' + result.update(inner) + elif tag == 0x82: + result['file_descriptor'] = _decode_file_descriptor(value) + elif tag == 0x83: + fid = value.hex().upper() + result['file_id'] = fid + if fid.lower() in KNOWN_FIDS: + result['file_id_name'] = KNOWN_FIDS[fid.lower()] + elif tag == 0x84: + result['df_name'] = value.hex().upper() + elif tag == 0x88: + result['sfi'] = f'0x{value[0]:02X}' if value else None + elif tag == 0x8A: + result['life_cycle'] = _LIFE_CYCLE.get(value[0] if value else 0, f'0x{(value[0] if value else 0):02X}') + elif tag == 0xAB: + result['short_ef_id'] = value.hex().upper() + return result + + +def _decode_auth_3g(data): + """Decode a 3G AUTHENTICATE response (tag DB success / DC sync-fail).""" + tag = data[0] + if tag == 0xDC: # synchronisation failure → length byte + AUTS + if len(data) >= 2: + auts_len = data[1] + return {'type': '3G', 'status': 'sync fail', + 'auts': data[2:2 + auts_len].hex().upper()} + return {'type': '3G', 'status': 'sync fail', 'auts': data[1:].hex().upper()} + # tag 0xDB: success → length-prefixed RES, CK, IK, (KC) + result = {'type': '3G', 'status': 'success'} + i = 1 + for name in ('res', 'ck', 'ik', 'kc'): + if i >= len(data): + break + ln = data[i] + i += 1 + result[name] = data[i:i + ln].hex().upper() + i += ln + return result + + +def _decode_auth(data): + """Decode an AUTHENTICATE response (GSM SRES+Kc, or 3G tag DB/DC).""" + if not data: + return {} + if data[0] in (0xDB, 0xDC): + return _decode_auth_3g(data) + # GSM: SRES (4 bytes) + Kc (8 bytes) + if len(data) >= 12: + return { + 'type': 'GSM', + 'sres': data[:4].hex().upper(), + 'kc': data[4:12].hex().upper(), + } + return {'type': 'GSM', 'raw': data.hex().upper()} + + +def _decode_response_for(ins, data): + """Decode response data using the command identified by INS.""" + if ins == 0xA4: # SELECT → FCP/FCI template + return _decode_fcp(data) + if ins in (0x88, 0x89): # AUTHENTICATE → SRES/Kc or RES/CK/IK + return _decode_auth(data) + return None + + # ──────────────────── Decode entry point ──────────────────── def decode_message(raw_data, prev=None): @@ -588,6 +762,7 @@ def decode_message(raw_data, prev=None): response_to = decode_tr_command(body) if response_to: result['response_to'] = response_to + result['response'] = _decode_tr_result(body) else: cat_command = decode_cat(ins, body) if cat_command: @@ -602,6 +777,11 @@ def decode_message(raw_data, prev=None): if ins == 0xC0: if prev and prev.get('sw1') == '61': result['response_for'] = prev.get('ins_name') + prev_ins = prev.get('ins') + if prev_ins is not None and cmd_body_len > 0: + response = _decode_response_for(prev_ins, remaining[:cmd_body_len]) + if response: + result['response'] = response else: result['response_for'] = None diff --git a/tests/test_decode.py b/tests/test_decode.py index dac9244..188f47c 100644 --- a/tests/test_decode.py +++ b/tests/test_decode.py @@ -154,5 +154,75 @@ class TestGetResponseContext(unittest.TestCase): self.assertIsNone(r['response_for']) +class TestFcpResponse(unittest.TestCase): + GET_RESPONSE = bytes.fromhex( + '00C000002962278202782183023F00A50A8001718302F9628701018A01058B032F0601C60990014083010183010A9000') + + def test_select_response_fcp(self): + r = decode_message(self.GET_RESPONSE, prev={'ins': 0xA4, 'ins_name': 'SELECT', 'sw1': '61'}) + self.assertEqual(r['response_for'], 'SELECT') + resp = r['response'] + self.assertEqual(resp['template'], 'FCP') + self.assertEqual(resp['file_id'], '3F00') + self.assertEqual(resp['file_id_name'], 'MF') + self.assertEqual(resp['file_descriptor']['file_type'], 'DF or ADF') + self.assertEqual(resp['file_descriptor']['shareable'], 'shareable') + self.assertEqual(resp['life_cycle'], 'operational state (activated)') + + def test_ef_descriptor_transparent(self): + from simtrace2_pysniff.server.decode import _decode_file_descriptor + fd = _decode_file_descriptor(bytes.fromhex('0121')) + self.assertEqual(fd['file_type'], 'Working EF') + self.assertEqual(fd['structure'], 'transparent') + + def test_ef_descriptor_linear_fixed(self): + from simtrace2_pysniff.server.decode import _decode_file_descriptor + fd = _decode_file_descriptor(bytes.fromhex('0221')) + self.assertEqual(fd['file_type'], 'Working EF') + self.assertEqual(fd['structure'], 'linear fixed') + + +class TestAuthResponse(unittest.TestCase): + def test_3g_success(self): + from simtrace2_pysniff.server.decode import _decode_auth + data = bytes.fromhex('DB084CFA9017FD0DD85A101A2622F60E8ABD2C2497B9A8EFAF55E510CAA393329FF97868B9537369D5266A4F084D417B05ABFAFAEE') + r = _decode_auth(data) + self.assertEqual(r['type'], '3G') + self.assertEqual(r['status'], 'success') + self.assertEqual(r['res'], '4CFA9017FD0DD85A') + self.assertEqual(r['ck'], '1A2622F60E8ABD2C2497B9A8EFAF55E5') + self.assertEqual(r['ik'], 'CAA393329FF97868B9537369D5266A4F') + + def test_3g_sync_fail(self): + from simtrace2_pysniff.server.decode import _decode_auth + data = bytes.fromhex('DC0E' + '00' * 14) + r = _decode_auth(data) + self.assertEqual(r['type'], '3G') + self.assertEqual(r['status'], 'sync fail') + self.assertEqual(r['auts'], '00' * 14) + + def test_gsm(self): + from simtrace2_pysniff.server.decode import _decode_auth + sres = bytes.fromhex('AABBCCDD') + kc = bytes.fromhex('0011223344556677') + r = _decode_auth(sres + kc) + self.assertEqual(r['type'], 'GSM') + self.assertEqual(r['sres'], 'AABBCCDD') + self.assertEqual(r['kc'], '0011223344556677') + + +class TestTrResult(unittest.TestCase): + def test_success(self): + r = decode_message(bytes.fromhex('801400000C8103010500020282810301009130')) + self.assertEqual(r['response']['code'], '0x00') + self.assertEqual(r['response']['name'], 'Command performed successfully') + + def test_session_terminated(self): + from simtrace2_pysniff.server.decode import _decode_tr_result + r = _decode_tr_result(bytes.fromhex('81030105000202828103010A')) + self.assertEqual(r['code'], '0x0A') + self.assertEqual(r['name'], 'Proactive UICC session terminated by the user') + + if __name__ == '__main__': unittest.main()