From bed66c8ff834bdc24a4b5a508cce760a01f9b7e2 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: Thu, 24 Sep 2026 20:52:49 +0300 Subject: [PATCH] fix: robust equip-state detection and review follow-ups (v3.5.4) Code-review follow-ups for v3.5.2/v3.5.1, plus a test flake found while re-running the suites: - F1: a half-initialized equip is no longer reported as success. cmd2 swallows exceptions raised inside the equip command (and prints no traceback by default), while PysimApp.equip() assigns card/rs before it registers the command sets - so app.card alone once let the "CommandSet ... is already installed" abort pass as done while /api/tree stayed broken. The auto-equip attempt now captures the output with cmd2 debug on (a swallowed error prints a traceback), requires the new profile's command-set instances to be installed (_app_equip_complete), and the manual /api/command equip branch applies the post-equip refresh only when that check passes (and reports it in the output when it does not). - F3: SCARD_E_SHARING_VIOLATION is recoverable (a rebuild cannot free another process's claim) instead of transport-fatal. - F4: SPI1 b2b1 = 11 (Digital Signature) is refused instead of building an unsigned packet; help notes RC is CRC-32 only (KID CRC-16 not offered). - F5: _clear_app_card_state removes the muted stdout again when the app object had no stdout attribute. - F7: the watchdog re-arm keeps its rate-limit window when the trigger is busy/disabled instead of consuming it. - F2: stale docstring in _auto_equip_attempt. - tests: equip-state units, half-equip and captured-traceback failures, unequip-on-failure, busy trigger, sharing violation, DS refusal. - bonus: the MCC/MNC random-pick test could fail because a dict keyed by (mcc, mnc) keeps one of two entries (the bundled list carries both a real and an MVNO entry for 234/18 and 234/28); the picker was correct - the assertion now checks the pair against the non-MVNO pairs. 549 frontend / 421 Python green; version 3.5.4; sw cache simple-v256. --- frontend/help-ru.html | 2 +- frontend/help.html | 2 +- frontend/index.html | 10 +++- frontend/sw.js | 2 +- frontend/tests/sp.test.js | 4 ++ pyproject.toml | 2 +- pysim_simple_server/server.py | 89 ++++++++++++++++++++++++++------- tests/test_card_recovery.py | 94 ++++++++++++++++++++++++++++++++++- tests/test_mcc_mnc.py | 8 ++- 9 files changed, 188 insertions(+), 25 deletions(-) diff --git a/frontend/help-ru.html b/frontend/help-ru.html index 40d5eb9..91ad414 100644 --- a/frontend/help-ru.html +++ b/frontend/help-ru.html @@ -304,7 +304,7 @@
  • AES-CBC шифрование (нулевой ICV, дополнение нулями до 16), ключи 16/24/32 байта (TS 102 225 §5.1.2, KIc x2)
  • Retail MAC (ISO 9797-1, MAC algorithm 3) для контрольной суммы DES/3DES
  • AES-CMAC (NIST SP 800-38B, усечённый до 8 октетов) для контрольной суммы AES (TS 102 225 §5.1.3.1, KID x2)
    • -
  • Redundancy Check (RC) — CRC-32 по тому же кадру заголовка, что и CC (TS 102 225 §5.1.3.2; ключ не нужен, SPI1 b2b1 = 01)
    • +
  • Redundancy Check (RC) — CRC-32 по тому же кадру заголовка, что и CC (TS 102 225 §5.1.3.2; ключ не нужен, SPI1 b2b1 = 01). Реализована только CRC-32 (как в pySim); кодирование CRC-16 по KID не поддерживается.
  • AES требует счётчик с защитой от повтора: биты SPI1 b5 b4 должны быть 10 (счётчик больше) или 11 (счётчик +1) согласно TS 102 225 §5.1.2/§5.1.3.1
  • Байт паддинга настраивается (00 по умолчанию или FF)
    • diff --git a/frontend/help.html b/frontend/help.html index 432c668..06198b0 100644 --- a/frontend/help.html +++ b/frontend/help.html @@ -303,7 +303,7 @@
  • AES-CBC encryption (zero ICV, zero-padded to 16), 16/24/32-byte keys (TS 102 225 §5.1.2, KIc x2)
  • Retail MAC (ISO 9797-1 MAC algorithm 3) for the DES/3DES cryptographic checksum
  • AES-CMAC (NIST SP 800-38B, truncated to 8 octets) for the AES cryptographic checksum (TS 102 225 §5.1.3.1, KID x2)
    • -
  • Redundancy Check (RC) — CRC-32 over the same header frame as the CC (TS 102 225 §5.1.3.2; needs no key, SPI1 b2b1 = 01)
    • +
  • Redundancy Check (RC) — CRC-32 over the same header frame as the CC (TS 102 225 §5.1.3.2; needs no key, SPI1 b2b1 = 01). Only CRC-32 is implemented, matching pySim; the KID's CRC-16 coding is not offered.
  • AES requires a replay-protected counter: SPI1 bits b5 b4 must be 10 (counter higher) or 11 (counter +1) per TS 102 225 §5.1.2/§5.1.3.1
  • Padding byte configurable (00 default, or FF)
    • diff --git a/frontend/index.html b/frontend/index.html index 0c1b228..7124798 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -1559,7 +1559,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 = '3.5.3'; +const SIMPLE_VERSION = '3.5.4'; document.getElementById('app-version').textContent = 'v' + SIMPLE_VERSION; // ===== Tab switching ===== @@ -4594,6 +4594,14 @@ function _genSpBuild() { const resultEl = document.getElementById('sp-result'); + // Digital Signature is not implemented (pySim has no DS dialect either): + // refuse rather than build a packet whose SPI claims a signature that is + // not present. The selector does not offer '11'; this guards the field. + if (rcCcDs === 0x03) { + resultEl.value = 'Error: SPI1 RC/CC/DS = 11 (Digital Signature) is not supported'; + return; + } + // Rel-18: AES requires a replay-protected counter (SPI1 b5b4 = 10 or 11) if ((kicIsAes && ciphering) || (kidIsAes && hasMac)) { const counterBits = (spi1 >> 3) & 0x03; diff --git a/frontend/sw.js b/frontend/sw.js index 254dcba..445f2a2 100644 --- a/frontend/sw.js +++ b/frontend/sw.js @@ -1,4 +1,4 @@ -const CACHE = 'simple-v255'; +const CACHE = 'simple-v256'; const URLS = [ 'index.html', 'help.html', diff --git a/frontend/tests/sp.test.js b/frontend/tests/sp.test.js index 4f4651f..16a0e2f 100644 --- a/frontend/tests/sp.test.js +++ b/frontend/tests/sp.test.js @@ -122,6 +122,10 @@ test('RC (SPI 01) computes CRC-32 over the CPL frame', () => { '00191101091515B0000000000000010050C942DC00A40000023F00'); }); +test('DS (SPI 03) is refused instead of building an unsigned packet', () => { + assert.match(makeRun({ 'sp-spi1': '03' }), /Digital Signature/); +}); + test('crc32Bytes known answer (TS 102 225 Annex B)', () => { assert.strictEqual(bytesToHex(crc32Bytes(hexToBytes('0102030405'))), '470B99F4'); }); diff --git a/pyproject.toml b/pyproject.toml index 7b371da..1b455f8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "pysim-simple-server" -version = "3.5.3" +version = "3.5.4" 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/server.py b/pysim_simple_server/server.py index 2f194de..678bfdb 100644 --- a/pysim_simple_server/server.py +++ b/pysim_simple_server/server.py @@ -29,7 +29,7 @@ from osmocom.tlv import BER_TLV_IE from osmocom.utils import rpad -VERSION = '3.5.3' +VERSION = '3.5.4' MAX_ENVELOPE_SEGMENTS = 5 # max SMS segments for outgoing C-APDU in ENVELOPE @@ -2670,9 +2670,11 @@ def _is_pcsc_error(exc): return getattr(exc, 'hresult', -1) not in (-1, None) -# PC/SC hresults describing the card, not the service: the context and the -# reader are still fine, so a plain reconnect on the same transport recovers. -_PCSC_CARD_LEVEL = ( +# PC/SC hresults where rebuilding the transport cannot help: card/media states +# (the next connect() recovers) and a card that another process holds +# exclusively (a fresh context cannot free someone else's claim). +_PCSC_RECOVERABLE = ( + 0x8010000B, # SCARD_E_SHARING_VIOLATION (another process holds the card) 0x8010000C, # SCARD_E_NO_SMARTCARD 0x80100066, # SCARD_W_UNRESPONSIVE_CARD 0x80100067, # SCARD_W_UNPOWERED_CARD @@ -2686,14 +2688,15 @@ def _is_transport_fatal(exc): Only service/context failures (pcscd restart, reader re-enumeration, dead handle) require a fresh transport. Card-level states - most importantly - ``SCARD_W_REMOVED_CARD`` on a normal card swap - are recoverable by the - next ``connect()`` on the existing link; rebuilding the transport there - used to hand the new link the removed card's handle (live bug 2026-09-24: - auto-equip failed with 0x80100069 until the server was restarted).""" + ``SCARD_W_REMOVED_CARD`` on a normal card swap - and a sharing conflict + with another process are recoverable on the existing link; rebuilding the + transport there used to hand the new link the removed card's handle (live + bug 2026-09-24: auto-equip failed with 0x80100069 until the server was + restarted).""" hr = getattr(exc, 'hresult', -1) if hr in (-1, None): return False - return hr not in _PCSC_CARD_LEVEL + return hr not in _PCSC_RECOVERABLE def _clear_app_card_state(app): @@ -2710,6 +2713,7 @@ def _clear_app_card_state(app): if app is None: return if getattr(app, 'rs', None) is not None and callable(getattr(app, 'equip', None)): + had_stdout = hasattr(app, 'stdout') old_stdout = getattr(app, 'stdout', None) try: app.stdout = StringIO() # mute the 'pySim-shell not equipped!' line @@ -2717,8 +2721,13 @@ def _clear_app_card_state(app): except Exception as e: sys.stderr.write('UNEQUIP: pySim unequip failed: %s\n' % e) finally: - if old_stdout is not None: + if had_stdout: app.stdout = old_stdout + else: + try: + del app.stdout + except Exception: + pass app.card = None app.rs = None app.lchan = None @@ -2974,15 +2983,47 @@ def _auto_equip_rearm(now=None): now = time.time() if now is None else now if now - _AUTO_EQUIP_LAST < _AUTO_EQUIP_BACKOFF: return False + if not _auto_equip_trigger(): + return False # busy/disabled: leave the window open for the next tick _AUTO_EQUIP_LAST = now - return _auto_equip_trigger() + return True + +def _app_equip_complete(app): + """True when pySim's shell really ended up equipped after an equip command. + + cmd2 swallows exceptions raised inside the equip command (it prints them, + no traceback, and returns normally), and PysimApp.equip() assigns + ``card``/``rs`` before it registers the command sets - so ``app.card`` + alone cannot tell a completed equip from a half-initialized shell (live + 2026-09-24: an abort at "CommandSet ... is already installed" passed the + card check, the worker reported done and /api/tree stayed broken). The + profile's shell command-set *instances* must be installed; those are + exactly what PysimApp.equip() registers. States that cannot be verified + (no profile command sets, foreign app objects) are trusted.""" + if app is None or getattr(app, 'card', None) is None or getattr(app, 'lchan', None) is None: + return False + rs = getattr(app, 'rs', None) + profile = getattr(rs, 'profile', None) + sets = list(getattr(profile, 'shell_cmdsets', []) or []) + finder = getattr(app, 'find_commandsets', None) + if not sets or not callable(finder): + return True # nothing to verify + try: + for cmd_set in sets: + if cmd_set not in finder(type(cmd_set)): + return False + return True + except Exception: + return True + def _auto_equip_attempt(server): """One equip attempt under _CARD_LOCK; True when connected/no-op. - A failed attempt leaves the previous card state intact (pySim's equip() - only swaps the command sets after a successful init), so a retry starts - from an explicit state.""" + cmd2 swallows exceptions raised inside the equip command, so success is + only reported when the shell really ended up equipped + (_app_equip_complete); a failed attempt unequips the half-initialized + shell before the next retry.""" with _CARD_LOCK: if _CARD_CONNECTED or not getattr(server, 'card_present', False): return True @@ -2995,17 +3036,28 @@ def _auto_equip_attempt(server): return False server.equipping = True try: + out = StringIO() old_stdout, old_stderr = app.stdout, sys.stderr - app.stdout = StringIO() - sys.stderr = app.stdout + old_debug = getattr(app, 'debug', None) + app.stdout = out + sys.stderr = out + if old_debug is not None: + app.debug = True # cmd2 prints a traceback for swallowed errors try: app.onecmd_plus_hooks('equip') finally: + if old_debug is not None: + app.debug = old_debug app.stdout = old_stdout sys.stderr = old_stderr if not getattr(server, 'card_present', False) or server.app.card is None: sys.stderr.write('AUTO-EQUIP: card gone during initialization\n') return True + if 'Traceback (most recent call last)' in out.getvalue(): + raise RuntimeError('equip reported an error (see the captured output)') + if not _app_equip_complete(app): + raise RuntimeError('equip finished with a half-initialized shell ' + '(command sets not registered)') _apply_equipped_card(server) sys.stderr.write('AUTO-EQUIP: done\n') return True @@ -4274,8 +4326,11 @@ class PysimHandler(BaseHTTPRequestHandler): status = 'OK' if not output or 'not a recognized command' not in output else 'ERROR' if is_equip: _tlog('equip: onecmd_plus_hooks %dms' % elapsed) - if is_equip and self.server.app and self.server.app.card and self.server.terminal_profile: + if is_equip and self.server.app and self.server.terminal_profile and _app_equip_complete(self.server.app): _apply_equipped_card(self.server) + elif is_equip: + output += ('EQUIP: the shell did not register its command sets - ' + 'see the output above\n') sys.stderr.write("CMD: %s → %s (%dms)\n" % (cmd, status, elapsed)) resp = {'output': output, 'stop': bool(stop)} self._send_json(resp) diff --git a/tests/test_card_recovery.py b/tests/test_card_recovery.py index b3d1eb5..8863732 100644 --- a/tests/test_card_recovery.py +++ b/tests/test_card_recovery.py @@ -2,6 +2,7 @@ """Tests for the PC/SC failure recovery: presence-monitor watchdog and transport recreation after a service failure (pcscd restart).""" +import sys import unittest from io import StringIO from types import SimpleNamespace @@ -36,6 +37,11 @@ class TransportFatalTests(unittest.TestCase): for hr in (0x80100069, 0x8010000C, 0x80100068, 0x80100066, 0x80100067): self.assertFalse(server._is_transport_fatal(self._exc(hr)), hex(hr)) + def test_sharing_violation_is_not_transport_fatal(self): + # Another process holds the card: a fresh context cannot free it, so + # rebuilding the transport would only burn the retry budget. + self.assertFalse(server._is_transport_fatal(self._exc(0x8010000B))) + def test_service_errors_rebuild_the_transport(self): for hr in (0x8010001D, 0x8010001E, 0x8010002E, 0x80100003): self.assertTrue(server._is_transport_fatal(self._exc(hr)), hex(hr)) @@ -44,6 +50,36 @@ class TransportFatalTests(unittest.TestCase): self.assertFalse(server._is_transport_fatal(RuntimeError('SW match failed'))) +class EquipStateTests(unittest.TestCase): + @staticmethod + def _app(card='card', lchan='lchan', cmdsets=None, profile_sets=None): + rs = None + if profile_sets is not None: + rs = SimpleNamespace(profile=SimpleNamespace(shell_cmdsets=profile_sets)) + app = SimpleNamespace(card=card, lchan=lchan, rs=rs) + if cmdsets is not None: + app.find_commandsets = lambda cls: cmdsets + return app + + def test_card_and_channel_are_required(self): + self.assertFalse(server._app_equip_complete(None)) + self.assertFalse(server._app_equip_complete(self._app(card=None))) + self.assertFalse(server._app_equip_complete(self._app(lchan=None))) + + def test_profile_command_sets_must_be_installed(self): + # PysimApp.equip() sets card/rs before registering the command sets; + # an abort during registration must not pass as success. + cs = object() + app = self._app(cmdsets=[], profile_sets=[cs]) + self.assertFalse(server._app_equip_complete(app)) + app.find_commandsets = lambda cls: [cs] + self.assertTrue(server._app_equip_complete(app)) + + def test_unverifiable_state_is_trusted(self): + # No profile command sets / foreign app object: cannot verify. + self.assertTrue(server._app_equip_complete(self._app())) + + class WatchdogTests(unittest.TestCase): def test_alive_monitor_is_left_alone(self): calls = [] @@ -210,7 +246,7 @@ class AutoEquipTests(unittest.TestCase): @staticmethod def make_server(onecmd): - app = SimpleNamespace(stdout=StringIO(), card=None) + app = SimpleNamespace(stdout=StringIO(), card=None, lchan='lchan') app.onecmd_plus_hooks = onecmd return SimpleNamespace(app=app, card=None, scc=None, card_present=True, equipping=False, terminal_profile='tp', card_session=1) @@ -224,6 +260,7 @@ class AutoEquipTests(unittest.TestCase): if state['calls'] == 1: raise RuntimeError('Failed to transmit with protocol T0. Card was removed.') srv.app.card = SimpleNamespace(_scc='scc') + srv.app.lchan = 'lchan' # the failed attempt unequipped the shell srv.app.onecmd_plus_hooks = onecmd applied = [] @@ -287,6 +324,61 @@ class AutoEquipTests(unittest.TestCase): self.assertEqual(state['calls'], 1) self.assertGreater(server._AUTO_EQUIP_BACKOFF, server._AUTO_EQUIP_REARM_DELAY) + def test_half_initialized_equip_is_a_failure(self): + # cmd2 swallows the equip exception, so a card without the new + # profile's command sets must not count as success (it broke + # /api/tree on 2026-09-24 while the worker reported done). + srv = self.make_server(None) + cs = object() + srv.app.rs = SimpleNamespace(profile=SimpleNamespace(shell_cmdsets=[cs])) + srv.app.find_commandsets = lambda cls: [] + + def onecmd(cmd): + srv.app.card = SimpleNamespace(_scc='scc') # set before the abort + + srv.app.onecmd_plus_hooks = onecmd + with mock.patch.object(server, '_ensure_transport', lambda s: True), \ + mock.patch.object(server, '_apply_equipped_card', + lambda s: self.fail('_apply_equipped_card must not run')): + self.assertFalse(server._auto_equip_attempt(srv)) + self.assertFalse(server._TRANSPORT_STALE) + + def test_captured_equip_traceback_is_a_failure(self): + # cmd2 prints swallowed exceptions (with cmd2 debug: a traceback) to + # sys.stderr, which the attempt captures. + srv = self.make_server(None) + + def onecmd(cmd): + srv.app.card = SimpleNamespace(_scc='scc') + sys.stderr.write('Traceback (most recent call last):\nRuntimeError: boom\n') + + srv.app.onecmd_plus_hooks = onecmd + with mock.patch.object(server, '_ensure_transport', lambda s: True), \ + mock.patch.object(server, '_apply_equipped_card', + lambda s: self.fail('_apply_equipped_card must not run')): + self.assertFalse(server._auto_equip_attempt(srv)) + + def test_failed_attempt_unequips_the_half_initialized_shell(self): + calls = [] + srv = self.make_server(None) + srv.app.rs = 'rs' + srv.app.equip = lambda c, r: calls.append((c, r)) + + def onecmd(cmd): + raise RuntimeError('CommandSet ShellCommands is already installed') + + srv.app.onecmd_plus_hooks = onecmd + with mock.patch.object(server, '_ensure_transport', lambda s: True): + self.assertFalse(server._auto_equip_attempt(srv)) + self.assertEqual(calls, [(None, None)]) + self.assertIsNone(srv.app.card) + + def test_busy_trigger_leaves_the_rearm_window_open(self): + server._server_ref = SimpleNamespace(card_present=True, equipping=False) + with mock.patch.object(server, '_auto_equip_trigger', lambda: False): + self.assertFalse(server._auto_equip_rearm(now=100.0)) + self.assertEqual(server._AUTO_EQUIP_LAST, 0.0) + def test_rearm_needs_a_present_card_and_no_session(self): calls = [] server._server_ref = SimpleNamespace(card_present=True, equipping=False) diff --git a/tests/test_mcc_mnc.py b/tests/test_mcc_mnc.py index 0f1fc52..36d07a1 100644 --- a/tests/test_mcc_mnc.py +++ b/tests/test_mcc_mnc.py @@ -90,10 +90,14 @@ class MccMncFilterTests(unittest.TestCase): def test_random_real_list_never_picks_mvno(self): data = server._mcc_mnc_load(__main__._default_mcc_mnc_list()) - by_pair = {(e['mcc'], e['mnc']): e for e in data} + # A few (mcc, mnc) pairs exist both as a real network and as an MVNO + # entry (e.g. 234/18, 234/28), so a dict keyed by the pair can keep the + # wrong entry: assert the picked pair has a non-MVNO entry behind it. + real_pairs = {(e['mcc'], e['mnc']) for e in data + if not server._mcc_mnc_is_mvno(e)} for _ in range(100): r = server._mcc_mnc_random(data) - self.assertFalse(server._mcc_mnc_is_mvno(by_pair[(r['mcc'], r['mnc'])])) + self.assertIn((r['mcc'], r['mnc']), real_pairs) if __name__ == '__main__':