catarrh/simple
1
0
Fork 0
Code Issues Pull Requests Actions 8 Packages Projects Releases Wiki Activity
Files
262041795ceba86ea53cab9e2ba4e21ea3c3ec54
simple/tests/test_testscript.py
T
catarrh cb40662e4e v3.26.0: per-TAR default keysets (the preset/SCP80/RAM rework) 
Every preset TAR entry carries `keyset` (int 1-15) - the keyset every
operation addressed to that TAR uses: mandatory for any non-zero MSL
(the counters are keyset-linked), optional for MSL 00 (allowed: some
cards reject the no-security 00/00 KIc/KID form).  The store fills it
automatically when the preset has a single keyset, the validation
refuses an entry naming an undefined keyset, and the schema is 6 (a v5
file loads leniently; editing it forces the fix).

The keyset selectors are gone: the SCP80 TAR field became a select of
the preset's entries (applying the TAR's keyset to the form and seeding
the SPI1 from its MSL); the RAM operations use the preset's ISD role
entry (the PWA sends only the preset id, the ISD TAR, SPI and the
counter override - the server resolves the keyset and the keys:
presets.tar_keyset, _preset_tar_keyset, _resolve_operation_keyset,
_op_keyset, used by send-ota's APDU path, ram-install, ram-install-app
and cap-compat); the counter probe and the test runner's scp80 steps
default to their TAR's keyset (an explicit kvn stays an override).  A
security TAR without a keyset refuses the operation ("set it in the
Cards tab") - never a silent keyless send - and the start buttons wait
for a usable operation-TAR keyset.

The TAR probe checks the preset's TAR list only (the standard
TAR_PROBE_TARS list and the custom-TAR localStorage are gone; a `tars`
list selects a subset, a TAR outside the preset is refused), each row
using its own MSL and default keyset.  The Cards editor's TAR table
gained a Keyset column (the only keyset preselected, an amber marker
when a security MSL lacks one).

Docs: help EN/RU, api.md's tar-probe contract, both AGENTS.  Tests: the
store (the field, the auto-fill, the schema, the string forms), the
server resolution (send-ota + the refusal, the probe's per-row keysets
and refusals), the PWA (the TAR select, the RAM ISD resolution, the
keyset-free requests) - several fixtures updated for the new schema.
891 frontend + 885 Python; sw cache simple-v434.
2026-10-14 12:21:16 +03:00

