diff --git a/docs/api.md b/docs/api.md index 400dd57..ca4d609 100644 --- a/docs/api.md +++ b/docs/api.md @@ -496,6 +496,24 @@ Returns the BIP/TLS event log (open/close, SEND/RECEIVE DATA hex, TLS handshake and HTTP request/response records). `?after=` returns only newer entries; `seq` echoes the latest sequence number. +### `POST /api/scp81/ram-install` + +Queue a RAM (GP) install as the SCP81 command script. The `.cap` is parsed +server-side (same parser as `/api/ram-install`) and expanded to the APDU +sequence INSTALL [for load] -> LOAD blocks (240-byte payloads) -> INSTALL +[for install]; the list runs on the card's next POST, one C-APDU per request. + +```json +{"cap_hex": "504B0304...", "sd_aid": "A000000003000000", "privileges": "00", + "install_params": "", "stk_params": "", "make_selectable": true, "force": false} +``` + +`sd_aid` empty = the ISD. Refused while a script is mid-run unless `force` is +true. Responds with `{"ok": true, "queued": true, "apdus": N, "load_file_aid": +..., "module_aid": ...}`; the results appear in `/api/scp81/script` and the +R-APDU log. `GET /api/scp81/script` reports the script `kind` +(`explore`/`none`/`custom`/`ram-install`). + ### `GET /api/scp81/script` Returns the active command script and the R-APDUs collected so far: diff --git a/docs/scp81-findings.md b/docs/scp81-findings.md index 31f7116..473bb11 100644 --- a/docs/scp81-findings.md +++ b/docs/scp81-findings.md @@ -202,8 +202,11 @@ Full session: 7/7 commands, all `X-Admin-Script-Status: ok`. 1. **UI:** group the per-page R-APDUs under their logical command in the SCP81 tab (page merging/decoding for ELF and application listings); expose the framing options in the tab. -2. **Load/store over SCP81:** RAM INSTALL/LOAD via command scripts using the - same recipe (one C-APDU per POST, pagination for long responses). +2. **Load/store over SCP81:** implemented - `POST /api/scp81/ram-install` + takes a `.cap`, expands it with the shared `_cap_apdu_sequence` helper + (INSTALL [for load] -> 240-byte LOAD blocks -> INSTALL [for install]) and + queues it as the command script, one C-APDU per POST. Live install test + pending (needs a push with a suitable applet). ## Tooling diff --git a/frontend/index.html b/frontend/index.html index 5da717a..d51b445 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -18,7 +18,7 @@
-

OTAMan SIM OTA with a Human Face v2.1.7

+

OTAMan SIM OTA with a Human Face v2.1.8

