WIP: Remote Card Procedure

Related: SYS#6959

README.md

@@ -97,7 +97,7 @@ Please install the following dependencies:
  - pyscard
  - pyserial
  - pytlv
- - pyyaml >= 5.1
+ - pyyaml >= 5.4
  - smpp.pdu (from `github.com/hologram-io/smpp.pdu`)
  - termcolor
 

contrib/jenkins.sh

@@ -10,6 +10,11 @@
 
 export PYTHONUNBUFFERED=1
 
+setup_venv() {
+	virtualenv -p python3 venv --system-site-packages
+	. venv/bin/activate
+}
+
 if [ ! -d "./tests/" ] ; then
 	echo "###############################################"
 	echo "Please call from pySim-prog top directory"
@@ -23,8 +28,7 @@ fi
 
 case "$JOB_TYPE" in
 "test")
-	virtualenv -p python3 venv --system-site-packages
-	. venv/bin/activate
+	setup_venv
 
 	pip install -r requirements.txt
 	pip install pyshark
@@ -32,23 +36,27 @@ case "$JOB_TYPE" in
 	# Execute automatically discovered unit tests first
 	python -m unittest discover -v -s tests/unittests
 
-	# Run pySim-prog integration tests (requires physical cards)
-	cd tests/pySim-prog_test/
-        ./pySim-prog_test.sh
-	cd ../../
-
 	# Run pySim-trace test
 	tests/pySim-trace_test/pySim-trace_test.sh
+	;;
+"card-test") # tests requiring physical cards
+	setup_venv
 
-	# Run pySim-shell integration tests (requires physical cards)
+	pip install -r requirements.txt
+
+	# Run pySim-prog integration tests
+	cd tests/pySim-prog_test/
+	./pySim-prog_test.sh
+	cd ../../
+
+	# Run pySim-shell integration tests
 	python3 -m unittest discover -v -s ./tests/pySim-shell_test/
 
 	# Run pySim-smpp2sim test
 	tests/pySim-smpp2sim_test/pySim-smpp2sim_test.sh
 	;;
 "distcheck")
-	virtualenv -p python3 venv --system-site-packages
-	. venv/bin/activate
+	setup_venv
 
 	pip install .
 	pip install pyshark
@@ -61,8 +69,7 @@ case "$JOB_TYPE" in
 	# Print pylint version
 	pip3 freeze | grep pylint
 
-	virtualenv -p python3 venv --system-site-packages
-	. venv/bin/activate
+	setup_venv
 
 	pip install .
 
