v3.19.0: the applet-agent script batch - por objects, alpha/sms checks, scp80 format
The test scripts get the pieces an applet test needed:
- scp80 check.por accepts an object {status?, sw?, data?} asserting the
decoded PoR of this very exchange (inline or via SEND SHORT MESSAGE);
the evaluation is shared with the expect-step por check
(_por_check_results). The editor's PoR select gains 'custom' with the
three fields (an empty object refuses on save).
- a new alpha content check reads the command's Alpha identifier (05/85,
Annex A coded) - the text check stays bound to the 8D Text string, so a
command carrying both keeps them distinct.
- a new sms content check asserts a SEND SHORT MESSAGE's submit TPDU:
DA / PID / DCS / UDL / UD (the parsed fields are logged and reported;
an applet's own SMS, not the OTA PoR the por check decodes).
- scp80.params.format wraps the C-APDU in the TS 102 226 5.2.1 Command
Scripting template (expanded = AA..., expanded-ae = AE 80...00 00); the
log shows WRAPPED=, the report C-APDU=<effective> format=<fmt>.
- docs: Worksheet 1 addresses an RFM TAR (B00000/B00001/B00010 compact,
B00120/B00140/B00130 expanded) with the own-TAR and MSL/spi1 notes, and
file-write is documented as a direct APDU that does not fire
EVENT_REMOTE_FILE_UPDATE.
Tests: 819 frontend + 724 Python; sw cache simple-v387.
This commit is contained in:
+129
-35
@@ -34,7 +34,7 @@ from osmocom.tlv import BER_TLV_IE
|
||||
from osmocom.utils import rpad
|
||||
|
||||
|
||||
VERSION = '3.18.4'
|
||||
VERSION = '3.19.0'
|
||||
|
||||
MAX_ENVELOPE_SEGMENTS = 5 # max SMS segments for outgoing C-APDU in ENVELOPE
|
||||
|
||||
@@ -2783,6 +2783,34 @@ def _decode_sms_ud(pid, dcs, ud, udhi, udl):
|
||||
return out
|
||||
|
||||
|
||||
def _sms_tpdu_fields(raw):
|
||||
"""Parse the TPDU of a SEND SHORT MESSAGE command into the fields a test
|
||||
script can assert (the same SMS_SUBMIT parse the log uses): the message
|
||||
type, DA digits, PID/DCS, UDL and the raw user data. None when the
|
||||
command carries no TPDU."""
|
||||
tpdu_hex = _find_sms_tpdu(raw)
|
||||
if not tpdu_hex:
|
||||
return None
|
||||
try:
|
||||
tpdu = bytes.fromhex(tpdu_hex)
|
||||
mti = tpdu[0] & 0x03
|
||||
out = {'mti': mti, 'tpdu': tpdu_hex.upper()}
|
||||
if mti != 1:
|
||||
return out
|
||||
from pySim.sms import SMS_SUBMIT
|
||||
s = SMS_SUBMIT.from_bytes(tpdu)
|
||||
out.update({
|
||||
'da': str(getattr(s.tp_da, 'digits', '') or '').rstrip('fF'),
|
||||
'pid': '%02X' % s.tp_pid,
|
||||
'dcs': '%02X' % s.tp_dcs,
|
||||
'udl': int(s.tp_udl),
|
||||
'ud': bytes(s.tp_ud).hex().upper(),
|
||||
})
|
||||
return out
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _decode_send_sm(raw):
|
||||
"""Decode a SEND SHORT MESSAGE command (TS 102 223 6.4.10): alpha
|
||||
identifier, address and the 3GPP-SMS TPDU with its user data."""
|
||||
@@ -5767,12 +5795,41 @@ def _test_finish_entry(entry, result):
|
||||
fields = {'status': result.get('status', 'ok'),
|
||||
'checks': result.get('checks', []),
|
||||
'ms': int((time.time() - (entry.get('started') or time.time())) * 1000)}
|
||||
for key in ('sent', 'sw', 'data', 'note', 'por', 'counter', 'command'):
|
||||
for key in ('sent', 'sw', 'data', 'note', 'por', 'sms', 'counter', 'command'):
|
||||
if result.get(key) is not None:
|
||||
fields[key] = result[key]
|
||||
_test_entry_update(entry, fields)
|
||||
|
||||
|
||||
def _por_check_results(spec, por, level, prefix='PoR'):
|
||||
"""The decoded-PoR assertion rows (status/sw/data), shared by the scp80
|
||||
action's `check.por` object and the expectation's `por` content check."""
|
||||
checks = []
|
||||
if por is None:
|
||||
checks.append(_test_check_result(
|
||||
prefix, False, 'a decoded PoR',
|
||||
'no PoR in this command (or no scp80 step before it)', level))
|
||||
return checks
|
||||
if 'status' in spec:
|
||||
actual = str(por.get('response_status') or '')
|
||||
want = spec['status']
|
||||
ok = (actual.lower() == want.lower()
|
||||
or (want.lower() == 'ok' and actual == 'por_ok'))
|
||||
checks.append(_test_check_result(prefix + ' status', ok, want,
|
||||
actual or '(none)', level))
|
||||
if 'sw' in spec:
|
||||
actual = str((por.get('decoded') or {}).get('last_status_word') or '')
|
||||
checks.append(_test_check_result(
|
||||
prefix + ' SW', testscript.match_value(spec['sw'], actual),
|
||||
spec['sw']['value'], actual or '(none)', level))
|
||||
if 'data' in spec:
|
||||
actual = str((por.get('decoded') or {}).get('last_response_data') or '')
|
||||
checks.append(_test_check_result(
|
||||
prefix + ' data', testscript.match_value(spec['data'], actual),
|
||||
spec['data']['value'], actual or '(none)', level))
|
||||
return checks
|
||||
|
||||
|
||||
def _test_action_checks(step, sw, data, por, kind):
|
||||
check = step['check']
|
||||
level = step['on_fail']
|
||||
@@ -5786,15 +5843,19 @@ def _test_action_checks(step, sw, data, por, kind):
|
||||
else:
|
||||
checks.append(_test_check_result('Data', True, check['data']['value'],
|
||||
'(not checked - SW mismatch)', 'ok'))
|
||||
if kind == 'scp80' and check.get('por') != 'any':
|
||||
want = check['por']
|
||||
if want == 'none':
|
||||
ok = por is None
|
||||
actual = 'none' if por is None else 'present'
|
||||
else:
|
||||
ok = bool(por) and por.get('response_status') == 'por_ok'
|
||||
actual = (por or {}).get('response_status') or 'none'
|
||||
checks.append(_test_check_result('PoR', ok, want, actual, level))
|
||||
if kind == 'scp80':
|
||||
want = check.get('por')
|
||||
if isinstance(want, dict):
|
||||
# the decoded-PoR assertion (inline or submit transport)
|
||||
checks.extend(_por_check_results(want, por, level))
|
||||
elif want != 'any':
|
||||
if want == 'none':
|
||||
ok = por is None
|
||||
actual = 'none' if por is None else 'present'
|
||||
else:
|
||||
ok = bool(por) and por.get('response_status') == 'por_ok'
|
||||
actual = (por or {}).get('response_status') or 'none'
|
||||
checks.append(_test_check_result('PoR', ok, want, actual, level))
|
||||
return checks
|
||||
|
||||
|
||||
@@ -5899,15 +5960,25 @@ def _test_run_scp80(server, step, ctx):
|
||||
ctx['last_scp80'] = {'spi1': spi1, 'spi2': spi2, 'kic': kic, 'kid': kid,
|
||||
'counter': counter, 'kicKey': kic_key, 'kidKey': kid_key,
|
||||
'tar': tar}
|
||||
fmt = p.get('format') or 'compact'
|
||||
apdu_hex = p.get('apdu') or ''
|
||||
if p.get('sp'):
|
||||
sp_hex = p['sp']
|
||||
source = 'sp'
|
||||
_test_log('SCP80 SP=%s' % sp_hex)
|
||||
else:
|
||||
sp_hex, _ = _build_secured_packet(spi1, spi2, kic, kid, tar, counter,
|
||||
p['apdu'], kic_key, kid_key)
|
||||
source = 'apdu'
|
||||
_test_log('SCP80 C-APDU=%s' % p['apdu'])
|
||||
if fmt != 'compact':
|
||||
# TS 102 226 5.2.1: wrap the C-APDU in the Command Scripting
|
||||
# template (the expanded remote-management format the RFM/GP
|
||||
# applications expect for multi-command payloads)
|
||||
apdu_hex = _ram_format_apdu(apdu_hex, fmt)
|
||||
_test_log('SCP80 C-APDU=%s WRAPPED=%s (%s)'
|
||||
% (p['apdu'], apdu_hex, fmt))
|
||||
else:
|
||||
_test_log('SCP80 C-APDU=%s' % apdu_hex)
|
||||
sp_hex, _ = _build_secured_packet(spi1, spi2, kic, kid, tar, counter,
|
||||
apdu_hex, kic_key, kid_key)
|
||||
_test_log('SCP80 SECURED=%s (%d B) TAR=%s SPI1=%s SPI2=%s CNTR=%s'
|
||||
% (sp_hex, len(sp_hex) // 2, tar or '-', spi1, spi2, counter))
|
||||
result = _send_secured_packet(scc, sp_hex, server.sms_oa, sm_sc=server.sms_sc,
|
||||
@@ -5936,10 +6007,12 @@ def _test_run_scp80(server, step, ctx):
|
||||
_test_log('PoR[inline] undecodable raw=%s' % data)
|
||||
else:
|
||||
_test_log('PoR[inline] none')
|
||||
sent = 'SCP80 %s TAR=%s SPI1=%s SPI2=%s cntr=%s%s %s' % (
|
||||
sent = 'SCP80 %s TAR=%s SPI1=%s SPI2=%s cntr=%s%s %s%s' % (
|
||||
source, tar or '-', spi1, spi2, counter,
|
||||
(' kvn=%d' % kvn) if kvn else '',
|
||||
('SP=%s' % sp_hex) if source == 'sp' else ('C-APDU=%s' % p['apdu']))
|
||||
('SP=%s' % sp_hex) if source == 'sp'
|
||||
else ('C-APDU=%s' % apdu_hex),
|
||||
(' format=%s' % fmt) if (source == 'apdu' and fmt != 'compact') else '')
|
||||
if sw and _counter_tracked(spi1):
|
||||
# The card answered and the packet asks for a counter check, so the
|
||||
# SCP80 counter was consumed: advance the working value of this key
|
||||
@@ -6158,6 +6231,12 @@ def _test_run_expect(server, step, pending, ctx=None):
|
||||
type_name = PROACTIVE_TYPE_NAMES.get(cmd_type, 'UNKNOWN')
|
||||
_test_log('CMD 0x%02X %s qual=%s' % (
|
||||
cmd_type, type_name, ('%02X' % cmd_qual) if cmd_qual is not None else '-'))
|
||||
sms_fields = _sms_tpdu_fields(raw) if cmd_type == 0x13 else None
|
||||
if sms_fields and sms_fields.get('mti') == 1:
|
||||
_test_log('SMS DA=%s PID=%s DCS=%s UDL=%s UD=%s'
|
||||
% (sms_fields.get('da'), sms_fields.get('pid'),
|
||||
sms_fields.get('dcs'), sms_fields.get('udl'),
|
||||
sms_fields.get('ud')))
|
||||
# A SEND SHORT MESSAGE carries the PoR when the packet asked for it in
|
||||
# submit mode (SPI2 bit 0x20): decode it with the triggering scp80 step's
|
||||
# context so the checks can assert the PoR contents.
|
||||
@@ -6224,29 +6303,44 @@ def _test_run_expect(server, step, pending, ctx=None):
|
||||
testscript.match_value({'mode': c['mode'], 'value': c['value']}, actual),
|
||||
c['value'], actual, c['on_fail']))
|
||||
elif c['kind'] == 'por':
|
||||
if por is None:
|
||||
checks.extend(_por_check_results(c, por, c['on_fail']))
|
||||
elif c['kind'] == 'alpha':
|
||||
alpha = _cmd_tlv(httpota.proactive_tlvs(raw), 0x05)
|
||||
text = None
|
||||
if alpha:
|
||||
try:
|
||||
text = _STK_DECODE._decode(alpha, {}, 'stk')
|
||||
except Exception:
|
||||
text = None
|
||||
checks.append(_test_check_result(
|
||||
'Alpha', testscript.match_text(c, text),
|
||||
c['value'], text if text is not None else '(none)', c['on_fail']))
|
||||
elif c['kind'] == 'sms':
|
||||
if sms_fields is None or sms_fields.get('mti') != 1:
|
||||
checks.append(_test_check_result(
|
||||
'PoR', False, 'a decoded PoR',
|
||||
'no PoR in this command (or no scp80 step before it)',
|
||||
'SMS', False, 'a SUBMIT TPDU', 'none (not an SMS-SUBMIT)',
|
||||
c['on_fail']))
|
||||
else:
|
||||
if 'status' in c:
|
||||
actual = str(por.get('response_status') or '')
|
||||
want = c['status']
|
||||
ok = (actual.lower() == want.lower()
|
||||
or (want.lower() == 'ok' and actual == 'por_ok'))
|
||||
checks.append(_test_check_result('PoR status', ok, want,
|
||||
actual or '(none)', c['on_fail']))
|
||||
if 'sw' in c:
|
||||
actual = str((por.get('decoded') or {}).get('last_status_word') or '')
|
||||
if 'da' in c:
|
||||
checks.append(_test_check_result(
|
||||
'PoR SW', testscript.match_value(c['sw'], actual),
|
||||
c['sw']['value'], actual or '(none)', c['on_fail']))
|
||||
if 'data' in c:
|
||||
actual = str((por.get('decoded') or {}).get('last_response_data') or '')
|
||||
'SMS DA', sms_fields.get('da') == c['da'], c['da'],
|
||||
sms_fields.get('da') or '(none)', c['on_fail']))
|
||||
for label, key in (('PID', 'pid'), ('DCS', 'dcs')):
|
||||
if key in c:
|
||||
actual = sms_fields.get(key) or ''
|
||||
checks.append(_test_check_result(
|
||||
'SMS ' + label, testscript.match_value(c[key], actual),
|
||||
c[key]['value'], actual or '(none)', c['on_fail']))
|
||||
if 'udl' in c:
|
||||
actual = sms_fields.get('udl')
|
||||
checks.append(_test_check_result(
|
||||
'PoR data', testscript.match_value(c['data'], actual),
|
||||
c['data']['value'], actual or '(none)', c['on_fail']))
|
||||
'SMS UDL', actual == c['udl'], str(c['udl']),
|
||||
'(none)' if actual is None else str(actual), c['on_fail']))
|
||||
if 'ud' in c:
|
||||
actual = sms_fields.get('ud') or ''
|
||||
checks.append(_test_check_result(
|
||||
'SMS UD', testscript.match_value(c['ud'], actual),
|
||||
c['ud']['value'], actual or '(none)', c['on_fail']))
|
||||
elif c['kind'] == 'files':
|
||||
paths = _parse_file_list(raw)
|
||||
want = c['files']
|
||||
@@ -6261,7 +6355,7 @@ def _test_run_expect(server, step, pending, ctx=None):
|
||||
status = testscript.combine_levels([c['level'] if not c['ok'] else 'ok' for c in checks])
|
||||
result = {'status': status, 'sw': tr_sw, 'data': raw.hex().upper(),
|
||||
'sent': 'TR ' + tr.hex().upper(), 'checks': checks,
|
||||
'por': por,
|
||||
'por': por, 'sms': sms_fields,
|
||||
'command': {'type_hex': '%02X' % cmd_type, 'type_name': type_name,
|
||||
'qualifier': '%02X' % cmd_qual if cmd_qual is not None else None,
|
||||
'raw': raw.hex().upper()}}
|
||||
|
||||
@@ -54,6 +54,21 @@ class ScriptError(ValueError):
|
||||
|
||||
# ─── 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
|
||||
@@ -312,6 +327,15 @@ def _normalise_params(kind, p):
|
||||
# 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')
|
||||
@@ -337,14 +361,20 @@ def _normalise_check(check, kind, params):
|
||||
sw = {'mode': 'exact', 'value': '9000'}
|
||||
data = _check_spec(check.get('data'), 'check.data')
|
||||
por = check.get('por')
|
||||
if por is None:
|
||||
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 or any')
|
||||
raise ScriptError('check.por must be none, ok, any or an object')
|
||||
return {'sw': sw, 'data': data, 'por': por}
|
||||
|
||||
|
||||
@@ -416,15 +446,36 @@ def _normalise_content_check(c, default_level):
|
||||
'on_fail': level}
|
||||
if kind == 'por':
|
||||
out = {'kind': 'por', 'on_fail': level}
|
||||
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'), 'por check sw')
|
||||
if c.get('data') not in (None, ''):
|
||||
out['data'] = _check_spec(c.get('data'), 'por check data')
|
||||
if not (out.get('status') or out.get('sw') or out.get('data')):
|
||||
raise ScriptError('por check: status, sw or data is required')
|
||||
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))
|
||||
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'))
|
||||
|
||||
Reference in New Issue
Block a user