catarrh/simple
1
0
Fork 0
Code Issues Pull Requests Actions 8 Packages Projects Releases Wiki Activity
Files
cc2edb3d8aedcf811ec15d5fc0c71ad56be59274
simple/tests/test_fetch_routes.py
T
catarrh cc2edb3d8a feat: one keyset per key version - indexed keys and counters (v3.8.0) 
A card usually holds several OTA key sets, and the SCP80 counter is tracked
per key version on the card, so a preset now carries one keyset per
GlobalPlatform key version - the b8..b5 nibble of KIc/KID (TS 102 225
5.1.2/A.2), the version is implicit in the bytes.  The spec is explicit:
"a dedicated counter shall be associated to each key version" (Annex A.1) and
"the key version number indicated in the KIc and KID (bits b8..b5) fields
shall be identical when different from 0" - a mismatch is rejected by the card
with "Unidentified security error" (A.2).

Server (presets.py, schema 2):
- a preset carries `keysets: [{kic, kid, kicKey, kidKey, cntr}]`; validation
  refuses a KIc/KID version mismatch (A.2), a version outside 01-0F ('00'
  means "no security" and is a packet-level choice, not a keyset), duplicate
  versions and missing keys/counters; a v3.8.0 flat preset converts into one
  keyset on load and on import;
- `set_counter(..., kvn)` writes the counter of that keyset (monotonic per
  keyset), the audit line carries the kvn, and `find_keyset()` serves the
  lookups;
- the operations derive the key version from the request's KIc byte and
  persist into that keyset: /api/send-ota, the RAM install/install-app/compat
  steps (_ram_send_gp_apdu) and the test runner's SCP80 steps (which may name
  a `kvn` per step and keep one working counter per keyset);
- a non-zero key version the named preset does not define, or a KIc/KID
  mismatch, is refused before anything is built (_preset_keyset_check /
  _keyset_guard: 400, or a step error in the test runner).  Without a
  preset_id only the mismatch rule applies.

PWA:
- the Cards form's KIc/KID/counter/keys inputs become a keyset editor (one row
  per version: KIc, KID, both keys, that version's counter, Add/Remove; the
  version is shown derived from the bytes and a mismatch is flagged);
- the SCP80 and RAM views get a Key-version selector beside the preset
  selector, driving the KIc/KID bytes and loading that keyset's keys and
  counter; the packet builder refuses a version the preset does not define
  (spKeysetCheck/spKeysetGuard, pure and unit-tested);
- the counter syncs are per keyset (spCntrLocalSync/ramSaveCntr), the SCP80
  form's counter edit posts the keysets, and the test-run write-back syncs
  every counter the server reports (scp80_counters).

Docs/help/AGENTS updated; sw cache simple-v335.  Tests: 610 Python (store
keysets, endpoint refusals, the counter landing in the right keyset) and 726
Node (keysets.test.js plus the updated harnesses).
2026-09-30 15:15:21 +03:00

665 lines
28 KiB
Python
Raw Blame History

