From 0f8c6dea50a7739f1dd5a4c931d0c8436bc389e7 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 08:57:06 +0300 Subject: [PATCH] fix: recover from a card swap without a server restart (v3.5.2) Equip was broken after a card swap: auto-equip (and manual Equip) failed with "Failed to transmit with protocol T0. Card was removed. (0x80100069)" until the server was restarted. Chain (v3.1.2 regression): - _handle_card_disconnect() cleared server.card/scc but not pySim's app.card/app.rs/app.lchan, so the removed card - and with it the old PC/SC link and its exclusive card handle - stayed referenced; handlers using app.rs (e.g. the PWA's /api/tree poll) kept transmitting over the dead card. - those errors carry hresult=0x80100069 (SCARD_W_REMOVED_CARD), but _is_pcsc_error() treated any PC/SC error as a dead service and set _TRANSPORT_STALE; the next auto-equip then called _ensure_transport(), which built a second PcscSimLink while the old one was still connected - the new link inherited the removed card's handle and failed on its first APDU. No retry existed, so every later equip repeated the failure. Fixes: - _is_transport_fatal(): only service/context hresults rebuild the transport (E_NO_SERVICE, E_SERVICE_STOPPED, E_NO_READERS_AVAILABLE, E_INVALID_HANDLE, ...); card-level states (W_REMOVED_CARD, E_NO_SMARTCARD, W_RESET_CARD, W_UNRESPONSIVE_CARD, W_UNPOWERED_CARD) reconnect on the existing link. All 8 disconnect call sites pass the new verdict. - _clear_app_card_state(): unequip through pySim's own equip(None, None) before clearing app.card/app.rs/app.lchan. Nulling them alone would make the next equip abort with "CommandSet ... is already installed" (PysimApp.equip() unregisters the previous profile's command sets from self.rs), which left /api/tree broken after a swap. A failed auto-equip attempt unequips the half-initialized shell as well. Handlers now answer "no card" instead of transmitting over the dead card, and the old link becomes collectable. - _ensure_transport(): disconnects the old link before building the new one (restoring it if the factory fails) - the rebuild path is now safe for the real pcscd-restart case. - auto-equip: _auto_equip_attempt()/_auto_equip_attempts() retry up to 3 times, 1 s apart (fatal failures mark the transport so the retry rebuilds); the watchdog re-arms it (_auto_equip_rearm) every 5 s while a card is present and the session is down, with exponential backoff to 60 s for a card that cannot be initialized at all. - fastinit.init_card_fast(): also retries once after a PC/SC link error (CardConnectionException/NoCardException), not only after SW mismatches. Tests: transport-fatal classification, app-state clearing, shell unequip, old-link release, auto-equip retry/backoff/re-arm, fastinit link retry. 547 frontend / 410 Python green; version 3.5.2; sw cache simple-v254. --- README.md | 3 +- README_RUS.md | 3 +- frontend/index.html | 2 +- frontend/sw.js | 2 +- pyproject.toml | 2 +- pysim_simple_server/fastinit.py | 12 ++ pysim_simple_server/server.py | 240 +++++++++++++++++++++++++------- tests/test_card_recovery.py | 198 ++++++++++++++++++++++++++ tests/test_fastinit.py | 24 ++++ 9 files changed, 434 insertions(+), 52 deletions(-) diff --git a/README.md b/README.md index 816c86d..c163ed6 100644 --- a/README.md +++ b/README.md @@ -752,7 +752,8 @@ pysim-simple-server --http-port 8080 - **"Failed to establish context: Access denied"** — `pcscd` isn't running or the user lacks permission: `sudo systemctl enable --now pcscd && sudo usermod -a -G pcscd $USER`. - **"device file /dev/ttyUSB0 does not exist"** — no serial reader; connect a USB reader or pass `-d` explicitly. The server still starts without a reader. -- **"Service not available" (0x8010001D) / reader gone after a `pcscd` restart** — the server recovers by itself: a watchdog revives pyscard's presence monitor and the next equip (automatic after re-insertion, or the **Equip** button) recreates the PC/SC transport. No server restart needed. +- **"Service not available" (0x8010001D) / reader gone after a `pcscd` restart** — the server recovers by itself: a watchdog revives pyscard's presence monitor and the next equip (automatic after re-insertion, or the **Equip** button) recreates the PC/SC transport (releasing the old link first). No server restart needed. +- **"Card was removed" errors / auto-equip fails after a card swap** — a normal card swap reconnects on the existing transport (card-level PC/SC errors no longer force a transport rebuild), a failed auto-equip is retried up to 3 times, and the watchdog re-arms it every 5 s while a card is present. A removed card also clears the card state completely, so handlers answer "no card" instead of transmitting over the dead card. ### API reference diff --git a/README_RUS.md b/README_RUS.md index 8a387ce..1fdc571 100644 --- a/README_RUS.md +++ b/README_RUS.md @@ -736,7 +736,8 @@ pysim-simple-server --http-port 8080 - **"Failed to establish context: Access denied"** — `pcscd` не запущен или нет прав: `sudo systemctl enable --now pcscd && sudo usermod -a -G pcscd $USER`. - **"device file /dev/ttyUSB0 does not exist"** — нет serial-ридера; подключите USB-ридер или укажите `-d`. Сервер всё равно стартует без ридера. -- **"Service not available" (0x8010001D) / ридер пропал после перезапуска `pcscd`** — сервер восстанавливается сам: сторож перезапускает монитор присутствия pyscard, а следующий equip (автоматически после повторной вставки карты или кнопкой **Equip**) пересоздаёт транспорт PC/SC. Перезапуск сервера не нужен. +- **"Service not available" (0x8010001D) / ридер пропал после перезапуска `pcscd`** — сервер восстанавливается сам: сторож перезапускает монитор присутствия pyscard, а следующий equip (автоматически после повторной вставки карты или кнопкой **Equip**) пересоздаёт транспорт PC/SC (сначала освобождая старый линк). Перезапуск сервера не нужен. +- **Ошибки "Card was removed" / equip не срабатывает после смены карты** — обычная смена карты переподключается на существующем транспорте (карточные ошибки PC/SC больше не вызывают пересоздание транспорта), неудавшийся авто-equip повторяется до 3 раз, а сторож перезапускает его каждые 5 с, пока карта в ридере. Снятие карты теперь полностью очищает состояние, поэтому обработчики отвечают «нет карты», а не шлют APDU мёртвой карте. ### Справочник API diff --git a/frontend/index.html b/frontend/index.html index b1d4e1e..e2b72fe 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.1'; +const SIMPLE_VERSION = '3.5.2'; document.getElementById('app-version').textContent = 'v' + SIMPLE_VERSION; // ===== Tab switching ===== diff --git a/frontend/sw.js b/frontend/sw.js index 7421825..ed58cb7 100644 --- a/frontend/sw.js +++ b/frontend/sw.js @@ -1,4 +1,4 @@ -const CACHE = 'simple-v253'; +const CACHE = 'simple-v254'; const URLS = [ 'index.html', 'help.html', diff --git a/pyproject.toml b/pyproject.toml index 5d941df..c0813d7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "pysim-simple-server" -version = "3.5.1" +version = "3.5.2" 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/fastinit.py b/pysim_simple_server/fastinit.py index 1d962f7..5ebf06b 100644 --- a/pysim_simple_server/fastinit.py +++ b/pysim_simple_server/fastinit.py @@ -17,6 +17,8 @@ keep a real reconnect/physical reset. import operator import sys +from smartcard.Exceptions import CardConnectionException, NoCardException + from pySim.cards import CardBase, SimCardBase, UiccCardBase, card_detect from pySim.commands import SimCardCommands from pySim.exceptions import ProtocolError, SwMatchError @@ -105,6 +107,16 @@ def init_card_fast(sl, skip_card_init=False, wait=True): sys.stderr.write('FAST-INIT: %s; retrying after physical reset\n' % e) sl.reset_card() return _init_card_once(sl, skip_card_init, wait=False) + except (CardConnectionException, NoCardException) as e: + # PC/SC link error during the first exchange (e.g. the card was + # swapped a moment ago): release the connection, wait for the card + # and retry once. + sys.stderr.write('FAST-INIT: link error (%s); retrying after reconnect\n' % e) + try: + sl.disconnect() + except Exception: + pass + return _init_card_once(sl, skip_card_init, wait=True) def _init_card_once(sl, skip_card_init, wait): diff --git a/pysim_simple_server/server.py b/pysim_simple_server/server.py index 8681e56..2371ac1 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.1' +VERSION = '3.5.2' MAX_ENVELOPE_SEGMENTS = 5 # max SMS segments for outgoing C-APDU in ENVELOPE @@ -1362,7 +1362,7 @@ def _do_status_poll(): _handle_proactive_chain(scc, st_sw) except Exception as e: sys.stderr.write('AUTO-STATUS error: %s\n' % e) - _handle_card_disconnect(stale=_is_pcsc_error(e)) + _handle_card_disconnect(stale=_is_transport_fatal(e)) _reset_poll_timer() def _poll_enable(): @@ -2666,17 +2666,70 @@ def _is_pcsc_error(exc): """True for a PC/SC-level failure. pyscard sets ``hresult`` on its exceptions (card-level errors such as - pySim's ``SwMatchError`` do not). After such a failure the PC/SC context - captured when the reader was opened is dead, so the transport has to be - recreated before the card can be used again.""" + pySim's ``SwMatchError`` do not).""" 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 = ( + 0x8010000C, # SCARD_E_NO_SMARTCARD + 0x80100066, # SCARD_W_UNRESPONSIVE_CARD + 0x80100067, # SCARD_W_UNPOWERED_CARD + 0x80100068, # SCARD_W_RESET_CARD + 0x80100069, # SCARD_W_REMOVED_CARD +) + + +def _is_transport_fatal(exc): + """True when a PC/SC failure means the transport itself is unusable. + + 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).""" + hr = getattr(exc, 'hresult', -1) + if hr in (-1, None): + return False + return hr not in _PCSC_CARD_LEVEL + + +def _clear_app_card_state(app): + """Unequip pySim's shell and drop its card/runtime-state references. + + The dead card must not stay referenced through ``app.card``/``app.rs``/ + ``app.lchan`` (handlers would keep transmitting over the removed card and + the old PC/SC link and its exclusive handle would never be released), but + the references cannot simply be nulled either: ``PysimApp.equip()`` + unregisters the previous profile's shell command sets from ``self.rs``, so + with ``rs`` already None the next equip aborts with "CommandSet ... is + already installed" and the file tree breaks. Route through pySim's own + unequip path (``equip(None, None)``) first, then clear the references.""" + if app is None: + return + if getattr(app, 'rs', None) is not None and callable(getattr(app, 'equip', None)): + old_stdout = getattr(app, 'stdout', None) + try: + app.stdout = StringIO() # mute the 'pySim-shell not equipped!' line + app.equip(None, None) + except Exception as e: + sys.stderr.write('UNEQUIP: pySim unequip failed: %s\n' % e) + finally: + if old_stdout is not None: + app.stdout = old_stdout + app.card = None + app.rs = None + app.lchan = None + + def _handle_card_disconnect(stale=False): """Tear down the card session; ``stale`` marks the transport as dead. - Called on card removal (no ``stale``) and on PC/SC errors (``stale``, see - ``_is_pcsc_error``): the next equip recreates the transport.""" + Called on card removal (no ``stale``) and on PC/SC errors + (``stale=_is_transport_fatal(e)``): the next equip recreates the + transport when it is really dead.""" global _CARD_CONNECTED, _TRANSPORT_STALE if stale: _TRANSPORT_STALE = True @@ -2685,6 +2738,7 @@ def _handle_card_disconnect(stale=False): _timer_cancel() _CARD_CONNECTED = False if _server_ref: + _clear_app_card_state(getattr(_server_ref, 'app', None)) _server_ref.card = None _server_ref.scc = None _server_ref.stk_pending = None @@ -2713,15 +2767,35 @@ def _ensure_transport(server): if factory is None: _TRANSPORT_STALE = False return True + # Release the previous link before building the new one: pyscard keeps the + # PC/SC context (and, while connected, an exclusive card handle) in the + # connection object. A second link created while the old one is still + # connected inherits the stale card handle and fails with + # SCARD_W_REMOVED_CARD on its first APDU. + app = getattr(server, 'app', None) + old = getattr(server, 'sl', None) + _clear_app_card_state(app) + server.sl = None + if app is not None: + app.sl = None + if old is not None: + try: + old.disconnect() + except Exception as e: + sys.stderr.write('TRANSPORT: releasing the old link failed: %s\n' % e) try: sl = factory() except Exception as e: sys.stderr.write('TRANSPORT: PC/SC reconnect failed: %s\n' % e) + # Keep the released link installed: a later equip can still reconnect + # it (PcscSimLink.connect() re-establishes the connection). + server.sl = old + if app is not None: + app.sl = old return False server.sl = sl server.card = None server.scc = None - app = getattr(server, 'app', None) if app is not None: app.sl = sl _TRANSPORT_STALE = False @@ -2865,6 +2939,12 @@ def _esim_verify_switch(app, action, iccid=None, isdp_aid=None): _AUTO_EQUIP = True _AUTO_EQUIP_BUSY = False +_AUTO_EQUIP_ATTEMPTS = 3 +_AUTO_EQUIP_RETRY_DELAY = 1.0 # seconds between attempts +_AUTO_EQUIP_REARM_DELAY = 5.0 # min seconds between watchdog re-arms +_AUTO_EQUIP_REARM_MAX = 60.0 # backoff ceiling after repeated failures +_AUTO_EQUIP_BACKOFF = _AUTO_EQUIP_REARM_DELAY +_AUTO_EQUIP_LAST = 0.0 def set_auto_equip(enabled): global _AUTO_EQUIP @@ -2874,44 +2954,104 @@ def _auto_equip_trigger(): """Spawn a one-shot worker; never run equip in the pyscard monitor thread.""" global _AUTO_EQUIP_BUSY if not _AUTO_EQUIP or _AUTO_EQUIP_BUSY: - return + return False _AUTO_EQUIP_BUSY = True threading.Thread(target=_auto_equip_worker, name='auto-equip', daemon=True).start() + return True + +def _auto_equip_rearm(now=None): + """Self-heal: re-arm auto-equip when a card is present but the session is + down (failed attempt, missed insertion event). Rate-limited so a genuinely + broken card is not retried in a tight loop.""" + global _AUTO_EQUIP_LAST + if not _AUTO_EQUIP or _AUTO_EQUIP_BUSY: + return False + server = _server_ref + if server is None or _CARD_CONNECTED: + return False + if not getattr(server, 'card_present', False) or getattr(server, 'equipping', False): + return False + now = time.time() if now is None else now + if now - _AUTO_EQUIP_LAST < _AUTO_EQUIP_BACKOFF: + return False + _AUTO_EQUIP_LAST = now + return _auto_equip_trigger() + +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.""" + with _CARD_LOCK: + if _CARD_CONNECTED or not getattr(server, 'card_present', False): + return True + app = server.app + if app is None or not getattr(server, 'terminal_profile', None): + return True + # A pcscd restart kills the PC/SC context: rebuild the transport + # before equipping, otherwise the equip can never succeed. + if not _ensure_transport(server): + return False + server.equipping = True + try: + old_stdout, old_stderr = app.stdout, sys.stderr + app.stdout = StringIO() + sys.stderr = app.stdout + try: + app.onecmd_plus_hooks('equip') + finally: + 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 + _apply_equipped_card(server) + sys.stderr.write('AUTO-EQUIP: done\n') + return True + except Exception as e: + sys.stderr.write('AUTO-EQUIP failed: %s\n' % e) + # A dead transport must be rebuilt before the next attempt; + # card-level failures reconnect on the existing link. Unequip the + # half-initialized shell as well: a failed init can leave command + # sets registered that the next equip would refuse to re-register. + if _is_transport_fatal(e): + _handle_card_disconnect(stale=True) + _clear_app_card_state(app) + return False + finally: + server.equipping = False + +def _auto_equip_attempts(server, attempts=None, sleep_fn=None): + """Bounded equip retries; returns True once the session is up.""" + attempts = _AUTO_EQUIP_ATTEMPTS if attempts is None else attempts + sleep_fn = time.sleep if sleep_fn is None else sleep_fn + if server is None: + return False + if _CARD_CONNECTED or not getattr(server, 'card_present', False) or getattr(server, 'app', None) is None: + return True + for attempt in range(1, attempts + 1): + if attempt == 1: + sys.stderr.write('AUTO-EQUIP: card inserted, initializing\n') + else: + sys.stderr.write('AUTO-EQUIP: retry %d/%d\n' % (attempt, attempts)) + if _auto_equip_attempt(server): + return True + if not _AUTO_EQUIP or _CARD_CONNECTED or not getattr(server, 'card_present', False): + return False + if attempt < attempts: + sleep_fn(_AUTO_EQUIP_RETRY_DELAY) + return False def _auto_equip_worker(): - global _AUTO_EQUIP_BUSY + global _AUTO_EQUIP_BUSY, _AUTO_EQUIP_BACKOFF try: - with _CARD_LOCK: - server = _server_ref - if not server or _CARD_CONNECTED or not getattr(server, 'card_present', False): - return - app = server.app - if app is None or not getattr(server, 'terminal_profile', None): - return - # A pcscd restart kills the PC/SC context: rebuild the transport - # before equipping, otherwise the equip can never succeed. - if not _ensure_transport(server): - return - server.equipping = True - try: - sys.stderr.write('AUTO-EQUIP: card inserted, initializing\n') - old_stdout, old_stderr = app.stdout, sys.stderr - app.stdout = StringIO() - sys.stderr = app.stdout - try: - app.onecmd_plus_hooks('equip') - finally: - 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 - _apply_equipped_card(server) - sys.stderr.write('AUTO-EQUIP: done\n') - except Exception as e: - sys.stderr.write('AUTO-EQUIP failed: %s\n' % e) - finally: - server.equipping = False + ok = _auto_equip_attempts(_server_ref) + # A card that cannot be initialized at all must not be retried every + # few seconds forever; the delay doubles up to the ceiling and is + # reset by any successful (or no-op) attempt. + _AUTO_EQUIP_BACKOFF = (_AUTO_EQUIP_REARM_DELAY if ok + else min(_AUTO_EQUIP_BACKOFF * 2, _AUTO_EQUIP_REARM_MAX)) finally: _AUTO_EQUIP_BUSY = False @@ -3002,6 +3142,12 @@ def _start_card_watchdog(reader_name, interval=5.0): _watchdog_tick(reader_name) except Exception as e: sys.stderr.write('CARD-WATCH: monitor restart failed: %s\n' % e) + try: + # Self-heal: a card may be present while the session is down + # (failed auto-equip, missed insertion event). + _auto_equip_rearm() + except Exception as e: + sys.stderr.write('CARD-WATCH: auto-equip re-arm failed: %s\n' % e) thread = threading.Thread(target=run, name='card-watchdog', daemon=True) thread.start() @@ -3241,7 +3387,7 @@ def _timer_expired(timer_id, elapsed): data, sw = scc._tp.send_apdu(apdu) except Exception as e: sys.stderr.write('TIMER-EXPIRATION send error: %s\n' % e) - _handle_card_disconnect(stale=_is_pcsc_error(e)) + _handle_card_disconnect(stale=_is_transport_fatal(e)) return sys.stderr.write('ENVELOPE(Timer Expiration): timer=%d elapsed=%ds -> %s\n' % (timer_id, elapsed, sw)) @@ -4197,7 +4343,7 @@ class PysimHandler(BaseHTTPRequestHandler): self._log_resp(resp) except Exception as e: sys.stderr.write('STATUS poll error: %s\n' % e) - _handle_card_disconnect(stale=_is_pcsc_error(e)) + _handle_card_disconnect(stale=_is_transport_fatal(e)) self._send_json({'sw': None, 'error': 'card disconnected'}) self._log_resp({'sw': None, 'error': 'card disconnected'}) elif self.path == '/api/net-sim': @@ -4229,7 +4375,7 @@ class PysimHandler(BaseHTTPRequestHandler): self._log_resp({'error': str(e)}) except Exception as e: sys.stderr.write('Net-sim error: %s\n' % e) - _handle_card_disconnect(stale=_is_pcsc_error(e)) + _handle_card_disconnect(stale=_is_transport_fatal(e)) self._send_json({'error': 'simulation failed: %s' % e}, 500) self._log_resp({'error': str(e)}) elif self.path == '/api/net-state-refresh': @@ -4584,7 +4730,7 @@ class PysimHandler(BaseHTTPRequestHandler): except Exception as e: sys.stderr.write('Handler error: %s\n' % e) if _is_pcsc_error(e) or 'Card' in str(e) or 'Transaction' in str(e) or 'Transmit' in str(e): - _handle_card_disconnect(stale=_is_pcsc_error(e)) + _handle_card_disconnect(stale=_is_transport_fatal(e)) err = {'success': False, 'error': str(e), 'exists': False} self._send_json(err, 500) self._log_resp(err) @@ -4663,7 +4809,7 @@ class PysimHandler(BaseHTTPRequestHandler): self._log_resp(resp) except Exception as e: sys.stderr.write('Event send error: %s\n' % e) - _handle_card_disconnect(stale=_is_pcsc_error(e)) + _handle_card_disconnect(stale=_is_transport_fatal(e)) self._send_json({'sw': None, 'error': 'card disconnected'}) self._log_resp({'sw': None, 'error': 'card disconnected'}) elif self.path == '/api/pli-dict': @@ -4791,7 +4937,7 @@ class PysimHandler(BaseHTTPRequestHandler): except Exception as e: sys.stderr.write('OTA send error: %s\n' % e) if _is_pcsc_error(e) or 'Card' in str(e) or 'Transaction' in str(e) or 'Transmit' in str(e): - _handle_card_disconnect(stale=_is_pcsc_error(e)) + _handle_card_disconnect(stale=_is_transport_fatal(e)) err = {'success': False, 'error': str(e)} self._send_json(err, 500) self._log_resp(err) diff --git a/tests/test_card_recovery.py b/tests/test_card_recovery.py index 6ca8439..b3d1eb5 100644 --- a/tests/test_card_recovery.py +++ b/tests/test_card_recovery.py @@ -3,7 +3,9 @@ transport recreation after a service failure (pcscd restart).""" import unittest +from io import StringIO from types import SimpleNamespace +from unittest import mock from pysim_simple_server import server @@ -21,6 +23,27 @@ class PcscErrorTests(unittest.TestCase): self.assertFalse(server._is_pcsc_error(Exception())) +class TransportFatalTests(unittest.TestCase): + @staticmethod + def _exc(hresult): + exc = RuntimeError('pcsc') + exc.hresult = hresult + return exc + + def test_card_level_errors_do_not_rebuild_the_transport(self): + # A normal card swap (SCARD_W_REMOVED_CARD) is recovered by a plain + # reconnect on the existing link - it must not trigger a rebuild. + for hr in (0x80100069, 0x8010000C, 0x80100068, 0x80100066, 0x80100067): + self.assertFalse(server._is_transport_fatal(self._exc(hr)), hex(hr)) + + 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)) + + def test_non_pcsc_errors_are_not_transport_fatal(self): + self.assertFalse(server._is_transport_fatal(RuntimeError('SW match failed'))) + + class WatchdogTests(unittest.TestCase): def test_alive_monitor_is_left_alone(self): calls = [] @@ -77,6 +100,26 @@ class EnsureTransportTests(unittest.TestCase): self.assertTrue(server._TRANSPORT_STALE) self.assertEqual(srv.sl, 'old-sl') + def test_old_link_is_released_before_the_rebuild(self): + released = [] + + class FakeLink: + def disconnect(self): + released.append(True) + + old = FakeLink() + app = SimpleNamespace(sl=old, card='card', rs='rs', lchan='lchan') + srv = SimpleNamespace(sl=old, scc='scc', card='card', app=app, + transport_factory=lambda: 'new-sl') + server._TRANSPORT_STALE = True + self.assertTrue(server._ensure_transport(srv)) + self.assertEqual(released, [True]) + self.assertEqual(srv.sl, 'new-sl') + self.assertEqual(app.sl, 'new-sl') + self.assertIsNone(app.card) + self.assertIsNone(app.rs) + self.assertIsNone(app.lchan) + def test_no_factory_clears_the_flag(self): srv = SimpleNamespace(sl='old-sl', app=None, transport_factory=None) server._TRANSPORT_STALE = True @@ -103,3 +146,158 @@ class DisconnectTests(unittest.TestCase): def test_plain_disconnect_leaves_the_transport_usable(self): server._handle_card_disconnect() self.assertFalse(server._TRANSPORT_STALE) + + def test_disconnect_unequips_the_shell_first(self): + # PysimApp.equip(None, None) unregisters the old profile's command + # sets; without it the next equip fails with "CommandSet ... is + # already installed" and the file tree breaks. + calls = [] + app = SimpleNamespace(rs='rs', card='card', lchan='lchan', + stdout=StringIO(), + equip=lambda c, r: calls.append((c, r))) + srv = SimpleNamespace(app=app, card='card', scc='scc', stk_pending=None, + menu_active=False, event_list=None, sim_menu=None, + iccid=None, net_state=None, equipping=False, card_session=1) + saved = server._server_ref + server._server_ref = srv + try: + server._handle_card_disconnect() + finally: + server._server_ref = saved + self.assertEqual(calls, [(None, None)]) + self.assertIsNone(app.card) + self.assertIsNone(app.rs) + self.assertIsNone(app.lchan) + + def test_disconnect_clears_the_app_card_state(self): + # The dead card must not stay reachable through app.card/app.rs: + # handlers using app.rs kept transmitting over the removed card and + # the old PC/SC link stayed connected (the v3.5.1 auto-equip bug). + app = SimpleNamespace(card='card', rs='rs', lchan='lchan') + srv = SimpleNamespace(app=app, card='card', scc='scc', stk_pending='x', + menu_active=True, event_list='e', sim_menu='m', + iccid='123', net_state='n', equipping=True, card_session=5) + saved = server._server_ref + server._server_ref = srv + try: + server._handle_card_disconnect() + finally: + server._server_ref = saved + self.assertIsNone(app.card) + self.assertIsNone(app.rs) + self.assertIsNone(app.lchan) + self.assertIsNone(srv.card) + self.assertIsNone(srv.scc) + + +class AutoEquipTests(unittest.TestCase): + def setUp(self): + self.saved = (server._server_ref, server._CARD_CONNECTED, server._TRANSPORT_STALE, + server._AUTO_EQUIP, server._AUTO_EQUIP_BUSY, server._AUTO_EQUIP_LAST, + server._AUTO_EQUIP_BACKOFF) + server._server_ref = None + server._CARD_CONNECTED = False + server._TRANSPORT_STALE = False + server._AUTO_EQUIP = True + server._AUTO_EQUIP_BUSY = False + server._AUTO_EQUIP_LAST = 0.0 + server._AUTO_EQUIP_BACKOFF = server._AUTO_EQUIP_REARM_DELAY + + def tearDown(self): + (server._server_ref, server._CARD_CONNECTED, server._TRANSPORT_STALE, + server._AUTO_EQUIP, server._AUTO_EQUIP_BUSY, server._AUTO_EQUIP_LAST, + server._AUTO_EQUIP_BACKOFF) = self.saved + + @staticmethod + def make_server(onecmd): + app = SimpleNamespace(stdout=StringIO(), card=None) + app.onecmd_plus_hooks = onecmd + return SimpleNamespace(app=app, card=None, scc=None, card_present=True, + equipping=False, terminal_profile='tp', card_session=1) + + def test_retries_after_a_transient_card_level_failure(self): + state = {'calls': 0} + srv = self.make_server(None) + + def onecmd(cmd): + state['calls'] += 1 + if state['calls'] == 1: + raise RuntimeError('Failed to transmit with protocol T0. Card was removed.') + srv.app.card = SimpleNamespace(_scc='scc') + + srv.app.onecmd_plus_hooks = onecmd + applied = [] + with mock.patch.object(server, '_ensure_transport', lambda s: True), \ + mock.patch.object(server, '_apply_equipped_card', lambda s: applied.append(s)): + self.assertTrue(server._auto_equip_attempts(srv, 3, lambda s: None)) + self.assertEqual(state['calls'], 2) + self.assertEqual(applied, [srv]) + self.assertFalse(server._TRANSPORT_STALE) + + def test_card_level_failure_keeps_the_transport(self): + srv = self.make_server(None) + + def onecmd(cmd): + exc = RuntimeError('Card was removed.') + exc.hresult = 0x80100069 + raise exc + + srv.app.onecmd_plus_hooks = onecmd + with mock.patch.object(server, '_ensure_transport', lambda s: True): + self.assertFalse(server._auto_equip_attempt(srv)) + self.assertFalse(server._TRANSPORT_STALE) + + def test_transport_fatal_failure_marks_the_transport(self): + srv = self.make_server(None) + + def onecmd(cmd): + exc = RuntimeError('service not available') + exc.hresult = 0x8010001D + raise exc + + srv.app.onecmd_plus_hooks = onecmd + with mock.patch.object(server, '_ensure_transport', lambda s: True): + self.assertFalse(server._auto_equip_attempt(srv)) + self.assertTrue(server._TRANSPORT_STALE) + + def test_rearm_triggers_with_cooldown(self): + calls = [] + server._server_ref = SimpleNamespace(card_present=True, equipping=False) + with mock.patch.object(server, '_auto_equip_trigger', + lambda: calls.append(1) or True): + self.assertTrue(server._auto_equip_rearm(now=100.0)) + self.assertFalse(server._auto_equip_rearm(now=100.5)) + self.assertTrue(server._auto_equip_rearm(now=106.0)) + self.assertEqual(len(calls), 2) + + def test_failed_worker_backs_off_the_rearm(self): + state = {'calls': 0} + srv = self.make_server(None) + + def onecmd(cmd): + state['calls'] += 1 + raise RuntimeError('card cannot be initialized') + + srv.app.onecmd_plus_hooks = onecmd + server._server_ref = srv + with mock.patch.object(server, '_ensure_transport', lambda s: True), \ + mock.patch.object(server, '_apply_equipped_card', lambda s: None), \ + mock.patch.object(server, '_AUTO_EQUIP_ATTEMPTS', 1): + server._auto_equip_worker() + self.assertEqual(state['calls'], 1) + self.assertGreater(server._AUTO_EQUIP_BACKOFF, server._AUTO_EQUIP_REARM_DELAY) + + def test_rearm_needs_a_present_card_and_no_session(self): + calls = [] + server._server_ref = SimpleNamespace(card_present=True, equipping=False) + with mock.patch.object(server, '_auto_equip_trigger', + lambda: calls.append(1) or True): + server._CARD_CONNECTED = True + self.assertFalse(server._auto_equip_rearm(now=100.0)) + server._CARD_CONNECTED = False + server._server_ref.card_present = False + self.assertFalse(server._auto_equip_rearm(now=100.0)) + server._server_ref.card_present = True + server._server_ref.equipping = True + self.assertFalse(server._auto_equip_rearm(now=100.0)) + self.assertEqual(calls, []) diff --git a/tests/test_fastinit.py b/tests/test_fastinit.py index 914968b..b34c5d9 100644 --- a/tests/test_fastinit.py +++ b/tests/test_fastinit.py @@ -214,6 +214,30 @@ class TestInitCardFastRetry(unittest.TestCase): self.assertEqual((rs, card), ('rs', 'card')) +class TestInitCardFastLinkRetry(unittest.TestCase): + def test_link_error_retries_after_reconnect(self): + from smartcard.Exceptions import CardConnectionException + calls = {'once': 0, 'disconnects': 0} + + class FakeLink: + def disconnect(self): + calls['disconnects'] += 1 + + def once(sl, skip, wait): + calls['once'] += 1 + if calls['once'] == 1: + raise CardConnectionException( + 'Failed to transmit with protocol T0. Card was removed.', + hresult=0x80100069) + return ('rs', 'card') + + with mock.patch.object(fastinit, '_init_card_once', side_effect=once): + rs, card = fastinit.init_card_fast(FakeLink(), wait=True) + self.assertEqual(calls['once'], 2) + self.assertEqual(calls['disconnects'], 1) + self.assertEqual((rs, card), ('rs', 'card')) + + class TestDoEquipFastFailure(unittest.TestCase): def test_failed_equip_keeps_previous_state(self): calls = []