catarrh
20fbfd99a5
SCP81: - New listener mode "passthru": no local listener - the card's BIP channels connect straight to a configured external platform (host/port required), which terminates TLS and runs the dialog. The status API reports mode/target (_SCP81_MODE/_SCP81_TARGET) and clears them on stop. PWA mode select, hint, Start validation; TLS PSK stays the default. Proactive command decoding (Phone tab log): - SEND SHORT MESSAGE (0x13) is now decoded: alpha, address (TON/NPI + number), 3GPP-SMS TPDU (type, TP-MR, TP-DA, TP-PID 0x7F flagged as SIM data download, TP-DCS, TP-VP, TP-UDL), UDH concatenation IEs, and the TP-UD as text (GSM-7 with a septet unpacker, UCS2, 8-bit) or as a TS 31.115 secured packet for PID 0x7F; malformed TPDUs fall back to the raw hex line. - PROVIDE LOCAL INFORMATION qualifier names completed per TS 102 223 V18.3.0: ESN (07), MEID (0B), Supported RATs (1A); 05 relabelled "Reserved for GSM (Timing Advance)". Fixed in the server dict and both frontend tables. Header: - Compact ADM badge next to the card indicator: "ADM ✓" green when pySim's adm_verified is set, "ADM ✗" red otherwise, hidden without a card session or when the server is down; updated ahead of the card state-key early return so it never disturbs the connect/reset flow. File manager: - Sort pills (FID/Name), Probe all files button and progress line are pinned above the tree instead of scrolling with it; the tree box cap grows from 420px to 65vh. SW cache otaman-v165; help EN/RU updated; tests 234 Python + 361 frontend.
1031 lines
46 KiB
Python
1031 lines
46 KiB
Python
#!/usr/bin/env python3
|
|
"""Unit tests for the OTA helper functions in pysim_otaman_server.server.
|
|
|
|
Reference vectors are key-free: synthetic dummy keys plus the already-public
|
|
sysmocom sample-key vectors that ship in pySim's own tests/unittests/test_ota.py.
|
|
No live/sample card keys and no ICCIDs appear here.
|
|
"""
|
|
|
|
import sys
|
|
import types
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
# pySim checkout is a sibling of this repo; put it on sys.path so the server
|
|
# module (which imports pySim at module level) can be exercised against it.
|
|
PROJECTS = Path(__file__).resolve().parents[2]
|
|
PY_SIM = PROJECTS / 'pysim'
|
|
if str(PY_SIM) not in sys.path:
|
|
sys.path.insert(0, str(PY_SIM))
|
|
|
|
from pysim_otaman_server.server import (
|
|
_build_sms_tpdu,
|
|
_build_tr,
|
|
_decode_cmd,
|
|
_decode_por,
|
|
_decode_tr,
|
|
_log_proactive,
|
|
_max_load_block_size,
|
|
_ota_reference,
|
|
_record_tr,
|
|
_spi_from_bytes,
|
|
_tr_data_only,
|
|
)
|
|
|
|
# Synthetic dummy key material (no real card keys).
|
|
K = '00112233445566778899AABBCCDDEEFF'
|
|
# Public sysmocom sample keys from pySim tests/unittests/test_ota.py.
|
|
KIC3 = 'C21DD66ACAC13CB3BC8B331B24AFB57B'
|
|
KID3 = '12110C78E678C25408233076AA033615'
|
|
# Public synthetic AES keys from pySim tests/unittests/test_ota.py.
|
|
KIC_AES = '200102030405060708090a0b0c0d0e0f'
|
|
KID_AES = '201102030405060708090a0b0c0d0e0f'
|
|
|
|
APDU = '00a40000023f00'
|
|
|
|
# (spi1, spi2) -> expected secured packet, generated with _ota_reference
|
|
# against pySim's OtaDialectSms.encode_cmd and cross-checked with the JS genSp().
|
|
REFERENCE_VECTORS = {
|
|
('06', '09'):
|
|
'00201506091515b00000c08f58c38860acb3a362fffe670ad13759a2a6b4c1a91116',
|
|
('16', '01'):
|
|
'00201516011515b00000e42573469e68a8462a57a505b0e2b1c09c1928c7a182311f',
|
|
('02', '09'):
|
|
'001d1502091515b0000000000000010085a8ca1a9828b0bb00a40000023f00',
|
|
}
|
|
|
|
# AES-128 reference vectors (public synthetic keys from pySim test_ota.py).
|
|
AES_APDU = '00a40004023f00'
|
|
AES_REFERENCE_VECTORS = {
|
|
('16', '19'):
|
|
'00281516192222b000115a47655527e96e832f1a5c698655715d4331454a0d83952c0ed35245706976b1',
|
|
('12', '09'):
|
|
'001d1512092222b0001100000000110029826122c7a0b79500a40004023f00',
|
|
('1e', '19'):
|
|
'0028151e192222b0001118b202ee47a3203e7370861c383b4142e704157b36e5c0eb4bb33eb6036cbaf8',
|
|
}
|
|
|
|
|
|
class TestSpiFromBytes(unittest.TestCase):
|
|
def test_06_09_ciphered_cc(self):
|
|
spi = _spi_from_bytes(0x06, 0x09)
|
|
self.assertEqual(spi, {
|
|
'counter': 'no_counter',
|
|
'ciphering': True,
|
|
'rc_cc_ds': 'cc',
|
|
'por_in_submit': False,
|
|
'por_shall_be_ciphered': False,
|
|
'por_rc_cc_ds': 'cc',
|
|
'por': 'por_required',
|
|
})
|
|
|
|
def test_16_01_counter_must_be_higher(self):
|
|
spi = _spi_from_bytes(0x16, 0x01)
|
|
self.assertEqual(spi['counter'], 'counter_must_be_higher')
|
|
self.assertTrue(spi['ciphering'])
|
|
self.assertEqual(spi['rc_cc_ds'], 'cc')
|
|
self.assertEqual(spi['por_rc_cc_ds'], 'no_rc_cc_ds')
|
|
|
|
def test_02_09_unciphered_cc(self):
|
|
spi = _spi_from_bytes(0x02, 0x09)
|
|
self.assertFalse(spi['ciphering'])
|
|
self.assertEqual(spi['rc_cc_ds'], 'cc')
|
|
self.assertEqual(spi['por_rc_cc_ds'], 'cc')
|
|
|
|
def test_04_19_ciphered_no_cc(self):
|
|
spi = _spi_from_bytes(0x04, 0x19)
|
|
self.assertTrue(spi['ciphering'])
|
|
self.assertEqual(spi['rc_cc_ds'], 'no_rc_cc_ds')
|
|
self.assertTrue(spi['por_shall_be_ciphered'])
|
|
self.assertEqual(spi['por_rc_cc_ds'], 'cc')
|
|
|
|
|
|
class TestBuildSmsTpdu(unittest.TestCase):
|
|
CHUNK = '00201506091515b00000c08f58c38860acb3a362fffe670ad13759a2a6b4c1a91116'
|
|
SCTS = bytes.fromhex('24051215173000')
|
|
|
|
def _build(self, *args, **kwargs):
|
|
with mock.patch('pysim_otaman_server.server._encode_scts', return_value=self.SCTS):
|
|
return _build_sms_tpdu(*args, **kwargs)
|
|
|
|
def test_single_message_with_cpi(self):
|
|
self.assertEqual(
|
|
self._build(self.CHUNK, include_cpi=True),
|
|
'4005812143f57ff6240512151730002502700000201506091515b00000'
|
|
'c08f58c38860acb3a362fffe670ad13759a2a6b4c1a91116')
|
|
|
|
def test_single_message_without_cpi(self):
|
|
self.assertEqual(
|
|
self._build(self.CHUNK, include_cpi=False),
|
|
'0405812143f57ff6240512151730002200201506091515b00000'
|
|
'c08f58c38860acb3a362fffe670ad13759a2a6b4c1a91116')
|
|
|
|
def test_first_chunk_has_cpi(self):
|
|
self.assertEqual(
|
|
self._build(self.CHUNK, chunk_total=3, chunk_num=1, include_cpi=True),
|
|
'4405812143f57ff6240512151730002a070003010301700000201506091515b00000'
|
|
'c08f58c38860acb3a362fffe670ad13759a2a6b4c1a91116')
|
|
|
|
def test_later_chunk_concat_only(self):
|
|
self.assertEqual(
|
|
self._build(self.CHUNK, chunk_total=3, chunk_num=2, include_cpi=True),
|
|
'4405812143f57ff6240512151730002805000301030200201506091515b00000'
|
|
'c08f58c38860acb3a362fffe670ad13759a2a6b4c1a91116')
|
|
|
|
|
|
class TestOtaReference(unittest.TestCase):
|
|
def test_ciphered_spi_06_09(self):
|
|
out, _ = _ota_reference('06', '09', '15', '15', 'b00000', '0000000001', APDU, K, K)
|
|
self.assertEqual(out, REFERENCE_VECTORS[('06', '09')])
|
|
|
|
def test_ciphered_spi_16_01(self):
|
|
out, _ = _ota_reference('16', '01', '15', '15', 'b00000', '0000000001', APDU, K, K)
|
|
self.assertEqual(out, REFERENCE_VECTORS[('16', '01')])
|
|
|
|
def test_unciphered_spi_02_09(self):
|
|
out, _ = _ota_reference('02', '09', '15', '15', 'b00000', '0000000001', APDU, K, K)
|
|
self.assertEqual(out, REFERENCE_VECTORS[('02', '09')])
|
|
|
|
def test_unciphered_cpl_is_0x001d(self):
|
|
# Regression: CPL counts octets from the CHL octet to the last octet
|
|
# of the secured data (29 here), it must NOT be len(out)-2 (27/0x001b).
|
|
out, _ = _ota_reference('02', '09', '15', '15', 'b00000', '0000000001', APDU, K, K)
|
|
self.assertEqual(out[:4], '001d')
|
|
self.assertEqual(len(out) // 2, 31)
|
|
|
|
def test_sysmocom_reference_vector(self):
|
|
# Public vector from pySim tests/unittests/test_ota.py (test_cmd_3des_ciphered).
|
|
out, _ = _ota_reference('04', '19', '35', '35', 'b00000', '0000000000', APDU, KIC3, KID3)
|
|
self.assertEqual(out, '00180d04193535b00000e3ec80a849b554421276af3883927c20')
|
|
|
|
def test_returns_spi_dict(self):
|
|
_, spi = _ota_reference('16', '01', '15', '15', 'b00000', '0000000001', APDU, K, K)
|
|
self.assertEqual(spi['counter'], 'counter_must_be_higher')
|
|
self.assertTrue(spi['ciphering'])
|
|
|
|
def test_aes128_ciphered_cc(self):
|
|
out, _ = _ota_reference('16', '19', '22', '22', 'b00011', '0000000011', AES_APDU, KIC_AES, KID_AES)
|
|
self.assertEqual(out, AES_REFERENCE_VECTORS[('16', '19')])
|
|
|
|
def test_aes128_unciphered_cc(self):
|
|
out, _ = _ota_reference('12', '09', '22', '22', 'b00011', '0000000011', AES_APDU, KIC_AES, KID_AES)
|
|
self.assertEqual(out, AES_REFERENCE_VECTORS[('12', '09')])
|
|
|
|
def test_aes128_counter_plus_one(self):
|
|
out, spi = _ota_reference('1e', '19', '22', '22', 'b00011', '0000000011', AES_APDU, KIC_AES, KID_AES)
|
|
self.assertEqual(out, AES_REFERENCE_VECTORS[('1e', '19')])
|
|
self.assertEqual(spi['counter'], 'counter_must_be_lower')
|
|
|
|
def test_max_load_block_size_fits_one_sms(self):
|
|
# LOAD blocks are too large for SCP80 at the 240-byte default (pySim
|
|
# refuses a secured packet above 140 octets), so the helper finds the
|
|
# largest payload that still encodes into a single SMS.
|
|
mx = _max_load_block_size('16', '01', '15', '15', 'b00000',
|
|
'0000000001', K, K)
|
|
self.assertGreater(mx, 0)
|
|
self.assertLessEqual(mx, 240)
|
|
def load_apdu(n):
|
|
return '80E80000%02X%s00' % (n, '00' * n)
|
|
out, _ = _ota_reference('16', '01', '15', '15', 'b00000',
|
|
'0000000001', load_apdu(mx), K, K)
|
|
self.assertLessEqual(len(out) // 2, 140)
|
|
with self.assertRaises(ValueError):
|
|
_ota_reference('16', '01', '15', '15', 'b00000',
|
|
'0000000001', load_apdu(mx + 1), K, K)
|
|
|
|
def test_max_load_block_size_respects_the_requested_cap(self):
|
|
mx = _max_load_block_size('16', '01', '15', '15', 'b00000',
|
|
'0000000001', K, K, requested=50)
|
|
self.assertLessEqual(mx, 50)
|
|
self.assertGreater(mx, 0)
|
|
|
|
|
|
class TestDecodePor(unittest.TestCase):
|
|
def test_plaintext_no_cc_synthetic(self):
|
|
r = _decode_por('02', '01', '15', '15', '0000000001', K, K,
|
|
'027100000e0ab0000000000000010000016e00')
|
|
self.assertEqual(r['response_status'], 'por_ok')
|
|
self.assertEqual(r['tar'], 'B00000')
|
|
self.assertEqual(r['decoded']['last_status_word'], '6e00')
|
|
|
|
def test_sysmocom_signed(self):
|
|
r = _decode_por('06', '09', '35', '35', '0000000001', KIC3, KID3,
|
|
'027100001612b000110000000000000055f47118381175fb01612f')
|
|
self.assertEqual(r['response_status'], 'por_ok')
|
|
self.assertEqual(r['decoded']['last_status_word'], '612f')
|
|
|
|
def test_sysmocom_ciphered(self):
|
|
r = _decode_por('06', '19', '35', '35', '0000000001', KIC3, KID3,
|
|
'027100001c12b000119660ebdb81be189b5e4389e9e7ab2bc0954f963ad869ed7c')
|
|
self.assertEqual(r['response_status'], 'por_ok')
|
|
self.assertEqual(r['decoded']['last_status_word'], '612f')
|
|
|
|
def test_sysmocom_no_cc(self):
|
|
r = _decode_por('06', '01', '35', '35', '0000000001', KIC3, KID3,
|
|
'027100000e0ab000110000000000000001612f')
|
|
self.assertEqual(r['response_status'], 'por_ok')
|
|
self.assertEqual(r['decoded']['last_status_word'], '612f')
|
|
|
|
def test_complete_field_report(self):
|
|
"""All parsed PoR fields are surfaced verbatim (v1.9.4)."""
|
|
raw = '027100000e0ab000110000000000000001612f'
|
|
r = _decode_por('06', '01', '35', '35', '0000000001', KIC3, KID3, raw)
|
|
self.assertEqual(r['response_status'], 'por_ok')
|
|
self.assertEqual(r['tar'], 'B00011')
|
|
self.assertEqual(r['cntr'], '0000000000')
|
|
self.assertEqual(r['pcntr'], 0)
|
|
self.assertEqual(r['rpl'], 14)
|
|
self.assertEqual(r['rhl'], 10)
|
|
self.assertEqual(r['cc_rc'], '')
|
|
self.assertEqual(r['raw'], raw)
|
|
self.assertNotIn('cntr_low', str(r))
|
|
|
|
def test_cntr_low_fields(self):
|
|
r = _decode_por('02', '01', '15', '15', '0000000001', K, K,
|
|
'027100000b0ab0000000000000070002')
|
|
self.assertEqual(r['response_status'], 'cntr_low')
|
|
self.assertEqual(r['tar'], 'B00000')
|
|
self.assertEqual(r['cntr'], '0000000007')
|
|
self.assertEqual(r['rpl'], 11)
|
|
self.assertEqual(r['rhl'], 10)
|
|
self.assertIsNone(r.get('decoded'))
|
|
|
|
def test_sysmocom_bad_cc_returns_none(self):
|
|
r = _decode_por('06', '09', '35', '35', '0000000001', KIC3, KID3,
|
|
'027100001612b000110000000000000055f47118381175fb02612f')
|
|
self.assertIsNone(r)
|
|
|
|
def test_aes128_ciphered(self):
|
|
r = _decode_por('06', '19', '22', '22', '0000000001', KIC_AES, KID_AES,
|
|
'027100002412b00011ebc6b497e2cad7aedf36ace0e3a29b38853f0fe9ccde81913be5702b73abce1f')
|
|
self.assertEqual(r['response_status'], 'por_ok')
|
|
self.assertEqual(r['decoded']['last_status_word'], '6132')
|
|
|
|
def test_malformed_returns_none(self):
|
|
for bad in ['', '00', '00027100000e0a', '027100000e0ab00000', 'garbage', 'zz']:
|
|
self.assertIsNone(
|
|
_decode_por('02', '01', '15', '15', '0000000001', K, K, bad),
|
|
msg='expected None for %r' % bad)
|
|
|
|
|
|
class TestProactiveDecode(unittest.TestCase):
|
|
"""Server-side proactive command/TR decode helpers (v1.8.0 log feature)."""
|
|
|
|
def setUp(self):
|
|
import pysim_otaman_server.server as srv
|
|
srv._PROACTIVE_SESSION_START = 1234.0
|
|
srv._PLI_DATA[0x00] = '93055210011000'
|
|
|
|
@staticmethod
|
|
def _cmd_raw(cmd_type, qualifier, extras=b''):
|
|
"""A D0-wrapped proactive command (header TLVs + extras)."""
|
|
body = (bytes([0x81, 0x03, 0x01, cmd_type, qualifier])
|
|
+ bytes([0x82, 0x02, 0x83, 0x81]) + extras)
|
|
return bytes([0xD0, len(body)]) + body
|
|
|
|
@staticmethod
|
|
def _decoded(cmd_type, raw, qualifier=None):
|
|
return {d['label']: d['value'] for d in _decode_cmd(cmd_type, raw, qualifier)}
|
|
|
|
def test_decode_cmd_poll_interval(self):
|
|
r = _decode_cmd(0x03, bytes.fromhex('d00d8103010300820283818402011e'), None)
|
|
self.assertEqual(r, [{'label': 'Interval', 'value': '30 s'}])
|
|
|
|
def test_decode_cmd_setup_event_list(self):
|
|
r = _decode_cmd(0x05, bytes.fromhex('d00c810301050082028381990101'), None)
|
|
self.assertEqual(r, [{'label': 'Events', 'value': 'Call connected'}])
|
|
|
|
def test_decode_cmd_send_short_message(self):
|
|
# SEND SHORT MESSAGE with an SMS-SUBMIT TPDU carrying GSM-7 text.
|
|
tpdu = bytes.fromhex('010006912143F5000005E8329BFD06')
|
|
raw = self._cmd_raw(0x13, 0, bytes([0x8B, len(tpdu)]) + tpdu)
|
|
r = self._decoded(0x13, raw)
|
|
self.assertEqual(r['Type'], 'SMS-SUBMIT')
|
|
self.assertEqual(r['TP-MR'], '0')
|
|
self.assertEqual(r['TP-DA'], '12345')
|
|
self.assertEqual(r['TP-PID'], '0x00')
|
|
self.assertEqual(r['TP-DCS'], '0x00')
|
|
self.assertEqual(r['TP-UDL'], '5')
|
|
self.assertEqual(r['Text'], 'hello')
|
|
self.assertEqual(r['SMS TPDU'], tpdu.hex().upper())
|
|
|
|
def test_decode_cmd_send_short_message_udh_8bit(self):
|
|
# UDHI + concatenation IE (16-bit ref) + 8-bit text data.
|
|
udh = bytes.fromhex('0608040001020341 42'.replace(' ', ''))
|
|
tpdu = (bytes.fromhex('4100' '06912143F5' '00' '04' '09') + udh)
|
|
raw = self._cmd_raw(0x13, 0, bytes([0x8B, len(tpdu)]) + tpdu)
|
|
r = self._decoded(0x13, raw)
|
|
self.assertEqual(r['Concat (16-bit ref)'], '1, part 2/3')
|
|
self.assertEqual(r['Text'], 'AB')
|
|
|
|
def test_decode_cmd_send_short_message_ucs2(self):
|
|
text = 'Тест'.encode('utf-16-be')
|
|
tpdu = (bytes.fromhex('0100' '06912143F5' '00' '08' '%02X' % len(text))
|
|
+ text)
|
|
raw = self._cmd_raw(0x13, 0, bytes([0x8B, len(tpdu)]) + tpdu)
|
|
r = self._decoded(0x13, raw)
|
|
self.assertEqual(r['TP-DCS'], '0x08')
|
|
self.assertEqual(r['Text'], 'Тест')
|
|
|
|
def test_decode_cmd_send_short_message_secured_packet(self):
|
|
# PID 0x7F = SIM data download: the UD is a secured packet (TS 31.115).
|
|
tpdu = bytes.fromhex('0100' '06912143F5' '7F' 'F6' '03' 'AABBCC')
|
|
raw = self._cmd_raw(0x13, 0, bytes([0x8B, len(tpdu)]) + tpdu)
|
|
r = self._decoded(0x13, raw)
|
|
self.assertEqual(r['TP-PID'], '0x7F (SIM data download)')
|
|
self.assertEqual(r['Secured packet (TS 31.115)'], '3 bytes: AABBCC')
|
|
|
|
def test_decode_cmd_send_short_message_malformed_falls_back(self):
|
|
# A malformed/garbage TPDU must not raise: the raw hex line remains.
|
|
raw = bytes.fromhex('d0158103011300820283818b0b916106152670f900a35f020101')
|
|
r = self._decoded(0x13, raw)
|
|
self.assertEqual(r['SMS TPDU'], '916106152670F900A35F02')
|
|
|
|
def test_decode_cmd_pli_qualifier_name(self):
|
|
r = _decode_cmd(0x26, b'\xd0', 0x00)
|
|
self.assertTrue(r[0]['value'].startswith('Location Information (MCC, MNC, LAC/TAC, Cell ID)'))
|
|
|
|
def test_decode_cmd_pli_all_standard_qualifiers_named(self):
|
|
# TS 102 223 V18.3.0 (PLI qualifier coding): names must exist even
|
|
# without a special data decoder, e.g. ESN (07) and MEID (0B).
|
|
cases = {
|
|
0x07: 'ESN',
|
|
0x0B: 'MEID',
|
|
0x1A: 'Supported Radio Access Technologies',
|
|
0x05: 'Reserved for GSM',
|
|
}
|
|
for qualifier, name in cases.items():
|
|
r = _decode_cmd(0x26, b'\xd0', qualifier)
|
|
self.assertIn(name, r[0]['value'], 'qualifier 0x%02X' % qualifier)
|
|
|
|
def test_decode_cmd_timer_management_start(self):
|
|
# TS 102 223 6.6.21/8.37/8.38: start timer 3 for 14:07:32
|
|
raw = bytes.fromhex('d011810301270082028182a40103a503417023')
|
|
self.assertEqual(_decode_cmd(0x27, raw, 0x00), [
|
|
{'label': 'Action', 'value': 'Start'},
|
|
{'label': 'Timer', 'value': '3'},
|
|
{'label': 'Value', 'value': '14:07:32'},
|
|
])
|
|
|
|
def test_decode_cmd_timer_management_plain_tags(self):
|
|
# Cards may use the plain (non comprehension-required) tag variant.
|
|
raw = bytes.fromhex('d00c010301270102028182240103')
|
|
self.assertEqual(_decode_cmd(0x27, raw, 0x01), [
|
|
{'label': 'Action', 'value': 'Deactivate'},
|
|
{'label': 'Timer', 'value': '3'},
|
|
])
|
|
|
|
def test_decode_cmd_open_channel_cr_tags(self):
|
|
# Same OPEN CHANNEL as the reference traces, but with CR-set TLVs.
|
|
raw = bytes.fromhex(
|
|
'd02b8103014001820281828500b50103b9020200c70b076d656761666f6e2e7275'
|
|
'bc03021f90be05217f000001')
|
|
r = _decode_cmd(0x40, raw, 0x01)
|
|
self.assertIn({'label': 'Bearer', 'value': '0x03'}, r)
|
|
self.assertIn({'label': 'Buffer size', 'value': '512'}, r)
|
|
self.assertIn({'label': 'APN', 'value': 'megafon.ru'}, r)
|
|
self.assertIn({'label': 'Destination', 'value': '127.0.0.1'}, r)
|
|
self.assertIn({'label': 'Transport', 'value': 'TCP client port 8080'}, r)
|
|
|
|
def test_decode_cmd_bip_channel_from_device_ids(self):
|
|
# Real trace: SEND DATA carries the channel in the device identities
|
|
# (source UICC 0x81, destination Channel 1 0x21).
|
|
raw = bytes.fromhex('d00e8103014301820281213701013603aabbcc')
|
|
r = _decode_cmd(0x43, raw, 0x01)
|
|
self.assertEqual(r[0], {'label': 'Channel', 'value': '1'})
|
|
self.assertEqual(r[1], {'label': 'Data bytes', 'value': '3'})
|
|
|
|
def test_parse_proactive_header_plain_tags(self):
|
|
import pysim_otaman_server.server as srv
|
|
raw = bytes.fromhex('d00c010301270102028182240103')
|
|
self.assertEqual(srv._parse_proactive_header(raw), (1, 0x27, 0x81, 0x82, 0x01))
|
|
|
|
def test_default_handler_logs_timer_management(self):
|
|
# pySim's auto-handler path: the parsed command object (not the empty
|
|
# collection) is re-encoded for the log and used for the response.
|
|
import pysim_otaman_server.server as srv
|
|
from pySim.cat import ProactiveCommand
|
|
from pySim.utils import h2b
|
|
srv._PROACTIVE_LOG.clear()
|
|
handler = srv._DefaultProactiveHandler()
|
|
pcmd = ProactiveCommand()
|
|
parsed = pcmd.from_tlv(h2b('d011810301270082028182a40103a503417023'))
|
|
ti = handler.receive_fetch_raw(pcmd, parsed)
|
|
tr = b''.join(x.to_tlv() for x in ti).hex()
|
|
self.assertTrue(tr.startswith('810301270082028281830100'), tr)
|
|
entry = srv._PROACTIVE_LOG[-1]
|
|
self.assertEqual(entry['type_hex'], '27')
|
|
self.assertEqual(entry['type_name'], 'TIMER MANAGEMENT')
|
|
self.assertEqual(entry['tr_result'], '00')
|
|
|
|
def test_default_handler_pli_includes_dict_data(self):
|
|
import pysim_otaman_server.server as srv
|
|
from pySim.cat import ProactiveCommand
|
|
from pySim.utils import h2b
|
|
srv._PROACTIVE_LOG.clear()
|
|
srv._PLI_DATA[0x00] = '93055210011000'
|
|
handler = srv._DefaultProactiveHandler()
|
|
pcmd = ProactiveCommand()
|
|
parsed = pcmd.from_tlv(h2b('d00d810301260082028182'))
|
|
ti = handler.receive_fetch_raw(pcmd, parsed)
|
|
tr = b''.join(x.to_tlv() for x in ti).hex()
|
|
self.assertIn('93055210011000', tr)
|
|
self.assertEqual(srv._PROACTIVE_LOG[-1]['tr_hex'], '93055210011000')
|
|
|
|
def test_decode_cmd_empty_raw(self):
|
|
self.assertEqual(_decode_cmd(0x26, b'', None), [])
|
|
self.assertEqual(_decode_cmd(0x03, None, None), [])
|
|
|
|
def test_decode_tr_pli_location(self):
|
|
r = _decode_tr('26', '00', '93055210011000')
|
|
self.assertEqual(r, [
|
|
{'label': 'MCC', 'value': '250'},
|
|
{'label': 'MNC', 'value': '11'},
|
|
{'label': 'LAC/TAC', 'value': '1000'},
|
|
])
|
|
|
|
def test_decode_tr_pli_imei(self):
|
|
r = _decode_tr('26', '01', '94082143658709214305')
|
|
self.assertEqual(r[0], {'label': 'IMEI', 'value': '123456789012345'})
|
|
|
|
def test_decode_tr_pli_access_technology(self):
|
|
r = _decode_tr('26', '06', 'bf0103')
|
|
self.assertEqual(r, [{'label': 'Access Technology', 'value': 'UTRAN (3)'}])
|
|
|
|
def test_decode_tr_pli_search_mode(self):
|
|
r = _decode_tr('26', '09', 'ad0101')
|
|
self.assertEqual(r, [{'label': 'Search Mode', 'value': 'Manual'}])
|
|
|
|
def test_decode_tr_poll_interval(self):
|
|
r = _decode_tr('03', None, '8402011e')
|
|
self.assertEqual(r, [{'label': 'Interval', 'value': '30 s'}])
|
|
|
|
def test_decode_tr_empty(self):
|
|
self.assertEqual(_decode_tr('26', '00', ''), [])
|
|
|
|
def test_tr_data_only_strips_boilerplate(self):
|
|
tr = bytes.fromhex('81030326008202818393055210011000030100')
|
|
self.assertEqual(_tr_data_only(tr).hex(), '93055210011000')
|
|
|
|
def test_build_and_record_tr(self):
|
|
entry = {'type_hex': '26', 'qualifier': '00'}
|
|
tr = _build_tr(None, 3, 0x26, 0x83, 0x81, 0x00)
|
|
_record_tr(entry, tr, '9000')
|
|
self.assertEqual(entry['tr_hex'], '93055210011000')
|
|
self.assertEqual(entry['tr_sw'], '9000')
|
|
self.assertEqual([f['label'] for f in entry['tr_decoded']], ['MCC', 'MNC', 'LAC/TAC'])
|
|
|
|
def test_record_tr_without_sw(self):
|
|
entry = {'type_hex': '26', 'qualifier': '00'}
|
|
_record_tr(entry, bytes.fromhex('93055210011000'))
|
|
self.assertNotIn('tr_sw', entry)
|
|
self.assertEqual(entry['tr_hex'], '93055210011000')
|
|
|
|
def test_log_proactive_fields(self):
|
|
entry = _log_proactive(0x26, b'\xd0', 0x00, 7)
|
|
self.assertEqual(entry['cmd_num'], 7)
|
|
self.assertEqual(entry['type_hex'], '26')
|
|
self.assertEqual(entry['qualifier'], '00')
|
|
self.assertEqual(entry['raw'], 'd0')
|
|
self.assertEqual(entry['cmd_decoded'][0]['label'], 'Qualifier')
|
|
self.assertIn('id', entry)
|
|
|
|
def test_log_proactive_malformed_raw(self):
|
|
entry = _log_proactive(0x21, bytes([0xD0, 0xFF]), 0x00)
|
|
self.assertEqual(entry['cmd_decoded'], [])
|
|
|
|
def test_log_proactive_no_cmd_num(self):
|
|
entry = _log_proactive(0x05, bytes.fromhex('990101'), None)
|
|
self.assertNotIn('cmd_num', entry)
|
|
|
|
def test_record_tr_basic_result(self):
|
|
entry = {'type_hex': '03', 'qualifier': None}
|
|
_record_tr(entry, bytes.fromhex('8103010300820281838402011e030100'))
|
|
self.assertEqual(entry['tr_result'], '00')
|
|
self.assertEqual(entry['tr_result_name'], 'Command performed successfully')
|
|
self.assertEqual(entry['tr_hex'], '8402011e')
|
|
|
|
def test_record_tr_general_result(self):
|
|
entry = {'type_hex': '26', 'qualifier': '00'}
|
|
_record_tr(entry, bytes.fromhex('8103032600820281839305521001100083022001'))
|
|
self.assertEqual(entry['tr_result'], '2001')
|
|
self.assertEqual(entry['tr_result_name'], 'ME currently unable to process command')
|
|
self.assertEqual(entry['tr_hex'], '93055210011000')
|
|
|
|
def test_record_tr_unknown_result(self):
|
|
entry = {'type_hex': '26', 'qualifier': '00'}
|
|
_record_tr(entry, bytes.fromhex('810303260082028181030107'))
|
|
self.assertEqual(entry['tr_result'], '07')
|
|
self.assertNotIn('tr_result_name', entry)
|
|
|
|
def test_record_tr_no_result_tlv(self):
|
|
entry = {'type_hex': '26', 'qualifier': '00'}
|
|
_record_tr(entry, bytes.fromhex('810303260082028181'))
|
|
self.assertNotIn('tr_result', entry)
|
|
|
|
|
|
class TestEventDownload(unittest.TestCase):
|
|
"""ENVELOPE (EVENT DOWNLOAD) assembly, TS 102 223 7.5.11."""
|
|
|
|
def _send(self, event_type, event_data):
|
|
import pysim_otaman_server.server as srv
|
|
calls = []
|
|
|
|
class Tp:
|
|
def send_apdu(self, apdu):
|
|
calls.append(apdu)
|
|
return '', '9000'
|
|
|
|
class Scc:
|
|
cat_cla = '80'
|
|
_tp = Tp()
|
|
|
|
data, sw = srv._send_event_download(Scc(), event_type, event_data)
|
|
return calls[0], sw
|
|
|
|
def test_channel_status_event(self):
|
|
# Event list + device identities + Channel status (8.56): channel 2,
|
|
# link established, info 05 = link dropped.
|
|
apdu, sw = self._send(0x0A, bytes.fromhex('b8028205'))
|
|
self.assertEqual(sw, '9000')
|
|
self.assertEqual(apdu, '80c200000dd60b99010a82028281b8028205')
|
|
|
|
def test_event_without_data(self):
|
|
apdu, sw = self._send(0x05, None)
|
|
self.assertEqual(sw, '9000')
|
|
self.assertEqual(apdu, '80c2000009d60799010582028281')
|
|
|
|
|
|
class TestTimerManagement(unittest.TestCase):
|
|
"""Terminal side of TIMER MANAGEMENT (TS 102 223 6.6.21, 6.8.13/14, 7.4).
|
|
|
|
The start vector is the live card's: timer 1, 60 s."""
|
|
|
|
START = bytes.fromhex('d011810301270082028182a40101a503001000')
|
|
|
|
def tearDown(self):
|
|
import pysim_otaman_server.server as srv
|
|
srv._timer_cancel()
|
|
|
|
def test_hms_bcd_roundtrip(self):
|
|
import pysim_otaman_server.server as srv
|
|
self.assertEqual(srv._hms_bcd(60).hex(), '001000')
|
|
self.assertEqual(srv._hms_bcd(3723).hex(), '102030')
|
|
self.assertEqual([srv._bcd_swap(b) for b in srv._hms_bcd(3723)], [1, 2, 3])
|
|
|
|
def test_start_returns_result_only_and_arms_timer(self):
|
|
import pysim_otaman_server.server as srv
|
|
tr = srv._handle_timer_command(1, 0x27, 0x00, self.START, 0x81, 0x82)
|
|
self.assertEqual(tr.hex(), '810301270082028281030100')
|
|
remaining = srv._timer_remaining(1)
|
|
self.assertTrue(55 <= remaining <= 60, remaining)
|
|
|
|
def test_get_returns_remaining_value(self):
|
|
import pysim_otaman_server.server as srv
|
|
srv._handle_timer_command(1, 0x27, 0x00, self.START, 0x81, 0x82)
|
|
tr = srv._handle_timer_command(1, 0x27, 0x02, self.START, 0x81, 0x82)
|
|
self.assertEqual(tr.hex(), '810301270282028281a40101a503001000030100')
|
|
|
|
def test_deactivate_stops_and_reports_value(self):
|
|
import pysim_otaman_server.server as srv
|
|
srv._handle_timer_command(1, 0x27, 0x00, 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.assertIn('a40101a503001000', tr.hex())
|
|
self.assertIsNone(srv._timer_remaining(1))
|
|
|
|
def test_get_on_stopped_timer_is_contradiction(self):
|
|
import pysim_otaman_server.server as srv
|
|
tr = srv._handle_timer_command(1, 0x27, 0x02, self.START, 0x81, 0x82)
|
|
self.assertEqual(tr.hex(), '810301270282028281030124')
|
|
|
|
def test_timer_expiration_envelope(self):
|
|
import pysim_otaman_server.server as srv
|
|
calls = []
|
|
|
|
class Tp:
|
|
def send_apdu(self, apdu):
|
|
calls.append(apdu)
|
|
return '', '9000'
|
|
|
|
ref = types.SimpleNamespace(
|
|
scc=types.SimpleNamespace(cat_cla='80', _tp=Tp()), stk_pending=None)
|
|
with mock.patch.object(srv, '_server_ref', ref):
|
|
with mock.patch.object(srv, '_CARD_CONNECTED', True):
|
|
srv._timer_expired(1, 60)
|
|
# D7 0C: device identities (terminal -> UICC), Timer id A4, value A5
|
|
self.assertEqual(calls, ['80c200000ed70c82028281a40101a503001000'])
|
|
|
|
def test_cancelled_timer_does_not_report(self):
|
|
import pysim_otaman_server.server as srv
|
|
calls = []
|
|
|
|
class Tp:
|
|
def send_apdu(self, apdu):
|
|
calls.append(apdu)
|
|
return '', '9000'
|
|
|
|
ref = types.SimpleNamespace(
|
|
scc=types.SimpleNamespace(cat_cla='80', _tp=Tp()), stk_pending=None)
|
|
with mock.patch.object(srv, '_server_ref', ref):
|
|
with mock.patch.object(srv, '_CARD_CONNECTED', True):
|
|
srv._timer_fire(1, 60) # never started/cancelled
|
|
self.assertEqual(calls, [])
|
|
|
|
def test_decode_tr_timer(self):
|
|
tr = bytes.fromhex('810301270082028281a40101a503001000030100')
|
|
data = _tr_data_only(tr).hex()
|
|
r = _decode_tr('27', '00', data)
|
|
self.assertEqual(r, [{'label': 'Timer', 'value': '1'},
|
|
{'label': 'Remaining', 'value': '00:01:00'}])
|
|
|
|
|
|
class TestExpandedRemoteResponse(unittest.TestCase):
|
|
"""Expanded Remote Response parsing (TS 102 226 §5.2.2)."""
|
|
|
|
def test_expanded_response_single_command(self):
|
|
entry = {'type_hex': '03', 'qualifier': None}
|
|
_record_tr(entry, bytes.fromhex('810301030082028183030100'))
|
|
self.assertEqual(entry['tr_result'], '00')
|
|
self.assertEqual(entry['tr_result_name'], 'Command performed successfully')
|
|
|
|
def test_expanded_response_parser_single_command(self):
|
|
# Simple test of the construct parsing structure
|
|
try:
|
|
from construct import Struct, Int8ub, Bytes, GreedyBytes, Optional, Array, this
|
|
from osmocom.utils import b2h
|
|
|
|
# Create sample expanded response data
|
|
secured_data = bytes.fromhex('01' '01' '9000' '11') # response_count, cmd#, SW, data
|
|
|
|
ExpandedRemoteResponse = Struct(
|
|
'response_count'/Int8ub,
|
|
'responses'/Array(this.response_count, Struct(
|
|
'command_number'/Int8ub,
|
|
'status_word'/Bytes(2),
|
|
'response_data'/GreedyBytes,
|
|
'error_details'/Optional(Struct(
|
|
'error_code'/Int8ub,
|
|
'error_info'/GreedyBytes
|
|
)),
|
|
'chaining_context'/Optional(Struct(
|
|
'script_id'/Bytes(4),
|
|
'is_first'/Int8ub,
|
|
'is_last'/Int8ub,
|
|
))
|
|
))
|
|
)
|
|
expanded = ExpandedRemoteResponse.parse(secured_data)
|
|
self.assertEqual(expanded.response_count, 1)
|
|
self.assertEqual(expanded.responses[0].command_number, 1)
|
|
self.assertEqual(b2h(expanded.responses[0].status_word).upper(), '9000')
|
|
self.assertEqual(b2h(expanded.responses[0].response_data).upper(), '11')
|
|
except Exception as e:
|
|
self.fail(f"ExpandedRemoteResponse parsing failed: {e}")
|
|
|
|
def test_expanded_response_with_error(self):
|
|
entry = {'type_hex': '26', 'qualifier': '01'}
|
|
# Result TLV: 03 01 6A (error_code 0x6A)
|
|
_record_tr(entry, bytes.fromhex('81030326008202818303016A'))
|
|
self.assertEqual(entry['tr_result'], '6a')
|
|
self.assertEqual(entry['tr_result_name'], 'Command performed with limited understanding')
|
|
|
|
def test_expanded_response_with_chaining(self):
|
|
entry = {'type_hex': '26', 'qualifier': '00'}
|
|
# Result: 03 01 00 + chaining context with script_id
|
|
_record_tr(entry, bytes.fromhex('81030326008202818303010093'))
|
|
self.assertEqual(entry['tr_result'], '00')
|
|
self.assertEqual(entry['tr_result_name'], 'Command performed successfully')
|
|
|
|
|
|
class TestSmsConcat(unittest.TestCase):
|
|
"""Tests for _parse_sms_concat — SMS UDH concatenation parsing."""
|
|
|
|
def test_no_udh_sms_submit(self):
|
|
"""SMS-SUBMIT without TP-UDHI: entire UD is payload."""
|
|
from pysim_otaman_server.server import _parse_sms_concat
|
|
# First octet 0x01: MTI=01 (SUBMIT), no UDH, no VP
|
|
# MR=00, DA_len=05, DA_type=90, DA=2143F5, PID=00, DCS=04, UDL=03, UD=AABBCC
|
|
tpdu = bytes.fromhex('0100' # first octet + MR
|
|
'05' # DA length
|
|
'90' # DA type
|
|
'2143F5' # DA data (3 bytes for 5 digits)
|
|
'0004' # PID + DCS
|
|
'03' # UDL
|
|
'AABBCC') # UD (payload)
|
|
ref, total, num, payload = _parse_sms_concat(tpdu)
|
|
self.assertIsNone(ref)
|
|
self.assertIsNone(total)
|
|
self.assertIsNone(num)
|
|
self.assertEqual(payload.hex(), 'aabbcc')
|
|
|
|
def test_8bit_concat_iei_0x00(self):
|
|
"""SMS-SUBMIT with IEI 0x00 (8-bit reference concatenation)."""
|
|
from pysim_otaman_server.server import _parse_sms_concat
|
|
# First octet 0x41: MTI=01 (SUBMIT), TP-UDHI=1, no VP
|
|
# MR=00, DA_len=05, DA_type=90, DA=2143F5, PID=00, DCS=04
|
|
# UDL=09, UDHL=05, UDH: 00 03 04 04 01 (concat IE), payload=AABBCC
|
|
tpdu = bytes.fromhex('4100' # first octet + MR
|
|
'05' # DA length
|
|
'90' # DA type
|
|
'2143F5' # DA data
|
|
'0004' # PID + DCS
|
|
'09' # UDL (1 UDHL + 5 UDH + 3 payload = 9)
|
|
'05' # UDHL = 5 bytes of UDH
|
|
'0003' # IEI=0x00, IEDL=3
|
|
'04' # ref
|
|
'04' # total (4 segments)
|
|
'01' # num (segment 1)
|
|
'AABBCC') # payload
|
|
ref, total, num, payload = _parse_sms_concat(tpdu)
|
|
self.assertEqual(ref, 0x04)
|
|
self.assertEqual(total, 4)
|
|
self.assertEqual(num, 1)
|
|
self.assertEqual(payload.hex(), 'aabbcc')
|
|
|
|
def test_16bit_concat_iei_0x08(self):
|
|
"""SMS-SUBMIT with IEI 0x08 (16-bit reference concatenation)."""
|
|
from pysim_otaman_server.server import _parse_sms_concat
|
|
# First octet 0x41: MTI=01, TP-UDHI=1
|
|
# UDH: 06 (UDHL) 08 04 01 02 03 04 (16-bit concat: ref=0x0102, total=3, num=4)
|
|
# payload=FF
|
|
tpdu = bytes.fromhex('4100'
|
|
'05'
|
|
'90'
|
|
'2143F5'
|
|
'0004'
|
|
'08' # UDL (1 UDHL + 6 UDH + 1 payload = 8)
|
|
'06' # UDHL
|
|
'0804' # IEI=0x08, IEDL=4
|
|
'0102' # ref (16-bit, big-endian)
|
|
'03' # total
|
|
'04' # num
|
|
'FF') # payload
|
|
ref, total, num, payload = _parse_sms_concat(tpdu)
|
|
self.assertEqual(ref, 0x0102)
|
|
self.assertEqual(total, 3)
|
|
self.assertEqual(num, 4)
|
|
self.assertEqual(payload.hex(), 'ff')
|
|
|
|
def test_udh_with_cpi(self):
|
|
"""UDH with concatenation IE + CPI IE (0x70)."""
|
|
from pysim_otaman_server.server import _parse_sms_concat
|
|
# First octet 0x41: MTI=01, TP-UDHI=1
|
|
# UDHL=07, UDH: 00 03 04 04 01 (concat) + 70 00 (CPI)
|
|
tpdu = bytes.fromhex('4100'
|
|
'05'
|
|
'90'
|
|
'2143F5'
|
|
'0004'
|
|
'0A' # UDL (1 UDHL + 7 UDH + 1 payload = 9? no: 1+5+2+1=9, but UDH=7 bytes)
|
|
'07' # UDHL = 7
|
|
'0003' # IEI=0x00, IEDL=3
|
|
'04' # ref
|
|
'04' # total
|
|
'01' # num
|
|
'7000' # CPI IE (IEI=0x70, IEDL=0)
|
|
'DD') # payload
|
|
ref, total, num, payload = _parse_sms_concat(tpdu)
|
|
self.assertEqual(ref, 0x04)
|
|
self.assertEqual(total, 4)
|
|
self.assertEqual(num, 1)
|
|
self.assertEqual(payload.hex(), 'dd')
|
|
|
|
def test_empty_payload(self):
|
|
"""Segment with empty payload after UDH."""
|
|
from pysim_otaman_server.server import _parse_sms_concat
|
|
# First octet 0x41: MTI=01, TP-UDHI=1
|
|
tpdu = bytes.fromhex('4100'
|
|
'05'
|
|
'90'
|
|
'2143F5'
|
|
'0004'
|
|
'06' # UDL (1 UDHL + 5 UDH + 0 payload = 6)
|
|
'05' # UDHL
|
|
'0003'
|
|
'01'
|
|
'02'
|
|
'01') # no payload after UDH
|
|
ref, total, num, payload = _parse_sms_concat(tpdu)
|
|
self.assertEqual(ref, 0x01)
|
|
self.assertEqual(total, 2)
|
|
self.assertEqual(num, 1)
|
|
self.assertEqual(len(payload), 0)
|
|
|
|
def test_short_tpdu(self):
|
|
"""Truncated TPDU returns gracefully."""
|
|
from pysim_otaman_server.server import _parse_sms_concat
|
|
ref, total, num, payload = _parse_sms_concat(b'\x01')
|
|
self.assertIsNone(ref)
|
|
self.assertIsNone(total)
|
|
self.assertIsNone(num)
|
|
|
|
def test_none_input(self):
|
|
"""None input returns empty payload."""
|
|
from pysim_otaman_server.server import _parse_sms_concat
|
|
ref, total, num, payload = _parse_sms_concat(None)
|
|
self.assertIsNone(ref)
|
|
self.assertIsNone(total)
|
|
self.assertIsNone(num)
|
|
self.assertEqual(len(payload), 0)
|
|
|
|
def test_short_tpdu(self):
|
|
"""Truncated TPDU returns gracefully."""
|
|
from pysim_otaman_server.server import _parse_sms_concat
|
|
ref, total, num, payload = _parse_sms_concat(b'\x44')
|
|
self.assertIsNone(ref)
|
|
self.assertIsNone(total)
|
|
self.assertIsNone(num)
|
|
|
|
def test_none_input(self):
|
|
"""None input returns empty payload."""
|
|
from pysim_otaman_server.server import _parse_sms_concat
|
|
ref, total, num, payload = _parse_sms_concat(None)
|
|
self.assertIsNone(ref)
|
|
self.assertIsNone(total)
|
|
self.assertIsNone(num)
|
|
self.assertEqual(len(payload), 0)
|
|
|
|
|
|
class TestSmsReassembly(unittest.TestCase):
|
|
"""Tests for SMS segment reassembly logic."""
|
|
|
|
def test_single_segment_no_concat(self):
|
|
"""Single segment without UDH → submit_tpdu_hex is set directly."""
|
|
from pysim_otaman_server.server import PoRSubmitHandler, _find_sms_tpdu, _parse_sms_concat
|
|
handler = PoRSubmitHandler()
|
|
# Build a simple D0 with tag 8B containing an SMS-SUBMIT without UDH
|
|
sms_tpdu = bytes.fromhex('040005902143F50004' # SMS-SUBMIT header
|
|
'03' # UDL
|
|
'AABBCC') # payload
|
|
# Wrap in D0 proactive command
|
|
d0 = bytes([0xD0, len(sms_tpdu) + 4, # approximate BER length
|
|
0x81, 0x03, 0x01, 0x13, 0x00, # Command Details
|
|
0x82, 0x02, 0x81, 0x83, # Device Identities
|
|
0x8B, len(sms_tpdu)]) # tag 8B
|
|
# Simulate _find_sms_tpdu extracting tag 8B
|
|
found = sms_tpdu.hex()
|
|
# Parse and check
|
|
ref, total, num, payload = _parse_sms_concat(sms_tpdu)
|
|
self.assertIsNone(ref)
|
|
handler.submit_tpdu_hex = found # single segment path
|
|
self.assertEqual(handler.submit_tpdu_hex, found)
|
|
|
|
def test_multi_segment_reassembly(self):
|
|
"""3 segments with IEI 0x00 in random order → assembled in correct order."""
|
|
from pysim_otaman_server.server import PoRSubmitHandler
|
|
handler = PoRSubmitHandler()
|
|
|
|
# Segment payloads (after UDH)
|
|
payloads = [b'\x01\x02', b'\x03\x04', b'\x05\x06']
|
|
ref = 0x42
|
|
total = 3
|
|
|
|
# Simulate receiving segments in random order: 2, 0, 1
|
|
for idx in [1, 0, 2]:
|
|
num = idx + 1
|
|
handler.sms_segments.append((ref, total, num, payloads[idx].hex()))
|
|
# Check if all segments collected
|
|
matching = [s for s in handler.sms_segments if s[0] == ref]
|
|
if len(matching) >= total:
|
|
sorted_segs = sorted(matching, key=lambda s: s[2])
|
|
assembled = b''.join(bytes.fromhex(s[3]) for s in sorted_segs)
|
|
handler.submit_tpdu_hex = assembled.hex()
|
|
|
|
self.assertEqual(handler.submit_tpdu_hex, '010203040506')
|
|
|
|
def test_independent_references(self):
|
|
"""Two different reference numbers are independent."""
|
|
from pysim_otaman_server.server import PoRSubmitHandler
|
|
handler = PoRSubmitHandler()
|
|
|
|
# Ref 0x01: 2 segments
|
|
handler.sms_segments.append((0x01, 2, 1, 'AA'))
|
|
handler.sms_segments.append((0x01, 2, 2, 'BB'))
|
|
matching = [s for s in handler.sms_segments if s[0] == 0x01]
|
|
if len(matching) >= 2:
|
|
sorted_segs = sorted(matching, key=lambda s: s[2])
|
|
assembled = b''.join(bytes.fromhex(s[3]) for s in sorted_segs)
|
|
handler.submit_tpdu_hex = assembled.hex()
|
|
|
|
self.assertEqual(handler.submit_tpdu_hex, 'aabb')
|
|
|
|
# Ref 0x02: 1 segment (independent)
|
|
handler.sms_segments.append((0x02, 1, 1, 'CC'))
|
|
matching2 = [s for s in handler.sms_segments if s[0] == 0x02]
|
|
if len(matching2) >= 1:
|
|
sorted_segs2 = sorted(matching2, key=lambda s: s[2])
|
|
assembled2 = b''.join(bytes.fromhex(s[3]) for s in sorted_segs2)
|
|
# Only update if ref 0x02 is complete
|
|
handler.submit_tpdu_hex = assembled2.hex()
|
|
|
|
# Last assembly was ref 0x02
|
|
self.assertEqual(handler.submit_tpdu_hex, 'cc')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|
|
|
|
|
|
class CapApduSequenceTest(unittest.TestCase):
|
|
"""RAM APDU sequence shared by the SCP80 and SCP81 install paths."""
|
|
|
|
def _mini_cap(self):
|
|
import io, zipfile
|
|
# Header: tag(1) size(2) magic(4) minor(1) major(1) flags(1)
|
|
# pkg minor(1) pkg major(1) aid_len(1) aid(N)
|
|
header = (b'\x01\x00\x11' + b'\xde\xca\xff\xed' + b'\x00\x01\x00' +
|
|
b'\x00\x01' + b'\x06' + b'\xa0\x00\x00\x01\x00\x01')
|
|
# Applet: tag(1) size(2) count(1) aid_len(1) module_aid(N) offset(2)
|
|
applet = (b'\x03\x00\x0a\x01\x05' + b'\xa0\x00\x00\x01\x00' + b'\x00\x08')
|
|
buf = io.BytesIO()
|
|
zf = zipfile.ZipFile(buf, 'w')
|
|
zf.writestr('pkg/Header.cap', header)
|
|
zf.writestr('pkg/Applet.cap', applet)
|
|
zf.close()
|
|
return buf.getvalue().hex().upper()
|
|
|
|
def test_cap_parse(self):
|
|
from pysim_otaman_server.server import _cap_parse
|
|
loadfile_aid, module_aid, data = _cap_parse(self._mini_cap())
|
|
self.assertEqual(loadfile_aid, 'A00000010001')
|
|
self.assertEqual(module_aid, 'A000000100')
|
|
# Header then Applet, per the CAP component order.
|
|
self.assertTrue(data.startswith('010011DECAFFED'))
|
|
self.assertIn('03000A01', data)
|
|
|
|
def test_sequence_install_load_install(self):
|
|
from pysim_otaman_server.server import _cap_apdu_sequence
|
|
seq = _cap_apdu_sequence('A00000010001', 'A000000100', 'AABBCCDD')
|
|
# INSTALL [for load]: lv(pkg aid) + lv(ISD) + 000000
|
|
self.assertEqual(seq[0],
|
|
'80E6020013' + '06A00000010001' + '08A000000003000000' + '000000' + '00')
|
|
# One LOAD block (small payload, last -> P1=0x80, P2=0)
|
|
self.assertEqual(seq[1][:8], '80E88000')
|
|
self.assertTrue(seq[1].endswith('00'))
|
|
# INSTALL [for install]: C9 00 install params appended to the lv chain
|
|
self.assertTrue(seq[2].startswith('80E60C00'))
|
|
self.assertIn('06A00000010001' + '05A000000100' + '05A000000100' + '0100', seq[2])
|
|
|
|
def test_load_blocks_split_and_counter(self):
|
|
from pysim_otaman_server.server import _cap_apdu_sequence, _ber_len as _ber_len_lower
|
|
data = ''.join('%02X' % (i % 256) for i in range(700))
|
|
seq = _cap_apdu_sequence('A00000010001', 'A000000100', data)
|
|
self.assertEqual(len(seq), 5) # INSTALL + 3 LOAD + INSTALL
|
|
self.assertEqual(seq[1][:8], '80E80000')
|
|
self.assertEqual(seq[2][:8], '80E80001')
|
|
self.assertEqual(seq[3][:8], '80E88002') # last block: P1=0x80
|
|
# The blocks are consecutive chunks and reassemble the load file TLV
|
|
# byte-for-byte (a shifted/overlapping split fails the card mid-load).
|
|
def payload(apdu):
|
|
lc = int(apdu[8:10], 16)
|
|
return apdu[10:10 + lc * 2]
|
|
joined = payload(seq[1]) + payload(seq[2]) + payload(seq[3])
|
|
self.assertTrue(joined.startswith('C482'))
|
|
expected = 'C4' + _ber_len_lower(700) + data # 700 = 0x2BC
|
|
self.assertEqual(joined.upper(), expected.upper())
|
|
self.assertEqual(int(seq[3][8:10], 16), len(expected) // 2 - 480)
|
|
|
|
def test_custom_block_size_splits_into_more_blocks(self):
|
|
# A smaller block size (SCP80: fit one SMS) slices the load file TLV
|
|
# into consecutive chunks of that size, the last block marked P1=0x80
|
|
# with the block counter in P2.
|
|
from pysim_otaman_server.server import _cap_apdu_sequence
|
|
data = ''.join('%02X' % (i % 256) for i in range(700)) # TLV = 704 bytes
|
|
seq = _cap_apdu_sequence('A00000010001', 'A000000100', data, block_size=100)
|
|
self.assertEqual(len(seq), 10) # INSTALL + 8 LOAD + INSTALL
|
|
loads = seq[1:-1]
|
|
self.assertEqual(len(loads), 8)
|
|
for i, apdu in enumerate(loads):
|
|
self.assertEqual(apdu[:8], '80E8%s%02X' % ('80' if i == 7 else '00', i))
|
|
def payload(apdu):
|
|
lc = int(apdu[8:10], 16)
|
|
return apdu[10:10 + lc * 2]
|
|
joined = ''.join(payload(a) for a in loads)
|
|
self.assertEqual(len(joined) // 2, 704) # C4 82 02BC + 700 data bytes
|
|
self.assertTrue(joined.startswith('C482'))
|
|
self.assertEqual(int(loads[0][8:10], 16), 100)
|
|
self.assertEqual(int(loads[-1][8:10], 16), 4) # 704 = 7*100 + 4
|
|
|
|
def test_gen_install_returns_the_apdu_list(self):
|
|
# /api/scp81/gen-install: build the INSTALL/LOAD/INSTALL list for a
|
|
# .cap without touching any listener or script state.
|
|
from pysim_otaman_server.server import _scp81_gen_install
|
|
resp = _scp81_gen_install({'cap_hex': self._mini_cap(), 'privileges': '01'})
|
|
self.assertTrue(resp['ok'], resp)
|
|
self.assertEqual(resp['load_file_aid'], 'A00000010001')
|
|
self.assertEqual(resp['module_aid'], 'A000000100')
|
|
self.assertEqual(len(resp['apdus']), 3)
|
|
self.assertTrue(resp['apdus'][0].startswith('80E60200'))
|
|
self.assertTrue(resp['apdus'][1].startswith('80E88000'))
|
|
self.assertTrue(resp['apdus'][2].startswith('80E60C00'))
|
|
self.assertNotIn('queued', resp) # generation only, no queueing
|
|
|
|
def test_gen_install_rejects_bad_input(self):
|
|
from pysim_otaman_server.server import _scp81_gen_install
|
|
self.assertFalse(_scp81_gen_install({})['ok'])
|
|
resp = _scp81_gen_install({'cap_hex': '00'})
|
|
self.assertFalse(resp['ok'])
|
|
self.assertIn('cap parse failed', resp['error'])
|