Revert "esim/http_json_api: extend JSON API with server functionality"

This reverts commit e00c0becca.

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 >= 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
+# 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
 else:
-    from cmd2 import style, Fg # pylint: disable=no-name-in-module
+    from cmd2 import Fg, Bg # pylint: disable=no-name-in-module
     RED = Fg.RED
     YELLOW = Fg.YELLOW
     LIGHT_RED = Fg.LIGHT_RED
@@ -76,19 +76,43 @@ from pySim.app import init_card
 
 log = PySimLogger.get(Path(__file__).stem)
 
-class PysimApp(cmd2.Cmd):
+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):
     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, include_ipy=True)
+                         auto_load_commands=False, startup_script=script, **kwargs)
         self.intro = style(self.BANNER, fg=RED)
         self.default_category = 'pySim-shell built-in commands'
         self.card = None
@@ -104,24 +128,18 @@ Online manual available at https://downloads.osmocom.org/docs/pysim/master/html/
         self.apdu_trace = False
         self.apdu_strict = False
 
-        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.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.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, KeyboardInterrupt):
+            except StopIteration:
                 print("%i APDUs parsed, stop iteration." % apdu_counter)
                 return 0
 

pySim/ara_m.py

@@ -300,51 +300,6 @@ 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):
@@ -389,15 +344,48 @@ 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."""
-            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)
+            # 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)
             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."""
-            res_do = ADF_ARAM.aram_delete_all(self._cmd.lchan.scc)
+            deldo = CommandDelete()
+            res_do = ADF_ARAM.store_data(self._cmd.lchan.scc, deldo)
             if res_do:
                 self._cmd.poutput_json(res_do.to_dict())
 
@@ -406,6 +394,7 @@ 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,12 +16,6 @@
 # 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
@@ -129,12 +123,10 @@ 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,
@@ -145,7 +137,6 @@ class DownloadOrder(Es2PlusApiFunction):
 class ConfirmOrder(Es2PlusApiFunction):
     path = '/gsma/rsp2/es2plus/confirmOrder'
     input_params = {
-        'header': JsonRequestHeader,
         'iccid': param.Iccid,
         'eid': param.Eid,
         'matchingId': param.MatchingId,
@@ -153,7 +144,7 @@ class ConfirmOrder(Es2PlusApiFunction):
         'smdsAddress': param.SmdsAddress,
         'releaseFlag': param.ReleaseFlag,
       }
-    input_mandatory = ['header', 'iccid', 'releaseFlag']
+    input_mandatory = ['iccid', 'releaseFlag']
     output_params = {
         'header': JsonResponseHeader,
         'eid': param.Eid,
@@ -166,13 +157,12 @@ 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 = ['header', 'finalProfileStatusIndicator', 'iccid']
+    input_mandatory = ['finalProfileStatusIndicator', 'iccid']
     output_params = {
         'header': JsonResponseHeader,
       }
@@ -182,10 +172,9 @@ class CancelOrder(Es2PlusApiFunction):
 class ReleaseProfile(Es2PlusApiFunction):
     path = '/gsma/rsp2/es2plus/releaseProfile'
     input_params = {
-        'header': JsonRequestHeader,
         'iccid': param.Iccid,
       }
-    input_mandatory = ['header', 'iccid']
+    input_mandatory = ['iccid']
     output_params = {
         'header': JsonResponseHeader,
       }
@@ -195,7 +184,6 @@ class ReleaseProfile(Es2PlusApiFunction):
 class HandleDownloadProgressInfo(Es2PlusApiFunction):
     path = '/gsma/rsp2/es2plus/handleDownloadProgressInfo'
     input_params = {
-        'header': JsonRequestHeader,
         'eid': param.Eid,
         'iccid': param.Iccid,
         'profileType': param.ProfileType,
@@ -204,9 +192,10 @@ class HandleDownloadProgressInfo(Es2PlusApiFunction):
         'notificationPointStatus': param.NotificationPointStatus,
         'resultData': param.ResultData,
     }
-    input_mandatory = ['header', 'iccid', 'profileType', 'timestamp', 'notificationPointId', 'notificationPointStatus']
+    input_mandatory = ['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):
@@ -217,17 +206,18 @@ class Es2pApiClient:
         if client_cert:
             self.session.cert = client_cert
 
-        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)
+        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)
 
     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())
@@ -247,116 +237,3 @@ 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 = 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)
+        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)
 
     def call_initiateAuthentication(self, data: dict) -> dict:
         return self.initiateAuthentication.call(data)

pySim/esim/http_json_api.py

@@ -19,10 +19,8 @@ import abc
 import requests
 import logging
 import json
-from typing import Optional, Tuple
+from typing import Optional
 import base64
