v3.25.2: the TR tag-form alignment with TS 102 223 6.8.0/6.8.1

The TR's response objects are all Min = N in the spec's own table, and
6.8.0 says such objects "should set the CR flag to comprehension not
required" - so the builders now emit the base tag forms:

- device identities 82 -> 02 (every builder)
- PLI data 93/94/AD/AE/BF/E2/E5 -> 13/14/2D/2E/3F/62/65 (26/63 already
  base); the PWA inline codecs follow
- POLL INTERVAL Duration 84 -> 04
- GET INKEY/INPUT Text string 8D -> 0D
- SELECT ITEM Item identifier 90 -> 10
- Timer identifier/value A4/A5 -> 24/25
(_bip_tr already used the reference traces' 01/02/03.)

The command details now echo the command's tag form (01/81) besides
its qualifier - 6.8.1 requires the object to be identical to the
command's "including the comprehension required flag": the parser
returns cd_tag and _build_tr, _handle_timer_command,
_handle_bip_command, the menu pendings and testscript.build_tr carry
it.  The Result keeps 03/83 (Min = Y).  _tr_data_only strips both tag
styles and the POLL Duration decode accepts 04.

Tests: the base forms per builder + the echo + the PWA drift guards;
883 frontend + 866 Python; sw cache simple-v426.
This commit is contained in:
2026-10-11 13:31:49 +03:00
parent beb001846f
commit 547ea61708
12 changed files with 238 additions and 161 deletions
+17 -17
View File
@@ -1905,7 +1905,7 @@
// ===== Version ===== // ===== Version =====
// Single source of truth for the PWA version: shown in the header and used // Single source of truth for the PWA version: shown in the header and used
// by the server version check in pysimConnect(). // by the server version check in pysimConnect().
const SIMPLE_VERSION = '3.25.1'; const SIMPLE_VERSION = '3.25.2';
document.getElementById('app-version').textContent = 'v' + SIMPLE_VERSION; document.getElementById('app-version').textContent = 'v' + SIMPLE_VERSION;
// ===== view history (the browser Back/Forward buttons) ===== // ===== view history (the browser Back/Forward buttons) =====
@@ -18140,24 +18140,24 @@ function encImeisvPli(digits) {
} }
const PLI_CODECS = { const PLI_CODECS = {
'00': { tlv: '9305', '00': { tlv: '1305',
decode(hex) { decode(hex) {
const v = hex.replace(/\s/g, '').slice(4); if (v.length < 10) return null; const v = hex.replace(/\s/g, '').slice(4); if (v.length < 10) return null;
const pl = decPlmn(v); return { mcc: pl[0], mnc: pl[1], lac: v.slice(6,10).toUpperCase() }; const pl = decPlmn(v); return { mcc: pl[0], mnc: pl[1], lac: v.slice(6,10).toUpperCase() };
}, },
encode(v) { return '9305' + encPlmn(v.mcc||'250',v.mnc||'01') + ((v.lac||'0000').toUpperCase()); }, encode(v) { return '1305' + encPlmn(v.mcc||'250',v.mnc||'01') + ((v.lac||'0000').toUpperCase()); },
fields: [ fields: [
{id:'mcc',label:'MCC',type:'text',pattern:'\\d{3}',placeholder:'e.g. 250'}, {id:'mcc',label:'MCC',type:'text',pattern:'\\d{3}',placeholder:'e.g. 250'},
{id:'mnc',label:'MNC',type:'text',pattern:'\\d{2,3}',placeholder:'e.g. 01'}, {id:'mnc',label:'MNC',type:'text',pattern:'\\d{2,3}',placeholder:'e.g. 01'},
{id:'lac',label:'LAC/TAC',type:'text',pattern:'[0-9A-Fa-f]{4}',placeholder:'hex'}, {id:'lac',label:'LAC/TAC',type:'text',pattern:'[0-9A-Fa-f]{4}',placeholder:'hex'},
], ],
}, },
'01': { tlv: '9408', '01': { tlv: '1408',
decode(hex) { decode(hex) {
const v = hex.replace(/\s/g, '').slice(4); if (v.length < 16) return null; const v = hex.replace(/\s/g, '').slice(4); if (v.length < 16) return null;
return { imei: decImei(v) }; return { imei: decImei(v) };
}, },
encode(v) { const imei = encImeiPli(v.imei||''); return imei ? '9408'+imei : ''; }, encode(v) { const imei = encImeiPli(v.imei||''); return imei ? '1408'+imei : ''; },
fields: [ fields: [
{id:'imei',label:'IMEI',type:'text',pattern:'\\d{15}',placeholder:'15 digits'}, {id:'imei',label:'IMEI',type:'text',pattern:'\\d{15}',placeholder:'15 digits'},
], ],
@@ -18210,39 +18210,39 @@ const PLI_CODECS = {
{id:'tz',label:'TZ offset',type:'text',pattern:'[-+]\\d{2}:?\\d{2}',placeholder:'+00:00'}, {id:'tz',label:'TZ offset',type:'text',pattern:'[-+]\\d{2}:?\\d{2}',placeholder:'+00:00'},
], ],
}, },
'04': { tlv: 'AD02', '04': { tlv: '2D02',
decode(hex) { decode(hex) {
const v = hex.replace(/\s/g, '').slice(4); if (v.length < 4) return null; const v = hex.replace(/\s/g, '').slice(4); if (v.length < 4) return null;
return { lang: String.fromCharCode(parseInt(v[0]+v[1],16), parseInt(v[2]+v[3],16)) }; return { lang: String.fromCharCode(parseInt(v[0]+v[1],16), parseInt(v[2]+v[3],16)) };
}, },
encode(v) { encode(v) {
const lang = (v.lang||'en').slice(0,2); const lang = (v.lang||'en').slice(0,2);
return 'AD02' + lang.charCodeAt(0).toString(16).padStart(2,'0') + lang.charCodeAt(1).toString(16).padStart(2,'0'); return '2D02' + lang.charCodeAt(0).toString(16).padStart(2,'0') + lang.charCodeAt(1).toString(16).padStart(2,'0');
}, },
fields: [ fields: [
{id:'lang',label:'Language',type:'text',pattern:'[a-zA-Z]{2,3}',placeholder:'en'}, {id:'lang',label:'Language',type:'text',pattern:'[a-zA-Z]{2,3}',placeholder:'en'},
], ],
}, },
'05': { tlv: 'AE02', '05': { tlv: '2E02',
decode(hex) { decode(hex) {
const v = hex.replace(/\s/g, '').slice(4); if (v.length < 4) return null; const v = hex.replace(/\s/g, '').slice(4); if (v.length < 4) return null;
return { status: parseInt(v[0]+v[1],16).toString(), ta: parseInt(v[2]+v[3],16).toString() }; return { status: parseInt(v[0]+v[1],16).toString(), ta: parseInt(v[2]+v[3],16).toString() };
}, },
encode(v) { return 'AE02' + (parseInt(v.status)||0).toString(16).padStart(2,'0') + (parseInt(v.ta)||0).toString(16).padStart(2,'0'); }, encode(v) { return '2E02' + (parseInt(v.status)||0).toString(16).padStart(2,'0') + (parseInt(v.ta)||0).toString(16).padStart(2,'0'); },
fields: [ fields: [
{id:'status',label:'ME Status',type:'number',min:'0',max:'255',placeholder:'0'}, {id:'status',label:'ME Status',type:'number',min:'0',max:'255',placeholder:'0'},
{id:'ta',label:'Timing Advance',type:'number',min:'0',max:'255',placeholder:'0'}, {id:'ta',label:'Timing Advance',type:'number',min:'0',max:'255',placeholder:'0'},
], ],
}, },
'06': { tlv: 'BF01', '06': { tlv: '3F01',
decode(hex) { decode(hex) {
const v = hex.replace(/\s/g, '').slice(4); if (v.length < 2) return null; const v = hex.replace(/\s/g, '').slice(4); if (v.length < 2) return null;
return { tech: parseInt(v[0]+v[1],16).toString() }; return { tech: parseInt(v[0]+v[1],16).toString() };
}, },
encode(v) { return 'BF01' + (parseInt(v.tech)||0).toString(16).padStart(2,'0'); }, encode(v) { return '3F01' + (parseInt(v.tech)||0).toString(16).padStart(2,'0'); },
fields: [{id:'tech',label:'Access Technology',type:'select',opts:ACCESSTECH}], fields: [{id:'tech',label:'Access Technology',type:'select',opts:ACCESSTECH}],
}, },
'08': { tlv: 'E208', '08': { tlv: '6208',
decode(hex) { decode(hex) {
const v = hex.replace(/\s/g, '').slice(4); if (v.length < 16) return null; const v = hex.replace(/\s/g, '').slice(4); if (v.length < 16) return null;
return { imeisv: decImei(v) + v.slice(14,16).toUpperCase() }; return { imeisv: decImei(v) + v.slice(14,16).toUpperCase() };
@@ -18253,17 +18253,17 @@ const PLI_CODECS = {
let h = ''; let h = '';
for (let i = 0; i < 16; i += 2) for (let i = 0; i < 16; i += 2)
h += (parseInt(raw[i+1])<<4 | parseInt(raw[i])).toString(16).padStart(2,'0'); h += (parseInt(raw[i+1])<<4 | parseInt(raw[i])).toString(16).padStart(2,'0');
return 'E208' + h; return '6208' + h;
}, },
fields: [{id:'imeisv',label:'IMEISV',type:'text',pattern:'\\d{16}',placeholder:'16 digits'}], fields: [{id:'imeisv',label:'IMEISV',type:'text',pattern:'\\d{16}',placeholder:'16 digits'}],
}, },
'09': { tlv: 'E501', '09': { tlv: '6501',
decode(hex) { decode(hex) {
const v = hex.replace(/\s/g, '').slice(4); if (v.length < 2) return null; const v = hex.replace(/\s/g, '').slice(4); if (v.length < 2) return null;
return { mode: parseInt(v[0]+v[1],16).toString() }; return { mode: parseInt(v[0]+v[1],16).toString() };
}, },
encode(v) { encode(v) {
return 'E501' + (parseInt(v.mode)||0).toString(16).padStart(2,'0'); return '6501' + (parseInt(v.mode)||0).toString(16).padStart(2,'0');
}, },
fields: [{id:'mode',label:'Search Mode',type:'select', fields: [{id:'mode',label:'Search Mode',type:'select',
opts:[{v:0,l:'Manual'},{v:1,l:'Automatic'}]}], opts:[{v:0,l:'Manual'},{v:1,l:'Automatic'}]}],
@@ -18281,7 +18281,7 @@ const PLI_CODECS = {
opts:[{v:0,l:'Very low'},{v:1,l:'Low'},{v:2,l:'Average'},{v:3,l:'Good'}, opts:[{v:0,l:'Very low'},{v:1,l:'Low'},{v:2,l:'Average'},{v:3,l:'Good'},
{v:4,l:'Full'},{v:0xFE,l:'Not battery powered'},{v:0xFF,l:'Unknown / charging'}]}], {v:4,l:'Full'},{v:0xFE,l:'Not battery powered'},{v:0xFF,l:'Unknown / charging'}]}],
}, },
'0E': { tlv: 'BF', '0E': { tlv: '3F',
decode(hex) { decode(hex) {
const v = hex.replace(/\s/g, '').slice(4); if (v.length < 2) return null; const v = hex.replace(/\s/g, '').slice(4); if (v.length < 2) return null;
const techs = []; for (let i = 0; i < v.length; i += 2) techs.push(parseInt(v[i]+v[i+1],16)); const techs = []; for (let i = 0; i < v.length; i += 2) techs.push(parseInt(v[i]+v[i+1],16));
@@ -18291,7 +18291,7 @@ const PLI_CODECS = {
const techs = (v.techs||'').split(',').filter(t => t !== '').map(t => parseInt(t)&0xFF); const techs = (v.techs||'').split(',').filter(t => t !== '').map(t => parseInt(t)&0xFF);
if (!techs.length) return ''; if (!techs.length) return '';
const len = techs.length.toString(16).padStart(2,'0'); const len = techs.length.toString(16).padStart(2,'0');
return 'BF' + len + techs.map(t => t.toString(16).padStart(2,'0')).join(''); return '3F' + len + techs.map(t => t.toString(16).padStart(2,'0')).join('');
}, },
fields: [ fields: [
{id:'techs',label:'Access Technologies (comma-separated)',type:'text', {id:'techs',label:'Access Technologies (comma-separated)',type:'text',
+1 -1
View File
@@ -1,4 +1,4 @@
const CACHE = 'simple-v425'; const CACHE = 'simple-v426';
const URLS = [ const URLS = [
'index.html', 'index.html',
'help.html', 'help.html',
+24 -15
View File
@@ -38,22 +38,31 @@ test('the PLI date/time codec uses the 8.39 TP-SCTS coding', () => {
assert.strictEqual(c.decode('2607123'), null); assert.strictEqual(c.decode('2607123'), null);
}); });
test('the PLI codec tags follow TS 101 220 Table 7.23', () => { test('the PLI codec tags follow TS 101 220 Table 7.23, base (CR=0) form', () => {
// 05 Timing Advance: 2E/AE (C6 is a 3GPP2 tag) // the TR's response objects are Min = N, so TS 102 223 6.8.0 says they
assert.strictEqual(PLI_CODECS['05'].tlv, 'AE02'); // "should set the CR flag to comprehension not required" (v3.25.2):
assert.strictEqual(PLI_CODECS['05'].encode({ status: '0', ta: '0' }), 'AE020000'); // the base tag values (13/14/2D/2E/3F/62/65/63/26), not the CR-set forms
// 08 IMEISV: 62/E2 (94 is the IMEI tag) assert.strictEqual(PLI_CODECS['00'].tlv, '1305');
assert.strictEqual(PLI_CODECS['08'].tlv, 'E208'); assert.strictEqual(PLI_CODECS['01'].tlv, '1408');
assert.strictEqual(PLI_CODECS['01'].encode({ imei: '490154203237518' }), '140894104502237315f8');
assert.strictEqual(PLI_CODECS['04'].tlv, '2D02');
assert.strictEqual(PLI_CODECS['06'].tlv, '3F01');
// 05 Timing Advance: 2E (C6 is a 3GPP2 tag)
assert.strictEqual(PLI_CODECS['05'].tlv, '2E02');
assert.strictEqual(PLI_CODECS['05'].encode({ status: '0', ta: '0' }), '2E020000');
// 08 IMEISV: 62 (94 is the IMEI tag)
assert.strictEqual(PLI_CODECS['08'].tlv, '6208');
assert.strictEqual(PLI_CODECS['08'].encode({ imeisv: '4901542032375108' }), assert.strictEqual(PLI_CODECS['08'].encode({ imeisv: '4901542032375108' }),
'E2089410450223731580'); '62089410450223731580');
// 09 Network search mode: 65/E5, '00' Manual / '01' Automatic (8.75) // 09 Network search mode: 65, '00' Manual / '01' Automatic (8.75)
assert.strictEqual(PLI_CODECS['09'].tlv, 'E501'); assert.strictEqual(PLI_CODECS['09'].tlv, '6501');
assert.strictEqual(PLI_CODECS['09'].fields[0].opts[0].l, 'Manual'); assert.strictEqual(PLI_CODECS['09'].fields[0].opts[0].l, 'Manual');
assert.strictEqual(PLI_CODECS['09'].fields[0].opts[1].l, 'Automatic'); assert.strictEqual(PLI_CODECS['09'].fields[0].opts[1].l, 'Automatic');
// 0A Battery state: 63/E3 is an enum, not a percentage (8.76) // 0A Battery state: 63/E3 is an enum, not a percentage (8.76)
assert.strictEqual(PLI_CODECS['0A'].tlv, '6301'); assert.strictEqual(PLI_CODECS['0A'].tlv, '6301');
assert.strictEqual(PLI_CODECS['0A'].encode({ charge: '4' }), '630104'); assert.strictEqual(PLI_CODECS['0A'].encode({ charge: '4' }), '630104');
assert.strictEqual(PLI_CODECS['0A'].encode({ charge: '0' }), '630100'); assert.strictEqual(PLI_CODECS['0A'].encode({ charge: '0' }), '630100');
assert.strictEqual(PLI_CODECS['0E'].tlv, '3F');
}); });
test('the IMEI encoder pads the odd 15th digit with F (TS 124 008)', () => { test('the IMEI encoder pads the odd 15th digit with F (TS 124 008)', () => {
@@ -65,16 +74,16 @@ test('the IMEI encoder pads the odd 15th digit with F (TS 124 008)', () => {
test('the server PLI defaults decode through their codecs', () => { test('the server PLI defaults decode through their codecs', () => {
// mirrors PLI_DEFAULTS in server.py (v3.23.1; a drift guard for the codecs) // mirrors PLI_DEFAULTS in server.py (v3.23.1; a drift guard for the codecs)
const defaults = { const defaults = {
'00': '930500F1100000', '01': '940894104502237315F8', '00': '130500F1100000', '01': '140894104502237315F8',
'04': 'AD02656E', '05': 'AE020000', '06': 'BF0108', '04': '2D02656E', '05': '2E020000', '06': '3F0108',
'08': 'E2089410450223731580', '09': 'E50101', '08': '62089410450223731580', '09': '650101',
'0A': '630104', '0E': 'BF020308', '0A': '630104', '0E': '3F020308',
}; };
for (const [code, hex] of Object.entries(defaults)) { for (const [code, hex] of Object.entries(defaults)) {
assert.ok(PLI_CODECS[code].decode(hex), code + ' must decode'); assert.ok(PLI_CODECS[code].decode(hex), code + ' must decode');
} }
assert.strictEqual(PLI_CODECS['01'].decode('940894104502237315F8').imei, assert.strictEqual(PLI_CODECS['01'].decode('140894104502237315F8').imei,
'490154203237518'); '490154203237518');
assert.deepStrictEqual(PLI_CODECS['09'].decode('E50101'), { mode: '1' }); assert.deepStrictEqual(PLI_CODECS['09'].decode('650101'), { mode: '1' });
assert.deepStrictEqual(PLI_CODECS['0A'].decode('630104'), { charge: '4' }); assert.deepStrictEqual(PLI_CODECS['0A'].decode('630104'), { charge: '4' });
}); });
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "pysim-simple-server" name = "pysim-simple-server"
version = "3.25.1" version = "3.25.2"
description = "HTTP REST server wrapping pysim for the SIMple PWA" description = "HTTP REST server wrapping pysim for the SIMple PWA"
requires-python = ">=3.8" requires-python = ">=3.8"
# pysim is a git-only dependency installed explicitly by setup.bat/setup.sh. # pysim is a git-only dependency installed explicitly by setup.bat/setup.sh.
+110 -69
View File
@@ -36,7 +36,7 @@ from osmocom.tlv import BER_TLV_IE
from osmocom.utils import rpad from osmocom.utils import rpad
VERSION = '3.25.1' VERSION = '3.25.2'
MAX_ENVELOPE_SEGMENTS = 5 # max SMS segments for outgoing C-APDU in ENVELOPE MAX_ENVELOPE_SEGMENTS = 5 # max SMS segments for outgoing C-APDU in ENVELOPE
@@ -2542,7 +2542,7 @@ class _DefaultProactiveHandler(ProactiveHandler):
try: try:
raw = cmd_obj.to_tlv() raw = cmd_obj.to_tlv()
if raw: if raw:
cmd_num, cmd_type, dev_src, dev_dst, cmd_qual = _parse_proactive_header(raw) cmd_num, cmd_type, dev_src, dev_dst, cmd_qual, _ = _parse_proactive_header(raw)
entry = _log_proactive(cmd_type, raw, cmd_qual, cmd_num) entry = _log_proactive(cmd_type, raw, cmd_qual, cmd_num)
except Exception: except Exception:
pass pass
@@ -2636,7 +2636,9 @@ PLI_QUALIFIER_NAMES = {
# Sensible PLI answers for an unconfigured dictionary - the qualifiers whose # Sensible PLI answers for an unconfigured dictionary - the qualifiers whose
# objects the pinned ETSI specs define (TS 102 223 8.x, tags per TS 101 220 # objects the pinned ETSI specs define (TS 102 223 8.x, tags per TS 101 220
# V18.3.0 Table 7.23). They are the *base layer*: a value set through # V18.3.0 Table 7.23, in the base (CR=0) form: the TR's response objects
# are Min = N and 6.8.0 says they "should set the CR flag to comprehension
# not required" - v3.25.2). They are the *base layer*: a value set through
# /api/pli-dict overrides its default, and clearing the field restores it # /api/pli-dict overrides its default, and clearing the field restores it
# (v3.23.1). Qualifier 03 stays empty - the date/time is answered from the # (v3.23.1). Qualifier 03 stays empty - the date/time is answered from the
# host clock (see _pli_data_hex). The 3GPP/3GPP2-specific qualifiers (02 # host clock (see _pli_data_hex). The 3GPP/3GPP2-specific qualifiers (02
@@ -2647,24 +2649,24 @@ PLI_QUALIFIER_NAMES = {
PLI_DEFAULTS = { PLI_DEFAULTS = {
# Location Information (8.19): PLMN 001/01 (the test network, TS 23.003) # Location Information (8.19): PLMN 001/01 (the test network, TS 23.003)
# + LAC 0000 # + LAC 0000
0x00: '9305' + '00F110' + '0000', 0x00: '1305' + '00F110' + '0000',
# IMEI (8.20): the classic test IMEI 490154203237518, nibble-swapped # IMEI (8.20): the classic test IMEI 490154203237518, nibble-swapped
# BCD with the odd 15th digit padded by F (TS 124 008 Mobile Identity) # BCD with the odd 15th digit padded by F (TS 124 008 Mobile Identity)
0x01: '9408' + '94104502237315F8', 0x01: '1408' + '94104502237315F8',
# Language (8.45): 'en' in the SMS default alphabet (bit 8 clear) # Language (8.45): 'en' in the SMS default alphabet (bit 8 clear)
0x04: 'AD02' + '656E', 0x04: '2D02' + '656E',
# Timing Advance (3GPP; tag 2E/AE): ME status 00 + TA 00 # Timing Advance (3GPP; base tag 2E): ME status 00 + TA 00
0x05: 'AE02' + '0000', 0x05: '2E02' + '0000',
# Access technology (8.61): E-UTRAN # Access technology (8.61): E-UTRAN
0x06: 'BF01' + '08', 0x06: '3F01' + '08',
# IMEISV (8.74, tag 62/E2): the test IMEI + SVN 08 # IMEISV (8.74, base tag 62): the test IMEI + SVN 08
0x08: 'E208' + '9410450223731580', 0x08: '6208' + '9410450223731580',
# Network search mode (8.75, tag 65/E5): automatic # Network search mode (8.75, base tag 65): automatic
0x09: 'E501' + '01', 0x09: '6501' + '01',
# Battery state (8.76, tag 63/E3): full # Battery state (8.76, base tag 63): full
0x0A: '6301' + '04', 0x0A: '6301' + '04',
# Access technologies (8.61): UTRAN + E-UTRAN # Access technologies (8.61): UTRAN + E-UTRAN
0x0E: 'BF02' + '0308', 0x0E: '3F02' + '0308',
} }
_PLI_DATA = {q: '' for q in PLI_QUALIFIER_NAMES} # the operator overrides _PLI_DATA = {q: '' for q in PLI_QUALIFIER_NAMES} # the operator overrides
@@ -2923,7 +2925,7 @@ _server_ref = None
def _tr_data_only(tr_tlv): def _tr_data_only(tr_tlv):
"""Strip boilerplate CTLVs (Command Details, Device IDs, Result) from TR, """Strip boilerplate CTLVs (Command Details, Device IDs, Result) from TR,
returning only command-specific data TLVs or empty bytes.""" returning only command-specific data TLVs or empty bytes."""
skip_tags = {0x81, 0x82, 0x03, 0x83} skip_tags = {0x01, 0x81, 0x02, 0x82, 0x03, 0x83}
data = bytearray() data = bytearray()
off = 0 off = 0
while off < len(tr_tlv) - 1: while off < len(tr_tlv) - 1:
@@ -3301,7 +3303,7 @@ def _decode_tr(type_hex, qual_hex, tr_hex):
cmd_type = int(type_hex, 16) if type_hex else None cmd_type = int(type_hex, 16) if type_hex else None
except ValueError: except ValueError:
return [{'label': 'Data', 'value': h}] return [{'label': 'Data', 'value': h}]
if cmd_type == 0x03 and len(h) >= 8 and h[0:2] == '84': if cmd_type == 0x03 and len(h) >= 8 and h[0:2] in ('84', '04'):
return [{'label': 'Interval', 'value': '%d s' % int(h[6:8], 16)}] return [{'label': 'Interval', 'value': '%d s' % int(h[6:8], 16)}]
if cmd_type == 0x27: if cmd_type == 0x27:
tlvs = _tlv_map(bytes.fromhex(h)) tlvs = _tlv_map(bytes.fromhex(h))
@@ -3378,13 +3380,17 @@ def _bip_channel_id(dev_dst):
return (dev_dst & 0x07) if 0x21 <= dev_dst <= 0x27 else None return (dev_dst & 0x07) if 0x21 <= dev_dst <= 0x27 else None
def _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst, result=0x00, info=None, extra=b''): def _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst, result=0x00, info=None, extra=b'',
cd_tag=None):
"""TERMINAL RESPONSE payload for a BIP command. """TERMINAL RESPONSE payload for a BIP command.
Tags follow the captured real-terminal traces (comprehension TLVs 01/02/03) The tags follow the captured real-terminal traces: the command details
and the result precedes the optional Channel status / data TLVs. echo the command's tag form (TS 102 223 6.8.1) and the response objects
use the base (CR=0) forms per 6.8.0 (Min = N); the result precedes the
optional Channel status / data TLVs.
""" """
tr = bytes([0x01, 0x03, cmd_num & 0xFF, cmd_type & 0xFF, (cmd_qual if cmd_qual is not None else 0) & 0xFF, tr = bytes([(cd_tag or 0x01), 0x03, cmd_num & 0xFF, cmd_type & 0xFF,
(cmd_qual if cmd_qual is not None else 0) & 0xFF,
0x02, 0x02, 0x82, 0x81]) 0x02, 0x02, 0x82, 0x81])
if info is None: if info is None:
tr += bytes([0x03, 0x01, result & 0xFF]) tr += bytes([0x03, 0x01, result & 0xFF])
@@ -3427,8 +3433,13 @@ def _decode_bip_cmd(cmd_type, raw):
return out return out
def _handle_bip_command(scc, cmd_num, cmd_type, cmd_qual, raw, dev_src, dev_dst): def _handle_bip_command(scc, cmd_num, cmd_type, cmd_qual, raw, dev_src, dev_dst, cd_tag=None):
"""Handle a BIP proactive command. Returns TR payload bytes, or None for the generic path.""" """Handle a BIP proactive command. Returns TR payload bytes, or None for the generic path."""
if cd_tag is None:
cd_tag = _cd_tag_of(raw)
def _tr(result=0x00, info=None, extra=b''):
return _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst,
result, info, extra, cd_tag=cd_tag)
tlvs = httpota.proactive_tlvs(raw) tlvs = httpota.proactive_tlvs(raw)
channel = _bip_channel_id(dev_dst) channel = _bip_channel_id(dev_dst)
if cmd_type == 0x40: if cmd_type == 0x40:
@@ -3439,7 +3450,7 @@ def _handle_bip_command(scc, cmd_num, cmd_type, cmd_qual, raw, dev_src, dev_dst)
extra += bytes([httpota.TAG_BUFFER_SIZE, 0x02]) + buffer_size.to_bytes(2, 'big') extra += bytes([httpota.TAG_BUFFER_SIZE, 0x02]) + buffer_size.to_bytes(2, 'big')
if not _BIP.enabled: if not _BIP.enabled:
_BIP.log('open-unavailable', reason='BIP not enabled') _BIP.log('open-unavailable', reason='BIP not enabled')
return _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst, 0x3A, 0x00, extra) return _tr(0x3A, 0x00, extra)
addr = httpota.parse_other_address(_cmd_tlv(tlvs, httpota.TAG_OTHER_ADDRESS)) addr = httpota.parse_other_address(_cmd_tlv(tlvs, httpota.TAG_OTHER_ADDRESS))
proto, port = httpota.parse_transport_level(_cmd_tlv(tlvs, httpota.TAG_TRANSPORT_LEVEL)) proto, port = httpota.parse_transport_level(_cmd_tlv(tlvs, httpota.TAG_TRANSPORT_LEVEL))
if not addr or port is None: if not addr or port is None:
@@ -3451,28 +3462,28 @@ def _handle_bip_command(scc, cmd_num, cmd_type, cmd_qual, raw, dev_src, dev_dst)
note='destination/transport not fully specified') note='destination/transport not fully specified')
cid, err = _BIP.open(addr or '-', port or 0, buffer_size, proto=proto) cid, err = _BIP.open(addr or '-', port or 0, buffer_size, proto=proto)
if cid is None: if cid is None:
return _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst, 0x3A, 0x00, extra) return _tr(0x3A, 0x00, extra)
if cmd_qual and (cmd_qual & 0x04): if cmd_qual and (cmd_qual & 0x04):
# Background mode: the terminal shall inform the UICC that the # Background mode: the terminal shall inform the UICC that the
# link was established (TS 102 223 7.5.11). # link was established (TS 102 223 7.5.11).
_BIP._queue_link_status(cid, status=0x80 | (cid & 0x07), info=0x00) _BIP._queue_link_status(cid, status=0x80 | (cid & 0x07), info=0x00)
status = bytes([httpota.TAG_CHANNEL_STATUS, 0x02, 0x80 | (cid & 0x07), 0x00]) status = bytes([httpota.TAG_CHANNEL_STATUS, 0x02, 0x80 | (cid & 0x07), 0x00])
return _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst, 0x00, None, status + extra) return _tr(0x00, None, status + extra)
if cmd_type == 0x43: if cmd_type == 0x43:
data = _cmd_tlv(tlvs, httpota.TAG_CHANNEL_DATA) data = _cmd_tlv(tlvs, httpota.TAG_CHANNEL_DATA)
if channel is None or not _BIP.send(channel, data): if channel is None or not _BIP.send(channel, data):
return _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst, 0x3A, 0x00) return _tr(0x3A, 0x00)
length = _BIP.send_capacity(channel) length = _BIP.send_capacity(channel)
return _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst, 0x00, None, return _tr(0x00, None,
bytes([httpota.TAG_CHANNEL_DATA_LENGTH, 0x01, length & 0xFF])) bytes([httpota.TAG_CHANNEL_DATA_LENGTH, 0x01, length & 0xFF]))
if cmd_type == 0x42: if cmd_type == 0x42:
req = _cmd_tlv(tlvs, httpota.TAG_CHANNEL_DATA_LENGTH) or b'\x00' req = _cmd_tlv(tlvs, httpota.TAG_CHANNEL_DATA_LENGTH) or b'\x00'
n = req[0] if req else 0 n = req[0] if req else 0
if channel is None: if channel is None:
return _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst, 0x3A, 0x00) return _tr(0x3A, 0x00)
data = _BIP.receive(channel, n) data = _BIP.receive(channel, n)
if data is None: if data is None:
return _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst, 0x3A, 0x00) return _tr(0x3A, 0x00)
remaining = _BIP.available(channel) remaining = _BIP.available(channel)
# The channel data TLV length is BER-encoded: a single byte only up # The channel data TLV length is BER-encoded: a single byte only up
# to 127, then the 0x81 long form (the reference terminal traces use # to 127, then the 0x81 long form (the reference terminal traces use
@@ -3486,16 +3497,16 @@ def _handle_bip_command(scc, cmd_num, cmd_type, cmd_qual, raw, dev_src, dev_dst)
else: else:
extra = bytes([httpota.TAG_CHANNEL_DATA, 0x81, len(data)]) + data extra = bytes([httpota.TAG_CHANNEL_DATA, 0x81, len(data)]) + data
extra += bytes([httpota.TAG_CHANNEL_DATA_LENGTH, 0x01, 0xFF if remaining > 0xFF else remaining]) extra += bytes([httpota.TAG_CHANNEL_DATA_LENGTH, 0x01, 0xFF if remaining > 0xFF else remaining])
return _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst, 0x00, None, extra) return _tr(0x00, None, extra)
if cmd_type == 0x41: if cmd_type == 0x41:
if channel is None or not _BIP.close(channel): if channel is None or not _BIP.close(channel):
return _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst, 0x3A, 0x00) return _tr(0x3A, 0x00)
return _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst, 0x00) return _tr(0x00)
if cmd_type == 0x44: if cmd_type == 0x44:
extra = b'' extra = b''
for cid in sorted(_BIP.channels): for cid in sorted(_BIP.channels):
extra += bytes([httpota.TAG_CHANNEL_STATUS, 0x02, 0x80 | (cid & 0x07), 0x00]) extra += bytes([httpota.TAG_CHANNEL_STATUS, 0x02, 0x80 | (cid & 0x07), 0x00])
return _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst, 0x00, None, extra) return _tr(0x00, None, extra)
return None return None
@@ -4279,20 +4290,23 @@ def _pli_data_hex(cmd_qual, dt=None):
return '' return ''
def _build_tr(scc, cmd_num, cmd_type, dev_src, dev_dst, cmd_qual, dt=None): def _build_tr(scc, cmd_num, cmd_type, dev_src, dev_dst, cmd_qual, dt=None, cd_tag=None):
"""Build the TERMINAL RESPONSE TLV payload for a fetched command. """Build the TERMINAL RESPONSE TLV payload for a fetched command.
The object order follows TS 102 223 6.8.0: command details, device The object order follows TS 102 223 6.8.0: command details, device
identities, Result, then the command-specific data objects (the echoed identities, Result, then the command-specific data objects (the echoed
Duration for POLL INTERVAL, the PLI data for PROVIDE LOCAL INFORMATION). Duration for POLL INTERVAL, the PLI data for PROVIDE LOCAL INFORMATION).
`dt` is a test seam for the PLI date/time default (see _pli_data_hex). `dt` is a test seam for the PLI date/time default (see _pli_data_hex).
The command details echo the command's qualifier: TS 102 223 6.8.1 The command details echo the command's tag form and qualifier (TS 102 223
requires the object to be identical to the command's (v3.25.1).""" 6.8.1: the object shall be identical to the command's); the response
base = bytes([0x81, 0x03, cmd_num, cmd_type, (cmd_qual or 0) & 0xFF, objects use the base (CR=0) forms per 6.8.0 ("for all COMPREHENSION-TLV
0x82, 0x02, dev_dst, dev_src, objects with Min = N the terminal should set the CR flag to comprehension
not required") - v3.25.2."""
base = bytes([(cd_tag or 0x81), 0x03, cmd_num, cmd_type, (cmd_qual or 0) & 0xFF,
0x02, 0x02, dev_dst, dev_src,
0x03, 0x01, 0x00]) 0x03, 0x01, 0x00])
if cmd_type == 0x03: if cmd_type == 0x03:
base += bytes([0x84, 0x02, 0x01, _POLL_INTERVAL]) base += bytes([0x04, 0x02, 0x01, _POLL_INTERVAL])
elif cmd_type == 0x26 and cmd_qual is not None: elif cmd_type == 0x26 and cmd_qual is not None:
pli_hex = _pli_data_hex(cmd_qual, dt) pli_hex = _pli_data_hex(cmd_qual, dt)
if pli_hex: if pli_hex:
@@ -5275,14 +5289,16 @@ def _timer_expired(timer_id, elapsed):
break break
def _handle_timer_command(cmd_num, cmd_type, cmd_qual, raw, dev_src, dev_dst): def _handle_timer_command(cmd_num, cmd_type, cmd_qual, raw, dev_src, dev_dst, cd_tag=None):
"""Terminal side of TIMER MANAGEMENT. Returns the TERMINAL RESPONSE payload.""" """Terminal side of TIMER MANAGEMENT. Returns the TERMINAL RESPONSE payload."""
tlvs = httpota.proactive_tlvs(raw) tlvs = httpota.proactive_tlvs(raw)
tid = _cmd_tlv(tlvs, 0x24) tid = _cmd_tlv(tlvs, 0x24)
timer_id = tid[0] if tid else 1 timer_id = tid[0] if tid else 1
action = (cmd_qual or 0) & 0x03 action = (cmd_qual or 0) & 0x03
base = bytes([0x81, 0x03, cmd_num, cmd_type, (cmd_qual or 0) & 0xFF, if cd_tag is None:
0x82, 0x02, dev_dst, dev_src]) cd_tag = _cd_tag_of(raw)
base = bytes([(cd_tag or 0x81), 0x03, cmd_num, cmd_type, (cmd_qual or 0) & 0xFF,
0x02, 0x02, dev_dst, dev_src])
if action == 0x00: if action == 0x00:
value = _cmd_tlv(tlvs, 0x25) value = _cmd_tlv(tlvs, 0x25)
if len(value) >= 3 and 1 <= timer_id <= 8: if len(value) >= 3 and 1 <= timer_id <= 8:
@@ -5296,7 +5312,7 @@ def _handle_timer_command(cmd_num, cmd_type, cmd_qual, raw, dev_src, dev_dst):
return base + bytes([0x03, 0x01, 0x24]) return base + bytes([0x03, 0x01, 0x24])
if action == 0x01: if action == 0x01:
_timer_cancel(timer_id) _timer_cancel(timer_id)
return (base + bytes([0xA4, 0x01, timer_id & 0xFF, 0xA5, 0x03]) return (base + bytes([0x24, 0x01, timer_id & 0xFF, 0x25, 0x03])
+ _hms_bcd(remaining) + bytes([0x03, 0x01, 0x00])) + _hms_bcd(remaining) + bytes([0x03, 0x01, 0x00]))
@@ -5326,15 +5342,31 @@ def _parse_proactive_header(raw):
cmd_num, cmd_type = 1, 0 cmd_num, cmd_type = 1, 0
cmd_qual = None cmd_qual = None
dev_src, dev_dst = 0x83, 0x81 dev_src, dev_dst = 0x83, 0x81
cd_tag = None
for tag, val in _ber_tlv_list(_proactive_body(raw)): for tag, val in _ber_tlv_list(_proactive_body(raw)):
# Cards use both the plain (01/02) and comprehension-required # Cards use both the plain (01/02) and comprehension-required
# (81/82) tag variants - TS 101 220 7.1.1 leaves the CR flag to the # (81/82) tag variants - TS 101 220 7.1.1 leaves the CR flag to the
# application, and the reference cards switch between them. # application, and the reference cards switch between them. The
# command-details tag is kept: the TR echoes it (TS 102 223 6.8.1,
# v3.25.2).
if tag in (0x01, 0x81) and len(val) >= 3: if tag in (0x01, 0x81) and len(val) >= 3:
cd_tag = tag
cmd_num, cmd_type, cmd_qual = val[0], val[1], val[2] cmd_num, cmd_type, cmd_qual = val[0], val[1], val[2]
elif tag in (0x02, 0x82) and len(val) >= 2: elif tag in (0x02, 0x82) and len(val) >= 2:
dev_src, dev_dst = val[0], val[1] dev_src, dev_dst = val[0], val[1]
return cmd_num, cmd_type, dev_src, dev_dst, cmd_qual return cmd_num, cmd_type, dev_src, dev_dst, cmd_qual, cd_tag
def _cd_tag_of(raw):
"""The command-details tag of a fetched command (0x01 or 0x81), or None
when it cannot be parsed: the TERMINAL RESPONSE echoes it (TS 102 223
6.8.1, v3.25.2)."""
if not raw:
return None
try:
return _parse_proactive_header(raw)[5]
except Exception:
return None
def _find_sms_tpdu(raw): def _find_sms_tpdu(raw):
@@ -5616,7 +5648,7 @@ def _run_proactive_chain(scc, sw91, on_fetch=None, status_poll=True, max_rounds=
fdata, sw = rv[0], rv[1] fdata, sw = rv[0], rv[1]
raw = bytes.fromhex(fdata) if fdata else None raw = bytes.fromhex(fdata) if fdata else None
action = None action = None
cmd_num, cmd_type, dev_src, dev_dst, cmd_qual = 1, 0, 0x83, 0x81, None cmd_num, cmd_type, dev_src, dev_dst, cmd_qual, cd_tag = 1, 0, 0x83, 0x81, None, None
entry = None entry = None
if raw: if raw:
# Every step is guarded: a decoder bug must never break the # Every step is guarded: a decoder bug must never break the
@@ -5625,7 +5657,7 @@ def _run_proactive_chain(scc, sw91, on_fetch=None, status_poll=True, max_rounds=
# card's command left unanswered). The exception and the # card's command left unanswered). The exception and the
# offending bytes go to the log; the TR is still sent. # offending bytes go to the log; the TR is still sent.
try: try:
cmd_num, cmd_type, dev_src, dev_dst, cmd_qual = _parse_proactive_header(raw) cmd_num, cmd_type, dev_src, dev_dst, cmd_qual, cd_tag = _parse_proactive_header(raw)
except Exception as e: except Exception as e:
sys.stderr.write('FETCH parse error: %s (raw=%s)\n' % (e, raw.hex())) sys.stderr.write('FETCH parse error: %s (raw=%s)\n' % (e, raw.hex()))
try: try:
@@ -5656,15 +5688,18 @@ def _run_proactive_chain(scc, sw91, on_fetch=None, status_poll=True, max_rounds=
tr_tlv = None tr_tlv = None
try: try:
if raw and cmd_type in (0x40, 0x41, 0x42, 0x43, 0x44): if raw and cmd_type in (0x40, 0x41, 0x42, 0x43, 0x44):
tr_tlv = _handle_bip_command(scc, cmd_num, cmd_type, cmd_qual, raw, dev_src, dev_dst) tr_tlv = _handle_bip_command(scc, cmd_num, cmd_type, cmd_qual, raw, dev_src, dev_dst,
cd_tag=cd_tag)
if cmd_type == 0x27: if cmd_type == 0x27:
tr_tlv = _handle_timer_command(cmd_num, cmd_type, cmd_qual, raw, dev_src, dev_dst) tr_tlv = _handle_timer_command(cmd_num, cmd_type, cmd_qual, raw, dev_src, dev_dst,
cd_tag=cd_tag)
except Exception as e: except Exception as e:
sys.stderr.write('FETCH handler error: %s (raw=%s)\n' % ( sys.stderr.write('FETCH handler error: %s (raw=%s)\n' % (
e, raw.hex() if raw else '(none)')) e, raw.hex() if raw else '(none)'))
tr_tlv = None tr_tlv = None
if tr_tlv is None: if tr_tlv is None:
tr_tlv = _build_tr(scc, cmd_num, cmd_type, dev_src, dev_dst, cmd_qual) tr_tlv = _build_tr(scc, cmd_num, cmd_type, dev_src, dev_dst, cmd_qual,
cd_tag=cd_tag)
tr_rv = scc._tp.send_apdu('%s140000%02x%s' % (scc.cat_cla, len(tr_tlv), tr_tlv.hex())) tr_rv = scc._tp.send_apdu('%s140000%02x%s' % (scc.cat_cla, len(tr_tlv), tr_tlv.hex()))
sys.stderr.write('TR: cmd=%02x type=%02x -> %s %s\n' % (cmd_num, cmd_type, tr_rv[1], ('(%d bytes)' % len(tr_tlv)))) sys.stderr.write('TR: cmd=%02x type=%02x -> %s %s\n' % (cmd_num, cmd_type, tr_rv[1], ('(%d bytes)' % len(tr_tlv))))
_record_tr(entry, tr_tlv, tr_rv[1]) _record_tr(entry, tr_tlv, tr_rv[1])
@@ -5735,6 +5770,7 @@ def _send_terminal_profile(scc, tp_hex):
fdata, sw = scc._tp.send_apdu('%s120000%02x' % (scc.cat_cla, fetch_len)) fdata, sw = scc._tp.send_apdu('%s120000%02x' % (scc.cat_cla, fetch_len))
cmd_num, cmd_type = 1, 0 cmd_num, cmd_type = 1, 0
cmd_qual = None cmd_qual = None
cd_tag = None
dev_src, dev_dst = 0x83, 0x81 dev_src, dev_dst = 0x83, 0x81
if fdata: if fdata:
raw = bytes.fromhex(fdata) raw = bytes.fromhex(fdata)
@@ -5745,6 +5781,7 @@ def _send_terminal_profile(scc, tp_hex):
val = raw[off + 2: off + 2 + tlen] val = raw[off + 2: off + 2 + tlen]
off += 2 + tlen off += 2 + tlen
if tag in (0x81, 0x01) and tlen >= 3: if tag in (0x81, 0x01) and tlen >= 3:
cd_tag = tag
cmd_num, cmd_type, cmd_qual = val[0], val[1], val[2] cmd_num, cmd_type, cmd_qual = val[0], val[1], val[2]
elif tag in (0x82, 0x02) and tlen >= 2: elif tag in (0x82, 0x02) and tlen >= 2:
dev_src, dev_dst = val[0], val[1] dev_src, dev_dst = val[0], val[1]
@@ -5760,7 +5797,8 @@ def _send_terminal_profile(scc, tp_hex):
entry = _log_proactive(cmd_type, raw, cmd_qual, cmd_num) entry = _log_proactive(cmd_type, raw, cmd_qual, cmd_num)
else: else:
entry = None entry = None
tr_tlv = _build_tr(scc, cmd_num, cmd_type, dev_src, dev_dst, cmd_qual) tr_tlv = _build_tr(scc, cmd_num, cmd_type, dev_src, dev_dst, cmd_qual,
cd_tag=cd_tag)
tr_rv = scc._tp.send_apdu('%s140000%02x%s' % (scc.cat_cla, len(tr_tlv), tr_tlv.hex())) tr_rv = scc._tp.send_apdu('%s140000%02x%s' % (scc.cat_cla, len(tr_tlv), tr_tlv.hex()))
sys.stderr.write('TR(tp): cmd=%02x type=%02x -> %s\n' % (cmd_num, cmd_type, tr_rv[1])) sys.stderr.write('TR(tp): cmd=%02x type=%02x -> %s\n' % (cmd_num, cmd_type, tr_rv[1]))
_record_tr(entry, tr_tlv, tr_rv[1]) _record_tr(entry, tr_tlv, tr_rv[1])
@@ -5814,9 +5852,11 @@ def _make_menu_fetch_handler(server, resp):
# 6.8.1); the on_fetch callback signature does not carry it, so parse # 6.8.1); the on_fetch callback signature does not carry it, so parse
# it here (v3.25.1) # it here (v3.25.1)
cmd_qual = None cmd_qual = None
cd_tag = None
if raw: if raw:
try: try:
cmd_qual = _parse_proactive_header(raw)[4] hdr = _parse_proactive_header(raw)
cmd_qual, cd_tag = hdr[4], hdr[5]
except Exception: except Exception:
cmd_qual = None cmd_qual = None
if cmd_type == 0x21: if cmd_type == 0x21:
@@ -5824,7 +5864,7 @@ def _make_menu_fetch_handler(server, resp):
if text: if text:
server.stk_pending = {'type': 'display_text', server.stk_pending = {'type': 'display_text',
'cmd_num': cmd_num, 'cmd_type': cmd_type, 'cmd_num': cmd_num, 'cmd_type': cmd_type,
'cmd_qual': cmd_qual, 'cmd_qual': cmd_qual, 'cd_tag': cd_tag,
'dev_src': dev_src, 'dev_dst': dev_dst, 'text': text} 'dev_src': dev_src, 'dev_dst': dev_dst, 'text': text}
resp.update(type='display_text', text=text) resp.update(type='display_text', text=text)
return 'pause' return 'pause'
@@ -5832,7 +5872,7 @@ def _make_menu_fetch_handler(server, resp):
items = _parse_select_item(raw) if raw else [] items = _parse_select_item(raw) if raw else []
server.stk_pending = {'type': 'select_item', server.stk_pending = {'type': 'select_item',
'cmd_num': cmd_num, 'cmd_type': cmd_type, 'cmd_num': cmd_num, 'cmd_type': cmd_type,
'cmd_qual': cmd_qual, 'cmd_qual': cmd_qual, 'cd_tag': cd_tag,
'dev_src': dev_src, 'dev_dst': dev_dst, 'items': items} 'dev_src': dev_src, 'dev_dst': dev_dst, 'items': items}
resp.update(type='select_item', items=items) resp.update(type='select_item', items=items)
return 'pause' return 'pause'
@@ -5840,7 +5880,7 @@ def _make_menu_fetch_handler(server, resp):
items = _parse_setup_menu_items(raw) if raw else [] items = _parse_setup_menu_items(raw) if raw else []
server.stk_pending = {'type': 'select_item', server.stk_pending = {'type': 'select_item',
'cmd_num': cmd_num, 'cmd_type': cmd_type, 'cmd_num': cmd_num, 'cmd_type': cmd_type,
'cmd_qual': cmd_qual, 'cmd_qual': cmd_qual, 'cd_tag': cd_tag,
'dev_src': dev_src, 'dev_dst': dev_dst, 'items': items} 'dev_src': dev_src, 'dev_dst': dev_dst, 'items': items}
resp.update(type='select_item', items=items) resp.update(type='select_item', items=items)
return 'pause' return 'pause'
@@ -5852,7 +5892,7 @@ def _make_menu_fetch_handler(server, resp):
if info: if info:
kind = 'get_inkey' if cmd_type == 0x22 else 'get_input' kind = 'get_inkey' if cmd_type == 0x22 else 'get_input'
pending = dict(info, type=kind, cmd_num=cmd_num, pending = dict(info, type=kind, cmd_num=cmd_num,
cmd_type=cmd_type, cmd_qual=cmd_qual, cmd_type=cmd_type, cmd_qual=cmd_qual, cd_tag=cd_tag,
dev_src=dev_src, dev_dst=dev_dst) dev_src=dev_src, dev_dst=dev_dst)
dur = _find_duration(raw) dur = _find_duration(raw)
pending['duration_unit'] = dur[0] if dur else None pending['duration_unit'] = dur[0] if dur else None
@@ -5906,7 +5946,7 @@ def _input_text_tlv(pd, text):
SMS default alphabet unpacked (DCS '04'), as UCS2 ('08') when the request SMS default alphabet unpacked (DCS '04'), as UCS2 ('08') when the request
asked for it, or packed ('00') for a packed GET INPUT.""" asked for it, or packed ('00') for a packed GET INPUT."""
if text == '': if text == '':
return bytes([0x8D, 0x00]) # null text string (empty input) return bytes([0x0D, 0x00]) # null text string (empty input)
if pd.get('yes_no'): if pd.get('yes_no'):
body = bytes([0x04, 0x01 if text == '01' else 0x00]) body = bytes([0x04, 0x01 if text == '01' else 0x00])
elif pd.get('ucs2'): elif pd.get('ucs2'):
@@ -5915,7 +5955,7 @@ def _input_text_tlv(pd, text):
body = bytes([0x00]) + _pack_gsm7(text.encode('gsm03.38')) body = bytes([0x00]) + _pack_gsm7(text.encode('gsm03.38'))
else: else:
body = bytes([0x04]) + text.encode('gsm03.38') body = bytes([0x04]) + text.encode('gsm03.38')
return bytes([0x8D, len(body)]) + body return bytes([0x0D, len(body)]) + body
def _menu_send_response(server, result, item_id=None, text=None): def _menu_send_response(server, result, item_id=None, text=None):
@@ -5949,16 +5989,16 @@ def _menu_send_response(server, result, item_id=None, text=None):
0x02: 10.0}.get(pd['duration_unit'], 1.0))) 0x02: 10.0}.get(pd['duration_unit'], 1.0)))
value = max(1, min(255, value)) value = max(1, min(255, value))
duration_tlv = bytes([0x04, 0x02, pd['duration_unit'], value]) duration_tlv = bytes([0x04, 0x02, pd['duration_unit'], value])
cd = bytes([0x81, 0x03, pd['cmd_num'], pd['cmd_type'], cd = bytes([(pd.get('cd_tag') or 0x81), 0x03, pd['cmd_num'], pd['cmd_type'],
(pd.get('cmd_qual') or 0) & 0xFF]) (pd.get('cmd_qual') or 0) & 0xFF])
di = bytes([0x82, 0x02, pd['dev_dst'], pd['dev_src']]) di = bytes([0x02, 0x02, pd['dev_dst'], pd['dev_src']])
# TS 102 223 6.8.0 object order: Result (C), Duration (D), Text string (E), # TS 102 223 6.8.0 object order: Result (C), Duration (D), Text string (E),
# Item identifier (F). # Item identifier (F).
tr_data = cd + di tr_data = cd + di
tr_data += bytes([0x83, 0x02, gr, 0x00]) tr_data += bytes([0x83, 0x02, gr, 0x00])
tr_data += duration_tlv + text_tlv tr_data += duration_tlv + text_tlv
if isinstance(item_id, int) and result == 'ok' and pd['type'] == 'select_item': if isinstance(item_id, int) and result == 'ok' and pd['type'] == 'select_item':
tr_data += bytes([0x90, 0x01, item_id]) tr_data += bytes([0x10, 0x01, item_id])
tr_hex = '%s140000%02x%s' % (scc.cat_cla, len(tr_data), tr_data.hex()) tr_hex = '%s140000%02x%s' % (scc.cat_cla, len(tr_data), tr_data.hex())
tr_rv = scc._tp.send_apdu(tr_hex) tr_rv = scc._tp.send_apdu(tr_hex)
sys.stderr.write('TR(menu): cmd=%02x type=%02x result=%02x -> %s\n' % (pd['cmd_num'], pd['cmd_type'], gr, tr_rv[1])) sys.stderr.write('TR(menu): cmd=%02x type=%02x result=%02x -> %s\n' % (pd['cmd_num'], pd['cmd_type'], gr, tr_rv[1]))
@@ -6552,7 +6592,7 @@ def _drain_require_label(req):
return name + ((' q=%s' % q['value']) if q else '') return name + ((' q=%s' % q['value']) if q else '')
def _test_terminal_response(cmd_num, cmd_type, dev_dst, dev_src, cmd_qual, respond): def _test_terminal_response(cmd_num, cmd_type, dev_dst, dev_src, cmd_qual, respond, cd_tag=None):
"""The runner's TERMINAL RESPONSE: the scripted answer plus the terminal's """The runner's TERMINAL RESPONSE: the scripted answer plus the terminal's
own command-specific object(s) - the same objects the terminal's answers own command-specific object(s) - the same objects the terminal's answers
carry (`_build_tr`), unless the script supplied `raw` TLVs, which win. carry (`_build_tr`), unless the script supplied `raw` TLVs, which win.
@@ -6569,7 +6609,8 @@ def _test_terminal_response(cmd_num, cmd_type, dev_dst, dev_src, cmd_qual, respo
elif cmd_type == 0x03: elif cmd_type == 0x03:
answer = '840201%02X' % _POLL_INTERVAL answer = '840201%02X' % _POLL_INTERVAL
return testscript.build_tr(cmd_num, cmd_type, dev_dst, dev_src, respond, return testscript.build_tr(cmd_num, cmd_type, dev_dst, dev_src, respond,
answer_tlvs=answer or None, cmd_qual=cmd_qual) answer_tlvs=answer or None, cmd_qual=cmd_qual,
cd_tag=cd_tag)
def _test_run_drain(server, step): def _test_run_drain(server, step):
@@ -6598,7 +6639,7 @@ def _test_run_drain(server, step):
break break
raw = bytes.fromhex(fdata) raw = bytes.fromhex(fdata)
_test_log('FETCH=%s -> SW=%s %s' % (fetch_apdu, fetch_sw, raw.hex().upper())) _test_log('FETCH=%s -> SW=%s %s' % (fetch_apdu, fetch_sw, raw.hex().upper()))
cmd_num, cmd_type, dev_src, dev_dst, cmd_qual = _parse_proactive_header(raw) cmd_num, cmd_type, dev_src, dev_dst, cmd_qual, cd_tag = _parse_proactive_header(raw)
name = PROACTIVE_TYPE_NAMES.get(cmd_type, 'UNKNOWN') name = PROACTIVE_TYPE_NAMES.get(cmd_type, 'UNKNOWN')
_test_log('CMD 0x%02X %s qual=%s' % ( _test_log('CMD 0x%02X %s qual=%s' % (
cmd_type, name, ('%02X' % cmd_qual) if cmd_qual is not None else '-')) cmd_type, name, ('%02X' % cmd_qual) if cmd_qual is not None else '-'))
@@ -6609,7 +6650,7 @@ def _test_run_drain(server, step):
server.sim_menu = menu server.sim_menu = menu
respond = p['first'] if (not consumed and p.get('first')) else p['respond'] respond = p['first'] if (not consumed and p.get('first')) else p['respond']
tr = _test_terminal_response(cmd_num, cmd_type, dev_dst, dev_src, tr = _test_terminal_response(cmd_num, cmd_type, dev_dst, dev_src,
cmd_qual, respond) cmd_qual, respond, cd_tag=cd_tag)
tr_rv = scc._tp.send_apdu('%s140000%02x%s' % (scc.cat_cla, len(tr), tr.hex())) tr_rv = scc._tp.send_apdu('%s140000%02x%s' % (scc.cat_cla, len(tr), tr.hex()))
_test_log('TR=%s -> SW=%s' % (tr.hex().upper(), tr_rv[1])) _test_log('TR=%s -> SW=%s' % (tr.hex().upper(), tr_rv[1]))
consumed.append({'type_hex': '%02X' % cmd_type, 'type_name': name, consumed.append({'type_hex': '%02X' % cmd_type, 'type_name': name,
@@ -7035,7 +7076,7 @@ def _test_run_expect(server, step, pending, ctx=None):
'returned no data')]}, None 'returned no data')]}, None
raw = bytes.fromhex(fdata) raw = bytes.fromhex(fdata)
_test_log('FETCH=%s -> SW=%s %s' % (fetch_apdu, fetch_sw, raw.hex().upper())) _test_log('FETCH=%s -> SW=%s %s' % (fetch_apdu, fetch_sw, raw.hex().upper()))
cmd_num, cmd_type, dev_src, dev_dst, cmd_qual = _parse_proactive_header(raw) cmd_num, cmd_type, dev_src, dev_dst, cmd_qual, cd_tag = _parse_proactive_header(raw)
log_entry = _log_proactive(cmd_type, raw, cmd_qual, cmd_num) log_entry = _log_proactive(cmd_type, raw, cmd_qual, cmd_num)
type_name = PROACTIVE_TYPE_NAMES.get(cmd_type, 'UNKNOWN') type_name = PROACTIVE_TYPE_NAMES.get(cmd_type, 'UNKNOWN')
_test_log('CMD 0x%02X %s qual=%s' % ( _test_log('CMD 0x%02X %s qual=%s' % (
@@ -7164,7 +7205,7 @@ def _test_run_expect(server, step, pending, ctx=None):
'Files', sorted(paths) == sorted(want), ', '.join(want), 'Files', sorted(paths) == sorted(want), ', '.join(want),
', '.join(paths) or '(none)', c['on_fail'])) ', '.join(paths) or '(none)', c['on_fail']))
tr = _test_terminal_response(cmd_num, cmd_type, dev_dst, dev_src, cmd_qual, tr = _test_terminal_response(cmd_num, cmd_type, dev_dst, dev_src, cmd_qual,
step['respond']) step['respond'], cd_tag=cd_tag)
tr_rv = scc._tp.send_apdu('%s140000%02x%s' % (scc.cat_cla, len(tr), tr.hex())) tr_rv = scc._tp.send_apdu('%s140000%02x%s' % (scc.cat_cla, len(tr), tr.hex()))
tr_sw = tr_rv[1] tr_sw = tr_rv[1]
_test_log('TR=%s -> SW=%s' % (tr.hex().upper(), tr_sw)) _test_log('TR=%s -> SW=%s' % (tr.hex().upper(), tr_sw))
@@ -7202,9 +7243,9 @@ def _test_drain_pending(scc, fetch_len, result=0x00, max_commands=3):
if not fdata: if not fdata:
break break
raw = bytes.fromhex(fdata) raw = bytes.fromhex(fdata)
cmd_num, cmd_type, dev_src, dev_dst, cmd_qual = _parse_proactive_header(raw) cmd_num, cmd_type, dev_src, dev_dst, cmd_qual, cd_tag = _parse_proactive_header(raw)
tr = _test_terminal_response(cmd_num, cmd_type, dev_dst, dev_src, tr = _test_terminal_response(cmd_num, cmd_type, dev_dst, dev_src,
cmd_qual, {'result': result}) cmd_qual, {'result': result}, cd_tag=cd_tag)
rv = scc._tp.send_apdu('%s140000%02x%s' % (scc.cat_cla, len(tr), tr.hex())) rv = scc._tp.send_apdu('%s140000%02x%s' % (scc.cat_cla, len(tr), tr.hex()))
name = PROACTIVE_TYPE_NAMES.get(cmd_type, '?') name = PROACTIVE_TYPE_NAMES.get(cmd_type, '?')
answered.append({'name': name, 'type': cmd_type, answered.append({'name': name, 'type': cmd_type,
+9 -8
View File
@@ -701,7 +701,7 @@ def _encode_text(text, dcs):
def build_tr(cmd_num, cmd_type, dev_dst, dev_src, respond, answer_tlvs=None, def build_tr(cmd_num, cmd_type, dev_dst, dev_src, respond, answer_tlvs=None,
cmd_qual=None): cmd_qual=None, cd_tag=None):
"""Flat COMPREHENSION-TLV TERMINAL RESPONSE payload (TS 102 223 6.8) in the """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 6.8.0 object order: command details, device identities, Result, then the
command-specific objects (the terminal's own `answer_tlvs` - the PLI data command-specific objects (the terminal's own `answer_tlvs` - the PLI data
@@ -709,11 +709,12 @@ def build_tr(cmd_num, cmd_type, dev_dst, dev_src, respond, answer_tlvs=None,
Text string for GET INKEY/GET INPUT, the Item identifier for SELECT ITEM) 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 and any extra raw TLVs. Matches the interactive menu TR layout; the
runner passes the terminal objects the script did not override. The runner passes the terminal objects the script did not override. The
command details echo the command's qualifier (TS 102 223 6.8.1: the command details echo the command's tag form and qualifier (TS 102 223
object shall be identical to the command's).""" 6.8.1: the object shall be identical to the command's); the response
out = bytearray([0x81, 0x03, cmd_num & 0xFF, cmd_type & 0xFF, objects use the base (CR=0) forms per 6.8.0 (Min = N) - v3.25.2."""
out = bytearray([(cd_tag or 0x81), 0x03, cmd_num & 0xFF, cmd_type & 0xFF,
(cmd_qual or 0) & 0xFF]) (cmd_qual or 0) & 0xFF])
out += bytes([0x82, 0x02, dev_dst & 0xFF, dev_src & 0xFF]) out += bytes([0x02, 0x02, dev_dst & 0xFF, dev_src & 0xFF])
result = int(respond.get('result', 0)) result = int(respond.get('result', 0))
add_info = int(respond.get('additional_info', 0) or 0) add_info = int(respond.get('additional_info', 0) or 0)
out += bytes([0x83, 0x02, result & 0xFF, add_info & 0xFF]) out += bytes([0x83, 0x02, result & 0xFF, add_info & 0xFF])
@@ -723,13 +724,13 @@ def build_tr(cmd_num, cmd_type, dev_dst, dev_src, respond, answer_tlvs=None,
text = str(respond['text']) text = str(respond['text'])
if text == '': if text == '':
# TS 102 223 8.15: a null text string is Length 00, no value part # TS 102 223 8.15: a null text string is Length 00, no value part
out += bytes([0x8D, 0x00]) out += bytes([0x0D, 0x00])
else: else:
dcs = int(respond.get('dcs', 0x00)) dcs = int(respond.get('dcs', 0x00))
body = _encode_text(text, dcs) body = _encode_text(text, dcs)
out += bytes([0x8D, len(body) + 1, dcs]) + body out += bytes([0x0D, len(body) + 1, dcs]) + body
if respond.get('item_id') is not None and result == 0x00: if respond.get('item_id') is not None and result == 0x00:
out += bytes([0x90, 0x01, respond['item_id'] & 0xFF]) out += bytes([0x10, 0x01, respond['item_id'] & 0xFF])
if respond.get('raw'): if respond.get('raw'):
out += bytes.fromhex(respond['raw']) out += bytes.fromhex(respond['raw'])
return bytes(out) return bytes(out)
+4 -4
View File
@@ -145,15 +145,15 @@ class PliDictHttpTests(unittest.TestCase):
try: try:
status, resp = self._req('GET', '/api/pli-dict') status, resp = self._req('GET', '/api/pli-dict')
self.assertEqual(status, 200, resp) self.assertEqual(status, 200, resp)
self.assertEqual(resp['00'], '930500F1100000') self.assertEqual(resp['00'], '130500F1100000')
self.assertEqual(resp['0A'], '630104') self.assertEqual(resp['0A'], '630104')
self.assertEqual(resp['03'], '') # the host clock answers it self.assertEqual(resp['03'], '') # the host clock answers it
self.assertEqual(resp['02'], '') # no default self.assertEqual(resp['02'], '') # no default
status, resp = self._req('POST', '/api/pli-dict', status, resp = self._req('POST', '/api/pli-dict',
{'01': '94080000000000000000'}) {'01': '14080000000000000000'})
self.assertEqual(status, 200, resp) self.assertEqual(status, 200, resp)
self.assertEqual(resp['01'], '94080000000000000000') self.assertEqual(resp['01'], '14080000000000000000')
self.assertEqual(srv._PLI_DATA.get(0x01), '94080000000000000000') self.assertEqual(srv._PLI_DATA.get(0x01), '14080000000000000000')
# clearing the field restores the default # clearing the field restores the default
status, resp = self._req('POST', '/api/pli-dict', {'01': ''}) status, resp = self._req('POST', '/api/pli-dict', {'01': ''})
self.assertEqual(status, 200, resp) self.assertEqual(status, 200, resp)
+6 -3
View File
@@ -30,6 +30,9 @@ OPEN_LOCALHOST = bytes.fromhex(
TR_OPEN_OK = '0103014001020282810301003802810035010339020200' TR_OPEN_OK = '0103014001020282810301003802810035010339020200'
TR_OPEN_FAIL = '01030140010202828103023a0035010339020200' TR_OPEN_FAIL = '01030140010202828103023a0035010339020200'
# the same vector as sent by the handler: the command details echo the
# command's tag form (the test raws use 81/82, TS 102 223 6.8.1)
TR_OPEN_OK_ECHO = '8103014001020282810301003802810035010339020200'
class PeerServer(threading.Thread): class PeerServer(threading.Thread):
@@ -314,13 +317,13 @@ class BipCommandFlowTest(unittest.TestCase):
def test_open_send_receive_close_flow(self): def test_open_send_receive_close_flow(self):
tr = server._handle_bip_command(None, 1, 0x40, 0x01, open_cmd('127.0.0.1', self.peer.port), 0x81, 0x82) tr = server._handle_bip_command(None, 1, 0x40, 0x01, open_cmd('127.0.0.1', self.peer.port), 0x81, 0x82)
self.assertEqual(tr.hex(), TR_OPEN_OK) self.assertEqual(tr.hex(), TR_OPEN_OK_ECHO)
tr = server._handle_bip_command(None, 1, 0x43, 0x01, tr = server._handle_bip_command(None, 1, 0x43, 0x01,
channel_cmd(0x43, 0x01, bytes([0x36, 0x08]) + b'CLIENTHE'), channel_cmd(0x43, 0x01, bytes([0x36, 0x08]) + b'CLIENTHE'),
0x81, 0x21) 0x81, 0x21)
tlvs = parse_tr(tr) tlvs = parse_tr(tr)
self.assertEqual(tlvs[0x01].hex(), '014301') self.assertEqual(tlvs[0x81].hex(), '014301')
self.assertEqual(tlvs[0x02].hex(), '8281') self.assertEqual(tlvs[0x02].hex(), '8281')
self.assertEqual(tlvs[0x03], b'\x00') self.assertEqual(tlvs[0x03], b'\x00')
self.assertEqual(tlvs[0x37], b'\xff') self.assertEqual(tlvs[0x37], b'\xff')
@@ -337,7 +340,7 @@ class BipCommandFlowTest(unittest.TestCase):
self.assertEqual(tlvs[0x37], b'\x00') self.assertEqual(tlvs[0x37], b'\x00')
tr = server._handle_bip_command(None, 1, 0x41, 0x00, channel_cmd(0x41, 0x00), 0x81, 0x21) tr = server._handle_bip_command(None, 1, 0x41, 0x00, channel_cmd(0x41, 0x00), 0x81, 0x21)
self.assertEqual(tr.hex(), '010301410002028281030100') self.assertEqual(tr.hex(), '810301410002028281030100')
self.peer.done.wait(3) self.peer.done.wait(3)
self.assertEqual(self.peer.received, b'CLIENTHE') self.assertEqual(self.peer.received, b'CLIENTHE')
+1 -1
View File
@@ -76,7 +76,7 @@ class TestMenuSendResponse(unittest.TestCase):
tr = sent[0] tr = sent[0]
self.assertTrue(tr.startswith('801400000d'), tr) self.assertTrue(tr.startswith('801400000d'), tr)
self.assertIn('8103012100', tr) self.assertIn('8103012100', tr)
self.assertIn('82028381', tr) self.assertIn('02028381', tr)
self.assertIn('83021200', tr) self.assertIn('83021200', tr)
def test_no_pending_returns_400(self): def test_no_pending_returns_400(self):
+44 -28
View File
@@ -802,7 +802,7 @@ class TestProactiveDecode(unittest.TestCase):
def test_parse_proactive_header_plain_tags(self): def test_parse_proactive_header_plain_tags(self):
import pysim_simple_server.server as srv import pysim_simple_server.server as srv
raw = bytes.fromhex('d00c010301270102028182240103') raw = bytes.fromhex('d00c010301270102028182240103')
self.assertEqual(srv._parse_proactive_header(raw), (1, 0x27, 0x81, 0x82, 0x01)) self.assertEqual(srv._parse_proactive_header(raw), (1, 0x27, 0x81, 0x82, 0x01, 0x01))
def test_default_handler_logs_timer_management(self): def test_default_handler_logs_timer_management(self):
# pySim's auto-handler path: the parsed command object (not the empty # pySim's auto-handler path: the parsed command object (not the empty
@@ -861,7 +861,23 @@ class TestProactiveDecode(unittest.TestCase):
# TS 102 223 6.8.0: command details, device identities, Result, then # TS 102 223 6.8.0: command details, device identities, Result, then
# the command-specific data objects (real terminals send them last). # the command-specific data objects (real terminals send them last).
tr = _build_tr(None, 1, 0x26, 0x81, 0x82, 0x00) tr = _build_tr(None, 1, 0x26, 0x81, 0x82, 0x00)
self.assertEqual(tr.hex(), '81030126008202828103010093055210011000') self.assertEqual(tr.hex(), '81030126000202828103010093055210011000')
def test_response_objects_use_the_base_cr0_forms(self):
# TS 102 223 6.8.0: the TR's response objects are Min = N ("should
# set the CR flag to comprehension not required"); 6.8.1: the
# command details echo the command's tag form (v3.25.2)
from pysim_simple_server import testscript as T
tr = _build_tr(None, 1, 0x03, 0x81, 0x82, None, cd_tag=0x01)
self.assertTrue(tr.hex().startswith('010301030002028281'), tr.hex())
self.assertIn('040201', tr.hex()) # POLL Duration base tag
self.assertTrue(_build_tr(None, 1, 0x26, 0x81, 0x82, 0x00, cd_tag=0x01)
.hex().startswith('010301260002028281'))
self.assertTrue(T.build_tr(1, 0x26, 0x82, 0x81, {}, cd_tag=0x01)
.hex().startswith('010301260002028281'))
# no tag known: the previous CR-set default keeps the old bytes
self.assertTrue(_build_tr(None, 1, 0x26, 0x81, 0x82, 0x00)
.hex().startswith('810301260002028281'))
def test_build_tr_echoes_the_command_qualifier(self): def test_build_tr_echoes_the_command_qualifier(self):
# TS 102 223 6.8.1: the TR's command details shall be identical to # TS 102 223 6.8.1: the TR's command details shall be identical to
@@ -881,7 +897,7 @@ class TestProactiveDecode(unittest.TestCase):
import pysim_simple_server.server as srv import pysim_simple_server.server as srv
tr = _build_tr(None, 1, 0x03, 0x81, 0x82, None) tr = _build_tr(None, 1, 0x03, 0x81, 0x82, None)
self.assertEqual(tr.hex(), self.assertEqual(tr.hex(),
'810301030082028281030100840201%02x' % srv._POLL_INTERVAL) '810301030002028281030100040201%02x' % srv._POLL_INTERVAL)
def test_build_tr_pli_datetime_from_host_clock_when_dict_empty(self): def test_build_tr_pli_datetime_from_host_clock_when_dict_empty(self):
# 8.39 coding: the same vectors as the PWA's eventDateTimeTlv test. # 8.39 coding: the same vectors as the PWA's eventDateTimeTlv test.
@@ -900,10 +916,10 @@ class TestProactiveDecode(unittest.TestCase):
srv._PLI_DATA[0x03] = saved srv._PLI_DATA[0x03] = saved
# the command details echo the qualifier (TS 102 223 6.8.1) # the command details echo the qualifier (TS 102 223 6.8.1)
self.assertEqual(tr_east.hex(), self.assertEqual(tr_east.hex(),
'810301260382028281030100260762900312550021') '810301260302028281030100260762900312550021')
# negative zone: the sign bit of the time zone byte (0x08) # negative zone: the sign bit of the time zone byte (0x08)
self.assertEqual(tr_west.hex(), self.assertEqual(tr_west.hex(),
'810301260382028281030100260762105080709009') '810301260302028281030100260762105080709009')
def test_build_tr_pli_dict_entry_overrides_host_clock(self): def test_build_tr_pli_dict_entry_overrides_host_clock(self):
import pysim_simple_server.server as srv import pysim_simple_server.server as srv
@@ -915,7 +931,7 @@ class TestProactiveDecode(unittest.TestCase):
datetime(2026, 9, 30, 21, 55, tzinfo=timezone.utc)) datetime(2026, 9, 30, 21, 55, tzinfo=timezone.utc))
finally: finally:
srv._PLI_DATA[0x03] = saved srv._PLI_DATA[0x03] = saved
self.assertEqual(tr.hex(), '810301260382028281030100130752f01000ff0001') self.assertEqual(tr.hex(), '810301260302028281030100130752f01000ff0001')
def test_build_tr_pli_empty_other_qualifier_sends_no_data(self): def test_build_tr_pli_empty_other_qualifier_sends_no_data(self):
# a qualifier with neither an override nor a built-in default (02 # a qualifier with neither an override nor a built-in default (02
@@ -928,7 +944,7 @@ class TestProactiveDecode(unittest.TestCase):
data = srv._pli_data_hex(0x02) data = srv._pli_data_hex(0x02)
finally: finally:
srv._PLI_DATA[0x02] = saved srv._PLI_DATA[0x02] = saved
self.assertEqual(tr.hex(), '810301260282028281030100') self.assertEqual(tr.hex(), '810301260202028281030100')
self.assertEqual(data, '') self.assertEqual(data, '')
def test_pli_defaults_cover_the_common_qualifiers(self): def test_pli_defaults_cover_the_common_qualifiers(self):
@@ -936,15 +952,15 @@ class TestProactiveDecode(unittest.TestCase):
# define (TS 102 223 8.x, tags per TS 101 220 V18.3.0 Table 7.23) # define (TS 102 223 8.x, tags per TS 101 220 V18.3.0 Table 7.23)
import pysim_simple_server.server as srv import pysim_simple_server.server as srv
d = srv.PLI_DEFAULTS d = srv.PLI_DEFAULTS
self.assertEqual(d[0x00], '930500F1100000') # 001/01 + LAC 0000 self.assertEqual(d[0x00], '130500F1100000') # 001/01 + LAC 0000
self.assertEqual(d[0x01], '940894104502237315F8') # IMEI, F pad self.assertEqual(d[0x01], '140894104502237315F8') # IMEI, F pad
self.assertEqual(d[0x04], 'AD02656E') # 'en' self.assertEqual(d[0x04], '2D02656E') # 'en'
self.assertEqual(d[0x05], 'AE020000') # Timing Advance self.assertEqual(d[0x05], '2E020000') # Timing Advance
self.assertEqual(d[0x06], 'BF0108') # E-UTRAN self.assertEqual(d[0x06], '3F0108') # E-UTRAN
self.assertEqual(d[0x08], 'E2089410450223731580') # IMEISV self.assertEqual(d[0x08], '62089410450223731580') # IMEISV
self.assertEqual(d[0x09], 'E50101') # automatic self.assertEqual(d[0x09], '650101') # automatic
self.assertEqual(d[0x0A], '630104') # battery full self.assertEqual(d[0x0A], '630104') # battery full
self.assertEqual(d[0x0E], 'BF020308') # UTRAN + E-UTRAN self.assertEqual(d[0x0E], '3F020308') # UTRAN + E-UTRAN
# the date/time stays dynamic; the 3GPP/3GPP2 qualifiers stay empty # the date/time stays dynamic; the 3GPP/3GPP2 qualifiers stay empty
self.assertNotIn(0x03, d) self.assertNotIn(0x03, d)
for q in (0x02, 0x07, 0x0B, 0x0C, 0x0D, 0x0F, 0x10, 0x11, 0x12, for q in (0x02, 0x07, 0x0B, 0x0C, 0x0D, 0x0F, 0x10, 0x11, 0x12,
@@ -1208,7 +1224,7 @@ class TestTimerManagement(unittest.TestCase):
def test_start_returns_result_only_and_arms_timer(self): def test_start_returns_result_only_and_arms_timer(self):
import pysim_simple_server.server as srv import pysim_simple_server.server as srv
tr = srv._handle_timer_command(1, 0x27, 0x00, self.START, 0x81, 0x82) tr = srv._handle_timer_command(1, 0x27, 0x00, self.START, 0x81, 0x82)
self.assertEqual(tr.hex(), '810301270082028281030100') self.assertEqual(tr.hex(), '810301270002028281030100')
remaining = srv._timer_remaining(1) remaining = srv._timer_remaining(1)
self.assertTrue(55 <= remaining <= 60, remaining) self.assertTrue(55 <= remaining <= 60, remaining)
@@ -1216,20 +1232,20 @@ class TestTimerManagement(unittest.TestCase):
import pysim_simple_server.server as srv import pysim_simple_server.server as srv
srv._handle_timer_command(1, 0x27, 0x00, self.START, 0x81, 0x82) srv._handle_timer_command(1, 0x27, 0x00, self.START, 0x81, 0x82)
tr = srv._handle_timer_command(1, 0x27, 0x02, self.START, 0x81, 0x82) tr = srv._handle_timer_command(1, 0x27, 0x02, self.START, 0x81, 0x82)
self.assertEqual(tr.hex(), '810301270282028281a40101a503001000030100') self.assertEqual(tr.hex(), '8103012702020282812401012503001000030100')
def test_deactivate_stops_and_reports_value(self): def test_deactivate_stops_and_reports_value(self):
import pysim_simple_server.server as srv import pysim_simple_server.server as srv
srv._handle_timer_command(1, 0x27, 0x00, self.START, 0x81, 0x82) srv._handle_timer_command(1, 0x27, 0x00, self.START, 0x81, 0x82)
tr = srv._handle_timer_command(1, 0x27, 0x01, self.START, 0x81, 0x82) tr = srv._handle_timer_command(1, 0x27, 0x01, self.START, 0x81, 0x82)
self.assertTrue(tr.hex().startswith('8103012701'), tr.hex()) self.assertTrue(tr.hex().startswith('8103012701'), tr.hex())
self.assertIn('a40101a503001000', tr.hex()) self.assertIn('2401012503001000', tr.hex())
self.assertIsNone(srv._timer_remaining(1)) self.assertIsNone(srv._timer_remaining(1))
def test_get_on_stopped_timer_is_contradiction(self): def test_get_on_stopped_timer_is_contradiction(self):
import pysim_simple_server.server as srv import pysim_simple_server.server as srv
tr = srv._handle_timer_command(1, 0x27, 0x02, self.START, 0x81, 0x82) tr = srv._handle_timer_command(1, 0x27, 0x02, self.START, 0x81, 0x82)
self.assertEqual(tr.hex(), '810301270282028281030124') self.assertEqual(tr.hex(), '810301270202028281030124')
def test_timer_expiration_envelope(self): def test_timer_expiration_envelope(self):
import pysim_simple_server.server as srv import pysim_simple_server.server as srv
@@ -2130,7 +2146,7 @@ class ProactiveLongTlvTests(unittest.TestCase):
TPDU + bytes(range(len(TPDU), 150))) TPDU + bytes(range(len(TPDU), 150)))
def test_header_parses_with_a_long_form_tlv(self): def test_header_parses_with_a_long_form_tlv(self):
self.assertEqual(_parse_proactive_header(self.RAW), (1, 0x13, 0x81, 0x83, 0)) self.assertEqual(_parse_proactive_header(self.RAW), (1, 0x13, 0x81, 0x83, 0, 0x81))
# the long TLV is found as one TLV, not mis-walked into its value # the long TLV is found as one TLV, not mis-walked into its value
self.assertEqual(sorted(_tlv_map(self.RAW[3:]).keys()), [0x81, 0x82, 0x86, 0x8B]) self.assertEqual(sorted(_tlv_map(self.RAW[3:]).keys()), [0x81, 0x82, 0x86, 0x8B])
self.assertEqual(len(_find_sms_tpdu(self.RAW)), 150 * 2) self.assertEqual(len(_find_sms_tpdu(self.RAW)), 150 * 2)
@@ -2138,7 +2154,7 @@ class ProactiveLongTlvTests(unittest.TestCase):
def test_a_truncated_tlv_does_not_raise(self): def test_a_truncated_tlv_does_not_raise(self):
# a declared length running past the data ends the walk # a declared length running past the data ends the walk
raw = bytes.fromhex('d004810381') raw = bytes.fromhex('d004810381')
self.assertEqual(_parse_proactive_header(raw), (1, 0, 0x83, 0x81, None)) self.assertEqual(_parse_proactive_header(raw), (1, 0, 0x83, 0x81, None, None))
self.assertEqual(_tlv_map(bytes.fromhex('810381')), {}) self.assertEqual(_tlv_map(bytes.fromhex('810381')), {})
# display text with a long-form text TLV (144 bytes -> 8D 81 90) # display text with a long-form text TLV (144 bytes -> 8D 81 90)
text = bytes([0x8D, 0x81, 0x90]) + b'\x00' * 0x90 text = bytes([0x8D, 0x81, 0x90]) + b'\x00' * 0x90
@@ -2385,19 +2401,19 @@ class StkInputTests(unittest.TestCase):
def test_input_text_tlv_codings(self): def test_input_text_tlv_codings(self):
pd = {'type': 'get_input', 'digits_only': True, 'hidden': True} pd = {'type': 'get_input', 'digits_only': True, 'hidden': True}
# digits: SMS default alphabet unpacked (DCS 04), one byte per char # digits: SMS default alphabet unpacked (DCS 04), one byte per char
self.assertEqual(_input_text_tlv(pd, '1234').hex().upper(), '8D050431323334') self.assertEqual(_input_text_tlv(pd, '1234').hex().upper(), '0D050431323334')
# empty input: the null text string (6.8.5) # empty input: the null text string (6.8.5)
self.assertEqual(_input_text_tlv(pd, '').hex().upper(), '8D00') self.assertEqual(_input_text_tlv(pd, '').hex().upper(), '0D00')
# UCS2 request: DCS 08 # UCS2 request: DCS 08
self.assertEqual(_input_text_tlv(dict(pd, ucs2=True, hidden=False), '12').hex().upper(), self.assertEqual(_input_text_tlv(dict(pd, ucs2=True, hidden=False), '12').hex().upper(),
'8D050800310032') '0D050800310032')
# packed request: DCS 00 + TS 23.038 septet packing # packed request: DCS 00 + TS 23.038 septet packing
self.assertEqual(_input_text_tlv(dict(pd, packed=True, hidden=False), '1234').hex().upper(), self.assertEqual(_input_text_tlv(dict(pd, packed=True, hidden=False), '1234').hex().upper(),
'8D050031D98C06') '0D050031D98C06')
# GET INKEY Yes/No: value 01 positive / 00 negative (6.8.5) # GET INKEY Yes/No: value 01 positive / 00 negative (6.8.5)
yesno = {'type': 'get_inkey', 'yes_no': True} yesno = {'type': 'get_inkey', 'yes_no': True}
self.assertEqual(_input_text_tlv(yesno, '01').hex().upper(), '8D020401') self.assertEqual(_input_text_tlv(yesno, '01').hex().upper(), '0D020401')
self.assertEqual(_input_text_tlv(yesno, '00').hex().upper(), '8D020400') self.assertEqual(_input_text_tlv(yesno, '00').hex().upper(), '0D020400')
def test_pack_gsm7_matches_the_sms_packing(self): def test_pack_gsm7_matches_the_sms_packing(self):
self.assertEqual(_pack_gsm7(b'1234').hex().upper(), '31D98C06') self.assertEqual(_pack_gsm7(b'1234').hex().upper(), '31D98C06')
@@ -2452,7 +2468,7 @@ class StkInputTests(unittest.TestCase):
resp, code = _menu_send_response(server, 'ok', None, '1234') resp, code = _menu_send_response(server, 'ok', None, '1234')
self.assertEqual(code, 200, resp) self.assertEqual(code, 200, resp)
apdu = server.scc._tp.sent[0].upper() apdu = server.scc._tp.sent[0].upper()
self.assertIn('8D050431323334', apdu) # the entered digits self.assertIn('0D050431323334', apdu) # the entered digits
self.assertIn('83020000', apdu) # successful result self.assertIn('83020000', apdu) # successful result
self.assertIsNone(server.stk_pending) self.assertIsNone(server.stk_pending)
# a rejected value sends nothing and keeps the command pending # a rejected value sends nothing and keeps the command pending
+9 -2
View File
@@ -94,8 +94,15 @@ class TagVariantMatrixTests(unittest.TestCase):
with self.subTest(label=label): with self.subTest(label=label):
plain = parser(raw) plain = parser(raw)
flipped = parser(flip_tags(raw)) flipped = parser(flip_tags(raw))
self.assertEqual(plain, flipped, if isinstance(plain, tuple) and len(plain) == 6:
'%s differs between the tag styles' % label) # the header parser also returns the command-details tag,
# which follows the style (the TR echoes it, v3.25.2) -
# the parsed values stay identical
self.assertEqual(plain[:5], flipped[:5],
'%s differs between the tag styles' % label)
else:
self.assertEqual(plain, flipped,
'%s differs between the tag styles' % label)
self.assertNotEqual(raw, flip_tags(raw), 'fixture has no TLVs?') self.assertNotEqual(raw, flip_tags(raw), 'fixture has no TLVs?')
def test_response_scripting_accepts_both_rapdu_tags(self): def test_response_scripting_accepts_both_rapdu_tags(self):
+12 -12
View File
@@ -163,21 +163,21 @@ class TestValidation(unittest.TestCase):
# then the command-specific objects (here Text), then the raw extras. # then the command-specific objects (here Text), then the raw extras.
tr = T.build_tr(4, 0x23, 0x81, 0x82, respond) tr = T.build_tr(4, 0x23, 0x81, 0x82, respond)
self.assertEqual(tr.hex().upper(), self.assertEqual(tr.hex().upper(),
'8103042300' + '82028182' + '83020000' '8103042300' + '02028182' + '83020000'
+ '8D0600' + 'E8329BFD06' + 'AA01BB') + '0D0600' + 'E8329BFD06' + 'AA01BB')
# 8 bits: one octet per character # 8 bits: one octet per character
tr = T.build_tr(4, 0x23, 0x81, 0x82, {'result': 0x00, 'text': 'hello', 'dcs': 0x04}) tr = T.build_tr(4, 0x23, 0x81, 0x82, {'result': 0x00, 'text': 'hello', 'dcs': 0x04})
self.assertEqual(tr.hex().upper(), self.assertEqual(tr.hex().upper(),
'8103042300' + '82028182' + '83020000' + '8D0604' + '68656C6C6F') '8103042300' + '02028182' + '83020000' + '0D0604' + '68656C6C6F')
# UCS2 (UTF-16-BE): 'A' U+0041, 'Ж' U+0416 # UCS2 (UTF-16-BE): 'A' U+0041, 'Ж' U+0416
tr = T.build_tr(4, 0x23, 0x81, 0x82, tr = T.build_tr(4, 0x23, 0x81, 0x82,
{'result': 0x00, 'text': 'A\u0416', 'dcs': 0x08}) {'result': 0x00, 'text': 'A\u0416', 'dcs': 0x08})
self.assertEqual(tr.hex().upper(), self.assertEqual(tr.hex().upper(),
'8103042300' + '82028182' + '83020000' + '8D0508' + '00410416') '8103042300' + '02028182' + '83020000' + '0D0508' + '00410416')
# an empty answer is the null text string (Length 00, no DCS) # an empty answer is the null text string (Length 00, no DCS)
tr = T.build_tr(4, 0x23, 0x81, 0x82, {'result': 0x00, 'text': ''}) tr = T.build_tr(4, 0x23, 0x81, 0x82, {'result': 0x00, 'text': ''})
self.assertEqual(tr.hex().upper(), self.assertEqual(tr.hex().upper(),
'8103042300' + '82028182' + '83020000' + '8D00') '8103042300' + '02028182' + '83020000' + '0D00')
def test_a_text_the_dcs_cannot_code_is_refused_at_load(self): 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 # v3.22.8: the load-time check uses the same coding the TERMINAL
@@ -427,15 +427,15 @@ class TestBuildTr(unittest.TestCase):
# TS 102 223 6.8.0: the item identifier (F) follows the Result (C) # 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}) tr = T.build_tr(3, 0x24, 0x82, 0x81, {'result': 0x00, 'item_id': 7})
self.assertEqual(tr.hex().upper(), self.assertEqual(tr.hex().upper(),
'81030324008202828183020000900107') '81030324000202828183020000100107')
def test_cancel_response_has_no_item_identifier(self): def test_cancel_response_has_no_item_identifier(self):
tr = T.build_tr(3, 0x24, 0x82, 0x81, {'result': 0x10, 'item_id': 7}) tr = T.build_tr(3, 0x24, 0x82, 0x81, {'result': 0x10, 'item_id': 7})
self.assertEqual(tr.hex().upper(), '81030324008202828183021000') self.assertEqual(tr.hex().upper(), '81030324000202828183021000')
def test_display_text_result_only(self): def test_display_text_result_only(self):
tr = T.build_tr(2, 0x21, 0x82, 0x81, {'result': 0x00}) tr = T.build_tr(2, 0x21, 0x82, 0x81, {'result': 0x00})
self.assertEqual(tr.hex().upper(), '81030221008202828183020000') self.assertEqual(tr.hex().upper(), '81030221000202828183020000')
def test_the_terminal_answer_objects_follow_the_result(self): def test_the_terminal_answer_objects_follow_the_result(self):
# The runner passes the terminal's own objects (the PLI data, the # The runner passes the terminal's own objects (the PLI data, the
@@ -444,18 +444,18 @@ class TestBuildTr(unittest.TestCase):
tr = T.build_tr(1, 0x26, 0x82, 0x81, {'result': 0x00}, tr = T.build_tr(1, 0x26, 0x82, 0x81, {'result': 0x00},
answer_tlvs='2607AABBCCDDEEFF') answer_tlvs='2607AABBCCDDEEFF')
self.assertEqual(tr.hex().upper(), self.assertEqual(tr.hex().upper(),
'810301260082028281830200002607AABBCCDDEEFF') '810301260002028281830200002607AABBCCDDEEFF')
# with a text object the answer object keeps its place # with a text object the answer object keeps its place
tr = T.build_tr(1, 0x23, 0x82, 0x81, tr = T.build_tr(1, 0x23, 0x82, 0x81,
{'result': 0x00, 'text': 'hi', 'dcs': 0x04}, {'result': 0x00, 'text': 'hi', 'dcs': 0x04},
answer_tlvs='8402011E') answer_tlvs='8402011E')
self.assertEqual(tr.hex().upper(), self.assertEqual(tr.hex().upper(),
'810301230082028281830200008402011E8D03046869') '810301230002028281830200008402011E0D03046869')
# and the script's raw TLVs stay last # and the script's raw TLVs stay last
tr = T.build_tr(1, 0x26, 0x82, 0x81, tr = T.build_tr(1, 0x26, 0x82, 0x81,
{'result': 0x00, 'raw': '9B0100'}, answer_tlvs='2607AA') {'result': 0x00, 'raw': '9B0100'}, answer_tlvs='2607AA')
self.assertEqual(tr.hex().upper(), self.assertEqual(tr.hex().upper(),
'810301260082028281830200002607AA9B0100') '810301260002028281830200002607AA9B0100')
# ─── runner with a fake card ───────────────────────────────────────────── # ─── runner with a fake card ─────────────────────────────────────────────
@@ -548,7 +548,7 @@ class TestRunnerDialogue(RunnerTestCase):
self.assertEqual(run['status'], 'ok', run['steps']) self.assertEqual(run['status'], 'ok', run['steps'])
self.assertEqual([s['status'] for s in run['steps']], ['ok', 'ok', 'ok']) self.assertEqual([s['status'] for s in run['steps']], ['ok', 'ok', 'ok'])
# the SELECT ITEM TR carries the item identifier # the SELECT ITEM TR carries the item identifier
self.assertIn('900101', run['steps'][1]['sent']) self.assertIn('100101', run['steps'][1]['sent'])
# the DISPLAY TEXT check decoded the text # the DISPLAY TEXT check decoded the text
text_check = [c for c in run['steps'][2]['checks'] if c['label'] == 'Text'][0] text_check = [c for c in run['steps'][2]['checks'] if c['label'] == 'Text'][0]
self.assertEqual(text_check['actual'], 'hello') self.assertEqual(text_check['actual'], 'hello')