Compare commits

..

5 Commits

Author SHA1 Message Date
catarrh 25787017e5 server: make reset-free fast init the default; add --full-pysim-init
Fast init (fastinit.py) is now the default for startup and equip: all
profile probes run on one connection, RuntimeState uses a software reset
and only explicit equip/reset reconnect the card. Measured on the
reference reader: equip 1729ms vs 9850ms and zero RESET events vs 7-8.

--full-pysim-init restores pysim's stock init_card/equip for
compatibility/debugging; the former --fast-init flag is kept as a hidden
no-op alias so existing command lines keep parsing. Docs updated.
2026-09-12 12:43:43 +03:00
catarrh b50be83da4 server: guarantee a TERMINAL RESPONSE for paused proactive commands
Every FETCHed proactive command must be answered, otherwise the card is
left in an unfinished session and stops issuing commands (e.g. it will
not deliver a PoR for SEND SM). The menu handlers now share
_menu_send_response, and a server-side watchdog (_arm_menu_timeout /
_menu_timeout_fire, --menu-timeout, default 60s, 0 disables) sends the
timeout result (0x12) when the user never answers. The timer is armed
while a command is pending and cancelled on any response, equip, rescue
and card disconnect.

Also fixes docs/api.md, which had back (0x11) and timeout (0x12) codes
swapped. Tests for arm/cancel/clamping and the flat timeout TR. SW cache
v89 -> v90.
2026-09-12 12:31:22 +03:00
catarrh 6c57ff4fd5 server: add --fast-init to skip redundant card resets
pysim's init_card() resets the card once per profile candidate in
CardProfile.pick(), once in RuntimeState.__init__ and again in
PysimApp.equip(); on typical readers each reconnect costs ~1.3s and a
normal init/equip does 7-8 of them (measured: equip 9.85s, startup
~14.8s).

fastinit.py mirrors pySim.app.init_card() with the resets removed:
pick_profile_no_reset() runs all profile probes back-to-back on one
connection, FastRuntimeState.reset() is a software reset (select MF,
clear selected_adf/scp, ATR from the transport) and the equip/reset
commands are routed through do_equip_fast (one reconnect via
wait_for_card) and do_reset_fast (always a physical reset). Enabled
with --fast-init; stock behavior remains the default.

Tests for the reset-free pick, the soft reset and the explicit reset
paths. Docs updated; SW cache v88 -> v89.
2026-09-12 12:27:03 +03:00
catarrh 895d0b7b36 server: serialize card access, fix poll-interval 0, add --timing
--poll-interval 0 now really disables background STATUS polling (it was
clamped to 1s, and the equip branch force-enabled it anyway). Request
handlers and the poll thread now share _CARD_LOCK so a poll can never
interleave a FETCH/TERMINAL RESPONSE pair — the baseline log showed
AUTO-STATUS chains and duplicate FETCHes inside the equip TP chain.