-from twisted.web.server import Request
-
 
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.DEBUG)
@@ -133,16 +131,6 @@ 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
@@ -173,118 +161,65 @@ class ApiError(Exception):
 
 class JsonHttpApiFunction(abc.ABC):
     """Base class for representing an HTTP[s] API Function."""
-    # 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.
+     # the below class variables are expected to be overridden in derived classes
 
-    # 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 = {}
 
-    # additional custom HTTP headers (server responses)
-    extra_http_res_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
 
-    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:
+    def encode(self, data: dict, func_call_id: Optional[str] = None) -> 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:
-                # 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
+                logger.warning('Unexpected/unsupported input parameter %s=%s', p, v)
+                output[p] = v
             else:
                 output[p] = p_class.encode(v)
         return output
 
-    def decode_client(self, data: dict) -> dict:
+    def decode(self, data: dict) -> dict:
         """[further] Decode and validate the JSON-Dict of the response body."""
         output = {}
-        output_mandatory = self.output_mandatory
+        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'])
 
-        # 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 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
             if not p in data:
                 raise ValueError('Mandatory output parameter "%s" missing' % p)
         for p, v in data.items():
@@ -296,202 +231,37 @@ 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 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
+        """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))
         req_headers = {
             'Content-Type': 'application/json',
             'X-Admin-Protocol': 'gsma/rsp/v2.5.0',
         }
-        req_headers.update(self.api_func.extra_http_req_headers)
+        req_headers.update(self.extra_http_req_headers)
 
-        # Perform HTTP request
         logger.debug("HTTP REQ %s - hdr: %s '%s'" % (url, req_headers, encoded))
-        response = self.session.request(self.api_func.http_method, url, data=encoded, headers=req_headers, timeout=timeout)
+        response = self.session.request(self.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))
 
-        # 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:
+        if response.status_code != self.expected_http_status:
             raise HttpStatusError(response)
+
         resp_content_type = response.headers.get('Content-Type')
-        if response.content and not resp_content_type.startswith(req_headers['Content-Type']):
+        if not resp_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'):
-                output = self.api_func.decode_client(response.json())
+                return self.decode(response.json())
             elif resp_content_type.startswith('text/plain;charset=UTF-8'):
-                output = { 'data': response.content.decode('utf-8') }
-            else:
-                raise HttpHeaderError(f'unimplemented response Content-Type: {response.headers=!r}')
-
-            # 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 { 'data': response.content.decode('utf-8') }
+            raise HttpHeaderError(f'unimplemented response Content-Type: {response.headers=!r}')
         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/personalization.py

@@ -493,7 +493,7 @@ class EnumParam(ConfigurableParameter):
 
     @classmethod
     def clean_name_str(cls, val):
-        return re.sub('[^0-9A-Za-z]', '', val).lower()
+        return re.sub('[^0-9A-Za-z-_]', '', val).lower()
 
 
 class Iccid(DecimalParam):
@@ -1280,7 +1280,7 @@ class SuciCalcInfoParameter(ConfigurableParameter):
     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": [], "hnet_pubkey_list": []}'
+    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")

pySim/log.py

@@ -24,15 +24,7 @@
 #
 
 import logging
-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
+from cmd2 import style
 
 class _PySimLogHandler(logging.Handler):
     def __init__(self, log_callback):
@@ -129,7 +121,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_31_102.py

@@ -285,14 +285,6 @@ 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
@@ -397,7 +389,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'] or [])
+        hnet_pubkey_list = self._compact_pubkey_list(hpkl.to_dict()['hnet_pubkey_list'])
 
         return {
             'prot_scheme_id_list': prot_scheme_id_list,

requirements.txt

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

setup.py

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

tests/unittests/test_configurable_parameters.py

@@ -55,8 +55,8 @@ class ConfigurableParameterTest(unittest.TestCase):
 
         upp_fnames = (
                 'SAIP2.1_gfmsuci.der',
-                'TS48v5_SAIP2.1A_NoBERTLV.der',
-                'TS48v5_SAIP2.3_BERTLV_SUCI.der',
+                'TS48v5_SAIP2.1B_NoBERTLV.der',
+                'TS48v5_SAIP2.3_NoBERTLV.der',
                )
 
         class Paramtest:
@@ -271,6 +271,7 @@ class ConfigurableParameterTest(unittest.TestCase):
                                  '11111111111111111111111111111111'
                                  '22222222222222222222222222222222'),
 
+
             Paramtest(param_cls=p13n.MncLen,
                       val='2',
                       expect_clean_val=2,
@@ -730,15 +731,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('usimtest'), 'usim-test')
-        self.assertEqual(p13n.AlgorithmID.name_normalize('usim_test'), 'usim-test')
-        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'), 'usimtest')
-        self.assertEqual(p13n.AlgorithmID.clean_name_str('usim_test'), 'usimtest')
+        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('Milenage'), 'milenage')
         self.assertEqual(p13n.AlgorithmID.clean_name_str('foo bar!'), 'foobar')