@@ -80,8 +87,7 @@ case "$JOB_TYPE" in
 		contrib/*.py
 	;;
 "docs")
-	virtualenv -p python3 venv --system-site-packages
-	. venv/bin/activate
+	setup_venv
 
 	pip install -r requirements.txt
 

contrib/rcp/rcp_client.py

@@ -0,0 +1,204 @@
+#!/usr/bin/env python3
+
+# (C) 2026 by sysmocom - s.f.m.c. GmbH
+# All Rights Reserved
+#
+# Author: Philipp Maier
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import sys
+import websockets
+import asyncio
+import argparse
+import logging
+from copy import deepcopy
+from pathlib import Path
+from pySim.log import PySimLogger
+from rcp_utils import CltConnHdlr, backtrace, pytype_to_type, load_ca_cert
+from pySim.transport import init_reader, argparse_add_reader_args, LinkBase
+
+SERVER_TIMEOUT = 10
+
+log = PySimLogger.get(Path(__file__).stem)
+option_parser = argparse.ArgumentParser(description='RCP Client',
+                                        formatter_class=argparse.ArgumentDefaultsHelpFormatter)
+argparse_add_reader_args(option_parser)
+option_parser.add_argument("--verbose", help="Enable verbose logging",
+                           action='store_true', default=False)
+option_parser.add_argument("--uri", help="URI of the RCP-Server")
+option_parser.add_argument("--ca-cert", help="SSL/TLS CA-Certificate of the RCP-Server")
+
+class RcpcCltConnHdlr(CltConnHdlr):
+    def __init__(self, sl, *args, **kwargs):
+        self.sl = sl
+        super().__init__(*args, **kwargs)
+
+    async def describe(self, suitable_for:dict) -> list:
+        log.info("Requesting module descriptions from RCP Server ...")
+        tx_json = {'rcpc_hello': {'suitable_for' : suitable_for}}
+        rx_json = await self._transact(tx_json)
+        module_descr = rx_json['rcpc_welcome']['module_descr']
+        if not module_descr:
+            raise ValueError("No RCP module available for this card")
+        return module_descr
+
+    async def run(self, cmd:str, cmd_argv) -> int:
+        log.info("Executing command with RCP Server ...")
+        tx_json = {'rcpc_command': {'cmd' : cmd, 'cmd_argv' : cmd_argv}}
+        while(True):
+            rx_json = await self._transact(tx_json)
+            tx_json = None
+            if 'rcpc_instr' in rx_json:
+                rcpc_instr = rx_json['rcpc_instr']
+                if 'c_apdu' in rcpc_instr:
+                    c_apdu = rx_json['rcpc_instr']['c_apdu']
+                    data, sw = sl.send_apdu(c_apdu)
+                    tx_json = {'rcpc_result': {'r_apdu' : {'data': data.upper(), 'sw': sw.upper()}}}
+                elif 'reset' in rcpc_instr:
+                    sl.reset_card()
+                    atr = sl.get_atr()
+                    tx_json = {'rcpc_result': {'atr' : atr.upper()}}
+                elif 'print' in rcpc_instr:
+                    log.info(str(self) + " -- %s", rx_json['rcpc_instr']['print'])
+                    tx_json = {'rcpc_result': {'empty' : None}}
+            elif 'rcpc_goodbye' in rx_json:
+                rc = rx_json['rcpc_goodbye']
+                log.info("Command execution done, rc: %d", rc)
+                return rc
+
+def check_if_user_needs_basic_help(argv):
+    """
+    The '--uri' argument is the minimum requirement to connect to the RCP Server to retrieve the information about the
+    dynamic commandline arguments. In case this argument is missing while '--help' or '-h' arguments are present. Then
+    we will fall back to display only a basic help that contains only the static commandline arguments (see above).
+    """
+
+    if '--help' in argv or '-h' in argv:
+        if '--uri' not in argv:
+            option_parser.parse_args()
+            sys.exit(1)
+
+def parse_known_arguemnts(argv):
+    """
+    Parse the commandline arguments we know so far. Ignore unknown arguments and filter out '--help' and '-h'
+    arguments, in case those are present.
+    """
+
+    argv_filtered = deepcopy(argv)
+    if '--help' in argv_filtered:
+        argv_filtered.remove('--help')
+    if '-h' in argv_filtered:
+        argv_filtered.remove('-h')
+    opts, unknown = option_parser.parse_known_args(argv_filtered)
+    return opts
+
+async def run_rcp_session(opts, sl, ssl_context) -> int:
+    """
+    Connect to the RCP Server, retrieve the module description, use the module description to complete the commandline
+    argument parser, execute the command that the user has selected.
+    """
+
+    # Request ATR from card
+    card_atr = sl.get_atr().upper()
+    log.info("Detected Card with ATR: %s" % card_atr)
+
+    # Connect to RCP server
+    log.info("RCP Server URI: %s" % opts.uri)
+    async with websockets.connect(opts.uri, ssl=ssl_context) as websocket:
+        client = RcpcCltConnHdlr(sl, websocket, SERVER_TIMEOUT)
+
+        # Retrieve module description
+        module_descrs = await client.describe({"atr" : card_atr})
+
+        # Complete the commandlie parser and set up a dict that we can use as filter
+        # TODO: Maybe it makes sense to integrate this as a method into the RcpcCltConnHdlr class?
+        option_subparsers = option_parser.add_subparsers(dest='command', help="RCP command to use", required=True)
+        sys_argv_filter = {}
+        for module_descr in module_descrs:
+            cmd_descr = module_descr['cmd_descr']
+            for cmd in cmd_descr:
+                command_name = module_descr['name'] + "_" + cmd['name']
+                option_parser_cmd = option_subparsers.add_parser(command_name, help=cmd['help'])
+                sys_argv_filter[command_name] = []
+                for arg in cmd['args']:
+                    arg['spec'] = pytype_to_type(arg['spec'])
+                    option_parser_cmd.add_argument(arg['name'], **arg['spec'])
+                    sys_argv_filter[command_name].append(arg['name'])
+
+        # Re-Parse commandline options with the completed commandline parser. In case commandline help is
+        # requested. The program is able to display the full helpscreen and exists.
+        opts = option_parser.parse_args()
+
+        # Filter the relevant command arguments from sys.argv
+        cmd_argv = []
+        next_is_value=False
+        for arg in sys.argv:
+            if arg in sys_argv_filter[opts.command]:
+                cmd_argv.append(arg)
+                next_is_value=True
+            elif next_is_value is True:
+                next_is_value=False
+                cmd_argv.append(arg)
+
+        # Run the command and close the connection
+        rc = await client.run(opts.command, cmd_argv)
+        await client.close()
+        return rc
+
+if __name__ == '__main__':
+
+    # Setup logging
+    PySimLogger.setup(print, {logging.WARN: "\033[33m", logging.DEBUG: "\033[90m"}, '--verbose' in sys.argv)
+
+    # Since parts of the commandline arguments are retrieved dynamically, we have to resolve a chicken-egg-problem.
+    # We cannot call option_parser.parse_args() at the beginning, since we haven't received all information to
+    # complete the option_parser yet. However in order to retrieve the arguments correctly we need to get the
+    # URI and the parameters for the smartcard reader before we make the connection. The situation is even further
+    # complicated in case the user requests commandline help.
+
+    # To resolve the problem we first check if the user needs basic help (no '--uri' parameter present). If this is the
+    # case, the program will exit with a basic helpscreen.
+    check_if_user_needs_basic_help(sys.argv)
+
+    # In all other cases we parse the arguments we know so far. In case the user requests commandline help, we will
+    # ignore this request and continue. The full help is then displayed later when the option_parser is completed
+    # afer we have requested the commandline argument descriptions from the RCP Server. (see below)
+    opts = parse_known_arguemnts(sys.argv)
+
+    # Load SSL/TLS CA certificate from file
+    if opts.ca_cert:
+        ssl_context = load_ca_cert("RCP Server CA", opts.ca_cert)
+    else:
+        ssl_context = None
+
+    # Initialize card reader
+    try:
+        sl = init_reader(opts)
+        sl.connect()
+    except Exception as e:
+        backtrace("Card reader initialization")
+        sys.exit(1)
+
+    # Run the RCP session
+    try:
+        rc = asyncio.run(run_rcp_session(opts, sl, ssl_context))
+        sys.exit(rc)
+    except SystemExit as rc:
+        sys.exit(rc)
+    except:
+        backtrace("RCP session")
+        sys.exit(1)
+
+

contrib/rcp/rcp_module_utils.py

@@ -0,0 +1,387 @@
+#!/usr/bin/env python3
+
+# (C) 2026 by sysmocom - s.f.m.c. GmbH
+# All Rights Reserved
+#
+# Author: Philipp Maier
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+
+import abc
+import os
+import argparse
+import logging
+import threading
+import asyncio
+import websockets
+from argparse import Namespace
+from copy import deepcopy
+from pathlib import Path
+from typing import Optional
+from osmocom.utils import Hexstr, is_hexstr
+from pySim.utils import ResTuple
+from pySim.transport import LinkBase
+from pySim.commands import SimCardCommands
+from pySim.log import PySimLogger
+from rcp_utils import SrvSyncConnHdlr, CltConnHdlr, backtrace, pytype_to_type, load_server_cert, load_ca_cert
+from rcp_utils import dict_from_key_value_pairs
+from websockets.sync.server import serve, ServerConnection
+
+# Response timeout towards the RCP Server (includes RCP Client latency)
+RCP_SERVER_TIMEOUT = 30 # sec.
+
+log = PySimLogger.get(Path(__file__).stem)
+
+class RcpsSimLink(LinkBase):
+    """
+    pySim: Transport Link for RCPM (Remote Card Procedure Module)
+    This is a 'headless' transport link implementation that can only be used from an RCPM module. It merely serves as
+    an adapter between the pySim transport API and the RCPM command server connection handler.
+    """
+
+    name = 'RCPM'
+
+    def __init__(self, conn_hdlr: SrvSyncConnHdlr, **kwargs):
+        self.conn_hdlr = conn_hdlr
+        self._atr = None
+        super().__init__(**kwargs)
+
+    def __str__(self) -> str:
+        return "rcpm:" + str(self.conn_hdlr)
+
+    def _send_apdu(self, apdu: Hexstr) -> ResTuple:
+        tx_json = {'rcpc_instr': {'c_apdu' : apdu.upper()}}
+        rx_json = self.conn_hdlr._transact(tx_json)
+        data = rx_json['rcpc_result']['r_apdu']['data']
+        sw = rx_json['rcpc_result']['r_apdu']['sw']
+        return data, sw
+
+    def wait_for_card(self, timeout: Optional[int] = None, newcardonly: bool = False):
+        # In this setting, we do not have/cannot to wait for a card since we are not the entity that handles the
+        # direct connection to the card. When the procedure begins, we assume that the remote end already has set up
+        # a connection to the card and made it ready to perform operations on it.
+        pass
+
+    def connect(self):
+        # In this setting, we do not have/cannot to connect because we are not the entity that handles the direct
+        # connection to the card. The connection is established by the remote end.
+        pass
+
+    def get_atr(self) -> Hexstr:
+        return self._atr
+
+    def disconnect(self):
+        # In this setting, we do not have/cannot disconnect because we are not the enitity that handles the direct
+        # connection to the card. The disconnect is eventually done by the remote end when the procedure has finished.
+        pass
+
+    def _reset_card(self):
+        tx_json = {'rcpc_instr': {'reset' : None}}
+        rx_json = self.conn_hdlr._transact(tx_json)
+        self._atr = rx_json['rcpc_result']['atr']
+        return 1
+
+class RcpsCltConnHdlr(CltConnHdlr):
+    """
+    The RCP Server client handler is used to connect to the RCP Server when RCP Module is started. The connection is
+    kept alive until the RCP Module is terminated. This connection is used to exchange management data with the RCP
+    Server.
+    """
+
+    def __init__(self, cmd_srv_addr: str, cmd_srv_port: int, module, *args, **kwargs):
+        self.cmd_srv_addr = cmd_srv_addr
+        self.cmd_srv_port = cmd_srv_port
+        self.module = module
+        super().__init__(*args, **kwargs)
+
+    async def describe(self):
+        """
+        Send a detailed description about this RCP Module to the RCP Server. This is also the initial message that
+        the RCP Server expects when an RCP Module connects.
+        """
+
+        # The rules (dict) in suitable_for (array of dict) may contain hexstrings. Here we go through those rules
+        # and convert those hexstrings to uppercase, since this is the standard we have set for the JSON messages.
+        suitable_for = []
+        for rule in self.module.suitable_for:
+            rule_filtered = {}
+            for k in rule:
+                if is_hexstr(rule[k]):
+                    rule_filtered[k] = rule[k].upper()
+                else:
+                    rule_filtered[k] = rule[k]
+            suitable_for.append(rule_filtered)
+
+        # Publish RCP Module description on the RCP server
+        tx_json = {'rcpm_hello':
+                   {'name' : self.module.name,
+                    'cmd_descr' : self.module.cmd_descr,
+                    'suitable_for' : suitable_for,
+                    'retrieve_keys' : {
+                        'euicc' : self.module.retrieve_euicc_keys,
+                        'uicc' : self.module.retrieve_uicc_keys
+                    },
+                    'addr' : self.cmd_srv_addr,
+                    'port' : self.cmd_srv_port
+                    }
+                   }
+        rx_json = await self._transact(tx_json)
+        if 'rcpm_welcome' not in rx_json:
+            raise ValueError("description not accepted by RCP Server")
+
+class RcpModule(abc.ABC):
+    """
+    Base class to implement to derive a concrete RCPM module class
+    """
+
+    # Module name used to identify the module in logs and user output. This module name should be short and concise.
+    name = "RCPM"
+
+    # Command description of this module. The command description consists of a short and concise command name, a
+    # helpstring and an argument specification in the form of a python dict. This specificaton dict is directly
+    # passed to agparse on the client side.
+    #
+    # Example:
+    # [{"name" : "reset",
+    #  "help": "reset the card",
+    #  "args" : []},
+    # {"name" : "read_binary",
+    #  "help": "read binary data from a transparent file.",
+    #  "args" : [ { "name" : "--fid",
+    #               "spec" : {"required" : True,
+    #                         "help" : "File identifier to of the file to read",
+    #                         "action" : "append"},
+    #              }
+    #            ]}
+    # ]
+    cmd_descr = []
+
+    # List with UICC (or eSIM) keys (columns) that the RCP Server shall retrieve before a command is executed.
+    # Execution will not continue in case any of the requested keys is not found.
+    # (see also: pySim.card_key_provider)
+    #
+    # Example: ['kic1', 'kid1', 'kik1']
+    retrieve_uicc_keys = []
+
+    # Same as retrieve_uicc_keys (see above), but only applicable with eUICCs
+    #
+    # Example: ['isdr_kic1', 'isdr_kid1', 'isdr_kik1']
+    retrieve_euicc_keys = []
+
+    # Card properties to determine if this module is suitable for a specific card type or card types. The RCP Server
+    # will match those properties against user requests to determine which module provides useful services to the
+    # user's card.
+    #
+    # Example: [{"atr" : "3b9f96803f87828031e073fe211f574543753130136502"}]
+    suitable_for = []
+
+    # In addition the above, the derived class must implement command methods for each command that is defined in the
+    # command description (see above). Each command method must begin with the prefix "cmd_" followed by the command
+    # name used in the command description. A command method must have the form as shown in the example shown below.
+    # Each method should return an integer value which will become the final return code of the RCP client program.
+    #
+    # Args:
+    #     hdlr:     RcpModuleHdlr object, this object is provided by the RcpmCmdSrvConnHdlr object, which calls
+    #               the command method of the module. Through the RcpModuleHdlr object, the API user gets access
+    #               to special service methods (e.g. print) and other required properties (e.g. the SimCardCommands
+    #               objects, key material and others (see above).
+    #
+    # Example:
+    # def cmd_reset(self, hdlr: RcpModuleHdlr) -> int: ...
+    # def cmd_read_binary(self, hdlr: RcpModuleHdlr) -> int: ...
+
+class RcpmCmdSrvConnHdlr(SrvSyncConnHdlr):
+    """
+    The RCP Module command server connection handler is used to handle dedicated connections from the RCP Server. Those
+    dedicated connections are technically transparent connections between the RCP Client and the RCP Module (this). The
+    RCP Server merely acts as a proxy at that point.
+    """
+
+    def __init__(self, module: RcpModule, *args, **kwargs):
+        SrvSyncConnHdlr.__init__(self, *args, *kwargs)
+        self.module = module
+
+    def _parse_cmd_argv(self, cmd_suffix: str, cmd_argv: list[str]) -> Namespace:
+        """ Parse (and validate) the received argument vector """
+        # Use the cmd_descr of the module to create a (temporary) argument parser for the received argument vector
+        cmd_parser = argparse.ArgumentParser()
+        for cmd in self.module.cmd_descr:
+            if cmd['name'] == cmd_suffix:
+                args = deepcopy(cmd['args'])
+                for arg in args:
+                    arg['spec'] = pytype_to_type(arg['spec'])
+                    cmd_parser.add_argument(arg['name'], **arg['spec'])
+
+        # Parse the arguments and return the parsed Namespace object.
+        try:
+            return cmd_parser.parse_args(cmd_argv)
+        except SystemExit:
+            raise ValueError("unable to parse arguments: %s", str(cmd_argv), )
+
+    def print(self, message: str):
+        """ Print a message on the client side """
+        log.info(str(self) + " -- %s" % message)
+        tx_json = {'rcpc_instr': {'print' : message}}
+        rx_json = self._transact(tx_json)
+        if rx_json != {'rcpc_result': {'empty' : None}}:
+            raise ValueError("unexpected response from RCP Client: %s", rx_json)
+
+    def procedure(self):
+        """ Receive and process a command from the RCP Client (via RCP Server) """
+
+        # Receive the command request
+        rx_json = self._recv()
+        cmd = rx_json['rcpc_command']['cmd']
+        cmd_argv = rx_json['rcpc_command']['cmd_argv']
+        keys = rx_json['rcpc_command']['keys']
+        keys_uicc = dict_from_key_value_pairs(keys['uicc'], keylabel='key', valuelabel='value')
+        keys_euicc = dict_from_key_value_pairs(keys['euicc'], keylabel='key', valuelabel='value')
+
+        log.info(str(self) + " -- executing command: %s %s", cmd, " ".join(cmd_argv))
+
+        try:
+            # Make sure the command actually addresses this module
+            cmd_prefix = self.module.name + "_"
+            if not cmd.startswith(cmd_prefix):
+                raise ValueError("invalid command: %s" % cmd)
+
+            # Make sure the module actually provides a command method for the requested command
+            cmd_suffix = cmd[len(cmd_prefix):]
+            cmd_method = "cmd_" + cmd_suffix
+            if not hasattr(self.module, cmd_method):
+                raise ValueError("missing command method: %s" % cmd_method)
+
+            # Parse and validate command arguments
+            cmd_args = self._parse_cmd_argv(cmd_suffix, cmd_argv)
+
+            # TODO: Perform a proper setup, similar to the one we have in pySim-shell, so that we have proper
+            # runtime states and full access to the pySim API
+            self.scc = SimCardCommands(transport=RcpsSimLink(self))
+            self.scc.cla_byte = "00"
+            self.scc.sel_ctrl = "0004"
+
+            # Hand over control to the command method provided by the specific module implementation
+            try:
+                rcp_module_hdlr = RcpModuleHdlr(self, cmd_args, keys_uicc, keys_euicc)
+                rc = getattr(self.module, cmd_method)(rcp_module_hdlr)
+            except Exception as e:
+                backtrace("command method")
+                rc = 1 # general error
+
+        except Exception as e:
+            backtrace("command parsing")
+            rc = 126 # cannot execute
+
+        # The prodedure is done, send "goodbye" message
+        log.info(str(self) + " -- command execution done, rc: %d" % rc)
+        tx_json = {'rcpc_goodbye': rc}
+        self._send(tx_json)
+
+class RcpModuleHdlr():
+    """
+    RCP Module handler class. This class is used by the RcpmCmdSrvConnHdlr to create the handler RcpModuleHdlr object
+    (hdlr), which is is passed to the command method. The RcpModuleHdlr gives the API user access to resources he can
+    use carry out the command.
+    """
+
+    # The scc property contains the SimCardCommands object may be used to send APDUs, retrieve the ATR, or even more
+    # complex tasks like selecting a file (see also pysim.commands)
+    scc = None
+
+    # The cmd_args property contains the parsed command arguments which were passed by the end-user to the RCP Client.
+    # The arguments are already parsed and validated against the cmd_dscr property of the RcpModule. The arguments are
+    # in the form of a Namespace object and can be accessed like any argparse output. However, since the arguments
+    # contain user input, some caution is required.
+    cmd_args = None
+
+    # In case the retrieve_uicc_keys property of the RcpModule is used retrieve UICC key material, this property will
+    # contain the key material in the form of a dictionary. The format is similar to the return value of
+    # card_key_provider_get() (see also pySim.card_key_provider)
+    keys_uicc = {}
+
+    # Same as self.keys_uicc, but contains eUICC related key material in case requested using retrieve_uicc_keys
+    keys_euicc = {}
+
+    def __init__(self, hdlr: RcpmCmdSrvConnHdlr, cmd_args: Namespace, keys_uicc: dict, keys_euicc: dict):
+        # The command method (API user) must not access the related RcpmCmdSrvConnHdlr (see below) directly. Only
+        # the resources below may be accessed.
+        self.__hdlr = hdlr
+
+        # Assign properties intended to be used by the command method (API user)
+        self.scc = self.__hdlr.scc
+        self.cmd_args = cmd_args
+        self.keys_uicc = keys_uicc
+        self.keys_euicc = keys_euicc
+
+    def print(self, message: str):
+        """ Print a message on the client side """
+        self.__hdlr.print(message)
+
+def rcpm_setup_argparse(description: str):
+    """Create argument parser and add the basic arguments all RCP Modules should have"""
+
+    option_parser = argparse.ArgumentParser(description='RCP Module: ' + description,
+                                            formatter_class=argparse.ArgumentDefaultsHelpFormatter)
+    option_parser.add_argument("--verbose", help="Enable verbose logging", action='store_true', default=False)
+    option_parser.add_argument("--uri", help="URI of the RCP-Server", required=True)
+    option_parser.add_argument("--rcps-ca-cert", help="SSL/TLS CA-Certificate of the RCP-Server", required=True)
+    option_parser.add_argument("--rcpm-cmd-server-addr", help="Local Host/IP to bind RCP-Module-Command-Server to",
+                               required=True)
+    option_parser.add_argument("--rcpm-cmd-server-port", help="Local TCP port to bind RCP-Module-Command-Server to",
+                               required=True, type=int)
+    option_parser.add_argument("--rcpm-cmd-server-cert", help="SSL/TLS Certificate of the RCP-Module-Command-Server",
+                               required=True)
+    return option_parser
+
+def rcpm_run_module(opts: Namespace, module: RcpModule, *args, **kwargs):
+
+    PySimLogger.setup(print, {logging.WARN: "\033[33m", logging.DEBUG: "\033[90m"}, opts.verbose)
+    log.info("RCP Module startup: %s", module.name)
+    log.debug("Main process ID: %d", os.getpid())
+
+    # Load SSL/TLS certificates
+    rcpm_cmd_ssl_context = load_server_cert("RCPM Command Server", opts.rcpm_cmd_server_cert)
+    ssl_context = load_ca_cert("RCPM Server Client", opts.rcps_ca_cert)
+
+    # Start local RCP Client Command Server
+    log.info("RCPC command server at: %s:%d" % (opts.rcpm_cmd_server_addr, opts.rcpm_cmd_server_port))
+    def rcpm_cmd_conn_hdlr(websocket: ServerConnection):
+        hdlr = RcpmCmdSrvConnHdlr(module(*args, *kwargs), websocket, RCP_SERVER_TIMEOUT)
+        hdlr.procedure()
+        hdlr.close()
+
+    server = serve(rcpm_cmd_conn_hdlr, opts.rcpm_cmd_server_addr, opts.rcpm_cmd_server_port, ssl=rcpm_cmd_ssl_context)
+    def rcpm_cmd_server():
+        log.debug("RCPC command server thread ID: %d", threading.get_native_id())
+        server.serve_forever()
+    rcpm_cmd_server_thread = threading.Thread(target = rcpm_cmd_server)
+    rcpm_cmd_server_thread.start()
+
+    # Connect to RCP Server and publish module description
+    async def rcps_client():
+        async with websockets.connect(opts.uri, ping_timeout=10.0, ping_interval=1.0, ssl=ssl_context) as websocket:
+            client = RcpsCltConnHdlr(opts.rcpm_cmd_server_addr, opts.rcpm_cmd_server_port, module, websocket,
+                                     RCP_SERVER_TIMEOUT)
+            await client.describe()
+            await client.wait_close()
+    try:
+        asyncio.run(rcps_client())
+    except Exception as e:
+        backtrace("RCPS client")
+
+    # Shutdown
+    server.shutdown()
+    rcpm_cmd_server_thread.join()
+    log.info("RCP Module shutdown: %s", module.name)

contrib/rcp/rcp_server.py

@@ -0,0 +1,361 @@
+#!/usr/bin/env python3
+
+# (C) 2026 by sysmocom - s.f.m.c. GmbH
+# All Rights Reserved
+#
+# Author: Philipp Maier
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import os
+import sys
+import argparse
+import asyncio
+import logging
+from osmocom.utils import Hexstr
+from pySim.utils import ResTuple
+from copy import deepcopy
+from pathlib import Path
+from pySim.log import PySimLogger
+from pySim.utils import dec_iccid
+import websockets
+from websockets.asyncio.server import serve, ServerConnection
+from rcp_utils import SrvConnHdlr, CltConnHdlr, JsonValidator
+from rcp_utils import load_json_schema, backtrace, pytype_to_type, load_server_cert, load_ca_cert
+from rcp_utils import key_value_pairs_from_dict
+from pySim.card_key_provider import argparse_add_card_key_provider_args, init_card_key_provider
+from pySim.card_key_provider import card_key_provider_get_field, card_key_provider_get
+
+# TODO: Logging is fine, however it may also be a good idea to log some higher level events to some sort of journal.
+# We could use OpenObserve for that.
+
+CLIENT_TIMEOUT = 10
+
+log = PySimLogger.get(Path(__file__).stem)
+runtime_state = None
+option_parser = argparse.ArgumentParser(description='RCP Server',
+                                        formatter_class=argparse.ArgumentDefaultsHelpFormatter)
+option_parser.add_argument("--verbose", help="Enable verbose logging",
+                           action='store_true', default=False)
+option_parser.add_argument("--rcpc-server-addr", help="Local Host/IP to bind RCP-Client-Server to",
+                           required=True)
+option_parser.add_argument("--rcpc-server-port", help="Local TCP port to bind RCP-Client-Server to",
+                           required=True, type=int)
+option_parser.add_argument("--rcpc-server-cert", help="SSL/TLS Certificate of the RCP-Client-Server",
+                           required=True)
+option_parser.add_argument("--rcpm-server-addr", help="Local Host/IP to bind RCP-Module-Server to",
+                           required=True)
+option_parser.add_argument("--rcpm-server-port", help="Local TCP port to bind RCP-Module-Server to",
+                           required=True, type=int)
+option_parser.add_argument("--rcpm-server-cert", help="SSL/TLS Certificate of the RCP-Module-Server",
+                           required=True)
+option_parser.add_argument("--rcpm-module-ca-cert", help="SSL/TLS CA-Certificate of the RCP-Module-Command-Server",
+                           required=True)
+argparse_add_card_key_provider_args(option_parser)
+
+# TODO move those into the RuntimeState?
+rcpc_rx_schema = None
+rcpc_tx_schema = None
+rcpm_ca_ssl_contextssl_context = None
+
+class ModuleRuntimeState:
+    def __init__(self, websocket:ServerConnection, name:str, cmd_descr:list, suitable_for:list, retrieve_keys:dict,
+                 addr:str, port:int):
+        self.name = name
+        self.websocket = websocket
+
+        # Run the cmd_descr through argparse to catch malformed arguments early
+        for cmd in cmd_descr:
+            args = deepcopy(cmd['args'])
+            cmd_parser = argparse.ArgumentParser()
+            for arg in args:
+                # TODO: wrap this into a try/catch block and log broken arguments?
+                arg['spec'] = pytype_to_type(arg['spec'])
+                cmd_parser.add_argument(arg['name'], **arg['spec'])
+
+        self.cmd_descr = cmd_descr
+        self.suitable_for = suitable_for
+        self.retrieve_keys = retrieve_keys
+        self.addr = addr
+        self.port = port
+        log.debug("new RCP module context created: '%s'", name)
+
+    def is_suitable(self, suitable_for:dict) -> bool:
+        if suitable_for in self.suitable_for:
+            return True
+        return False
+
+    def describe(self) -> dict:
+        return {'name': self.name,
+                'cmd_descr': self.cmd_descr}
+
+    def __str__(self) -> str:
+        return self.name
+
+    def __del__(self):
+        log.debug("RCP module context destroyed: '%s'", self.name)
+
+class RuntimeState:
+    def __init__(self):
+        self.module_runtime_states = []
+        log.debug("new runtime context created.")
+
+    def __log_modules_available(self) -> str:
+        if self.module_runtime_states:
+            modules_str = ""
+            for module in self.module_runtime_states:
+                modules_str += "'" + str(module) + "', "
+            return "RCP modules available: %s" % modules_str[:-2]
+        else:
+            return "RCP modules available: none"
+
+    def module_add(self, module: ModuleRuntimeState):
+        self.module_runtime_states.append(module)
+        log.info("new RCP module, %s", self.__log_modules_available())
+
+    def module_remove(self, websocket:ServerConnection):
+        for module in self.module_runtime_states:
+            if module.websocket == websocket:
+                self.module_runtime_states.remove(module)
+                log.info("RCP module removed, %s", self.__log_modules_available())
+                return
+        log.warning("cannot remove RCP module, no RCP module associated with RCPC connection: %s:%d, %s" %
+                    (*websocket.remote_address, self.__log_modules_available()))
+
+    def modules_find(self, suitable_for:dict) -> list[dict]:
+        modules = []
+        for module in self.module_runtime_states:
+            if module.is_suitable(suitable_for):
+                modules.append(module.describe())
+        if modules:
+            return modules
+        # It is absolutely tolerable if no suitable RCP module can be found. If this is the case, the client should
+        # display an empty help screen and exit normally.
+        log.warning("no suitable RCP module found, %s", self.__log_modules_available())
+        return []
+
+    def module_find(self, suitable_for:dict, cmd:str) -> ModuleRuntimeState:
+        modules = self.modules_find(suitable_for)
+        for m in modules:
+            module_name = m['name']
+            cmd_descr = m['cmd_descr']
+            for c in cmd_descr:
+                cmd_name = c['name']
+                if module_name + "_" + cmd_name == cmd:
+                    break
+        for module_runtime_state in self.module_runtime_states:
+            if module_runtime_state.name == module_name:
+                return module_runtime_state
+        # Normally we should find the RCP module. When this method is called, we have already called modules_find
+        # before because we had to return the command descriptions to the client. If we cannot find the RCP module
+        # now, the module have been disconnected or the client somehow called a command that does not exist. In any
+        # case, ending up here means we cannot continue.
+        raise ValueError("RCP module not found for command: %s, ", cmd, self.__log_modules_available())
+
+class RcpmCltConnHdlr(CltConnHdlr):
+    """
+    The RCP Module client connection handler is the dedicated client that is used by the RCP Client connection handler
+    to handle the dedicated connection towards the RCP Module (see below)
+    """
+
+class RcpcSrvConnHdlr(SrvConnHdlr):
+    """
+    The RCP Client connection handler takes care of the handling of client requests. Througout the lifetime of a
+    connection, the client will request a description of the available commands and then request the execution of a
+    procedure. To execute the procedure, the handler will make a dedicated connection to the RCP Module and then
+    transparently pass the messages from the RCP Client to the RCP Module and vice versa.
+    """
+
+    async def describe(self):
+        """
+        Collect the command/argument description of suitable modules and forward that definition to the RCP client. The
+        RCP client will then build an argument parser (commmandlien help, argument validation) from this information.
+        """
+        rx_json = await self._recv()
+        self.suitable_for = rx_json['rcpc_hello']['suitable_for']
+        modules = runtime_state.modules_find(self.suitable_for)
+        tx_json = {'rcpc_welcome':
+                   {'module_descr' : modules}
+                   }
+        await self._send(tx_json)
+
+    async def _transact_apdu(self, apdu: Hexstr) -> ResTuple:
+        """Private low level method to exchange an APDU"""
+        tx_json = {'rcpc_instr': {'c_apdu' : apdu.upper()}}
+        rx_json = await self._transact(tx_json)
+        data = rx_json['rcpc_result']['r_apdu']['data']
+        sw = rx_json['rcpc_result']['r_apdu']['sw']
+        return data, sw
+
+    async def _reset(self) -> Hexstr:
+        """Private low level method to reset the UICC/eUICC"""
+        tx_json = {'rcpc_instr': {'reset' : None}}
+        rx_json = await self._transact(tx_json)
+        return rx_json['rcpc_result']['atr']
+
+    async def _read_iccid(self) -> Hexstr:
+        """Private low level method to read the EID from an UICC (or eSIM)"""
+        data, sw = await self._transact_apdu("00A40000022FE200")
+        if sw != "9000":
+            raise ValueError("Unable to select EF.ICCID, sw: %s, " % sw)
+        data, sw = await self._transact_apdu("00B000000A")
+        if sw != "9000":
+            raise ValueError("Unable to read EF.ICCID, sw: %s, " % sw)
+        return dec_iccid(data)
+
+    async def _read_eid(self) -> Hexstr:
+        """Private low level method to read the EID from an eUICC"""
+        data, sw = await self._transact_apdu("00A4040410A0000005591010FFFFFFFF890000010000")
+        if sw != "9000":
+            raise ValueError("Unable to select ISD-R, sw: %s, " % sw)
+        data, sw = await self._transact_apdu("80E2910006BF3E035C015A00")
+        if sw != "9000":
+            raise ValueError("Unable to retrieve EID, sw: %s, " % sw)
+        return data[10:]
+
+    async def print(self, message: str):
+        """ Print a message on the client side """
+        tx_json = {'rcpc_instr': {'print' : message}}
+        rx_json = await self._transact(tx_json)
+        if rx_json != {'rcpc_result': {'empty' : None}}:
+            raise ValueError("unexpected response from RCP Client: %s", rx_json)
+
+    async def procedure(self):
+        """
+        Receive a command from the client, pick a matching module, make a decdicated connection to that module and
+        forward instruction/response messages between RCP Client and RCP Module until the procedure is done.
+        """
+
+        # Expect a command from the client
+        rx_json = await self._recv()
+        if rx_json is None:
+            log.debug(str(self) + " -- RCP client has closed the connection, no procedure executed")
+            return
+        command = rx_json['rcpc_command']
+
+        # Pick the matching RCP Module
+        module = runtime_state.module_find(self.suitable_for, command['cmd'])
+
+        # Retrieve keys (if module requires them)
+        keys = {}
+        if module.retrieve_keys['uicc']:
+            iccid = await self._read_iccid()
+            keys_uicc = card_key_provider_get(module.retrieve_keys['uicc'], 'ICCID', iccid)
+            keys['uicc'] = key_value_pairs_from_dict(keys_uicc, keylabel='key', valuelabel='value')
+        else:
+            keys['uicc'] = []
+        if module.retrieve_keys['euicc']:
+            eid = await self._read_eid()
+            keys_euicc = card_key_provider_get(module.retrieve_keys['euicc'], 'EID', eid)
+            keys['euicc'] = key_value_pairs_from_dict(keys_euicc, keylabel='key', valuelabel='value')
+        else:
+            keys['euicc'] = []
+        command['keys'] = keys
+
+        # Resetting card to ensure the card is in a defined state
+        await self._reset()
+
+        # Transparently forward messages between RCP Client and RCP Module
+        module_uri = "wss://%s:%d" % (module.addr, module.port)
+        log.info(str(self) + " -- executing procedure for command \"%s\" on module \"%s\" at: %s" %
+                 (command['cmd'], module.name, module_uri))
+        async with websockets.connect(module_uri, ssl=rcpm_ca_ssl_context) as websocket:
+            module_client = RcpmCltConnHdlr(websocket, CLIENT_TIMEOUT)
+            rx_json = {'rcpc_command' : command}
+            while(True):
+                module_rx_json = await module_client._transact(rx_json)
+                await self._send(module_rx_json)
+                if 'rcpc_goodbye' in module_rx_json:
+                    log.info(str(self) + " -- command execution done, rc: %d" % module_rx_json['rcpc_goodbye'])
+                    break
+                rx_json = await self._recv()
+            await module_client.close()
+
+class RcpmSrvConnHdlr(SrvConnHdlr):
+    """
+    The RCP Module connection handler is responsible to handle connect and disconnect events of RCP Modules. This
+    connection between the RCP Module and the RCP Server is used for management purposes only.
+    """
+
+    async def describe(self):
+        """
+        Receive the module description from an RCP Module. This description will be stored in an internal list until
+        the module is disconnected from the server.
+        """
+        rx_json = await self._recv()
+        runtime_state.module_add(module = ModuleRuntimeState(self.websocket, **rx_json['rcpm_hello']))
+        tx_json = {'rcpm_welcome': {}}
+        await self._send(tx_json)
+
+    def __del__(self):
+        """
+        Remove RCPM from internal list when the connection is closed (and the handler is deleted)
+        """
+        runtime_state.module_remove(self.websocket)
+        super().__del__()
+
+async def rcpc_conn_hdlr(websocket: ServerConnection):
+    # TODO: Implement some sort of rate limit to protect against DoS. We may count the requests for each requesting
+    # IP address and reject the connection once a certain threshold is reached. (we plan to use the CardKeyProvider
+    # together with a database)
+    try:
+        json_validator = JsonValidator(rcpc_rx_schema, rcpc_tx_schema)
+        hdlr = RcpcSrvConnHdlr(websocket, CLIENT_TIMEOUT, json_validator)
+        await hdlr.describe()
+        await hdlr.procedure()
+        await hdlr.close()
+    except:
+        backtrace("RCPC connection handler")
+
+async def rcpm_conn_hdlr(websocket: ServerConnection):
+    try:
+        hdlr = RcpmSrvConnHdlr(websocket, CLIENT_TIMEOUT)
+        await hdlr.describe()
+        await hdlr.close()
+    except:
+        backtrace("RCPM connection handler")
+
+if __name__ == '__main__':
+    opts = option_parser.parse_args()
+
+    PySimLogger.setup(print, {logging.WARN: "\033[33m", logging.DEBUG: "\033[90m"}, opts.verbose)
+    runtime_state = RuntimeState()
+
+    # TODO: Modularize the JSON schemas. We already repeat ourselves with multiple definitions of the ATR fields.
+    rcpc_rx_schema = load_json_schema(os.path.join(Path(__file__).parent.resolve(), "rcpc_rx_schema.json"))
+    rcpc_tx_schema = load_json_schema(os.path.join(Path(__file__).parent.resolve(), "rcpc_tx_schema.json"))
+
+    # Load SSL/TLS certificates
+    rcpc_ssl_context = load_server_cert("RCP Client Server", opts.rcpc_server_cert)
+    rcpm_ssl_context = load_server_cert("RCP Module Server", opts.rcpm_server_cert)
+    rcpm_ca_ssl_context = load_ca_cert("RCP Module Command Server Client", opts.rcpm_module_ca_cert)
+
+    # Init card key provider for automatic card key retrieval
+    init_card_key_provider(opts)
+
+    # Start RCP server
+    async def rcp_server():
+        log.info("RCP Client Server at: %s:%d" % (opts.rcpc_server_addr, opts.rcpc_server_port))
+        log.info("RCP Module server at: %s:%d" % (opts.rcpm_server_addr, opts.rcpm_server_port))
+        async with serve(rcpc_conn_hdlr, opts.rcpc_server_addr, opts.rcpc_server_port, ssl=rcpc_ssl_context), \
+                   serve(rcpm_conn_hdlr, opts.rcpm_server_addr, opts.rcpm_server_port, ssl=rcpm_ssl_context):
+            await asyncio.get_running_loop().create_future()
+    try:
+        asyncio.run(rcp_server())
+    except SystemExit:
+        pass
+    except:
+        backtrace("RCP Server")
+        sys.exit(1)
+

contrib/rcp/rcp_utils.py

@@ -0,0 +1,254 @@
+#!/usr/bin/env python3
+
+# (C) 2026 by sysmocom - s.f.m.c. GmbH
+# All Rights Reserved
+#
+# Author: Philipp Maier
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import sys
+import ssl
+import json
+import abc
+import asyncio
+import websockets
+import traceback
+import threading
+from copy import deepcopy
+from websockets.asyncio.server import ServerConnection
+from websockets.asyncio.client import ClientConnection
+from pathlib import Path
+from jsonschema import validate
+from pySim.log import PySimLogger
+from ssl import SSLContext
+
+log = PySimLogger.get(Path(__file__).stem)
+
+# TODO: Might be helpful for others as well, move this to pySim.utils?
+def backtrace(what: str):
+    log.error("%s failed with an exception:", what)
+    log.error("---------------------8<---------------------")
+    traceback_lines = traceback.format_exc()
+    for line in traceback_lines.split("\n"):
+        if line:
+            log.error(line)
+    log.error("---------------------8<---------------------")
+
+# TODO: Might be helpful for others as well, move this to pySim.utils?
+def key_value_pairs_from_dict(keys: dict, keylabel: str='key', valuelabel: str='value') -> list:
+    key_list = []
+    for key in keys:
+        key_list.append({keylabel : key, valuelabel : keys[key]})
+    return key_list
+
+# TODO: Might be helpful for others as well, move this to pySim.utils?
+def dict_from_key_value_pairs(keys: list, keylabel: str='key', valuelabel: str='value') -> dict:
+    key_dict = {}
+    for key in keys:
+        key_dict[key[keylabel]] = key[valuelabel]
+    return key_dict
+
+def pytype_to_type(dict_in: dict) -> dict:
+    """
+    There is no way to properly express python types in JSON. This function can be used to replace
+    each ocurrence of "pytype", with "type", where the string type name is replaced with an actual
+    python type.
+    """
+    dict_out = deepcopy(dict_in)
+    if dict_out.get('pytype'):
+        if dict_out['pytype'] == "str":
+            dict_out.pop('pytype')
+            dict_out['type'] = str
+        elif dict_out['pytype'] == "int":
+            dict_out.pop('pytype')
+            dict_out['type'] = int
+        else:
+            raise ValueError("invalid type in command argument specification: %s" % arg['spec']['type'])
+    return dict_out
+
+def load_json_schema(filename: str) -> dict:
+    """Load a JSON schema from file"""
+    log.info("loading JSON schema: %s", filename)
+    try:
+        with open(filename) as schema_file:
+            return json.load(schema_file)
+    except Exception as e:
+        backtrace("JSON schema load")
+        sys.exit(1)
+
+def load_server_cert(what: str, filename: str) -> SSLContext:
+    """Load an SSL/TLS server certificate"""
+    log.info("loading SSL/TLS server certificate (%s): %s", what, filename)
+    ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+    ssl_context.load_cert_chain(filename)
+    return ssl_context
+
+def load_ca_cert(what: str, filename: str) -> SSLContext:
+    """Load an SSL/TLS CA certificate"""
+    log.info("loading SSL/TLS CA certificate (%s): %s", what, filename)
+    ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+    ssl_context.load_verify_locations(filename)
+    return ssl_context
+
+class JsonValidator():
+    """
+    JSON validator class, can be passed to any ConnHdlr object to automatically validate the JSON messages which are
+    sent and and received.
+    """
+
+    def __init__(self, rx_schema: dict, tx_schema: dict = None):
+        self.rx_schema = rx_schema
+        if tx_schema:
+            self.tx_schema = tx_schema
+        else:
+            self.tx_schema = None
+
+    def valid_rx_json(self, rx_json: dict):
+        validate(instance = rx_json, schema = self.rx_schema)
+
+    def valid_tx_json(self, tx_json: dict):
+        if self.tx_schema:
+            # We intentionally do not prevent the sending of an invalid JSON message. It is the responsibility of the
+            # receiving end to detect an invalid message and react accordingly. The purpose of this validation is to
+            # make developers/users aware of the problem.
+            try:
+                validate(instance = tx_json, schema = self.tx_schema)
+            except Exception as e:
+                backtrace("JSON schema validation (TX)")
+
+class ConnHdlr(abc.ABC):
+    """Base class that can be used to create a connection handler"""
+
+    def __init__(self, websocket: ServerConnection | ClientConnection, timeout: int,
+                 json_validator: JsonValidator = None):
+        self.websocket = websocket
+        self.timeout = timeout
+        self.json_validator = json_validator
+        log.debug(str(self) + " -- new handler, timeout: %d sec.", self.timeout)
+
+    def _log_recv_peer(self, rx_json_str: str):
+        peer = "%s:%d<-%s:%d" % (self.websocket.local_address[0],
+                           self.websocket.local_address[1],
+                           self.websocket.remote_address[0],
+                           self.websocket.remote_address[1])
+        log.debug(str(self) + " -- RX(%s): %s", peer, rx_json_str)
+
+    def _log_send_peer(self, tx_json_str: str):
+        peer = "%s:%d->%s:%d" % (self.websocket.local_address[0],
+                           self.websocket.local_address[1],
+                           self.websocket.remote_address[0],
+                           self.websocket.remote_address[1])
+        log.debug(str(self) + " -- TX(%s): %s", peer, tx_json_str)
+
+    def __str__(self) -> str:
+        return "%s(%d)" % (type(self).__name__, id(self))
+
+    def __del__(self):
+        log.debug(str(self) + " -- closed handler")
+
+class SrvConnHdlr(ConnHdlr):
+    """Base class that can be used to create a connection handler for a server"""
+
+    async def _recv(self) -> dict:
+        """Receive JSON message from client"""
+        async with asyncio.timeout(self.timeout):
+            try:
+                rx_json_str = await self.websocket.recv()
+            except websockets.exceptions.ConnectionClosedOK:
+                log.debug(str(self) + " -- no data received, connection is closed")
+                return None
+        self._log_recv_peer(rx_json_str)
+        rx_json = json.loads(rx_json_str)
+        if self.json_validator:
+            self.json_validator.valid_rx_json(rx_json)
+        return rx_json
+
+    async def _send(self, tx_json: dict):
+        """Send JSON message to client"""
+        if self.json_validator:
+            self.json_validator.valid_tx_json(tx_json)
+        tx_json_str = json.dumps(tx_json)
+        self._log_send_peer(tx_json_str)
+        await self.websocket.send(tx_json_str)
+
+    async def _transact(self, tx_json: dict) -> dict:
+        """Exchange JSON message with client"""
+        await self._send(tx_json)
+        return await self._recv()
+
+    async def close(self):
+        """Wait for a connecion to close normally"""
+        await self.websocket.wait_closed()
+        log.debug(str(self) + " -- closed connection")
+
+class SrvSyncConnHdlr(ConnHdlr):
+    """Base class that can be used to create a synchronous connection handler for a server"""
+
+    def _recv(self) -> dict:
+        """Receive JSON message from client"""
+        # TODO: we do not have a timeout here (the self.timeout is currently useless). Check if we can do something
+        # about this or if we have to implement some watchdog functionality elsewhere.
+        rx_json_str = self.websocket.recv()
+        self._log_recv_peer(rx_json_str)
+        rx_json = json.loads(rx_json_str)
+        if self.json_validator:
+            self.json_validator.valid_rx_json(rx_json)
+        return rx_json
+
+    def _send(self, tx_json: dict):
+        """Send JSON message to client"""
+        if self.json_validator:
+            self.json_validator.valid_tx_json(tx_json)
+        tx_json_str = json.dumps(tx_json)
+        self._log_send_peer(tx_json_str)
+        self.websocket.send(tx_json_str)
+
+    def _transact(self, tx_json: dict) -> dict:
+        """Exchange JSON message with client"""
+        self._send(tx_json)
+        return self._recv()
+
+    def close(self):
+        """Close connection normally"""
+        self.websocket.close()
+        log.debug(str(self) + " -- closed connection")
+
+class CltConnHdlr(ConnHdlr):
+    """Base class that can be used to create a connection handler for a client"""
+
+    async def _transact(self, tx_json: dict) -> dict:
+        """Exchange JSON message with server"""
+        if self.json_validator:
+            self.json_validator.valid_tx_json(tx_json)
+        tx_json_str = json.dumps(tx_json)
+        self._log_send_peer(tx_json_str)
+        async with asyncio.timeout(self.timeout):
+            await self.websocket.send(tx_json_str)
+            rx_json_str = await self.websocket.recv()
+        self._log_recv_peer(rx_json_str)
+        rx_json = json.loads(rx_json_str);
+        if self.json_validator:
+            self.json_validator.valid_rx_json(rx_json)
+        return rx_json
+
+    async def close(self):
+        """Close connection normally"""
+        await self.websocket.close()
+        log.debug(str(self) + " -- closed connection")
+
+    async def wait_close(self):
+        """Wait for a connecion to close normally"""
+        await self.websocket.wait_closed()
+        log.debug(str(self) + " -- closed connection")

contrib/rcp/rcpc_rx_schema.json

@@ -0,0 +1,69 @@
+{
+    "$schema": "https://json-schema.org/draft/2020-12/schema",
+    "title": "RCP Client RX",
+    "type": "object",
+    "properties": {
+	"rcpc_hello": {
+	    "type": "object",
+	    "properties": {
+		"suitable_for": {
+		    "type": "object",
+		    "properties": {
+			"atr": {
+			    "type": "string",
+			    "pattern": "^[0-9,A-F]{0,66}$"
+			}
+		    },
+		    "additionalProperties": false
+		}
+	    },
+	    "additionalProperties": false
+	},
+	"rcpc_command": {
+	    "type": "object",
+	    "properties": {
+		"cmd": {
+		    "type": "string",
+		    "pattern": "^[0-9,A-Z,a-z,_]{0,40}$"
+		},
+		"cmd_argv": {
+		    "type": "array",
+		    "items": {
+			"type": "string",
+			"pattern": "^.{0,512}$"
+		    },
+		    "maxItems": 255
+		}
+	    },
+	    "additionalProperties": false
+	},
+	"rcpc_result": {
+	    "type": "object",
+	    "properties": {
+		"r_apdu": {
+		    "type": "object",
+		    "properties": {
+			"data": {
+			    "type": "string",
+			    "pattern": "^[0-9,A-F]{0,512}$"
+			},
+			"sw": {
+			    "type": "string",
+			    "pattern": "^[0-9,A-F]{0,4}$"
+			}
+		    },
+		    "additionalProperties": false
+		},
+		"atr": {
+		    "type": "string",
+		    "pattern": "^[0-9,A-F]{0,66}$"
+		},
+		"empty": {
+		    "type": "null"
+		}
+	    },
+	    "additionalProperties": false
+	}
+    },
+    "additionalProperties": false
+}

contrib/rcp/rcpc_tx_schema.json

@@ -0,0 +1,89 @@
+{
+    "$schema": "https://json-schema.org/draft/2020-12/schema",
+    "title": "RCP Client TX",
+    "type": "object",
+    "properties": {
+	"rcpc_welcome": {
+	    "type": "object",
+	    "properties": {
+		"module_descr": {
+		    "type": "array",
+		    "items": {
+			"type": "object",
+			"properties": {
+			    "name": {
+				"type": "string"
+			    },
+			    "cmd_descr": {
+				"type": "array",
+				"items": {
+				    "type": "object",
+				    "properties": {
+					"name": {
+					    "type": "string"
+					},
+					"help": {
+					    "type": "string"
+					},
+					"args": {
+					    "type": "array",
+					    "items": {
+						"name": {
+						    "type": "string"
+						},
+						"spec": {
+						    "type": "object",
+						    "properties": {
+							"required" : {
+							    "type": "boolean"
+							},
+							"help": {
+							    "type": "string"
+							},
+							"action": {
+							    "type": "string"
+							},
+							"pytype": {
+							    "type": "string"
+							},
+							"default" : {
+							    "type": ["string", "integer"]
+							}
+						    },
+						    "additionalProperties": false
+						}
+					    }
+					}
+				    },
+				    "additionalProperties": false
+				}
+			    }
+			},
+			"additionalProperties": false
+		    }
+		}
+	    },
+	    "additionalProperties": false
+	},
+	"rcpc_instr": {
+	    "type": "object",
+	    "properties": {
+		"print": {
+		    "type": "string"
+		},
+		"reset": {
+		    "type": "null"
+		},
+		"c_apdu": {
+		    "type": "string",
+		    "pattern": "^[0-9,A-F]{0,512}$"
+		}
+	    },
+	    "additionalProperties": false
+	},
+	"rcpc_goodbye": {
+	    "type": "integer"
+	}
+    },
+    "additionalProperties": false
+}

contrib/rcp/usage_example/certs/example_ssl_rcp_ca_cert.crt

@@ -0,0 +1,20 @@
+-----BEGIN CERTIFICATE-----
+MIIDSzCCAjOgAwIBAgIUEv1f0yjVtkr+RNYLItZ33eTJwHMwDQYJKoZIhvcNAQEL
+BQAwFjEUMBIGA1UEAwwLRWFzeS1SU0EgQ0EwHhcNMjYwNDI5MTIwOTM0WhcNMzYw
+NDI2MTIwOTM0WjAWMRQwEgYDVQQDDAtFYXN5LVJTQSBDQTCCASIwDQYJKoZIhvcN
+AQEBBQADggEPADCCAQoCggEBANXdkSyQlDzuo2cJmnBmFiZpc0V9tYBcNkpZd3Ac
+R0WljazKKgXDWNmOcSO7891bi+1HZzz+nDfV0mJY776ScGkTqF43Hzpg9eZakMAx
+yC24mT4h+uyRcPWZrBwaQhpiQrvZy4MRyuUB+BEgBSmhoDiuXP44kWiuEJHuzpOq
+X6Q2dW8RIeQPDGGK6XPZIQLqx+krxkaqphd/vHgT1/yd7Ol5xxMc4x2UuPaVCj0D
+OzslFsbb0Zu77ffCtHOVVnzSCzeEGGx1MPQm6hDVW+KUXXTwke1K55fmFZhu0gKO
+HYSEjgPj6X8muDb+GvOAQX3fHmS6KvFS4fwWd2InZ3v2f3cCAwEAAaOBkDCBjTAM
+BgNVHRMEBTADAQH/MB0GA1UdDgQWBBS6zY4Dd0pJFrvWLmyjn0vDTFqVqzBRBgNV
+HSMESjBIgBS6zY4Dd0pJFrvWLmyjn0vDTFqVq6EapBgwFjEUMBIGA1UEAwwLRWFz
+eS1SU0EgQ0GCFBL9X9Mo1bZK/kTWCyLWd93kycBzMAsGA1UdDwQEAwIBBjANBgkq
+hkiG9w0BAQsFAAOCAQEAGJUXlbnVhh+xL+pyTyjwtd8nxhUcHzYZl+OT0bkGY9zT
+S3NjHkKBbdnEftuYDYqp0uBuGFQ1WIOKiM3rp4IePKe84lSivZMVh9ObtNalcEQr
+sqxBziNOMJM2mh5V2NdxiK2E1gCZ959wOQ8yzM6gGC+wW8w4zwULhv4JimQDjk+G
+kAdiGL7+WAxrNWUulvm8khFt2nOlucJg4IAYVt2SI1AFMt/YSXoA4wMwM9QcHGj0
+1A069IxX93WVhUpIL1Avwz+KJK0BPY6SM8LYUy6V50Hojp76BB7VD6SxQrSoceUo
+6cRNDtCmofOlltfeUJLr1mI4S2tM50bQVsHD92EJBA==
+-----END CERTIFICATE-----

contrib/rcp/usage_example/certs/example_ssl_rcpc_rcps_cert.pem

@@ -0,0 +1,115 @@
+Certificate:
+    Data:
+        Version: 3 (0x2)
+        Serial Number:
+            42:38:a5:6f:70:53:40:e4:a4:1a:2c:0f:fc:81:13:42
+        Signature Algorithm: sha256WithRSAEncryption
+        Issuer: CN=Easy-RSA CA
+        Validity
+            Not Before: Apr 29 12:09:35 2026 GMT
+            Not After : Aug  1 12:09:35 2028 GMT
+        Subject: CN=example_ssl_rcpc_rcps_cert
+        Subject Public Key Info:
+            Public Key Algorithm: rsaEncryption
+                Public-Key: (2048 bit)
+                Modulus:
+                    00:ae:0f:e1:ee:fc:f6:db:75:45:c0:f4:49:72:46:
+                    3d:e3:db:0c:c4:34:d2:9e:49:d4:86:4f:19:0d:55:
+                    70:50:81:e4:e6:64:56:a8:58:e8:e6:54:0a:16:bc:
+                    f4:4b:84:cd:1d:b9:2e:ed:62:b6:cd:62:35:8b:81:
+                    18:ab:ff:63:f5:c1:dc:16:3e:a8:dc:ac:11:dd:43:
+                    12:f8:ef:f2:f1:af:84:fd:83:fe:a8:d3:46:7d:77:
+                    e6:ae:95:61:a6:c9:99:6b:40:61:8d:6e:7e:66:1e:
+                    97:77:b0:e8:b7:3d:3a:d5:d7:d3:ee:66:95:62:83:
+                    14:cc:5e:32:ff:9e:bd:f1:06:e6:8d:6a:7c:0a:27:
+                    22:19:b9:06:09:cf:ef:c7:dc:e8:8f:04:4b:83:0d:
+                    cc:8d:b1:c2:cf:ab:40:25:6e:f2:bf:b7:c6:1d:8f:
+                    d2:fc:3d:c8:a1:be:4a:09:b9:91:e3:76:4f:c7:9b:
+                    fc:2f:de:d9:bb:eb:df:d3:d8:8c:72:79:bd:bf:10:
+                    8b:01:e6:0f:7f:bb:f6:75:31:5a:40:ad:df:e1:07:
+                    e6:12:12:b2:d3:99:d0:bd:24:5a:9a:ce:62:4f:da:
+                    fe:0d:df:09:ae:da:04:83:54:e8:cb:68:c0:57:78:
+                    c2:f4:68:42:d7:f4:81:4a:a3:b4:4e:0b:49:95:26:
+                    1d:15
+                Exponent: 65537 (0x10001)
+        X509v3 extensions:
+            X509v3 Basic Constraints: 
+                CA:FALSE
+            X509v3 Subject Key Identifier: 
+                8E:99:9D:C0:70:98:57:16:08:8E:DF:6E:51:78:A6:86:18:FF:06:52
+            X509v3 Authority Key Identifier: 
+                keyid:BA:CD:8E:03:77:4A:49:16:BB:D6:2E:6C:A3:9F:4B:C3:4C:5A:95:AB
+                DirName:/CN=Easy-RSA CA
+                serial:12:FD:5F:D3:28:D5:B6:4A:FE:44:D6:0B:22:D6:77:DD:E4:C9:C0:73
+            X509v3 Extended Key Usage: 
+                TLS Web Server Authentication
+            X509v3 Key Usage: 
+                Digital Signature, Key Encipherment
+            X509v3 Subject Alternative Name: 
+                DNS:127.0.0.1, IP Address:127.0.0.1
+    Signature Algorithm: sha256WithRSAEncryption
+    Signature Value:
+        3e:56:20:f9:3b:fa:13:6e:7e:a9:80:a6:15:18:01:82:f1:b8:
+        4d:1b:f1:ee:da:ed:50:f7:3b:13:01:a5:14:f9:4c:0e:34:57:
+        dc:e6:d1:7e:02:30:af:3b:fd:c9:ae:18:16:c9:3b:0a:4e:20:
+        da:cd:e8:cc:05:0c:b3:7d:6f:e5:15:ff:66:59:6b:fe:ff:1a:
+        ef:ca:b5:3a:1a:ad:dd:f6:19:43:d9:2b:61:18:29:95:b4:0c:
+        1e:b2:4a:ce:80:d3:1b:59:dc:62:ec:50:21:37:9c:2f:7a:4d:
+        c2:ac:de:1b:1d:a3:25:e0:e8:33:42:cf:77:31:2a:f2:44:36:
+        ef:59:89:da:6c:3e:9a:e8:d7:06:39:17:d5:78:82:6d:b6:63:
+        3f:9a:40:3b:e6:12:58:52:3d:63:4e:85:0b:02:cb:40:d2:8a:
+        59:8d:8f:ee:4a:c8:97:91:51:a9:2f:1b:15:81:9c:20:dd:94:
+        08:6f:ac:fa:c6:28:90:6c:17:5a:23:87:9a:5b:e5:c6:2e:f3:
+        09:66:de:76:1b:60:42:c1:5c:71:88:87:f6:7b:cb:e3:7e:14:
+        67:c9:a0:15:98:b6:7b:75:40:9a:08:fc:77:39:3a:23:cb:e3:
+        78:7d:57:f9:a7:66:36:b4:b5:07:de:61:3a:dd:07:58:b3:4f:
+        41:f6:f4:d9
+-----BEGIN CERTIFICATE-----
+MIIDhDCCAmygAwIBAgIQQjilb3BTQOSkGiwP/IETQjANBgkqhkiG9w0BAQsFADAW
+MRQwEgYDVQQDDAtFYXN5LVJTQSBDQTAeFw0yNjA0MjkxMjA5MzVaFw0yODA4MDEx
+MjA5MzVaMCUxIzAhBgNVBAMMGmV4YW1wbGVfc3NsX3JjcGNfcmNwc19jZXJ0MIIB
+IjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEArg/h7vz223VFwPRJckY949sM
+xDTSnknUhk8ZDVVwUIHk5mRWqFjo5lQKFrz0S4TNHbku7WK2zWI1i4EYq/9j9cHc
+Fj6o3KwR3UMS+O/y8a+E/YP+qNNGfXfmrpVhpsmZa0BhjW5+Zh6Xd7Dotz061dfT
+7maVYoMUzF4y/5698QbmjWp8CiciGbkGCc/vx9zojwRLgw3MjbHCz6tAJW7yv7fG
+HY/S/D3Iob5KCbmR43ZPx5v8L97Zu+vf09iMcnm9vxCLAeYPf7v2dTFaQK3f4Qfm
+EhKy05nQvSRams5iT9r+Dd8JrtoEg1Toy2jAV3jC9GhC1/SBSqO0TgtJlSYdFQID
+AQABo4G+MIG7MAkGA1UdEwQCMAAwHQYDVR0OBBYEFI6ZncBwmFcWCI7fblF4poYY
+/wZSMFEGA1UdIwRKMEiAFLrNjgN3SkkWu9YubKOfS8NMWpWroRqkGDAWMRQwEgYD
+VQQDDAtFYXN5LVJTQSBDQYIUEv1f0yjVtkr+RNYLItZ33eTJwHMwEwYDVR0lBAww
+CgYIKwYBBQUHAwEwCwYDVR0PBAQDAgWgMBoGA1UdEQQTMBGCCTEyNy4wLjAuMYcE
+fwAAATANBgkqhkiG9w0BAQsFAAOCAQEAPlYg+Tv6E25+qYCmFRgBgvG4TRvx7trt
+UPc7EwGlFPlMDjRX3ObRfgIwrzv9ya4YFsk7Ck4g2s3ozAUMs31v5RX/Zllr/v8a
+78q1Ohqt3fYZQ9krYRgplbQMHrJKzoDTG1ncYuxQITecL3pNwqzeGx2jJeDoM0LP
+dzEq8kQ271mJ2mw+mujXBjkX1XiCbbZjP5pAO+YSWFI9Y06FCwLLQNKKWY2P7krI
+l5FRqS8bFYGcIN2UCG+s+sYokGwXWiOHmlvlxi7zCWbedhtgQsFccYiH9nvL434U
+Z8mgFZi2e3VAmgj8dzk6I8vjeH1X+admNrS1B95hOt0HWLNPQfb02Q==
+-----END CERTIFICATE-----
+-----BEGIN PRIVATE KEY-----
+MIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQCuD+Hu/PbbdUXA
+9ElyRj3j2wzENNKeSdSGTxkNVXBQgeTmZFaoWOjmVAoWvPRLhM0duS7tYrbNYjWL
+gRir/2P1wdwWPqjcrBHdQxL47/Lxr4T9g/6o00Z9d+aulWGmyZlrQGGNbn5mHpd3
+sOi3PTrV19PuZpVigxTMXjL/nr3xBuaNanwKJyIZuQYJz+/H3OiPBEuDDcyNscLP
+q0AlbvK/t8Ydj9L8PcihvkoJuZHjdk/Hm/wv3tm769/T2Ixyeb2/EIsB5g9/u/Z1
+MVpArd/hB+YSErLTmdC9JFqazmJP2v4N3wmu2gSDVOjLaMBXeML0aELX9IFKo7RO
+C0mVJh0VAgMBAAECggEAGBhmQhdeE+Cu1Ihsn2dWW3PAF2wpiNR3GVWbRfOBHf/x
+QCx9K4ZNTU8ua1niZo7edyIiuyVaYWGaQHLRR8QNoiBhN2oapZujSHInzvKmeqr9
+ubt7NgMzQ5ykwB+5OiW3uXda2cGFOV08QgspF+6ftakQMzUbslyrdSQIIscmi5Ya
+uTDvE6+lBFinxy3RHFKVCZ3UrsDwfHR4eTUmgCHRB27joB7DFXL32amv0M8HjoGz
+EZKGJgTwmRf9U4z4D4wCnOfVAPlsuthKUqMuTlBg0ZEstMZrzlP4suT2ieku0Usv
+0XbJ38VozPYYFdR7nApVVvrJgHzI9cpoUbGto4BLOQKBgQDbXjFVLffOec8hv9dN
+2VGZQmK61S9OrbvTnEOlxJd+kRid71X1pV5TuPKJJQtUJXf429bQOs40YbLeOmJt
+BiRSR5yIBH7hDDC/c0ynqunstwDlgz+QX2Oh2B4alvVaWy0rZYF6NpBiI0+R5r5V
+C2fHRS4LLPoflg83+CMubyLS6QKBgQDLIOXxlp1JQTzXhJkrkytLkafmEHAafovt
+wbRD50/s+dl16BRX12sK0gXj2vwu0FleUD6Z7afDfspmvQdg3fyDxYw9Q+vw5LYQ
+7BvoVU99o1m468yXwX/v36peCt4nOpwkJZKJfjgxjnMJByyeSUgL9uW4K+0D0LBV
+a5Iv7QklTQKBgH30BkVPIHKIE/rfyIJlXemuaTu2/fOh4y9sEJdUWluMeeLssaFa
+ct+FWJSQFYIaBVl4+E0VBqKi2e2o/ix1E1O+1ExwsF0M/8xdKk024BtPNA+TnWKK
+so0Rpq9Dr9pScYvyOzZtr9b5SU2PfAcehlavDPHTwEV0hoZvTdvyab9JAoGADMBJ
+7vp3cSvJN/Y470VTyHCiS4zonKEpA4nPWRviJowgnIgvDryVGZ7Jg94xSncFxSfg
+ZiVHDLye1Ag1uFz3BwaVoRrsarjQvQs1TUZdsRNaBIO42iXpdBNkTHb+LxQ8zQAW
+zM7BlErO6dgrctxCy416Ki+Ht1+YUiRojt2gX1kCgYBqytUy+XkPi5j3Ga29xcvP
+WI3Uc8RI2GmoAmrw5QFiSG6lNXAzfo2ZNQbFnxgxeMOG9fV9yzBdIjXWNWr0E/KH
+Fsb65R8iIrXQB9BZjuQqjz9nDm7eZZUBNGGbQ4DgSepnp194gXC5DoAElzuwOXbE
+pY/kM1KwlpUR3J3LeF3i+Q==
+-----END PRIVATE KEY-----

contrib/rcp/usage_example/certs/example_ssl_rcpm_rcps_cert.pem

@@ -0,0 +1,115 @@
+Certificate:
+    Data:
+        Version: 3 (0x2)
+        Serial Number:
+            e7:09:ab:70:b5:dc:1f:11:d9:2a:23:04:39:87:34:f3
+        Signature Algorithm: sha256WithRSAEncryption
+        Issuer: CN=Easy-RSA CA
+        Validity
+            Not Before: Apr 29 12:09:35 2026 GMT
+            Not After : Aug  1 12:09:35 2028 GMT
+        Subject: CN=example_ssl_rcpm_rcps_cert
+        Subject Public Key Info:
+            Public Key Algorithm: rsaEncryption
+                Public-Key: (2048 bit)
+                Modulus:
+                    00:cc:79:9b:d3:f3:1f:41:9f:00:48:cd:47:0b:ae:
+                    b9:1c:4e:3e:55:e2:4e:5f:a8:cc:13:d5:dd:bd:f0:
+                    01:4c:19:ae:e3:a9:09:06:89:92:49:f7:bb:90:28:
+                    fb:8c:22:69:b5:f5:a0:50:3d:97:0f:1e:1d:b1:a8:
+                    57:9b:d7:e2:0d:99:67:7f:02:82:0c:9c:8e:dd:13:
+                    03:28:93:b5:cb:7e:b5:78:06:10:bf:7b:55:c3:f7:
+                    10:8b:20:4a:1c:f9:f1:b2:fa:f1:c7:44:9d:0a:ce:
+                    ef:8d:f9:e8:ff:d1:c1:69:ec:8e:5f:11:cc:c9:98:
+                    d5:1c:33:e2:5b:7a:4d:34:dc:76:c3:cd:db:4c:93:
+                    d1:08:78:6f:3c:9a:ee:74:39:1e:cd:65:1e:c9:35:
+                    cc:3b:2b:9e:d7:49:10:8e:58:85:b0:10:5b:90:1e:
+                    f1:5e:d5:92:04:93:f9:33:c6:9d:77:63:d1:33:46:
+                    5b:98:ff:9a:a8:f5:df:f7:84:21:e2:88:28:7a:a4:
+                    c6:0d:9f:25:7e:0d:73:5b:d5:53:4a:90:79:94:37:
+                    14:f3:c8:75:76:d4:1c:32:51:bf:58:16:74:d5:8d:
+                    18:b6:53:f4:ab:cb:91:a8:8c:a3:ca:3c:5c:35:b6:
+                    5f:62:57:37:5a:75:28:b7:4d:26:aa:ea:50:da:a4:
+                    1c:55
+                Exponent: 65537 (0x10001)
+        X509v3 extensions:
+            X509v3 Basic Constraints: 
+                CA:FALSE
+            X509v3 Subject Key Identifier: 
+                47:92:B5:81:8B:5C:14:98:B3:83:B6:EB:06:9F:43:F3:3A:7E:ED:24
+            X509v3 Authority Key Identifier: 
+                keyid:BA:CD:8E:03:77:4A:49:16:BB:D6:2E:6C:A3:9F:4B:C3:4C:5A:95:AB
+                DirName:/CN=Easy-RSA CA
+                serial:12:FD:5F:D3:28:D5:B6:4A:FE:44:D6:0B:22:D6:77:DD:E4:C9:C0:73
+            X509v3 Extended Key Usage: 
+                TLS Web Server Authentication
+            X509v3 Key Usage: 
+                Digital Signature, Key Encipherment
+            X509v3 Subject Alternative Name: 
+                DNS:127.0.0.1, IP Address:127.0.0.1
+    Signature Algorithm: sha256WithRSAEncryption
+    Signature Value:
+        6d:31:e6:29:d2:3b:a8:90:5c:4b:ac:61:15:95:5d:70:66:a5:
+        77:9d:88:47:49:73:75:be:70:69:d8:2f:62:82:5e:83:86:3b:
+        a8:48:3f:f1:5f:22:ae:81:23:64:c4:f2:2b:dd:4d:be:e5:6a:
+        26:a5:ea:c7:ba:1b:3e:6a:34:03:5a:f1:49:28:5f:56:4a:a6:
+        0e:1b:7a:07:48:76:95:b6:4b:f5:3f:b9:67:2e:e0:33:06:80:
+        d4:d6:01:a5:76:01:c0:a5:18:e5:38:8b:52:73:6e:6d:45:50:
+        b7:9a:ab:86:5d:e3:65:b4:b8:c7:ee:b2:dc:bf:e3:d5:bb:e4:
+        91:eb:f5:0c:38:22:5e:37:54:9e:ba:96:25:10:04:18:23:f7:
+        ae:73:4d:d0:aa:03:81:b4:89:36:97:15:da:1a:60:a0:98:5f:
+        03:f8:1b:22:83:57:41:4b:12:28:7d:8d:ea:88:74:24:28:5c:
+        53:41:89:5e:9a:da:fd:7b:bf:60:dc:de:9b:49:ce:5c:a3:b2:
+        01:7d:1d:cb:28:8c:ba:f4:7b:5d:2b:cb:15:5b:2a:97:1a:d1:
+        f9:e7:12:e3:43:b9:f4:2a:88:dd:6d:b6:a0:72:d3:bd:63:23:
+        e9:d7:f0:ac:b5:6d:0d:f2:d9:8b:2c:c4:35:5b:4d:83:dc:e8:
+        7d:0b:3d:a3
+-----BEGIN CERTIFICATE-----
+MIIDhTCCAm2gAwIBAgIRAOcJq3C13B8R2SojBDmHNPMwDQYJKoZIhvcNAQELBQAw
+FjEUMBIGA1UEAwwLRWFzeS1SU0EgQ0EwHhcNMjYwNDI5MTIwOTM1WhcNMjgwODAx
+MTIwOTM1WjAlMSMwIQYDVQQDDBpleGFtcGxlX3NzbF9yY3BtX3JjcHNfY2VydDCC
+ASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAMx5m9PzH0GfAEjNRwuuuRxO
+PlXiTl+ozBPV3b3wAUwZruOpCQaJkkn3u5Ao+4wiabX1oFA9lw8eHbGoV5vX4g2Z
+Z38Cggycjt0TAyiTtct+tXgGEL97VcP3EIsgShz58bL68cdEnQrO74356P/RwWns
+jl8RzMmY1Rwz4lt6TTTcdsPN20yT0Qh4bzya7nQ5Hs1lHsk1zDsrntdJEI5YhbAQ
+W5Ae8V7VkgST+TPGnXdj0TNGW5j/mqj13/eEIeKIKHqkxg2fJX4Nc1vVU0qQeZQ3
+FPPIdXbUHDJRv1gWdNWNGLZT9KvLkaiMo8o8XDW2X2JXN1p1KLdNJqrqUNqkHFUC
+AwEAAaOBvjCBuzAJBgNVHRMEAjAAMB0GA1UdDgQWBBRHkrWBi1wUmLODtusGn0Pz
+On7tJDBRBgNVHSMESjBIgBS6zY4Dd0pJFrvWLmyjn0vDTFqVq6EapBgwFjEUMBIG
+A1UEAwwLRWFzeS1SU0EgQ0GCFBL9X9Mo1bZK/kTWCyLWd93kycBzMBMGA1UdJQQM
+MAoGCCsGAQUFBwMBMAsGA1UdDwQEAwIFoDAaBgNVHREEEzARggkxMjcuMC4wLjGH
+BH8AAAEwDQYJKoZIhvcNAQELBQADggEBAG0x5inSO6iQXEusYRWVXXBmpXediEdJ
+c3W+cGnYL2KCXoOGO6hIP/FfIq6BI2TE8ivdTb7laial6se6Gz5qNANa8UkoX1ZK
+pg4begdIdpW2S/U/uWcu4DMGgNTWAaV2AcClGOU4i1Jzbm1FULeaq4Zd42W0uMfu
+sty/49W75JHr9Qw4Il43VJ66liUQBBgj965zTdCqA4G0iTaXFdoaYKCYXwP4GyKD
+V0FLEih9jeqIdCQoXFNBiV6a2v17v2Dc3ptJzlyjsgF9HcsojLr0e10ryxVbKpca
+0fnnEuNDufQqiN1ttqBy071jI+nX8Ky1bQ3y2YssxDVbTYPc6H0LPaM=
+-----END CERTIFICATE-----
+-----BEGIN PRIVATE KEY-----
+MIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQDMeZvT8x9BnwBI
+zUcLrrkcTj5V4k5fqMwT1d298AFMGa7jqQkGiZJJ97uQKPuMImm19aBQPZcPHh2x
+qFeb1+INmWd/AoIMnI7dEwMok7XLfrV4BhC/e1XD9xCLIEoc+fGy+vHHRJ0Kzu+N
++ej/0cFp7I5fEczJmNUcM+Jbek003HbDzdtMk9EIeG88mu50OR7NZR7JNcw7K57X
+SRCOWIWwEFuQHvFe1ZIEk/kzxp13Y9EzRluY/5qo9d/3hCHiiCh6pMYNnyV+DXNb
+1VNKkHmUNxTzyHV21BwyUb9YFnTVjRi2U/Sry5GojKPKPFw1tl9iVzdadSi3TSaq
+6lDapBxVAgMBAAECggEADJFN6K9OWhYX1PcEWUgOLxqdCLd95Iccsfxot7ekcMUP
+A4WnHRyACLqor9c2V3o2//IpU2fnB2IXu6ISmRd3WKl3hm4vnZmoIJeTpQm9Iv/g
++fqkyrbIgktcHDJUySal+n+jiYFNW2B1h1xXUT/scMz+FthNJg1Azfi0vorMFjCk
+SBOSo7BQ2hiQ83FneVJU1TsxD4S4IBLx9fF6AW05norRmvm17Ip2pKk4QYzKiOI3
+NIoSwbOgU0Vp1X3MMlilM5ZZdN3a9lI6lfBZE682WOxBmH67mKMQnR8aC+nwynQ+
+45pUQIn8Fjx2hQHehbD7ZNw9Nob9AyGWqWGKFV74IQKBgQDox/j7dCHNm1mz1QRm
+Q2YyN4OGXJI97t9Gd8UdNCv5bAkdx2cR2lmP3tRc6iAzrfNIpPCGaXo1vwJx477X
+wO95W2b4hfm3j6v0cqRbsFzzVIHZUB77pXfAJfZeICpoqu0vxn5nb+yPgzgmToLX
+pbIDdqWafzzrDLTmLCfwfKDI9QKBgQDg3tgpAXo8WCL4phNuW44XQ172lPTijpn+
+wj1Z0rBrS806gL/+QvZZLS1WCym/QBV7TgGlxIJbmAfghcGyin3NliskSHAiccxG
+/9eCSQes8czfsVj6qmBMwyff5r+wmk662qV0u07UHmuykYk3Dgs/zYdwq2SsTlL4
+Y9eRjutp4QKBgQCONg0wYcR8/hmROeRULXzz1OJvZYKaf6K8RFOSAduTp6LyJG4d
+hA4PTQzkLsy5hd4JVWr0UuAskaMGvSJMYTxsIaEI16C1ufpNfvRWZ6qBpfEmOEKV
+boN4Sjj3TCNcioAZHeT/gGs/SeU10eUxpbLZVtTZTD6FQuAJdpR34UvBOQKBgFNM
+mXxPLM2vxHyhYK9PwQoDDel/8lr+gjMqFvnwHyQP911FllmEyqbsIlAuYG+VOJ/t
+nJSgf72YSsq0IbWWsdV3XFHbd5Z62zYtzdJYZTx+cesnUhPBC11EKcA6RSYRczqq
+hgIA5MmU30ZNvSukyyv+Yb6t7uQZO4kByzgDXldhAoGBALgRkAHxgbKUXp5XyCDJ
+e8dwVx0g9tfDM/DEZtU/Si5oUaunBaPV/Byov7OXOT02V8JLnA5ChUYgwUFI030x
+QL/3eK12Qh5Gb9VabvYCicDRk4GzmqZU9Wcvm1zgbUr5jY8Lou44nFjol/Y1m70n
+51WZbVkkWmBZO5m3NqN66SkJ
+-----END PRIVATE KEY-----

contrib/rcp/usage_example/certs/example_ssl_rcps_rcpm_cert.pem

@@ -0,0 +1,115 @@
+Certificate:
+    Data:
+        Version: 3 (0x2)
+        Serial Number:
+            28:96:2e:a1:40:e0:7e:f1:fb:63:1a:f4:53:6f:ce:fb
+        Signature Algorithm: sha256WithRSAEncryption
+        Issuer: CN=Easy-RSA CA
+        Validity
+            Not Before: Apr 29 12:09:35 2026 GMT
+            Not After : Aug  1 12:09:35 2028 GMT
+        Subject: CN=example_ssl_rcps_rcpm_cert
+        Subject Public Key Info:
+            Public Key Algorithm: rsaEncryption
+                Public-Key: (2048 bit)
+                Modulus:
+                    00:ae:11:46:ef:d1:81:34:dd:23:5d:54:40:f3:9c:
+                    85:35:95:a6:91:57:92:5c:bf:eb:40:34:69:eb:db:
+                    c0:86:3c:7b:ff:9c:d7:ba:0e:41:57:84:15:cd:94:
+                    f1:48:63:50:9c:34:97:ee:be:be:b0:27:d8:fd:cd:
+                    8a:cf:85:ff:08:1f:07:d8:28:96:0e:e4:2d:d0:8b:
+                    df:a8:fa:41:47:a0:a2:80:2e:2e:58:01:cc:6f:43:
+                    5c:c2:fb:84:a7:ff:9e:97:bb:b3:a3:1f:63:64:73:
+                    8d:73:dd:f4:7e:96:d7:6b:b3:cb:e2:35:59:55:e0:
+                    e7:e3:c0:41:f8:b6:0f:c5:46:4c:cd:0e:91:80:ef:
+                    e3:43:f0:72:26:12:10:be:83:a2:db:23:2d:b4:b1:
+                    07:5a:b1:b3:10:9c:09:69:98:42:79:81:77:5e:22:
+                    e4:71:47:70:27:15:2c:a7:13:c2:6d:44:59:b4:73:
+                    c9:bb:27:7f:d6:e8:3d:85:bb:36:f6:cb:71:36:11:
+                    b1:99:1a:1d:1a:15:dd:cd:65:7f:cd:cc:10:00:49:
+                    ed:07:2d:7b:15:88:be:73:ba:1d:15:69:bc:d3:02:
+                    55:ea:dc:2c:3f:0b:cd:18:57:59:7a:e3:09:b2:89:
+                    cd:d6:e7:f6:95:c4:2e:8a:53:2b:a8:96:82:94:53:
+                    00:77
+                Exponent: 65537 (0x10001)
+        X509v3 extensions:
+            X509v3 Basic Constraints: 
+                CA:FALSE
+            X509v3 Subject Key Identifier: 
+                60:BD:48:06:68:15:D4:DC:ED:EE:E4:C7:B1:9F:C4:93:6D:50:3A:77
+            X509v3 Authority Key Identifier: 
+                keyid:BA:CD:8E:03:77:4A:49:16:BB:D6:2E:6C:A3:9F:4B:C3:4C:5A:95:AB
+                DirName:/CN=Easy-RSA CA
+                serial:12:FD:5F:D3:28:D5:B6:4A:FE:44:D6:0B:22:D6:77:DD:E4:C9:C0:73
+            X509v3 Extended Key Usage: 
+                TLS Web Server Authentication
+            X509v3 Key Usage: 
+                Digital Signature, Key Encipherment
+            X509v3 Subject Alternative Name: 
+                DNS:127.0.0.1, IP Address:127.0.0.1
+    Signature Algorithm: sha256WithRSAEncryption
+    Signature Value:
+        c5:35:61:58:23:e2:69:da:6c:d5:41:ab:a8:70:f4:dd:cc:a0:
+        a3:3d:84:89:93:b6:7f:69:7d:10:35:9d:c5:d1:0d:db:d2:d7:
+        36:af:d4:54:30:14:a7:5d:31:ca:5c:13:92:d5:60:50:f8:56:
+        4a:cb:16:b1:b3:b1:03:bf:96:53:77:1f:4a:0f:9c:29:2b:bf:
+        a4:e0:da:6f:ad:13:c7:2d:8e:18:c4:72:50:17:ed:1f:36:51:
+        7a:12:9f:fc:a6:d6:c8:55:e0:db:ea:16:d6:22:0d:a2:cb:eb:
+        b2:ba:07:92:2f:db:33:d6:a2:0c:ec:89:29:f1:96:40:e5:0b:
+        e6:1f:08:50:d6:29:87:a8:20:b2:e2:17:50:25:ff:53:36:ee:
+        7f:ce:e6:1d:ed:b3:16:61:18:42:a9:17:9e:a6:86:0d:a5:fc:
+        f9:42:c8:50:48:74:72:35:eb:8c:ff:4d:e8:98:88:a0:b4:b3:
+        d0:82:b3:2f:ea:19:d7:d5:ac:47:35:96:24:37:34:0c:7a:a2:
+        e0:4d:99:a7:55:61:85:1e:7e:6a:23:77:f5:07:13:e6:50:5c:
+        65:00:13:f6:b5:4b:5b:8c:11:c3:5d:af:ba:41:e9:84:1d:f1:
+        a4:70:16:28:c2:be:6e:d8:67:38:c5:a0:ba:8a:64:6f:27:ce:
+        63:a0:92:9b
+-----BEGIN CERTIFICATE-----
+MIIDhDCCAmygAwIBAgIQKJYuoUDgfvH7Yxr0U2/O+zANBgkqhkiG9w0BAQsFADAW
+MRQwEgYDVQQDDAtFYXN5LVJTQSBDQTAeFw0yNjA0MjkxMjA5MzVaFw0yODA4MDEx
+MjA5MzVaMCUxIzAhBgNVBAMMGmV4YW1wbGVfc3NsX3JjcHNfcmNwbV9jZXJ0MIIB
+IjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEArhFG79GBNN0jXVRA85yFNZWm
+kVeSXL/rQDRp69vAhjx7/5zXug5BV4QVzZTxSGNQnDSX7r6+sCfY/c2Kz4X/CB8H
+2CiWDuQt0IvfqPpBR6CigC4uWAHMb0NcwvuEp/+el7uzox9jZHONc930fpbXa7PL
+4jVZVeDn48BB+LYPxUZMzQ6RgO/jQ/ByJhIQvoOi2yMttLEHWrGzEJwJaZhCeYF3
+XiLkcUdwJxUspxPCbURZtHPJuyd/1ug9hbs29stxNhGxmRodGhXdzWV/zcwQAEnt
+By17FYi+c7odFWm80wJV6twsPwvNGFdZeuMJsonN1uf2lcQuilMrqJaClFMAdwID
+AQABo4G+MIG7MAkGA1UdEwQCMAAwHQYDVR0OBBYEFGC9SAZoFdTc7e7kx7GfxJNt
+UDp3MFEGA1UdIwRKMEiAFLrNjgN3SkkWu9YubKOfS8NMWpWroRqkGDAWMRQwEgYD
+VQQDDAtFYXN5LVJTQSBDQYIUEv1f0yjVtkr+RNYLItZ33eTJwHMwEwYDVR0lBAww
+CgYIKwYBBQUHAwEwCwYDVR0PBAQDAgWgMBoGA1UdEQQTMBGCCTEyNy4wLjAuMYcE
+fwAAATANBgkqhkiG9w0BAQsFAAOCAQEAxTVhWCPiadps1UGrqHD03cygoz2EiZO2
+f2l9EDWdxdEN29LXNq/UVDAUp10xylwTktVgUPhWSssWsbOxA7+WU3cfSg+cKSu/
+pODab60Txy2OGMRyUBftHzZRehKf/KbWyFXg2+oW1iINosvrsroHki/bM9aiDOyJ
+KfGWQOUL5h8IUNYph6ggsuIXUCX/Uzbuf87mHe2zFmEYQqkXnqaGDaX8+ULIUEh0
+cjXrjP9N6JiIoLSz0IKzL+oZ19WsRzWWJDc0DHqi4E2Zp1VhhR5+aiN39QcT5lBc
+ZQAT9rVLW4wRw12vukHphB3xpHAWKMK+bthnOMWguopkbyfOY6CSmw==
+-----END CERTIFICATE-----
+-----BEGIN PRIVATE KEY-----
+MIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQCuEUbv0YE03SNd
+VEDznIU1laaRV5Jcv+tANGnr28CGPHv/nNe6DkFXhBXNlPFIY1CcNJfuvr6wJ9j9
+zYrPhf8IHwfYKJYO5C3Qi9+o+kFHoKKALi5YAcxvQ1zC+4Sn/56Xu7OjH2Nkc41z
+3fR+ltdrs8viNVlV4OfjwEH4tg/FRkzNDpGA7+ND8HImEhC+g6LbIy20sQdasbMQ
+nAlpmEJ5gXdeIuRxR3AnFSynE8JtRFm0c8m7J3/W6D2Fuzb2y3E2EbGZGh0aFd3N
+ZX/NzBAASe0HLXsViL5zuh0VabzTAlXq3Cw/C80YV1l64wmyic3W5/aVxC6KUyuo
+loKUUwB3AgMBAAECggEABGfLiSZJmYeUmDgZrLtkDlx16sJx9zGkR+u2V+cn6D3U
++uiCoo2EedfjSrYKT/AI35Xf19sGrc6ptJgPJfwY3aEWFxJv5HtB7ZVHWS98QiPT
++QqHgb1fPzGlQgoQ7Bo8GVBW1joPz0Bdbsv0ntTn1CyowauNUe8Z71mzpxJ0iQ7u
+Z8LNoD7INEAZDBjHVov6pLGDHS9KFxGy20WG549mE37I4QxyxetJEgmuyhJkZOQ4
+noEWiCMQjGsSg4YuSc1GS1jAVf3p2g3/TiheD/31r1jleY/T5s2qYC6MJ4vY+7yA
+5sl7m8A47i0lHSKBdR3nWz+vXEsxB0nXK3Sulyt9AQKBgQDYRZ1nEe32CCZcWn/7
+nG+9e6XNOe9E25skuvY/uEpikt92kdnyZOgHFfwM9Nv+w2IWGLZ8MmDDJ6/4hGvj
+fJjcOUb3d/SJiivPvdRC9GYMrDUZe5AJ3p6fPqi4IeZOw7bMTcVB6aHAr9KoO//J
+2t0WNvwzkOyl6KlhaQEIDK8bOQKBgQDOCvWvUEg8nHaKcmN0uQaKuPvCorhnyqUT
+VFqLlSrYC79ffHCwp2y8nkFUnxpeHXENrtHtcrdnKBNkgGEn2l4xs63CjhKrBuIG
+bDjrtp3vKlHRhQjX6HkrEEuUk52wYzuX7CfU8nTZnrtV74MocOEewJVW+84Rtwiw
+vIUcgfgJLwKBgAbgj9TLOSntsGqXZiJ2Iwd/exI/mWAzK4fLejEkhxkDWp/Gm4ud
+sdMn28/9qVE8nU3ek073uyP5ixr3+wZM2/+EwsDzy47kGeiNPMa0Rtp4T2f0CeyG
+a7zcnTjduxkeGB3/CxrBdydNcAFxhvzAPO+L6BErtpq//0Ldt+6tmJPhAoGAFxEh
+Cjx5qdd2ae9+dO3V7qfg/5xJ+sy0CGL0NBZCEqfWB/GdiBlmUgOBmuCpCgpPwtFk
+jSm/oJva9/BrcBPBYd0Uweg37M+7dC6ffLwYGFNrj4JOSCWtkwWjAII6MCob3NlC
+aFOwg0CDBo7m5xskCNZUocVU/6S3I1onqNZgF18CgYAczBf1NS4VPZvebrutTBEH
+zyUX3XU9mR+dy0ncCGNVS8zYMtz7cweZInzNB2cTOfisIHzzdOnazm7D9uzPsREi
+pKgaL+ErWYDlGiDTxMtGSRPTWGocYBYdU6y/0bobhZb0qyvyRhpGvPK0ReMUuvqu
+FkNgoQ1lo0n6vawvxWW8Mw==
+-----END PRIVATE KEY-----

contrib/rcp/usage_example/certs/make_certs.sh

@@ -0,0 +1,49 @@
+#!/bin/bash
+EASYRSA=/usr/share/easy-rsa/easyrsa
+CA_NAME="example_ssl_rcp_ca_cert"
+
+export EASYRSA_PASSIN=pass:test
+export EASYRSA_PASSOUT=pass:test
+
+echo "Cleaning up..."
+rm -rf ./ca
+rm -rf ./*.pem
+rm -rf ./*.key
+rm -rf ./*.crt
+
+echo "Creating CA cert..."
+mkdir -p ./ca
+cd ./ca
+$EASYRSA init-pki
+cp ../vars ./pki/
+$EASYRSA --batch build-ca
+cp ./pki/ca.crt  ../$CA_NAME.crt
+
+echo "Creating server certs..."
+# Secures connection between RCP-Client and RCP-Server:
+$EASYRSA --batch --subject-alt-name="DNS:127.0.0.1,IP:127.0.0.1" build-server-full example_ssl_rcpc_rcps_cert nopass
+
+# Secures connection between RCP-Module and RCP-Server (module description):
+$EASYRSA --batch --subject-alt-name="DNS:127.0.0.1,IP:127.0.0.1" build-server-full example_ssl_rcpm_rcps_cert nopass
+
+# Secures connection between RCP-Server and RCP-Module (command execution):
+$EASYRSA --batch --subject-alt-name="DNS:127.0.0.1,IP:127.0.0.1" build-server-full example_ssl_rcps_rcpm_cert nopass
+
+echo "Collecting server certs..."
+cp ./pki/issued/* ../
+cp ./pki/private/* ../
+cd ..
+rm ./ca.key
+
+echo "Merging server certs..."
+for CRT in ./*.crt; do
+    CRT_NAME=`basename ${CRT%.*}`
+    if [ -f $CRT_NAME.key ]; then
+	cat $CRT_NAME.crt $CRT_NAME.key > $CRT_NAME.pem
+	rm $CRT_NAME.key
+	rm $CRT_NAME.crt
+    fi
+done
+
+echo "Finalizing..."
+rm -rf ./ca

contrib/rcp/usage_example/params.cfg

@@ -0,0 +1,36 @@
+# PYSIM_DIR passed to all components
+PYSIM_DIR=../../../ # Points to the psyim top directory
+
+# Verbosity switch passed to all components (comment-out to disable verbode mode)
+#VERBOSE="--verbose"
+
+# PCSC reader that the RCP Client shall use
+PCSC_READER=0
+
+# Since RCP Modules are custom implementations, they will most likely reside
+# in a dedicated directory. This directory is passed together with PYSIM_DIR
+# via PYTHONPATH to the module.
+RCP_DIR=../
+
+# CA of the certificates used in this example
+CA_CERT="./certs/example_ssl_rcp_ca_cert.crt"
+
+# Network interface where RCP Clients connect
+RCPC_SERVER_PORT=8000
+RCPC_SERVER_ADDR="127.0.0.1"
+RCPC_SERVER_CERT="./certs/example_ssl_rcpc_rcps_cert.pem"
+RCPC_SERVER_URI="wss://$RCPC_SERVER_ADDR:$RCPC_SERVER_PORT"
+
+# Network interface where RCP Modules connect
+RCPM_SERVER_PORT=8010
+RCPM_SERVER_ADDR="127.0.0.1"
+RCPM_SERVER_CERT="./certs/example_ssl_rcpm_rcps_cert.pem"
+RCPM_SERVER_URI="wss://$RCPM_SERVER_ADDR:$RCPM_SERVER_PORT"
+
+# Network interface where the (example) RCP Module binds its Command Server to.
+# The command server is used by the RCP Server to run the command requested
+# by the user. Each module needs a dedicated port. The address and port is
+# automatically forwarded to the RCP Server.
+RCPM_CMD_SERVER_PORT=8020
+RCPM_CMD_SERVER_ADDR="127.0.0.1"
+RCPM_CMD_SERVER_CERT="./certs/example_ssl_rcps_rcpm_cert.pem"

contrib/rcp/usage_example/rcp_module.py

@@ -0,0 +1,86 @@
+#!/usr/bin/env python3
+
+# (C) 2026 by sysmocom - s.f.m.c. GmbH
+# All Rights Reserved
+#
+# Author: Philipp Maier
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import logging
+from pathlib import Path
+from pySim.log import PySimLogger
+from argparse import Namespace
+from rcp_module_utils import rcpm_setup_argparse, rcpm_run_module, RcpModule, RcpmCmdSrvConnHdlr
+
+log = PySimLogger.get(Path(__file__).stem)
+
+class ExmpleModule(RcpModule):
+
+    name = Path(__file__).stem
+    cmd_descr = [{"name" : "reset",
+                  "help": "reset the card",
+                  "args" : []},
+                 {"name" : "read_binary",
+                  "help": "read binary data from a transparent file.",
+                  "args" : [{ "name" : "--fid",
+                               "spec" : {"required" : True,
+                                         "help" : "File identifier to of the file to read",
+                                         "action" : "append",
+                                         "pytype" : "str"},
+                             }
+                            ]},
+                 {"name" : "read_record",
+                  "help": "read binary data from a transparent file.",
+                  "args" : [{ "name" : "--fid",
+                               "spec" : {"required" : True,
+                                         "help" : "File identifier to of the file to read",
+                                         "action" : "append",
+                                         "pytype" : "str"},
+                             },
+                            { "name" : "--record",
+                               "spec" : {"required" : True,
+                                         "help" : "File record to read",
+                                         "default" : 1,
+                                         "pytype" : "int"},
+                              }
+                            ]}
+                 ]
+    suitable_for = [{"atr" : "3b9f96803f87828031e073fe211f574543753130136502"}]
+
+    def cmd_reset(self, hdlr: RcpmCmdSrvConnHdlr) -> int:
+        hdlr.print("resetting UICC/eUICC")
+        hdlr.scc.reset_card()
+        hdlr.print("ATR is: %s" % hdlr.scc.get_atr())
+        return 0
+
+    def cmd_read_binary(self, hdlr: RcpmCmdSrvConnHdlr) -> int:
+        fid = hdlr.cmd_args.fid
+        hdlr.print("reading transparent file: %s" % fid)
+        (res, _) = hdlr.scc.read_binary(fid)
+        hdlr.print("file content is: %s" % res)
+        return 0
+
+    def cmd_read_record(self, hdlr: RcpmCmdSrvConnHdlr) -> int:
+        fid = hdlr.cmd_args.fid
+        record = hdlr.cmd_args.record
+        hdlr.print("reading linear-fixed file: %s" % fid)
+        (res, _) = hdlr.scc.read_record(fid, record)
+        hdlr.print("file content is: %s" % res)
+        return 0
+
+if __name__ == '__main__':
+    option_parser = rcpm_setup_argparse("Example Module")
+    opts = option_parser.parse_args()
+    rcpm_run_module(opts, ExmpleModule)

contrib/rcp/usage_example/readme.txt

@@ -0,0 +1,14 @@
+How to try:
+
+Go to the directory that contains the usage example:
+cd pysim/contrib/rcp/usage_example
+
+Start the RCP Server:
+./start_rcp_server.sh
+
+Start the RCP Module:
+ ./start_rcp_module.sh
+
+Run the exmple scripts:
+./run_rcp_client.sh
+(it is also possible to call the run_rcp_client_*.sh scripts individually)

contrib/rcp/usage_example/run_rcp_client.sh

@@ -0,0 +1,29 @@
+#!/bin/bash
+
+echo "basic help"
+echo "===================================================================================="
+./run_rcp_client_help.sh
+echo "===================================================================================="
+echo ""
+echo ""
+
+echo "help for which commands are available"
+echo "===================================================================================="
+./run_rcp_client_help_cmd.sh
+echo "===================================================================================="
+echo ""
+echo ""
+
+echo "help for specific commands"
+echo "===================================================================================="
+./run_rcp_client_help_cmd_specific.sh
+echo "===================================================================================="
+echo ""
+echo ""
+
+echo "run specific RCP commands"
+echo "===================================================================================="
+./run_rcp_client_cmd.sh
+echo "===================================================================================="
+echo ""
+echo ""

contrib/rcp/usage_example/run_rcp_client_cmd.sh

@@ -0,0 +1,22 @@
+#!/bin/bash
+. ./params.cfg
+
+set -x
+
+PYTHONPATH=$PYSIM_DIR ../rcp_client.py $VERBOSE \
+		      --uri $RCPC_SERVER_URI\
+		      --ca-cert $CA_CERT \
+		      -p $PCSC_READER \
+		      rcp_module_reset
+
+PYTHONPATH=$PYSIM_DIR ../rcp_client.py $VERBOSE \
+		      --uri $RCPC_SERVER_URI \
+		      --ca-cert $CA_CERT \
+		      -p $PCSC_READER \
+		      rcp_module_read_binary --fid 3f00 --fid 2fe2
+
+PYTHONPATH=$PYSIM_DIR ../rcp_client.py $VERBOSE \
+		      --uri $RCPC_SERVER_URI \
+		      --ca-cert $CA_CERT \
+		      -p $PCSC_READER \
+		      rcp_module_read_record --fid 3f00 --fid 2f00 --record 1

contrib/rcp/usage_example/run_rcp_client_help.sh

@@ -0,0 +1,6 @@
+#!/bin/bash
+. ./params.cfg
+
+set -x
+PYTHONPATH=$PYSIM_DIR ../rcp_client.py $VERBOSE \
+		      -h

contrib/rcp/usage_example/run_rcp_client_help_cmd.sh

@@ -0,0 +1,9 @@
+#!/bin/bash
+. ./params.cfg
+
+set -x
+PYTHONPATH=$PYSIM_DIR ../rcp_client.py $VERBOSE \
+		      --uri $RCPC_SERVER_URI \
+		      --ca-cert $CA_CERT \
+		      -p $PCSC_READER \
+		      -h

contrib/rcp/usage_example/run_rcp_client_help_cmd_specific.sh

@@ -0,0 +1,22 @@
+#!/bin/bash
+. ./params.cfg
+
+set -x
+
+PYTHONPATH=$PYSIM_DIR ../rcp_client.py $VERBOSE \
+		      --uri $RCPC_SERVER_URI \
+		      --ca-cert $CA_CERT \
+		      -p $PCSC_READER \
+		      rcp_module_reset --help
+
+PYTHONPATH=$PYSIM_DIR ../rcp_client.py $VERBOSE \
+		      --uri $RCPC_SERVER_URI \
+		      --ca-cert $CA_CERT \
+		      -p $PCSC_READER \
+		      rcp_module_read_binary --help
+
+PYTHONPATH=$PYSIM_DIR ../rcp_client.py $VERBOSE \
+		      --uri $RCPC_SERVER_URI \
+		      --ca-cert $CA_CERT \
+		      -p $PCSC_READER \
+		      rcp_module_read_record --help

contrib/rcp/usage_example/start_rcp_module.sh

@@ -0,0 +1,11 @@
+#!/bin/bash
+. ./params.cfg
+
+set -x
+PYTHONPATH=$PYSIM_DIR:$RCP_DIR ./rcp_module.py $VERBOSE \
+			       --uri $RCPM_SERVER_URI \
+			       --rcps-ca-cert $CA_CERT \
+			       --rcpm-cmd-server-addr $RCPM_CMD_SERVER_ADDR \
+			       --rcpm-cmd-server-port $RCPM_CMD_SERVER_PORT \
+			       --rcpm-cmd-server-cert $RCPM_CMD_SERVER_CERT
+

contrib/rcp/usage_example/start_rcp_server.sh

@@ -0,0 +1,13 @@
+#!/bin/bash
+. ./params.cfg
+
+set -x
+PYTHONPATH=$PYSIM_DIR ../rcp_server.py $VERBOSE \
+		      --rcpc-server-addr $RCPC_SERVER_ADDR \
+		      --rcpc-server-port $RCPC_SERVER_PORT \
+		      --rcpc-server-cert $RCPC_SERVER_CERT \
+		      --rcpm-server-addr $RCPM_SERVER_ADDR \
+		      --rcpm-server-port $RCPM_SERVER_PORT \
+		      --rcpm-server-cert $RCPM_SERVER_CERT \
+		      --rcpm-module-ca-cert $CA_CERT
+

pySim-prog.py

@@ -27,7 +27,6 @@
 import hashlib
 import argparse
 import os
-import random
 import re
 import sys
 import traceback
@@ -436,7 +435,7 @@ def gen_parameters(opts):
         if not re.match('^[0-9a-fA-F]{32}$', ki):
             raise ValueError('Ki needs to be 128 bits, in hex format')
     else:
-        ki = ''.join(['%02x' % random.randrange(0, 256) for i in range(16)])
+        ki = os.urandom(16).hex()
 
     # OPC (random)
     if opts.opc is not None:
@@ -447,7 +446,7 @@ def gen_parameters(opts):
     elif opts.op is not None:
         opc = derive_milenage_opc(ki, opts.op)
     else:
-        opc = ''.join(['%02x' % random.randrange(0, 256) for i in range(16)])
+        opc = os.urandom(16).hex()
 
     pin_adm = sanitize_pin_adm(opts.pin_adm, opts.pin_adm_hex)
 

pySim-shell.py

@@ -69,8 +69,8 @@ from pySim.ts_102_222 import Ts102222Commands
 from pySim.gsm_r import DF_EIRENE
 from pySim.cat import ProactiveCommand
 
-from pySim.card_key_provider import CardKeyProviderCsv, CardKeyProviderPgsql
-from pySim.card_key_provider import card_key_provider_register, card_key_provider_get_field, card_key_provider_get
+from pySim.card_key_provider import argparse_add_card_key_provider_args, init_card_key_provider
+from pySim.card_key_provider import card_key_provider_get_field, card_key_provider_get
 
 from pySim.app import init_card
 
@@ -1121,15 +1121,12 @@ class Iso7816Commands(CommandSet):
         fcp_dec = self._cmd.lchan.status()
         self._cmd.poutput_json(fcp_dec)
 
-
 class Proact(ProactiveHandler):
     def receive_fetch(self, pcmd: ProactiveCommand):
         # print its parsed representation
         print(pcmd.decoded)
         # TODO: implement the basics, such as SMS Sending, ...
 
-
-
 option_parser = argparse.ArgumentParser(description='interactive SIM card shell',
                                         formatter_class=argparse.ArgumentDefaultsHelpFormatter)
 argparse_add_reader_args(option_parser)
@@ -1146,18 +1143,6 @@ global_group.add_argument("--skip-card-init", help="Skip all card/profile initia
 global_group.add_argument("--verbose", help="Enable verbose logging",
                           action='store_true', default=False)
 
-card_key_group = option_parser.add_argument_group('Card Key Provider Options')
-card_key_group.add_argument('--csv', metavar='FILE',
-                            default="~/.osmocom/pysim/card_data.csv",
-                            help='Read card data from CSV file')
-card_key_group.add_argument('--pgsql', metavar='FILE',
-                            default="~/.osmocom/pysim/card_data_pgsql.cfg",
-                            help='Read card data from PostgreSQL database (config file)')
-card_key_group.add_argument('--csv-column-key', metavar='FIELD:AES_KEY_HEX', default=[], action='append',
-                            help=argparse.SUPPRESS, dest='column_key')
-card_key_group.add_argument('--column-key', metavar='FIELD:AES_KEY_HEX', default=[], action='append',
-                            help='per-column AES transport key', dest='column_key')
-
 adm_group = global_group.add_mutually_exclusive_group()
 adm_group.add_argument('-a', '--pin-adm', metavar='PIN_ADM1', dest='pin_adm', default=None,
                        help='ADM PIN used for provisioning (overwrites default)')
@@ -1170,6 +1155,7 @@ option_parser.add_argument("command", nargs='?',
                            help="A pySim-shell command that would optionally be executed at startup")
 option_parser.add_argument('command_args', nargs=argparse.REMAINDER,
                            help="Optional Arguments for command")
+argparse_add_card_key_provider_args(option_parser)
 
 if __name__ == '__main__':
     startup_errors = False
@@ -1178,16 +1164,8 @@ if __name__ == '__main__':
     # Ensure that we are able to print formatted warnings from the beginning.
     PySimLogger.setup(print, {logging.WARN: YELLOW}, opts.verbose)
 
-    # Register csv-file as card data provider, either from specified CSV
-    # or from CSV file in home directory
-    column_keys = {}
-    for par in opts.column_key:
-        name, key = par.split(':')
-        column_keys[name] = key
-    if os.path.isfile(os.path.expanduser(opts.csv)):
-        card_key_provider_register(CardKeyProviderCsv(os.path.expanduser(opts.csv), column_keys))
-    if os.path.isfile(os.path.expanduser(opts.pgsql)):
-        card_key_provider_register(CardKeyProviderPgsql(os.path.expanduser(opts.pgsql), column_keys))
+    # Init card key provider for automatic card key retrieval
+    init_card_key_provider(opts)
 
     # Init card reader driver
     sl = init_reader(opts, proactive_handler = Proact())

pySim/card_key_provider.py

@@ -33,10 +33,12 @@ from Cryptodome.Cipher import AES
 from osmocom.utils import h2b, b2h
 from pySim.log import PySimLogger
 
+import os
 import abc
 import csv
 import logging
 import yaml
+import argparse
 
 log = PySimLogger.get(__name__)
 
@@ -148,6 +150,15 @@ class CardKeyProvider(abc.ABC):
                 fond None shall be returned.
         """
 
+    @staticmethod
+    def argparse_add_args(arg_parser: argparse.ArgumentParser):
+        """
+        Add the commandline arguments relevant for this card key provider.
+
+        Args:
+                arg_parser : argument parser group
+        """
+
     def __str__(self):
         return type(self).__name__
 
@@ -188,6 +199,12 @@ class CardKeyProviderCsv(CardKeyProvider):
             return None
         return return_dict
 
+    @staticmethod
+    def argparse_add_args(arg_parser: argparse.ArgumentParser):
+        arg_parser.add_argument('--csv', metavar='FILE',
+                                default="~/.osmocom/pysim/card_data.csv",
+                                help='Read card data from CSV file')
+
 class CardKeyProviderPgsql(CardKeyProvider):
     """Card key provider implementation that allows to query against a specified PostgreSQL database table."""
 
@@ -252,6 +269,11 @@ class CardKeyProviderPgsql(CardKeyProvider):
             result[k] = self.crypt.decrypt_field(k, result.get(k))
         return result
 
+    @staticmethod
+    def argparse_add_args(arg_parser: argparse.ArgumentParser):
+        arg_parser.add_argument('--pgsql', metavar='FILE',
+                                default="~/.osmocom/pysim/card_data_pgsql.cfg",
+                                help='Read card data from PostgreSQL database (config file)')
 
 def card_key_provider_register(provider: CardKeyProvider, provider_list=card_key_providers):
     """Register a new card key provider.
@@ -264,7 +286,6 @@ def card_key_provider_register(provider: CardKeyProvider, provider_list=card_key
         raise ValueError("provider is not a card data provider")
     provider_list.append(provider)
 
-
 def card_key_provider_get(fields: list[str], key: str, value: str, provider_list=card_key_providers) -> Dict[str, str]:
     """Query all registered card data providers for card-individual [key] data.
 
@@ -289,7 +310,6 @@ def card_key_provider_get(fields: list[str], key: str, value: str, provider_list
 
     raise ValueError("Unable to find card key data (key=%s, value=%s, fields=%s)" % (key, value, str(fields)))
 
-
 def card_key_provider_get_field(field: str, key: str, value: str, provider_list=card_key_providers) -> str:
     """Query all registered card data providers for a single field.
 
@@ -305,3 +325,25 @@ def card_key_provider_get_field(field: str, key: str, value: str, provider_list=
     fields = [field]
     result = card_key_provider_get(fields, key, value, card_key_providers)
     return result.get(field.upper())
+
+def argparse_add_card_key_provider_args(arg_parser: argparse.ArgumentParser):
+    """Add card key provider commandline options to the given argument parser"""
+    card_key_group = arg_parser.add_argument_group('Card Key Provider Options')
+    CardKeyProviderCsv.argparse_add_args(card_key_group)
+    CardKeyProviderPgsql.argparse_add_args(card_key_group)
+    card_key_group.add_argument('--column-key', metavar='FIELD:AES_KEY_HEX', default=[], action='append',
+                                help='per-column AES transport key', dest='column_key')
+    # Depprecated argument, replaced by --column-key (see above)
+    card_key_group.add_argument('--csv-column-key', metavar='FIELD:AES_KEY_HEX', default=[], action='append',
+                                help=argparse.SUPPRESS, dest='column_key')
+
+def init_card_key_provider(opts: argparse.Namespace):
+    """Initialize card key provider depending on the user provided commandline options"""
+    column_keys = {}
+    for par in opts.column_key:
+        name, key = par.split(':')
+        column_keys[name] = key
+    if os.path.isfile(os.path.expanduser(opts.csv)):
+        card_key_provider_register(CardKeyProviderCsv(os.path.expanduser(opts.csv), column_keys))
+    if os.path.isfile(os.path.expanduser(opts.pgsql)):
+        card_key_provider_register(CardKeyProviderPgsql(os.path.expanduser(opts.pgsql), column_keys))

pySim/esim/es2p.py

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

pySim/esim/es9p.py

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

pySim/esim/http_json_api.py

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

pySim/esim/saip/batch.py

@@ -19,20 +19,10 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import copy
-import pprint
-import logging
-import traceback
-import inspect
-from typing import List, Generator
+from typing import Generator
 from pySim.esim.saip.personalization import ConfigurableParameter
 from pySim.esim.saip import param_source
-from pySim.esim.saip import ProfileElementSequence, ProfileElementSD
-from pySim.global_platform import KeyUsageQualifier
-from osmocom.utils import b2h
-
-logger = logging.getLogger(__name__)
-def _func_():
-    return inspect.currentframe().f_back.f_code.co_name
+from pySim.esim.saip import ProfileElementSequence
 
 class BatchPersonalization:
     """Produce a series of eSIM profiles from predefined parameters.
@@ -40,9 +30,9 @@ class BatchPersonalization:
 
     Usage example:
 
-    der_input = some_file.open('rb').read()
+    der_input = open('some_file', 'rb').read()
     pes = ProfileElementSequence.from_der(der_input)
-    p = pers.BatchPersonalization(
+    p = BatchPersonalization(
         n=10,
         src_pes=pes,
         csv_rows=get_csv_reader())
@@ -64,9 +54,12 @@ class BatchPersonalization:
     """
 
     class ParamAndSrc:
-        'tie a ConfigurableParameter to a source of actual values'
+        """tie a ConfigurableParameter to a source of actual values"""
         def __init__(self, param: ConfigurableParameter, src: param_source.ParamSource):
-            self.param = param
+            if isinstance(param, type):
+                self.param_cls = param
+            else:
+                self.param_cls = param.__class__
             self.src = src
 
     def __init__(self,
@@ -81,10 +74,10 @@ class BatchPersonalization:
                  copied.
         params: list of ParamAndSrc instances, defining a ConfigurableParameter and corresponding ParamSource to fill in
                 profile values.
-        csv_rows: A list or generator producing all CSV rows one at a time, starting with a row containing the column
-                  headers. This is compatible with the python csv.reader. Each row gets passed to
-                  ParamSource.get_next(), such that ParamSource implementations can access the row items.
-                  See param_source.CsvSource.
+        csv_rows: A generator (e.g. iter(list_of_rows)) producing all CSV rows one at a time, starting with a row
+                  containing the column headers. This is compatible with the python csv.reader. Each row gets passed to
+                  ParamSource.get_next(), such that ParamSource implementations can access the row items. See
+                  param_source.CsvSource.
         """
         self.n = n
         self.params = params or []
@@ -92,7 +85,7 @@ class BatchPersonalization:
         self.csv_rows = csv_rows
 
     def add_param_and_src(self, param:ConfigurableParameter, src:param_source.ParamSource):
-        self.params.append(BatchPersonalization.ParamAndSrc(param=param, src=src))
+        self.params.append(BatchPersonalization.ParamAndSrc(param, src))
 
     def generate_profiles(self):
         # get first row of CSV: column names
@@ -119,248 +112,9 @@ class BatchPersonalization:
                 try:
                     input_value = p.src.get_next(csv_row=csv_row)
                     assert input_value is not None
-                    value = p.param.__class__.validate_val(input_value)
-                    p.param.__class__.apply_val(pes, value)
+                    value = p.param_cls.validate_val(input_value)
+                    p.param_cls.apply_val(pes, value)
                 except Exception as e:
-                    print(traceback.format_exc())
-                    logger.error('during %s: %r', _func_(), e)
-                    raise ValueError(f'{p.param.name} fed by {p.src.name}: {e!r}') from e
+                    raise ValueError(f'{p.param_cls.get_name()} fed by {p.src.name}: {e}') from e
 
             yield pes
-
-
-class UppAudit(dict):
-    """
-    Key-value pairs collected from a single UPP DER or PES.
-
-    UppAudit itself is a dict, callers may use the standard python dict API to access key-value pairs read from the UPP.
-    """
-
-    @classmethod
-    def from_der(cls, der: bytes, params: List, der_size=False, additional_sd_keys=False):
-        '''return a dict of parameter name and set of selected parameter values found in a DER encoded profile. Note:
-        some ConfigurableParameter implementations return more than one key-value pair, for example, Imsi returns
-        both 'IMSI' and 'IMSI-ACC' parameters.
-
-        e.g.
-            UppAudit.from_der(my_der, [Imsi, ])
-            --> {'IMSI': '001010000000023', 'IMSI-ACC': '5'}
-
-        (where 'IMSI' == Imsi.name)
-
-        Read all parameters listed in params. params is a list of either ConfigurableParameter classes or
-        ConfigurableParameter class instances. This calls only classmethods, so each entry in params can either be the
-        class itself, or a class-instance of, a (non-abstract) ConfigurableParameter subclass.
-        For example, params = [Imsi, ] is equivalent to params = [Imsi(), ].
-
-        For der_size=True, also include a {'der_size':12345} entry.
-
-        For additional_sd_keys=True, output also all Security Domain KVN that there are *no* ConfigurableParameter
-        subclasses for. For example, SCP80 has reserved kvn 0x01..0x0f, but we offer only Scp80Kvn01, Scp80Kvn02,
-        Scp80Kvn03. So we would not show kvn 0x04..0x0f in an audit. additional_sd_keys=True includes audits of all SD
-        key KVN there may be in the UPP. This helps to spot SD keys that may already be present in a UPP template, with
-        unexpected / unusual kvn.
-        '''
-
-        # make an instance of this class
-        upp_audit = cls()
-
-        if der_size:
-            upp_audit['der_size'] = set((len(der), ))
-
-        pes = ProfileElementSequence.from_der(der)
-        for param in params:
-            try:
-                for valdict in param.get_values_from_pes(pes):
-                    upp_audit.add_values(valdict)
-            except Exception as e:
-                raise ValueError(f'Error during audit for parameter {param}: {e}') from e
-
-        if not additional_sd_keys:
-            return upp_audit
-
-        # additional_sd_keys
-        for pe in pes.pe_list:
-            if pe.type != 'securityDomain':
-                continue
-            assert isinstance(pe, ProfileElementSD)
-
-            for key in pe.keys:
-                audit_key = f'SdKey_KVN{key.key_version_number:02x}_ID{key.key_identifier:02x}'
-                kuq_bin = KeyUsageQualifier.build(key.key_usage_qualifier).hex()
-                audit_val = f'{key.key_components=!r} key_usage_qualifier=0x{kuq_bin}={key.key_usage_qualifier!r}'
-                upp_audit[audit_key] = set((audit_val, ))
-
-        return upp_audit
-
-    def get_single_val(self, key, validate=True, allow_absent=False, absent_val=None):
-        """
-        Return the audit's value for the given audit key (like 'IMSI' or 'IMSI-ACC').
-        Any kind of value may occur multiple times in a profile. When all of these agree to the same unambiguous value,
-        return that value. When they do not agree, raise a ValueError.
-        """
-        # key should be a string, but if someone passes a ConfigurableParameter, just use its default name
-        if ConfigurableParameter.is_super_of(key):
-            key = key.get_name()
-
-        assert isinstance(key, str)
-        v = self.get(key)
-        if v is None and allow_absent:
-            return absent_val
-        if not isinstance(v, set):
-            raise ValueError(f'audit value should be a set(), got {v!r}')
-        if len(v) != 1:
-            raise ValueError(f'expected a single value for {key}, got {v!r}')
-        v = tuple(v)[0]
-        return v
-
-    @staticmethod
-    def audit_val_to_str(v):
-        """
-        Usually, we want to see a single value in an audit. Still, to be able to collect multiple ambiguous values,
-        audit values are always python sets. Turn it into a nice string representation: only the value when it is
-        unambiguous, otherwise a list of the ambiguous values.
-        A value may also be completely absent, then return 'not present'.
-        """
-        def try_single_val(w):
-            'change single-entry sets to just the single value'
-            if isinstance(w, set):
-                if len(w) == 1:
-                    return tuple(w)[0]
-                if len(w) == 0:
-                    return None
-            return w
-
-        v = try_single_val(v)
-        if isinstance(v, bytes):
-            v = bytes_to_hexstr(v)
-        if v is None:
-            return 'not present'
-        return str(v)
-
-    def get_val_str(self, key):
-        """Return a string of the value stored for the given key"""
-        return UppAudit.audit_val_to_str(self.get(key))
-
-    def add_values(self, src:dict):
-        """self and src are both a dict of sets.
-        For example from
-            self == { 'a': set((123,)) }
-        and
-            src == { 'a': set((456,)), 'b': set((789,)) }
-        then after this function call:
-            self == { 'a': set((123, 456,)), 'b': set((789,)) }
-        """
-        assert isinstance(src, dict)
-        for key, srcvalset in src.items():
-            dstvalset = self.get(key)
-            if dstvalset is None:
-                dstvalset = set()
-                self[key] = dstvalset
-            dstvalset.add(srcvalset)
-
-    def __str__(self):
-        return '\n'.join(f'{key}: {self.get_val_str(key)}' for key in sorted(self.keys()))
-
-class BatchAudit(list):
-    """
-    Collect UppAudit instances for a batch of UPP, for example from a personalization.BatchPersonalization.
-    Produce an output CSV.
-
-    Usage example:
-
-        ba = BatchAudit(params=(personalization.Iccid, ))
-        for upp_der in upps:
-            ba.add_audit(upp_der)
-        print(ba.summarize())
-
-        with open('output.csv', 'wb') as csv_data:
-            csv_str = io.TextIOWrapper(csv_data, 'utf-8', newline='')
-            csv.writer(csv_str).writerows( ba.to_csv_rows() )
-            csv_str.flush()
-
-    BatchAudit itself is a list, callers may use the standard python list API to access the UppAudit instances.
-    """
-
-    def __init__(self, params:List):
-        assert params
-        self.params = params
-
-    def add_audit(self, upp_der:bytes):
-        audit = UppAudit.from_der(upp_der, self.params)
-        self.append(audit)
-        return audit
-
-    def summarize(self):
-        batch_audit = UppAudit()
-
-        audits = self
-
-        if len(audits) > 2:
-            val_sep = ', ..., '
-        else:
-            val_sep = ', '
-
-        first_audit = None
-        last_audit = None
-        if len(audits) >= 1:
-            first_audit = audits[0]
-        if len(audits) >= 2:
-            last_audit = audits[-1]
-
-        if first_audit:
-            if last_audit:
-                for key in first_audit.keys():
-                    first_val = first_audit.get_val_str(key)
-                    last_val = last_audit.get_val_str(key)
-
-                    if first_val == last_val:
-                        val = first_val
-                    else:
-                        val_sep_with_newline = f"{val_sep.rstrip()}\n{' ' * (len(key) + 2)}"
-                        val = val_sep_with_newline.join((first_val, last_val))
-                    batch_audit[key] = val
-            else:
-                batch_audit.update(first_audit)
-
-        return batch_audit
-
-    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())
-
-        columns = tuple(sorted(columns, key=sort_key))
-
-        if headers:
-            yield columns
-
-        for audit in self:
-            yield (audit.get_single_val(col, allow_absent=True, absent_val="") for col in columns)
-
-def bytes_to_hexstr(b:bytes, sep=''):
-    return sep.join(f'{x:02x}' for x in b)
-
-def esim_profile_introspect(upp):
-    pes = ProfileElementSequence.from_der(upp.read())
-    d = {}
-    d['upp'] = repr(pes)
-
-    def show_bytes_as_hexdump(item):
-        if isinstance(item, bytes):
-            return bytes_to_hexstr(item)
-        if isinstance(item, list):
-            return list(show_bytes_as_hexdump(i) for i in item)
-        if isinstance(item, tuple):
-            return tuple(show_bytes_as_hexdump(i) for i in item)
-        if isinstance(item, dict):
-            d = {}
-            for k, v in item.items():
-                d[k] = show_bytes_as_hexdump(v)
-            return d
-        return item
-
-    l = list((pe.type, show_bytes_as_hexdump(pe.decoded)) for pe in pes)
-    d['pp'] = pprint.pformat(l, width=120)
-    return d

pySim/esim/saip/param_source.py

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

pySim/esim/saip/personalization.py

@@ -16,17 +16,14 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import abc
+import enum
 import io
-import os
 import re
-import pprint
 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
 from pySim.ts_51_011 import EF_SMSP
 from pySim.esim.saip import param_source
 from pySim.esim.saip import ProfileElement, ProfileElementSD, ProfileElementSequence
@@ -55,6 +52,7 @@ class ClassVarMeta(abc.ABCMeta):
         x = super().__new__(metacls, name, bases, namespace)
         for k, v in kwargs.items():
             setattr(x, k, v)
+        setattr(x, 'name', camel_to_snake(name))
         return x
 
 class ConfigurableParameter(abc.ABC, metaclass=ClassVarMeta):
@@ -74,7 +72,6 @@ class ConfigurableParameter(abc.ABC, metaclass=ClassVarMeta):
     min_len: minimum length of an input str; min_len = 4
     max_len: maximum length of an input str; max_len = 8
     allow_len: permit only specific lengths; allow_len = (8, 16, 32)
-    numeric_base: indicate hex / decimal, if any; numeric_base = None; numeric_base = 10; numeric_base = 16
 
     Subclasses may change the meaning of these by overriding validate_val(), for example that the length counts
     resulting bytes instead of a hexstring length. Most subclasses will be covered by the default validate_val().
@@ -130,7 +127,6 @@ class ConfigurableParameter(abc.ABC, metaclass=ClassVarMeta):
     allow_len = None # a list of specific lengths
     example_input = None
     default_source = None # a param_source.ParamSource subclass
-    numeric_base = None # or 10 or 16
 
     def __init__(self, input_value=None):
         self.input_value = input_value # the raw input value as given by caller
@@ -192,28 +188,19 @@ class ConfigurableParameter(abc.ABC, metaclass=ClassVarMeta):
             if cls.allow_chars is not None:
                 if any(c not in cls.allow_chars for c in val):
                     raise ValueError(f"invalid characters in input value {val!r}, valid chars are {cls.allow_chars}")
-        elif isinstance(val, io.BytesIO):
-            val = val.getvalue()
-
-        if hasattr(val, '__len__'):
-            val_len = len(val)
-        else:
-            # e.g. int length
-            val_len = len(str(val))
-
         if cls.allow_len is not None:
             l = cls.allow_len
             # cls.allow_len could be one int, or a tuple of ints. Wrap a single int also in a tuple.
             if not isinstance(l, (tuple, list)):
                 l = (l,)
-            if val_len not in l:
-                raise ValueError(f'length must be one of {cls.allow_len}, not {val_len}: {val!r}')
+            if len(val) not in l:
+                raise ValueError(f'length must be one of {cls.allow_len}, not {len(val)}: {val!r}')
         if cls.min_len is not None:
-            if val_len < cls.min_len:
-                raise ValueError(f'length must be at least {cls.min_len}, not {val_len}: {val!r}')
+            if len(val) < cls.min_len:
+                raise ValueError(f'length must be at least {cls.min_len}, not {len(val)}: {val!r}')
         if cls.max_len is not None:
-            if val_len > cls.max_len:
-                raise ValueError(f'length must be at most {cls.max_len}, not {val_len}: {val!r}')
+            if len(val) > cls.max_len:
+                raise ValueError(f'length must be at most {cls.max_len}, not {len(val)}: {val!r}')
         return val
 
     @classmethod
@@ -222,26 +209,6 @@ class ConfigurableParameter(abc.ABC, metaclass=ClassVarMeta):
         Write the given val in the right format in all the right places in pes."""
         pass
 
-    @classmethod
-    def get_value_from_pes(cls, pes: ProfileElementSequence):
-        """Same as get_values_from_pes() but expecting a single value.
-           get_values_from_pes() may return values like this:
-             [{ 'AlgorithmID': 'Milenage' }, { 'AlgorithmID': 'Milenage' }]
-           This ensures that all these entries are identical and would return only
-              { 'AlgorithmID': 'Milenage' }.
-
-           This is relevant for any profile element that may appear multiple times in the same PES (only a few),
-           where each occurrence should reflect the same value (all currently known parameters).
-        """
-
-        val = None
-        for v in cls.get_values_from_pes(pes):
-            if val is None:
-                val = v
-            elif val != v:
-                    raise ValueError(f'get_value_from_pes(): got distinct values: {val!r} != {v!r}')
-        return val
-
     @classmethod
     @abc.abstractmethod
     def get_values_from_pes(cls, pes: ProfileElementSequence) -> Generator:
@@ -293,20 +260,12 @@ class ConfigurableParameter(abc.ABC, metaclass=ClassVarMeta):
            '''
         return cls.get_len_range()[1] or 16
 
-    @classmethod
-    def is_super_of(cls, other_class):
-        try:
-            return issubclass(other_class, cls)
-        except TypeError:
-            return False
-
 class DecimalParam(ConfigurableParameter):
     """Decimal digits. The input value may be a string of decimal digits like '012345', or an int. The output of
     validate_val() is a string with only decimal digits 0-9, in the required length with leading zeros if necessary.
     """
     allow_types = (str, int)
     allow_chars = '0123456789'
-    numeric_base = 10
 
     @classmethod
     def validate_val(cls, val):
@@ -330,6 +289,7 @@ class DecimalHexParam(DecimalParam):
     @classmethod
     def validate_val(cls, val):
         val = super().validate_val(val)
+        assert isinstance(val, str)
         val = ''.join('%02x' % ord(x) for x in val)
         if cls.rpad is not None:
             c = cls.rpad_char
@@ -339,7 +299,7 @@ class DecimalHexParam(DecimalParam):
 
     @classmethod
     def decimal_hex_to_str(cls, val):
-        'useful for get_values_from_pes() implementations of subclasses'
+        """useful for get_values_from_pes() implementations of subclasses"""
         if isinstance(val, bytes):
             val = b2h(val)
         assert isinstance(val, hexstr)
@@ -351,7 +311,6 @@ class DecimalHexParam(DecimalParam):
 class IntegerParam(ConfigurableParameter):
     allow_types = (str, int)
     allow_chars = '0123456789'
-    numeric_base = 10
 
     # two integers, if the resulting int should be range limited
     min_val = None
@@ -381,19 +340,14 @@ class IntegerParam(ConfigurableParameter):
             yield valdict
 
 class BinaryParam(ConfigurableParameter):
-    allow_types = (str, io.BytesIO, bytes, bytearray, int)
+    allow_types = (str, io.BytesIO, bytes, bytearray)
     allow_chars = '0123456789abcdefABCDEF'
     strip_chars = ' \t\r\n'
-    numeric_base = 16
     default_source = param_source.RandomHexDigitSource
 
     @classmethod
     def validate_val(cls, val):
         # take care that min_len and max_len are applied to the binary length by converting to bytes first
-        if isinstance(val, int):
-            min_len, _max_len = cls.get_len_range()
-            val = '%0*d' % (min_len, val)
-
         if isinstance(val, str):
             if cls.strip_chars is not None:
                 val = ''.join(c for c in val if c not in cls.strip_chars)
@@ -420,67 +374,69 @@ class BinaryParam(ConfigurableParameter):
 
 
 class EnumParam(ConfigurableParameter):
-    value_map = {
-            # For example:
-            #'Meaningful label for value 23': 0x23,
-            # Where 0x23 is a valid value to use for apply_val().
-        }
-    _value_map_reverse = None
+    """ConfigurableParameter for named integer enumeration values.
+
+    Subclasses must define a nested enum.IntEnum named 'Values' listing all valid names and their
+    integer codes.  apply_val() and get_values_from_pes() are not implemented here and this must
+    be inherited from another mixin."""
+
+    class Values(enum.IntEnum):
+        pass  # subclasses override this
 
     @classmethod
-    def validate_val(cls, val):
-        orig_val = val
-        enum_val = None
-        if isinstance(val, str):
-            enum_name = val
-            enum_val = cls.map_name_to_val(enum_name)
+    def validate_val(cls, val) -> int:
+        if isinstance(val, int):
+            try:
+                return int(cls.Values(val))
+            except ValueError:
+                pass
+        elif isinstance(val, str):
+            member = cls.map_name_to_val(val, strict=False)
+            if member is not None:
+                return member
 
-        # if the str is not one of the known value_map.keys(), is it maybe one of value_map.keys()?
-        if enum_val is None and val in cls.value_map.values():
-            enum_val = val
-
-        if enum_val not in cls.value_map.values():
-            raise ValueError(f"{cls.get_name()}: invalid argument: {orig_val!r}. Valid arguments are:"
-                             f" {', '.join(cls.value_map.keys())}")
-
-        return enum_val
+        valid = ', '.join(m.name for m in cls.Values)
+        raise ValueError(f"{cls.get_name()}: invalid argument: {val!r}. Valid arguments are: {valid}")
 
     @classmethod
-    def map_name_to_val(cls, name:str, strict=True):
-        val = cls.value_map.get(name)
-        if val is not None:
-            return val
+    def map_name_to_val(cls, name: str, strict=True) -> int:
+        """Return the integer value for a given enum member name. Performs an exact match first,
+        then falls back to fuzzy matching (case-insensitive, punctuation-insensitive)."""
+        try:
+            return int(cls.Values[name])
+        except KeyError:
+            pass
 
-        clean_name = cls.clean_name_str(name)
-        for k, v in cls.value_map.items():
-            if clean_name == cls.clean_name_str(k):
-                return v
+        clean = cls.clean_name_str(name)
+        for member in cls.Values:
+            if cls.clean_name_str(member.name) == clean:
+                return int(member)
 
         if strict:
-            raise ValueError(f"Problem in {cls.get_name()}: {name!r} is not a known value."
-                    f" Known values are: {cls.value_map.keys()!r}")
+            valid = ', '.join(m.name for m in cls.Values)
+            raise ValueError(f"{cls.get_name()}: {name!r} is not a known value. Known values are: {valid}")
         return None
 
     @classmethod
     def map_val_to_name(cls, val, strict=False) -> str:
-        if cls._value_map_reverse is None:
-            cls._value_map_reverse = dict((v, k) for k, v in cls.value_map.items())
-
-        name = cls._value_map_reverse.get(val)
-        if name:
-            return name
-        if strict:
-            raise ValueError(f"Problem in {cls.get_name()}: {val!r} ({type(val)}) is not a known value."
-                    f" Known values are: {cls.value_map.values()!r}")
-        return None
+        """Return the enum member name for a given integer value."""
+        try:
+            return cls.Values(val).name
+        except ValueError:
+            if strict:
+                raise ValueError(f"{cls.get_name()}: {val!r} ({type(val).__name__}) is not a known value.")
+            return None
 
     @classmethod
-    def name_normalize(cls, name:str) -> str:
-        return cls.map_val_to_name(cls.map_name_to_val(name))
+    def name_normalize(cls, name: str) -> str:
+        """Map a (possibly fuzzy) name to its canonical enum member name."""
+        return cls.Values(cls.map_name_to_val(name)).name
 
     @classmethod
-    def clean_name_str(cls, val):
-        return re.sub('[^0-9A-Za-z-_]', '', val).lower()
+    def clean_name_str(cls, val: str) -> str:
+        """Strip punctuation and case for fuzzy name comparison.
+        Treats hyphens and underscores as equivalent (both removed)."""
+        return re.sub('[^0-9A-Za-z]', '', val).lower()
 
 
 class Iccid(DecimalParam):
@@ -556,7 +512,6 @@ class SmspTpScAddr(ConfigurableParameter):
     name = 'SMSP-TP-SC-ADDR'
     allow_chars = '+0123456789'
     strip_chars = ' \t\r\n'
-    numeric_base = 10
     max_len = 21 # '+' and 20 digits
     min_len = 1
     example_input = '+49301234567'
@@ -618,7 +573,6 @@ class SmspTpScAddr(ConfigurableParameter):
             ef_smsp_dec['tp_sc_addr']['ton_npi']['type_of_number'] = 'international' if international else 'unknown'
             # ensure the parameter_indicators.tp_sc_addr is True
             ef_smsp_dec['parameter_indicators']['tp_sc_addr'] = True
-            ef_smsp_dec['alpha_id'] = ''
             # re-encode into the File body
             f_smsp.body = ef_smsp.encode_record_bin(ef_smsp_dec, 1)
             #print("SMSP  (new): %s" % f_smsp.body)
@@ -628,101 +582,35 @@ class SmspTpScAddr(ConfigurableParameter):
     @classmethod
     def get_values_from_pes(cls, pes: ProfileElementSequence):
         for pe in pes.get_pes_for_type('usim'):
-            f_smsp = pe.files.get('ef-smsp', None)
-            if f_smsp is None:
-                continue
-
-            try:
-                ef_smsp = EF_SMSP()
-                ef_smsp_dec = ef_smsp.decode_record_bin(f_smsp.body, 1)
-            except IndexError:
-                continue
+            f_smsp = pe.files['ef-smsp']
+            ef_smsp = EF_SMSP()
+            ef_smsp_dec = ef_smsp.decode_record_bin(f_smsp.body, 1)
 
             tp_sc_addr = ef_smsp_dec.get('tp_sc_addr', None)
+            if not tp_sc_addr:
+                continue
 
             digits = tp_sc_addr.get('call_number', None)
+            if not digits:
+                continue
 
             ton_npi = tp_sc_addr.get('ton_npi', None)
+            if not ton_npi:
+                continue
             international = ton_npi.get('type_of_number', None)
+            if international is None:
+                continue
             international = (international == 'international')
 
             yield { cls.name: cls.tuple_to_str((international, digits)) }
 
 
-class MncLen(ConfigurableParameter):
-    """MNC length. Must be either 2 or 3. Sets only the MNC length field in EF-AD (Administrative Data)."""
-    name = 'MNC-LEN'
-    allow_chars = '23'
-    strip_chars = ' \t\r\n'
-    numeric_base = 10
-    max_len = 1
-    min_len = 1
-    example_input = '2'
-    default_source = param_source.ConstantSource
-
-    @classmethod
-    def validate_val(cls, val):
-        val = super().validate_val(val)
-        val = int(val)
-        if val not in (2, 3):
-            raise ValueError(f"MNC-LEN must be either 2 or 3, not {val!r}")
-        return val
-
-    @classmethod
-    def apply_val(cls, pes: ProfileElementSequence, val):
-        """val must be an int: either 2 or 3"""
-        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:
-                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:
-                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)
-            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)
-            if f_ad is None:
-                continue
-
-            try:
-                ef_ad = EF_AD()
-                ef_ad_dec = ef_ad.decode_bin(f_ad.body)
-            except StreamError:
-                continue
-
-            mnc_len = ef_ad_dec.get('mnc_len', None)
-            if mnc_len is None:
-                continue
-
-            yield { cls.name: str(mnc_len) }
-
-
-class SdKey(BinaryParam):
-    """Configurable Security Domain (SD) Key.  Value is presented as bytes.
-       Non-abstract implementations are generated in SdKey.generate_sd_key_classes"""
+class SdKey(BinaryParam, metaclass=ClassVarMeta):
+    """Configurable Security Domain (SD) Key.  Value is presented as bytes."""
     # these will be set by subclasses
     key_type = None
-    kvn = None
-    reserved_kvn = tuple() # tuple of all reserved kvn for a given SCPxx
     key_id = None
+    kvn = None
     key_usage_qual = None
 
     @classmethod
@@ -740,16 +628,11 @@ class SdKey(BinaryParam):
                 key = SecurityDomainKey(
                         key_version_number=cls.kvn,
                         key_id=cls.key_id,
-                        key_usage_qualifier=cls.key_usage_qual,
+                        key_usage_qualifier=KeyUsageQualifier.build(cls.key_usage_qual),
                         key_components=set_components,
                         )
                 pe.add_key(key)
             else:
-                # A key of this KVN and ID already exists in the profile.
-
-                # Keep the key_usage_qualifier as it was in the profile, so skip this here:
-                # key.key_usage_qualifier = cls.key_usage_qual
-
                 key.key_components = set_components
 
     @classmethod
@@ -766,162 +649,60 @@ class SdKey(BinaryParam):
             if kc:
                 yield { cls.name: b2h(kc) }
 
+class SdKeyScp80_01(SdKey, kvn=0x01, key_type=0x88, permitted_len=[16,24,32]): # AES key type
+    pass
+class SdKeyScp80_01Kic(SdKeyScp80_01, key_id=0x01, key_usage_qual=0x18): # FIXME: ordering?
+    pass
+class SdKeyScp80_01Kid(SdKeyScp80_01, key_id=0x02, key_usage_qual=0x14):
+    pass
+class SdKeyScp80_01Kik(SdKeyScp80_01, key_id=0x03, key_usage_qual=0x48):
+    pass
 
-    NO_OP = (('', {}))
+class SdKeyScp81_01(SdKey, kvn=0x81): # FIXME
+    pass
+class SdKeyScp81_01Psk(SdKeyScp81_01, key_id=0x01, key_type=0x85, key_usage_qual=0x3C):
+    pass
+class SdKeyScp81_01Dek(SdKeyScp81_01, key_id=0x02, key_type=0x88, key_usage_qual=0x48):
+    pass
 
-    LEN_128 = (16,)
-    LEN_128_192_256 = (16, 24, 32)
-    LEN_128_256 = (16, 32)
+class SdKeyScp02_20(SdKey, kvn=0x20, key_type=0x88, permitted_len=[16,24,32]): # AES key type
+    pass
+class SdKeyScp02_20Enc(SdKeyScp02_20, key_id=0x01, key_usage_qual=0x18):
+    pass
+class SdKeyScp02_20Mac(SdKeyScp02_20, key_id=0x02, key_usage_qual=0x14):
+    pass
+class SdKeyScp02_20Dek(SdKeyScp02_20, key_id=0x03, key_usage_qual=0x48):
+    pass
 
-    DES =    ('DES', dict(key_type=KeyType.des, allow_len=LEN_128) )
-    AES =    ('AES', dict(key_type=KeyType.aes, allow_len=LEN_128_192_256) )
+class SdKeyScp03_30(SdKey, kvn=0x30, key_type=0x88, permitted_len=[16,24,32]): # AES key type
+    pass
+class SdKeyScp03_30Enc(SdKeyScp03_30, key_id=0x01, key_usage_qual=0x18):
+    pass
+class SdKeyScp03_30Mac(SdKeyScp03_30, key_id=0x02, key_usage_qual=0x14):
+    pass
+class SdKeyScp03_30Dek(SdKeyScp03_30, key_id=0x03, key_usage_qual=0x48):
+    pass
 
-    ENC =    ('ENC', dict(key_id=0x01, key_usage_qual=0x18) )
-    MAC =    ('MAC', dict(key_id=0x02, key_usage_qual=0x14) )
-    DEK =    ('DEK', dict(key_id=0x03, key_usage_qual=0x48) )
+class SdKeyScp03_31(SdKey, kvn=0x31, key_type=0x88, permitted_len=[16,24,32]): # AES key type
+    pass
+class SdKeyScp03_31Enc(SdKeyScp03_31, key_id=0x01, key_usage_qual=0x18):
+    pass
+class SdKeyScp03_31Mac(SdKeyScp03_31, key_id=0x02, key_usage_qual=0x14):
+    pass
+class SdKeyScp03_31Dek(SdKeyScp03_31, key_id=0x03, key_usage_qual=0x48):
+    pass
 
-    TLSPSK_PSK = ('TLSPSK', dict(key_type=KeyType.tls_psk, key_id=0x01, key_usage_qual=0x3c, allow_len=LEN_128_192_256) )
-    TLSPSK_DEK = ('DEK',    dict(key_id=0x02, key_usage_qual=0x48) )
-
-    # THIS IS THE LIST that controls which SdKeyXxx subclasses exist:
-    SD_KEY_DEFS = (
-        # name    KVN                     x variants            x variants
-        ('SCP02', (0x20, 0x21, 0x22, 0xff), (AES, ),              (ENC, MAC, DEK) ),
-        ('SCP03', (0x30, 0x31, 0x32),       (AES, ),              (ENC, MAC, DEK) ),
-        ('SCP80', (0x01, 0x02, 0x03),       (DES, AES),           (ENC, MAC, DEK) ),
-
-        # key_id=1
-        ('SCP81', (0x40, 0x41, 0x42),       (TLSPSK_PSK, ),                       ),
-        # key_id=2
-        ('SCP81', (0x40, 0x41, 0x42),       (DES, AES),           (TLSPSK_DEK, )  ),
-    )
-
-    all_implementations = None
-
-    @classmethod
-    def generate_sd_key_classes(cls, sd_key_defs=SD_KEY_DEFS):
-        '''This generates python classes to be exported in this module, as subclasses of class SdKey.
-
-        We create SdKey subclasses dynamically from a list.
-        You can list all of them via:
-          from pySim.esim.saip.personalization import SdKey
-          SdKey.all_implementations
-        or
-          print('\n'.join(sorted(f'{x.__name__}\t{x.name}' for x in SdKey.all_implementations)))
-
-        at time of writing this comment, this prints:
-
-        SdKeyScp02Kvn20AesDek	SCP02-KVN20-AES-DEK
-        SdKeyScp02Kvn20AesEnc	SCP02-KVN20-AES-ENC
-        SdKeyScp02Kvn20AesMac	SCP02-KVN20-AES-MAC
-        SdKeyScp02Kvn21AesDek	SCP02-KVN21-AES-DEK
-        SdKeyScp02Kvn21AesEnc	SCP02-KVN21-AES-ENC
-        SdKeyScp02Kvn21AesMac	SCP02-KVN21-AES-MAC
-        SdKeyScp02Kvn22AesDek	SCP02-KVN22-AES-DEK
-        SdKeyScp02Kvn22AesEnc	SCP02-KVN22-AES-ENC
-        SdKeyScp02Kvn22AesMac	SCP02-KVN22-AES-MAC
-        SdKeyScp02KvnffAesDek	SCP02-KVNff-AES-DEK
-        SdKeyScp02KvnffAesEnc	SCP02-KVNff-AES-ENC
-        SdKeyScp02KvnffAesMac	SCP02-KVNff-AES-MAC
-        SdKeyScp03Kvn30AesDek	SCP03-KVN30-AES-DEK
-        SdKeyScp03Kvn30AesEnc	SCP03-KVN30-AES-ENC
-        SdKeyScp03Kvn30AesMac	SCP03-KVN30-AES-MAC
-        SdKeyScp03Kvn31AesDek	SCP03-KVN31-AES-DEK
-        SdKeyScp03Kvn31AesEnc	SCP03-KVN31-AES-ENC
-        SdKeyScp03Kvn31AesMac	SCP03-KVN31-AES-MAC
-        SdKeyScp03Kvn32AesDek	SCP03-KVN32-AES-DEK
-        SdKeyScp03Kvn32AesEnc	SCP03-KVN32-AES-ENC
-        SdKeyScp03Kvn32AesMac	SCP03-KVN32-AES-MAC
-        SdKeyScp80Kvn01AesDek	SCP80-KVN01-AES-DEK
-        SdKeyScp80Kvn01AesEnc	SCP80-KVN01-AES-ENC
-        SdKeyScp80Kvn01AesMac	SCP80-KVN01-AES-MAC
-        SdKeyScp80Kvn01DesDek	SCP80-KVN01-DES-DEK
-        SdKeyScp80Kvn01DesEnc	SCP80-KVN01-DES-ENC
-        SdKeyScp80Kvn01DesMac	SCP80-KVN01-DES-MAC
-        SdKeyScp80Kvn02AesDek	SCP80-KVN02-AES-DEK
-        SdKeyScp80Kvn02AesEnc	SCP80-KVN02-AES-ENC
-        SdKeyScp80Kvn02AesMac	SCP80-KVN02-AES-MAC
-        SdKeyScp80Kvn02DesDek	SCP80-KVN02-DES-DEK
-        SdKeyScp80Kvn02DesEnc	SCP80-KVN02-DES-ENC
-        SdKeyScp80Kvn02DesMac	SCP80-KVN02-DES-MAC
-        SdKeyScp80Kvn03AesDek	SCP80-KVN03-AES-DEK
-        SdKeyScp80Kvn03AesEnc	SCP80-KVN03-AES-ENC
-        SdKeyScp80Kvn03AesMac	SCP80-KVN03-AES-MAC
-        SdKeyScp80Kvn03DesDek	SCP80-KVN03-DES-DEK
-        SdKeyScp80Kvn03DesEnc	SCP80-KVN03-DES-ENC
-        SdKeyScp80Kvn03DesMac	SCP80-KVN03-DES-MAC
-        SdKeyScp81Kvn40AesDek	SCP81-KVN40-AES-DEK
-        SdKeyScp81Kvn40DesDek	SCP81-KVN40-DES-DEK
-        SdKeyScp81Kvn40Tlspsk	SCP81-KVN40-TLSPSK
-        SdKeyScp81Kvn41AesDek	SCP81-KVN41-AES-DEK
-        SdKeyScp81Kvn41DesDek	SCP81-KVN41-DES-DEK
-        SdKeyScp81Kvn41Tlspsk	SCP81-KVN41-TLSPSK
-        SdKeyScp81Kvn42AesDek	SCP81-KVN42-AES-DEK
-        SdKeyScp81Kvn42DesDek	SCP81-KVN42-DES-DEK
-        SdKeyScp81Kvn42Tlspsk	SCP81-KVN42-TLSPSK
-        '''
-
-        SdKey.all_implementations = []
-
-        def camel(s):
-            return s[:1].upper() + s[1:].lower()
-
-        def do_variants(name, kvn, remaining_variants, labels=[], attrs={}):
-            'recurse to unfold as many variants as there may be'
-            if remaining_variants:
-                # not a leaf node, collect more labels and attrs
-                variants = remaining_variants[0]
-                remaining_variants = remaining_variants[1:]
-
-                for label, valdict in variants:
-                    # pass copies to recursion
-                    inner_labels = list(labels)
-                    inner_attrs = dict(attrs)
-
-                    inner_labels.append(label)
-                    inner_attrs.update(valdict)
-                    do_variants(name, kvn, remaining_variants,
-                                labels=inner_labels,
-                                attrs=inner_attrs)
-                return
-
-            # leaf node. create a new class with all the accumulated vals
-            parts = [name, f'KVN{kvn:02x}',] + labels
-            cls_label = '-'.join(p for p in parts if p)
-
-            parts = ['Sd', 'Key', name, f'Kvn{kvn:02x}'] + labels
-            clsname = ''.join(camel(p) for p in parts)
-
-            max_key_len = attrs.get('allow_len')[-1]
-
-            attrs.update({
-                'name' : cls_label,
-                'kvn': kvn,
-                'example_input': f'00*{max_key_len}',
-                })
-
-            # below line is like
-            # class SdKeyScpNNKvnXXYyyZzz(SdKey):
-            #     <set attrs>
-            cls_def = type(clsname, (cls,), attrs)
-
-            # for some unknown reason, subclassing from abc.ABC makes cls_def.__module__ == 'abc',
-            # but we don't want 'abc.SdKeyScp03Kvn32AesEnc'.
-            # Make sure it is 'pySim.esim.saip.personalization.SdKeyScp03Kvn32AesEnc'
-            cls_def.__module__ = __name__
-
-            globals()[clsname] = cls_def
-            SdKey.all_implementations.append(cls_def)
+class SdKeyScp03_32(SdKey, kvn=0x32, key_type=0x88, permitted_len=[16,24,32]): # AES key type
+    pass
+class SdKeyScp03_32Enc(SdKeyScp03_32, key_id=0x01, key_usage_qual=0x18):
+    pass
+class SdKeyScp03_32Mac(SdKeyScp03_32, key_id=0x02, key_usage_qual=0x14):
+    pass
+class SdKeyScp03_32Dek(SdKeyScp03_32, key_id=0x03, key_usage_qual=0x48):
+    pass
 
 
-        for items in sd_key_defs:
-            name, kvns = items[:2]
-            variants = items[2:]
-            for kvn in kvns:
-                do_variants(name, kvn, variants)
 
-# this creates all of the classes named like SdKeyScp02Kvn20AesDek to be published in this python module:
-SdKey.generate_sd_key_classes()
 
 def obtain_all_pe_from_pelist(l: List[ProfileElement], wanted_type: str) -> ProfileElement:
     return (pe for pe in l if pe.type == wanted_type)
@@ -1008,7 +789,7 @@ class Pin(DecimalHexParam):
 
             for pinCode in pinCodes.decoded['pinCodes'][1]:
                 if pinCode['keyReference'] == cls.keyReference:
-                     yield { cls.name: cls.decimal_hex_to_str(pinCode['pinValue']) }
+                    yield { cls.name: cls.decimal_hex_to_str(pinCode['pinValue']) }
 
     @classmethod
     def get_values_from_pes(cls, pes: ProfileElementSequence):
@@ -1086,22 +867,20 @@ class AlgoConfig(ConfigurableParameter):
             yield { cls.name: val }
 
 class AlgorithmID(EnumParam, AlgoConfig):
-    '''use validate_val() from EnumParam, and apply_val() from AlgoConfig.
-    In get_values_from_pes(), return enum value names, not raw values.'''
+    """use validate_val() from EnumParam, and apply_val() from AlgoConfig.
+    In get_values_from_pes(), return enum value names, not raw values."""
     name = "Algorithm"
-
-    # as in pySim/esim/asn1/saip/PE_Definitions-3.3.1.asn
-    value_map = {
-            "Milenage" : 1,
-            "TUAK" : 2,
-            "usim-test" : 3,
-        }
+    algo_config_key = 'algorithmID'
     example_input = "Milenage"
     default_source = param_source.ConstantSource
 
-    algo_config_key = 'algorithmID'
+    # as in pySim/esim/asn1/saip/PE_Definitions-3.3.1.asn
+    class Values(enum.IntEnum):
+        Milenage  = 1
+        TUAK      = 2
+        usim_test = 3  # input 'usim-test' also accepted via fuzzy matching
 
-    # EnumParam.validate_val() returns the int values from value_map
+    # EnumParam.validate_val() returns the int values from Values
 
     @classmethod
     def get_values_from_pes(cls, pes: ProfileElementSequence):

pySim/legacy/cards.py

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

pySim/transport/__init__.py

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

pySim/ts_51_011.py

@@ -251,6 +251,16 @@ class EF_SMSP(LinFixedEF):
                                                       "numbering_plan_id": "isdn_e164" },
                             "call_number": "4915790109999" },
             "tp_pid": b"\x00", "tp_dcs": b"\x00", "tp_vp_minutes": 4320 } ),
+        ( 'e1ffffffffffffffffffffffff0891945197109099f9ffffff0000a9',
+          { "alpha_id": "", "parameter_indicators": { "tp_dest_addr": False, "tp_sc_addr": True,
+                                                          "tp_pid": True, "tp_dcs": True, "tp_vp": True },
+            "tp_dest_addr": { "length": 255, "ton_npi": { "ext": True, "type_of_number": "reserved_for_extension",
+                                                          "numbering_plan_id": "reserved_for_extension" },
+                              "call_number": "" },
+            "tp_sc_addr": { "length": 8, "ton_npi": { "ext": True, "type_of_number": "international",
+                                                      "numbering_plan_id": "isdn_e164" },
+                            "call_number": "4915790109999" },
+            "tp_pid": b"\x00", "tp_dcs": b"\x00", "tp_vp_minutes": 4320 } ),
         ( '454e6574776f726b73fffffffffffffff1ffffffffffffffffffffffffffffffffffffffffffffffff0000a7',
           { "alpha_id": "ENetworks", "parameter_indicators": { "tp_dest_addr": False, "tp_sc_addr": True,
                                                                "tp_pid": True, "tp_dcs": True, "tp_vp": False },
@@ -331,7 +341,8 @@ class EF_SMSP(LinFixedEF):
                         'ton_npi'/TonNpi, 'call_number'/PaddedBcdAdapter(Rpad(Bytes(10))))
         DestAddr = Struct('length'/Rebuild(Int8ub, lambda ctx: EF_SMSP.dest_addr_len(ctx)),
                         'ton_npi'/TonNpi, 'call_number'/PaddedBcdAdapter(Rpad(Bytes(10))))
-        self._construct = Struct('alpha_id'/COptional(GsmOrUcs2Adapter(Rpad(Bytes(this._.total_len-28)))),
+        # (see comment below)
+        self._construct = Struct('alpha_id'/GsmOrUcs2Adapter(Rpad(Bytes(this._.total_len-28))),
                                  'parameter_indicators'/InvertAdapter(BitStruct(
                                                                         Const(7, BitsInteger(3)),
                                                                         'tp_vp'/Flag,
@@ -345,6 +356,25 @@ class EF_SMSP(LinFixedEF):
                                  'tp_dcs'/Bytes(1),
                                  'tp_vp_minutes'/EF_SMSP.ValidityPeriodAdapter(Byte))
 
+    # Ensure 'alpha_id' is always present
+    def encode_record_hex(self, abstract_data: dict, record_nr: int, total_len: int = None) -> str:
+        # Problem: TS 51.011 Section 10.5.6 describes the 'alpha_id' field as optional. However, this is only true
+        # at the time when the record length of the file is set up in the file system. A card manufacturer may decide
+        # to remove the field by setting the record length to 28. Likewise, the card manaufacturer may also decide to
+        # set the field to a distinct length by setting the record length to a value greater than 28 (e.g. 14 bytes
+        # 'alpha_id' + 28 bytes). Due to the fixed nature of the record length, this eventually means that in practice
+        # 'alpha_id' is a mandatory field with a fixed length.
+        #
+        # Due to the problematic specification of 'alpha_id' as a pseudo-optional field at the beginning of a
+        # fixed-size memory, the construct definition in self._construct has been incorrectly implemented and the field
+        # has been marked as COptional. We may correct the problem by removing COptional. But to maintain compatibility,
+        # we then have to ensure that in case the field is not provided (None), it is set to an empty string ('').
+        #
+        # See also ts_31_102.py, class EF_OCI for a correct example.
+        if abstract_data['alpha_id'] is None:
+            abstract_data['alpha_id'] = ''
+        return super().encode_record_hex(abstract_data, record_nr, total_len)
+
 # TS 51.011 Section 10.5.7
 class EF_SMSS(TransparentEF):
     class MemCapAdapter(Adapter):

requirements.txt

@@ -6,7 +6,7 @@ jsonpath-ng
 construct>=2.10.70
 bidict
 pyosmocom>=0.0.12
-pyyaml>=5.1
+pyyaml>=5.4
 termcolor
 colorlog
 pycryptodomex

setup.py

@@ -26,7 +26,7 @@ setup(
         "construct >= 2.10.70",
         "bidict",
         "pyosmocom >= 0.0.12",
-        "pyyaml >= 5.1",
+        "pyyaml >= 5.4",
         "termcolor",
         "colorlog",
         "pycryptodomex",

tests/pySim-prog_test/Fairwaves-SIM.ok

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

tests/pySim-prog_test/Wavemobile-SIM.ok

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

tests/pySim-shell_test/apdu/test_apdu.script

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

tests/pySim-smpp2sim_test/pySim-smpp2sim_test.sh

@@ -1,6 +1,6 @@
 #!/bin/bash
 
-# Utility to verify the functionality of pySim-trace.py
+# Utility to verify the functionality of pySim-smpp2sim.py
 #
 # (C) 2026 by sysmocom - s.f.m.c. GmbH
 # All Rights Reserved

tests/unittests/smdpp_data

@@ -1 +0,0 @@
-../../smdpp-data

tests/unittests/test_configurable_parameters.py

@@ -1,461 +0,0 @@
-#!/usr/bin/env python3
-
-# (C) 2025 by sysmocom - s.f.m.c. GmbH <info@sysmocom.de>
-#
-# Author: Neels Hofmeyr
-#
-# This program is free software: you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation, either version 2 of the License, or
-# (at your option) any later version.
-#
-# This program is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-# GNU General Public License for more details.
-#
-# You should have received a copy of the GNU General Public License
-# along with this program.  If not, see <http://www.gnu.org/licenses/>.
-
-import io
-import sys
-import unittest
-import io
-from importlib import resources
-from osmocom.utils import hexstr
-from pySim.esim.saip import ProfileElementSequence
-import pySim.esim.saip.personalization as p13n
-import smdpp_data.upp
-
-import xo
-update_expected_output = False
-
-def valstr(val):
-    if isinstance(val, io.BytesIO):
-        val = val.getvalue()
-    if isinstance(val, bytearray):
-        val = bytes(val)
-    return f'{val!r}'
-
-def valtypestr(val):
-    if isinstance(val, dict):
-        types = []
-        for v in val.values():
-            types.append(f'{type(v).__name__}')
-
-        val_type = '{' + ', '.join(types) + '}'
-    else:
-        val_type = f'{type(val).__name__}'
-    return f'{valstr(val)}:{val_type}'
-
-class ConfigurableParameterTest(unittest.TestCase):
-
-    def test_parameters(self):
-
-        upp_fnames = (
-                'TS48v5_SAIP2.1A_NoBERTLV.der',
-                'TS48v5_SAIP2.3_BERTLV_SUCI.der',
-                'TS48v5_SAIP2.1B_NoBERTLV.der',
-                'TS48v5_SAIP2.3_NoBERTLV.der',
-               )
-
-        class Paramtest:
-            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
-
-        param_tests = [
-            Paramtest(param_cls=p13n.Imsi, val='123456',
-                      expect_clean_val=str('123456'),
-                      expect_val={'IMSI': hexstr('123456'),
-                                  'IMSI-ACC': '0040'}),
-            Paramtest(param_cls=p13n.Imsi, val=int(123456),
-                      expect_val={'IMSI': hexstr('123456'),
-                                  'IMSI-ACC': '0040'}),
-
-            Paramtest(param_cls=p13n.Imsi, val='123456789012345',
-                      expect_clean_val=str('123456789012345'),
-                      expect_val={'IMSI': hexstr('123456789012345'),
-                                  'IMSI-ACC': '0020'}),
-            Paramtest(param_cls=p13n.Imsi, val=int(123456789012345),
-                      expect_val={'IMSI': hexstr('123456789012345'),
-                                  'IMSI-ACC': '0020'}),
-
-            Paramtest(param_cls=p13n.Puk1,
-                      val='12345678',
-                      expect_clean_val=b'12345678',
-                      expect_val='12345678'),
-            Paramtest(param_cls=p13n.Puk1,
-                      val=int(12345678),
-                      expect_clean_val=b'12345678',
-                      expect_val='12345678'),
-
-            Paramtest(param_cls=p13n.Puk2,
-                      val='12345678',
-                      expect_clean_val=b'12345678',
-                      expect_val='12345678'),
-
-            Paramtest(param_cls=p13n.Pin1,
-                      val='1234',
-                      expect_clean_val=b'1234\xff\xff\xff\xff',
-                      expect_val='1234'),
-            Paramtest(param_cls=p13n.Pin1,
-                      val='123456',
-                      expect_clean_val=b'123456\xff\xff',
-                      expect_val='123456'),
-            Paramtest(param_cls=p13n.Pin1,
-                      val='12345678',
-                      expect_clean_val=b'12345678',
-                      expect_val='12345678'),
-            Paramtest(param_cls=p13n.Pin1,
-                      val=int(1234),
-                      expect_clean_val=b'1234\xff\xff\xff\xff',
-                      expect_val='1234'),
-            Paramtest(param_cls=p13n.Pin1,
-                      val=int(123456),
-                      expect_clean_val=b'123456\xff\xff',
-                      expect_val='123456'),
-            Paramtest(param_cls=p13n.Pin1,
-                      val=int(12345678),
-                      expect_clean_val=b'12345678',
-                      expect_val='12345678'),
-
-            Paramtest(param_cls=p13n.Adm1,
-                      val='1234',
-                      expect_clean_val=b'1234\xff\xff\xff\xff',
-                      expect_val='1234'),
-            Paramtest(param_cls=p13n.Adm1,
-                      val='123456',
-                      expect_clean_val=b'123456\xff\xff',
-                      expect_val='123456'),
-            Paramtest(param_cls=p13n.Adm1,
-                      val='12345678',
-                      expect_clean_val=b'12345678',
-                      expect_val='12345678'),
-            Paramtest(param_cls=p13n.Adm1,
-                      val=int(123456),
-                      expect_clean_val=b'123456\xff\xff',
-                      expect_val='123456'),
-
-            Paramtest(param_cls=p13n.AlgorithmID,
-                      val='Milenage',
-                      expect_clean_val=1,
-                      expect_val='Milenage'),
-            Paramtest(param_cls=p13n.AlgorithmID,
-                      val='TUAK',
-                      expect_clean_val=2,
-                      expect_val='TUAK'),
-            Paramtest(param_cls=p13n.AlgorithmID,
-                      val='usim-test',
-                      expect_clean_val=3,
-                      expect_val='usim-test'),
-
-            Paramtest(param_cls=p13n.AlgorithmID,
-                      val=1,
-                      expect_clean_val=1,
-                      expect_val='Milenage'),
-            Paramtest(param_cls=p13n.AlgorithmID,
-                      val=2,
-                      expect_clean_val=2,
-                      expect_val='TUAK'),
-            Paramtest(param_cls=p13n.AlgorithmID,
-                      val=3,
-                      expect_clean_val=3,
-                      expect_val='usim-test'),
-
-            Paramtest(param_cls=p13n.K,
-                      val='01020304050607080910111213141516',
-                      expect_clean_val=b'\x01\x02\x03\x04\x05\x06\x07\x08\x09\x10\x11\x12\x13\x14\x15\x16',
-                      expect_val='01020304050607080910111213141516'),
-            Paramtest(param_cls=p13n.K,
-                      val=b'\x01\x02\x03\x04\x05\x06\x07\x08\x09\x10\x11\x12\x13\x14\x15\x16',
-                      expect_clean_val=b'\x01\x02\x03\x04\x05\x06\x07\x08\x09\x10\x11\x12\x13\x14\x15\x16',
-                      expect_val='01020304050607080910111213141516'),
-            Paramtest(param_cls=p13n.K,
-                      val=bytearray(b'\x01\x02\x03\x04\x05\x06\x07\x08\x09\x10\x11\x12\x13\x14\x15\x16'),
-                      expect_clean_val=b'\x01\x02\x03\x04\x05\x06\x07\x08\x09\x10\x11\x12\x13\x14\x15\x16',
-                      expect_val='01020304050607080910111213141516'),
-            Paramtest(param_cls=p13n.K,
-                      val=io.BytesIO(b'\x01\x02\x03\x04\x05\x06\x07\x08\x09\x10\x11\x12\x13\x14\x15\x16'),
-                      expect_clean_val=b'\x01\x02\x03\x04\x05\x06\x07\x08\x09\x10\x11\x12\x13\x14\x15\x16',
-                      expect_val='01020304050607080910111213141516'),
-            Paramtest(param_cls=p13n.K,
-                      val=int(11020304050607080910111213141516),
-                      expect_clean_val=b'\x11\x02\x03\x04\x05\x06\x07\x08\x09\x10\x11\x12\x13\x14\x15\x16',
-                      expect_val='11020304050607080910111213141516'),
-
-            Paramtest(param_cls=p13n.Opc,
-                      val='01020304050607080910111213141516',
-                      expect_clean_val=b'\x01\x02\x03\x04\x05\x06\x07\x08\x09\x10\x11\x12\x13\x14\x15\x16',
-                      expect_val='01020304050607080910111213141516'),
-            Paramtest(param_cls=p13n.Opc,
-                      val=b'\x01\x02\x03\x04\x05\x06\x07\x08\x09\x10\x11\x12\x13\x14\x15\x16',
-                      expect_clean_val=b'\x01\x02\x03\x04\x05\x06\x07\x08\x09\x10\x11\x12\x13\x14\x15\x16',
-                      expect_val='01020304050607080910111213141516'),
-            Paramtest(param_cls=p13n.Opc,
-                      val=bytearray(b'\x01\x02\x03\x04\x05\x06\x07\x08\x09\x10\x11\x12\x13\x14\x15\x16'),
-                      expect_clean_val=b'\x01\x02\x03\x04\x05\x06\x07\x08\x09\x10\x11\x12\x13\x14\x15\x16',
-                      expect_val='01020304050607080910111213141516'),
-            Paramtest(param_cls=p13n.Opc,
-                      val=io.BytesIO(b'\x01\x02\x03\x04\x05\x06\x07\x08\x09\x10\x11\x12\x13\x14\x15\x16'),
-                      expect_clean_val=b'\x01\x02\x03\x04\x05\x06\x07\x08\x09\x10\x11\x12\x13\x14\x15\x16',
-                      expect_val='01020304050607080910111213141516'),
-
-            Paramtest(param_cls=p13n.SmspTpScAddr,
-                      val='+1234567',
-                      expect_clean_val=(True, '1234567'),
-                      expect_val='+1234567'),
-            Paramtest(param_cls=p13n.SmspTpScAddr,
-                      val=1234567,
-                      expect_clean_val=(False, '1234567'),
-                      expect_val='1234567'),
-
-            Paramtest(param_cls=p13n.TuakNumberOfKeccak,
-                      val='123',
-                      expect_clean_val=123,
-                      expect_val='123'),
-            Paramtest(param_cls=p13n.TuakNumberOfKeccak,
-                      val=123,
-                      expect_clean_val=123,
-                      expect_val='123'),
-
-            Paramtest(param_cls=p13n.MilenageRotationConstants,
-                      val='0a 0b 0c 01 02',
-                      expect_clean_val=b'\x0a\x0b\x0c\x01\x02',
-                      expect_val='0a0b0c0102'),
-            Paramtest(param_cls=p13n.MilenageRotationConstants,
-                      val=b'\x0a\x0b\x0c\x01\x02',
-                      expect_clean_val=b'\x0a\x0b\x0c\x01\x02',
-                      expect_val='0a0b0c0102'),
-            Paramtest(param_cls=p13n.MilenageRotationConstants,
-                      val=bytearray(b'\x0a\x0b\x0c\x01\x02'),
-                      expect_clean_val=b'\x0a\x0b\x0c\x01\x02',
-                      expect_val='0a0b0c0102'),
-
-            Paramtest(param_cls=p13n.MilenageXoringConstants,
-                      val='aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'
-                          ' bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb'
-                          ' cccccccccccccccccccccccccccccccc'
-                          ' 11111111111111111111111111111111'
-                          ' 22222222222222222222222222222222',
-                      expect_clean_val=b'\xaa' * 16
-                                       + b'\xbb' * 16
-                                       + b'\xcc' * 16
-                                       + b'\x11' * 16
-                                       + b'\x22' * 16,
-                      expect_val='aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'
-                                 'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb'
-                                 'cccccccccccccccccccccccccccccccc'
-                                 '11111111111111111111111111111111'
-                                 '22222222222222222222222222222222'),
-            Paramtest(param_cls=p13n.MilenageXoringConstants,
-                      val=b'\xaa' * 16
-                          + b'\xbb' * 16
-                          + b'\xcc' * 16
-                          + b'\x11' * 16
-                          + b'\x22' * 16,
-                      expect_clean_val=b'\xaa' * 16
-                                       + b'\xbb' * 16
-                                       + b'\xcc' * 16
-                                       + b'\x11' * 16
-                                       + b'\x22' * 16,
-                      expect_val='aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'
-                                 'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb'
-                                 'cccccccccccccccccccccccccccccccc'
-                                 '11111111111111111111111111111111'
-                                 '22222222222222222222222222222222'),
-
-
-            Paramtest(param_cls=p13n.MncLen,
-                      val='2',
-                      expect_clean_val=2,
-                      expect_val='2'),
-            Paramtest(param_cls=p13n.MncLen,
-                      val=3,
-                      expect_clean_val=3,
-                      expect_val='3'),
-
-            ]
-
-        for sdkey_cls in (
-            # thin out the number of tests, as a compromise between completeness and test runtime
-            p13n.SdKeyScp02Kvn20AesDek,
-           #p13n.SdKeyScp02Kvn20AesEnc,
-           #p13n.SdKeyScp02Kvn20AesMac,
-           #p13n.SdKeyScp02Kvn21AesDek,
-            p13n.SdKeyScp02Kvn21AesEnc,
-           #p13n.SdKeyScp02Kvn21AesMac,
-           #p13n.SdKeyScp02Kvn22AesDek,
-           #p13n.SdKeyScp02Kvn22AesEnc,
-            p13n.SdKeyScp02Kvn22AesMac,
-           #p13n.SdKeyScp02KvnffAesDek,
-           #p13n.SdKeyScp02KvnffAesEnc,
-           #p13n.SdKeyScp02KvnffAesMac,
-            p13n.SdKeyScp03Kvn30AesDek,
-           #p13n.SdKeyScp03Kvn30AesEnc,
-           #p13n.SdKeyScp03Kvn30AesMac,
-           #p13n.SdKeyScp03Kvn31AesDek,
-            p13n.SdKeyScp03Kvn31AesEnc,
-           #p13n.SdKeyScp03Kvn31AesMac,
-           #p13n.SdKeyScp03Kvn32AesDek,
-           #p13n.SdKeyScp03Kvn32AesEnc,
-            p13n.SdKeyScp03Kvn32AesMac,
-           #p13n.SdKeyScp80Kvn01AesDek,
-           #p13n.SdKeyScp80Kvn01AesEnc,
-           #p13n.SdKeyScp80Kvn01AesMac,
-            p13n.SdKeyScp80Kvn01DesDek,
-           #p13n.SdKeyScp80Kvn01DesEnc,
-           #p13n.SdKeyScp80Kvn01DesMac,
-           #p13n.SdKeyScp80Kvn02AesDek,
-            p13n.SdKeyScp80Kvn02AesEnc,
-           #p13n.SdKeyScp80Kvn02AesMac,
-           #p13n.SdKeyScp80Kvn02DesDek,
-           #p13n.SdKeyScp80Kvn02DesEnc,
-            p13n.SdKeyScp80Kvn02DesMac,
-           #p13n.SdKeyScp80Kvn03AesDek,
-           #p13n.SdKeyScp80Kvn03AesEnc,
-           #p13n.SdKeyScp80Kvn03AesMac,
-            p13n.SdKeyScp80Kvn03DesDek,
-           #p13n.SdKeyScp80Kvn03DesEnc,
-           #p13n.SdKeyScp80Kvn03DesMac,
-           #p13n.SdKeyScp81Kvn40AesDek,
-            p13n.SdKeyScp81Kvn40DesDek,
-           #p13n.SdKeyScp81Kvn40Tlspsk,
-           #p13n.SdKeyScp81Kvn41AesDek,
-           #p13n.SdKeyScp81Kvn41DesDek,
-            p13n.SdKeyScp81Kvn41Tlspsk,
-           #p13n.SdKeyScp81Kvn42AesDek,
-           #p13n.SdKeyScp81Kvn42DesDek,
-           #p13n.SdKeyScp81Kvn42Tlspsk,
-            ):
-
-            for key_len in sdkey_cls.allow_len:
-                val = '0102030405060708091011121314151617181920212223242526272829303132'
-                expect_clean_val = (b'\x01\x02\x03\x04\x05\x06\x07\x08\x09\x10\x11\x12\x13\x14\x15\x16'
-                                    b'\x17\x18\x19\x20\x21\x22\x23\x24\x25\x26\x27\x28\x29\x30\x31\x32')
-                expect_val = '0102030405060708091011121314151617181920212223242526272829303132'
-
-                val = val[:key_len*2]
-                expect_clean_val = expect_clean_val[:key_len]
-                expect_val = val
-
-                param_tests.append(Paramtest(param_cls=sdkey_cls, val=val, expect_clean_val=expect_clean_val, expect_val=expect_val))
-
-                # test bytes input
-                val = expect_clean_val
-                param_tests.append(Paramtest(param_cls=sdkey_cls, val=val, expect_clean_val=expect_clean_val, expect_val=expect_val))
-
-                # test bytearray input
-                val = bytearray(expect_clean_val)
-                param_tests.append(Paramtest(param_cls=sdkey_cls, val=val, expect_clean_val=expect_clean_val, expect_val=expect_val))
-
-                # test BytesIO input
-                val = io.BytesIO(expect_clean_val)
-                param_tests.append(Paramtest(param_cls=sdkey_cls, val=val, expect_clean_val=expect_clean_val, expect_val=expect_val))
-
-                if key_len == 16:
-                    # test huge integer input.
-                    # needs to start with nonzero.. stupid
-                    val = 11020304050607080910111213141516
-                    expect_clean_val = (b'\x11\x02\x03\x04\x05\x06\x07\x08\x09\x10\x11\x12\x13\x14\x15\x16')
-                    expect_val = '11020304050607080910111213141516'
-                    param_tests.append(Paramtest(param_cls=sdkey_cls, val=val, expect_clean_val=expect_clean_val, expect_val=expect_val))
-
-        outputs = []
-
-        for upp_fname in upp_fnames:
-            test_idx = -1
-            try:
-
-                der = resources.read_binary(smdpp_data.upp, upp_fname)
-
-                for t in param_tests:
-                    test_idx += 1
-                    logloc = f'{upp_fname} {t.param_cls.__name__}(val={valtypestr(t.val)})'
-
-                    param = None
-                    try:
-                        param = t.param_cls()
-                        param.input_value = t.val
-                        param.validate()
-                    except ValueError as e:
-                        raise ValueError(f'{logloc}: {e}') from e
-
-                    clean_val = param.value
-                    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'{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)
-                    try:
-                        param.apply(pes)
-                    except ValueError as e:
-                        raise ValueError(f'{logloc} apply_val(clean_val): {e}') from e
-
-                    changed_der = pes.to_der()
-
-                    pes2 = ProfileElementSequence.from_der(changed_der)
-
-                    read_back_val = t.param_cls.get_value_from_pes(pes2)
-
-                    # compose log string to show the precise type of dict values
-                    if isinstance(read_back_val, dict):
-                        types = set()
-                        for v in read_back_val.values():
-                            types.add(f'{type(v).__name__}')
-
-                        read_back_val_type = '{' + ', '.join(types) + '}'
-                    else:
-                        read_back_val_type = f'{type(read_back_val).__name__}'
-
-                    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'{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'{logloc}: expected {expect_val=!r}:{type(t.expect_val).__name__}')
-
-                    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)
-
-            except Exception as e:
-                raise RuntimeError(f'Error while testing UPP {upp_fname} {test_idx=}: {e}') from e
-
-        output = '\n'.join(outputs) + '\n'
-        xo_name = 'test_configurable_parameters'
-        if update_expected_output:
-            with resources.path(xo, xo_name) as xo_path:
-                with open(xo_path, 'w', encoding='utf-8') as f:
-                    f.write(output)
-        else:
-            xo_str = resources.read_text(xo, xo_name)
-            if xo_str != output:
-                at = 0
-                while at < len(output):
-                    if output[at] == xo_str[at]:
-                        at += 1
-                        continue
-                    break
-
-                raise RuntimeError(f'output differs from expected output at position {at}: "{output[at:at+20]}" != "{xo_str[at:at+20]}"')
-
-
-if __name__ == "__main__":
-    if '-u' in sys.argv:
-        update_expected_output = True
-        sys.argv.remove('-u')
-    unittest.main()

tests/unittests/test_esim_saip.py

@@ -21,7 +21,7 @@ import copy
 from osmocom.utils import h2b, b2h
 
 from pySim.esim.saip import *
-from pySim.esim.saip import personalization
+from pySim.esim.saip.personalization import *
 from pprint import pprint as pp
 
 
@@ -55,56 +55,14 @@ class SaipTest(unittest.TestCase):
     def test_personalization(self):
         """Test some of the personalization operations."""
         pes = copy.deepcopy(self.pes)
-        params = [personalization.Puk1('01234567'),
-                  personalization.Puk2(98765432),
-                  personalization.Pin1('1111'),
-                  personalization.Pin2(2222),
-                  personalization.Adm1('11111111'),
-                  personalization.K(h2b('000102030405060708090a0b0c0d0e0f')),
-                  personalization.Opc(h2b('101112131415161718191a1b1c1d1e1f'))]
+        params = [Puk1('01234567'), Puk2(98765432), Pin1('1111'), Pin2(2222), Adm1('11111111'),
+                  K(h2b('000102030405060708090a0b0c0d0e0f')), Opc(h2b('101112131415161718191a1b1c1d1e1f'))]
         for p in params:
             p.validate()
             p.apply(pes)
         # TODO: we don't actually test the results here, but we just verify there is no exception
         pes.to_der()
 
-    def test_personalization2(self):
-        """Test some of the personalization operations."""
-        cls = personalization.SdKeyScp80Kvn01DesEnc
-        pes = ProfileElementSequence.from_der(self.per_input)
-        prev_val = tuple(cls.get_values_from_pes(pes))
-        print(f'{prev_val=}')
-        self.assertTrue(prev_val)
-
-        set_val = '42342342342342342342342342342342'
-        param = cls(set_val)
-        param.validate()
-        param.apply(pes)
-
-        get_val1 = tuple(cls.get_values_from_pes(pes))
-        print(f'{get_val1=} {set_val=}')
-        self.assertEqual(get_val1, ({cls.name: set_val},))
-
-        get_val1b = tuple(cls.get_values_from_pes(pes))
-        print(f'{get_val1b=} {set_val=}')
-        self.assertEqual(get_val1b, ({cls.name: set_val},))
-
-        der = pes.to_der()
-
-        get_val1c = tuple(cls.get_values_from_pes(pes))
-        print(f'{get_val1c=} {set_val=}')
-        self.assertEqual(get_val1c, ({cls.name: set_val},))
-
-        # assertTrue to not dump the entire der.
-        # Expecting the modified DER to be different. If this assertion fails, then no change has happened in the output
-        # DER and the ConfigurableParameter subclass is buggy.
-        self.assertTrue(der != self.per_input)
-
-        pes2 = ProfileElementSequence.from_der(der)
-        get_val2 = tuple(cls.get_values_from_pes(pes2))
-        print(f'{get_val2=} {set_val=}')
-        self.assertEqual(get_val2, ({cls.name: set_val},))
-
     def test_constructor_encode(self):
         """Test that DER-encoding of PE created by "empty" constructor works without raising exception."""
         for cls in [ProfileElementMF, ProfileElementPuk, ProfileElementPin, ProfileElementTelecom,

tests/unittests/test_param_src.py

@@ -1,216 +0,0 @@
-#!/usr/bin/env python3
-
-# (C) 2025 by sysmocom - s.f.m.c. GmbH <info@sysmocom.de>
-#
-# Author: Neels Hofmeyr
-#
-# This program is free software: you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation, either version 2 of the License, or
-# (at your option) any later version.
-#
-# This program is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-# GNU General Public License for more details.
-#
-# You should have received a copy of the GNU General Public License
-# along with this program.  If not, see <http://www.gnu.org/licenses/>.
-
-import sys
-import math
-from importlib import resources
-import unittest
-from pySim.esim.saip import param_source
-
-import xo
-update_expected_output = False
-
-class D:
-    mandatory = set()
-    optional = set()
-
-    def __init__(self, **kwargs):
-        if (set(kwargs.keys()) - set(self.optional)) != set(self.mandatory):
-            raise RuntimeError(f'{self.__class__.__name__}.__init__():'
-                               f' {set(kwargs.keys())=!r} - {self.optional=!r} != {self.mandatory=!r}')
-        for k, v in kwargs.items():
-            setattr(self, k, v)
-        for k in self.optional:
-            if not hasattr(self, k):
-                setattr(self, k, None)
-
-decimals = '0123456789'
-hexadecimals = '0123456789abcdefABCDEF'
-
-class FakeRandom:
-    vals = b'\xab\xcfm\xf0\x98J_\xcf\x96\x87fp5l\xe7f\xd1\xd6\x97\xc1\xf9]\x8c\x86+\xdb\t^ke\xc1r'
-    i = 0
-
-    @classmethod
-    def next(cls):
-        cls.i = (cls.i + 1) % len(cls.vals)
-        return cls.vals[cls.i]
-
-    @staticmethod
-    def randint(a, b):
-        d = b - a
-        n_bytes = math.ceil(math.log(d, 2))
-        r = int.from_bytes( bytes(FakeRandom.next() for i in range(n_bytes)) )
-        return a + (r % (b - a))
-
-    @staticmethod
-    def randbytes(n):
-        return bytes(FakeRandom.next() for i in range(n))
-
-
-class ParamSourceTest(unittest.TestCase):
-
-    def test_param_source(self):
-
-        class ParamSourceTest(D):
-            mandatory = (
-                    'param_source',
-                    'n',
-                    'expect',
-                    )
-            optional = (
-                    'expect_arg',
-                    'csv_rows',
-                    )
-
-        def expect_const(t, vals):
-            return tuple(t.expect_arg) == tuple(vals)
-
-        def expect_random(t, vals):
-            chars = t.expect_arg.get('digits')
-            repetitions = (t.n - len(set(vals)))
-            if repetitions:
-                raise RuntimeError(f'expect_random: there are {repetitions} repetitions in the returned values: {vals}')
-            for val_i in range(len(vals)):
-                v = vals[val_i]
-                val_minlen = t.expect_arg.get('val_minlen')
-                val_maxlen = t.expect_arg.get('val_maxlen')
-                if len(v) < val_minlen or len(v) > val_maxlen:
-                    raise RuntimeError(f'expect_random: invalid length {len(v)} for value [{val_i}]: {v!r}, expecting'
-                                       f' {val_minlen}..{val_maxlen}')
-
-                if chars is not None and not all(c in chars for c in v):
-                    raise RuntimeError(f'expect_random: invalid char in value [{val_i}]: {v!r}')
-            return True
-
-        param_source_tests = [
-            ParamSourceTest(param_source=param_source.ConstantSource.from_str('123'),
-                            n=3,
-                            expect=expect_const,
-                            expect_arg=('123', '123', '123')
-                           ),
-            ParamSourceTest(param_source=param_source.RandomDigitSource.from_str('12345'),
-                            n=3,
-                            expect=expect_random,
-                            expect_arg={'digits': decimals,
-                                        'val_minlen': 5,
-                                        'val_maxlen': 5,
-                                        },
-                           ),
-            ParamSourceTest(param_source=param_source.RandomDigitSource.from_str('1..999'),
-                            n=10,
-                            expect=expect_random,
-                            expect_arg={'digits': decimals,
-                                        'val_minlen': 1,
-                                        'val_maxlen': 3,
-                                        },
-                           ),
-            ParamSourceTest(param_source=param_source.RandomDigitSource.from_str('001..999'),
-                            n=10,
-                            expect=expect_random,
-                            expect_arg={'digits': decimals,
-                                        'val_minlen': 3,
-                                        'val_maxlen': 3,
-                                        },
-                           ),
-            ParamSourceTest(param_source=param_source.RandomHexDigitSource.from_str('12345678'),
-                            n=3,
-                            expect=expect_random,
-                            expect_arg={'digits': hexadecimals,
-                                        'val_minlen': 8,
-                                        'val_maxlen': 8,
-                                       },
-                           ),
-            ParamSourceTest(param_source=param_source.RandomHexDigitSource.from_str('0*8'),
-                            n=3,
-                            expect=expect_random,
-                            expect_arg={'digits': hexadecimals,
-                                        'val_minlen': 8,
-                                        'val_maxlen': 8,
-                                       },
-                           ),
-            ParamSourceTest(param_source=param_source.RandomHexDigitSource.from_str('00*4'),
-                            n=3,
-                            expect=expect_random,
-                            expect_arg={'digits': hexadecimals,
-                                        'val_minlen': 8,
-                                        'val_maxlen': 8,
-                                       },
-                           ),
-            ParamSourceTest(param_source=param_source.IncDigitSource.from_str('10001'),
-                            n=3,
-                            expect=expect_const,
-                            expect_arg=('10001', '10002', '10003')
-                           ),
-            ParamSourceTest(param_source=param_source.CsvSource('column_name'),
-                            n=3,
-                            expect=expect_const,
-                            expect_arg=('first val', 'second val', 'third val'),
-                            csv_rows=(
-                                      {'column_name': 'first val',},
-                                      {'column_name': 'second val',},
-                                      {'column_name': 'third val',},
-                                     )
-                           ),
-            ]
-
-        outputs = []
-
-        for t in param_source_tests:
-            try:
-                if hasattr(t.param_source, 'random_impl'):
-                    t.param_source.random_impl = FakeRandom
-
-                vals = []
-                for i in range(t.n):
-                    csv_row = None
-                    if t.csv_rows is not None:
-                        csv_row = t.csv_rows[i]
-                    vals.append( t.param_source.get_next(csv_row=csv_row) )
-                if not t.expect(t, vals):
-                    raise RuntimeError(f'invalid values returned: returned {vals}')
-                output = f'ok: {t.param_source.__class__.__name__} {vals=!r}'
-                outputs.append(output)
-                print(output)
-            except RuntimeError as e:
-                raise RuntimeError(f'{t.param_source.__class__.__name__} {t.n=} {t.expect.__name__}({t.expect_arg!r}): {e}') from e
-
-        output = '\n'.join(outputs) + '\n'
-        xo_name = 'test_param_src'
-        if update_expected_output:
-            with resources.path(xo, xo_name) as xo_path:
-                with open(xo_path, 'w', encoding='utf-8') as f:
-                    f.write(output)
-        else:
-            xo_str = resources.read_text(xo, xo_name)
-            if xo_str != output:
-                at = 0
-                while at < len(output):
-                    if output[at] == xo_str[at]:
-                        at += 1
-                        continue
-                    break
-
-                raise RuntimeError(f'output differs from expected output at position {at}: {xo_str[at:at+128]!r}')
-
-if __name__ == "__main__":
-    if '-u' in sys.argv:
-        update_expected_output = True
-        sys.argv.remove('-u')
-    unittest.main()

tests/unittests/xo/test_param_src

@@ -1,9 +0,0 @@
-ok: ConstantSource vals=['123', '123', '123']
-ok: RandomDigitSource vals=['13987', '49298', '55670']
-ok: RandomDigitSource vals=['650', '580', '49', '885', '497', '195', '320', '137', '245', '663']
-ok: RandomDigitSource vals=['638', '025', '232', '779', '826', '972', '650', '580', '049', '885']
-ok: RandomHexDigitSource vals=['6b65c172', 'abcf6df0', '984a5fcf']
-ok: RandomHexDigitSource vals=['96876670', '356ce766', 'd1d697c1']
-ok: RandomHexDigitSource vals=['f95d8c86', '2bdb095e', '6b65c172']
-ok: IncDigitSource vals=['10001', '10002', '10003']
-ok: CsvSource vals=['first val', 'second val', 'third val']