Compare commits
5 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 25787017e5 | |||
| b50be83da4 | |||
| 6c57ff4fd5 | |||
| 895d0b7b36 | |||
| ef00b2c3f4 |
@@ -4,3 +4,4 @@ __pycache__/
|
||||
*.egg-info/
|
||||
dist/
|
||||
build/
|
||||
AGENTS.md
|
||||
|
||||
@@ -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
@@ -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
@@ -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`
|
||||
|
||||
|
||||
@@ -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, вверху этого представления появляется блок «Меню STK» с изумрудной кнопкой <strong>STK: <название></strong>, открывающей оверлей меню (браузер STK-меню карты). Если карта не задала меню, вместо кнопки показывается «Меню не задано картой». Состояние меню обновляется при каждом открытии представления.</p>
|
||||
<p class="text-sm mb-3">Если карта выдала команду SET UP MENU, вверху этого представления появляется блок «Меню STK» с изумрудной кнопкой <strong>STK: <название></strong>, открывающей оверлей меню (браузер STK-меню карты). Если карта не задала меню, вместо кнопки показывается «Меню не задано картой». Состояние меню обновляется при каждом открытии представления. Интерактивные проактивные команды всегда получают 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–255 с, по умолчанию 30 с) сервер отправляет 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–255 с, по умолчанию 30 с, <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
@@ -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 “STK menu” block appears at the top of this view with an emerald <strong>STK: <title></strong> button that opens the menu overlay (same as the card’s STK menu browser). If the card has not set up a menu, the block shows “No menu set by the card” 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 “STK menu” block appears at the top of this view with an emerald <strong>STK: <title></strong> button that opens the menu overlay (same as the card’s STK menu browser). If the card has not set up a menu, the block shows “No menu set by the card” 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–255 s, default 30 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–255 s, default 30 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
@@ -1,4 +1,4 @@
|
||||
const CACHE = 'otaman-v88';
|
||||
const CACHE = 'otaman-v90';
|
||||
const URLS = [
|
||||
'index.html',
|
||||
'help.html',
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
@@ -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
|
||||
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user