## ## Copyright (C) 2020 Sven Soltermann ## ## This program is free software; you can redistribute it and/or modify ## it under the terms of the GNU General Public License as published by ## the Free Software Foundation; either version 2 of the License, or ## (at your option) any later version. ## ## This program is distributed in the hope that it will be useful, ## but WITHOUT ANY WARRANTY; without even the implied warranty of ## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the ## GNU General Public License for more details. ## ## You should have received a copy of the GNU General Public License ## along with this program; if not, see . ## import sigrokdecode as srd import struct import sys import time import codecs import traceback from collections import deque from .gsmtap_stream import (GsmtapStreamSender, GSMTAP_SIM_APDU, GSMTAP_SIM_ATR, GSMTAP_SIM_PPS, GSMTAP_SIM_RST_EVENT, GSMTAP_SIM_VCC_EVENT, GSMTAP_FLAG_BAD_FCS) VERSION = '1.6.0' class Ann: '''Annotation and binary output classes.''' ( ANN_WARN, ANN_BYTE, ANN_ATR, ANN_PPS, ANN_T0, ANN_T1, ANN_T1IBLOCK, ANN_T1RBLOCK, ANN_T1SBLOCK, ANN_APDU, ANN_RST, ANN_VCC, ) = range(12) class pcap_udp_pkt(): # GSMTAP-SIM wrapped in a synthetic Ethernet/IP/UDP header so the binary # output dissects as a UDP stream in Wireshark. The IPv4 header checksum # is left 0 and the endpoints are fixed loopback + the default GSMTAP # ports on purpose: Wireshark tolerates the zero checksum and the values # are only cosmetic. Field bytes are patched in set_data() below. # GSM TAP h = b'' # layer_2 h += b'\x00\x00\x00\x00\x00\x00' # destination_mac h += b'\x00\x00\x00\x00\x00\x00' # source_mac h += b'\x08\x00' # layer_3_protocol # layer_3 h += b'\x45' # version h += b'\x00' # DiffServField h += b'\xFF\xFF' # total_length h += b'\x2B\x0D' # Identification h += b'\x40\x00' # Flags h += b'\x40' # TTL h += b'\x11' # layer_4_protocol h += b'\x00\x00' # header_checksum h += b'\x7F\x00\x00\x01' # source_ip h += b'\x7F\x00\x00\x01' # dest_ip # layer_4 h += b'\xcc\x46' # source_port h += b'\x12\x79' # dest_port h += b'\x00\x00' # datagram_length h += b'\x00\x00' # checksum def __init__(self, ts, data, sub_type=GSMTAP_SIM_APDU, flags=0): self.header = bytearray(pcap_udp_pkt.h) self.data = b'' self.set_timestamp(ts) self.set_data(data, sub_type, flags) def set_timestamp(self, ts): self.timestamp = ts def set_data(self, data, sub_type=GSMTAP_SIM_APDU, flags=0): gsmtap = bytearray(b'\x02\x04\x04\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00') gsmtap[12] = sub_type gsmtap[15] = flags self.data = list(gsmtap + bytes(data)); self.header[16:18] = struct.pack('>H', len(self.data) + 28) self.header[38:40] = struct.pack('>H', len(self.data) + 8) def packet(self): return bytes(self.header) + bytes(self.data) def record_header(self): # See https://wiki.wireshark.org/Development/LibpcapFileFormat. (secs, usecs) = self.timestamp h = struct.pack(' MAX_TPDU_LEN: return (False, 'exceeds T=0 max length') cla = packet[0] ins = packet[1] sw1 = packet[-2] if not plausible_cla(cla): return (False, 'invalid CLA') if not plausible_ins(ins): return (False, 'invalid INS') if not plausible_sw(sw1): return (False, 'invalid SW1') return (True, None) def is_desynced(packet): '''Classify an emitted T=0 packet as mis-framed. All-zero packets are idle-low noise (suppressed elsewhere), NOT desync. A real TPDU must reach its status word: >= 7 bytes ending in a plausible SW1, never exceed the T=0 maximum, and start with a valid CLA/INS pair.''' if not packet: return False if all(b == 0 for b in packet): return False valid, _ = validate_t0_apdu(packet) return not valid # --- embedded-exchange (CONCAT) detector --- # A byte-drop mis-frame can swallow several real T=0 exchanges into one # structurally valid packet: each swallowed header survives verbatim in the # data region, followed by its P3 data bytes and an interior status word. # These constants bound detector precision (empirically tuned on all 10 # example traces: 0 hits on clean traces, every Samsung/xiaomi family hit). # A swallowed header's CLA is validated with the same structural rule as the # decoder itself (see plausible_cla): interindustry classes including the # logical-channel numbers 0x01-0x03 are legitimate first bytes of a command. EMBEDDED_CLA = frozenset(cla for cla in range(256) if plausible_cla(cla)) # Commands that actually occur on these cards' bus (T=0 + STK/CAT). EMBEDDED_INS = frozenset(( 0x12, 0x14, 0x20, 0x44, 0x70, 0x88, 0xA2, 0xA4, 0xB0, 0xB2, 0xB6, 0xC0, 0xC2, 0xCA, 0xCB, 0xD6, 0xDC, 0xF2, )) # Real SW1 values that are not also frequent data bytes (BER-TLV tags). EMBEDDED_SW1 = frozenset(( 0x60, 0x61, 0x62, 0x63, 0x67, 0x68, 0x69, 0x6A, 0x6B, 0x6D, 0x6E, 0x6F, 0x90, 0x91, 0x93, 0x98, 0x99, )) EMBEDDED_SW1_SLACK = 3 # SW1 search window after the nominal data end def embedded_exchanges(buf): '''Return [(header_start, sw1_index), ...] for complete T=0 exchanges embedded inside a packet's data region (index 5 onward). Pattern: header (CLA INS P1 P2 P3) + P3 data bytes + interior SW1/SW2 before the packet's own final status word. CLA must be structurally valid (same rule as plausible_cla); INS must be a real command byte; SW1 is accepted within a small slack window after the nominal data end and must be interior (trailing bytes follow it).''' n = len(buf) out = [] if n < 12: return out q = 5 while q <= n - 9: s = q cla, ins, _p1, _p2, p3 = buf[s:s + 5] if cla not in EMBEDDED_CLA: q += 1 continue if ins not in EMBEDDED_INS: q += 1 continue if p3 == 0: q += 1 continue d1 = s + 5 + p3 if d1 + 3 > n: q += 1 continue sw1 = -1 for e in range(d1, min(n - 1, d1 + EMBEDDED_SW1_SLACK + 1)): if buf[e] in EMBEDDED_SW1: sw1 = e break if sw1 == -1 or sw1 >= n - 2: q += 1 continue out.append((s, sw1)) q += 1 return out MAX_TPDU_LEN = 271 # ISO 7816-3 T=0: header(5) + proc + 255 + SW(2) < 271 IDLE_RESYNC_ETU = 4 # idle bit-periods marking an exchange boundary MAX_RESYNC_ATTEMPTS = 1024 # bail out if no idle gap found in that many tries # ATR-hunt failure bounds. When the default-rate (372 CLK/bit) hunt keeps # failing, the card is either idle or already running at a non-default F/D # (e.g. Samsung F=512/D=32 -> 16 CLK/bit). Without a bound the hunt reads # idle 0xFF forever at the wrong rate and can never reach the DATA state, # where the live ETU is measured -- a deadlock. These cap the retries. ATR_HUNT_ETU_ATTEMPTS = 3 # live-ETU recovery attempts before resuming DATA ATR_HUNT_IDLE_RESETS = 3 # all-0xFF hunt resets before resuming DATA ATR_HUNT_FAIL_LIMIT = 12 # total failed hunt rounds before resuming DATA class Decoder(srd.Decoder): api_version = 3 id = 'iso7816' name = 'ISO 7816' longname = 'Smartcard' desc = 'ISO 7816 decoder (smartcard)' license = 'gplv2+' inputs = ['logic'] outputs = ['iso7816'] tags = ['Embedded/industrial'] channels = ( {'id': 'clk', 'name': 'CLK', 'desc': 'clock'}, {'id': 'data', 'name': 'data', 'desc': 'data'}, ) optional_channels = ( {'id': 'rst', 'name': 'RST', 'desc': 'reset (enable via rst_detect=true)'}, {'id': 'vcc', 'name': 'VCC', 'desc': 'power supply, digital only (enable via vcc_detect=true)'}, ) options = ( {'id': 'clock_option', 'desc': 'Clock option', 'default': 'native', 'values': ('native', 'detect', 'sample_as_clock')}, {'id': 'protocol', 'desc': 'Protocol', 'default': 'auto', 'values': ('auto', 'T=0', 'T=1')}, {'id': 'gsmtap_enable', 'desc': 'Live GSMTAP UDP stream (simtrace2-sniff compatible)', 'default': 'true', 'values': ('true', 'false')}, {'id': 'gsmtap_host', 'desc': 'GSMTAP destination host', 'default': '127.0.0.1'}, {'id': 'gsmtap_port', 'desc': 'GSMTAP destination port', 'default': 4729}, {'id': 'pcap_file', 'desc': 'Write PCAP to file in addition to binary output', 'default': ''}, {'id': 'rst_detect', 'desc': 'Track RST pin events (needs rst channel assigned)', 'default': 'false', 'values': ('false', 'true')}, {'id': 'vcc_detect', 'desc': 'Track VCC pin events (needs vcc channel assigned)', 'default': 'false', 'values': ('false', 'true')}, ) annotations = ( ('warning', 'Human-readable warnings'), ('byte', 'Byte'), ('atr', 'ATR (Answer to Reset)'), ('pps', 'PPS (Protocol and parameters selection)'), ('t0', 'T=0 packet'), ('t1', 'T=1 packet'), ('t1-iblock', 'T=1 I-Block'), ('t1-rblock', 'T=1 R-Block'), ('t1-sblock', 'T=1 S-Block'), ('apdu', 'APDU'), ('rst', 'RST event'), ('vcc', 'VCC event'), ) annotation_rows = ( ('warnings', 'Warnings', (Ann.ANN_WARN,)), ('bytes', 'Bytes', (Ann.ANN_BYTE,)), ('type', 'Type', (Ann.ANN_ATR,Ann.ANN_PPS,Ann.ANN_T0,Ann.ANN_T1)), ('t1blocks', 'T=1 Decode', (Ann.ANN_T1IBLOCK,Ann.ANN_T1RBLOCK,Ann.ANN_T1SBLOCK)), ('apdus', 'apdu', (Ann.ANN_APDU,)), ('lines', 'Lines', (Ann.ANN_RST, Ann.ANN_VCC)), ) binary = ( ('pcap', 'PCAP format'), ) clock_rate = { 0: 372, 1: 372, 2: 558, 3: 744, 4: 1116, 5: 1488, 6: 1860, 9: 512, 10: 768, 11: 1024, 12: 1536, 13: 2048 } baud_rate = { 0: 1, 1: 1, 2: 2, 3: 4, 4: 8, 5: 16, 6: 32, 7: 64, 8: 12, 9: 20 } # Channel indices within the pins tuple returned by self.wait() # (declaration order of self.channels). CLK_IDX = 0 DATA_IDX = 1 RST_IDX = 2 VCC_IDX = 3 def __init__(self): self.reset() def log(self, *args): print(args, file=sys.stderr, flush=True) def _note_discard(self, value, reason, idle=False): # Report bytes consumed but not emitted as part of an APDU. Idle # line noise (0x00/0xFF on an idle period) is counted and reported # periodically so a long idle line cannot flood the session log; # every other discard (real data) is logged and annotated always. if idle: self._idle_discard_total += 1 if self._idle_discard_total == 1 or self._idle_discard_total % 256 == 0: self.log("discarded idle byte 0x%02x: %s (idle discards so far: %d)" % (value, reason, self._idle_discard_total)) return self._discard_total += 1 self.log("discarded byte 0x%02x: %s (discards so far: %d)" % (value, reason, self._discard_total)) if self.ss is not None and self.ss >= 0: self.put(self.ss, self.ss, self.out_ann, [Ann.ANN_WARN, ["dropped 0x%02x: %s" % (value, reason)]]) return value def _consume_first(self): '''Consume the byte the last peek_byte() returned, removing it from whichever source it came from (the _replay buffer or the peeked byte). The decode-step consume paths use this so re-framed bytes queued in _replay cannot loop forever.''' if self._replay: return self._replay.popleft() b = self.peeked_byte self.peeked_byte = None self.peeked_samplenum = -1 return b def reset(self): self.peeked_byte = None self.peeked_samplenum = -1 # Replay buffer for bytes read ahead but not consumed (used by the # command-detect path in the ATR hunt to hand a command header back # to the DATA state without dropping it). self._replay = deque() self.wrote_pcap_header = False self.samplerate = None self.sample_as_clock = False self.detect_clock = False self.fi = self.clock_rate[0] self.di = self.baud_rate[0] self.ss = self.es = -1 self.state = 'FIND START' self.bits = [] self.clock_skip = 372 # Measured bit period (samples or CLK edges) from the first frame; # 372 = spec default until measured. Must exist before any PPS # handling -- native mode never measures it explicitly. self.detected_clock_skip = 372 self.bit_samples = None self._etu_confirm_count = 0 self._etu_fail_count = 0 self._edge_spacings = deque(maxlen=128) self._samples_per_clock = None # True when clock_skip is known to be correct: after a parsed ATR # (372 default), an accepted PPS (FI/DI), or a successful ETU-based # recovery on a mid-session capture. The CLK-sync reader is only # trusted when this is set; otherwise the edge-list reader (which # uses bit_samples) is the safe fallback. self._clock_skip_confident = False self._start_fall = None self._edge_read = True self._last_sn = -1 self._eof = False self._min_pulse_samples = 2 # tuned for 16 MHz sample rate self._glitch_rejects = 0 self._prev_wait_sn = None self._stall = 0 # One-shot: at the first decode_step the initial RST/VCC levels are # primed from the real first sample instead of the defaults below. # Without this, last_rst=0/last_vcc=0 make "capture started while the # card was already in reset" indistinguishable from a live session. self._primed = False self.lastSamplePositive = True # Rolling parity-error monitor: warns when the wire degrades # (e.g. GSM TX bursts coupling into the probe wiring). self.parity_hist = deque(maxlen=64) self.sig_warned = False # Stuck-line watchdog: consecutive all-zero frames usually mean # the I/O probe lost contact (floating low), not data. self.zero_run = 0 # Idle-low APDU suppression counter (all-zero T=0 reads). self._idle_low = 0 # Set after a desync so the next DATA step re-aligns on an # inter-frame idle gap before resuming. self._resync = False # Discard accounting: every consumed byte that is not emitted as # part of an APDU is reported. Non-idle discards (real data) are # logged + annotated at every occurrence; idle 0x00/0xFF line noise # is counted and reported periodically so a long idle line cannot # flood the session log. self._discard_total = 0 self._idle_discard_total = 0 # Set by handle_pps() when an accepted PPS changes the bit rate. # Forces _measure_etu() on the next decode_step() cycle so the # decoder re-locks to the new ETU from the live DATA line. self._pps_speed_changed = False # Count of times the ATR hunt failed to find a valid TS byte -- # bounds the "keep hunting" loop so a mid-session sniffer (no ATR # ever present) eventually falls back to T=0 parsing instead of # looping forever. self._atr_hunt_count = 0 # Persistent hunt-failure accounting (reset only when an ATR is # parsed, NOT on RST re-arm): a device that power-cycles RST while # never sending a valid ATR would otherwise starve the escape below. # _atr_fail_total counts every failed hunt round across activations; # _hunt_etu_attempts caps live-ETU recovery; _atr_idle_resets caps the # all-0xFF reset loop. self._atr_fail_total = 0 self._hunt_etu_attempts = 0 self._atr_idle_resets = 0 # PPS is only attempted once after an ATR/warm reset. Prevents # idle-line 0xFF bytes from being parsed as an endless stream of # PPS requests when the card does not actually perform PPS. self._pps_attempted = False self.gsmtap = None self.pcap_fh = None self.track_rst = False self.track_vcc = False # RST/VCC line tracking uses level-stability hysteresis (see # _line_hyst): a level must stay continuously stable for # vcc_min_gap samples before it is "confirmed" and reported. This # makes power-up and power-down symmetric -- a VCC sense line that # rings on the rising edge (noisy/floating tap) no longer loses its # "VCC on" event; it simply fires once the line settles HIGH. # last_rst/last_vcc hold the *confirmed* level; *_cand / *_cand_sn # track a pending level and the sample it first appeared. self.last_rst = 0 self.last_vcc = 0 self.rst_cand = None self.rst_cand_sn = 0 self.vcc_cand = None self.vcc_cand_sn = 0 # Minimum quiet window (samples) a NEW level must hold before it is # honored. 10 ms @ 16 MHz; set from samplerate in metadata(). self.vcc_min_gap = 0 # 0 = hysteresis disabled (no samplerate) def metadata(self, key, value): if key == srd.SRD_CONF_SAMPLERATE: self.samplerate = value if self.samplerate: self.secs_per_sample = float(1) / float(self.samplerate) # 10 ms debounce window, scaled to samples. self.vcc_min_gap = int(self.samplerate * 0.010) def start(self): self.log("=== iso7816 decoder v{} ===".format(VERSION)) self._pcap_base_time = time.time() self.out_python = self.register(srd.OUTPUT_PYTHON) self.out_ann = self.register(srd.OUTPUT_ANN) self.out_binary = self.register(srd.OUTPUT_BINARY) self.sample_as_clock = self.options['clock_option'] == "sample_as_clock" self.detect_clock = self.options['clock_option'] == "detect" self.state = 'DATA' self._atr_parsed = False # Inverse convention (TS=0x3F): logic 1 = LOW, logic 0 = HIGH. # Set when TS=0x3F is detected; all subsequent byte reads invert bits. self._inverse_convention = False # clock_skip is not trusted until recovered (the card may have already # PPS'd to a faster ETU before the capture began). if (self.options['protocol'] == "T=0"): self.hasT0 = True self.hasT1 = False elif (self.options['protocol'] == "T=1"): self.hasT0 = False self.hasT1 = True else: self.hasT0 = False self.hasT1 = False # True only when the user explicitly set protocol=T=0/T=1 (not 'auto'). # Used to enable mid-session raw-burst emission (below) without # affecting the default auto-protocol path. self.explicit_protocol = self.options['protocol'] in ("T=0", "T=1") # Live GSMTAP UDP stream (default on, mimics simtrace2-sniff). if (self.options['gsmtap_enable'] == "true" and self.options['gsmtap_host']): try: self.gsmtap = GsmtapStreamSender(self.options['gsmtap_host'], int(self.options['gsmtap_port'])) self.log("GSMTAP stream to", "%s:%s" % (self.options['gsmtap_host'], self.options['gsmtap_port'])) except OSError as e: self.gsmtap = None self.log("GSMTAP stream disabled:", str(e)) # Optional direct PCAP file output. if (self.options['pcap_file']): try: self.pcap_fh = open(self.options['pcap_file'], 'wb') self.pcap_fh.write(self.pcap_global_header()) except OSError as e: self.pcap_fh = None self.log("PCAP file disabled:", str(e)) # Optional RST/VCC line tracking. self.track_rst = self.options['rst_detect'] == "true" self.track_vcc = self.options['vcc_detect'] == "true" if (self.track_rst and hasattr(self, 'has_channel') and not self.has_channel(self.RST_IDX)): self.log("rst_detect=true but 'rst' channel not assigned - disabled") self.track_rst = False if (self.track_vcc and hasattr(self, 'has_channel') and not self.has_channel(self.VCC_IDX)): self.log("vcc_detect=true but 'vcc' channel not assigned - disabled") self.track_vcc = False def finish(self): if self.pcap_fh is not None: try: self.pcap_fh.close() except OSError: pass self.pcap_fh = None if self.gsmtap is not None: self.gsmtap.close() self.gsmtap = None def emit_packet(self, sub_type, payload, ss, es, flags=0): '''Emit one event as PCAP binary stream + PCAP file + GSMTAP UDP. `flags` is written into the GSMTAP header `res` byte (see GSMTAP_FLAG_BAD_FCS in gsmtap_stream).''' ts = self.ts_from_samplenum(ss) pkt = pcap_udp_pkt(ts, payload, sub_type, flags) self.put(ss, es, self.out_binary, [0, pkt.record_header()]) self.put(ss, es, self.out_binary, [0, pkt.packet()]) if self.pcap_fh is not None: self.pcap_fh.write(pkt.record_header()) self.pcap_fh.write(pkt.packet()) if self.gsmtap is not None: self.gsmtap.send(sub_type, payload, flags) def _line_hyst(self, lvl, confirmed, cand, cand_sn): '''Level-stability hysteresis for RST/VCC line tracking. A level different from `confirmed` becomes a *candidate*. It is confirmed (and returned for emission) once it has been continuously stable for >= vcc_min_gap samples -- either because it persists that long, or because it ends (the line moves again) after having been stable that long. This matters because the decoder is edge-driven: the stable window between a fall and the next rise is never re-sampled, so a candidate must be confirmable when it ENDS. Returns (emit_lvl, edge_sn, new_confirmed, new_cand, new_cand_sn): emit_lvl is the level just confirmed (or None); edge_sn is the sample the (now confirmed) level first appeared.''' now = self.samplenum if lvl == confirmed: # Candidate (if any) ended by returning to the confirmed level. # If it had been stable long enough, report it; the current line # level (lvl) is now the state to track as a fresh candidate, so # a genuine rise (e.g. power-up) is not lost on the same call. if cand is not None and ((not self.vcc_min_gap) or (now - cand_sn) >= self.vcc_min_gap): return (cand, cand_sn, cand, lvl, now) return (None, 0, confirmed, None, 0) if cand is None: # First divergence from confirmed. if not self.vcc_min_gap: # Hysteresis disabled: report the edge immediately. return (lvl, now, lvl, None, 0) return (None, 0, confirmed, lvl, now) if cand == lvl: # Same candidate persists: confirm once stable long enough. if (not self.vcc_min_gap) or (now - cand_sn) >= self.vcc_min_gap: return (lvl, cand_sn, lvl, None, 0) return (None, 0, confirmed, cand, cand_sn) # lvl differs from both confirmed and the current candidate: the # candidate just ended and a new level appeared. If the candidate # was stable, report it; then start tracking the new level. if (not self.vcc_min_gap) or (now - cand_sn) >= self.vcc_min_gap: return (cand, cand_sn, cand, lvl, now) return (None, 0, confirmed, lvl, now) def line_events(self, pins): '''Check RST/VCC levels via level-stability hysteresis and emit events + annotations for confirmed level changes. An edge is only reported once the new level has been continuously stable for >= vcc_min_gap samples, so a noisy/floating sense line that rings on a transition no longer drops the event. Returns True if any event fired.''' fired = False if (self.track_rst): lvl = pins[self.RST_IDX] emit_lvl, edge_sn, self.last_rst, self.rst_cand, self.rst_cand_sn = \ self._line_hyst(lvl, self.last_rst, self.rst_cand, self.rst_cand_sn) if emit_lvl is not None: direction = 1 if emit_lvl == 0 else 0 # 1=asserted, 0=deasserted self.emit_packet(GSMTAP_SIM_RST_EVENT, bytes([direction, emit_lvl, 0]), edge_sn, edge_sn) text = ('RST asserted' if emit_lvl == 0 else 'RST deasserted') self.put(edge_sn, edge_sn, self.out_ann, [Ann.ANN_RST, [text]]) self.log(text) fired = True if (emit_lvl == 0): # RST asserted (LOW): card entering reset state. A reset # makes the card send its ATR at the DEFAULT ETU (372 CLK # cycles/bit), so clock_skip must be restored to 372 -- # otherwise a post-PPS clock_skip (e.g. 16) would make the # decoder hunt for the fresh ATR at the wrong rate and # mis-frame every byte. Prepare for the next ATR by # re-arming the hunt and clearing _atr_parsed so the next # RST deassert will also re-arm (first cycle at startup). self._atr_parsed = False self._inverse_convention = False self.clock_skip = 372 self.bit_samples = None self.state = 'FIND START' self._atr_hunt_count = 0 self._pps_attempted = False if (emit_lvl == 1): # RST deasserted (HIGH): card released from reset and # will send a fresh ATR. However, the 10ms hysteresis # means this event fires AFTER the ATR has already been # parsed and the decoder entered DATA state. In that # case, stay in DATA so the terminal's T=0 commands # (which follow the ATR) are decoded. Only re-arm if # _atr_parsed is False (startup / first cycle, where no # ATR has been parsed yet). if self._atr_parsed: self.log("warm reset: staying in DATA (ATR already parsed)") else: # _atr_parsed is False: a fresh ATR is expected. The # 10ms hysteresis confirms the deassert ~10ms after # the physical edge -- typically in the MIDDLE of that # very ATR. By then the hunt is already armed with # the correct default rate (clock_skip=372, # bit_samples=None): either by the RST assert handler # (warm reset, edge observed) or by the "RST low at # start" cold-boot path. Re-arming here would clear # the in-flight ATR's timing and garble it, so only # re-arm if we are still sitting in DATA with nothing # armed (the one case the assert handler/cold-boot # path never covered: a deassert without a preceding # observed assert or hunt). if self.state == 'DATA': self.state = 'FIND START' self._atr_hunt_count = 0 self._pps_attempted = False self.bit_samples = None self._etu_confirm_count = 0 self._etu_fail_count = 0 self.clock_skip = 372 self._inverse_convention = False self._clock_skip_confident = \ not (self.sample_as_clock or self.detect_clock) self.log("RST deasserted with no hunt armed; re-arming ATR hunt") else: self.log("RST deasserted (hunt/ATR already in progress)") if (self.track_vcc): lvl = pins[self.VCC_IDX] emit_lvl, edge_sn, self.last_vcc, self.vcc_cand, self.vcc_cand_sn = \ self._line_hyst(lvl, self.last_vcc, self.vcc_cand, self.vcc_cand_sn) if emit_lvl is not None: direction = emit_lvl # 1=power applied, 0=power removed self.emit_packet(GSMTAP_SIM_VCC_EVENT, bytes([direction, emit_lvl, 0]), edge_sn, edge_sn) text = ('VCC on' if emit_lvl == 1 else 'VCC off') self.put(edge_sn, edge_sn, self.out_ann, [Ann.ANN_VCC, [text]]) self.log(text) fired = True return fired def pcap_global_header(self): # See https://wiki.wireshark.org/Development/LibpcapFileFormat. h = b'\xd4\xc3\xb2\xa1' # Magic, indicate microsecond ts resolution h += b'\x02\x00' # Major version 2 h += b'\x04\x00' # Minor version 4 h += b'\x00\x00\x00\x00' # Correction vs. UTC, seconds h += b'\x00\x00\x00\x00' # Timestamp accuracy h += b'\xff\xff\x00\x00' # Max packet len h += b'\x01\x00\x00\x00' # Link layer return h def get_bytes(self, bits): byte = 0 for bit in list(reversed(bits)): byte = (byte << 1) | bit return byte def write_pcap_header(self): if not self.wrote_pcap_header: self.put(0, 0, self.out_binary, [0, self.pcap_global_header()]) self.wrote_pcap_header = True def ts_from_samplenum(self, sample): # Use the full sample number. The previous c_uint32 truncation (plus # an overflow counter that was never incremented) made pcap timestamps # wrap and go backwards once a capture passed 2^32 samples (~4.5 min # at 16 MHz). float64 holds sample numbers well past 2^53, so the # whole sample domain is representable. ts = float(sample) * self.secs_per_sample + self._pcap_base_time return (int(ts), int((ts % 1.0) * 1e6)) def signal_quality(self, err): '''Feed one frame's parity result (1 = bad) into the rolling monitor; warn once per degradation episode when >= 8 of the last 64 frames failed parity.''' self.parity_hist.append(err) if len(self.parity_hist) < 32: return errs = sum(self.parity_hist) if errs >= 8 and not self.sig_warned: self.sig_warned = True self.log("SIGNAL INTEGRITY WARNING: {}/{} recent frames failed " "parity (noise? RF burst? contact?)".format( errs, len(self.parity_hist))) try: wss = max(self.ss, 0) wes = max(self.es, wss + 1) self.put(wss, wes, self.out_ann, [Ann.ANN_WARN, ["noisy signal"]]) except SystemError: raise except Exception: pass elif errs <= 2: self.sig_warned = False def read_first_byte(self): '''Find the first start-bit falling edge and read the first byte with the standard re-anchored reader. (Mid-session ETU recovery is done by _measure_etu, called once from the mid-session branch of decode_step before the normal T=0/T=1 framing state machine takes over.)''' self.wait_data_falling() return self.read_byte_no_wait() def _measure_clock_period(self): '''Measure the period of the native CLK signal in samples. For native CLK mode, the actual bit period is clock_skip * samples_per_clock. Measuring it directly removes the hard-coded 3× approximation and gives the correct ETU for the ATR and for any post-PPS speed. Returns True if a stable period was found. Collects multiple CLK rising-edge spacings and averages them, which handles CLK periods that alternate between samples (e.g. a 4,3,3 pattern for ~5.33 MHz at 16 MHz sample rate).''' if self.sample_as_clock or self.detect_clock: return False spacings = [] last_sn = None try: # Collect up to 16 consecutive CLK rising-edge spacings. for _ in range(16): pins = self.wait({self.CLK_IDX: 'r'}) if pins is None: return False sn = self.samplenum if last_sn is not None: spacings.append(sn - last_sn) last_sn = sn if len(spacings) < 3: return False # Use average period instead of robust_min to handle varying CLK periods # (e.g., 4,3,3 alternating pattern). period = sum(spacings) / len(spacings) if 2 <= period <= 100000: self._samples_per_clock = period self.log("clock period (samples):", period) return True except SystemError: raise except Exception: pass return False def _compute_bit_samples(self): '''Return the best available bit-period estimate in samples. Uses the measured clock period if available; otherwise falls back to the legacy clock_skip * 3 estimate.''' if self._samples_per_clock is not None: return int(self.clock_skip * self._samples_per_clock) return int(self.clock_skip * 3) def _derive_clock_skip_from_etu(self): '''Derive clock_skip (CLK edges per ETU) from the recovered ETU. For a mid-session capture (no ATR) the card may already be PPS'd to a fast ETU, so the default clock_skip=372 is wrong. Once _measure_etu has recovered bit_samples (samples per ETU) and _measure_clock_period has recovered _samples_per_clock (samples per CLK cycle), clock_skip = bit_samples / _samples_per_clock. Sets _clock_skip_confident on success so the CLK-sync reader is used; leaves it clear (edge-list fallback) if the CLK period could not be measured.''' if self._samples_per_clock and self.bit_samples: cs = int(round(float(self.bit_samples) / float(self._samples_per_clock))) if cs >= 1: if cs != self.clock_skip: self.log("clock_skip recovered from ETU:", cs, "(bit_samples", self.bit_samples, "/ spc", self._samples_per_clock, ")") self.clock_skip = cs self._clock_skip_confident = True return True return False def _measure_etu(self): '''Recover the live bit period (ETU, in samples) from the DATA line for a mid-session capture that has no ATR to derive timing from. Jumps to each DATA transition (O(edges), immune to Clock Stop Mode) and collects edge spacings until an idle gap ends the burst. The shortest *recurring* spacing equals one etu, so this recovers fast / non-standard ETUs (e.g. a phone SIM already PPS'd to a short etu) that the spec-default 372-clock etu would mis-read by ~20x. The result is **locked after 3 successful decodes**: bit_samples is set after measurement, then validated by decoding subsequent bytes. If 2 consecutive parity errors occur before validation, the measurement is discarded and re-measured from the next burst. Consumes one burst for the measurement; the normal framing resumes on the next start bit (the first exchange may therefore be mis-framed -- expected for a mid-session capture).''' if self.bit_samples is not None and self._etu_confirm_count >= 3: return if self._start_fall is None: self.wait_data_falling() if self._eof or self._start_fall is None: return e0 = self._start_fall self._start_fall = None # consumed; framing re-waits on its own self.ss = e0 idle_limit = int((self.bit_samples or self._compute_bit_samples()) * 20) if idle_limit < 3000: idle_limit = 3000 edges = [(e0, 0)] last_sn = e0 while True: pins = self.wait({self.DATA_IDX: 'e'}) if pins is None: break sn = self.samplenum if sn <= last_sn: break # no progress (end of capture) -> stop scan last_sn = sn d = pins[self.DATA_IDX] if (sn - edges[-1][0]) > idle_limit: # Big gap => burst ended; leave this edge for the framing. break edges.append((sn, d)) if len(edges) >= 2: spacings = [edges[i + 1][0] - edges[i][0] for i in range(len(edges) - 1)] # Filter glitches (< MIN_ETU) then check if the first spacing # is an outlier (scan started in a noisy pre-burst region). MIN_ETU = 12 valid = [s for s in spacings if s >= MIN_ETU] if len(valid) >= 2: med = sorted(valid)[len(valid) // 2] # If the first valid spacing is >10x the median, the scan # started at a burst boundary — drop leading outliers. while valid and valid[0] > med * 10: valid.pop(0) spacings = valid if valid else spacings etu = self._robust_min(spacings) # Sanity clamp: a real ISO 7816 etu is >= a few samples and below # ~100k samples even at extreme F/D. Anything else is a glitch. if 4 <= etu <= 100000: self.bit_samples = etu self._etu_confirm_count = 0 self._etu_fail_count = 0 if self.bit_samples is None: self.bit_samples = self._compute_bit_samples() self._etu_confirm_count = 0 self._etu_fail_count = 0 # The live line is gated/stopped CLK; use the edge-list bit reader # (immune to the drift a fixed sample-skip cadence would suffer) for # every byte from here on. self._edge_read = True self.log("etu recovered (samples):", self.bit_samples) def _recover_etu_in_hunt(self): '''Recover the live bit period from the DATA line while still in the ATR hunt. The hunt reads at the spec-default 372 CLK/bit. A device already running at a non-default F/D (Samsung F=512/D=32 -> 16 CLK/bit) or a mid-session capture with no ATR sends no 372-rate bytes, so every candidate reads as idle 0xFF and the hunt can never reach the DATA state -- where _measure_etu() normally lives. Measure the ETU here (after the default-rate hunt has already failed a few rounds) and re-derive clock_skip so subsequent hunt reads use the true rate. Reuses the DATA-path measurement and its _confirm_etu() validation.''' if self._samples_per_clock is None: self._measure_clock_period() self._measure_etu() self._derive_clock_skip_from_etu() # Start a fresh default-rate counting window at the recovered rate. self._atr_hunt_count = 0 self.log("hunt ETU recovery attempt", self._hunt_etu_attempts, "bit_samples", self.bit_samples, "clock_skip", self.clock_skip) def _atr_hunt_failure_action(self, byte): '''Decide what to do after a failed ATR hunt round. Returns one of: 'recover' -- stop and recover the live ETU, then re-hunt 'retry' -- go back to FIND START (bounded) 'resume_data' -- give up on the ATR and resume T=0 framing The counters are persistent across RST re-arms (reset only when an ATR is parsed) so a phone that cycles RST while never emitting a valid ATR cannot starve the escape. Pure function of state + byte so it is unit-testable without libsigrokdecode.''' self._atr_fail_total += 1 # Live-ETU recovery every 2 failures, up to ATR_HUNT_ETU_ATTEMPTS. if (self.bit_samples is None and self._hunt_etu_attempts < ATR_HUNT_ETU_ATTEMPTS and self._atr_fail_total >= 2 * (self._hunt_etu_attempts + 1)): self._hunt_etu_attempts += 1 return 'recover' # Hard bound: never hunt forever on an idle/unsyncable line. if self._atr_fail_total >= ATR_HUNT_FAIL_LIMIT: return 'resume_data' if byte == 0xFF and self._atr_idle_resets >= ATR_HUNT_IDLE_RESETS: return 'resume_data' if self._atr_hunt_count < 8: return 'retry' return 'resume_data' def _confirm_etu(self, parity_ok): '''Validate ETU measurement after _measure_etu. Call after each byte decode in _read_byte_edges. After 3 successful decodes the measurement is locked. If 2 consecutive parity errors occur before validation, the measurement is discarded and re-measured from the next burst.''' if self._etu_confirm_count >= 3: return if parity_ok: self._etu_confirm_count += 1 if self._etu_confirm_count >= 3: self.log("ETU confirmed:", self.bit_samples) else: self._etu_fail_count += 1 if self._etu_fail_count >= 2: self.log("ETU re-measure (bad initial measurement):", self.bit_samples) self.bit_samples = None self._etu_confirm_count = 0 self._etu_fail_count = 0 def _robust_min(self, spacings): '''Smallest spacing that recurs (>=3x): the true etu. Two consecutive differing bits produce a 1-etu gap, so a real bit period recurs often; a glitch is unique. Falls back to the raw minimum. Spacings below MIN_ETU (12 samples at 16 MHz) are rejected as metastability glitches — they are too short for any valid ISO 7816 bit period and would otherwise be picked as the "minimum recurring" spacing by the counter.''' MIN_ETU = 12 from collections import Counter c = Counter(s for s in spacings if s >= MIN_ETU) for v in sorted(c): if c[v] >= 3: return v # Fallback: accept any spacing >= MIN_ETU, or raw min if nothing qualifies. valid = [s for s in spacings if s >= MIN_ETU] return min(valid) if valid else min(spacings) def _deglitch_edges(self, edges, bs): '''Remove pulses shorter than a fraction of an ETU from an edge list. Phone traces (especially Samsung at 16 CLK/ETU) contain 1-sample (62.5 ns) LOW glitches in inter-byte guard times. These are far shorter than the shortest real ISO 7816 pulse (about 1 ETU) and would otherwise be accepted as false start-bit falls. The filter uses an iterative narrowest-first strategy: at each step it finds the shortest pulse, and if it is shorter than 0.2 * bs it removes the edge that *ends* the pulse. Removing the end edge lets the previous (real) edge's value persist, re-merging the surrounding fragments back into the original ETU-multiple pulse. The first edge (the start-bit fall passed to _read_byte_edges) is never removed. Returns a new edge list; the input list is not modified.''' if not edges: return edges threshold = max(1, int(0.2 * bs)) # Work with a mutable copy. Each item is [sample, value]. out = [[s, v] for (s, v) in edges] changed = True while changed: changed = False # Find the shortest pulse (distance between consecutive edges). min_width = None min_idx = None n = len(out) for i in range(n - 1): width = out[i + 1][0] - out[i][0] if width < threshold and (min_width is None or width < min_width): min_width = width min_idx = i if min_idx is not None: # Remove the edge that ends the short pulse. This keeps the # preceding edge (the real transition) and lets its value # persist until the next edge. If the short pulse starts at # the first edge, keep the first edge and remove the second. del out[min_idx + 1] changed = True return [(s, v) for [s, v] in out] def _bits_at(self, e, edges, bs): '''Reconstruct the 10 bits (start + 8 data + parity) of the byte whose start-bit fall is at sample e, from the DATA edge list. Returns None only if `e` is not a genuine start (no following edge).''' bits = [] for k in range(10): m = e + round((k + 0.5) * bs) val = 1 for (s, v) in edges: if s <= m: val = v else: break # Inverse convention (TS=0x3F): logic 1 = LOW, logic 0 = HIGH. if self._inverse_convention: val = 1 - val bits.append(val) return bits def _next_start_fall(self, edges, after, bs): '''First GENUINE byte-start FALL (1->0) at or after sample `after`, = the next byte start. A real start bit is LOW for a full ETU. A sub-ETU fall inside the stop-bit / inter-byte guard time is a phantom (line bounce) and must not be treated as the next byte's start -- latching onto it shifts the whole frame. A fall is accepted only if the line stays LOW for at least half an ETU after it (a genuine start bit's low hold is ~1 ETU; the first data-bit transition arrives no sooner).''' thresh = int(0.5 * bs) for i in range(len(edges)): s, v = edges[i] if s < after or v != 0: continue nxt = edges[i + 1][0] if i + 1 < len(edges) else s + thresh + 1 if (nxt - s) < thresh: continue return s return None def _wait_clk_rising(self, n): '''Wait for n CLK rising edges. Returns False on end-of-capture. CLK runs continuously for the whole of a character (Clock Stop only happens between characters), so within a byte it is safe to skip most of the way by samples and then land on the exact rising edge. This keeps the edge-exactness of CLK-edge timing without paying one wait() per CLK cycle. Lands within +/-1 CLK cycle of the target edge, which is far inside the stable region of a bit (a bit is held for the whole ETU).''' if n <= 0: return True if self._samples_per_clock and n > 2: # Skip (n - 2) cycles by samples, then take the last 2 edges # exactly. Undershooting by a cycle is harmless (the bit is # stable for the full ETU), and the final two edge waits re-align # us to a genuine CLK edge. skip = int((n - 2) * self._samples_per_clock) if skip > 0: self.wait({'skip': skip}) n = 2 for _ in range(n): pins = self.wait({self.CLK_IDX: 'r'}) if pins is None: self._eof = True return False return True def _read_byte_clk(self): '''Decode a single byte using CLK-edge counting for bit timing. Counts clock_skip CLK rising edges per ETU and samples DATA at the bit centers (1.5, 2.5, ..., 9.5 ETU from the start-bit fall). Because the bit value is held stable for the whole ETU, sampling near the centre is exact for ANY byte -- including 0x00 / 0xff, which have too few DATA transitions for the edge-list reader to anchor reliably. Safe across Clock Stop Mode: CLK is never stopped mid-character, and it always resumes a few cycles before the next start bit (the guard time), so once wait_data_falling() has found a start bit, CLK is running for the whole frame. clock_skip is protocol-defined (372 for the ATR, FI/DI after PPS), so no runtime ETU measurement is needed on the native path -- this is what makes decoding deterministic.''' if self._start_fall is None: self.wait_data_falling() if self._eof: return 0x00 e0 = self._start_fall self._start_fall = None self.ss = e0 bs = self.clock_skip self.bits = [0] # start bit (always low) prev_edges = 0 for i in range(9): # bit centres at (1.5 + i) ETU from the start fall total = int(round((1.5 + i) * bs)) if not self._wait_clk_rising(total - prev_edges): # End of capture mid-byte: pad so the frame stays 10 bits. while len(self.bits) < 10: self.bits.append(self.bits[-1] if self.bits else 1) break prev_edges = total pins = self.wait({'skip': 0}) if pins is None: self._eof = True while len(self.bits) < 10: self.bits.append(self.bits[-1] if self.bits else 1) break bit = pins[self.DATA_IDX] # Inverse convention (TS=0x3F): logic 1 = LOW, logic 0 = HIGH. if self._inverse_convention: bit = 1 - bit self.bits.append(bit) self.es = self.samplenum self.signal_quality(1 if (self.bits.count(1) % 2 != 0) else 0) parity_ok = (self.bits.count(1) % 2 == 0) if (self.bits.count(1) % 2 != 0): self.log(self.samplenum, "CHKSUM ERROR: ", 0, "bits: ", self.bits) self.put(self.ss, self.samplenum, self.out_ann, [0, ["CHKSUM ERROR bits={bits}".format(bits=self.bits)]]) if self.bit_samples is not None: self._confirm_etu(parity_ok) byte = self.get_bytes(self.bits[1:9]) # Stuck-line watchdog (same as the other readers). if (sum(self.bits) == 0): self.zero_run += 1 if (self.zero_run == 32): self.log("I/O line reads all-zero for", self.zero_run, "frames - stuck low? check pogo contact") try: self.put(self.ss, self.samplenum, self.out_ann, [Ann.ANN_WARN, ["I/O stuck low?"]]) except SystemError: raise except Exception: pass else: self.zero_run = 0 self.log(self.samplenum, self.bits[1:9], " : ", "0x{:02x}".format(byte)) self.put(self.ss, self.samplenum, self.out_ann, [1, [hex(byte)]]) return byte def _read_byte_edges(self, bs): '''Decode a single byte from the DATA line using edge-list reconstruction (robust to Clock Stop Mode / gated CLK, where fixed sample-skip timing drifts). Returns the byte, having set self.ss/es and queued the next start-bit position in self._start_fall. Jumps straight to DATA edges around the current start fall (O(edges), not O(samples)), collects ~11 etu of edges (one byte plus the gap to the next start bit), reconstructs the 10 bits from their actual positions, and leaves self._start_fall at the next byte so subsequent reads stay re-anchored (no cross-byte drift). A large turn-around idle (CLK stopped between command and response) simply leaves _start_fall unset here, and the next read re-hunts via wait_data_falling().''' # Safety: if we were called without a queued start bit, hunt for one. if self._start_fall is None: self.wait_data_falling() if self._eof: return 0x00 e0 = self._start_fall self._start_fall = None self.ss = e0 # Collect DATA edges for this byte + the inter-byte gap. # Use a 15 ETU window so a 2 ETU turnaround idle (observed on # Samsung phone at 16 CLK/bit) still leaves the next start-bit fall # inside the collected edge list. edges = [(e0, 0)] limit = e0 + int(15 * bs) last_sn = e0 while True: pins = self.wait({self.DATA_IDX: 'e'}) if pins is None: break sn = self.samplenum if sn <= last_sn: break last_sn = sn edges.append((sn, pins[self.DATA_IDX])) if sn > limit: break # Remove sub-ETU glitch pulses (common on phone SIMs at fast ETUs # like 16 CLK/bit). Without this the edge reader may latch onto a # 1-sample bounce in an inter-byte guard time and read the wrong # byte, desynchronising the whole T=0 frame. edges = self._deglitch_edges(edges, bs) bits = self._bits_at(e0, edges, bs) # Keep the spacings for ETU locking; the first few bytes of an ATR # provide the most reliable reference for the actual bit period. if len(edges) >= 2: for i in range(1, len(edges)): self._edge_spacings.append(edges[i][0] - edges[i - 1][0]) nxt = None if bits is not None and bits[0] == 0: self.bits = bits byte = self.get_bytes(bits[1:9]) nxt = self._next_start_fall(edges, e0 + int(10 * bs), bs) else: # bits[0] is the start bit; it must be 0 after _bits_at() because # _bits_at() already inverts the value for inverse convention. # bits[0] != 0 means we latched onto a spurious edge (phantom). # Do NOT drop the byte: dropping replaces a real byte with a fake # 0x00 / shifts the exchange boundary, which is exactly how a # whole exchange gets swallowed (CONCAT). Instead slide forward # to the next genuine start-bit fall inside this same edge window # and re-anchor there. for (s, v) in edges: if s <= e0 or v != 0: continue cand = self._bits_at(s, edges, bs) if cand is not None and cand[0] == 0: e0 = s bits = cand self.bits = bits byte = self.get_bytes(bits[1:9]) nxt = self._next_start_fall(edges, e0 + int(10 * bs), bs) break else: # No genuine start in the window (EOF / pathological line): # keep the legacy drop; the next read re-hunts. self.bits = [] byte = 0 self.ss = e0 if nxt is not None: self._start_fall = nxt self.es = self.samplenum self.signal_quality(1 if (self.bits.count(1) % 2 != 0) else 0) parity_ok = (self.bits.count(1) % 2 == 0) if (self.bits.count(1) % 2 != 0): self.log(self.samplenum, "CHKSUM ERROR: ", 0, "bits: ", self.bits) self.put(self.ss, self.samplenum, self.out_ann, [0, ["CHKSUM ERROR bits={bits}".format(bits=self.bits)]]) if self.bit_samples is not None: self._confirm_etu(parity_ok) if (sum(self.bits) == 0): self.zero_run += 1 if (self.zero_run == 32): self.log("I/O line reads all-zero for", self.zero_run, "frames - stuck low? check pogo contact") try: self.put(self.ss, self.samplenum, self.out_ann, [Ann.ANN_WARN, ["I/O stuck low?"]]) except SystemError: raise except Exception: pass else: self.zero_run = 0 self.log(self.samplenum, self.bits[1:9], " : ", "0x{:02x}".format(byte)) self.put(self.ss, self.samplenum, self.out_ann, [1, [hex(byte)]]) return byte def wait_data_falling(self): '''Wait for the I/O start-bit falling edge, servicing optional RST/VCC line events while hunting. The bit period is measured directly from the DATA line in SAMPLES (not by counting CLK edges), so a frozen CLK during Clock Stop Mode cannot desync the decoder. A start bit is accepted only after DATA has been HIGH for >= 1.5 bit periods (inter-character guard, rejects intra-byte edges / idle glitches).''' min_high = 4 if self.bit_samples: if self._edge_read: # DATA must be HIGH (idle) for at least half an etu before a # fall is accepted as a start bit. This rejects intra-byte # glitch falls yet still accepts the legitimate inter-byte gap # (the ~1-etu stop bit), which a 1.5-etu guard would wrongly # reject on fast ETUs (e.g. a phone SIM at ~53 samples/etu), # causing bytes to be skipped / 0x00 inserted. min_high = max(int(0.5 * self.bit_samples), 4) else: min_high = max(int(1.5 * self.bit_samples), 4) while True: # Wait for DATA idle, servicing RST/VCC. # In direct convention: idle = HIGH, start bit = falling edge (HIGH→LOW). # In inverse convention: idle = LOW, start bit = rising edge (LOW→HIGH). if self._inverse_convention: idle_cond = {self.DATA_IDX: 'l'} # Wait for LOW (idle in inverse) start_edge = 'r' # Rising edge = start bit in inverse else: idle_cond = {self.DATA_IDX: 'h'} # Wait for HIGH (idle in direct) start_edge = 'f' # Falling edge = start bit in direct cond = [idle_cond] if (self.track_rst): cond.append({self.RST_IDX: 'e'}) if (self.track_vcc): cond.append({self.VCC_IDX: 'e'}) while True: pins = self.wait(cond) if self._stall_check(pins): self._eof = True return if (self.line_events(pins)): continue break hi_start = self.samplenum # Wait for the start-bit edge; re-anchor the high-start across # any line events so the measured guard stays accurate. cond2 = [{self.DATA_IDX: start_edge}] if (self.track_rst): cond2.append({self.RST_IDX: 'e'}) if (self.track_vcc): cond2.append({self.VCC_IDX: 'e'}) while True: pins = self.wait(cond2) if self._stall_check(pins): self._eof = True return if (self.line_events(pins)): hi_start = self.samplenum continue break if (self.samplenum - hi_start) < min_high: continue if self._min_pulse_samples > 0: pins = self.wait({'skip': self._min_pulse_samples}) if pins is not None: bit = pins[self.DATA_IDX] idle_val = 1 if not self._inverse_convention else 0 if bit == idle_val: self._glitch_rejects += 1 self.log("start-bit glitch rejected (#%d)" % self._glitch_rejects) continue self._start_fall = self.samplenum return def _stall_check(self, pins): '''Detect end-of-capture: libsigrokdecode stops advancing the sample counter once the feed is exhausted, so a wait that returns the same sample number repeatedly means there is no more data.''' if pins is None: return True if self._prev_wait_sn is not None and self.samplenum == self._prev_wait_sn: self._stall += 1 else: self._stall = 0 self._prev_wait_sn = self.samplenum return self._stall > 4 def read_byte(self): if self._replay: return self._replay.popleft() if (self.peeked_byte != None): byte = self.peeked_byte self.peeked_byte = None self.peeked_samplenum = -1 return byte; # In edge-read (mid-session, gated-CLK) mode the next start bit is # already queued by _read_byte_edges, so re-hunting would wrongly # reject the short inter-byte gap. Otherwise re-hunt normally. if not (self._edge_read and self._start_fall is not None): self.wait_data_falling() if self._eof: return None return self.read_byte_no_wait() def peek_byte(self): if self._replay: return self._replay[0] if (self.peeked_byte != None): return self.peeked_byte if not (self._edge_read and self._start_fall is not None): self.wait_data_falling() if self._eof: return None self.peeked_samplenum = self._start_fall self.peeked_byte = self.read_byte_no_wait() return self.peeked_byte def read_byte_no_wait(self): self.bits = [] if (self.sample_as_clock): # Experimental mode: DATA doubles as the clock. Keep the # original sample-as-clock reader (bits sampled on the DATA # edge, with no separate timing measurement). self.ss = self.samplenum self.clock_skip = 0 pins = self.wait({1: 'r'}) self.clock_skip = self.samplenum - self.ss + 2 self.detected_clock_skip = self.clock_skip self.wait({'skip': int(self.clock_skip / 3)}) self.bits.append(0) for x in range(9): pins = self.wait({'skip': 0}) bit = pins[1] # Inverse convention (TS=0x3F): logic 1 = LOW, logic 0 = HIGH. if self._inverse_convention: bit = 1 - bit self.bits.append(bit) self.wait({'skip': self.clock_skip - 4}) self.es = self.samplenum self.signal_quality(1 if (self.bits.count(1) % 2 != 0) else 0) if (self.bits.count(1) % 2 != 0): self.log(self.samplenum, "CHKSUM ERROR: ", pins[1], "bits: ", self.bits) self.put(self.ss, self.samplenum, self.out_ann, [0, ["CHKSUM ERROR bits={bits}".format(bits=self.bits)]]) byte = self.get_bytes(self.bits[1:9]) if (sum(self.bits) == 0): self.zero_run += 1 if (self.zero_run == 32): self.log("I/O line reads all-zero for", self.zero_run, "frames - stuck low? check pogo contact") try: self.put(self.ss, self.samplenum, self.out_ann, [Ann.ANN_WARN, ["I/O stuck low?"]]) except SystemError: raise except Exception: pass else: self.zero_run = 0 self.log(self.samplenum, self.bits[1:9], " : ", "0x{:02x}".format(byte)) self.put(self.ss, self.samplenum, self.out_ann, [1, [hex(byte)]]) return byte # Normal path. Two bit readers are available: # - CLK-synchronous (_read_byte_clk): exact CLK-edge timing, used in # native mode where a real CLK channel is present. Handles 0x00 / # 0xff perfectly and is deterministic. # - Edge-list (_read_byte_edges): DATA-edge reconstruction, used for # the non-native clock modes (sample_as_clock / detect) and as a # fallback when clock_skip has not been established yet. # Callers (read_byte / peek_byte / read_first_byte) ensure # _start_fall is set before calling here. if self._use_clk_sync(): return self._read_byte_clk() return self._read_byte_edges(self.bit_samples or self._compute_bit_samples()) def _use_clk_sync(self): '''True when the CLK-synchronous reader should be used. Only in native clock mode (a real CLK channel) AND only when clock_skip is known to be correct (_clock_skip_confident). In sample_as_clock / detect modes there is no usable CLK signal, and on a mid-session capture clock_skip is unknown until recovered -- in both cases the edge-list DATA reader (driven by bit_samples) is the safe fallback.''' return (not (self.sample_as_clock or self.detect_clock) and self._clock_skip_confident) def resync_idle(self): '''Re-align framing after a desync: wait for DATA to go to idle level so the next peek lands on a genuine start bit of a fresh exchange instead of a mid-burst byte. Gating-immune: we poll DATA in samples (which advance even when CLK is frozen), and bound the hunt by a sample budget so a pathologically-busy (or stuck) line never stalls the decoder.''' bs = self.bit_samples or self._compute_bit_samples() budget = int(IDLE_RESYNC_ETU * bs) * 8 step = max(int(bs), 1) loops = budget // step + 16 # In direct convention: idle = HIGH (mark). # In inverse convention: idle = LOW (mark is inverted). idle_level = 0 if self._inverse_convention else 1 for _ in range(loops): pins = self.wait({'skip': 0}) if pins[self.DATA_IDX] == idle_level: return # Could not find an idle gap -- resume normal decode anyway. def handle_atr(self, pins): atr_start = self.samplenum; self.log("START ATR:", self.samplenum); self.state = 'ATR' tA = [] tB = [] tC = [] tD = [] historicalBytes = [] self.ATR = [] # Clear any stale edge spacings; the first few ATR bytes give us the # most reliable ETU reference for the whole session. self._edge_spacings.clear() # Try to lock the ETU from the actual CLK frequency before reading the # first ATR byte. This removes the hard-coded clock_skip*3 estimate. if self._samples_per_clock is None: self._measure_clock_period() # Ensure the sample-domain ETU is available for framing # (wait_data_falling / resync_idle) even though the CLK-sync reader # itself only needs clock_skip. if self.bit_samples is None: self.bit_samples = self._compute_bit_samples() if (self.peeked_byte != None): atr_start = self.peeked_samplenum byte = self.read_byte() else: byte = self.read_first_byte() atr_start = self.ss # TS must be 0x3B (direct) or 0x3F (inverse) per TS 102 221 §5.2.2. # Pre-activation traffic (reader card-detect polling, a stray PPS # `ff 10 95 7a`, power-up noise) can appear on DATA before RST even # deasserts. Resync past those instead of mis-parsing them as the # ATR and permanently desyncing the session. ts_attempts = 0 while (byte not in (0x3b, 0x3f) and ts_attempts < 64): # Command-detect: a plausible CLA immediately followed by a # plausible INS is the start of a command, not an ATR. Hand the # header back to the DATA state for normal T=0 framing instead of # silently skipping it as an "invalid TS" byte (this is how a # SELECT was being lost while the decoder hunted for an ATR that # was not coming). The full 5-byte header is still validated in # the DATA state; ambiguous frames are emitted flagged, not dropped. # # CLA=0x00 is excluded: it's technically valid for interindustry # commands, but during ATR hunt it's overwhelmingly noise (the # real SELECT that was lost used CLA=0x00 and appeared *after* the # ATR, not during the hunt). Including it causes false positives # on noise like the 0x00 0x1E seen in the live test. if plausible_cla(byte) and byte != 0x00: nxt = self.peek_byte() if plausible_ins(nxt): self._replay.appendleft(byte) self.hasT0 = True self.hasT1 = False self.state = 'DATA' self.log("Command header 0x{cla:02x} 0x{ins:02x} during ATR hunt; decoding as command".format(cla=byte, ins=nxt)) return self.log("Invalid TS byte 0x{byte:02x} (pre-activation?), resyncing".format(byte=byte)) self.put(atr_start, self.samplenum, self.out_ann, [Ann.ANN_WARN, ["Invalid TS byte 0x{byte:02x} - skipped".format(byte=byte)]]) byte = self.read_byte() if byte is None: # End of capture mid-hunt: stop cleanly instead of raising # TypeError (None & 0x80) in the next loop iteration. return atr_start = self.ss ts_attempts += 1 if (byte not in (0x3b, 0x3f)): self._atr_hunt_count += 1 # No valid ATR found this round. Never fabricate a synthetic # ATR: a stray 0x3B/0x3F byte inside command traffic (e.g. the # FID '3F00' of a SELECT) would otherwise be mis-committed as a # fake 2-byte ATR and destroy the surrounding command. action = self._atr_hunt_failure_action(byte) if action == 'recover': # Default-rate hunt has failed a few rounds: the card may be # idle or already at a non-default F/D. Measure the live ETU # and re-hunt at that rate instead of looping on idle 0xFF. self.log("ATR hunt failing; recovering live ETU") self._recover_etu_in_hunt() self.state = 'FIND START' return if action == 'retry': if byte == 0xFF: self._atr_idle_resets += 1 self.log("line appears idle (all 0xFF), resetting ATR " "hunt (#{n})".format(n=self._atr_idle_resets)) self._atr_hunt_count = 0 else: self.log("No valid TS byte, back to FIND START " "(hunt {count})".format(count=self._atr_hunt_count)) self.state = 'FIND START' return # Give up hunting for an ATR here without emitting anything; # resume T=0 framing from the next byte instead of inventing one. # Clear bit_samples so the DATA path measures the ETU fresh. self.log("No valid ATR after {count} hunts (fails={fails}); " "resuming DATA without synthetic ATR".format( count=self._atr_hunt_count, fails=self._atr_fail_total)) self.bit_samples = None self._etu_confirm_count = 0 self._etu_fail_count = 0 self.hasT0 = True self.hasT1 = False self.state = 'DATA' self._resync = True if (self.options['protocol'] == "T=0"): self.hasT0 = True self.hasT1 = False elif (self.options['protocol'] == "T=1"): self.hasT0 = False self.hasT1 = True return else: self._atr_hunt_count = 0 self._atr_fail_total = 0 self._hunt_etu_attempts = 0 self._atr_idle_resets = 0 self.ATR.append(byte) # TS=0x3F means inverse convention: logic 1 = LOW, logic 0 = HIGH. # All subsequent byte reads must invert bits. self._inverse_convention = (byte == 0x3f) self.log("TS=0x{ts:02x} ({conv} convention)".format( ts=byte, conv="inverse" if self._inverse_convention else "direct")) t0 = self.read_byte() if t0 is None: return self.ATR.append(t0) firstT0 = t0 self.hasT0 = False self.hasT1 = False self.hasT15 = False while (firstT0 & 0b11110000): # Defense-in-depth: a valid ISO 7816 ATR is at most 33 bytes. # If the parse runs past that (e.g. a noise byte slipped in as # TS and the TD chain never terminates), abandon the parse and # hunt again rather than commit a giant bogus ATR. if (len(self.ATR) > 33): self.log("ATR exceeds 33 bytes (noise-induced?), abandoning parse") self._atr_hunt_count += 1 self.state = 'FIND START' return if (firstT0 & 0b00010000): byte = self.read_byte() tA.append(byte) self.ATR.append(byte) if (firstT0 & 0b00100000): byte = self.read_byte() tB.append(byte) self.ATR.append(byte) if (firstT0 & 0b01000000): byte = self.read_byte() tC.append(byte) self.ATR.append(byte) if (firstT0 & 0b10000000): byte = self.read_byte() if ((byte & 0x0F) == 0): self.hasT0 = True elif ((byte & 0x0F) == 1): self.hasT1 = True elif ((byte & 0x0F) == 15): self.hasT15 = True else: self.log("Invalid Protocol in ATR: ", "T=",(byte & 0x0F)) self.put(atr_start, self.samplenum, self.out_ann, [0, ["Invalid Protocol in ATR T={protocol}".format(protocol=(byte & 0x0F))]]) tD.append(byte) self.ATR.append(byte) firstT0 = byte self.log("TD("+str(len(tD))+"): ", hex(firstT0), "T=",(byte & 0x0F)) else: firstT0 = 0; for _ in range(0, t0 & 0x0F): byte = self.read_byte() self.ATR.append(byte) if (self.hasT1 == False and self.hasT0 == False): self.hasT0 = True if (self.hasT1 == True or self.hasT15): byte = self.read_byte() self.ATR.append(byte) xor = 0 for i in range(1, len(self.ATR)): xor = xor ^ self.ATR[i] if (xor != 0): self.put(atr_start, self.samplenum, self.out_ann, [0, ["Invalid TCK in ATR, got={tck:02x} expected={xor:02x}".format(tck=byte,xor=xor)]]) self.log("Invalid TCK in ATR", hex(byte), hex(xor)) self.put(atr_start, self.samplenum, self.out_ann, [2, ["ATR", "ATR={atr}".format(atr=codecs.encode(bytes(self.ATR), 'hex'))]]) self.put(atr_start, self.samplenum, self.out_python, [0, self.ATR]) self.emit_packet(GSMTAP_SIM_ATR, bytes(self.ATR), atr_start, self.samplenum) self.log("ENDATR", codecs.encode(bytes(self.ATR), 'hex')) self._atr_parsed = True # A real ATR was parsed at the default rate, so clock_skip=372 is now # confirmed correct -- the CLK-sync reader is trusted from here on. self._clock_skip_confident = True # The ATR is always sent at the default rate (372 CLK/bit). # _measure_etu() may have inflated clock_skip via inter-byte gap # averaging; reset to 372 so PPS (also at the default rate) is # read at the correct bit boundary. self.clock_skip = 372 # Also lock bit_samples so the mid-session re-measure in decode_step # does not re-run and overwrite clock_skip=372 with a wrong value # derived from pre-ATR edge spacings. _confirm_etu() may have # cleared bit_samples during ATR hunt (parity failures on noise), # which would trigger _measure_etu() on the next decode_step. self.bit_samples = self.clock_skip * (self._samples_per_clock or 3) # Lock the ETU so _confirm_etu() cannot clear bit_samples after # parity failures on post-ATR noise bytes. Without this, two # consecutive parity errors would trigger an ETU re-measure that # overwrites clock_skip=372 with a wrong value from pre-ATR data. self._etu_confirm_count = 3 self._etu_fail_count = 0 self.state = 'DATA' self._resync = True # Wait for inter-frame idle gap before reading commands # PPS may be attempted once after this ATR. self._pps_attempted = False if (self.options['protocol'] == "T=0"): self.hasT0 = True self.hasT1 = False elif (self.options['protocol'] == "T=1"): self.hasT0 = False self.hasT1 = True def t1_parse_block(self, es): packet = [] lrc = 0; nad = self.read_byte() lrc = lrc ^ nad self.put(es, self.samplenum, self.out_ann, [1, "NAD:" + hex(nad)]) sad = nad & 0x70 dad = nad & 0x07 self.log("T=1 NAD=", hex(nad), "SAD=", hex(sad), "DAD=", hex(dad)) pcb = self.read_byte() lrc = lrc ^ pcb isSBlock = False isRBlock = False isIBlock = False if (pcb & 0b11000000 == 0b11000000): # S-Block isSBlock = True self.put(es, self.samplenum, self.out_ann, [3, "PCB S " + hex(pcb)]) self.log("PCB S-Block", hex(pcb & 0b00111111)) elif (pcb & 0b10000000 == 0b10000000): # R-Block isRBlock = True self.put(es, self.samplenum, self.out_ann, [3, "PCB R " + hex(pcb)]) self.log("PCB R-Block", hex(pcb & 0b00111111)) else: # I-Block isIBlock = True self.put(es, self.samplenum, self.out_ann, [3, "PCB I " + hex(pcb)]) self.log("PCB I-Block", hex(pcb)) bLen = self.read_byte() lrc = lrc ^ bLen if (bLen > 0): for _ in range(bLen): byte = self.read_byte() lrc = lrc ^ byte if (isIBlock): packet.append(byte) # CRC to implement bLrc = self.read_byte() lrc = lrc ^ bLrc if (lrc != 0): self.put(es, self.samplenum, self.out_ann, [0, ["Invalid checksum on T=1 block, , got={got:02x} expected={expected:02x}".format(got=lrc,expected=bLrc)]]) self.log("Invalid checksum on T=1 block", hex(lrc), hex(bLrc)) self.log("block_content", codecs.encode(bytes(packet), 'hex')) if (isIBlock): self.put(es, self.samplenum, self.out_ann, [6, ["I-Block", "I-Block len={len} isMultiBlock={multi}".format(len=bLen,multi=(pcb & 0b00100000 > 0))]]) if (isRBlock): self.put(es, self.samplenum, self.out_ann, [7, ["R-Block", "R-Block flag={flag:02x}".format(flag=pcb & 0b00111111)]]) if (isSBlock): self.put(es, self.samplenum, self.out_ann, [8, ["S-Block", "S-Block flag={flag:02x}".format(flag=pcb & 0b00111111)]]) if (isIBlock and pcb & 0b00100000): # m-flag self.log("T=1 Multiblock flag", hex(pcb)) for _ in range(32): # bounded: garbage must not loop forever isIBlock2,packet2 = self.t1_parse_block(self.samplenum) if (isIBlock2): packet = packet + packet2 break return isIBlock,packet; def handle_pps(self): # Mark PPS attempted so idle 0xFF bytes after a failed/no-PPS # scenario do not loop back into PPS detection indefinitely. self._pps_attempted = True lrc = 0 ss = self.peeked_samplenum pps = self.read_byte() pps_req = [pps] pps0 = self.read_byte() pps_req.append(pps0) # An idle I/O line idles HIGH (repeated 0xFF). A genuine PPS0 always # has its reserved bits clear (bit7 and bits 0..3 == 0), so the value # is one of 0x00/0x10/0x20/.../0x70. 0xFF (or any byte with bit7 or # low nibble set) is not a PPS0 -- it is idle noise. Without this # guard an idle line is mis-read as an endless stream of "PPS accepted" # and real APDUs are never reached. if (pps0 & 0x8F) != 0: self.log("0xFF followed by 0x%02x is not a valid PPS0 - idle line, skipping" % pps0) self._resync = True return pps1 = 0; pps2 = 0; pps3 = 0 if (pps0 & 0b00010000): pps1 = self.read_byte() pps_req.append(pps1) lrc = lrc ^ pps1 if (pps0 & 0b00100000): pps2 = self.read_byte() pps_req.append(pps2) lrc = lrc ^ pps2 if (pps0 & 0b01000000): pps3 = self.read_byte() pps_req.append(pps3) lrc = lrc ^ pps3 pck = self.read_byte() pps_req.append(pck) lrc = lrc ^ pps ^ pps0 ^ pck if (lrc != 0): self.put(ss, self.samplenum, self.out_ann, [0, ["INVALID Checksum on PPS Request, got={got:02x} expected={expected:02x}".format(got=pck,expected=(lrc ^ pps ^ pps0))]]) self.log("INVALID Checksum on PPS Request", hex(lrc)) r_lrc = 0 r_pps = self.read_byte() pps_rsp = [r_pps] r_pps1 = 0; r_pps2 = 0; r_pps3 = 0 if (r_pps != 0xFF): self.put(ss, self.samplenum, self.out_ann, [0, ["PPS Request not confirmed"]]) self.log("PPS Request not confirmed", r_pps) r_pps0 = self.read_byte() pps_rsp.append(r_pps0) if (r_pps0 & 0b00010000): r_pps1 = self.read_byte() pps_rsp.append(r_pps1) r_lrc = r_lrc ^ r_pps1 if (r_pps0 & 0b00100000): r_pps2 = self.read_byte() pps_rsp.append(r_pps2) r_lrc = r_lrc ^ r_pps2 if (r_pps0 & 0b01000000): r_pps3 = self.read_byte() pps_rsp.append(r_pps3) r_lrc = r_lrc ^ r_pps3 r_pck = self.read_byte() pps_rsp.append(r_pck) r_lrc = r_lrc ^ r_pps ^ r_pps0 ^ r_pck if (r_lrc != 0): self.put(ss, self.samplenum, self.out_ann, [0, ["INVALID Checksum on PPS Response, got={got:02x} expected={expected:02x}".format(got=r_pck,expected=(r_lrc ^ r_pps ^ r_pps0))]]) self.log("INVALID Checksum on PPS Response", hex(r_lrc)) if (pps0 == r_pps0 and pps1 == r_pps1 and pps2 == r_pps2 and pps3 == r_pps3): if (self.detect_clock or self.sample_as_clock): # .get() with spec defaults: unknown FI/DI codes from a # corrupted PPS1 must not raise KeyError or divide by zero. tmp_fi = self.clock_rate.get(int(pps1 >> 4), 372) tmp_di = self.baud_rate.get(int(pps1 & 0x0F), 1) if tmp_di == 0: self.log("Corrupted PPS: DI=0 (invalid)") return tmp_clock_skip = int(tmp_fi / tmp_di) self.log("Received PPS change: FI", tmp_fi, "DI", tmp_di, "clock_skip", tmp_clock_skip) self.clock_skip = int(tmp_clock_skip * self.detected_clock_skip / 372) self.fi = tmp_fi self.di = tmp_di self.log("PPS Success new settings (calculated): FI", self.fi, "DI", self.di, "clock_skip", self.clock_skip) else: # Native CLK mode: clock_skip counts real CLK edges per # data bit. After an ACCEPTED PPS both sides switch the # UART to F/D cycles per bit (e.g. 512/16 = 32 edges), # so the decoder must follow the switch. Live capture # proof: ATR+PPS decode cleanly at 372, everything after # aliases into garbage if clock_skip stays at 372. self.fi = self.clock_rate.get(int(pps1 >> 4), 372) self.di = self.baud_rate.get(int(pps1 & 0x0F), 1) self.clock_skip = max(int(self.fi // self.di), 1) self.log("PPS accepted: FI", self.fi, "DI", self.di, "clock_skip", self.clock_skip) # PPS changed the bit rate. How we follow it depends on the reader: # # - Native CLK-sync reader (_use_clk_sync): clock_skip is the ONLY # timing input -- the CLK period is unchanged by PPS, so the new # ETU is exactly clock_skip * samples_per_clock. No DATA # re-measurement is needed (and none should happen: consuming a # burst here would silently drop the first post-PPS command). # Just refresh bit_samples for wait_data_falling / edge fallback. # # - Edge-list reader (non-native modes): there is no usable CLK # reference, so re-measure the ETU from the live DATA line on # the next decode cycle; that burst is emitted flagged, not # silently consumed. # PPS changed the bit rate. How we follow it depends on the mode: # # - Native CLK mode (real CLK channel): after an ACCEPTED PPS the # card and terminal switch to FI/DI CLK cycles per bit, and the # CLK period itself is unchanged. So clock_skip = FI/DI is # exact -- no DATA re-measurement is needed (and none should # happen: consuming a burst here would silently drop the first # post-PPS command). Just refresh bit_samples for # wait_data_falling / the edge-list fallback. # # - sample_as_clock / detect modes: there is no usable CLK # reference, so re-measure the ETU from the live DATA line on # the next decode cycle; that burst is emitted flagged, not # silently consumed. if not (self.sample_as_clock or self.detect_clock): self.bit_samples = self._compute_bit_samples() self._pps_speed_changed = False # Re-hunt for the next start bit after the PPS response. self._start_fall = None else: self._pps_speed_changed = True self.bit_samples = None self._start_fall = None # clock_skip now reflects the negotiated FI/DI, so it is known to # be correct regardless of which reader is active. self._clock_skip_confident = True else: self.log("INVALID PPS. Request & Response not matching.", hex(r_lrc)) self.put(ss, self.samplenum, self.out_ann, [0, ["INVALID PPS. Request & Response not matching"]]) self.put(ss, self.samplenum, self.out_ann, [3, ["PPS", "PPS DI={di} FI={fi} clock_skip={clock_skip}".format(di=self.di,fi=self.fi,clock_skip=self.clock_skip)]]) self.emit_packet(GSMTAP_SIM_PPS, bytes(pps_req + pps_rsp), ss, self.samplenum) def decode_step(self): '''Run one state-machine step. Returns False when decoding ends.''' # End-of-capture detection: libsigrokdecode stops advancing the sample # counter once the feed is exhausted, so a stuck sample number means # we have consumed everything (and any pending wait is non-blocking). if self.samplenum == self._last_sn or self._eof: return False self._last_sn = self.samplenum # Prime the initial RST/VCC levels from the actual first sample (no # event emission, no advancement). The reset() defaults are both 0, # which is indistinguishable from a card that is genuinely in reset; # a capture started during power-on (RST already LOW, the assertion # edge in the past) would otherwise miss the assertion and try to # measure the ETU while the card is inactive. With the true initial # levels, RST-low-at-start means "expect a cold-boot ATR on deassert" # and the DATA state waits for that deassert before measuring. # Re-anchor the hysteresis candidates relative to the primed level so # the first confirmed transition is measured from the true state. if not self._primed: self._primed = True pins = self.wait({'skip': 0}) if pins is not None: if self.track_rst: self.last_rst = pins[self.RST_IDX] if self.track_vcc: self.last_vcc = pins[self.VCC_IDX] self.rst_cand = None self.rst_cand_sn = 0 self.vcc_cand = None self.vcc_cand_sn = 0 if (self.track_rst and self.last_rst == 0 and self.state == 'DATA' and self.bit_samples is None): # Card is sensed in reset at capture start (RST primed LOW; # the assertion edge is in the past -- e.g. the capture began # during power-on). The card is inactive, so measuring the # ETU now would lock a bogus period from pre-ATR power-up # transitions on DATA. Expect the fresh ATR instead: re-arm # the hunt at the spec-default rate and let FIND START wait # for the card to power on (read_first_byte/wait_data_falling # service RST/VCC events while DATA is idle). This is the # "a mid-session capture may still begin with a reset" case. self.state = 'FIND START' self._atr_hunt_count = 0 self._pps_attempted = False self.bit_samples = None self._etu_confirm_count = 0 self._etu_fail_count = 0 self.clock_skip = 372 self._inverse_convention = False self._clock_skip_confident = \ not (self.sample_as_clock or self.detect_clock) self.log("RST low at start: expecting cold-boot ATR") # State machine. if self.state == 'FIND START': # Find the first frame (sample-based, gating-immune) and hand it # to the ATR hunter. RST/VCC edges are still serviced inside # wait_data_falling via track_rst/track_vcc. byte = self.read_first_byte() self.peeked_byte = byte self.peeked_samplenum = self.ss self.handle_atr(self.wait({'skip': 0})) elif self.state == 'DATA': packet = []; # Mid-session capture with an explicit protocol: there is no ATR to # derive the bit period from, so recover it once from the live line # and then fall through to the normal T=0/T=1 framing state machine # below, which reassembles each full command+response exchange into # a single GSMTAP APDU. (Emitting each idle-split burst verbatim # produced fragments, because the card stops CLK during the # turn-around and every exchange is split across idle gaps.) # # Also re-measure after an accepted PPS changes the bit rate: # the old ETU no longer applies and the new one must be measured # from the live DATA line. if self.bit_samples is None or self._pps_speed_changed: self._pps_speed_changed = False # For the CLK-sync reader we also need clock_skip, which is # unknown on a mid-session capture. Measure the CLK period # (samples per CLK cycle) and the ETU (samples per bit), then # derive clock_skip = ETU / CLK-period so the CLK-sync reader # runs at the true rate. If the CLK period cannot be measured # (fully gated CLK), clock_skip stays unconfident and the # edge-list reader (bit_samples-driven) is used instead. if self._samples_per_clock is None: self._measure_clock_period() self._measure_etu() if self._eof: return False self._derive_clock_skip_from_etu() return True # ETU now known: run the standard framing below. # After a desync, re-align on an inter-frame idle gap so the # next exchange is read from a real command boundary. if (self._resync): self._resync = False self.resync_idle() # Opportunistic RST/VCC level check at packet boundary. # Note: edges occurring entirely inside a packet transfer # are only detected here as a net level change. if (self.track_rst or self.track_vcc): self.line_events(self.wait({'skip': 0})) if not self.hasT0 and not self.hasT1: bs = self.bit_samples or self._compute_bit_samples() self.wait({'skip': int(bs)}) return True firstByte = self.peek_byte(); if firstByte is None: return True if (firstByte == 0xFF): # PPS Request # Only one PPS attempt is made after an ATR/warm reset. # Idle 0xFF bytes after a failed/no-PPS scenario would # otherwise trigger PPS detection in an infinite loop. if not self._pps_attempted: self.handle_pps(); else: # Idle 0xFF noise after the PPS window has closed. # Consume it and re-sync on the next real start bit. self.read_byte() self._resync = True return True elif (firstByte == 0x3b) and not self._atr_parsed: # Probably ATR (only if no ATR yet) self.handle_atr(self.wait({'skip': 0})) return True elif bin(firstByte).count('1') >= 6: # Idle-line noise: byte has 6+ bits set (all-1s with # 0-2 bit errors). Re-sync instead of mis-parsing as # a T=0 header (which would pack idle noise into # garbage APDUs). self._note_discard(firstByte, 'idle-line noise (6+ bits set)', idle=True) self._consume_first() self._resync = True return True elif not plausible_cla(firstByte): # Non-standard CLA: byte is not a recognized ISO 7816 # class. Likely mis-framed data — attempt to resync. self._note_discard(firstByte, 'suspicious CLA (not ISO 7816-4)') self._consume_first() self.log("Suspicious CLA 0x{:02x}, attempting resync".format(firstByte)) self._resync = True return True elif firstByte == 0x00 and self.zero_run >= 32: # Stuck-low noise: I/O held low for many consecutive frames. # Don't start a T=0 frame with leading zeros. self._note_discard(firstByte, 'stuck-low noise (zero_run >= 32)', idle=True) self._consume_first() self._resync = True return True es = self.peeked_samplenum; if (self.hasT0): # Idle-line check before reading the command header. # DATA idles HIGH in direct convention (0xFF) or LOW in # inverse convention (0x00). If the first byte is idle, # the card is not transmitting — skip it and let the RST # handler re-arm ATR hunt. idle_byte = 0xFF if not self._inverse_convention else 0x00 first = self.peek_byte() if first is None: return True if first == idle_byte: self._note_discard(first, 'idle byte during header wait', idle=True) self._consume_first() self._resync = True return True bClass = self.read_byte() if bClass is None: return True bIns = self.read_byte() if bIns is None: return True p1 = self.read_byte() if p1 is None: return True p2 = self.read_byte() if p2 is None: return True p3 = self.read_byte() if p3 is None: return True packet.extend((bClass, bIns, p1, p2, p3)) # All-zero header is never a valid T=0 command. if all(b == 0 for b in packet[:5]): self.put(es, self.samplenum, self.out_ann, [Ann.ANN_WARN, ["all-zero T=0 header (stuck low?)"]]) self._resync = True return True # T=0 header validation per ISO 7816-4 §12.1.1. # Rejects noise headers that slip past the CLA filter. # - INS=0x00 is reserved/invalid. # - MSB nibble 0x6/0x9 conflicts with status words. # - LSB must be 0 (all standard interindustry commands). if (bIns == 0x00 or (bIns & 0xF0) == 0x60 or (bIns & 0xF0) == 0x90 or (bIns & 0x01) != 0): # Byte-slip re-frame: the whole 5-byte window is a valid # size but the INS is unusable, so the window is # probably shifted by one (a stray leading byte from the # previous exchange, e.g. a leftover SW2). Drop ONLY the # leading byte and re-frame the remaining four so a real # command that would otherwise be swept into a garbage # header (or swallowed by a bogus P3) is recovered. self._note_discard(bClass, 'invalid INS 0x%02x in header; re-framing' % bIns, idle=False) self.peeked_samplenum = es self.peeked_byte = None self._replay.appendleft(p3) self._replay.appendleft(p2) self._replay.appendleft(p1) self._replay.appendleft(bIns) return True # T=0 procedure-byte transport (ISO 7816-3 §10.3.3/10.3.4). # Reassemble the exchange to ISO 7816-4 form (C-APDU = header # + command data, R-APDU = response data + SW1 SW2) into one # event. Procedure bytes (ACK == INS, NULL 0x60, 9E/9F) are # framing and omitted. got_sw = False if self._edge_read: # Mid-session (gated-CLK) path. # # T=0 case 3/4 commands carry Lc (= P3) command-data bytes # on the wire IMMEDIATELY after the 5-byte header, BEFORE # the card's first procedure byte. The original loop (which # treats the first post-header byte as a procedure byte) # therefore mis-frames them. We distinguish the cases by # the card's first procedure byte: # * ACK (== INS or ~INS) right after the header -> case 2/4, the # terminal sent no command data and the card is # acknowledging; read P3 response bytes (+SW), or, for # Le == 0 (GET RESPONSE), read until the SW. # * any other byte -> case 3/4, it # is terminal command data; read P3 command-data bytes, # then the card's procedure byte (ACK/SW/9x). pb0 = self.read_byte() if pb0 is None: return True # Skip NULL (0x60) procedure bytes and turnaround idle # bytes (0xFF): the card may send NULLs while processing # before the ACK, or the bus may idle at xFF during # turnaround on the bidirectional DATA line. Without # this the NULLs / idle bytes are mis-identified as # case-3/4 command data. for _ in range(MAX_TPDU_LEN): if pb0 != 0x60 and pb0 != 0xFF: break self._note_discard(pb0, 'idle/NULL byte before procedure byte', idle=True) pb0 = self.read_byte() if pb0 is None: return True if (pb0 == bIns or pb0 == (bIns ^ 0xFF)): # ACK (full match or ~INS): case 2/4 (response data # follows). Read exactly P3 response bytes, then # scan for SW1 SW2, skipping any turnaround idle # bytes (e.g. 0xFF) between data and SW. for _ in range(p3): b = self.read_byte() if b is None: return True packet.append(b) sw1 = None for _ in range(MAX_TPDU_LEN): sw1 = self.read_byte() if sw1 is None: return True if ((sw1 & 0xF0) == 0x60 or (sw1 & 0xF0) == 0x90): break self._note_discard(sw1, 'non-SW byte before status word ' '(P3-bounded response)', idle=(sw1 in (0x00, 0xFF))) if sw1 is None: return True sw2 = self.read_byte() if sw2 is None: return True packet.append(sw1) # SW1 packet.append(sw2) # SW2 got_sw = True else: # Case 2/3/4: determine data direction from P3 and # the first post-header byte. packet.append(pb0) if (p3 == 0): # P3=0 (Le=0): no command data. Card sends # response data + SW, or SW directly (no ACK). if ((pb0 & 0xF0) == 0x60 or (pb0 & 0xF0) == 0x90): # pb0 is SW1: card sent SW directly. sw2 = self.read_byte() if sw2 is None: return True packet.append(sw2) got_sw = True else: # pb0 is response data: read until SW, # skipping turnaround idle bytes. for _ in range(MAX_TPDU_LEN): b = self.read_byte() if b is None: return True if ((b & 0xF0) == 0x60 or (b & 0xF0) == 0x90): sw2 = self.read_byte() if sw2 is None: return True packet.append(b) packet.append(sw2) got_sw = True break self._note_discard(b, 'non-SW byte in P3=0 ' 'response', idle=(b in (0x00, 0xFF))) else: # Case 3/4: P3 command-data bytes, then # procedure byte. for _ in range(max(p3 - 1, 0)): b = self.read_byte() if b is None: return True packet.append(b) for _proc in range(32): pb = self.read_byte() if pb is None: return True if (pb == bIns or pb == (bIns ^ 0xFF)): # ACK (full match / ~INS): case 4 -> read response until SW, # skipping turnaround idle bytes. for _ in range(MAX_TPDU_LEN): b = self.read_byte() if b is None: return True if ((b & 0xF0) == 0x60 or (b & 0xF0) == 0x90): sw2 = self.read_byte() if sw2 is None: return True packet.append(b) packet.append(sw2) got_sw = True break self._note_discard(b, 'non-SW byte in case-4 ' 'response', idle=(b in (0x00, 0xFF))) break elif (pb == (bIns ^ 0x01) or pb == ((bIns ^ 0xFF) ^ 0x01)): # Single-byte match (INS^0x01 / ~INS^0x01): # read 1 byte, continue handshake. b = self.read_byte() if b is None: return True packet.append(b) continue elif (pb == 0x60 or pb == 0xFF): # NULL (0x60): card busy. Turnaround # idle (0xFF): bidirectional bus turnaround # between terminal command data and the # card's procedure byte. Both: consume and # wait for next procedure byte. continue elif (pb == 0x9e or pb == 0x9f): # Extended-length procedure byte: L # response data bytes follow, then SW. L = self.read_byte() if L is None: return True for _ in range(L): b = self.read_byte() if b is None: return True packet.append(b) continue elif ((pb & 0xF0) == 0x60 or (pb & 0xF0) == 0x90): # SW1: case 3, no response data. sw2 = self.read_byte() if sw2 is None: return True packet.append(pb) packet.append(sw2) got_sw = True break else: # Unexpected byte: scan for SW, # skip non-SW bytes (e.g. 0xFF # turnaround artifacts). for _ in range(MAX_TPDU_LEN): if ((pb & 0xF0) == 0x60 or (pb & 0xF0) == 0x90): sw2 = self.read_byte() if sw2 is None: return True packet.append(pb) packet.append(sw2) got_sw = True break self._note_discard(pb, 'non-SW byte during ' 'unexpected-byte scan', idle=(pb in (0x00, 0xFF))) pb = self.read_byte() if pb is None: return True break else: # ATR-based captures (test_8 / test_7): keep the original # procedure loop unchanged. data_read = False for _proc in range(8): pb = self.read_byte() if pb is None: break if (pb == bIns or pb == (bIns ^ 0xFF)) and not data_read: # ACK (full match / ~INS): the P3 bytes follow on # the wire (Lc command data for case 3, Le response # data for case 2). data_read = True for _ in range(p3): b = self.read_byte() if b is None: break packet.append(b) elif (pb == (bIns ^ 0x01) or pb == ((bIns ^ 0xFF) ^ 0x01)): # Single-byte match (INS^0x01 / ~INS^0x01): # read 1 byte, continue handshake. b = self.read_byte() if b is None: break packet.append(b) elif (pb == 0x60): # NULL: card busy; wait for the next procedure byte. pass elif (pb == 0x9e or pb == 0x9f): # T=0 extended-length procedure byte: the next byte # L is the number of response data bytes the card # now sends, followed by the status word. L = self.read_byte() if L is None: break for _ in range(L): b = self.read_byte() if b is None: break packet.append(b) elif ((pb & 0xF0) == 0x60 or (pb & 0xF0) == 0x90): # SW1: append SW2; the exchange is complete. sw2 = self.read_byte() if sw2 is None: break packet.append(pb) packet.append(sw2) got_sw = True break else: # Mis-framed procedure byte: stop reading. The # partial packet is flagged desynced downstream and # the decoder re-aligns on the next idle gap. self.put(es, self.samplenum, self.out_ann, [0, ["INVALID Procedure Byte"]]) self.log("INVALID Procedure Byte", hex(pb)) self._resync = True break self.put(es, self.samplenum, self.out_ann, [4, ["T=0"]]) if (got_sw and len(packet) >= 7): self.put(es, self.samplenum, self.out_ann, [9, ["APDU", "APDU cls={cls:02x} ins={ins:02x}".format(cls=bClass,ins=bIns), "APDU cls={cls:02x} ins={ins:02x} p1={p1:02x} p2={p2:02x} p3={p3:02x} len={len} status={sw1:02x}{sw2:02x}".format(cls=bClass,ins=bIns,p1=p1,p2=p2,p3=p3,len=p3,sw1=packet[-2],sw2=packet[-1])]]) elif (self.hasT1): isIBlock,packet = self.t1_parse_block(es) if (isIBlock): for _ in range(32): # bounded chain follow-up isIBlock,packet2 = self.t1_parse_block(es) if (isIBlock): packet = packet + packet2 break self.put(es, self.samplenum, self.out_ann, [4, ["T=1", "T=1 (reassembled)"]]) if (len(packet) >= 8): self.put(es, self.samplenum, self.out_ann, [9, ["APDU", "APDU cls={cls:02x} ins={ins:02x}".format(cls=packet[0],ins=packet[1]), "APDU cls={cls:02x} ins={ins:02x} p1={p1:02x} p2={p2:02x} p3={p3:02x} len={len} status={sw1:02x}{sw2:02x}".format(cls=packet[0],ins=packet[1],p1=packet[2],p2=packet[3],p3=packet[4],len=len(packet) - 7,sw1=packet[-2],sw2=packet[-1])]]) if (len(packet) > 0): # All-zero first: an I/O-held-low stretch is idle noise, # not a TPDU -- suppress it entirely (and re-sync later). if (all(b == 0 for b in packet)): self._idle_low += 1 if self._idle_low == 1 or self._idle_low % 256 == 0: self.put(es, self.samplenum, self.out_ann, [Ann.ANN_WARN, ["line idle-low (all-zero reads suppressed)"]]) self._resync = True elif is_desynced(packet): # Keep the bytes but flag them via GSMTAP_FLAG_BAD_FCS # (res byte) so downstream tools can tell desynced # TPDUs apart; sub_type stays 0x00. Then re-align. self.put(es, self.samplenum, self.out_ann, [Ann.ANN_WARN, ["desynced APDU (flagged BAD_FCS)"]]) self.emit_packet(GSMTAP_SIM_APDU, bytes(packet), es, self.samplenum, flags=GSMTAP_FLAG_BAD_FCS) self._resync = True else: emb = embedded_exchanges(packet) if emb: # CONCAT: a byte-drop mis-frame swallowed several # real exchanges into this structurally valid # packet (their headers survive in the data region). # Flag BAD_FCS + log so downstream tools can # split/reject the wrong-APDU group. self.put(es, self.samplenum, self.out_ann, [Ann.ANN_WARN, ["CONCAT APDU (embedded exchange(s))"]]) self.log("CONCAT APDU: %d embedded exchange(s) at %r; packet %s" % (len(emb), [(s, e) for s, e in emb], codecs.encode(bytes(packet), 'hex'))) self.emit_packet(GSMTAP_SIM_APDU, bytes(packet), es, self.samplenum, flags=GSMTAP_FLAG_BAD_FCS) else: self._idle_low = 0 self.emit_packet(GSMTAP_SIM_APDU, bytes(packet), es, self.samplenum) self.log("PACKETEND", codecs.encode(bytes(packet), 'hex')) else: return False return True def decode(self): try: self.write_pcap_header(); failures = 0 while True: # Crash protection: garbage from a noisy live signal must # never kill the whole sigrok-cli process. Log, warn and # resync. SystemError from wait() is the end-of-sample- # stream signal and must propagate for a clean exit; # KeyboardInterrupt (Ctrl+C) likewise passes through. try: if (not self.decode_step()): break failures = 0 except SystemError: raise except Exception as e: traceback.print_exc(file=sys.stderr) self.log("DECODE ERROR:", repr(e), "- resyncing") try: wss = max(self.ss, 0) wes = max(self.es, wss + 1) self.put(wss, wes, self.out_ann, [Ann.ANN_WARN, ["decode error, resynced"]]) except SystemError: raise except Exception: pass self.state = 'DATA' failures += 1 if (failures >= 50): self.log("decode failed 50 times in a row, stopping") break finally: self.finish();