catarrh
e0dfc1032e
The step editor's result select (the expect's respond, the proactive-drain's respond/first) offered five symbolic values with no hex. It now offers the full TS 102 223 v18.3.0 8.12.0 list (00-3F), labelled '<hex> - <alias>' (or '<hex> - <spec name>'), plus a manual hex entry, and an optional additional-information byte for the general result (83 02 <result> <info>, default 00; the 8.12.2 terminal-problem values are suggested through a datalist). The script aliases were corrected to the spec: refused 03->22, not_understood 04->32, modified 06->07, no_response 22->12 - a stored script using one of them now answers with the corrected value; no_response stays accepted for compatibility but is no longer offered by the editor. The server's TR decode tables (_RESULT_NAMES_BASIC/_GENERAL, a GSM 11.14 mix displayed in the proactive log) are replaced by the spec table plus the 8.12.2 additional-information names; the dcs field parses as hex (0A used to be refused and 10 read as decimal 10); the drain summary's 22 label is fixed. The result labels are not translated (spec text); the unused RU keys were removed. Shared fixture frontend/tests/tr_results.json, drift-guarded by both suites. 855 frontend + 838 Python; sw cache simple-v418.
732 lines
33 KiB
Python
732 lines
33 KiB
Python
"""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
|
|
|
|
import gsm0338 # registers the 'gsm03.38' codec (the default alphabet coding)
|
|
|
|
from pysim_simple_server import events
|
|
|
|
__all__ = [
|
|
'ScriptError', 'ACTION_KINDS', 'FAIL_LEVELS', 'POR_CHECKS', 'RESULT_NAMES',
|
|
'TR_RESULTS', 'TR_RESULT_ADD_INFO',
|
|
'normalise_script', 'normalise_step', 'normalise_respond',
|
|
'match_value', 'match_text', 'match_item', 'combine_levels', 'build_tr',
|
|
'pack_gsm7',
|
|
]
|
|
|
|
ACTION_KINDS = ('envelope', 'event', 'menu-select', 'file-write', 'file-read',
|
|
'apdu', 'scp80', 'status', 'proactive-drain')
|
|
FAIL_LEVELS = ('error', 'warning')
|
|
POR_CHECKS = ('none', 'ok', 'any')
|
|
|
|
# TERMINAL RESPONSE general results (TS 102 223 v18.3.0 8.12.0) and the
|
|
# terminal-problem additional information (8.12.2), pinned by
|
|
# frontend/tests/tr_results.json (both suites drift-guard against it).
|
|
TR_RESULTS = {
|
|
'00': 'Command performed successfully',
|
|
'01': 'Command performed with partial comprehension',
|
|
'02': 'Command performed, with missing information',
|
|
'03': 'REFRESH performed with additional EFs read',
|
|
'04': 'Command performed successfully, but requested icon could not be displayed',
|
|
'05': 'Command performed, but modified by call control by NAA',
|
|
'06': 'Command performed successfully, limited service',
|
|
'07': 'Command performed with modification',
|
|
'08': 'REFRESH performed but indicated NAA was not active',
|
|
'09': 'Command performed successfully, tone not played',
|
|
'10': 'Proactive UICC session terminated by the user',
|
|
'11': 'Backward move in the proactive UICC session requested by the user',
|
|
'12': 'No response from user',
|
|
'13': 'Help information required by the user',
|
|
'14': 'Reserved for GSM/3G',
|
|
'15': 'Reserved for 3GPP (for future usage)',
|
|
'16': 'Reserved for 3GPP (for future usage)',
|
|
'20': 'Terminal currently unable to process command',
|
|
'21': 'Network currently unable to process command',
|
|
'22': 'User did not accept the proactive command',
|
|
'23': 'User cleared down call before connection or network release',
|
|
'24': 'Action in contradiction with the current timer state',
|
|
'25': 'Interaction with call control by NAA, temporary problem',
|
|
'26': 'Launch browser generic error',
|
|
'27': 'MMS temporary problem',
|
|
'28': 'Reserved for 3GPP (BIP temporary error, T timer in the additional information)',
|
|
'29': 'Reserved for 3GPP (for future usage)',
|
|
'30': "Command beyond terminal's capabilities",
|
|
'31': 'Command type not understood by terminal',
|
|
'32': 'Command data not understood by terminal',
|
|
'33': 'Command number not known by terminal',
|
|
'34': 'Reserved for GSM/3G',
|
|
'35': 'Reserved for GSM/3G',
|
|
'36': 'Error, required values are missing',
|
|
'37': 'Reserved for GSM/3G',
|
|
'38': 'MultipleCard commands error',
|
|
'39': 'Interaction with call control by NAA, permanent problem',
|
|
'3A': 'Bearer Independent Protocol permanent error',
|
|
'3B': 'Access Technology unable to process command',
|
|
'3C': 'Frames error',
|
|
'3D': 'MMS Error',
|
|
'3E': 'Reserved for 3GPP (for future usage)',
|
|
'3F': 'Reserved for 3GPP (for future usage)',
|
|
}
|
|
|
|
TR_RESULT_ADD_INFO = {
|
|
'00': 'No specific cause can be given',
|
|
'01': 'Screen is busy',
|
|
'02': 'Terminal currently busy on call',
|
|
'03': 'Reserved for GSM/3G',
|
|
'04': 'No service',
|
|
'05': 'Access control class bar',
|
|
'06': 'Radio resource not granted',
|
|
'07': 'Not in speech call',
|
|
'08': 'Reserved for GSM/3G',
|
|
'09': 'Terminal currently busy on SEND DTMF command',
|
|
'0A': 'No NAA active',
|
|
}
|
|
|
|
# The script authoring aliases: a convenience for the common values (the
|
|
# script may also use the two-digit hex value directly). `no_response` is
|
|
# kept as an alias of 12 for backward compatibility but is not offered by the
|
|
# editor (12's canonical alias is `timeout`); the v3.23.5 review corrected
|
|
# `refused` (03 -> 22), `not_understood` (04 -> 32), `modified` (06 -> 07)
|
|
# and `no_response` (22 -> 12) to their TS 102 223 8.12.0 meanings.
|
|
RESULT_NAMES = {
|
|
'ok': 0x00, 'partial': 0x01, 'missing': 0x02, 'refused': 0x22,
|
|
'not_understood': 0x32, 'modified': 0x07,
|
|
'cancel': 0x10, 'back': 0x11, 'timeout': 0x12, 'no_response': 0x12,
|
|
}
|
|
|
|
_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 _por_fields(c, what='por check'):
|
|
"""The decoded-PoR assertion fields (status/sw/data); at least one."""
|
|
out = {}
|
|
status = c.get('status')
|
|
if status is not None and str(status).strip():
|
|
out['status'] = str(status).strip()
|
|
if c.get('sw') not in (None, ''):
|
|
out['sw'] = _check_spec(c.get('sw'), what + ' sw')
|
|
if c.get('data') not in (None, ''):
|
|
out['data'] = _check_spec(c.get('data'), what + ' data')
|
|
if not (out.get('status') or out.get('sw') or out.get('data')):
|
|
raise ScriptError(what + ': status, sw or data is required')
|
|
return out
|
|
|
|
|
|
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:
|
|
text = str(value).strip()
|
|
v = int(text, 16 if text.lower().startswith('0x') else 10)
|
|
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 _hex_byte(value, what):
|
|
"""A byte given as an int or as a hex string ('0A' / '0a' / '0x0A')."""
|
|
if isinstance(value, str):
|
|
text = value.strip().lower().replace('0x', '')
|
|
if not re.fullmatch(r'[0-9a-f]{1,2}', text):
|
|
raise ScriptError('%s must be a hex byte' % what)
|
|
return int(text, 16)
|
|
return _int(value, what, 0, 255)
|
|
|
|
|
|
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, command_resolver)
|
|
if typ == 'expect':
|
|
return _normalise_expect(step, command_resolver)
|
|
raise ScriptError("step type must be 'action' or 'expect'")
|
|
|
|
|
|
def _normalise_action(step, command_resolver=None):
|
|
kind = str(step.get('kind') or '').lower()
|
|
if kind not in ACTION_KINDS:
|
|
raise ScriptError("unknown action kind %r" % step.get('kind'))
|
|
params = _normalise_params(kind, step.get('params') or {}, command_resolver)
|
|
on_fail = _fail_level(step.get('on_fail'))
|
|
check = _normalise_check(step.get('check'), kind, params)
|
|
out = {'type': 'action', 'kind': kind, 'params': params,
|
|
'check': check, 'on_fail': on_fail}
|
|
if step.get('label'):
|
|
out['label'] = str(step['label'])
|
|
return out
|
|
|
|
|
|
def _normalise_params(kind, p, command_resolver=None):
|
|
if not isinstance(p, dict):
|
|
raise ScriptError('%s: params must be an object' % kind)
|
|
if kind == 'envelope':
|
|
if p.get('event') is None:
|
|
raise ScriptError('envelope: event is required')
|
|
out = {'event': _int(p['event'], 'envelope event', 0, 255),
|
|
'data': _data_hex(p.get('data'), 'envelope data', allow_empty=True)}
|
|
src = str(p.get('src') or '').strip().upper()
|
|
if src:
|
|
if not re.fullmatch(r'[0-9A-F]{2}', src):
|
|
raise ScriptError('envelope: src must be one hex byte '
|
|
'(82 terminal / 83 network)')
|
|
out['src'] = src
|
|
return out
|
|
if kind == 'event':
|
|
# The semantic event download: the same forms the Simulator uses
|
|
# (EVENT_FORMS). The four events events.py models are built from
|
|
# `fields` (validated here, so a script fails before the run starts);
|
|
# every other event carries the form-built `data` hex the PWA
|
|
# produces - the single-envelope budget is checked here too.
|
|
code = events.resolve_event(p.get('event'))
|
|
if code is None:
|
|
raise ScriptError('event: a hex byte or a known event name is required')
|
|
raw = p.get('fields')
|
|
if raw is not None and not isinstance(raw, dict):
|
|
raise ScriptError('event: fields must be an object')
|
|
out = {'event': code}
|
|
if code in events.BUILDERS:
|
|
try:
|
|
out['fields'] = events.normalise(code, raw or {})
|
|
except events.EventError as e:
|
|
raise ScriptError('event 0x%02X: %s' % (code, e))
|
|
else:
|
|
data = re.sub(r'\s', '', str(p.get('data') or '')).upper()
|
|
if not re.fullmatch(r'(?:[0-9A-F]{2})+', data):
|
|
raise ScriptError('event 0x%02X: no server-side builder for this '
|
|
'event - pass the form-built `data` hex (or '
|
|
'use the raw envelope action)' % code)
|
|
if len(data) // 2 > events.EVENT_DATA_MAX:
|
|
raise ScriptError('event 0x%02X: the event data does not fit one '
|
|
'ENVELOPE (max %d bytes; chained delivery is '
|
|
'not implemented)' % (code, events.EVENT_DATA_MAX))
|
|
out['data'] = data
|
|
src = str(p.get('src') or '').strip().upper()
|
|
if src:
|
|
if not re.fullmatch(r'[0-9A-F]{2}', src):
|
|
raise ScriptError('event: src must be one hex byte '
|
|
'(82 terminal / 83 network)')
|
|
out['src'] = src
|
|
return out
|
|
if kind == 'menu-select':
|
|
# The item is named by its id, or by its text - the id varies with the
|
|
# applet's install parameters, so a script can match the text the card
|
|
# actually shows (the runner resolves it against the cached menu).
|
|
out = {}
|
|
if p.get('item_id') not in (None, ''):
|
|
out['item_id'] = _int(p.get('item_id'), 'menu item_id', 1, 255)
|
|
text = p.get('text')
|
|
if text is not None and str(text).strip():
|
|
out['text'] = str(text)
|
|
mode = str(p.get('mode') or 'exact').lower()
|
|
if mode not in ('exact', 'contains'):
|
|
raise ScriptError('menu-select: mode must be exact or contains')
|
|
out['mode'] = mode
|
|
# Menu labels are UI text: a case slip should not fail a run, so
|
|
# the match is case-insensitive by default; `case_sensitive: true`
|
|
# forces an exact-case match.
|
|
out['case_sensitive'] = bool(p.get('case_sensitive', False))
|
|
if not out:
|
|
raise ScriptError('menu-select: item_id or text is required')
|
|
if 'item_id' in out and 'text' in out:
|
|
raise ScriptError('menu-select: give item_id or text, not both')
|
|
return out
|
|
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 = {}
|
|
source = str(p.get('source') or '').lower()
|
|
if source not in ('', 'apdu', 'sp'):
|
|
raise ScriptError('scp80: source must be apdu or sp')
|
|
if source == 'sp' or (not source and p.get('sp')):
|
|
if not p.get('sp'):
|
|
raise ScriptError('scp80: secured packet is required')
|
|
out['sp'] = _data_hex(p.get('sp'), 'secured packet')
|
|
else:
|
|
if not p.get('apdu'):
|
|
raise ScriptError('scp80: apdu is required')
|
|
out['apdu'] = _data_hex(p.get('apdu'), 'scp80 apdu')
|
|
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)
|
|
# The keyset number (KIc/KID b8..b5, TS 102 225 5.1.2): the keys and
|
|
# the counter stay preset-owned, the step only picks which keyset.
|
|
if p.get('kvn') not in (None, ''):
|
|
out['kvn'] = _int(p.get('kvn'), 'scp80 keyset number', 1, 15)
|
|
# The command format: the C-APDU verbatim (compact, the default) or
|
|
# wrapped in the expanded Command Scripting template (TS 102 226
|
|
# 5.2.1) - `expanded` = AA/<len>/22/..., `expanded-ae` = AE 80/22/00 00.
|
|
fmt = str(p.get('format') or '').strip().lower()
|
|
if fmt:
|
|
if fmt not in ('compact', 'expanded', 'expanded-ae'):
|
|
raise ScriptError('scp80: format must be compact, expanded or '
|
|
'expanded-ae')
|
|
out['format'] = fmt
|
|
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}
|
|
if kind == 'proactive-drain':
|
|
# The proactive drain: consume whatever the card announces (the
|
|
# TERMINAL RESPONSE is `first` for the first command, `respond` for
|
|
# the rest) and confirm the card is idle; `require` asserts what was
|
|
# drained (at least one match - an empty drain fails it).
|
|
out = {'respond': normalise_respond(p.get('respond') or {}),
|
|
'attempts': _int(3 if p.get('attempts') is None else p.get('attempts'),
|
|
'proactive-drain attempts', 1, 1000),
|
|
'interval_ms': _int(200 if p.get('interval_ms') is None
|
|
else p.get('interval_ms'),
|
|
'proactive-drain interval_ms', 0, 10000)}
|
|
if p.get('first') is not None:
|
|
out['first'] = normalise_respond(p['first'])
|
|
req = p.get('require')
|
|
if req is not None:
|
|
if not isinstance(req, dict):
|
|
raise ScriptError('proactive-drain require must be an object')
|
|
ctype, cname = _resolve_command(req.get('command'), 'proactive-drain require',
|
|
command_resolver)
|
|
qualifier = _check_spec(req.get('qualifier'), 'proactive-drain require.qualifier')
|
|
if qualifier and '?' not in qualifier['value'] and len(qualifier['value']) != 2:
|
|
raise ScriptError('proactive-drain require.qualifier must be one byte')
|
|
out['require'] = {'type': ctype, 'name': cname, 'qualifier': qualifier}
|
|
return out
|
|
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')
|
|
if kind == 'proactive-drain' and data is not None:
|
|
# The proactive drain's result data is the drained command list, not
|
|
# a single response: a `data` check would be silently ignored, so it
|
|
# is refused here (assert the drained commands with `require`).
|
|
raise ScriptError('check.data is not used by the proactive drain action - '
|
|
'assert what was drained with require, or the final '
|
|
'status word with check.sw')
|
|
por = check.get('por')
|
|
if isinstance(por, dict):
|
|
# The decoded-PoR assertion (available on scp80 actions whether the
|
|
# PoR arrived inline or via SEND SHORT MESSAGE).
|
|
if kind != 'scp80':
|
|
raise ScriptError('check.por is only valid for scp80 actions')
|
|
por = _por_fields(por, 'check.por')
|
|
elif 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, any or an object')
|
|
return {'sw': sw, 'data': data, 'por': por}
|
|
|
|
|
|
def _resolve_command(cmd, what, command_resolver=None):
|
|
"""A proactive command spec: a name (resolved to a type code through
|
|
``command_resolver``), a hex type code or ANY. Returns (type|None,
|
|
name|None)."""
|
|
if cmd is None or isinstance(cmd, bool):
|
|
raise ScriptError('%s: command is required' % what)
|
|
if isinstance(cmd, int):
|
|
return cmd, None
|
|
s = str(cmd).strip()
|
|
up = s.upper()
|
|
if up in ('ANY', '*'):
|
|
return None, 'ANY'
|
|
if re.fullmatch(r'(0X)?[0-9A-F]{2}', up):
|
|
return int(up.replace('0X', ''), 16), None
|
|
if command_resolver is not None:
|
|
ctype = command_resolver(up)
|
|
if ctype is None:
|
|
raise ScriptError('%s: unknown proactive command %r' % (what, s))
|
|
return ctype, up
|
|
raise ScriptError('%s: command must be a hex type code' % what)
|
|
|
|
|
|
def _normalise_expect(step, command_resolver=None):
|
|
ctype, cname = _resolve_command(step.get('command'), 'expect', command_resolver)
|
|
qualifier = _check_spec(step.get('qualifier'), 'qualifier')
|
|
if qualifier and '?' not in qualifier['value'] and len(qualifier['value']) != 2:
|
|
raise ScriptError('qualifier must be one byte')
|
|
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}
|
|
if kind == 'por':
|
|
out = {'kind': 'por', 'on_fail': level}
|
|
out.update(_por_fields(c, 'por check'))
|
|
return out
|
|
if kind == 'alpha':
|
|
mode = str(c.get('mode') or 'contains').lower()
|
|
if mode not in ('contains', 'exact'):
|
|
raise ScriptError('alpha check: mode must be contains or exact')
|
|
if c.get('value') is None:
|
|
raise ScriptError('alpha check: value is required')
|
|
return {'kind': 'alpha', 'mode': mode, 'value': str(c['value']),
|
|
'case_sensitive': bool(c.get('case_sensitive', True)),
|
|
'on_fail': level}
|
|
if kind == 'sms':
|
|
# The SEND SHORT MESSAGE TPDU fields (an applet's own SMS, as opposed
|
|
# to the PoR of an OTA exchange).
|
|
out = {'kind': 'sms', 'on_fail': level}
|
|
da = c.get('da')
|
|
if da is not None and str(da).strip():
|
|
d = re.sub(r'\s', '', str(da))
|
|
# The international flag lives in the TPDU's type-of-number, not
|
|
# in the digits: a leading '+' is accepted and dropped.
|
|
if d.startswith('+'):
|
|
d = d[1:]
|
|
if not re.fullmatch(r'[0-9*#]{1,20}', d):
|
|
raise ScriptError('sms check: da must be up to 20 digits')
|
|
out['da'] = d
|
|
for key in ('pid', 'dcs'):
|
|
if c.get(key) not in (None, ''):
|
|
out[key] = _check_spec(c.get(key), 'sms check %s' % key)
|
|
if c.get('udl') not in (None, ''):
|
|
out['udl'] = _int(c.get('udl'), 'sms check udl', 0, 255)
|
|
if c.get('ud') not in (None, ''):
|
|
out['ud'] = _check_spec(c.get('ud'), 'sms check ud')
|
|
if not any(k in out for k in ('da', 'pid', 'dcs', 'udl', 'ud')):
|
|
raise ScriptError('sms check: da, pid, dcs, udl or ud is required')
|
|
return out
|
|
if kind == 'files':
|
|
val = c.get('files', c.get('value'))
|
|
if isinstance(val, str):
|
|
val = [x for x in re.split(r'[,\s]+', val) if x]
|
|
if not isinstance(val, list) or not val:
|
|
raise ScriptError('files check: a list of file paths is required')
|
|
files = []
|
|
for f in val:
|
|
path = re.sub(r'\s', '', str(f)).upper()
|
|
if not re.fullmatch(r'(?:[0-9A-F]{2})+', path) or len(path) < 4:
|
|
raise ScriptError('files check: each path must be hex FID bytes: %r' % f)
|
|
files.append(path)
|
|
return {'kind': 'files', 'files': files, '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('additional_info') is not None:
|
|
out['additional_info'] = _hex_byte(respond['additional_info'],
|
|
'respond additional_info')
|
|
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')
|
|
# the field is a hex byte (the PWA's input strips to hex): parse it as
|
|
# hex, not decimal - '0A' used to be refused and '10' read as 0x0A
|
|
out['dcs'] = _hex_byte(dcs, 'respond dcs') if dcs is not None else 0x00
|
|
# The same coding the TERMINAL RESPONSE builder uses: a text the
|
|
# selected dcs cannot code (a non-GSM character with 00/04) must fail
|
|
# the script at load, not mid-run after the command was fetched
|
|
# (v3.22.8 review).
|
|
try:
|
|
_encode_text(out['text'], out['dcs'])
|
|
except Exception:
|
|
raise ScriptError(
|
|
'respond: the text cannot be coded with dcs %02X - use dcs 08 '
|
|
'(UCS2) for non-GSM characters' % out['dcs'])
|
|
extra = respond.get('raw')
|
|
if extra not in (None, ''):
|
|
out['raw'] = _data_hex(extra, 'respond raw')
|
|
return out
|
|
|
|
|
|
# ─── scripted TERMINAL RESPONSE ─────────────────────────────────────────
|
|
|
|
def pack_gsm7(septets):
|
|
"""Pack GSM 03.38 septets into 7-bit octets (TS 23.038 4, SMS packing)."""
|
|
out = bytearray()
|
|
acc = 0
|
|
bits = 0
|
|
for s in septets:
|
|
acc |= (s & 0x7F) << bits
|
|
bits += 7
|
|
while bits >= 8:
|
|
out.append(acc & 0xFF)
|
|
acc >>= 8
|
|
bits -= 8
|
|
if bits:
|
|
out.append(acc & 0xFF)
|
|
return bytes(out)
|
|
|
|
|
|
def _encode_text(text, dcs):
|
|
"""Text string coding for a scripted TERMINAL RESPONSE (TS 102 223 8.15):
|
|
'00' = GSM default alphabet 7 bits packed, '04' (or any other 8-bit
|
|
scheme) = GSM default alphabet 8 bits (TS 23.038 via the gsm03.38 codec,
|
|
bit 8 clear), '08' = UCS2. Matches the interactive GET INKEY/GET INPUT
|
|
response codings."""
|
|
if (dcs & 0x0C) == 0x08:
|
|
return text.encode('utf-16-be')
|
|
if dcs == 0x00:
|
|
return pack_gsm7(text.encode('gsm03.38'))
|
|
return text.encode('gsm03.38')
|
|
|
|
|
|
def build_tr(cmd_num, cmd_type, dev_dst, dev_src, respond, answer_tlvs=None):
|
|
"""Flat COMPREHENSION-TLV TERMINAL RESPONSE payload (TS 102 223 6.8) in the
|
|
6.8.0 object order: command details, device identities, Result, then the
|
|
command-specific objects (the terminal's own `answer_tlvs` - the PLI data
|
|
of PROVIDE LOCAL INFORMATION, the echoed Duration of POLL INTERVAL - the
|
|
Text string for GET INKEY/GET INPUT, the Item identifier for SELECT ITEM)
|
|
and any extra raw TLVs. Matches the interactive menu TR layout; the
|
|
runner passes the terminal objects the script did not override."""
|
|
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))
|
|
add_info = int(respond.get('additional_info', 0) or 0)
|
|
out += bytes([0x83, 0x02, result & 0xFF, add_info & 0xFF])
|
|
if answer_tlvs:
|
|
out += bytes.fromhex(answer_tlvs)
|
|
if respond.get('text') is not None:
|
|
text = str(respond['text'])
|
|
if text == '':
|
|
# TS 102 223 8.15: a null text string is Length 00, no value part
|
|
out += bytes([0x8D, 0x00])
|
|
else:
|
|
dcs = int(respond.get('dcs', 0x00))
|
|
body = _encode_text(text, dcs)
|
|
out += bytes([0x8D, len(body) + 1, dcs]) + body
|
|
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'])
|
|
return bytes(out)
|