Compare commits

...

2 Commits

Author SHA1 Message Date
catarrh fe10603aea server+ui: auto-equip on card insertion; reset card views on session change
- _apply_equipped_card() centralizes the post-equip refresh + TERMINAL
  PROFILE (shared by the /api/command equip branch and auto-equip)
- server tracks card_session (bumped on equip and disconnect) and
  equipping; /api/status exposes connected, card_present, card_session,
  equipping, auto_equip and is exempt from _CARD_LOCK (pure cached state)
- auto-equip is on by default (--no-auto-equip; off with --no-card-init):
  the presence observer spawns a one-shot worker after insertion, which
  runs equip under _CARD_LOCK and applies the same refresh; the monitor
  starts after the startup init so pyscard's initial 'already present'
  event does not re-equip a fresh session
- UI: /api/status polls every 2s (other views stay at 5s); when
  card_session changes it runs pysimResetCardData() (STK overlay, file
  tree, events, proactive log, PLI, status) — the same reset as a manual
  Equip; messages: initializing / press Equip / no card

Tests for the observer and auto-equip rules, session bumps,
_apply_equipped_card, and the UI state machine. SW cache v91 -> v92.
2026-09-12 13:34:40 +03:00
catarrh 57eb412b6d 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.
2026-09-12 13:21:33 +03:00
8 changed files with 462 additions and 29 deletions
+1
View File
@@ -628,6 +628,7 @@ pysim-otaman-server --http-port 8080
| `--terminal-profile` | TERMINAL PROFILE payload hex (default 10-byte GSM profile) | | `--terminal-profile` | TERMINAL PROFILE payload hex (default 10-byte GSM profile) |
| `--poll-interval` | Idle interval before automatic STATUS polling (default 30s; `0` disables polling) | | `--poll-interval` | Idle interval before automatic STATUS polling (default 30s; `0` disables polling) |
| `--full-pysim-init` | Use pysim's stock init/equip (redundant card resets). The default init/equip is reset-free — only explicit equip/reset reconnect the card | | `--full-pysim-init` | Use pysim's stock init/equip (redundant card resets). The default init/equip is reset-free — only explicit equip/reset reconnect the card |
| `--no-auto-equip` | Do not initialize a card automatically right after it is inserted (default: auto-equip on) |
| `--menu-timeout` | Auto-answer a paused STK command with a timeout TERMINAL RESPONSE (default 60s; `0` disables) | | `--menu-timeout` | Auto-answer a paused STK command with a timeout TERMINAL RESPONSE (default 60s; `0` disables) |
| `--timing` | Log phase durations, card resets and APDU counters with elapsed timestamps | | `--timing` | Log phase durations, card resets and APDU counters with elapsed timestamps |
+1
View File
@@ -608,6 +608,7 @@ pysim-otaman-server --http-port 8080
| `--log-requests` | Лог запросов/ответов в stderr | | `--log-requests` | Лог запросов/ответов в stderr |
| `--poll-interval` | Интервал автоопроса STATUS (по умолчанию 30с; `0` отключает опрос) | | `--poll-interval` | Интервал автоопроса STATUS (по умолчанию 30с; `0` отключает опрос) |
| `--full-pysim-init` | Штатная инициализация/equip из pysim (с лишними сбросами карты). По умолчанию инициализация без лишних сбросов — карта переподключается только по явным equip/reset | | `--full-pysim-init` | Штатная инициализация/equip из pysim (с лишними сбросами карты). По умолчанию инициализация без лишних сбросов — карта переподключается только по явным equip/reset |
| `--no-auto-equip` | Не инициализировать карту автоматически сразу после вставки (по умолчанию автоинициализация включена) |
| `--menu-timeout` | Автоответ timeout TERMINAL RESPONSE на приостановленную STK-команду (по умолчанию 60с; `0` отключает) | | `--menu-timeout` | Автоответ timeout TERMINAL RESPONSE на приостановленную STK-команду (по умолчанию 60с; `0` отключает) |
| `--timing` | Лог длительности фаз, сбросов карты и счётчиков APDU с отметками времени | | `--timing` | Лог длительности фаз, сбросов карты и счётчиков APDU с отметками времени |
+52 -9
View File
@@ -4455,6 +4455,7 @@ async function pysimConnect() {
function pysimDisconnect() { function pysimDisconnect() {
if (_pysimPollTimer) { clearInterval(_pysimPollTimer); _pysimPollTimer = null; } if (_pysimPollTimer) { clearInterval(_pysimPollTimer); _pysimPollTimer = null; }
if (_pysimStatusTimer) { clearInterval(_pysimStatusTimer); _pysimStatusTimer = null; }
pysimShowConnected(false); pysimShowConnected(false);
document.getElementById('pysim-status').textContent = ''; document.getElementById('pysim-status').textContent = '';
pysimSetConnected(false); pysimSetConnected(false);
@@ -6400,6 +6401,9 @@ async function pysimStatusPoll() {
} }
let _pysimPollTimer = null; let _pysimPollTimer = null;
let _pysimStatusTimer = null;
let _pysimCardStateKey = null;
let _pysimCardSession = null;
async function pysimPollToggle() { async function pysimPollToggle() {
const btn = document.getElementById('pli-pause-btn'); const btn = document.getElementById('pli-pause-btn');
@@ -6435,9 +6439,54 @@ async function pysimPollStatusInit() {
} catch (e) { /* ignore */ } } catch (e) { /* ignore */ }
} }
async function pysimResetCardData(refreshStatus) {
stkMenuClose();
stkMenuStack = [];
pysimFsRefresh();
stkCheckMenu();
pysimEventsRender();
pysimProactiveLogRender();
pysimPliRender();
if (refreshStatus) await pysimRefresh();
}
function pysimCardStateUpdate(status) {
if (!status || typeof status.connected !== 'boolean') return;
const key = [status.connected, !!status.card_present, !!status.equipping, !!status.auto_equip, status.card_session].join('|');
if (key === _pysimCardStateKey) return;
_pysimCardStateKey = key;
const sessionChanged = _pysimCardSession !== null
&& status.card_session !== undefined && status.card_session !== _pysimCardSession;
if (status.card_session !== undefined) _pysimCardSession = status.card_session;
if (status.connected) {
pysimSetConnected(true);
pysimResetCardData(true);
return;
}
pysimSetConnected(false);
const statusEl = document.getElementById('pysim-status');
if (statusEl && statusEl.textContent.trim() !== '') {
if (status.equipping || status.auto_equip) {
statusEl.innerHTML = '<span class="text-gray-500">' + esc(t('Card inserted — initializing...')) + '</span>';
} else 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>';
}
}
if (sessionChanged) pysimResetCardData(false);
}
function pysimStartBackendPoll() { function pysimStartBackendPoll() {
if (_pysimPollTimer) clearInterval(_pysimPollTimer); if (_pysimPollTimer) clearInterval(_pysimPollTimer);
let _hadData = false; if (_pysimStatusTimer) clearInterval(_pysimStatusTimer);
_pysimCardStateKey = null;
_pysimCardSession = null;
_pysimStatusTimer = setInterval(async () => {
try {
pysimCardStateUpdate(await pysimFetch('/api/status'));
} catch (e) { /* ignore */ }
}, 2000);
_pysimPollTimer = setInterval(async () => { _pysimPollTimer = setInterval(async () => {
try { try {
const [log, stk, ps] = await Promise.all([ const [log, stk, ps] = await Promise.all([
@@ -6446,14 +6495,6 @@ function pysimStartBackendPoll() {
pysimFetch('/api/poll-status'), pysimFetch('/api/poll-status'),
]); ]);
pysimUpdatePollUI(ps.enabled, ps.interval); pysimUpdatePollUI(ps.enabled, ps.interval);
if (Array.isArray(log) && log.length > 0) _hadData = true;
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); }, 5000);
} }
@@ -8746,6 +8787,8 @@ 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': 'Карта вставлена — нажмите «Подключить карту»',
'Card inserted — initializing...': 'Карта вставлена — инициализация...',
'Checking...': 'Проверка...', 'Checking...': 'Проверка...',
'Resetting...': 'Сброс...', 'Resetting...': 'Сброс...',
'Equipping...': 'Подключение...', 'Equipping...': 'Подключение...',
+1 -1
View File
@@ -1,4 +1,4 @@
const CACHE = 'otaman-v90'; const CACHE = 'otaman-v92';
const URLS = [ const URLS = [
'index.html', 'index.html',
'help.html', 'help.html',
+100
View File
@@ -0,0 +1,100 @@
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 _pysimCardStateKey = null;\nvar _pysimCardSession = 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: [], resets: [], refreshStatus: [] };
_pysimCardStateKey = null;
_pysimCardSession = null;
globalThis.document = { getElementById: () => el };
globalThis.pysimSetConnected = v => calls.connected.push(v);
globalThis.pysimResetCardData = refresh => calls.resets.push(refresh);
return { el, calls };
}
function status(extra) {
return Object.assign({ connected: false, card_present: false, equipping: false, auto_equip: false, card_session: 1 }, extra);
}
test('disconnect without card shows the no-card message', () => {
const { el, calls } = setup();
pysimCardStateUpdate(status({}));
assert.deepStrictEqual(calls.connected, [false]);
assert.ok(el.innerHTML.includes('No card detected'), el.innerHTML);
});
test('disconnect with card present shows the Equip hint when auto-equip is off', () => {
const { el } = setup();
pysimCardStateUpdate(status({ card_present: true }));
assert.ok(el.innerHTML.includes('Card inserted — press Equip'), el.innerHTML);
});
test('disconnect with auto-equip shows the initializing message', () => {
const { el } = setup();
pysimCardStateUpdate(status({ card_present: true, auto_equip: true }));
assert.ok(el.innerHTML.includes('initializing'), el.innerHTML);
});
test('unchanged state key does not touch the UI again', () => {
const { el, calls } = setup();
pysimCardStateUpdate(status({ card_session: 7 }));
el.innerHTML = 'unchanged';
calls.connected.length = 0;
pysimCardStateUpdate(status({ card_session: 7 }));
assert.deepStrictEqual(calls.connected, []);
assert.strictEqual(el.innerHTML, 'unchanged');
});
test('connected restores the UI and reloads card data', () => {
const { calls } = setup();
pysimCardStateUpdate(status({ connected: true, card_present: true, card_session: 2 }));
assert.deepStrictEqual(calls.connected, [true]);
assert.deepStrictEqual(calls.resets, [true]);
});
test('card session change triggers a data reset', () => {
const { calls } = setup();
pysimCardStateUpdate(status({ card_session: 3 }));
calls.resets.length = 0;
pysimCardStateUpdate(status({ card_session: 4 }));
assert.deepStrictEqual(calls.resets, [false]);
});
test('first observation does not trigger a reset on its own', () => {
const { calls } = setup();
pysimCardStateUpdate(status({ card_session: 9 }));
assert.deepStrictEqual(calls.resets, []);
});
test('payload without connected flag is ignored', () => {
const { calls } = setup();
pysimCardStateUpdate({ reader: 'x' });
pysimCardStateUpdate(null);
assert.deepStrictEqual(calls.connected, []);
});
+15 -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, set_auto_equip
_server_start = 0 _server_start = 0
@@ -55,6 +55,8 @@ def main():
help="Use pysim's stock init_card/equip (multiple physical card resets) instead of the default reset-free fast init") help="Use pysim's stock init_card/equip (multiple physical card resets) instead of the default reset-free fast init")
parser.add_argument('--menu-timeout', type=int, default=60, metavar='SECS', parser.add_argument('--menu-timeout', type=int, default=60, metavar='SECS',
help='Auto-send a timeout TERMINAL RESPONSE if a paused STK command is not answered (default: 60, 0 disables)') help='Auto-send a timeout TERMINAL RESPONSE if a paused STK command is not answered (default: 60, 0 disables)')
parser.add_argument('--no-auto-equip', action='store_true', default=False,
help='Do not automatically initialize a card right after it is inserted (default: auto-equip on)')
opts = parser.parse_args() opts = parser.parse_args()
opts.skip_card_init = opts.no_card_init opts.skip_card_init = opts.no_card_init
@@ -63,6 +65,7 @@ def main():
_timing_on() _timing_on()
if opts.menu_timeout is not None: if opts.menu_timeout is not None:
_set_menu_timeout(opts.menu_timeout) _set_menu_timeout(opts.menu_timeout)
set_auto_equip(not opts.no_auto_equip and not opts.skip_card_init)
sl = None sl = None
scc = None scc = None
card = None card = None
@@ -179,15 +182,24 @@ 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
server.card_session = 1 if card is not None else 0
server.equipping = False
# 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)
if server.scc and server.card and opts.poll_interval != 0: if server.scc and server.card and opts.poll_interval != 0:
pysim_otaman_server.server._poll_enable() pysim_otaman_server.server._poll_enable()
# Start presence monitoring only after the startup init: pyscard reports an
# already-present card as "added" on the first pass, and we must not
# auto-equip over a session we just initialized. If startup init failed,
# that event triggers auto-equip instead — the desired retry.
if sl is not None and getattr(sl, '_reader', None) is not None:
start_card_monitor(str(sl._reader))
print("" * 70) print("" * 70)
print(" pysim-otaman-server v%s listening on http://%s:%s" % (VERSION, opts.http_host, opts.http_port)) print(" pysim-otaman-server v%s listening on http://%s:%s" % (VERSION, opts.http_host, opts.http_port))
print(" Open http://%s:%s in your browser for the OTAMan UI (served by this server)." print(" Open http://%s:%s in your browser for the OTAMan UI (served by this server)."
+142 -16
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
@@ -1031,9 +1032,128 @@ def _handle_card_disconnect():
_server_ref.menu_active = False _server_ref.menu_active = False
_server_ref.event_list = None _server_ref.event_list = None
_server_ref.sim_menu = None _server_ref.sim_menu = None
_server_ref.equipping = False
_server_ref.card_session = getattr(_server_ref, 'card_session', 0) + 1
_reset_proactive_log() _reset_proactive_log()
def _apply_equipped_card(server):
"""Common post-equip state refresh + TERMINAL PROFILE, shared by the
/api/command equip branch and the auto-equip worker."""
global _CARD_CONNECTED
server.stk_pending = None
server.menu_active = False
_cancel_menu_timeout()
server.event_list = None
_reset_proactive_log()
server.card = server.app.card
server.scc = server.app.card._scc
server.scc.cat_cla = '80' if isinstance(server.card, UiccCardBase) else 'a0'
_CARD_CONNECTED = True
server.card_present = True
server.card_session = getattr(server, 'card_session', 0) + 1
_poll_enable()
sm, el = _send_terminal_profile(server.scc, server.terminal_profile)
server.sim_menu = sm
server.event_list = el
_tlog('equip: terminal profile done')
_AUTO_EQUIP = True
_AUTO_EQUIP_BUSY = False
def set_auto_equip(enabled):
global _AUTO_EQUIP
_AUTO_EQUIP = bool(enabled)
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
_AUTO_EQUIP_BUSY = True
threading.Thread(target=_auto_equip_worker, name='auto-equip', daemon=True).start()
def _auto_equip_worker():
global _AUTO_EQUIP_BUSY
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
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
finally:
_AUTO_EQUIP_BUSY = False
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:
trigger_auto = False
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\n' % self.reader_name)
with _CARD_LOCK:
if _server_ref:
_server_ref.card_present = True
trigger_auto = True
if trigger_auto and _AUTO_EQUIP:
_auto_equip_trigger()
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()
@@ -1508,6 +1628,13 @@ class PysimHandler(BaseHTTPRequestHandler):
self.wfile.write(data) self.wfile.write(data)
def do_GET(self): def do_GET(self):
# /api/status is pure cached state (no card I/O); keeping it out of the
# lock lets the UI report 'initializing' while a long equip holds the
# card lock. Result-shaping masks everything card-derived when the
# session is not connected.
if self.path == '/api/status':
self._do_GET()
return
# Serialize all card access: the background STATUS poll runs in its own # Serialize all card access: the background STATUS poll runs in its own
# thread and must never interleave with a FETCH/TERMINAL RESPONSE pair. # thread and must never interleave with a FETCH/TERMINAL RESPONSE pair.
with _CARD_LOCK: with _CARD_LOCK:
@@ -1526,9 +1653,22 @@ 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_session': int(getattr(self.server, 'card_session', 0)),
'equipping': bool(getattr(self.server, 'equipping', False)),
'auto_equip': bool(_AUTO_EQUIP),
'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,
@@ -1662,21 +1802,7 @@ class PysimHandler(BaseHTTPRequestHandler):
if is_equip: if is_equip:
_tlog('equip: onecmd_plus_hooks %dms' % elapsed) _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.app.card and self.server.terminal_profile:
global _CARD_CONNECTED _apply_equipped_card(self.server)
self.server.stk_pending = None
self.server.menu_active = False
_cancel_menu_timeout()
self.server.event_list = None
_reset_proactive_log()
self.server.card = self.server.app.card
self.server.scc = self.server.app.card._scc
self.server.scc.cat_cla = '80' if isinstance(self.server.card, UiccCardBase) else 'a0'
_CARD_CONNECTED = True
_poll_enable()
sm, el = _send_terminal_profile(self.server.scc, self.server.terminal_profile)
self.server.sim_menu = sm
self.server.event_list = el
_tlog('equip: terminal profile done')
sys.stderr.write("CMD: %s%s (%dms)\n" % (cmd, status, elapsed)) sys.stderr.write("CMD: %s%s (%dms)\n" % (cmd, status, elapsed))
resp = {'output': output, 'stop': bool(stop)} resp = {'output': output, 'stop': bool(stop)}
self._send_json(resp) self._send_json(resp)
+150
View File
@@ -0,0 +1,150 @@
#!/usr/bin/env python3
"""Tests for the passive PC/SC card-presence observer and auto-equip state."""
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()
self.trigger = mock.patch.object(S, '_auto_equip_trigger')
self.trigger_mock = self.trigger.start()
self.saved_auto = S._AUTO_EQUIP
S._AUTO_EQUIP = True
def tearDown(self):
S._AUTO_EQUIP = self.saved_auto
self.trigger.stop()
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)
self.trigger_mock.assert_not_called()
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, [])
self.trigger_mock.assert_not_called()
def test_insertion_sets_card_present_and_triggers_auto_equip(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, [])
self.trigger_mock.assert_called_once()
def test_insertion_does_not_trigger_when_disabled(self):
S._AUTO_EQUIP = False
self.server.card_present = False
self.observer.update(None, ([FakeCard('Test Reader 00 00')], []))
self.assertTrue(self.server.card_present)
self.trigger_mock.assert_not_called()
def test_missing_reader_attribute_is_ignored(self):
self.observer.update(None, ([], [types.SimpleNamespace()]))
self.assertTrue(self.server.card_present)
self.assertEqual(self.disconnects, [])
self.trigger_mock.assert_not_called()
class TestAutoEquipTrigger(unittest.TestCase):
def tearDown(self):
S._AUTO_EQUIP = True
S._AUTO_EQUIP_BUSY = False
def test_disabled_does_not_spawn(self):
S._AUTO_EQUIP = False
with mock.patch.object(S.threading, 'Thread') as thread:
S._auto_equip_trigger()
thread.assert_not_called()
def test_busy_does_not_spawn_twice(self):
S._AUTO_EQUIP = True
S._AUTO_EQUIP_BUSY = True
with mock.patch.object(S.threading, 'Thread') as thread:
S._auto_equip_trigger()
thread.assert_not_called()
def test_spawns_worker_once(self):
S._AUTO_EQUIP = True
S._AUTO_EQUIP_BUSY = False
with mock.patch.object(S.threading, 'Thread') as thread:
S._auto_equip_trigger()
thread.assert_called_once_with(target=S._auto_equip_worker, name='auto-equip', daemon=True)
class TestCardSession(unittest.TestCase):
def test_disconnect_bumps_session_and_clears_equipping(self):
server = types.SimpleNamespace(
card_session=5, card=object(), scc=object(), stk_pending=object(),
menu_active=True, event_list=[1], sim_menu={}, equipping=True)
saved = S._server_ref
S._server_ref = server
try:
S._handle_card_disconnect()
finally:
S._server_ref = saved
self.assertEqual(server.card_session, 6)
self.assertFalse(server.equipping)
self.assertIsNone(server.card)
class TestApplyEquippedCard(unittest.TestCase):
def test_updates_state_and_bumps_session(self):
scc = types.SimpleNamespace(cat_cla=None)
card = types.SimpleNamespace(_scc=scc, name='Card')
app = types.SimpleNamespace(card=card)
server = types.SimpleNamespace(
app=app, card=None, scc=None, stk_pending=object(), menu_active=True,
event_list=[1], sim_menu={}, card_session=2, card_present=False,
equipping=False, terminal_profile='7F')
saved_ref, saved_conn = S._server_ref, S._CARD_CONNECTED
S._server_ref = server
S._CARD_CONNECTED = False
try:
with mock.patch.object(S, '_send_terminal_profile', return_value=('menu', ['ev'])):
with mock.patch.object(S, '_poll_enable'):
S._apply_equipped_card(server)
connected_after = S._CARD_CONNECTED
finally:
S._server_ref = saved_ref
S._CARD_CONNECTED = saved_conn
self.assertTrue(connected_after)
self.assertIs(server.card, card)
self.assertIs(server.scc, scc)
self.assertEqual(server.card_session, 3)
self.assertTrue(server.card_present)
self.assertEqual(server.sim_menu, 'menu')
self.assertEqual(server.event_list, ['ev'])
if __name__ == '__main__':
unittest.main()