diff --git a/simtrace2_pysniff/server/database.py b/simtrace2_pysniff/server/database.py index c3982f9..3ba5af0 100644 --- a/simtrace2_pysniff/server/database.py +++ b/simtrace2_pysniff/server/database.py @@ -128,14 +128,18 @@ class Database: 'SELECT id, session_id, elapsed, type, data, flags FROM messages ' 'WHERE session_id=? ORDER BY elapsed, id LIMIT ? OFFSET ?', (session_id, limit, offset)).fetchall() - return [self._row_to_message(r) for r in rows] + initial_prev = None + if offset > 0 and rows: + initial_prev = self._last_tpdu_context(session_id, rows[0][0]) + return self._decode_rows(rows, initial_prev=initial_prev) def get_messages_after(self, session_id, after_id): rows = self._conn.execute( 'SELECT id, session_id, elapsed, type, data, flags FROM messages ' 'WHERE session_id=? AND id > ? ORDER BY elapsed, id', (session_id, after_id)).fetchall() - return [self._row_to_message(r) for r in rows] + initial_prev = self._last_tpdu_context(session_id, after_id + 1) + return self._decode_rows(rows, initial_prev=initial_prev) def count_messages(self, session_id): row = self._conn.execute( @@ -166,7 +170,7 @@ class Database: (session_id,)).fetchall() return [{'type': r[0], 'count': r[1]} for r in rows] - def _row_to_message(self, row): + def _row_to_message(self, row, prev=None): data_blob = row[4] flags = row[5] msg = { @@ -180,9 +184,36 @@ class Database: if row[3] in ('tpdu', 'change', 'fidi', 'atr', 'pps'): try: from .decode import decode_sniff_msg - msg['decoded'] = decode_sniff_msg(data_blob, row[3], flags) + msg['decoded'] = decode_sniff_msg(data_blob, row[3], flags, prev=prev) except Exception as e: import sys print(f'Decode error for msg {row[0]} ({row[3]}, {len(data_blob)} bytes): {e}', file=sys.stderr) return msg + + def _last_tpdu_context(self, session_id, before_id): + """Return previous-command context from the last TPDU with id < before_id.""" + row = self._conn.execute( + "SELECT data FROM messages WHERE session_id=? AND id < ? AND type='tpdu' " + "ORDER BY id DESC LIMIT 1", (session_id, before_id)).fetchone() + if row is None: + return None + from .decode import decode_sniff_msg + 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')} + + def _decode_rows(self, rows, initial_prev=None): + msgs = [] + prev = initial_prev + for row in rows: + 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'), + } + msgs.append(msg) + return msgs diff --git a/simtrace2_pysniff/server/decode.py b/simtrace2_pysniff/server/decode.py index 4eb0d41..41ac924 100644 --- a/simtrace2_pysniff/server/decode.py +++ b/simtrace2_pysniff/server/decode.py @@ -497,17 +497,24 @@ def decode_cat(ins, body): return CAT_COMMAND_TYPES.get(v2[1]) elif ins == 0xC2: # ENVELOPE → envelope type return ENVELOPE_TYPES.get(body[0]) + elif ins == 0x14: # TERMINAL RESPONSE → echoes proactive command type + for tag, _length, value in parse_tlv(body): + if tag == 0x81 and len(value) >= 2: + return CAT_COMMAND_TYPES.get(value[1]) return None # ──────────────────── Decode entry point ──────────────────── -def decode_message(raw_data): +def decode_message(raw_data, prev=None): """Decode a raw TPDU byte string into a structured dict. Returns None for non-TPDU messages (ATR, PPS, CHANGE, FIDI) or un-parseable data. Returns a dict with ins_name, cla, p1, p2, p3, body, and sw keys for valid TPDU messages. + + *prev* optionally carries the previous TPDU's decoded context + (``{'ins_name': str, 'sw1': str}``) for resolving GET RESPONSE. """ if not raw_data or len(raw_data) < 5: return None @@ -576,6 +583,14 @@ def decode_message(raw_data): if sw_bytes: result['sw'] = decode_sw(sw_bytes) + # GET RESPONSE context: the response belongs to the previous command + # if that command ended with SW1 '61' (response data available). + if ins == 0xC0: + if prev and prev.get('sw1') == '61': + result['response_for'] = prev.get('ins_name') + else: + result['response_for'] = None + return result @@ -624,14 +639,17 @@ def decode_fidi(raw_data): } -def decode_sniff_msg(raw_data, msg_type, flags=0): +def decode_sniff_msg(raw_data, msg_type, flags=0, prev=None): """Decode a raw sniff message of the given type. Returns a structured dict with at least a 'type' key, or None if the message type has no structured decode. + + *prev* optionally carries the previous TPDU's decoded context for + resolving GET RESPONSE. """ if msg_type == 'tpdu' and raw_data: - return decode_message(raw_data) + return decode_message(raw_data, prev=prev) if msg_type == 'change': return decode_change(flags) if msg_type == 'fidi' and raw_data: diff --git a/tests/test_decode.py b/tests/test_decode.py index 18e509a..86d6c55 100644 --- a/tests/test_decode.py +++ b/tests/test_decode.py @@ -66,11 +66,11 @@ class TestCatDecoding(unittest.TestCase): self.assertEqual(r['ins_name'], 'ENVELOPE') self.assertEqual(r['cat_command'], 'EVENT DOWNLOAD') - def test_tr_no_cat(self): + def test_tr_has_cat(self): r = decode_message(bytes.fromhex( '801400000C8103010500020282810301009130')) self.assertEqual(r['ins_name'], 'TERMINAL RESPONSE') - self.assertNotIn('cat_command', r) + self.assertEqual(r['cat_command'], 'SET UP EVENT LIST') def test_select_no_cat(self): r = decode_message(bytes.fromhex('a0a40000023f009000')) @@ -119,5 +119,39 @@ class TestChangeAndFidiDecoding(unittest.TestCase): self.assertEqual(r['di_val'], 64) +class TestTerminalResponse(unittest.TestCase): + def test_tr_setup_event_list(self): + r = decode_message(bytes.fromhex( + '801400000C8103010500020282810301009130')) + self.assertEqual(r['ins_name'], 'TERMINAL RESPONSE') + self.assertEqual(r['cat_command'], 'SET UP EVENT LIST') + + def test_tr_setup_menu(self): + r = decode_message(bytes.fromhex( + '801400000C810301250002028281830100910F')) + self.assertEqual(r['cat_command'], 'SET UP MENU') + + def test_tr_poll_interval(self): + r = decode_message(bytes.fromhex( + '80140000108103010300020282810301000402011E9000')) + self.assertEqual(r['cat_command'], 'POLL INTERVAL') + + +class TestGetResponseContext(unittest.TestCase): + GET_RESPONSE = bytes.fromhex('00C000002962278202782183023F00A50A800171') + + def test_with_prev_select(self): + r = decode_message(self.GET_RESPONSE, prev={'ins_name': 'SELECT', 'sw1': '61'}) + self.assertEqual(r['response_for'], 'SELECT') + + def test_with_prev_non_61(self): + r = decode_message(self.GET_RESPONSE, prev={'ins_name': 'SELECT', 'sw1': '90'}) + self.assertIsNone(r['response_for']) + + def test_orphaned(self): + r = decode_message(self.GET_RESPONSE, prev=None) + self.assertIsNone(r['response_for']) + + if __name__ == '__main__': unittest.main()