esim: drive the profile switch ourselves; guard the FCP metadata

Disabling a profile failed with 6985 and left the card stuck until an
equip.  pySim's send_apdu_checksw auto-handler keeps flushing proactive
commands after the REFRESH TERMINAL RESPONSE; the card is then mid-switch
and answers 6985 to the next FETCH, which propagated as a 500 and skipped
the re-initialization.  Per SGP.22 v2.6 5.7.16/5.7.17 a 91XX answer is
the ISD-R's 'result OK before REFRESH' (step 6) and the switch completes
on the TERMINAL RESPONSE or the following RESET (step 8) - lpac treats
91XX the same way and never retries.

- esim.py: build_switch_apdu/parse_switch_response/switch_profile split
  out of set_profile_state; the switch is one raw STORE DATA via
  scc._tp.send_apdu, a 91XX runs our own FETCH/TR chain (status_poll=False)
  and is reported as ok, the STORE DATA is never retried and a chain
  failure still counts the accepted switch.
- server.py: /api/esim/profile answers the REFRESH with our chain, then
  re-initializes the card and re-reads the profile list, returning
  verified/state_after; _handle_proactive_chain grew status_poll.
- /api/status and /api/select: FCP metadata via _fcp_value - an ADF or a
  failed select (card with the active profile disabled) has no
  file_descriptor and used to crash the request handler; _get_file_type
  no longer raises either.
- esim._restore logs a failed selection restore instead of swallowing it.
- PWA: esimSwitchStatus shows the verified state / not-confirmed warning.
- tests: the switch flow (9000 / 91XX / error SW / chain failure), the
  FCP guards and the status helper; docs and sw simple-v230.
