tests: stop test_log from leaking the print callback

PySimLogger.setup() installs a process-global print callback.
PySimLogger_Test sets one, a helper that asserts the message equals a global
expected_message, and never removes it, so from the moment test_log runs,
every PySimLogger message emitted anywhere in the process is checked against
whatever string that global happens to hold.

Fortunately unittest discovery runs modules in sorted order, and today
the PySimLogger users that log during tests all sort before test_log, so
this only breaks as soon as I try to add tests, just like anything else
breaks as soon as I try to use it.

Change-Id: I481e2c443fe0f412380b0f1acf6da5971ffca147

osmo-smdpp.py

@@ -640,7 +640,7 @@ class SmDppHttpServer:
                 # look up profile based on matchingID.  We simply check if a given file exists for now..
                 path = os.path.join(self.upp_dir, matchingId) + '.der'
                 # prevent directory traversal attack
-                if os.path.commonprefix((os.path.realpath(path),self.upp_dir)) != self.upp_dir:
+                if os.path.commonpath((os.path.realpath(path),self.upp_dir)) != self.upp_dir:
                     raise ApiError('8.2.6', '3.8', 'Refused')
                 if not os.path.isfile(path) or not os.access(path, os.R_OK):
                     raise ApiError('8.2.6', '3.8', 'Refused')

pySim-shell.py

@@ -24,21 +24,21 @@ import traceback
 import re
 import cmd2
 from packaging import version
-from cmd2 import style
-
 import logging
 from pySim.log import PySimLogger
 from osmocom.utils import auto_uint8
 
-# cmd2 >= 2.3.0 has deprecated the bg/fg in favor of Bg/Fg :(
-if version.parse(cmd2.__version__) < version.parse("2.3.0"):
-    from cmd2 import fg, bg # pylint: disable=no-name-in-module
-    RED = fg.red
-    YELLOW = fg.yellow
-    LIGHT_RED = fg.bright_red
-    LIGHT_GREEN = fg.bright_green
+# cmd2 >= 3.0 replaced Fg + style() with Color + stylize()
+if version.parse(cmd2.__version__) >= version.parse("3.0.0"):
+    from cmd2 import Color, stylize # pylint: disable=no-name-in-module
+    RED = Color.RED
+    YELLOW = Color.YELLOW
+    LIGHT_RED = Color.BRIGHT_RED
+    LIGHT_GREEN = Color.BRIGHT_GREEN
+    def style(text, fg=None, bg=None, bold=False): # pylint: disable=function-redefined
+        return stylize(text, fg) if fg else text
 else:
-    from cmd2 import Fg, Bg # pylint: disable=no-name-in-module
+    from cmd2 import style, Fg # pylint: disable=no-name-in-module
     RED = Fg.RED
     YELLOW = Fg.YELLOW
     LIGHT_RED = Fg.LIGHT_RED
@@ -69,50 +69,26 @@ from pySim.ts_102_222 import Ts102222Commands
 from pySim.gsm_r import DF_EIRENE
 from pySim.cat import ProactiveCommand
 
-from pySim.card_key_provider import CardKeyProviderCsv, CardKeyProviderPgsql
-from pySim.card_key_provider import card_key_provider_register, card_key_provider_get_field, card_key_provider_get
+from pySim.card_key_provider import card_key_provider_argparse_add_args, card_key_provider_init
+from pySim.card_key_provider import card_key_provider_get_field, card_key_provider_get
 
 from pySim.app import init_card
 
 log = PySimLogger.get(Path(__file__).stem)
 
-class Cmd2Compat(cmd2.Cmd):
-    """Backwards-compatibility wrapper around cmd2.Cmd to support older and newer
-    releases. See https://github.com/python-cmd2/cmd2/blob/master/CHANGELOG.md"""
-    def run_editor(self, file_path: Optional[str] = None) -> None:
-        if version.parse(cmd2.__version__) < version.parse("2.0.0"):
-            return self._run_editor(file_path) # pylint: disable=no-member
-        else:
-            return super().run_editor(file_path) # pylint: disable=no-member
-
-class Settable2Compat(cmd2.Settable):
-    """Backwards-compatibility wrapper around cmd2.Settable to support older and newer
-    releases. See https://github.com/python-cmd2/cmd2/blob/master/CHANGELOG.md"""
-    def __init__(self, name, val_type, description, settable_object, **kwargs):
-        if version.parse(cmd2.__version__) < version.parse("2.0.0"):
-            super().__init__(name, val_type, description, **kwargs) # pylint: disable=no-value-for-parameter
-        else:
-            super().__init__(name, val_type, description, settable_object, **kwargs) # pylint: disable=too-many-function-args
-
-class PysimApp(Cmd2Compat):
+class PysimApp(cmd2.Cmd):
     CUSTOM_CATEGORY = 'pySim Commands'
     BANNER = """Welcome to pySim-shell!
 (C) 2021-2023 by Harald Welte, sysmocom - s.f.m.c. GmbH and contributors
 Online manual available at https://downloads.osmocom.org/docs/pysim/master/html/shell.html """
 
     def __init__(self, verbose, card, rs, sl, ch, script=None):
-        if version.parse(cmd2.__version__) < version.parse("2.0.0"):
-            kwargs = {'use_ipython': True}
-        else:
-            kwargs = {'include_ipy': True}
-
         self.verbose = verbose
         PySimLogger.setup(self.poutput, {logging.WARN: YELLOW})
         self._onchange_verbose('verbose', False, self.verbose)
 
-        # pylint: disable=unexpected-keyword-arg
         super().__init__(persistent_history_file='~/.pysim_shell_history', allow_cli_args=False,
-                         auto_load_commands=False, startup_script=script, **kwargs)
+                         auto_load_commands=False, startup_script=script, include_ipy=True)
         self.intro = style(self.BANNER, fg=RED)
         self.default_category = 'pySim-shell built-in commands'
         self.card = None
@@ -128,18 +104,24 @@ Online manual available at https://downloads.osmocom.org/docs/pysim/master/html/
         self.apdu_trace = False
         self.apdu_strict = False
 
-        self.add_settable(Settable2Compat('numeric_path', bool, 'Print File IDs instead of names', self,
-                                          onchange_cb=self._onchange_numeric_path))
-        self.add_settable(Settable2Compat('conserve_write', bool, 'Read and compare before write', self,
-                                          onchange_cb=self._onchange_conserve_write))
-        self.add_settable(Settable2Compat('json_pretty_print', bool, 'Pretty-Print JSON output', self))
-        self.add_settable(Settable2Compat('apdu_trace', bool, 'Trace and display APDUs exchanged with card', self,
-                                          onchange_cb=self._onchange_apdu_trace))
-        self.add_settable(Settable2Compat('apdu_strict', bool,
-                                          'Strictly apply APDU format according to ISO/IEC 7816-3, table 12', self))
-        self.add_settable(Settable2Compat('verbose', bool,
-                                          'Enable/disable verbose logging', self,
-                                          onchange_cb=self._onchange_verbose))
+        self.add_settable(cmd2.Settable('numeric_path', bool,
+                                        'Print File IDs instead of names',
+                                        self, onchange_cb=self._onchange_numeric_path))
+        self.add_settable(cmd2.Settable('conserve_write', bool,
+                                        'Read and compare before write',
+                                        self, onchange_cb=self._onchange_conserve_write))
+        self.add_settable(cmd2.Settable('json_pretty_print', bool,
+                                        'Pretty-Print JSON output',
+                                        self))
+        self.add_settable(cmd2.Settable('apdu_trace', bool,
+                                        'Trace and display APDUs exchanged with card',
+                                        self, onchange_cb=self._onchange_apdu_trace))
+        self.add_settable(cmd2.Settable('apdu_strict', bool,
+                                        'Strictly apply APDU format according to ISO/IEC 7816-3, table 12',
+                                        self))
+        self.add_settable(cmd2.Settable('verbose', bool,
+                                        'Enable/disable verbose logging',
+                                        self, onchange_cb=self._onchange_verbose))
         self.equip(card, rs)
 
     def equip(self, card, rs):
