diff --git a/README.md b/README.md index e9db440..2f77e85 100644 --- a/README.md +++ b/README.md @@ -30,9 +30,12 @@ GSMTAP into this analyser. **APDU decoder** (server-side, rendered by the PWA): -- Full APDU decode — CLA per ISO 7816-4 / ETSI coding (logical channel, - secure messaging, command chaining), INS names from a spec table, - SFI-aware P1/P2 for READ/UPDATE BINARY and record commands +- Full APDU decode — CLA per ISO 7816-4 / ETSI coding with the **logical + channel 0-19** (standard channels from b2-b1, the extended channels 4-19 + from b4-b1 per TS 102 221 §10.3), secure messaging and command chaining, + INS names from a spec table, SFI-aware P1/P2 for READ/UPDATE BINARY and + record commands; MANAGE CHANNEL replies show the channel the card + assigned - Status words — full ISO 7816-4 set plus UICC-specific values: `91XX` (proactive command pending), `62F1/F2/F3`, `63F1/F2` (more data), `9300` (SAT busy), `9850`, `9862–64` @@ -55,7 +58,12 @@ GSMTAP into this analyser. decoded as a stream of **C-APDUs** (SELECT, UPDATE BINARY, …) with full CLA/INS/P1/P2/Lc/Le breakdown — plaintext immediately, or after decryption for ciphered packets -- ATR parse (clock-rate conversion, Fi/Di, T-bitmask) and FCP/FCI TLV +- ATR parse (clock-rate conversion, Fi/Di, T-bitmask, truncated-ATR safe) + and FCP/FCI TLV; the historical bytes are decoded as the UICC's + COMPACT-TLV objects (TS 102 221 §6.3.1) — card service data, the **card + capabilities** (ISO 7816-4 §8.1.1.2.7 Table 88: the assignment methods + and the announced maximum number of logical channels, `8 or more` for the + extended-capable cards) and the trailing status indicator (LCS/SW) **SIMtrace Analyser PWA** ([`frontend/`](frontend/)): @@ -121,7 +129,10 @@ with the local time): `Capture started` on begin, a periodic `Capture stopped` (or `ERROR: capture thread stopped on error …` if the capture thread died unexpectedly). A mis-framed or non-SIM GSMTAP datagram is dropped and counted (`dropped`) rather than silently -killing the capture. The heartbeat interval defaults to 60 s and is +killing the capture; a `tpdu` packet that is not APDU-shaped at all +(shorter than the header, the invalid CLA `FF` or an ISO-reserved class) +is skipped and counted (`skipped`) — a character-level or mis-sampled +stream no longer pollutes the decode. The heartbeat interval defaults to 60 s and is tunable via `--log-interval SECONDS`. The GSMTAP listener also accepts the **sigrok-iso7816-stream** decoder diff --git a/README_RUS.md b/README_RUS.md index ac2326d..3b3308d 100644 --- a/README_RUS.md +++ b/README_RUS.md @@ -31,9 +31,11 @@ SIM/UICC из UDP-потока GSMTAP (или импортирует файлы **Декодер APDU** (на стороне сервера, отображается в PWA): -- Полное декодирование APDU — CLA по ISO 7816-4 / ETSI (логический канал, - secure messaging, chaining команд), имена INS из таблицы спецификаций, - учёт SFI в P1/P2 для READ/UPDATE BINARY и команд с записями +- Полное декодирование APDU — CLA по ISO 7816-4 / ETSI с **логическим + каналом 0-19** (стандартные каналы из b2-b1, расширенные 4-19 из b4-b1 + по TS 102 221 §10.3), secure messaging и chaining команд, имена INS из + таблицы спецификаций, учёт SFI в P1/P2 для READ/UPDATE BINARY и команд + с записями; ответ MANAGE CHANNEL показывает назначенный картой канал - Статусные слова — полный набор ISO 7816-4 плюс специфичные для UICC значения: `91XX` (ожидается proactive-команда), `62F1/F2/F3`, `63F1/F2` (больше данных), `9300` (SAT busy), `9850`, `9862–64` @@ -57,8 +59,13 @@ SIM/UICC из UDP-потока GSMTAP (или импортирует файлы сообщение приложения декодируется как поток **C-APDU** (SELECT, UPDATE BINARY, …) с полной разбивкой CLA/INS/P1/P2/Lc/Le — сразу для открытого текста или после расшифровки для зашифрованных пакетов -- Разбор ATR (пересчёт тактовой частоты, Fi/Di, битовая маска T) - и FCP/FCI TLV +- Разбор ATR (пересчёт тактовой частоты, Fi/Di, битовая маска T, + безопасен для усечённых ATR) и FCP/FCI TLV; исторические байты + разбираются как COMPACT-TLV объекты UICC (TS 102 221 §6.3.1) — card + service data, **card capabilities** (ISO 7816-4 §8.1.1.2.7 таблица 88: + способы назначения и заявленное максимальное число логических каналов, + `8 or more` у карт с расширенными каналами) и завершающий индикатор + состояния (LCS/SW) **PWA SIMtrace Analyser** ([`frontend/`](frontend/)): @@ -122,12 +129,15 @@ simtrace-analyser-server --capture disabled Сервер пишет в stderr информацию о жизненном цикле захвата (каждая строка с локальным временем): `Capture started` при старте, периодический -heartbeat `Capture alive: session=N messages=M bytes=B dropped=D` +heartbeat `Capture alive: session=N messages=M bytes=B dropped=D skipped=S` и `Capture stopped` (либо `ERROR: capture thread stopped on error …`, если поток захвата неожиданно умер). Неверно сформированные или не-SIM датаграммы GSMTAP отбрасываются и считаются (`dropped`), а не молча -убивают захват. Интервал heartbeat по умолчанию 60 с, настраивается -через `--log-interval SECONDS`. +убивают захват; `tpdu`-пакет, вовсе не похожий на APDU (короче заголовка, +недопустимый CLA `FF` или зарезервированный ISO класс `001x`), +пропускается и считается (`skipped`) — посимвольный или миссемплированный +поток больше не засоряет декодирование. Интервал heartbeat по умолчанию +60 с, настраивается через `--log-interval SECONDS`. GSMTAP-листенер также принимает декодер **sigrok-iso7816-stream** (https://github.com/anttro/sigrok_iso7816_stream) — пассивный SIM-сниффер diff --git a/frontend/index.html b/frontend/index.html index 7575fac..58fe7e0 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -1302,6 +1302,7 @@ function responseRows(resp) { if (resp.life_cycle) rows.push(['Life cycle', resp.life_cycle]); if (resp.file_size != null) rows.push(['File Size', resp.file_size + ' B']); if (resp.short_ef_id) rows.push(['Short EF ID', resp.short_ef_id]); + if (resp.channel != null) rows.push(['Assigned channel', resp.channel]); if (resp.type === 'GSM' || resp.type === '3G/EPS/5G') { rows.push(['Auth', resp.type + (resp.status ? ' (' + resp.status + ')' : '')]); @@ -1710,6 +1711,24 @@ function atrPpsRows(decoded) { if (h.life_cycle) v += ' — ' + h.life_cycle; if (h.sw) v += ' (SW ' + h.sw + ')'; rows.push(['Historical', v + ' · ' + h.raw]); + if (h.card_capabilities) { + const caps = h.card_capabilities; + const cv = []; + if (caps.max_channels != null) { + cv.push('logical channels: ' + (typeof caps.max_channels === 'number' + ? 'up to ' + caps.max_channels : caps.max_channels)); + } + if (caps.channel_assignment && caps.channel_assignment.length) { + cv.push('assigned ' + caps.channel_assignment.join(', ')); + } + if (caps.command_chaining) cv.push('command chaining'); + if (caps.extended_lc_le) cv.push('extended Lc/Le'); + rows.push(['Card capabilities', cv.join(' · ') + (cv.length ? ' · ' : '') + caps.raw]); + } + if (h.objects && h.objects.length) { + const names = h.objects.map(o => o.name + (o.value ? ' (' + o.value + ')' : '')); + rows.push(['Historical objects', names.join(', ')]); + } } if (decoded.tck) { rows.push(['TCK', '0x' + decoded.tck + (decoded.tck_valid ? ' (valid)' : ' (mismatch)')]); @@ -2780,6 +2799,10 @@ function importSuccessText(data) { if (data.time_base === 'application-layer') { text += ' · ' + t('exact timestamps (Application Layer)'); } + if (data.skipped) { + text += ' · ' + t('{n} packets skipped (not APDU-shaped)') + .replace('{n}', data.skipped); + } return text; } @@ -3168,7 +3191,7 @@ function onFilterChange() { } // ──────────────────── Versioning ──────────────────── -const PWA_VERSION = '2.5.0'; +const PWA_VERSION = '2.6.0'; const MIN_SERVER_VERSION = '2.3.0'; function parseVersion(v) { @@ -3282,6 +3305,7 @@ const LANG_RU = { 'This looks like a COMPRION export — use \'Import COMPRION trace\'.': 'Похоже на экспорт COMPRION — используйте «Импорт трейсов COMPRION».', 'exact timestamps (Application Layer)': 'точные метки времени (ISO Application Layer)', + '{n} packets skipped (not APDU-shaped)': 'пропущено пакетов (не APDU): {n}', 'New capture': 'Новый захват', 'Tiles': 'Плитки', 'List': 'Список', diff --git a/frontend/sw.js b/frontend/sw.js index 6e54c08..0a0396a 100644 --- a/frontend/sw.js +++ b/frontend/sw.js @@ -1,4 +1,4 @@ -const CACHE = 'simtrace-analyser-v71'; +const CACHE = 'simtrace-analyser-v72'; const PRECACHE = [ 'index.html', diff --git a/frontend/tests/comprion.test.js b/frontend/tests/comprion.test.js index a5676eb..fb70f4c 100644 --- a/frontend/tests/comprion.test.js +++ b/frontend/tests/comprion.test.js @@ -519,6 +519,13 @@ test('stopCaptureIfRunning tolerates "no active capture" (400)', async () => { assert.equal(calls, 2); }); +test('importSuccessText mentions skipped non-APDU packets', () => { + const text = importSuccessText({name: 'x', message_count: 3, skipped: 2}); + assert.match(text, /3 messages/); + assert.match(text, /2 packets skipped \(not APDU-shaped\)/); + assert.doesNotMatch(importSuccessText({name: 'x', message_count: 3}), /skipped/); +}); + test('loadSession stops a capture tolerantly', () => { const body = extractFunc(src, 'loadSession'); assert.match(body, /await stopCaptureIfRunning\(\)/); diff --git a/frontend/tests/decode_rows.test.js b/frontend/tests/decode_rows.test.js new file mode 100644 index 0000000..02860f0 --- /dev/null +++ b/frontend/tests/decode_rows.test.js @@ -0,0 +1,80 @@ +// ATR/historical rows, the card-capabilities summary and the MANAGE CHANNEL +// response row (server-decoded fields rendered by the detail panel). + +const { test } = require('node:test'); +const assert = require('node:assert'); +const { loadSrc, extractFunc } = require('./helpers.js'); + +const src = loadSrc(); +eval( + extractFunc(src, 'ifaceValue') + '\n' + + extractFunc(src, 'atrPpsRows') + '\n' + + extractFunc(src, 'responseRows') +); + +function row(rows, label) { + const found = rows.find(r => r[0] === label); + return found ? found[1] : undefined; +} + +test('atrPpsRows: COMPACT-TLV historical and card capabilities', () => { + const rows = atrPpsRows({ + type: 'atr', + historical: { + raw: '8031E073FE211B66D0022A761300', + category: 'COMPACT-TLV objects', + card_capabilities: { + raw: 'FE211B', tables: 3, + max_channels: 4, + channel_assignment: ['by the card', 'by the interface device'], + }, + objects: [ + { tag: '3', name: 'Card service data', value: 'E0' }, + { tag: '7', name: 'Card capabilities', value: 'FE211B' }, + { tag: '6', name: 'Pre-issuing data', value: 'D0022A761300' }, + ], + }, + }); + assert.match(row(rows, 'Historical'), /COMPACT-TLV objects/); + const caps = row(rows, 'Card capabilities'); + assert.match(caps, /logical channels: up to 4/); + assert.match(caps, /assigned by the card, by the interface device/); + assert.match(caps, /FE211B/); + assert.match(row(rows, 'Historical objects'), /Card service data \(E0\)/); + assert.match(row(rows, 'Historical objects'), /Pre-issuing data/); +}); + +test('atrPpsRows: "8 or more" logical channels', () => { + const rows = atrPpsRows({ + type: 'atr', + historical: { + raw: '8031E073FE2157', + category: 'COMPACT-TLV objects', + card_capabilities: { + raw: 'FE2157', tables: 3, + max_channels: '8 or more', + channel_assignment: ['by the card'], + extended_lc_le: true, + }, + }, + }); + const caps = row(rows, 'Card capabilities'); + assert.match(caps, /logical channels: 8 or more/); + assert.match(caps, /extended Lc\/Le/); +}); + +test('atrPpsRows: a plain status-indicator historical stays short', () => { + const rows = atrPpsRows({ + type: 'atr', + historical: { raw: '808105', category: 'COMPACT-TLV objects', + life_cycle: 'operational state (activated)' }, + }); + assert.match(row(rows, 'Historical'), /operational state \(activated\)/); + assert.equal(row(rows, 'Card capabilities'), undefined); +}); + +test('responseRows: MANAGE CHANNEL assigned channel', () => { + const html = responseRows({ channel: 5 }); + assert.match(html, /Assigned channel/); + assert.match(html, /5<\/td>/); +}); diff --git a/pyproject.toml b/pyproject.toml index eaf75dc..5789ae6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.backends._legacy:_Backend" [project] name = "simtrace-analyser" -version = "2.5.0" +version = "2.6.0" description = "SIMtrace Analyser — APDU trace analyser for GSMTAP/PCAP captures (bundles the simtrace2-pysniff utility)" readme = {file = "README.md", content-type = "text/markdown"} license = {text = "GPL-2.0-or-later"} diff --git a/simtrace_analyser/capture.py b/simtrace_analyser/capture.py index ee217e6..740f0ec 100644 --- a/simtrace_analyser/capture.py +++ b/simtrace_analyser/capture.py @@ -10,6 +10,8 @@ from simtrace2_pysniff.gsmtap import (GsmtapReceiver, GSMTAP_SIM_ATR, GSMTAP_SIM_PPS_REQ, GSMTAP_SIM_PPS_RSP, GSMTAP_SIM_RST_EVENT, GSMTAP_SIM_VCC_EVENT) +from .decode import looks_like_apdu + def _ts(): """Local-time stamp for server-side capture log lines.""" @@ -120,6 +122,7 @@ class CaptureManager: self._latest_msg_id = 0 self._msg_count = 0 self._byte_count = 0 + self._skipped = 0 self._log_interval = log_interval @property @@ -138,6 +141,11 @@ class CaptureManager: def latest_msg_id(self): return self._latest_msg_id + @property + def skipped(self): + """Packets dropped because they were not APDU-shaped.""" + return self._skipped + @property def device_connected(self): """True while the capture device is connected, or None when the @@ -156,6 +164,7 @@ class CaptureManager: self._latest_msg_id = 0 self._msg_count = 0 self._byte_count = 0 + self._skipped = 0 self._backend.start() self._thread = threading.Thread(target=self._capture_loop, daemon=True) @@ -187,7 +196,7 @@ class CaptureManager: return None self._db.close_session(sid) _log(f'Capture stopped: session={sid} messages={self._msg_count} ' - f'bytes={self._byte_count}') + f'bytes={self._byte_count} skipped={self._skipped}') return sid def _capture_loop(self): @@ -195,6 +204,14 @@ class CaptureManager: for msg_type, data, flags in self._backend.iter_messages(): if self._session_id is None: break + if msg_type == 'tpdu' and not looks_like_apdu(data): + # Not APDU-shaped at all: a mis-sampled or character-level + # GSMTAP stream would only pollute the decode. + self._skipped += 1 + if self._skipped <= 1 or self._skipped % 100 == 0: + _log(f'Capture: skipped packet {self._skipped} that is ' + f'not APDU-shaped — check the sniffer stream') + continue try: elapsed = round(time.monotonic() - self._start_time, 3) mid = self._db.insert_message( @@ -226,4 +243,4 @@ class CaptureManager: dropped = getattr(self._backend, 'dropped', 0) _log(f'Capture alive: session={self._session_id} ' f'messages={self._msg_count} bytes={self._byte_count} ' - f'dropped={dropped}') + f'dropped={dropped} skipped={self._skipped}') diff --git a/simtrace_analyser/comprion.py b/simtrace_analyser/comprion.py index 7d1e374..b94794c 100644 --- a/simtrace_analyser/comprion.py +++ b/simtrace_analyser/comprion.py @@ -30,7 +30,7 @@ pairing, FCP decoding and the GET RESPONSE filters all work as usual. import re import time -from .decode import _fidi_dict +from .decode import _fidi_dict, cla_channel # Unit scale for the Delay-derived timeline: the COMPRION export states no # clock, so this only converts the Delay column's etu into seconds (card/line @@ -442,11 +442,16 @@ def _emit_apdu(exchange, messages, skipped, warn_bucket, warnings): if incoming and outgoing: # Case 4 on the wire (T=0): the command carries no Le, the card # answers 61 and the terminal fetches with GET RESPONSE. - # The GET RESPONSE keeps the command's logical channel (CLA low bits). + # The GET RESPONSE keeps the command's logical channel: the terminal + # answers with a plain first-interindustry CLA for channels 0-3 and + # the further-interindustry coding for the extended channels 4-19 + # (real traces drop the class/SM bits, TS 102 221 §10.3). _push(messages, t1, 'tpdu', head + incoming + bytes([0x61, len(outgoing)])) t2 = max(items[-1]['t'], t1) - cmd = bytes([head[0] & 0x03, 0xC0, 0x00, 0x00, len(outgoing)]) + channel = cla_channel(head[0]) + gr_cla = channel if channel < 4 else 0x40 | (channel - 4) + cmd = bytes([gr_cla, 0xC0, 0x00, 0x00, len(outgoing)]) _push(messages, t2, 'tpdu', cmd + outgoing + sw) elif incoming: _push(messages, t1, 'tpdu', head + incoming + sw) diff --git a/simtrace_analyser/database.py b/simtrace_analyser/database.py index a70313a..6433c42 100644 --- a/simtrace_analyser/database.py +++ b/simtrace_analyser/database.py @@ -288,7 +288,7 @@ class Database: selection, DF, SFI map, and the last RST-reported CLK frequency. Returns ``(states, clk_hz)``.""" - from .decode import decode_sniff_msg + from .decode import cla_channel, decode_sniff_msg rows = self._conn.execute( "SELECT data, type FROM messages WHERE session_id=? AND id < ? " "AND type IN ('tpdu','atr','change','gap','rst','vcc') " @@ -307,7 +307,7 @@ class Database: if d and d.get('clk_hz'): clk_hz = d['clk_hz'] continue - channel = data[0] & 0x03 if data and len(data) >= 1 else 0 + channel = cla_channel(data[0]) if data else 0 st = self._channel_state(states, channel) ctx = dict(st['prev']) if st['prev'] else {} ctx['sel'] = st['sel'] @@ -387,7 +387,7 @@ class Database: return sel def _decode_rows(self, rows, initial_states=None, clk_hz=None, on_message=None): - from .decode import atr_rate_fields + from .decode import atr_rate_fields, cla_channel msgs = [] states = dict(initial_states or {}) last_atr_i = None # index of the most recent ATR in *msgs* @@ -395,7 +395,7 @@ class Database: for row in rows: channel = 0 if row[3] == 'tpdu' and row[4] and len(row[4]) >= 5: - channel = row[4][0] & 0x03 + channel = cla_channel(row[4][0]) st = self._channel_state(states, channel) ctx = dict(st['prev']) if st['prev'] else {} ctx['sel'] = st['sel'] diff --git a/simtrace_analyser/decode.py b/simtrace_analyser/decode.py index c56055a..7269324 100644 --- a/simtrace_analyser/decode.py +++ b/simtrace_analyser/decode.py @@ -108,48 +108,86 @@ def decode_sw(raw_sw): # ──────────────────── CLA byte ──────────────────── +# ISO 7816-4 §5.1.1 (Tables 2/3) and TS 102 221 §10.1.1/§10.3: the logical +# channel number lives in the CLA byte in two codings — +# * standard channels 0-3: the two least significant bits (CLA '0X', +# '8X' (UICC class) and 'AX' (SIM class), TS 102 221 Table 10.3); +# * extended channels 4-19: the four least significant bits plus four +# (CLA '01x0 xxxx' further interindustry and '11x0 xxxx' ETSI-defined, +# TS 102 221 Table 10.4a). +_SM_NAMES = {0: 'none', 1: 'proprietary', 2: 'SM header not authenticated', + 3: 'SM header authenticated'} +_CHAIN_NAMES = {0: 'last or only', 1: 'first or continuing'} + + +def cla_channel(cla): + """The logical channel number a CLA byte addresses (0-19, TS 102 221 §10.3).""" + if cla & 0x40: # b7: 01x0 xxxx / 11x0 xxxx codings + return (cla & 0x0F) + 4 # b4-b1 + 4 = extended channels 4-19 + return cla & 0x03 # b2-b1 = standard channels 0-3 + + def decode_cla(cla): - if cla == 0x80: + if cla == 0xff: + interclass = 'invalid (ISO 7816-4)' + elif 0x80 <= cla <= 0x8f: interclass = 'ETSI-defined (UICC/USIM)' - elif cla == 0xa0: + elif 0xa0 <= cla <= 0xaf: interclass = 'ETSI-defined (SIM/GSM)' + elif 0xc0 <= cla <= 0xef: + interclass = 'ETSI-defined (extended logical channels)' + elif 0x40 <= cla <= 0x7f: + interclass = 'inter-industry (further format)' + elif 0x20 <= cla <= 0x3f: + interclass = 'reserved (ISO 7816-4)' elif cla & 0x80: interclass = 'proprietary' - elif (cla >> 6) & 0x03 == 0x00: + else: interclass = 'inter-industry (ISO 7816-4)' - elif (cla >> 6) & 0x03 == 0x01: - interclass = 'inter-industry (further format)' - else: - interclass = 'reserved' - if cla & 0x80: - # proprietary / ETSI-defined: no standard SM/chaining/channel coding - result = { - 'hex': f'{cla:02x}', - 'interclass': interclass, - 'channel': cla & 0x03, - 'secure_messaging': 'none', - 'chain': 'last or only', - } - else: - # ISO 7816-4 §5.1.1 first/further interindustry - chain_bit = (cla >> 5) & 0x01 - chain_names = {0: 'last or only', 1: 'first or continuing'} - sm_names = {0: 'none', 1: 'proprietary', 2: 'SM header not authenticated', - 3: 'SM header authenticated'} - sm_ind = (cla >> 3) & 0x03 - channel = cla & 0x03 - if (cla >> 6) & 0x03 == 0x01: - channel += 4 # further format: b4-b1 = channel + 4 - result = { - 'hex': f'{cla:02x}', - 'interclass': interclass, - 'channel': channel, - 'secure_messaging': sm_names.get(sm_ind, 'none'), - 'chain': chain_names.get(chain_bit, 'last or only'), - } + + result = { + 'hex': f'{cla:02x}', + 'interclass': interclass, + 'channel': cla_channel(cla), + 'secure_messaging': 'n/a', + 'chain': 'n/a', + } + if 0x00 <= cla <= 0x1f: + # First interindustry values: b5 chaining, b4-b3 SM, b2-b1 channel. + result['secure_messaging'] = _SM_NAMES[(cla >> 2) & 0x03] + result['chain'] = _CHAIN_NAMES[(cla >> 4) & 0x01] + elif 0x40 <= cla <= 0x7f: + # Further interindustry values: b6 SM, b5 chaining, b4-b1 channel. + result['secure_messaging'] = ('none' if not cla & 0x20 + else 'SM header not authenticated') + result['chain'] = _CHAIN_NAMES[(cla >> 4) & 0x01] + elif 0x80 <= cla <= 0x8f or 0xa0 <= cla <= 0xaf: + # UICC/SIM classes (TS 102 221 Table 10.3/10.4): b4-b3 SM, b5 = 0. + result['secure_messaging'] = _SM_NAMES[(cla >> 2) & 0x03] + result['chain'] = 'last or only' + elif 0xc0 <= cla <= 0xef: + # Extended-channel classes (TS 102 221 Table 10.4a/10.4b): b6 SM, + # b5 = 0. + result['secure_messaging'] = ('none' if not cla & 0x20 + else 'SM header not authenticated') + result['chain'] = 'last or only' return result +def looks_like_apdu(data): + """Conservative sanity check for a captured TPDU byte string. + + Rejects only shapes that cannot be a command APDU: anything shorter than + the 5-byte header, the invalid CLA 'FF' (ISO 7816-4 §5.1.1) and the + reserved '001x xxxx' class range. Used by the capture/import paths to + drop non-APDU streams (mis-sampled or character-level GSMTAP capturers) + instead of storing them as decodable exchanges. + """ + if not data or len(data) < 5: + return False + return not (data[0] == 0xFF or 0x20 <= data[0] <= 0x3F) + + # ──────────────────── Known FIDs ──────────────────── # The spec-sourced file/DF/SFI tables live in ef_tables.py (see its # header for the sources); _file_name() resolves DF-first. @@ -461,6 +499,7 @@ APDU_SPEC = { }, 0x70: { 'name': 'MANAGE CHANNEL', + 'le': True, 'p1': {0x00: 'Open channel', 0x80: 'Close channel'}, 'p2': {0x00: 'Auto-assign channel', **{n: f'Channel {n}' for n in range(1, 20)}}, 'body': None, @@ -3898,6 +3937,9 @@ def decode_message(raw_data, prev=None): fcp = _decode_fcp(body) if fcp: result['response'] = fcp + elif ins == 0x70: # MANAGE CHANNEL → the channel the card assigned + if p1 == 0x00 and cmd_body_len == 1: + result['response'] = {'channel': body[0]} elif ins == 0x10: # TERMINAL PROFILE → device characteristics tp = _decode_terminal_profile(body) if tp: @@ -4201,26 +4243,149 @@ def atr_rate_fields(clk_hz, decoded): } +# COMPACT-TLV data object names (ISO 7816-4 §8.1.1.2; the UICC always uses +# this form — category indicator '80', TS 102 221 §6.3.1). +_COMPACT_TLV_NAMES = { + 0x1: 'Country or issuer indicator', 0x2: 'Issuer identification', + 0x3: 'Card service data', 0x4: 'Initial access data', + 0x5: "Card issuer's data", 0x6: 'Pre-issuing data', + 0x7: 'Card capabilities', 0x8: 'Status indicator', + 0xf: 'Application identifier', +} +# Card capabilities, first software function table (ISO 7816-4 §8.1.1.2.7 +# Table 86): selection methods. +_CARD_CAP_SELECTION = [ + (0x80, 'full DF name'), (0x40, 'partial DF name'), (0x20, 'by path'), + (0x10, 'by file identifier'), (0x08, 'implicit DF selection'), + (0x04, 'short EF identifier'), (0x02, 'record number'), + (0x01, 'record identifier'), +] +# ...second table (Table 87): b7-b6 the write behaviour. +_CARD_CAP_WRITE = {0x00: 'one-time', 0x20: 'proprietary', 0x40: 'write OR', + 0x60: 'write AND'} + + +def _decode_card_capabilities(value): + """Card capabilities on the compact TLV '7Y': up to three software + function tables (ISO 7816-4 §8.1.1.2.7) — selection methods, data + coding, and the chaining / length / logical-channels table.""" + out = {'raw': value.hex().upper(), 'tables': len(value)} + if len(value) >= 1: + out['selection_methods'] = [name for bit, name in _CARD_CAP_SELECTION + if value[0] & bit] + if len(value) >= 2: + unit = value[1] & 0x0F + out['data_coding'] = { + 'tlv_structure': bool(value[1] & 0x80), + 'write': _CARD_CAP_WRITE.get(value[1] & 0x60, 'reserved'), + 'ff_first_byte': bool(value[1] & 0x10), + 'data_unit_bytes': 2 ** (unit - 1) if unit else None, + } + if len(value) >= 3: + t = value[2] + out['command_chaining'] = bool(t & 0x80) + out['extended_lc_le'] = bool(t & 0x40) + # b5 = assignment by the card, b4 = by the interface device, b3-b1 + # y,z,t: 4y+2z+t+1 channels (all three set = eight or more). + assignment = [] + if t & 0x10: + assignment.append('by the card') + if t & 0x08: + assignment.append('by the interface device') + out['channel_assignment'] = assignment + y, z, v = (t >> 2) & 1, (t >> 1) & 1, t & 1 + out['max_channels'] = ('8 or more' if y and z and v + else 4 * y + 2 * z + v + 1) + return out + + +def _parse_compact_tlvs(data): + """Parse COMPACT-TLV objects: one byte = tag (b8-b5) and length (b4-b1), + then the value (ISO 7816-4 §8.1.1.2). A truncated object is reported, + never invented.""" + objects = [] + i = 0 + while i < len(data): + head = data[i] + tag, length = head >> 4, head & 0x0F + value = data[i + 1:i + 1 + length] + obj = {'tag': f'{tag:X}', 'length': length, + 'name': _COMPACT_TLV_NAMES.get(tag, 'Reserved'), + 'value': value.hex().upper()} + if len(value) < length: + obj['truncated'] = True + elif tag == 0x7: + obj['card_capabilities'] = _decode_card_capabilities(value) + elif tag == 0x3 and length == 1: + obj['card_service'] = { + 'selection': [name for bit, name in + ((0x80, 'full DF name'), + (0x40, 'partial DF name')) if value[0] & bit], + 'ber_tlv_in_dir': bool(value[0] & 0x20), + 'ber_tlv_in_atr': bool(value[0] & 0x10), + 'card_with_mf': not (value[0] & 0x01), + } + objects.append(obj) + i += 1 + length + return objects + + +def _status_indicator(lcs, sw1, sw2): + """The card's status (life cycle + processing status), ISO 7816-4 + §8.1.1.3: '00' is not reported.""" + out = {} + if lcs: + name = _LIFE_CYCLE.get(lcs) + if name: + out['life_cycle'] = name + if sw1 is not None and sw2 is not None and (sw1 or sw2): + out['sw'] = f'{sw1:02X}{sw2:02X}' + return out + + def _decode_historical(hist): - """Decode ATR historical bytes (ISO 7816-4 §8.1.1).""" + """Decode ATR historical bytes (ISO 7816-4 §8.1.1). + + Category '00' carries optional COMPACT-TLVs and a mandatory trailing + three-byte status indicator; '80' (the value a UICC always uses, + TS 102 221 §6.3.1) carries COMPACT-TLVs with an optional trailing '8Y' + status indicator; '10' carries the DIR data reference; the rest are + proprietary or reserved. + """ result = {'raw': hist.hex().upper()} cat = hist[0] - if cat == 0x80 and len(hist) >= 3: - result['category'] = 'status information (life cycle)' - result['sw'] = f'{hist[1]:02X}{hist[2]:02X}' - life_cycle = _LIFE_CYCLE.get(hist[2]) - if life_cycle: - result['life_cycle'] = life_cycle - elif cat == 0x8F: - result['category'] = 'TLV list' - elif 0x00 <= cat <= 0x7F or 0x90 <= cat <= 0x9F: - result['category'] = 'proprietary' - elif 0x81 <= cat <= 0x8E or 0xA0 <= cat <= 0xFE: - result['category'] = 'reserved' - elif cat == 0xFF: - result['category'] = 'default' + body = hist[1:] + if cat == 0x00: + result['category'] = 'COMPACT-TLV objects + status indicator' + if len(body) >= 3: + result.update(_status_indicator(body[-3], body[-2], body[-1])) + body = body[:-3] + result['objects'] = _parse_compact_tlvs(body) + elif cat == 0x80: + result['category'] = 'COMPACT-TLV objects' + objects = _parse_compact_tlvs(body) + result['objects'] = objects + if objects and objects[-1]['tag'] == '8' \ + and not objects[-1].get('truncated'): + last = objects.pop() + value = bytes.fromhex(last['value']) + if last['length'] == 1: + result.update(_status_indicator(value[0], None, None)) + elif last['length'] == 2: + result.update(_status_indicator(None, value[0], value[1])) + elif last['length'] == 3: + result.update(_status_indicator(*value)) + for obj in objects: + if 'card_capabilities' in obj: + result['card_capabilities'] = obj['card_capabilities'] + elif cat == 0x10: + result['category'] = 'DIR data reference' + if body: + result['dir_reference'] = f'{body[0]:02X}' + elif 0x81 <= cat <= 0x8f: + result['category'] = 'reserved (ISO 7816-4)' else: - result['category'] = f'0x{cat:02X}' + result['category'] = 'proprietary' return result @@ -4266,6 +4431,12 @@ def _decode_atr(data, clk_hz=None): cur_t = None # protocol T governing the TA/TB/TC bytes at this level (None = global) while flags and i < len(body): + # The interface bytes indicated by b5-b8 of the previous byte must + # all fit; a truncated ATR (a mis-sampled capture) stops here + # instead of raising. + if i + bin(flags >> 4).count('1') > len(body): + result['truncated'] = True + break if flags & 0x10: # TA b = body[i] i += 1 diff --git a/simtrace_analyser/server.py b/simtrace_analyser/server.py index 776c769..d1a648a 100644 --- a/simtrace_analyser/server.py +++ b/simtrace_analyser/server.py @@ -9,8 +9,9 @@ from datetime import datetime from http.server import HTTPServer, BaseHTTPRequestHandler from urllib.parse import urlparse, parse_qs, quote -from .database import Database from .capture import CaptureManager, GsmtapListener, DirectSniffer, _log +from .database import Database +from .decode import looks_like_apdu from .comprion import (ComprionError, decode_text, looks_like_comprion, parse_comprion) from simtrace2_pysniff.gsmtap import build_gsmtap_packet, GSMTAP_SIM_ATR, GSMTAP_SIM_APDU @@ -367,7 +368,13 @@ class RequestHandler(BaseHTTPRequestHandler): first_ts = packets[0][0] last_ts = packets[-1][0] rows = [] + skipped = 0 for ts, msg_type, payload, flags in packets: + # A captured 'tpdu' must at least be APDU-shaped; mis-sampled or + # character-level streams would only pollute the decode. + if msg_type == 'tpdu' and not looks_like_apdu(payload): + skipped += 1 + continue elapsed = round(ts - first_ts, 6) if ts is not None else 0.0 rows.append((max(0.0, elapsed), msg_type, payload, flags)) self.db.insert_messages(session_id, rows) @@ -382,12 +389,14 @@ class RequestHandler(BaseHTTPRequestHandler): self.db.close_session(session_id, duration) _log(f'Session imported: session={session_id} ' - f'name={name or "(untitled)"} messages={len(packets)} ' - f'bytes={sum(len(p) for _, _, p, _ in packets)}') + f'name={name or "(untitled)"} messages={len(rows)} ' + f'skipped={skipped} ' + f'bytes={sum(len(r[2]) for r in rows)}') self._send_json({ 'session_id': session_id, 'name': name, - 'message_count': len(packets), + 'message_count': len(rows), + 'skipped': skipped, }) def _handle_import_comprion(self, params, raw): diff --git a/simtrace_analyser/version.py b/simtrace_analyser/version.py index 8b1e131..6c5b69c 100644 --- a/simtrace_analyser/version.py +++ b/simtrace_analyser/version.py @@ -1,3 +1,3 @@ """Single source of truth for the SIMtrace Analyser version.""" -__version__ = '2.5.0' +__version__ = '2.6.0' diff --git a/tests/test_capture.py b/tests/test_capture.py index ce8529c..d6428e8 100644 --- a/tests/test_capture.py +++ b/tests/test_capture.py @@ -66,6 +66,20 @@ class TestCaptureManager(unittest.TestCase): self.assertIsNotNone(db.get_session(sid)) self.assertEqual(db.count_messages(sid), 1) + def test_non_apdu_packets_are_skipped(self): + # A character-level or mis-sampled stream is not APDU-shaped and must + # not be stored as decodable exchanges; valid rows pass. + backend = _FakeBackend([ + ('tpdu', bytes.fromhex('ff10957aff10957a'), 0), + ('tpdu', bytes.fromhex('2d6ccfb0ff'), 0), + ('tpdu', bytes.fromhex('00a40004023f00'), 0), + ]) + mgr, db = self._manager(backend) + sid = mgr.start_session() + self.assertEqual(mgr.stop_session(), sid) + self.assertEqual(db.count_messages(sid), 1) + self.assertEqual(mgr.skipped, 2) + def test_direct_sniffer_device_disconnect(self): from simtrace_analyser.capture import DirectSniffer sn = DirectSniffer() diff --git a/tests/test_comprion.py b/tests/test_comprion.py index d5befc0..a1b71e1 100644 --- a/tests/test_comprion.py +++ b/tests/test_comprion.py @@ -252,6 +252,22 @@ class TestMessageEncoding(unittest.TestCase): self.assertIn('01a40004023f00612a', pairs) self.assertIn('01c000002a' + FCP.lower() + '9000', pairs) + def test_get_response_keeps_extended_channel(self): + # Extended channels (4-19) are fetched with the further-interindustry + # CLA: channel 4 (command 0x60 = further + SM) becomes a plain 0x40, + # channel 12 (0x48) stays 0x48 (TS 102 221 §10.3). + extra = apdu(10, '60 A4 00 04', '3F 00', FCP) + msgs = rows_of(build_trace(extra=extra)) + pairs = [(data.hex()) for _t, typ, data in msgs if typ == 'tpdu'] + self.assertIn('60a40004023f00612a', pairs) + self.assertIn('40c000002a' + FCP.lower() + '9000', pairs) + + extra = apdu(10, '48 A4 00 04', '3F 00', FCP) + msgs = rows_of(build_trace(extra=extra)) + pairs = [(data.hex()) for _t, typ, data in msgs if typ == 'tpdu'] + self.assertIn('48a40004023f00612a', pairs) + self.assertIn('48c000002a' + FCP.lower() + '9000', pairs) + def test_continuation_rows_join(self): # A long outgoing block wraps onto rows with an empty description; # the parser must fold them back into the element's value. diff --git a/tests/test_decode.py b/tests/test_decode.py index a583c9c..c7eb48b 100644 --- a/tests/test_decode.py +++ b/tests/test_decode.py @@ -950,11 +950,101 @@ class TestAtrPps(unittest.TestCase): def test_atr_historical_life_cycle(self): from simtrace_analyser.decode import _decode_atr - # T0 = 0x03 (3 historical bytes); historical = 80 00 05 (life cycle: operational) - r = _decode_atr(bytes.fromhex('3B03800005')) + # T0 = 0x03 (3 historical bytes). Category '80' (the value a UICC + # always uses, TS 102 221 §6.3.1) carries COMPACT-TLV objects; the + # trailing '81' object is a status indicator holding the LCS + # (ISO 7816-4 §8.1.1.3). + r = _decode_atr(bytes.fromhex('3B03808105')) self.assertEqual(r['historical_len'], 3) - self.assertEqual(r['historical']['category'], 'status information (life cycle)') - self.assertEqual(r['historical']['life_cycle'], 'operational state (activated)') + self.assertEqual(r['historical']['category'], 'COMPACT-TLV objects') + self.assertEqual(r['historical']['life_cycle'], + 'operational state (activated)') + self.assertEqual(r['historical']['objects'], []) + + def test_atr_historical_status_bytes(self): + from simtrace_analyser.decode import _decode_atr + # '82' = SW1-SW2, '83' = LCS + SW1-SW2 (ISO 7816-4 §8.1.1.3). + r = _decode_atr(bytes.fromhex('3B0480829000')) + self.assertEqual(r['historical']['sw'], '9000') + r = _decode_atr(bytes.fromhex('3B058083059000')) + self.assertEqual(r['historical']['life_cycle'], + 'operational state (activated)') + self.assertEqual(r['historical']['sw'], '9000') + + def test_atr_historical_status_category_00(self): + from simtrace_analyser.decode import _decode_atr + # Category '00': the last three bytes are the status indicator, the + # bytes before it optional COMPACT-TLVs. + r = _decode_atr(bytes.fromhex('3B0400059000')) + self.assertEqual(r['historical']['category'], + 'COMPACT-TLV objects + status indicator') + self.assertEqual(r['historical']['life_cycle'], + 'operational state (activated)') + self.assertEqual(r['historical']['sw'], '9000') + + def test_atr_card_capabilities(self): + from simtrace_analyser.decode import _decode_atr + # A real workspace-card ATR's historical bytes: '31 E0' card service + # data, '73 FE 21 1B' card capabilities (3 tables: selection methods + # FE, data coding 21, 1B = card-assigned channels, 4 maximum) and + # '66 ...' pre-issuing data. TS 102 221 §6.3.1 pins the first two + # objects. + r = _decode_atr(bytes.fromhex('3B0E8031E073FE211B66D0022A761300')) + hist = r['historical'] + self.assertEqual(hist['category'], 'COMPACT-TLV objects') + self.assertEqual([o['tag'] for o in hist['objects']], ['3', '7', '6']) + cs = hist['objects'][0]['card_service'] + self.assertEqual(cs['selection'], ['full DF name', 'partial DF name']) + self.assertTrue(cs['ber_tlv_in_dir']) + self.assertFalse(cs['ber_tlv_in_atr']) + self.assertTrue(cs['card_with_mf']) + caps = hist['card_capabilities'] + self.assertEqual(caps['tables'], 3) + self.assertEqual(caps['max_channels'], 4) + # b5/b4 declare the assignment methods the card supports (here both). + self.assertEqual(caps['channel_assignment'], + ['by the card', 'by the interface device']) + self.assertFalse(caps['command_chaining']) + self.assertFalse(caps['extended_lc_le']) + self.assertIn('full DF name', caps['selection_methods']) + self.assertEqual(caps['data_coding']['write'], 'proprietary') + + def test_atr_card_capabilities_extended(self): + from simtrace_analyser.decode import _decode_atr + # '73 FE 21 57': y=z=t=1 → "eight or more" logical channels + # (ISO 7816-4 Table 88) — the extended-channel capable cards. + r = _decode_atr(bytes.fromhex('3B0E8031E073FE215766D0022A761300')) + caps = r['historical']['card_capabilities'] + self.assertEqual(caps['max_channels'], '8 or more') + self.assertEqual(caps['channel_assignment'], ['by the card']) + self.assertTrue(caps['extended_lc_le']) + + def test_atr_historical_reserved_and_proprietary(self): + from simtrace_analyser.decode import _decode_atr + r = _decode_atr(bytes.fromhex('3B018F')) + self.assertEqual(r['historical']['category'], + 'reserved (ISO 7816-4)') + r = _decode_atr(bytes.fromhex('3B01A0')) + self.assertEqual(r['historical']['category'], 'proprietary') + r = _decode_atr(bytes.fromhex('3B021001')) + self.assertEqual(r['historical']['category'], 'DIR data reference') + self.assertEqual(r['historical']['dir_reference'], '01') + + def test_manage_channel_response(self): + # MANAGE CHANNEL open: the card returns the assigned channel as the + # response data byte (ISO 7816-4 §7.1.5); a close carries none. + from simtrace_analyser.decode import decode_message + r = decode_message(bytes.fromhex('0070000001019110')) + self.assertEqual(r['p1']['name'], 'Open channel') + self.assertEqual(r['response'], {'channel': 1}) + r = decode_message(bytes.fromhex('0070000001029000')) + self.assertEqual(r['response'], {'channel': 2}) + r = decode_message(bytes.fromhex('00708002009000')) + self.assertEqual(r['p1']['name'], 'Close channel') + self.assertNotIn('response', r) + # The extended channels are assigned the same way (1-19). + r = decode_message(bytes.fromhex('0070000001139000')) + self.assertEqual(r['response'], {'channel': 0x13}) def test_atr_t15_is_pps_capability(self): from simtrace_analyser.decode import _decode_atr @@ -1659,6 +1749,49 @@ class TestSelectionTracking(unittest.TestCase): # channel 0 still selects EF_MSISDN, not channel 1's EF_IMPI. self.assertEqual(up['decoded']['file']['ef'], 'EF_MSISDN') + def test_extended_channel_selection_isolated(self): + # CLA 0x44 = further-interindustry coding, extended channel 8 + # (TS 102 221 §10.3: b4-b1 + 4); 0x48 = channel 12. Each channel + # keeps its own selection — before the channel fix the state was + # keyed by CLA & 0x03, so these all clobbered channel 0. + import tempfile + from simtrace_analyser.database import Database + with tempfile.NamedTemporaryFile(suffix='.db') as tmp: + db = Database(tmp.name) + sid = db.create_session('capture') + db.insert_message(sid, 0.0, 'tpdu', + bytes.fromhex('00a40804047fff6f409000')) # ch0: MSISDN + db.insert_message(sid, 0.1, 'tpdu', + bytes.fromhex('44a40804047fff6f029000')) # ch8: OCST + db.insert_message(sid, 0.2, 'tpdu', + bytes.fromhex('48a40804047fff6fb79000')) # ch12: ECC + read = (bytes.fromhex('b2010427') + + bytes.fromhex('8109089cbeb920bdbebcb5c0ffffffffffffffffffffffffff07919752738981f7ffffffffffff') + + bytes.fromhex('9000')) + db.insert_message(sid, 0.3, 'tpdu', bytes([0x44]) + read) + db.insert_message(sid, 0.4, 'tpdu', bytes([0x00]) + read) + db.insert_message(sid, 0.5, 'tpdu', bytes([0x48]) + read) + msgs = db.get_messages(sid) + self.assertEqual(msgs[3]['decoded']['file']['ef'], 'EF_OCST') + self.assertEqual(msgs[4]['decoded']['file']['ef'], 'EF_MSISDN') + self.assertEqual(msgs[5]['decoded']['file']['ef'], 'EF_ECC') + self.assertEqual(msgs[3]['decoded']['cla']['channel'], 8) + self.assertEqual(msgs[5]['decoded']['cla']['channel'], 12) + + def test_manage_channel_opens_use_their_channel(self): + # The MANAGE CHANNEL open/close exchanges themselves run on the + # basic channel; a later extended-channel command must not disturb it. + import tempfile + from simtrace_analyser.database import Database + with tempfile.NamedTemporaryFile(suffix='.db') as tmp: + db = Database(tmp.name) + sid = db.create_session('capture') + db.insert_message(sid, 0.0, 'tpdu', bytes.fromhex('0070000001059000')) + db.insert_message(sid, 0.1, 'tpdu', bytes.fromhex('44a40804047fff6f079000')) + msgs = db.get_messages(sid) + self.assertEqual(msgs[0]['decoded']['response'], {'channel': 5}) + self.assertEqual(msgs[1]['decoded']['cla']['channel'], 8) + class TestSfiResolution(unittest.TestCase): def test_sfi_resolved(self): @@ -2721,28 +2854,107 @@ class TestClaDecode(unittest.TestCase): self.assertEqual(r['secure_messaging'], 'SM header authenticated') def test_sm_proprietary(self): - # 0x0C = 0000 1100: b4-b3 = 01 → proprietary secure messaging - r = self._cla(0x0c) + # 0x04 = 0000 0100: b4-b3 = 01 → proprietary secure messaging; + # 0x0C = b4-b3 = 11 → header authenticated (ISO 7816-4 Table 2). + r = self._cla(0x04) self.assertEqual(r['secure_messaging'], 'proprietary') + r = self._cla(0x0c) + self.assertEqual(r['channel'], 0) + self.assertEqual(r['secure_messaging'], 'SM header authenticated') def test_chain_bit_b5(self): - # 0x20 = 0010 0000: b5=1 → first or continuing command of chain - r = self._cla(0x20) + # 0x10 = 0001 0000: b5=1 → first or continuing command of chain + # (ISO 7816-4 §5.1.1 Table 2). + r = self._cla(0x10) self.assertEqual(r['chain'], 'first or continuing') + def test_reserved_class_001x(self): + # 0x20 = 0010 0000: the '001x xxxx' range is reserved by ISO + # 7816-4; its bits carry no defined SM/chaining meaning. + r = self._cla(0x20) + self.assertEqual(r['interclass'], 'reserved (ISO 7816-4)') + self.assertEqual(r['chain'], 'n/a') + self.assertEqual(r['secure_messaging'], 'n/a') + def test_further_interindustry_channel_offset(self): # 0x40 = 0100 0000: further format, channel = b4-b1 + 4 = 4 r = self._cla(0x40) self.assertEqual(r['interclass'], 'inter-industry (further format)') self.assertEqual(r['channel'], 4) + def test_extended_channels_iso(self): + # TS 102 221 §10.3: the further-interindustry coding carries the + # extended channels 4-19 in b4-b1 (ISO 7816-4 §5.1.1 Table 3). + for cla, channel in ((0x44, 8), (0x49, 13), (0x4a, 14), (0x4f, 19), + (0x60, 4), (0x6f, 19)): + r = self._cla(cla) + self.assertEqual(r['channel'], channel, f'CLA {cla:#04x}') + # b6 of the further format is the SM indication, b5 the chaining bit. + self.assertEqual(self._cla(0x60)['secure_messaging'], + 'SM header not authenticated') + self.assertEqual(self._cla(0x70)['chain'], 'first or continuing') + + def test_extended_channels_etsi(self): + # TS 102 221 Table 10.4a: the ETSI '11x0 xxxx' coding addresses the + # extended channels too (b4-b1 + 4). + for cla, channel in ((0xc0, 4), (0xc5, 9), (0xcf, 19), (0xe0, 4), + (0xe5, 9), (0xef, 19)): + r = self._cla(cla) + self.assertEqual(r['interclass'], + 'ETSI-defined (extended logical channels)', + f'CLA {cla:#04x}') + self.assertEqual(r['channel'], channel, f'CLA {cla:#04x}') + # b6 is the SM indication (Table 10.4b), b5 must be 0. + self.assertEqual(self._cla(0xe5)['secure_messaging'], + 'SM header not authenticated') + self.assertEqual(self._cla(0xc5)['secure_messaging'], 'none') + def test_etsi_uicc(self): r = self._cla(0x80) self.assertEqual(r['interclass'], 'ETSI-defined (UICC/USIM)') + self.assertEqual(r['channel'], 0) + # The low nibble of the ETSI classes carries SM (b4-b3) and the + # standard channel (b2-b1), TS 102 221 Table 10.3/10.4. + r = self._cla(0x8c) + self.assertEqual(r['interclass'], 'ETSI-defined (UICC/USIM)') + self.assertEqual(r['channel'], 0) + self.assertEqual(r['secure_messaging'], 'SM header authenticated') + r = self._cla(0x83) + self.assertEqual(r['channel'], 3) def test_etsi_sim(self): r = self._cla(0xa0) self.assertEqual(r['interclass'], 'ETSI-defined (SIM/GSM)') + r = self._cla(0xa3) + self.assertEqual(r['interclass'], 'ETSI-defined (SIM/GSM)') + self.assertEqual(r['channel'], 3) + + def test_proprietary_and_invalid(self): + # b5 = 1 contradicts the ETSI class codings (TS 102 221 §10.1.1) and + # 0xFF is invalid; both keep the best-effort channel reading. + self.assertEqual(self._cla(0x90)['interclass'], 'proprietary') + self.assertEqual(self._cla(0xd5)['channel'], 9) + self.assertEqual(self._cla(0xff)['interclass'], 'invalid (ISO 7816-4)') + + def test_cla_channel_helper(self): + # The tracker/state paths share this helper with decode_cla. + from simtrace_analyser.decode import cla_channel + self.assertEqual(cla_channel(0x00), 0) + self.assertEqual(cla_channel(0x03), 3) + self.assertEqual(cla_channel(0x80), 0) + self.assertEqual(cla_channel(0x44), 8) + self.assertEqual(cla_channel(0x4f), 19) + self.assertEqual(cla_channel(0xef), 19) + + def test_looks_like_apdu(self): + from simtrace_analyser.decode import looks_like_apdu + self.assertTrue(looks_like_apdu(bytes.fromhex('00a40004023f00'))) + self.assertTrue(looks_like_apdu(bytes.fromhex('80f2000000'))) + self.assertFalse(looks_like_apdu(b'')) + self.assertFalse(looks_like_apdu(bytes.fromhex('00a4'))) + self.assertFalse(looks_like_apdu(bytes.fromhex('ff10957aff10957a'))) + self.assertFalse(looks_like_apdu(bytes.fromhex('2d6ccfb0ff'))) + self.assertFalse(looks_like_apdu(bytes.fromhex('31e073fe211b66'))) class TestLineEvents(unittest.TestCase): diff --git a/tests/test_filetrack.py b/tests/test_filetrack.py index 2149886..1f6e466 100644 --- a/tests/test_filetrack.py +++ b/tests/test_filetrack.py @@ -512,6 +512,25 @@ class TestNaming(TrackerCase): self.assertEqual(entry['path'], 'ADF/EF_SMS (6F3C)') self.assertEqual(entry['counters']['select'], 1) + def test_selections_are_per_extended_channel(self): + # CLA 0x44 / 0x48 address the extended channels 8 / 12 (TS 102 221 + # §10.3); each channel's selection drives its own reads. Before the + # channel fix both CLAs collapsed into channel 0's state, so both + # reads resolved to the last selected file. + data = self.track( + '44' + sel_fid('7fff')[2:], '44' + gr(fcp('7fff'))[2:], + '44' + sel_fid('6f07')[2:], '44' + gr(fcp('6f07', size=9))[2:], + '48' + sel_fid('7fff')[2:], '48' + gr(fcp('7fff'))[2:], + '48' + sel_fid('6fb7')[2:], '48' + gr(fcp('6fb7', size=2))[2:], + '44' + read_binary(0, '082905102143658709')[2:], + '48' + read_binary(0, '0A0B')[2:], + ) + self.assertIn('7fff/6f07', data['files']) + self.assertIn('7fff/6fb7', data['files']) + self.assertEqual(data['files']['7fff/6f07']['counters']['read'], 1) + self.assertEqual(data['files']['7fff/6fb7']['counters']['read'], 1) + self.assertEqual(data['files']['7fff/6f07']['name'], 'EF_IMSI') + def test_sim_df_names(self): # The same FID resolves against the SIM DF_GSM table. data = self.track( diff --git a/tests/test_pcap_import.py b/tests/test_pcap_import.py index c39b37a..126085a 100644 --- a/tests/test_pcap_import.py +++ b/tests/test_pcap_import.py @@ -224,7 +224,7 @@ class TestImportEndpoint(unittest.TestCase): class TestVersion(unittest.TestCase): def test_package_version(self): from simtrace2_pysniff import __version__ - self.assertEqual(__version__, '2.5.0') + self.assertEqual(__version__, '2.6.0') def test_status_includes_version(self): import os