catarrh/sigrok_iso7816_stream
1
0 Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity
Files
v1.9.7
sigrok_iso7816_stream/pd.py
T
catarrh 6e6f465841 v1.9.7: spec fixes, T=1/parity, EOF guards, dead code + rate scaling 
The audit's P2 items:

- ODD_INS: add 0x73 MANAGE SECURE CHANNEL and 0x75 TRANSACT DATA
  (TS 102 221 Table 10.5); the tools/vs_reader.py mirror and the pinned
  test are updated.
- 9E/9F are SW1: ISO 7816-3 10.3.3 Table 11 makes every 9X value SW1.
  Remove the fabricated "extended-length procedure byte" that consumed
  SW2 as a length and lost the status word.
- T=1: EOF-guard every block read, bound the M-bit chain with
  T1_CHAIN_MAX, and drop the caller's duplicate chain loop -- it read the
  next block after every I-block (usually the peer's R-block) and merged
  the following I-block's INF into the packet.  R/S blocks emit nothing;
  the (possibly chained) I-block INF is the APDU payload.
- Parity: a packet containing a parity-error byte is emitted with
  GSMTAP_FLAG_BAD_FCS (the CHKSUM annotation was invisible downstream),
  with a log note when the exchange is otherwise clean.
- EOF guards: handle_atr (TA/TB/TC/TD, historical, TCK), handle_pps
  (request/response fields incl. PPS1-3) and the T=1 block parser return
  cleanly on a truncated capture instead of raising on None.
- Dead code: _edge_read was initialized True and never cleared, so the
  ATR-based T=0 procedure loop (58 lines) and the 1.5-ETU guard branch
  were unreachable.  Remove the flag, both dead paths and the branch;
  dedent the live loop (behavior unchanged).
- Samplerate scaling: _samples_for_us() replaces the hard-coded 16 MHz
  thresholds -- start-bit deglitch (0.125 us), min_high floor (0.25 us)
  and the ETU glitch floor (0.75 us).  At 16 MHz the values are
  unchanged.

Validation: 130/130 unit tests (13 new: 9E/9F SW1, parity BAD_FCS,
T=1 I/R/S blocks + M chain + truncations + one-block-per-step, PPS
truncations, truncated ATR TA, samplerate/ETU-floor vectors); the
10-trace suite is unchanged from v1.9.6 (identical APDU/CHKSUM/CONCAT/
BAD_FCS counts, RESULT: OK); compare_baseline.sh reports IDENTICAL.
VERSION -> 1.9.7.
2026-10-20 19:02:20 +03:00

2958 lines
136 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
##
## Copyright (C) 2020 Sven Soltermann <sven@handyman.ch>
##
## 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 <http://www.gnu.org/licenses/>.
##
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_REQ, GSMTAP_SIM_PPS_RSP,
GSMTAP_SIM_RST_EVENT, GSMTAP_SIM_VCC_EVENT,
GSMTAP_FLAG_BAD_FCS)
VERSION = '1.9.7'
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('<I', secs) # TS seconds
h += struct.pack('<I', usecs) # TS microseconds
# No truncation, so both lengths are the same.
h += struct.pack('<I', len(self)) # Captured len
h += struct.pack('<I', len(self)) # Original len
return h
def __len__(self):
return len(self.h) + len(self.data)
def plausible_sw(sw1):
'''T=0 status word first byte sanity check.
Per ISO 7816-3:2006 §10.3.3 Table 11, an SW1 byte is '6X' or '9X'
EXCEPT '60': that value is the NULL procedure byte (card busy /
waiting-time extension), and ISO 7816-4 enforces '60' as an invalid
SW1. Accepting it mis-frames "NULL + real SW1" as the status word --
the bogus "60 61" APDU ending, where the real SW1 (0x61) is eaten as
SW2. The 6X classes are '61'..'6F' (normal '61XX'; warning '62XX'/
'63XX'; execution error '64XX'-'66XX'; checking error '67XX'-'6FXX'),
plus '9X' for normal processing.'''
return (0x61 <= sw1 <= 0x6F) or (0x90 <= sw1 <= 0x9F)
def plausible_cla(cla):
'''Check if CLA byte is a structurally valid ISO 7816-4 / UICC class.
Accepted:
0x00-0x1F first interindustry: logical channels 0-3 in b2b1,
secure messaging in b4b3, chaining in b5
[ISO 7816-4 §5.1.1 Table 2].
0x40-0x4F further interindustry: logical channels 4-19 (b4..b1
encode the channel number minus four)
[ISO Table 3; TS 102 221 Table 10.4a].
0x50-0x5F the legacy command-chaining variants of 0x40-0x4F
[ISO Table 3 only: TS 102 221 keeps b5 = 0].
0x80/0x84/0xA0/0xB0 SIM/UICC classes (0xB0 has no pinned-spec
basis; kept for backward compatibility).
0xC0-0xCF/0xE0-0xEF the UICC extended classes 'CX'/'EX': logical
channels 4-19 in b4..b1, secure messaging in b6
[TS 102 221 Table 10.4a].
Not accepted, deliberately:
0x20-0x3F reserved for future use by ISO/IEC.
0x60-0x6F extended channels 4-19 with secure messaging in b6
('6X'): the class overlaps printable ASCII ('a'-'o'),
and stray menu-text bytes started bogus frames on the
mis-framed phone traces (0x61 on xiaomi_mi_a1_coldboot).
Extended channels without CLA-level SM ('4X') and the
'CX'/'EX' classes are accepted instead.
0x70-0x7F chaining + secure messaging (not used by UICC: b5=0).
0x81-0x83/0x85-0x8F/0xA1-0xAF the '8X'/'AX' standard-channel
variants (channels 0-3) other than the legacy
0x80/0x84/0xA0: no trace uses them, and the same
precision trade-off applies (the FCP byte 0xA5 on
xiaomi_mi_a1_coldboot).
0x90-0x9F, 0xD0-0xDF, 0xF0-0xFF not SIM/UICC classes.
Anything else means the decoder likely mis-framed the byte stream and
should attempt to resync.'''
if not (cla & 0x80):
return (cla & 0xF0) in (0x00, 0x10, 0x40, 0x50)
return (cla & 0xF0) in (0xC0, 0xE0) or cla in (0x80, 0x84, 0xA0, 0xB0)
# Odd INS codes defined by the pinned specs, i.e. the alternate variants of
# ISO 7816-4 Table 4 pairs (READ BINARY 'B1' carries the offset in a data
# object, ENVELOPE 'C3', UPDATE RECORD 'DD', ...) plus TS 102 221
# AUTHENTICATE '89' (chunked authentication data). Odd values outside this
# set stay rejected: no command uses them, and a mis-framed boundary often
# lands on a data byte -- the text run "I M" (0x49 0x4D) started a bogus
# frame on xiaomi_phone_sample when every odd INS was accepted.
ODD_INS = frozenset((
0x0F, 0x21, 0x73, 0x75, 0x87, 0x89, 0xA1, 0xB1, 0xB3,
0xC3, 0xCB, 0xDB, 0xD1, 0xD7, 0xDD,
))
def plausible_ins(ins):
'''Check if INS byte is a structurally valid T=0 instruction.
Per ISO 7816-4 §5.1.2 / TS 102 221 §10.1.2:
- 0x00 is reserved/invalid.
- MSB nibble 0x6/0x9 collides with status words.
- Odd INS codes are accepted only as the defined alternate command
variants (Table 4 pairs such as READ BINARY 'B1'; TS 102 221
AUTHENTICATE '89').
Used together with plausible_cla() to distinguish a real command header
from noise while hunting for an ATR.'''
if ins == 0x00:
return False
if (ins & 0xF0) == 0x60 or (ins & 0xF0) == 0x90:
return False
if (ins & 0x01) != 0 and ins not in ODD_INS:
return False
return True
def ts_from_bits(bits):
'''Return the TS value (0x3B direct / 0x3F inverse) when the raw ten
character bits match an ISO 7816-3 8.1 initial-character pattern, else
None.
Both TS patterns share moments 1-4 LHHL and 8-10 LLH; moments 5-7 are
LLL (inverse: state L encodes 1, msb first) or HHH (direct: state H
encodes 1, lsb first). The decoded byte alone cannot tell them apart:
the inverse TS reads as 0x03 through a direct lsb-first assembly.'''
if len(bits) < 10:
return None
if bits[0:4] != [0, 1, 1, 0] or bits[7:10] != [0, 0, 1]:
return None
if not any(bits[4:7]):
return 0x3f
if all(bits[4:7]):
return 0x3b
return None
def is_idle_noise_byte(first_byte):
'''Check if a byte looks like an idle-line artifact rather than a
command class byte.
Idle-line noise is an all-1s byte with a few dropped bits, so 6 or
more set bits. Valid CLAs can be bit-dense as well (0x5F, 0xCF,
0xE7, 0xEB, 0xED-0xEF), so plausibility wins over density: such a
byte takes the normal CLA/header path instead of the noise resync.'''
return bin(first_byte).count('1') >= 6 and not plausible_cla(first_byte)
def validate_t0_apdu(packet):
'''Return (valid, reason) for a reassembled T=0 APDU.
Validation is purely structural: we parse the 5-byte header and
verify that the packet length, CLA, INS and status word are all
plausible. No content scanning of command/response data is performed,
so the check has very low false-positive risk. It catches truncation,
missing status words, and obviously bogus header bytes. Concatenated
APDUs (several real exchanges swallowed into one "valid" packet) pass
this structural check; they are caught separately by
embedded_exchanges() and flagged CONCAT/BAD_FCS at emit time.
'''
n = len(packet)
if n < 7:
return (False, 'too short (< 7)')
if n > 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).
# This CLA set is deliberately NARROWER than plausible_cla(): the CONCAT
# scan runs across every payload byte, where the extended-channel classes
# (0x40-0x5F, 0xC0-0xEF) false-hit -- the Alcor secured-packet payload
# "40 70 70 61 15" + an SW-like byte 21 later. The command classes that
# really occur on these wires stay covered: interindustry 0x00-0x1F
# (including logical channels 1-3) and the SIM/UICC classes
# 0x80/0x84/0xA0/0xB0.
EMBEDDED_CLA = frozenset(
list(range(0x00, 0x20)) + [0x80, 0x84, 0xA0, 0xB0])
# 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 one of the
pinned EMBEDDED_CLA classes (deliberately narrower than
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
# --- line-event payload extension (RST/VCC, sub_type 0x10/0x11) ---
# The base payload is [direction, level, flags]; byte 2 used to be reserved
# and always 0. Bit 0 of it now marks an appended CLK frequency (big-endian
# uint32 Hz), so consumers that only read the first two bytes keep working.
LINE_EVENT_FLAG_CLK_HZ = 0x01
def line_event_payload(direction, level, clk_hz=None):
'''Build a RST/VCC line-event payload (GSMTAP sub_type 0x10/0x11).
Base form: [direction, level, flags] (3 bytes). When `clk_hz` is
known, flags bit 0 is set and the frequency follows as a big-endian
uint32 in Hz, giving a 7-byte payload. Old consumers read only the
first two bytes and ignore the rest, so the extension is backward
compatible.'''
flags = 0
extra = b''
if clk_hz:
flags |= LINE_EVENT_FLAG_CLK_HZ
extra = struct.pack('>I', int(clk_hz))
return bytes([direction & 0xFF, level & 0xFF, flags]) + extra
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
T1_CHAIN_MAX = 32 # T=1 I-block M-bit chain depth bound
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 _note_pre_sw(self, value, reason):
'''Log a byte consumed while scanning for the status word.
A T=0 NULL (0x60) is a legitimate wait extension -- cards emit them
in bursts while busy -- so it is reported like idle noise
(rate-limited) with its own reason instead of raising a warning.
Other non-SW bytes keep the call-site reason.'''
if value == 0x60:
self._note_discard(value,
'NULL procedure byte (card busy) before '
'status word', idle=True)
else:
self._note_discard(value, reason, idle=(value in (0x00, 0xFF)))
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
# True only when bit_samples came from a real DATA-edge
# measurement (not a protocol/fallback guess). The hunt-side ETU
# recovery and the clock_skip commit key off this, so a pre-seeded
# default cannot mask a failed measurement.
self._etu_measured = False
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._last_sn = -1
# Bounded allowance for steps that consume only queued bytes
# (_replay / peeked_byte) and therefore do not advance the sample
# counter. Without it the "stuck sample number == EOF" test below
# would mistake a re-frame step for end-of-capture.
self._no_advance = 0
self._eof = False
self._min_pulse_samples = None # None = scale with samplerate (2 @ 16 MHz)
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
# Total parity failures (ATR validation) and "APDU traffic has
# started" (a DATA-state 0x3B is only an ATR before the first APDU;
# afterwards it is data misread at a frame boundary).
self._parity_errors = 0
self._apdu_seen = 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
# Set by handle_pps() on the native path: some readers also change
# the CLK frequency at the PPS speed switch (Alcor reader: 4.0 ->
# 4.8 MHz with FI/DI 512/32), which invalidates the samples-per-clock
# period measured at the ATR. _read_byte_clk() re-measures it inside
# the first post-PPS character, where CLK is guaranteed to be running.
self._clk_remeasure_pending = 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 _samples_for_us(self, us):
'''Return the sample count for a duration in microseconds, scaled
to the capture samplerate.
The glitch/pulse thresholds used to be hard-coded for 16 MHz
(`_min_pulse_samples = 2`, `min_high = 4`, `MIN_ETU = 12`); at
another rate those become far too large or too small. The examples
are 16 MHz, which is also the fallback when the framework has not
reported a rate (unit tests, older sigrok-cli).'''
sr = getattr(self, 'samplerate', None) or 16000000
return max(1, int(round(float(sr) * us / 1000000.0)))
def _clk_hz(self):
'''Return the measured CLK frequency in Hz, or None when unknown.
Uses the 16-edge average from _measure_clock_period(). The value is
a session constant: PPS changes F/D (CLK cycles per bit), not the
physical CLK rate. Native mode only (the measurement is skipped in
sample_as_clock / detect modes).'''
spc = self._samples_per_clock
sr = self.samplerate
if spc and sr:
hz = int(round(float(sr) / float(spc)))
if 1 <= hz <= 100000000:
return hz
return None
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,
line_event_payload(direction, emit_lvl,
self._clk_hz()),
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._etu_measured = False
self._apdu_seen = False
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_measured = False
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):
bits = list(bits)
if self._inverse_convention:
# ISO 7816-3 7.1: direct sends the lsb first, inverse the msb
# first; the readers have already applied the polarity.
byte = 0
for bit in bits:
byte = (byte << 1) | bit
return byte
byte = 0
for bit in 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 err:
self._parity_errors += 1
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, EOFError):
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, EOFError):
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) from real DATA edges and
_measure_clock_period has recovered _samples_per_clock (samples per
CLK cycle), clock_skip = bit_samples / _samples_per_clock. Only a
real DATA measurement is trusted (_etu_measured); a failed or
rejected measurement never commits a rate. A later parity failure
invalidates it (_confirm_etu).'''
if (self._etu_measured and 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 _etu_candidate(self, pairs):
'''Return (etu, ok) from (spacing, level) edge pairs.
Filters glitches, drops a leading burst-boundary outlier, and
rejects a burst whose minimum recurrence is the inter-character
guard rather than a bit period: all-zero data produces long LOW
runs and a short HIGH guard, so the smallest recurring spacing is
the guard (2x or more the true ETU). A burst with bit transitions
has a smaller spacing somewhere, so the guard pattern is
distinguishable when every non-minimum spacing is at least 2x the
minimum.'''
# ~0.75 us: 12 samples at 16 MHz. A real ETU is far longer, but a
# glitch can be shorter; scale with the capture rate.
MIN_ETU = self._samples_for_us(0.75)
valid = [p for p in pairs if p[0] >= MIN_ETU]
if len(valid) >= 2:
med = sorted(p[0] for p in 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][0] > med * 10:
valid.pop(0)
if len(valid) < 2:
# A single spacing cannot prove a recurring bit period (a lone
# 0x00 character yields one 10-ETU run -> 10x the true ETU).
return (None, False)
spacings = [p[0] for p in valid]
etu = self._robust_min(spacings)
guard_dominated = (
all(lvl == 1 for sp, lvl in valid if sp == etu)
and all(sp >= 2 * etu for sp, lvl in valid if sp != etu))
if guard_dominated or not (4 <= etu <= 100000):
return (etu, False)
return (etu, True)
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:
pairs = [(edges[i + 1][0] - edges[i][0], edges[i][1])
for i in range(len(edges) - 1)]
etu, ok = self._etu_candidate(pairs)
if ok:
self.bit_samples = etu
self._etu_measured = True
self._etu_confirm_count = 0
self._etu_fail_count = 0
else:
self.log("ETU measurement rejected", etu,
"spacings", [p[0] for p in pairs][:8])
# The live line is gated/stopped CLK; use the edge-list bit reader
# (immune to the drift of fixed sample-skip timing) for every byte.
# The CLK-sync reader is used once _derive_clock_skip_from_etu()
# has a real measurement; a parity failure invalidates it.
if self._etu_measured:
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.
# A real DATA measurement is required: bit_samples may be a
# pre-seeded default from handle_atr, which must not mask the need
# to recover the live rate.
if (not self._etu_measured
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)
if self._etu_measured:
# The CLK-sync reader may trust clock_skip now.
self._clock_skip_confident = True
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_measured = False
# Never keep sampling at a rejected rate: drop back to the
# edge reader until a fresh measurement is confirmed.
self._clock_skip_confident = False
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). The n-th edge is found by projecting
n periods ahead of the edge we are sitting on and landing on the
nearest CLK edge: skip to (target - half a period), then take one
edge. The projection keeps the edge count exact for ANY clock
period -- "skip (n - 2) cycles, then take 2 edges" loses ~1 edge per
call when the period is a fraction of a sample (e.g. 3.333 samples
at 4.8 MHz / 16 MHz), and the deficit accumulates across the byte
(the parity sample then reads the last data bit). Without a period
estimate, fall back to waiting each edge exactly.'''
if n <= 0:
return True
spc = self._samples_per_clock
if spc:
target = self.samplenum + n * spc
skip = int(target - 0.5 * spc - self.samplenum)
if skip > 0:
self.wait({'skip': skip})
else:
for _ in range(n - 1):
pins = self.wait({self.CLK_IDX: 'r'})
if pins is None:
self._eof = True
return False
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
if self._clk_remeasure_pending:
# A PPS may also change the CLK frequency (see handle_pps).
# CLK runs for the whole frame here; the measurement consumes
# exactly 16 rising edges, accounted as the first edges of the
# frame so the bit loop stays edge-exact.
self._clk_remeasure_pending = False
self._measure_clock_period()
if not self._eof:
prev_edges = 16
self.bit_samples = self._compute_bit_samples()
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
# A TS's moment 10 is fixed by the ISO 7816-3 8.1 pattern (the
# inverse TS is odd over moments 2-10); it is not a parity-checked
# data character, so it must neither warn nor feed the ETU check.
is_ts = ts_from_bits(self.bits) is not None
parity_ok = is_ts or (self.bits.count(1) % 2 == 0)
self.signal_quality(0 if parity_ok else 1)
if not parity_ok:
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 and self.bits:
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, EOFError):
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
# A TS is not a parity-checked data character (see _read_byte_edges).
is_ts = ts_from_bits(self.bits) is not None
parity_ok = is_ts or (self.bits.count(1) % 2 == 0)
self.signal_quality(0 if parity_ok else 1)
if not parity_ok:
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, EOFError):
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 >= half an ETU: 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).'''
min_high = self._samples_for_us(0.25)
if self.bit_samples:
min_high = max(int(0.5 * self.bit_samples), min_high)
while True:
# Wait for DATA idle, servicing RST/VCC.
# ISO 7816-3 7.1: the line idles HIGH and the start bit is
# always LOW (falling edge) in BOTH conventions; only the
# data-bit polarity and order differ.
idle_cond = {self.DATA_IDX: 'h'}
start_edge = 'f'
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, track=False):
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
pulse = self._min_pulse_samples
if pulse is None:
pulse = self._samples_for_us(0.125) # 2 samples @ 16 MHz
if pulse > 0:
pins = self.wait({'skip': pulse})
if pins is not None:
bit = pins[self.DATA_IDX]
idle_val = 1 # line HIGH is idle in both conventions
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, track=True):
'''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.
`track=False` is used for idle-LEVEL waits: a level condition that is
already satisfied returns immediately at the current sample without
advancing, which is normal on an idle line and NOT end-of-capture.
Only a ``None`` pin set (true end-of-stream) is treated as EOF there;
the stall window is reset so a following edge wait starts fresh.'''
if pins is None:
return True
if not track:
self._prev_wait_sn = None
self._stall = 0
return False
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;
# The next start bit is already queued by _read_byte_edges, so
# re-hunting would wrongly reject the short inter-byte gap.
if self._start_fall is 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 self._start_fall is 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
is_ts = ts_from_bits(self.bits) is not None
bad_parity = (not is_ts) and (self.bits.count(1) % 2 != 0)
self.signal_quality(1 if bad_parity else 0)
if bad_parity:
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, EOFError):
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 return to its
idle level so the next peek lands on a genuine start bit of a fresh
exchange instead of a mid-burst byte. The wait is a level condition
(plus optional RST/VCC edges), so it advances the sample counter
instead of spinning on the current sample; a bounded loop guards
against a pathologically busy or stuck line.'''
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
# ISO 7816-3 7.1: the line idles HIGH in both conventions.
idle_level = 1
cond = [{self.DATA_IDX: 'h' if idle_level else 'l'}]
if self.track_rst:
cond.append({self.RST_IDX: 'e'})
if self.track_vcc:
cond.append({self.VCC_IDX: 'e'})
for _ in range(loops):
pins = self.wait(cond)
if pins is None:
# End of capture / session stopped mid-resync: stop cleanly
# instead of raising TypeError on the None pin state.
self._eof = True
return
if self._stall_check(pins, track=False):
self._eof = True
return
if self.line_events(pins):
continue
return
# Could not find an idle level -- resume normal decode anyway.
def handle_atr(self, pins):
atr_start = self.samplenum;
self.log("START ATR:", self.samplenum);
self.state = 'ATR'
bad_parity_before = self._parity_errors
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
# ISO 7816-3 8.1: an inverse-convention TS (moments 5-7 = LLL) is
# decoded as 0x03 by the direct reader; take the convention from
# the raw pattern so both TS forms are recognised.
raw_ts = ts_from_bits(self.bits)
if raw_ts is not None:
byte = raw_ts
# 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()
# nxt is None at end-of-capture: keep hunting (the read
# below returns None and the hunt ends cleanly) instead of
# calling plausible_ins(None), which raises TypeError.
if nxt is not None and 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
raw_ts = ts_from_bits(self.bits)
if raw_ts is not None:
byte = raw_ts
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_measured = False
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()
if byte is None:
return
tA.append(byte)
self.ATR.append(byte)
if (firstT0 & 0b00100000):
byte = self.read_byte()
if byte is None:
return
tB.append(byte)
self.ATR.append(byte)
if (firstT0 & 0b01000000):
byte = self.read_byte()
if byte is None:
return
tC.append(byte)
self.ATR.append(byte)
if (firstT0 & 0b10000000):
byte = self.read_byte()
if byte is None:
return
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()
if byte is None:
return
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()
if byte is None:
return
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))
if self._parity_errors != bad_parity_before:
# A real ATR is parity-clean; a data run misread at a frame
# boundary is not. Ignore the ATR instead of committing 372.
self.log("ATR rejected: parity error(s) in",
codecs.encode(bytes(self.ATR), 'hex'))
self.state = 'DATA'
self._resync = True
return
if len(self.ATR) < 2 or (len(self.ATR) < 3
and (self.ATR[1] & 0xF0) == 0):
# A bare TS+T0 ATR is legal ISO but not credible for a SIM/UICC:
# a 0x3B data byte at a frame boundary would otherwise fabricate
# an ATR and force clock_skip=372 for the rest of the capture.
self.log("implausible bare ATR ignored:",
codecs.encode(bytes(self.ATR), 'hex'))
self.state = 'DATA'
self._resync = True
return
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)
self._etu_measured = True
# 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, depth=0):
'''Parse one T=1 block (NAD, PCB, LEN, INF, LRC) and return
(is_i_block, inf_bytes).
An I-block with the M bit set is followed by the chain of further
I-blocks in the same direction; their INF fields are concatenated
(the T=1 application APDU may span them). Every read is
EOF-guarded -- a truncated capture returns no half block instead of
raising -- and the chain is bounded by T1_CHAIN_MAX so a garbage
M bit cannot recurse without limit.'''
packet = []
nad = self.read_byte()
if nad is None:
return (False, packet)
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()
if pcb is None:
return (False, packet)
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()
if bLen is None:
return (False, packet)
lrc = lrc ^ bLen
if (bLen > 0):
for _ in range(bLen):
byte = self.read_byte()
if byte is None:
self.log("T=1 block truncated in INF at EOF")
return (False, [])
lrc = lrc ^ byte
if (isIBlock): packet.append(byte)
# CRC to implement (TC1 b0); LRC only for now.
bLrc = self.read_byte()
if bLrc is None:
self.log("T=1 block truncated before the LRC at EOF")
return (False, [])
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 and depth < T1_CHAIN_MAX):
self.log("T=1 Multiblock flag", hex(pcb))
next_is_i, packet2 = self.t1_parse_block(self.samplenum, depth + 1)
if next_is_i:
packet = packet + packet2
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()
if pps is None:
return
pps_req = [pps]
pps0 = self.read_byte()
if pps0 is None:
return
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()
if pps1 is None:
return
pps_req.append(pps1)
lrc = lrc ^ pps1
if (pps0 & 0b00100000):
pps2 = self.read_byte()
if pps2 is None:
return
pps_req.append(pps2)
lrc = lrc ^ pps2
if (pps0 & 0b01000000):
pps3 = self.read_byte()
if pps3 is None:
return
pps_req.append(pps3)
lrc = lrc ^ pps3
pck = self.read_byte()
if pck is None:
return
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))
# End of the request frame: the response starts on the next byte.
# Keep the boundary sample so the two halves can be emitted as the
# separate standard GSMTAP subtypes (PPS_REQ / PPS_RSP).
es_req = self.samplenum
r_lrc = 0
r_pps = self.read_byte()
if r_pps is None:
return
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()
if r_pps0 is None:
return
pps_rsp.append(r_pps0)
if (r_pps0 & 0b00010000):
r_pps1 = self.read_byte()
if r_pps1 is None:
return
pps_rsp.append(r_pps1)
r_lrc = r_lrc ^ r_pps1
if (r_pps0 & 0b00100000):
r_pps2 = self.read_byte()
if r_pps2 is None:
return
pps_rsp.append(r_pps2)
r_lrc = r_lrc ^ r_pps2
if (r_pps0 & 0b01000000):
r_pps3 = self.read_byte()
if r_pps3 is None:
return
pps_rsp.append(r_pps3)
r_lrc = r_lrc ^ r_pps3
r_pck = self.read_byte()
if r_pck is None:
return
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 mode:
#
# - Native CLK mode (real CLK channel): clock_skip = FI/DI is
# protocol-defined, so the new ETU is clock_skip *
# samples_per_clock -- no DATA re-measurement is needed (and
# none should happen here: consuming a burst at this point
# would silently drop the first post-PPS command). But some
# readers also change the CLK FREQUENCY at the speed switch
# (Alcor reader: 4.0 -> 4.8 MHz with FI/DI 512/32), which
# invalidates the samples-per-clock measured at the ATR; the
# first post-PPS character re-measures it (see _read_byte_clk).
#
# - 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
# Some readers also change the CLK frequency at the switch
# (Alcor 4.0 -> 4.8 MHz with FI/DI 512/32); re-measure it in
# the first post-PPS character, where CLK is running.
self._clk_remeasure_pending = True
else:
self._pps_speed_changed = True
self.bit_samples = None
self._etu_measured = False
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)]])
# Standard GSMTAP-SIM framing: request and response are separate
# packets (sub_type 0x02 / 0x03), each with its own sample span.
self.emit_packet(GSMTAP_SIM_PPS_REQ, bytes(pps_req), ss, es_req)
self.emit_packet(GSMTAP_SIM_PPS_RSP, bytes(pps_rsp), es_req, self.samplenum)
def _at_eof(self):
'''True when the sample stream is exhausted.
libsigrokdecode stops advancing the sample counter once the feed is
gone, so a stuck sample number normally means end-of-capture. But a
step may legitimately consume only queued bytes (_replay /
peeked_byte) without touching the stream (e.g. the invalid-INS
one-byte re-frame); allow a bounded number of those so a re-frame is
never mistaken for EOF.'''
if self._eof:
return True
if self.samplenum != self._last_sn:
self._no_advance = 0
return False
if ((self._replay or self.peeked_byte is not None)
and self._no_advance < 64):
self._no_advance += 1
return False
return True
def _t0_read_data_until_sw(self, packet):
'''Append every card byte until SW1 SW2, keeping the data.
Used after an INS ACK ("all remaining data follows", ISO 7816-3
Table 11) and for P3=0 / case-4 response data: from this point on
the card sends data bytes, then the status word. A plausible SW1
ends the response; data bytes are appended verbatim. Returns True
when the status word was read.'''
for _ in range(MAX_TPDU_LEN):
pb = self.read_byte()
if pb is None:
return False
if plausible_sw(pb):
sw2 = self.read_byte()
if sw2 is None:
return False
packet.extend((pb, sw2))
return True
packet.append(pb)
return False
def _t0_read_response_data(self, packet, ins):
'''Append the card's T=0 response data bytes until SW1 SW2.
Used for the per-byte (~INS) ACK form, where every data byte is
followed by another procedure byte. Procedure bytes are consumed
and never stored: NULL 0x60 / turnaround 0xFF are skipped, ~INS
means "the next data byte only", and the deprecated INS^0x01 forms
consume one data byte. An INS ACK ("all remaining data follows")
switches to _t0_read_data_until_sw(). Any other byte is response
data. Returns True when the status word was read.'''
for _ in range(MAX_TPDU_LEN):
pb = self.read_byte()
if pb is None:
return False
if pb in (0x60, 0xFF):
self._note_pre_sw(pb, 'NULL/turnaround before response data')
continue
if pb == ins:
return self._t0_read_data_until_sw(packet)
if pb == (ins ^ 0xFF):
d = self.read_byte()
if d is None:
return False
packet.append(d)
continue
if pb == (ins ^ 0x01) or pb == ((ins ^ 0xFF) ^ 0x01):
d = self.read_byte()
if d is None:
return False
packet.append(d)
continue
if plausible_sw(pb):
sw2 = self.read_byte()
if sw2 is None:
return False
packet.extend((pb, sw2))
return True
packet.append(pb)
return False
def decode_step(self):
'''Run one state-machine step. Returns False when decoding ends.'''
if self._at_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_measured = False
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 = [];
forced_bad = False
# A parity error on any byte of this exchange flags the emitted
# packet BAD_FCS (the annotation/log alone is not visible to
# downstream tooling).
parity_before = self._parity_errors
# 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 in (0x3b, 0x3f)
or ts_from_bits(self.bits) is not None)
and not self._atr_parsed and not self._apdu_seen): # ATR candidate (direct/inverse), only before traffic
self.handle_atr(self.wait({'skip': 0}))
return True
elif is_idle_noise_byte(firstByte):
# Idle-line noise: an all-1s byte with a few dropped bits.
# Re-sync instead of mis-parsing as a T=0 header (which
# would pack idle noise into garbage APDUs). Valid
# bit-dense CLAs (0x5F/0xCF/0xE7/0xEB/0xED-0xEF:
# extended logical channels) are not noise and fall
# through to the CLA/header path below.
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):
# CLA byte is not a recognized class (no ISO 7816-4 /
# TS 102 221 coding). Likely mis-framed data — attempt
# to resync.
self._note_discard(firstByte, 'suspicious CLA (not a recognized class)')
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 §5.1.2: the same
# INS rule as plausible_ins() (0x00 and 0x6x/0x9x are
# invalid; odd codes only as the defined alternate
# variants). A window that fails it is almost always
# shifted by one stray leading byte.
if not plausible_ins(bIns):
# 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
# 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 plausible_sw(pb0):
# Immediate status word: ISO 7816-3 §10.3.3 lets the
# card finish the command at the first procedure
# byte; a case-2 rejection ('6A82', '6CXX', ...)
# carries no ACK and must not be read as data.
sw2 = self.read_byte()
if sw2 is None:
return True
packet.append(pb0)
packet.append(sw2)
got_sw = True
elif pb0 == bIns:
if (p3 == 0):
# P3=0 in an outgoing command is Le=256
# (ISO 7816-3 §10.3.2): the card ACKs and sends
# response data until the status word.
if self._t0_read_data_until_sw(packet):
got_sw = True
else:
# ACK: read the P3 response bytes, then the SW.
# The card may return FEWER bytes than Le
# (6282 end-of-file; Lr <= Le, TS 102 221
# §10.1.6): if the byte after the read is not
# SW1, split the buffer at the first plausible
# SW1 followed by a plausible CLA and replay the
# tail (flagged BAD_FCS).
resp = []
eof = False
for _ in range(p3):
b = self.read_byte()
if b is None:
eof = True
break
resp.append(b)
s1 = None if eof else self.read_byte()
if s1 is None:
eof = True
while s1 in (0x60, 0xFF):
s1 = self.read_byte()
if s1 is None:
eof = True
break
if s1 is not None and plausible_sw(s1):
sw2 = self.read_byte()
if sw2 is None:
return True
packet.extend(resp)
packet.extend((s1, sw2))
got_sw = True
else:
buf = resp + ([s1] if s1 is not None else [])
split = None
for j in range(len(buf) - 1):
if not plausible_sw(buf[j]):
continue
k = j + 2
if k >= len(buf) or plausible_cla(buf[k]):
split = j
break
if split is not None:
packet.extend(buf[:split])
packet.extend(buf[split:split + 2])
for t in reversed(buf[split + 2:]):
self._replay.appendleft(t)
got_sw = True
forced_bad = True
self.log(
"SHORT RESPONSE: split %d data byte(s) "
"before SW1 0x%02x (Le %d)"
% (split, buf[split], p3))
else:
packet.extend(buf)
forced_bad = True
self._resync = True
self.log(
"SHORT RESPONSE: no status word in "
"%d byte(s) read for Le %d"
% (len(buf), p3))
elif pb0 == (bIns ^ 0xFF):
# ~INS ACK: one data byte at a time. Read data
# until the SW, consuming the procedure bytes in
# between (ISO 7816-3 Table 11).
if self._t0_read_response_data(packet, bIns):
got_sw = True
else:
# Case 3/4: the first post-header byte is command
# data (P3 = Lc).
packet.append(pb0)
if (p3 == 0):
# P3=0 and no procedure byte: treat the byte as
# response data (Le=256, permissive) and read
# until the status word.
if self._t0_read_data_until_sw(packet):
got_sw = True
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: case-4 response data follows.
# Append it until the status word (it
# used to be discarded). INS means all
# remaining data; ~INS is per byte.
if pb == bIns:
if self._t0_read_data_until_sw(packet):
got_sw = True
elif self._t0_read_response_data(packet, bIns):
got_sw = True
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 plausible_sw(pb):
# 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 plausible_sw(pb):
sw2 = self.read_byte()
if sw2 is None:
return True
packet.append(pb)
packet.append(sw2)
got_sw = True
break
self._note_pre_sw(pb,
'non-SW byte during '
'unexpected-byte scan')
pb = self.read_byte()
if pb is None:
return 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):
# One block per step (the M-bit chain is followed inside
# t1_parse_block); the old extra chain loop read the next
# block -- usually the other side's R-block -- and merged
# the following I-block into this packet.
isIBlock, packet = self.t1_parse_block(es)
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):
# APDU traffic has started: a later DATA-state 0x3B is
# data, not an ATR.
self._apdu_seen = True
# 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
parity_bad = (self._parity_errors != parity_before)
if parity_bad and not forced_bad:
self.log("APDU flagged BAD_FCS: parity error "
"in the exchange")
self.emit_packet(GSMTAP_SIM_APDU, bytes(packet),
es, self.samplenum,
flags=GSMTAP_FLAG_BAD_FCS
if (forced_bad or parity_bad) else 0)
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. End-of-sample-stream must propagate for a clean
# exit: older libsigrokdecode raises SystemError from
# put()/wait(), newer versions raise EOFError('samples
# exhausted') from wait() -- libsigrokdecode accepts either
# as normal decode() termination. KeyboardInterrupt
# (Ctrl+C) likewise passes through.
try:
if (not self.decode_step()):
break
failures = 0
except (SystemError, EOFError):
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, EOFError):
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();
Reference in new issue View Git Blame Copy Permalink