--timing adds elapsed timestamps, per-reset logging (RESET #n) and
phase durations for startup (init_reader, card_init, pysim_app,
terminal_profile_drain) and equip (onecmd, terminal profile).

AGENTS.md documents the proactive-TR invariant: never fetch without
answering, never fetch twice, one APDU conversation at a time.
2026-09-12 12:25:07 +03:00
catarrh ef00b2c3f4 docs: keep a local AGENTS.md out of the repo; fix RAM-install note
Add AGENTS.md to .gitignore so the local agent guide (goal, implemented
features, pySim usage) stays untracked. Correct the /api/ram-install
description: the .cap archive is parsed server-side by _cap_parse, not
by pySim.javacard/global_platform.
2026-09-12 11:46:32 +03:00
13 changed files with 656 additions and 108 deletions
+1
View File
@@ -4,3 +4,4 @@ __pycache__/
*.egg-info/
dist/
build/
AGENTS.md
+4 -1
View File
@@ -626,7 +626,10 @@ pysim-otaman-server --http-port 8080
| `--log-requests` | Log request/response payloads to stderr |
| `--sms-oa` / `--sms-sm-sc` | SMS-DELIVER originating address / SM-SC for PoR-in-submit |
| `--terminal-profile` | TERMINAL PROFILE payload hex (default 10-byte GSM profile) |
| `--poll-interval` | Idle interval before automatic STATUS polling (default 30s) |
| `--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 |
| `--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 |
### Troubleshooting
+4 -1
View File
@@ -606,7 +606,10 @@ pysim-otaman-server --http-port 8080
| `--no-card-init` | Пропустить инициализацию карты (сохранить CAT-сессию) |
| `--apdu-trace` | Лог APDU-трафика в stderr |
| `--log-requests` | Лог запросов/ответов в stderr |
| `--poll-interval` | Интервал автоопроса STATUS (по умолчанию 30с) |
| `--poll-interval` | Интервал автоопроса STATUS (по умолчанию 30с; `0` отключает опрос) |
| `--full-pysim-init` | Штатная инициализация/equip из pysim (с лишними сбросами карты). По умолчанию инициализация без лишних сбросов — карта переподключается только по явным equip/reset |
| `--menu-timeout` | Автоответ timeout TERMINAL RESPONSE на приостановленную STK-команду (по умолчанию 60с; `0` отключает) |
| `--timing` | Лог длительности фаз, сбросов карты и счётчиков APDU с отметками времени |
### Устранение неполадок
+7 -5
View File
@@ -143,7 +143,7 @@ The SPI2 `por_in_submit` bit (0x20) selects submit-mode PoR.
### `POST /api/ram-install`
Install a Java Card `.cap` file on the card via GlobalPlatform commands (INSTALL[for load] → LOAD ×N → INSTALL[for install (+ make selectable)]) wrapped in SCP80 secured packets. Each step is sent via ENVELOPE and its PoR is checked; the sequence aborts on the first PoR error. Requires pySim with `pySim.javacard.CapFile` and `pySim.global_platform` available on the server.
Install a Java Card `.cap` file on the card via GlobalPlatform commands (INSTALL[for load] → LOAD ×N → INSTALL[for install (+ make selectable)]) wrapped in SCP80 secured packets. Each step is sent via ENVELOPE and its PoR is checked; the sequence aborts on the first PoR error. The `.cap` archive (a ZIP of nested components) is parsed server-side in `_cap_parse`; no external tooling is required.
**Request body:**
```json
@@ -245,7 +245,9 @@ or
### `POST /api/menu-respond`
Sends `TERMINAL RESPONSE` to the current proactive command with the given result
code. Continues the proactive chain if the card responds with `91XX`.
code. Continues the proactive chain if the card responds with `91XX`. If no
response arrives within `--menu-timeout` seconds (default 60, `0` disables), the
server watchdog sends the `timeout` result itself.
```json
{"result": "ok", "item_id": 1}
@@ -254,9 +256,9 @@ code. Continues the proactive chain if the card responds with `91XX`.
| `result` | TERMINAL RESPONSE code | Meaning |
|---|---|---|
| `ok` | `0x00` | Command performed successfully |
| `back` | `0x12` | Backward move requested |
| `cancel` | `0x10` | Proactive session terminated |
| `timeout` | `0x11` | No response from user |
| `cancel` | `0x10` | Proactive session terminated by the user |
| `back` | `0x11` | Backward move in the proactive session requested by the user |
| `timeout` | `0x12` | No response from the user |
### `GET /api/stk-status`
+2 -2
View File
@@ -336,7 +336,7 @@
<p class="text-sm mb-3">Работа с сессией Card Application Toolkit: меню STK, подписанные события, журнал проактивных команд, словарь данных PROVIDE LOCAL INFORMATION и опрос STATUS.</p>
<h4 id="stk-menu" class="font-medium mb-1">5.5.1 Меню STK</h4>
<p class="text-sm mb-3">Если карта выдала команду SET UP MENU, вверху этого представления появляется блок &laquo;Меню STK&raquo; с изумрудной кнопкой <strong>STK: &lt;название&gt;</strong>, открывающей оверлей меню (браузер STK-меню карты). Если карта не задала меню, вместо кнопки показывается &laquo;Меню не задано картой&raquo;. Состояние меню обновляется при каждом открытии представления.</p>
<p class="text-sm mb-3">Если карта выдала команду SET UP MENU, вверху этого представления появляется блок &laquo;Меню STK&raquo; с изумрудной кнопкой <strong>STK: &lt;название&gt;</strong>, открывающей оверлей меню (браузер STK-меню карты). Если карта не задала меню, вместо кнопки показывается &laquo;Меню не задано картой&raquo;. Состояние меню обновляется при каждом открытии представления. Интерактивные проактивные команды всегда получают TERMINAL RESPONSE: оверлей ждёт вашего выбора, и если вы не ответили и не нажали <strong>Timeout</strong>, сервер сам отвечает результатом timeout через <code class="font-mono text-sm">--menu-timeout</code> секунд (по умолчанию 60, <code class="font-mono text-sm">0</code> отключает).</p>
<h4 id="subscribed-events" class="font-medium mb-1">5.5.2 Подписанные события (SET UP EVENT LIST)</h4>
<p class="text-sm mb-2">События, которые отслеживает карта. У каждого события есть кнопка <strong>Отправить</strong>, открывающая форму, специфичную для типа события:</p>
@@ -362,7 +362,7 @@
<p class="text-sm mb-3">Значения хранятся на сервере до перезапуска. Когда карта выдаёт PLI, сервер вставляет значения словаря в TERMINAL RESPONSE.</p>
<h4 id="status-polling" class="font-medium mb-1">5.5.5 Опрос STATUS</h4>
<p class="text-sm mb-3">Кнопка <strong>Отправить STATUS</strong> отправляет STATUS (F2) вручную. Переключатель <strong>Опрос</strong> включает фоновый опрос: после настраиваемого интервала бездействия (аргумент сервера <code class="font-mono text-sm">--poll-interval</code>, 1&ndash;255&nbsp;с, по умолчанию 30&nbsp;с) сервер отправляет STATUS и обрабатывает любую ожидающую проактивную команду. При извлечении карты опрос останавливается, а состояние карты сбрасывается.</p>
<p class="text-sm mb-3">Кнопка <strong>Отправить STATUS</strong> отправляет STATUS (F2) вручную. Переключатель <strong>Опрос</strong> включает фоновый опрос: после настраиваемого интервала бездействия (аргумент сервера <code class="font-mono text-sm">--poll-interval</code>, 1&ndash;255&nbsp;с, по умолчанию 30&nbsp;с, <code class="font-mono text-sm">0</code> отключает опрос) сервер отправляет STATUS и обрабатывает любую ожидающую проактивную команду. При извлечении карты опрос останавливается, а состояние карты сбрасывается.</p>
<h3 id="profiler" class="text-lg font-medium mb-2">5.6 Профайлер</h3>
<p class="text-sm mb-2">Проверяет соответствие карты именованному <strong>профилю</strong> — упорядоченному набору правил, описывающих ожидаемую файловую систему и (опционально) содержимое файлов. Профили хранятся в <code class="font-mono text-sm">localStorage</code>.</p>
+2 -2
View File
@@ -336,7 +336,7 @@
<p class="text-sm mb-3">Interacts with the Card Application Toolkit session: the STK menu, subscribed events, the proactive command log, the PROVIDE LOCAL INFORMATION data dictionary, and STATUS polling.</p>
<h4 id="stk-menu" class="font-medium mb-1">5.5.1 STK menu</h4>
<p class="text-sm mb-3">When the card has issued a SET UP MENU command, a &ldquo;STK menu&rdquo; block appears at the top of this view with an emerald <strong>STK: &lt;title&gt;</strong> button that opens the menu overlay (same as the card&rsquo;s STK menu browser). If the card has not set up a menu, the block shows &ldquo;No menu set by the card&rdquo; instead. The menu state is refreshed each time the view is opened.</p>
<p class="text-sm mb-3">When the card has issued a SET UP MENU command, a &ldquo;STK menu&rdquo; block appears at the top of this view with an emerald <strong>STK: &lt;title&gt;</strong> button that opens the menu overlay (same as the card&rsquo;s STK menu browser). If the card has not set up a menu, the block shows &ldquo;No menu set by the card&rdquo; instead. The menu state is refreshed each time the view is opened. User-interactive proactive commands always get a TERMINAL RESPONSE: the overlay pauses for your choice, and if you neither answer nor press <strong>Timeout</strong>, the server answers with a timeout result after the <code class="font-mono text-sm">--menu-timeout</code> seconds (default 60, <code class="font-mono text-sm">0</code> disables).</p>
<h4 id="subscribed-events" class="font-medium mb-1">5.5.2 Subscribed events (SET UP EVENT LIST)</h4>
<p class="text-sm mb-2">The events the card monitors. Each event has a <strong>Send</strong> button that opens a form specific to the event type:</p>
@@ -362,7 +362,7 @@
<p class="text-sm mb-3">Values persist server-side until restart. When the card issues PLI, the server injects the dictionary values into the TERMINAL RESPONSE.</p>
<h4 id="status-polling" class="font-medium mb-1">5.5.5 STATUS polling</h4>
<p class="text-sm mb-3">A <strong>Send STATUS</strong> button issues a manual STATUS (F2). A <strong>Polling</strong> toggle enables background polling: after a configurable idle interval (server CLI <code class="font-mono text-sm">--poll-interval</code>, 1&ndash;255&nbsp;s, default 30&nbsp;s) the server sends STATUS and handles any pending proactive command. Polling stops and card state resets if the card is removed.</p>
<p class="text-sm mb-3">A <strong>Send STATUS</strong> button issues a manual STATUS (F2). A <strong>Polling</strong> toggle enables background polling: after a configurable idle interval (server CLI <code class="font-mono text-sm">--poll-interval</code>, 1&ndash;255&nbsp;s, default 30&nbsp;s, <code class="font-mono text-sm">0</code> disables polling) the server sends STATUS and handles any pending proactive command. Polling stops and card state resets if the card is removed.</p>
<h3 id="profiler" class="text-lg font-medium mb-2">5.6 Profiler</h3>
<p class="text-sm mb-2">Verifies that a card matches a named <strong>profile</strong> — an ordered set of rules describing the expected file system and (optionally) file contents. Profiles are stored in <code class="font-mono text-sm">localStorage</code>.</p>
+1 -1
View File
@@ -1,4 +1,4 @@
const CACHE = 'otaman-v88';
const CACHE = 'otaman-v90';
const URLS = [
'index.html',
'help.html',
+29 -3
View File
@@ -11,7 +11,8 @@ from pySim.log import PySimLogger
from pySim.cards import UiccCardBase
from .shell import load_pysim_app
from .server import PysimHandler, StderrApduTracer, _LoggingApduTracer, VERSION, _send_terminal_profile, _DefaultProactiveHandler, _handle_proactive_chain, _send_status, _init_proactive_session
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
_server_start = 0
@@ -47,9 +48,21 @@ def main():
help='Idle interval before automatic STATUS polling (1-255 seconds, default: 30). Disable with --poll-interval 0')
parser.add_argument('--no-card-init', action='store_true', default=False,
help='Skip pysim card initialization (preserve CAT session — no file manager)')
parser.add_argument('--timing', action='store_true', default=False,
help='Log phase durations, card resets and APDU counters with elapsed timestamps')
parser.add_argument('--fast-init', action='store_true', help=argparse.SUPPRESS)
parser.add_argument('--full-pysim-init', action='store_true', default=False,
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',
help='Auto-send a timeout TERMINAL RESPONSE if a paused STK command is not answered (default: 60, 0 disables)')
opts = parser.parse_args()
opts.skip_card_init = opts.no_card_init
opts.fast_init = not opts.full_pysim_init
if opts.timing:
_timing_on()
if opts.menu_timeout is not None:
_set_menu_timeout(opts.menu_timeout)
sl = None
scc = None
card = None
@@ -77,27 +90,39 @@ def main():
kwargs = {}
if opts.apdu_trace:
kwargs['apdu_tracer'] = _LoggingApduTracer()
t_phase = time.time()
sl = mod.init_reader(opts, **kwargs)
_tlog('init_reader: %.0fms' % ((time.time() - t_phase) * 1000))
scc = SimCardCommands(sl)
scc.cat_cla = '80' # UICC CLA default; overridden for SIM after init_card
scc._tp.proactive_handler = _DefaultProactiveHandler()
sl.wait_for_card(3)
rs, card = mod.init_card(sl, opts.skip_card_init)
t_phase = time.time()
if opts.fast_init:
rs, card = fastinit.init_card_fast(sl, opts.skip_card_init, wait=True)
else:
sl.wait_for_card(3)
rs, card = mod.init_card(sl, opts.skip_card_init)
_tlog('card_init: %.0fms' % ((time.time() - t_phase) * 1000))
scc.cat_cla = '80' if isinstance(card, UiccCardBase) else 'a0'
except Exception:
print("Warning: reader/card initialization failed:", file=sys.stderr)
traceback.print_exc()
ch = CardHandler(sl) if sl else None
t_phase = time.time()
try:
app = mod.PysimApp(verbose=opts.verbose, card=card, rs=rs, sl=sl, ch=ch)
except Exception:
print("Warning: PysimApp creation failed:", file=sys.stderr)
traceback.print_exc()
app = None
_tlog('pysim_app: %.0fms' % ((time.time() - t_phase) * 1000))
if app is not None and opts.fast_init:
fastinit.install(app)
if scc and hasattr(scc, '_tp'):
scc._tp.apdu_tracer = _LoggingApduTracer()
try:
_init_proactive_session()
t_phase = time.time()
sys.stderr.write('INIT: sending TERMINAL PROFILE %s (CLA=%s)\n' % (opts.terminal_profile, scc.cat_cla))
sm, el = _send_terminal_profile(scc, opts.terminal_profile)
sys.stderr.write('INIT: TP done, menu=%s events=%s\n' % ('yes' if sm else 'no', 'yes' if el else 'no'))
@@ -109,6 +134,7 @@ def main():
if not st_sw.startswith('91'):
break
_handle_proactive_chain(scc, st_sw)
_tlog('terminal_profile_drain: %.0fms' % ((time.time() - t_phase) * 1000))
except Exception:
traceback.print_exc(file=sys.stderr)
if app is not None and opts.apdu_trace:
+161
View File
@@ -0,0 +1,161 @@
"""Fast card initialization for pysim-otaman-server.
pySim's ``init_card()`` performs several physical card resets: one per profile
candidate tried by ``CardProfile.pick()`` plus one at the end of
``RuntimeState.__init__``, and ``PysimApp.equip()`` resets yet again. On common
readers each disconnect/connect costs around a second, so the stock path spends
most of its time re-establishing a clean state (MF selected) that can also be
restored in software.
This module mirrors ``pySim.app.init_card()`` with those resets removed: all
profile probes run back-to-back on the same connection and the runtime state
uses a software reset. It is the default init/equip path; ``--full-pysim-init``
restores pysim's stock behavior, and the explicit ``equip``/``reset`` commands
keep a real reconnect/physical reset.
"""
import operator
from pySim.cards import CardBase, SimCardBase, UiccCardBase, card_detect
from pySim.commands import SimCardCommands
from pySim.exceptions import SwMatchError
from pySim.filesystem import CardApplication, CardModel
from pySim.profile import CardProfile
from pySim.runtime import RuntimeState
from pySim.ts_102_221 import CardProfileUICC
from pySim.utils import all_subclasses
import pySim.euicc
from .server import _tlog
class FastRuntimeState(RuntimeState):
"""RuntimeState whose reset() restores software state (selects MF) instead
of power-cycling the card. Use hard_reset() for an explicit reset."""
def reset(self, cmd_app=None):
return self.soft_reset(cmd_app)
def soft_reset(self, cmd_app=None):
for lchan_nr in list(self.lchan.keys()):
self.lchan[lchan_nr].scc.scp = None
if lchan_nr == 0:
continue
del self.lchan[lchan_nr]
self.adm_verified = False
try:
atr = self.card._scc.get_atr()
except Exception:
atr = None
if cmd_app:
cmd_app.lchan = self.lchan[0]
self.lchan[0].select('MF', cmd_app)
self.lchan[0].selected_adf = None
self.identity['ATR'] = atr
return atr
def hard_reset(self, cmd_app=None):
return super().reset(cmd_app)
def pick_profile_no_reset(scc):
"""Like CardProfile.pick(), but without a physical reset between
candidates. Each probe selects its own discriminating file, so a reset only
costs a reconnect without changing the outcome."""
original_reset = scc.reset_card
scc.reset_card = lambda: None
try:
profiles = sorted(all_subclasses(CardProfile), key=operator.attrgetter('ORDER'))
for p in profiles:
if p.match_with_card(scc):
return p()
return None
finally:
scc.reset_card = original_reset
def init_card_fast(sl, skip_card_init=False, wait=True):
"""Replacement for pySim.app.init_card() that avoids redundant resets.
``wait`` performs the single disconnect/connect of this init (explicit
equip passes True; startup already connects via wait_for_card)."""
scc = SimCardCommands(transport=sl)
if wait:
sl.wait_for_card(3)
if skip_card_init:
return None, CardBase(scc)
generic_card = False
card = card_detect(scc)
if card is None:
card = SimCardBase(scc)
generic_card = True
profile = pick_profile_no_reset(scc)
if profile is None:
return None, card
if generic_card and isinstance(profile, CardProfileUICC):
card._adm_chv_num = 0x0A
if isinstance(profile, CardProfileUICC):
for app_cls in all_subclasses(CardApplication):
if hasattr(app_cls, '_' + app_cls.__name__ + '__intermediate'):
continue
profile.add_application(app_cls())
if generic_card:
card = UiccCardBase(scc)
rs = FastRuntimeState(card, profile)
CardModel.apply_matching_models(scc, rs)
sl.set_sw_interpreter(rs)
isd_r = rs.mf.applications.get(pySim.euicc.AID_ISD_R.lower(), None)
if isd_r:
rs.lchan[0].select_file(isd_r)
try:
rs.identity['EID'] = pySim.euicc.CardApplicationISDR.get_eid(scc)
except SwMatchError:
pass
finally:
rs.soft_reset()
return rs, card
def do_equip_fast(app):
"""Explicit equip: one real reconnect (wait_for_card) then reset-free init."""
if app.rs and app.rs.profile:
for cmd_set in app.rs.profile.shell_cmdsets:
app.unregister_command_set(cmd_set)
rs, card = init_card_fast(app.sl, wait=True)
app.equip(card, rs)
def do_reset_fast(app):
"""Explicit reset: always a physical card reset."""
if app.rs is None:
app.card._scc.reset_card()
atr = app.card._scc.get_atr()
else:
atr = app.rs.hard_reset(app)
app.poutput('Card ATR: %s' % atr)
def install(app):
"""Route the pySim-shell equip/reset commands through the fast paths."""
def _do_equip(statement):
_tlog('do_equip_fast: start')
do_equip_fast(app)
_tlog('do_equip_fast: done')
def _do_reset(statement):
_tlog('do_reset_fast: start')
do_reset_fast(app)
_tlog('do_reset_fast: done')
app.do_equip = _do_equip
app.do_reset = _do_reset
+176 -93
View File
@@ -41,6 +41,21 @@ _STATIC_MIME = {
}
_T0 = time.time()
_TIMING = False
_APDU_N = 0
_RESET_N = 0
def _timing_on():
global _TIMING
_TIMING = True
def _tlog(msg):
if not _TIMING:
return
sys.stderr.write('TIMING [+%7.3fs] %s\n' % (time.time() - _T0, msg))
class StderrApduTracer(ApduTracer):
def __init__(self):
super().__init__()
@@ -49,9 +64,20 @@ class StderrApduTracer(ApduTracer):
def trace_command(self, cmd):
self._cmd_start = time.time()
def trace_reset(self):
global _RESET_N
_RESET_N += 1
if _TIMING:
sys.stderr.write('TIMING [+%7.3fs] RESET #%d\n' % (time.time() - _T0, _RESET_N))
def trace_response(self, cmd, sw, resp):
global _APDU_N
_APDU_N += 1
elapsed = int((time.time() - self._cmd_start) * 1000)
msg = 'APDU-TRACE(%dms): %s → SW: %s' % (elapsed, cmd, sw)
if _TIMING:
msg = 'APDU-TRACE(+%7.3fs #%d, %dms): %s → SW: %s' % (time.time() - _T0, _APDU_N, elapsed, cmd, sw)
else:
msg = 'APDU-TRACE(%dms): %s → SW: %s' % (elapsed, cmd, sw)
if resp:
msg += ' RESP: %s' % resp
os.write(2, (msg + '\n').encode())
@@ -618,19 +644,19 @@ _PLI_DATA = {q: '' for q in PLI_QUALIFIER_NAMES}
_POLL_ENABLED = False
_POLL_INTERVAL = 30
_POLL_TIMER = None
_POLL_LOCK = threading.Lock()
_CARD_LOCK = threading.RLock()
_CARD_CONNECTED = False
def _set_poll_interval(seconds):
global _POLL_INTERVAL
_POLL_INTERVAL = max(1, min(255, int(seconds)))
_POLL_INTERVAL = max(0, min(255, int(seconds)))
def _reset_poll_timer():
global _POLL_TIMER
if _POLL_TIMER is not None:
_POLL_TIMER.cancel()
_POLL_TIMER = None
if _POLL_ENABLED:
if _POLL_ENABLED and _POLL_INTERVAL > 0:
_POLL_TIMER = threading.Timer(_POLL_INTERVAL, _do_status_poll)
_POLL_TIMER.daemon = True
_POLL_TIMER.start()
@@ -640,7 +666,7 @@ def _do_status_poll():
_POLL_TIMER = None
if not _POLL_ENABLED:
return
with _POLL_LOCK:
with _CARD_LOCK:
try:
scc = getattr(_server_ref, 'scc', None) if _server_ref else None
if not scc:
@@ -656,9 +682,51 @@ def _do_status_poll():
def _poll_enable():
global _POLL_ENABLED
if _POLL_INTERVAL <= 0:
_POLL_ENABLED = False
return
_POLL_ENABLED = True
_reset_poll_timer()
_MENU_TIMEOUT = 60
_MENU_TIMER = None
def _set_menu_timeout(seconds):
global _MENU_TIMEOUT
_MENU_TIMEOUT = max(0, min(3600, int(seconds)))
def _cancel_menu_timeout():
global _MENU_TIMER
if _MENU_TIMER is not None:
_MENU_TIMER.cancel()
_MENU_TIMER = None
def _arm_menu_timeout():
"""Watchdog: a paused proactive command must always get a TERMINAL RESPONSE,
even if the user never answers. Fires 0x12 ('timeout') via the same path as
an explicit user response."""
global _MENU_TIMER
_cancel_menu_timeout()
if _MENU_TIMEOUT <= 0:
return
_MENU_TIMER = threading.Timer(_MENU_TIMEOUT, _menu_timeout_fire)
_MENU_TIMER.daemon = True
_MENU_TIMER.start()
def _menu_timeout_fire():
global _MENU_TIMER
_MENU_TIMER = None
with _CARD_LOCK:
server = _server_ref
if not server or not getattr(server, 'stk_pending', None) or not getattr(server, 'scc', None):
return
pd = server.stk_pending
sys.stderr.write('MENU-TIMEOUT: auto TR timeout (cmd=%02x type=%02x)\n' % (pd['cmd_num'], pd['cmd_type']))
try:
_menu_send_response(server, 'timeout', None)
except Exception as e:
sys.stderr.write('MENU-TIMEOUT error: %s\n' % e)
def _poll_disable():
global _POLL_ENABLED, _POLL_TIMER
_POLL_ENABLED = False
@@ -954,6 +1022,7 @@ def _record_tr(entry, tr_tlv, tr_sw=None):
def _handle_card_disconnect():
global _CARD_CONNECTED
_poll_disable()
_cancel_menu_timeout()
_CARD_CONNECTED = False
if _server_ref:
_server_ref.card = None
@@ -1303,6 +1372,79 @@ def _send_terminal_profile(scc, tp_hex):
return sim_menu, event_list
def _make_menu_fetch_handler(server, resp):
"""on_fetch callback for the menu chain: pauses on user-interactive commands
and stores the pending command so a TERMINAL RESPONSE can be sent later."""
def _on_menu_fetch(raw, cmd_num, cmd_type, dev_src, dev_dst):
if cmd_type == 0x21:
text = _parse_display_text(raw) if raw else None
if text:
server.stk_pending = {'type': 'display_text',
'cmd_num': cmd_num, 'cmd_type': cmd_type,
'dev_src': dev_src, 'dev_dst': dev_dst, 'text': text}
resp.update(type='display_text', text=text)
return 'pause'
elif cmd_type == 0x24:
items = _parse_select_item(raw) if raw else []
server.stk_pending = {'type': 'select_item',
'cmd_num': cmd_num, 'cmd_type': cmd_type,
'dev_src': dev_src, 'dev_dst': dev_dst, 'items': items}
resp.update(type='select_item', items=items)
return 'pause'
elif cmd_type == 0x25:
items = _parse_setup_menu_items(raw) if raw else []
server.stk_pending = {'type': 'select_item',
'cmd_num': cmd_num, 'cmd_type': cmd_type,
'dev_src': dev_src, 'dev_dst': dev_dst, 'items': items}
resp.update(type='select_item', items=items)
return 'pause'
return _on_menu_fetch
def _menu_send_response(server, result, item_id=None):
"""Send the pending command's TERMINAL RESPONSE and continue the chain.
Shared by /api/menu-respond and the user-input timeout watchdog. Returns
(payload, http_status)."""
if not server.stk_pending:
return {'error': 'no pending command'}, 400
scc = server.scc
RESULT_MAP = {'ok': 0x00, 'cancel': 0x10, 'back': 0x11, 'timeout': 0x12}
gr = RESULT_MAP.get(result, 0x00)
pd = server.stk_pending
cd = bytes([0x81, 0x03, pd['cmd_num'], pd['cmd_type'], 0x00])
di = bytes([0x82, 0x02, pd['dev_dst'], pd['dev_src']])
tr_data = cd + di
if isinstance(item_id, int) and result == 'ok' and pd['type'] == 'select_item':
tr_data += bytes([0x90, 0x01, item_id])
tr_data += bytes([0x83, 0x02, gr, 0x00])
tr_hex = '%s140000%02x%s' % (scc.cat_cla, len(tr_data), tr_data.hex())
tr_rv = scc._tp.send_apdu(tr_hex)
sys.stderr.write('TR(menu): cmd=%02x type=%02x result=%02x -> %s\n' % (pd['cmd_num'], pd['cmd_type'], gr, tr_rv[1]))
for entry in reversed(_PROACTIVE_LOG):
if (entry.get('cmd_num') == pd['cmd_num']
and entry.get('type_hex') == '%02x' % pd['cmd_type']
and 'tr_hex' not in entry):
_record_tr(entry, tr_data, tr_rv[1])
break
sw = tr_rv[1]
resp = {'sw': sw}
if result == 'cancel':
server.stk_pending = None
server.menu_active = False
else:
server.stk_pending = None
if sw.startswith('91'):
_handle_proactive_chain(scc, sw, _make_menu_fetch_handler(server, resp))
else:
server.menu_active = False
resp['type'] = 'done'
if server.stk_pending:
_arm_menu_timeout()
else:
_cancel_menu_timeout()
return resp, 200
class PysimHandler(BaseHTTPRequestHandler):
def _send_json(self, data, status=200):
self.send_response(status)
@@ -1366,6 +1508,12 @@ class PysimHandler(BaseHTTPRequestHandler):
self.wfile.write(data)
def do_GET(self):
# Serialize all card access: the background STATUS poll runs in its own
# thread and must never interleave with a FETCH/TERMINAL RESPONSE pair.
with _CARD_LOCK:
self._do_GET()
def _do_GET(self):
lang = _get_lang(self.headers)
if self.path == '/api/version':
self._log_req()
@@ -1477,8 +1625,14 @@ class PysimHandler(BaseHTTPRequestHandler):
self._serve_static()
def do_POST(self):
# Serialize all card access: the background STATUS poll runs in its own
# thread and must never interleave with a FETCH/TERMINAL RESPONSE pair.
with _CARD_LOCK:
_reset_poll_timer()
self._do_POST()
def _do_POST(self):
lang = _get_lang(self.headers)
_reset_poll_timer()
if self.path == '/api/command':
app = self.server.app
if not app:
@@ -1504,10 +1658,14 @@ class PysimHandler(BaseHTTPRequestHandler):
sys.stderr = old_stderr
elapsed = int((time.time() - t0) * 1000)
status = 'OK' if not output or 'not a recognized command' not in output else 'ERROR'
if str(cmd).strip().startswith('equip') and self.server.app and self.server.app.card and self.server.terminal_profile:
is_equip = str(cmd).strip().startswith('equip')
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:
global _CARD_CONNECTED
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
@@ -1518,6 +1676,7 @@ class PysimHandler(BaseHTTPRequestHandler):
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))
resp = {'output': output, 'stop': bool(stop)}
self._send_json(resp)
@@ -1579,6 +1738,7 @@ class PysimHandler(BaseHTTPRequestHandler):
sys.stderr.write('RESCUE: re-sending TERMINAL PROFILE\n')
self.server.stk_pending = None
self.server.menu_active = False
_cancel_menu_timeout()
self.server.event_list = None
_reset_proactive_log()
sm, el = _send_terminal_profile(scc, self.server.terminal_profile)
@@ -1857,31 +2017,9 @@ class PysimHandler(BaseHTTPRequestHandler):
env_hex = '%sc20000%02x%s' % (scc.cat_cla, len(menu_tlv), menu_tlv.hex())
data, sw = scc._tp.send_apdu(env_hex)
resp = {'type': 'done', 'sw': sw}
on_fetch = _make_menu_fetch_handler(self.server, resp)
if sw.startswith('91'):
def _on_menu_fetch(raw, cmd_num, cmd_type, dev_src, dev_dst):
if cmd_type == 0x21:
text = _parse_display_text(raw) if raw else None
if text:
self.server.stk_pending = {'type': 'display_text',
'cmd_num': cmd_num, 'cmd_type': cmd_type,
'dev_src': dev_src, 'dev_dst': dev_dst, 'text': text}
resp.update(type='display_text', text=text)
return 'pause'
elif cmd_type == 0x24:
items = _parse_select_item(raw) if raw else []
self.server.stk_pending = {'type': 'select_item',
'cmd_num': cmd_num, 'cmd_type': cmd_type,
'dev_src': dev_src, 'dev_dst': dev_dst, 'items': items}
resp.update(type='select_item', items=items)
return 'pause'
elif cmd_type == 0x25:
items = _parse_setup_menu_items(raw) if raw else []
self.server.stk_pending = {'type': 'select_item',
'cmd_num': cmd_num, 'cmd_type': cmd_type,
'dev_src': dev_src, 'dev_dst': dev_dst, 'items': items}
resp.update(type='select_item', items=items)
return 'pause'
_handle_proactive_chain(scc, sw, _on_menu_fetch)
_handle_proactive_chain(scc, sw, on_fetch)
else:
self.server.menu_active = False
self.server.stk_pending = None
@@ -1890,73 +2028,18 @@ class PysimHandler(BaseHTTPRequestHandler):
st_data, st_sw = _send_status(scc)
sys.stderr.write('STATUS -> %s\n' % st_sw)
if st_sw.startswith('91'):
_handle_proactive_chain(scc, st_sw, _on_menu_fetch)
_handle_proactive_chain(scc, st_sw, on_fetch)
if self.server.stk_pending:
_arm_menu_timeout()
else:
_cancel_menu_timeout()
self._send_json(resp)
self._log_resp(resp)
elif self.path == '/api/menu-respond':
scc = self.server.scc
if not self.server.stk_pending:
self._send_json({'error': 'no pending command'}, 400)
return
body = self._read_body()
self._log_req(body)
result = body.get('result', 'ok')
item_id = body.get('item_id')
RESULT_MAP = {'ok': 0x00, 'cancel': 0x10, 'back': 0x11, 'timeout': 0x12}
gr = RESULT_MAP.get(result, 0x00)
pd = self.server.stk_pending
# Build TERMINAL RESPONSE
cd = bytes([0x81, 0x03, pd['cmd_num'], pd['cmd_type'], 0x00])
di = bytes([0x82, 0x02, pd['dev_dst'], pd['dev_src']])
tr_data = cd + di
if isinstance(item_id, int) and result == 'ok' and pd['type'] == 'select_item':
tr_data += bytes([0x90, 0x01, item_id])
tr_data += bytes([0x83, 0x02, gr, 0x00])
tr_hex = '%s140000%02x%s' % (scc.cat_cla, len(tr_data), tr_data.hex())
tr_rv = scc._tp.send_apdu(tr_hex)
sys.stderr.write('TR(menu): cmd=%02x type=%02x result=%02x -> %s\n' % (pd['cmd_num'], pd['cmd_type'], gr, tr_rv[1]))
for entry in reversed(_PROACTIVE_LOG):
if (entry.get('cmd_num') == pd['cmd_num']
and entry.get('type_hex') == '%02x' % pd['cmd_type']
and 'tr_hex' not in entry):
_record_tr(entry, tr_data, tr_rv[1])
break
sw = tr_rv[1]
resp = {'sw': sw}
if result == 'cancel':
self.server.stk_pending = None
self.server.menu_active = False
else:
self.server.stk_pending = None
if sw.startswith('91'):
def _on_menu_fetch(raw, cmd_num, cmd_type, dev_src, dev_dst):
if cmd_type == 0x21:
text = _parse_display_text(raw) if raw else None
if text:
self.server.stk_pending = {'type': 'display_text',
'cmd_num': cmd_num, 'cmd_type': cmd_type,
'dev_src': dev_src, 'dev_dst': dev_dst, 'text': text}
resp.update(type='display_text', text=text)
return 'pause'
elif cmd_type == 0x24:
items = _parse_select_item(raw) if raw else []
self.server.stk_pending = {'type': 'select_item',
'cmd_num': cmd_num, 'cmd_type': cmd_type,
'dev_src': dev_src, 'dev_dst': dev_dst, 'items': items}
resp.update(type='select_item', items=items)
return 'pause'
elif cmd_type == 0x25:
items = _parse_setup_menu_items(raw) if raw else []
self.server.stk_pending = {'type': 'select_item',
'cmd_num': cmd_num, 'cmd_type': cmd_type,
'dev_src': dev_src, 'dev_dst': dev_dst, 'items': items}
resp.update(type='select_item', items=items)
return 'pause'
_handle_proactive_chain(scc, sw, _on_menu_fetch)
else:
self.server.menu_active = False
resp['type'] = 'done'
self._send_json(resp)
resp, code = _menu_send_response(self.server, body.get('result', 'ok'), body.get('item_id'))
self._send_json(resp, code)
self._log_resp(resp)
elif self.path == '/api/event-send':
scc = self.server.scc
+125
View File
@@ -0,0 +1,125 @@
#!/usr/bin/env python3
"""Tests for the reset-free fast initialization helpers."""
import sys
import types
import unittest
from pathlib import Path
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))
from pySim.exceptions import SwMatchError
from pySim.ts_102_221 import CardProfileUICC
from pysim_otaman_server.fastinit import (
FastRuntimeState,
do_reset_fast,
pick_profile_no_reset,
)
class FakeScc:
def __init__(self):
self.sel_ctrl = '0004'
self.cla_byte = '00'
self.resets = 0
self.selected = []
def reset_card(self):
self.resets += 1
def select_file(self, fid):
self.selected.append(fid)
return ('', '9000')
def select_adf(self, aid):
raise SwMatchError('6a82', '9000')
class TestPickProfileNoReset(unittest.TestCase):
def test_uicc_selected_without_any_reset(self):
scc = FakeScc()
profile = pick_profile_no_reset(scc)
self.assertIsInstance(profile, CardProfileUICC)
self.assertEqual(scc.resets, 0)
self.assertIn('3f00', scc.selected)
def test_reset_card_restored_after_pick(self):
scc = FakeScc()
pick_profile_no_reset(scc)
scc.reset_card()
self.assertEqual(scc.resets, 1)
class FakeLchan:
def __init__(self):
self.scc = types.SimpleNamespace(scp=object())
self.selected_adf = 'SOMETHING'
self.selected = []
def select(self, path, cmd_app=None):
self.selected.append(path)
class TestFastRuntimeStateSoftReset(unittest.TestCase):
def make_rs(self):
rs = FastRuntimeState.__new__(FastRuntimeState)
rs.lchan = {0: FakeLchan(), 1: FakeLchan()}
rs.adm_verified = True
rs.card = types.SimpleNamespace(_scc=types.SimpleNamespace(get_atr=lambda: 'AABB'))
rs.identity = {}
return rs
def test_soft_reset_selects_mf_without_physical_reset(self):
rs = self.make_rs()
atr = rs.soft_reset()
self.assertEqual(atr, 'AABB')
self.assertEqual(rs.identity['ATR'], 'AABB')
self.assertEqual(rs.lchan[0].selected, ['MF'])
self.assertIsNone(rs.lchan[0].selected_adf)
self.assertFalse(rs.adm_verified)
self.assertNotIn(1, rs.lchan)
def test_reset_is_soft(self):
rs = self.make_rs()
rs.card = types.SimpleNamespace(_scc=types.SimpleNamespace(get_atr=lambda: 'EEFF'))
self.assertEqual(rs.reset(), 'EEFF')
self.assertEqual(rs.lchan[0].selected, ['MF'])
class FakeCardScc:
def __init__(self):
self.resets = 0
def reset_card(self):
self.resets += 1
return 'ATR'
def get_atr(self):
return 'AABB'
class TestDoResetFast(unittest.TestCase):
def test_explicit_reset_is_physical(self):
scc = FakeCardScc()
out = []
app = types.SimpleNamespace(rs=None, card=types.SimpleNamespace(_scc=scc), poutput=out.append)
do_reset_fast(app)
self.assertEqual(scc.resets, 1)
self.assertEqual(out, ['Card ATR: AABB'])
def test_explicit_reset_uses_hard_reset_with_runtime_state(self):
calls = []
rs = types.SimpleNamespace(hard_reset=lambda cmd_app=None: calls.append(cmd_app) or 'CCDD')
out = []
app = types.SimpleNamespace(rs=rs, card=None, poutput=out.append)
do_reset_fast(app)
self.assertEqual(calls, [app])
self.assertEqual(out, ['Card ATR: CCDD'])
if __name__ == '__main__':
unittest.main()
+89
View File
@@ -0,0 +1,89 @@
#!/usr/bin/env python3
"""Tests for the paused-command (STK menu) timeout watchdog."""
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 TestMenuTimeout(unittest.TestCase):
def setUp(self):
self.saved = (S._MENU_TIMEOUT, S._MENU_TIMER)
def tearDown(self):
S._cancel_menu_timeout()
S._MENU_TIMEOUT, S._MENU_TIMER = self.saved
def test_clamped(self):
S._set_menu_timeout(30)
self.assertEqual(S._MENU_TIMEOUT, 30)
S._set_menu_timeout(0)
self.assertEqual(S._MENU_TIMEOUT, 0)
S._set_menu_timeout(-1)
self.assertEqual(S._MENU_TIMEOUT, 0)
S._set_menu_timeout(99999)
self.assertEqual(S._MENU_TIMEOUT, 3600)
def test_arm_starts_timer(self):
S._set_menu_timeout(30)
with mock.patch.object(S.threading, 'Timer') as timer:
S._arm_menu_timeout()
timer.assert_called_once_with(30, S._menu_timeout_fire)
def test_zero_disables_arming(self):
S._set_menu_timeout(0)
with mock.patch.object(S.threading, 'Timer') as timer:
S._arm_menu_timeout()
timer.assert_not_called()
def test_cancel(self):
timer = mock.Mock()
S._MENU_TIMER = timer
S._cancel_menu_timeout()
timer.cancel.assert_called_once()
self.assertIsNone(S._MENU_TIMER)
class TestMenuSendResponse(unittest.TestCase):
def test_timeout_tr_is_flat_with_general_result(self):
sent = []
def send_apdu(hexstr):
sent.append(hexstr)
return ('', '9000')
server = types.SimpleNamespace(
stk_pending={'type': 'display_text', 'cmd_num': 1, 'cmd_type': 0x21,
'dev_src': 0x81, 'dev_dst': 0x83},
menu_active=True,
scc=types.SimpleNamespace(cat_cla='80', _tp=types.SimpleNamespace(send_apdu=send_apdu)),
)
resp, code = S._menu_send_response(server, 'timeout', None)
self.assertEqual(code, 200)
self.assertEqual(resp['sw'], '9000')
self.assertEqual(resp['type'], 'done')
self.assertIsNone(server.stk_pending)
self.assertFalse(server.menu_active)
tr = sent[0]
self.assertTrue(tr.startswith('801400000d'), tr)
self.assertIn('8103012100', tr)
self.assertIn('82028381', tr)
self.assertIn('83021200', tr)
def test_no_pending_returns_400(self):
resp, code = S._menu_send_response(types.SimpleNamespace(stk_pending=None), 'ok')
self.assertEqual(code, 400)
self.assertIn('error', resp)
if __name__ == '__main__':
unittest.main()
+55
View File
@@ -0,0 +1,55 @@
#!/usr/bin/env python3
"""Tests for the background STATUS polling interval semantics."""
import sys
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 TestPollInterval(unittest.TestCase):
def setUp(self):
self.saved = (S._POLL_ENABLED, S._POLL_INTERVAL, S._POLL_TIMER)
def tearDown(self):
S._poll_disable()
S._POLL_ENABLED, S._POLL_INTERVAL, S._POLL_TIMER = self.saved
def test_zero_interval_disables_polling(self):
S._set_poll_interval(0)
self.assertEqual(S._POLL_INTERVAL, 0)
with mock.patch.object(S.threading, 'Timer') as timer:
S._poll_enable()
timer.assert_not_called()
self.assertFalse(S._POLL_ENABLED)
self.assertIsNone(S._POLL_TIMER)
def test_negative_interval_clamped_to_zero(self):
S._set_poll_interval(-5)
self.assertEqual(S._POLL_INTERVAL, 0)
def test_positive_interval_starts_timer(self):
S._set_poll_interval(30)
with mock.patch.object(S.threading, 'Timer') as timer:
S._poll_enable()
self.assertTrue(S._POLL_ENABLED)
timer.assert_called_once_with(30, S._do_status_poll)
def test_reset_timer_skipped_when_disabled(self):
S._set_poll_interval(0)
S._POLL_ENABLED = True
with mock.patch.object(S.threading, 'Timer') as timer:
S._reset_poll_timer()
timer.assert_not_called()
self.assertIsNone(S._POLL_TIMER)
if __name__ == '__main__':
unittest.main()