@@ -1146,18 +1128,6 @@ global_group.add_argument("--skip-card-init", help="Skip all card/profile initia
 global_group.add_argument("--verbose", help="Enable verbose logging",
                           action='store_true', default=False)
 
-card_key_group = option_parser.add_argument_group('Card Key Provider Options')
-card_key_group.add_argument('--csv', metavar='FILE',
-                            default="~/.osmocom/pysim/card_data.csv",
-                            help='Read card data from CSV file')
-card_key_group.add_argument('--pgsql', metavar='FILE',
-                            default="~/.osmocom/pysim/card_data_pgsql.cfg",
-                            help='Read card data from PostgreSQL database (config file)')
-card_key_group.add_argument('--csv-column-key', metavar='FIELD:AES_KEY_HEX', default=[], action='append',
-                            help=argparse.SUPPRESS, dest='column_key')
-card_key_group.add_argument('--column-key', metavar='FIELD:AES_KEY_HEX', default=[], action='append',
-                            help='per-column AES transport key', dest='column_key')
-
 adm_group = global_group.add_mutually_exclusive_group()
 adm_group.add_argument('-a', '--pin-adm', metavar='PIN_ADM1', dest='pin_adm', default=None,
                        help='ADM PIN used for provisioning (overwrites default)')
@@ -1170,6 +1140,7 @@ option_parser.add_argument("command", nargs='?',
                            help="A pySim-shell command that would optionally be executed at startup")
 option_parser.add_argument('command_args', nargs=argparse.REMAINDER,
                            help="Optional Arguments for command")
+card_key_provider_argparse_add_args(option_parser)
 
 if __name__ == '__main__':
     startup_errors = False
@@ -1178,16 +1149,8 @@ if __name__ == '__main__':
     # Ensure that we are able to print formatted warnings from the beginning.
     PySimLogger.setup(print, {logging.WARN: YELLOW}, opts.verbose)
 
-    # Register csv-file as card data provider, either from specified CSV
-    # or from CSV file in home directory
-    column_keys = {}
-    for par in opts.column_key:
-        name, key = par.split(':')
-        column_keys[name] = key
-    if os.path.isfile(os.path.expanduser(opts.csv)):
-        card_key_provider_register(CardKeyProviderCsv(os.path.expanduser(opts.csv), column_keys))
-    if os.path.isfile(os.path.expanduser(opts.pgsql)):
-        card_key_provider_register(CardKeyProviderPgsql(os.path.expanduser(opts.pgsql), column_keys))
+    # Init card key provider for automatic card key retrieval
+    card_key_provider_init(opts)
 
     # Init card reader driver
     sl = init_reader(opts, proactive_handler = Proact())

pySim-trace.py

@@ -117,7 +117,7 @@ class Tracer:
             try:
                 apdu = self.source.read()
                 apdu_counter = apdu_counter + 1
-            except StopIteration:
+            except (StopIteration, KeyboardInterrupt):
                 print("%i APDUs parsed, stop iteration." % apdu_counter)
                 return 0
 

pySim/app.py

@@ -26,6 +26,9 @@ from pySim.cdma_ruim import CardProfileRUIM
 from pySim.ts_102_221 import CardProfileUICC
 from pySim.utils import all_subclasses
 from pySim.exceptions import SwMatchError
+from pySim.log import PySimLogger
+
+log = PySimLogger.get(__name__)
 
 # we need to import this module so that the SysmocomSJA2 sub-class of
 # CardModel is created, which will add the ATR-based matching and
@@ -54,7 +57,7 @@ def init_card(sl: LinkBase, skip_card_init: bool = False) -> Tuple[RuntimeState,
 
     # Wait up to three seconds for a card in reader and try to detect
     # the card type.
-    print("Waiting for card...")
+    log.info("Waiting for card...")
     sl.wait_for_card(3)
 
     # The user may opt to skip all card initialization. In this case only the
@@ -66,7 +69,7 @@ def init_card(sl: LinkBase, skip_card_init: bool = False) -> Tuple[RuntimeState,
     generic_card = False
     card = card_detect(scc)
     if card is None:
-        print("Warning: Could not detect card type - assuming a generic card type...")
+        log.warning("Could not detect card type - assuming a generic card type...")
         card = SimCardBase(scc)
         generic_card = True
 
@@ -76,7 +79,7 @@ def init_card(sl: LinkBase, skip_card_init: bool = False) -> Tuple[RuntimeState,
         # just means that pySim was unable to recognize the card profile. This
         # may happen in particular with unprovisioned cards that do not have
         # any files on them yet.
-        print("Unsupported card type!")
+        log.warning("Unsupported card type!")
         return None, card
 
     # ETSI TS 102 221, Table 9.3 specifies a default for the PIN key
@@ -87,7 +90,7 @@ def init_card(sl: LinkBase, skip_card_init: bool = False) -> Tuple[RuntimeState,
     if generic_card and isinstance(profile, CardProfileUICC):
         card._adm_chv_num = 0x0A
 
-    print("Info: Card is of type: %s" % str(profile))
+    log.info("Card is of type: %s", str(profile))
 
     # FIXME: this shouldn't really be here but somewhere else/more generic.
     # We cannot do it within pySim/profile.py as that would create circular

pySim/ara_m.py

@@ -300,6 +300,51 @@ class ADF_ARAM(CardADF):
                                'major': v_major, 'minor': v_minor, 'patch': v_patch}}])
         return ADF_ARAM.xceive_apdu_tlv(scc, '80cadf21', cmd_do, ResponseAramConfigDO)
 
+    @staticmethod
+    def store_ref_ar_do(scc, aid:Hexstr, aid_empty:bool, device_app_id:Hexstr, pkg_ref:str,
+                        apdu_filter:Hexstr, apdu_never:bool, apdu_always:bool,
+                        nfc_always:bool, nfc_never:bool, android_permissions:Hexstr):
+        # REF
+        ref_do_content = []
+        if aid is not None:
+            ref_do_content += [{'aid_ref_do': aid}]
+        elif aid_empty:
+            ref_do_content += [{'aid_ref_empty_do': None}]
+        ref_do_content += [{'dev_app_id_ref_do': device_app_id}]
+        if pkg_ref:
+            ref_do_content += [{'pkg_ref_do': {'package_name_string': pkg_ref}}]
+        # AR
+        ar_do_content = []
+        if apdu_never:
+            ar_do_content += [{'apdu_ar_do': {'generic_access_rule': 'never'}}]
+        elif apdu_always:
+            ar_do_content += [{'apdu_ar_do': {'generic_access_rule': 'always'}}]
+        elif apdu_filter:
+            if len(apdu_filter) % 16:
+                raise ValueError(f'Invalid non-modulo-16 length of APDU filter: {len(apdu_filter)}')
+            offset = 0
+            apdu_filter_list = []
+            while offset < len(apdu_filter):
+                apdu_filter_list += [{'header': apdu_filter[offset:offset+8],
+                                      'mask': apdu_filter[offset+8:offset+16]}]
+                offset += 16 # Move offset to the beginning of the next apdu_filter object
+            ar_do_content += [{'apdu_ar_do': {'apdu_filter': apdu_filter_list}}]
+        if nfc_never:
+            ar_do_content += [{'nfc_ar_do': {'nfc_event_access_rule': 'never'}}]
+        elif nfc_always:
+            ar_do_content += [{'nfc_ar_do': {'nfc_event_access_rule': 'always'}}]
+        if android_permissions:
+            ar_do_content += [{'perm_ar_do': {'permissions': android_permissions}}]
+        d = [{'ref_ar_do': [{'ref_do': ref_do_content}, {'ar_do': ar_do_content}]}]
+        csrado = CommandStoreRefArDO()
+        csrado.from_val_dict(d)
+        return ADF_ARAM.store_data(scc, csrado)
+
+    @staticmethod
+    def aram_delete_all(scc):
+        deldo = CommandDelete()
+        return ADF_ARAM.store_data(scc, deldo)
+
     @with_default_category('Application-Specific Commands')
     class AddlShellCommands(CommandSet):
         def do_aram_get_all(self, _opts):
@@ -334,58 +379,25 @@ class ADF_ARAM(CardADF):
         apdu_grp.add_argument(
             '--apdu-filter', help='APDU filter: multiple groups of 8 hex bytes (4 byte CLA/INS/P1/P2 followed by 4 byte mask)')
         nfc_grp = store_ref_ar_do_parse.add_mutually_exclusive_group()
-        nfc_grp.add_argument('--nfc-always', action='store_true',
-                             help='NFC event access is allowed')
         nfc_grp.add_argument('--nfc-never', action='store_true',
                              help='NFC event access is not allowed')
+        nfc_grp.add_argument('--nfc-always', action='store_true',
+                             help='NFC event access is allowed')
         store_ref_ar_do_parse.add_argument(
             '--android-permissions', help='Android UICC Carrier Privilege Permissions (8 hex bytes)')
 
         @cmd2.with_argparser(store_ref_ar_do_parse)
         def do_aram_store_ref_ar_do(self, opts):
             """Perform STORE DATA [Command-Store-REF-AR-DO] to store a (new) access rule."""
-            # REF
-            ref_do_content = []
-            if opts.aid is not None:
-                ref_do_content += [{'aid_ref_do': opts.aid}]
-            elif opts.aid_empty:
-                ref_do_content += [{'aid_ref_empty_do': None}]
-            ref_do_content += [{'dev_app_id_ref_do': opts.device_app_id}]
-            if opts.pkg_ref:
-                ref_do_content += [{'pkg_ref_do': {'package_name_string': opts.pkg_ref}}]
-            # AR
-            ar_do_content = []
-            if opts.apdu_never:
-                ar_do_content += [{'apdu_ar_do': {'generic_access_rule': 'never'}}]
-            elif opts.apdu_always:
-                ar_do_content += [{'apdu_ar_do': {'generic_access_rule': 'always'}}]
-            elif opts.apdu_filter:
-                if len(opts.apdu_filter) % 16:
-                    raise ValueError(f'Invalid non-modulo-16 length of APDU filter: {len(opts.apdu_filter)}')
-                offset = 0
-                apdu_filter = []
-                while offset < len(opts.apdu_filter):
-                    apdu_filter += [{'header': opts.apdu_filter[offset:offset+8],
-                                     'mask': opts.apdu_filter[offset+8:offset+16]}]
-                    offset += 16 # Move offset to the beginning of the next apdu_filter object
-                ar_do_content += [{'apdu_ar_do': {'apdu_filter': apdu_filter}}]
-            if opts.nfc_always:
-                ar_do_content += [{'nfc_ar_do': {'nfc_event_access_rule': 'always'}}]
-            elif opts.nfc_never:
-                ar_do_content += [{'nfc_ar_do': {'nfc_event_access_rule': 'never'}}]
-            if opts.android_permissions:
-                ar_do_content += [{'perm_ar_do': {'permissions': opts.android_permissions}}]
-            d = [{'ref_ar_do': [{'ref_do': ref_do_content}, {'ar_do': ar_do_content}]}]
-            csrado = CommandStoreRefArDO()
-            csrado.from_val_dict(d)
-            res_do = ADF_ARAM.store_data(self._cmd.lchan.scc, csrado)
+            res_do = ADF_ARAM.store_ref_ar_do(self._cmd.lchan.scc, opts.aid, opts.aid_empty, opts.device_app_id,
+                                              opts.pkg_ref, opts.apdu_filter, opts.apdu_never, opts.apdu_always,
+                                              opts.nfc_always, opts.nfc_never, opts.android_permissions)
             if res_do:
                 self._cmd.poutput_json(res_do.to_dict())
 
         def do_aram_delete_all(self, _opts):
             """Perform STORE DATA [Command-Delete[all]] to delete all access rules."""
-            deldo = CommandDelete()
-            res_do = ADF_ARAM.store_data(self._cmd.lchan.scc, deldo)
+            res_do = ADF_ARAM.aram_delete_all(self._cmd.lchan.scc)
             if res_do:
                 self._cmd.poutput_json(res_do.to_dict())
 
@@ -394,7 +406,6 @@ class ADF_ARAM(CardADF):
             (Proprietary feature that is specific to sysmocom's fork of Bertrand Martel’s ARA-M implementation.)"""
             self._cmd.lchan.scc.send_apdu_checksw('80e2900001A1', '9000')
 
-
 # SEAC v1.1 Section 4.1.2.2 + 5.1.2.2
 sw_aram = {
     'ARA-M': {

pySim/card_key_provider.py

@@ -33,10 +33,12 @@ from Cryptodome.Cipher import AES
 from osmocom.utils import h2b, b2h
 from pySim.log import PySimLogger
 
+import os
 import abc
 import csv
 import logging
 import yaml
+import argparse
 
 log = PySimLogger.get(__name__)
 
@@ -130,6 +132,31 @@ class CardKeyFieldCryptor:
         cipher = AES.new(h2b(self.transport_keys[field_name.upper()]), AES.MODE_CBC, self.__IV)
         return b2h(cipher.encrypt(h2b(plaintext_val)))
 
+    @staticmethod
+    def argparse_add_args(arg_parser: argparse.ArgumentParser):
+        arg_parser.add_argument('--column-key', metavar='FIELD:AES_KEY_HEX', default=[], action='append',
+                                help='per-column AES transport key', dest='column_key')
+        # Depprecated argument, replaced by --column-key (see above)
+        arg_parser.add_argument('--csv-column-key', metavar='FIELD:AES_KEY_HEX', default=[], action='append',
+                                help=argparse.SUPPRESS, dest='column_key')
+
+    @staticmethod
+    def transport_keys_from_opts(opts: argparse.Namespace) -> dict:
+        """
+        Transport keys are passed via the commandline using the '--column-key' option. Each column requires a
+        dedicated transport key. This method can be used to extract the column keys parameters from the commandline
+        options into a dict that can be directly passed to the construtor with the transport_keys argument.
+
+        Args:
+                opts: parsed commandline options (Namespace)
+        """
+
+        transport_keys = {}
+        for par in opts.column_key:
+            name, key = par.split(':')
+            transport_keys[name] = key
+        return transport_keys
+
 class CardKeyProvider(abc.ABC):
     """Base class, not containing any concrete implementation."""
 
@@ -148,24 +175,33 @@ class CardKeyProvider(abc.ABC):
                 fond None shall be returned.
         """
 
+    @staticmethod
+    def argparse_add_args(arg_parser: argparse.ArgumentParser):
+        """
+        Add the commandline arguments relevant for this card key provider.
+
+        Args:
+                arg_parser : argument parser group
+        """
+
     def __str__(self):
         return type(self).__name__
 
 class CardKeyProviderCsv(CardKeyProvider):
     """Card key provider implementation that allows to query against a specified CSV file."""
 
-    def __init__(self, csv_filename: str, transport_keys: dict):
+    def __init__(self, csv_filename: str, field_cryptor: CardKeyFieldCryptor):
         """
         Args:
                 csv_filename : file name (path) of CSV file containing card-individual key/data
-                transport_keys : (see class CardKeyFieldCryptor)
+                field_cryptor : (see class CardKeyFieldCryptor)
         """
         log.info("Using CSV file as card key data source: %s" % csv_filename)
         self.csv_file = open(csv_filename, 'r')
         if not self.csv_file:
             raise RuntimeError("Could not open CSV file '%s'" % csv_filename)
         self.csv_filename = csv_filename
-        self.crypt = CardKeyFieldCryptor(transport_keys)
+        self.crypt = field_cryptor
 
     def get(self, fields: List[str], key: str, value: str) -> Dict[str, str]:
         self.csv_file.seek(0)
@@ -188,14 +224,20 @@ class CardKeyProviderCsv(CardKeyProvider):
             return None
         return return_dict
 
+    @staticmethod
+    def argparse_add_args(arg_parser: argparse.ArgumentParser):
+        arg_parser.add_argument('--csv', metavar='FILE',
+                                default="~/.osmocom/pysim/card_data.csv",
+                                help='Read card data from CSV file')
+
 class CardKeyProviderPgsql(CardKeyProvider):
     """Card key provider implementation that allows to query against a specified PostgreSQL database table."""
 
-    def __init__(self, config_filename: str, transport_keys: dict):
+    def __init__(self, config_filename: str, field_cryptor: CardKeyFieldCryptor):
         """
         Args:
                 config_filename : file name (path) of CSV file containing card-individual key/data
-                transport_keys : (see class CardKeyFieldCryptor)
+                field_cryptor : (see class CardKeyFieldCryptor)
         """
         import psycopg2
         log.info("Using SQL database as card key data source: %s" % config_filename)
@@ -212,7 +254,7 @@ class CardKeyProviderPgsql(CardKeyProvider):
                                          host=config.get('host'))
             self.tables = config.get('table_names')
             log.info("Card key database tables: %s" % str(self.tables))
-            self.crypt = CardKeyFieldCryptor(transport_keys)
+            self.crypt = field_cryptor
 
     def get(self, fields: List[str], key: str, value: str) -> Dict[str, str]:
         import psycopg2
@@ -252,6 +294,11 @@ class CardKeyProviderPgsql(CardKeyProvider):
             result[k] = self.crypt.decrypt_field(k, result.get(k))
         return result
 
+    @staticmethod
+    def argparse_add_args(arg_parser: argparse.ArgumentParser):
+        arg_parser.add_argument('--pgsql', metavar='FILE',
+                                default="~/.osmocom/pysim/card_data_pgsql.cfg",
+                                help='Read card data from PostgreSQL database (config file)')
 
 def card_key_provider_register(provider: CardKeyProvider, provider_list=card_key_providers):
     """Register a new card key provider.
@@ -305,3 +352,19 @@ def card_key_provider_get_field(field: str, key: str, value: str, provider_list=
     fields = [field]
     result = card_key_provider_get(fields, key, value, card_key_providers)
     return result.get(field.upper())
+
+def card_key_provider_argparse_add_args(arg_parser: argparse.ArgumentParser):
+    """Add card key provider commandline options to the given argument parser"""
+    card_key_group = arg_parser.add_argument_group('Card Key Provider Options')
+    CardKeyProviderCsv.argparse_add_args(card_key_group)
+    CardKeyProviderPgsql.argparse_add_args(card_key_group)
+    CardKeyFieldCryptor.argparse_add_args(card_key_group)
+
+def card_key_provider_init(opts: argparse.Namespace):
+    """Initialize card key provider depending on the user provided commandline options"""
+    transport_keys = CardKeyFieldCryptor.transport_keys_from_opts(opts)
+    card_key_field_cryptor = CardKeyFieldCryptor(transport_keys)
+    if os.path.isfile(os.path.expanduser(opts.csv)):
+        card_key_provider_register(CardKeyProviderCsv(os.path.expanduser(opts.csv), card_key_field_cryptor))
+    if os.path.isfile(os.path.expanduser(opts.pgsql)):
+        card_key_provider_register(CardKeyProviderPgsql(os.path.expanduser(opts.pgsql), card_key_field_cryptor))

pySim/esim/es2p.py

@@ -16,6 +16,12 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import requests
+from klein import Klein
+from twisted.internet import defer, protocol, ssl, task, endpoints, reactor
+from twisted.internet.posixbase import PosixReactorBase
+from pathlib import Path
+from twisted.web.server import Site, Request
+
 import logging
 from datetime import datetime
 import time
@@ -123,10 +129,12 @@ class Es2PlusApiFunction(JsonHttpApiFunction):
 class DownloadOrder(Es2PlusApiFunction):
     path = '/gsma/rsp2/es2plus/downloadOrder'
     input_params = {
+        'header': JsonRequestHeader,
         'eid': param.Eid,
         'iccid': param.Iccid,
         'profileType': param.ProfileType
       }
+    input_mandatory = ['header']
     output_params = {
         'header': JsonResponseHeader,
         'iccid': param.Iccid,
@@ -137,6 +145,7 @@ class DownloadOrder(Es2PlusApiFunction):
 class ConfirmOrder(Es2PlusApiFunction):
     path = '/gsma/rsp2/es2plus/confirmOrder'
     input_params = {
+        'header': JsonRequestHeader,
         'iccid': param.Iccid,
         'eid': param.Eid,
         'matchingId': param.MatchingId,
@@ -144,7 +153,7 @@ class ConfirmOrder(Es2PlusApiFunction):
         'smdsAddress': param.SmdsAddress,
         'releaseFlag': param.ReleaseFlag,
       }
-    input_mandatory = ['iccid', 'releaseFlag']
+    input_mandatory = ['header', 'iccid', 'releaseFlag']
     output_params = {
         'header': JsonResponseHeader,
         'eid': param.Eid,
@@ -157,12 +166,13 @@ class ConfirmOrder(Es2PlusApiFunction):
 class CancelOrder(Es2PlusApiFunction):
     path = '/gsma/rsp2/es2plus/cancelOrder'
     input_params = {
+        'header': JsonRequestHeader,
         'iccid': param.Iccid,
         'eid': param.Eid,
         'matchingId': param.MatchingId,
         'finalProfileStatusIndicator': param.FinalProfileStatusIndicator,
       }
-    input_mandatory = ['finalProfileStatusIndicator', 'iccid']
+    input_mandatory = ['header', 'finalProfileStatusIndicator', 'iccid']
     output_params = {
         'header': JsonResponseHeader,
       }
@@ -172,9 +182,10 @@ class CancelOrder(Es2PlusApiFunction):
 class ReleaseProfile(Es2PlusApiFunction):
     path = '/gsma/rsp2/es2plus/releaseProfile'
     input_params = {
+        'header': JsonRequestHeader,
         'iccid': param.Iccid,
       }
-    input_mandatory = ['iccid']
+    input_mandatory = ['header', 'iccid']
     output_params = {
         'header': JsonResponseHeader,
       }
@@ -184,6 +195,7 @@ class ReleaseProfile(Es2PlusApiFunction):
 class HandleDownloadProgressInfo(Es2PlusApiFunction):
     path = '/gsma/rsp2/es2plus/handleDownloadProgressInfo'
     input_params = {
+        'header': JsonRequestHeader,
         'eid': param.Eid,
         'iccid': param.Iccid,
         'profileType': param.ProfileType,
@@ -192,10 +204,9 @@ class HandleDownloadProgressInfo(Es2PlusApiFunction):
         'notificationPointStatus': param.NotificationPointStatus,
         'resultData': param.ResultData,
     }
-    input_mandatory = ['iccid', 'profileType', 'timestamp', 'notificationPointId', 'notificationPointStatus']
+    input_mandatory = ['header', 'iccid', 'profileType', 'timestamp', 'notificationPointId', 'notificationPointStatus']
     expected_http_status = 204
 
-
 class Es2pApiClient:
     """Main class representing a full ES2+ API client. Has one method for each API function."""
     def __init__(self, url_prefix:str, func_req_id:str, server_cert_verify: str = None, client_cert: str = None):
@@ -206,18 +217,17 @@ class Es2pApiClient:
         if client_cert:
             self.session.cert = client_cert
 
-        self.downloadOrder = DownloadOrder(url_prefix, func_req_id, self.session)
-        self.confirmOrder = ConfirmOrder(url_prefix, func_req_id, self.session)
-        self.cancelOrder = CancelOrder(url_prefix, func_req_id, self.session)
-        self.releaseProfile = ReleaseProfile(url_prefix, func_req_id, self.session)
-        self.handleDownloadProgressInfo = HandleDownloadProgressInfo(url_prefix, func_req_id, self.session)
+        self.downloadOrder = JsonHttpApiClient(DownloadOrder(), url_prefix, func_req_id, self.session)
+        self.confirmOrder = JsonHttpApiClient(ConfirmOrder(), url_prefix, func_req_id, self.session)
+        self.cancelOrder = JsonHttpApiClient(CancelOrder(), url_prefix, func_req_id, self.session)
+        self.releaseProfile = JsonHttpApiClient(ReleaseProfile(), url_prefix, func_req_id, self.session)
+        self.handleDownloadProgressInfo = JsonHttpApiClient(HandleDownloadProgressInfo(), url_prefix, func_req_id, self.session)
 
     def _gen_func_id(self) -> str:
         """Generate the next function call id."""
         self.func_id += 1
         return 'FCI-%u-%u' % (time.time(), self.func_id)
 
-
     def call_downloadOrder(self, data: dict) -> dict:
         """Perform ES2+ DownloadOrder function (SGP.22 section 5.3.1)."""
         return self.downloadOrder.call(data, self._gen_func_id())
@@ -237,3 +247,116 @@ class Es2pApiClient:
     def call_handleDownloadProgressInfo(self, data: dict) -> dict:
         """Perform ES2+ HandleDownloadProgressInfo function (SGP.22 section 5.3.5)."""
         return self.handleDownloadProgressInfo.call(data, self._gen_func_id())
+
+class Es2pApiServerHandlerSmdpp(abc.ABC):
+    """ES2+ (SMDP+ side) API Server handler class. The API user is expected to override the contained methods."""
+
+    @abc.abstractmethod
+    def call_downloadOrder(self, data: dict) -> (dict, str):
+        """Perform ES2+ DownloadOrder function (SGP.22 section 5.3.1)."""
+        pass
+
+    @abc.abstractmethod
+    def call_confirmOrder(self, data: dict) -> (dict, str):
+        """Perform ES2+ ConfirmOrder function (SGP.22 section 5.3.2)."""
+        pass
+
+    @abc.abstractmethod
+    def call_cancelOrder(self, data: dict) -> (dict, str):
+        """Perform ES2+ CancelOrder function (SGP.22 section 5.3.3)."""
+        pass
+
+    @abc.abstractmethod
+    def call_releaseProfile(self, data: dict) -> (dict, str):
+        """Perform ES2+ CancelOrder function (SGP.22 section 5.3.4)."""
+        pass
+
+class Es2pApiServerHandlerMno(abc.ABC):
+    """ES2+ (MNO side) API Server handler class. The API user is expected to override the contained methods."""
+
+    @abc.abstractmethod
+    def call_handleDownloadProgressInfo(self, data: dict) -> (dict, str):
+        """Perform ES2+ HandleDownloadProgressInfo function (SGP.22 section 5.3.5)."""
+        pass
+
+class Es2pApiServer(abc.ABC):
+    """Main class representing a full ES2+ API server. Has one method for each API function."""
+    app = None
+
+    def __init__(self, port: int, interface: str, server_cert: str = None, client_cert_verify: str = None):
+        logger.debug("HTTP SRV: starting ES2+ API server on %s:%s" % (interface, port))
+        self.port = port
+        self.interface = interface
+        if server_cert:
+            self.server_cert = ssl.PrivateCertificate.loadPEM(Path(server_cert).read_text())
+        else:
+            self.server_cert = None
+        if client_cert_verify:
+            self.client_cert_verify = ssl.Certificate.loadPEM(Path(client_cert_verify).read_text())
+        else:
+            self.client_cert_verify = None
+
+    def reactor(self, reactor: PosixReactorBase):
+        logger.debug("HTTP SRV: listen on %s:%s" % (self.interface, self.port))
+        if self.server_cert:
+            if self.client_cert_verify:
+                reactor.listenSSL(self.port, Site(self.app.resource()), self.server_cert.options(self.client_cert_verify),
+                                  interface=self.interface)
+            else:
+                reactor.listenSSL(self.port, Site(self.app.resource()), self.server_cert.options(),
+                                  interface=self.interface)
+        else:
+            reactor.listenTCP(self.port, Site(self.app.resource()), interface=self.interface)
+        return defer.Deferred()
+
+class Es2pApiServerSmdpp(Es2pApiServer):
+    """ES2+ (SMDP+ side) API Server."""
+    app = Klein()
+
+    def __init__(self, port: int, interface: str, handler: Es2pApiServerHandlerSmdpp,
+                 server_cert: str = None, client_cert_verify: str = None):
+        super().__init__(port, interface, server_cert, client_cert_verify)
+        self.handler = handler
+        self.downloadOrder = JsonHttpApiServer(DownloadOrder(), handler.call_downloadOrder)
+        self.confirmOrder = JsonHttpApiServer(ConfirmOrder(), handler.call_confirmOrder)
+        self.cancelOrder = JsonHttpApiServer(CancelOrder(), handler.call_cancelOrder)
+        self.releaseProfile = JsonHttpApiServer(ReleaseProfile(), handler.call_releaseProfile)
+        task.react(self.reactor)
+
+    @app.route(DownloadOrder.path)
+    def call_downloadOrder(self, request: Request) -> dict:
+        """Perform ES2+ DownloadOrder function (SGP.22 section 5.3.1)."""
+        return self.downloadOrder.call(request)
+
+    @app.route(ConfirmOrder.path)
+    def call_confirmOrder(self, request: Request) -> dict:
+        """Perform ES2+ ConfirmOrder function (SGP.22 section 5.3.2)."""
+        return self.confirmOrder.call(request)
+
+    @app.route(CancelOrder.path)
+    def call_cancelOrder(self, request: Request) -> dict:
+        """Perform ES2+ CancelOrder function (SGP.22 section 5.3.3)."""
+        return self.cancelOrder.call(request)
+
+    @app.route(ReleaseProfile.path)
+    def call_releaseProfile(self, request: Request) -> dict:
+        """Perform ES2+ CancelOrder function (SGP.22 section 5.3.4)."""
+        return self.releaseProfile.call(request)
+
+class Es2pApiServerMno(Es2pApiServer):
+    """ES2+ (MNO side) API Server."""
+
+    app = Klein()
+
+    def __init__(self, port: int, interface: str, handler: Es2pApiServerHandlerMno,
+                 server_cert: str = None, client_cert_verify: str = None):
+        super().__init__(port, interface, server_cert, client_cert_verify)
+        self.handler = handler
+        self.handleDownloadProgressInfo = JsonHttpApiServer(HandleDownloadProgressInfo(),
+                                                            handler.call_handleDownloadProgressInfo)
+        task.react(self.reactor)
+
+    @app.route(HandleDownloadProgressInfo.path)
+    def call_handleDownloadProgressInfo(self, request: Request) -> dict:
+        """Perform ES2+ HandleDownloadProgressInfo function (SGP.22 section 5.3.5)."""
+        return self.handleDownloadProgressInfo.call(request)

pySim/esim/es9p.py

@@ -155,11 +155,11 @@ class Es9pApiClient:
         if server_cert_verify:
             self.session.verify = server_cert_verify
 
-        self.initiateAuthentication = InitiateAuthentication(url_prefix, '', self.session)
-        self.authenticateClient = AuthenticateClient(url_prefix, '', self.session)
-        self.getBoundProfilePackage = GetBoundProfilePackage(url_prefix, '', self.session)
-        self.handleNotification = HandleNotification(url_prefix, '', self.session)
-        self.cancelSession = CancelSession(url_prefix, '', self.session)
+        self.initiateAuthentication = JsonHttpApiClient(InitiateAuthentication(), url_prefix, '', self.session)
+        self.authenticateClient = JsonHttpApiClient(AuthenticateClient(), url_prefix, '', self.session)
+        self.getBoundProfilePackage = JsonHttpApiClient(GetBoundProfilePackage(), url_prefix, '', self.session)
+        self.handleNotification = JsonHttpApiClient(HandleNotification(), url_prefix, '', self.session)
+        self.cancelSession = JsonHttpApiClient(CancelSession(), url_prefix, '', self.session)
 
     def call_initiateAuthentication(self, data: dict) -> dict:
         return self.initiateAuthentication.call(data)

pySim/esim/http_json_api.py

@@ -19,8 +19,10 @@ import abc
 import requests
 import logging
 import json
-from typing import Optional
+from typing import Optional, Tuple
 import base64
+from twisted.web.server import Request
+
 
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.DEBUG)
@@ -131,6 +133,16 @@ class JsonResponseHeader(ApiParam):
         if status not in ['Executed-Success', 'Executed-WithWarning', 'Failed', 'Expired']:
             raise ValueError('Unknown/unspecified status "%s"' % status)
 
+class JsonRequestHeader(ApiParam):
+    """SGP.22 section 6.5.1.3."""
+    @classmethod
+    def verify_decoded(cls, data):
+        func_req_id = data.get('functionRequesterIdentifier')
+        if not func_req_id:
+            raise ValueError('Missing mandatory functionRequesterIdentifier in header')
+        func_call_id = data.get('functionCallIdentifier')
+        if not func_call_id:
+            raise ValueError('Missing mandatory functionCallIdentifier in header')
 
 class HttpStatusError(Exception):
     pass
@@ -161,65 +173,118 @@ class ApiError(Exception):
 
 class JsonHttpApiFunction(abc.ABC):
     """Base class for representing an HTTP[s] API Function."""
-     # the below class variables are expected to be overridden in derived classes
+    # The below class variables are used to describe the properties of the API function. Derived classes are expected
+    # to orverride those class properties with useful values. The prefixes "input_" and "output_" refer to the API
+    # function from an abstract point of view. Seen from the client perspective, "input_" will refer to parameters the
+    # client sends to a HTTP server. Seen from the server perspective, "input_" will refer to parameters the server
+    # receives from the a requesting client. The same applies vice versa to class variables that have an "output_"
+    # prefix.
 
+    # path of the API function (e.g. '/gsma/rsp2/es2plus/confirmOrder', see also method rewrite_url).
     path = None
+
     # dictionary of input parameters. key is parameter name, value is ApiParam class
     input_params = {}
+
     # list of mandatory input parameters
     input_mandatory = []
+
     # dictionary of output parameters. key is parameter name, value is ApiParam class
     output_params = {}
+
     # list of mandatory output parameters (for successful response)
     output_mandatory = []
+
+    # list of mandatory output parameters (for failed response)
+    output_mandatory_failed = []
+
     # expected HTTP status code of the response
     expected_http_status = 200
+
     # the HTTP method used (GET, OPTIONS, HEAD, POST, PUT, PATCH or DELETE)
     http_method = 'POST'
+
+    # additional custom HTTP headers (client requests)
     extra_http_req_headers = {}
 
-    def __init__(self, url_prefix: str, func_req_id: Optional[str], session: requests.Session):
-        self.url_prefix = url_prefix
-        self.func_req_id = func_req_id
-        self.session = session
+    # additional custom HTTP headers (server responses)
+    extra_http_res_headers = {}
 
-    def encode(self, data: dict, func_call_id: Optional[str] = None) -> dict:
+    def __new__(cls, *args, role = 'legacy_client', **kwargs):
+        """
+        Args:
+                args: (see JsonHttpApiClient and JsonHttpApiServer)
+                role: role ('server' or 'client') in which the JsonHttpApiFunction should be created.
+                kwargs: (see JsonHttpApiClient and JsonHttpApiServer)
+        """
+
+        # Create a dictionary with the class attributes of this class (the properties listed above and the encode_
+        # decode_ methods below). The dictionary will not include any dunder/magic methods
+        cls_attr = {attr_name: getattr(cls, attr_name) for attr_name in dir(cls) if not attr_name.startswith('__')}
+
+        # Normal instantiation as JsonHttpApiFunction:
+        if len(args) == 0 and len(kwargs) == 0:
+            return type(cls.__name__, (abc.ABC,), cls_attr)()
+
+        # Instantiation as as JsonHttpApiFunction with a JsonHttpApiClient or JsonHttpApiServer base
+        if role == 'legacy_client':
+            # Deprecated: With the advent of the server role (JsonHttpApiServer) the API had to be changed. To maintain
+            # compatibility with existing code (out-of-tree) the original behaviour and API interface and behaviour had
+            # to be preserved. Already existing JsonHttpApiFunction definitions will still work and the related objects
+            # may still be created on the original way: my_api_func = MyApiFunc(url_prefix, func_req_id, self.session)
+            logger.warning('implicit role (falling back to legacy JsonHttpApiClient) is deprecated, please specify role explcitly')
+            result = type(cls.__name__, (JsonHttpApiClient,), cls_attr)(None, *args, **kwargs)
+            result.api_func = result
+            result.legacy = True
+            return result
+        elif role == 'client':
+            # Create a JsonHttpApiFunction in client role
+            # Example: my_api_func = MyApiFunc(url_prefix, func_req_id, self.session, role='client')
+            result = type(cls.__name__, (JsonHttpApiClient,), cls_attr)(None, *args, **kwargs)
+            result.api_func = result
+            return result
+        elif role == 'server':
+            # Create a JsonHttpApiFunction in server role
+            # Example: my_api_func = MyApiFunc(url_prefix, func_req_id, self.session, role='server')
+            result = type(cls.__name__, (JsonHttpApiServer,), cls_attr)(None, *args, **kwargs)
+            result.api_func = result
+            return result
+        else:
+            raise ValueError('Invalid role \'%s\' specified' % role)
+
+    def encode_client(self, data: dict) -> dict:
         """Validate an encode input dict into JSON-serializable dict for request body."""
         output = {}
-        if func_call_id:
-            output['header'] = {
-                    'functionRequesterIdentifier': self.func_req_id,
-                    'functionCallIdentifier': func_call_id
-                }
-
         for p in self.input_mandatory:
             if not p in data:
                 raise ValueError('Mandatory input parameter %s missing' % p)
         for p, v in data.items():
             p_class = self.input_params.get(p)
             if not p_class:
-                logger.warning('Unexpected/unsupported input parameter %s=%s', p, v)
-                output[p] = v
+                # pySim/esim/http_json_api.py:269:47: E1101: Instance of 'JsonHttpApiFunction' has no 'legacy' member (no-member)
+                # pylint: disable=no-member
+                if hasattr(self, 'legacy') and self.legacy:
+                    output[p] = JsonRequestHeader.encode(v)
+                else:
+                    logger.warning('Unexpected/unsupported input parameter %s=%s', p, v)
+                    output[p] = v
             else:
                 output[p] = p_class.encode(v)
         return output
 
-    def decode(self, data: dict) -> dict:
+    def decode_client(self, data: dict) -> dict:
         """[further] Decode and validate the JSON-Dict of the response body."""
         output = {}
-        if 'header' in self.output_params:
-            # let's first do the header, it's special
-            if not 'header' in data:
-                raise ValueError('Mandatory output parameter "header" missing')
-            hdr_class = self.output_params.get('header')
-            output['header'] = hdr_class.decode(data['header'])
+        output_mandatory = self.output_mandatory
 
-            if output['header']['functionExecutionStatus']['status'] not in ['Executed-Success','Executed-WithWarning']:
-                raise ApiError(output['header']['functionExecutionStatus'])
-        # we can only expect mandatory parameters to be present in case of successful execution
-        for p in self.output_mandatory:
-            if p == 'header':
-                continue
+        # In case a provided header (may be optional) indicates that the API function call was unsuccessful, a
+        # different set of mandatory parameters applies.
+        header = data.get('header')
+        if header:
+            if data['header']['functionExecutionStatus']['status'] not in ['Executed-Success','Executed-WithWarning']:
+                output_mandatory = self.output_mandatory_failed
+
+        for p in output_mandatory:
             if not p in data:
                 raise ValueError('Mandatory output parameter "%s" missing' % p)
         for p, v in data.items():
@@ -231,35 +296,195 @@ class JsonHttpApiFunction(abc.ABC):
                 output[p] = p_class.decode(v)
         return output
 
+    def encode_server(self, data: dict) -> dict:
+        """Validate an encode input dict into JSON-serializable dict for response body."""
+        output = {}
+        output_mandatory = self.output_mandatory
+
+        # In case a provided header (may be optional) indicates that the API function call was unsuccessful, a
+        # different set of mandatory parameters applies.
+        header = data.get('header')
+        if header:
+            if data['header']['functionExecutionStatus']['status'] not in ['Executed-Success','Executed-WithWarning']:
+                output_mandatory = self.output_mandatory_failed
+
+        for p in output_mandatory:
+            if not p in data:
+                raise ValueError('Mandatory output parameter %s missing' % p)
+        for p, v in data.items():
+            p_class = self.output_params.get(p)
+            if not p_class:
+                logger.warning('Unexpected/unsupported output parameter %s=%s', p, v)
+                output[p] = v
+            else:
+                output[p] = p_class.encode(v)
+        return output
+
+    def decode_server(self, data: dict) -> dict:
+        """[further] Decode and validate the JSON-Dict of the request body."""
+        output = {}
+
+        for p in self.input_mandatory:
+            if not p in data:
+                raise ValueError('Mandatory input parameter "%s" missing' % p)
+        for p, v in data.items():
+            p_class = self.input_params.get(p)
+            if not p_class:
+                logger.warning('Unexpected/unsupported input parameter "%s"="%s"', p, v)
+                output[p] = v
+            else:
+                output[p] = p_class.decode(v)
+        return output
+
+    def rewrite_url(self, data: dict, url: str) -> Tuple[dict, str]:
+        """
+        Rewrite a static URL using information passed in the data dict. This method may be overloaded by a derived
+        class to allow fully dynamic URLs. The input parameters required for the URL rewriting may be passed using
+        data parameter. In case those parameters are additional parameters that are not intended to be passed to
+        the encode_client method later, they must be removed explcitly.
+
+        Args:
+                data: (see JsonHttpApiClient and JsonHttpApiServer)
+                url: statically generated URL string (see comment in JsonHttpApiClient)
+        """
+
+        # This implementation is a placeholder in which we do not perform any URL rewriting. We just pass through data
+        # and url unmodified.
+        return data, url
+
+class JsonHttpApiClient():
+    def __init__(self, api_func: JsonHttpApiFunction, url_prefix: str, func_req_id: Optional[str],
+                 session: requests.Session):
+        """
+        Args:
+                api_func : API function definition (JsonHttpApiFunction)
+                url_prefix : prefix to be put in front of the API function path (see JsonHttpApiFunction)
+                func_req_id : function requestor id to use for requests
+                session : session object (requests)
+        """
+        self.api_func = api_func
+        self.url_prefix = url_prefix
+        self.func_req_id = func_req_id
+        self.session = session
+
     def call(self, data: dict, func_call_id: Optional[str] = None, timeout=10) -> Optional[dict]:
-        """Make an API call to the HTTP API endpoint represented by this object.
-        Input data is passed in `data` as json-serializable dict.  Output data
-        is returned as json-deserialized dict."""
-        url = self.url_prefix + self.path
-        encoded = json.dumps(self.encode(data, func_call_id))
+        """
+        Make an API call to the HTTP API endpoint represented by this object. Input data is passed in `data` as
+        json-serializable fields. `data` may also contain additional parameters required for URL rewriting (see
+        rewrite_url in class JsonHttpApiFunction). Output data is returned as json-deserialized dict.
+
+        Args:
+                data: Input data required to perform the request.
+                func_call_id: Function Call Identifier, if present a header field is generated automatically.
+                timeout: Maximum amount of time to wait for the request to complete.
+        """
+
+        # In case a function caller ID is supplied, use it together with the stored function requestor ID to generate
+        # and prepend the header field according to SGP.22, section 6.5.1.1 and 6.5.1.3. (the presence of the header
+        # field is checked by the encode_client method)
+        if func_call_id:
+            data = {'header' : {'functionRequesterIdentifier': self.func_req_id,
+                                'functionCallIdentifier': func_call_id}} | data
+
+        # The URL used for the HTTP request (see below) normally consists of the initially given url_prefix
+        # concatenated with the path defined by the JsonHttpApiFunction definition. This static URL path may be
+        # rewritten by rewrite_url method defined in the JsonHttpApiFunction.
+        data, url = self.api_func.rewrite_url(data, self.url_prefix + self.api_func.path)
+
+        # Encode the message (the presence of mandatory fields is checked during encoding)
+        encoded = json.dumps(self.api_func.encode_client(data))
+
+        # Apply HTTP request headers according to SGP.22, section 6.5.1
         req_headers = {
             'Content-Type': 'application/json',
             'X-Admin-Protocol': 'gsma/rsp/v2.5.0',
         }
-        req_headers.update(self.extra_http_req_headers)
+        req_headers.update(self.api_func.extra_http_req_headers)
 
+        # Perform HTTP request
         logger.debug("HTTP REQ %s - hdr: %s '%s'" % (url, req_headers, encoded))
-        response = self.session.request(self.http_method, url, data=encoded, headers=req_headers, timeout=timeout)
+        response = self.session.request(self.api_func.http_method, url, data=encoded, headers=req_headers, timeout=timeout)
         logger.debug("HTTP RSP-STS: [%u] hdr: %s" % (response.status_code, response.headers))
         logger.debug("HTTP RSP: %s" % (response.content))
 
-        if response.status_code != self.expected_http_status:
+        # Check HTTP response status code and make sure that the returned HTTP headers look plausible (according to
+        # SGP.22, section 6.5.1)
+        if response.status_code != self.api_func.expected_http_status:
             raise HttpStatusError(response)
-        if not response.headers.get('Content-Type').startswith(req_headers['Content-Type']):
+        if response.content and not response.headers.get('Content-Type').startswith(req_headers['Content-Type']):
             raise HttpHeaderError(response)
         if not response.headers.get('X-Admin-Protocol', 'gsma/rsp/v2.unknown').startswith('gsma/rsp/v2.'):
             raise HttpHeaderError(response)
 
+        # Decode response and return the result back to the caller
         if response.content:
-            if response.headers.get('Content-Type').startswith('application/json'):
-                return self.decode(response.json())
-            elif response.headers.get('Content-Type').startswith('text/plain;charset=UTF-8'):
-                return { 'data': response.content.decode('utf-8') }
-            raise HttpHeaderError(f'unimplemented response Content-Type: {response.headers=!r}')
-
+            output = self.api_func.decode_client(response.json())
+            # In case the response contains a header, check it to make sure that the API call was executed successfully
+            # (the presence of the header field is checked by the decode_client method)
+            if 'header' in output:
+                if output['header']['functionExecutionStatus']['status'] not in ['Executed-Success','Executed-WithWarning']:
+                    raise ApiError(output['header']['functionExecutionStatus'])
+            return output
         return None
+
+class JsonHttpApiServer():
+    def __init__(self, api_func: JsonHttpApiFunction, call_handler = None):
+        """
+        Args:
+                api_func : API function definition (JsonHttpApiFunction)
+                call_handler : handler function to process the request. This function must accept the
+                               decoded request as a dictionary. The handler function must return a tuple consisting
+                               of the response in the form of a dictionary (may be empty), and a function execution
+                               status string ('Executed-Success', 'Executed-WithWarning', 'Failed' or 'Expired')
+        """
+        self.api_func = api_func
+        if call_handler:
+            self.call_handler = call_handler
+        else:
+            self.call_handler = self.default_handler
+
+    def default_handler(self, data: dict) -> (dict, str):
+        """default handler, used in case no call handler is provided."""
+        logger.error("no handler function for request: %s" % str(data))
+        return {}, 'Failed'
+
+    def call(self, request: Request) -> str:
+        """ Process an incoming request.
+        Args:
+                request : request object as received using twisted.web.server
+        Returns:
+                encoded JSON string (HTTP response code and headers are set by calling the appropriate methods on the
+                provided the request object)
+        """
+
+        # Make sure the request is done with the correct HTTP method
+        if (request.method.decode() != self.api_func.http_method):
+            raise ValueError('Wrong HTTP method %s!=%s' % (request.method.decode(), self.api_func.http_method))
+
+        # Decode the request
+        decoded_request = self.api_func.decode_server(json.loads(request.content.read()))
+
+        # Run call handler (see above)
+        data, fe_status = self.call_handler(decoded_request)
+
+        # In case a function execution status is returned, use it to generate and prepend the header field according to
+        # SGP.22, section 6.5.1.2 and 6.5.1.4 (the presence of the header filed is checked by the encode_server method)
+        if fe_status:
+            data = {'header' : {'functionExecutionStatus': {'status' : fe_status}}} | data
+
+        # Encode the message (the presence of mandatory fields is checked during encoding)
+        encoded = json.dumps(self.api_func.encode_server(data))
+
+        # Apply HTTP request headers according to SGP.22, section 6.5.1
+        res_headers = {
+            'Content-Type': 'application/json',
+            'X-Admin-Protocol': 'gsma/rsp/v2.5.0',
+        }
+        res_headers.update(self.api_func.extra_http_res_headers)
+        for header, value in res_headers.items():
+            request.setHeader(header, value)
+        request.setResponseCode(self.api_func.expected_http_status)
+
+        # Return the encoded result back to the caller for sending (using twisted/klein)
+        return encoded
+

pySim/esim/saip/__init__.py

@@ -1517,11 +1517,8 @@ class ProfileElementHeader(ProfileElement):
     def mandatory_service_add(self, service_name):
         self.decoded['eUICC-Mandatory-services'][service_name] = None
 
-    def mandatory_service_present(self, service_name):
-        return service_name in self.decoded['eUICC-Mandatory-services'].keys()
-
     def mandatory_service_remove(self, service_name):
-        if self.mandatory_service_present(service_name):
+        if service_name in self.decoded['eUICC-Mandatory-services'].keys():
             del self.decoded['eUICC-Mandatory-services'][service_name]
         else:
             raise ValueError("service not in eUICC-Mandatory-services list, cannot remove")
@@ -1737,11 +1734,11 @@ class ProfileElementSequence:
         #   - calculated in the USIM (EF.UST 125 = true),
         #   then eUICC-Mandatory-services needs 'get-identity'.
         # - 'get-identity' implies that the eUICC must support ONE OF profile-A OR profile-B.
-        #   So, when SUCI-CalcInfo for USIM in DF.SAIP contains both key types,
-        #   then no profile-A or B services need to be requested explicitly.
-        # - When the SUCI-CalcInfo for USIM (DF.SAIP) contains ONLY a key of profile-A ("identifier": 1),
+        #   (One might assume from this that, when SUCI-CalcInfo for USIM in DF.SAIP contains both key types, then no
+        #   profile-A or B services need to be requested explicitly. However, the correct logic is:)
+        # - Iff the SUCI-CalcInfo for USIM (DF.SAIP) contains a key of profile-A ("identifier": 1),
         #   then eUICC-Mandatory-services needs 'profile-a-x25519'.
-        # - Same: ONLY profile-B ("identifier": 2) needs 'profile-b-p256'.
+        # - Same: profile-B ("identifier": 2) needs 'profile-b-p256'.
         # - (When SUCI is calculated in the UE, then the eUICC does not need to provide any of these services.)
         suci_in_usim_enabled = False
         try:
@@ -1752,7 +1749,7 @@ class ProfileElementSequence:
             pass
         if suci_in_usim_enabled:
             svc_set.add('get-identity')
-            # now check for profile-a and profile-b
+            # now check for profile-a and profile-b presence
             suci_calcinfo_has_profile_a = False
             suci_calcinfo_has_profile_b = False
             try:
@@ -1768,25 +1765,18 @@ class ProfileElementSequence:
                         suci_calcinfo_has_profile_b = True
             except (KeyError, AttributeError):
                 pass
-            if suci_calcinfo_has_profile_a and suci_calcinfo_has_profile_b:
-                # 'get-identity' implies that the eUICC supports one of the above. Do not require a specific one.
-                pass
-            elif suci_calcinfo_has_profile_a:
-                # The profile has only a profile-A key, so require that
+            if suci_calcinfo_has_profile_a:
+                # The profile has a profile-A key, so require that
                 svc_set.add('profile-a-x25519')
-            elif suci_calcinfo_has_profile_b:
-                # The profile has only a profile-B key, so require that
+            if suci_calcinfo_has_profile_b:
+                # The profile has a profile-B key, so require that
                 svc_set.add('profile-b-p256')
 
         hdr_pe = self.get_pe_for_type('header')
         # patch in the 'manual' services from the existing list:
-        old_svc_set = set()
         for old_svc in hdr_pe.decoded['eUICC-Mandatory-services'].keys():
             if old_svc in manual_services:
-                old_svc_set.add(old_svc)
-        logger.debug(f"{svc_set=}  +  {old_svc_set=}")
-        svc_set = svc_set.union(old_svc_set)
-        logger.debug(f"{svc_set=}")
+                svc_set.add(old_svc)
         hdr_pe.decoded['eUICC-Mandatory-services'] = {x: None for x in svc_set}
 
     def rebuild_mandatory_gfstelist(self):

pySim/esim/saip/batch.py

@@ -20,19 +20,15 @@
 
 import copy
 import pprint
-import logging
-import traceback
-import inspect
-from typing import List, Generator
+from typing import Generator, Union
 from pySim.esim.saip.personalization import ConfigurableParameter
 from pySim.esim.saip import param_source
 from pySim.esim.saip import ProfileElementSequence, ProfileElementSD
 from pySim.global_platform import KeyUsageQualifier
 from osmocom.utils import b2h
 
-logger = logging.getLogger(__name__)
-def _func_():
-    return inspect.currentframe().f_back.f_code.co_name
+# a list of ConfigurableParameter classes and/or ConfigurableParameter class instances
+ParamList = list[Union[type[ConfigurableParameter], ConfigurableParameter]]
 
 class BatchPersonalization:
     """Produce a series of eSIM profiles from predefined parameters.
@@ -40,9 +36,9 @@ class BatchPersonalization:
 
     Usage example:
 
-    der_input = some_file.open('rb').read()
+    der_input = open('some_file', 'rb').read()
     pes = ProfileElementSequence.from_der(der_input)
-    p = pers.BatchPersonalization(
+    p = BatchPersonalization(
         n=10,
         src_pes=pes,
         csv_rows=get_csv_reader())
@@ -64,9 +60,12 @@ class BatchPersonalization:
     """
 
     class ParamAndSrc:
-        'tie a ConfigurableParameter to a source of actual values'
+        """tie a ConfigurableParameter to a source of actual values"""
         def __init__(self, param: ConfigurableParameter, src: param_source.ParamSource):
-            self.param = param
+            if isinstance(param, type):
+                self.param_cls = param
+            else:
+                self.param_cls = param.__class__
             self.src = src
 
     def __init__(self,
@@ -81,10 +80,10 @@ class BatchPersonalization:
                  copied.
         params: list of ParamAndSrc instances, defining a ConfigurableParameter and corresponding ParamSource to fill in
                 profile values.
-        csv_rows: A list or generator producing all CSV rows one at a time, starting with a row containing the column
-                  headers. This is compatible with the python csv.reader. Each row gets passed to
-                  ParamSource.get_next(), such that ParamSource implementations can access the row items.
-                  See param_source.CsvSource.
+        csv_rows: A generator (e.g. iter(list_of_rows)) producing all CSV rows one at a time, starting with a row
+                  containing the column headers. This is compatible with the python csv.reader. Each row gets passed to
+                  ParamSource.get_next(), such that ParamSource implementations can access the row items. See
+                  param_source.CsvSource.
         """
         self.n = n
         self.params = params or []
@@ -92,7 +91,7 @@ class BatchPersonalization:
         self.csv_rows = csv_rows
 
     def add_param_and_src(self, param:ConfigurableParameter, src:param_source.ParamSource):
-        self.params.append(BatchPersonalization.ParamAndSrc(param=param, src=src))
+        self.params.append(BatchPersonalization.ParamAndSrc(param, src))
 
     def generate_profiles(self):
         # get first row of CSV: column names
@@ -119,12 +118,10 @@ class BatchPersonalization:
                 try:
                     input_value = p.src.get_next(csv_row=csv_row)
                     assert input_value is not None
-                    value = p.param.__class__.validate_val(input_value)
-                    p.param.__class__.apply_val(pes, value)
+                    value = p.param_cls.validate_val(input_value)
+                    p.param_cls.apply_val(pes, value)
                 except Exception as e:
-                    print(traceback.format_exc())
-                    logger.error('during %s: %r', _func_(), e)
-                    raise ValueError(f'{p.param.name} fed by {p.src.name}: {e!r}') from e
+                    raise ValueError(f'{p.param_cls.get_name()} fed by {p.src.name}: {e}') from e
 
             pes.rebuild_mandatory_services()
 
@@ -139,14 +136,14 @@ class UppAudit(dict):
     """
 
     @classmethod
-    def from_der(cls, der: bytes, params: List, der_size=False, additional_sd_keys=False):
-        '''return a dict of parameter name and set of selected parameter values found in a DER encoded profile. Note:
+    def from_der(cls, der: bytes, params: ParamList, der_size=False, additional_sd_keys=False):
+        """return a dict of parameter name and set of selected parameter values found in a DER encoded profile. Note:
         some ConfigurableParameter implementations return more than one key-value pair, for example, Imsi returns
         both 'IMSI' and 'IMSI-ACC' parameters.
 
         e.g.
             UppAudit.from_der(my_der, [Imsi, ])
-            --> {'IMSI': '001010000000023', 'IMSI-ACC': '5'}
+            --> {'IMSI': {'001010000000023'}, 'IMSI-ACC': {'5'}}
 
         (where 'IMSI' == Imsi.name)
 
@@ -162,7 +159,7 @@ class UppAudit(dict):
         Scp80Kvn03. So we would not show kvn 0x04..0x0f in an audit. additional_sd_keys=True includes audits of all SD
         key KVN there may be in the UPP. This helps to spot SD keys that may already be present in a UPP template, with
         unexpected / unusual kvn.
-        '''
+        """
 
         # make an instance of this class
         upp_audit = cls()
@@ -191,11 +188,11 @@ class UppAudit(dict):
                 audit_key = f'SdKey_KVN{key.key_version_number:02x}_ID{key.key_identifier:02x}'
                 kuq_bin = KeyUsageQualifier.build(key.key_usage_qualifier).hex()
                 audit_val = f'{key.key_components=!r} key_usage_qualifier=0x{kuq_bin}={key.key_usage_qualifier!r}'
-                upp_audit[audit_key] = set((audit_val, ))
+                upp_audit.add_values({audit_key: audit_val})
 
         return upp_audit
 
-    def get_single_val(self, key, validate=True, allow_absent=False, absent_val=None):
+    def get_single_val(self, key, allow_absent=False, absent_val=None):
         """
         Return the audit's value for the given audit key (like 'IMSI' or 'IMSI-ACC').
         Any kind of value may occur multiple times in a profile. When all of these agree to the same unambiguous value,
@@ -235,7 +232,7 @@ class UppAudit(dict):
 
         v = try_single_val(v)
         if isinstance(v, bytes):
-            v = bytes_to_hexstr(v)
+            v = b2h(v)
         if v is None:
             return 'not present'
         return str(v)
@@ -245,21 +242,21 @@ class UppAudit(dict):
         return UppAudit.audit_val_to_str(self.get(key))
 
     def add_values(self, src:dict):
-        """self and src are both a dict of sets.
+        """Merge a plain dict of values into self, which is a dict of sets.
         For example from
-            self == { 'a': set((123,)) }
+            self == { 'a': {123} }
         and
-            src == { 'a': set((456,)), 'b': set((789,)) }
+            src == { 'a': 456, 'b': 789 }
         then after this function call:
-            self == { 'a': set((123, 456,)), 'b': set((789,)) }
+            self == { 'a': {123, 456}, 'b': {789} }
         """
         assert isinstance(src, dict)
-        for key, srcvalset in src.items():
+        for key, srcval in src.items():
             dstvalset = self.get(key)
             if dstvalset is None:
                 dstvalset = set()
                 self[key] = dstvalset
-            dstvalset.add(srcvalset)
+            dstvalset.add(srcval)
 
     def __str__(self):
         return '\n'.join(f'{key}: {self.get_val_str(key)}' for key in sorted(self.keys()))
@@ -284,7 +281,7 @@ class BatchAudit(list):
     BatchAudit itself is a list, callers may use the standard python list API to access the UppAudit instances.
     """
 
-    def __init__(self, params:List):
+    def __init__(self, params: ParamList):
         assert params
         self.params = params
 
@@ -327,15 +324,12 @@ class BatchAudit(list):
 
         return batch_audit
 
-    def to_csv_rows(self, headers=True, sort_key=None, column_blacklist=None):
-        '''generator that yields all audits' values as rows, useful feed to a csv.writer.'''
+    def to_csv_rows(self, headers=True, sort_key=None):
+        """generator that yields all audits' values as rows, useful feed to a csv.writer."""
         columns = set()
         for audit in self:
             columns.update(audit.keys())
 
-        if column_blacklist:
-            columns.difference_update(set(column_blacklist))
-
         columns = tuple(sorted(columns, key=sort_key))
 
         if headers:
@@ -344,9 +338,6 @@ class BatchAudit(list):
         for audit in self:
             yield (audit.get_single_val(col, allow_absent=True, absent_val="") for col in columns)
 
-def bytes_to_hexstr(b:bytes, sep=''):
-    return sep.join(f'{x:02x}' for x in b)
-
 def esim_profile_introspect(upp):
     pes = ProfileElementSequence.from_der(upp.read())
     d = {}
@@ -354,7 +345,7 @@ def esim_profile_introspect(upp):
 
     def show_bytes_as_hexdump(item):
         if isinstance(item, bytes):
-            return bytes_to_hexstr(item)
+            return b2h(item)
         if isinstance(item, list):
             return list(show_bytes_as_hexdump(i) for i in item)
         if isinstance(item, tuple):

pySim/esim/saip/param_source.py

@@ -37,13 +37,10 @@ class ParamSource:
     name = "none"
     numeric_base = None # or 10 or 16
 
-    @classmethod
-    def from_str(cls, s:str):
-        """Subclasses implement this:
-           if a parameter source defines some string input magic, override this function.
-           For example, a RandomDigitSource derives the number of digits from the string length,
-           so the user can enter '0000' to get a four digit random number."""
-        return cls(s)
+    def __init__(self, input_str:str):
+        """Subclasses should call super().__init__(input_str) before evaluating self.input_str. Each subclass __init__()
+        may in turn manipulate self.input_str to apply expansions or decodings."""
+        self.input_str = input_str
 
     def get_next(self, csv_row:dict=None):
         """Subclasses implement this: return the next value from the parameter source.
@@ -51,146 +48,143 @@ class ParamSource:
            This default implementation is an empty source."""
         raise ParamSourceExhaustedExn()
 
+    @classmethod
+    def from_str(cls, input_str:str):
+        """compatibility with earlier version of ParamSource. Just use the constructor."""
+        return cls(input_str)
 
 class ConstantSource(ParamSource):
     """one value for all"""
     name = "constant"
 
-    def __init__(self, val:str):
-        self.val = val
-
     def get_next(self, csv_row:dict=None):
-        return self.val
+        return self.input_str
 
 class InputExpandingParamSource(ParamSource):
 
+    def __init__(self, input_str:str):
+        super().__init__(input_str)
+        self.input_str = self.expand_input_str(self.input_str)
+
     @classmethod
-    def expand_str(cls, s:str):
+    def expand_input_str(cls, input_str:str):
         # user convenience syntax '0*32' becomes '00000000000000000000000000000000'
-        if "*" not in s:
-            return s
-        tokens = re.split(r"([^ \t]+)[ \t]*\*[ \t]*([0-9]+)", s)
+        if "*" not in input_str:
+            return input_str
+        # re:                "XX            *        123" with optional spaces
+        tokens = re.split(r"([^ \t]+)[ \t]*\*[ \t]*([0-9]+)", input_str)
         if len(tokens) < 3:
-            return s
+            return input_str
         parts = []
         for unchanged, snippet, repeat_str in zip(tokens[0::3], tokens[1::3], tokens[2::3]):
             parts.append(unchanged)
             repeat = int(repeat_str)
             parts.append(snippet * repeat)
-        return "".join(parts)
 
-    @classmethod
-    def from_str(cls, s:str):
-        return cls(cls.expand_str(s))
+        return "".join(parts)
 
 class DecimalRangeSource(InputExpandingParamSource):
     """abstract: decimal numbers with a value range"""
 
     numeric_base = 10
 
-    def __init__(self, num_digits, first_value, last_value):
-        """
-        See also from_str().
+    def __init__(self, input_str:str=None, num_digits:int=None, first_value:int=None, last_value:int=None):
+        """Constructor to set up values from a (user entered) string: DecimalRangeSource(input_str).
+        Constructor to set up values directly: DecimalRangeSource(num_digits=3, first_value=123, last_value=456)
 
-        All arguments are integer values, and are converted to int if necessary, so a string of an integer is fine.
-        num_digits: fixed number of digits (possibly with leading zeros) to generate.
-        first_value, last_value: the decimal range in which to provide digits.
+        num_digits produces leading zeros when first_value..last_value are shorter.
         """
-        num_digits = int(num_digits)
-        first_value = int(first_value)
-        last_value = int(last_value)
+        assert ((input_str is not None and (num_digits, first_value, last_value) == (None, None, None))
+                or (input_str is None and None not in (num_digits, first_value, last_value)))
+
+        if input_str is not None:
+            super().__init__(input_str)
+
+            input_str = self.input_str
+
+            if ".." in input_str:
+                first_str, last_str = input_str.split('..')
+                first_str = first_str.strip()
+                last_str = last_str.strip()
+            else:
+                first_str = input_str.strip()
+                last_str = None
+
+            num_digits = len(first_str)
+            first_value = int(first_str)
+            last_value = int(last_str if last_str is not None else "9" * num_digits)
+
         assert num_digits > 0
         assert first_value <= last_value
         self.num_digits = num_digits
-        self.val_first_last = (first_value, last_value)
+        self.first_value = first_value
+        self.last_value = last_value
 
     def val_to_digit(self, val:int):
         return "%0*d" % (self.num_digits, val)  # pylint: disable=consider-using-f-string
 
-    @classmethod
-    def from_str(cls, s:str):
-        s = cls.expand_str(s)
-
-        if ".." in s:
-            first_str, last_str = s.split('..')
-            first_str = first_str.strip()
-            last_str = last_str.strip()
-        else:
-            first_str = s.strip()
-            last_str = None
-
-        first_value = int(first_str)
-        last_value = int(last_str) if last_str is not None else "9" * len(first_str)
-        return cls(num_digits=len(first_str), first_value=first_value, last_value=last_value)
-
 class RandomSourceMixin:
     random_impl = secrets.SystemRandom()
 
 class RandomDigitSource(DecimalRangeSource, RandomSourceMixin):
     """return a different sequence of random decimal digits each"""
     name = "random decimal digits"
-    used_keys = set()
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.used_keys = set()
 
     def get_next(self, csv_row:dict=None):
-        # try to generate random digits that are always different from previously produced random bytes
-        attempts = 10
-        while True:
-            val = self.random_impl.randint(*self.val_first_last)
-            if val in RandomDigitSource.used_keys:
-                attempts -= 1
-                if attempts:
-                    continue
-            RandomDigitSource.used_keys.add(val)
-            break
+        # try to generate random digits that are always different from previously produced random digits
+        for _ in range(10):
+            val = self.random_impl.randint(self.first_value, self.last_value)
+            if val not in self.used_keys:
+                break
+        self.used_keys.add(val)
         return self.val_to_digit(val)
 
 class RandomHexDigitSource(InputExpandingParamSource, RandomSourceMixin):
     """return a different sequence of random hexadecimal digits each"""
     name = "random hexadecimal digits"
     numeric_base = 16
-    used_keys = set()
+    def __init__(self, input_str:str):
+        super().__init__(input_str)
+        input_str = self.input_str
 
-    def __init__(self, num_digits):
-        """see from_str()"""
-        num_digits = int(num_digits)
+        num_digits = len(input_str.strip())
         if num_digits < 1:
             raise ValueError("zero number of digits")
         # hex digits always come in two
         if (num_digits & 1) != 0:
             raise ValueError(f"hexadecimal value should have even number of digits, not {num_digits}")
         self.num_digits = num_digits
+        self.used_keys = set()
 
     def get_next(self, csv_row:dict=None):
         # try to generate random bytes that are always different from previously produced random bytes
-        attempts = 10
-        while True:
+        for _ in range(10):
             val = self.random_impl.randbytes(self.num_digits // 2)
-            if val in RandomHexDigitSource.used_keys:
-                attempts -= 1
-                if attempts:
-                    continue
-            RandomHexDigitSource.used_keys.add(val)
-            break
+            if val not in self.used_keys:
+                break
+        self.used_keys.add(val)
 
         return b2h(val)
 
-    @classmethod
-    def from_str(cls, s:str):
-        s = cls.expand_str(s)
-        return cls(num_digits=len(s.strip()))
-
 class IncDigitSource(DecimalRangeSource):
     """incrementing sequence of digits"""
     name = "incrementing decimal digits"
 
-    def __init__(self, num_digits, first_value, last_value):
-        super().__init__(num_digits, first_value, last_value)
+    def __init__(self, input_str:str=None, num_digits:int=None, first_value:int=None, last_value:int=None):
+        """input_str: the range of values to iterate. Format: 'FIRST..LAST' (e.g. '0001..9999') or
+        just 'FIRST' (iterates to the maximum value for the given digit width). Leading zeros in
+        FIRST determine the digit width and are preserved in returned values."""
+        super().__init__(input_str, num_digits, first_value, last_value)
         self.next_val = None
         self.reset()
 
     def reset(self):
         """Restart from the first value of the defined range passed to __init__()."""
-        self.next_val = self.val_first_last[0]
+        self.next_val = self.first_value
 
     def get_next(self, csv_row:dict=None):
         val = self.next_val
@@ -200,7 +194,7 @@ class IncDigitSource(DecimalRangeSource):
         returnval = self.val_to_digit(val)
 
         val += 1
-        if val > self.val_first_last[1]:
+        if val > self.last_value:
             self.next_val = None
         else:
             self.next_val = val
@@ -211,18 +205,17 @@ class CsvSource(ParamSource):
     """apply a column from a CSV row, as passed in to ParamSource.get_next(csv_row)"""
     name = "from CSV"
 
-    def __init__(self, csv_column):
-        """
-        csv_column: column name indicating the column to use for this parameter.
-                    This name is used in get_next(): the caller passes the current CSV row to get_next(), from which
-                    CsvSource picks the column with the name matching csv_column.
-        """
-        self.csv_column = csv_column
+    def __init__(self, input_str:str):
+        """input_str: the CSV column name to read values from.
+        The caller passes the current CSV row to get_next(), from which CsvSource picks the column matching
+        this name."""
+        super().__init__(input_str)
+        self.csv_column = self.input_str
 
     def get_next(self, csv_row:dict=None):
         val = None
         if csv_row:
             val = csv_row.get(self.csv_column)
-        if not val:
+        if val is None:
             raise ParamSourceUndefinedExn(f"no value for CSV column {self.csv_column!r}")
         return val

pySim/esim/saip/personalization.py

@@ -16,30 +16,22 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import abc
+import enum
 import io
-import os
 import re
-import pprint
-import json
 from typing import List, Tuple, Generator, Optional
 
 from construct.core import StreamError
 from osmocom.tlv import camel_to_snake
 from osmocom.utils import hexstr
 from pySim.utils import enc_iccid, dec_iccid, enc_imsi, dec_imsi, h2b, b2h, rpad, sanitize_iccid
-from pySim.ts_31_102 import EF_AD, EF_UST, EF_Routing_Indicator, EF_SUCI_Calc_Info, DF_USIM_5GS
+from pySim.ts_31_102 import EF_AD
 from pySim.ts_51_011 import EF_SMSP
 from pySim.esim.saip import param_source
 from pySim.esim.saip import ProfileElement, ProfileElementSD, ProfileElementSequence
-from pySim.esim.saip import ProfileElementHeader
 from pySim.esim.saip import SecurityDomainKey, SecurityDomainKeyComponent
 from pySim.global_platform import KeyUsageQualifier, KeyType
 
-# optimization: instantiate class instance to get the fid only once.
-file_path_df_5gs = bytes.fromhex(DF_USIM_5GS().fid)
-fid_ri = bytes.fromhex(EF_Routing_Indicator().fid)
-fid_sucici = bytes.fromhex(EF_SUCI_Calc_Info().fid)
-
 def unrpad(s: hexstr, c='f') -> hexstr:
     return hexstr(s.rstrip(c))
 
@@ -246,7 +238,7 @@ class ConfigurableParameter(abc.ABC, metaclass=ClassVarMeta):
             if val is None:
                 val = v
             elif val != v:
-                    raise ValueError(f'get_value_from_pes(): got distinct values: {val!r} != {v!r}')
+                raise ValueError(f'get_value_from_pes(): got distinct values: {val!r} != {v!r}')
         return val
 
     @classmethod
@@ -298,9 +290,7 @@ class ConfigurableParameter(abc.ABC, metaclass=ClassVarMeta):
            May be overridden by subclasses.
            This default implementation returns the maximum allowed value length -- a good fit for most subclasses.
            '''
-        l = cls.get_len_range()[1] or 16
-        l = min(10*80, l)
-        return l
+        return cls.get_len_range()[1] or 16
 
     @classmethod
     def is_super_of(cls, other_class):
@@ -339,6 +329,7 @@ class DecimalHexParam(DecimalParam):
     @classmethod
     def validate_val(cls, val):
         val = super().validate_val(val)
+        assert isinstance(val, str)
         val = ''.join('%02x' % ord(x) for x in val)
         if cls.rpad is not None:
             c = cls.rpad_char
@@ -348,7 +339,7 @@ class DecimalHexParam(DecimalParam):
 
     @classmethod
     def decimal_hex_to_str(cls, val):
-        'useful for get_values_from_pes() implementations of subclasses'
+        """useful for get_values_from_pes() implementations of subclasses"""
         if isinstance(val, bytes):
             val = b2h(val)
         assert isinstance(val, hexstr)
@@ -429,67 +420,69 @@ class BinaryParam(ConfigurableParameter):
 
 
 class EnumParam(ConfigurableParameter):
-    value_map = {
-            # For example:
-            #'Meaningful label for value 23': 0x23,
-            # Where 0x23 is a valid value to use for apply_val().
-        }
-    _value_map_reverse = None
+    """ConfigurableParameter for named integer enumeration values.
+
+    Subclasses must define a nested enum.IntEnum named 'Values' listing all valid names and their
+    integer codes.  apply_val() and get_values_from_pes() are not implemented here and this must
+    be inherited from another mixin."""
+
+    class Values(enum.IntEnum):
+        pass  # subclasses override this
 
     @classmethod
-    def validate_val(cls, val):
-        orig_val = val
-        enum_val = None
-        if isinstance(val, str):
-            enum_name = val
-            enum_val = cls.map_name_to_val(enum_name)
+    def validate_val(cls, val) -> int:
+        if isinstance(val, int):
+            try:
+                return int(cls.Values(val))
+            except ValueError:
+                pass
+        elif isinstance(val, str):
+            member = cls.map_name_to_val(val, strict=False)
+            if member is not None:
+                return member
 
-        # if the str is not one of the known value_map.keys(), is it maybe one of value_map.keys()?
-        if enum_val is None and val in cls.value_map.values():
-            enum_val = val
-
-        if enum_val not in cls.value_map.values():
-            raise ValueError(f"{cls.get_name()}: invalid argument: {orig_val!r}. Valid arguments are:"
-                             f" {', '.join(cls.value_map.keys())}")
-
-        return enum_val
+        valid = ', '.join(m.name for m in cls.Values)
+        raise ValueError(f"{cls.get_name()}: invalid argument: {val!r}. Valid arguments are: {valid}")
 
     @classmethod
-    def map_name_to_val(cls, name:str, strict=True):
-        val = cls.value_map.get(name)
-        if val is not None:
-            return val
+    def map_name_to_val(cls, name: str, strict=True) -> int:
+        """Return the integer value for a given enum member name. Performs an exact match first,
+        then falls back to fuzzy matching (case-insensitive, punctuation-insensitive)."""
+        try:
+            return int(cls.Values[name])
+        except KeyError:
+            pass
 
-        clean_name = cls.clean_name_str(name)
-        for k, v in cls.value_map.items():
-            if clean_name == cls.clean_name_str(k):
-                return v
+        clean = cls.clean_name_str(name)
+        for member in cls.Values:
+            if cls.clean_name_str(member.name) == clean:
+                return int(member)
 
         if strict:
-            raise ValueError(f"Problem in {cls.get_name()}: {name!r} is not a known value."
-                    f" Known values are: {cls.value_map.keys()!r}")
+            valid = ', '.join(m.name for m in cls.Values)
+            raise ValueError(f"{cls.get_name()}: {name!r} is not a known value. Known values are: {valid}")
         return None
 
     @classmethod
     def map_val_to_name(cls, val, strict=False) -> str:
-        if cls._value_map_reverse is None:
-            cls._value_map_reverse = dict((v, k) for k, v in cls.value_map.items())
-
-        name = cls._value_map_reverse.get(val)
-        if name:
-            return name
-        if strict:
-            raise ValueError(f"Problem in {cls.get_name()}: {val!r} ({type(val)}) is not a known value."
-                    f" Known values are: {cls.value_map.values()!r}")
-        return None
+        """Return the enum member name for a given integer value."""
+        try:
+            return cls.Values(val).name
+        except ValueError:
+            if strict:
+                raise ValueError(f"{cls.get_name()}: {val!r} ({type(val).__name__}) is not a known value.")
+            return None
 
     @classmethod
-    def name_normalize(cls, name:str) -> str:
-        return cls.map_val_to_name(cls.map_name_to_val(name))
+    def name_normalize(cls, name: str) -> str:
+        """Map a (possibly fuzzy) name to its canonical enum member name."""
+        return cls.Values(cls.map_name_to_val(name)).name
 
     @classmethod
-    def clean_name_str(cls, val):
-        return re.sub('[^0-9A-Za-z-_]', '', val).lower()
+    def clean_name_str(cls, val: str) -> str:
+        """Strip punctuation and case for fuzzy name comparison.
+        Treats hyphens and underscores as equivalent (both removed)."""
+        return re.sub('[^0-9A-Za-z]', '', val).lower()
 
 
 class Iccid(DecimalParam):
@@ -642,28 +635,21 @@ class SmspTpScAddr(ConfigurableParameter):
             # - To generate the right amount of fillFileContent, pass total_len=42 to encode_record_bin().
             # - To show the right size in the PES, set f_smsp.rec_len = 42
             ef_smsp_dec['alpha_id'] = ''
-
-            # we can set this to choose a fixed length:
-            #f_smsp.rec_len = 42
-            # but leave rec_len unchanged to keep the same length as was found in the eSIM template.
+            f_smsp.rec_len = 42
 
             # re-encode into the File body.
-            f_smsp.body = ef_smsp.encode_record_bin(ef_smsp_dec, 1, total_len=f_smsp.rec_len)
+            #
+            #print("SMSP  (new): %s" % f_smsp.body)
             # re-generate the pe.decoded member from the File instance
+            f_smsp.body = ef_smsp.encode_record_bin(ef_smsp_dec, 1, total_len=f_smsp.rec_len)
             pe.file2pe(f_smsp)
 
     @classmethod
     def get_values_from_pes(cls, pes: ProfileElementSequence):
         for pe in pes.get_pes_for_type('usim'):
-            f_smsp = pe.files.get('ef-smsp', None)
-            if f_smsp is None:
-                continue
-
-            try:
-                ef_smsp = EF_SMSP()
-                ef_smsp_dec = ef_smsp.decode_record_bin(f_smsp.body, 1)
-            except IndexError:
-                continue
+            f_smsp = pe.files['ef-smsp']
+            ef_smsp = EF_SMSP()
+            ef_smsp_dec = ef_smsp.decode_record_bin(f_smsp.body, 1)
 
             tp_sc_addr = ef_smsp_dec.get('tp_sc_addr', None)
 
@@ -677,57 +663,69 @@ class SmspTpScAddr(ConfigurableParameter):
 
 
 class MncLen(EnumParam):
-    """MNC length. Must be either 2 or 3. Sets only the MNC length field in EF-AD (Administrative Data)."""
+    """MNC length.  Sets only the MNC length field in EF.AD (Administrative Data).
+    Accepted values: integer 2 or 3, digit strings '2' or '3', or enum names 'MNC2'/'MNC3'.
+    """
     name = 'MNC-LEN'
-    value_map = { '2': 2, '3': 3 }
-    default_source = param_source.ConstantSource
     example_input = '2'
+    default_source = param_source.ConstantSource
+
+    class Values(enum.IntEnum):
+        MNC2 = 2
+        MNC3 = 3
 
     @classmethod
-    def apply_val(cls, pes: ProfileElementSequence, val):
-        """val must be an int: either 2 or 3"""
+    def validate_val(cls, val):
+        if isinstance(val, str) and val.isdigit():
+            val = int(val)
+        return super().validate_val(val)
+
+    @classmethod
+    def _get_f_ad(cls, pe: ProfileElement):
+        if not hasattr(pe, 'files'):
+            return None
+        f_ad = pe.files.get('ef-ad', None)
+        if f_ad and f_ad.body:
+            return f_ad
+        return None
+
+    @classmethod
+    def _decode_f_ad(cls, f_ad):
+        try:
+            ef_ad_dec = EF_AD().decode_bin(f_ad.body)
+        except StreamError:
+            return None
+        if 'mnc_len' not in ef_ad_dec:
+            return None
+        return ef_ad_dec
+
+    @classmethod
+    def apply_val(cls, pes: ProfileElementSequence, val: int):
         for pe in pes.get_pes_for_type('usim'):
-            if not hasattr(pe, 'files'):
-                continue
-            f_ad = pe.files.get('ef-ad')
-            if not f_ad:
+            f_ad = cls._get_f_ad(pe)
+            if f_ad is None:
                 continue
             # decode existing values
-            if not f_ad.body:
-                continue
-            try:
-                ef_ad = EF_AD()
-                ef_ad_dec = ef_ad.decode_bin(f_ad.body)
-            except StreamError:
-                continue
-            if 'mnc_len' not in ef_ad_dec:
+            ef_ad_dec = cls._decode_f_ad(f_ad)
+            if ef_ad_dec is None:
                 continue
             # change mnc_len
             ef_ad_dec['mnc_len'] = val
             # re-encode into the File body
-            f_ad.body = ef_ad.encode_bin(ef_ad_dec)
+            f_ad.body = EF_AD().encode_bin(ef_ad_dec)
             pe.file2pe(f_ad)
 
     @classmethod
     def get_values_from_pes(cls, pes: ProfileElementSequence):
         for pe in pes.get_pes_for_type('usim'):
-            if not hasattr(pe, 'files'):
-                continue
-            f_ad = pe.files.get('ef-ad', None)
+            f_ad = cls._get_f_ad(pe)
             if f_ad is None:
                 continue
-
-            try:
-                ef_ad = EF_AD()
-                ef_ad_dec = ef_ad.decode_bin(f_ad.body)
-            except StreamError:
+            ef_ad_dec = cls._decode_f_ad(f_ad)
+            if ef_ad_dec is None:
                 continue
-
-            mnc_len = ef_ad_dec.get('mnc_len', None)
-            if mnc_len is None:
-                continue
-
-            yield { cls.name: cls.map_val_to_name(int(mnc_len)) }
+            mnc_len = ef_ad_dec.get('mnc_len')
+            yield { cls.name: str(mnc_len) }
 
 
 class SdKey(BinaryParam):
@@ -736,7 +734,6 @@ class SdKey(BinaryParam):
     # these will be set by subclasses
     key_type = None
     kvn = None
-    reserved_kvn = tuple() # tuple of all reserved kvn for a given SCPxx
     key_id = None
     key_usage_qual = None
 
@@ -782,8 +779,6 @@ class SdKey(BinaryParam):
                 yield { cls.name: b2h(kc) }
 
 
-    NO_OP = (('', {}))
-
     LEN_128 = (16,)
     LEN_128_192_256 = (16, 24, 32)
     LEN_128_256 = (16, 32)
@@ -1023,7 +1018,7 @@ class Pin(DecimalHexParam):
 
             for pinCode in pinCodes.decoded['pinCodes'][1]:
                 if pinCode['keyReference'] == cls.keyReference:
-                     yield { cls.name: cls.decimal_hex_to_str(pinCode['pinValue']) }
+                    yield { cls.name: cls.decimal_hex_to_str(pinCode['pinValue']) }
 
     @classmethod
     def get_values_from_pes(cls, pes: ProfileElementSequence):
@@ -1101,22 +1096,20 @@ class AlgoConfig(ConfigurableParameter):
             yield { cls.name: val }
 
 class AlgorithmID(EnumParam, AlgoConfig):
-    '''use validate_val() from EnumParam, and apply_val() from AlgoConfig.
-    In get_values_from_pes(), return enum value names, not raw values.'''
+    """use validate_val() from EnumParam, and apply_val() from AlgoConfig.
+    In get_values_from_pes(), return enum value names, not raw values."""
     name = "Algorithm"
-
-    # as in pySim/esim/asn1/saip/PE_Definitions-3.3.1.asn
-    value_map = {
-            "Milenage" : 1,
-            "TUAK" : 2,
-            "usim-test" : 3,
-        }
+    algo_config_key = 'algorithmID'
     example_input = "Milenage"
     default_source = param_source.ConstantSource
 
-    algo_config_key = 'algorithmID'
+    # as in pySim/esim/asn1/saip/PE_Definitions-3.3.1.asn
+    class Values(enum.IntEnum):
+        Milenage  = 1
+        TUAK      = 2
+        usim_test = 3  # input 'usim-test' also accepted via fuzzy matching
 
-    # EnumParam.validate_val() returns the int values from value_map
+    # EnumParam.validate_val() returns the int values from Values
 
     @classmethod
     def get_values_from_pes(cls, pes: ProfileElementSequence):
@@ -1191,411 +1184,3 @@ class TuakNumberOfKeccak(IntegerParam, AlgoConfig):
     max_val = 255
     example_input = '1'
     default_source = param_source.ConstantSource
-    numeric_base = None  # indicate that this won't need random number sources
-
-
-class EfUstServiceParam(EnumParam):
-    """superclass for EF-UST service flag parameters"""
-    service_idx = 0
-    value_map = { 'enabled': True, 'disabled': False }
-    default_source = param_source.ConstantSource
-    example_input = sorted(value_map.keys())[0]
-
-    @classmethod
-    def apply_val(cls, pes: ProfileElementSequence, val):
-        for pe in pes.get_pes_for_type('usim'):
-            f_ust = pe.files['ef-ust']
-            ef_ust = EF_UST()
-            ust = ef_ust.decode_bin(f_ust.body)
-
-            ust[cls.service_idx]['activated'] = val
-
-            f_ust.body = ef_ust.encode_bin(ust)
-            pe.file2pe(f_ust)
-
-    @classmethod
-    def get_values_from_pes(cls, pes: ProfileElementSequence):
-        for pe in pes.get_pes_for_type('usim'):
-            f_ust = pe.files.get('ef-ust', None)
-            if not f_ust:
-                continue
-            ef_ust = EF_UST()
-            try:
-                ust = ef_ust.decode_bin(f_ust.body)
-
-                service_flag = ust[cls.service_idx]['activated']
-                yield { cls.name: cls.map_val_to_name(service_flag) }
-            except:
-                pass
-
-class SuciActive(EfUstServiceParam):
-    """EF-UST service nr 124: enable or disable the SUCI service."""
-    service_idx = 124
-    name = '5G-SUCI-active'
-    value_map = { 'SUCI-off': False, 'SUCI-on': True }
-    example_input = 'SUCI-on'
-
-class SuciInUsim(EfUstServiceParam):
-    """EF-UST service nr 125: calculate SUCI in UE or in USIM"""
-    service_idx = 125
-    name = '5G-SUCI-in-USIM'
-    value_map = { 'SUCI-in-UE': False, 'SUCI-in-USIM': True }
-    example_input = 'SUCI-in-USIM'
-
-class SuciRi(ConfigurableParameter):
-    """SUCI Routing Indicator as in section 4.4.11.11 of 3GPP TS 31.102"""
-    name = '5G-SUCI-RI'
-    allow_chars = '0123456789'
-    min_len = 1
-    max_len = 4
-    allow_types = (str,)
-    example_input = '0'
-    default_source = param_source.ConstantSource
-
-    KEY_RI = "routing_indicator"
-
-    @classmethod
-    def apply_val(cls, pes: ProfileElementSequence, val):
-        for pe in pes.get_pes_for_type('df-5gs'):
-            f_ri = pe.files.get('ef-routing-indicator', None)
-            if f_ri is None:
-                continue
-            ef_ri = EF_Routing_Indicator()
-            ri = ef_ri.decode_bin(f_ri.body)
-
-            ri[cls.KEY_RI] = str(val)
-
-            f_ri.body = ef_ri.encode_bin(ri)
-            pe.file2pe(f_ri)
-
-    @classmethod
-    def get_values_from_pes(cls, pes: ProfileElementSequence):
-        for pe in pes.get_pes_for_type('df-5gs'):
-            f_ri = pe.files.get('ef-routing-indicator', None)
-            if f_ri is None:
-                continue
-            ef_ri = EF_Routing_Indicator()
-            try:
-                ri = ef_ri.decode_bin(f_ri.body)
-                yield { cls.name: ri.get(cls.KEY_RI) }
-            except:
-                pass
-
-class SuciCalcInfoParameter(ConfigurableParameter):
-    """SUCI Calculation Information as in section 4.4.11.8 of 3GPP TS 31.102"""
-    name = '5G-SUCI-CalcInfo'
-    default_source = param_source.ConstantSource
-    allow_types = (str,)
-    max_len = 4096  # to indicate a large input field to UI renderers
-    example_input = '{"prot_scheme_id_list": [{"priority": 0, "identifier": 0, "key_index": 0}], "hnet_pubkey_list": []}'
-
-    PE_IN_UE = ("df-5gs", "ef-suci-calc-info")
-    PE_IN_USIM = ("df-saip", "ef-suci-calc-info-usim")
-    suci_calc_info_pe = None
-
-    @classmethod
-    def validate_val(cls, val):
-        val = super().validate_val(val)
-
-        if not val:
-            val = "{}"
-
-        # check that it is a dict something like
-        # {
-        #     "prot_scheme_id_list": [
-        #         {"priority": 0, "identifier": 2, "key_index": 1},
-        #         {"priority": 1, "identifier": 1, "key_index": 2},
-        #         ],
-        #     "hnet_pubkey_list": [
-        #         {"hnet_pubkey_identifier": 27,
-        #          "hnet_pubkey": "0472DA71976234CE833A6907425867B82E074D44EF907DFB4B3E21C1C2256EBCD15A7DED52FCBB097A4ED250E036C7B9C8C7004C4EEDC4F068CD7BF8D3F900E3B4"},
-        #         {"hnet_pubkey_identifier": 30,
-        #          "hnet_pubkey": "5A8D38864820197C3394B92613B20B91633CBD897119273BF8E4A6F4EEC0A650"},
-        #          ],
-        # }
-
-        try:
-            d = json.loads(val)
-        except json.decoder.JSONDecodeError as e:
-            raise ValueError(f"Cannot parse SUCI Calc Info: {e}") from e
-
-        KEY_PSI_LIST = 'prot_scheme_id_list'
-        KEY_HPK_LIST = 'hnet_pubkey_list'
-        KEYS_D = set((KEY_HPK_LIST, KEY_PSI_LIST))
-        KEYS_PSI = set(('identifier', 'key_index', 'priority'))
-        KEYS_HPK = set(('hnet_pubkey_identifier', 'hnet_pubkey'))
-
-        if not d:
-            d = { KEY_PSI_LIST: [], KEY_HPK_LIST: [] }
-
-        if not (isinstance(d, dict)
-                and set(d.keys()) == KEYS_D):
-            raise ValueError(f"Unexpected structure in SUCI Calc Info: expected dict with entries {KEYS_D}")
-
-        psi = d.get(KEY_PSI_LIST, None)
-        if not all((set(e.keys()) == KEYS_PSI) for e in psi):
-            raise ValueError("Unexpected structure in SUCI Calc Info:"
-                             f" in {KEY_PSI_LIST}, expected dict with entries {KEYS_PSI}")
-
-        hpk = d.get(KEY_HPK_LIST, None)
-        if not all((set(e.keys()) == KEYS_HPK) for e in hpk):
-            raise ValueError("Unexpected structure in SUCI Calc Info:"
-                             f" in {KEY_HPK_LIST}, expected dict with entries {KEYS_HPK}")
-        return d
-
-    @classmethod
-    def _apply_suci(cls, pes: ProfileElementSequence, val, pe_type="df-5gs", pe_file="ef-suci-calc-info"):
-        for pe in pes.get_pes_for_type(pe_type):
-            f_sucici = pe.files.get(pe_file, None)
-            if not f_sucici:
-                continue
-            ef_sucici = EF_SUCI_Calc_Info()
-            body = ef_sucici.encode_bin(val)
-
-            # 0xff pad up to the existing file size, so that the underlying template doesn't come through
-            is_size = f_sucici.file_size
-            pad_n = is_size - len(body)
-            if pad_n > 0:
-                body = body + b'\xff' * pad_n
-
-            f_sucici.body = body
-            pe.file2pe(f_sucici)
-
-    @classmethod
-    def apply_val(cls, pes: ProfileElementSequence, val):
-        cls._apply_suci(pes, val, *cls.suci_calc_info_pe)
-
-    @staticmethod
-    def normalize_sucici(sucici:dict):
-        """Normalize the CalcInfo dict so it can be json encoded:
-           convert bytes to hex strings."""
-        if not sucici:
-            sucici = {}
-
-        for hnet_pubkey in sucici.get('hnet_pubkey_list', ()):
-            val = hnet_pubkey['hnet_pubkey']
-            if isinstance(val, bytes):
-                val = b2h(val)
-                hnet_pubkey['hnet_pubkey'] = val
-
-        return sucici
-
-    @classmethod
-    def _get_suci(cls, pes: ProfileElementSequence, pe_type="df-5gs", pe_file="ef-suci-calc-info"):
-        for pe in pes.get_pes_for_type(pe_type):
-            f_sucici = pe.files.get(pe_file, None)
-            if not f_sucici:
-                continue
-            ef_sucici = EF_SUCI_Calc_Info()
-            sucici = ef_sucici.decode_bin(f_sucici.body)
-
-            # normalize to string (bytes cannot go into json)
-            sucici = cls.normalize_sucici(sucici)
-
-            yield { cls.name: json.dumps(sucici) }
-
-    @classmethod
-    def get_values_from_pes(cls, pes: ProfileElementSequence):
-        yield from cls._get_suci(pes, *cls.suci_calc_info_pe)
-
-class SuciCalcInfoUe(SuciCalcInfoParameter):
-    """SUCI Calculation Information as in section 4.4.11.8 of 3GPP TS 31.102, readable by UE (DF-5GS)"""
-    name = '5G-SUCI-CalcInfo-UE'
-    suci_calc_info_pe = SuciCalcInfoParameter.PE_IN_UE
-
-class SuciCalcInfoUsim(SuciCalcInfoParameter):
-    """SUCI Calculation Information as in section 4.4.11.8 of 3GPP TS 31.102, readable only by USIM (DF-SAIP)"""
-    name = '5G-SUCI-CalcInfo-USIM'
-    suci_calc_info_pe = SuciCalcInfoParameter.PE_IN_USIM
-
-def gfm_find(pes: ProfileElementSequence, file_path:bytes, ef_fid:bytes):
-    """look through genericFileManagement PE and return the fmc list with start and end indexes as
-       (fmc_list, first_idx, after_last_idx)
-       so that fmc_list[first_idx:after_last_idx] is the slice of file management commands relevant to the given
-       file_path/ef_fid.
-    """
-    for pe in pes.get_pes_for_type('genericFileManagement'):
-        path_match = False
-        creating_fid = False
-
-        for fmc in pe.decoded['fileManagementCMD']:
-            first = None
-            last = None
-
-            for idx in range(len(fmc)):
-                cmd, arg = fmc[idx]
-
-                if cmd == 'filePath':
-                    path_match = (arg == file_path)
-                    if not path_match:
-                        creating_fid = False
-                elif path_match and cmd == 'createFCP':
-                    creating_fid = (arg.get('fileID') == ef_fid)
-                if creating_fid:
-                    if first is None:
-                        first = idx
-                        last = idx
-                    first = min(first, idx)
-                    last = max(last, idx)
-
-            if first is not None:
-                yield fmc, first, last + 1
-
-# genericFileManagement 5G params
-
-def pes_get_adf_fid(pes:ProfileElementSequence, naa_name="usim", adf_name="adf-usim"):
-    adf = pes.get_pe_for_type(naa_name)
-    return adf.decoded[adf_name][0][1]['fileID']
-
-def mk_adf_df_path(pes, naa:str, adf:str, file_path:bytes) -> bytes:
-    adf_file_id = pes_get_adf_fid(pes, naa, adf)
-    return b''.join((adf_file_id, file_path))
-
-def gfm_get_file_content(pes: ProfileElementSequence, naa:str, adf:str, file_path:bytes, ef_fid:bytes) -> bytes:
-    '''find a given file in the genericFileManagement section, and return the bytes from the first fillFileContent
-       item.
-       TODO: implement File.from_gfm() and return the full resulting bytes?
-    '''
-    adf_df_path = mk_adf_df_path(pes, naa, adf, file_path)
-
-    data = []
-    for fmc, first_idx, after_last_idx in gfm_find(pes, adf_df_path, ef_fid):
-        assert fmc[first_idx][0] == 'createFCP'
-        assert after_last_idx > first_idx
-
-        idx = first_idx + 1
-        while idx < after_last_idx:
-            if fmc[idx][0] == 'fillFileContent':
-                data.append(fmc[idx][1])
-            idx += 1
-
-    return data
-
-def gfm_set_file_content(pes: ProfileElementSequence, naa:str, adf:str, file_path:bytes, ef_fid:bytes, file_content:bytes) -> int:
-    adf_df_path = mk_adf_df_path(pes, naa, adf, file_path)
-
-    found = 0
-    for fmc, first_idx, after_last_idx in gfm_find(pes, adf_df_path, ef_fid):
-        assert fmc[first_idx][0] == 'createFCP'
-        assert after_last_idx > first_idx
-
-        new_fmc = [
-                fmc[first_idx],
-                ('fillFileContent', file_content),
-            ]
-        new_fmc[0][1]['efFileSize'] = bytes((len(file_content), ))
-
-        fmc[first_idx:after_last_idx] = new_fmc
-
-        found += 1
-    return found
-
-class GfmSuciRi(SuciRi):
-    """SUCI Routing Indicator as in section 4.4.11.11 of 3GPP TS 31.102,
-       applied via General File Management. Intended for SAIP 2.1 profiles."""
-    name = 'GFM-5G-SUCI-RI'
-
-    @classmethod
-    def apply_val(cls, pes: ProfileElementSequence, val):
-        ri = {
-            "routing_indicator": str(val),
-            "rfu": "ffff"
-        }
-        ef_ri = EF_Routing_Indicator()
-        found = gfm_set_file_content(pes, 'usim', 'adf-usim', file_path_df_5gs, fid_ri,
-                                     ef_ri.encode_bin(ri))
-        if not found:
-            raise ValueError(f"No target file found, Cannot apply {cls.name} = {ri}")
-
-        data = gfm_get_file_content(pes, 'usim', 'adf-usim', file_path_df_5gs, fid_ri)
-        val = ef_ri.decode_bin(b''.join(data))
-
-    @classmethod
-    def get_values_from_pes(cls, pes: ProfileElementSequence):
-        data = gfm_get_file_content(pes, 'usim', 'adf-usim', file_path_df_5gs, fid_ri)
-        if not data:
-            return
-
-        data = b''.join(data)
-        if not data:
-            return
-
-        ef_ri = EF_Routing_Indicator()
-        ri = ef_ri.decode_bin(data)
-        yield { cls.name: ri.get(cls.KEY_RI) }
-
-class GfmSuciCalcInfoUe(SuciCalcInfoUe):
-    """SUCI Calculation Information as in section 4.4.11.8 of 3GPP TS 31.102, readable by UE (DF-5GS),
-       applied via General File Management. Intended for SAIP 2.1 profiles."""
-    name = 'GFM-5G-SUCI-CalcInfo-UE'
-
-    @classmethod
-    def apply_val(cls, pes: ProfileElementSequence, val):
-        if not isinstance(val, dict):
-            raise ValueError("val should be a dict, after 'val = SuciCalcInfoParameter.validate_val(val)'")
-
-        ef_sucici = EF_SUCI_Calc_Info()
-        body = ef_sucici.encode_bin(val)
-        gfm_set_file_content(pes, 'usim', 'adf-usim', file_path_df_5gs, fid_sucici,
-                             body)
-
-    @classmethod
-    def get_values_from_pes(cls, pes: ProfileElementSequence):
-        data = gfm_get_file_content(pes, 'usim', 'adf-usim', file_path_df_5gs, fid_sucici)
-        if not data:
-            return
-
-        data = b''.join(data)
-        if not data:
-            return
-
-        ef_sucici = EF_SUCI_Calc_Info()
-        sucici = ef_sucici.decode_bin(data)
-        sucici = cls.normalize_sucici(sucici)
-        yield { cls.name: json.dumps(sucici) }
-
-
-class EuiccMandatoryServiceParam(EnumParam):
-    """superclass for managing items of the ProfileHeader / eUICC-Mandatory-services ServicesList"""
-    service_name = None
-    value_map = { 'mandatory': True, 'optional': False }
-    default_source = param_source.ConstantSource
-    example_input = sorted(value_map.keys())[0]
-
-    @classmethod
-    def apply_val(cls, pes: ProfileElementSequence, val):
-        for pe in pes.get_pes_for_type('header'):
-            assert isinstance(pe, ProfileElementHeader)
-            if val:
-                pe.mandatory_service_add(cls.service_name)
-            else:
-                # explicitly check to avoid exception when then service is already not present
-                if pe.mandatory_service_present(cls.service_name):
-                    pe.mandatory_service_remove(cls.service_name)
-
-    @classmethod
-    def get_values_from_pes(cls, pes: ProfileElementSequence):
-        for pe in pes.get_pes_for_type('header'):
-            assert isinstance(pe, ProfileElementHeader)
-            val = bool(pe.mandatory_service_present(cls.service_name))
-            yield { cls.name: cls.map_val_to_name(val) }
-
-class EuiccMandatoryServiceGetIdentity(EuiccMandatoryServiceParam):
-    """eUICC Mandatory Services: get-identity. The eUICC must be capable of providing a 5G identity using SUCI-CalcInfo
-    located in the USIM's DF-SAIP, see parameter 5G-SUCI-CalcInfo-USIM."""
-    name = '5G-eUICC-get-identity'
-    service_name = 'get-identity'
-
-class EuiccMandatoryServiceProfileA(EuiccMandatoryServiceParam):
-    """eUICC Mandatory Services: profile-a-x25519. The eUICC must be able to estblish a 5G identity using an X25519 key,
-    as provided in a profile-A ("identifier": 1) key in SUCI-CalcInfo located in the USIM's DF-SAIP, see parameter
-    5G-SUCI-CalcInfo-USIM."""
-    name = '5G-eUICC-profile-a-x25519'
-    service_name = 'profile-a-x25519'
-
-class EuiccMandatoryServiceProfileB(EuiccMandatoryServiceParam):
-    """eUICC Mandatory Services: profile-b-p256. The eUICC must be able to estblish a 5G identity using a P256 key, as
-    provided in a profile-B ("identifier": 2) key in SUCI-CalcInfo located in the USIM's DF-SAIP, see parameter
-    5G-SUCI-CalcInfo-USIM."""
-    name = '5G-eUICC-profile-b-p256'
-    service_name = 'profile-b-p256'

pySim/euicc.py

@@ -226,9 +226,28 @@ class Icon(BER_TLV_IE, tag=0x94):
     _construct = GreedyBytes
 class ProfileClass(BER_TLV_IE, tag=0x95):
     _construct = Enum(Int8ub, test=0, provisioning=1, operational=2)
+class ProfilePolicyRules(BER_TLV_IE, tag=0x99):
+    _construct = GreedyBytes
+class NotificationConfigurationInfo(BER_TLV_IE, tag=0xb6):
+    _construct = GreedyBytes
+
+# ProfileOwner
+class ProfileOwnerPLMN(BER_TLV_IE, tag=0x80):
+    _construct = PlmnAdapter(Bytes(3))
+class ProfileOwnerGID1(BER_TLV_IE, tag=0x81):
+    _construct = GreedyBytes
+class ProfileOwnerGID2(BER_TLV_IE, tag=0x82):
+    _construct = GreedyBytes
+class ProfileOwner(BER_TLV_IE, tag=0xb7, nested=[ProfileOwnerPLMN, ProfileOwnerGID1, ProfileOwnerGID2]):
+    _construct = GreedyBytes
+
+class SMDPPProprietaryData(BER_TLV_IE, tag=0xb8):
+    _construct = GreedyBytes
+
 class ProfileInfo(BER_TLV_IE, tag=0xe3, nested=[Iccid, IsdpAid, ProfileState, ProfileNickname,
                                                 ServiceProviderName, ProfileName, IconType, Icon,
-                                                ProfileClass]): # FIXME: more IEs
+                                                ProfileClass, ProfilePolicyRules, NotificationConfigurationInfo,
+                                                ProfileOwner, SMDPPProprietaryData]):
     pass
 class ProfileInfoSeq(BER_TLV_IE, tag=0xa0, nested=[ProfileInfo]):
     pass
@@ -444,9 +463,28 @@ class CardApplicationISDR(pySim.global_platform.CardApplicationSD):
             d = rn.to_dict()
             self._cmd.poutput_json(flatten_dict_lists(d['notification_sent_resp']))
 
-        def do_get_profiles_info(self, _opts):
+        get_profiles_info_parser = argparse.ArgumentParser()
+        get_profiles_info_parser.add_argument('--all', action='store_true', help='Retrieve all known tags of a profile')
+
+        @cmd2.with_argparser(get_profiles_info_parser)
+        def do_get_profiles_info(self, opts):
             """Perform an ES10c GetProfilesInfo function."""
-            pi = CardApplicationISDR.store_data_tlv(self._cmd.lchan.scc, ProfileInfoListReq(), ProfileInfoListResp)
+            if opts.all:
+                tags = [nest.tag for nest in ProfileInfo.nested_collection_cls().nested]
+                u8tags = []
+                # TODO: rework TagList to support 2 byte tags to not filter it into u8 tags
+                for tag in tags:
+                    if tag <= 255:
+                        u8tags.append(tag)
+                    elif tag <= 65535:
+                        u8tags.append(tag >> 8)
+                        u8tags.append(tag & 0xff)
+                    # Ignoring 3 byte tags
+                req = ProfileInfoListReq(children=[TagList(decoded=u8tags)])
+            else:
+                req = ProfileInfoListReq()
+
+            pi = CardApplicationISDR.store_data_tlv(self._cmd.lchan.scc, req, ProfileInfoListResp)
             d = pi.to_dict()
             self._cmd.poutput_json(flatten_dict_lists(d['profile_info_list_resp']))
 

pySim/filesystem.py

@@ -44,6 +44,7 @@ from pySim.utils import sw_match, decomposeATR
 from pySim.jsonpath import js_path_modify
 from pySim.commands import SimCardCommands
 from pySim.exceptions import SwMatchError
+from pySim.log import PySimLogger
 
 # int: a single service is associated with this file
 # list: any of the listed services requires this file
@@ -52,6 +53,8 @@ CardFileService = Union[int, List[int], Tuple[int, ...]]
 
 Size = Tuple[int, Optional[int]]
 
+log = PySimLogger.get(__name__)
+
 class CardFile:
     """Base class for all objects in the smart card filesystem.
     Serve as a common ancestor to all other file types; rarely used directly.
@@ -1609,14 +1612,14 @@ class CardModel(abc.ABC):
         card_atr = scc.get_atr()
         for atr in cls._atrs:
             if atr == card_atr:
-                print("Detected CardModel:", cls.__name__)
+                log.info("Detected CardModel: %s", cls.__name__)
                 return True
         # if nothing found try to just compare the Historical Bytes of the ATR
         card_atr_hb = decomposeATR(card_atr)['hb']
         for atr in cls._atrs:
             atr_hb = decomposeATR(atr)['hb']
             if atr_hb == card_atr_hb:
-                print("Detected CardModel:", cls.__name__)
+                log.info("Detected CardModel: %s", cls.__name__)
                 return True
         return False
 

pySim/global_platform/__init__.py

@@ -18,10 +18,12 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 
 import io
+import hashlib
 from copy import deepcopy
 from typing import Optional, List, Dict, Tuple
 from construct import Optional as COptional
 from construct import Struct, GreedyRange, FlagsEnum, Int16ub, Int24ub, Padding, Bit, Const
+from construct import Construct, stream_read, stream_write
 from Cryptodome.Random import get_random_bytes
 from Cryptodome.Cipher import DES, DES3, AES
 from osmocom.utils import *
@@ -148,6 +150,24 @@ sw_table = {
     },
 }
 
+class PutKeyLength(Construct):
+    """A length field of a PUT KEY data field, GP CardSpec v2.3.1 11.8.2.3.1
+    - all lengths ASN.1 BER-TLV (ITU-T X.690 Section 8.1.3)
+    - except that the length 128 may also be coded on one byte as '80' for backwards compatibility
+    80 does not introduce the indefinite form here which is unused in GP as far as i know.
+    That legacy form is accepted when parsing, but never generated, which agrees with the spec"""
+    def _parse(self, stream, context, path):
+        first = stream_read(stream, 1, path)[0]
+        if first <= 0x80:
+            return first
+        return int.from_bytes(stream_read(stream, first & 0x7f, path), 'big')
+
+    def _build(self, obj, stream, context, path):
+        data = bertlv_encode_len(obj)
+        stream_write(stream, data, len(data), path)
+        return obj
+
+
 # GlobalPlatform 2.1.1 Section 9.1.6
 KeyType = Enum(Byte,    des=0x80,
                         tls_psk=0x85,                           # v2.3.1 Section 11.1.8
@@ -602,8 +622,8 @@ class ADF_SD(CardADF):
             See GlobalPlatform CardSpecification v2.3 Section 11.8 for details.
 
             The KCV (Key Check Values) can either be explicitly specified using `--key-check`, or will
-            otherwise be automatically generated for DES and AES keys.  You can suppress the latter using
-            `--suppress-key-check`.
+            otherwise be automatically generated for DES, AES and TLS-PSK keys.  You can suppress the
+            latter using `--suppress-key-check`.
 
             Example (SCP80 KIC/KID/KIK):
                 put_key --key-version-nr 1 --key-id 0x01    --key-type aes --key-data 000102030405060708090a0b0c0d0e0f
@@ -620,33 +640,81 @@ class ADF_SD(CardADF):
             kdb = []
             for i in range(0, len(opts.key_type)):
                 if opts.key_check and len(opts.key_check) > i:
-                    kcv = opts.key_check[i]
+                    kcv = h2b(opts.key_check[i])
                 elif opts.suppress_key_check:
-                    kcv = ''
+                    kcv = b''
                 else:
-                    kcv_bin = compute_kcv(opts.key_type[i], h2b(opts.key_data[i])) or b''
-                    kcv = b2h(kcv_bin)
-                if self._cmd.lchan.scc.scp:
-                    # encrypted key data with DEK of current SCP
-                    kcb = b2h(self._cmd.lchan.scc.scp.encrypt_key(h2b(opts.key_data[i])))
-                else:
-                    # (for example) during personalization, DEK might not be required)
-                    kcb = opts.key_data[i]
-                kdb.append({'key_type': opts.key_type[i], 'kcb': kcb, 'kcv': kcv})
+                    kcv = compute_kcv(opts.key_type[i], h2b(opts.key_data[i])) or b''
+                kdb.append({'key_type': opts.key_type[i], 'clear_key': h2b(opts.key_data[i]), 'kcv': kcv})
             p2 = opts.key_id
             if len(opts.key_type) > 1:
                 p2 |= 0x80
             self.put_key(opts.old_key_version_nr, opts.key_version_nr, p2, kdb)
 
-        # Table 11-68: Key Data Field - Format 1 (Basic Format)
-        KeyDataBasic = GreedyRange(Struct('key_type'/KeyType,
-                                          'kcb'/Prefixed(Int8ub, GreedyBytes),
-                                          'kcv'/Prefixed(Int8ub, GreedyBytes)))
+        # Table 11-68: Key Data Field - Format 1 (Basic Format).  The key component block length is
+        # BER-TLV coded (Section 11.8.2.3.1), the key check value length is always '00' - '7F'.
+        KeyDataBasic = Struct('key_type'/KeyType,
+                              'kcb'/Prefixed(PutKeyLength(), GreedyBytes),
+                              'kcv'/Prefixed(Int8ub, GreedyBytes))
 
-        def put_key(self, old_kvn:int, kvn: int, kid: int, key_dict: dict) -> bytes:
+        @classmethod
+        def encode_key_data_basic(cls, key_type: str, kcb: bytes, kcv: bytes) -> bytes:
+            """Generic Basic key data field, GP CardSpec v2.3 Table 11-68):
+                tag || L1 || <maybe L2> KCB || <1-byte length> KCV"""
+            return cls.KeyDataBasic.build({'key_type': key_type, 'kcb': kcb, 'kcv': kcv})
+
+        @classmethod
+        def encode_key_data_psk(cls, clear_key: bytes, ciphered_key: bytes, kcv: bytes) -> bytes:
+            """Single PSK TLS '85' key data field per GP Amendment B 1.2, 3.9.1 / Table 3-13:
+                85 | L1 | <L2>  <ciphered PSK key> | <KCV length> | <KCV>
+            - framing is like Basic Format, but the kcb is always GP CardSpec Table 11-70
+            so always with the length of the clear text key value, even without padding!
+            - 'ciphered_key' is DEK(block-padded clear key), no additional length prefix."""
+            kcb = bertlv_encode_len(len(clear_key)) + ciphered_key
+            return cls.encode_key_data_basic('tls_psk', kcb, kcv)
+
+        @classmethod
+        def build_put_key_data(cls, kvn: int, keys: List[dict], scp) -> bytes:
+            """Assemble the PUT KEY data field, mixed PSK + DES DEK is supported:
+            - new KVN followed by one key data field per key.
+            - tls_psk keys per GP Amendment B
+            - other key types generic Basic format
+            Param 'keys' is a dict:
+            - 'key_type' (str)
+            - 'clear_key' (bytes)
+            - 'kcv' (bytes / empty).
+            'scp' may be None (e.g. during personalization, when the DEK may not be required)."""
+            key_data = kvn.to_bytes(1, 'big')
+            for k in keys:
+                clear = k['clear_key']
+                if k['key_type'] == 'tls_psk':
+                    # len always part of the data see CardSpec Table 11-70 vs Table 11-71
+                    if scp:
+                        ciphered = scp.dek_encrypt(scp.pad_to_blocksize(clear))
+                    else:
+                        ciphered = clear
+                    key_data += cls.encode_key_data_psk(clear, ciphered, k['kcv'])
+                else:
+                    if scp:
+                        ciphered = scp.encrypt_key(clear)
+                    else:
+                        # (for example) during personalization, DEK might not be required
+                        ciphered = clear
+                    key_data += cls.encode_key_data_basic(k['key_type'], ciphered, k['kcv'])
+            return key_data
+
+        def put_key(self, old_kvn:int, kvn: int, kid: int, keys: List[dict]) -> bytes:
             """Perform the GlobalPlatform PUT KEY command in order to store a new key on the card.
             See GlobalPlatform CardSpecification v2.3 Section 11.8 for details."""
-            key_data = kvn.to_bytes(1, 'big') + build_construct(ADF_SD.AddlShellCommands.KeyDataBasic, key_dict)
+            key_data = self.build_put_key_data(kvn, keys, self._cmd.lchan.scc.scp)
+            # Lc of Table 11-64 is a single byte, while LOAD or STORE DATA splits we can't:
+            # 11.8.2.3.3 splits a key at component boundaries -> not helping here
+            max_cmd_len = self._cmd.lchan.scc.max_cmd_len
+            if len(key_data) > max_cmd_len:
+                raise ValueError('key data field of %u bytes exceeds the maximum command length of %u '
+                                 '(limited by the overhead of the current secure channel); use fewer '
+                                 'keys per command, a single key component that large needs STORE DATA' %
+                                 (len(key_data), max_cmd_len))
             hdr = "80D8%02x%02x%02x" % (old_kvn, kid, len(key_data))
             data, _sw = self._cmd.lchan.scc.send_apdu_checksw(hdr + b2h(key_data) + "00")
             return data
@@ -826,23 +894,32 @@ class ADF_SD(CardADF):
         load_parser_from_grp.add_argument('--from-hex', type=is_hexstr, help='load from hex string')
         load_parser_from_grp.add_argument('--from-file', type=argparse.FileType('rb', 0), help='load from binary file')
         load_parser_from_grp.add_argument('--from-cap-file', type=argparse.FileType('rb', 0), help='load from JAVA-card CAP file')
+        load_parser.add_argument('--chunk-len', type=auto_uint8, default=None,
+                                 help='Block size for the LOAD command; default: as large as the current secure channel overhead permits, at most 240')
 
         @cmd2.with_argparser(load_parser)
         def do_load(self, opts):
             """Perform a GlobalPlatform LOAD command. (We currently only support loading without DAP and
             without ciphering.)"""
             if opts.from_hex is not None:
-                self.load(h2b(opts.from_hex))
+                self.load(h2b(opts.from_hex), opts.chunk_len)
             elif opts.from_file is not None:
-                self.load(opts.from_file.read())
+                self.load(opts.from_file.read(), opts.chunk_len)
             elif opts.from_cap_file is not None:
                 cap = CapFile(opts.from_cap_file)
-                self.load(cap.get_loadfile())
+                self.load(cap.get_loadfile(), opts.chunk_len)
             else:
                 raise ValueError('load source not specified!')
 
-        def load(self, contents:bytes, chunk_len:int = 240):
-            # TODO:tune chunk_len based on the overhead of the used SCP?
+        def load(self, contents:bytes, chunk_len:Optional[int] = None):
+            # scc.max_cmd_len knows the overhead the currently active SCP
+            # 240 is the old default, keep it for now.
+            max_chunk_len = self._cmd.lchan.scc.max_cmd_len
+            if chunk_len is None:
+                chunk_len = min(240, max_chunk_len)
+            elif not 1 <= chunk_len <= max_chunk_len:
+                raise ValueError('chunk_len must be in range 1..%u (limited by the overhead of the current secure channel)' %
+                                 max_chunk_len)
             # build TLV according to GPC_SPE_034 section 11.6.2.3 / Table 11-58 for unencrypted case
             remainder = b'\xC4' + bertlv_encode_len(len(contents)) + contents
             # transfer this in various chunks to the card
@@ -881,6 +958,8 @@ class ADF_SD(CardADF):
         install_cap_parser_inst_prm_grp.add_argument('--install-parameters-stk',
                                                      type=is_hexstr, default=None,
                                                      help='Load Parameters (ETSI TS 102 226, section 8.2.1.3.2.1)')
+        install_cap_parser.add_argument('--chunk-len', type=auto_uint8, default=None,
+                                        help='Block size for the LOAD command; default: as large as the current secure channel overhead permits, at most 240')
 
         @cmd2.with_argparser(install_cap_parser)
         def do_install_cap(self, opts):
@@ -919,7 +998,7 @@ class ADF_SD(CardADF):
             self._cmd.poutput("step #1: install for load...")
             self.do_install_for_load("--load-file-aid %s --security-domain-aid %s" % (load_file_aid, security_domain_aid))
             self._cmd.poutput("step #2: load...")
-            self.load(load_file)
+            self.load(load_file, opts.chunk_len)
             self._cmd.poutput("step #3: install_for_install (and make selectable)...")
             self.do_install_for_install("--load-file-aid %s --module-aid %s --application-aid %s --install-parameters %s --make-selectable" %
                                         (load_file_aid, module_aid, application_aid, install_parameters))
@@ -1065,10 +1144,16 @@ def compute_kcv_aes(key:bytes) -> bytes:
     cipher = AES.new(key, AES.MODE_ECB)
     return cipher.encrypt(plaintext)
 
+def compute_kcv_psk(key:bytes) -> bytes:
+    # GP Amendment B v1.2, 3.9.1 / Table 3-13
+    # KCV of a PSK TLS key is the 3 highest-order bytes of the SHA-1 digest of the clear key value.
+    return hashlib.sha1(key).digest()
+
 # dict is keyed by the string name of the KeyType enum above in this file
 KCV_CALCULATOR = {
         'aes': compute_kcv_aes,
         'des': compute_kcv_des,
+        'tls_psk': compute_kcv_psk,
     }
 
 def compute_kcv(key_type: str, key: bytes) -> Optional[bytes]:

pySim/global_platform/scp.py

@@ -27,9 +27,9 @@ from osmocom.utils import b2h
 from osmocom.tlv import bertlv_parse_len, bertlv_encode_len
 from pySim.utils import parse_command_apdu
 from pySim.secure_channel import SecureChannel
+from pySim.log import PySimLogger
 
-logger = logging.getLogger(__name__)
-logger.setLevel(logging.DEBUG)
+log = PySimLogger.get(__name__)
 
 def scp02_key_derivation(constant: bytes, counter: int, base_key: bytes) -> bytes:
     assert len(constant) == 2
@@ -75,7 +75,7 @@ class Scp02SessionKeys:
             h = e.encrypt(strxor(h, bytes(padded_data[8*i:8*(i+1)])))
         h = d.decrypt(h)
         h = e.encrypt(h)
-        logger.debug("mac_1des(%s,icv=%s) -> %s", b2h(data), b2h(icv), b2h(h))
+        log.debug("mac_1des(%s,icv=%s) -> %s", b2h(data), b2h(icv), b2h(h))
         if self.des_icv_enc:
             self.icv = self.des_icv_enc.encrypt(h)
         else:
@@ -89,7 +89,7 @@ class Scp02SessionKeys:
         h = b'\x00' * 8
         for i in range(q):
             h = e.encrypt(strxor(h, bytes(padded_data[8*i:8*(i+1)])))
-        logger.debug("mac_3des(%s) -> %s", b2h(data), b2h(h))
+        log.debug("mac_3des(%s) -> %s", b2h(data), b2h(h))
         return h
 
     def __init__(self, counter: int, card_keys: 'GpCardKeyset', icv_encrypt=True):
@@ -182,6 +182,29 @@ class SCP(SecureChannel, abc.ABC):
         """Should we perform R-ENC?"""
         return self.security_level & 0x20
 
+    @property
+    @abc.abstractmethod
+    def mac_len(self) -> int:
+        """Length of the appended C-MAC, to be provided by derived class."""
+
+    @property
+    def overhead(self) -> int:
+        """Worst-case len that wrapping a command APDU adds to its data field at the
+        current sec level is (255 - overhead), C-MAC + C-DECRYPTION encryption padding."""
+        if not self.do_cmac:
+            return 0
+        if not self.do_cenc:
+            return self.mac_len
+        # see Secure Channel Protocol '03' Card Specification v2.3 - Amendment D v1.1.2
+        # which defers to GPCS v2.3 Section B.2 which then defers to
+        # NIST SP 800-38B for encryption and points out that
+        # the padding is, as expected, just the usual padding from NIST SP 800-38A
+        # C-DECRYPTION pads with ('80'+['00'...] at least 1 byte) up to
+        # the cipher block size + C-MAC on top -> largest usable data field
+        # is one byte less than the largest block-size multiple within 255 - mac_len.
+        bs = self.sk.blocksize
+        return 255 - ((255 - self.mac_len) // bs * bs - 1)
+
     def __str__(self) -> str:
         return "%s[%02x]" % (self.__class__.__name__, self.security_level)
 
@@ -215,11 +238,20 @@ class SCP(SecureChannel, abc.ABC):
     def gen_ext_auth_apdu(self, security_level: int = 0x01) -> bytes:
         pass
 
+    def pad_to_blocksize(self, data: bytes) -> bytes:
+        """Right pad the data with zero bytes to a multiple of the DEK cipher block size."""
+        if len(data) % self.sk.blocksize:
+            # not '+=' which would mutate the callers bytearray in place..
+            data = data + b'\x00' * (self.sk.blocksize - len(data) % self.sk.blocksize)
+        return data
+
     def encrypt_key(self, key: bytes) -> bytes:
         """Encrypt a key with the DEK."""
-        num_pad = len(key) % self.sk.blocksize
-        if num_pad:
-            return bertlv_encode_len(len(key)) + self.dek_encrypt(key + b'\x00'*num_pad)
+        if len(key) % self.sk.blocksize:
+            # The kcv is right padded before encryption and the kcb
+            # is formatted as described in Table 11-70: preceded by the actual length of the
+            # clear text kcv.
+            return bertlv_encode_len(len(key)) + self.dek_encrypt(self.pad_to_blocksize(key))
         return self.dek_encrypt(key)
 
     def decrypt_key(self, encrypted_key:bytes) -> bytes:
@@ -232,9 +264,8 @@ class SCP(SecureChannel, abc.ABC):
             # Block provides the actual length of the key component value, which allows recovering the
             # clear-text key component value after decryption of the encrypted key component value and removal
             # of padding bytes.
-            decrypted = self.dek_decrypt(encrypted_key)
-            key_len, remainder = bertlv_parse_len(decrypted)
-            return remainder[:key_len]
+            key_len, remainder = bertlv_parse_len(encrypted_key)
+            return self.dek_decrypt(remainder)[:key_len]
         else:
             # If the length of the Key Component Block is a multiple of the block size of the encryption
             # algorithm (i.e.  8 bytes for DES, 16 bytes for AES), then it shall be assumed that no padding
@@ -260,10 +291,8 @@ class SCP02(SCP):
     # Key Version Number 0x70 is a non-spec special-case of sysmoISIM-SJA2/SJA5 and possibly more sysmocom products
     # Key Version Number 0x01 is a non-spec special-case of sysmoUSIM-SJS1
     kvn_ranges = [[0x01, 0x01], [0x20, 0x2f], [0x70, 0x70]]
-
-    def __init__(self, *args, **kwargs):
-        self.overhead = 8
-        super().__init__(*args, **kwargs)
+    # C-MAC (Single DES + final 3DES, B.1.2.2) is always one full DES block
+    mac_len = 8
 
     def dek_encrypt(self, plaintext:bytes) -> bytes:
         # See also GPC section B.1.1.2, E.4.7, and E.4.1
@@ -276,10 +305,10 @@ class SCP02(SCP):
         return cipher.decrypt(ciphertext)
 
     def _compute_cryptograms(self, card_challenge: bytes, host_challenge: bytes):
-        logger.debug("host_challenge(%s), card_challenge(%s)", b2h(host_challenge), b2h(card_challenge))
+        log.debug("host_challenge(%s), card_challenge(%s)", b2h(host_challenge), b2h(card_challenge))
         self.host_cryptogram = self.sk.calc_mac_3des(self.sk.counter.to_bytes(2, 'big') + card_challenge + host_challenge)
         self.card_cryptogram = self.sk.calc_mac_3des(self.host_challenge + self.sk.counter.to_bytes(2, 'big') + card_challenge)
-        logger.debug("host_cryptogram(%s), card_cryptogram(%s)", b2h(self.host_cryptogram), b2h(self.card_cryptogram))
+        log.debug("host_cryptogram(%s), card_cryptogram(%s)", b2h(self.host_cryptogram), b2h(self.card_cryptogram))
 
     def gen_init_update_apdu(self, host_challenge: bytes = b'\x00'*8) -> bytes:
         """Generate INITIALIZE UPDATE APDU."""
@@ -291,7 +320,7 @@ class SCP02(SCP):
         resp = self.constr_iur.parse(resp_bin)
         self.card_challenge = resp['card_challenge']
         self.sk = Scp02SessionKeys(resp['seq_counter'], self.card_keys)
-        logger.debug(self.sk)
+        log.debug(self.sk)
         self._compute_cryptograms(self.card_challenge, self.host_challenge)
         if self.card_cryptogram != resp['card_cryptogram']:
             raise ValueError("card cryptogram doesn't match")
@@ -311,7 +340,7 @@ class SCP02(SCP):
 
     def _wrap_cmd_apdu(self, apdu: bytes, *args, **kwargs) -> bytes:
         """Wrap Command APDU for SCP02: calculate MAC and encrypt."""
-        logger.debug("wrap_cmd_apdu(%s)", b2h(apdu))
+        log.debug("wrap_cmd_apdu(%s)", b2h(apdu))
 
         if not self.do_cmac:
             return apdu
@@ -338,10 +367,16 @@ class SCP02(SCP):
             # CMAC on modified APDU
             mlc = lc + 8
             clac = cla | CLA_SM
+        if mlc >= 256:
+            raise ValueError('Modified Lc (%u) would exceed maximum when appending 8 bytes of mac' % mlc)
         mac = self.sk.calc_mac_1des(bytes([clac]) + apdu[1:4] + bytes([mlc]) + data)
         if self.do_cenc:
+            padded_data = pad80(data, 8)
+            if len(padded_data) + 8 >= 256:
+                raise ValueError('Modified Lc (%u) would exceed maximum when appending padding and mac' %
+                                 (len(padded_data) + 8))
             k = DES3.new(self.sk.enc, DES.MODE_CBC, b'\x00'*8)
-            data = k.encrypt(pad80(data, 8))
+            data = k.encrypt(padded_data)
             lc = len(data)
 
         lc += 8
@@ -378,7 +413,7 @@ def scp03_key_derivation(constant: bytes, context: bytes, base_key: bytes, l: Op
     if l is None:
         l = len(base_key) * 8
 
-    logger.debug("scp03_kdf(constant=%s, context=%s, base_key=%s, l=%u)", b2h(constant), b2h(context), b2h(base_key), l)
+    log.debug("scp03_kdf(constant=%s, context=%s, base_key=%s, l=%u)", b2h(constant), b2h(context), b2h(base_key), l)
     output_len = l // 8
     # SCP03 Section 4.1.5 defines a different parameter order than NIST SP 800-108, so we cannot use the
     # existing Cryptodome.Protocol.KDF.SP800_108_Counter function :(
@@ -451,7 +486,7 @@ class Scp03SessionKeys:
         # This block SHALL be encrypted with S-ENC to produce the ICV for command encryption.
         cipher = AES.new(self.s_enc, AES.MODE_CBC, iv)
         icv = cipher.encrypt(data)
-        logger.debug("_get_icv(data=%s, is_resp=%s) -> icv=%s", b2h(data), is_response, b2h(icv))
+        log.debug("_get_icv(data=%s, is_resp=%s) -> icv=%s", b2h(data), is_response, b2h(icv))
         return icv
 
     # TODO: Resolve duplication with pySim.esim.bsp.BspAlgoCryptAES128 which provides pad80-wrapping
@@ -477,9 +512,13 @@ class SCP03(SCP):
 
     def __init__(self, *args, **kwargs):
         self.s_mode = kwargs.pop('s_mode', 8)
-        self.overhead = self.s_mode
         super().__init__(*args, **kwargs)
 
+    @property
+    def mac_len(self) -> int:
+        # C-MAC truncated to 8 in S8 or 16 bytes in S16 mode
+        return self.s_mode
+
     def dek_encrypt(self, plaintext:bytes) -> bytes:
         cipher = AES.new(self.card_keys.dek, AES.MODE_CBC, b'\x00'*16)
         return cipher.encrypt(plaintext)
@@ -489,12 +528,12 @@ class SCP03(SCP):
         return cipher.decrypt(ciphertext)
 
     def _compute_cryptograms(self):
-        logger.debug("host_challenge(%s), card_challenge(%s)", b2h(self.host_challenge), b2h(self.card_challenge))
+        log.debug("host_challenge(%s), card_challenge(%s)", b2h(self.host_challenge), b2h(self.card_challenge))
         # Card + Host Authentication Cryptogram: Section 6.2.2.2 + 6.2.2.3
         context = self.host_challenge + self.card_challenge
         self.card_cryptogram = scp03_key_derivation(self.sk.DERIV_CONST_AUTH_CGRAM_CARD, context, self.sk.s_mac, l=self.s_mode*8)
         self.host_cryptogram = scp03_key_derivation(self.sk.DERIV_CONST_AUTH_CGRAM_HOST, context, self.sk.s_mac, l=self.s_mode*8)
-        logger.debug("host_cryptogram(%s), card_cryptogram(%s)", b2h(self.host_cryptogram), b2h(self.card_cryptogram))
+        log.debug("host_cryptogram(%s), card_cryptogram(%s)", b2h(self.host_cryptogram), b2h(self.card_cryptogram))
 
     def gen_init_update_apdu(self, host_challenge: Optional[bytes] = None) -> bytes:
         """Generate INITIALIZE UPDATE APDU."""
@@ -514,7 +553,7 @@ class SCP03(SCP):
         self.i_param = resp['i_param']
         # derive session keys and compute cryptograms
         self.sk = Scp03SessionKeys(self.card_keys, self.host_challenge, self.card_challenge)
-        logger.debug(self.sk)
+        log.debug(self.sk)
         self._compute_cryptograms()
         # verify computed cryptogram matches received cryptogram
         if self.card_cryptogram != resp['card_cryptogram']:
@@ -529,7 +568,7 @@ class SCP03(SCP):
 
     def _wrap_cmd_apdu(self, apdu: bytes, skip_cenc: bool = False) -> bytes:
         """Wrap Command APDU for SCP03: calculate MAC and encrypt."""
-        logger.debug("wrap_cmd_apdu(%s)", b2h(apdu))
+        log.debug("wrap_cmd_apdu(%s)", b2h(apdu))
 
         if not self.do_cmac:
             return apdu
@@ -584,7 +623,7 @@ class SCP03(SCP):
         # status word: in this case only the status word shall be returned in the response. All status words
         # except '9000' and warning status words (i.e. '62xx' and '63xx') shall be interpreted as error status
         # words.
-        logger.debug("unwrap_rsp_apdu(sw=%s, rsp_apdu=%s)", sw, rsp_apdu)
+        log.debug("unwrap_rsp_apdu(sw=%s, rsp_apdu=%s)", sw, rsp_apdu)
         if not self.do_rmac:
             assert not self.do_renc
             return rsp_apdu
@@ -600,9 +639,9 @@ class SCP03(SCP):
         if self.do_renc:
             # decrypt response data
             decrypted = self.sk._decrypt(response_data)
-            logger.debug("decrypted: %s", b2h(decrypted))
+            log.debug("decrypted: %s", b2h(decrypted))
             # remove padding
             response_data = unpad80(decrypted)
-            logger.debug("response_data: %s", b2h(response_data))
+            log.debug("response_data: %s", b2h(response_data))
 
         return response_data

pySim/legacy/cards.py

@@ -152,7 +152,8 @@ class SimCard(SimCardBase):
         return sw
 
     def update_smsp(self, smsp):
-        data, sw = self._scc.update_record(EF['SMSP'], 1, rpad(smsp, 84))
+        print("using update_smsp")
+        data, sw = self._scc.update_record(EF['SMSP'], 1, smsp, leftpad=True)
         return sw
 
     def update_ad(self, mnc=None, opmode=None, ofm=None, path=EF['AD']):

pySim/log.py

@@ -24,7 +24,15 @@
 #
 
 import logging
-from cmd2 import style
+import cmd2
+from packaging import version
+
+if version.parse(cmd2.__version__) >= version.parse("3.0.0"):
+    from cmd2 import stylize as _stylize # pylint: disable=no-name-in-module
+    def _style(text, fg=None): # pylint: disable=function-redefined
+        return _stylize(text, fg) if fg else text
+else: # cmd2>=2.6.2
+    from cmd2 import style as _style # pylint: disable=no-name-in-module
 
 class _PySimLogHandler(logging.Handler):
     def __init__(self, log_callback):
@@ -44,7 +52,7 @@ class PySimLogger:
     """
 
     LOG_FMTSTR = "%(levelname)s: %(message)s"
-    LOG_FMTSTR_VERBOSE = "%(module)s.%(lineno)d -- " + LOG_FMTSTR
+    LOG_FMTSTR_VERBOSE = "%(name)s.%(lineno)d -- " + LOG_FMTSTR
     __formatter = logging.Formatter(LOG_FMTSTR)
     __formatter_verbose = logging.Formatter(LOG_FMTSTR_VERBOSE)
 
@@ -121,7 +129,7 @@ class PySimLogger:
                 if isinstance(color, str):
                     PySimLogger.print_callback(color + formatted_message + "\033[0m")
                 else:
-                    PySimLogger.print_callback(style(formatted_message, fg = color))
+                    PySimLogger.print_callback(_style(formatted_message, fg = color))
             else:
                 PySimLogger.print_callback(formatted_message)
 

pySim/transport/__init__.py

@@ -301,24 +301,54 @@ class LinkBaseTpdu(LinkBase):
 
         prev_tpdu = tpdu
         data, sw = self.send_tpdu(tpdu)
+        log.debug("T0: case #%u TPDU: %s => %s %s", case, tpdu, data or "(no data)", sw or "(no status word)")
+        if sw is None:
+            raise ValueError("no status word received")
 
-        # When we have sent the first APDU, the SW may indicate that there are response bytes
-        # available. There are two SWs commonly used for this 9fxx (sim) and 61xx (usim), where
-        # xx is the number of response bytes available.
-        # See also:
-        if sw is not None:
-            while (sw[0:2] in ['9f', '61', '62', '63']):
-                # SW1=9F: 3GPP TS 51.011 9.4.1, Responses to commands which are correctly executed
-                # SW1=61: ISO/IEC 7816-4, Table 5 — General meaning of the interindustry values of SW1-SW2
-                # SW1=62: ETSI TS 102 221 7.3.1.1.4 Clause 4b): 62xx, 63xx, 9xxx != 9000
-                tpdu_gr = tpdu[0:2] + 'c00000' + sw[2:4]
+        # After sending the APDU/TPDU the UICC/eUICC or SIM may response with a status word that indicates that further
+        # TPDUs have to be sent in order to complete the task.
+        if case == 4 or self.apdu_strict == False:
+            # In case the APDU is a case #4 APDU, the UICC/eUICC/SIM may indicate that there is response data
+            # available which has to be retrieved using a GET RESPONSE command TPDU.
+            #
+            # ETSI TS 102 221, section 7.3.1.1.4 is very cleare about the fact that the GET RESPONSE mechanism
+            # shall only apply on case #4 APDUs but unfortunately it is impossible to distinguish between case #3
+            # and case #4 when the APDU format is not strictly followed. In order to be able to detect case #4
+            # correctly the Le byte (usually 0x00) must be present, is often forgotten. To avoid problems with
+            # legacy scripts that use raw APDU strings, we will still loosely apply GET RESPONSE based on what
+            # the status word indicates. Unless the user explicitly enables the strict mode (set apdu_strict true)
+            while True:
+                if sw in ['9000', '9100']:
+                    # A status word of 9000 (or 9100 in case there is pending data from a proactive SIM command)
+                    # indicates that either no response data was returnd or all response data has been retrieved
+                    # successfully. We may discontinue the processing at this point.
+                    break;
+                if sw[0:2] in ['61', '9f']:
+                    # A status word of 61xx or 9fxx indicates that there is (still) response data available. We
+                    # send a GET RESPONSE command with the length value indicated in the second byte of the status
+                    # word. (see also ETSI TS 102 221, section 7.3.1.1.4, clause 4a and 3GPP TS 51.011 9.4.1 and
+                    # ISO/IEC 7816-4, Table 5)
+                    le_gr = sw[2:4]
+                elif sw[0:2] in ['62', '63']:
+                    # There are corner cases (status word is 62xx or 63xx) where the UICC/eUICC/SIM asks us
+                    # to send a dummy GET RESPONSE command. We send a GET RESPONSE command with a length of 0.
+                    # (see also ETSI TS 102 221, section 7.3.1.1.4, clause 4b and ETSI TS 151 011, section 9.4.1)
+                    le_gr = '00'
+                else:
+                    # A status word other then the ones covered by the above logic may indicate an error. In this
+                    # case we will discontinue the processing as well.
+                    # (see also ETSI TS 102 221, section 7.3.1.1.4, clause 4c)
+                    break
+                tpdu_gr = tpdu[0:2] + 'c00000' + le_gr
                 prev_tpdu = tpdu_gr
-                d, sw = self.send_tpdu(tpdu_gr)
-                data += d
-            if sw[0:2] == '6c':
-                # SW1=6C: ETSI TS 102 221 Table 7.1: Procedure byte coding
-                tpdu_gr = prev_tpdu[0:8] + sw[2:4]
-                data, sw = self.send_tpdu(tpdu_gr)
+                data_gr, sw = self.send_tpdu(tpdu_gr)
+                log.debug("T0: GET RESPONSE TPDU: %s => %s %s", tpdu_gr, data_gr or "(no data)", sw or "(no status word)")
+                data += data_gr
+        if sw[0:2] == '6c':
+            # SW1=6C: ETSI TS 102 221 Table 7.1: Procedure byte coding
+            tpdu_gr = prev_tpdu[0:8] + sw[2:4]
+            data, sw = self.send_tpdu(tpdu_gr)
+            log.debug("T0: repated case #%u TPDU: %s => %s %s", case, tpdu_gr, data or "(no data)", sw or "(no status word)")
 
         return data, sw
 

pySim/ts_102_221.py

@@ -17,6 +17,7 @@ You should have received a copy of the GNU General Public License
 along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 from bidict import bidict
+import copy
 
 from construct import Select, Const, Bit, Struct, Int16ub, FlagsEnum, GreedyString, ValidationError
 from construct import Optional as COptional, Computed
@@ -335,6 +336,8 @@ class TerminalCapability(BER_TLV_IE, tag=0xa9, nested=[TerminalPowerSupply, Exte
 
 # ETSI TS 102 221 Section 9.2.7 + ISO7816-4 9.3.3/9.3.4
 class _AM_DO_DF(DataObject):
+    """ISO7816-4:2005 5.4.3.1 Table 16"""
+
     def __init__(self):
         super().__init__('access_mode', 'Access Mode', tag=0x80)
 
@@ -381,7 +384,7 @@ class _AM_DO_DF(DataObject):
 
 
 class _AM_DO_EF(DataObject):
-    """ISO7816-4 9.3.2 Table 18 + 9.3.3.1 Table 31"""
+    """ISO7816-4:2005 5.4.3.1 Table 17"""
 
     def __init__(self):
         super().__init__('access_mode', 'Access Mode', tag=0x80)
@@ -429,7 +432,7 @@ class _AM_DO_EF(DataObject):
 
 
 class _AM_DO_CHDR(DataObject):
-    """Command Header Access Mode DO according to ISO 7816-4 Table 32."""
+    """Command Header Access Mode DO according to ISO 7816-4:2005 5.4.3.2 Table 22."""
 
     def __init__(self, tag):
         super().__init__('command_header', 'Command Header Description', tag=tag)
@@ -543,8 +546,9 @@ class CRT_DO(DataObject):
         pin = pin_names.inverse[self.decoded]
         return b'\x83\x01' + pin.to_bytes(1, 'big') + b'\x95\x01\x08'
 
-# ISO7816-4 9.3.3 Table 33
 class SecCondByte_DO(DataObject):
+    """ISO7816-4:2005 5.4.3.1 Table 20"""
+
     def __init__(self, tag=0x9d):
         super().__init__('security_condition_byte', tag=tag)
 
@@ -732,36 +736,57 @@ class EF_ARR(LinFixedEF):
                 raise ValueError
         return by_mode
 
+    @staticmethod
+    def __get_do_sequence(decode_for_df : bool = False):
+        if decode_for_df:
+            return DataObjectSequence('arr', sequence=[AM_DO_DF, SC_DO])
+        else:
+            return DataObjectSequence('arr', sequence=[AM_DO_EF, SC_DO])
+
     def _decode_record_bin(self, raw_bin_data, **kwargs):
-        # we can only guess if we should decode for EF or DF here :(
-        arr_seq = DataObjectSequence('arr', sequence=[AM_DO_EF, SC_DO])
+        # we can only guess if we should decode for EF or DF here, but our caller may
+        # be able to pass us a hint:
+        arr_seq = self.__get_do_sequence(kwargs.get('decode_for_df', False))
         dec = arr_seq.decode_multi(raw_bin_data)
         # we cannot pass the result through flatten() here, as we don't have a related
         # 'un-flattening' decoder, and hence would be unable to encode :(
         return dec[0]
 
     def _encode_record_bin(self, in_json, **kwargs):
-        # we can only guess if we should decode for EF or DF here :(
-        arr_seq = DataObjectSequence('arr', sequence=[AM_DO_EF, SC_DO])
+        # we can only guess if we should decode for EF or DF here, but our caller may
+        # be able to pass us a hint:
+        arr_seq = self.__get_do_sequence(kwargs.get('encode_for_df', False))
         return arr_seq.encode_multi(in_json)
 
     @with_default_category('File-Specific Commands')
     class AddlShellCommands(CommandSet):
-        @cmd2.with_argparser(LinFixedEF.ShellCommands.read_rec_dec_parser)
+        read_arr_argparser = copy.deepcopy(LinFixedEF.ShellCommands.read_rec_dec_parser)
+        read_arr_argparser.add_argument('--decode-for-df', action='store_true',
+                                        help='Decode EF.ARR record as if used by a DF (default: EF)')
+
+        @cmd2.with_argparser(read_arr_argparser)
         def do_read_arr_record(self, opts):
             """Read one EF.ARR record in flattened, human-friendly form."""
-            (data, _sw) = self._cmd.lchan.read_record_dec(opts.RECORD_NR)
+            (hexdata, _sw) = self._cmd.lchan.read_record(opts.RECORD_NR)
+            data = self._cmd.lchan.selected_file._decode_record_bin(h2b(hexdata),
+                                                                    decode_for_df = opts.decode_for_df)
             data = self._cmd.lchan.selected_file.flatten(data)
             self._cmd.poutput_json(data, opts.oneline)
 
-        @cmd2.with_argparser(LinFixedEF.ShellCommands.read_recs_dec_parser)
+        read_arrs_argparser = copy.deepcopy(LinFixedEF.ShellCommands.read_recs_dec_parser)
+        read_arrs_argparser.add_argument('--decode-for-df', action='store_true',
+                                        help='Decode EF.ARR records as if used by a DF (default: EF)')
+
+        @cmd2.with_argparser(read_arrs_argparser)
         def do_read_arr_records(self, opts):
             """Read + decode all EF.ARR records in flattened, human-friendly form."""
             num_of_rec = self._cmd.lchan.selected_file_num_of_rec()
             # collect all results in list so they are rendered as JSON list when printing
             data_list = []
             for recnr in range(1, 1 + num_of_rec):
-                (data, _sw) = self._cmd.lchan.read_record_dec(recnr)
+                (hexdata, _sw) = self._cmd.lchan.read_record(recnr)
+                data = self._cmd.lchan.selected_file._decode_record_bin(h2b(hexdata),
+                                                                        decode_for_df = opts.decode_for_df)
                 data = self._cmd.lchan.selected_file.flatten(data)
                 data_list.append(data)
             self._cmd.poutput_json(data_list, opts.oneline)

pySim/ts_31_102.py

@@ -285,6 +285,14 @@ class EF_SUCI_Calc_Info(TransparentEF):
                                 {"hnet_pubkey_identifier": 11, "hnet_pubkey":
                                  h2b("d1bc365f4997d17ce4374e72181431cbfeba9e1b98d7618f79d48561b144672a")}]} ),
     ]
+    _test_decode = [
+        ( 'A000FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF',
+          {"prot_scheme_id_list": [],
+           "hnet_pubkey_list": []} ),
+        ( 'A000A100FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF',
+          {"prot_scheme_id_list": [],
+           "hnet_pubkey_list": []} ),
+    ]
     # 3GPP TS 31.102 Section 4.4.11.8
     class ProtSchemeIdList(BER_TLV_IE, tag=0xa0):
         # FIXME: 3GPP TS 24.501 Protection Scheme Identifier
@@ -389,7 +397,7 @@ class EF_SUCI_Calc_Info(TransparentEF):
         # remaining data holds Home Network Public Key Data Object
         hpkl = EF_SUCI_Calc_Info.HnetPubkeyList()
         hpkl.from_tlv(in_bytes[pos:])
-        hnet_pubkey_list = self._compact_pubkey_list(hpkl.to_dict()['hnet_pubkey_list'])
+        hnet_pubkey_list = self._compact_pubkey_list(hpkl.to_dict()['hnet_pubkey_list'] or [])
 
         return {
             'prot_scheme_id_list': prot_scheme_id_list,

pySim/ts_51_011.py

@@ -1263,9 +1263,11 @@ class CardProfileSIM(CardProfile):
 
     @staticmethod
     def decode_select_response(resp_hex: str) -> object:
-        # we try to build something that resembles a dict resulting from the TLV decoder
-        # of TS 102.221 (FcpTemplate), so that higher-level code only has to deal with one
-        # format of SELECT response
+        """
+        Decode the select response to a dict representation, similar to the one of TS 102.221 (see ts_102_221.py,
+        class FcpTemplate), so that higher-level code only has to deal with one respresentation. See also
+        3GPP TS 51.011, section 9.2.1
+        """
         resp_bin = h2b(resp_hex)
         struct_of_file_map = {
             0: 'transparent',
@@ -1303,13 +1305,24 @@ class CardProfileSIM(CardProfile):
                 record_len = resp_bin[14]
                 ret['file_descriptor']['record_len'] = record_len
                 ret['file_descriptor']['num_of_rec'] = ret['file_size'] // record_len
-            ret['access_conditions'] = b2h(resp_bin[8:10])
-            if resp_bin[11] & 0x01 == 0:
+            ret['access_conditions'] = b2h(resp_bin[8:11])
+
+            # Life cycle status integer, see also ETSI TS 102 221, table 11.7b
+            lcsi = resp_bin[11]
+            if lcsi == 0x00:
+                ret['life_cycle_status_int'] = 'no_information'
+            elif lcsi == 0x01:
+                ret['life_cycle_status_int'] = 'creation'
+            elif lcsi == 0x03:
+                ret['life_cycle_status_int'] = 'initialization'
+            elif lcsi & 0xFD == 0x05:
                 ret['life_cycle_status_int'] = 'operational_activated'
-            elif resp_bin[11] & 0x04:
+            elif lcsi & 0xFD == 0x04:
                 ret['life_cycle_status_int'] = 'operational_deactivated'
+            elif lcsi & 0xFC == 0x0C:
+                ret['life_cycle_status_int'] = 'termination'
             else:
-                ret['life_cycle_status_int'] = 'terminated'
+                ret['life_cycle_status_int'] = lcsi
         return ret
 
     @classmethod

pyproject.toml

@@ -4,3 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [tool.pylint.main]
 ignored-classes = ["twisted.internet.reactor"]
+
+[tool.pylint.TYPECHECK]
+# SdKey subclasses are generated dynamically via SdKey.generate_sd_key_classes()
+generated-members = ["SdKey[A-Za-z0-9]+"]

requirements.txt

@@ -1,7 +1,7 @@
 pyscard
 pyserial
 pytlv
-cmd2>=2.6.2,<3.0
+cmd2>=2.6.2,<4.0
 jsonpath-ng
 construct>=2.10.70
 bidict

setup.py

@@ -21,7 +21,7 @@ setup(
         "pyscard",
         "pyserial",
         "pytlv",
-        "cmd2 >= 1.5.0, < 3.0",
+        "cmd2 >= 2.6.2, < 4.0",
         "jsonpath-ng",
         "construct >= 2.10.70",
         "bidict",

tests/pySim-prog_test/Fairwaves-SIM.ok

@@ -5,7 +5,7 @@ ICCID: 8988219000000117833
 IMSI: 001010000000111
 GID1: ffffffffffffffff
 GID2: ffffffffffffffff
-SMSP: e1ffffffffffffffffffffffff0581005155f5ffffffffffff000000ffffffffffffffffffffffffffff
+SMSP: ffffffffffffffffffffffffffffe1ffffffffffffffffffffffff0581005155f5ffffffffffff000000
 SMSC: 0015555
 SPN: Fairwaves
 Show in HPLMN: False

tests/pySim-prog_test/Wavemobile-SIM.ok

@@ -5,7 +5,7 @@ ICCID: 89445310150011013678
 IMSI: 001010000000102
 GID1: Can't read file -- SW match failed! Expected 9000 and got 6a82.
 GID2: Can't read file -- SW match failed! Expected 9000 and got 6a82.
-SMSP: e1ffffffffffffffffffffffff0581005155f5ffffffffffff000000ffffffffffffffffffffffffffff
+SMSP: ffffffffffffffffffffffffffffe1ffffffffffffffffffffffff0581005155f5ffffffffffff000000
 SMSC: 0015555
 SPN: wavemobile
 Show in HPLMN: False

tests/pySim-shell_test/apdu/test_apdu.script

@@ -7,10 +7,24 @@ set apdu_strict true
 # No command data field, No response data field present
 apdu 00700001 --expect-sw 9000 --expect-response-regex '^$'
 
+# Case #1: (verify pin)
+# This command returns the number of remaining authentication attempts in the
+# form of a status that has the form 63cX, where X is the number of remaining
+# attempts. Such a status word can be easily confused with the response to a
+# case #4 APDU. This test checks if the transport layer correctly distinguishes
+# the between APDU case #1 and APDU case #4.
+apdu 0020000A --expect-sw 63c? --expect-response-regex '^$'
+
 # Case #2: (status)
 # No command data field, Response data field present
 apdu 80F2000000 --expect-sw 9000 --expect-response-regex '^[a-fA-F0-9]+$'
 
+# Case #2: (verify pin)
+# (see also above). This test checks if the transport layer is also able to
+# distinguish correctly between APDU case #2 (with zero length response) and
+# APDU case #4.
+apdu 0020000A00 --expect-sw 63c? --expect-response-regex '^$'
+
 # Case #3: (terminal capability)
 # Command data field present, No response data field
 apdu 80AA000005a903830180 --expect-sw 9000 --expect-response-regex '^$'

tests/unittests/test_card_key_provider.py

@@ -20,7 +20,8 @@ class TestCardKeyProviderCsv(unittest.TestCase):
                        "KIK3" : "00010204040506070809488B0C0D0E0F"}
 
         csv_file_path = os.path.dirname(os.path.abspath(__file__)) + "/test_card_key_provider.csv"
-        card_key_provider_register(CardKeyProviderCsv(csv_file_path, column_keys))
+        card_key_field_cryptor = CardKeyFieldCryptor(column_keys)
+        card_key_provider_register(CardKeyProviderCsv(csv_file_path, card_key_field_cryptor))
         super().__init__(*args, **kwargs)
 
     def test_card_key_provider_get(self):

tests/unittests/test_configurable_parameters.py

@@ -17,11 +17,10 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+import enum
 import io
 import sys
 import unittest
-import io
-import json
 from importlib import resources
 from osmocom.utils import hexstr
 from pySim.esim.saip import ProfileElementSequence
@@ -54,21 +53,16 @@ class ConfigurableParameterTest(unittest.TestCase):
     def test_parameters(self):
 
         upp_fnames = (
-                'SAIP2.1_gfmsuci.der',
-                'TS48v5_SAIP2.1B_NoBERTLV.der',
-                'TS48v5_SAIP2.3_NoBERTLV.der',
+                'TS48v5_SAIP2.1A_NoBERTLV.der',
+                'TS48v5_SAIP2.3_BERTLV_SUCI.der',
                )
 
         class Paramtest:
-            iff_present_default = False
-            def __init__(self, param_cls, val, expect_val, expect_clean_val=None, iff_present=None):
+            def __init__(self, param_cls, val, expect_val, expect_clean_val=None):
                 self.param_cls = param_cls
                 self.val = val
                 self.expect_clean_val = expect_clean_val
                 self.expect_val = expect_val
-                if iff_present is None:
-                    iff_present = Paramtest.iff_present_default
-                self.iff_present = iff_present
 
         param_tests = [
             Paramtest(param_cls=p13n.Imsi, val='123456',
@@ -154,7 +148,7 @@ class ConfigurableParameterTest(unittest.TestCase):
             Paramtest(param_cls=p13n.AlgorithmID,
                       val='usim-test',
                       expect_clean_val=3,
-                      expect_val='usim-test'),
+                      expect_val='usim_test'),
 
             Paramtest(param_cls=p13n.AlgorithmID,
                       val=1,
@@ -167,7 +161,7 @@ class ConfigurableParameterTest(unittest.TestCase):
             Paramtest(param_cls=p13n.AlgorithmID,
                       val=3,
                       expect_clean_val=3,
-                      expect_val='usim-test'),
+                      expect_val='usim_test'),
 
             Paramtest(param_cls=p13n.K,
                       val='01020304050607080910111213141516',
@@ -271,7 +265,6 @@ class ConfigurableParameterTest(unittest.TestCase):
                                  '11111111111111111111111111111111'
                                  '22222222222222222222222222222222'),
 
-
             Paramtest(param_cls=p13n.MncLen,
                       val='2',
                       expect_clean_val=2,
@@ -281,102 +274,7 @@ class ConfigurableParameterTest(unittest.TestCase):
                       expect_clean_val=3,
                       expect_val='3'),
 
-            Paramtest(param_cls=p13n.EuiccMandatoryServiceGetIdentity,
-                      val='mandatory',
-                      expect_clean_val=True,
-                      expect_val='mandatory'),
-            Paramtest(param_cls=p13n.EuiccMandatoryServiceGetIdentity,
-                      val='optional',
-                      expect_clean_val=False,
-                      expect_val='optional'),
-
-            Paramtest(param_cls=p13n.EuiccMandatoryServiceProfileA,
-                      val='mandatory',
-                      expect_clean_val=True,
-                      expect_val='mandatory'),
-            Paramtest(param_cls=p13n.EuiccMandatoryServiceProfileA,
-                      val='optional',
-                      expect_clean_val=False,
-                      expect_val='optional'),
-
-            Paramtest(param_cls=p13n.EuiccMandatoryServiceProfileB,
-                      val='mandatory',
-                      expect_clean_val=True,
-                      expect_val='mandatory'),
-            Paramtest(param_cls=p13n.EuiccMandatoryServiceProfileB,
-                      val='optional',
-                      expect_clean_val=False,
-                      expect_val='optional'),
-        ]
-
-        Paramtest.iff_present_default = True
-
-        sucici = {
-             "prot_scheme_id_list": [
-                 {"priority": 0, "identifier": 2, "key_index": 1},
-                 {"priority": 1, "identifier": 1, "key_index": 2},
-                 ],
-             "hnet_pubkey_list": [
-                 {"hnet_pubkey_identifier": 27,
-                  "hnet_pubkey": "0472da71976234ce833a6907425867b82e074d44ef907dfb4b3e21c1c2256ebcd15a7ded52fcbb097a4ed250e036c7b9c8c7004c4eedc4f068cd7bf8d3f900e3b4"},
-                 {"hnet_pubkey_identifier": 30,
-                  "hnet_pubkey": "5a8d38864820197c3394b92613b20b91633cbd897119273bf8e4a6f4eec0a650"},
-                  ],
-        }
-
-        param_tests.extend([
-            Paramtest(param_cls=p13n.SuciActive, val='SUCI-on',
-                      expect_clean_val=True,
-                      expect_val={'5G-SUCI-active': 'SUCI-on'}),
-            Paramtest(param_cls=p13n.SuciActive, val='SUCI-off',
-                      expect_clean_val=False,
-                      expect_val={'5G-SUCI-active': 'SUCI-off'}),
-
-            Paramtest(param_cls=p13n.SuciInUsim, val='SUCI-in-UE',
-                      expect_clean_val=False,
-                      expect_val={'5G-SUCI-in-USIM': 'SUCI-in-UE'}),
-            Paramtest(param_cls=p13n.SuciInUsim, val='SUCI-in-USIM',
-                      expect_clean_val=True,
-                      expect_val={'5G-SUCI-in-USIM': 'SUCI-in-USIM'}),
-
-            Paramtest(param_cls=p13n.SuciRi, val='123',
-                      expect_clean_val='123',
-                      expect_val={'5G-SUCI-RI': '123'}),
-            Paramtest(param_cls=p13n.SuciRi, val='0',
-                      expect_clean_val='0',
-                      expect_val={'5G-SUCI-RI': '0'}),
-            Paramtest(param_cls=p13n.SuciRi, val='9999',
-                      expect_clean_val='9999',
-                      expect_val={'5G-SUCI-RI': '9999'}),
-
-            Paramtest(param_cls=p13n.SuciCalcInfoUe,
-                      val=json.dumps(sucici),
-                      expect_clean_val=sucici,
-                      expect_val={'5G-SUCI-CalcInfo-UE': json.dumps(sucici)}),
-
-            Paramtest(param_cls=p13n.SuciCalcInfoUsim,
-                      val=json.dumps(sucici),
-                      expect_clean_val=sucici,
-                      expect_val={'5G-SUCI-CalcInfo-USIM': json.dumps(sucici)}),
-
-            Paramtest(param_cls=p13n.GfmSuciRi, val='123',
-                      expect_clean_val='123',
-                      expect_val={'GFM-5G-SUCI-RI': '123'}),
-            Paramtest(param_cls=p13n.GfmSuciRi, val='0',
-                      expect_clean_val='0',
-                      expect_val={'GFM-5G-SUCI-RI': '0'}),
-            Paramtest(param_cls=p13n.GfmSuciRi, val='9999',
-                      expect_clean_val='9999',
-                      expect_val={'GFM-5G-SUCI-RI': '9999'}),
-
-            Paramtest(param_cls=p13n.GfmSuciCalcInfoUe,
-                      val=json.dumps(sucici),
-                      expect_clean_val=sucici,
-                      expect_val={'GFM-5G-SUCI-CalcInfo-UE': json.dumps(sucici)}),
-
-        ])
-
-        Paramtest.iff_present_default = False
+            ]
 
         for sdkey_cls in (
             # thin out the number of tests, as a compromise between completeness and test runtime
@@ -419,14 +317,11 @@ class ConfigurableParameterTest(unittest.TestCase):
             p13n.SdKeyScp80Kvn03DesDek,
            #p13n.SdKeyScp80Kvn03DesEnc,
            #p13n.SdKeyScp80Kvn03DesMac,
-           #p13n.SdKeyScp81Kvn40AesDek,
-            p13n.SdKeyScp81Kvn40DesDek,
+            p13n.SdKeyScp81Kvn40AesDek,
            #p13n.SdKeyScp81Kvn40Tlspsk,
            #p13n.SdKeyScp81Kvn41AesDek,
-           #p13n.SdKeyScp81Kvn41DesDek,
             p13n.SdKeyScp81Kvn41Tlspsk,
            #p13n.SdKeyScp81Kvn42AesDek,
-           #p13n.SdKeyScp81Kvn42DesDek,
            #p13n.SdKeyScp81Kvn42Tlspsk,
             ):
 
@@ -472,8 +367,7 @@ class ConfigurableParameterTest(unittest.TestCase):
 
                 for t in param_tests:
                     test_idx += 1
-                    testlog = []
-                    testlog.append(f'{upp_fname} {t.param_cls.__name__}(val={valtypestr(t.val)})')
+                    logloc = f'{upp_fname} {t.param_cls.__name__}(val={valtypestr(t.val)})'
 
                     param = None
                     try:
@@ -481,32 +375,21 @@ class ConfigurableParameterTest(unittest.TestCase):
                         param.input_value = t.val
                         param.validate()
                     except ValueError as e:
-                        raise ValueError(f'{" ".join(testlog)}: {e}') from e
+                        raise ValueError(f'{logloc}: {e}') from e
 
                     clean_val = param.value
-                    testlog.append(f'clean_val={valtypestr(clean_val)}')
+                    logloc = f'{logloc} clean_val={valtypestr(clean_val)}'
                     if t.expect_clean_val is not None and t.expect_clean_val != clean_val:
-                        raise ValueError(f'{" ".join(testlog)}: expected'
+                        raise ValueError(f'{logloc}: expected'
                                          f' expect_clean_val={valtypestr(t.expect_clean_val)}')
 
                     # on my laptop, deepcopy is about 30% slower than decoding the DER from scratch:
                     # pes = copy.deepcopy(orig_pes)
                     pes = ProfileElementSequence.from_der(der)
-
-                    found = list((t.param_cls.get_value_from_pes(pes) or {}).values())
-                    testlog.append(f"previous value: {found}")
-
-                    if t.iff_present and not found:
-                        testlog.append("skipping, param not in template.")
-                        output = "\nskip: " + "\n  ".join(testlog)
-                        outputs.append(output)
-                        print(output)
-                        continue
-
                     try:
                         param.apply(pes)
                     except ValueError as e:
-                        raise ValueError(f'{" ".join(testlog)} apply_val(clean_val): {e}') from e
+                        raise ValueError(f'{logloc} apply_val(clean_val): {e}') from e
 
                     changed_der = pes.to_der()
 
@@ -524,18 +407,22 @@ class ConfigurableParameterTest(unittest.TestCase):
                     else:
                         read_back_val_type = f'{type(read_back_val).__name__}'
 
-                    testlog.append(f'read_back_val={valtypestr(read_back_val)}')
+                    logloc = (f'{logloc} read_back_val={valtypestr(read_back_val)}')
 
                     if isinstance(read_back_val, dict) and not t.param_cls.get_name() in read_back_val.keys():
-                        raise ValueError(f'{" ".join(testlog)}: expected to find name {t.param_cls.get_name()!r} in read_back_val')
+                        raise ValueError(f'{logloc}: expected to find name {t.param_cls.get_name()!r} in read_back_val')
 
                     expect_val = t.expect_val
                     if not isinstance(expect_val, dict):
                         expect_val = { t.param_cls.get_name(): expect_val }
                     if read_back_val != expect_val:
-                        raise ValueError(f'{" ".join(testlog)}: expected {expect_val=!r}:{type(t.expect_val).__name__}')
+                        raise ValueError(f'{logloc}: expected {expect_val=!r}:{type(t.expect_val).__name__}')
 
-                    output = "\nok: " + "\n  ".join(testlog)
+                    ok = logloc.replace(' clean_val', '\n\tclean_val'
+                              ).replace(' read_back_val', '\n\tread_back_val'
+                              ).replace('=', '=\t'
+                              )
+                    output = f'\nok: {ok}'
                     outputs.append(output)
                     print(output)
 
@@ -561,6 +448,191 @@ class ConfigurableParameterTest(unittest.TestCase):
                 raise RuntimeError(f'output differs from expected output at position {at}: "{output[at:at+20]}" != "{xo_str[at:at+20]}"')
 
 
+class TestValidateVal(unittest.TestCase):
+    """validate_val() tests for various ConfigurableParameter subclasses."""
+
+    def _ok(self, cls, val, expected=None):
+        result = cls.validate_val(val)
+        if expected is not None:
+            self.assertEqual(result, expected)
+        return result
+
+    def _err(self, cls, val):
+        with self.assertRaises(ValueError):
+            cls.validate_val(val)
+
+    # --- Iccid ---
+
+    def test_iccid_18digits_adds_luhn(self):
+        result = self._ok(p13n.Iccid, '998877665544332211')
+        self.assertIsInstance(result, str)
+        self.assertEqual(len(result), 19)
+        self.assertTrue(result.isdecimal())
+
+    def test_iccid_19digits_passthrough(self):
+        result = self._ok(p13n.Iccid, '9988776655443322110')
+        self.assertIsInstance(result, str)
+        self.assertEqual(len(result), 19)
+
+    def test_iccid_too_short(self):
+        self._err(p13n.Iccid, '12345678901234567')  # 17 digits
+
+    def test_iccid_too_long(self):
+        self._err(p13n.Iccid, '1' * 21)
+
+    def test_iccid_non_digits(self):
+        self._err(p13n.Iccid, '99887766554433221X')
+
+    # --- Imsi ---
+
+    def test_imsi_valid_short(self):
+        self._ok(p13n.Imsi, '001010', '001010')
+
+    def test_imsi_valid_long(self):
+        self._ok(p13n.Imsi, '001010123456789', '001010123456789')
+
+    def test_imsi_too_short(self):
+        self._err(p13n.Imsi, '12345')  # 5 digits, min is 6
+
+    def test_imsi_too_long(self):
+        self._err(p13n.Imsi, '1' * 16)
+
+    def test_imsi_non_digits(self):
+        self._err(p13n.Imsi, '00101A123456789')
+
+    # --- Pin1 ---
+
+    def test_pin1_4digits(self):
+        # DecimalHexParam encodes each digit as its ASCII byte, then rpad to 8 bytes with 0xff
+        self._ok(p13n.Pin1, '1234', b'1234\xff\xff\xff\xff')
+
+    def test_pin1_8digits(self):
+        self._ok(p13n.Pin1, '12345678', b'12345678')
+
+    def test_pin1_too_short(self):
+        self._err(p13n.Pin1, '123')
+
+    def test_pin1_too_long(self):
+        self._err(p13n.Pin1, '123456789')
+
+    def test_pin1_non_digits(self):
+        self._err(p13n.Pin1, '123A')
+
+    # --- Puk1 ---
+
+    def test_puk1_8digits(self):
+        self._ok(p13n.Puk1, '12345678', b'12345678')
+
+    def test_puk1_wrong_length(self):
+        self._err(p13n.Puk1, '1234567')   # 7 digits
+        self._err(p13n.Puk1, '123456789')  # 9 digits
+
+    def test_puk1_non_digits(self):
+        self._err(p13n.Puk1, '1234567X')
+
+    # --- K (BinaryParam) ---
+
+    def test_k_valid_hex_str(self):
+        self._ok(p13n.K, '000102030405060708090a0b0c0d0e0f',
+                 b'\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f')
+
+    def test_k_valid_bytes(self):
+        raw = bytes(range(16))
+        self._ok(p13n.K, raw, raw)
+
+    def test_k_wrong_length(self):
+        self._err(p13n.K, '00' * 15)  # 15 bytes, allow_len requires 16 or 32
+
+    def test_k_non_hex(self):
+        self._err(p13n.K, 'gg' * 16)
+
+    def test_k_odd_hex_digits(self):
+        self._err(p13n.K, '0' * 31)   # odd number of hex digits
+
+
+class TestEnumParam(unittest.TestCase):
+    """Tests for the EnumParam machinery, using AlgorithmID as the concrete subclass."""
+
+    # --- validate_val ---
+
+    def test_validate_by_name_exact(self):
+        self.assertEqual(p13n.AlgorithmID.validate_val('Milenage'), 1)
+        self.assertEqual(p13n.AlgorithmID.validate_val('TUAK'), 2)
+        self.assertEqual(p13n.AlgorithmID.validate_val('usim_test'), 3)
+
+    def test_validate_by_int(self):
+        self.assertEqual(p13n.AlgorithmID.validate_val(1), 1)
+        self.assertEqual(p13n.AlgorithmID.validate_val(2), 2)
+        self.assertEqual(p13n.AlgorithmID.validate_val(3), 3)
+
+    def test_validate_fuzzy_case(self):
+        self.assertEqual(p13n.AlgorithmID.validate_val('milenage'), 1)
+        self.assertEqual(p13n.AlgorithmID.validate_val('MILENAGE'), 1)
+        self.assertEqual(p13n.AlgorithmID.validate_val('tuak'), 2)
+
+    def test_validate_fuzzy_hyphen_underscore(self):
+        # 'usim-test' has a hyphen; enum member is 'usim_test' — must fuzzy-match
+        self.assertEqual(p13n.AlgorithmID.validate_val('usim-test'), 3)
+
+    def test_validate_invalid_name(self):
+        with self.assertRaises(ValueError):
+            p13n.AlgorithmID.validate_val('unknown')
+
+    def test_validate_invalid_int(self):
+        with self.assertRaises(ValueError):
+            p13n.AlgorithmID.validate_val(99)
+
+    def test_validate_returns_int(self):
+        result = p13n.AlgorithmID.validate_val('Milenage')
+        self.assertIsInstance(result, int)
+        self.assertNotIsInstance(result, enum.Enum)
+
+    # --- map_name_to_val ---
+
+    def test_map_name_exact(self):
+        self.assertEqual(p13n.AlgorithmID.map_name_to_val('Milenage'), 1)
+
+    def test_map_name_fuzzy(self):
+        self.assertEqual(p13n.AlgorithmID.map_name_to_val('milenage'), 1)
+        self.assertEqual(p13n.AlgorithmID.map_name_to_val('usim-test'), 3)
+
+    def test_map_name_strict_raises(self):
+        with self.assertRaises(ValueError):
+            p13n.AlgorithmID.map_name_to_val('unknown', strict=True)
+
+    def test_map_name_nonstrict_returns_none(self):
+        self.assertIsNone(p13n.AlgorithmID.map_name_to_val('unknown', strict=False))
+
+    # --- map_val_to_name ---
+
+    def test_map_val_known(self):
+        self.assertEqual(p13n.AlgorithmID.map_val_to_name(1), 'Milenage')
+        self.assertEqual(p13n.AlgorithmID.map_val_to_name(2), 'TUAK')
+        self.assertEqual(p13n.AlgorithmID.map_val_to_name(3), 'usim_test')
+
+    def test_map_val_unknown_nonstrict(self):
+        self.assertIsNone(p13n.AlgorithmID.map_val_to_name(99))
+
+    def test_map_val_unknown_strict(self):
+        with self.assertRaises(ValueError):
+            p13n.AlgorithmID.map_val_to_name(99, strict=True)
+
+    # --- name_normalize ---
+
+    def test_name_normalize(self):
+        self.assertEqual(p13n.AlgorithmID.name_normalize('Milenage'), 'Milenage')
+        self.assertEqual(p13n.AlgorithmID.name_normalize('milenage'), 'Milenage')
+        self.assertEqual(p13n.AlgorithmID.name_normalize('usim-test'), 'usim_test')
+
+    # --- clean_name_str ---
+
+    def test_clean_name_str(self):
+        self.assertEqual(p13n.AlgorithmID.clean_name_str('usim-test'), 'usimtest')
+        self.assertEqual(p13n.AlgorithmID.clean_name_str('usim_test'), 'usimtest')
+        self.assertEqual(p13n.AlgorithmID.clean_name_str('Milenage'), 'milenage')
+        self.assertEqual(p13n.AlgorithmID.clean_name_str('foo bar!'), 'foobar')
+
+
 if __name__ == "__main__":
     if '-u' in sys.argv:
         update_expected_output = True

tests/unittests/test_decode_select_resp.py

@@ -0,0 +1,78 @@
+#!/usr/bin/env python3
+
+# (C) 2026 by sysmocom - s.f.m.c. GmbH
+# All Rights Reserved
+#
+# Author: Philipp Maier <pmaier@sysmocom.de>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import unittest
+import os
+from pySim.profile import CardProfile
+from pySim.ts_51_011 import CardProfileSIM
+from pySim.ts_102_221 import CardProfileUICC
+
+class TestDecodeSelectResponse_CardProfile(unittest.TestCase):
+
+    def decode_select_response(self, card_Profile: CardProfile, testcases: list[dict]):
+        for testcase in testcases:
+            resp_hex = testcase['resp_hex']
+            decoded = card_Profile.decode_select_response(resp_hex)
+            if testcase['decoded']:
+                self.assertEqual(decoded, testcase['decoded'])
+            else:
+                print("no testvector to compare against, assuming the following output is correct:")
+                print("resp_hex:", resp_hex)
+                print("decoded:", decoded)
+
+    def test_CardProfileSIM(self):
+        testcases = [
+            # MF
+            {"resp_hex" : "000000003f000100000000000981020c0400838a838a",
+             "decoded" : {'file_descriptor': {'file_descriptor_byte': {'file_type': 'mf'}}, 'proprietary_info': {'available_memory': 0}, 'file_id': '3f00', 'file_characteristics': '81', 'num_direct_child_df': 2, 'num_direct_child_ef': 12, 'num_chv_unblock_adm_codes': 4}},
+            # DF.TELECOM
+            {"resp_hex" : "000000007f100200000000000981000d0400838a838a",
+             "decoded" : {'file_descriptor': {'file_descriptor_byte': {'file_type': 'df'}}, 'proprietary_info': {'available_memory': 0}, 'file_id': '7f10', 'file_characteristics': '81', 'num_direct_child_df': 0, 'num_direct_child_ef': 13, 'num_chv_unblock_adm_codes': 4}},
+            # EF.MSISDN
+            {"resp_hex" : "000000346f40040011ffff0102011a",
+             "decoded" : {'file_descriptor': {'file_descriptor_byte': {'file_type': 'working_ef', 'structure': 'linear_fixed'}, 'record_len': 26, 'num_of_rec': 2}, 'proprietary_info': {}, 'file_id': '6f40', 'file_size': 52, 'access_conditions': '11ffff', 'life_cycle_status_int': 'creation'}},
+            # EF.ICCID
+            {"resp_hex" : "0000000a2fe204000cffff01020000",
+             "decoded" : {'file_descriptor': {'file_descriptor_byte': {'file_type': 'working_ef', 'structure': 'transparent'}}, 'proprietary_info': {}, 'file_id': '2fe2', 'file_size': 10, 'access_conditions': '0cffff', 'life_cycle_status_int': 'creation'}},
+        ]
+        self.decode_select_response(CardProfileSIM, testcases)
+
+    def test_CardProfileUICC(self):
+        testcases = [
+            # MF
+            {"resp_hex" : "622c8202782183023f00a50c80017183040003a7388701018a01058b032f0601c60c90016083010183010a83010b",
+             "decoded" : {'file_descriptor': {'file_descriptor_byte': {'shareable': True, 'file_type': 'df', 'structure': 'no_info_given'}, 'record_len': None, 'num_of_rec': None}, 'file_identifier': b'?\x00', 'proprietary_information': {'uicc_characteristics': b'q', 'available_memory': 239416, 'supported_filesystem_commands': {'terminal_capability': True}}, 'life_cycle_status_integer': 'operational_activated', 'security_attrib_referenced': {'ef_arr_file_id': b'/\x06', 'ef_arr_record_nr': 1}, 'pin_status_template_do': [{'ps_do': b'`'}, {'key_reference': 1}, {'key_reference': 10}, {'key_reference': 11}]}},
+            # ADF.USIM
+            {"resp_hex" : "623d8202782183027fd0840ca0000000871002ff49ff0589a50c80017183040003a7388701018a01058b032f0601c60f90017083010183018183010a83010b",
+             "decoded" : {'file_descriptor': {'file_descriptor_byte': {'shareable': True, 'file_type': 'df', 'structure': 'no_info_given'}, 'record_len': None, 'num_of_rec': None}, 'file_identifier': b'\x7f\xd0', 'df_name': b'\xa0\x00\x00\x00\x87\x10\x02\xffI\xff\x05\x89', 'proprietary_information': {'uicc_characteristics': b'q', 'available_memory': 239416, 'supported_filesystem_commands': {'terminal_capability': True}}, 'life_cycle_status_integer': 'operational_activated', 'security_attrib_referenced': {'ef_arr_file_id': b'/\x06', 'ef_arr_record_nr': 1}, 'pin_status_template_do': [{'ps_do': b'p'}, {'key_reference': 1}, {'key_reference': 129}, {'key_reference': 10}, {'key_reference': 11}]}},
+            # ADF.ISIM
+            {"resp_hex" : "623d8202782183027fb0840ca0000000871004ff49ff0589a50c80017183040003a7388701018a01058b032f0601c60f90017083010183018183010a83010b",
+             "decoded" : {'file_descriptor': {'file_descriptor_byte': {'shareable': True, 'file_type': 'df', 'structure': 'no_info_given'}, 'record_len': None, 'num_of_rec': None}, 'file_identifier': b'\x7f\xb0', 'df_name': b'\xa0\x00\x00\x00\x87\x10\x04\xffI\xff\x05\x89', 'proprietary_information': {'uicc_characteristics': b'q', 'available_memory': 239416, 'supported_filesystem_commands': {'terminal_capability': True}}, 'life_cycle_status_integer': 'operational_activated', 'security_attrib_referenced': {'ef_arr_file_id': b'/\x06', 'ef_arr_record_nr': 1}, 'pin_status_template_do': [{'ps_do': b'p'}, {'key_reference': 1}, {'key_reference': 129}, {'key_reference': 10}, {'key_reference': 11}]}},
+            # EF.IMSI
+            {"resp_hex" : "62178202412183026f078a01058b036f060a80020009880138",
+             "decoded" : {'file_descriptor': {'file_descriptor_byte': {'shareable': True, 'file_type': 'working_ef', 'structure': 'transparent'}, 'record_len': None, 'num_of_rec': None}, 'file_identifier': b'o\x07', 'life_cycle_status_integer': 'operational_activated', 'security_attrib_referenced': {'ef_arr_file_id': b'o\x06', 'ef_arr_record_nr': 10}, 'file_size': 9, 'short_file_identifier': 7}},
+            # EF.ECC
+            {"resp_hex" : "621a82054221000e0283026fb78a01058b036f06088002001c880108",
+             "decoded" : {'file_descriptor': {'file_descriptor_byte': {'shareable': True, 'file_type': 'working_ef', 'structure': 'linear_fixed'}, 'record_len': 14, 'num_of_rec': 2}, 'file_identifier': b'o\xb7', 'life_cycle_status_integer': 'operational_activated', 'security_attrib_referenced': {'ef_arr_file_id': b'o\x06', 'ef_arr_record_nr': 8}, 'file_size': 28, 'short_file_identifier': 1}},
+        ]
+        self.decode_select_response(CardProfileUICC, testcases)
+
+if __name__ == "__main__":
+	unittest.main()

tests/unittests/test_globalplatform.py

@@ -17,7 +17,10 @@
 
 import unittest
 import logging
+import hashlib
+from types import SimpleNamespace
 from osmocom.utils import b2h, h2b
+from osmocom.tlv import bertlv_encode_len
 
 from pySim.global_platform import *
 from pySim.global_platform.scp import *
@@ -283,6 +286,41 @@ class SCP03_Test_AES256_33(SCP03_Test, unittest.TestCase):
 # FIXME: test auth with random (0x60) vs pseudo-random (0x70) challenge
 
 
+class KeyComponentBlock_Test(unittest.TestCase):
+    """Tests for the kcb of GP CardSpec v2.3
+    - Table 11-70 kcv that required padding, preceded by its clear-text length
+    - Table 11-71 no padding required"""
+
+    def setUp(self):
+        # SCP02 (3DES DEK, 8 byte blocks), same vectors as SCP02_Test
+        self.scp02 = SCP02(card_keys=ck_3des_70)
+        self.scp02.gen_init_update_apdu(host_challenge=h2b('40A62C37FA6304F8'))
+        self.scp02.parse_init_update_resp(h2b('00000000000000000000700200016B4524ABEE7CF32EA3838BC148F3'))
+        self.scp02.gen_ext_auth_apdu()
+        # SCP03 (AES DEK, 16 byte blocks), same vectors as SCP03_Test_AES128_11
+        self.scp03 = SCP03(card_keys=KEYSET_AES128)
+        self.scp03.gen_init_update_apdu(h2b('b13e5f938fc108c4'))
+        self.scp03.parse_init_update_resp(h2b('000000000000000000003003703eb51047495b249f66c484c1d2ef1948000002'))
+        self.scp03.gen_ext_auth_apdu(0x11)
+
+    def test_encrypt_decrypt_key(self):
+        for scp in (self.scp02, self.scp03):
+            bs = scp.sk.blocksize
+            for keylen in range(1, 3 * bs + 1):
+                with self.subTest(scp=type(scp).__name__, keylen=keylen):
+                    key = bytes(range(keylen))
+                    kcb = scp.encrypt_key(key)
+                    if keylen % bs:
+                        # Table 11-70: <length of clear key component> || <encrypted padded value>
+                        self.assertEqual(kcb[0], keylen)
+                        self.assertEqual((len(kcb) - 1) % bs, 0)
+                        self.assertEqual(len(kcb) - 1, keylen + (bs - keylen % bs))
+                    else:
+                        # Table 11-71: only the encrypted key component value
+                        self.assertEqual(len(kcb), keylen)
+                    self.assertEqual(scp.decrypt_key(kcb), key)
+
+
 class SCP03_KCV_Test(unittest.TestCase):
     def test_kcv(self):
         self.assertEqual(compute_kcv('aes', KEYSET_AES128.enc), h2b('C35280'))
@@ -290,6 +328,208 @@ class SCP03_KCV_Test(unittest.TestCase):
         self.assertEqual(compute_kcv('aes', KEYSET_AES128.dek), h2b('840DE5'))
 
 
+class PutKey_PSK_Test(unittest.TestCase):
+    """Tests for the PUT KEY command data field encoding, in particular the PSK TLS ('85') key data
+    field defined by GlobalPlatform Amendment B (Remote Application Management over HTTP) Table 3-13."""
+
+    # the PUT KEY encoder we exercise
+    C = ADF_SD.AddlShellCommands
+
+    # SCP80 TLS-PSK example key from the do_put_key docstring (16 bytes)
+    PSK_CLEAR = h2b('303132333435363738393a3b3c3d3e3f')
+    # its DEK ciphertext + Table 3-13 KCV with SCP02 session set up below
+    PSK_CIPHERED = h2b('15abf1fe16ccc5aa13743394442942cd')
+    PSK_KCV = h2b('06125d')  # = SHA-1(PSK_CLEAR)[:3]
+
+    def setUp(self):
+        # SCP02 with the same vectors as SCP02_Test, so that the whole PUT KEY data field is reproducible.
+        self.scp02 = SCP02(card_keys=ck_3des_70)
+        self.scp02.gen_init_update_apdu(host_challenge=h2b('40A62C37FA6304F8'))
+        self.scp02.parse_init_update_resp(h2b('00000000000000000000700200016B4524ABEE7CF32EA3838BC148F3'))
+        self.scp02.gen_ext_auth_apdu()
+
+    def test_psk_kcv_is_sha1(self):
+        # GP Amendment B Table 3-13: KCV = 3 most significant bytes of SHA-1(clear key)
+        self.assertEqual(compute_kcv('tls_psk', self.PSK_CLEAR), hashlib.sha1(self.PSK_CLEAR).digest()[:3])
+        self.assertEqual(compute_kcv('tls_psk', self.PSK_CLEAR), self.PSK_KCV)
+
+    def test_encode_psk_framing_golden(self):
+        # assert the exact Table 3-13 layout
+        #   85 | L1 | L2 | <ciphered> | 03 | <SHA-1(clear)[:3]>
+        clear = self.PSK_CLEAR
+        ciphered = h2b('aabbccddeeff00112233445566778899')  # arbitrary 16-byte ciphertext
+        kcv = hashlib.sha1(clear).digest()[:3]
+        field = self.C.encode_key_data_psk(clear, ciphered, kcv)
+        #                            85   L1   L2   <---------- ciphered ----------->  03  <-kcv->
+        self.assertEqual(b2h(field),'85' '11' '10' 'aabbccddeeff00112233445566778899' '03' + b2h(kcv))
+        self.assertEqual(b2h(field),'851110aabbccddeeff0011223344556677889903' + '06125d')
+
+    def test_psk_golden_over_scp02(self):
+        # Full PUT KEY data field (KVN 0x40 + single PSK key) enciphered with the SCP02 DEK.
+        keys = [{'key_type': 'tls_psk', 'clear_key': self.PSK_CLEAR,
+                 'kcv': compute_kcv('tls_psk', self.PSK_CLEAR)}]
+        data = self.C.build_put_key_data(0x40, keys, self.scp02)
+        self.assertEqual(b2h(data),
+                         '40' '85' '11' '10' + b2h(self.PSK_CIPHERED) + '03' + b2h(self.PSK_KCV))
+
+    def test_wrong_basic_format_differs(self):
+        # regression test, the generic "Basic format" does NOT match Table 3-13 for a PSK key
+        # rejected by card with with 6a88
+        wrong_basic = self.C.encode_key_data_basic('tls_psk', self.PSK_CIPHERED, b'')
+        right_psk = self.C.encode_key_data_psk(self.PSK_CLEAR, self.PSK_CIPHERED, self.PSK_KCV)
+        self.assertEqual(b2h(wrong_basic), '8510' + b2h(self.PSK_CIPHERED) + '00')
+        self.assertEqual(b2h(right_psk), '8511' '10' + b2h(self.PSK_CIPHERED) + '03' + b2h(self.PSK_KCV))
+        self.assertNotEqual(wrong_basic, right_psk)
+
+    def test_key_component_block_length_is_bertlv(self):
+        # GP CardSpec v2.3.1 Section 11.8.2.3.1: all lengths ofPUT KEY are always BER TLV coded
+        for kcb_len, exp_len_field in [(127, '7f'), (128, '8180'), (129, '8181'), (256, '820100')]:
+            with self.subTest(kcb_len=kcb_len):
+                kcb = bytes(kcb_len)
+                field = self.C.encode_key_data_basic('rsa_modulus_n', kcb, b'')
+                self.assertEqual(b2h(field), 'a2' + exp_len_field + b2h(kcb) + '00')
+                # 85 field of Amendment B Table 3-13 uses the same coding
+                # single byte inner length (clear key < 128) == block kcb_len bytes long
+                psk = self.C.encode_key_data_psk(bytes(120), bytes(kcb_len - 1), b'')
+                self.assertEqual(b2h(psk)[:2 + len(exp_len_field)], '85' + exp_len_field)
+
+    def test_basic_format_unchanged(self):
+        # as before
+        for kt, clear in [('des', h2b('404142434445464748494a4b4c4d4e4f')),
+                          ('aes', h2b('000102030405060708090a0b0c0d0e0f'))]:
+            ciph = self.scp02.encrypt_key(clear)
+            kcv = compute_kcv(kt, clear)
+            via_construct = build_construct(self.C.KeyDataBasic, {'key_type': kt, 'kcb': b2h(ciph), 'kcv': b2h(kcv)})
+            via_helper = self.C.encode_key_data_basic(kt, ciph, kcv)
+            self.assertEqual(via_helper, via_construct)
+
+    def test_psk_padding_no_double_length(self):
+        # A PSK key whose length is not a multiple of the DEK block size (DES: 8) is right-padded before
+        # ciphering. Table 3-13 states the clear key length (L2) in the '85' DO itself, so the ciphered
+        # key field is the bare cryptogram:
+        # - ciphered field == padded ciphertext (no duplicated length prefix),
+        # - clear key == first L2 bytes.
+        for keylen in (18, 20):
+            with self.subTest(keylen=keylen):
+                clear = bytes(range(keylen))
+                padded_len = keylen + (-keylen % 8)
+                field = self.C.build_put_key_data(0x40, [{'key_type': 'tls_psk', 'clear_key': clear,
+                                                          'kcv': compute_kcv('tls_psk', clear)}], self.scp02)[1:]
+                self.assertEqual(field[0], 0x85)
+                l1 = field[1]
+                l2 = field[2]
+                self.assertEqual(l2, keylen)                # single-byte BER length of clear key
+                ciphered = field[3:3 + (l1 - 1)]            # value = L2 (1 byte) || ciphered key
+                self.assertEqual(len(ciphered), padded_len) # padded to the 8-byte DES block size
+                self.assertEqual(l1, 1 + padded_len)        # no duplicated length prefix
+                self.assertEqual(self.scp02.dek_decrypt(ciphered)[:keylen], clear)
+
+    def test_psk_clear_key_is_not_padded_in_place(self):
+        # padding the bytearray in place would make L2 the padded length,
+        # then stored as key material and rejected thanks to the KCV
+        clear = h2b('000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d')  # 30, not %8
+        kcv = compute_kcv('tls_psk', clear)
+        field = self.C.build_put_key_data(0x40, [{'key_type': 'tls_psk', 'clear_key': clear,
+                                                  'kcv': kcv}], self.scp02)[1:]
+        self.assertEqual(len(clear), 30)
+        self.assertEqual(field[2], 30)                       # L2 == clear key length, not 32
+        self.assertEqual(self.scp02.dek_decrypt(field[3:3 + field[1] - 1])[:30], clear)
+
+    def test_kcv_suppressed(self):
+        # --suppress-key-check -> KCV length 00 and no KCV bytes
+        field = self.C.build_put_key_data(0x40, [{'key_type': 'tls_psk', 'clear_key': self.PSK_CLEAR,
+                                                  'kcv': b''}], self.scp02)[1:]
+        self.assertEqual(b2h(field), '8511' '10' + b2h(self.PSK_CIPHERED) + '00')
+
+    def test_multikey_psk_plus_des_dek(self):
+        # load a PSK TLS key (KID 1, Amendment B format) together with its DES DEK
+        # (KID 2, Basic format) in one PUT KEY.
+        # Verify the concatenated data field parses back into the two components with proper type formats.
+        dek = h2b('404142434445464748494a4b4c4d4e4f')
+        keys = [{'key_type': 'tls_psk', 'clear_key': self.PSK_CLEAR, 'kcv': compute_kcv('tls_psk', self.PSK_CLEAR)},
+                {'key_type': 'des', 'clear_key': dek, 'kcv': compute_kcv('des', dek)}]
+        data = self.C.build_put_key_data(0x40, keys, self.scp02)
+
+        b = data
+        self.assertEqual(b[0], 0x40)            # KVN
+        b = b[1:]
+        # component 1: PSK TLS (Table 3-13)
+        self.assertEqual(b[0], 0x85)
+        self.assertEqual(b[1], 0x11)            # L1 = 17
+        self.assertEqual(b[2], 0x10)            # L2 = 16 (clear key length)
+        self.assertEqual(b[3:3 + 16], self.PSK_CIPHERED)
+        self.assertEqual(b[3 + 16], 0x03)       # KCV length
+        self.assertEqual(b[3 + 16 + 1:3 + 16 + 1 + 3], self.PSK_KCV)
+        b = b[3 + 16 + 1 + 3:]
+        # component 2: DES DEK (Basic format)
+        self.assertEqual(b[0], 0x80)            # key type des
+        kcb_len = b[1]
+        self.assertEqual(kcb_len, 16)
+        self.assertEqual(b[2:2 + kcb_len], self.scp02.encrypt_key(dek))
+        b = b[2 + kcb_len:]
+        self.assertEqual(b[0], 0x03)            # KCV length
+        self.assertEqual(b[1:1 + 3], compute_kcv('des', dek))
+        self.assertEqual(b[1 + 3:], b'')        # no trailing bytes
+
+    def test_no_scp_leaves_key_clear(self):
+        # During personalization (no SCP) the key is not enciphered, framing still follows Table 3-13.
+        field = self.C.build_put_key_data(0x40, [{'key_type': 'tls_psk', 'clear_key': self.PSK_CLEAR,
+                                                  'kcv': self.PSK_KCV}], None)[1:]
+        self.assertEqual(b2h(field), '8511' '10' + b2h(self.PSK_CLEAR) + '03' + b2h(self.PSK_KCV))
+
+
+class PutKey_Length_Test(unittest.TestCase):
+    """Tests for the length of the PUT KEY command APDU.  Lc of GP CardSpec v2.3 Table 11-64 is a
+    single byte, so an oversized key data field cannot be sent."""
+
+    class PutKeyOnly(ADF_SD.AddlShellCommands):
+        """ADF_SD.AddlShellCommands with a canned scc to drive put_key()"""
+        def __init__(self, scp=None, max_cmd_len=255):
+            super().__init__()
+            self.sent = []
+            self.scc = SimpleNamespace(scp=scp, max_cmd_len=max_cmd_len,
+                                       send_apdu_checksw=lambda pdu: (self.sent.append(pdu), ('', '9000'))[1])
+
+        @property
+        def _cmd(self):
+            return SimpleNamespace(lchan=SimpleNamespace(scc=self.scc))
+
+    # KVN, key type, two byte BER length of the key component block, KCV length; KCV suppressed
+    FRAMING = 1 + 1 + 2 + 1
+
+    @staticmethod
+    def key(nbytes: int):
+        return [{'key_type': 'rsa_modulus_n', 'clear_key': bytes(nbytes), 'kcv': b''}]
+
+    def test_lc_matches_data_field(self):
+        # largest key component block that still fits without a secure channel
+        sd = self.PutKeyOnly()
+        sd.put_key(0, 0x40, 1, self.key(255 - self.FRAMING))
+        apdu = sd.sent[0]
+        self.assertEqual(apdu[:8], '80D80001')
+        lc = int(apdu[8:10], 16)
+        self.assertEqual(lc, 255)                       # Lc ...
+        self.assertEqual(len(apdu[10:-2]) // 2, lc)     # ... and it matches the actual data field
+
+    def test_oversized_key_data_raises(self):
+        # real world fat example: RSA-2048 modulus does not fit, led to 3 nibble Lc 106,
+        # which silently shifted and broke the whole APDU by half a byte.
+        sd = self.PutKeyOnly()
+        with self.assertRaises(ValueError) as ctx:
+            sd.put_key(0, 0x40, 1, self.key(256))
+        self.assertIn('262', str(ctx.exception))
+        self.assertIn('255', str(ctx.exception))
+        self.assertEqual(sd.sent, [])                   # nothing was sent to the card
+
+    def test_secure_channel_overhead_lowers_the_limit(self):
+        # scc.max_cmd_len shrinks by the C-MAC + encryption padding of active SCP
+        sd = self.PutKeyOnly(max_cmd_len=239)
+        sd.put_key(0, 0x40, 1, self.key(239 - self.FRAMING))
+        self.assertEqual(int(sd.sent[0][8:10], 16), 239)
+        with self.assertRaises(ValueError):
+            sd.put_key(0, 0x40, 1, self.key(239 - self.FRAMING + 1))
+
+
 class Install_param_Test(unittest.TestCase):
     def test_gen_install_parameters(self):
         load_parameters = gen_install_parameters(256, 256, '010001001505000000000000000000000000')
@@ -298,5 +538,180 @@ class Install_param_Test(unittest.TestCase):
         load_parameters = gen_install_parameters()
         self.assertEqual(load_parameters, 'c900')
 
+class SCP_Overhead_Test(unittest.TestCase):
+    """SCP.overhead varies according to the current security level:
+    C-MAC + at level >= 3 the worst-case padding!
+    """
+
+    def _scp02(self, security_level):
+        scp = SCP02(card_keys=ck_3des_70)
+        scp.sk = Scp02SessionKeys(0x0001, ck_3des_70)
+        scp.security_level = security_level
+        return scp
+
+    def _scp03(self, security_level, s_mode=8):
+        scp = SCP03(card_keys=KEYSET_AES128, s_mode=s_mode)
+        scp.sk = Scp03SessionKeys(KEYSET_AES128, b'\x00' * s_mode, b'\x11' * s_mode)
+        scp.security_level = security_level
+        return scp
+
+    def test_scp02(self):
+        self.assertEqual(self._scp02(0x00).overhead, 0)   # no wrapping at all
+        self.assertEqual(self._scp02(0x01).overhead, 8)   # C-MAC
+        self.assertEqual(self._scp02(0x03).overhead, 16)  # C-MAC + C-DEC: pad80 to 8, largest fit 239
+
+    def test_scp03_s8(self):
+        self.assertEqual(self._scp03(0x00).overhead, 0)
+        self.assertEqual(self._scp03(0x01).overhead, 8)
+        self.assertEqual(self._scp03(0x03).overhead, 16)  # pad80 to 16 within 247 -> 240, minus pad byte
+        self.assertEqual(self._scp03(0x33).overhead, 16)  # R-MAC/R-ENC add no *command* overhead
+
+    def test_scp03_s16(self):
+        self.assertEqual(self._scp03(0x01, s_mode=16).overhead, 16)
+        self.assertEqual(self._scp03(0x03, s_mode=16).overhead, 32)  # pad80 to 16 within 239 -> 224, minus pad byte
+
+
+class SCP_Lc_Limit_Test_Base(unittest.TestCase):
+    """Test wrap_cmd_apdu() boundary handling: data of (255 - overhead) must produce Lc <= 255 else ValueError"""
+
+    def _load_apdu(self, data_len):
+        return h2b('80E80000') + bytes([data_len]) + b'\xa5' * data_len
+
+    def _check_boundary(self, scp):
+        fits = 255 - scp.overhead
+        wrapped = scp.wrap_cmd_apdu(self._load_apdu(fits))
+        self.assertLessEqual(wrapped[4], 255)
+        self.assertEqual(len(wrapped), 5 + wrapped[4])  # case #3: header + Lc bytes, no Le
+        with self.assertRaises(ValueError) as ctx:
+            scp.wrap_cmd_apdu(self._load_apdu(fits + 1))
+        self.assertIn('Lc', str(ctx.exception))
+
+
+class SCP02_Lc_Limit_Test(SCP_Lc_Limit_Test_Base):
+    """Same session vectors as SCP02_Auth_Test"""
+
+    def setUp(self):
+        self.scp02 = SCP02(card_keys=ck_3des_70)
+        self.scp02.gen_init_update_apdu(host_challenge=h2b('40A62C37FA6304F8'))
+        self.scp02.parse_init_update_resp(h2b('00000000000000000000700200016B4524ABEE7CF32EA3838BC148F3'))
+        self.scp02.gen_ext_auth_apdu()
+
+    def test_cmac_only(self):
+        self.scp02.security_level = 0x01
+        self._check_boundary(self.scp02)  # 247 fits, 248 raises
+
+    def test_cmac_cdec(self):
+        self.scp02.security_level = 0x03
+        self._check_boundary(self.scp02)  # 239 fits (-> Lc 248), 240 raises (would be 256)
+
+    def test_cmac_cdec_wrapped_lc(self):
+        # my actual failing case: 240 bytes at level 3
+        self.scp02.security_level = 0x03
+        wrapped = self.scp02.wrap_cmd_apdu(self._load_apdu(239))
+        self.assertEqual(wrapped[4], 248)  # 239 -> pad80 -> 240 ciphertext + 8 mac
+
+
+class SCP03_Lc_Limit_Test(SCP_Lc_Limit_Test_Base):
+    """Session keys derived directly"""
+
+    def _scp03(self, security_level, s_mode):
+        scp = SCP03(card_keys=KEYSET_AES128, s_mode=s_mode)
+        scp.sk = Scp03SessionKeys(KEYSET_AES128, b'\x00' * s_mode, b'\x11' * s_mode)
+        scp.security_level = security_level
+        return scp
+
+    def test_s8_cmac_only(self):
+        self._check_boundary(self._scp03(0x01, 8))    # 247 fits, 248 raises
+
+    def test_s8_cmac_cdec(self):
+        self._check_boundary(self._scp03(0x03, 8))    # 239 fits, 240 raises
+
+    def test_s16_cmac_only(self):
+        self._check_boundary(self._scp03(0x01, 16))   # 239 fits, 240 raises
+
+    def test_s16_cmac_cdec(self):
+        self._check_boundary(self._scp03(0x03, 16))   # 223 fits, 224 raises
+
+
+class _FakeSccForLoad:
+    """mock lchan.scc: records LOAD APDUs, optionally wrapping them through a real SCP
+    instance first where the Lc overflow used to blow up"""
+
+    def __init__(self, max_cmd_len=255, scp=None):
+        self.max_cmd_len = max_cmd_len
+        self.scp = scp
+        self.sent = []
+        self.wrapped = []
+
+    def send_apdu_checksw(self, apdu, sw='9000'):
+        self.sent.append(apdu.lower())
+        if self.scp:
+            self.wrapped.append(self.scp.wrap_cmd_apdu(h2b(apdu)))
+        return ('', '9000')
+
+
+class Load_ChunkLen_Test(unittest.TestCase):
+    """ADF_SD.load() chunking: block size must use scc.max_cmd_len"""
+
+    payload = b'\xaa' * 500  # actual real world case LOAD TLV: C4 + 8201f4 + 500 = 504 total
+
+    def _sd(self, scc):
+        cmd = type('_Cmd', (), {'lchan': type('_Lchan', (), {'scc': scc})(),
+                                'poutput': lambda self, *args: None})()
+        # cmd2 CommandSet has a r/o _cmd property -> shadow it
+        _SD = type('_SD', (ADF_SD.AddlShellCommands,), {'_cmd': cmd})
+        return _SD.__new__(_SD)
+
+    def _blocks(self, scc):
+        """Get (p1, p2, lc) from LOAD APDU"""
+        for apdu in scc.sent:
+            self.assertEqual(apdu[0:4], '80e8')
+            yield int(apdu[4:6], 16), int(apdu[6:8], 16), int(apdu[8:10], 16)
+
+    def test_default_no_scp(self):
+        """Without SCP the old 240 byte block size is kept, no idea what else might rely on this number"""
+        scc = _FakeSccForLoad(max_cmd_len=255)
+        self._sd(scc).load(self.payload)
+        blocks = list(self._blocks(scc))
+        self.assertEqual([b[2] for b in blocks], [240, 240, 24])
+        self.assertEqual([b[0] for b in blocks], [0x00, 0x00, 0x80])  # P1: last block flagged
+        self.assertEqual([b[1] for b in blocks], [0, 1, 2])           # P2: block num
+
+    def test_default_scp02_level3(self):
+        """max_cmd_len 239 (SCP02 lvl 3) squeezes the blocks"""
+        scc = _FakeSccForLoad(max_cmd_len=239)
+        self._sd(scc).load(self.payload)
+        self.assertEqual([b[2] for b in list(self._blocks(scc))], [239, 239, 26])
+
+    def test_explicit_chunk_len(self):
+        scc = _FakeSccForLoad(max_cmd_len=255)
+        self._sd(scc).load(self.payload, chunk_len=100)
+        self.assertEqual([b[2] for b in list(self._blocks(scc))], [100] * 5 + [4])
+
+    def test_explicit_chunk_len_too_large(self):
+        scc = _FakeSccForLoad(max_cmd_len=239)
+        with self.assertRaises(ValueError):
+            self._sd(scc).load(self.payload, chunk_len=240)
+        self.assertEqual(scc.sent, [])  # nothing sent!
+
+    def test_explicit_chunk_len_zero(self):
+        scc = _FakeSccForLoad(max_cmd_len=255)
+        with self.assertRaises(ValueError):
+            self._sd(scc).load(self.payload, chunk_len=0)
+
+    def test_end_to_end_scp02_level3(self):
+        """original failure: 286 byte CAP + SCP02 lvl 3"""
+        scp02 = SCP02(card_keys=ck_3des_70)
+        scp02.gen_init_update_apdu(host_challenge=h2b('40A62C37FA6304F8'))
+        scp02.parse_init_update_resp(h2b('00000000000000000000700200016B4524ABEE7CF32EA3838BC148F3'))
+        scp02.gen_ext_auth_apdu()
+        scp02.security_level = 0x03
+        scc = _FakeSccForLoad(max_cmd_len=255 - scp02.overhead, scp=scp02)
+        self._sd(scc).load(b'\x5a' * 286)
+        self.assertEqual(len(scc.sent), 2)  # 289 byte TLV in blocks of 239
+        for wrapped in scc.wrapped:
+            self.assertLessEqual(wrapped[4], 255)
+
+
 if __name__ == "__main__":
 	unittest.main()

tests/unittests/test_log.py

@@ -37,6 +37,17 @@ expected_message = None
 
 class PySimLogger_Test(unittest.TestCase):
 
+    def setUp(self):
+        # PySimLogger.setup() is global, so a print callback left installed here fires for
+        # every PySimLogger message emitted by any test module that runs later in the same process
+        # ... where it asserts against a stale 'expected_message' and fails a test that has nothing
+        # to do with logging. Great fun!
+        # Restore before each test.
+        saved = (PySimLogger.print_callback, PySimLogger.verbose)
+        def _restore():
+            PySimLogger.print_callback, PySimLogger.verbose = saved
+        self.addCleanup(_restore)
+
     def __test_01_safe_defaults_one(self, callback, message:str):
         # When log messages are sent to an unconfigured PySimLogger class, we expect the unmodified message being
         # logged to stdout, just as if it were printed via a normal print() statement.

tests/unittests/test_param_src.py

@@ -68,7 +68,7 @@ class ParamSourceTest(unittest.TestCase):
 
     def test_param_source(self):
 
-        class ParamSourceTest(D):
+        class Paramtest(D):
             mandatory = (
                     'param_source',
                     'n',
@@ -78,6 +78,11 @@ class ParamSourceTest(unittest.TestCase):
                     'expect_arg',
                     'csv_rows',
                     )
+            param_source: param_source.ParamSource
+            n: int
+            expect: object
+            expect_arg: object
+            csv_rows: object
 
         def expect_const(t, vals):
             return tuple(t.expect_arg) == tuple(vals)
@@ -100,74 +105,59 @@ class ParamSourceTest(unittest.TestCase):
             return True
 
         param_source_tests = [
-            ParamSourceTest(param_source=param_source.ConstantSource.from_str('123'),
-                            n=3,
-                            expect=expect_const,
-                            expect_arg=('123', '123', '123')
-                           ),
-            ParamSourceTest(param_source=param_source.RandomDigitSource.from_str('12345'),
-                            n=3,
-                            expect=expect_random,
-                            expect_arg={'digits': decimals,
-                                        'val_minlen': 5,
-                                        'val_maxlen': 5,
-                                        },
-                           ),
-            ParamSourceTest(param_source=param_source.RandomDigitSource.from_str('1..999'),
-                            n=10,
-                            expect=expect_random,
-                            expect_arg={'digits': decimals,
-                                        'val_minlen': 1,
-                                        'val_maxlen': 3,
-                                        },
-                           ),
-            ParamSourceTest(param_source=param_source.RandomDigitSource.from_str('001..999'),
-                            n=10,
-                            expect=expect_random,
-                            expect_arg={'digits': decimals,
-                                        'val_minlen': 3,
-                                        'val_maxlen': 3,
-                                        },
-                           ),
-            ParamSourceTest(param_source=param_source.RandomHexDigitSource.from_str('12345678'),
-                            n=3,
-                            expect=expect_random,
-                            expect_arg={'digits': hexadecimals,
-                                        'val_minlen': 8,
-                                        'val_maxlen': 8,
-                                       },
-                           ),
-            ParamSourceTest(param_source=param_source.RandomHexDigitSource.from_str('0*8'),
-                            n=3,
-                            expect=expect_random,
-                            expect_arg={'digits': hexadecimals,
-                                        'val_minlen': 8,
-                                        'val_maxlen': 8,
-                                       },
-                           ),
-            ParamSourceTest(param_source=param_source.RandomHexDigitSource.from_str('00*4'),
-                            n=3,
-                            expect=expect_random,
-                            expect_arg={'digits': hexadecimals,
-                                        'val_minlen': 8,
-                                        'val_maxlen': 8,
-                                       },
-                           ),
-            ParamSourceTest(param_source=param_source.IncDigitSource.from_str('10001'),
-                            n=3,
-                            expect=expect_const,
-                            expect_arg=('10001', '10002', '10003')
-                           ),
-            ParamSourceTest(param_source=param_source.CsvSource('column_name'),
-                            n=3,
-                            expect=expect_const,
-                            expect_arg=('first val', 'second val', 'third val'),
-                            csv_rows=(
-                                      {'column_name': 'first val',},
-                                      {'column_name': 'second val',},
-                                      {'column_name': 'third val',},
-                                     )
-                           ),
+            Paramtest(param_source=param_source.ConstantSource.from_str('123'),
+                      n=3,
+                      expect=expect_const,
+                      expect_arg=('123', '123', '123')),
+            Paramtest(param_source=param_source.RandomDigitSource.from_str('12345'),
+                      n=3,
+                      expect=expect_random,
+                      expect_arg={'digits': decimals,
+                                  'val_minlen': 5,
+                                  'val_maxlen': 5}),
+            Paramtest(param_source=param_source.RandomDigitSource.from_str('1..999'),
+                      n=10,
+                      expect=expect_random,
+                      expect_arg={'digits': decimals,
+                                  'val_minlen': 1,
+                                  'val_maxlen': 3}),
+            Paramtest(param_source=param_source.RandomDigitSource.from_str('001..999'),
+                      n=10,
+                      expect=expect_random,
+                      expect_arg={'digits': decimals,
+                                  'val_minlen': 3,
+                                  'val_maxlen': 3}),
+            Paramtest(param_source=param_source.RandomHexDigitSource.from_str('12345678'),
+                      n=3,
+                      expect=expect_random,
+                      expect_arg={'digits': hexadecimals,
+                                  'val_minlen': 8,
+                                  'val_maxlen': 8}),
+            Paramtest(param_source=param_source.RandomHexDigitSource.from_str('0*8'),
+                      n=3,
+                      expect=expect_random,
+                      expect_arg={'digits': hexadecimals,
+                                  'val_minlen': 8,
+                                  'val_maxlen': 8}),
+            Paramtest(param_source=param_source.RandomHexDigitSource.from_str('00*4'),
+                      n=3,
+                      expect=expect_random,
+                      expect_arg={'digits': hexadecimals,
+                                  'val_minlen': 8,
+                                  'val_maxlen': 8}),
+            Paramtest(param_source=param_source.IncDigitSource.from_str('10001'),
+                      n=3,
+                      expect=expect_const,
+                      expect_arg=('10001', '10002', '10003')),
+            Paramtest(param_source=param_source.CsvSource('column_name'),
+                      n=3,
+                      expect=expect_const,
+                      expect_arg=('first val', 'second val', 'third val'),
+                      csv_rows=(
+                          {'column_name': 'first val'},
+                          {'column_name': 'second val'},
+                          {'column_name': 'third val'},
+                      )),
             ]
 
         outputs = []