diff --git a/frontend/index.html b/frontend/index.html index 7833d04..41b6fd1 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -525,6 +525,7 @@ function $on(id, ev, fn) { el(id).addEventListener(ev, fn); } function responseRows(resp) { if (!resp) return ''; const rows = []; + if (resp.qualifier) rows.push(['Qualifier', resp.qualifier]); const fd = resp.file_descriptor; if (fd) { rows.push(['File type', fd.file_type]); @@ -610,7 +611,7 @@ function cmdRows(cmd) { } // Proactive command fields (FETCH) - if (cmd.refresh_mode) rows.push(['Refresh mode', cmd.refresh_mode]); + if (cmd.qualifier) rows.push(['Qualifier', cmd.qualifier]); if (cmd.file_list) rows.push(['File list', cmd.file_list.join(', ')]); if (cmd.aid) rows.push(['AID', cmd.aid]); if (cmd.title) rows.push(['Title', cmd.title]); @@ -867,6 +868,10 @@ function showDetail(msg, entry) { if (decoded.p1p2) { if (decoded.p1p2.unused) { p1p2 = `P1, P20x${hx(decoded.p1p2.value, 4)} (RFU)`; + } else if (decoded.p1p2.sfi !== undefined) { + p1p2 = `P1-P20x${hx(decoded.p1p2.value, 4)} — SFI ${decoded.p1p2.sfi}, offset ${decoded.p1p2.offset}`; + } else if (decoded.p1p2.offset !== undefined) { + p1p2 = `P1-P20x${hx(decoded.p1p2.value, 4)} — Offset: ${decoded.p1p2.offset}`; } else { p1p2 = `P1-P20x${hx(decoded.p1p2.value, 4)} — ${decoded.p1p2.label}: ${decoded.p1p2.value}`; } diff --git a/simtrace2_pysniff/server/capture.py b/simtrace2_pysniff/server/capture.py index 03503bd..f5b0efb 100644 --- a/simtrace2_pysniff/server/capture.py +++ b/simtrace2_pysniff/server/capture.py @@ -46,10 +46,14 @@ class DirectSniffer: self._session.close() def iter_messages(self): - for msg in self._session.iter_messages(): - if not self._running: - break - yield msg.type, msg.data, msg.flags + from ..device import DeviceDisconnected + try: + for msg in self._session.iter_messages(): + if not self._running: + break + yield msg.type, msg.data, msg.flags + except DeviceDisconnected: + return class CaptureManager: diff --git a/simtrace2_pysniff/server/decode.py b/simtrace2_pysniff/server/decode.py index c815053..7cb74d9 100644 --- a/simtrace2_pysniff/server/decode.py +++ b/simtrace2_pysniff/server/decode.py @@ -867,22 +867,37 @@ PLI_QUALIFIERS = { } -def _command_details_name(value): - """Decode a Command Details TLV value (number, type, qualifier). +# TS 102 223 §8.6 — Command Qualifier tables per Type of Command. +# Only commands with a known qualifier coding are listed; others fall back +# to a raw '0xXX' in _command_qualifier. +COMMAND_QUALIFIERS = { + 0x01: REFRESH_MODES, # REFRESH + 0x26: PLI_QUALIFIERS, # PROVIDE LOCAL INFORMATION +} - Returns the command type name, with the PLI qualifier description - appended for PROVIDE LOCAL INFORMATION (type 0x26). + +def _command_qualifier(cmd_type, qualifier): + """Decode a Command Qualifier (Command Details byte 3) into a string. + + Returns None when no meaningful qualifier is present, a decoded name + when a table exists (including a table's own '00' entry), or a raw + '0xXX' for a non-zero qualifier without a table. """ + if qualifier is None: + return None + table = COMMAND_QUALIFIERS.get(cmd_type) + if table: + return table.get(qualifier, f'0x{qualifier:02X}') + if qualifier == 0x00: + return None + return f'0x{qualifier:02X}' + + +def _command_details_name(value): + """Decode a Command Details TLV value into the command type name.""" if len(value) < 2: return None - name = CAT_COMMAND_TYPES.get(value[1]) - if not name: - return None - if value[1] == 0x26 and len(value) >= 3: # PROVIDE LOCAL INFORMATION - q = PLI_QUALIFIERS.get(value[2]) - if q: - name = f'{name} — {q}' - return name + return CAT_COMMAND_TYPES.get(value[1]) def decode_cat(ins, body): @@ -920,6 +935,16 @@ def decode_tr_command(body): return None +def decode_tr_qualifier(body): + """Extract the Command Qualifier echoed in a TERMINAL RESPONSE body.""" + if not body: + return None + for tag, _length, value in parse_tlv(body): + if tag == 0x81 and len(value) >= 3: + return _command_qualifier(value[1], value[2]) + return None + + # ──────────────────── Response decoders ──────────────────── # TS 102 223 Result (TERMINAL RESPONSE) — general result codes @@ -2069,8 +2094,9 @@ def _decode_proactive(body): if cmd_type is None: return None result['type'] = CAT_COMMAND_TYPES.get(cmd_type, f'0x{cmd_type:02X}') - if cmd_type == 0x01 and qualifier is not None: # REFRESH - result['refresh_mode'] = REFRESH_MODES.get(qualifier, f'0x{qualifier:02X}') + qualifier_desc = _command_qualifier(cmd_type, qualifier) + if qualifier_desc: + result['qualifier'] = qualifier_desc if file_list: result['file_list'] = file_list if aid: @@ -2701,8 +2727,16 @@ def _build_summary(result): p1 = result.get('p1') p2 = result.get('p2') p1txt = p2txt = None - if 'p1p2' in result and not result['p1p2'].get('unused'): - p1txt = f"{result['p1p2']['label']}: 0x{result['p1p2']['value']:04X}" + p1p2 = result.get('p1p2') + if p1p2 and not p1p2.get('unused'): + if 'sfi' in p1p2: + p1txt = f"SFI {p1p2['sfi']}" + if p1p2.get('offset'): + p1txt += f", offset {p1p2['offset']}" + elif 'offset' in p1p2: + p1txt = f"Offset: 0x{p1p2['offset']:04X}" + else: + p1txt = f"{p1p2['label']}: 0x{p1p2['value']:04X}" else: if p1: p1txt = p1.get('name') @@ -2738,8 +2772,8 @@ def _build_summary(result): if cmd: if cmd.get('context'): parts.append(cmd['context']) - if cmd.get('refresh_mode'): - parts.append(cmd['refresh_mode']) + if cmd.get('qualifier'): + parts.append(cmd['qualifier']) if cmd.get('title'): parts.append(cmd['title']) items = cmd.get('items') @@ -2770,6 +2804,9 @@ def _build_summary(result): if result.get('response_to'): parts.append('\u2192 ' + result['response_to']) + response = result.get('response') + if response and response.get('qualifier'): + parts.append(response['qualifier']) if result.get('response_for'): response = result.get('response') or {} fd = response.get('file_descriptor') @@ -2777,7 +2814,6 @@ def _build_summary(result): parts.append(_fcp_summary(result['response_for'], fd, response)) else: parts.append('response for ' + result['response_for']) - response = result.get('response') if response and response.get('name'): parts.append(response['name']) @@ -2820,7 +2856,9 @@ def decode_message(raw_data, prev=None): } # P1 / P2 decode - if spec: + if ins in (0xB0, 0xD6): # READ/UPDATE BINARY → SFI/offset P1-P2 + result['p1p2'] = _decode_binary_offset(p1, p2) + elif spec: if 'p1p2' in spec: offset = (p1 << 8) | p2 if spec['p1p2'].get('unused'): @@ -2873,7 +2911,11 @@ 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) + response = _decode_tr_result(body) or {} + tr_qualifier = decode_tr_qualifier(body) + if tr_qualifier: + response['qualifier'] = tr_qualifier + result['response'] = response or None else: cat_command = decode_cat(ins, body) if cat_command: @@ -2891,7 +2933,7 @@ def decode_message(raw_data, prev=None): # File data decode for READ/UPDATE using the current selection. if ins in (0xB0, 0xB2, 0xD6, 0xDC) and prev and prev.get('sel'): - offset = (result.get('p1p2') or {}).get('value', 0) + offset = (result.get('p1p2') or {}).get('offset', 0) if ins in (0xB2, 0xDC) or offset == 0: if ins in (0xB2, 0xDC) and not _is_record_file(prev['sel']): result['file'] = _stale_file_note(prev['sel']) @@ -3219,6 +3261,18 @@ def decode_sniff_msg(raw_data, msg_type, flags=0, prev=None): return None +def _decode_binary_offset(p1, p2): + """Decode READ/UPDATE BINARY P1/P2 (TS 102 221 §11.1.3.2 Table 11.10). + + b8 of P1 = 0 → offset = b7..b1 of P1 << 8 | P2. + b8 of P1 = 1 → SFI referencing: b5..b1 of P1 = SFI, P2 = offset. + """ + raw = (p1 << 8) | p2 + if p1 & 0x80: + return {'value': raw, 'sfi': p1 & 0x1F, 'offset': p2} + return {'value': raw, 'offset': ((p1 & 0x7F) << 8) | p2} + + def _decode_field(spec, value): """Decode a single P1 or P2 field from its byte value.""" result = {'raw': f'{value:02x}'} diff --git a/tests/test_capture.py b/tests/test_capture.py index 68843d2..66f86fb 100644 --- a/tests/test_capture.py +++ b/tests/test_capture.py @@ -22,6 +22,13 @@ class _FakeBackend: yield from self._messages +class _DisconnectingSession: + def iter_messages(self): + from simtrace2_pysniff.device import DeviceDisconnected + raise DeviceDisconnected('disconnected') + yield + + class TestCaptureManager(unittest.TestCase): def _manager(self, backend): tmp = tempfile.NamedTemporaryFile(suffix='.db', delete=False) @@ -44,6 +51,12 @@ class TestCaptureManager(unittest.TestCase): self.assertIsNotNone(db.get_session(sid)) self.assertEqual(db.count_messages(sid), 1) + def test_direct_sniffer_device_disconnect(self): + from simtrace2_pysniff.server.capture import DirectSniffer + sn = DirectSniffer() + sn._session = _DisconnectingSession() + self.assertEqual(list(sn.iter_messages()), []) + if __name__ == '__main__': unittest.main() diff --git a/tests/test_decode.py b/tests/test_decode.py index 29992d6..8caee59 100644 --- a/tests/test_decode.py +++ b/tests/test_decode.py @@ -71,8 +71,9 @@ class TestCatDecoding(unittest.TestCase): def test_fetch_provide_local_info(self): r = decode_message(bytes.fromhex( '801200000BD0098103012600820281829000')) - self.assertEqual(r['cat_command'], - 'PROVIDE LOCAL INFORMATION — Location Info (MCC, MNC, LAC/TAC, Cell ID)') + self.assertEqual(r['cat_command'], 'PROVIDE LOCAL INFORMATION') + self.assertEqual(r['cmd']['qualifier'], + 'Location Info (MCC, MNC, LAC/TAC, Cell ID)') def test_envelope_event_download(self): r = decode_message(bytes.fromhex( @@ -171,19 +172,21 @@ class TestPliQualifier(unittest.TestCase): def test_tr_pli_imei(self): r = decode_message(bytes.fromhex( '8014000016810301260102028281030106130952F0991EC57A68009F')) - self.assertEqual(r['response_to'], 'PROVIDE LOCAL INFORMATION — IMEI') + self.assertEqual(r['response_to'], 'PROVIDE LOCAL INFORMATION') + self.assertEqual(r['response']['qualifier'], 'IMEI') def test_fetch_pli_location(self): r = decode_message(bytes.fromhex( '801200000BD0098103012600820281829000')) - self.assertEqual(r['cat_command'], - 'PROVIDE LOCAL INFORMATION — Location Info (MCC, MNC, LAC/TAC, Cell ID)') + self.assertEqual(r['cat_command'], 'PROVIDE LOCAL INFORMATION') + self.assertEqual(r['cmd']['qualifier'], + 'Location Info (MCC, MNC, LAC/TAC, Cell ID)') def test_fetch_pli_date_time(self): r = decode_message(bytes.fromhex( '801200000BD0098103012603820281829000')) - self.assertEqual(r['cat_command'], - 'PROVIDE LOCAL INFORMATION — Date, time and time zone') + self.assertEqual(r['cat_command'], 'PROVIDE LOCAL INFORMATION') + self.assertEqual(r['cmd']['qualifier'], 'Date, time and time zone') def test_non_pli_no_qualifier(self): r = decode_message(bytes.fromhex( @@ -1040,6 +1043,27 @@ class TestFileDecoders(unittest.TestCase): prev={'sel': {'fid': '6f07'}}) self.assertNotIn('file', r) + def test_read_binary_offset_form(self): + # b8 of P1 = 0 → offset = b7..b1 << 8 | P2. + r = decode_message(bytes.fromhex('00b0010000'), prev=None) + self.assertEqual(r['p1p2'], {'value': 0x0100, 'offset': 256}) + self.assertIn('Offset: 0x0100', r['summary']) + + def test_read_binary_sfi_form(self): + # b8 of P1 = 1 → SFI referencing: SFI = b5..b1, P2 = offset. + r = decode_message(bytes.fromhex('00d68b000b') + + bytes.fromhex('0000000000000000000000') + bytes.fromhex('9000'), + prev=None) + self.assertEqual(r['p1p2'], {'value': 0x8B00, 'sfi': 11, 'offset': 0}) + self.assertIn('SFI 11', r['summary']) + self.assertNotIn('offset 0', r['summary']) + + def test_read_binary_sfi_offset(self): + # SFI referencing with a non-zero offset (P2 = 5). + r = decode_message(bytes.fromhex('00b08b0500'), prev=None) + self.assertEqual(r['p1p2'], {'value': 0x8B05, 'sfi': 11, 'offset': 5}) + self.assertIn('SFI 11, offset 5', r['summary']) + def test_read_via_get_response(self): iccid = bytes.fromhex('98891020000000460012') gr = bytes.fromhex('00c00000') + bytes([len(iccid)]) + iccid + bytes.fromhex('9000') @@ -1239,13 +1263,13 @@ class TestRefresh(unittest.TestCase): def test_uicc_reset(self): r = decode_message(bytes.fromhex('801200000bd0098103010104820281829000')) self.assertEqual(r['cat_command'], 'REFRESH') - self.assertEqual(r['cmd']['refresh_mode'], 'UICC Reset') + self.assertEqual(r['cmd']['qualifier'], 'UICC Reset') self.assertNotIn('file_list', r['cmd']) self.assertNotIn('aid', r['cmd']) def test_file_change_notification(self): r = decode_message(bytes.fromhex('8012000011d00f81030101018202818292046f076f209000')) - self.assertEqual(r['cmd']['refresh_mode'], 'File Change Notification') + self.assertEqual(r['cmd']['qualifier'], 'File Change Notification') self.assertEqual(r['cmd']['file_list'], ['6F07', '6F20'])