Add context-aware decoding for GET RESPONSE and TERMINAL RESPONSE
decode.py: - decode_cat: TERMINAL RESPONSE (0x14) now extracts the proactive command type from its echoed Command Details (tag 81) — no history lookup needed since TR carries its own type - decode_message gains prev param; GET RESPONSE (0xC0) resolves 'response_for' from the previous command when its SW1 was 0x61, else None (orphaned) - decode_sniff_msg threads prev through database.py: - _row_to_message accepts prev; new _decode_rows tracks previous-TPDU context in sequence; get_messages/get_messages_after seed context from the message before the window/cursor tests: TR cat_command + GET RESPONSE context/orphaned cases (49 total)
This commit is contained in:
1 parent
2a58fec0a3
commit
083924c976
3 files changed
+92
-9
No files matched your search
@@ -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
|
||||
@@ -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:
|
||||
|
||||
Reference in new issue
Block a user