diff --git a/frontend/index.html b/frontend/index.html index 7a70e68..10334ab 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -1414,7 +1414,7 @@ // ===== Version ===== // Single source of truth for the PWA version: shown in the header and used // by the server version check in pysimConnect(). -const SIMPLE_VERSION = '2.7.0'; +const SIMPLE_VERSION = '2.7.1'; document.getElementById('app-version').textContent = 'v' + SIMPLE_VERSION; // ===== Tab switching ===== @@ -12690,6 +12690,7 @@ async function netSimRun(scenario) { } pysimEventsRender(); if (res.net_state) netStateRender(res.net_state); + else netStateFetch(); } catch (e) { statusEl.textContent = t('Error') + ': ' + e.message; statusEl.className = 'text-xs text-red-500 mb-1'; diff --git a/frontend/sw.js b/frontend/sw.js index 6f18509..53f61e6 100644 --- a/frontend/sw.js +++ b/frontend/sw.js @@ -1,4 +1,4 @@ -const CACHE = 'simple-v203'; +const CACHE = 'simple-v204'; const URLS = [ 'index.html', 'help.html', diff --git a/pyproject.toml b/pyproject.toml index 87415fb..e0b8e20 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "pysim-simple-server" -version = "2.7.0" +version = "2.7.1" description = "HTTP REST server wrapping pysim for the SIMple PWA" requires-python = ">=3.8" # pysim is a git-only dependency installed explicitly by setup.bat/setup.sh. diff --git a/pysim_simple_server/__main__.py b/pysim_simple_server/__main__.py index 3467ab0..9407744 100644 --- a/pysim_simple_server/__main__.py +++ b/pysim_simple_server/__main__.py @@ -13,7 +13,7 @@ from pySim.cards import UiccCardBase from .shell import load_pysim_app from . import fastinit -from .server import PysimHandler, StderrApduTracer, _LoggingApduTracer, VERSION, _send_terminal_profile, _DefaultProactiveHandler, _handle_proactive_chain, _send_status, _init_proactive_session, _timing_on, _tlog, _set_menu_timeout, start_card_monitor, set_auto_equip, _read_iccid, _LineFilter +from .server import PysimHandler, StderrApduTracer, _LoggingApduTracer, VERSION, _send_terminal_profile, _DefaultProactiveHandler, _handle_proactive_chain, _send_status, _init_proactive_session, _timing_on, _tlog, _set_menu_timeout, start_card_monitor, set_auto_equip, _read_iccid, _netstate_read, _netstate_install, _LineFilter _server_start = 0 @@ -167,6 +167,7 @@ def main(): if app is not None and opts.fast_init: fastinit.install(app) iccid = None + netstate_files = None if scc and card is not None and hasattr(scc, '_tp'): scc._tp.apdu_tracer = _LoggingApduTracer() try: @@ -176,6 +177,12 @@ def main(): iccid = _read_iccid(app) if iccid: sys.stderr.write('INIT: ICCID %s\n' % iccid) + # Network state monitor: read the network-related EFs in the + # same CAT-free window (skipped without a readable ICCID). + try: + netstate_files = _netstate_read(app) + except Exception: + netstate_files = None t_phase = time.time() sys.stderr.write('INIT: sending TERMINAL PROFILE %s (CLA=%s)\n' % (opts.terminal_profile, scc.cat_cla)) sm, el = _send_terminal_profile(scc, opts.terminal_profile) @@ -231,6 +238,14 @@ def main(): server.card_present = card is not None server.card_session = 1 if card is not None else 0 server.iccid = iccid + # Network state monitor: install the state read during the startup init + # (right after the ICCID, before the TERMINAL PROFILE). No readable + # ICCID means the card is considered unusable - give up. + try: + _netstate_install(server, netstate_files if iccid else None) + except Exception as e: + server.net_state = None + sys.stderr.write('INIT: network state failed: %s\n' % e) server.equipping = False # Set server reference for polling timer and mark the card session state import pysim_simple_server.server diff --git a/pysim_simple_server/server.py b/pysim_simple_server/server.py index 510411e..bbe9264 100644 --- a/pysim_simple_server/server.py +++ b/pysim_simple_server/server.py @@ -25,7 +25,7 @@ from osmocom.construct import GsmOrUcs2Adapter from osmocom.tlv import BER_TLV_IE -VERSION = '2.7.0' +VERSION = '2.7.1' MAX_ENVELOPE_SEGMENTS = 5 # max SMS segments for outgoing C-APDU in ENVELOPE @@ -492,14 +492,13 @@ def _mcc_mnc_random(data, exclude=None): ('countryName', 'countryCode', 'mcc', 'mnc', 'brand', 'operator', 'status')} -def _netstate_read(server, keys=None): +def _netstate_read(app, keys=None): """Read the monitored network-state EFs (best effort per file). The Network state panel caches them: the full set is read once at equip (after a readable ICCID), on demand via /api/net-state-refresh, and only the card-side files (EF.IMSI) after net-sim / Location-status operations. """ - app = getattr(server, 'app', None) rs = getattr(app, 'rs', None) if app else None if not app or not rs: return {} @@ -553,24 +552,53 @@ def _netstate_compute(server): return netstate.compute_network(state, data) +def _netstate_install(server, files, source='init'): + """Install a freshly read monitor state (None/{} = nothing readable).""" + state = netstate.new_state() + netstate.merge_read(state, files or {}, source=source) + netstate.set_read_time(state) + server.net_state = state + _netstate_compute(server) + + +def _netstate_init(server): + """Create the monitor state from a fresh read (best effort).""" + try: + _netstate_install(server, _netstate_read(server.app)) + except Exception as e: + server.net_state = None + _tlog('network state read failed: %s' % e) + + +def _netstate_ensure(server): + """Monitor state of the current session, created on demand for equip + paths that predate it (e.g. the startup init).""" + state = getattr(server, 'net_state', None) + if state is None and getattr(server, 'iccid', None): + _netstate_init(server) + state = getattr(server, 'net_state', None) + return state + + def _netstate_after_net_sim(server, scenario, result): """Update the cached Network state after a scenario run: the bytes we wrote are known without re-reading; the card may update EF.IMSI itself.""" - state = getattr(server, 'net_state', None) + state = _netstate_ensure(server) if state is None: return netstate.apply_steps(state, (result or {}).get('steps') or []) service = netsim.SCENARIO_SERVICE.get(scenario) if service and (result or {}).get('success'): netstate.set_service(state, service, 'net-sim:' + scenario) - netstate.merge_read(state, _netstate_read(server, ['imsi']), source='read') + netstate.merge_read(state, _netstate_read(server.app, ['imsi']), + source='read') _netstate_compute(server) def _netstate_after_event(server, event_type, event_data): """Location status changes the simulated service state; the card may also switch EF.IMSI (multi-IMSI applets) after such an event.""" - state = getattr(server, 'net_state', None) + state = _netstate_ensure(server) if state is None: return try: @@ -587,7 +615,8 @@ def _netstate_after_event(server, event_type, event_data): 0x02: netstate.SERVICE_NONE}.get(status) if service: netstate.set_service(state, service, 'event:location-status') - netstate.merge_read(state, _netstate_read(server, ['imsi']), source='read') + netstate.merge_read(state, _netstate_read(server.app, ['imsi']), + source='read') _netstate_compute(server) @@ -2481,15 +2510,7 @@ def _apply_equipped_card(server): # Network state monitor: read the network-related EFs right after the # ICCID (still before the TERMINAL PROFILE opens a CAT session). A # card without a readable ICCID is considered unusable - give up. - try: - server.net_state = netstate.new_state() - netstate.merge_read(server.net_state, _netstate_read(server), - source='init') - netstate.set_read_time(server.net_state) - _netstate_compute(server) - except Exception as e: - server.net_state = None - _tlog('equip: network state read failed: %s' % e) + _netstate_init(server) else: server.net_state = None _tlog('equip: ICCID not readable - network state skipped') @@ -3625,7 +3646,7 @@ class PysimHandler(BaseHTTPRequestHandler): state = netstate.new_state() self.server.net_state = state try: - files = _netstate_read(self.server, keys) + files = _netstate_read(self.server.app, keys) except Exception as e: self._send_json({'error': str(e)}, 500) self._log_resp({'error': str(e)}) diff --git a/tests/test_netstate_server.py b/tests/test_netstate_server.py new file mode 100644 index 0000000..56f3192 --- /dev/null +++ b/tests/test_netstate_server.py @@ -0,0 +1,97 @@ +# coding=utf-8 +"""Tests for the server-side Network-state monitor plumbing. + +Covers the read helper (candidate order, transparent vs record files, absent +files) and the install/ensure helpers shared by the equip paths and the HTTP +endpoints. The card I/O is faked; ``_select_path`` is stubbed so no pySim +file tree is needed. +""" + +import unittest +from types import SimpleNamespace + +from pysim_simple_server import server + + +class FakeLchan: + def __init__(self, files): + self.files = files # path -> ('transparent', hex) | ('record', [hex]) + self.calls = [] + self.selected = None + self.selected_file = None + self.selected_file_fcp = True + self.selected_file_type = lambda: 'ef' + self.selected_file_structure = lambda: self.files[self.selected][0] + self.selected_file_num_of_rec = lambda: len(self.files[self.selected][1]) + self.read_binary = lambda: (self.files[self.selected][1], '9000') + self.read_record = lambda i: (self.files[self.selected][1][i - 1], '9000') + + def select(self, path): + self.calls.append(('select', path)) + if path not in self.files: + raise RuntimeError('file not found: %s' % path) + self.selected = path + + +class NetStateServerTests(unittest.TestCase): + def setUp(self): + self._orig_select = server._select_path + + def fake_select(lchan, path, app): + lchan.select(path) + return None, None + + server._select_path = fake_select + + def tearDown(self): + server._select_path = self._orig_select + + def test_read_walks_candidates_and_marks_missing(self): + lchan = FakeLchan({ + 'DF.GSM/6F07': ('transparent', '0829AA'), # 2nd IMSI candidate + 'ADF.USIM/6FE4': ('linear_fixed', ['AABB', 'CCDD']), # EPSNSC records + }) + app = SimpleNamespace(rs=SimpleNamespace(lchan=[lchan])) + out = server._netstate_read(app, ['imsi', 'epsnsc', 'spdi']) + self.assertTrue(out['imsi']['present']) + self.assertEqual(out['imsi']['path'], 'DF.GSM/6F07') + self.assertEqual(out['imsi']['kind'], 'transparent') + self.assertEqual(out['imsi']['data'], '0829AA') + self.assertEqual(out['epsnsc']['kind'], 'record') + self.assertEqual(out['epsnsc']['records'], + [{'num': 1, 'data': 'AABB'}, + {'num': 2, 'data': 'CCDD'}]) + self.assertFalse(out['spdi']['present']) + # candidates are walked in FILE_DEFS order (spdi sits before epsnsc) + self.assertEqual(lchan.calls, + [('select', 'ADF.USIM/6F07'), + ('select', 'DF.GSM/6F07'), + ('select', 'ADF.USIM/6FCD'), + ('select', 'ADF.USIM/6FE4')]) + + def test_install_builds_the_state_and_network(self): + files = {'imsi': {'name': 'EF.IMSI', 'fid': '6F07', 'present': True, + 'kind': 'transparent', 'data': '082982608200002080'}} + srv = SimpleNamespace(mcc_mnc_path=None) + server._netstate_install(srv, files) + self.assertEqual(srv.net_state['files']['imsi']['source'], 'init') + self.assertEqual(srv.net_state['files']['imsi']['data'], + '082982608200002080') + self.assertIsNotNone(srv.net_state['read_at']) + self.assertIsNotNone(srv.net_state['network']) + + def test_ensure_creates_the_state_only_with_a_readable_iccid(self): + lchan = FakeLchan({}) + app = SimpleNamespace(rs=SimpleNamespace(lchan=[lchan])) + srv = SimpleNamespace(iccid='8970119000004600098', app=app, + mcc_mnc_path=None) + state = server._netstate_ensure(srv) + self.assertIsNotNone(state) + self.assertEqual(srv.net_state, state) + srv2 = SimpleNamespace(iccid=None, app=app, mcc_mnc_path=None) + self.assertIsNone(server._netstate_ensure(srv2)) + self.assertFalse(hasattr(srv2, 'net_state')) + + +if __name__ == '__main__': + unittest.main()