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.
This commit is contained in:
2026-09-24 20:52:49 +03:00
parent 337e5df770
commit bed66c8ff8
9 changed files with 188 additions and 25 deletions
+1 -1
View File
@@ -304,7 +304,7 @@
<li><strong>AES-CBC</strong> шифрование (нулевой ICV, дополнение нулями до 16), ключи 16/24/32 байта (TS 102 225 &sect;5.1.2, KIc <code class="font-mono text-sm">x2</code>)</li>
<li><strong>Retail MAC</strong> (ISO 9797-1, MAC algorithm 3) для контрольной суммы DES/3DES</li>
<li><strong>AES-CMAC</strong> (NIST SP 800-38B, усечённый до 8 октетов) для контрольной суммы AES (TS 102 225 &sect;5.1.3.1, KID <code class="font-mono text-sm">x2</code>)</li>
<li><strong>Redundancy Check (RC)</strong> &mdash; CRC-32 по тому же кадру заголовка, что и CC (TS 102 225 &sect;5.1.3.2; ключ не нужен, SPI1 b2b1 = <code class="font-mono text-sm">01</code>)</li>
<li><strong>Redundancy Check (RC)</strong> &mdash; CRC-32 по тому же кадру заголовка, что и CC (TS 102 225 &sect;5.1.3.2; ключ не нужен, SPI1 b2b1 = <code class="font-mono text-sm">01</code>). Реализована только CRC-32 (как в pySim); кодирование CRC-16 по KID не поддерживается.</li>
<li>AES требует счётчик с защитой от повтора: биты SPI1 b5&nbsp;b4 должны быть <code class="font-mono text-sm">10</code> (счётчик больше) или <code class="font-mono text-sm">11</code> (счётчик +1) согласно TS 102 225 &sect;5.1.2/&sect;5.1.3.1</li>
<li>Байт паддинга настраивается (<code class="font-mono text-sm">00</code> по умолчанию или <code class="font-mono text-sm">FF</code>)</li>
</ul>
+1 -1
View File
@@ -303,7 +303,7 @@
<li><strong>AES-CBC</strong> encryption (zero ICV, zero-padded to 16), 16/24/32-byte keys (TS 102 225 &sect;5.1.2, KIc <code class="font-mono text-sm">x2</code>)</li>
<li><strong>Retail MAC</strong> (ISO 9797-1 MAC algorithm 3) for the DES/3DES cryptographic checksum</li>
<li><strong>AES-CMAC</strong> (NIST SP 800-38B, truncated to 8 octets) for the AES cryptographic checksum (TS 102 225 &sect;5.1.3.1, KID <code class="font-mono text-sm">x2</code>)</li>
<li><strong>Redundancy Check (RC)</strong> &mdash; CRC-32 over the same header frame as the CC (TS 102 225 &sect;5.1.3.2; needs no key, SPI1 b2b1 = <code class="font-mono text-sm">01</code>)</li>
<li><strong>Redundancy Check (RC)</strong> &mdash; CRC-32 over the same header frame as the CC (TS 102 225 &sect;5.1.3.2; needs no key, SPI1 b2b1 = <code class="font-mono text-sm">01</code>). Only CRC-32 is implemented, matching pySim; the KID's CRC-16 coding is not offered.</li>
<li>AES requires a replay-protected counter: SPI1 bits b5&nbsp;b4 must be <code class="font-mono text-sm">10</code> (counter higher) or <code class="font-mono text-sm">11</code> (counter +1) per TS 102 225 &sect;5.1.2/&sect;5.1.3.1</li>
<li>Padding byte configurable (<code class="font-mono text-sm">00</code> default, or <code class="font-mono text-sm">FF</code>)</li>
</ul>
+9 -1
View File
@@ -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;
+1 -1
View File
@@ -1,4 +1,4 @@
const CACHE = 'simple-v255';
const CACHE = 'simple-v256';
const URLS = [
'index.html',
'help.html',
+4
View File
@@ -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');
});
+1 -1
View File
@@ -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.
+72 -17
View File
@@ -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)
+93 -1
View File
@@ -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)
+6 -2
View File
@@ -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__':