simple/pysim_simple_server/esim.py

# 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
import sys

from osmocom.tlv import BER_TLV_IE, bertlv_parse_one_rawtag, 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.

    The shell app is passed as cmd_app so pySim keeps its command-set
    bookkeeping in sync with the selection: ``equip()`` only unregisters the
    sets of the file selected at that moment, so a selection made without
    cmd_app leaves the old file's sets registered and the next equip dies
    re-registering them (cmd2: 'Attribute already exists')."""
    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, app)
    return lchan.scc


def _restore(app):
    """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.  The shell app
    is passed as cmd_app so the command-set bookkeeping follows the selection."""
    try:
        app.rs.soft_reset(app)
    except Exception as e:
        sys.stderr.write('ESIM: selection restore failed: %s\n' % e)


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)


# ---- SGP.22 EUICCInfo decoding -------------------------------------------
#
# pySim's EuiccInfo1/2 classes are incomplete (the capability fields are raw
# GreedyBytes, several SGP.22 TLVs are missing from the class), so the chip
# endpoint requests the raw TLVs and decodes them here, per SGP.22 v2.6
# §5.7.8/§5.7.13 and cross-checked against lpac's euicc/es10c_ex.c.  Unknown
# TLVs are preserved in `raw_tlvs`, never dropped.

UICC_CAPABILITY_BITS = [
    'contactlessSupport', 'usimSupport', 'isimSupport', 'csimSupport',
    'akaMilenage', 'akaCave', 'akaTuak128', 'akaTuak256', 'rfu1', 'rfu2',
    'gbaAuthenUsim', 'gbaAuthenISim', 'mbmsAuthenUsim', 'eapClient',
    'javacard', 'multos', 'multipleUsimSupport', 'multipleIsimSupport',
    'multipleCsimSupport', 'berTlvFileSupport', 'dfLinkSupport', 'catTp',
    'getIdentity', 'profile-a-x25519', 'profile-b-p256', 'suciCalculatorApi',
]
RSP_CAPABILITY_BITS = [
    'additionalProfile', 'crlSupport', 'rpmSupport', 'testProfileSupport',
    'deviceInfoExtensibilitySupport', 'serviceSpecificDataSupport',
]
PPR_ID_BITS = ['pprUpdateControl', 'ppr1', 'ppr2', 'ppr3']
PPR_FLAG_BITS = ['consentRequired']
TRE_PROPERTY_BITS = ['isDiscrete', 'isIntegrated', 'usesRemoteMemory']
EUICC_CATEGORIES = {0: 'other', 1: 'basicEuicc', 2: 'mediumEuicc',
                    3: 'contactlessEuicc'}


class _GetRatRequest(BER_TLV_IE, tag=0xbf43):
    """ES10b GetRat request (no input data, SGP.22 §5.7.13)."""


def _tlvs(data):
    """Walk a BER-TLV buffer -> [(tag, value)]; multi-byte tags kept raw."""
    out = []
    rest = bytes(data or b'')
    while rest:
        tag, _length, value, rest = bertlv_parse_one_rawtag(rest)
        out.append((tag, value))
    return out


def _tlv_value(data, tag):
    """Value bytes of the first `tag` TLV in `data` (b'' when absent)."""
    for t, value in _tlvs(data):
        if t == tag:
            return value
    return b''


def _decode_version(data):
    """VersionType: major/minor/revision bytes -> 'M.m.r'."""
    if len(data) != 3:
        return None
    return '%d.%d.%d' % (data[0], data[1], data[2])


def _decode_bit_string(data, names):
    """ASN.1 BIT STRING content -> list of set bit names.

    The first octet is the number of unused bits in the final octet; bits are
    numbered MSB-first within each octet (SGP.22 v2.6 §5.7.8)."""
    if not data:
        return []
    unused = data[0]
    body = data[1:]
    out = []
    for j, byte in enumerate(body):
        b = byte
        if j == len(body) - 1 and unused:
            b &= ~(0xFF >> (8 - unused)) & 0xFF
        for i in range(8):
            idx = j * 8 + i
            if idx >= len(names):
                break
            if b & 0x80:
                out.append(names[idx])
            b = (b << 1) & 0xFF
    return out


def _decode_ski_list(data):
    """SEQUENCE OF SubjectKeyIdentifier -> hex strings."""
    return [value.hex().upper() for _tag, value in _tlvs(data)]


def _decode_ext_card_resource(data):
    """ETSI TS 102 226 Extended Card Resource Information (inner 81/82/83)."""
    out = {}
    raw = {}
    for tag, value in _tlvs(data):
        if tag == 0x81:
            out['installed_application'] = int.from_bytes(value, 'big')
        elif tag == 0x82:
            out['free_non_volatile_memory'] = int.from_bytes(value, 'big')
        elif tag == 0x83:
            out['free_volatile_memory'] = int.from_bytes(value, 'big')
        else:
            raw['%02X' % tag] = value.hex().upper()
    if raw:
        out['raw_tlvs'] = raw
    return out


def _decode_certification_data_object(data):
    """CertificationDataObject (SGP.22 v2.6 §5.7.8): platform label + DLOA URL."""
    out = {}
    raw = {}
    for tag, value in _tlvs(data):
        if tag == 0x80:
            out['platform_label'] = value.decode('utf-8', 'replace')
        elif tag == 0x81:
            out['discovery_base_url'] = value.decode('utf-8', 'replace')
        else:
            raw['%02X' % tag] = value.hex().upper()
    if raw:
        out['raw_tlvs'] = raw
    return out


def _decode_info1(raw_hex):
    """EUICCInfo1 (BF20): SVN and the CI PKI lists."""
    out = {'svn': None, 'euicc_ci_pki_list_for_verification': [],
           'euicc_ci_pki_list_for_signing': []}
    raw = {}
    for tag, value in _tlvs(_tlv_value(bytes.fromhex(raw_hex or ''), 0xBF20)):
        if tag == 0x82:
            out['svn'] = _decode_version(value)
        elif tag == 0xA9:
            out['euicc_ci_pki_list_for_verification'] = _decode_ski_list(value)
        elif tag == 0xAA:
            out['euicc_ci_pki_list_for_signing'] = _decode_ski_list(value)
        else:
            raw['%02X' % tag] = value.hex().upper()
    if raw:
        out['raw_tlvs'] = raw
    return out


def _decode_info2(raw_hex):
    """EUICCInfo2 (BF22) with every SGP.22 v2.6 field decoded."""
    out = {}
    raw = {}
    for tag, value in _tlvs(_tlv_value(bytes.fromhex(raw_hex or ''), 0xBF22)):
        if tag == 0x81:
            out['profile_version'] = _decode_version(value)
        elif tag == 0x82:
            out['svn'] = _decode_version(value)
        elif tag == 0x83:
            out['euicc_firmware_ver'] = _decode_version(value)
        elif tag == 0x84:
            out['ext_card_resource'] = _decode_ext_card_resource(value)
        elif tag == 0x85:
            out['uicc_capability'] = _decode_bit_string(value, UICC_CAPABILITY_BITS)
        elif tag == 0x86:
            out['ts102241_version'] = _decode_version(value)
        elif tag == 0x87:
            out['globalplatform_version'] = _decode_version(value)
        elif tag == 0x88:
            out['rsp_capability'] = _decode_bit_string(value, RSP_CAPABILITY_BITS)
        elif tag == 0xA9:
            out['euicc_ci_pki_list_for_verification'] = _decode_ski_list(value)
        elif tag == 0xAA:
            out['euicc_ci_pki_list_for_signing'] = _decode_ski_list(value)
        elif tag in (0x8B, 0xAB):   # implicit and explicit category encodings
            out['euicc_category'] = EUICC_CATEGORIES.get(
                int.from_bytes(value, 'big') if value else 0, 'other')
        elif tag == 0x99:
            out['forbidden_profile_policy_rules'] = _decode_bit_string(value, PPR_ID_BITS)
        elif tag == 0x04:           # ppVersion has no context tag
            out['pp_version'] = _decode_version(value)
        elif tag == 0x0C:           # sasAcreditationNumber is a bare UTF8String
            out['ss_acreditation_number'] = value.decode('utf-8', 'replace')
        elif tag == 0xAC:
            out['certification_data_object'] = _decode_certification_data_object(value)
        elif tag == 0xAD:
            out['tre_properties'] = _decode_bit_string(value, TRE_PROPERTY_BITS)
        elif tag == 0xAE:
            out['tre_product_reference'] = value.decode('utf-8', 'replace')
        elif tag == 0xAF:
            out['additional_euicc_profile_package_versions'] = [
                _decode_version(v) for _t, v in _tlvs(value)]
        else:
            raw['%02X' % tag] = value.hex().upper()
    if raw:
        out['raw_tlvs'] = raw
    return out


def _decode_addresses(raw_hex):
    """ES10a GetEuiccConfiguredAddresses (BF3C)."""
    out = {'default_dp_address': None, 'root_ds_address': None}
    raw = {}
    for tag, value in _tlvs(_tlv_value(bytes.fromhex(raw_hex or ''), 0xBF3C)):
        if tag == 0x80:
            out['default_dp_address'] = value.decode('utf-8', 'replace')
        elif tag == 0x81:
            out['root_ds_address'] = value.decode('utf-8', 'replace')
        else:
            raw['%02X' % tag] = value.hex().upper()
    if raw:
        out['raw_tlvs'] = raw
    return out


def _decode_rat(raw_hex):
    """ES10b GetRat (BF43): the Rules Authorisation Table (SGP.22 §5.7.13)."""
    out = []
    table = _tlv_value(_tlv_value(bytes.fromhex(raw_hex or ''), 0xBF43), 0xA0)
    for _tag, rule in _tlvs(table):
        entry = {'ppr_ids': [], 'allowed_operators': [], 'ppr_flags': []}
        for tag, value in _tlvs(rule):
            if tag == 0x80:
                entry['ppr_ids'] = _decode_bit_string(value, PPR_ID_BITS)
            elif tag == 0xA1:
                operators = []
                for _t, op in _tlvs(value):
                    ident = {'plmn': None, 'gid1': None, 'gid2': None}
                    for t2, v2 in _tlvs(op):
                        if t2 == 0x80:
                            ident['plmn'] = v2.hex().upper()
                        elif t2 == 0x81:
                            ident['gid1'] = v2.hex().upper()
                        elif t2 == 0x82:
                            ident['gid2'] = v2.hex().upper()
                    operators.append(ident)
                entry['allowed_operators'] = operators
            elif tag == 0x82:
                entry['ppr_flags'] = _decode_bit_string(value, PPR_FLAG_BITS)
        out.append(entry)
    return out


def _raw_request(scc, cmd_cls):
    """Raw response hex of a request TLV (no pySim response decoding)."""
    return CardApplicationISDR.store_data_tlv(scc, cmd_cls(), None)


def chip_info(app):
    """EID (ES10c GetEuiccData), EUICCInfo1/2, configured addresses and RAT."""
    out = {'eid': None, 'info1': None, 'info2': None, 'addresses': None,
           'rat': None, 'errors': {}}
    scc = _select_isdr(app)
    try:
        parts = (
            ('eid', lambda: CardApplicationISDR.get_eid(scc)),
            ('info1', lambda: _decode_info1(_raw_request(scc, EuiccInfo1))),
            ('info2', lambda: _decode_info2(_raw_request(scc, EuiccInfo2))),
            ('addresses', lambda: _decode_addresses(
                _raw_request(scc, EuiccConfiguredAddresses))),
            ('rat', lambda: _decode_rat(_raw_request(scc, _GetRatRequest))),
        )
        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')
        icon = p.get('icon')
        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'),
            'icon': icon,
            'icon_size': len(icon) // 2 if icon else None,
            'owner': owner.get('profile_owner_plmn') if isinstance(owner, dict) else None,
        })
    return {'profiles': out, 'error': None}


# ProfileMgmtOperation flags in bit order (SGP.22 §5.7.9).  The TLV carries a
# padding-bits octet followed by the flags octet, so a TLV parsed from the
# card nests the flags under 'pmo' while an object built from decoded flags
# carries them directly.
PROFILE_MGMT_OPERATIONS = ('install', 'enable', 'disable', 'delete')


def _profile_operations(op):
    """ProfileMgmtOperation flags -> operation names ([] when unknown)."""
    if not isinstance(op, dict):
        return []
    flags = op.get('pmo') if isinstance(op.get('pmo'), dict) else op
    return [name for name in PROFILE_MGMT_OPERATIONS if flags.get(name)]


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'):
        out.append({
            'seq_number': n.get('seq_number'),
            'operations': _profile_operations(n.get('profile_mgmt_operation')),
            'address': n.get('notification_address'),
            'iccid': n.get('iccid'),
        })
    return {'notifications': out, 'error': None}


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 = []
    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)
    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(app, send_apdu, run_chain, action, iccid=None,
                   isdp_aid=None, refresh=True):
    """Run an ES10c Enable/DisableProfile switch on the ISD-R.

    ``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).  The ISD-R is
    selected first and the previous selection restored afterwards, like the
    other ES10 functions.

    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)
    _select_isdr(app)
    try:
        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}
    finally:
        _restore(app)