2142 lines
109 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""Tests for the test-script engine (pure parts) and the server-side runner."""
import contextlib
import io
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.events as events
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())
def preset_tars(isd='000000', msl='16', keyset=1):
"""The three mandatory role entries of a card preset's TAR table (v3.26.0:
each carries its default keyset - the fixtures name keyset 1)."""
return [{'role': 'isd', 'tar': isd, 'msl': msl, 'desc': '', 'keyset': keyset},
{'role': 'uiccRfm', 'tar': 'B00000', 'msl': '16', 'desc': '', 'keyset': keyset},
{'role': 'usimRfm', 'tar': 'B00001', 'msl': '16', 'desc': '', 'keyset': keyset}]
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_scp80_defaults_to_an_exact_9000_sw(self):
# The default must not drift to a mask: a GP INSTALL/LOAD/DELETE
# envelope answers 91XX and the script gives those steps the 9???
# mask explicitly (the docs recipe); applet scripts keep asserting
# exact 9000.
script = T.normalise_script({'steps': [
{'type': 'action', 'kind': 'scp80', 'params': {'apdu': '80E2900000'}},
]}, _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_proactive_drain_action_normalises(self):
script = T.normalise_script({'steps': [
{'type': 'action', 'kind': 'proactive-drain',
'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-drain'}]}, _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-drain',
'params': bad}]}, _resolver)
def test_the_proactive_drain_action_refuses_a_data_check(self):
# The proactive drain's data is the drained-command list, not a
# single response: a `data` check would be silently ignored, so the
# engine refuses it (assert what was drained with `require`).
with self.assertRaises(T.ScriptError) as cm:
T.normalise_script({'steps': [
{'type': 'action', 'kind': 'proactive-drain',
'check': {'data': 'AABB'}}]}, _resolver)
self.assertIn('check.data is not used by the proactive drain action',
str(cm.exception))
def test_integer_fields_accept_decimals_with_leading_zeros_and_hex(self):
script = T.normalise_script({'steps': [
{'type': 'action', 'kind': 'status',
'params': {'attempts': '08', 'interval_ms': '0x10'}},
]}, _resolver)
self.assertEqual(script['steps'][0]['params']['attempts'], 8)
self.assertEqual(script['steps'][0]['params']['interval_ms'], 16)
with self.assertRaises(T.ScriptError):
T.normalise_script({'steps': [
{'type': 'action', 'kind': 'status', 'params': {'attempts': 'x'}}]}, _resolver)
def test_respond_text_codings_and_raw(self):
# TS 102 223 8.15 recommended codings: '00' GSM default alphabet
# 7 bits packed, '04' GSM default alphabet 8 bits, '08' UCS2.
respond = T.normalise_respond({'result': 'ok', 'text': 'hello', 'dcs': '00',
'raw': 'aa01bb'})
self.assertEqual(respond['text'], 'hello')
self.assertEqual(respond['raw'], 'AA01BB')
# 6.8.0 object order: command details, device identities, Result,
# then the command-specific objects (here Text), then the raw extras.
tr = T.build_tr(4, 0x23, 0x81, 0x82, respond)
self.assertEqual(tr.hex().upper(),
'8103042300' + '02028182' + '83020000'
+ '0D0600' + 'E8329BFD06' + 'AA01BB')
# 8 bits: one octet per character
tr = T.build_tr(4, 0x23, 0x81, 0x82, {'result': 0x00, 'text': 'hello', 'dcs': 0x04})
self.assertEqual(tr.hex().upper(),
'8103042300' + '02028182' + '83020000' + '0D0604' + '68656C6C6F')
# UCS2 (UTF-16-BE): 'A' U+0041, 'Ж' U+0416
tr = T.build_tr(4, 0x23, 0x81, 0x82,
{'result': 0x00, 'text': 'A\u0416', 'dcs': 0x08})
self.assertEqual(tr.hex().upper(),
'8103042300' + '02028182' + '83020000' + '0D0508' + '00410416')
# an empty answer is the null text string (Length 00, no DCS)
tr = T.build_tr(4, 0x23, 0x81, 0x82, {'result': 0x00, 'text': ''})
self.assertEqual(tr.hex().upper(),
'8103042300' + '02028182' + '83020000' + '0D00')
def test_a_text_the_dcs_cannot_code_is_refused_at_load(self):
# v3.22.8: the load-time check uses the same coding the TERMINAL
# RESPONSE builder does - a non-GSM character with dcs 00/04 must
# fail the script at load, not mid-run after the command was fetched.
with self.assertRaises(T.ScriptError) as cm:
T.normalise_respond({'result': 'ok', 'text': 'Пароль'})
self.assertIn('dcs 00', str(cm.exception))
self.assertIn('dcs 08', str(cm.exception))
with self.assertRaises(T.ScriptError):
T.normalise_respond({'result': 'ok', 'text': 'Пароль', 'dcs': '04'})
# UCS2 accepts it; ASCII and the empty answer pass as before
self.assertEqual(T.normalise_respond({'text': 'Пароль', 'dcs': '08'})['dcs'], 0x08)
self.assertEqual(T.normalise_respond({'text': 'hello'})['text'], 'hello')
self.assertEqual(T.normalise_respond({'text': ''})['text'], '')
# a whole script is refused with the same message at run start
with self.assertRaises(T.ScriptError) as cm:
T.normalise_script({'steps': [
{'type': 'expect', 'command': 'GET INPUT',
'respond': {'result': 'ok', 'text': 'Пароль'}}]}, _resolver)
self.assertIn('cannot be coded', str(cm.exception))
def test_menu_select_params_accept_id_or_text(self):
script = T.normalise_script({'steps': [
{'type': 'action', 'kind': 'menu-select', 'params': {'item_id': 3}},
{'type': 'action', 'kind': 'menu-select',
'params': {'text': 'My menu', 'mode': 'contains'}},
{'type': 'action', 'kind': 'menu-select',
'params': {'text': 'Exact', 'mode': 'exact', 'case_sensitive': True}},
]}, _resolver)
self.assertEqual(script['steps'][0]['params'], {'item_id': 3})
# the text match is case-insensitive by default (menu labels are UI
# text); an explicit true is kept for an exact-case match
self.assertEqual(script['steps'][1]['params'],
{'text': 'My menu', 'mode': 'contains', 'case_sensitive': False})
self.assertEqual(script['steps'][2]['params'],
{'text': 'Exact', 'mode': 'exact', 'case_sensitive': True})
with self.assertRaises(T.ScriptError):
T.normalise_script({'steps': [{'type': 'action', 'kind': 'menu-select',
'params': {}}]}, _resolver)
with self.assertRaises(T.ScriptError):
T.normalise_script({'steps': [{'type': 'action', 'kind': 'menu-select',
'params': {'item_id': 1, 'text': 'x'}}]}, _resolver)
with self.assertRaises(T.ScriptError):
T.normalise_script({'steps': [{'type': 'action', 'kind': 'menu-select',
'params': {'text': 'x', 'mode': 'nope'}}]}, _resolver)
def test_event_action_normalises_the_semantic_fields(self):
script = T.normalise_script({'steps': [
{'type': 'action', 'kind': 'event',
'params': {'event': 'location status',
'fields': {'status': 1, 'mcc': '250'}}},
{'type': 'action', 'kind': 'event',
'params': {'event': '0x1D', 'src': '83',
'fields': {'status': '0', 'type': '0', 'ti': '00',
'loc_status': '0'}}},
]}, _resolver)
p = script['steps'][0]['params']
self.assertEqual(p['event'], 0x03)
self.assertEqual(p['fields'], {'status': 1, 'mcc': '250'})
self.assertNotIn('src', p)
p = script['steps'][1]['params']
self.assertEqual(p['event'], 0x1D)
self.assertEqual(p['src'], '83')
# an unknown name, a bad field and a bad source are refused before the run
with self.assertRaises(T.ScriptError):
T.normalise_script({'steps': [{'type': 'action', 'kind': 'event',
'params': {'event': 'nope'}}]}, _resolver)
with self.assertRaises(T.ScriptError):
T.normalise_script({'steps': [{'type': 'action', 'kind': 'event',
'params': {'event': '0x03', 'fields': {'status': 9}}}]}, _resolver)
with self.assertRaises(T.ScriptError):
T.normalise_script({'steps': [{'type': 'action', 'kind': 'event',
'params': {'event': 3, 'src': 'zz'}}]}, _resolver)
# the raw envelope action carries the same source override
script = T.normalise_script({'steps': [
{'type': 'action', 'kind': 'envelope',
'params': {'event': 0x00, 'data': '', 'src': '83'}}]}, _resolver)
self.assertEqual(script['steps'][0]['params'],
{'event': 0x00, 'data': '', 'src': '83'})
with self.assertRaises(T.ScriptError):
T.normalise_script({'steps': [{'type': 'action', 'kind': 'envelope',
'params': {'event': 3, 'src': 'zz'}}]}, _resolver)
def test_event_action_accepts_the_form_built_data(self):
# The events events.py does not model carry the PWA form's built hex;
# a modelled event builds from `fields` (a stray data hex is ignored).
script = T.normalise_script({'steps': [
{'type': 'action', 'kind': 'event',
'params': {'event': '0x00', 'data': '1c0100', 'src': '83'}}]}, _resolver)
self.assertEqual(script['steps'][0]['params'],
{'event': 0x00, 'data': '1C0100', 'src': '83'})
script = T.normalise_script({'steps': [
{'type': 'action', 'kind': 'event',
'params': {'event': '0x03', 'fields': {'status': 1}, 'data': 'AA'}}]}, _resolver)
self.assertNotIn('data', script['steps'][0]['params'])
self.assertEqual(script['steps'][0]['params']['fields'], {'status': 1})
# missing, malformed and oversized data are refused before the run
for bad in ({'event': '0x00'},
{'event': '0x00', 'data': 'ABC'},
{'event': '0x00', 'data': 'AA' * (events.EVENT_DATA_MAX + 1)}):
with self.assertRaises(T.ScriptError):
T.normalise_script({'steps': [{'type': 'action', 'kind': 'event',
'params': bad}]}, _resolver)
def test_scp80_keyset_number_and_new_content_checks(self):
# kvn (1..15) selects the preset's keyset; `por`/`files` are expect
# content checks (the PoR contents / the REFRESH file list).
script = T.normalise_script({'steps': [
{'type': 'action', 'kind': 'scp80',
'params': {'apdu': '80E2900000', 'kvn': 2}},
{'type': 'expect', 'command': 'SEND SHORT MESSAGE',
'checks': [{'kind': 'por', 'status': 'por_ok', 'sw': '6A8?',
'data': {'mode': 'mask', 'value': 'AA??'}}]},
{'type': 'expect', 'command': 'REFRESH',
'checks': [{'kind': 'files', 'value': '3F007F106F3A, 3F002FE2'}]},
]}, _resolver)
self.assertEqual(script['steps'][0]['params']['kvn'], 2)
por = script['steps'][1]['checks'][0]
self.assertEqual(por['status'], 'por_ok')
self.assertEqual(por['sw'], {'mode': 'mask', 'value': '6A8?'})
self.assertEqual(por['data'], {'mode': 'mask', 'value': 'AA??'})
self.assertEqual(script['steps'][2]['checks'][0]['files'],
['3F007F106F3A', '3F002FE2'])
# a keyset number outside 1..15, an empty por check and a bad path are
# refused before anything runs
with self.assertRaises(T.ScriptError):
T.normalise_script({'steps': [
{'type': 'action', 'kind': 'scp80',
'params': {'apdu': '80E2', 'kvn': 0}}]}, _resolver)
with self.assertRaises(T.ScriptError):
T.normalise_script({'steps': [
{'type': 'expect', 'command': 'ANY',
'checks': [{'kind': 'por'}]}]}, _resolver)
with self.assertRaises(T.ScriptError):
T.normalise_script({'steps': [
{'type': 'expect', 'command': 'ANY',
'checks': [{'kind': 'files', 'value': ['zz']}]}]}, _resolver)
def test_scp80_format_param_and_por_object(self):
# v3.19.0: the scp80 step may wrap the C-APDU in the expanded Command
# Scripting template (TS 102 226 5.2.1) and assert the decoded PoR
# inline (`check.por` as an object).
script = T.normalise_script({'steps': [
{'type': 'action', 'kind': 'scp80',
'params': {'apdu': '80E2900000', 'format': 'expanded'},
'check': {'por': {'status': 'por_ok', 'sw': '6A8?',
'data': {'mode': 'mask', 'value': 'AA??'}}}},
]}, _resolver)
params = script['steps'][0]['params']
self.assertEqual(params['format'], 'expanded')
por = script['steps'][0]['check']['por']
self.assertEqual(por['status'], 'por_ok')
self.assertEqual(por['sw'], {'mode': 'mask', 'value': '6A8?'})
self.assertEqual(por['data'], {'mode': 'mask', 'value': 'AA??'})
# both codings are accepted; an unknown one is refused
for fmt in ('compact', 'expanded-ae'):
script = T.normalise_script({'steps': [
{'type': 'action', 'kind': 'scp80',
'params': {'apdu': '80E2', 'format': fmt}}]}, _resolver)
self.assertEqual(script['steps'][0]['params']['format'], fmt)
with self.assertRaises(T.ScriptError):
T.normalise_script({'steps': [
{'type': 'action', 'kind': 'scp80',
'params': {'apdu': '80E2', 'format': 'auto'}}]}, _resolver)
# an empty PoR object has nothing to assert; the object is scp80-only
with self.assertRaises(T.ScriptError):
T.normalise_script({'steps': [
{'type': 'action', 'kind': 'scp80', 'params': {'apdu': '80E2'},
'check': {'por': {}}}]}, _resolver)
with self.assertRaises(T.ScriptError):
T.normalise_script({'steps': [
{'type': 'action', 'kind': 'status', 'params': {'attempts': 1},
'check': {'por': {'status': 'por_ok'}}}]}, _resolver)
def test_alpha_and_sms_content_checks(self):
# `alpha` reads the Alpha identifier ('05', Annex A coded); `sms`
# asserts the SEND SHORT MESSAGE TPDU fields (submit only).
script = T.normalise_script({'steps': [
{'type': 'expect', 'command': 'DISPLAY TEXT',
'checks': [{'kind': 'alpha', 'mode': 'contains', 'value': 'Alfa',
'case_sensitive': False}]},
{'type': 'expect', 'command': 'SEND SHORT MESSAGE',
'checks': [{'kind': 'sms', 'da': '12 345', 'pid': '7F', 'dcs': '00',
'udl': 5, 'ud': '0271??AABB'}]},
]}, _resolver)
alpha = script['steps'][0]['checks'][0]
self.assertEqual(alpha['kind'], 'alpha')
self.assertEqual(alpha['mode'], 'contains')
self.assertFalse(alpha['case_sensitive'])
sms = script['steps'][1]['checks'][0]
self.assertEqual(sms['da'], '12345')
self.assertEqual(sms['pid'], {'mode': 'exact', 'value': '7F'})
self.assertEqual(sms['dcs'], {'mode': 'exact', 'value': '00'})
self.assertEqual(sms['udl'], 5)
self.assertEqual(sms['ud'], {'mode': 'mask', 'value': '0271??AABB'})
# a leading '+' (the TPDU keeps the international flag in its
# type-of-number) and whitespace are dropped from `da`
script = T.normalise_script({'steps': [
{'type': 'expect', 'command': 'SEND SHORT MESSAGE',
'checks': [{'kind': 'sms', 'da': '+79 332 505 884'}]},
]}, _resolver)
self.assertEqual(script['steps'][0]['checks'][0]['da'], '79332505884')
for check in ({'kind': 'alpha'}, # no value
{'kind': 'alpha', 'mode': 'nope', 'value': 'x'},
{'kind': 'sms'}, # no fields
{'kind': 'sms', 'da': '12X'},
{'kind': 'sms', 'udl': 256}):
with self.assertRaises(T.ScriptError):
T.normalise_script({'steps': [
{'type': 'expect', 'command': 'ANY', 'checks': [check]}]}, _resolver)
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):
# TS 102 223 6.8.0: the item identifier (F) follows the Result (C)
tr = T.build_tr(3, 0x24, 0x82, 0x81, {'result': 0x00, 'item_id': 7})
self.assertEqual(tr.hex().upper(),
'81030324000202828183020000100107')
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(), '81030324000202828183021000')
def test_display_text_result_only(self):
tr = T.build_tr(2, 0x21, 0x82, 0x81, {'result': 0x00})
self.assertEqual(tr.hex().upper(), '81030221000202828183020000')
def test_the_terminal_answer_objects_follow_the_result(self):
# The runner passes the terminal's own objects (the PLI data, the
# POLL INTERVAL echo) - they sit right after the Result, before the
# text/item/raw objects (TS 102 223 6.8.0).
tr = T.build_tr(1, 0x26, 0x82, 0x81, {'result': 0x00},
answer_tlvs='2607AABBCCDDEEFF')
self.assertEqual(tr.hex().upper(),
'810301260002028281830200002607AABBCCDDEEFF')
# with a text object the answer object keeps its place
tr = T.build_tr(1, 0x23, 0x82, 0x81,
{'result': 0x00, 'text': 'hi', 'dcs': 0x04},
answer_tlvs='8402011E')
self.assertEqual(tr.hex().upper(),
'810301230002028281830200008402011E0D03046869')
# and the script's raw TLVs stay last
tr = T.build_tr(1, 0x26, 0x82, 0x81,
{'result': 0x00, 'raw': '9B0100'}, answer_tlvs='2607AA')
self.assertEqual(tr.hex().upper(),
'810301260002028281830200002607AA9B0100')
# ─── 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.card_present = True
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,
'suite': None, 'log': [], 'log_prefix': '', 'adm': 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('100101', 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.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_a_command_mismatch_honours_on_fail(self):
# v3.22.3: a different command pending is a check failure at the
# step's on_fail level, like every other expectation check - so
# on_fail "warning" expresses "the applet either asks X or does Y"
# (the mismatched command is still fetched and answered).
scc = FakeScc()
scc.push('80F2', '', '9102') # STATUS: DISPLAY TEXT pending
scc.push('8012', DISPLAY_TEXT_CMD, '9000')
scc.push('8014', '', '9000')
run = self.run_script(FakeServer(scc), [
{'type': 'action', 'kind': 'status', 'params': {'attempts': 2}},
{'type': 'expect', 'command': 'SEND SHORT MESSAGE',
'respond': {'result': 'ok'}, 'on_fail': 'warning'},
])
self.assertEqual(run['status'], 'warning', run['steps'])
check = run['steps'][1]['checks'][0]
self.assertEqual(check['label'], 'Command')
self.assertFalse(check['ok'])
self.assertEqual(check['level'], 'warning')
self.assertIn('DISPLAY TEXT', check['actual'])
self.assertTrue(run['steps'][1]['sent'].startswith('TR '))
# the default (error) still stops the run
scc2 = FakeScc()
scc2.push('80F2', '', '9102')
scc2.push('8012', DISPLAY_TEXT_CMD, '9000')
scc2.push('8014', '', '9000')
run2 = self.run_script(FakeServer(scc2), [
{'type': 'action', 'kind': 'status', 'params': {'attempts': 2}},
{'type': 'expect', 'command': 'SEND SHORT MESSAGE',
'respond': {'result': 'ok'}},
])
self.assertEqual(run2['status'], 'error')
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 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('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()
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_failed_thread_start_unblocks_the_card(self):
script = T.normalise_script({'steps': [
{'type': 'action', 'kind': 'status', 'params': {}}]}, S._test_command_type)
server = FakeServer(FakeScc())
with mock.patch.object(S.threading, 'Thread', side_effect=RuntimeError('no threads')):
with self.assertRaises(RuntimeError):
S._test_run_start(server, script, {})
self.assertFalse(S._TEST_RUNNING)
self.assertFalse(S._TEST_RUN['running'])
self.assertEqual(S._TEST_RUN['status'], 'error')
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 TestRunnerProactiveDrain(RunnerTestCase):
"""The proactive drain 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-drain', '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-drain', '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_a_drain_consumes_a_command_left_pending(self):
# v3.22.4: the drain is exempt from the runner's pending guard - a
# step that ended 91XX is consumed by the next drain (its STATUS poll
# re-announces the unfetched command), no "unexpected pending" error.
scc = FakeScc()
scc.push('80F2', '', '9102') # the status step: pending
scc.push('80F2', '', '9102') # the drain's first STATUS
scc.push('8012', DISPLAY_TEXT_CMD, '9000')
scc.push('8014', '', '9000')
scc.push('80F2', '', '9000') # the drain's confirming STATUS
run = self.run_script(FakeServer(scc), [
{'type': 'action', 'kind': 'status', 'params': {'attempts': 2}},
{'type': 'action', 'kind': 'proactive-drain',
'params': {'attempts': 3, 'interval_ms': 0}},
])
self.assertEqual(run['status'], 'ok', run['steps'])
self.assertEqual(len(run['steps']), 2)
entry = run['steps'][1]
self.assertEqual(entry['status'], 'ok')
self.assertEqual(len(entry['drained']), 1)
self.assertEqual(entry['drained'][0]['type_name'], 'DISPLAY TEXT')
self.assertEqual(entry['sent'], 'STATUS x2; TR x1')
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-drain',
'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-drain',
'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-drain',
'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-drain',
'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-drain', '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 TestRunnerTerminalAnswers(RunnerTestCase):
"""The runner's TERMINAL RESPONSE inherits the terminal's own objects
(v3.22.2 review): the PLI dictionary / host clock and the POLL INTERVAL
echo - unless the script's `raw` overrides them."""
PLI_01 = 'D009810301260182028281' # PROVIDE LOCAL INFORMATION, qual 01
PLI_03 = 'D009810301260382028281' # qual 03: date/time
POLL = 'D009810301030082028281' # POLL INTERVAL
def _tr(self, scc):
return [a for a in scc.sent if a.startswith('8014')][-1]
def _pending(self, scc, cmd):
scc.push('80F2', '', '9109') # STATUS: the command is pending
scc.push('8012', cmd, '9000')
scc.push('8014', '', '9000')
def test_a_pli_expectation_carries_the_dictionary_entry(self):
scc = FakeScc()
self._pending(scc, self.PLI_01)
with mock.patch.dict(S._PLI_DATA, {1: '140853937640650700F4'}):
run = self.run_script(FakeServer(scc), [
{'type': 'action', 'kind': 'status', 'params': {'attempts': 2}},
{'type': 'expect', 'command': 'PROVIDE LOCAL INFORMATION',
'respond': {'result': 'ok'}},
])
self.assertEqual(run['status'], 'ok', run['steps'])
self.assertIn('140853937640650700F4', self._tr(scc))
def test_a_pli_raw_overrides_the_dictionary(self):
scc = FakeScc()
self._pending(scc, self.PLI_01)
with mock.patch.dict(S._PLI_DATA, {1: '140853937640650700F4'}):
run = self.run_script(FakeServer(scc), [
{'type': 'action', 'kind': 'status', 'params': {'attempts': 2}},
{'type': 'expect', 'command': 'PROVIDE LOCAL INFORMATION',
'respond': {'result': 'ok', 'raw': '2607AABBCCDDEEFF'}},
])
self.assertEqual(run['status'], 'ok', run['steps'])
tr = self._tr(scc)
self.assertIn('2607AABBCCDDEEFF', tr)
self.assertNotIn('140853937640650700F4', tr)
def test_a_pli_date_time_answer_uses_the_host_clock(self):
scc = FakeScc()
self._pending(scc, self.PLI_03)
run = self.run_script(FakeServer(scc), [
{'type': 'action', 'kind': 'status', 'params': {'attempts': 2}},
{'type': 'expect', 'command': 'PROVIDE LOCAL INFORMATION',
'respond': {'result': 'ok'}},
])
self.assertEqual(run['status'], 'ok', run['steps'])
self.assertRegex(self._tr(scc), r'2607[0-9A-F]{14}')
def test_a_refused_pli_gets_no_data_object(self):
scc = FakeScc()
self._pending(scc, self.PLI_01)
with mock.patch.dict(S._PLI_DATA, {1: '140853937640650700F4'}):
run = self.run_script(FakeServer(scc), [
{'type': 'action', 'kind': 'status', 'params': {'attempts': 2}},
{'type': 'expect', 'command': 'PROVIDE LOCAL INFORMATION',
'respond': {'result': 'refused'}},
])
self.assertEqual(run['status'], 'ok', run['steps'])
self.assertNotIn('140853937640650700F4', self._tr(scc))
def test_a_poll_interval_expectation_echoes_the_duration(self):
scc = FakeScc()
self._pending(scc, self.POLL)
run = self.run_script(FakeServer(scc), [
{'type': 'action', 'kind': 'status', 'params': {'attempts': 2}},
{'type': 'expect', 'command': 'POLL INTERVAL',
'respond': {'result': 'ok'}},
])
self.assertEqual(run['status'], 'ok', run['steps'])
self.assertRegex(self._tr(scc), r'840201[0-9A-F]{2}')
def test_a_drained_pli_carries_the_dictionary_entry(self):
scc = FakeScc()
scc.push('80F2', '', '9109') # STATUS: PLI pending
scc.push('8012', self.PLI_01, '9000')
scc.push('8014', '', '9000')
scc.push('80F2', '', '9000') # confirming STATUS: idle
with mock.patch.dict(S._PLI_DATA, {1: '140853937640650700F4'}):
run = self.run_script(FakeServer(scc), [
{'type': 'action', 'kind': 'proactive-drain',
'params': {'attempts': 3, 'interval_ms': 0}},
])
self.assertEqual(run['status'], 'ok', run['steps'])
self.assertIn('140853937640650700F4', self._tr(scc))
class TestRunnerActions(RunnerTestCase):
def test_script_menu_select_mirrors_menu_active(self):
# 9000 -> the menu dialogue is over
server = FakeServer(FakeScc().push('80C2', '', '9000'))
self.run_script(server, [{'type': 'action', 'kind': 'menu-select', 'params': {'item_id': 1}}])
self.assertFalse(server.menu_active)
# 91XX -> a menu dialogue is pending (drained at the end)
scc = FakeScc()
scc.push('80C2', '', '9103')
scc.push('8012', SELECT_ITEM_CMD, '9000')
scc.push('8014', '', '9000')
server = FakeServer(scc)
self.run_script(server, [{'type': 'action', 'kind': 'menu-select',
'params': {'item_id': 1}, 'check': {'sw': '91??'}}])
self.assertTrue(server.menu_active)
# SET UP MENU (title 'Menu', item 1 = 'One') as the card sends it.
SETUP_MENU_CMD = 'D01581030125008202828185044D656E758F04014F6E65'
def test_expect_updates_the_cached_menu_and_text_selection_resolves(self):
# The item ids vary with the applet's install parameters: the menu the
# card sends must refresh the cache, and a text selection resolves
# against it (the live id is not in the script).
scc = FakeScc()
scc.push('80F2', '', '9102')
scc.push('8012', self.SETUP_MENU_CMD, '9000')
scc.push('8014', '', '9000')
scc.push('80C2', '', '9000')
server = FakeServer(scc)
run = self.run_script(server, [
{'type': 'action', 'kind': 'status', 'params': {'attempts': 2},
'check': {'sw': {'mode': 'mask', 'value': '91??'}}},
{'type': 'expect', 'command': 'SET UP MENU', 'respond': {'result': 'ok'}},
{'type': 'action', 'kind': 'menu-select', 'params': {'text': 'One'}},
])
self.assertEqual(run['status'], 'ok', run['steps'])
self.assertEqual(server.sim_menu['items'],
[{'id': 1, 'text': 'One'}])
# the resolved Menu Selection TLV carries the item id the card gave
self.assertIn('D30702020181900101', ''.join(scc.sent))
def test_menu_text_selection_reports_the_available_items(self):
scc = FakeScc()
server = FakeServer(scc)
server.sim_menu = {'title': 'Menu', 'items': [{'id': 3, 'text': 'Alpha'}]}
run = self.run_script(server, [
{'type': 'action', 'kind': 'menu-select', 'params': {'text': 'Beta'}},
])
self.assertEqual(run['status'], 'error')
note = run['steps'][0].get('note') or ''
self.assertIn("no menu item matches 'Beta'", note)
self.assertIn("3='Alpha'", note)
self.assertEqual(scc.sent, [])
# The live card's menu (captured 2026-10-04 20:58): Annex A Variant 2
# Cyrillic items under the card's ids 128..130 - the compressed
# `81 <num_chars> 08 <chars>` form (the menu-select text must resolve
# against it; the card's ids stay out of the script).
LIVE_MENU = {'title': 'GPB MOBILE', 'items': [
{'id': 128, 'text': 'Выбор сети'}, {'id': 129, 'text': 'Язык'},
{'id': 130, 'text': 'Кофе'}]}
def test_menu_text_selection_matches_variant2_cyrillic(self):
# a lowercase exact match resolves by default (case-insensitive)
scc = FakeScc().push('80C2', '', '9000')
server = FakeServer(scc)
server.sim_menu = dict(self.LIVE_MENU)
run = self.run_script(server, [
{'type': 'action', 'kind': 'menu-select',
'params': {'text': 'кофе'}},
])
self.assertEqual(run['status'], 'ok', run['steps'])
self.assertIn('D30702020181900182', ''.join(scc.sent))
# a lowercase contains match resolves too (item 128)
scc = FakeScc().push('80C2', '', '9000')
server = FakeServer(scc)
server.sim_menu = dict(self.LIVE_MENU)
run = self.run_script(server, [
{'type': 'action', 'kind': 'menu-select',
'params': {'text': 'ыбор', 'mode': 'contains'}},
])
self.assertEqual(run['status'], 'ok', run['steps'])
self.assertIn('D30702020181900180', ''.join(scc.sent))
# an explicit case-sensitive spec misses and names the loose match
server = FakeServer(FakeScc())
server.sim_menu = dict(self.LIVE_MENU)
run = self.run_script(server, [
{'type': 'action', 'kind': 'menu-select',
'params': {'text': 'кофе', 'case_sensitive': True}},
])
self.assertEqual(run['status'], 'error')
note = run['steps'][0].get('note') or ''
self.assertIn("no menu item matches 'кофе'", note)
self.assertIn('case-insensitive match exists', note)
self.assertIn("130='Кофе'", note)
def test_menu_text_selection_refuses_an_ambiguous_match(self):
scc = FakeScc()
server = FakeServer(scc)
server.sim_menu = {'items': [{'id': 1, 'text': 'Alpha one'},
{'id': 2, 'text': 'Alpha two'}]}
run = self.run_script(server, [
{'type': 'action', 'kind': 'menu-select',
'params': {'text': 'Alpha', 'mode': 'contains'}},
])
self.assertEqual(run['status'], 'error')
note = run['steps'][0].get('note') or ''
self.assertIn('2 items match', note)
self.assertIn('use the item id', note)
def test_event_action_sends_the_built_object(self):
scc = FakeScc()
scc.push('80C2', '', '9000')
buf = io.StringIO()
with contextlib.redirect_stderr(buf):
run = self.run_script(FakeServer(scc), [
{'type': 'action', 'kind': 'event',
'params': {'event': 'location status',
'fields': {'status': '0', 'mcc': '250', 'mnc': '01',
'lac': '00FF', 'cell': '0001'}}},
])
self.assertEqual(run['status'], 'ok', run['steps'])
apdu = scc.sent[0].upper()
# event list + device identities (terminal default) + the built object
self.assertIn('99010382028281' + '9B0100930752F01000FF0001', apdu)
# the runner logs one step-annotated ENVELOPE line, not the generic
# sender pair as well (its line carries the name, src and SW)
log = buf.getvalue()
self.assertIn('TEST-RUN step 1: ENVELOPE(Event Download) type=0x03 ' '(location_status) data=9B0100930752F01000FF0001 src=82 -> SW=9000',
log)
self.assertNotIn('ENVELOPE(Event Download): type=', log)
self.assertNotIn('ENVELOPE SW:', log)
# the src override names the network as the source
scc2 = FakeScc()
scc2.push('80C2', '', '9000')
buf2 = io.StringIO()
with contextlib.redirect_stderr(buf2):
run = self.run_script(FakeServer(scc2), [
{'type': 'action', 'kind': 'event',
'params': {'event': '0x0B', 'fields': {'tech': '8'}, 'src': '83'}},
])
self.assertEqual(run['status'], 'ok', run['steps'])
apdu = scc2.sent[0].upper()
self.assertIn('99010B82028381' + 'BF0108', apdu)
self.assertIn('src=83 -> SW=9000', buf2.getvalue())
# the raw envelope action takes the same source override
scc3 = FakeScc()
scc3.push('80C2', '', '9000')
run = self.run_script(FakeServer(scc3), [
{'type': 'action', 'kind': 'envelope',
'params': {'event': 0x00, 'data': '', 'src': '83'}},
])
self.assertEqual(run['status'], 'ok', run['steps'])
self.assertIn('99010082028381', scc3.sent[0].upper())
def test_event_action_sends_the_form_built_data(self):
# The events events.py does not model (e.g. MT call) run with the
# PWA form's built hex and its source (network).
scc = FakeScc()
scc.push('80C2', '', '9000')
buf = io.StringIO()
with contextlib.redirect_stderr(buf):
run = self.run_script(FakeServer(scc), [
{'type': 'action', 'kind': 'event',
'params': {'event': '0x00', 'data': '1C0100', 'src': '83'}},
])
self.assertEqual(run['status'], 'ok', run['steps'])
self.assertIn('99010082028381' + '1C0100', scc.sent[0].upper())
self.assertIn('type=0x00 (0x00) data=1C0100 src=83 -> SW=9000',
buf.getvalue())
def test_scp80_uses_the_preset_and_advances_the_counter(self):
scc = FakeScc()
server = FakeServer(scc)
# the ISD entry carries a distinctive TAR: the assertion below must
# fail if the runner ever falls back to a built-in default
preset = {'kic': '15', 'kid': '15', 'kicKey': 'AA' * 16, 'kidKey': 'BB' * 16,
'counter': '0000000A', 'tars': preset_tars(isd='B00000')}
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_source_selects_between_apdu_and_packet(self):
script = T.normalise_script({'steps': [
{'type': 'action', 'kind': 'scp80',
'params': {'source': 'apdu', 'apdu': '80E2900000', 'sp': 'AABB'}},
]}, _resolver)
params = script['steps'][0]['params']
self.assertIn('apdu', params)
self.assertNotIn('sp', params)
script = T.normalise_script({'steps': [
{'type': 'action', 'kind': 'scp80', 'params': {'source': 'sp', 'sp': 'AABB', 'apdu': '80E2'}},
]}, _resolver)
self.assertIn('sp', script['steps'][0]['params'])
self.assertNotIn('apdu', script['steps'][0]['params'])
with self.assertRaises(T.ScriptError):
T.normalise_script({'steps': [
{'type': 'action', 'kind': 'scp80', 'params': {'source': 'sp', 'apdu': '80E2'}}]}, _resolver)
with self.assertRaises(T.ScriptError):
T.normalise_script({'steps': [
{'type': 'action', 'kind': 'scp80', 'params': {'source': 'nope', 'apdu': '80E2'}}]}, _resolver)
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', 'tars': preset_tars()}
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_scp80_step_selects_the_keyset_by_kvn(self):
# the step's keyset number picks the keyset (v3.15.0: the normaliser
# dropped `kvn`, so scripts silently used the preset's first keyset)
scc = FakeScc()
preset = {'keysets': [
{'kic': '15', 'kid': '15', 'kicKey': 'AA' * 16,
'kidKey': 'BB' * 16, 'cntr': '00000001'},
{'kic': '25', 'kid': '25', 'kicKey': 'CC' * 16,
'kidKey': 'DD' * 16, 'cntr': '00000020'}],
'tars': preset_tars()}
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}):
run = self.run_script(FakeServer(scc), [
{'type': 'action', 'kind': 'scp80',
'params': {'apdu': '80E2900000', 'kvn': 2}},
], preset)
self.assertEqual(run['status'], 'ok', run['steps'])
args = build.call_args[0]
self.assertEqual(args[2], '25') # the second keyset's KIc/KID
self.assertEqual(args[3], '25')
self.assertEqual(args[5], '00000020') # and its counter
self.assertEqual(args[7], 'CC' * 16)
self.assertEqual(args[8], 'DD' * 16)
self.assertEqual(run['scp80_counters'], {2: '00000021'})
self.assertEqual(run['scp80_counter'], '00000021')
def test_preset_completeness_validates_the_step_keyset(self):
script = T.normalise_script({'steps': [
{'type': 'action', 'kind': 'scp80',
'params': {'apdu': '80E2900000', 'kvn': 3}}]}, S._test_command_type)
preset = {'keysets': [{'kic': '15', 'kid': '15', 'kicKey': 'AA', 'kidKey': 'BB',
'cntr': '00000000'}],
'tars': preset_tars()}
self.assertIn('keyset 3', S._test_preset_error(script, preset) or '')
good = T.normalise_script({'steps': [
{'type': 'action', 'kind': 'scp80',
'params': {'apdu': '80E2900000', 'kvn': 1}}]}, S._test_command_type)
self.assertIsNone(S._test_preset_error(good, preset))
def test_run_preset_resolves_the_stored_id(self):
class Store:
def get(self, pid):
return {'id': pid, 'name': 'Card 1'} if pid == 'p1' else None
import types
server = types.SimpleNamespace(card_presets=Store())
preset, err = S._test_run_preset_from_body(server, {'preset_id': 'p1'})
self.assertEqual(preset['name'], 'Card 1')
self.assertIsNone(err)
preset, err = S._test_run_preset_from_body(server, {'preset_id': 'nope'})
self.assertIsNone(preset)
self.assertIn('not found', err)
inline = {'name': 'inline'}
preset, err = S._test_run_preset_from_body(server, {'preset': inline})
self.assertIs(preset, inline)
self.assertIsNone(err)
def test_run_script_resolves_the_stored_id(self):
import types
entry = {'id': 'a' * 32, 'name': 'stored',
'steps': [{'type': 'action', 'kind': 'status',
'params': {'attempts': 1}}]}
class Store:
def get(self, sid):
return entry if sid == entry['id'] else None
server = types.SimpleNamespace(test_scripts=Store())
# the inline script wins (the PWA sends the edited copy) - the id is
# carried for the report
raw, sid, err = S._test_run_script_from_body(
server, {'script': {'name': 'inline', 'steps': [{}]},
'script_id': entry['id']})
self.assertEqual(raw['name'], 'inline')
self.assertEqual(sid, entry['id'])
self.assertIsNone(err)
# script_id alone resolves from the store
raw, sid, err = S._test_run_script_from_body(server, {'script_id': entry['id']})
self.assertEqual(raw, {'name': 'stored', 'steps': entry['steps'],
'require_adm': False})
self.assertEqual(sid, entry['id'])
self.assertIsNone(err)
raw, sid, err = S._test_run_script_from_body(server, {'script_id': 'nope'})
self.assertIsNone(raw)
self.assertIn('not found', err)
# neither form: the normaliser reports the missing script
raw, sid, err = S._test_run_script_from_body(server, {})
self.assertIsNone(raw)
self.assertIsNone(sid)
self.assertIsNone(err)
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', 'tars': preset_tars()}
self.assertIsNone(S._test_preset_error(script, full))
no_tar = dict(full, tars=[])
self.assertIn('TAR', S._test_preset_error(script, no_tar) or '')
# A SEND SHORT MESSAGE announcing the PoR-in-submit (SPI2 bit 0x20): the
# TPDU carries a submit PDUs with UD = 02 71 00 AA BB (the deliver-style
# PoR header); the runner must decode it with the scp80 step's context.
SEND_SM_POR_CMD = 'D0188103011300820283810B0D01000181127F0005027100AABB'
# A REFRESH (qualifier 00 = NAA init + full FCN) with the file list
# 12 07 01 3F007F106F3A (one file: MF/7F10/6F3A).
REFRESH_FCN_CMD = 'D0128103010100820283811207013F007F106F3A'
def test_the_run_log_carries_the_card_exchange_details(self):
# the semantic lines a verification needs: the plaintext C-APDU, the
# built secured packet, the FETCH/CMD/TR pair of an expectation and
# the menu selection's ENVELOPE with the resolved item
scc = FakeScc()
scc.push('80F2', '', '9102')
scc.push('8012', self.SETUP_MENU_CMD, '9000')
scc.push('8014', '', '9000')
scc.push('80C2', '', '9000')
preset = {'kic': '15', 'kid': '15', 'kicKey': 'AA' * 16, 'kidKey': 'BB' * 16,
'counter': '0000000A', 'tars': preset_tars()}
buf = io.StringIO()
with mock.patch.object(S, '_build_secured_packet', return_value=('CC' * 10, {})), \
mock.patch.object(S, '_send_secured_packet',
return_value={'success': True, 'sw': '9000',
'response_data': None}):
with contextlib.redirect_stderr(buf):
run = self.run_script(FakeServer(scc), [
{'type': 'action', 'kind': 'scp80',
'params': {'apdu': '80E2900000'}},
{'type': 'action', 'kind': 'status', 'params': {'attempts': 2},
'check': {'sw': {'mode': 'mask', 'value': '91??'}}},
{'type': 'expect', 'command': 'SET UP MENU',
'respond': {'result': 'ok'}},
{'type': 'action', 'kind': 'menu-select',
'params': {'text': 'One'}},
], preset)
self.assertEqual(run['status'], 'ok', run['steps'])
log = buf.getvalue()
self.assertIn('TEST-RUN step 1: SCP80 C-APDU=80E2900000', log)
self.assertIn('TEST-RUN step 1: SCP80 SECURED=' + 'CC' * 10, log)
self.assertIn('TEST-RUN step 1: SCP80 -> SW=9000', log)
self.assertIn('TEST-RUN step 1: PoR[inline] none', log)
self.assertIn('TEST-RUN step 2: STATUS 1/2 -> 9102', log)
self.assertIn('TEST-RUN step 3: FETCH=', log)
self.assertIn('TEST-RUN step 3: CMD 0x25 SET UP MENU', log)
self.assertIn('TEST-RUN step 3: TR=', log)
self.assertIn('TEST-RUN step 4: MENU-SELECT ENVELOPE=', log)
self.assertIn("item=1 ('One')", log)
# the report's label carries the plaintext APDU too (log/report parity)
self.assertIn('C-APDU=80E2900000', run['steps'][0]['sent'])
def test_the_run_log_carries_the_inline_por(self):
# the decoded PoR and its R-APDU are logged with the raw packet
scc = FakeScc()
preset = {'kic': '15', 'kid': '15', 'kicKey': 'AA' * 16, 'kidKey': 'BB' * 16,
'counter': '0000000A', 'tars': preset_tars()}
decoded = {'response_status': 'por_ok', 'tar': '000000', 'cntr': '0000000A',
'raw': '0271000021AABB',
'decoded': {'last_status_word': '9000', 'last_response_data': 'AABB'}}
buf = io.StringIO()
with mock.patch.object(S, '_build_secured_packet', return_value=('AA' * 10, {})), \
mock.patch.object(S, '_send_secured_packet',
return_value={'success': True, 'sw': '9000',
'response_data': '0271000021AABB'}), \
mock.patch.object(S, '_decode_por', return_value=decoded):
with contextlib.redirect_stderr(buf):
run = self.run_script(FakeServer(scc), [
{'type': 'action', 'kind': 'scp80',
'params': {'apdu': '80E2900000'}},
], preset)
self.assertEqual(run['status'], 'ok', run['steps'])
log = buf.getvalue()
self.assertIn('TEST-RUN step 1: PoR[inline] status=por_ok', log)
self.assertIn('raw=0271000021AABB', log)
self.assertIn('TEST-RUN step 1: R-APDU SW=9000 data=AABB', log)
def test_expect_decodes_the_submit_por(self):
scc = FakeScc()
preset = {'kic': '15', 'kid': '15', 'kicKey': 'AA' * 16, 'kidKey': 'BB' * 16,
'counter': '0000000A', 'tars': preset_tars()}
decoded = {'response_status': 'por_ok', 'tar': '000000',
'decoded': {'last_status_word': '9000', 'last_response_data': 'AABB'}}
buf = io.StringIO()
with mock.patch.object(S, '_build_secured_packet', return_value=('AA' * 10, {})), \
mock.patch.object(S, '_send_secured_packet',
return_value={'success': True, 'sw': '9102',
'response_data': None}), \
mock.patch.object(S, '_decode_por', return_value=decoded) as dec:
scc.push('8012', self.SEND_SM_POR_CMD, '9000')
scc.push('8014', '', '9000')
with contextlib.redirect_stderr(buf):
run = self.run_script(FakeServer(scc), [
{'type': 'action', 'kind': 'scp80', 'params': {'apdu': '80E2900000'},
'check': {'sw': {'mode': 'mask', 'value': '91??'}, 'por': 'none'}},
{'type': 'expect', 'command': 'SEND SHORT MESSAGE',
'checks': [{'kind': 'por', 'status': 'por_ok', 'sw': '9000'}],
'respond': {'result': 'ok'}},
], preset)
self.assertEqual(run['status'], 'ok', run['steps'])
log = buf.getvalue()
self.assertIn('TEST-RUN step 2: PoR[sms-submit] status=por_ok', log)
self.assertIn('TEST-RUN step 2: R-APDU SW=9000 data=AABB', log)
self.assertEqual(run['steps'][1]['por']['response_status'], 'por_ok')
labels = {c['label']: c for c in run['steps'][1]['checks']}
self.assertTrue(labels['PoR status']['ok'], labels)
self.assertTrue(labels['PoR SW']['ok'], labels)
args = dec.call_args[0]
self.assertEqual(args[4], '0000000A') # the command's counter
self.assertEqual(args[5], 'AA' * 16) # the keyset's keys
self.assertEqual(args[6], 'BB' * 16)
self.assertEqual(args[-1], '027100027100aabb')
def test_expect_por_check_without_a_prior_scp80_fails(self):
scc = FakeScc()
scc.push('80F2', '', '9102')
scc.push('8012', self.SEND_SM_POR_CMD, '9000')
scc.push('8014', '', '9000')
run = self.run_script(FakeServer(scc), [
{'type': 'action', 'kind': 'status', 'params': {'attempts': 2},
'check': {'sw': {'mode': 'mask', 'value': '91??'}}},
{'type': 'expect', 'command': 'SEND SHORT MESSAGE',
'checks': [{'kind': 'por', 'status': 'por_ok'}],
'respond': {'result': 'ok'}},
])
self.assertIsNone(run['steps'][1].get('por'))
por_check = [c for c in run['steps'][1]['checks'] if c['label'] == 'PoR'][0]
self.assertFalse(por_check['ok'])
self.assertIn('no PoR', por_check['actual'])
def test_expect_checks_the_refresh_file_list(self):
scc = FakeScc()
scc.push('80F2', '', '9102')
scc.push('8012', self.REFRESH_FCN_CMD, '9000')
scc.push('8014', '', '9000')
run = self.run_script(FakeServer(scc), [
{'type': 'action', 'kind': 'status', 'params': {'attempts': 2},
'check': {'sw': {'mode': 'mask', 'value': '91??'}}},
{'type': 'expect', 'command': 'REFRESH', 'qualifier': '00',
'checks': [{'kind': 'files', 'value': '3F007F106F3A'}],
'respond': {'result': 'ok'}},
])
self.assertEqual(run['status'], 'ok', run['steps'])
labels = {c['label']: c for c in run['steps'][1]['checks']}
self.assertTrue(labels['Command']['ok'], labels)
self.assertTrue(labels['Qualifier']['ok'], labels)
self.assertTrue(labels['Files']['ok'], labels)
self.assertEqual(labels['Files']['actual'], '3F007F106F3A')
# a DISPLAY TEXT carrying the Alpha identifier ('05' = 'Alfa') plus the
# text string ('8D' = 'hello') - the alpha check must not read the text
DISPLAY_TEXT_ALPHA_CMD = 'D0178103012100820281820504416C66618D060468656C6C6F'
# a SEND SHORT MESSAGE with an SMS-SUBMIT: DA 12345, PID 7F, DCS 00,
# UDL 5, UD 02 71 00 AA BB
SEND_SM_SUBMIT_CMD = 'D01A8103011300820283810B0F010005812143F57F0005027100AABB'
def test_expect_alpha_and_sms_checks(self):
# the Alpha identifier ('05', TS 102 223 8.2) and the SEND SHORT
# MESSAGE TPDU fields (an applet's own SMS, not the OTA PoR)
scc = FakeScc()
scc.push('80F2', '', '9102')
scc.push('8012', self.DISPLAY_TEXT_ALPHA_CMD, '9000')
scc.push('8014', '', '9000')
scc.push('80F2', '', '9102')
scc.push('8012', self.SEND_SM_SUBMIT_CMD, '9000')
scc.push('8014', '', '9000')
buf = io.StringIO()
with contextlib.redirect_stderr(buf):
run = self.run_script(FakeServer(scc), [
{'type': 'action', 'kind': 'status', 'params': {'attempts': 2},
'check': {'sw': {'mode': 'mask', 'value': '91??'}}},
{'type': 'expect', 'command': 'DISPLAY TEXT',
'checks': [{'kind': 'alpha', 'mode': 'exact', 'value': 'Alfa'}],
'respond': {'result': 'ok'}},
{'type': 'action', 'kind': 'status', 'params': {'attempts': 2},
'check': {'sw': {'mode': 'mask', 'value': '91??'}}},
{'type': 'expect', 'command': 'SEND SHORT MESSAGE',
'checks': [{'kind': 'sms', 'da': '12345', 'pid': '7F', 'dcs': '00',
'udl': 5, 'ud': '027100AABB'}],
'respond': {'result': 'ok'}},
])
self.assertEqual(run['status'], 'ok', run['steps'])
labels = {c['label']: c for c in run['steps'][1]['checks']}
self.assertTrue(labels['Alpha']['ok'], labels)
self.assertEqual(labels['Alpha']['actual'], 'Alfa')
labels = {c['label']: c for c in run['steps'][3]['checks']}
for label in ('SMS DA', 'SMS PID', 'SMS DCS', 'SMS UDL', 'SMS UD'):
self.assertTrue(labels[label]['ok'], labels)
self.assertEqual(run['steps'][3]['sms']['da'], '12345')
log = buf.getvalue()
self.assertIn('TEST-RUN step 4: SMS DA=12345 PID=7F DCS=00 UDL=5 UD=027100AABB', log)
def test_expect_sms_check_mismatch_terminates(self):
scc = FakeScc()
scc.push('80F2', '', '9102')
scc.push('8012', self.SEND_SM_SUBMIT_CMD, '9000')
scc.push('8014', '', '9000')
run = self.run_script(FakeServer(scc), [
{'type': 'action', 'kind': 'status', 'params': {'attempts': 2},
'check': {'sw': {'mode': 'mask', 'value': '91??'}}},
{'type': 'expect', 'command': 'SEND SHORT MESSAGE',
'checks': [{'kind': 'sms', 'dcs': 'AA'}], 'respond': {'result': 'ok'}},
])
self.assertNotEqual(run['status'], 'ok')
labels = {c['label']: c for c in run['steps'][1]['checks']}
self.assertFalse(labels['SMS DCS']['ok'])
self.assertEqual(labels['SMS DCS']['expected'], 'AA')
self.assertEqual(labels['SMS DCS']['actual'], '00')
# The live applet's SEND SHORT MESSAGE (captured 2026-10-04 20:58): the
# Alpha identifier is Annex A Variant 2 Cyrillic ("Отправка...") and the
# TPDU is a real SMS-SUBMIT to +79332505884 with 7-bit user data.
APP_SEND_SM_CMD = ('D02F810301130082028183050E810B089EC2BFC0B0B2BAB02E2E2E'
'0B1401000B919733525088F40000084B1C12579C9D83')
def test_expect_alpha_and_sms_checks_on_the_live_applet_sms(self):
scc = FakeScc()
scc.push('80F2', '', '9102')
scc.push('8012', self.APP_SEND_SM_CMD, '9000')
scc.push('8014', '', '9000')
buf = io.StringIO()
with contextlib.redirect_stderr(buf):
run = self.run_script(FakeServer(scc), [
{'type': 'action', 'kind': 'status', 'params': {'attempts': 2},
'check': {'sw': {'mode': 'mask', 'value': '91??'}}},
{'type': 'expect', 'command': 'SEND SHORT MESSAGE',
'checks': [{'kind': 'alpha', 'mode': 'exact', 'value': 'Отправка...'},
{'kind': 'sms', 'da': '+79 332 505 884', 'pid': '00',
'dcs': '00', 'udl': 8, 'ud': '4B1C12579C9D83'}],
'respond': {'result': 'ok'}},
])
self.assertEqual(run['status'], 'ok', run['steps'])
labels = {c['label']: c for c in run['steps'][1]['checks']}
self.assertTrue(labels['Alpha']['ok'], labels)
self.assertEqual(labels['Alpha']['actual'], 'Отправка...')
for label in ('SMS DA', 'SMS PID', 'SMS DCS', 'SMS UDL', 'SMS UD'):
self.assertTrue(labels[label]['ok'], labels)
self.assertEqual(run['steps'][1]['sms']['da'], '79332505884')
log = buf.getvalue()
self.assertIn('TEST-RUN step 2: SMS DA=79332505884 PID=00 DCS=00 UDL=8 '
'UD=4B1C12579C9D83', log)
# the live alfa-dsa applet's response packet (2026-10-04): the secured
# data `80 01 10` is the applet's own - the compact structure's command
# count (0x80) cannot fit the 1-byte command script
APP_POR_RAW = '027100000e0aaf4d0100020000190000800110'
def test_scp80_por_object_check_on_an_applet_response(self):
# the live alfa-dsa applet's response (2026-10-04): the secured data
# `80 01 10` is the applet's own, so the compact parse must not claim
# 128 commands - the runner exposes it as the PoR data
preset = {'kic': '15', 'kid': '15', 'kicKey': 'AA' * 16,
'kidKey': 'BB' * 16, 'counter': '00000001',
'tars': preset_tars()}
buf = io.StringIO()
with mock.patch.object(S, '_build_secured_packet', return_value=('AA' * 10, {})), \
mock.patch.object(S, '_send_secured_packet',
return_value={'success': True, 'sw': '9000',
'response_data': self.APP_POR_RAW}):
with contextlib.redirect_stderr(buf):
run = self.run_script(FakeServer(FakeScc()), [
{'type': 'action', 'kind': 'scp80', 'params': {'apdu': '01'},
'check': {'por': {'status': 'por_ok', 'data': '800110'}}},
], preset)
self.assertEqual(run['status'], 'ok', run['steps'])
labels = {c['label']: c for c in run['steps'][0]['checks']}
self.assertTrue(labels['PoR status']['ok'], labels)
self.assertTrue(labels['PoR data']['ok'], labels)
self.assertEqual(run['steps'][0]['por']['response_type'], 'raw')
# the log names application data instead of inventing an R-APDU line
self.assertIn('APP DATA=800110 (no R-APDU SW)', buf.getvalue())
self.assertNotIn('R-APDU SW=', buf.getvalue())
def test_scp80_por_sw_check_on_applet_data_reports_the_hint(self):
# a `por.sw` expectation cannot be satisfied by an applet's own TAR -
# the report explains why on the check row (the live confusion)
preset = {'kic': '15', 'kid': '15', 'kicKey': 'AA' * 16,
'kidKey': 'BB' * 16, 'counter': '00000001',
'tars': preset_tars()}
with mock.patch.object(S, '_build_secured_packet', return_value=('AA' * 10, {})), \
mock.patch.object(S, '_send_secured_packet',
return_value={'success': True, 'sw': '9000',
'response_data': self.APP_POR_RAW}):
run = self.run_script(FakeServer(FakeScc()), [
{'type': 'action', 'kind': 'scp80', 'params': {'apdu': '01'},
'check': {'por': {'sw': '9000'}}},
], preset)
self.assertNotEqual(run['status'], 'ok')
sw_check = [c for c in run['steps'][0]['checks'] if c['label'] == 'PoR SW'][0]
self.assertFalse(sw_check['ok'])
self.assertIn('application data', sw_check.get('detail') or '')
self.assertIn("assert it with 'data'", sw_check.get('detail') or '')
def test_scp80_format_wraps_the_apdu(self):
# `format` wraps the C-APDU in the expanded Command Scripting template
# (TS 102 226 5.2.1): `expanded` = AA/<len>/22/..., `expanded-ae` =
# AE 80 ... 00 00. A compact step keeps the APDU verbatim.
for fmt, want in (('', '80E2900000'),
('expanded', 'AA07220580E2900000'),
('expanded-ae', 'AE80220580E29000000000')):
scc = FakeScc()
preset = {'kic': '15', 'kid': '15', 'kicKey': 'AA' * 16,
'kidKey': 'BB' * 16, 'counter': '00000001',
'tars': preset_tars()}
params = {'apdu': '80E2900000'}
if fmt:
params['format'] = fmt
buf = io.StringIO()
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}):
with contextlib.redirect_stderr(buf):
run = self.run_script(FakeServer(scc), [
{'type': 'action', 'kind': 'scp80', 'params': params},
], preset)
self.assertEqual(run['status'], 'ok', run['steps'])
self.assertEqual(build.call_args[0][6], want)
log = buf.getvalue()
if fmt:
self.assertIn('TEST-RUN step 1: SCP80 C-APDU=80E2900000 '
'WRAPPED=%s (%s)' % (want, fmt), log)
self.assertIn('C-APDU=%s format=%s' % (want, fmt),
run['steps'][0]['sent'])
else:
self.assertIn('TEST-RUN step 1: SCP80 C-APDU=80E2900000', log)
self.assertNotIn('WRAPPED', log)
self.assertNotIn('format=', run['steps'][0]['sent'])
def test_scp80_inline_por_object_check(self):
# `check.por` as an object asserts the decoded PoR (status/SW/data) of
# this very exchange - inline here, the SEND SHORT MESSAGE transport
# works the same
decoded = {'response_status': 'por_ok', 'tar': '000000',
'cntr': '0000000A', 'raw': '0271000021AABB',
'decoded': {'last_status_word': '9000',
'last_response_data': 'AABB'}}
preset = {'kic': '15', 'kid': '15', 'kicKey': 'AA' * 16,
'kidKey': 'BB' * 16, 'counter': '0000000A',
'tars': preset_tars()}
with mock.patch.object(S, '_build_secured_packet', return_value=('AA' * 10, {})), \
mock.patch.object(S, '_send_secured_packet',
return_value={'success': True, 'sw': '9000',
'response_data': '0271000021AABB'}), \
mock.patch.object(S, '_decode_por', return_value=decoded):
run = self.run_script(FakeServer(FakeScc()), [
{'type': 'action', 'kind': 'scp80', 'params': {'apdu': '80E2900000'},
'check': {'por': {'status': 'por_ok', 'sw': '9000',
'data': 'AABB'}}},
], preset)
self.assertEqual(run['status'], 'ok', run['steps'])
labels = {c['label']: c for c in run['steps'][0]['checks']}
self.assertTrue(labels['PoR status']['ok'], labels)
self.assertTrue(labels['PoR SW']['ok'], labels)
self.assertTrue(labels['PoR data']['ok'], labels)
# a mismatch fails the step; without a decoded PoR the check names it
with mock.patch.object(S, '_build_secured_packet', return_value=('AA' * 10, {})), \
mock.patch.object(S, '_send_secured_packet',
return_value={'success': True, 'sw': '9000',
'response_data': '0271000021AABB'}), \
mock.patch.object(S, '_decode_por', return_value=decoded):
run = self.run_script(FakeServer(FakeScc()), [
{'type': 'action', 'kind': 'scp80', 'params': {'apdu': '80E2900000'},
'check': {'por': {'status': 'por_error'}}},
], preset)
self.assertNotEqual(run['status'], 'ok')
labels = {c['label']: c for c in run['steps'][0]['checks']}
self.assertFalse(labels['PoR status']['ok'])
self.assertEqual(labels['PoR status']['actual'], 'por_ok')
with mock.patch.object(S, '_build_secured_packet', return_value=('AA' * 10, {})), \
mock.patch.object(S, '_send_secured_packet',
return_value={'success': True, 'sw': '9000',
'response_data': None}):
run = self.run_script(FakeServer(FakeScc()), [
{'type': 'action', 'kind': 'scp80', 'params': {'apdu': '80E2900000'},
'check': {'por': {'status': 'por_ok'}}},
], preset)
labels = {c['label']: c for c in run['steps'][0]['checks']}
self.assertFalse(labels['PoR']['ok'])
self.assertIn('no PoR', labels['PoR']['actual'])
def test_parse_file_list_splits_concatenated_paths(self):
raw = bytes.fromhex('D016810301010082028381120B023F002FE23F007F106F3A')
self.assertEqual(S._parse_file_list(raw), ['3F002FE2', '3F007F106F3A'])
self.assertEqual(S._parse_file_list(bytes.fromhex('D009810301010082028381')), [])
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_a_raised_card_error_honours_on_fail(self):
# v3.22.7: a card error raised during a step (an absent optional
# file's 6A82, a refused SELECT) is a step failure like any other -
# on_fail "warning" warns and the run continues.
lchan = object() # _select_path (patched) raises before its use
server = FakeServer(FakeScc())
server.app = type('App', (), {'rs': type('Rs', (), {'lchan': [lchan]})()})()
server.scc.push('80F2', '', '9000')
with mock.patch.object(S, '_select_path', side_effect=RuntimeError(
'6a82: Wrong parameters - File not found')):
run = self.run_script(server, [
{'type': 'action', 'kind': 'file-read',
'params': {'path': 'MF/A153/49AC'}, 'on_fail': 'warning'},
{'type': 'action', 'kind': 'status', 'params': {}},
])
self.assertEqual(run['status'], 'warning', run['steps'])
self.assertEqual(len(run['steps']), 2)
entry = run['steps'][0]
self.assertEqual(entry['status'], 'warning')
self.assertIn('6a82', entry['note'])
check = entry['checks'][0]
self.assertEqual(check['label'], 'Error')
self.assertFalse(check['ok'])
self.assertEqual(check['level'], 'warning')
self.assertIn('6a82', check['actual'])
# the following step ran
self.assertEqual(run['steps'][1]['status'], 'ok')
def test_a_raised_card_error_still_stops_by_default(self):
lchan = object() # _select_path (patched) raises before its use
server = FakeServer(FakeScc())
server.app = type('App', (), {'rs': type('Rs', (), {'lchan': [lchan]})()})()
with mock.patch.object(S, '_select_path', side_effect=RuntimeError(
'6a82: Wrong parameters - File not found')):
run = self.run_script(server, [
{'type': 'action', 'kind': 'file-read',
'params': {'path': 'MF/A153/49AC'}},
{'type': 'action', 'kind': 'status', 'params': {}},
])
self.assertEqual(run['status'], 'error', run['steps'])
self.assertEqual(len(run['steps']), 1)
self.assertEqual(run['steps'][0]['status'], 'error')
self.assertIn('6a82', run['steps'][0]['note'])
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(''), '')
OK_STEP = {'type': 'action', 'kind': 'apdu', 'params': {'apdu': '00A40000023F00'},
'check': {'sw': '9000'}}
BAD_STEP = {'type': 'action', 'kind': 'apdu', 'params': {'apdu': '00B0000000'},
'check': {'sw': '9000'}} # no queue -> 6D00
class SuiteRunnerTestCase(RunnerTestCase):
"""Suite runner tests (v3.22.0): setup -> members (per-member on_fail) ->
teardown, one card session, no resume."""
def members(self, specs):
"""specs: [(role, ok, on_fail)] -> the resolved member list."""
out = []
for i, spec in enumerate(specs):
role, ok = spec[0], spec[1]
on_fail = spec[2] if len(spec) > 2 else 'stop'
step = dict(OK_STEP if ok else BAD_STEP)
out.append(({'script_id': '%032x' % (i + 1), 'role': role,
'on_fail': on_fail},
T.normalise_script({'name': 'script %d' % (i + 1),
'steps': [step]}, S._test_command_type)))
return out
def suite_run(self, server, members, suite=None, preset=None, preset_id=None):
suite = suite or {'name': 'suite'}
S._TEST_RUN.clear()
S._TEST_RUN.update({'running': True, 'stop': False, 'name': 'suite',
'status': None, 'session': server.card_session,
'index': 0, 'total': 0, 'steps': [],
'started': time.time(), 'finished': None, 'error': None,
'scp80_counter': None, 'scp80_counters': {},
'script_id': None, 'preset': None,
'suite': S._test_suite_state(suite, members,
preset or {}, preset_id,
server.card_session),
'log': [], 'log_prefix': '', 'adm': None})
S._TEST_RUNNING = True
try:
S._test_suite_execute(server, suite, members, preset or {}, preset_id)
finally:
S._TEST_RUNNING = False
return S._TEST_RUN
def push_ok(self, scc, n):
for _ in range(n):
scc.push('00A4', '', '9000')
class TestSuiteRunner(SuiteRunnerTestCase):
def test_setup_members_teardown_run_in_order(self):
scc = FakeScc()
self.push_ok(scc, 4)
members = self.members([('setup', True), ('member', True),
('member', True), ('teardown', True)])
run = self.suite_run(FakeServer(scc), members)
self.assertEqual(run['status'], 'ok')
state = run['suite']
self.assertEqual([m['role'] for m in state['members']],
['setup', 'member', 'member', 'teardown'])
self.assertEqual([m['status'] for m in state['members']],
['ok', 'ok', 'ok', 'ok'])
self.assertEqual(state['summary']['setup'], 'ok')
self.assertEqual(state['summary']['teardown'], 'ok')
self.assertEqual(state['summary']['ok'], 4)
self.assertEqual(state['summary']['error'], 0)
self.assertFalse(state['summary']['stopped'])
self.assertEqual(state['summary']['counters_before'], {})
# the member steps are copied into the report (the live view is reset
# between members)
self.assertEqual(len(state['members'][1]['steps']), 1)
self.assertEqual(state['members'][1]['steps'][0]['status'], 'ok')
# the run log carries the member prefix and is sliced per member
joined = '\n'.join(state['members'][1]['log'])
self.assertIn('[2/4 member script 2]', joined)
def test_on_fail_stop_skips_the_rest_and_still_runs_teardown(self):
scc = FakeScc()
self.push_ok(scc, 1) # setup ok
self.push_ok(scc, 1) # first member ok
# the failing member gets no queue entry -> 6D00 (error)
self.push_ok(scc, 1) # teardown ok
members = self.members([('setup', True), ('member', True),
('member', False, 'stop'), ('member', True),
('teardown', True)])
run = self.suite_run(FakeServer(scc), members)
self.assertEqual(run['status'], 'error')
statuses = [m['status'] for m in run['suite']['members']]
self.assertEqual(statuses, ['ok', 'ok', 'error', 'skipped', 'ok'])
self.assertEqual(run['suite']['members'][3]['note'],
'a previous script failed (on_fail: stop)')
# no member ran after the error: the queue holds exactly the teardown's
self.assertEqual(sum(1 for a in scc.sent if a.startswith('00A4')), 3)
def test_on_fail_continue_runs_the_next_member(self):
scc = FakeScc()
self.push_ok(scc, 1) # member 1 ok
# member 2 fails (no queue entry)
self.push_ok(scc, 1) # member 3 ok
members = self.members([('member', True), ('member', False, 'continue'),
('member', True)])
run = self.suite_run(FakeServer(scc), members)
self.assertEqual(run['status'], 'error') # the failure is reported
self.assertEqual([m['status'] for m in run['suite']['members']],
['ok', 'error', 'ok'])
def test_setup_failure_skips_members_but_runs_teardown(self):
scc = FakeScc()
# setup fails (no queue), members never run, teardown ok
self.push_ok(scc, 1)
members = self.members([('setup', False), ('member', True),
('teardown', True)])
run = self.suite_run(FakeServer(scc), members)
self.assertEqual([m['status'] for m in run['suite']['members']],
['error', 'skipped', 'ok'])
self.assertEqual(run['suite']['members'][1]['note'],
'the setup script failed')
def test_stop_skips_the_members_and_runs_the_teardown(self):
scc = FakeScc()
self.push_ok(scc, 1) # setup ok (runs first)
self.push_ok(scc, 1) # teardown ok
members = self.members([('setup', True), ('member', True),
('teardown', True)])
server = FakeServer(scc)
real = S._test_run_script
calls = []
def wrapped(srv, script, preset, shared=None):
ctx, stopped = real(srv, script, preset, shared=shared)
calls.append(script['name'])
if srv is server and len(calls) == 1:
S._TEST_RUN['stop'] = True # the operator stops here
return ctx, stopped
with mock.patch.object(S, '_test_run_script', wrapped):
run = self.suite_run(server, members)
statuses = [m['status'] for m in run['suite']['members']]
self.assertEqual(statuses, ['ok', 'skipped', 'ok'])
self.assertTrue(run['suite']['summary']['stopped'])
self.assertEqual(run['status'], 'stopped')
def test_a_card_reset_stops_the_suite_and_flags_it(self):
scc = FakeScc()
self.push_ok(scc, 1) # member 1 ok
self.push_ok(scc, 1) # teardown ok (new session)
members = self.members([('member', True), ('member', True),
('teardown', True)])
server = FakeServer(scc)
real = S._test_run_script
def wrapped(srv, script, preset, shared=None):
ctx, stopped = real(srv, script, preset, shared=shared)
if srv is server and len(scc.sent) <= 1:
srv.card_session += 1 # the card reset mid-suite
return ctx, stopped
with mock.patch.object(S, '_test_run_script', wrapped):
run = self.suite_run(server, members)
state = run['suite']
self.assertEqual([m['status'] for m in state['members']],
['ok', 'skipped', 'ok'])
self.assertEqual(state['members'][1]['note'],
'card session changed (card removed or re-equipped)')
self.assertTrue(state['session_changed'])
self.assertEqual(state['session_start'], 1)
self.assertEqual(state['session_end'], 2)
self.assertTrue(state['summary']['session_changed'])
def test_teardown_is_skipped_when_the_card_is_gone(self):
scc = FakeScc()
self.push_ok(scc, 1)
members = self.members([('member', True), ('teardown', True)])
server = FakeServer(scc)
server.card_present = False
run = self.suite_run(server, members)
self.assertEqual([m['status'] for m in run['suite']['members']],
['ok', 'skipped'])
self.assertEqual(run['suite']['members'][1]['note'],
'no card in the reader')
def test_scp80_members_share_the_counter_state(self):
# The counter is per keyset and per card session: the second SCP80
# script must send what the first consumed + 1, not the suite-start
# snapshot again (the store persists each packet's counter, but the
# in-memory preset copy does not move).
built = []
server = FakeServer(FakeScc())
step = {'type': 'action', 'kind': 'scp80',
'params': {'apdu': '80E2900000', 'tar': 'B00000'}}
script = T.normalise_script({'name': 'scp80', 'steps': [step]},
S._test_command_type)
preset = {'id': 'p1',
'keysets': [{'kic': '15', 'kid': '15', 'kicKey': '00' * 16,
'kidKey': '00' * 16, 'cntr': '0000000010'}],
'tars': [{'role': 'isd', 'tar': 'B00000', 'msl': '16',
'keyset': 1}]}
members = [({'script_id': 'a' * 32, 'role': 'member', 'on_fail': 'stop'}, script),
({'script_id': 'b' * 32, 'role': 'member', 'on_fail': 'stop'}, script)]
with mock.patch.object(S, '_build_secured_packet',
side_effect=lambda *a, **k: (
built.append(a[5]) or ('00' * 20, {}))), \
mock.patch.object(S, '_send_secured_packet', return_value={
'success': True, 'sw': '9000', 'response_data': '',
'bytes': 10, 'segments': 1}), \
mock.patch.object(S, '_decode_por', return_value=None), \
mock.patch.object(S, '_preset_counter_persist', return_value=None):
run = self.suite_run(server, members, preset=preset)
self.assertEqual(built, ['0000000010', '0000000011'])
self.assertEqual([m['status'] for m in run['suite']['members']], ['ok', 'ok'])
self.assertEqual(run['scp80_counters'], {1: '0000000012'})
class TestAdmGate(unittest.TestCase):
"""The run prerequisite: a script/suite flagged `require_adm` runs only
with the ADM verified for the current card session - the preset's key is
tried once, never retried automatically."""
def setUp(self):
self.server = FakeServer(FakeScc())
self.server.app = object()
def test_not_needed_passes(self):
ok, info, err = S._ensure_adm(self.server, {}, False)
self.assertTrue(ok)
self.assertIsNone(err)
self.assertFalse(info['required'])
def test_verified_latch_passes(self):
self.server.adm_session = 1
ok, info, err = S._ensure_adm(self.server, {}, True)
self.assertTrue(ok)
self.assertTrue(info['verified'])
def test_verifies_once_with_the_preset_key(self):
with mock.patch.object(S, '_verify_adm', return_value={'ok': True,
'sw': '9000'}) as v:
ok, info, err = S._ensure_adm(self.server, {'adm': 'AABBCCDD'}, True)
self.assertTrue(ok, err)
self.assertTrue(info['verified'])
self.assertTrue(info['checked'])
self.assertEqual(self.server.adm_session, self.server.card_session)
v.assert_called_once()
# the second gate in the same session does not verify again
with mock.patch.object(S, '_verify_adm', return_value={'ok': True}) as v2:
ok, _info, _err = S._ensure_adm(self.server, {'adm': 'AABBCCDD'}, True)
self.assertTrue(ok)
v2.assert_not_called()
def test_a_failure_is_not_retried_automatically(self):
with mock.patch.object(S, '_verify_adm', return_value={
'ok': False, 'sw': '63C2', 'attempts_left': 2}) as v:
ok, info, err = S._ensure_adm(self.server, {'adm': 'AABBCCDD'}, True)
self.assertFalse(ok)
self.assertIn('63C2', err)
self.assertIn('2 attempt(s) left', err)
self.assertEqual(info['attempts_left'], 2)
self.assertEqual(self.server.adm_failed_session, self.server.card_session)
# the next gate refuses without touching the card again
with mock.patch.object(S, '_verify_adm') as v2:
ok, _info, err = S._ensure_adm(self.server, {'adm': 'AABBCCDD'}, True)
self.assertFalse(ok)
self.assertIn('verify it manually', err)
v2.assert_not_called()
def test_no_key_in_the_preset_refuses(self):
ok, _info, err = S._ensure_adm(self.server, {}, True)
self.assertFalse(ok)
self.assertIn('no ADM key', err)
def test_a_new_session_clears_the_failure_stamp(self):
self.server.adm_failed_session = 1
self.server.card_session = 2
with mock.patch.object(S, '_verify_adm', return_value={'ok': True}):
ok, _info, _err = S._ensure_adm(self.server, {'adm': 'AABBCCDD'}, True)
self.assertTrue(ok)
class TestRunnerBipCommands(RunnerTestCase):
"""The runner's steps let the terminal answer the card's own BIP commands
(v3.25.3): a channel the card opens during a run really opens and its data
really flows - unless the script overrides the answer with `raw` TLVs or a
non-ok result (an applet error path to test)."""
OPEN = 'D02B8103014001820281828500B50103B9020200C70B076D656761666F6E2E7275'
BIP_TR = bytes.fromhex('010301400002028281030100')
def _tr(self, scc):
return [a for a in scc.sent if a.startswith('8014')][-1]
def _pending(self, scc, cmd):
scc.push('80F2', '', '9109') # STATUS: the command is pending
scc.push('8012', cmd, '9000')
scc.push('8014', '', '9000')
def _run_bip(self, scc, steps):
with mock.patch.object(S, '_handle_bip_command',
return_value=self.BIP_TR) as bip:
run = self.run_script(FakeServer(scc), steps)
return run, bip
def test_a_drain_lets_the_terminal_answer_bip(self):
scc = FakeScc()
scc.push('80F2', '', '9109').push('8012', self.OPEN, '9000')
scc.push('8014', '', '9000').push('80F2', '', '9000')
run, bip = self._run_bip(scc, [
{'type': 'action', 'kind': 'proactive-drain',
'params': {'attempts': 2, 'respond': {'result': 'ok'}}},
])
self.assertEqual(run['status'], 'ok', run['steps'])
self.assertEqual(bip.call_count, 1)
self.assertEqual(bip.call_args[0][2], 0x40)
self.assertTrue(self._tr(scc).endswith(self.BIP_TR.hex().upper()),
self._tr(scc))
def test_an_expectation_lets_the_terminal_answer_bip(self):
scc = FakeScc()
self._pending(scc, self.OPEN)
run, bip = self._run_bip(scc, [
{'type': 'action', 'kind': 'status', 'params': {'attempts': 2}},
{'type': 'expect', 'command': 'ANY', 'respond': {'result': 'ok'}},
])
self.assertEqual(run['status'], 'ok', run['steps'])
self.assertEqual(bip.call_count, 1)
def test_a_raw_answer_overrides_the_terminal_bip(self):
scc = FakeScc()
self._pending(scc, self.OPEN)
_run, bip = self._run_bip(scc, [
{'type': 'action', 'kind': 'status', 'params': {'attempts': 2}},
{'type': 'expect', 'command': 'ANY',
'respond': {'result': 'ok', 'raw': '010301400002028281030100'}},
])
self.assertEqual(bip.call_count, 0)
self.assertTrue(self._tr(scc).endswith('010301400002028281030100'),
self._tr(scc))
def test_a_non_ok_result_overrides_the_terminal_bip(self):
scc = FakeScc()
self._pending(scc, self.OPEN)
_run, bip = self._run_bip(scc, [
{'type': 'action', 'kind': 'status', 'params': {'attempts': 2}},
{'type': 'expect', 'command': 'ANY', 'respond': {'result': 'refused'}},
])
self.assertEqual(bip.call_count, 0)
self.assertIn('83022200', self._tr(scc))
def test_the_cleanup_drain_handles_bip(self):
scc = FakeScc()
scc.push('8012', self.OPEN, '9000')
with mock.patch.object(S, '_handle_bip_command',
return_value=self.BIP_TR) as bip:
drained = S._test_drain_pending(scc, 9)
self.assertEqual(bip.call_count, 1)
self.assertIn('OPEN CHANNEL', drained.get('note', ''))
if __name__ == '__main__':
unittest.main()
Reference in New Issue View Git Blame Copy Permalink