● @@ -997,6 +997,17 @@
+
+
+ + +
+
+ + +
+ +
The key is only sent to the local server and is never stored or logged.
@@ -7251,6 +7262,32 @@ async function scp81Start() { } } +async function scp81QueueCapInstall() { + const fileInput = document.getElementById('scp81-cap-file'); + const file = fileInput.files[0]; + if (!file) { scp81Msg(t('Select a .cap file'), 'text-red-500'); return; } + if (file.size > 48 * 1024) { scp81Msg(t('CAP file exceeds 48 kB limit'), 'text-red-500'); return; } + scp81Msg(t('Reading CAP file...'), 'text-gray-500'); + try { + const capHex = await ramReadFileHex(file); + const resp = await pysimFetch('/api/scp81/ram-install', { + cap_hex: capHex, + sd_aid: (document.getElementById('scp81-cap-sd').value || '').replace(/[^0-9a-fA-F]/g, ''), + }); + if (resp.ok) { + scp81Msg(t('Queued') + ': ' + resp.apdus + ' ' + t('APDUs') + ' (' + + t('INSTALL / LOAD / INSTALL') + ') - ' + t('send a push to run it'), + 'text-emerald-600 dark:text-emerald-400'); + } else { + scp81Msg(resp.error || t('Error'), 'text-red-500'); + } + scp81ResultsRefresh(); + scp81LogRefresh(); + } catch (err) { + scp81Msg(String(err.message || err), 'text-red-500'); + } +} + async function scp81Stop() { try { await pysimFetch('/api/scp81/bip', { action: 'stop' }); @@ -9723,6 +9760,13 @@ const LANG_RU = { 'cascade': 'каскадно', 'Select a card preset with keys first (RAM subtab → Card preset)': 'Сначала выберите пресет карты с ключами (RAM → Пресет карты)', 'Select a .cap file': 'Выберите файл .cap', + 'RAM install (CAP file)': 'RAM-установка (файл CAP)', + 'Queue CAP install': 'Поставить установку CAP в очередь', + 'Queued': 'В очереди', + 'APDUs': 'APDU', + 'INSTALL / LOAD / INSTALL': 'INSTALL / LOAD / INSTALL', + 'send a push to run it': 'отправьте push для запуска', + 'Script results (R-APDUs)': 'Результаты скрипта (R-APDU)', 'CAP file exceeds 48 kB limit': 'Файл CAP превышает лимит 48 кБ', 'Reading CAP file...': 'Чтение файла CAP...', 'Sending to server for install...': 'Отправка на сервер для установки...', diff --git a/frontend/sw.js b/frontend/sw.js index ffc7ca7..befdee6 100644 --- a/frontend/sw.js +++ b/frontend/sw.js @@ -1,4 +1,4 @@ -const CACHE = 'otaman-v144'; +const CACHE = 'otaman-v145'; const URLS = [ 'index.html', 'help.html', diff --git a/pyproject.toml b/pyproject.toml index af00cf6..af0602c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "pysim-otaman-server" -version = "2.1.7" +version = "2.1.8" description = "HTTP REST server wrapping pysim for the OTAMan PWA" requires-python = ">=3.8" # pysim is a git-only dependency installed explicitly by setup.bat/setup.sh. diff --git a/pysim_otaman_server/server.py b/pysim_otaman_server/server.py index 12d2024..9a2ede5 100644 --- a/pysim_otaman_server/server.py +++ b/pysim_otaman_server/server.py @@ -21,7 +21,7 @@ from osmocom.construct import GsmOrUcs2Adapter from osmocom.tlv import BER_TLV_IE -VERSION = '2.1.7' +VERSION = '2.1.8' MAX_ENVELOPE_SEGMENTS = 5 # max SMS segments for outgoing C-APDU in ENVELOPE @@ -403,6 +403,34 @@ def _encode_scts(dt=None): ]) +def _cap_apdu_sequence(loadfile_aid, module_aid, loadfile_data, sd_aid='', + privileges='00', install_params='', stk_params='', + make_selectable=True, block_size=240, instance_aid=None): + """RAM (GP) APDU sequence for a parsed .cap: INSTALL [for load], LOAD + blocks (240-byte payloads, block counter in P2, last block P1=0x80), + INSTALL [for install]. Shared by the SCP80 delivery path and the SCP81 + command-script path; keep byte-compatible with /api/ram-install.""" + sd = sd_aid or 'A000000003000000' + ifl_data = _lv(loadfile_aid) + _lv(sd) + '00' + '00' + '00' + apdus = ['80E60200%02X%s00' % (len(ifl_data) // 2, ifl_data)] + loadfile_tlv = 'C4' + _ber_len(len(loadfile_data) // 2) + loadfile_data + total_bytes = len(loadfile_tlv) // 2 + blocks = [loadfile_tlv[i * 2:(i + block_size) * 2] + for i in range(0, (total_bytes + block_size - 1) // block_size)] + for i, block in enumerate(blocks): + p1 = 0x80 if i == len(blocks) - 1 else 0x00 + apdus.append('80E8%02X%02X%02X%s00' % (p1, i % 256, len(block) // 2, block)) + instance = instance_aid or module_aid + params = install_params if install_params else 'C900' + if stk_params: + params += stk_params + p1_install = 0x0C if make_selectable else 0x04 + ifi_data = (_lv(loadfile_aid) + _lv(module_aid) + _lv(instance) + + _lv(privileges or '00') + _lv(params) + '00') + apdus.append('80E6%02X00%02X%s00' % (p1_install, len(ifi_data) // 2, ifi_data)) + return apdus + + def _lv(hex_str): """Length-prefix a hex string (1-byte length).""" n = len(hex_str) // 2 @@ -1366,6 +1394,26 @@ _SCP81_SCRIPT_RESULTS = [] _SCP81_SCRIPT_INSERTED = [] _SCP81_PAGES = 0 SCP81_MAX_PAGES = 24 +# What the queued script is ('explore', 'none', 'custom' or 'ram-install'). +_SCP81_SCRIPT_KIND = 'explore' + + +def _scp81_queue_script(apdus, kind='custom', force=False): + """Replace the SCP81 command script with a new APDU list. Refuses while + a script is mid-run unless forced; the list runs on the card's next POST.""" + global _SCP81_SCRIPT, _SCP81_SCRIPT_SENT, _SCP81_SCRIPT_RESULTS + global _SCP81_SCRIPT_INSERTED, _SCP81_PAGES, _SCP81_SCRIPT_KIND + if not force and 0 < _SCP81_SCRIPT_SENT < len(_SCP81_SCRIPT): + return {'queued': False, 'reason': 'script in progress', + 'sent': _SCP81_SCRIPT_SENT, 'of': len(_SCP81_SCRIPT)} + _SCP81_SCRIPT = [a.upper().replace(' ', '') for a in apdus] + _SCP81_SCRIPT_KIND = kind + _SCP81_SCRIPT_SENT = 0 + _SCP81_SCRIPT_RESULTS = [] + _SCP81_SCRIPT_INSERTED = [] + _SCP81_PAGES = 0 + _BIP.log('script-queued', script_kind=kind, apdus=len(_SCP81_SCRIPT)) + return {'queued': True, 'apdus': len(_SCP81_SCRIPT)} _SCP81_SCRIPT_TEMPLATE = 'indefinite' _SCP81_SCRIPT_CR_TAG = False # None = short per-command Next-URI ('/N'); '' = omit the header (spec: the @@ -1583,6 +1631,7 @@ def _scp81_response_headers(): def _scp81_bip_control(body): global _SCP81_LISTENER, _SCP81_PSK global _SCP81_SCRIPT, _SCP81_SCRIPT_SENT, _SCP81_SCRIPT_RESULTS + global _SCP81_SCRIPT_INSERTED, _SCP81_PAGES, _SCP81_SCRIPT_KIND global _SCP81_SCRIPT_TEMPLATE, _SCP81_SCRIPT_CR_TAG, _SCP81_NEXT_URI global _SCP81_LINK_EVENTS, _SCP81_TARGETED_APP, _SCP81_APACHE_HEADERS global _SCP81_CHUNKED @@ -1621,12 +1670,16 @@ def _scp81_bip_control(body): script = body.get('script', 'explore') if isinstance(script, list): _SCP81_SCRIPT = [re.sub(r'\s', '', s) for s in script if s] + _SCP81_SCRIPT_KIND = 'custom' elif script in _SCP81_SCRIPTS: _SCP81_SCRIPT = list(_SCP81_SCRIPTS[script]) + _SCP81_SCRIPT_KIND = script else: return {'ok': False, 'error': 'unknown script preset: %s' % script} _SCP81_SCRIPT_SENT = 0 _SCP81_SCRIPT_RESULTS = [] + _SCP81_SCRIPT_INSERTED = [] + _SCP81_PAGES = 0 template = body.get('script_template', 'indefinite') if template not in ('indefinite', 'definite'): return {'ok': False, 'error': 'script_template must be indefinite or definite'} @@ -2731,6 +2784,7 @@ class PysimHandler(BaseHTTPRequestHandler): elif self.path == '/api/scp81/script': self._log_req() resp = {'script': _SCP81_SCRIPT, 'sent': _SCP81_SCRIPT_SENT, + 'kind': _SCP81_SCRIPT_KIND, 'template': _SCP81_SCRIPT_TEMPLATE, 'cr_tag': _SCP81_SCRIPT_CR_TAG, 'results': _SCP81_SCRIPT_RESULTS} self._send_json(resp) @@ -3456,53 +3510,29 @@ class PysimHandler(BaseHTTPRequestHandler): if submit_handler and hasattr(scc, '_tp'): scc._tp.proactive_handler = old_proactive - # Step 1: INSTALL [for load] - sys.stderr.write('RAM-INSTALL: Step 1 — INSTALL [for load] loadfile_aid=%s\n' % loadfile_aid) - ifl_data = _lv(loadfile_aid) + _lv(sd_aid) + '00' + '00' + '00' - ifl_apdu = '80E60200%02X%s00' % (len(ifl_data) // 2, ifl_data) - if not _send_gp_apdu(ifl_apdu, 'INSTALL [for load]'): - resp = {'success': False, 'steps': steps, 'failed_step': len(steps), - 'error': 'INSTALL [for load] failed', 'load_file_aid': loadfile_aid, 'module_aid': module_aid} - self._send_json(resp) - self._log_resp(resp) - return - - # Step 2: LOAD blocks - loadfile_tlv = 'C4' + _ber_len(len(loadfile_data) // 2) + loadfile_data - block_size = 240 - total_bytes = len(loadfile_tlv) // 2 - blocks = [loadfile_tlv[i * 2:(i + block_size) * 2] for i in range(0, (total_bytes + block_size - 1) // block_size)] - sys.stderr.write('RAM-INSTALL: Step 2 — LOAD %d bytes in %d blocks\n' % (total_bytes, len(blocks))) - for block_idx, block in enumerate(blocks): - is_last = (block_idx == len(blocks) - 1) - p1 = 0x80 if is_last else 0x00 - p2 = block_idx % 256 - load_apdu = '80E8%02X%02X%02X%s00' % (p1, p2, len(block) // 2, block) - if not _send_gp_apdu(load_apdu, 'LOAD (%d/%d)' % (block_idx + 1, len(blocks))): + # INSTALL [for load] -> LOAD blocks -> INSTALL [for install] + seq = _cap_apdu_sequence( + loadfile_aid, module_aid, loadfile_data, sd_aid=sd_aid, + privileges=privileges_hex, + install_params=install_params_hex, stk_params=stk_params_hex, + make_selectable=make_selectable) + sys.stderr.write('RAM-INSTALL: %d APDUs (INSTALL / %d x LOAD / INSTALL) loadfile_aid=%s\n' % ( + len(seq), len(seq) - 2, loadfile_aid)) + for apdu_idx, gp_apdu in enumerate(seq): + if apdu_idx == 0: + step_name = 'INSTALL [for load]' + elif apdu_idx == len(seq) - 1: + step_name = 'INSTALL [for install]' + else: + step_name = 'LOAD (%d/%d)' % (apdu_idx, len(seq) - 2) + if not _send_gp_apdu(gp_apdu, step_name): resp = {'success': False, 'steps': steps, 'failed_step': len(steps), - 'error': 'LOAD block %d failed' % (block_idx + 1), + 'error': '%s failed' % step_name, 'load_file_aid': loadfile_aid, 'module_aid': module_aid} self._send_json(resp) self._log_resp(resp) return - # Step 3: INSTALL [for install] - sys.stderr.write('RAM-INSTALL: Step 3 — INSTALL [for install]\n') - instance_aid = module_aid - privileges = privileges_hex - inst_params = install_params_hex if install_params_hex else 'C900' - if stk_params_hex: - inst_params += stk_params_hex - p1_install = 0x0C if make_selectable else 0x04 - ifi_data = _lv(loadfile_aid) + _lv(module_aid) + _lv(instance_aid) + _lv(privileges) + _lv(inst_params) + '00' - ifi_apdu = '80E6%02X00%02X%s00' % (p1_install, len(ifi_data) // 2, ifi_data) - if not _send_gp_apdu(ifi_apdu, 'INSTALL [for install]'): - resp = {'success': False, 'steps': steps, 'failed_step': len(steps), - 'error': 'INSTALL [for install] failed', 'load_file_aid': loadfile_aid, 'module_aid': module_aid} - self._send_json(resp) - self._log_resp(resp) - return - resp = {'success': True, 'steps': steps, 'load_file_aid': loadfile_aid, 'module_aid': module_aid, 'final_cntr': cntr} sys.stderr.write('RAM-INSTALL: Complete — loadfile_aid=%s module_aid=%s cntr=%s\n' % ( @@ -3524,6 +3554,36 @@ class PysimHandler(BaseHTTPRequestHandler): resp = {'ok': False, 'error': str(e)} self._send_json(resp) self._log_resp(resp) + elif self.path == '/api/scp81/ram-install': + body = self._read_body() + self._log_req(body) + cap_hex = (body.get('cap_hex') or '').replace(' ', '') + if not cap_hex: + resp = {'ok': False, 'error': 'No cap_hex provided'} + elif _SCP81_LISTENER is None: + resp = {'ok': False, 'error': 'SCP81 listener is not running'} + else: + try: + loadfile_aid, module_aid, loadfile_data = _cap_parse(cap_hex) + seq = _cap_apdu_sequence( + loadfile_aid, module_aid, loadfile_data, + sd_aid=(body.get('sd_aid') or '').replace(' ', ''), + privileges=(body.get('privileges') or '').replace(' ', '') or '00', + install_params=(body.get('install_params') or '').replace(' ', ''), + stk_params=(body.get('stk_params') or '').replace(' ', ''), + make_selectable=bool(body.get('make_selectable', True))) + queued = _scp81_queue_script(seq, kind='ram-install', + force=bool(body.get('force', False))) + resp = dict(queued, ok=bool(queued.get('queued')), + load_file_aid=loadfile_aid, + module_aid=module_aid, apdus=len(seq)) + if queued.get('queued'): + resp['note'] = ('queued as the SCP81 command script; ' + 'runs on the card next POST (push/trigger)') + except Exception as e: + resp = {'ok': False, 'error': 'cap parse failed: %s' % e} + self._send_json(resp) + self._log_resp(resp) elif self.path == '/api/scp81/log-clear': body = self._read_body() self._log_req(body) diff --git a/tests/test_ota_helpers.py b/tests/test_ota_helpers.py index 710c172..2304477 100644 --- a/tests/test_ota_helpers.py +++ b/tests/test_ota_helpers.py @@ -836,3 +836,53 @@ class TestSmsReassembly(unittest.TestCase): 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 + data = 'AB' * 700 # 700 bytes -> C4 TLV 703 -> 3 x 240-byte blocks + 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 diff --git a/tests/test_scp81.py b/tests/test_scp81.py index 2a093fb..035ceb7 100644 --- a/tests/test_scp81.py +++ b/tests/test_scp81.py @@ -755,3 +755,38 @@ class TargetedAppTest(unittest.TestCase): self.assertEqual(rapdus[0][1], '6A88') # A status-only POST (no body, e.g. unknown-application) parses empty. self.assertEqual(server._scp81_parse_response(b''), (0, [])) + + +class QueueScriptTest(unittest.TestCase): + def test_queue_replaces_and_resets(self): + server._SCP81_SCRIPT = ['80CAFF2100'] + server._SCP81_SCRIPT_SENT = 1 + server._SCP81_SCRIPT_RESULTS = [{'index': 1, 'sw': '9000', 'apdu': '80CAFF2100', 'rapdu': ''}] + try: + r = server._scp81_queue_script(['80E6020013' + '00' * 20, '80E88000' + '00' * 4], + kind='ram-install') + self.assertTrue(r['queued']) + self.assertEqual(server._SCP81_SCRIPT_KIND, 'ram-install') + self.assertEqual(server._SCP81_SCRIPT_SENT, 0) + self.assertEqual(server._SCP81_SCRIPT_RESULTS, []) + self.assertEqual(len(server._SCP81_SCRIPT), 2) + finally: + server._SCP81_SCRIPT = list(server._SCP81_SCRIPTS['explore']) + server._SCP81_SCRIPT_SENT = 0 + server._SCP81_SCRIPT_RESULTS = [] + server._SCP81_SCRIPT_KIND = 'explore' + + def test_queue_refuses_while_running(self): + server._SCP81_SCRIPT = ['80CAFF2100', '80F28002024F0000'] + server._SCP81_SCRIPT_SENT = 1 + try: + r = server._scp81_queue_script(['80E60200'], kind='ram-install') + self.assertFalse(r['queued']) + self.assertEqual(r['sent'], 1) + r = server._scp81_queue_script(['80E60200'], kind='ram-install', force=True) + self.assertTrue(r['queued']) + self.assertEqual(server._SCP81_SCRIPT, ['80E60200']) + finally: + server._SCP81_SCRIPT = list(server._SCP81_SCRIPTS['explore']) + server._SCP81_SCRIPT_SENT = 0 + server._SCP81_SCRIPT_KIND = 'explore'