diff --git a/docs/api.md b/docs/api.md index 5df9fd1..c42459c 100644 --- a/docs/api.md +++ b/docs/api.md @@ -39,6 +39,10 @@ a 3.x PWA). | `/api/send-ota` | POST | SCP80 OTA secured packet delivery | | `/api/ram-install` | POST | Install a Java Card `.cap` file via SCP80 (INSTALL[for load] → LOAD ×N → INSTALL[for install]) | | `/api/cap-info` | POST | Validate a `.cap` archive and estimate its code/NVRAM/RAM requirements (read-only) | +| `/api/test/run` | POST | Start a test script (actions + proactive expectations) | +| `/api/test/status` | GET | Test script run state and per-step results | +| `/api/test/stop` | POST | Request a running test script to stop | +| `/api/test/clear` | POST | Clear the finished run report | | `/api/sp-verify` | POST | Verify secured packet against pySim reference | | `/api/menu` | GET | Current STK menu (title + items + active) | | `/api/menu-select` | POST | ENVELOPE(Menu Selection) with item_id | @@ -362,6 +366,82 @@ the structural parse but not the analyzer. The numbers are an estimate: the model assumes 2-byte references, a 6-byte object header and NVM cell rounding, and does not include applet-created runtime objects/arrays. +### `POST /api/test/run` + +Runs a **test script**: an ordered list of actions and proactive-command +expectations, executed server-side on the equipped card. While a run is +active the card is owned by the script - other card endpoints answer +`409 {"error": "test script running ..."}` and background STATUS polling is +suspended; only `/api/test/*`, `/api/status`, `/api/poll-status`, +`/api/version` and static files stay available. + +```json +{"script": {"name": "STK menu browsing", "steps": [ + {"type": "action", "kind": "menu-select", "params": {"item_id": 128}, + "check": {"sw": "91??"}}, + {"type": "expect", "command": "SELECT ITEM", + "checks": [{"kind": "item", "id": 1, "text": "test"}], + "respond": {"result": "ok", "item_id": 1}}, + {"type": "expect", "command": "DISPLAY TEXT", + "checks": [{"kind": "text", "value": "hello"}], + "respond": {"result": "ok"}} + ]}, + "preset": {"name": "lab card", "kic": "15", "kid": "15", "kicKey": "...", + "kidKey": "...", "counter": "0000000A", "tar": "B00000", + "spi1": "16", "spi2": "01"}} +``` + +**Action steps** (`type: "action"`): `kind` is `envelope` (`event`, `data`), +`menu-select` (`item_id` 1-255), `file-write` (`path`, `data`, `mode` +`auto`/`binary`/`record`, `record`), `file-read` (same, verifies `check.data`), +`apdu` (raw transport, no auto-handler), `scp80` (`apdu` or `sp`, optional +`tar`/`spi1`/`spi2` overrides - KIc/KID and the counter always come from the +`preset`, which must match the equipped card and be complete) or `status` +(`attempts`, `interval_ms` - when `attempts > 1` the default SW check is the +mask `91??`, i.e. poll until the card announces a command). + +`check` is `{"sw": ..., "data": ...}` (exact or `{"mode": "mask", "value": +"91??"}`, `?` = per-nibble wildcard) plus `"por": "none"|"ok"|"any"` for +SCP80. `on_fail` is `error` (terminates the script) or `warning` (continues). + +**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` +may be `text` (contains/exact), `item` (`id`/`text` for SELECT ITEM / SET UP +MENU) or `raw` (mask); `respond` is the TERMINAL RESPONSE (`result` name or +value, `item_id`, `text`+`dcs`, raw TLVs). + +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. + +### `GET /api/test/status` + +The current (or last) run: + +```json +{"running": true, "name": "STK menu browsing", "status": null, + "index": 1, "total": 3, "scp80_counter": null, + "steps": [{"index": 0, "type": "action", "label": "ENVELOPE(Menu Selection)", + "status": "ok", "sw": "9103", "sent": "80C2000009...", + "checks": [{"label": "SW", "ok": true, "expected": "91??", + "actual": "9103", "level": "error"}], "ms": 4}]} +``` + +`status` becomes `ok`/`warning`/`error`/`stopped` when the run finishes; +expected/actual pairs are reported per check. + +### `POST /api/test/stop` + +Requests a stop (`{"stop": true}` is set on the run); the runner finishes the +current step, answers any pending proactive command with a cancel TERMINAL +RESPONSE and reports the run as `stopped`. + +### `POST /api/test/clear` + +Clears a finished run report (409 while a run is active). + ### `POST /api/ram-install` Install a Java Card `.cap` file on the card via GlobalPlatform commands (INSTALL[for load] → LOAD ×N → INSTALL[for install (+ make selectable)]) wrapped in SCP80 secured packets. Each step is sent via ENVELOPE and its PoR is checked; the sequence aborts on the first PoR error. The `.cap` archive (a ZIP of nested components) is parsed server-side in `_cap_parse`; no external tooling is required. diff --git a/pysim_simple_server/server.py b/pysim_simple_server/server.py index 12c4c38..c889eb2 100644 --- a/pysim_simple_server/server.py +++ b/pysim_simple_server/server.py @@ -20,6 +20,7 @@ from pysim_simple_server import netstate from pysim_simple_server import scp81 from pysim_simple_server import esim from pysim_simple_server import capmem +from pysim_simple_server import testscript from smartcard.CardMonitoring import CardMonitor, CardObserver from cmd2.exceptions import CommandSetRegistrationError @@ -843,7 +844,8 @@ def _build_sms_tpdu(chunk_hex, chunk_total=1, chunk_num=1, oa_number='12345', in return tpdu.hex() -def _send_envelope(tpdu_hex, scc, sm_sc='12345678912', submit_handler=None): +def _send_envelope(tpdu_hex, scc, sm_sc='12345678912', submit_handler=None, + handle_proactive=True): from pySim.ts_31_102 import SMSPPDownload from pySim.cat import DeviceIdentities, Address from osmocom.tlv import COMPR_TLV_IE @@ -869,7 +871,7 @@ def _send_envelope(tpdu_hex, scc, sm_sc='12345678912', submit_handler=None): if sw.startswith('61'): get_len = int(sw[2:], 16) if len(sw) == 4 else 0x100 data, sw = scc._tp.send_apdu('00c00000%02x' % get_len) - elif sw.startswith('91'): + elif handle_proactive and sw.startswith('91'): def _capture_sms_tpdu(raw, cmd_num, cmd_type, dev_src, dev_dst): if submit_handler: tpdu_hex = _find_sms_tpdu(raw) @@ -887,7 +889,7 @@ def _send_envelope(tpdu_hex, scc, sm_sc='12345678912', submit_handler=None): submit_handler.submit_tpdu_hex = tpdu_hex _handle_proactive_chain(scc, sw, _capture_sms_tpdu) data, sw = '', '9000' - if sw == '9000' and submit_handler and not submit_handler.submit_tpdu_hex: + if handle_proactive and sw == '9000' and submit_handler and not submit_handler.submit_tpdu_hex: sys.stderr.write('STATUS poll (PoR not captured)\n') st_data, st_sw = _send_status(scc) sys.stderr.write('STATUS -> %s\n' % st_sw) @@ -1047,7 +1049,8 @@ def _build_secured_packet(spi1, spi2, kic, kid, tar_hex, cntr_hex, apdu_hex, def _send_secured_packet(scc, sp_hex, oa_number, sm_sc=None, include_cpi=True, - submit_handler=None, max_segments=SCP80_MAX_SEGMENTS): + submit_handler=None, max_segments=SCP80_MAX_SEGMENTS, + handle_proactive=True): """Send a secured packet as SMS-PP download ENVELOPEs, one per segment. TS 31.115 4.3: the whole command packet is split into SMS user-data @@ -1082,7 +1085,8 @@ def _send_secured_packet(scc, sp_hex, oa_number, sm_sc=None, include_cpi=True, sys.stderr.write('OTA SEND: ENVELOPE %d/%d (%d B)%s\n' % ( i + 1, total, len(part), ' + CPI' if i == 0 and include_cpi else '')) data, sw = _send_envelope(tpdu, scc, sm_sc=sm_sc or '12345678912', - submit_handler=submit_handler) + submit_handler=submit_handler, + handle_proactive=handle_proactive) if sw != '9000' and not sw.startswith('91'): return {'success': False, 'sw': sw, 'bytes': len(pkt), 'segments': total, @@ -1417,7 +1421,7 @@ def _reset_poll_timer(): def _do_status_poll(): global _POLL_TIMER _POLL_TIMER = None - if not _POLL_ENABLED or _POLL_DISABLED_BY_CARD: + if not _POLL_ENABLED or _POLL_DISABLED_BY_CARD or _TEST_RUNNING: return with _CARD_LOCK: try: @@ -3458,7 +3462,7 @@ def _verify_adm(scc, app, adm_hex): 'error': 'Security status not satisfied' if sw == '6982' else 'Error'} -def _send_event_download(scc, event_type, event_data=None): +def _send_event_download(scc, event_type, event_data=None, drain=True): """Send ENVELOPE(Event Download) for the given event type. Builds: CLA C2 0000 Lc D6 [len] (99 01 [type] 82 02 82 81 [extra])""" inner = bytearray() @@ -3470,8 +3474,11 @@ def _send_event_download(scc, event_type, event_data=None): env_hex = '%sc20000%02x%s' % (scc.cat_cla, len(d6_tlv), d6_tlv.hex()) sys.stderr.write('ENVELOPE(Event Download): type=0x%02x data=%s\n' % (event_type, event_data.hex() if event_data else '(none)')) data, sw = scc._tp.send_apdu(env_hex) + if sw.startswith('61'): + get_len = int(sw[2:], 16) if len(sw) == 4 else 0x100 + data, sw = scc._tp.send_apdu('00c00000%02x' % get_len) sys.stderr.write('ENVELOPE SW: %s\n' % sw) - if sw.startswith('91'): + if drain and sw.startswith('91'): _handle_proactive_chain(scc, sw) sw = '9000' return data, sw @@ -3772,7 +3779,7 @@ def _parse_display_text(raw): while off < len(raw) - 1: tag, tlen = raw[off], raw[off + 1] val = raw[off + 2: off + 2 + tlen]; off += 2 + tlen - if tag == 0x8D and tlen >= 1: + if tag in (0x8D, 0x0D) and tlen >= 1: return _decode_dcs_text(val) return None @@ -4109,6 +4116,487 @@ def _finish_pending_menu(server, scc): _handle_proactive_chain(scc, sw) +# ===== Test scripts (scripted card dialogue) ===== +# A script runs server-side in a worker thread and owns the card while it +# runs: other card endpoints answer 409 (_TEST_BLOCKED_PATHS) and background +# STATUS polling is suspended until the run finishes. Steps and results live +# in _TEST_RUN (polled by the PWA via GET /api/test/status); the pure engine +# (validation, checks, TERMINAL RESPONSE building) is in testscript.py. + +_TEST_RUNNING = False +_TEST_RUN = { + 'running': False, 'stop': False, 'name': None, 'status': None, + 'session': None, 'index': 0, 'total': 0, 'steps': [], + 'preset': None, 'scp80_counter': None, + 'started': None, 'finished': None, 'error': None, +} +_TEST_LOCK = threading.Lock() +_TEST_THREAD = None +_TEST_KIND_LABELS = { + 'envelope': 'ENVELOPE(Event Download)', 'menu-select': 'ENVELOPE(Menu Selection)', + 'file-write': 'UPDATE FILE', 'file-read': 'READ FILE', 'apdu': 'APDU', + 'scp80': 'SCP80', 'status': 'STATUS', 'cleanup': 'CLEANUP', +} +# Card-touching endpoints refused while a script owns the card. +_TEST_BLOCKED_PATHS = frozenset([ + '/api/command', '/api/cardinfo', '/api/tree', '/api/select', '/api/read', + '/api/write', '/api/apdu', '/api/verify-adm', '/api/send-ota', + '/api/ram-install', '/api/sp-verify', '/api/menu-select', + '/api/menu-respond', '/api/event-send', '/api/net-sim', + '/api/net-state-refresh', '/api/status-poll', '/api/rescue', + '/api/terminal-profile', '/api/poll-toggle', '/api/esim/chip', + '/api/esim/profiles', '/api/esim/notifications', '/api/esim/profile', + '/api/scp81/bip', '/api/scp81/queue', '/api/scp81/psk-map', + '/api/scp81/gen-install', '/api/scp81/log-clear', +]) +_TEST_COMMAND_TYPES = {name.upper(): code for code, name in PROACTIVE_TYPE_NAMES.items()} + + +def _test_command_type(name): + return _TEST_COMMAND_TYPES.get((name or '').upper()) + + +def _test_state_snapshot(): + with _TEST_LOCK: + return json.loads(json.dumps(_TEST_RUN)) + + +def _test_request_blocked(path): + if not _TEST_RUNNING: + return False + return path.split('?', 1)[0] in _TEST_BLOCKED_PATHS + + +def _increment_counter_hex(counter_hex): + """SCP80 counter + 1, keeping the pattern's width (hex string).""" + c = re.sub(r'\s', '', str(counter_hex or '')).upper() + if not c or not re.fullmatch(r'[0-9A-F]+', c): + return counter_hex + width = len(c) + return '%0*X' % (width, (int(c, 16) + 1) & ((1 << (4 * width)) - 1)) + + +def _test_preset_error(script, preset): + """The SCP80 steps need a complete card preset; steps may override + TAR/SPI1/SPI2 only (KIc/KID and the counter stay preset-owned).""" + preset = preset or {} + for key in ('kic', 'kid', 'kicKey', 'kidKey'): + if not preset.get(key): + return 'SCP80 preset is incomplete: %s is missing' % key + if not (preset.get('counter') or preset.get('cntr')): + return 'SCP80 preset has no counter' + for i, step in enumerate(script['steps']): + if step['type'] != 'action' or step['kind'] != 'scp80': + continue + p = step['params'] + if not (p.get('tar') or preset.get('tar')): + return 'step %d: no TAR (neither in the step nor in the preset)' % (i + 1) + if not (p.get('spi1') or preset.get('spi1')) or not (p.get('spi2') or preset.get('spi2')): + return 'step %d: no SPI1/SPI2 (neither in the step nor in the preset)' % (i + 1) + return None + + +def _test_check_result(label, ok, expected, actual, level, detail=None): + res = {'label': label, 'ok': bool(ok), 'expected': str(expected), + 'actual': str(actual), 'level': level} + if detail: + res['detail'] = detail + return res + + +def _test_entry(step, index): + if step['type'] == 'expect': + ctype = step['command'].get('type') + kind = step['command'].get('name') or (('0x%02X' % ctype) if ctype is not None else 'ANY') + label = 'EXPECT ' + kind + else: + kind = step['kind'] + label = step.get('label') or _TEST_KIND_LABELS.get(kind, kind) + return {'index': index, 'type': step['type'], 'kind': kind, 'label': label, + 'status': 'running', 'checks': [], 'note': None, 'sent': None, + 'sw': None, 'data': None, 'ms': None, 'started': time.time()} + + +def _test_finish_entry(entry, result): + entry['status'] = result.get('status', 'ok') + entry['checks'] = result.get('checks', []) + if result.get('sent') is not None: + entry['sent'] = result['sent'] + if result.get('sw') is not None: + entry['sw'] = result['sw'] + if result.get('data') is not None: + entry['data'] = result['data'] + if result.get('note'): + entry['note'] = result['note'] + if result.get('por') is not None: + entry['por'] = result['por'] + if result.get('counter'): + entry['counter'] = result['counter'] + if result.get('command'): + entry['command'] = result['command'] + entry['ms'] = int((time.time() - (entry.get('started') or time.time())) * 1000) + + +def _test_action_checks(step, sw, data, por, kind): + check = step['check'] + level = step['on_fail'] + checks = [] + sw_ok = testscript.match_value(check['sw'], sw) + checks.append(_test_check_result('SW', sw_ok, check['sw']['value'], sw or '(none)', level)) + if check.get('data'): + if sw_ok: + checks.append(_test_check_result('Data', testscript.match_value(check['data'], data), + check['data']['value'], data or '(none)', level)) + else: + checks.append(_test_check_result('Data', True, check['data']['value'], + '(not checked - SW mismatch)', 'ok')) + if kind == 'scp80' and check.get('por') != 'any': + want = check['por'] + if want == 'none': + ok = por is None + actual = 'none' if por is None else 'present' + else: + ok = bool(por) and por.get('response_status') == 'por_ok' + actual = (por or {}).get('response_status') or 'none' + checks.append(_test_check_result('PoR', ok, want, actual, level)) + return checks + + +def _test_run_status(scc, step): + p = step['params'] + data, sw = '', '' + used = 0 + for n in range(p['attempts']): + data, sw = _send_status(scc) + used = n + 1 + if testscript.match_value(step['check']['sw'], sw): + break + if n + 1 < p['attempts'] and p['interval_ms']: + time.sleep(p['interval_ms'] / 1000.0) + return data or '', sw, ('STATUS x%d' % used if used > 1 else 'STATUS') + + +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).""" + app = server.app + lchan = app.rs.lchan[0] + p = step['params'] + cleanup = None + try: + _, cleanup = _select_path(lchan, p['path'], app) + is_record = _get_file_type(lchan, lchan.selected_file) in ('linear_fixed', 'cyclic') + use_record = p['mode'] == 'record' or (p['mode'] == 'auto' and is_record) + record = p.get('record') or 1 + if step['kind'] == 'file-write': + if use_record: + _out, sw = lchan.update_record(record, p['data']) + sent = 'UPDATE RECORD %d %s' % (record, p['path']) + else: + _out, sw = lchan.update_binary(p['data']) + sent = 'UPDATE BINARY %s' % p['path'] + return _out or '', sw, sent + if use_record: + data, sw = lchan.read_record(record) + return data or '', sw, 'READ RECORD %d %s' % (record, p['path']) + data, sw = lchan.read_binary() + return data or '', sw, 'READ BINARY %s' % p['path'] + finally: + if cleanup: + try: + cleanup() + except Exception: + pass + + +def _test_run_scp80(server, step, ctx): + scc = server.scc + preset = ctx['preset'] + p = step['params'] + spi1 = p.get('spi1') or preset.get('spi1') or '16' + spi2 = p.get('spi2') or preset.get('spi2') or '01' + tar = p.get('tar') or preset.get('tar') or '' + counter = ctx['counter'] or '00000000' + if p.get('sp'): + sp_hex = p['sp'] + source = 'sp' + else: + sp_hex, _ = _build_secured_packet(spi1, spi2, preset.get('kic', ''), preset.get('kid', ''), + tar, counter, p['apdu'], + preset.get('kicKey', ''), preset.get('kidKey', '')) + source = 'apdu' + result = _send_secured_packet(scc, sp_hex, server.sms_oa, sm_sc=server.sms_sc, + handle_proactive=False) + sw = result.get('sw') or '' + data = result.get('response_data') or '' + por = None + if data: + por = _decode_por(spi1, spi2, preset.get('kic', ''), preset.get('kid', ''), + counter, preset.get('kicKey', ''), preset.get('kidKey', ''), data) + sent = 'SCP80 %s TAR=%s SPI1=%s SPI2=%s cntr=%s' % ( + source, tar or '-', spi1, spi2, counter) + if sw: + # The card answered, so the SCP80 counter was consumed: advance the + # working value (the PWA writes the final one back to the preset). + ctx['counter'] = _increment_counter_hex(counter) + return data, sw, sent, por, counter + + +def _test_run_action(server, step, ctx): + scc = server.scc + kind = step['kind'] + p = step['params'] + por = None + counter = None + if kind == 'envelope': + data, sw = _send_event_download(scc, p['event'], + bytes.fromhex(p['data']) if p['data'] else None, + drain=False) + sent = 'ENVELOPE(Event Download) type=0x%02X' % p['event'] + elif kind == 'menu-select': + tlv = bytes([0xD3, 0x07, 0x02, 0x02, 0x01, 0x81, 0x90, 0x01, p['item_id']]) + sent = '%sc20000%02x%s' % (scc.cat_cla, len(tlv), tlv.hex()) + data, sw = scc._tp.send_apdu(sent) + elif kind == 'apdu': + sent = p['apdu'] + data, sw = scc._tp.send_apdu(sent) + elif kind == 'status': + data, sw, sent = _test_run_status(scc, step) + elif kind in ('file-write', 'file-read'): + data, sw, sent = _test_run_file(server, step) + elif kind == 'scp80': + data, sw, sent, por, counter = _test_run_scp80(server, step, ctx) + else: + raise testscript.ScriptError('unknown action kind %r' % kind) + checks = _test_action_checks(step, sw, data, por, kind) + status = testscript.combine_levels([c['level'] if not c['ok'] else 'ok' for c in checks]) + result = {'status': status, 'sw': sw or '', 'data': data or '', 'sent': sent, + 'checks': checks, 'por': por} + if counter: + result['counter'] = counter + pending = int(sw[2:], 16) if (sw or '').startswith('91') else None + return result, pending + + +def _test_run_expect(server, step, pending): + 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)') + fdata, fetch_sw = scc._tp.send_apdu('%s120000%02x' % (scc.cat_cla, pending)) + if not fdata: + raise testscript.ScriptError('FETCH returned no data (SW %s)' % fetch_sw) + raw = bytes.fromhex(fdata) + cmd_num, cmd_type, dev_src, dev_dst, cmd_qual = _parse_proactive_header(raw) + log_entry = _log_proactive(cmd_type, raw, cmd_qual, cmd_num) + type_name = PROACTIVE_TYPE_NAMES.get(cmd_type, 'UNKNOWN') + checks = [] + want_type = step['command'].get('type') + type_ok = want_type is None or cmd_type == want_type + if type_ok: + checks.append(_test_check_result('Command', True, step['command'].get('name') or type_name, + '%s (0x%02X)' % (type_name, cmd_type), 'ok')) + else: + checks.append(_test_check_result( + 'Command', False, step['command'].get('name') or '0x%02X' % want_type, + '%s (0x%02X)' % (type_name, cmd_type), 'error')) + if step.get('qualifier'): + actual_q = '%02X' % cmd_qual if cmd_qual is not None else None + checks.append(_test_check_result('Qualifier', + testscript.match_value(step['qualifier'], actual_q), + step['qualifier']['value'], actual_q or '(none)', + step['on_fail'])) + for c in step['checks']: + if c['kind'] == 'text': + text = _parse_display_text(raw) + ok = testscript.match_text(c, text) + checks.append(_test_check_result('Text', ok, c['value'], + text if text is not None else '(none)', + c['on_fail'])) + elif c['kind'] == 'item': + items = None + if cmd_type == 0x24: + items = _parse_select_item(raw) + elif cmd_type == 0x25: + items = _parse_setup_menu_items(raw) + if items is None: + checks.append(_test_check_result('Item', False, + 'id=%s text=%r' % (c.get('id'), c.get('text')), + 'command carries no items', c['on_fail'])) + else: + ok, detail = testscript.match_item(items, c) + checks.append(_test_check_result('Item', ok, + 'id=%s text=%r' % (c.get('id'), c.get('text')), + detail, c['on_fail'])) + elif c['kind'] == 'raw': + actual = raw.hex().upper() + checks.append(_test_check_result('Raw', + testscript.match_value({'mode': c['mode'], 'value': c['value']}, actual), + c['value'], actual, c['on_fail'])) + tr = testscript.build_tr(cmd_num, cmd_type, dev_dst, dev_src, step['respond']) + tr_rv = scc._tp.send_apdu('%s140000%02x%s' % (scc.cat_cla, len(tr), tr.hex())) + tr_sw = tr_rv[1] + _record_tr(log_entry, tr, tr_sw) + status = testscript.combine_levels([c['level'] if not c['ok'] else 'ok' for c in checks]) + result = {'status': status, 'sw': tr_sw, 'data': raw.hex().upper(), + 'sent': 'TR ' + tr.hex().upper(), 'checks': checks, + 'command': {'type_hex': '%02X' % cmd_type, 'type_name': type_name, + 'qualifier': '%02X' % cmd_qual if cmd_qual is not None else None, + 'raw': raw.hex().upper()}} + pending_next = int(tr_sw[2:], 16) if tr_sw.startswith('91') else 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).""" + 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)} + except Exception as e: + return {'note': 'drain failed: %s' % e} + + +def _test_run_execute(server, script, preset): + scc = server.scc + ctx = {'preset': dict(preset or {}), + 'counter': str((preset or {}).get('counter') or (preset or {}).get('cntr') or '').upper(), + 'uses_scp80': any(s['type'] == 'action' and s['kind'] == 'scp80' + for s in script['steps'])} + if ctx['uses_scp80']: + with _TEST_LOCK: + _TEST_RUN['scp80_counter'] = ctx['counter'] + pending = None + stopped = False + session = getattr(server, 'card_session', 0) + for index, step in enumerate(script['steps']): + with _TEST_LOCK: + if _TEST_RUN['stop']: + stopped = True + _TEST_RUN['index'] = index + if stopped: + break + entry = _test_entry(step, index) + with _TEST_LOCK: + _TEST_RUN['steps'].append(entry) + if pending is not None and step['type'] != 'expect': + entry['status'] = 'error' + entry['note'] = ('unexpected proactive command pending (SW 91XX) - ' + 'add an expect step or a status action') + drained = _test_drain_pending(scc, pending) + entry.update({k: v for k, v in drained.items() if k != 'note'}) + entry['ms'] = int((time.time() - entry['started']) * 1000) + pending = None + break + if getattr(server, 'card_session', 0) != session: + entry['status'] = 'error' + entry['note'] = 'card session changed (card removed or re-equipped)' + entry['ms'] = int((time.time() - entry['started']) * 1000) + break + try: + with _CARD_LOCK: + if step['type'] == 'action': + result, pending = _test_run_action(server, step, ctx) + else: + result, pending = _test_run_expect(server, step, pending) + _test_finish_entry(entry, result) + if entry['status'] == 'error': + # Error terminates the script; Warning and OK continue. + break + except Exception as e: + entry['status'] = 'error' + entry['note'] = str(e) + entry['ms'] = int((time.time() - entry['started']) * 1000) + sys.stderr.write('TEST-RUN step %d failed: %s\n' % (index + 1, e)) + if pending is not None: + drained = _test_drain_pending(scc, pending) + entry.update({k: v for k, v in drained.items() if k != 'note'}) + pending = None + break + if pending is not None: + drained = _test_drain_pending(scc, pending) + with _TEST_LOCK: + _TEST_RUN['steps'].append(dict( + {'index': len(script['steps']), 'type': 'cleanup', 'kind': 'cleanup', + 'label': _TEST_KIND_LABELS['cleanup'], 'status': 'error' if not stopped else 'warning', + 'checks': [], 'started': time.time(), 'ms': 0, 'note': None, + 'sent': None, 'sw': None, 'data': None}, **drained)) + with _TEST_LOCK: + _TEST_RUN['running'] = False + _TEST_RUN['finished'] = time.time() + if ctx['uses_scp80']: + _TEST_RUN['scp80_counter'] = ctx['counter'] + if _TEST_RUN['status'] is None: + levels = [s.get('status') for s in _TEST_RUN['steps']] + if stopped: + _TEST_RUN['status'] = 'stopped' + elif 'error' in levels: + _TEST_RUN['status'] = 'error' + elif 'warning' in levels: + _TEST_RUN['status'] = 'warning' + else: + _TEST_RUN['status'] = 'ok' + global _TEST_RUNNING + _TEST_RUNNING = False + if _POLL_ENABLED and not _POLL_DISABLED_BY_CARD: + _reset_poll_timer() + + +def _test_run_worker(server, script, preset): + try: + _test_run_execute(server, script, preset) + except Exception as e: + traceback.print_exc() + global _TEST_RUNNING + _TEST_RUNNING = False + with _TEST_LOCK: + _TEST_RUN['error'] = str(e) + _TEST_RUN['running'] = False + _TEST_RUN['finished'] = time.time() + if not _TEST_RUN['status']: + _TEST_RUN['status'] = 'error' + if _POLL_ENABLED and not _POLL_DISABLED_BY_CARD: + _reset_poll_timer() + + +def _test_run_start(server, script, preset): + """Start a run: suspend background polling, answer a paused interactive + command, own the card via the worker thread.""" + global _TEST_THREAD, _TEST_RUNNING, _POLL_TIMER + with _TEST_LOCK: + _TEST_RUN.update({ + 'running': True, 'stop': False, 'name': script['name'], 'status': None, + 'session': getattr(server, 'card_session', 0), 'index': 0, + 'total': len(script['steps']), 'steps': [], 'started': time.time(), + 'finished': None, 'error': None, 'scp80_counter': None, + 'preset': (preset or {}).get('name') or (preset or {}).get('iccid'), + }) + _TEST_RUNNING = True + if _POLL_TIMER is not None: + _POLL_TIMER.cancel() + _POLL_TIMER = None + with _CARD_LOCK: + try: + _finish_pending_menu(server, server.scc) + except Exception as e: + sys.stderr.write('TEST-RUN: finishing pending menu failed: %s\n' % e) + _TEST_THREAD = threading.Thread(target=_test_run_worker, + args=(server, script, preset), + name='test-script', daemon=True) + _TEST_THREAD.start() + + class PysimHandler(BaseHTTPRequestHandler): def _send_json(self, data, status=200): self.send_response(status) @@ -4172,11 +4660,15 @@ class PysimHandler(BaseHTTPRequestHandler): self.wfile.write(data) def do_GET(self): - # /api/status is pure cached state (no card I/O); keeping it out of the - # lock lets the UI report 'initializing' while a long equip holds the - # card lock. Result-shaping masks everything card-derived when the - # session is not connected. - if self.path == '/api/status': + if _test_request_blocked(self.path): + self._send_json({'error': 'test script running - card commands are ' + 'blocked until it finishes'}, 409) + return + # /api/status and /api/test/status are pure cached state (no card I/O); + # keeping them out of the lock lets the UI report 'initializing' while a + # long equip (or a test script step) holds the card lock. Result-shaping + # masks everything card-derived when the session is not connected. + if self.path in ('/api/status', '/api/test/status'): self._do_GET() return # Serialize all card access: the background STATUS poll runs in its own @@ -4190,6 +4682,11 @@ class PysimHandler(BaseHTTPRequestHandler): self._log_req() self._send_json({'version': VERSION}) self._log_resp({'version': VERSION}) + elif self.path == '/api/test/status': + self._log_req() + resp = _test_state_snapshot() + self._send_json(resp) + self._log_resp(resp) elif self.path.startswith('/api/mcc-mnc'): self._log_req() q = '' @@ -4403,6 +4900,60 @@ class PysimHandler(BaseHTTPRequestHandler): resp = {('%02X' % q): v for q, v in _PLI_DATA.items()} self._send_json(resp) self._log_resp(resp) + elif self.path == '/api/test/run': + body = self._read_body() + self._log_req(body) + if _TEST_RUNNING: + resp = {'error': 'a test script is already running'} + self._send_json(resp, 409) + self._log_resp(resp) + return + try: + script = testscript.normalise_script(body.get('script'), _test_command_type) + except testscript.ScriptError as e: + resp = {'error': str(e)} + self._send_json(resp, 400) + self._log_resp(resp) + return + preset = body.get('preset') or {} + needs_scp80 = any(s['type'] == 'action' and s['kind'] == 'scp80' + for s in script['steps']) + err = _test_preset_error(script, preset) if needs_scp80 else None + if err: + resp = {'error': err} + self._send_json(resp, 400) + self._log_resp(resp) + return + _test_run_start(self.server, script, preset) + resp = _test_state_snapshot() + self._send_json(resp) + self._log_resp(resp) + elif self.path == '/api/test/stop': + self._log_req() + with _TEST_LOCK: + if _TEST_RUN['running']: + _TEST_RUN['stop'] = True + resp = _test_state_snapshot() + self._send_json(resp) + self._log_resp(resp) + elif self.path == '/api/test/clear': + self._log_req() + with _TEST_LOCK: + busy = bool(_TEST_RUN['running']) + if not busy: + _TEST_RUN.update({ + 'running': False, 'stop': False, 'name': None, 'status': None, + 'session': None, 'index': 0, 'total': 0, 'steps': [], + 'preset': None, 'scp80_counter': None, + 'started': None, 'finished': None, 'error': None, + }) + if busy: + resp = {'error': 'test script is running'} + self._send_json(resp, 409) + else: + resp = _test_state_snapshot() + self._send_json(resp) + self._log_resp(resp) elif self.path == '/api/poll-status': resp = {'enabled': _POLL_ENABLED, 'interval': _POLL_INTERVAL, 'card_disabled': _POLL_DISABLED_BY_CARD} @@ -4449,6 +5000,10 @@ class PysimHandler(BaseHTTPRequestHandler): self._serve_static() def do_POST(self): + if _test_request_blocked(self.path): + self._send_json({'error': 'test script running - card commands are ' + 'blocked until it finishes'}, 409) + return # Serialize all card access: the background STATUS poll runs in its own # thread and must never interleave with a FETCH/TERMINAL RESPONSE pair. with _CARD_LOCK: diff --git a/pysim_simple_server/testscript.py b/pysim_simple_server/testscript.py new file mode 100644 index 0000000..b9b0086 --- /dev/null +++ b/pysim_simple_server/testscript.py @@ -0,0 +1,407 @@ +"""Test script engine (pure): validation, response checks and scripted TRs. + +A test script drives a deterministic dialogue with the card: + +* an **action** step sends something (ENVELOPE, Menu Selection, raw APDU, + SCP80 secured packet, file update/read, STATUS) and checks the response + (SW exact/mask, data exact/mask, PoR for SCP80); +* an **expectation** step fetches the proactive command announced by the + previous step (SW ``91XX`` - TS 102 221 7.4.2.1 / TS 102 223 6.3: the UICC + announces a pending command in the response to a command and re-announces + it with ``91XX`` until it is fetched; it never pushes unsolicited), checks + its contents and answers it with a scripted TERMINAL RESPONSE. + +This module has no card access - ``server.py`` runs the steps; only the pure +parts live here so they can be tested without hardware. + +Check syntax (SW, data, qualifier): a plain hex string is an exact match, +``?`` in mask mode is a per-nibble wildcard (same convention as the +profiler), e.g. ``{"mode": "mask", "value": "91??"}``. +""" + +import re + +__all__ = [ + 'ScriptError', 'ACTION_KINDS', 'FAIL_LEVELS', 'POR_CHECKS', 'RESULT_NAMES', + 'normalise_script', 'normalise_step', 'normalise_respond', + 'match_value', 'match_text', 'match_item', 'combine_levels', 'build_tr', +] + +ACTION_KINDS = ('envelope', 'menu-select', 'file-write', 'file-read', 'apdu', + 'scp80', 'status') +FAIL_LEVELS = ('error', 'warning') +POR_CHECKS = ('none', 'ok', 'any') + +# TERMINAL RESPONSE result values commonly used by scripts (TS 102 223 8.12). +RESULT_NAMES = { + 'ok': 0x00, 'partial': 0x01, 'missing': 0x02, 'refused': 0x03, + 'not_understood': 0x04, 'modified': 0x06, + 'cancel': 0x10, 'back': 0x11, 'timeout': 0x12, 'no_response': 0x22, +} + +_HEX_MASK_RE = re.compile(r'^[0-9A-F?]+$') +_HEX_RE = re.compile(r'^[0-9A-F]+$') + + +class ScriptError(ValueError): + """Invalid test script - reported to the client before a run starts.""" + + +# ─── normalisation helpers ────────────────────────────────────────────── + +def _fail_level(value, default='error'): + if value in (None, ''): + return default + v = str(value).lower() + if v not in FAIL_LEVELS: + raise ScriptError("on_fail must be 'error' or 'warning'") + return v + + +def _int(value, what, lo, hi): + try: + v = int(str(value).strip(), 0) + except (TypeError, ValueError): + raise ScriptError('%s must be an integer' % what) + if not lo <= v <= hi: + raise ScriptError('%s must be %d..%d' % (what, lo, hi)) + return v + + +def _data_hex(value, what, allow_empty=False): + if value in (None, ''): + if allow_empty: + return '' + raise ScriptError('%s is required' % what) + if not isinstance(value, str): + raise ScriptError('%s must be a hex string' % what) + v = re.sub(r'\s', '', value).upper() + if not v: + if allow_empty: + return '' + raise ScriptError('%s is required' % what) + if not _HEX_RE.match(v) or len(v) % 2: + raise ScriptError('%s must be hex with an even number of digits' % what) + return v + + +def _check_spec(value, what, default_mode='exact'): + """Normalise a check: hex string or {'mode', 'value'}.""" + if value is None: + return None + if isinstance(value, dict): + mode = str(value.get('mode') or default_mode).lower() + val = value.get('value') + else: + val = value + # a plain string with '?' is a mask ("91??"), no need to spell it out + mode = 'mask' if (default_mode == 'exact' and isinstance(val, str) + and '?' in val) else default_mode + if mode not in ('exact', 'mask'): + raise ScriptError('%s: mode must be exact or mask' % what) + if not isinstance(val, str) or not val.strip(): + raise ScriptError('%s: value is required' % what) + v = re.sub(r'\s', '', val).upper() + if not _HEX_MASK_RE.match(v) or len(v) % 2: + raise ScriptError('%s: value must be hex (even length, "?" = wildcard)' % what) + if mode == 'exact' and '?' in v: + raise ScriptError('%s: "?" is only allowed in mask mode' % what) + return {'mode': mode, 'value': v} + + +# ─── matching (pure) ──────────────────────────────────────────────────── + +def match_value(spec, actual): + """Exact/mask hex comparison; no spec means 'no check'.""" + if not spec: + return True + if actual is None: + return False + a = re.sub(r'\s', '', str(actual)).upper() + v = spec['value'] + if len(a) != len(v): + return False + if spec['mode'] == 'exact': + return a == v + return all(vc == '?' or vc == ac for vc, ac in zip(v, a)) + + +def match_text(spec, text): + if not spec: + return True + if text is None: + return False + want, got = spec['value'], str(text) + if not spec.get('case_sensitive', True): + want, got = want.lower(), got.lower() + return want == got if spec['mode'] == 'exact' else want in got + + +def match_item(items, spec): + """Find a parsed item (SELECT ITEM / SET UP MENU) matching id and/or text. + + Returns ``(ok, detail)`` - the detail names the matching item or lists the + decoded items so the report is useful without the raw bytes.""" + decoded = ', '.join('%s=%r' % (it.get('id'), it.get('text')) for it in (items or [])) + if not items: + return False, 'no items decoded' + for it in items: + if spec.get('id') is not None and int(it.get('id', -1)) != spec['id']: + continue + if spec.get('text') is not None: + text_spec = {'mode': spec['mode'], 'value': spec['text'], + 'case_sensitive': spec.get('case_sensitive', True)} + if not match_text(text_spec, it.get('text')): + continue + return True, 'item %s %r' % (it.get('id'), it.get('text')) + return False, 'no matching item (decoded: %s)' % (decoded or 'none') + + +def combine_levels(levels): + """Worst outcome of a step: any error wins, then warning, else ok.""" + if 'error' in levels: + return 'error' + if 'warning' in levels: + return 'warning' + return 'ok' + + +# ─── script validation ────────────────────────────────────────────────── + +def normalise_script(raw, command_resolver=None): + """Validate/normalise a script. ``command_resolver(name) -> int|None`` + resolves proactive command names (provided by server.py).""" + if not isinstance(raw, dict): + raise ScriptError('script must be an object') + name = str(raw.get('name') or '').strip() or 'test script' + steps_raw = raw.get('steps') + if not isinstance(steps_raw, list) or not steps_raw: + raise ScriptError('script must have at least one step') + steps = [normalise_step(s, command_resolver) for s in steps_raw] + return {'name': name, 'steps': steps} + + +def normalise_step(step, command_resolver=None): + if not isinstance(step, dict): + raise ScriptError('each step must be an object') + typ = step.get('type') + if typ == 'action': + return _normalise_action(step) + if typ == 'expect': + return _normalise_expect(step, command_resolver) + raise ScriptError("step type must be 'action' or 'expect'") + + +def _normalise_action(step): + 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 {}) + on_fail = _fail_level(step.get('on_fail')) + check = _normalise_check(step.get('check'), kind, params) + out = {'type': 'action', 'kind': kind, 'params': params, + 'check': check, 'on_fail': on_fail} + if step.get('label'): + out['label'] = str(step['label']) + return out + + +def _normalise_params(kind, p): + if not isinstance(p, dict): + raise ScriptError('%s: params must be an object' % kind) + if kind == 'envelope': + if p.get('event') is None: + raise ScriptError('envelope: event is required') + return {'event': _int(p['event'], 'envelope event', 0, 255), + 'data': _data_hex(p.get('data'), 'envelope data', allow_empty=True)} + if kind == 'menu-select': + return {'item_id': _int(p.get('item_id'), 'menu item_id', 1, 255)} + if kind in ('file-write', 'file-read'): + path = str(p.get('path') or '').strip() + if not path: + raise ScriptError('%s: path is required' % kind) + mode = str(p.get('mode') or 'auto').lower() + if mode not in ('auto', 'binary', 'record'): + raise ScriptError('%s: mode must be auto, binary or record' % kind) + out = {'path': path, 'mode': mode} + if mode == 'record' or p.get('record') is not None: + out['record'] = _int(p.get('record') or 1, '%s record' % kind, 1, 255) + if kind == 'file-write': + out['data'] = _data_hex(p.get('data'), 'file-write data') + return out + if kind == 'apdu': + return {'apdu': _data_hex(p.get('apdu'), 'apdu')} + if kind == 'scp80': + out = {} + if p.get('sp'): + out['sp'] = _data_hex(p.get('sp'), 'secured packet') + elif p.get('apdu'): + out['apdu'] = _data_hex(p.get('apdu'), 'scp80 apdu') + else: + raise ScriptError('scp80: apdu or sp is required') + for key in ('tar', 'spi1', 'spi2'): + if p.get(key) not in (None, ''): + out[key] = _data_hex(p[key], 'scp80 %s' % key) + if out.get('tar') and len(out['tar']) != 6: + raise ScriptError('scp80: tar must be 3 bytes') + for key in ('spi1', 'spi2'): + if out.get(key) and len(out[key]) != 2: + raise ScriptError('scp80: %s must be 1 byte' % key) + return out + if kind == 'status': + attempts = p.get('attempts') + attempts = _int(1 if attempts is None else attempts, 'status attempts', 1, 1000) + 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} + raise ScriptError('unknown action kind %r' % kind) + + +def _normalise_check(check, kind, params): + if check is None: + check = {} + if not isinstance(check, dict): + raise ScriptError('check must be an object') + sw = _check_spec(check.get('sw'), 'check.sw') + if sw is None: + # A STATUS poll for a pending proactive command ends on 91XX + # (TS 102 221 7.4.2.1); a single STATUS normally ends on 9000. + if kind == 'status' and params.get('attempts', 1) > 1: + sw = {'mode': 'mask', 'value': '91??'} + else: + sw = {'mode': 'exact', 'value': '9000'} + data = _check_spec(check.get('data'), 'check.data') + por = check.get('por') + if por is None: + por = 'any' + else: + por = str(por).lower() + if kind != 'scp80': + raise ScriptError('check.por is only valid for scp80 actions') + if por not in POR_CHECKS: + raise ScriptError('check.por must be none, ok or any') + return {'sw': sw, 'data': data, 'por': por} + + +def _normalise_expect(step, command_resolver=None): + cmd = step.get('command') + if cmd is None or isinstance(cmd, bool): + raise ScriptError('expect: command is required') + 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') + 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') + on_fail = _fail_level(step.get('on_fail')) + checks_raw = step.get('checks') or [] + if not isinstance(checks_raw, list): + raise ScriptError('expect: checks must be a list') + checks = [_normalise_content_check(c, on_fail) for c in checks_raw] + respond = normalise_respond(step.get('respond') or {}, ctype) + return {'type': 'expect', 'command': {'type': ctype, 'name': cname}, + 'qualifier': qualifier, 'checks': checks, 'respond': respond, + 'on_fail': on_fail} + + +def _normalise_content_check(c, default_level): + if not isinstance(c, dict): + raise ScriptError('expect: each check must be an object') + kind = str(c.get('kind') or '').lower() + level = _fail_level(c.get('on_fail'), default_level) + if kind == 'text': + mode = str(c.get('mode') or 'contains').lower() + if mode not in ('contains', 'exact'): + raise ScriptError('text check: mode must be contains or exact') + if c.get('value') is None: + raise ScriptError('text check: value is required') + return {'kind': 'text', 'mode': mode, 'value': str(c['value']), + 'case_sensitive': bool(c.get('case_sensitive', True)), + 'on_fail': level} + if kind == 'item': + item_id = c.get('id') + if item_id is not None: + item_id = _int(item_id, 'item check id', 1, 255) + text = c.get('text') + if item_id is None and text is None: + raise ScriptError('item check: id or text is required') + mode = str(c.get('mode') or 'contains').lower() + if mode not in ('contains', 'exact'): + raise ScriptError('item check: mode must be contains or exact') + return {'kind': 'item', 'id': item_id, + 'text': str(text) if text is not None else None, 'mode': mode, + 'case_sensitive': bool(c.get('case_sensitive', True)), + 'on_fail': level} + if kind == 'raw': + spec = _check_spec(c.get('value') if 'value' in c else c, 'raw check') + return {'kind': 'raw', 'mode': spec['mode'], 'value': spec['value'], + 'on_fail': level} + raise ScriptError('unknown check kind %r' % c.get('kind')) + + +def normalise_respond(respond, cmd_type=None): + """Validate the scripted TERMINAL RESPONSE for an expected command.""" + if not isinstance(respond, dict): + raise ScriptError('respond must be an object') + res = respond.get('result', 0x00) + if isinstance(res, str): + key = res.strip().lower() + if re.fullmatch(r'(0x)?[0-9a-f]{2}', key): + res = int(key.replace('0x', ''), 16) + elif key in RESULT_NAMES: + res = RESULT_NAMES[key] + else: + raise ScriptError('respond: unknown result %r' % respond.get('result')) + else: + res = _int(res, 'respond result', 0, 255) + out = {'result': res} + if respond.get('item_id') is not None: + out['item_id'] = _int(respond['item_id'], 'respond item_id', 1, 255) + if respond.get('text') is not None: + out['text'] = str(respond['text']) + dcs = respond.get('dcs') + out['dcs'] = _int(dcs, 'respond dcs', 0, 255) if dcs is not None else 0x00 + extra = respond.get('raw') + if extra not in (None, ''): + out['raw'] = _data_hex(extra, 'respond raw') + return out + + +# ─── scripted TERMINAL RESPONSE ───────────────────────────────────────── + +def _encode_text(text, dcs): + if (dcs & 0x0C) == 0x08: + return text.encode('utf-16-be') + return text.encode('latin-1', 'replace') + + +def build_tr(cmd_num, cmd_type, dev_dst, dev_src, respond): + """Flat COMPREHENSION-TLV TERMINAL RESPONSE payload (TS 102 223 6.8): + command details + device identities + optional item identifier / text + string / raw TLVs + result. Matches the interactive menu TR layout.""" + out = bytearray([0x81, 0x03, cmd_num & 0xFF, cmd_type & 0xFF, 0x00]) + out += bytes([0x82, 0x02, dev_dst & 0xFF, dev_src & 0xFF]) + result = int(respond.get('result', 0)) + if respond.get('item_id') is not None and result == 0x00: + out += bytes([0x90, 0x01, respond['item_id'] & 0xFF]) + if respond.get('raw'): + out += bytes.fromhex(respond['raw']) + if respond.get('text') is not None: + dcs = int(respond.get('dcs', 0x00)) + body = _encode_text(respond['text'], dcs) + out += bytes([0x0D, len(body) + 1, dcs]) + body + out += bytes([0x83, 0x02, result & 0xFF, 0x00]) + return bytes(out) diff --git a/tests/test_ota_helpers.py b/tests/test_ota_helpers.py index 9430474..9422bd6 100644 --- a/tests/test_ota_helpers.py +++ b/tests/test_ota_helpers.py @@ -273,7 +273,7 @@ class TestSmsConcatenation(unittest.TestCase): '0000000001', apdu, K, K) sent = [] - def fake_envelope(tpdu_hex, scc, sm_sc=None, submit_handler=None): + def fake_envelope(tpdu_hex, scc, sm_sc=None, submit_handler=None, **kwargs): sent.append(tpdu_hex.upper()) return '', '9000' diff --git a/tests/test_testscript.py b/tests/test_testscript.py new file mode 100644 index 0000000..e5e1e58 --- /dev/null +++ b/tests/test_testscript.py @@ -0,0 +1,428 @@ +#!/usr/bin/env python3 +"""Tests for the test-script engine (pure parts) and the server-side runner.""" + +import sys +import time +import unittest +from pathlib import Path +from unittest import mock + +PROJECTS = Path(__file__).resolve().parents[2] +PY_SIM = PROJECTS / 'pysim' +if str(PY_SIM) not in sys.path: + sys.path.insert(0, str(PY_SIM)) + +import pysim_simple_server.server as S +import pysim_simple_server.testscript as T + + +# ─── pure engine ───────────────────────────────────────────────────────── + +def _resolver(name): + return {n.upper(): c for c, n in S.PROACTIVE_TYPE_NAMES.items()}.get(name.upper()) + + +class TestValidation(unittest.TestCase): + def test_minimal_script_normalises(self): + script = T.normalise_script({'name': 'demo', 'steps': [ + {'type': 'action', 'kind': 'menu-select', 'params': {'item_id': 128}}, + {'type': 'expect', 'command': 'SELECT ITEM', + 'checks': [{'kind': 'item', 'id': 1, 'text': 'test'}], + 'respond': {'result': 'ok', 'item_id': 1}}, + ]}, _resolver) + self.assertEqual(script['name'], 'demo') + self.assertEqual(script['steps'][0]['params']['item_id'], 128) + self.assertEqual(script['steps'][0]['check']['sw'], {'mode': 'exact', 'value': '9000'}) + self.assertEqual(script['steps'][1]['command']['type'], 0x24) + self.assertEqual(script['steps'][1]['respond'], {'result': 0x00, 'item_id': 1}) + + def test_status_poll_defaults_to_a_91xx_mask(self): + script = T.normalise_script({'steps': [ + {'type': 'action', 'kind': 'status', 'params': {'attempts': 5}}, + ]}, _resolver) + self.assertEqual(script['steps'][0]['check']['sw'], {'mode': 'mask', 'value': '91??'}) + script = T.normalise_script({'steps': [ + {'type': 'action', 'kind': 'status', 'params': {}}, + ]}, _resolver) + self.assertEqual(script['steps'][0]['check']['sw'], {'mode': 'exact', 'value': '9000'}) + + def test_rejects_bad_scripts(self): + bad = [ + 'not a dict', + {'steps': []}, + {'steps': [{'type': 'nope'}]}, + {'steps': [{'type': 'action', 'kind': 'nope'}]}, + {'steps': [{'type': 'action', 'kind': 'apdu', 'params': {}}]}, + {'steps': [{'type': 'action', 'kind': 'menu-select', 'params': {'item_id': 0}}]}, + {'steps': [{'type': 'action', 'kind': 'apdu', 'params': {'apdu': 'ABC'}}]}, + {'steps': [{'type': 'expect', 'command': 'NOT A COMMAND'}]}, + {'steps': [{'type': 'expect', 'command': 'SELECT ITEM', 'respond': {'result': 'bogus'}}]}, + {'steps': [{'type': 'action', 'kind': 'apdu', 'params': {'apdu': '00'}, + 'check': {'por': 'ok'}}]}, + {'steps': [{'type': 'action', 'kind': 'status', 'params': {'attempts': 0}}]}, + {'steps': [{'type': 'expect', 'command': 'SELECT ITEM', + 'checks': [{'kind': 'item'}]}]}, + ] + for raw in bad: + with self.assertRaises(T.ScriptError, msg=repr(raw)): + T.normalise_script(raw if isinstance(raw, dict) else raw, _resolver) + + def test_check_strings_and_masks(self): + script = T.normalise_script({'steps': [ + {'type': 'action', 'kind': 'apdu', 'params': {'apdu': '00A4'}, + 'check': {'sw': '91??', 'data': {'mode': 'mask', 'value': 'aa??'}}}, + ]}, _resolver) + check = script['steps'][0]['check'] + self.assertEqual(check['sw'], {'mode': 'mask', 'value': '91??'}) + self.assertEqual(check['data'], {'mode': 'mask', 'value': 'AA??'}) + + def test_respond_text_and_raw(self): + respond = T.normalise_respond({'result': 'ok', 'text': 'hello', 'dcs': '00', + 'raw': 'aa01bb'}) + self.assertEqual(respond['text'], 'hello') + self.assertEqual(respond['raw'], 'AA01BB') + tr = T.build_tr(4, 0x23, 0x81, 0x82, respond) + self.assertEqual(tr.hex().upper(), + '8103042300' + '82028182' + 'AA01BB' + '0D0600' + '68656C6C6F' + + '83020000') + + +class TestMatchers(unittest.TestCase): + def test_match_value(self): + self.assertTrue(T.match_value({'mode': 'exact', 'value': '9000'}, '9000')) + self.assertFalse(T.match_value({'mode': 'exact', 'value': '9000'}, '9001')) + self.assertTrue(T.match_value({'mode': 'mask', 'value': '91??'}, '9102')) + self.assertFalse(T.match_value({'mode': 'mask', 'value': '91??'}, '9002')) + self.assertFalse(T.match_value({'mode': 'exact', 'value': '9000'}, None)) + self.assertFalse(T.match_value({'mode': 'mask', 'value': '91??'}, '91')) + self.assertTrue(T.match_value(None, 'anything')) + + def test_match_text(self): + self.assertTrue(T.match_text({'mode': 'contains', 'value': 'ell'}, 'hello')) + self.assertFalse(T.match_text({'mode': 'contains', 'value': 'ELL'}, 'hello')) + self.assertTrue(T.match_text({'mode': 'contains', 'value': 'ELL', 'case_sensitive': False}, 'hello')) + self.assertTrue(T.match_text({'mode': 'exact', 'value': 'hello'}, 'hello')) + self.assertFalse(T.match_text({'mode': 'exact', 'value': 'hell'}, 'hello')) + self.assertFalse(T.match_text({'mode': 'contains', 'value': 'x'}, None)) + + def test_match_item(self): + items = [{'id': 1, 'text': 'test'}, {'id': 2, 'text': 'other'}] + self.assertTrue(T.match_item(items, {'id': 1, 'text': None, 'mode': 'contains'})[0]) + self.assertTrue(T.match_item(items, {'id': None, 'text': 'test', 'mode': 'contains'})[0]) + self.assertFalse(T.match_item(items, {'id': 3, 'text': None, 'mode': 'contains'})[0]) + self.assertFalse(T.match_item([], {'id': 1, 'text': None, 'mode': 'contains'})[0]) + + def test_combine_levels(self): + self.assertEqual(T.combine_levels(['ok', 'ok']), 'ok') + self.assertEqual(T.combine_levels(['ok', 'warning']), 'warning') + self.assertEqual(T.combine_levels(['warning', 'error']), 'error') + + +class TestBuildTr(unittest.TestCase): + def test_select_item_response_carries_the_identifier(self): + tr = T.build_tr(3, 0x24, 0x82, 0x81, {'result': 0x00, 'item_id': 7}) + self.assertEqual(tr.hex().upper(), '81030324008202828190010783020000') + + def test_cancel_response_has_no_item_identifier(self): + tr = T.build_tr(3, 0x24, 0x82, 0x81, {'result': 0x10, 'item_id': 7}) + self.assertEqual(tr.hex().upper(), '81030324008202828183021000') + + def test_display_text_result_only(self): + tr = T.build_tr(2, 0x21, 0x82, 0x81, {'result': 0x00}) + self.assertEqual(tr.hex().upper(), '81030221008202828183020000') + + +# ─── runner with a fake card ───────────────────────────────────────────── + +class FakeScc: + """APDU queue keyed by prefix; unmatched commands answer 6D00.""" + cat_cla = '80' + + def __init__(self): + self._tp = self + self.queues = {} + self.sent = [] + + def push(self, prefix, data='', sw='9000'): + self.queues.setdefault(prefix.upper(), []).append((data, sw)) + return self + + def send_apdu(self, apdu): + up = apdu.upper() + self.sent.append(up) + for prefix, queue in self.queues.items(): + if up.startswith(prefix) and queue: + return queue.pop(0) + return '', '6D00' + + +class FakeServer: + def __init__(self, scc): + self.scc = scc + self.app = None + self.sms_oa = '12345' + self.sms_sc = '12345678912' + self.card_session = 1 + self.stk_pending = None + self.menu_active = False + + +SELECT_ITEM_CMD = 'D0108103012400820281828F050174657374' # item 1: 'test' +DISPLAY_TEXT_CMD = 'D0118103022100820281828D060468656C6C6F' # 'hello' + + +class RunnerTestCase(unittest.TestCase): + def setUp(self): + self.saved = (S._TEST_RUNNING, dict(S._TEST_RUN), S._POLL_ENABLED) + S._POLL_ENABLED = False + S._TEST_RUNNING = False + + def tearDown(self): + S._TEST_RUNNING, S._POLL_ENABLED = self.saved[0], self.saved[2] + S._TEST_RUN.clear() + S._TEST_RUN.update(self.saved[1]) + + def run_script(self, server, steps, preset=None, stop=False): + script = T.normalise_script({'name': 'test', 'steps': steps}, S._test_command_type) + S._TEST_RUN.update({'running': True, 'stop': stop, 'name': 'test', + 'status': None, 'steps': [], 'index': 0, + 'total': len(script['steps']), 'scp80_counter': None, + 'started': time.time(), 'finished': None, 'error': None}) + S._TEST_RUNNING = True + try: + S._test_run_execute(server, script, preset or {}) + finally: + S._TEST_RUNNING = False + return S._TEST_RUN + + +class TestRunnerDialogue(RunnerTestCase): + def test_stk_menu_browsing_script(self): + # 1) menu selection -> SELECT ITEM announced; 2) expect it, answer with + # item 1; the TR leaves DISPLAY TEXT pending; 3) expect it, answer 00. + scc = FakeScc() + scc.push('80C2', '', '9103') # ENVELOPE(Menu Selection) + scc.push('8012', SELECT_ITEM_CMD, '9000') # FETCH + scc.push('8014', '', '9105') # TR -> DISPLAY TEXT pending + scc.push('8012', DISPLAY_TEXT_CMD, '9000') # FETCH + scc.push('8014', '', '9000') # TR -> session over + server = FakeServer(scc) + run = self.run_script(server, [ + {'type': 'action', 'kind': 'menu-select', 'params': {'item_id': 128}, + 'check': {'sw': '91??'}}, + {'type': 'expect', 'command': 'SELECT ITEM', + 'checks': [{'kind': 'item', 'id': 1, 'text': 'test'}], + 'respond': {'result': 'ok', 'item_id': 1}}, + {'type': 'expect', 'command': 'DISPLAY TEXT', + 'checks': [{'kind': 'text', 'value': 'hello'}], + 'respond': {'result': 'ok'}}, + ]) + self.assertEqual(run['status'], 'ok', run['steps']) + self.assertEqual([s['status'] for s in run['steps']], ['ok', 'ok', 'ok']) + # the SELECT ITEM TR carries the item identifier + self.assertIn('900101', run['steps'][1]['sent']) + # the DISPLAY TEXT check decoded the text + text_check = [c for c in run['steps'][2]['checks'] if c['label'] == 'Text'][0] + self.assertEqual(text_check['actual'], 'hello') + + def test_expectation_without_pending_command_is_an_error(self): + server = FakeServer(FakeScc()) + run = self.run_script(server, [ + {'type': 'expect', 'command': 'DISPLAY TEXT', + 'checks': [{'kind': 'text', 'value': 'hello'}], 'respond': {}}, + ]) + self.assertEqual(run['status'], 'error') + self.assertIn('no proactive command pending', run['steps'][0]['note']) + + def test_text_mismatch_fails_the_expectation(self): + scc = FakeScc() + scc.push('80C2', '', '9105') + scc.push('8012', DISPLAY_TEXT_CMD, '9000') + scc.push('8014', '', '9000') + run = self.run_script(FakeServer(scc), [ + {'type': 'action', 'kind': 'menu-select', 'params': {'item_id': 1}, + 'check': {'sw': '91??'}}, + {'type': 'expect', 'command': 'DISPLAY TEXT', + 'checks': [{'kind': 'text', 'value': 'goodbye'}], 'respond': {}}, + ]) + self.assertEqual(run['status'], 'error') + self.assertEqual(run['steps'][1]['status'], 'error') + text_check = [c for c in run['steps'][1]['checks'] if c['label'] == 'Text'][0] + self.assertFalse(text_check['ok']) + self.assertEqual(text_check['actual'], 'hello') + + def test_unexpected_pending_command_terminates_and_is_drained(self): + scc = FakeScc() + scc.push('80C2', '', '9103') # announces a command + scc.push('8012', SELECT_ITEM_CMD, '9000') + scc.push('8014', '', '9000') + run = self.run_script(FakeServer(scc), [ + {'type': 'action', 'kind': 'menu-select', 'params': {'item_id': 1}, + 'check': {'sw': '91??'}}, + {'type': 'action', 'kind': 'menu-select', 'params': {'item_id': 2}}, + ]) + 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) + self.assertTrue(any(a.startswith('8012') for a in scc.sent[1:])) + self.assertTrue(any('83021000' in a for a in scc.sent)) + + def test_sw_mismatch_error_stops_the_script_warning_continues(self): + scc = FakeScc() + scc.push('00A4', '6A82', '6A82') + run = self.run_script(FakeServer(scc), [ + {'type': 'action', 'kind': 'apdu', 'params': {'apdu': '00A4040008A00000015100'}, + 'check': {'sw': '9000'}, 'on_fail': 'error'}, + {'type': 'action', 'kind': 'apdu', 'params': {'apdu': '00A4040008A00000015100'}}, + ]) + self.assertEqual(run['status'], 'error') + self.assertEqual(len(run['steps']), 1) # second step never ran + + scc = FakeScc() + scc.push('00A4', 'AA', '9000') + scc.push('00A4', '', '6982') + run = self.run_script(FakeServer(scc), [ + {'type': 'action', 'kind': 'apdu', 'params': {'apdu': '00A4040008A00000015100'}, + 'check': {'sw': '6982'}, 'on_fail': 'warning'}, + {'type': 'action', 'kind': 'apdu', 'params': {'apdu': '00A4040008A00000015100'}, + 'check': {'sw': '6982'}}, + ]) + self.assertEqual(len(run['steps']), 2) # warning continued + self.assertEqual(run['status'], 'warning') + + def test_status_poll_waits_for_91xx(self): + scc = FakeScc() + scc.push('80F2', '', '9000') + scc.push('80F2', '', '9000') + scc.push('80F2', '', '9103') + scc.push('8012', SELECT_ITEM_CMD, '9000') + scc.push('8014', '', '9000') + run = self.run_script(FakeServer(scc), [ + {'type': 'action', 'kind': 'status', 'params': {'attempts': 3, 'interval_ms': 0}}, + {'type': 'expect', 'command': 'SELECT ITEM', 'respond': {'result': 'cancel'}}, + ]) + self.assertEqual(run['status'], 'ok', run['steps']) + self.assertEqual(sum(1 for a in scc.sent if a.startswith('80F2')), 3) + self.assertEqual(run['steps'][0]['sent'], 'STATUS x3') + + def test_stop_before_the_first_step(self): + run = self.run_script(FakeServer(FakeScc()), [ + {'type': 'action', 'kind': 'status', 'params': {}}, + ], stop=True) + self.assertEqual(run['status'], 'stopped') + self.assertEqual(run['steps'], []) + + +class TestRunnerActions(RunnerTestCase): + def test_scp80_uses_the_preset_and_advances_the_counter(self): + scc = FakeScc() + server = FakeServer(scc) + preset = {'kic': '15', 'kid': '15', 'kicKey': 'AA' * 16, 'kidKey': 'BB' * 16, + 'counter': '0000000A', 'tar': 'B00000', 'spi1': '16', 'spi2': '01'} + with mock.patch.object(S, '_build_secured_packet', + return_value=('AA' * 10, {})) as build, \ + mock.patch.object(S, '_send_secured_packet', + return_value={'success': True, 'sw': '9102', + 'response_data': None}) as send: + scc.push('8012', DISPLAY_TEXT_CMD, '9000') + scc.push('8014', '', '9000') + run = self.run_script(server, [ + {'type': 'action', 'kind': 'scp80', 'params': {'apdu': '80E2900000'}, + 'check': {'sw': {'mode': 'mask', 'value': '91??'}, 'por': 'none'}}, + {'type': 'expect', 'command': 'DISPLAY TEXT', + 'checks': [{'kind': 'text', 'value': 'hello'}], + 'respond': {'result': 'ok'}}, + ], preset) + self.assertEqual(run['status'], 'ok', run['steps']) + self.assertEqual(run['steps'][0]['counter'], '0000000A') + self.assertEqual(run['scp80_counter'], '0000000B') + # the counter is the preset's, TAR comes from the preset when not overridden + self.assertEqual(build.call_args[0][4], 'B00000') + self.assertEqual(build.call_args[0][5], '0000000A') + self.assertFalse(send.call_args.kwargs.get('handle_proactive', True)) + + def test_scp80_step_overrides_tar_and_spi_only(self): + scc = FakeScc() + preset = {'kic': '15', 'kid': '15', 'kicKey': 'AA' * 16, 'kidKey': 'BB' * 16, + 'counter': '00000001', 'tar': 'B00000', 'spi1': '16', 'spi2': '01'} + with mock.patch.object(S, '_build_secured_packet', return_value=('AA' * 10, {})) as build, \ + mock.patch.object(S, '_send_secured_packet', + return_value={'success': True, 'sw': '9000', + 'response_data': None}): + self.run_script(FakeServer(scc), [ + {'type': 'action', 'kind': 'scp80', + 'params': {'apdu': '80E2900000', 'tar': 'B00001', 'spi1': '17', 'spi2': '01'}}, + ], preset) + args = build.call_args[0] + self.assertEqual(args[0], '17') # spi1 override + self.assertEqual(args[4], 'B00001') # tar override + self.assertEqual(args[2], '15') # kic stays preset-owned + + def test_preset_completeness_is_validated(self): + script = T.normalise_script({'steps': [ + {'type': 'action', 'kind': 'scp80', 'params': {'apdu': '80E2900000'}}, + ]}, S._test_command_type) + self.assertIn('incomplete', S._test_preset_error(script, {}) or '') + full = {'kic': '15', 'kid': '15', 'kicKey': 'AA', 'kidKey': 'BB', + 'counter': '00000000', 'tar': 'B00000', 'spi1': '16', 'spi2': '01'} + self.assertIsNone(S._test_preset_error(script, full)) + no_tar = dict(full, tar='') + self.assertIn('TAR', S._test_preset_error(script, no_tar) or '') + + def test_file_write_and_read_actions(self): + class FakeLchan: + selected_file = None + + def __init__(self): + self.written = None + self.content = 'AA55' + def update_binary(self, data): + self.written = data + return '', '9000' + def read_binary(self): + return self.content, '9000' + + lchan = FakeLchan() + server = FakeServer(FakeScc()) + with mock.patch.object(S, '_select_path', return_value=(None, None)), \ + mock.patch.object(S, '_get_file_type', return_value='transparent'): + server.app = type('App', (), {'rs': type('Rs', (), { + 'lchan': [lchan]})()})() + run = self.run_script(server, [ + {'type': 'action', 'kind': 'file-write', + 'params': {'path': 'ADF.USIM/EF.TEST', 'data': 'AA55'}}, + {'type': 'action', 'kind': 'file-read', + 'params': {'path': 'ADF.USIM/EF.TEST'}, + 'check': {'sw': '9000', 'data': {'mode': 'mask', 'value': 'AA??'}}}, + ]) + self.assertEqual(run['status'], 'ok', run['steps']) + self.assertEqual(lchan.written, 'AA55') + self.assertEqual(run['steps'][1]['data'], 'AA55') + + def test_run_guard_blocks_card_endpoints_only(self): + S._TEST_RUNNING = True + try: + for path in ('/api/apdu', '/api/read', '/api/send-ota', '/api/menu-select', + '/api/esim/profiles', '/api/scp81/bip'): + self.assertTrue(S._test_request_blocked(path), path) + for path in ('/api/test/status', '/api/test/stop', '/api/status', + '/api/version', '/api/proactive-log', '/index.html'): + self.assertFalse(S._test_request_blocked(path), path) + self.assertFalse(S._test_request_blocked('/api/apdu?x=1') is False) + finally: + S._TEST_RUNNING = False + + def test_state_snapshot_is_a_copy(self): + S._TEST_RUN['steps'] = [{'index': 0, 'status': 'ok'}] + snap = S._test_state_snapshot() + snap['steps'][0]['status'] = 'mutated' + self.assertEqual(S._TEST_RUN['steps'][0]['status'], 'ok') + + def test_increment_counter_hex_keeps_width(self): + self.assertEqual(S._increment_counter_hex('000000FF'), '00000100') + self.assertEqual(S._increment_counter_hex('FF'), '00') + self.assertEqual(S._increment_counter_hex(''), '') + + +if __name__ == '__main__': + unittest.main()