fix: recover the SMS-SUBMIT listings in RAM Explore (v3.6.4)

Explore returned partial data (no ELF/module entries, the installed package
missing) and then failed with cntr_low.  Two causes, both in the
actual-response SMS-SUBMIT path the card uses for big listings:

- `_find_sms_tpdu` read TLV lengths as a single byte; the FETCH carries
  `8B 81 97 ...` (BER long form) for the big pages, so it returned a
  corrupted, truncated TPDU.
- `_calc_ud_offset` treated the SMS-SUBMIT relative validity period (VPF=10)
  as 7 bytes instead of 1, shifting the UD offset: `_parse_sms_concat` then
  read a bogus UDH and reported no concatenation, so the segments never
  assembled and the PoR/data was dropped ("RAM RESPONSE-PACKET: empty").
  The step was marked failed, the counter did not advance, the same counter
  was retried and the card answered `cntr_low`, which also blocked P1=10
  (modules) where the installed package appears.

- the captured segments now always store the assembled UD (`submit_ud_hex`);
  `_sms_submit_por()` rebuilds the DELIVER-style packet (`02 71 00` + UD) for
  `_decode_por`.  Verified against the live capture: por_ok, remote SW 6310,
  438 hex chars of listing data (the exact bytes of the P1=20 page).
- `spPorAccepted` counts `actual_response_sms_submit` (0x0B) as accepted, so
  the counter advances when the data follows via SMS-SUBMIT.

Explore queries follow GP Card Spec v2.3.1 11.4.2.2: compact listings
(P2.b2=0) with the chained GET RESPONSE (`ramGetStatusApdu`, paging P2=00 ->
P2=01), the malformed P2=02 attempt is gone, and the ISD-only query (P1=80)
never pages with next-occurrence (the card shall reject it).

Tests: Python SmsSubmitCaptureTest with the live FETCH bytes (TPDU length,
UD offset, concat reassembly, 027100+UD decode); ram.test.js checks the APDU
builder and the missing P2=02 attempt; cards_counter covers 0x0B.

