From bb13ed45f8f7c8b744957d098ca928b4f5b044ba Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=90=D0=BD=D1=82=D0=BE=D0=BD=20=D0=A2=D1=80=D0=BE=D1=88?= =?UTF-8?q?=D0=B8=D0=BD?= Date: Sun, 20 Sep 2026 13:12:00 +0300 Subject: [PATCH] net: initialize the monitor in the startup equip path (v2.7.1) The startup init (server booted with the card already in the reader) equips the card and reads the ICCID itself, but never created server.net_state - only /api/command equip and the auto-equip worker did. With a card present at boot the monitor stayed empty, net-sim responses carried net_state: null and the panel only started updating after pressing Refresh (which lazily created the state). Fix: - factor _netstate_read(app) / _netstate_install() / _netstate_init() in server.py; _apply_equipped_card uses _netstate_init() - __main__ startup: read the monitored EFs in the same CAT-free window as the ICCID and install the state once the server object exists - _netstate_ensure() lazily initializes the state in the net-sim/event hooks so any equip path predating the monitor cannot leave it dead - netSimRun falls back to netStateFetch() when a response has no net_state - tests: tests/test_netstate_server.py (candidate order, record vs transparent, absent files, install, ensure); SW cache simple-v204 --- frontend/index.html | 3 +- frontend/sw.js | 2 +- pyproject.toml | 2 +- pysim_simple_server/__main__.py | 17 +++++- pysim_simple_server/server.py | 55 +++++++++++++------ tests/test_netstate_server.py | 97 +++++++++++++++++++++++++++++++++ 6 files changed, 155 insertions(+), 21 deletions(-) create mode 100644 tests/test_netstate_server.py 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()