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

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
@@ -76,43 +76,19 @@ 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):

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/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):
@@ -344,48 +389,15 @@ class ADF_ARAM(CardADF):
         @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/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,37 +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)
-
-        resp_content_type = response.headers.get('Content-Type')
-        if not resp_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 resp_content_type.startswith('application/json'):
-                return self.decode(response.json())
-            elif resp_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")
@@ -1777,13 +1774,9 @@ class ProfileElementSequence:
 
         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,9 +20,6 @@
 
 import copy
 import pprint
-import logging
-import traceback
-import inspect
 from typing import Generator, Union
 from pySim.esim.saip.personalization import ConfigurableParameter
 from pySim.esim.saip import param_source
@@ -30,10 +27,6 @@ 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]]
 
@@ -128,8 +121,6 @@ class BatchPersonalization:
                     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_cls.get_name()} fed by {p.src.name}: {e}') from e
 
             pes.rebuild_mandatory_services()
@@ -333,15 +324,12 @@ class BatchAudit(list):
 
         return batch_audit
 
-    def to_csv_rows(self, headers=True, sort_key=None, column_blacklist=None):
+    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:

pySim/esim/saip/personalization.py

@@ -19,26 +19,19 @@ import abc
 import enum
 import io
 import re
-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))
 
@@ -297,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):
@@ -429,71 +420,69 @@ class BinaryParam(ConfigurableParameter):
 
 
 class EnumParam(ConfigurableParameter):
-    """ConfigurableParameter for named value enumerations.
+    """ConfigurableParameter for named integer enumeration values.
 
-    Subclasses define an own value_map, and implement their own apply_val() and get_values_from_pes().
-    """
-    value_map = {
-            # For example:
-            #'Meaningful label for value 23': 0x23,
-            # Where 0x23 is a valid value to use for apply_val(), of any valid type.
-        }
-    _value_map_reverse = None
+    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):
@@ -646,14 +635,13 @@ 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
@@ -675,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):
@@ -1099,17 +1099,17 @@ 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."""
     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
 
-    # EnumParam.validate_val() returns the int values from value_map
+    # 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 Values
 
     @classmethod
     def get_values_from_pes(cls, pes: ProfileElementSequence):
@@ -1184,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/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

@@ -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
@@ -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
@@ -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)

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):
@@ -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/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,

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/unittests/test_configurable_parameters.py

@@ -21,7 +21,6 @@ import enum
 import io
 import sys
 import unittest
-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
@@ -469,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:
@@ -478,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()
 
@@ -521,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)
 
@@ -668,7 +558,7 @@ class TestEnumParam(unittest.TestCase):
     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)
+        self.assertEqual(p13n.AlgorithmID.validate_val('usim_test'), 3)
 
     def test_validate_by_int(self):
         self.assertEqual(p13n.AlgorithmID.validate_val(1), 1)
@@ -681,6 +571,7 @@ class TestEnumParam(unittest.TestCase):
         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):
@@ -717,7 +608,7 @@ class TestEnumParam(unittest.TestCase):
     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')
+        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))
@@ -731,13 +622,13 @@ class TestEnumParam(unittest.TestCase):
     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')
+        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'), 'usim-test')
-        self.assertEqual(p13n.AlgorithmID.clean_name_str('usim_test'), 'usim_test')
+        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')
 

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.