"""PWA <-> server HTTP contract tests.
`pysimFetch(path, body)` sends POST whenever a body is passed and GET
otherwise; the server's routes live in `PysimHandler._do_GET` / `_do_POST`.
The test-script endpoints were registered in `_do_GET` while the PWA POSTs
to them, so every Start answered 404 "Not found" (live 2026-09-29; present
since the phase-1 commit and never exercised over HTTP). The matrix below
is the guard for that contract; the smoke test boots the real handler and
checks the three endpoints.
"""
import json
import pathlib
import re
import threading
import unittest
import urllib.error
import urllib.request
from unittest import mock
from pysim_simple_server.__main__ import _build_http_server
from pysim_simple_server import presets
from pysim_simple_server.server import PysimHandler, _CARD_FREE_GET, _CARD_LOCK
ROOT = pathlib.Path(__file__).resolve().parents[1]
def _pwa_calls():
"""[(path, has_body)] for every literal pysimFetch()/pysimFetchOk() call in
the PWA."""
html = (ROOT / 'frontend' / 'index.html').read_text(encoding='utf-8')
return [(m.group(1).split('?')[0], m.group(2) == ',')
for m in re.finditer(r"pysimFetch(?:Ok)?\(\s*'([^']+)'\s*(,|\))", html)]
def _server_routes():
"""((GET exact, GET prefixes), (POST exact, POST prefixes)) of /api paths."""
src = (ROOT / 'pysim_simple_server' / 'server.py').read_text(encoding='utf-8')
get_seg = src[src.index('def _do_GET('):src.index('def do_POST(')]
post_seg = src[src.index('def _do_POST('):src.index('def log_message(')]
def routes(seg):
exact = set(re.findall(r"self\.path == '(/api/[^']+)'", seg))
for group in re.findall(r"self\.path in \(([^)]*)\)", seg):
exact |= set(re.findall(r"'(/api/[^']+)'", group))
prefixes = set(re.findall(r"self\.path\.startswith\('(/api/[^']+)'\)", seg))
return exact, prefixes - {'/api/'}
return routes(get_seg), routes(post_seg)
class FetchRouteMatrixTests(unittest.TestCase):
def test_every_fetch_matches_a_route_with_the_right_method(self):
(get, get_pref), (post, post_pref) = _server_routes()
self.assertTrue(get and post, 'route extraction found nothing')
def in_routes(path, exact, prefixes):
return path in exact or any(path.startswith(p) for p in prefixes)
for path, has_body in _pwa_calls():
with self.subTest(path=path, post=has_body):
if has_body:
self.assertTrue(in_routes(path, post, post_pref),
'%s is POSTed (body) but has no POST route' % path)
else:
self.assertTrue(
in_routes(path, get, get_pref) or in_routes(path, post, post_pref),
'%s is fetched without a body but has no GET/POST route' % path)
class TestScriptHttpTests(unittest.TestCase):
"""The test-script endpoints must answer on POST (the PWA's method)."""
def setUp(self):
self.server = _build_http_server('127.0.0.1', 0, PysimHandler)
self.server.log_requests = False
self.server.app = None
threading.Thread(target=self.server.serve_forever, daemon=True).start()
self.port = self.server.server_address[1]
def tearDown(self):
self.server.shutdown()
self.server.server_close()
def _post(self, path, body):
req = urllib.request.Request(
'http://127.0.0.1:%d%s' % (self.port, path),
data=json.dumps(body).encode(),
headers={'Content-Type': 'application/json'}, method='POST')
try:
with urllib.request.urlopen(req, timeout=5) as res:
return res.status, json.loads(res.read() or b'{}')
except urllib.error.HTTPError as e:
return e.code, json.loads(e.read() or b'{}')
def test_start_stop_clear_answer_on_post(self):
# an empty script is invalid, but it proves the route exists (400)
status, resp = self._post('/api/test/run', {})
self.assertEqual(status, 400, resp)
self.assertIn('error', resp)
status, resp = self._post('/api/test/stop', {})
self.assertEqual(status, 200, resp)
status, resp = self._post('/api/test/clear', {})
self.assertEqual(status, 200, resp)
def test_the_test_endpoints_are_not_get_routes(self):
with self.assertRaises(urllib.error.HTTPError) as cm:
urllib.request.urlopen('http://127.0.0.1:%d/api/test/run' % self.port, timeout=5)
self.assertEqual(cm.exception.code, 404)
cm.exception.close()
if __name__ == '__main__':
unittest.main()
class CardFreeGetTests(unittest.TestCase):
"""Cached GETs must answer while the card lock is held.
During a long RAM install the UI polls several cached endpoints every few
seconds; while they queued behind the install they ate the browser's
per-host connection budget and starved /api/status - the modal's progress
bar froze around 70% and jumped to 100% at the end (live 2026-09-29).
"""
def setUp(self):
self.server = _build_http_server('127.0.0.1', 0, PysimHandler)
self.server.log_requests = False
self.server.app = None
self.server.sl = None
self.server.scc = None
self.server.stk_pending = None
self.server.menu_active = False
self.server.sim_menu = None
self.server.event_list = None
self.server.terminal_profile = None
self.server.cli_terminal_profile = None
threading.Thread(target=self.server.serve_forever, daemon=True).start()
self.port = self.server.server_address[1]
def tearDown(self):
self.server.shutdown()
self.server.server_close()
def _get(self, path, timeout=2):
return urllib.request.urlopen(
'http://127.0.0.1:%d%s' % (self.port, path), timeout=timeout)
def test_cached_endpoints_answer_while_the_card_lock_is_held(self):
with _CARD_LOCK:
for path in ('/api/status', '/api/stk-status', '/api/poll-status',
'/api/proactive-log', '/api/menu', '/api/events'):
with self.subTest(path=path):
with self._get(path) as res:
self.assertEqual(res.status, 200)
def test_card_touching_gets_stay_locked(self):
# cardinfo runs a card command; the ES10 endpoints select/STORE DATA
for path in ('/api/cardinfo', '/api/esim/chip', '/api/esim/profiles'):
self.assertNotIn(path, _CARD_FREE_GET)
def _mini_cap_hex():
"""A minimal parseable CAP archive (Header + Applet + one Import)."""
import io as _io
import struct as _struct
import zipfile as _zipfile
def comp(tag, payload):
return bytes([tag]) + _struct.pack('>H', len(payload)) + payload
aid = bytes.fromhex('0102030405')
header = comp(0x01, _struct.pack('>I', 0xDECAFFED) + bytes([1, 2, 0, 1, 2]) +
bytes([len(aid)]) + aid)
applet = comp(0x03, bytes([1, len(aid)]) + aid + b'\x00\x01')
imports = comp(0x04, bytes([1]) + bytes([0, 1, 7]) + bytes.fromhex('A0000000620101'))
buf = _io.BytesIO()
with _zipfile.ZipFile(buf, 'w') as z:
z.writestr('Header.cap', header)
z.writestr('Applet.cap', applet)
z.writestr('Import.cap', imports)
return buf.getvalue().hex().upper()
class CapCompatHttpTests(unittest.TestCase):
"""POST /api/cap-compat exists (the CAP compatibility test, v3.6.48)."""
def setUp(self):
self.server = _build_http_server('127.0.0.1', 0, PysimHandler)
self.server.log_requests = False
self.server.app = None
self.server.sl = None
self.server.scc = None
threading.Thread(target=self.server.serve_forever, daemon=True).start()
self.port = self.server.server_address[1]
def tearDown(self):
self.server.shutdown()
self.server.server_close()
def _post(self, path, body):
req = urllib.request.Request(
'http://127.0.0.1:%d%s' % (self.port, path),
data=json.dumps(body).encode(),
headers={'Content-Type': 'application/json'}, method='POST')
try:
with urllib.request.urlopen(req, timeout=5) as res:
return res.status, json.loads(res.read() or b'{}')
except urllib.error.HTTPError as e:
return e.code, json.loads(e.read() or b'{}')
def test_cap_compat_route_exists(self):
# without a card session it answers 503 (route present), not 404
status, resp = self._post('/api/cap-compat', {'cap_hex': '00'})
self.assertEqual(status, 503, resp)
self.assertIn('error', resp)
def test_cap_compat_rejects_a_bad_probe(self):
# a parseable CAP + a bad override -> 400 from the probe validation,
# before any card I/O (a dummy scc object is enough)
self.server.scc = object()
status, resp = self._post('/api/cap-compat',
{'cap_hex': _mini_cap_hex(), 'probe_imports': {'all': 'x'}})
self.assertEqual(status, 400, resp)
self.assertIn('import probe', resp['error'])
def test_cap_compat_reports_unmatched_probe_lines_with_versions(self):
# v3.6.52: probe overrides that match no Import entry come back as a
# per-AID list ({aid, version}) for the modal's detail block.
from pysim_simple_server import server as srv
self.server.scc = object()
patches = [
mock.patch.object(srv, '_ram_detect_format', lambda *a, **k: 'compact'),
mock.patch.object(srv, '_ram_send_gp_apdu', lambda *a, **k: True),
]
for p in patches:
p.start()
try:
status, resp = self._post('/api/cap-compat', {
'cap_hex': _mini_cap_hex(),
'probe_imports': {'A0000000620101': '0.0', 'dead beef': '1.2'},
})
finally:
for p in reversed(patches):
p.stop()
self.assertEqual(status, 200, resp)
self.assertTrue(resp['success'], resp)
self.assertEqual(resp['probe_unmatched'], [{'aid': 'DEADBEEF', 'version': '1.2'}])
self.assertEqual(resp['probe_imports'],
[{'aid': 'A0000000620101', 'from': '1.0', 'to': '0.0'}])
class RamInstallNvFootprintHttpTests(unittest.TestCase):
"""The RAM install response carries the measured NV footprint (v3.6.51):
the server reads GET DATA FF21 before and after the chain (best effort)
and reports the free-NV values and their delta."""
def setUp(self):
from pysim_simple_server import server as srv
self.srv = srv
self.server = _build_http_server('127.0.0.1', 0, PysimHandler)
self.server.log_requests = False
self.server.app = None
self.server.sl = None
self.server.scc = object()
self.server.sms_oa = '12345'
self.server.sms_sc = '12345678912'
threading.Thread(target=self.server.serve_forever, daemon=True).start()
self.port = self.server.server_address[1]
def tearDown(self):
self.server.shutdown()
self.server.server_close()
def _post(self, path, body):
req = urllib.request.Request(
'http://127.0.0.1:%d%s' % (self.port, path),
data=json.dumps(body).encode(),
headers={'Content-Type': 'application/json'}, method='POST')
try:
with urllib.request.urlopen(req, timeout=5) as res:
return res.status, json.loads(res.read() or b'{}')
except urllib.error.HTTPError as e:
return e.code, json.loads(e.read() or b'{}')
def _install(self, fail_at=None, nv_before=50000, nv_after=20000):
def fake_send(server, scc, sp, state, name, apdu, silent=False):
if fail_at == name:
state['steps'].append({'name': name, 'por_status': 'por_error', 'por_sw': '6A88'})
return False
state['steps'].append({'name': name, 'por_status': 'por_ok', 'por_sw': '9000'})
return True
def fake_read(server, scc, sp, state, ram_format, label):
return nv_before if label == 'NV before' else nv_after
patches = [
mock.patch.object(self.srv, '_cap_parse',
lambda cap_hex: ('F0414C46416101', 'F0414C4641610101', b'\x01\x02')),
mock.patch.object(self.srv, '_cap_apdu_sequence',
lambda *a, **k: ['80E60200', '80E80000', '80E60000']),
mock.patch.object(self.srv, '_ram_detect_format', lambda *a, **k: 'compact'),
mock.patch.object(self.srv, '_ram_send_gp_apdu', fake_send),
mock.patch.object(self.srv, '_ram_read_ff21', fake_read),
]
for p in patches:
p.start()
try:
return self._post('/api/ram-install', {'cap_hex': 'AA'})
finally:
for p in reversed(patches):
p.stop()
def test_success_reports_the_nv_delta(self):
status, resp = self._install()
self.assertEqual(status, 200, resp)
self.assertTrue(resp['success'], resp)
self.assertEqual(resp['nv_before'], 50000)
self.assertEqual(resp['nv_after'], 20000)
self.assertEqual(resp['nv_delta'], 30000)
def test_failed_install_still_reports_the_nv_delta(self):
status, resp = self._install(fail_at='LOAD (1/1)')
self.assertEqual(status, 200, resp)
self.assertFalse(resp['success'], resp)
self.assertEqual(resp['failed_step'], 2)
self.assertEqual(resp['nv_before'], 50000)
self.assertEqual(resp['nv_after'], 20000)
self.assertEqual(resp['nv_delta'], 30000)
def test_a_card_without_ff21_has_no_nv_fields(self):
status, resp = self._install(nv_before=None, nv_after=None)
self.assertEqual(status, 200, resp)
self.assertTrue(resp['success'], resp)
for key in ('nv_before', 'nv_after', 'nv_delta'):
self.assertNotIn(key, resp)
class CntrLowGuardHttpTests(unittest.TestCase):
"""The low-counter guard on the HTTP paths (v3.6.58): /api/send-ota
reports the card's verdict as data, and a RAM chain stops at it."""
def setUp(self):
from pysim_simple_server import server as srv
self.srv = srv
self.server = _build_http_server('127.0.0.1', 0, PysimHandler)
self.server.log_requests = False
self.server.app = None
self.server.sl = None
self.server.scc = object()
self.server.sms_oa = '12345'
self.server.sms_sc = '12345678912'
threading.Thread(target=self.server.serve_forever, daemon=True).start()
self.port = self.server.server_address[1]
def tearDown(self):
self.server.shutdown()
self.server.server_close()
def _post(self, path, body):
req = urllib.request.Request(
'http://127.0.0.1:%d%s' % (self.port, path),
data=json.dumps(body).encode(),
headers={'Content-Type': 'application/json'}, method='POST')
try:
with urllib.request.urlopen(req, timeout=5) as res:
return res.status, json.loads(res.read() or b'{}')
except urllib.error.HTTPError as e:
return e.code, json.loads(e.read() or b'{}')
def test_send_ota_reports_the_low_counter_verdict(self):
patches = [
mock.patch.object(self.srv, '_send_secured_packet', lambda *a, **k: {
'success': True, 'sw': '9000', 'bytes': 34, 'segments': 1,
'response_data': '027100000b0a00000000000000bb0002'}),
mock.patch.object(self.srv, '_decode_por', lambda *a, **k: {
'response_status': 'cntr_low', 'tar': '000000', 'cntr': '00000000BB',
'pcntr': 0, 'rpl': 11, 'rhl': 10, 'raw': ''}),
]
self.server.app = object() # the send path only checks truthiness
for p in patches:
p.start()
try:
status, resp = self._post('/api/send-ota', {
'sp': '00', 'spi1': '16', 'spi2': '01', 'kic': '15', 'kid': '15',
'tar': '000000', 'cntr': '00000000BB'})
finally:
for p in reversed(patches):
p.stop()
self.assertEqual(status, 200, resp)
self.assertTrue(resp['cntr_low'], resp)
self.assertEqual(resp['card_cntr'], '00000000BB')
self.assertEqual(resp['suggested_cntr'], '00000000BC')
def test_the_ram_chain_stops_at_a_low_counter(self):
calls = []
def fake_send(server, scc, sp, state, name, apdu, silent=False):
calls.append(name)
if len(calls) == 2:
# the card's verdict: stop - no further step may be sent
state['steps'].append({'name': name, 'por_status': 'cntr_low',
'por_cntr': '00000000BB'})
return False
state['steps'].append({'name': name, 'por_status': 'por_ok', 'por_sw': '9000'})
return True
patches = [
mock.patch.object(self.srv, '_cap_parse',
lambda cap_hex: ('F0414C46416101', 'F0414C4641610101', b'\x01\x02')),
mock.patch.object(self.srv, '_cap_apdu_sequence',
lambda *a, **k: ['80E60200', '80E80000', '80E80000', '80E60000']),
mock.patch.object(self.srv, '_ram_detect_format', lambda *a, **k: 'compact'),
mock.patch.object(self.srv, '_ram_send_gp_apdu', fake_send),
mock.patch.object(self.srv, '_ram_read_ff21', lambda *a, **k: None),
]
for p in patches:
p.start()
try:
status, resp = self._post('/api/ram-install', {'cap_hex': 'AA'})
finally:
for p in reversed(patches):
p.stop()
self.assertEqual(status, 200, resp)
self.assertFalse(resp['success'], resp)
self.assertEqual(len(calls), 2, calls) # nothing after the verdict
self.assertTrue(resp['cntr_low'], resp)
self.assertEqual(resp['card_cntr'], '00000000BB')
self.assertEqual(resp['suggested_cntr'], '00000000BC')
class PresetStoreHttpTests(unittest.TestCase):
"""The card preset endpoints (v3.8.0): the PWA's Cards tab talks to the
server store instead of localStorage, and the counter an operation
accepted is persisted server-side."""
def setUp(self):
import tempfile
from pysim_simple_server import presets
from pysim_simple_server import server as srv
self.srv = srv
self.tmp = tempfile.TemporaryDirectory()
self.store = presets.PresetStore(pathlib.Path(self.tmp.name) / 'card_presets.json')
self.server = _build_http_server('127.0.0.1', 0, PysimHandler)
self.server.log_requests = False
self.server.app = None
self.server.sl = None
self.server.scc = object()
self.server.sms_oa = '12345'
self.server.sms_sc = '12345678912'
self.server.card_presets = self.store
threading.Thread(target=self.server.serve_forever, daemon=True).start()
self.port = self.server.server_address[1]
def tearDown(self):
self.server.shutdown()
self.server.server_close()
self.tmp.cleanup()
def _post(self, path, body):
req = urllib.request.Request(
'http://127.0.0.1:%d%s' % (self.port, path),
data=json.dumps(body).encode(),
headers={'Content-Type': 'application/json'}, method='POST')
try:
with urllib.request.urlopen(req, timeout=5) as res:
return res.status, json.loads(res.read() or b'{}')
except urllib.error.HTTPError as e:
body = json.loads(e.read() or b'{}')
e.close()
return e.code, body
def _get(self, path, timeout=5):
try:
with urllib.request.urlopen(
'http://127.0.0.1:%d%s' % (self.port, path), timeout=timeout) as res:
return res.status, json.loads(res.read() or b'{}')
except urllib.error.HTTPError as e:
body = json.loads(e.read() or b'{}')
e.close()
return e.code, body
def _keyset(self, **kw):
ks = {'kic': '15', 'kid': '15', 'kicKey': 'AA', 'kidKey': 'BB'}
ks.update(kw)
return ks
def _preset(self, **kw):
p = {'name': 'C', 'keysets': [self._keyset()]}
p.update(kw)
return p
def test_the_get_route_is_card_free_and_reports_the_store(self):
self.assertIn('/api/presets', _CARD_FREE_GET)
with _CARD_LOCK:
status, resp = self._get('/api/presets')
self.assertEqual(status, 200, resp)
self.assertEqual(resp['presets'], [])
self.assertEqual(resp['path'], str(self.store.path))
def test_create_update_delete_round_trip(self):
status, resp = self._post('/api/presets', self._preset(iccid='8970119000004600098'))
self.assertEqual(status, 200, resp)
pid = resp['preset']['id']
self.assertTrue(pid)
self.assertEqual(resp['preset']['keysets'][0]['cntr'], '0000000001')
# a duplicate ICCID (raw-hex form) is refused
status, resp = self._post('/api/presets',
self._preset(name='Other', iccid='980711090000640090F8'))
self.assertEqual(status, 400, resp)
self.assertIn('already exists', resp['error'])
status, resp = self._post('/api/presets/update',
{'id': pid, 'fields': {'name': 'Renamed'}})
self.assertEqual(status, 200, resp)
self.assertEqual(resp['preset']['name'], 'Renamed')
self.assertEqual(resp['preset']['id'], pid)
status, resp = self._post('/api/presets/update', {'id': 'nope', 'fields': {}})
self.assertEqual(status, 404, resp)
status, resp = self._post('/api/presets/delete', {'id': pid})
self.assertEqual(status, 200, resp)
self.assertTrue(resp['removed'])
self.assertEqual(self.store.list(), [])
def test_import_accepts_the_old_localstorage_export(self):
status, resp = self._post('/api/presets/import', {'presets': [
{'name': 'A', 'iccid': '8970119000004600098', 'kic': '15', 'kid': '15',
'kicKey': 'AA', 'kidKey': 'BB', 'cntr': '0000000007'},
{'name': 'Broken'},
]})
self.assertEqual(status, 200, resp)
self.assertEqual(resp['added'], 1)
self.assertEqual(resp['skipped'], 1)
imported = self.store.find_by_iccid('8970119000004600098')
self.assertEqual(imported['keysets'][0]['cntr'], '0000000007')
self.assertEqual(presets.keyset_kvn(imported['keysets'][0]), 1)
status, resp = self._post('/api/presets/import', {'presets': 'nope'})
self.assertEqual(status, 400, resp)
def test_the_counter_an_operation_accepted_is_persisted(self):
p = self.store.add(self._preset(iccid='8970119000004600098'))
patches = [
mock.patch.object(self.srv, '_send_secured_packet', lambda *a, **k: {
'success': True, 'sw': '9000', 'bytes': 34, 'segments': 1,
'response_data': '027100000000'}),
mock.patch.object(self.srv, '_decode_por', lambda *a, **k: {
'response_status': 'por_ok', 'tar': '000000', 'cntr': '0000000001',
'pcntr': 0, 'rpl': 11, 'rhl': 10, 'raw': '',
'decoded': {'number_of_commands': 1, 'last_status_word': '9000',
'last_response_data': ''}}),
]
self.server.app = object()
for patch in patches:
patch.start()
try:
status, resp = self._post('/api/send-ota', {
'sp': '00', 'spi1': '16', 'spi2': '01', 'kic': '15', 'kid': '15',
'tar': '000000', 'cntr': '0000000001', 'preset_id': p['id']})
finally:
for patch in reversed(patches):
patch.stop()
self.assertEqual(status, 200, resp)
self.assertEqual(resp['final_cntr'], '0000000002')
self.assertEqual(self.store.find_keyset(p['id'], 1)['cntr'], '0000000002')
def test_a_counter_less_send_ota_body_still_answers(self):
# a pre-built packet without a counter: no final_cntr, no persist,
# and above all no int('') crash (v3.8.0 review fix)
p = self.store.add(self._preset(iccid='8970119000004600098'))
patches = [
mock.patch.object(self.srv, '_send_secured_packet', lambda *a, **k: {
'success': True, 'sw': '9000', 'bytes': 34, 'segments': 1,
'response_data': ''}),
mock.patch.object(self.srv, '_decode_por', lambda *a, **k: None),
]
self.server.app = object()
for patch in patches:
patch.start()
try:
status, resp = self._post('/api/send-ota', {
'sp': '00', 'spi1': '16', 'spi2': '01', 'kic': '15', 'kid': '15',
'tar': '000000', 'preset_id': p['id']})
finally:
for patch in reversed(patches):
patch.stop()
self.assertEqual(status, 200, resp)
self.assertNotIn('final_cntr', resp)
self.assertEqual(self.store.find_keyset(p['id'], 1)['cntr'], '0000000001')
def test_send_ota_refuses_a_kic_kid_version_mismatch(self):
# TS 102 225 A.2: the versions must be identical - invalid input
self.server.app = object()
status, resp = self._post('/api/send-ota', {
'sp': '00', 'spi1': '16', 'spi2': '01', 'kic': '15', 'kid': '25',
'tar': '000000', 'cntr': '0000000001'})
self.assertEqual(status, 400, resp)
self.assertIn('same key version', resp['error'])
def test_send_ota_refuses_an_undefined_key_version(self):
p = self.store.add(self._preset(iccid='8970119000004600098'))
self.server.app = object()
status, resp = self._post('/api/send-ota', {
'sp': '00', 'spi1': '16', 'spi2': '01', 'kic': '35', 'kid': '35',
'tar': '000000', 'cntr': '0000000001', 'preset_id': p['id']})
self.assertEqual(status, 400, resp)
self.assertIn('key version 3 is not defined', resp['error'])
def test_the_counter_lands_in_the_keyset_of_the_packet(self):
p = self.store.add(self._preset(keysets=[
self._keyset(kic='15', kid='15'),
self._keyset(kic='29', kid='29', kicKey='CC', kidKey='DD', cntr='0000000005')]))
patches = [
mock.patch.object(self.srv, '_send_secured_packet', lambda *a, **k: {
'success': True, 'sw': '9000', 'bytes': 34, 'segments': 1,
'response_data': '027100000000'}),
mock.patch.object(self.srv, '_decode_por', lambda *a, **k: {
'response_status': 'por_ok', 'tar': '000000', 'cntr': '0000000005',
'pcntr': 0, 'rpl': 11, 'rhl': 10, 'raw': '',
'decoded': {'number_of_commands': 1, 'last_status_word': '9000',
'last_response_data': ''}}),
]
self.server.app = object()
for patch in patches:
patch.start()
try:
status, resp = self._post('/api/send-ota', {
'sp': '00', 'spi1': '16', 'spi2': '01', 'kic': '29', 'kid': '29',
'tar': '000000', 'cntr': '0000000005', 'kicKey': 'CC', 'kidKey': 'DD',
'preset_id': p['id']})
finally:
for patch in reversed(patches):
patch.stop()
self.assertEqual(status, 200, resp)
self.assertEqual(resp['final_cntr'], '0000000006')
self.assertEqual(self.store.find_keyset(p['id'], 2)['cntr'], '0000000006')
self.assertEqual(self.store.find_keyset(p['id'], 1)['cntr'], '0000000001')
def test_a_ram_install_refuses_an_undefined_key_version(self):
p = self.store.add(self._preset(iccid='8970119000004600098'))
status, resp = self._post('/api/ram-install', {
'cap_hex': _mini_cap_hex(), 'kic': '35', 'kid': '35', 'preset_id': p['id']})
self.assertEqual(status, 400, resp)
self.assertIn('key version 3 is not defined', resp['error'])
def test_a_store_write_failure_answers_a_json_500(self):
with mock.patch.object(self.store, 'add', side_effect=OSError('read-only file system')):
status, resp = self._post('/api/presets', self._preset())
self.assertEqual(status, 500, resp)
self.assertIn('preset store write failed', resp['error'])
def test_the_store_endpoints_answer_503_without_a_store(self):
self.server.card_presets = None
status, resp = self._get('/api/presets')
self.assertEqual(status, 503, resp)
status, resp = self._post('/api/presets', self._preset())
self.assertEqual(status, 503, resp)
if __name__ == '__main__':
unittest.main()
Reference in New Issue View Git Blame Copy Permalink