This commit is contained in:
2026-09-22 00:14:02 +03:00
parent 91c642a646
commit 86808ce55d
11 changed files with 338 additions and 87 deletions
+57 -15
View File
@@ -13,6 +13,7 @@ restored afterwards. The caller holds ``_CARD_LOCK``.
"""
import re
import sys
from osmocom.tlv import BER_TLV_IE, bertlv_parse_one_rawtag, flatten_dict_lists
from pySim.euicc import (
@@ -65,11 +66,15 @@ def _select_isdr(app):
def _restore(app):
"""Return to MF so later server operations start from a known selection."""
"""Return to MF so later server operations start from a known selection.
Best effort: a card whose active profile is disabled has no filesystem to
select, so the restore can legitimately fail - the selection metadata is
then guarded by the status endpoint instead of crashing it."""
try:
app.rs.soft_reset()
except Exception:
pass
except Exception as e:
sys.stderr.write('ESIM: selection restore failed: %s\n' % e)
def _run(app, fn):
@@ -449,8 +454,8 @@ def notifications(app):
return {'notifications': out, 'error': None}
def set_profile_state(app, action, iccid=None, isdp_aid=None, refresh=True):
"""ES10c Enable/DisableProfile for one profile (by ICCID or ISD-P AID)."""
def build_switch_apdu(action, iccid=None, isdp_aid=None, refresh=True):
"""STORE DATA APDU hex for ES10c Enable/DisableProfile (no sending)."""
if action not in ('enable', 'disable'):
raise EsimError('bad_action')
ident = []
@@ -469,14 +474,51 @@ def set_profile_state(app, action, iccid=None, isdp_aid=None, refresh=True):
else:
raise EsimError('missing_profile')
flag = RefreshFlag(decoded=1 if refresh else 0)
if action == 'enable':
cmd = EnableProfileReq(children=[ProfileIdentifier(children=ident), flag])
resp_cls, key = EnableProfileResp, 'enable_result'
else:
cmd = DisableProfileReq(children=[ProfileIdentifier(children=ident), flag])
resp_cls, key = DisableProfileResp, 'disable_result'
flat = _flatten(_transceive(app, cmd, resp_cls))
result = flat.get(key)
ok = result == 'ok'
return {'ok': ok, 'result': result if isinstance(result, str) else 'undefinedError',
req_cls = EnableProfileReq if action == 'enable' else DisableProfileReq
tx_do = req_cls(children=[ProfileIdentifier(children=ident), flag]).to_tlv()
return '80E29100%02x%s00' % (len(tx_do), tx_do.hex().upper())
def parse_switch_response(action, data_hex):
"""STORE DATA response TLV -> {'ok', 'result', 'message'}."""
resp_cls = EnableProfileResp if action == 'enable' else DisableProfileResp
key = 'enable_result' if action == 'enable' else 'disable_result'
result = None
if data_hex:
resp = resp_cls()
resp.from_tlv(bytes.fromhex(data_hex))
result = _flatten(resp).get(key)
return {'ok': result == 'ok',
'result': result if isinstance(result, str) else 'undefinedError',
'message': _error_text(result)}
def switch_profile(send_apdu, run_chain, action, iccid=None, isdp_aid=None,
refresh=True):
"""Run an ES10c Enable/DisableProfile switch.
``send_apdu(apdu_hex) -> (data_hex, sw)`` performs one raw STORE DATA and
``run_chain(sw91)`` answers the proactive command(s) the card sends
alongside the switch (True when a REFRESH was answered).
With the refresh flag set the ISD-R returns OK *before* the REFRESH
(SGP.22 v2.6 §5.7.16/§5.7.17 step 6) and the switch completes upon the
TERMINAL RESPONSE or the following RESET (step 8). A 91XX status is that
OK: the STORE DATA is never retried (the mid-switch card answers 6985 to
the retry) and the caller re-initializes the card afterwards."""
apdu = build_switch_apdu(action, iccid, isdp_aid, refresh)
data, sw = send_apdu(apdu)
if sw == '9000':
out = parse_switch_response(action, data)
out['refresh_seen'] = False
return out
if sw and sw.startswith('91'):
refresh_seen = False
try:
refresh_seen = bool(run_chain(sw))
except Exception as e:
sys.stderr.write('ESIM: REFRESH chain failed: %s\n' % e)
return {'ok': True, 'result': 'ok', 'message': 'ok',
'refresh_seen': refresh_seen}
return {'ok': False, 'result': 'undefinedError', 'sw': sw,
'message': 'SW %s' % sw}
+100 -29
View File
@@ -255,20 +255,34 @@ def _parse_help_text(text):
def _get_file_type(lchan, cur_file):
if cur_file and cur_file.name:
if cur_file.name.startswith('EF.'):
if lchan and lchan.selected_file_fcp:
ft = lchan.selected_file_type()
if ft != 'df':
return lchan.selected_file_structure()
return 'transparent'
if cur_file.name.startswith('DF.') or cur_file.name.startswith('ADF.') or cur_file.name == 'MF':
return 'df'
if lchan and lchan.selected_file_fcp:
return lchan.selected_file_structure()
try:
if cur_file and cur_file.name:
if cur_file.name.startswith('EF.'):
if lchan and lchan.selected_file_fcp:
ft = lchan.selected_file_type()
if ft != 'df':
return lchan.selected_file_structure()
return 'transparent'
if cur_file.name.startswith('DF.') or cur_file.name.startswith('ADF.') or cur_file.name == 'MF':
return 'df'
if lchan and lchan.selected_file_fcp:
return lchan.selected_file_structure()
except Exception:
# No usable FCP (an ADF, or a failed select while the card has no
# active profile) - report no file type instead of crashing.
return None
return None
def _fcp_value(lchan, name):
"""One FCP-derived selection value, or None when the card delivered no
usable FCP (an ADF, or a failed select) - never raises."""
try:
return getattr(lchan, name)()
except Exception:
return None
def _fid4(sel):
"""True if sel is a 4-digit hex FID."""
return bool(re.fullmatch(r'[0-9a-fA-F]{4}', str(sel or '')))
@@ -2589,6 +2603,52 @@ def _esim_reinit(server):
server.equipping = False
def _esim_refresh_chain(server, sw91):
"""Answer the proactive command(s) that accompany a profile switch.
With the refresh flag set the card sends REFRESH (SGP.22 §5.7.16 step 7);
it is answered and the chain stops - the card reset that follows performs
the switch. No STATUS poll is sent: the card is mid-switch and refuses
further commands (6985) until the reset."""
seen = {'refresh': False}
def on_fetch(raw, cmd_num, cmd_type, dev_src, dev_dst):
if cmd_type == 0x01:
seen['refresh'] = True
return 'exit'
return None
_handle_proactive_chain(server.scc, sw91, on_fetch=on_fetch,
status_poll=False)
return seen['refresh']
def _norm_iccid(value):
"""ICCID comparison form: hex digits without the trailing F pad."""
s = re.sub(r'[^0-9a-fA-F]', '', str(value or '')).upper()
return s[:-1] if s.endswith('F') else s
def _esim_verify_switch(app, action, iccid=None, isdp_aid=None):
"""Re-read the profile list and check the requested state took effect."""
expected = 'enabled' if action == 'enable' else 'disabled'
try:
profs = esim.profiles(app).get('profiles') or []
except Exception as e:
sys.stderr.write('ESIM: state verification failed: %s\n' % e)
return {'verified': None, 'state_after': None}
want_iccid = _norm_iccid(iccid) if iccid else None
want_aid = re.sub(r'[^0-9a-fA-F]', '', str(isdp_aid or '')).upper() or None
state_after = None
for p in profs:
if (want_iccid and _norm_iccid(p.get('iccid')) == want_iccid) or \
(want_aid and re.sub(r'[^0-9a-fA-F]', '',
str(p.get('isdp_aid') or '')).upper() == want_aid):
state_after = p.get('state')
break
return {'verified': state_after == expected, 'state_after': state_after}
_AUTO_EQUIP = True
_AUTO_EQUIP_BUSY = False
@@ -3112,18 +3172,20 @@ def _parse_setup_menu_items(raw):
return items
def _handle_proactive_chain(scc, sw91, on_fetch=None):
def _handle_proactive_chain(scc, sw91, on_fetch=None, status_poll=True):
"""Run a FETCH/TERMINAL RESPONSE chain; marks the card as busy so that
terminal-initiated ENVELOPEs (Data available, Channel status, timers) wait."""
terminal-initiated ENVELOPEs (Data available, Channel status, timers) wait.
``status_poll=False`` skips the trailing STATUS (used by the profile
switch, where the card is mid-switch and must not be queried further)."""
global _PROACTIVE_BUSY
_PROACTIVE_BUSY = True
try:
return _run_proactive_chain(scc, sw91, on_fetch)
return _run_proactive_chain(scc, sw91, on_fetch, status_poll)
finally:
_PROACTIVE_BUSY = False
def _run_proactive_chain(scc, sw91, on_fetch=None):
def _run_proactive_chain(scc, sw91, on_fetch=None, status_poll=True):
sys.stderr.write('91XX chain: sw=%s\n' % sw91)
sw = sw91
paused = False
@@ -3155,7 +3217,7 @@ def _run_proactive_chain(scc, sw91, on_fetch=None):
sys.stderr.write('TR: cmd=%02x type=%02x -> %s %s\n' % (cmd_num, cmd_type, tr_rv[1], ('(%d bytes)' % len(tr_tlv))))
_record_tr(entry, tr_tlv, tr_rv[1])
sw = tr_rv[1]
if sw == '9000':
if sw == '9000' and status_poll:
sys.stderr.write('STATUS poll (chain ended)\n')
st_data, st_sw = _send_status(scc)
sys.stderr.write('STATUS -> %s\n' % st_sw)
@@ -3508,9 +3570,9 @@ class PysimHandler(BaseHTTPRequestHandler):
'type': cur_file.__class__.__name__ if cur_file else None,
'path': str(lchan.get_cwd()) if lchan else None,
'file_type': _get_file_type(lchan, cur_file),
'file_size': lchan.selected_file_size() if lchan else None,
'record_len': lchan.selected_file_record_len() if lchan else None,
'num_of_rec': lchan.selected_file_num_of_rec() if lchan else None,
'file_size': _fcp_value(lchan, 'selected_file_size'),
'record_len': _fcp_value(lchan, 'selected_file_record_len'),
'num_of_rec': _fcp_value(lchan, 'selected_file_num_of_rec'),
} if cur_file else None,
'channels': [str(i) for i, ch in rs.lchan.items() if ch] if rs else [],
}
@@ -3798,24 +3860,33 @@ class PysimHandler(BaseHTTPRequestHandler):
with _CARD_LOCK:
_finish_pending_menu(self.server, self.server.scc)
cursor = _PROACTIVE_ENTRY_ID
resp = esim.set_profile_state(
app, action,
iccid=body.get('iccid'), isdp_aid=body.get('isdp_aid'),
resp = esim.switch_profile(
self.server.scc._tp.send_apdu,
lambda sw: _esim_refresh_chain(self.server, sw),
action, iccid=body.get('iccid'),
isdp_aid=body.get('isdp_aid'),
refresh=body.get('refresh', True))
# A REFRESH during the command means the card wants the
# terminal to re-initialize; a lost STORE DATA response
# (T=0 after REFRESH) is covered by the same re-init.
resp['refresh_seen'] = any(
# terminal to re-initialize; the switch itself completes on
# the TERMINAL RESPONSE or the reset (SGP.22 5.7.16 step 8).
resp['refresh_seen'] = resp.get('refresh_seen') or any(
e.get('type_hex') == '01' and e.get('id', 0) > cursor
for e in _PROACTIVE_LOG)
resp['reinitialized'] = False
resp['verified'] = None
resp['state_after'] = None
if resp['ok'] or resp['refresh_seen']:
resp['reinitialized'] = _esim_reinit(self.server)
if resp['reinitialized']:
resp.update(_esim_verify_switch(
app, action, iccid=body.get('iccid'),
isdp_aid=body.get('isdp_aid')))
resp['iccid'] = getattr(self.server, 'iccid', None)
resp['card_session'] = getattr(self.server, 'card_session', None)
self._send_json(resp)
self._log_resp({k: resp.get(k) for k in
('ok', 'result', 'refresh_seen', 'reinitialized')})
('ok', 'result', 'refresh_seen', 'reinitialized',
'verified', 'state_after')})
except esim.EsimError as e:
resp = {'ok': False, 'error': e.code, 'message': str(e)}
self._send_json(resp, 400)
@@ -4005,9 +4076,9 @@ class PysimHandler(BaseHTTPRequestHandler):
'name': cur.name if cur else None,
'fid': cur.fid.upper() if cur and cur.fid else None,
'file_type': _get_file_type(lchan, cur),
'file_size': lchan.selected_file_size() if lchan else None,
'record_len': lchan.selected_file_record_len() if lchan else None,
'num_of_rec': lchan.selected_file_num_of_rec() if lchan else None,
'file_size': _fcp_value(lchan, 'selected_file_size'),
'record_len': _fcp_value(lchan, 'selected_file_record_len'),
'num_of_rec': _fcp_value(lchan, 'selected_file_num_of_rec'),
'fci_hex': (lchan.selected_file_fcp_hex or '').upper() if lchan and lchan.selected_file_fcp_hex else None,
'apdu_times': apdu_times,
'exists': True,