608 frontend / 492 Python green; version 3.6.4; sw simple-v277.
This commit is contained in:
2026-09-28 00:09:34 +03:00
parent 39c82f26f3
commit 9e85f522f6
8 changed files with 224 additions and 72 deletions
+8
View File
@@ -294,6 +294,14 @@ segments are sent in order. A packet that would need more than 5 segments
is refused (the card's concatenation buffer is the limit). With `sp` a is refused (the card's concatenation buffer is the limit). With `sp` a
pre-built packet is delivered the same way. pre-built packet is delivered the same way.
The card may answer with PoR status `actual_response_sms_submit` (`0x0B`):
the real response (a big GET STATUS listing, for example) then arrives as one
or more proactive SEND SHORT MESSAGE commands. The server captures those
SMS-SUBMIT TPDUs, reassembles the concatenated segments and returns the
decoded response in `por` (as if it had arrived in the ENVELOPE), so callers
see a normal `por.response_status == "por_ok"` with the remote status word
and response data.
**Request body:** **Request body:**
```json ```json
{ {
+58 -57
View File
@@ -1665,7 +1665,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.6.3'; const SIMPLE_VERSION = '3.6.4';
document.getElementById('app-version').textContent = 'v' + SIMPLE_VERSION; document.getElementById('app-version').textContent = 'v' + SIMPLE_VERSION;
// ===== Tab switching ===== // ===== Tab switching =====
@@ -8730,66 +8730,54 @@ async function ramExplore(sp) {
const isd = [], apps = [], elfs = [], modules = []; const isd = [], apps = [], elfs = [], modules = [];
async function paginate(p1, collector, parser, label) { async function paginate(p1, collector, parser, label) {
// Chain GET STATUS + GET RESPONSE into a single SCP80 payload. // Compact listing format (GP 11.4.2.2: P2.b2=0) with the chained GET
// The card's SCP80 layer executes both: GET STATUS returns 61XX, // RESPONSE - see ramGetStatusApdu(). The card's SCP80 layer runs both
// then GET RESPONSE fetches the data — the PoR captures the final // commands, so the PoR carries the listing. ELF queries (P1=20/10) use
// result (9000 + response data) without the frontend handling 61XX. // SPI2=0x21 (PoR via SMS-SUBMIT) because the listing won't fit in the
// ELF queries (P1=20/10) use SPI2=0x21 (PoR via SMS-SUBMIT) because // ENVELOPE response.
// ELF data won't fit in the ENVELOPE response.
const isElf = (p1 === '20' || p1 === '10'); const isElf = (p1 === '20' || p1 === '10');
const spi2 = isElf ? '21' : '01'; const spi2 = isElf ? '21' : '01';
// Try TLV format (P2=02) first for structured data; fall back to raw let p2 = '00';
// (P2=00) if the card doesn't support TLV GET STATUS. let guard = 0;
for (const p2Init of ['02', '00']) { while (guard++ < 32) {
let p2 = p2Init; const apdu = ramGetStatusApdu(p1, p2);
let guard = 0; ramShowProgress(label + ' P1=' + p1 + ' P2=' + p2 + '...');
let tlvFailed = false; const res = await ramSendOta(apdu, Object.assign({}, sp, { cntr, spi2 }));
while (guard++ < 32) { // A packet the card accepted advances the counter even when the
// P2=02: no data field (card returns all tags). P2=00: Lc=02 TagList=4F00 (ignored by card). // page itself is incomplete: the card consumed it, and a retry with
const dataField = p2 === '02' ? '' : '024F0000'; // the same counter is rejected (cntr_low).
const apdu = '80F2' + p1 + p2 + dataField + 'C0000000'; if (res.success && spPorAccepted(res.por)) cntr = ramIncrementCntr(cntr);
ramShowProgress(label + ' P1=' + p1 + ' P2=' + p2 + '...'); if (!res.success || !res.por || res.por.response_status !== 'por_ok') {
const res = await ramSendOta(apdu, Object.assign({}, sp, { cntr, spi2 })); const errorMsg = res.por ? res.por.response_status : (res.error || t('no data'));
if (!res.success || !res.por || res.por.response_status !== 'por_ok') { errors.push(label + ': ' + errorMsg);
const errorMsg = res.por ? res.por.response_status : (res.error || t('no data')); break;
errors.push(label + ': ' + errorMsg); }
tlvFailed = true;
break;
}
// the card consumed the packet: advance for the next step
cntr = ramIncrementCntr(cntr);
const data = res.por.decoded ? res.por.decoded.last_response_data : ''; const data = res.por.decoded ? res.por.decoded.last_response_data : '';
const sw = (res.por.decoded ? res.por.decoded.last_status_word : '').toUpperCase(); const sw = (res.por.decoded ? res.por.decoded.last_status_word : '').toUpperCase();
// If first attempt with P2=02 gets an unsupported error, retry with P2=00. // Defensive: a remote 61XX means the data is still pending (the
if (guard === 1 && p2Init === '02' && (sw === '6A86' || sw === '6A88')) { // chained GET RESPONSE normally prevents this).
tlvFailed = true; if (sw && sw.startsWith('61')) {
break; if (data) {
const parsed61 = parser(data);
if (parsed61.length) collector.push(...parsed61);
} }
// Defensive: 61XX means more data available (shouldn't happen with if (p1 === '80') break; // the ISD is a single occurrence (11.4.2.2)
// chained GET RESPONSE, but handle it if the card responds this way).
if (sw && sw.startsWith('61')) {
if (data) {
const parsed61 = parser(data);
if (parsed61.length) collector.push(...parsed61);
}
p2 = '01';
continue;
}
if (sw === '6F00') { tlvFailed = true; break; }
if (!data) {
if (sw !== '9000') {
errors.push(label + ': ' + t('(no data)') + ' — SW ' + sw);
tlvFailed = true;
}
break;
}
const parsed = parser(data);
if (!parsed.length) break;
collector.push(...parsed);
if (sw === '9000') break;
p2 = '01'; p2 = '01';
continue;
} }
if (!tlvFailed) break; if (sw === '6F00') break;
if (!data) {
if (sw !== '9000') errors.push(label + ': ' + t('(no data)') + ' — SW ' + sw);
break;
}
const parsed = parser(data);
if (!parsed.length) break;
collector.push(...parsed);
if (sw === '9000') break;
// next occurrence (P2.b1=1) - never for the ISD-only query: the
// card shall reject it (GP 11.4.2.2)
if (p1 === '80') break;
p2 = '01';
} }
} }
@@ -9459,10 +9447,23 @@ function spRefreshFromPreset(selId) {
} }
// A send counts as accepted when there is no PoR to check (the SPI requests // A send counts as accepted when there is no PoR to check (the SPI requests
// none) or the PoR is por_ok; anything else (cntr_low, Por error) must leave // none) or the card reports success: por_ok, or actual_response_sms_submit
// the counter untouched. // (0x0B - the real response follows as an SMS-SUBMIT; the packet was still
// consumed and the counter must advance). Anything else (cntr_low, PoR
// error) leaves the counter untouched.
function spPorAccepted(por) { function spPorAccepted(por) {
return !por || por.response_status === 'por_ok'; return !por || por.response_status === 'por_ok' ||
por.response_status === 'actual_response_sms_submit';
}
// GET STATUS APDU for the compact listing format: P2.b2=0 (GP Card Spec
// v2.3.1 11.4.2.2) with the mandatory '4F00' search criterion (all
// occurrences) and a chained GET RESPONSE (00C0000000) in the same secured
// payload. The expanded TLV format (P2.b2=1, P2=02/03) must NOT carry a GET
// RESPONSE - it returns the E3 templates directly - so this builder is only
// for the compact path.
function ramGetStatusApdu(p1, p2) {
return '80F2' + p1 + p2 + '024F0000' + 'C0000000';
} }
function downloadJson(name, obj) { function downloadJson(name, obj) {
+1 -1
View File
@@ -1,4 +1,4 @@
const CACHE = 'simple-v276'; const CACHE = 'simple-v277';
const URLS = [ const URLS = [
'index.html', 'index.html',
'help.html', 'help.html',
+2
View File
@@ -84,6 +84,8 @@ test('cardsApplyFields fills the form without generating a packet', () => {
test('spPorAccepted treats a missing PoR and por_ok as accepted', () => { test('spPorAccepted treats a missing PoR and por_ok as accepted', () => {
assert.strictEqual(spPorAccepted(undefined), true); assert.strictEqual(spPorAccepted(undefined), true);
assert.strictEqual(spPorAccepted({ response_status: 'por_ok' }), true); assert.strictEqual(spPorAccepted({ response_status: 'por_ok' }), true);
// 0x0B: the real response follows as an SMS-SUBMIT - the packet was consumed
assert.strictEqual(spPorAccepted({ response_status: 'actual_response_sms_submit' }), true);
assert.strictEqual(spPorAccepted({ response_status: 'cntr_low' }), false); assert.strictEqual(spPorAccepted({ response_status: 'cntr_low' }), false);
assert.strictEqual(spPorAccepted({ response_status: 'rc_cc_ds_failed' }), false); assert.strictEqual(spPorAccepted({ response_status: 'rc_cc_ds_failed' }), false);
}); });
+15 -1
View File
@@ -22,7 +22,7 @@ function extractFunc(src, name) {
} }
// Extract chain builder functions and dependencies // Extract chain builder functions and dependencies
const FNS = ['berLenStr', 'buildApdu', 'escHtml', 'esc', 'chainInit', 'chainRamBuildRowHex', 'ramFmtLifecycle', 'ramFmtPrivileges', 'ramRenderExploreHtml', 'ramStepLine', const FNS = ['berLenStr', 'buildApdu', 'escHtml', 'esc', 'chainInit', 'chainRamBuildRowHex', 'ramFmtLifecycle', 'ramFmtPrivileges', 'ramRenderExploreHtml', 'ramStepLine', 'ramGetStatusApdu',
'ramCardIdxAfterRemove', 'ramClearResults', 'ramHideProgress', 'ramOpChanged', 'ramRender', 'ramApplyCard', 'ramExecute', 'ramCardIdxAfterRemove', 'ramClearResults', 'ramHideProgress', 'ramOpChanged', 'ramRender', 'ramApplyCard', 'ramExecute',
'jcAidNorm', 'jcAidName', 'jcAidSuffix', 'jcAidHtml']; 'jcAidNorm', 'jcAidName', 'jcAidSuffix', 'jcAidHtml'];
let code = ''; let code = '';
@@ -265,6 +265,20 @@ function fakeRamDocument(ids) {
return els; return els;
} }
test('ramGetStatusApdu builds the compact chain, never the expanded form', () => {
// GP 11.4.2.2: compact listings (P2.b2=0) carry the chained GET RESPONSE
assert.strictEqual(ramGetStatusApdu('80', '00'), '80F28000024F0000C0000000');
assert.strictEqual(ramGetStatusApdu('20', '01'), '80F22001024F0000C0000000');
// the expanded TLV form (P2=02/03) takes no GET RESPONSE and is not built:
// its GET STATUS bytes would be `80F2<P1>02 04 4F00 5C.. 00`
assert.ok(!ramGetStatusApdu('20', '01').startsWith('80F2200204'),
'expanded form must not be generated by this builder');
assert.ok(!html.includes("for (const p2Init of ['02', '00'])"),
'the P2=02 attempt must be gone');
// the ISD-only query is a single occurrence: no next-occurrence paging
assert.ok(/if \(p1 === '80'\) break;/.test(html), 'ISD next-occurrence guard missing');
});
test('ramStepLine shows the PoR verdict and the remote status word', () => { test('ramStepLine shows the PoR verdict and the remote status word', () => {
globalThis.t = s => s; globalThis.t = s => s;
globalThis.lookupSw = (a, b) => (a + b === '6700' ? 'Wrong length in Lc' : ''); globalThis.lookupSw = (a, b) => (a + b === '6700' ? 'Wrong length in Lc' : '');
+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.6.3" version = "3.6.4"
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.
+68 -11
View File
@@ -31,7 +31,7 @@ from osmocom.tlv import BER_TLV_IE
from osmocom.utils import rpad from osmocom.utils import rpad
VERSION = '3.6.3' VERSION = '3.6.4'
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
@@ -883,9 +883,12 @@ def _send_envelope(tpdu_hex, scc, sm_sc='12345678912', submit_handler=None,
if len(matching) >= total: if len(matching) >= total:
sorted_segs = sorted(matching, key=lambda s: s[2]) sorted_segs = sorted(matching, key=lambda s: s[2])
assembled = b''.join(bytes.fromhex(s[3]) for s in sorted_segs) assembled = b''.join(bytes.fromhex(s[3]) for s in sorted_segs)
submit_handler.submit_tpdu_hex = assembled.hex() submit_handler.submit_ud_hex = assembled.hex()
sys.stderr.write('SMS concat: assembled %d segments (ref=%s)\n' % (total, ref)) submit_handler.submit_tpdu_hex = submit_handler.submit_tpdu_hex or tpdu_hex
sys.stderr.write('SMS concat: assembled %d segments (ref=%s, %d B)\n' % (
total, ref, len(assembled)))
else: else:
submit_handler.submit_ud_hex = payload.hex()
submit_handler.submit_tpdu_hex = tpdu_hex submit_handler.submit_tpdu_hex = tpdu_hex
_handle_proactive_chain(scc, sw, _capture_sms_tpdu) _handle_proactive_chain(scc, sw, _capture_sms_tpdu)
data, sw = '', '9000' data, sw = '', '9000'
@@ -1168,6 +1171,23 @@ def _ram_step_result(step_name, last_sw, por, por_hex, bytes_, segments):
return step, error return step, error
def _sms_submit_por(submit_handler):
"""Response packet carried by an actual-response SMS-SUBMIT, in the
DELIVER-style form `_decode_por` expects.
The submit UD has no RPI UDH (the RPI is a UDH IE there) and starts at
RPL/RHL/TAR/CNTR/PCNTR/STS, so the `02 71 00` RPI UDH is prepended. Returns
'' when no submit response was captured."""
ud_hex = (getattr(submit_handler, 'submit_ud_hex', None) or '').strip()
if not ud_hex:
return ''
try:
bytes.fromhex(ud_hex)
except ValueError:
return ''
return '027100' + ud_hex
def _decode_por(spi1, spi2, kic, kid, cntr_hex, kic_key_hex, kid_key_hex, response_hex): def _decode_por(spi1, spi2, kic, kid, cntr_hex, kic_key_hex, kid_key_hex, response_hex):
from pySim.ota import OtaDialectSms, CompactRemoteResp from pySim.ota import OtaDialectSms, CompactRemoteResp
from osmocom.utils import h2b, b2h from osmocom.utils import h2b, b2h
@@ -1272,6 +1292,9 @@ class PoRSubmitHandler(ProactiveHandler):
def __init__(self): def __init__(self):
super().__init__() super().__init__()
self.submit_tpdu_hex = None self.submit_tpdu_hex = None
# Assembled UD data of the actual-response SMS-SUBMIT(s), UDH stripped:
# the DELIVER-style response packet is `02 71 00` + this data.
self.submit_ud_hex = None
self.sms_segments = [] # [(ref, total, num, payload_hex), ...] self.sms_segments = [] # [(ref, total, num, payload_hex), ...]
@@ -3685,6 +3708,19 @@ def _bip_flush_channel_events(scc):
_FLUSHING_CHANNEL_EVENTS = False _FLUSHING_CHANNEL_EVENTS = False
def _ber_len_at(raw, off):
"""Return (length, value_offset) of a BER length field at raw[off]."""
if off >= len(raw):
return 0, off
first = raw[off]
if first < 0x80:
return first, off + 1
n = first & 0x7F
if n < 1 or off + 1 + n > len(raw):
return 0, off + 1
return int.from_bytes(raw[off + 1:off + 1 + n], 'big'), off + 1 + n
def _skip_ber_len(raw, off): def _skip_ber_len(raw, off):
if off >= len(raw): if off >= len(raw):
return off return off
@@ -3852,13 +3888,22 @@ def _parse_proactive_header(raw):
def _find_sms_tpdu(raw): def _find_sms_tpdu(raw):
if raw[0] == 0xD0: """Extract the SMS TPDU (tag 0x8B) from a FETCH response.
FETCH responses may use BER long-form lengths (`8B 81 97 ...` for the big
listings), so the TLV length must be parsed, not read as one byte: a
single-byte read silently truncates the TPDU and the PoR/data is lost."""
if raw and raw[0] == 0xD0:
off = _skip_ber_len(raw, 1) off = _skip_ber_len(raw, 1)
while off < len(raw) - 1: while off < len(raw) - 1:
tag, tlen = raw[off], raw[off + 1] tag = raw[off]
val = raw[off + 2: off + 2 + tlen]; off += 2 + tlen tlen, val_off = _ber_len_at(raw, off + 1)
if tag == 0x8B and tlen >= 1: if tag == 0x8B and tlen >= 1:
return val.hex() return raw[val_off:val_off + tlen].hex()
nxt = val_off + tlen
if nxt <= off: # no forward progress: stop
break
off = nxt
return None return None
@@ -3875,8 +3920,10 @@ def _calc_ud_offset(tpdu):
da_len_digits = tpdu[2] da_len_digits = tpdu[2]
da_data_bytes = (da_len_digits + 1) // 2 da_data_bytes = (da_len_digits + 1) // 2
off = 1 + 1 + 1 + 1 + da_data_bytes + 1 + 1 off = 1 + 1 + 1 + 1 + da_data_bytes + 1 + 1
if vpf in (0x01, 0x02): # relative or absolute # TP-VP: none (0), enhanced (7), relative (1), absolute (7) - reading
off += 7 # the relative form as 7 bytes shifted the UD offset and made the
# concatenation UDH unparseable.
off += {0x00: 0, 0x01: 7, 0x02: 1, 0x03: 7}[vpf]
if off >= len(tpdu): if off >= len(tpdu):
return None return None
return off + 1 # skip UDL byte return off + 1 # skip UDL byte
@@ -5935,7 +5982,12 @@ class PysimHandler(BaseHTTPRequestHandler):
'bytes': result['bytes'], 'segments': result['segments']} 'bytes': result['bytes'], 'segments': result['segments']}
por_src = 'envelope' por_src = 'envelope'
por_hex = resp['response_data'] por_hex = resp['response_data']
if submit_handler and submit_handler.submit_tpdu_hex: submit_hex = _sms_submit_por(submit_handler)
if submit_hex:
por_hex = submit_hex
por_src = 'sms-submit'
elif submit_handler and submit_handler.submit_tpdu_hex:
# no assembled UD: fall back to a raw RPI packet
tpdu_b = bytes.fromhex(submit_handler.submit_tpdu_hex) tpdu_b = bytes.fromhex(submit_handler.submit_tpdu_hex)
idx = tpdu_b.find(b'\x02\x71\x00') idx = tpdu_b.find(b'\x02\x71\x00')
if idx >= 0: if idx >= 0:
@@ -6099,7 +6151,12 @@ class PysimHandler(BaseHTTPRequestHandler):
# Decode PoR # Decode PoR
por_src = 'envelope' por_src = 'envelope'
por_hex = last_data por_hex = last_data
if submit_handler and submit_handler.submit_tpdu_hex: submit_hex = _sms_submit_por(submit_handler)
if submit_hex:
por_hex = submit_hex
por_src = 'sms-submit'
elif submit_handler and submit_handler.submit_tpdu_hex:
# no assembled UD: fall back to a raw RPI packet
tpdu_b = bytes.fromhex(submit_handler.submit_tpdu_hex) tpdu_b = bytes.fromhex(submit_handler.submit_tpdu_hex)
idx = tpdu_b.find(b'\x02\x71\x00') idx = tpdu_b.find(b'\x02\x71\x00')
if idx >= 0: if idx >= 0:
+71 -1
View File
@@ -30,9 +30,13 @@ from pysim_simple_server.server import (
_ota_reference, _ota_reference,
_parse_select_item, _parse_select_item,
_parse_setup_menu_items, _parse_setup_menu_items,
_calc_ud_offset,
_find_sms_tpdu,
_parse_sms_concat,
_por_remote_sw, _por_remote_sw,
_ram_next_cntr, _ram_next_cntr,
_ram_remote_sw_ok, _ram_remote_sw_ok,
_sms_submit_por,
_ram_step_result, _ram_step_result,
_record_tr, _record_tr,
_send_secured_packet, _send_secured_packet,
@@ -1032,7 +1036,7 @@ class TestSmsReassembly(unittest.TestCase):
def test_single_segment_no_concat(self): def test_single_segment_no_concat(self):
"""Single segment without UDH → submit_tpdu_hex is set directly.""" """Single segment without UDH → submit_tpdu_hex is set directly."""
from pysim_simple_server.server import PoRSubmitHandler, _find_sms_tpdu, _parse_sms_concat from pysim_simple_server.server import PoRSubmitHandler
handler = PoRSubmitHandler() handler = PoRSubmitHandler()
# Build a simple D0 with tag 8B containing an SMS-SUBMIT without UDH # Build a simple D0 with tag 8B containing an SMS-SUBMIT without UDH
sms_tpdu = bytes.fromhex('040005902143F50004' # SMS-SUBMIT header sms_tpdu = bytes.fromhex('040005902143F50004' # SMS-SUBMIT header
@@ -1311,3 +1315,69 @@ class RamPorStepTest(unittest.TestCase):
'responses': [{'status_word': '9000'}, {'status_word': '6A82'}]} 'responses': [{'status_word': '9000'}, {'status_word': '6A82'}]}
self.assertEqual(_por_remote_sw(por), '6A82') self.assertEqual(_por_remote_sw(por), '6A82')
class SmsSubmitCaptureTest(unittest.TestCase):
"""The actual-response SMS-SUBMIT path: the card answers the ENVELOPE with
PoR status 0x0B and delivers the listing in SMS-SUBMIT TPDUs (SEND SHORT
MESSAGE proactive commands). The FETCH bytes below are from a live RAM
explore (the two 274-byte segments of the ELF listing)."""
FETCH_1 = ('d081ae81030113008202818305000607812143658719f28b8197510005812143f57ff6'
'058c070003130201710000e90a000000000000030600000263100bd276000005aaffcafe'
'0001010007a000000151535001000bd276000005aaffcafe001001000bd276000005aaff'
'cafe0304010010d2760001180002ff491ff3890000010101000bd276000005aaffcafe00'
'02010007a0000000620001010007a0000000620002010007a0000000620101010006')
FETCH_2 = ('d0818e81030113008202818305000607812143658719f28b78510005812143f57ff6'
'056d050003130202a00000015100010007a0000000620102010007a000000062020101'
'0008a000000062020801010009a00000006202080101010010a0000000090003ffffff'
'ff8910710001010010a0000000090003ffffffff891071000201000bd276000005aaff'
'cafe00030100')
def test_find_sms_tpdu_reads_ber_long_form_lengths(self):
tpdu = bytes.fromhex(_find_sms_tpdu(bytes.fromhex(self.FETCH_1)))
self.assertEqual(len(tpdu), 0x97) # 8B 81 97 <151 bytes>
self.assertEqual(tpdu[:6].hex(), '510005812143'.lower())
tpdu2 = bytes.fromhex(_find_sms_tpdu(bytes.fromhex(self.FETCH_2)))
self.assertEqual(len(tpdu2), 0x78) # 8B 78 <120 bytes>
def test_sms_submit_ud_offset_with_relative_validity(self):
# VPF=10 (relative) = 1 byte, not 7: 0x51 & 0x18 = 0x10 -> relative
tpdu = bytes.fromhex(_find_sms_tpdu(bytes.fromhex(self.FETCH_1)))
self.assertEqual(_calc_ud_offset(tpdu), 11)
# synthetic absolute/enhanced forms still take 7 bytes
abs_tpdu = bytes.fromhex('5900048111227ff6' + '00' * 7 + '00' + '00' * 8)
self.assertEqual(_calc_ud_offset(abs_tpdu), 16)
def test_parse_sms_concat_segments(self):
ref, total, num, payload = _parse_sms_concat(
bytes.fromhex(_find_sms_tpdu(bytes.fromhex(self.FETCH_1))))
self.assertEqual((ref, total, num), (0x13, 2, 1))
self.assertEqual(len(payload), 132)
ref2, total2, num2, payload2 = _parse_sms_concat(
bytes.fromhex(_find_sms_tpdu(bytes.fromhex(self.FETCH_2))))
self.assertEqual((ref2, total2, num2), (0x13, 2, 2))
self.assertEqual(len(payload2), 103)
def test_sms_submit_por_decodes_to_the_listing(self):
class Handler:
submit_ud_hex = None
handler = Handler()
segs = {}
for hx in (self.FETCH_1, self.FETCH_2):
_, tot, num, payload = _parse_sms_concat(bytes.fromhex(_find_sms_tpdu(bytes.fromhex(hx))))
segs[num] = payload
handler.submit_ud_hex = b''.join(segs[k] for k in sorted(segs)).hex()
por_hex = _sms_submit_por(handler)
self.assertTrue(por_hex.startswith('027100'))
por = _decode_por('15', '21', '25', '25', '0000000306',
'00' * 16, '00' * 16, por_hex)
self.assertEqual(por['response_status'], 'por_ok')
self.assertEqual(por['decoded']['last_status_word'], '6310')
data = por['decoded']['last_response_data']
self.assertEqual(len(data), 438)
self.assertTrue(data.lower().startswith('0bd2760000'), data[:20])
# no submit response captured -> empty string, never a bogus packet
handler.submit_ud_hex = None
self.assertEqual(_sms_submit_por(handler), '')