esim: local eUICC operations in the Phone simulator tab (v2.8.0)
New eSIM pill (Phone simulator) for SGP.22/32 cards, built on pySim's ES10 static API — no lpac, no new dependencies, no pysim patches, no SM-DP+ interaction: - Chip: EID, EUICCInfo1/2, configured addresses (ES10a/b). - Profiles: GetProfilesInfo with metadata (state, nickname, provider, ICCID, ISD-P AID, class, owner, icon). - Notifications: read-only ListNotification viewer. - Switch: Enable/DisableProfile with RefreshFlag=1; the card's REFRESH (fetched/answered/logged by the transport's proactive handler) or an ok result triggers _esim_reinit() — reset + equip + _apply_equipped_card — so the ICCID, network state and cached views are re-read. - Server: pysim_simple_server/esim.py, GET /api/esim/chip|profiles| notifications, POST /api/esim/profile (all under _CARD_LOCK; 400 not_an_euicc), euicc/eid in /api/status. - Tests: tests/test_esim.py (fake scc, monkeypatched store_data_tlv), frontend/tests/esim.test.js; help EN/RU, docs/api.md, AGENTS.
This commit is contained in:
@@ -0,0 +1,246 @@
|
||||
# coding=utf-8
|
||||
"""eSIM / LPA local operations (ES10a/b/c) on the equipped eUICC.
|
||||
|
||||
Scope: chip details (EID, EUICCInfo1/2, configured addresses), profile
|
||||
listing with metadata (GetProfilesInfo), profile switching (Enable/Disable)
|
||||
and a read-only notification viewer (ListNotification). No profile
|
||||
downloads, no notification processing/removal and no SM-DP+ interaction.
|
||||
|
||||
Every command goes through pySim's ES10 static API
|
||||
(``CardApplicationISDR.store_data_tlv`` / ``get_eid``) on the card's own
|
||||
logical channel; the ISD-R is selected first and the previous selection is
|
||||
restored afterwards. The caller holds ``_CARD_LOCK``.
|
||||
"""
|
||||
|
||||
import re
|
||||
|
||||
from osmocom.tlv import flatten_dict_lists
|
||||
from pySim.euicc import (
|
||||
AID_ISD_R, CardApplicationISDR, DisableProfileReq, DisableProfileResp,
|
||||
EnableProfileReq, EnableProfileResp, EuiccConfiguredAddresses, EuiccInfo1,
|
||||
EuiccInfo2, Iccid, IsdpAid, ListNotificationReq, ListNotificationResp,
|
||||
ProfileIdentifier, ProfileInfo, ProfileInfoListReq, ProfileInfoListResp,
|
||||
RefreshFlag, TagList,
|
||||
)
|
||||
from pySim.utils import b2h
|
||||
|
||||
|
||||
class EsimError(Exception):
|
||||
"""eSIM operation failed; ``code`` is a stable identifier for the API."""
|
||||
|
||||
def __init__(self, code, message=None):
|
||||
super().__init__(message or code)
|
||||
self.code = code
|
||||
|
||||
|
||||
# ES10c result codes -> short English fallback (the PWA localizes by code).
|
||||
RESULT_MESSAGES = {
|
||||
'ok': 'ok',
|
||||
'iccidOrAidNotFound': 'profile not found',
|
||||
'profileNotInDisabledState': 'profile is not disabled',
|
||||
'profileNotInEnabledState': 'profile is not enabled',
|
||||
'disallowedByPolicy': 'disallowed by policy',
|
||||
'wrongProfileReenabling': 'wrong profile re-enabling',
|
||||
'catBusy': 'card is busy with a CAT session',
|
||||
'undefinedError': 'undefined error',
|
||||
}
|
||||
|
||||
|
||||
def is_euicc(app):
|
||||
"""True when the equipped card runs an eUICC profile (SGP.02/22/32)."""
|
||||
rs = getattr(app, 'rs', None)
|
||||
return 'eUICC' in str(getattr(rs, 'profile', '') or '')
|
||||
|
||||
|
||||
def _select_isdr(app):
|
||||
"""Select ISD-R on logical channel 0; returns the lchan's scc."""
|
||||
rs = getattr(app, 'rs', None)
|
||||
apps = getattr(getattr(rs, 'mf', None), 'applications', None) or {}
|
||||
isd_r = apps.get(AID_ISD_R.lower())
|
||||
if isd_r is None:
|
||||
raise EsimError('not_an_euicc')
|
||||
lchan = rs.lchan[0]
|
||||
lchan.select_file(isd_r)
|
||||
return lchan.scc
|
||||
|
||||
|
||||
def _restore(app):
|
||||
"""Return to MF so later server operations start from a known selection."""
|
||||
try:
|
||||
app.rs.soft_reset()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def _run(app, fn):
|
||||
scc = _select_isdr(app)
|
||||
try:
|
||||
return fn(scc)
|
||||
finally:
|
||||
_restore(app)
|
||||
|
||||
|
||||
def _normalize(value):
|
||||
"""bytes -> hex string, recursively; scalars pass through."""
|
||||
if isinstance(value, (bytes, bytearray)):
|
||||
return b2h(bytes(value)).upper()
|
||||
if isinstance(value, dict):
|
||||
return {k: _normalize(v) for k, v in value.items()}
|
||||
if isinstance(value, (list, tuple)):
|
||||
return [_normalize(v) for v in value]
|
||||
return value
|
||||
|
||||
|
||||
def _flatten(resp):
|
||||
"""ES10 response object -> flat dict (single root key unwrapped)."""
|
||||
if resp is None:
|
||||
return {}
|
||||
flat = flatten_dict_lists(resp.to_dict())
|
||||
if len(flat) == 1:
|
||||
flat = next(iter(flat.values()))
|
||||
return _normalize(flat)
|
||||
|
||||
|
||||
def _as_list(value):
|
||||
"""A repeated TLV child decodes to a dict for one entry, a list for many."""
|
||||
if value is None:
|
||||
return []
|
||||
return value if isinstance(value, list) else [value]
|
||||
|
||||
|
||||
def _repeated(container, key):
|
||||
"""Collect `key` children from a container that may itself be a dict or a
|
||||
list of dicts (flatten_dict_lists collapses repeated parents either way)."""
|
||||
out = []
|
||||
for item in _as_list(container):
|
||||
if isinstance(item, dict):
|
||||
out.extend(_as_list(item.get(key)))
|
||||
return out
|
||||
|
||||
|
||||
def _transceive(app, cmd_do, resp_cls):
|
||||
return _run(app, lambda scc: CardApplicationISDR.store_data_tlv(scc, cmd_do, resp_cls))
|
||||
|
||||
|
||||
def _error_text(result):
|
||||
if isinstance(result, int):
|
||||
return RESULT_MESSAGES.get('undefinedError')
|
||||
return RESULT_MESSAGES.get(result, result)
|
||||
|
||||
|
||||
def chip_info(app):
|
||||
"""EID (ES10b GetEuiccData), EUICCInfo1/2 and the configured addresses."""
|
||||
out = {'eid': None, 'info1': None, 'info2': None, 'addresses': None,
|
||||
'errors': {}}
|
||||
scc = _select_isdr(app)
|
||||
try:
|
||||
parts = (
|
||||
('eid', lambda: CardApplicationISDR.get_eid(scc)),
|
||||
('info1', lambda: _flatten(CardApplicationISDR.store_data_tlv(
|
||||
scc, EuiccInfo1(), EuiccInfo1))),
|
||||
('info2', lambda: _flatten(CardApplicationISDR.store_data_tlv(
|
||||
scc, EuiccInfo2(), EuiccInfo2))),
|
||||
('addresses', lambda: _flatten(CardApplicationISDR.store_data_tlv(
|
||||
scc, EuiccConfiguredAddresses(), EuiccConfiguredAddresses))),
|
||||
)
|
||||
for key, fn in parts:
|
||||
try:
|
||||
out[key] = fn()
|
||||
except Exception as e: # one unsupported part must not fail the rest
|
||||
out['errors'][key] = str(e)
|
||||
finally:
|
||||
_restore(app)
|
||||
return out
|
||||
|
||||
|
||||
def _profile_tag_list():
|
||||
"""TagList requesting every ProfileInfo tag (pySim-shell's --all set)."""
|
||||
tags = [nest.tag for nest in ProfileInfo.nested_collection_cls().nested]
|
||||
u8 = []
|
||||
for tag in tags:
|
||||
if tag <= 255:
|
||||
u8.append(tag)
|
||||
elif tag <= 65535:
|
||||
u8.append(tag >> 8)
|
||||
u8.append(tag & 0xff)
|
||||
return TagList(decoded=u8)
|
||||
|
||||
|
||||
def profiles(app):
|
||||
"""ES10c GetProfilesInfo: the profile list with its metadata."""
|
||||
resp = _transceive(app, ProfileInfoListReq(children=[_profile_tag_list()]),
|
||||
ProfileInfoListResp)
|
||||
flat = _flatten(resp)
|
||||
err = flat.get('profile_info_list_error')
|
||||
if err is not None:
|
||||
return {'profiles': [], 'error': _error_text(err)}
|
||||
seq = flat.get('profile_info_seq')
|
||||
out = []
|
||||
for p in _repeated(seq, 'profile_info'):
|
||||
owner = p.get('profile_owner')
|
||||
out.append({
|
||||
'iccid': p.get('iccid'),
|
||||
'isdp_aid': p.get('isdp_aid'),
|
||||
'state': p.get('profile_state'),
|
||||
'nickname': p.get('profile_nickname'),
|
||||
'provider': p.get('service_provider_name'),
|
||||
'name': p.get('profile_name'),
|
||||
'class': p.get('profile_class'),
|
||||
'icon_type': p.get('icon_type'),
|
||||
'owner': owner.get('profile_owner_plmn') if isinstance(owner, dict) else None,
|
||||
})
|
||||
return {'profiles': out, 'error': None}
|
||||
|
||||
|
||||
def notifications(app):
|
||||
"""ES10b ListNotification: read-only list of pending notifications."""
|
||||
resp = _transceive(app, ListNotificationReq(), ListNotificationResp)
|
||||
flat = _flatten(resp)
|
||||
err = flat.get('list_notifications_result_error')
|
||||
if err is not None:
|
||||
return {'notifications': [], 'error': _error_text(err)}
|
||||
lst = flat.get('notification_metadata_list')
|
||||
out = []
|
||||
for n in _repeated(lst, 'notification_metadata'):
|
||||
op = n.get('profile_mgmt_operation')
|
||||
operations = sorted(k for k, v in op.items() if v) if isinstance(op, dict) else []
|
||||
out.append({
|
||||
'seq_number': n.get('seq_number'),
|
||||
'operations': operations,
|
||||
'address': n.get('notification_address'),
|
||||
'iccid': n.get('iccid'),
|
||||
})
|
||||
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)."""
|
||||
if action not in ('enable', 'disable'):
|
||||
raise EsimError('bad_action')
|
||||
ident = []
|
||||
if isdp_aid:
|
||||
aid = re.sub(r'[^0-9a-fA-F]', '', str(isdp_aid))
|
||||
if not aid:
|
||||
raise EsimError('bad_aid')
|
||||
ident.append(IsdpAid(decoded=bytes.fromhex(aid)))
|
||||
elif iccid:
|
||||
digits = re.sub(r'[^0-9a-fA-F]', '', str(iccid))
|
||||
if digits[-1:] in ('f', 'F'):
|
||||
digits = digits[:-1]
|
||||
if not digits.isdigit():
|
||||
raise EsimError('bad_iccid')
|
||||
ident.append(Iccid(decoded=digits))
|
||||
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',
|
||||
'message': _error_text(result)}
|
||||
Reference in New Issue
Block a user