diff --git a/docs/api.md b/docs/api.md index e5b777c..ae58f96 100644 --- a/docs/api.md +++ b/docs/api.md @@ -592,12 +592,16 @@ response carries them too, plus an `error`). fetched via a proactive command (FETCH). The response contains the same `por` structure if decoding succeeds. -When the secured data cannot be the TS 102 226 compact remote response (its -command count exceeds the command script that was sent - a third-party -applet's own TAR answers with application bytes), `por.response_type` is -`raw`, the whole secured data is reported as `decoded.last_response_data` -and no status word is decoded; `secured_data` carries the response bytes in -every case. +The response form follows the command's TAR (v3.21.0): a remote-management +TAR (the preset's role entries and the standard ISD/RFM allocations of +TS 101 220 Annex D) is decoded as an RM response; any other TAR's secured +data is the receiving application's own (TS 102 226 4) and is reported with +`por.response_type` `raw` - the whole secured data in +`decoded.last_response_data` and no status word (`secured_data` carries the +response bytes in every case). The same `raw` classification covers a +secured data that cannot be the compact structure (its command count exceeds +the command script that was sent, or it is too short for Table 5.1), while a +genuinely malformed packet or a failed RC/CC/DS check still yields no `por`. The SPI2 `por_in_submit` bit (0x20) selects submit-mode PoR. @@ -722,20 +726,29 @@ number 1-15), `tar`/`spi1`/`spi2` overrides and `format` (`compact` default / `expanded` / `expanded-ae` - the TS 102 226 5.2.1 Command Scripting template around the C-APDU) - KIc/KID and the counter always come from the resolved preset, which must match the equipped card and be -complete) or `status` +complete), `status` (`attempts`, `interval_ms` - when `attempts > 1` the default SW check is the -mask `91??`, i.e. poll until the card announces a command). +mask `91??`, i.e. poll until the card announces a command) or `proactive` +(the proactive cleanup: poll STATUS and consume every announced command - +`respond`/`first` TERMINAL RESPONSE results, `attempts`/`interval_ms` bound, +optional `require` `{command, qualifier?}` asserting at least one drained +command; the final SW is the step's `check.sw`, so a still-pending card fails +and hands the command to the next step - an empty drain passes unless +`require` was given). `check` is `{"sw": ..., "data": ...}` (exact or `{"mode": "mask", "value": "91??"}`, `?` = per-nibble wildcard) plus `"por"` for SCP80 - `"none"`/`"ok"`/`"any"` or an object `{"status"?, "sw"?, "data"?}` asserting the decoded PoR of this exchange (inline or SEND SHORT MESSAGE; at least one -field). `on_fail` is `error` (terminates the script) or `warning` (continues). +field). `on_fail` is `error` (terminates the script) or `warning` (continues) +and applies to every step failure - including an expectation that finds no +pending command (v3.21.0; `warning` is the tolerant consume-if-pending form). **Expectation steps** (`type: "expect"`) require a command pending from the previous step (`91XX`); they never poll - a `9000` response means no command -and is an error (TS 102 221 7.4.2.1 / TS 102 223 6.3; add a `status` action -if the card delivers on poll). `command` is a name or type code; `checks` +and fails the step at its `on_fail` level (TS 102 221 7.4.2.1 / TS 102 223 +6.3; a `status` action with `attempts > 1` polls when the card delivers on +poll). `command` is a name or type code; `checks` may be `text` (contains/exact), `alpha` (the command's Alpha identifier `05`/`85`, contains/exact - distinct from the `8D` text string), `item` (`id`/`text` for SELECT ITEM / SET UP MENU), `raw` (mask), `por` (a SEND @@ -756,6 +769,14 @@ The response is the initial state (`running: true`), the final counter (`scp80_counter`) and the step list; poll `/api/test/status`. The PWA writes `scp80_counter` back to the card preset after the run. +A run that ends (an error, a stop, an unexpected pending command) with a +command still announced **drains it with `ok` (0x00)** - fetching while the +TERMINAL RESPONSE answers `91XX` (bounded); a cancel is never sent +automatically (it tells the card the user aborted the session and can push +assertive applications onto an error path). The cleanup is reported as a +`cleanup` step; a script that wants the refusal path answers explicitly (an +expectation's `respond`, or the `proactive` action's `first`). + ### `GET /api/test/scripts` and the script store Test scripts live in a server-side store (`~/.pysim-simple-server/test_scripts.json`, diff --git a/docs/test-scripts.md b/docs/test-scripts.md index 8908638..f3f04b7 100644 --- a/docs/test-scripts.md +++ b/docs/test-scripts.md @@ -54,6 +54,13 @@ secured data is exposed as `por.data`. Assert such a response with `data` and leave `sw` empty; the step's `check.sw` still asserts the envelope exchange. +The form is decided by the command's **TAR** (v3.21.0): the preset's role +TARs and the standard ISD/RFM allocations (TS 101 220 Annex D) are decoded as +RM responses; any other TAR - an applet's own - is reported `raw` regardless +of whether its bytes happen to parse as a compact response, so no status word +is ever fabricated (a 2-byte reply cannot be one: Table 5.1 needs the command +count plus two status bytes - it is reported raw too). + A `62XX`/`63XX` envelope answer does not guarantee a response packet: an application that refuses at the envelope level (e.g. an applet answering `6200` to an unknown command) sends none, while other warning paths still @@ -172,6 +179,9 @@ TEST-RUN step 4: MENU-SELECT ENVELOPE=80c2000009d30702020181900101 item=1 ('One' - Other actions log their send/response pair too (`APDU TX=… -> SW=…`, `ENVELOPE(Event Download) … -> SW=…`, `READ BINARY `, `READ RECORD `, `UPDATE BINARY `, each with `-> SW=…`). +- A `proactive` action logs its rounds (`STATUS i/N -> …`, `FETCH=…`, + `CMD 0x…`, `TR=… -> SW=…`); the run cleanup logs + `TEST-RUN drain: (0xNN) TR=… -> SW=…`. - A failing step adds `TEST-RUN step N failed: …` with the reason. - The shared secured-packet sender logs each segment's answer (`OTA SEND: ENVELOPE i/N -> `). @@ -193,7 +203,11 @@ TEST-RUN step 4: MENU-SELECT ENVELOPE=80c2000009d30702020181900101 item=1 ('One' Every step may carry `on_fail`: `"error"` (default - stops the run) or `"warning"` (continues, the report shows ⚠). Each entry of `checks` may -override it with its own `on_fail`. +override it with its own `on_fail`. The level applies to **every** step +failure, including an expectation that finds no pending command (v3.21.0) - +the tolerant "consume a command if there is one" form is an expectation with +`on_fail: "warning"`; to drain and confirm the card is idle, use the +`proactive` action. ## Presets: where the keys, TARs and counters come from @@ -233,6 +247,7 @@ consecutive runs continue correctly. | `apdu` | `apdu` hex | raw transport, no auto-handler | | `scp80` | `apdu` **xor** `sp` (optional `source`), `kvn`, `tar`, `spi1`, `spi2`, `format` | secured packet to the TAR; `format`: `compact` (default - the C-APDU verbatim), `expanded` (the `AA`-prefixed Command Scripting template) or `expanded-ae` (the `AE 80 … 00 00` form), TS 102 226 §5.2.1 | | `status` | `attempts` 1–1000, `interval_ms` 0–10000 | with `attempts > 1` the default SW check is `91??` (poll until a command) | +| `proactive` | `respond`, `first`, `attempts` 1–1000 (default 3), `interval_ms` (default 200), `require` | drain the card's announced proactive commands and confirm it is idle (see below) | `path` is `/`-separated: `MF` (or `3F00`) or an ADF name/AID first, then FIDs or file names - e.g. `MF/7F20/6F07`, `ADF.USIM/EF.TEST`. @@ -266,6 +281,37 @@ actions take the same optional `src` device-identities override. `src` overrides the device-identities source (`82` terminal, `83` network); the default is the terminal, matching the Phone tab's forms. +### Proactive cleanup (`kind: "proactive"`) + +Consume whatever the card announces and leave it idle - at the start, in the +middle or at the end of a script (a leftover command from a previous run must +not derail the next step). Each round polls STATUS; a `91XX` answer fetches +the command and answers it (the first command with `first`, the rest with +`respond` - both `{"result": …}` like an expectation's response). The loop +ends when STATUS answers `9000` (idle), a non-`91XX` SW appears or `attempts` +rounds are spent; the **final SW is the step's `check.sw`** (default `9000`), +so a still-pending card fails the step and hands the pending command to the +next step. + +```json +{"type": "action", "kind": "proactive", + "params": {"respond": {"result": "ok"}, + "first": {"result": "cancel"}, + "attempts": 5, "interval_ms": 200, + "require": {"command": "REFRESH", "qualifier": "00"}}, + "check": {"sw": "9000"}, + "label": "serve whatever is pending, leave the card idle"} +``` + +- An **empty drain passes** (that is the point) - unless `require` is given: + the step then fails with "nothing was pending". +- `require` asserts **at least one** consumed command matches (`command` name + or hex type, `ANY` allowed; optional `qualifier`, exact/mask). +- `first` lets a script test the card's refusal path for the first command + (e.g. `{"result": "cancel"}`) while the rest are answered `ok`. +- Every consumed command is reported (type, qualifier, the TR's SW) and + logged (`STATUS`/`FETCH`/`CMD`/`TR` lines under the step). + ### `check` on actions ```json @@ -303,6 +349,12 @@ runner never polls by itself. `command` is a name (as shown in the proactive log) or a hex type; `ANY`/`*` skips the type check. `qualifier` is the command-details byte 3 (one byte, exact/mask). +When nothing is pending the step fails at its own `on_fail` level (the +default stops the run; `on_fail: "warning"` warns and continues - the +tolerant consume-if-pending form, v3.21.0). "Add a `status` action" applies +when the card delivers on poll: `status` with `attempts > 1` polls until a +command appears; it cannot make one appear when the card has none. + ### Content checks | kind | fields | checks | @@ -365,6 +417,10 @@ UCS2 answer to a GET INPUT: "respond": {"result": "ok", "text": "Пароль", "dcs": "08"}} ``` +The `proactive` action's `respond`/`first` use the same shape (the command +type varies per drained command, so `text`/`item_id` apply where the command +carries them). + ## The two PoR transports (SPI2) 1. **Inline** - the card answers the ENVELOPE `9000` with the PoR in the @@ -487,14 +543,29 @@ PoR-in-submit (with the transport asserted by the expectation): is outside the preset's keysets. - `no MSL for TAR … - set it in the Cards tab or pass spi1` - the step's TAR has no entry in the preset and the step carries no `spi1`. -- `no proactive command pending …` - an expectation ran without a `91XX` from - the previous step (add a `status` action when the card delivers on poll). +- `no proactive command pending` - an expectation ran without a `91XX` from + the previous step; it fails at the step's `on_fail` level (the check row + names it). A `status` action with `attempts > 1` polls until a command + appears when the card delivers on poll; `on_fail: "warning"` tolerates the + empty case, and the `proactive` action drains and confirms idle. - `menu-select: no menu item matches …` - the error lists the cached menu; a miss on an explicit `case_sensitive: true` names the case-insensitive candidates. - `card verdict: no PoR (the card sent none for SPI2 …)` - an SCP80 operation expected a PoR that never arrived (check SPI2 and the transport). +## Run cleanup + +When a run ends (an error, a stop request, or an unexpected pending command) +with a command still announced, the runner fetches it and answers **`ok` +(0x00)** and keeps draining while the TERMINAL RESPONSE answers `91XX` +(bounded) - a cancel (`0x10`) is not sent automatically: it tells the card +the user aborted the proactive session, which can push assertive applications +onto an error path and make them queue a further command. The cleanup is +reported like a normal step (`TR ok …`, `pending REFRESH (0x01) answered with +ok`); a script that wants to test the refusal path answers explicitly (an +expectation's `respond`, or the proactive action's `first`). + ## Known limitations - The runner executes one command at a time and never polls by itself; a diff --git a/frontend/help-ru.html b/frontend/help-ru.html index b56c91b..58fffbf 100644 --- a/frontend/help-ru.html +++ b/frontend/help-ru.html @@ -563,12 +563,12 @@

8.9 Тест-скрипт

Тест-скрипт — упорядоченный список действий и ожиданий проактивных команд, выполняемый на подключённой карте на стороне сервера. Пока запуск активен, карта принадлежит скрипту: остальные операции с картой отклоняются (HTTP 409), фоновый опрос STATUS приостанавливается; отчёт о запуске хранится в памяти (в том числе после перезагрузки страницы) до нажатия Очистить отчёт. Скрипты хранятся в хранилище на сервере (~/.pysim-simple-server/test_scripts.json, --test-scripts), поэтому все браузеры работают с одним набором; изменения записываются в хранилище по мере правки (кнопки «Сохранить» нет — черновик без валидного шага остаётся локальным); экспорт/импорт JSON остаётся для резервных копий и переноса скриптов между стендами (старый экспорт из localStorage импортируется как есть).

    -
  • Действия — Envelope (Event Download: сырой тип события + данные, с опциональным переопределением источника — 82 терминал / 83 сеть) или Event (семантические формы: Location status, Access technology change, Network rejection, Data connection status change — поля редактируются как объект JSON в редакторе шагов, а для остальных событий форма события на вкладке «Телефон» показывает готовый hex), Menu Selection (item id 1–255 или текст пункта с точным/частичным совпадением, по умолчанию без учёта регистра — выбор «Учёт регистра» в редакторе шагов включает точное совпадение; текст подбирается по меню карты, которое следует за параметрами установки апплета), UPDATE/READ файла (путь + режим; чтение проверяет данные файла), сырой APDU (без авто-обработчика), SCP80 (C-APDU или готовый защищённый пакет, опциональные переопределения номера keyset (kvn) / TAR / SPI1 / SPI2 и формат команды — компактный или обёрнутый в шаблон Command Scripting TS 102 226 §5.2.1 expanded (AA) / expanded AE 80; ключи и счётчик берутся из пресета карты, подобранного по ICCID подключённой карты — запуск называет пресет в хранилище, поэтому всегда используются его keysets, TAR и счётчик — а счётчик записывается обратно после запуска) и STATUS (при попытках > 1 опрашивает, пока карта не объявит команду). У каждого действия есть проверка ответа: ожидаемые SW и данные (точно или маска с подстановкой ? — 91??) и проверка PoR для SCP80 (любой / без PoR / por_ok или своя проверка декодированного PoR — статус / SW / данные; транспорт через SEND SHORT MESSAGE работает так же).
    • -
  • Ожидания — карта должна объявить проактивную команду на предыдущем шаге (SW 91XX, TS 102 221 §7.4.2.1 / TS 102 223 §6.3); 9000 означает «команды нет» и считается ошибкой — добавьте действие STATUS, если апплет выдаёт команду при опросе. Проверяются тип команды, квалификатор и содержимое: текст / пункт / raw, проверка Alpha (идентификатор Alpha команды, тег 05/85 — отличается от текстовой строки 8D), проверка SMS TPDU (submit-TPDU команды SEND SHORT MESSAGE: получатель / PID / DCS / UDL / данные — для собственного SMS апплета), проверка PoR (SEND SHORT MESSAGE несёт PoR, когда пакет запросил PoR-in-submit, бит SPI2 20: статус ответа, SW R-APDU и данные ответа; перед этим должен выполниться шаг SCP80 — SW R-APDU находится внутри полезной нагрузки ответа: собственный TAR апплета возвращает данные приложения, поэтому проверяйте данные, а SW оставляйте пустым; отклонённый ENVELOPE (62XX/63XX) может вовсе не нести пакет ответа — для такого шага оставляйте проверку PoR «любой» или ожидайте «без PoR») и проверка Files (File List команды REFRESH, например квалификатор 00 = NAA initialization with full file change notification). TERMINAL RESPONSE (результат, идентификатор пункта, текст, доп. TLV) является частью инструкции.
    • -
  • Уровни ошибок — несовпадение помечает шаг как Ошибка (скрипт останавливается) или Предупреждение (продолжает); каждая проверка показывается с ожидаемым и фактическим значением.
    • +
  • Действия — Envelope (Event Download: сырой тип события + данные, с опциональным переопределением источника — 82 терминал / 83 сеть) или Event (семантические формы: Location status, Access technology change, Network rejection, Data connection status change — поля редактируются как объект JSON в редакторе шагов, а для остальных событий форма события на вкладке «Телефон» показывает готовый hex), Menu Selection (item id 1–255 или текст пункта с точным/частичным совпадением, по умолчанию без учёта регистра — выбор «Учёт регистра» в редакторе шагов включает точное совпадение; текст подбирается по меню карты, которое следует за параметрами установки апплета), UPDATE/READ файла (путь + режим; чтение проверяет данные файла), сырой APDU (без авто-обработчика), SCP80 (C-APDU или готовый защищённый пакет, опциональные переопределения номера keyset (kvn) / TAR / SPI1 / SPI2 и формат команды — компактный или обёрнутый в шаблон Command Scripting TS 102 226 §5.2.1 expanded (AA) / expanded AE 80; ключи и счётчик берутся из пресета карты, подобранного по ICCID подключённой карты — запуск называет пресет в хранилище, поэтому всегда используются его keysets, TAR и счётчик — а счётчик записывается обратно после запуска), STATUS (при попытках > 1 опрашивает, пока карта не объявит команду) и Очистка proactive (опрос STATUS и поглощение всех объявленных команд — результат TERMINAL RESPONSE для каждой, опционально другой результат для первой; граница попыток/интервала; опциональная проверка require по поглощённому. Пустой опрос проходит, финальный STATUS должен быть idle, а оставшаяся команда передаётся следующему шагу — используйте в начале, середине или конце скрипта, чтобы убедиться, что карта перестала слать команды). У каждого действия есть проверка ответа: ожидаемые SW и данные (точно или маска с подстановкой ? — 91??) и проверка PoR для SCP80 (любой / без PoR / por_ok или своя проверка декодированного PoR — статус / SW / данные; транспорт через SEND SHORT MESSAGE работает так же).
    • +
  • Ожидания — карта должна объявить проактивную команду на предыдущем шаге (SW 91XX, TS 102 221 §7.4.2.1 / TS 102 223 §6.3); 9000 означает «команды нет»: шаг падает на своём уровне ошибок — Ошибка останавливает скрипт, Предупреждение продолжает (терпимая форма «поглотить команду, если она есть») — а действие STATUS с попытками > 1 опрашивает, когда апплет выдаёт команду при опросе. Проверяются тип команды, квалификатор и содержимое: текст / пункт / raw, проверка Alpha (идентификатор Alpha команды, тег 05/85 — отличается от текстовой строки 8D), проверка SMS TPDU (submit-TPDU команды SEND SHORT MESSAGE: получатель / PID / DCS / UDL / данные — для собственного SMS апплета), проверка PoR (SEND SHORT MESSAGE несёт PoR, когда пакет запросил PoR-in-submit, бит SPI2 20: статус ответа, SW R-APDU и данные ответа; перед этим должен выполниться шаг SCP80 — SW R-APDU находится внутри полезной нагрузки ответа: собственный TAR апплета возвращает данные приложения, поэтому проверяйте данные, а SW оставляйте пустым; отклонённый ENVELOPE (62XX/63XX) может вовсе не нести пакет ответа — для такого шага оставляйте проверку PoR «любой» или ожидайте «без PoR») и проверка Files (File List команды REFRESH, например квалификатор 00 = NAA initialization with full file change notification). TERMINAL RESPONSE (результат, идентификатор пункта, текст, доп. TLV) является частью инструкции.
    • +
  • Уровни ошибок — несовпадение помечает шаг как Ошибка (скрипт останавливается) или Предупреждение (продолжает); уровень применяется к любой ошибке шага, включая ожидание без объявленной команды; каждая проверка показывается с ожидаемым и фактическим значением.
  • Шаблоны — Empty, Обзор меню STK, Апплет через SCP80; редактор шагов — формный (поля действия Event вводятся объектом JSON), отдельные скрипты экспортируются/импортируются как JSON.
-

Каждый шаг пишет семантическую строку в журнал сервера под своим номером (TEST-RUN step N: …), запрос перед ответом: отправленный APDU/данные и полученный SW, для SCP80 — открытый C-APDU, собранный защищённый пакет и R-APDU из PoR, для ожидания — тройка FETCH / команда / TERMINAL RESPONSE, а для выбора пункта меню — полный ENVELOPE с определённым пунктом. Запустите сервер с --apdu-trace, чтобы добавить и журнал сырых транспортных APDU.

+

Каждый шаг пишет семантическую строку в журнал сервера под своим номером (TEST-RUN step N: …), запрос перед ответом: отправленный APDU/данные и полученный SW, для SCP80 — открытый C-APDU, собранный защищённый пакет и R-APDU из PoR, для ожидания — тройка FETCH / команда / TERMINAL RESPONSE, очистка proactive — каждый раунд STATUS/FETCH/команда/TR, а очистка в конце запуска — поглощённые команды (отвечаются ok — cancel отправляется только по явному запросу скрипта), а для выбора пункта меню — полный ENVELOPE с определённым пунктом. Запустите сервер с --apdu-trace, чтобы добавить и журнал сырых транспортных APDU.

8.10 BIP-терминал (обычный TCP)

Пилюля BIP запускает терминальную сторону Bearer Independent Protocol для карт, использующих TCP без HTTP OTA (SCP81): апплет или пуш-триггер может открыть канал и обменяться сырыми данными. Одновременно работает только одна BIP-сессия — запуск здесь останавливает слушатель SCP81, а запуск слушателя SCP81 останавливает эту сессию (в обоих случаях сообщается, что было заменено).

diff --git a/frontend/help.html b/frontend/help.html index 1be3517..033cb78 100644 --- a/frontend/help.html +++ b/frontend/help.html @@ -562,12 +562,12 @@

8.9 Test script

A test script is an ordered list of actions and proactive-command expectations executed server-side on the equipped card. While a run is active the card belongs to the script: other card actions are refused (HTTP 409) and background STATUS polling is paused; the run report is kept in memory (also after reloading the page) until Clear report. Scripts live in the server-side store (~/.pysim-simple-server/test_scripts.json, --test-scripts), so every browser manages the same set; edits are written to the store as you change them (there is no Save button — a draft without a valid step stays local until it is valid); JSON export/import remains for backup and for moving scripts between benches (the old localStorage export imports as-is).

    -
  • Actions — Envelope (Event Download: raw event type + data, with an optional source override — 82 terminal / 83 network) or Event (the semantic forms: Location status, Access technology change, Network rejection, Data connection status change — the fields are edited as a JSON object in the step editor, and the Phone tab's event form shows the built hex for the other events), Menu Selection (item id 1–255, or the item's text with an exact/contains match, case-insensitive by default — the step editor's Match case choice switches to an exact-case match; the text is resolved against the card's menu, which follows the applet's install parameters), UPDATE/READ file (path + mode; reads verify the file data), raw APDU (no auto-handler), SCP80 (C-APDU or a pre-built secured packet, optional keyset number (kvn) / TAR / SPI1 / SPI2 overrides and a command format — compact, or the TS 102 226 §5.2.1 Command Scripting template expanded (AA) / expanded AE 80 around the C-APDU; the keys and the counter come from the card preset matched by the equipped card's ICCID — the run names the stored preset, so it always uses its keysets, TARs and counter — and the counter is written back to it after the run) and STATUS (with attempts > 1 it polls until the card announces a command). Every action has a response check: expected SW and data (exact, or a mask with ? nibble wildcards — 91??) plus a PoR check for SCP80 (any / no PoR / por_ok, or a custom assertion of the decoded PoR — status / SW / data; the SEND SHORT MESSAGE transport works the same).
    • -
  • Expectations — the card must have announced a proactive command in the previous step (SW 91XX, TS 102 221 §7.4.2.1 / TS 102 223 §6.3); a 9000 means “no command” and is an error — add a STATUS action when the applet delivers on poll. The command type, qualifier and contents are verified: text / item / raw checks, an Alpha check (the command's Alpha identifier, tag 05/85 — distinct from the 8D text string), an SMS TPDU check (the submit TPDU of a SEND SHORT MESSAGE: destination / PID / DCS / UDL / user data — for an applet's own SMS), a PoR check (a SEND SHORT MESSAGE carries the PoR when the packet requested PoR-in-submit, SPI2 bit 20: the response status, the R-APDU SW and the response data; make sure an SCP80 step ran first — the R-APDU SW lives inside the response payload: an applet's own TAR returns application data, so check its data and leave the SW empty; a refused ENVELOPE (62XX/63XX) may carry no response packet at all — leave the PoR check at any or expect no PoR for such a step) and a Files check (the File List of a REFRESH, e.g. qualifier 00 = NAA initialization with full file change notification). The scripted TERMINAL RESPONSE (result, item identifier, text, extra TLVs) is part of the instruction.
    • -
  • Fail levels — a mismatch flags the step as Error (terminates the script) or Warning (continues); each check is reported with expected vs actual values.
    • +
  • Actions — Envelope (Event Download: raw event type + data, with an optional source override — 82 terminal / 83 network) or Event (the semantic forms: Location status, Access technology change, Network rejection, Data connection status change — the fields are edited as a JSON object in the step editor, and the Phone tab's event form shows the built hex for the other events), Menu Selection (item id 1–255, or the item's text with an exact/contains match, case-insensitive by default — the step editor's Match case choice switches to an exact-case match; the text is resolved against the card's menu, which follows the applet's install parameters), UPDATE/READ file (path + mode; reads verify the file data), raw APDU (no auto-handler), SCP80 (C-APDU or a pre-built secured packet, optional keyset number (kvn) / TAR / SPI1 / SPI2 overrides and a command format — compact, or the TS 102 226 §5.2.1 Command Scripting template expanded (AA) / expanded AE 80 around the C-APDU; the keys and the counter come from the card preset matched by the equipped card's ICCID — the run names the stored preset, so it always uses its keysets, TARs and counter — and the counter is written back to it after the run), STATUS (with attempts > 1 it polls until the card announces a command) and Proactive cleanup (poll STATUS and consume every announced command — the TERMINAL RESPONSE result for each, optionally a different result for the first; the attempt/interval bound; an optional require assertion over what was drained. An empty poll passes, the final STATUS must be idle, and a still-pending command is handed to the next step — use it at the start, in the middle or at the end of a script to make sure the card has stopped sending). Every action has a response check: expected SW and data (exact, or a mask with ? nibble wildcards — 91??) plus a PoR check for SCP80 (any / no PoR / por_ok, or a custom assertion of the decoded PoR — status / SW / data; the SEND SHORT MESSAGE transport works the same).
    • +
  • Expectations — the card must have announced a proactive command in the previous step (SW 91XX, TS 102 221 §7.4.2.1 / TS 102 223 §6.3); a 9000 means “no command”: the step fails at its fail level — Error stops the script, Warning continues (the tolerant “consume a command if there is one” form) — and a STATUS action with attempts > 1 polls when the applet delivers on poll. The command type, qualifier and contents are verified: text / item / raw checks, an Alpha check (the command's Alpha identifier, tag 05/85 — distinct from the 8D text string), an SMS TPDU check (the submit TPDU of a SEND SHORT MESSAGE: destination / PID / DCS / UDL / user data — for an applet's own SMS), a PoR check (a SEND SHORT MESSAGE carries the PoR when the packet requested PoR-in-submit, SPI2 bit 20: the response status, the R-APDU SW and the response data; make sure an SCP80 step ran first — the R-APDU SW lives inside the response payload: an applet's own TAR returns application data, so check its data and leave the SW empty; a refused ENVELOPE (62XX/63XX) may carry no response packet at all — leave the PoR check at any or expect no PoR for such a step) and a Files check (the File List of a REFRESH, e.g. qualifier 00 = NAA initialization with full file change notification). The scripted TERMINAL RESPONSE (result, item identifier, text, extra TLVs) is part of the instruction.
    • +
  • Fail levels — a mismatch flags the step as Error (terminates the script) or Warning (continues); every step failure honours it, including an expectation that finds no pending command; each check is reported with expected vs actual values.
  • Templates — Empty, STK menu browsing, Applet via SCP80; the step editor is form-based (the event action's fields are a JSON object) and individual scripts export/import as JSON.
-

Every step logs a semantic line to the server log under its step number (TEST-RUN step N: …), request before response: the APDU/data sent and the SW received, for SCP80 the plaintext C-APDU, the built secured packet and the PoR's R-APDU, for an expectation the FETCH / command / TERMINAL RESPONSE trio, and for a menu selection the complete ENVELOPE with the resolved item. Start the server with --apdu-trace to add the raw transport APDU log as well.

+

Every step logs a semantic line to the server log under its step number (TEST-RUN step N: …), request before response: the APDU/data sent and the SW received, for SCP80 the plaintext C-APDU, the built secured packet and the PoR's R-APDU, for an expectation the FETCH / command / TERMINAL RESPONSE trio, a proactive cleanup every STATUS/FETCH/command/TR round and the run cleanup the commands it drained (answered ok — a cancel is only sent when a script asks for it), and for a menu selection the complete ENVELOPE with the resolved item. Start the server with --apdu-trace to add the raw transport APDU log as well.

8.10 BIP terminal (generic TCP)

The BIP pill runs the terminal side of the Bearer Independent Protocol for cards that use TCP without HTTP OTA (SCP81): an applet or a push trigger can open a channel and exchange raw data. Only one BIP session runs at a time — starting here stops a running SCP81 listener, and starting the SCP81 listener stops this session (both report what was replaced).

diff --git a/frontend/index.html b/frontend/index.html index c2e9eb4..d626c4b 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -1877,7 +1877,7 @@ // ===== Version ===== // Single source of truth for the PWA version: shown in the header and used // by the server version check in pysimConnect(). -const SIMPLE_VERSION = '3.20.0'; +const SIMPLE_VERSION = '3.21.0'; document.getElementById('app-version').textContent = 'v' + SIMPLE_VERSION; // ===== Tab switching ===== @@ -6913,7 +6913,7 @@ function decodeResponse() { // through /api/test/status. This UI edits the script and monitors the run; // SCP80 steps use the card preset matched by the equipped card's ICCID, and // the final SCP80 counter is written back to it. -const TEST_ACTION_KINDS = ['envelope', 'event', 'menu-select', 'file-write', 'file-read', 'apdu', 'scp80', 'status']; +const TEST_ACTION_KINDS = ['envelope', 'event', 'menu-select', 'file-write', 'file-read', 'apdu', 'scp80', 'status', 'proactive']; let _testScripts = null; let _testCurrentIdx = -1; let _testRunState = null; @@ -7072,6 +7072,8 @@ function testScriptProblem(script) { } } if ((s.kind === 'file-write' || s.kind === 'file-read') && !String(p.path || '').trim()) return at + t('file path is empty'); + if (s.kind === 'proactive' && String(p.attempts === undefined ? '' : p.attempts).trim() !== '' + && !(parseInt(p.attempts, 10) >= 1)) return at + t('attempts must be at least 1'); } else if (s.type === 'expect') { if (!String(s.command || '').trim()) return at + t('command is required'); } else { @@ -7147,6 +7149,15 @@ function testStepSummary(step) { else if (step.kind === 'file-write') s = 'UPDATE ' + (p.path || '?') + (p.mode && p.mode !== 'auto' ? ' (' + p.mode + (p.record ? ' ' + p.record : '') + ')' : '') + ' = ' + (p.data || ''); else if (step.kind === 'file-read') s = 'READ ' + (p.path || '?') + (p.mode && p.mode !== 'auto' ? ' (' + p.mode + (p.record ? ' ' + p.record : '') + ')' : ''); else if (step.kind === 'status') s = 'STATUS' + ((p.attempts || 1) > 1 ? ' x' + p.attempts + ' (poll)' : ''); + else if (step.kind === 'proactive') { + const parts = ['PROACTIVE x' + (p.attempts === undefined ? 3 : p.attempts)]; + const rq = p.require || {}; + if (rq.command) parts.push('require ' + rq.command + (rq.qualifier ? ' q=' + (rq.qualifier.value || rq.qualifier) : '')); + const rr = p.respond || {}; + parts.push('TR ' + (rr.result === undefined ? 'ok' : rr.result)); + if (p.first && p.first.result) parts.push('first ' + p.first.result); + s = parts.join(' '); + } else s = String(step.kind || '?'); const c = step.check || {}; const ck = []; @@ -7464,6 +7475,24 @@ function testStepRender() { } else if (step.kind === 'status') { html += testFormRow('Attempts (1 = single STATUS)', testFormInput('test-f-attempts', p.attempts === undefined ? 1 : p.attempts, '1..1000')); html += testFormRow('Interval (ms, when polling)', testFormInput('test-f-interval', p.interval_ms === undefined ? 200 : p.interval_ms, '0..10000')); + } else if (step.kind === 'proactive') { + const pr = p.respond || {}; + html += testFormRow('TERMINAL RESPONSE result (each command)', + testFormSelect('test-f-presult', pr.result === undefined ? 'ok' : String(pr.result), [ + {v: 'ok', l: t('ok')}, {v: 'cancel', l: t('cancel')}, {v: 'back', l: t('back')}, + {v: 'timeout', l: t('timeout')}, {v: 'no_response', l: t('no response')}])); + const pf = p.first || {}; + html += testFormRow('First command only (optional)', + testFormSelect('test-f-pfirst', pf.result === undefined ? '' : String(pf.result), [ + {v: '', l: t('same as above')}, {v: 'ok', l: t('ok')}, {v: 'cancel', l: t('cancel')}, + {v: 'back', l: t('back')}, {v: 'timeout', l: t('timeout')}, + {v: 'no_response', l: t('no response')}])); + html += testFormRow('Attempts (STATUS rounds)', testFormInput('test-f-pattempts', p.attempts === undefined ? 3 : p.attempts, '1..1000')); + html += testFormRow('Interval (ms, between rounds)', testFormInput('test-f-pinterval', p.interval_ms === undefined ? 200 : p.interval_ms, '0..10000')); + const rq = p.require || {}; + html += testFormRow('Require a command (optional)', '
' + + testFormInput('test-f-prequire', rq.command, 'REFRESH') + + testFormInput('test-f-prequal', rq.qualifier && rq.qualifier.value, '00') + '
'); } const c = step.check || {}; const sw = c.sw || {}; @@ -7720,6 +7749,20 @@ function testStepCollect() { if (fmt && fmt !== 'compact') p.format = fmt; } else if (kind === 'status') { p.attempts = num('test-f-attempts', 1); p.interval_ms = num('test-f-interval', 200); } + else if (kind === 'proactive') { + p.respond = {result: val('test-f-presult') || 'ok'}; + const first = val('test-f-pfirst'); + if (first) p.first = {result: first}; + p.attempts = num('test-f-pattempts', 3); + p.interval_ms = num('test-f-pinterval', 200); + const requireCmd = val('test-f-prequire'); + if (requireCmd) { + const rq = {command: requireCmd.toUpperCase()}; + const rqHex = hex('test-f-prequal'); + if (rqHex) rq.qualifier = rqHex; + p.require = rq; + } + } step.params = p; const check = {}; const swVal = hex('test-f-sw'); @@ -7793,6 +7836,7 @@ function testStepFormError() { } if ((step.kind === 'file-write' || step.kind === 'file-read') && !p.path) return t('file path is empty'); if (step.kind === 'status' && !(p.attempts >= 1)) return t('attempts must be at least 1'); + if (step.kind === 'proactive' && !(p.attempts >= 1)) return t('attempts must be at least 1'); } else if (!step.command) { return t('command is required'); } @@ -21276,6 +21320,12 @@ const LANG_RU = { 'SPI2 override (hex, optional)': 'Переопределение SPI2 (hex, опционально)', 'Attempts (1 = single STATUS)': 'Попытки (1 = один STATUS)', 'Interval (ms, when polling)': 'Интервал (мс, при опросе)', + 'Attempts (STATUS rounds)': 'Попытки (раунды STATUS)', + 'Interval (ms, between rounds)': 'Интервал (мс, между раундами)', + 'TERMINAL RESPONSE result (each command)': 'Результат TERMINAL RESPONSE (для каждой команды)', + 'First command only (optional)': 'Только для первой команды (необязательно)', + 'same as above': 'как выше', + 'Require a command (optional)': 'Требовать команду (необязательно)', 'attempts must be at least 1': 'число попыток должно быть ≥ 1', 'Expected SW': 'Ожидаемое SW', 'default: 9000 (91?? when polling)': 'по умолчанию: 9000 (91?? при опросе)', diff --git a/frontend/sw.js b/frontend/sw.js index 56a87cd..2b51a26 100644 --- a/frontend/sw.js +++ b/frontend/sw.js @@ -1,4 +1,4 @@ -const CACHE = 'simple-v395'; +const CACHE = 'simple-v396'; const URLS = [ 'index.html', 'help.html', diff --git a/frontend/tests/testscripts.test.js b/frontend/tests/testscripts.test.js index 141e103..14df888 100644 --- a/frontend/tests/testscripts.test.js +++ b/frontend/tests/testscripts.test.js @@ -107,6 +107,10 @@ test('testScriptProblem accepts good scripts and names bad ones', () => { assert.match(testScriptProblem({ steps: [{ type: 'action', kind: 'file-read', params: {} }] }), /file path/); assert.match(testScriptProblem({ steps: [{ type: 'expect' }] }), /command is required/); + assert.match(testScriptProblem({ steps: [{ type: 'action', kind: 'proactive', + params: { attempts: 0 } }] }), /attempts/); + assert.strictEqual(testScriptProblem({ steps: [{ type: 'action', kind: 'proactive', + params: {} }] }), ''); // the selected SCP80 source decides which value is required assert.match(testScriptProblem({ steps: [{ type: 'action', kind: 'scp80', params: { source: 'apdu', sp: 'AA' } }] }), /SCP80/); @@ -116,6 +120,29 @@ test('testScriptProblem accepts good scripts and names bad ones', () => { params: { source: 'sp', sp: 'AA', apdu: '80E2' } }] }), ''); }); +test('the proactive cleanup step collects its fields', () => { + _testEditStep = { type: 'action', kind: 'proactive', params: {} }; + _testEditChecks = []; + _testEditStepIndex = 0; + fakeForm({ 'test-step-kind': 'proactive', 'test-f-presult': 'ok', 'test-f-pfirst': 'cancel', + 'test-f-pattempts': '5', 'test-f-pinterval': '0', 'test-f-prequire': 'refresh', + 'test-f-prequal': '00', 'test-f-sw': '', 'test-f-cdata': '', 'test-f-fail': 'error' }); + testStepCollect(); + const p = _testEditStep.params; + assert.strictEqual(p.respond.result, 'ok'); + assert.strictEqual(p.first.result, 'cancel'); + assert.strictEqual(p.attempts, 5); + assert.strictEqual(p.interval_ms, 0); + assert.deepStrictEqual(p.require, { command: 'REFRESH', qualifier: '00' }); + // an empty require field drops the assertion + fakeForm({ 'test-step-kind': 'proactive', 'test-f-presult': 'ok', 'test-f-pfirst': '', + 'test-f-pattempts': '3', 'test-f-pinterval': '200', 'test-f-prequire': '', + 'test-f-prequal': '', 'test-f-sw': '', 'test-f-cdata': '', 'test-f-fail': 'error' }); + testStepCollect(); + assert.ok(!('first' in _testEditStep.params)); + assert.ok(!('require' in _testEditStep.params)); +}); + test('testStepSummary renders actions', () => { assert.strictEqual( testStepSummary({ type: 'action', kind: 'menu-select', params: { item_id: 3 }, @@ -135,6 +162,15 @@ test('testStepSummary renders actions', () => { assert.strictEqual( testStepSummary({ type: 'action', kind: 'status', params: { attempts: 5 } }), 'STATUS x5 (poll)'); + assert.strictEqual( + testStepSummary({ type: 'action', kind: 'proactive', + params: { attempts: 5, respond: { result: 'ok' }, + require: { command: 'REFRESH', qualifier: '00' } } }), + 'PROACTIVE x5 require REFRESH q=00 TR ok'); + assert.strictEqual( + testStepSummary({ type: 'action', kind: 'proactive', + params: { attempts: 3, first: { result: 'cancel' } } }), + 'PROACTIVE x3 TR ok first cancel'); assert.strictEqual( testStepSummary({ type: 'action', kind: 'file-read', params: { path: 'MF/2FE2' } }), 'READ MF/2FE2'); diff --git a/pyproject.toml b/pyproject.toml index 21a7d35..4cb99f6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "pysim-simple-server" -version = "3.20.0" +version = "3.21.0" description = "HTTP REST server wrapping pysim for the SIMple PWA" requires-python = ">=3.8" # pysim is a git-only dependency installed explicitly by setup.bat/setup.sh. diff --git a/pysim_simple_server/server.py b/pysim_simple_server/server.py index 845920a..4c0e79e 100644 --- a/pysim_simple_server/server.py +++ b/pysim_simple_server/server.py @@ -34,7 +34,7 @@ from osmocom.tlv import BER_TLV_IE from osmocom.utils import rpad -VERSION = '3.20.0' +VERSION = '3.21.0' MAX_ENVELOPE_SEGMENTS = 5 # max SMS segments for outgoing C-APDU in ENVELOPE @@ -1910,7 +1910,8 @@ def _counter_probe(server, scc, preset, body, send_fn=None): por_hex = submit_hex por = _decode_por(p['spi1'], p['spi2'], p['kic'], p['kid'], cntr, p['kic_key'], p['kid_key'], por_hex, - cmd_len=len(p['apdu']) // 2) if por_hex else None + cmd_len=len(p['apdu']) // 2, + rm=_tar_is_rm(preset, p['tar'])) if por_hex else None pstatus = str((por or {}).get('response_status') or '') out['packets'] += 1 attempt = {'cntr': cntr, 'por_status': pstatus or None, @@ -2121,7 +2122,9 @@ def _ram_send_gp_apdu(server, scc, sp, state, step_name, apdu_hex, silent=False) por_src = 'sms-submit' por = _decode_por(spi1, spi2, sp['kic'], sp['kid'], state['cntr'], sp['kic_key'], sp['kid_key'], por_hex, - cmd_len=len(apdu_hex) // 2) + cmd_len=len(apdu_hex) // 2, + rm=_tar_is_rm(_preset_by_id(server, state.get('preset_id')), + sp.get('tar'))) step, step_error = _ram_step_result(step_name, last_sw, por, por_hex, result['bytes'], result['segments']) state['last_step'] = step @@ -2262,33 +2265,94 @@ def _sms_submit_por(submit_handler): return '027100' + ud_hex +def _por_header_parse(data, otak, spi): + """Header-only parse of an SMS response packet (TS 102 225 5.2 Table 3). + + Everything `OtaDialectSms.decode_resp` validates *before* the compact + response parse: the UDH/RPI check, the response header, deciphering, the + PCNTR padding and the RC/CC/DS verification. Used when the secured data + is not a compact/scripting RM response (TS 102 226 4: the receiving + application's own format) - a valid packet is then reported raw instead + of "undecodable". Returns the parsed response or None.""" + from pySim.ota import OtaDialectSms + from pySim.sms import UserDataHeader + try: + if not data or data[0] != 0x02: + return None + udhd, remainder = UserDataHeader.from_bytes(data) + if not udhd.has_ie(0x71): + return None + res = OtaDialectSms.SmsResponsePacket.parse(remainder) + if spi['por_shall_be_ciphered']: + deciph = otak.crypt.decrypt(remainder[6:]) + temp_data = remainder[:6] + deciph + res = OtaDialectSms.SmsResponsePacket.parse(temp_data) + if res['pcntr'] != 0: + res['secured_data'] = res['secured_data'][:-res['pcntr']] + remainder = temp_data + len_sig = res['rhl'] - 10 + if spi['por_rc_cc_ds'] == 'no_rc_cc_ds': + if len_sig: + return None + elif spi['por_rc_cc_ds'] == 'cc': + # UDH + RPL/RHL/TAR/CNTR/PCNTR/STS are part of the CC input + udh = data[:3] + header = remainder[:13] + otak.auth.check_sig(udh + header + remainder[13 + len_sig:], + res['cc_rc']) + else: + return None + except Exception: + # a malformed packet or a failed check is not a response packet + return None + return res + + def _decode_por(spi1, spi2, kic, kid, cntr_hex, kic_key_hex, kid_key_hex, - response_hex, cmd_len=None): - """Decode a response packet (PoR). `cmd_len` is the length in bytes of the - plaintext command script the packet carried (None when unknown): the - TS 102 226 5.1.2 compact response's command count cannot exceed it, so a - larger count means the data is the application's own (a third-party - applet's TAR) and is reported as `raw` - no status word is invented.""" - from pySim.ota import OtaDialectSms, CompactRemoteResp - from osmocom.utils import h2b, b2h + response_hex, cmd_len=None, rm=None): + """Decode a response packet (PoR). + + `rm` tells whether the command was addressed to a remote-management TAR + (TS 102 226 4: only those answer with the compact/scripting structures of + 5.1.2/5.2.2): + + - ``True``: decode as an RM response; + - ``False``: any other TAR - the secured data is the receiving + application's own (application specific, "not defined" in the spec) and + is reported raw; no status word is invented; + - ``None`` (unknown): decode, falling back to raw when the compact parse + cannot apply. + + `cmd_len` is the length in bytes of the plaintext command script the + packet carried (None when unknown): the TS 102 226 5.1.2 compact + response's command count cannot exceed it, so a larger count means the + data is the application's own and is reported as `raw`.""" + from pySim.ota import OtaDialectSms + from osmocom.utils import h2b if not response_hex: return None try: otak = _ota_keyset(spi1, spi2, kic, kid, cntr_hex, kic_key_hex, kid_key_hex) spi = _spi_from_bytes(int(spi1, 16), int(spi2, 16)) - except Exception: - # a malformed counter/key (e.g. a hand-made request) is not a PoR - # either - the same rule as the decode below: never raise here. - return None - try: data = h2b(response_hex) - if not data or data[0] != 0x02: - return None - res, dec = OtaDialectSms().decode_resp(otak, spi, data) except Exception: - # any malformed/garbage PoR (non-hex, bad UDL, truncated fields, - # bad CC) is not a POR; never let decoding crash the request handler. + # a malformed counter/key/hex (e.g. a hand-made request) is not a PoR return None + dec = None + if rm is False: + res = _por_header_parse(data, otak, spi) + if res is None: + return None + else: + try: + res, dec = OtaDialectSms().decode_resp(otak, spi, data) + except Exception: + # pySim's decode fails when the secured data is not a compact + # response (e.g. a third-party applet's short reply): fall back + # to the header parse and report the data raw, not lost. + res = _por_header_parse(data, otak, spi) + if res is None: + return None out = { 'response_status': str(res['response_status']), 'tar': res['tar'].hex().upper(), @@ -2315,9 +2379,11 @@ def _decode_por(spi1, spi2, kic, kid, cntr_hex, kic_key_hex, kid_key_hex, # R-APDU(s) of the executed remote command(s) this way instead of the # plain compact response. The R-APDU's own SW and data are the useful # result (a bare CompactRemoteResp parse would read `AB` as the command - # count and produce garbage). + # count and produce garbage). A non-RM TAR's payload is application data + # (rm False): never parse it as an RM structure. if res.response_status == 'por_ok' and len(res['secured_data']): - scripted = _parse_response_scripting(bytes(res['secured_data'])) + scripted = (_parse_response_scripting(bytes(res['secured_data'])) + if rm is not False else None) if scripted is not None: out['response_type'] = 'scripting' out['decoded'] = { @@ -5761,6 +5827,7 @@ _TEST_KIND_LABELS = { 'menu-select': 'ENVELOPE(Menu Selection)', 'file-write': 'UPDATE FILE', 'file-read': 'READ FILE', 'apdu': 'APDU', 'scp80': 'SCP80', 'status': 'STATUS', 'cleanup': 'CLEANUP', + 'proactive': 'PROACTIVE', } # Card-touching endpoints refused while a script owns the card. _TEST_BLOCKED_PATHS = frozenset([ @@ -5813,6 +5880,28 @@ TAR_PROBE_TARS = [ ] TAR_PROBE_DEFAULT_APDU = '00A40000023F00' # SELECT MF: harmless, applet-visible +# The TARs the tool treats as remote-management targets: the preset's role +# entries (checked at call time - a card may use non-standard values) plus +# the ISD/RFM allocations of TS 101 220 V18.3.0 Annex D, Table D.1. A command +# sent to any other TAR goes to a receiving application whose data format is +# application-specific (TS 102 226 4), so its response is reported raw - +# never decoded as an RM response with a fabricated status word. +RM_TAR_ALLOCATIONS = ('000000', 'B20100', 'B00000', 'B00001', 'B00010', + 'B00120', 'B00130', 'B00140') + + +def _tar_is_rm(preset, tar): + """True when `tar` is a remote-management TAR (a preset role entry or a + standard ISD/RFM allocation), False for any other TAR; None when the TAR + is empty (unknown). Drives the response-form decision of `_decode_por`.""" + t = str(tar or '').strip().upper() + if not t: + return None + for entry in (preset or {}).get('tars') or []: + if entry.get('role') and str(entry.get('tar') or '').strip().upper() == t: + return True + return t in RM_TAR_ALLOCATIONS + def _tar_probe_verdict(step): """Classify one TAR probe step: the card's ENVELOPE verdict plus the PoR @@ -5935,7 +6024,8 @@ def _test_finish_entry(entry, result): fields = {'status': result.get('status', 'ok'), 'checks': result.get('checks', []), 'ms': int((time.time() - (entry.get('started') or time.time())) * 1000)} - for key in ('sent', 'sw', 'data', 'note', 'por', 'sms', 'counter', 'command'): + for key in ('sent', 'sw', 'data', 'note', 'por', 'sms', 'counter', + 'command', 'drained'): if result.get(key) is not None: fields[key] = result[key] _test_entry_update(entry, fields) @@ -6029,6 +6119,102 @@ def _test_run_status(scc, step): return data or '', sw, ('STATUS x%d' % used if used > 1 else 'STATUS') +def _proactive_require_match(req, consumed): + """Whether one consumed command satisfies a proactive action's `require` + spec: the command type/name (ANY matches all) and an optional qualifier + (exact/mask).""" + if req.get('type') is not None and consumed.get('type_hex') != '%02X' % req['type']: + return False + q = req.get('qualifier') + if q and not testscript.match_value(q, consumed.get('qualifier')): + return False + return True + + +def _proactive_require_label(req): + """The `require` spec as the report's expected string.""" + if req.get('type') is not None: + name = req.get('name') or '0x%02X' % req['type'] + else: + name = 'ANY' + q = req.get('qualifier') + return name + ((' q=%s' % q['value']) if q else '') + + +def _test_run_proactive(server, step): + """The proactive cleanup action: poll STATUS and consume every announced + command - the TERMINAL RESPONSE is `first` for the first one, `respond` + for the rest - until the card answers 9000 (idle), a non-91XX SW appears + or the attempt budget is spent. The final SW is the step's SW check; an + optional `require` asserts what was drained (at least one match; an empty + drain fails it).""" + scc = server.scc + p = step['params'] + consumed = [] + sw = '' + polls = 0 + for n in range(p['attempts']): + data, sw = _send_status(scc) + polls = n + 1 + _test_log('STATUS %d/%d -> %s' % (polls, p['attempts'], sw)) + if not (sw or '').startswith('91'): + break + fetch_apdu = '%s120000%02x' % (scc.cat_cla, int(sw[2:], 16)) + fdata, fetch_sw = scc._tp.send_apdu(fetch_apdu) + if not fdata: + _test_log('FETCH=%s -> SW=%s (no data)' % (fetch_apdu, fetch_sw)) + sw = fetch_sw + break + raw = bytes.fromhex(fdata) + _test_log('FETCH=%s -> SW=%s %s' % (fetch_apdu, fetch_sw, raw.hex().upper())) + cmd_num, cmd_type, dev_src, dev_dst, cmd_qual = _parse_proactive_header(raw) + name = PROACTIVE_TYPE_NAMES.get(cmd_type, 'UNKNOWN') + _test_log('CMD 0x%02X %s qual=%s' % ( + cmd_type, name, ('%02X' % cmd_qual) if cmd_qual is not None else '-')) + if cmd_type == 0x25: + # keep the cached menu (text-based selection) current + menu = _parse_setup_menu_command(raw) + if menu: + server.sim_menu = menu + respond = p['first'] if (not consumed and p.get('first')) else p['respond'] + tr = testscript.build_tr(cmd_num, cmd_type, dev_dst, dev_src, respond) + tr_rv = scc._tp.send_apdu('%s140000%02x%s' % (scc.cat_cla, len(tr), tr.hex())) + _test_log('TR=%s -> SW=%s' % (tr.hex().upper(), tr_rv[1])) + consumed.append({'type_hex': '%02X' % cmd_type, 'type_name': name, + 'qualifier': ('%02X' % cmd_qual) if cmd_qual is not None else None, + 'tr_sw': tr_rv[1], 'raw': raw.hex().upper()}) + sw = tr_rv[1] + if not (sw or '').startswith('91') and sw != '9000': + # a TERMINAL RESPONSE the card rejected: stop and let the SW + # check name it + break + if n + 1 < p['attempts'] and p['interval_ms']: + time.sleep(p['interval_ms'] / 1000.0) + checks = [_test_check_result('SW', testscript.match_value(step['check']['sw'], sw), + step['check']['sw']['value'], sw or '(none)', + step['on_fail'])] + req = p.get('require') + if req: + matched = any(_proactive_require_match(req, c) for c in consumed) + if consumed: + actual = ', '.join( + '%s%s' % (c['type_name'], + (' q=%s' % c['qualifier']) if c['qualifier'] else '') + for c in consumed) + else: + actual = '(nothing was pending)' + checks.append(_test_check_result('Require', matched, + _proactive_require_label(req), actual, + step['on_fail'])) + status = testscript.combine_levels([c['level'] if not c['ok'] else 'ok' + for c in checks]) + result = {'status': status, 'sw': sw or '', 'data': '', + 'sent': 'STATUS x%d; TR x%d' % (polls, len(consumed)), + 'checks': checks, 'drained': consumed} + pending = int(sw[2:], 16) if (sw or '').startswith('91') else None + return result, pending + + def _test_run_file(server, step): """UPDATE/READ a file by path (extends the network simulator's writes to arbitrary files); returns (data, sw, sent).""" @@ -6107,7 +6293,8 @@ def _test_run_scp80(server, step, ctx): cmd_len = len(p.get('apdu') or '') // 2 or None ctx['last_scp80'] = {'spi1': spi1, 'spi2': spi2, 'kic': kic, 'kid': kid, 'counter': counter, 'kicKey': kic_key, 'kidKey': kid_key, - 'tar': tar, 'cmd_len': cmd_len} + 'tar': tar, 'cmd_len': cmd_len, + 'rm': _tar_is_rm(preset, tar)} fmt = p.get('format') or 'compact' apdu_hex = p.get('apdu') or '' if p.get('sp'): @@ -6139,7 +6326,7 @@ def _test_run_scp80(server, step, ctx): por = None if data: por = _decode_por(spi1, spi2, kic, kid, counter, kic_key, kid_key, data, - cmd_len=cmd_len) + cmd_len=cmd_len, rm=_tar_is_rm(preset, tar)) if por: dec = por.get('decoded') or {} extra = '' @@ -6273,6 +6460,8 @@ def _test_run_action(server, step, ctx): (' RESP=%dB' % (len(data) // 2)) if data else '')) elif kind == 'scp80': data, sw, sent, por, counter = _test_run_scp80(server, step, ctx) + elif kind == 'proactive': + return _test_run_proactive(server, step) else: raise testscript.ScriptError('unknown action kind %r' % kind) checks = _test_action_checks(step, sw, data, por, kind) @@ -6375,21 +6564,37 @@ def _test_expect_por(ctx, raw): por_hex = '027100' + ud_hex return _decode_por(last['spi1'], last['spi2'], last['kic'], last['kid'], last['counter'], last['kicKey'], last['kidKey'], por_hex, - cmd_len=last.get('cmd_len')) + cmd_len=last.get('cmd_len'), rm=last.get('rm')) def _test_run_expect(server, step, pending, ctx=None): scc = server.scc if not pending: - raise testscript.ScriptError( - 'no proactive command pending - the previous step must end with SW 91XX ' - '(or add a status action); the UICC announces pending commands in the ' - 'response to a command (TS 102 221 7.4.2.1)') + # Nothing to fetch: the step fails at its own on_fail level. The + # tolerant "consume a pending command if there is one" form is an + # expectation with on_fail "warning" (or the proactive cleanup action, + # which tolerates an empty card by design). The UICC announces + # pending commands in the response to a command (TS 102 221 7.4.2.1). + want = step['command'].get('name') + if not want and step['command'].get('type') is not None: + want = '0x%02X' % step['command']['type'] + return {'status': step['on_fail'], 'checks': [_test_check_result( + 'Command', False, want or 'a pending command', '(none)', + step['on_fail'], + detail='no proactive command pending - the previous step ' + 'did not end with SW 91XX (the card announces ' + 'pending commands in the response to a command, ' + 'TS 102 221 7.4.2.1); a status action with ' + 'attempts > 1 polls until one appears')]}, None fetch_apdu = '%s120000%02x' % (scc.cat_cla, pending) fdata, fetch_sw = scc._tp.send_apdu(fetch_apdu) if not fdata: _test_log('FETCH=%s -> SW=%s (no data)' % (fetch_apdu, fetch_sw)) - raise testscript.ScriptError('FETCH returned no data (SW %s)' % fetch_sw) + return {'status': step['on_fail'], 'checks': [_test_check_result( + 'Fetch', False, 'the announced command', + 'no data (SW %s)' % fetch_sw, step['on_fail'], + detail='the card announced a command (SW 91XX) but FETCH ' + 'returned no data')]}, None raw = bytes.fromhex(fdata) _test_log('FETCH=%s -> SW=%s %s' % (fetch_apdu, fetch_sw, raw.hex().upper())) cmd_num, cmd_type, dev_src, dev_dst, cmd_qual = _parse_proactive_header(raw) @@ -6533,23 +6738,50 @@ def _test_run_expect(server, step, pending, ctx=None): return result, pending_next -def _test_drain_pending(scc, fetch_len): - """Fetch an announced command the script did not expect and answer it - with a cancel TR, so the card is not left re-announcing it (6.3).""" +def _test_drain_pending(scc, fetch_len, result=0x00, max_commands=3): + """Fetch the commands the script did not expect and answer them, so the + card is not left re-announcing one (TS 102 223 6.3). + + The answer defaults to 'ok' (0x00, command performed successfully): a + cancel (0x10) tells the card the user terminated the proactive session, + which makes assertive applications take an error path - a REFRESH + answered cancel can queue a further command. Answering one command can + make the application emit another, so the drain keeps fetching while the + TERMINAL RESPONSE answers 91XX (bounded by `max_commands`). A script that + wants to test the refusal path answers explicitly (an expectation's + `respond`, or the proactive action's `first`).""" + answered = [] try: - fdata, sw = scc._tp.send_apdu('%s120000%02x' % (scc.cat_cla, fetch_len or 0x100)) - if not fdata: - return {'note': 'nothing to drain (FETCH SW %s)' % sw} - raw = bytes.fromhex(fdata) - cmd_num, cmd_type, dev_src, dev_dst, cmd_qual = _parse_proactive_header(raw) - tr = testscript.build_tr(cmd_num, cmd_type, dev_dst, dev_src, {'result': 0x10}) - rv = scc._tp.send_apdu('%s140000%02x%s' % (scc.cat_cla, len(tr), tr.hex())) - return {'sent': 'TR cancel ' + tr.hex().upper(), 'sw': rv[1], - 'data': raw.hex().upper(), - 'note': 'pending %s (0x%02X) answered with cancel' - % (PROACTIVE_TYPE_NAMES.get(cmd_type, '?'), cmd_type)} + while len(answered) < max_commands: + fdata, sw = scc._tp.send_apdu('%s120000%02x' % (scc.cat_cla, + fetch_len or 0x100)) + if not fdata: + break + raw = bytes.fromhex(fdata) + cmd_num, cmd_type, dev_src, dev_dst, cmd_qual = _parse_proactive_header(raw) + tr = testscript.build_tr(cmd_num, cmd_type, dev_dst, dev_src, {'result': result}) + rv = scc._tp.send_apdu('%s140000%02x%s' % (scc.cat_cla, len(tr), tr.hex())) + name = PROACTIVE_TYPE_NAMES.get(cmd_type, '?') + answered.append({'name': name, 'type': cmd_type, + 'raw': raw.hex().upper(), 'tr': tr.hex().upper(), + 'sw': rv[1]}) + sys.stderr.write('TEST-RUN drain: %s (0x%02X) TR=%s -> SW=%s\n' + % (name, cmd_type, tr.hex().upper(), rv[1])) + if not (rv[1] or '').startswith('91'): + break + fetch_len = int(rv[1][2:], 16) except Exception as e: return {'note': 'drain failed: %s' % e} + if not answered: + return {'note': 'nothing to drain'} + result_name = {0x00: 'ok', 0x10: 'cancel', 0x11: 'back', + 0x12: 'timeout', 0x22: 'no response'}.get(result, + '0x%02X' % result) + return {'sent': '; '.join('TR %s %s' % (result_name, a['tr']) for a in answered), + 'sw': answered[-1]['sw'], 'data': answered[0]['raw'], + 'note': '; '.join('pending %s (0x%02X) answered with %s' + % (a['name'], a['type'], result_name) + for a in answered)} def _test_run_execute(server, script, preset): @@ -7903,6 +8135,12 @@ class PysimHandler(BaseHTTPRequestHandler): # no counter is advanced or persisted. counter_tracked = _counter_tracked(spi1) include_cpi = body.get('includeCpi', True) + # The response form follows the command's TAR (TS 102 226 4): only + # a remote-management TAR answers with the RM structures; a hand + # send may name any TAR (the pre-built `sp` path may name none - + # unknown, decode with the structural fallback). + rm = _tar_is_rm(_preset_by_id(self.server, body.get('preset_id')), + body.get('tar') or ('000000' if apdu else '')) ram_format = None try: if apdu: @@ -7986,7 +8224,8 @@ class PysimHandler(BaseHTTPRequestHandler): body.get('kic', ''), body.get('kid', ''), cntr, body.get('kicKey', ''), body.get('kidKey', ''), failed_por_hex, - cmd_len=len(body.get('apdu') or '') // 2 or None) + cmd_len=len(body.get('apdu') or '') // 2 or None, + rm=rm) if failed_por: resp['por'] = failed_por low = _cntr_low_fields(failed_por.get('response_status')) @@ -8024,7 +8263,8 @@ class PysimHandler(BaseHTTPRequestHandler): por = _decode_por(spi1, spi2, body.get('kic', ''), body.get('kid', ''), cntr, body.get('kicKey', ''), body.get('kidKey', ''), por_hex, - cmd_len=len(body.get('apdu') or '') // 2 or None) + cmd_len=len(body.get('apdu') or '') // 2 or None, + rm=rm) # Check for SPI2=0x21 (PoR required) but got 9000 with no PoR → card refuses PoR por_required = bool(spi2_val & 0x01) no_por_received = not por_hex and not (submit_handler and submit_handler.submit_tpdu_hex) diff --git a/pysim_simple_server/testscript.py b/pysim_simple_server/testscript.py index 4aaeb0d..f32c777 100644 --- a/pysim_simple_server/testscript.py +++ b/pysim_simple_server/testscript.py @@ -33,7 +33,7 @@ __all__ = [ ] ACTION_KINDS = ('envelope', 'event', 'menu-select', 'file-write', 'file-read', - 'apdu', 'scp80', 'status') + 'apdu', 'scp80', 'status', 'proactive') FAIL_LEVELS = ('error', 'warning') POR_CHECKS = ('none', 'ok', 'any') @@ -207,17 +207,17 @@ def normalise_step(step, command_resolver=None): raise ScriptError('each step must be an object') typ = step.get('type') if typ == 'action': - return _normalise_action(step) + return _normalise_action(step, command_resolver) if typ == 'expect': return _normalise_expect(step, command_resolver) raise ScriptError("step type must be 'action' or 'expect'") -def _normalise_action(step): +def _normalise_action(step, command_resolver=None): kind = str(step.get('kind') or '').lower() if kind not in ACTION_KINDS: raise ScriptError("unknown action kind %r" % step.get('kind')) - params = _normalise_params(kind, step.get('params') or {}) + params = _normalise_params(kind, step.get('params') or {}, command_resolver) on_fail = _fail_level(step.get('on_fail')) check = _normalise_check(step.get('check'), kind, params) out = {'type': 'action', 'kind': kind, 'params': params, @@ -227,7 +227,7 @@ def _normalise_action(step): return out -def _normalise_params(kind, p): +def _normalise_params(kind, p, command_resolver=None): if not isinstance(p, dict): raise ScriptError('%s: params must be an object' % kind) if kind == 'envelope': @@ -346,6 +346,30 @@ def _normalise_params(kind, p): interval = p.get('interval_ms') interval = _int(200 if interval is None else interval, 'status interval_ms', 0, 10000) return {'attempts': attempts, 'interval_ms': interval} + if kind == 'proactive': + # The proactive cleanup: consume whatever the card announces (the + # TERMINAL RESPONSE is `first` for the first command, `respond` for + # the rest) and confirm the card is idle; `require` asserts what was + # drained (at least one match - an empty drain fails it). + out = {'respond': normalise_respond(p.get('respond') or {}), + 'attempts': _int(3 if p.get('attempts') is None else p.get('attempts'), + 'proactive attempts', 1, 1000), + 'interval_ms': _int(200 if p.get('interval_ms') is None + else p.get('interval_ms'), + 'proactive interval_ms', 0, 10000)} + if p.get('first') is not None: + out['first'] = normalise_respond(p['first']) + req = p.get('require') + if req is not None: + if not isinstance(req, dict): + raise ScriptError('proactive require must be an object') + ctype, cname = _resolve_command(req.get('command'), 'proactive require', + command_resolver) + qualifier = _check_spec(req.get('qualifier'), 'proactive require.qualifier') + if qualifier and '?' not in qualifier['value'] and len(qualifier['value']) != 2: + raise ScriptError('proactive require.qualifier must be one byte') + out['require'] = {'type': ctype, 'name': cname, 'qualifier': qualifier} + return out raise ScriptError('unknown action kind %r' % kind) @@ -381,26 +405,30 @@ def _normalise_check(check, kind, params): return {'sw': sw, 'data': data, 'por': por} -def _normalise_expect(step, command_resolver=None): - cmd = step.get('command') +def _resolve_command(cmd, what, command_resolver=None): + """A proactive command spec: a name (resolved to a type code through + ``command_resolver``), a hex type code or ANY. Returns (type|None, + name|None).""" if cmd is None or isinstance(cmd, bool): - raise ScriptError('expect: command is required') + raise ScriptError('%s: command is required' % what) if isinstance(cmd, int): - ctype, cname = cmd, None - else: - s = str(cmd).strip() - up = s.upper() - if up in ('ANY', '*'): - ctype, cname = None, 'ANY' - elif re.fullmatch(r'(0X)?[0-9A-F]{2}', up): - ctype, cname = int(up.replace('0X', ''), 16), None - elif command_resolver is not None: - ctype = command_resolver(up) - if ctype is None: - raise ScriptError('expect: unknown proactive command %r' % s) - cname = up - else: - raise ScriptError('expect: command must be a hex type code') + return cmd, None + s = str(cmd).strip() + up = s.upper() + if up in ('ANY', '*'): + return None, 'ANY' + if re.fullmatch(r'(0X)?[0-9A-F]{2}', up): + return int(up.replace('0X', ''), 16), None + if command_resolver is not None: + ctype = command_resolver(up) + if ctype is None: + raise ScriptError('%s: unknown proactive command %r' % (what, s)) + return ctype, up + raise ScriptError('%s: command must be a hex type code' % what) + + +def _normalise_expect(step, command_resolver=None): + ctype, cname = _resolve_command(step.get('command'), 'expect', command_resolver) qualifier = _check_spec(step.get('qualifier'), 'qualifier') if qualifier and '?' not in qualifier['value'] and len(qualifier['value']) != 2: raise ScriptError('qualifier must be one byte') diff --git a/tests/test_ota_helpers.py b/tests/test_ota_helpers.py index 316e3da..deb375a 100644 --- a/tests/test_ota_helpers.py +++ b/tests/test_ota_helpers.py @@ -25,6 +25,7 @@ from pysim_simple_server.server import ( _build_tr, _decode_cmd, _decode_por, + _tar_is_rm, _decode_tr, _log_proactive, _ota_reference, @@ -535,6 +536,59 @@ class TestDecodePor(unittest.TestCase): self.assertEqual(r['decoded']['last_status_word'], '6a86') self.assertEqual(r['secured_data'], '016A86') + def test_non_rm_tar_is_application_data(self): + # v3.21.0: the response form follows the command's TAR. A non-RM TAR + # answers with application data (TS 102 226 4); `01 00 0B 91 ...` + # happens to parse as a compact response (count 1, SW 000B) and must + # not be split into a fabricated status word (live 2026-10-05). + raw = '027100000e0aaf4d010000000001000001000B919733525088F4' + r = _decode_por('16', '01', '15', '15', '0000000001', K, K, raw, rm=False) + self.assertEqual(r['response_type'], 'raw') + self.assertEqual(r['response_status'], 'por_ok') + self.assertEqual(r['secured_data'], '01000B919733525088F4') + self.assertEqual(r['decoded']['last_response_data'], '01000B919733525088F4') + self.assertEqual(r['decoded']['last_status_word'], '') + self.assertIsNone(r['decoded']['number_of_commands']) + # without the TAR context the same bytes still parse as compact + r = _decode_por('16', '01', '15', '15', '0000000001', K, K, raw) + self.assertEqual(r['response_type'], 'compact') + self.assertEqual(r['decoded']['last_status_word'], '000b') + + def test_short_applet_reply_falls_back_to_raw(self): + # a 2-byte secured data cannot be a compact response (TS 102 226 5.1.2 + # Table 5.1 needs a count plus two status bytes): pySim's decode raises + # and the packet is reported raw instead of "undecodable" + raw = '027100000e0aaf4d01000000000100000400' + r = _decode_por('16', '01', '15', '15', '0000000001', K, K, raw) + self.assertEqual(r['response_type'], 'raw') + self.assertEqual(r['response_status'], 'por_ok') + self.assertEqual(r['secured_data'], '0400') + self.assertEqual(r['decoded']['last_response_data'], '0400') + + +class TestTarIsRm(unittest.TestCase): + """v3.21.0: the response form follows the command's TAR - the preset's + role entries plus the standard ISD/RFM allocations of TS 101 220 Annex D.""" + + def test_standard_allocations(self): + for tar in ('000000', 'B00000', 'B00001', 'B00010', 'B20100', + 'B00120', 'B00130', 'B00140'): + self.assertTrue(_tar_is_rm(None, tar), tar) + + def test_application_allocations_are_not_rm(self): + # the Annex D application TARs (USAT interpreter, multiplexing, CASD, + # OMA) and any custom TAR: application-specific data (TS 102 226 4) + for tar in ('AF4D01', 'B20000', 'B20200', 'B20201', 'B20202', + 'B20203', 'B00200'): + self.assertFalse(_tar_is_rm(None, tar), tar) + + def test_preset_roles_win(self): + preset = {'tars': [{'role': 'isd', 'tar': 'A11300', 'msl': '16'}, + {'role': '', 'tar': 'AF4D01', 'msl': '16'}]} + self.assertTrue(_tar_is_rm(preset, 'A11300')) + self.assertFalse(_tar_is_rm(preset, 'AF4D01')) + self.assertIsNone(_tar_is_rm(preset, '')) + class TestPresetCounterSeed(unittest.TestCase): """v3.20.0: the effective counter of an operation - explicit `cntr` wins, diff --git a/tests/test_testscript.py b/tests/test_testscript.py index 8f203ed..c7eed08 100644 --- a/tests/test_testscript.py +++ b/tests/test_testscript.py @@ -85,6 +85,39 @@ class TestValidation(unittest.TestCase): self.assertEqual(check['sw'], {'mode': 'mask', 'value': '91??'}) self.assertEqual(check['data'], {'mode': 'mask', 'value': 'AA??'}) + def test_proactive_action_normalises(self): + script = T.normalise_script({'steps': [ + {'type': 'action', 'kind': 'proactive', + 'params': {'respond': {'result': 'ok'}, 'first': {'result': 'cancel'}, + 'attempts': 5, 'interval_ms': 0, + 'require': {'command': 'REFRESH', 'qualifier': '00'}}}, + ]}, _resolver) + p = script['steps'][0]['params'] + self.assertEqual(p['respond'], {'result': 0x00}) + self.assertEqual(p['first'], {'result': 0x10}) + self.assertEqual(p['attempts'], 5) + self.assertEqual(p['interval_ms'], 0) + self.assertEqual(p['require']['type'], 0x01) + self.assertEqual(p['require']['name'], 'REFRESH') + self.assertEqual(p['require']['qualifier'], {'mode': 'exact', 'value': '00'}) + # the defaults: ok / 3 attempts / 200 ms, no first, no require + script = T.normalise_script({'steps': [ + {'type': 'action', 'kind': 'proactive'}]}, _resolver) + p = script['steps'][0]['params'] + self.assertEqual(p['respond'], {'result': 0x00}) + self.assertNotIn('first', p) + self.assertNotIn('require', p) + self.assertEqual(p['attempts'], 3) + self.assertEqual(p['interval_ms'], 200) + # a bad require is refused + for bad in ({'require': {'qualifier': '00'}}, + {'require': 'REFRESH'}, + {'require': {'command': 'NOT A COMMAND'}}, + {'attempts': 0}): + with self.assertRaises(T.ScriptError, msg=repr(bad)): + T.normalise_script({'steps': [{'type': 'action', 'kind': 'proactive', + 'params': bad}]}, _resolver) + def test_integer_fields_accept_decimals_with_leading_zeros_and_hex(self): script = T.normalise_script({'steps': [ {'type': 'action', 'kind': 'status', @@ -440,7 +473,29 @@ class TestRunnerDialogue(RunnerTestCase): 'checks': [{'kind': 'text', 'value': 'hello'}], 'respond': {}}, ]) self.assertEqual(run['status'], 'error') - self.assertIn('no proactive command pending', run['steps'][0]['note']) + self.assertEqual(run['steps'][0]['status'], 'error') + check = run['steps'][0]['checks'][0] + self.assertEqual(check['label'], 'Command') + self.assertFalse(check['ok']) + self.assertEqual(check['expected'], 'DISPLAY TEXT') + self.assertEqual(check['actual'], '(none)') + self.assertIn('no proactive command pending', check['detail']) + + def test_expectation_warning_without_pending_continues(self): + # on_fail "warning" is the tolerant consume-if-pending form (v3.21.0): + # the empty poll warns and the script continues + scc = FakeScc() + scc.push('80F2', '', '9000') + run = self.run_script(FakeServer(scc), [ + {'type': 'expect', 'command': 'ANY', 'respond': {}, + 'on_fail': 'warning'}, + {'type': 'action', 'kind': 'status', 'params': {}}, + ]) + self.assertEqual(run['status'], 'warning') + self.assertEqual(len(run['steps']), 2) + self.assertEqual(run['steps'][0]['status'], 'warning') + self.assertIn('no proactive command pending', + run['steps'][0]['checks'][0]['detail']) def test_text_mismatch_fails_the_expectation(self): scc = FakeScc() @@ -471,9 +526,31 @@ class TestRunnerDialogue(RunnerTestCase): ]) self.assertEqual(run['status'], 'error') self.assertIn('unexpected proactive command pending', run['steps'][1]['note']) - # the drain FETCHed the command and answered with a cancel TR (0x10) + # the drain FETCHed the command and answered with an ok TR (0x00, + # v3.21.0: a cancel told the card the user aborted the session) self.assertTrue(any(a.startswith('8012') for a in scc.sent[1:])) - self.assertTrue(any('83021000' in a for a in scc.sent)) + self.assertTrue(any('83020000' in a for a in scc.sent)) + self.assertIn('TR ok', run['steps'][1]['sent']) + + def test_drain_answers_ok_and_follows_the_next_command(self): + # answering one command can make the application emit another; the + # drain keeps fetching while the TR answers 91XX (bounded, v3.21.0) + scc = FakeScc() + scc.push('80C2', '', '9103') # menu selection + scc.push('8012', SELECT_ITEM_CMD, '9000') # FETCH #1 + scc.push('8014', '', '9102') # TR -> another pending + scc.push('8012', DISPLAY_TEXT_CMD, '9000') # FETCH #2 + scc.push('8014', '', '9000') # TR -> idle + run = self.run_script(FakeServer(scc), [ + {'type': 'action', 'kind': 'menu-select', 'params': {'item_id': 1}, + 'check': {'sw': '91??'}}, + {'type': 'action', 'kind': 'apdu', 'params': {'apdu': '00A40000023F00'}}, + ]) + self.assertEqual(run['status'], 'error') # the unexpected pending stops the script + entry = run['steps'][1] + self.assertEqual(entry['sent'].count('TR ok'), 2) + self.assertEqual(entry['sw'], '9000') + self.assertIn(SELECT_ITEM_CMD, entry['data']) def test_sw_mismatch_error_stops_the_script_warning_continues(self): scc = FakeScc() @@ -532,6 +609,117 @@ class TestRunnerDialogue(RunnerTestCase): self.assertEqual(run['steps'], []) +class TestRunnerProactive(RunnerTestCase): + """The proactive cleanup action (v3.21.0): consume whatever the card + announces, answer it and confirm the card is idle.""" + + def test_empty_drain_passes(self): + # the point of the action: an empty first poll is fine + scc = FakeScc() + scc.push('80F2', '', '9000') + run = self.run_script(FakeServer(scc), [ + {'type': 'action', 'kind': 'proactive', 'params': {}}, + ]) + self.assertEqual(run['status'], 'ok', run['steps']) + entry = run['steps'][0] + self.assertEqual(entry['status'], 'ok') + self.assertEqual(entry['sent'], 'STATUS x1; TR x0') + self.assertEqual(entry['drained'], []) + self.assertEqual(entry['checks'][0]['label'], 'SW') + self.assertTrue(entry['checks'][0]['ok']) + + def test_drains_a_pending_command_and_confirms_idle(self): + scc = FakeScc() + scc.push('80F2', '', '9102') # STATUS: DISPLAY TEXT pending + scc.push('8012', DISPLAY_TEXT_CMD, '9000') + scc.push('8014', '', '9000') # TR ok + scc.push('80F2', '', '9000') # confirming STATUS: idle + run = self.run_script(FakeServer(scc), [ + {'type': 'action', 'kind': 'proactive', 'params': {'attempts': 3, 'interval_ms': 0}}, + ]) + self.assertEqual(run['status'], 'ok', run['steps']) + entry = run['steps'][0] + self.assertEqual(entry['sent'], 'STATUS x2; TR x1') + self.assertEqual(len(entry['drained']), 1) + self.assertEqual(entry['drained'][0]['type_name'], 'DISPLAY TEXT') + self.assertEqual(entry['drained'][0]['qualifier'], '00') + self.assertTrue(any('83020000' in a for a in scc.sent)) + + def test_require_matches_a_drained_command(self): + scc = FakeScc() + scc.push('80F2', '', '9102') + scc.push('8012', DISPLAY_TEXT_CMD, '9000') + scc.push('8014', '', '9000') + scc.push('80F2', '', '9000') + run = self.run_script(FakeServer(scc), [ + {'type': 'action', 'kind': 'proactive', + 'params': {'attempts': 3, 'interval_ms': 0, 'require': {'command': 'DISPLAY TEXT'}}}, + ]) + self.assertEqual(run['status'], 'ok', run['steps']) + req = [c for c in run['steps'][0]['checks'] if c['label'] == 'Require'][0] + self.assertTrue(req['ok']) + + def test_require_fails_on_a_mismatch(self): + scc = FakeScc() + scc.push('80F2', '', '9102') + scc.push('8012', DISPLAY_TEXT_CMD, '9000') + scc.push('8014', '', '9000') + scc.push('80F2', '', '9000') + run = self.run_script(FakeServer(scc), [ + {'type': 'action', 'kind': 'proactive', + 'params': {'attempts': 3, 'interval_ms': 0, + 'require': {'command': 'REFRESH', 'qualifier': '00'}}}, + ]) + self.assertEqual(run['status'], 'error') + req = [c for c in run['steps'][0]['checks'] if c['label'] == 'Require'][0] + self.assertFalse(req['ok']) + self.assertIn('DISPLAY TEXT', req['actual']) + + def test_require_fails_on_an_empty_drain(self): + scc = FakeScc() + scc.push('80F2', '', '9000') + run = self.run_script(FakeServer(scc), [ + {'type': 'action', 'kind': 'proactive', + 'params': {'require': {'command': 'REFRESH'}}}, + ]) + self.assertEqual(run['status'], 'error') + req = [c for c in run['steps'][0]['checks'] if c['label'] == 'Require'][0] + self.assertEqual(req['actual'], '(nothing was pending)') + + def test_first_uses_a_different_result(self): + scc = FakeScc() + scc.push('80F2', '', '9102') # STATUS: DISPLAY TEXT pending + scc.push('8012', DISPLAY_TEXT_CMD, '9000') + scc.push('8014', '', '9102') # TR cancel -> another pending + scc.push('80F2', '', '9102') # STATUS: SELECT ITEM pending + scc.push('8012', SELECT_ITEM_CMD, '9000') + scc.push('8014', '', '9000') # TR ok + scc.push('80F2', '', '9000') # confirming STATUS + run = self.run_script(FakeServer(scc), [ + {'type': 'action', 'kind': 'proactive', + 'params': {'attempts': 4, 'interval_ms': 0, 'respond': {'result': 'ok'}, + 'first': {'result': 'cancel'}}}, + ]) + self.assertEqual(run['status'], 'ok', run['steps']) + self.assertEqual(len(run['steps'][0]['drained']), 2) + self.assertTrue(any('83021000' in a for a in scc.sent)) # the first TR + self.assertTrue(any('83020000' in a for a in scc.sent)) # the second + + def test_bound_leaves_the_pending_for_the_next_step(self): + scc = FakeScc() + scc.push('80F2', '', '9102') + scc.push('8012', DISPLAY_TEXT_CMD, '9000') + scc.push('8014', '', '9102') # another command pending + run = self.run_script(FakeServer(scc), [ + {'type': 'action', 'kind': 'proactive', 'params': {'attempts': 1}}, + ]) + # the bound stopped with 91XX: the SW check fails and the pending is + # handed to the next step + self.assertEqual(run['status'], 'error') + self.assertEqual(run['steps'][0]['sw'], '9102') + self.assertFalse(run['steps'][0]['checks'][0]['ok']) + + class TestRunnerActions(RunnerTestCase): def test_script_menu_select_mirrors_menu_active(self): # 9000 -> the menu dialogue is over