server+ui: detect card removal passively and reflect it within 2s

The UI only noticed a removed card when some user action ran a real card
command (e.g. Check status); /api/status is a cached-state read that kept
returning the old card, and _handle_card_disconnect() did not clear
app.card/rs.

- start_card_monitor() registers a pyscard CardObserver for our reader;
  it only polls SCardGetStatusChange (no APDU, no connection, no extra
  process), and on removal sets server.card_present=False and calls
  _handle_card_disconnect() under _CARD_LOCK
- /api/status now exposes connected (session usable) and card_present
  (physically inserted) and masks card/profile/atr/selection when not
  connected; _CARD_CONNECTED is initialized from card presence instead of
  being unconditionally True
- the 2s UI poll includes /api/status; on disconnect it switches to the
  existing 'No card detected. Insert card and click Equip' state, or the
  new 'Card inserted — press Equip' hint when the card is back; the old
  _hadData heuristic is gone

Tests for the observer (filtering, removal, insertion) and the UI state
transitions. SW cache v90 -> v91.
This commit is contained in:
2026-09-12 13:21:33 +03:00
parent 7288c22830
commit 57eb412b6d
6 changed files with 228 additions and 16 deletions
+27 -11
View File
@@ -6400,6 +6400,7 @@ async function pysimStatusPoll() {
} }
let _pysimPollTimer = null; let _pysimPollTimer = null;
let _pysimLastConnected = null;
async function pysimPollToggle() { async function pysimPollToggle() {
const btn = document.getElementById('pli-pause-btn'); const btn = document.getElementById('pli-pause-btn');
@@ -6435,27 +6436,41 @@ async function pysimPollStatusInit() {
} catch (e) { /* ignore */ } } catch (e) { /* ignore */ }
} }
function pysimCardStateUpdate(status) {
if (!status || typeof status.connected !== 'boolean') return;
const statusEl = document.getElementById('pysim-status');
if (status.connected === _pysimLastConnected) return;
_pysimLastConnected = status.connected;
if (status.connected) {
pysimSetConnected(true);
pysimRefresh();
return;
}
pysimSetConnected(false);
if (statusEl && statusEl.textContent.trim() !== '') {
if (status.card_present) {
statusEl.innerHTML = '<span class="text-amber-600">' + esc(t('Card inserted — press Equip')) + '</span>';
} else {
statusEl.innerHTML = '<span class="text-red-500">' + esc(t('No card detected. Insert card and click Equip card button')) + '</span>';
}
}
}
function pysimStartBackendPoll() { function pysimStartBackendPoll() {
if (_pysimPollTimer) clearInterval(_pysimPollTimer); if (_pysimPollTimer) clearInterval(_pysimPollTimer);
let _hadData = false; _pysimLastConnected = null;
_pysimPollTimer = setInterval(async () => { _pysimPollTimer = setInterval(async () => {
try { try {
const [log, stk, ps] = await Promise.all([ const [log, stk, ps, status] = await Promise.all([
pysimFetch('/api/proactive-log'), pysimFetch('/api/proactive-log'),
pysimFetch('/api/stk-status'), pysimFetch('/api/stk-status'),
pysimFetch('/api/poll-status'), pysimFetch('/api/poll-status'),
pysimFetch('/api/status'),
]); ]);
pysimUpdatePollUI(ps.enabled, ps.interval); pysimUpdatePollUI(ps.enabled, ps.interval);
if (Array.isArray(log) && log.length > 0) _hadData = true; pysimCardStateUpdate(status);
if (_hadData && !ps.enabled && Array.isArray(log) && log.length === 0) {
_hadData = false;
const statusEl = document.getElementById('pysim-status');
if (statusEl && statusEl.textContent.trim() !== '') {
statusEl.innerHTML = '<span class="text-red-500">Card disconnected</span>';
}
}
} catch (e) { /* ignore */ } } catch (e) { /* ignore */ }
}, 5000); }, 2000);
} }
// ===== proactive command log ===== // ===== proactive command log =====
@@ -8746,6 +8761,7 @@ const LANG_RU = {
'pysim-blocked-hint': 'Браузер, вероятно, заблокировал доступ к локальному серверу карт. Разрешите доступ к локальной сети для этого сайта (Chrome/Edge/Vivaldi: Настройки сайта → Доступ к локальной сети → Разрешить). Также проверьте, что pysim-otaman-server запущен по адресу {url}.', 'pysim-blocked-hint': 'Браузер, вероятно, заблокировал доступ к локальному серверу карт. Разрешите доступ к локальной сети для этого сайта (Chrome/Edge/Vivaldi: Настройки сайта → Доступ к локальной сети → Разрешить). Также проверьте, что pysim-otaman-server запущен по адресу {url}.',
'pysim-unreachable-hint': 'Не удалось подключиться к серверу карт. Проверьте, что pysim-otaman-server запущен и адрес указан верно ({url}). Если эта страница открыта по HTTPS, также разрешите доступ к локальной сети в браузере (Chrome/Edge/Vivaldi: Настройки сайта → Доступ к локальной сети → Разрешить).', 'pysim-unreachable-hint': 'Не удалось подключиться к серверу карт. Проверьте, что pysim-otaman-server запущен и адрес указан верно ({url}). Если эта страница открыта по HTTPS, также разрешите доступ к локальной сети в браузере (Chrome/Edge/Vivaldi: Настройки сайта → Доступ к локальной сети → Разрешить).',
'No card detected. Insert card and click Equip card button': 'Карта не обнаружена. Вставьте карту и нажмите Подключить карту', 'No card detected. Insert card and click Equip card button': 'Карта не обнаружена. Вставьте карту и нажмите Подключить карту',
'Card inserted — press Equip': 'Карта вставлена — нажмите «Подключить карту»',
'Checking...': 'Проверка...', 'Checking...': 'Проверка...',
'Resetting...': 'Сброс...', 'Resetting...': 'Сброс...',
'Equipping...': 'Подключение...', 'Equipping...': 'Подключение...',
+1 -1
View File
@@ -1,4 +1,4 @@
const CACHE = 'otaman-v90'; const CACHE = 'otaman-v91';
const URLS = [ const URLS = [
'index.html', 'index.html',
'help.html', 'help.html',
+77
View File
@@ -0,0 +1,77 @@
const { test } = require('node:test');
const assert = require('node:assert');
const fs = require('node:fs');
const path = require('node:path');
const html = fs.readFileSync(path.join(__dirname, '..', 'index.html'), 'utf8');
function extractFunc(src, name) {
const re = new RegExp('function\\s+' + name + '\\s*\\([^)]*\\)\\s*\\{');
const m = re.exec(src);
if (!m) throw new Error('function ' + name + ' not found');
let i = m.index + m[0].length - 1;
let depth = 0;
for (; i < src.length; i++) {
if (src[i] === '{') depth++;
else if (src[i] === '}') {
depth--;
if (depth === 0) break;
}
}
return src.slice(m.index, i + 1);
}
let code = 'var _pysimLastConnected = null;\n';
code += extractFunc(html, 'pysimCardStateUpdate') + '\n';
code += '\nglobalThis.esc = s => s;\n';
code += 'globalThis.t = s => s;\n';
eval(code);
function setup() {
const el = { textContent: 'status line', innerHTML: '' };
const calls = { connected: [], refresh: 0 };
globalThis.document = { getElementById: () => el };
globalThis.pysimSetConnected = v => calls.connected.push(v);
globalThis.pysimRefresh = () => { calls.refresh++; };
return { el, calls };
}
test('disconnect without card shows the no-card message', () => {
_pysimLastConnected = true;
const { el, calls } = setup();
pysimCardStateUpdate({ connected: false, card_present: false });
assert.deepStrictEqual(calls.connected, [false]);
assert.ok(el.innerHTML.includes('No card detected'), el.innerHTML);
});
test('disconnect with card present shows the Equip hint', () => {
_pysimLastConnected = true;
const { el } = setup();
pysimCardStateUpdate({ connected: false, card_present: true });
assert.ok(el.innerHTML.includes('Card inserted'), el.innerHTML);
});
test('unchanged state does not touch the UI again', () => {
_pysimLastConnected = false;
const { el, calls } = setup();
el.innerHTML = 'unchanged';
pysimCardStateUpdate({ connected: false, card_present: false });
assert.deepStrictEqual(calls.connected, []);
assert.strictEqual(el.innerHTML, 'unchanged');
});
test('reconnect restores the connected UI and refreshes', () => {
_pysimLastConnected = false;
const { calls } = setup();
pysimCardStateUpdate({ connected: true, card_present: true });
assert.deepStrictEqual(calls.connected, [true]);
assert.strictEqual(calls.refresh, 1);
});
test('payload without connected flag is ignored', () => {
_pysimLastConnected = null;
const { calls } = setup();
pysimCardStateUpdate({ reader: 'x' });
pysimCardStateUpdate(null);
assert.deepStrictEqual(calls.connected, []);
});
+6 -3
View File
@@ -12,7 +12,7 @@ from pySim.cards import UiccCardBase
from .shell import load_pysim_app from .shell import load_pysim_app
from . import fastinit 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 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
_server_start = 0 _server_start = 0
@@ -93,6 +93,8 @@ def main():
t_phase = time.time() t_phase = time.time()
sl = mod.init_reader(opts, **kwargs) sl = mod.init_reader(opts, **kwargs)
_tlog('init_reader: %.0fms' % ((time.time() - t_phase) * 1000)) _tlog('init_reader: %.0fms' % ((time.time() - t_phase) * 1000))
if getattr(sl, '_reader', None) is not None:
start_card_monitor(str(sl._reader))
scc = SimCardCommands(sl) scc = SimCardCommands(sl)
scc.cat_cla = '80' # UICC CLA default; overridden for SIM after init_card scc.cat_cla = '80' # UICC CLA default; overridden for SIM after init_card
scc._tp.proactive_handler = _DefaultProactiveHandler() scc._tp.proactive_handler = _DefaultProactiveHandler()
@@ -179,10 +181,11 @@ def main():
server.event_list = event_list server.event_list = event_list
server.menu_active = False server.menu_active = False
server.stk_pending = None server.stk_pending = None
# Set server reference for polling timer and mark card as connected server.card_present = card is not None
# Set server reference for polling timer and mark the card session state
import pysim_otaman_server.server import pysim_otaman_server.server
pysim_otaman_server.server._server_ref = server pysim_otaman_server.server._server_ref = server
pysim_otaman_server.server._CARD_CONNECTED = True pysim_otaman_server.server._CARD_CONNECTED = card is not None
if opts.poll_interval is not None: if opts.poll_interval is not None:
pysim_otaman_server.server._set_poll_interval(opts.poll_interval) pysim_otaman_server.server._set_poll_interval(opts.poll_interval)
# Auto-enable polling if card initialized successfully (unless interval is 0) # Auto-enable polling if card initialized successfully (unless interval is 0)
+55 -1
View File
@@ -11,6 +11,7 @@ from http.server import HTTPServer, BaseHTTPRequestHandler
from io import StringIO from io import StringIO
from pySim.transport import ApduTracer, ProactiveHandler from pySim.transport import ApduTracer, ProactiveHandler
from pySim.cards import UiccCardBase from pySim.cards import UiccCardBase
from smartcard.CardMonitoring import CardMonitor, CardObserver
import gsm0338 # registers 'gsm03.38' codec import gsm0338 # registers 'gsm03.38' codec
from construct import GreedyBytes from construct import GreedyBytes
@@ -1034,6 +1035,48 @@ def _handle_card_disconnect():
_reset_proactive_log() _reset_proactive_log()
class _CardPresenceObserver(CardObserver):
"""Passive PC/SC presence watcher: pyscard's CardMonitor only polls
SCardGetStatusChange (no connection, no APDUs), so it can never interleave
with our APDU traffic. We only update flags and tear down card state."""
def __init__(self, reader_name):
self.reader_name = reader_name
def update(self, observable, handlers):
addedcards, removedcards = handlers
try:
for card in removedcards:
if str(getattr(card, 'reader', '')) == self.reader_name:
sys.stderr.write('CARD-WATCH: card removed from %s\n' % self.reader_name)
with _CARD_LOCK:
if _server_ref:
_server_ref.card_present = False
_handle_card_disconnect()
for card in addedcards:
if str(getattr(card, 'reader', '')) == self.reader_name:
sys.stderr.write('CARD-WATCH: card inserted into %s (press Equip)\n' % self.reader_name)
with _CARD_LOCK:
if _server_ref:
_server_ref.card_present = True
except Exception as e:
sys.stderr.write('CARD-WATCH error: %s\n' % e)
_card_presence_observer = None
def start_card_monitor(reader_name):
"""Start the process-wide pyscard monitor (one daemon thread, no process)
and register our reader's presence observer."""
global _card_presence_observer
if not reader_name:
return None
if _card_presence_observer is None:
_card_presence_observer = _CardPresenceObserver(reader_name)
CardMonitor().addObserver(_card_presence_observer)
return _card_presence_observer
def _init_proactive_session(): def _init_proactive_session():
global _PROACTIVE_SESSION_START global _PROACTIVE_SESSION_START
_PROACTIVE_SESSION_START = time.time() _PROACTIVE_SESSION_START = time.time()
@@ -1526,9 +1569,19 @@ class PysimHandler(BaseHTTPRequestHandler):
lchan = rs.lchan[0] if rs else None lchan = rs.lchan[0] if rs else None
cur_file = lchan.selected_file if lchan else None cur_file = lchan.selected_file if lchan else None
scc = app.card._scc if app and app.card else None scc = app.card._scc if app and app.card else None
card = app.card if app else None
connected = bool(_CARD_CONNECTED and card is not None)
if not connected:
rs = None
lchan = None
cur_file = None
scc = None
card = None
data = { data = {
'reader': str(self.server.sl) if self.server.sl else None, 'reader': str(self.server.sl) if self.server.sl else None,
'card': app.card.name if app and app.card else None, 'connected': connected,
'card_present': bool(getattr(self.server, 'card_present', False)),
'card': card.name if card else None,
'profile': str(rs.profile) if rs and rs.profile else None, 'profile': str(rs.profile) if rs and rs.profile else None,
'app_ready': app is not None, 'app_ready': app is not None,
'adm_verified': rs.adm_verified if rs else False, 'adm_verified': rs.adm_verified if rs else False,
@@ -1672,6 +1725,7 @@ class PysimHandler(BaseHTTPRequestHandler):
self.server.scc = self.server.app.card._scc self.server.scc = self.server.app.card._scc
self.server.scc.cat_cla = '80' if isinstance(self.server.card, UiccCardBase) else 'a0' self.server.scc.cat_cla = '80' if isinstance(self.server.card, UiccCardBase) else 'a0'
_CARD_CONNECTED = True _CARD_CONNECTED = True
self.server.card_present = True
_poll_enable() _poll_enable()
sm, el = _send_terminal_profile(self.server.scc, self.server.terminal_profile) sm, el = _send_terminal_profile(self.server.scc, self.server.terminal_profile)
self.server.sim_menu = sm self.server.sim_menu = sm
+62
View File
@@ -0,0 +1,62 @@
#!/usr/bin/env python3
"""Tests for the passive PC/SC card-presence observer."""
import sys
import types
import unittest
from pathlib import Path
from unittest import mock
PROJECTS = Path(__file__).resolve().parents[2]
PY_SIM = PROJECTS / 'pysim'
if str(PY_SIM) not in sys.path:
sys.path.insert(0, str(PY_SIM))
import pysim_otaman_server.server as S
class FakeCard:
def __init__(self, reader):
self.reader = reader
class TestCardPresenceObserver(unittest.TestCase):
def setUp(self):
self.observer = S._CardPresenceObserver('Test Reader 00 00')
self.server = types.SimpleNamespace(card_present=True)
self.saved_ref = S._server_ref
S._server_ref = self.server
self.disconnects = []
self.patcher = mock.patch.object(
S, '_handle_card_disconnect',
side_effect=lambda: self.disconnects.append(True))
self.patcher.start()
def tearDown(self):
self.patcher.stop()
S._server_ref = self.saved_ref
def test_removal_of_our_reader_disconnects(self):
self.observer.update(None, ([], [FakeCard('Test Reader 00 00')]))
self.assertFalse(self.server.card_present)
self.assertEqual(len(self.disconnects), 1)
def test_removal_of_other_reader_ignored(self):
self.observer.update(None, ([], [FakeCard('Other Reader 00 00')]))
self.assertTrue(self.server.card_present)
self.assertEqual(self.disconnects, [])
def test_insertion_sets_card_present(self):
self.server.card_present = False
self.observer.update(None, ([FakeCard('Test Reader 00 00')], []))
self.assertTrue(self.server.card_present)
self.assertEqual(self.disconnects, [])
def test_missing_reader_attribute_is_ignored(self):
self.observer.update(None, ([], [types.SimpleNamespace()]))
self.assertTrue(self.server.card_present)
self.assertEqual(self.disconnects, [])
if __name__ == '__main__':
unittest.main()