pySim-smpp2sim: Implement handle_send-/receivedata

Change-Id: Icec9265520a6ab20cb4764e9cab5cdab31841862

contrib/csv-encrypt-columns.py

@@ -24,20 +24,12 @@ import argparse
 from Cryptodome.Cipher import AES
 from osmocom.utils import h2b, b2h, Hexstr
 
-from pySim.card_key_provider import CardKeyProviderCsv
+from pySim.card_key_provider import CardKeyFieldCryptor
 
-def dict_keys_to_upper(d: dict) -> dict:
-    return {k.upper():v for k,v in d.items()}
-
-class CsvColumnEncryptor:
+class CsvColumnEncryptor(CardKeyFieldCryptor):
     def __init__(self, filename: str, transport_keys: dict):
         self.filename = filename
-        self.transport_keys = dict_keys_to_upper(transport_keys)
-
-    def encrypt_col(self, colname:str, value: str) -> Hexstr:
-        key = self.transport_keys[colname]
-        cipher = AES.new(h2b(key), AES.MODE_CBC, CardKeyProviderCsv.IV)
-        return b2h(cipher.encrypt(h2b(value)))
+        self.crypt = CardKeyFieldCryptor(transport_keys)
 
     def encrypt(self) -> None:
         with open(self.filename, 'r') as infile:
@@ -49,9 +41,8 @@ class CsvColumnEncryptor:
                 cw.writeheader()
 
                 for row in cr:
-                    for key_colname in self.transport_keys:
-                        if key_colname in row:
-                            row[key_colname] = self.encrypt_col(key_colname, row[key_colname])
+                    for fieldname in cr.fieldnames:
+                            row[fieldname] = self.crypt.encrypt_field(fieldname, row[fieldname])
                     cw.writerow(row)
 
 if __name__ == "__main__":
@@ -71,9 +62,5 @@ if __name__ == "__main__":
         print("You must specify at least one key!")
         sys.exit(1)
 
-    csv_column_keys = CardKeyProviderCsv.process_transport_keys(csv_column_keys)
-    for name, key in csv_column_keys.items():
-        print("Encrypting column %s using AES key %s" % (name, key))
-
     cce = CsvColumnEncryptor(opts.CSVFILE, csv_column_keys)
     cce.encrypt()

contrib/es9p_client.py

@@ -123,8 +123,8 @@ class Es9pClient:
             'profileManagementOperation': PMO(self.opts.operation).to_bitstring(),
             'notificationAddress': self.opts.notification_address or urlparse(self.opts.url).netloc,
         }
-        if opts.iccid:
-            ntf_metadata['iccid'] = h2b(swap_nibbles(opts.iccid))
+        if self.opts.iccid:
+            ntf_metadata['iccid'] = h2b(swap_nibbles(self.opts.iccid))
 
         if self.opts.operation == 'download':
             pird = {

contrib/esim_gen_metadata.py

@@ -0,0 +1,40 @@
+#!/usr/bin/env python3
+
+# (C) 2025 by Harald Welte <laforge@osmocom.org>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 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 Affero General Public License for more details.
+#
+# 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 argparse
+from osmocom.utils import h2b, swap_nibbles
+from pySim.esim.es8p import ProfileMetadata
+
+parser = argparse.ArgumentParser(description="""Utility program to generate profile metadata in the
+StoreMetadataRequest format based on input values from the command line.""")
+parser.add_argument('--iccid', required=True, help="ICCID of eSIM profile");
+parser.add_argument('--spn', required=True, help="Service Provider Name");
+parser.add_argument('--profile-name', required=True, help="eSIM Profile Name");
+parser.add_argument('--profile-class', choices=['test', 'operational', 'provisioning'],
+                    default='operational', help="Profile Class");
+parser.add_argument('--outfile', required=True, help="Output File Name");
+
+if __name__ == '__main__':
+    opts = parser.parse_args()
+
+    iccid_bin = h2b(swap_nibbles(opts.iccid))
+    pmd = ProfileMetadata(iccid_bin, spn=opts.spn, profile_name=opts.profile_name,
+                          profile_class=opts.profile_class)
+
+    with open(opts.outfile, 'wb') as f:
+        f.write(pmd.gen_store_metadata_request())
+    print("Written StoreMetadataRequest to '%s'" % opts.outfile)

docs/conf.py

@@ -18,7 +18,7 @@ sys.path.insert(0, os.path.abspath('..'))
 # -- Project information -----------------------------------------------------
 
 project = 'osmopysim-usermanual'
-copyright = '2009-2023 by Sylvain Munaut, Harald Welte, Philipp Maier, Supreeth Herle, Merlin Chlosta'
+copyright = '2009-2025 by Sylvain Munaut, Harald Welte, Philipp Maier, Supreeth Herle, Merlin Chlosta'
 author = 'Sylvain Munaut, Harald Welte, Philipp Maier, Supreeth Herle, Merlin Chlosta'
 
 # PDF: Avoid that the authors list exceeds the page by inserting '\and'

docs/osmo-smdpp.rst

@@ -40,16 +40,21 @@ osmo-smdpp currently
 Running osmo-smdpp
 ------------------
 
-osmo-smdpp does not have built-in TLS support as the used *twisted* framework appears to have
-problems when using the example elliptic curve certificates (both NIST and Brainpool) from GSMA.
+osmo-smdpp comes with built-in TLS support which is enabled by default. However, it is always possible to
+disable the built-in TLS support if needed.
+
+In order to use osmo-smdpp without the built-in TLS support, it has to be put behind a TLS reverse proxy,
+which terminates the ES9+ HTTPS traffic from the LPA, and then forwards it as plain HTTP to osmo-smdpp.
+
+NOTE: The built in TLS support in osmo-smdpp makes use of the python *twisted* framework. Older versions
+of this framework appear to have problems when using the example elliptic curve certificates (both NIST and
+Brainpool) from GSMA.
 
-So in order to use it, you have to put it behind a TLS reverse proxy, which terminates the ES9+
-HTTPS from the LPA, and then forwards it as plain HTTP to osmo-smdpp.
 
 nginx as TLS proxy
 ~~~~~~~~~~~~~~~~~~
 
-If you use `nginx` as web server, you can use the following configuration snippet::
+If you chose to use `nginx` as TLS reverse proxy, you can use the following configuration snippet::
 
   upstream smdpp {
           server localhost:8000;
@@ -92,32 +97,43 @@ The `smdpp-data/upp` directory contains the UPP (Unprotected Profile Package) us
 commandline options
 ~~~~~~~~~~~~~~~~~~~
 
-Typically, you just run it without any arguments, and it will bind its plain-HTTP ES9+ interface to
-`localhost` TCP port 8000.
+Typically, you just run osmo-smdpp without any arguments, and it will bind its built-in HTTPS ES9+ interface to
+`localhost` TCP port 443. In this case an external TLS reverse proxy is not needed.
 
 osmo-smdpp currently doesn't have any configuration file.
 
 There are command line options for binding:
 
-Bind the HTTP ES9+ to a port other than 8000::
+Bind the HTTPS ES9+ to a port other than 443::
 
-  ./osmo-smdpp.py -p 8001
+  ./osmo-smdpp.py -p 8443
+
+Disable the built-in TLS support and bind the plain-HTTP ES9+ to a port 8000::
+
+  ./osmo-smdpp.py -p 8000 --nossl
 
 Bind the HTTP ES9+ to a different local interface::
 
-  ./osmo-smdpp.py -H 127.0.0.1
+  ./osmo-smdpp.py -H 127.0.0.2
 
 DNS setup for your LPA
 ~~~~~~~~~~~~~~~~~~~~~~
 
 The LPA must resolve `testsmdpplus1.example.com` to the IP address of your TLS proxy.
 
-It must also accept the TLS certificates used by your TLS proxy.
+It must also accept the TLS certificates used by your TLS proxy. In case osmo-smdpp is used with built-in TLS support,
+it will use the certificates provided in smdpp-data.
+
+NOTE: The HTTPS ES9+ interface cannot be addressed by the LPA directly via its IP address. The reason for this is that
+the included SGP.26 (DPtls) test certificates explicitly restrict the hostname to `testsmdpplus1.example.com` in the
+`X509v3 Subject Alternative Name` extension. Using a bare IP address as hostname may cause the certificate to be
+rejected by the LPA.
+
 
 Supported eUICC
 ~~~~~~~~~~~~~~~
 
-If you run osmo-smdpp with the included SGP.26 certificates, you must use an eUICC with matching SGP.26
+If you run osmo-smdpp with the included SGP.26 (DPauth, DPpb) certificates, you must use an eUICC with matching SGP.26
 certificates, i.e. the EUM certificate must be signed by a SGP.26 test root CA and the eUICC certificate
 in turn must be signed by that SGP.26 EUM certificate.
 

osmo-ras.py

@@ -0,0 +1,46 @@
+#!/usr/bin/env python3
+
+# Remote Application Server for Remote Application Management over HTTP
+# See Amendment B of the GlobalPlatform Card Specification v2.2
+#
+# (C) 2025 sysmocom s.f.m.c.
+# Author: Daniel Willmann <dwillmann@sysmocom.de>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 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 Affero General Public License for more details.
+#
+# 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/>.
+
+from http.server import HTTPServer, SimpleHTTPRequestHandler
+from ssl import PROTOCOL_TLS_SERVER, SSLContext, TLSVersion
+
+
+context = SSLContext(PROTOCOL_TLS_SERVER)
+context.maximum_version = TLSVersion.TLSv1_2
+CIPHERS_1_0 = "TLS_PSK_WITH_3DES_EDE_CBC_SHA,TLS_PSK_WITH_AES_128_CBC_SHA,TLS_PSK_WITH_NULL_SHA"
+CIPHERS_1_2 = "TLS_PSK_WITH_AES_128_CBC_SHA256,TLS_PSK_WITH_NULL_SHA256"
+context.set_ciphers(CIPHERS_1_2)
+
+# A table using the identity of the client:
+psk_table = { 'ClientId_1': bytes.fromhex('c0ffee'),
+              'ClientId_2': bytes.fromhex('facade')
+}
+
+def get_psk(ident):
+    """ Get the PSK for the client """
+    print(f"Get PSK for {ident}")
+    return psk_table.get(ident, b'')
+
+context.set_psk_server_callback(get_psk)
+
+server = HTTPServer(("0.0.0.0", 8080), SimpleHTTPRequestHandler)
+server.socket = context.wrap_socket(server.socket, server_side=True)
+server.serve_forever()

osmo-smdpp.py

@@ -629,6 +629,7 @@ class SmDppHttpServer:
         # If ctxParams1 contains a ctxParamsForCommonAuthentication data object, the SM-DP+ Shall [...]
         # TODO: We really do a very simplistic job here, this needs to be properly implemented later,
         # considering all the various cases, profile state, etc.
+        iccid_str = None
         if euiccSigned1['ctxParams1'][0] == 'ctxParamsForCommonAuthentication':
             cpca = euiccSigned1['ctxParams1'][1]
             matchingId = cpca.get('matchingId', None)
@@ -860,10 +861,10 @@ class SmDppHttpServer:
 
 def main(argv):
     parser = argparse.ArgumentParser()
-    parser.add_argument("-H", "--host", help="Host/IP to bind HTTP to", default="localhost")
-    parser.add_argument("-p", "--port", help="TCP port to bind HTTP to", default=8000)
+    parser.add_argument("-H", "--host", help="Host/IP to bind HTTP(S) to", default="localhost")
+    parser.add_argument("-p", "--port", help="TCP port to bind HTTP(S) to", default=443)
     parser.add_argument("-c", "--certdir", help=f"cert subdir relative to {DATA_DIR}", default="certs")
-    parser.add_argument("-s", "--nossl", help="do NOT use ssl", action='store_true', default=False)
+    parser.add_argument("-s", "--nossl", help="disable built in SSL/TLS support", action='store_true', default=False)
     parser.add_argument("-v", "--verbose", help="dump more raw info", action='store_true', default=False)
     parser.add_argument("-b", "--brainpool", help="Use Brainpool curves instead of NIST",
                         action='store_true', default=False)

pySim-shell.py

@@ -22,19 +22,25 @@ from typing import List, Optional
 import json
 import traceback
 import re
-
 import cmd2
 from packaging import version
 from cmd2 import style
+
+import logging
+from pySim.log import PySimLogger
+from osmocom.utils import auto_uint8
+
 # cmd2 >= 2.3.0 has deprecated the bg/fg in favor of Bg/Fg :(
 if version.parse(cmd2.__version__) < version.parse("2.3.0"):
     from cmd2 import fg, bg # pylint: disable=no-name-in-module
     RED = fg.red
+    YELLOW = fg.yellow
     LIGHT_RED = fg.bright_red
     LIGHT_GREEN = fg.bright_green
 else:
     from cmd2 import Fg, Bg # pylint: disable=no-name-in-module
     RED = Fg.RED
+    YELLOW = Fg.YELLOW
     LIGHT_RED = Fg.LIGHT_RED
     LIGHT_GREEN = Fg.LIGHT_GREEN
 from cmd2 import CommandSet, with_default_category, with_argparser
@@ -63,10 +69,12 @@ 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, card_key_provider_register, card_key_provider_get_field
+from pySim.card_key_provider import CardKeyProviderCsv
+from pySim.card_key_provider import card_key_provider_register, card_key_provider_get_field, card_key_provider_get
 
 from pySim.app import init_card
 
+log = PySimLogger.get("main")
 
 class Cmd2Compat(cmd2.Cmd):
     """Backwards-compatibility wrapper around cmd2.Cmd to support older and newer
@@ -92,15 +100,19 @@ class PysimApp(Cmd2Compat):
 (C) 2021-2023 by Harald Welte, sysmocom - s.f.m.c. GmbH and contributors
 Online manual available at https://downloads.osmocom.org/docs/pysim/master/html/shell.html """
 
-    def __init__(self, card, rs, sl, ch, script=None):
+    def __init__(self, verbose, card, rs, sl, ch, script=None):
         if version.parse(cmd2.__version__) < version.parse("2.0.0"):
             kwargs = {'use_ipython': True}
         else:
             kwargs = {'include_ipy': True}
 
+        self.verbose = verbose
+        self._onchange_verbose('verbose', False, self.verbose);
+
         # pylint: disable=unexpected-keyword-arg
         super().__init__(persistent_history_file='~/.pysim_shell_history', allow_cli_args=False,
                          auto_load_commands=False, startup_script=script, **kwargs)
+        PySimLogger.setup(self.poutput, {logging.WARN: YELLOW})
         self.intro = style(self.BANNER, fg=RED)
         self.default_category = 'pySim-shell built-in commands'
         self.card = None
@@ -126,6 +138,9 @@ Online manual available at https://downloads.osmocom.org/docs/pysim/master/html/
         self.add_settable(Settable2Compat('apdu_strict', bool,
                                           'Enforce APDU responses according to ISO/IEC 7816-3, table 12', self,
                                           onchange_cb=self._onchange_apdu_strict))
+        self.add_settable(Settable2Compat('verbose', bool,
+                                          'Enable/disable verbose logging', self,
+                                          onchange_cb=self._onchange_verbose))
         self.equip(card, rs)
 
     def equip(self, card, rs):
@@ -210,6 +225,13 @@ Online manual available at https://downloads.osmocom.org/docs/pysim/master/html/
             else:
                 self.card._scc._tp.apdu_strict = False
 
+    def _onchange_verbose(self, param_name, old, new):
+        PySimLogger.set_verbose(new)
+        if new == True:
+            PySimLogger.set_level(logging.DEBUG)
+        else:
+            PySimLogger.set_level(logging.INFO)
+
     class Cmd2ApduTracer(ApduTracer):
         def __init__(self, cmd2_app):
             self.cmd2 = cmd2_app
@@ -477,6 +499,23 @@ Online manual available at https://downloads.osmocom.org/docs/pysim/master/html/
         """Echo (print) a string on the console"""
         self.poutput(' '.join(opts.STRING))
 
+    query_card_key_parser = argparse.ArgumentParser()
+    query_card_key_parser.add_argument('FIELDS', help="fields to query", type=str, nargs='+')
+    query_card_key_parser.add_argument('--key', help='lookup key (typically \'ICCID\' or \'EID\')',
+                                       type=str, required=True)
+    query_card_key_parser.add_argument('--value', help='lookup key match value (e.g \'8988211000000123456\')',
+                                       type=str, required=True)
+    @cmd2.with_argparser(query_card_key_parser)
+    @cmd2.with_category(CUSTOM_CATEGORY)
+    def do_query_card_key(self, opts):
+        """Manually query the Card Key Provider"""
+        result = card_key_provider_get(opts.FIELDS, opts.key, opts.value)
+        self.poutput("Result:")
+        if result == {}:
+            self.poutput(" (none)")
+        for k in result.keys():
+            self.poutput(" %s: %s" % (str(k), str(result.get(k))))
+
     @cmd2.with_category(CUSTOM_CATEGORY)
     def do_version(self, opts):
         """Print the pySim software version."""
@@ -915,36 +954,53 @@ class Iso7816Commands(CommandSet):
             raise RuntimeError("cannot find %s for ICCID '%s'" % (field, iccid))
         return result
 
+    @staticmethod
+    def __select_pin_nr(pin_type:str, pin_nr:int) -> int:
+        if pin_type:
+            # pylint: disable=unsubscriptable-object
+            return pin_names.inverse[pin_type]
+        return pin_nr
+
+    @staticmethod
+    def __add_pin_nr_to_ArgumentParser(chv_parser):
+        group = chv_parser.add_mutually_exclusive_group()
+        group.add_argument('--pin-type',
+                           choices=[x for x in pin_names.values()
+                                    if (x.startswith('PIN') or x.startswith('2PIN'))],
+                           help='Specifiy pin type (default is PIN1)')
+        group.add_argument('--pin-nr', type=auto_uint8, default=0x01,
+                           help='PIN Number, 1=PIN1, 0x81=2PIN1 or custom value (see also TS 102 221, Table 9.3")')
+
     verify_chv_parser = argparse.ArgumentParser()
-    verify_chv_parser.add_argument(
-        '--pin-nr', type=int, default=1, help='PIN Number, 1=PIN1, 2=PIN2 or custom value (decimal)')
     verify_chv_parser.add_argument('PIN', nargs='?', type=is_decimal,
                                    help='PIN code value. If none given, CSV file will be queried')
+    __add_pin_nr_to_ArgumentParser(verify_chv_parser)
 
     @cmd2.with_argparser(verify_chv_parser)
     def do_verify_chv(self, opts):
         """Verify (authenticate) using specified CHV (PIN) code, which is how the specifications
         call it if you authenticate yourself using the specified PIN.  There usually is at least PIN1 and
-        PIN2."""
-        pin = self.get_code(opts.PIN, "PIN" + str(opts.pin_nr))
-        (data, sw) = self._cmd.lchan.scc.verify_chv(opts.pin_nr, h2b(pin))
+        2PIN1 (see also TS 102 221 Section 9.5.1 / Table 9.3)."""
+        pin_nr = self.__select_pin_nr(opts.pin_type, opts.pin_nr)
+        pin = self.get_code(opts.PIN, "PIN" + str(pin_nr))
+        (data, sw) = self._cmd.lchan.scc.verify_chv(pin_nr, h2b(pin))
         self._cmd.poutput("CHV verification successful")
 
     unblock_chv_parser = argparse.ArgumentParser()
-    unblock_chv_parser.add_argument(
-        '--pin-nr', type=int, default=1, help='PUK Number, 1=PIN1, 2=PIN2 or custom value (decimal)')
     unblock_chv_parser.add_argument('PUK', nargs='?', type=is_decimal,
                                    help='PUK code value. If none given, CSV file will be queried')
     unblock_chv_parser.add_argument('NEWPIN', nargs='?', type=is_decimal,
                                    help='PIN code value. If none given, CSV file will be queried')
+    __add_pin_nr_to_ArgumentParser(unblock_chv_parser)
 
     @cmd2.with_argparser(unblock_chv_parser)
     def do_unblock_chv(self, opts):
         """Unblock PIN code using specified PUK code"""
-        new_pin = self.get_code(opts.NEWPIN, "PIN" + str(opts.pin_nr))
-        puk = self.get_code(opts.PUK, "PUK" + str(opts.pin_nr))
+        pin_nr = self.__select_pin_nr(opts.pin_type, opts.pin_nr)
+        new_pin = self.get_code(opts.NEWPIN, "PIN" + str(pin_nr))
+        puk = self.get_code(opts.PUK, "PUK" + str(pin_nr))
         (data, sw) = self._cmd.lchan.scc.unblock_chv(
-            opts.pin_nr, h2b(puk), h2b(new_pin))
+            pin_nr, h2b(puk), h2b(new_pin))
         self._cmd.poutput("CHV unblock successful")
 
     change_chv_parser = argparse.ArgumentParser()
@@ -952,42 +1008,42 @@ class Iso7816Commands(CommandSet):
                                    help='PIN code value. If none given, CSV file will be queried')
     change_chv_parser.add_argument('PIN', nargs='?', type=is_decimal,
                                    help='PIN code value. If none given, CSV file will be queried')
-    change_chv_parser.add_argument(
-        '--pin-nr', type=int, default=1, help='PUK Number, 1=PIN1, 2=PIN2 or custom value (decimal)')
+    __add_pin_nr_to_ArgumentParser(change_chv_parser)
 
     @cmd2.with_argparser(change_chv_parser)
     def do_change_chv(self, opts):
         """Change PIN code to a new PIN code"""
-        new_pin = self.get_code(opts.NEWPIN, "PIN" + str(opts.pin_nr))
-        pin = self.get_code(opts.PIN, "PIN" + str(opts.pin_nr))
+        pin_nr = self.__select_pin_nr(opts.pin_type, opts.pin_nr)
+        new_pin = self.get_code(opts.NEWPIN, "PIN" + str(pin_nr))
+        pin = self.get_code(opts.PIN, "PIN" + str(pin_nr))
         (data, sw) = self._cmd.lchan.scc.change_chv(
-            opts.pin_nr, h2b(pin), h2b(new_pin))
+            pin_nr, h2b(pin), h2b(new_pin))
         self._cmd.poutput("CHV change successful")
 
     disable_chv_parser = argparse.ArgumentParser()
-    disable_chv_parser.add_argument(
-        '--pin-nr', type=int, default=1, help='PIN Number, 1=PIN1, 2=PIN2 or custom value (decimal)')
     disable_chv_parser.add_argument('PIN', nargs='?', type=is_decimal,
                                    help='PIN code value. If none given, CSV file will be queried')
+    __add_pin_nr_to_ArgumentParser(disable_chv_parser)
 
     @cmd2.with_argparser(disable_chv_parser)
     def do_disable_chv(self, opts):
         """Disable PIN code using specified PIN code"""
-        pin = self.get_code(opts.PIN, "PIN" + str(opts.pin_nr))
-        (data, sw) = self._cmd.lchan.scc.disable_chv(opts.pin_nr, h2b(pin))
+        pin_nr = self.__select_pin_nr(opts.pin_type, opts.pin_nr)
+        pin = self.get_code(opts.PIN, "PIN" + str(pin_nr))
+        (data, sw) = self._cmd.lchan.scc.disable_chv(pin_nr, h2b(pin))
         self._cmd.poutput("CHV disable successful")
 
     enable_chv_parser = argparse.ArgumentParser()
-    enable_chv_parser.add_argument(
-        '--pin-nr', type=int, default=1, help='PIN Number, 1=PIN1, 2=PIN2 or custom value (decimal)')
+    __add_pin_nr_to_ArgumentParser(enable_chv_parser)
     enable_chv_parser.add_argument('PIN', nargs='?', type=is_decimal,
                                    help='PIN code value. If none given, CSV file will be queried')
 
     @cmd2.with_argparser(enable_chv_parser)
     def do_enable_chv(self, opts):
         """Enable PIN code using specified PIN code"""
-        pin = self.get_code(opts.PIN, "PIN" + str(opts.pin_nr))
-        (data, sw) = self._cmd.lchan.scc.enable_chv(opts.pin_nr, h2b(pin))
+        pin_nr = self.__select_pin_nr(opts.pin_type, opts.pin_nr)
+        pin = self.get_code(opts.PIN, "PIN" + str(pin_nr))
+        (data, sw) = self._cmd.lchan.scc.enable_chv(pin_nr, h2b(pin))
         self._cmd.poutput("CHV enable successful")
 
     def do_deactivate_file(self, opts):
@@ -1071,16 +1127,23 @@ argparse_add_reader_args(option_parser)
 global_group = option_parser.add_argument_group('General Options')
 global_group.add_argument('--script', metavar='PATH', default=None,
                           help='script with pySim-shell commands to be executed automatically at start-up')
-global_group.add_argument('--csv', metavar='FILE',
-                          default=None, help='Read card data from CSV file')
-global_group.add_argument('--csv-column-key', metavar='FIELD:AES_KEY_HEX', default=[], action='append',
-                          help='per-CSV-column AES transport key')
 global_group.add_argument("--card_handler", dest="card_handler_config", metavar="FILE",
                           help="Use automatic card handling machine")
 global_group.add_argument("--noprompt", help="Run in non interactive mode",
                           action='store_true', default=False)
 global_group.add_argument("--skip-card-init", help="Skip all card/profile initialization",
                           action='store_true', default=False)
+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=str(Path.home()) + "/.osmocom/pysim/card_data.csv",
+                            help='Read card data from CSV 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,
@@ -1095,23 +1158,27 @@ option_parser.add_argument("command", nargs='?',
 option_parser.add_argument('command_args', nargs=argparse.REMAINDER,
                            help="Optional Arguments for command")
 
-
 if __name__ == '__main__':
-
     startup_errors = False
     opts = option_parser.parse_args()
 
+    # Ensure that we are able to print formatted warnings from the beginning.
+    PySimLogger.setup(print, {logging.WARN: YELLOW})
+    if (opts.verbose):
+        PySimLogger.set_verbose(True)
+        PySimLogger.set_level(logging.DEBUG)
+    else:
+        PySimLogger.set_verbose(False)
+        PySimLogger.set_level(logging.INFO)
+
     # Register csv-file as card data provider, either from specified CSV
     # or from CSV file in home directory
-    csv_column_keys = {}
-    for par in opts.csv_column_key:
+    column_keys = {}
+    for par in opts.column_key:
         name, key = par.split(':')
-        csv_column_keys[name] = key
-    csv_default = str(Path.home()) + "/.osmocom/pysim/card_data.csv"
-    if opts.csv:
-        card_key_provider_register(CardKeyProviderCsv(opts.csv, csv_column_keys))
-    if os.path.isfile(csv_default):
-        card_key_provider_register(CardKeyProviderCsv(csv_default, csv_column_keys))
+        column_keys[name] = key
+    if os.path.isfile(opts.csv):
+        card_key_provider_register(CardKeyProviderCsv(opts.csv, column_keys))
 
     # Init card reader driver
     sl = init_reader(opts, proactive_handler = Proact())
@@ -1127,7 +1194,7 @@ if __name__ == '__main__':
     # able to tolerate and recover from that.
     try:
         rs, card = init_card(sl, opts.skip_card_init)
-        app = PysimApp(card, rs, sl, ch)
+        app = PysimApp(opts.verbose, card, rs, sl, ch)
     except:
         startup_errors = True
         print("Card initialization (%s) failed with an exception:" % str(sl))
@@ -1139,7 +1206,7 @@ if __name__ == '__main__':
             print(" it should also be noted that some readers may behave strangely when no card")
             print(" is inserted.)")
             print("")
-        app = PysimApp(None, None, sl, ch)
+        app = PysimApp(opts.verbose, None, None, sl, ch)
 
     # If the user supplies an ADM PIN at via commandline args authenticate
     # immediately so that the user does not have to use the shell commands

pySim-smpp2sim.py

@@ -53,7 +53,7 @@ from pySim.cards import UiccCardBase
 from pySim.exceptions import *
 from pySim.cat import ProactiveCommand, SendShortMessage, SMS_TPDU, SMSPPDownload, BearerDescription
 from pySim.cat import DeviceIdentities, Address, OtherAddress, UiccTransportLevel, BufferSize
-from pySim.cat import ChannelStatus, ChannelData, ChannelDataLength
+from pySim.cat import ChannelStatus, ChannelData, ChannelDataLength, EventDownload, EventList
 from pySim.utils import b2h, h2b
 
 logger = logging.getLogger(__name__)
@@ -71,24 +71,46 @@ class MyApduTracer(ApduTracer):
         print("-> %s %s" % (cmd[:10], cmd[10:]))
         print("<- %s: %s" % (sw, resp))
 
-class TcpProtocol(protocol.Protocol):
-    def dataReceived(self, data):
-        pass
-
-    def connectionLost(self, reason):
-        pass
-
-
 def tcp_connected_callback(p: protocol.Protocol):
     """called by twisted TCP client."""
     logger.error("%s: connected!" % p)
+    for data in p.pending_tx:
+        p.transport.write(data)
 
-class ProactChannel:
-    """Representation of a single protective channel."""
+class ProactChannel(protocol.Protocol):
+    """Representation of a single proective channel."""
     def __init__(self, channels: 'ProactChannels', chan_nr: int):
         self.channels = channels
         self.chan_nr = chan_nr
         self.ep = None
+        self.pending_tx = []
+        self.pending_rx = bytearray()
+
+    def write(self, data: bytes):
+        if self.connected:
+            self.transport.write(data)
+        else:
+            self.pending_tx.append(data)
+
+    def dataReceived(self, data: bytes):
+        logger.error(f"Got data (len={len(data)}): {data}")
+        self.pending_rx.extend(data)
+        # Send ENVELOPE with EventDownload Data available
+        event_list_ie = EventList(decoded=[ EventList.Event.data_available])
+        channel_status_ie = ChannelStatus(decoded='8100')
+        channel_data_len_ie = ChannelDataLength(decoded=min(255,len(self.pending_rx)))
+        dev_ids = DeviceIdentities(decoded={'source_dev_id': 'network', 'dest_dev_id': 'uicc'})
+        event_dl = EventDownload(children=[event_list_ie, dev_ids, channel_status_ie, channel_data_len_ie])
+        # 3) send to the card
+        envelope_hex = b2h(event_dl.to_tlv())
+        logger.info("ENVELOPE Event: %s" % envelope_hex)
+        global g_ms
+        (data, sw) = g_ms.scc.envelope(envelope_hex)
+        logger.info("SW %s: %s" % (sw, data))
+        # FIXME: Handle result?!
+
+    def connectionLost(self, reason):
+        logger.error("connection lost: %s" % reason)
 
     def close(self):
         """Close the channel."""
@@ -174,14 +196,13 @@ class Proact(ProactiveHandler):
             raise ValueError('Unsupported protocol_type')
         if other_addr_ie.decoded.get('type_of_address', None) != 'ipv4':
             raise ValueError('Unsupported type_of_address')
-        ipv4_bytes = h2b(other_addr_ie.decoded['address'])
+        ipv4_bytes = other_addr_ie.decoded['address']
         ipv4_str = '%u.%u.%u.%u' % (ipv4_bytes[0], ipv4_bytes[1], ipv4_bytes[2], ipv4_bytes[3])
         port_nr = transp_lvl_ie.decoded['port_number']
-        print("%s:%u" % (ipv4_str, port_nr))
+        logger.error("OpenChannel opening with %s:%u" % (ipv4_str, port_nr))
         channel = self.channels.channel_create()
         channel.ep = endpoints.TCP4ClientEndpoint(reactor, ipv4_str, port_nr)
-        channel.prot = TcpProtocol()
-        d = endpoints.connectProtocol(channel.ep, channel.prot)
+        d = endpoints.connectProtocol(channel.ep, channel)
         # FIXME: why is this never called despite the client showing the inbound connection?
         d.addCallback(tcp_connected_callback)
 
@@ -213,6 +234,17 @@ class Proact(ProactiveHandler):
         #                  ]}
         logger.info("ReceiveData")
         logger.info(pcmd)
+        dev_id_ie = Proact._find_first_element_of_type(pcmd.children, DeviceIdentities)
+        chan_data_len_ie = Proact._find_first_element_of_type(pcmd.children, ChannelDataLength)
+        len_requested = chan_data_len_ie.decoded
+        chan_str = dev_id_ie.decoded['dest_dev_id']
+        chan_nr = 1 # FIXME
+        chan = self.channels.channels.get(chan_nr, None)
+
+        requested = chan.pending_rx[:len_requested]
+        chan.pending_rx = chan.pending_rx[len_requested:]
+        resp = self.prepare_response(pcmd) + [ChannelData(decoded=requested), ChannelDataLength(decoded=min(255, len(chan.pending_rx)))]
+
         # Terminal Response example: [
         #  {'command_details': {'command_number': 1,
         #                       'type_of_command': 'receive_data',
@@ -222,7 +254,8 @@ class Proact(ProactiveHandler):
         #  {'channel_data': '16030100040e000000'},
         #  {'channel_data_length': 0}
         # ]
-        return self.prepare_response(pcmd) + []
+        resp = self.prepare_response(pcmd) + [ChannelData(decoded=requested), ChannelDataLength(decoded=min(255, len(chan.pending_rx)))]
+        return resp
 
     def handle_SendData(self, pcmd: ProactiveCommand):
         """Send/write data received from the SIM to the socket."""
@@ -240,7 +273,10 @@ class Proact(ProactiveHandler):
         chan_str = dev_id_ie.decoded['dest_dev_id']
         chan_nr = 1 # FIXME
         chan = self.channels.channels.get(chan_nr, None)
-        # FIXME chan.prot.transport.write(h2b(chan_data_ie.decoded))
+        # FIXME
+        logger.error(f"Chan data received: {chan_data_ie.decoded}")
+        chan.write(chan_data_ie.decoded)
+        #chan.write(h2b(chan_data_ie.decoded))
         # Terminal Response example: [
         #  {'command_details': {'command_number': 1,
         #                       'type_of_command': 'send_data',
@@ -425,4 +461,3 @@ if __name__ == '__main__':
     g_ms = MyServer(opts.smpp_bind_port, opts.smpp_bind_ip, opts.smpp_system_id, opts.smpp_password)
     g_ms.connect_to_card(tp)
     reactor.run()
-

pySim/card_key_provider.py

@@ -10,7 +10,7 @@ the need of manually entering the related card-individual data on every
 operation with pySim-shell.
 """
 
-# (C) 2021-2024 by Sysmocom s.f.m.c. GmbH
+# (C) 2021-2025 by Sysmocom s.f.m.c. GmbH
 # All Rights Reserved
 #
 # Author: Philipp Maier, Harald Welte
@@ -31,16 +31,29 @@ operation with pySim-shell.
 from typing import List, Dict, Optional
 from Cryptodome.Cipher import AES
 from osmocom.utils import h2b, b2h
+from pySim.log import PySimLogger
 
 import abc
 import csv
+import logging
+
+log = PySimLogger.get("CARDKEY")
 
 card_key_providers = []  # type: List['CardKeyProvider']
 
+class CardKeyFieldCryptor:
+    """
+    A Card key field encryption class that may be used by Card key provider implementations to add support for
+    a column-based encryption to protect sensitive material (cryptographic key material, ADM keys, etc.).
+    The sensitive material is encrypted using a "key-encryption key", occasionally also known as "transport key"
+    before it is stored into a file or database (see also GSMA FS.28). The "transport key" is then used to decrypt
+    the key material on demand.
+    """
+
     # well-known groups of columns relate to a given functionality.  This avoids having
     # to specify the same transport key N number of times, if the same key is used for multiple
     # fields of one group, like KIC+KID+KID of one SD.
-CRYPT_GROUPS = {
+    __CRYPT_GROUPS = {
             'UICC_SCP02': ['UICC_SCP02_KIC1', 'UICC_SCP02_KID1', 'UICC_SCP02_KIK1'],
             'UICC_SCP03': ['UICC_SCP03_KIC1', 'UICC_SCP03_KID1', 'UICC_SCP03_KIK1'],
             'SCP03_ISDR': ['SCP03_ENC_ISDR', 'SCP03_MAC_ISDR', 'SCP03_DEK_ISDR'],
@@ -48,111 +61,131 @@ CRYPT_GROUPS = {
             'SCP03_ECASD': ['SCP03_ENC_ECASD', 'SCP03_MAC_ECASD', 'SCP03_DEK_ECASD'],
     }
 
-class CardKeyProvider(abc.ABC):
-    """Base class, not containing any concrete implementation."""
-
-    VALID_KEY_FIELD_NAMES = ['ICCID', 'EID', 'IMSI' ]
-
-    # check input parameters, but do nothing concrete yet
-    def _verify_get_data(self, fields: List[str] = [], key: str = 'ICCID', value: str = "") -> Dict[str, str]:
-        """Verify multiple fields for identified card.
-
-        Args:
-                fields : list of valid field names such as 'ADM1', 'PIN1', ... which are to be obtained
-                key : look-up key to identify card data, such as 'ICCID'
-                value : value for look-up key to identify card data
-        Returns:
-                dictionary of {field, value} strings for each requested field from 'fields'
-        """
-
-        if key not in self.VALID_KEY_FIELD_NAMES:
-            raise ValueError("Key field name '%s' is not a valid field name, valid field names are: %s" %
-                             (key, str(self.VALID_KEY_FIELD_NAMES)))
-
-        return {}
-
-    def get_field(self, field: str, key: str = 'ICCID', value: str = "") -> Optional[str]:
-        """get a single field from CSV file using a specified key/value pair"""
-        fields = [field]
-        result = self.get(fields, key, value)
-        return result.get(field)
-
-    @abc.abstractmethod
-    def get(self, fields: List[str], key: str, value: str) -> Dict[str, str]:
-        """Get multiple card-individual fields for identified card.
-
-        Args:
-                fields : list of valid field names such as 'ADM1', 'PIN1', ... which are to be obtained
-                key : look-up key to identify card data, such as 'ICCID'
-                value : value for look-up key to identify card data
-        Returns:
-                dictionary of {field, value} strings for each requested field from 'fields'
-        """
-
-
-class CardKeyProviderCsv(CardKeyProvider):
-    """Card key provider implementation that allows to query against a specified CSV file.
-    Supports column-based encryption as it is generally a bad idea to store cryptographic key material in
-    plaintext.  Instead, the key material should be encrypted by a "key-encryption key", occasionally also
-    known as "transport key" (see GSMA FS.28)."""
-    IV = b'\x23' * 16
-    csv_file = None
-    filename = None
-
-    def __init__(self, filename: str, transport_keys: dict):
-        """
-        Args:
-                filename : file name (path) of CSV file containing card-individual key/data
-                transport_keys : a dict indexed by field name, whose values are hex-encoded AES keys for the
-                                 respective field (column) of the CSV.  This is done so that different fields
-                                 (columns) can use different transport keys, which is strongly recommended by
-                                 GSMA FS.28
-        """
-        self.csv_file = open(filename, 'r')
-        if not self.csv_file:
-            raise RuntimeError("Could not open CSV file '%s'" % filename)
-        self.filename = filename
-        self.transport_keys = self.process_transport_keys(transport_keys)
+    __IV = b'\x23' * 16
 
     @staticmethod
-    def process_transport_keys(transport_keys: dict):
+    def __dict_keys_to_upper(d: dict) -> dict:
+            return {k.upper():v for k,v in d.items()}
+
+    @staticmethod
+    def __process_transport_keys(transport_keys: dict, crypt_groups: dict):
         """Apply a single transport key to multiple fields/columns, if the name is a group."""
         new_dict = {}
         for name, key in transport_keys.items():
-            if name in CRYPT_GROUPS:
-                for field in CRYPT_GROUPS[name]:
+            if name in crypt_groups:
+                for field in crypt_groups[name]:
                     new_dict[field] = key
             else:
                 new_dict[name] = key
         return new_dict
 
-    def _decrypt_field(self, field_name: str, encrypted_val: str) -> str:
-        """decrypt a single field, if we have a transport key for the field of that name."""
-        if not field_name in self.transport_keys:
+    def __init__(self, transport_keys: dict):
+        """
+        Create new field encryptor/decryptor object and set transport keys, usually one for each column. In some cases
+        it is also possible to use a single key for multiple columns (see also __CRYPT_GROUPS)
+
+        Args:
+                transport_keys : a dict indexed by field name, whose values are hex-encoded AES keys for the
+                                 respective field (column) of the CSV. This is done so that different fields
+                                 (columns) can use different transport keys, which is strongly recommended by
+                                 GSMA FS.28
+        """
+        self.transport_keys = self.__process_transport_keys(self.__dict_keys_to_upper(transport_keys),
+                                                            self.__CRYPT_GROUPS)
+        for name, key in self.transport_keys.items():
+                log.debug("Encrypting/decrypting field %s using AES key %s" % (name, key))
+
+    def decrypt_field(self, field_name: str, encrypted_val: str) -> str:
+        """
+        Decrypt a single field. The decryption is only applied if we have a transport key is known under the provided
+        field name, otherwise the field is treated as plaintext and passed through as it is.
+
+        Args:
+                field_name : name of the field to decrypt (used to identify which key to use)
+                encrypted_val : encrypted field value
+
+        Returns:
+                plaintext field value
+        """
+        if not field_name.upper() in self.transport_keys:
             return encrypted_val
-        cipher = AES.new(h2b(self.transport_keys[field_name]), AES.MODE_CBC, self.IV)
+        cipher = AES.new(h2b(self.transport_keys[field_name.upper()]), AES.MODE_CBC, self.__IV)
         return b2h(cipher.decrypt(h2b(encrypted_val)))
 
-    def get(self, fields: List[str], key: str, value: str) -> Dict[str, str]:
-        super()._verify_get_data(fields, key, value)
+    def encrypt_field(self, field_name: str, plaintext_val: str) -> str:
+        """
+        Encrypt a single field. The encryption is only applied if we have a transport key is known under the provided
+        field name, otherwise the field is treated as non sensitive and passed through as it is.
 
+        Args:
+                field_name : name of the field to decrypt (used to identify which key to use)
+                encrypted_val : encrypted field value
+
+        Returns:
+                plaintext field value
+        """
+        if not field_name.upper() in self.transport_keys:
+            return plaintext_val
+        cipher = AES.new(h2b(self.transport_keys[field_name.upper()]), AES.MODE_CBC, self.__IV)
+        return b2h(cipher.encrypt(h2b(plaintext_val)))
+
+class CardKeyProvider(abc.ABC):
+    """Base class, not containing any concrete implementation."""
+
+    @abc.abstractmethod
+    def get(self, fields: List[str], key: str, value: str) -> Dict[str, str]:
+        """
+        Get multiple card-individual fields for identified card. This method should not fail with an exception in
+        case the entry, columns or even the key column itsself is not found.
+
+        Args:
+                fields : list of valid field names such as 'ADM1', 'PIN1', ... which are to be obtained
+                key : look-up key to identify card data, such as 'ICCID'
+                value : value for look-up key to identify card data
+        Returns:
+                dictionary of {field : value, ...} strings for each requested field from 'fields'. In case nothing is
+                fond None shall be returned.
+        """
+
+    def __str__(self):
+        return type(self).__name__
+
+class CardKeyProviderCsv(CardKeyProvider):
+    """Card key provider implementation that allows to query against a specified CSV file."""
+
+    def __init__(self, csv_filename: str, transport_keys: dict):
+        """
+        Args:
+                csv_filename : file name (path) of CSV file containing card-individual key/data
+                transport_keys : (see class CardKeyFieldCryptor)
+        """
+        self.csv_file = open(csv_filename, 'r')
+        if not self.csv_file:
+            raise RuntimeError("Could not open CSV file '%s'" % csv_filename)
+        self.csv_filename = csv_filename
+        self.crypt = CardKeyFieldCryptor(transport_keys)
+
+    def get(self, fields: List[str], key: str, value: str) -> Dict[str, str]:
         self.csv_file.seek(0)
         cr = csv.DictReader(self.csv_file)
         if not cr:
-            raise RuntimeError(
-                "Could not open DictReader for CSV-File '%s'" % self.filename)
+            raise RuntimeError("Could not open DictReader for CSV-File '%s'" % self.csv_filename)
         cr.fieldnames = [field.upper() for field in cr.fieldnames]
 
-        rc = {}
+        if key not in cr.fieldnames:
+            return None
+        return_dict = {}
         for row in cr:
             if row[key] == value:
                 for f in fields:
                     if f in row:
-                        rc.update({f: self._decrypt_field(f, row[f])})
+                        return_dict.update({f: self.crypt.decrypt_field(f, row[f])})
                     else:
-                        raise RuntimeError("CSV-File '%s' lacks column '%s'" %
-                                           (self.filename, f))
-        return rc
+                        raise RuntimeError("CSV-File '%s' lacks column '%s'" % (self.csv_filename, f))
+        if return_dict == {}:
+            return None
+        return return_dict
+
 
 
 def card_key_provider_register(provider: CardKeyProvider, provider_list=card_key_providers):
@@ -167,7 +200,7 @@ def card_key_provider_register(provider: CardKeyProvider, provider_list=card_key
     provider_list.append(provider)
 
 
-def card_key_provider_get(fields, key: str, value: str, provider_list=card_key_providers) -> Dict[str, str]:
+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.
 
     Args:
@@ -178,17 +211,21 @@ def card_key_provider_get(fields, key: str, value: str, provider_list=card_key_p
     Returns:
             dictionary of {field, value} strings for each requested field from 'fields'
     """
+    key = key.upper()
+    fields = [f.upper() for f in fields]
     for p in provider_list:
         if not isinstance(p, CardKeyProvider):
-            raise ValueError(
-                "provider list contains element which is not a card data provider")
+            raise ValueError("Provider list contains element which is not a card data provider")
+        log.debug("Searching for card key data (key=%s, value=%s, provider=%s)" % (key, value, str(p)))
         result = p.get(fields, key, value)
         if result:
+            log.debug("Found card data: %s" % (str(result)))
             return result
-    return {}
+
+    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) -> Optional[str]:
+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.
 
     Args:
@@ -199,11 +236,7 @@ def card_key_provider_get_field(field: str, key: str, value: str, provider_list=
     Returns:
             dictionary of {field, value} strings for the requested field
     """
-    for p in provider_list:
-        if not isinstance(p, CardKeyProvider):
-            raise ValueError(
-                "provider list contains element which is not a card data provider")
-        result = p.get_field(field, key, value)
-        if result:
-            return result
-    return None
+
+    fields = [field]
+    result = card_key_provider_get(fields, key, value, card_key_providers)
+    return result.get(field.upper())

pySim/commands.py

@@ -141,7 +141,7 @@ class SimCardCommands:
         Returns:
                 Tuple of (decoded_data, sw)
         """
-        cmd = cmd_constr.build(cmd_data) if cmd_data else ''
+        cmd = cmd_constr.build(cmd_data) if cmd_data else b''
         lc = i2h([len(cmd)]) if cmd_data else ''
         le = '00' if resp_constr else ''
         pdu = ''.join([cla, ins, p1, p2, lc, b2h(cmd), le])

pySim/esim/es8p.py

@@ -76,10 +76,11 @@ def gen_replace_session_keys(ppk_enc: bytes, ppk_cmac: bytes, initial_mcv: bytes
 class ProfileMetadata:
     """Representation of Profile metadata. Right now only the mandatory bits are
     supported, but in general this should follow the StoreMetadataRequest of SGP.22 5.5.3"""
-    def __init__(self, iccid_bin: bytes, spn: str, profile_name: str):
+    def __init__(self, iccid_bin: bytes, spn: str, profile_name: str, profile_class = 'operational'):
         self.iccid_bin = iccid_bin
         self.spn = spn
         self.profile_name = profile_name
+        self.profile_class = profile_class
         self.icon = None
         self.icon_type = None
         self.notifications = []
@@ -105,6 +106,14 @@ class ProfileMetadata:
             'serviceProviderName': self.spn,
             'profileName': self.profile_name,
         }
+        if self.profile_class == 'test':
+            smr['profileClass'] = 0
+        elif self.profile_class == 'provisioning':
+            smr['profileClass'] = 1
+        elif self.profile_class == 'operational':
+            smr['profileClass'] = 2
+        else:
+            raise ValueError('Unsupported Profile Class %s' % self.profile_class)
         if self.icon:
             smr['icon'] = self.icon
             smr['iconType'] = self.icon_type

pySim/log.py

@@ -0,0 +1,125 @@
+# -*- coding: utf-8 -*-
+
+""" pySim: Logging
+"""
+
+#
+# (C) 2025 by Sysmocom s.f.m.c. GmbH
+# All Rights Reserved
+#
+# Author: Philipp Maier <pmaier@sysmocom.de>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+
+import logging
+from cmd2 import style
+
+class _PySimLogHandler(logging.Handler):
+    def __init__(self, log_callback):
+        super().__init__()
+        self.log_callback = log_callback
+
+    def emit(self, record):
+        formatted_message = self.format(record)
+        self.log_callback(formatted_message, record)
+
+class PySimLogger:
+    """
+    Static class to centralize the log output of PySim applications. This class can be used to print log messages from
+    any pySim module. Configuration of the log behaviour (see setup and set_ methods) is entirely optional. In case no
+    print callback is set (see setup method), the logger will pass the log messages directly to print() without applying
+    any formatting to the original log message.
+    """
+
+    LOG_FMTSTR = "%(levelname)s: %(message)s"
+    LOG_FMTSTR_VERBOSE = "%(module)s.%(lineno)d -- %(name)s - " + LOG_FMTSTR
+    __formatter = logging.Formatter(LOG_FMTSTR)
+    __formatter_verbose = logging.Formatter(LOG_FMTSTR_VERBOSE)
+
+    # No print callback by default, means that log messages are passed directly to print()
+    print_callback = None
+
+    # No specific color scheme by default
+    colors = {}
+
+    # The logging default is non-verbose logging on logging level DEBUG. This is a safe default that works for
+    # applications that ignore the presence of the PySimLogger class.
+    verbose = False
+    logging.root.setLevel(logging.DEBUG)
+
+    def __init__(self):
+        raise RuntimeError('static class, do not instantiate')
+
+    @staticmethod
+    def setup(print_callback = None, colors:dict = {}):
+        """
+        Set a print callback function and color scheme. This function call is optional. In case this method is not
+        called, default settings apply.
+        Args:
+            print_callback : A callback function that accepts the resulting log string as input. The callback should
+                             have the following format: print_callback(message:str)
+            colors : An optional dict through which certain log levels can be assigned a color.
+                     (e.g. {logging.WARN: YELLOW})
+        """
+        PySimLogger.print_callback = print_callback
+        PySimLogger.colors = colors
+
+    @staticmethod
+    def set_verbose(verbose:bool = False):
+        """
+        Enable/disable verbose logging. (has no effect in case no print callback is set, see method setup)
+        Args:
+            verbose: verbosity (True = verbose logging, False = normal logging)
+        """
+        PySimLogger.verbose = verbose;
+
+    @staticmethod
+    def set_level(level:int = logging.DEBUG):
+        """
+        Set the logging level.
+        Args:
+            level: Logging level, valis log leves are: DEBUG, INFO, WARNING, ERROR and CRITICAL
+        """
+        logging.root.setLevel(level)
+
+    @staticmethod
+    def _log_callback(message, record):
+        if not PySimLogger.print_callback:
+            # In case no print callback has been set display the message as if it were printed trough a normal
+            # python print statement.
+            print(record.message)
+        else:
+            # When a print callback is set, use it to display the log line. Apply color if the API user chose one
+            if PySimLogger.verbose:
+                formatted_message = logging.Formatter.format(PySimLogger.__formatter_verbose, record)
+            else:
+                formatted_message = logging.Formatter.format(PySimLogger.__formatter, record)
+            color = PySimLogger.colors.get(record.levelno)
+            if color:
+                PySimLogger.print_callback(style(formatted_message, fg = color))
+            else:
+                PySimLogger.print_callback(formatted_message)
+
+    @staticmethod
+    def get(log_facility: str):
+        """
+        Set up and return a new python logger object
+        Args:
+            log_facility : Name of log facility (e.g. "MAIN", "RUNTIME"...)
+        """
+        logger = logging.getLogger(log_facility)
+        handler = _PySimLogHandler(log_callback=PySimLogger._log_callback)
+        logger.addHandler(handler)
+        return logger

pySim/runtime.py

@@ -23,6 +23,9 @@ from osmocom.tlv import bertlv_parse_one
 
 from pySim.exceptions import *
 from pySim.filesystem import *
+from pySim.log import PySimLogger
+
+log = PySimLogger.get("RUNTIME")
 
 def lchan_nr_from_cla(cla: int) -> int:
     """Resolve the logical channel number from the CLA byte."""
@@ -44,6 +47,7 @@ class RuntimeState:
             card : pysim.cards.Card instance
             profile : CardProfile instance
         """
+
         self.mf = CardMF(profile=profile)
         self.card = card
         self.profile = profile
@@ -66,7 +70,7 @@ class RuntimeState:
         for addon_cls in self.profile.addons:
             addon = addon_cls()
             if addon.probe(self.card):
-                print("Detected %s Add-on \"%s\"" % (self.profile, addon))
+                log.info("Detected %s Add-on \"%s\"" % (self.profile, addon))
                 for f in addon.files_in_mf:
                     self.mf.add_file(f)
 
@@ -100,18 +104,18 @@ class RuntimeState:
         apps_taken = []
         if aids_card:
             aids_taken = []
-            print("AIDs on card:")
+            log.info("AIDs on card:")
             for a in aids_card:
                 for f in apps_profile:
                     if f.aid in a:
-                        print(" %s: %s (EF.DIR)" % (f.name, a))
+                        log.info(" %s: %s (EF.DIR)" % (f.name, a))
                         aids_taken.append(a)
                         apps_taken.append(f)
             aids_unknown = set(aids_card) - set(aids_taken)
             for a in aids_unknown:
-                print(" unknown: %s (EF.DIR)" % a)
+                log.info(" unknown: %s (EF.DIR)" % a)
         else:
-            print("warning: EF.DIR seems to be empty!")
+            log.warn("EF.DIR seems to be empty!")
 
         # Some card applications may not be registered in EF.DIR, we will actively
         # probe for those applications
@@ -126,7 +130,7 @@ class RuntimeState:
                 _data, sw = self.card.select_adf_by_aid(f.aid)
                 self.selected_adf = f
                 if sw == "9000":
-                    print(" %s: %s" % (f.name, f.aid))
+                    log.info(" %s: %s" % (f.name, f.aid))
                     apps_taken.append(f)
             except (SwMatchError, ProtocolError):
                 pass
@@ -510,6 +514,47 @@ class RuntimeLchan:
         dec_data = self.selected_file.decode_hex(data)
         return (dec_data, sw)
 
+    def __get_writeable_size(self):
+        """ Determine the writable size (file or record) using the cached FCP parameters of the currently selected
+            file. Return None in case the writeable size cannot be determined (no FCP available, FCP lacks size
+            information).
+        """
+        fcp = self.selected_file_fcp
+        if not fcp:
+            return None
+
+        structure = fcp.get('file_descriptor', {}).get('file_descriptor_byte', {}).get('structure')
+        if not structure:
+            return None
+
+        if structure == 'transparent':
+            return fcp.get('file_size')
+        elif structure == 'linear_fixed':
+            return fcp.get('file_descriptor', {}).get('record_len')
+        else:
+            return None
+
+    def __check_writeable_size(self, data_len):
+        """ Guard against unsuccessful writes caused by attempts to write data that exceeds the file limits. """
+
+        writeable_size = self.__get_writeable_size()
+        if not writeable_size:
+            return
+
+        if isinstance(self.selected_file, TransparentEF):
+            writeable_name = "file"
+        elif isinstance(self.selected_file, LinFixedEF):
+            writeable_name = "record"
+        else:
+            writeable_name = "object"
+
+        if data_len > writeable_size:
+            raise TypeError("Data length (%u) exceeds %s size (%u) by %u bytes" %
+                            (data_len, writeable_name, writeable_size, data_len - writeable_size))
+        elif data_len < writeable_size:
+            log.warn("Data length (%u) less than %s size (%u), leaving %u unwritten bytes at the end of the %s" %
+                     (data_len, writeable_name, writeable_size, writeable_size - data_len, writeable_name))
+
     def update_binary(self, data_hex: str, offset: int = 0):
         """Update transparent EF binary data.
 
@@ -520,6 +565,7 @@ class RuntimeLchan:
         if not isinstance(self.selected_file, TransparentEF):
             raise TypeError("Only works with TransparentEF, but %s is %s" % (self.selected_file,
                                                                              self.selected_file.__class__.__mro__))
+        self.__check_writeable_size(len(data_hex) // 2 + offset)
         return self.scc.update_binary(self.selected_file.fid, data_hex, offset, conserve=self.rs.conserve_write)
 
     def update_binary_dec(self, data: dict):
@@ -567,6 +613,7 @@ class RuntimeLchan:
         if not isinstance(self.selected_file, LinFixedEF):
             raise TypeError("Only works with Linear Fixed EF, but %s is %s" % (self.selected_file,
                                                                                self.selected_file.__class__.__mro__))
+        self.__check_writeable_size(len(data_hex) // 2)
         return self.scc.update_record(self.selected_file.fid, rec_nr, data_hex,
 					       conserve=self.rs.conserve_write,
 					       leftpad=self.selected_file.leftpad)

pyproject.toml

@@ -1,3 +1,6 @@
 [build-system]
 requires = ["setuptools", "wheel"]
 build-backend = "setuptools.build_meta"
+
+[tool.pylint.main]
+ignored-classes = ["twisted.internet.reactor"]

smpp_ota_apdu.py

@@ -0,0 +1,107 @@
+#!/usr/bin/env python3
+import logging
+import sys
+from pprint import pprint as pp
+
+from pySim.ota import OtaKeyset, OtaDialectSms
+from pySim.utils import b2h, h2b
+
+import smpplib.gsm
+import smpplib.client
+import smpplib.consts
+import argparse
+
+logger = logging.getLogger(__name__)
+
+# if you want to know what's happening
+logging.basicConfig(level='DEBUG')
+
+
+
+class Foo:
+    def smpp_rx_handler(self, pdu):
+        sys.stdout.write('delivered {}\n'.format(pdu.receipted_message_id))
+        if pdu.short_message:
+            try:
+                dec = self.ota_dialect.decode_resp(self.ota_keyset, self.spi, pdu.short_message)
+            except ValueError:
+                spi = self.spi.copy()
+                spi['por_shall_be_ciphered'] = False
+                spi['por_rc_cc_ds'] = 'no_rc_cc_ds'
+                dec = self.ota_dialect.decode_resp(self.ota_keyset, spi, pdu.short_message)
+            pp(dec)
+        return None
+
+    def __init__(self, kic, kid, idx, tar):
+        # Two parts, UCS2, SMS with UDH
+        #parts, encoding_flag, msg_type_flag = smpplib.gsm.make_parts(u'Привет мир!\n'*10)
+
+        client = smpplib.client.Client('localhost', 2775, allow_unknown_opt_params=True)
+
+        # Print when obtain message_id
+        client.set_message_sent_handler(
+            lambda pdu: sys.stdout.write('sent {} {}\n'.format(pdu.sequence, pdu.message_id)))
+        #client.set_message_received_handler(
+        #    lambda pdu: sys.stdout.write('delivered {}\n'.format(pdu.receipted_message_id)))
+        client.set_message_received_handler(self.smpp_rx_handler)
+
+        client.connect()
+        client.bind_transceiver(system_id='test', password='test')
+
+        self.client = client
+
+
+        self.ota_keyset = OtaKeyset(algo_crypt='triple_des_cbc2', kic_idx=idx, kic=h2b(kic),
+                                    algo_auth='triple_des_cbc2', kid_idx=idx, kid=h2b(kid))
+        self.ota_keyset.cntr = 0xdadb
+        self.tar = h2b(tar)
+
+        self.ota_dialect = OtaDialectSms()
+        self.spi = {'counter':'no_counter', 'ciphering':True, 'rc_cc_ds': 'cc', 'por_in_submit':False,
+                    'por_shall_be_ciphered':True, 'por_rc_cc_ds': 'cc', 'por': 'por_required'}
+
+
+    def tx_sms_tpdu(self, tpdu: bytes):
+        self.client.send_message(
+            source_addr_ton=smpplib.consts.SMPP_TON_INTL,
+            #source_addr_npi=smpplib.consts.SMPP_NPI_ISDN,
+            # Make sure it is a byte string, not unicode:
+            source_addr='12',
+
+            dest_addr_ton=smpplib.consts.SMPP_TON_INTL,
+            #dest_addr_npi=smpplib.consts.SMPP_NPI_ISDN,
+            # Make sure thease two params are byte strings, not unicode:
+            destination_addr='23',
+            short_message=tpdu,
+
+            data_coding=smpplib.consts.SMPP_ENCODING_BINARY,
+            esm_class=smpplib.consts.SMPP_GSMFEAT_UDHI,
+            protocol_id=0x7f,
+            #registered_delivery=True,
+        )
+
+    def tx_c_apdu(self, apdu: bytes):
+        logger.info("C-APDU: %s" % b2h(apdu))
+        # translate to Secured OTA RFM
+        secured = self.ota_dialect.encode_cmd(self.ota_keyset, self.tar, self.spi, apdu=apdu)
+        # add user data header
+        tpdu = b'\x02\x70\x00' + secured
+        # send via SMPP
+        self.tx_sms_tpdu(tpdu)
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+
+    parser.add_argument('-c', '--kic')
+    parser.add_argument('-d', '--kid')
+    parser.add_argument('-i', '--idx', type=int, default=1)
+    parser.add_argument('-t', '--tar', default='b00011')
+    parser.add_argument('apdu', default="", nargs='+')
+    args = parser.parse_args()
+
+    f = Foo(args.kic, args.kid, args.idx, args.tar)
+    print("initialized, sending APDU")
+    f.tx_c_apdu(h2b("".join(args.apdu)))
+
+    f.client.listen()

tests/pySim-trace_test/pySim-trace_test_gsmtap.pcapng.ok

@@ -2,7 +2,7 @@ Detected UICC Add-on "SIM"
 Detected UICC Add-on "GSM-R"
 Detected UICC Add-on "RUIM"
 Can't read AIDs from SIM -- 'list' object has no attribute 'lower'
-warning: EF.DIR seems to be empty!
+EF.DIR seems to be empty!
  ADF.ECASD: a0000005591010ffffffff8900000200
  ADF.ISD-R: a0000005591010ffffffff8900000100
  ISIM: a0000000871004

tests/unittests/test_card_key_provider.csv

@@ -0,0 +1,5 @@
+"card_type_id","formfactor_id","imsi","iccid","pin1","puk1","pin2","puk2","ki","adm1","adm2","proprietary","kic1","kic2","kic3","kid1","kid2","kid3","kik1","kik2","kik3","msisdn","acc","opc"
+"myCardType","3FF","901700000000001","8988211000000000001","1234","12345678","1223","12345678","AAAAAAAAAAA5435425AAAAAAAAAAAAAA","10101010","9999999999999999","proprietary data 01","BBBBBBBBBB3324BBBBBBBB21212BBBBB","CC7654CCCCCCCCCCCCCCCCCCCCCCCCCC","DDDD90DDDDDDDDDDDDDDDDDD767DDDDD","EEEEEE567657567567EEEEEEEEEEEEEE","FFFFFFFFFFFFFFFFFFF56765765FFFFF","11111567811111111111111111111111","22222222222222222227669999222222","33333333333333333333333333333333","44444444444444445234544444444444","55555555555","0001","66666666666666666666666666666666"
+"myCardType","3FF","901700000000002","8988211000000000002","1234","12345678","1223","12345678","AAAAAAAAAAAAAAAAAAAAAAAA3425AAAA","10101010","9999999999999999","proprietary data 02","BBBBBB421BBBBBBBBBB12BBBBBBBBBBB","CCCCCCCCCC3456CCCCCCCCCCCCCCCCCC","DDDDDDDDD567657DDDD2DDDDDDDDDDDD","EEEEEEEE56756EEEEEEEEE567657EEEE","FFFFF567657FFFFFFFFFFFFFFFFFFFFF","11111111111146113433411576511111","22222222222223432225765222222222","33333333333333523453333333333333","44425435234444444444446544444444","55555555555","0001","66666666666666266666666666666666"
+"myCardType","3FF","901700000000003","8988211000000000003","1234","12345678","1223","12345678","AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA","10101010","9999999999999999","proprietary data 03","BBBBBBB45678BBBB756765BBBBBBBBBB","CCCCCCCCCCCCCC76543CCCC56765CCCC","DDDDDDDDDDDDDDDDDD5676575DDDDDDD","EEEEEEEEEEEEEEEEEE56765EEEEEEEEE","FFFFFFFFFFFFFFF567657FFFFFFFFFFF","11111111119876511111111111111111","22222222222444422222222222576522","33333332543333576733333333333333","44444444444567657567444444444444","55555555555","0001","66666675676575666666666666666666"
+

tests/unittests/test_card_key_provider.py

@@ -0,0 +1,152 @@
+#!/usr/bin/env python3
+
+import unittest
+import os
+from pySim.card_key_provider import *
+
+class TestCardKeyProviderCsv(unittest.TestCase):
+
+    def __init__(self, *args, **kwargs):
+        column_keys = {"KI" : "000424252525535532532A0B0C0D0E0F",
+                       "OPC" : "000102030405065545645645645D0E0F",
+                       "KIC1" : "06410203546406456456450B0C0D0E0F",
+                       "KID1" : "00040267840507667609045645645E0F",
+                       "KIK1" : "0001020307687607668678678C0D0E0F",
+                       "KIC2" : "000142457594860706090A0B0688678F",
+                       "KID2" : "600102030405649468690A0B0C0D648F",
+                       "KIK2" : "00010203330506070496330B08640E0F",
+                       "KIC3" : "000104030405064684686A068C0D0E0F",
+                       "KID3" : "00010243048468070809060B0C0D0E0F",
+                       "KIK3" : "00010204040506070809488B0C0D0E0F"}
+
+        csv_file_path = os.path.dirname(os.path.abspath(__file__)) + "/test_card_key_provider.csv"
+        card_key_provider_register(CardKeyProviderCsv(csv_file_path, column_keys))
+        super().__init__(*args, **kwargs)
+
+    def test_card_key_provider_get(self):
+        test_data = [{'EXPECTED' : {'PIN1': '1234', 'PUK1': '12345678', 'PIN2': '1223', 'PUK2': '12345678',
+                                    'KI': '48a6d5f60567d45299e3ba08594009e7', 'ADM1': '10101010',
+                                    'ADM2': '9999999999999999', 'KIC1': '3eb8567fa0b4b1e63bcab13bff5f2702',
+                                    'KIC2': 'fd6c173a5b3f04b563808da24237fb46',
+                                    'KIC3': '66c8c848e5dff69d70689d155d44f323',
+                                    'KID1': 'd78accce870332dced467c173244dd94',
+                                    'KID2': 'b3bf050969747b2d2c9389e127a3d791',
+                                    'KID3': '40a77deb50d260b3041bbde1b5040625',
+                                    'KIK1': '451b503239d818ea34421aa9c2a8887a',
+                                    'KIK2': '967716f5fca8ae179f87f76524d1ae6b',
+                                    'KIK3': '0884db5eee5409a00fc1bbc57ac52541',
+                                    'OPC': '81817574c1961dd272ad080eb2caf279'}, 'ICCID' :"8988211000000000001"},
+                     {'EXPECTED' : {'PIN1': '1234', 'PUK1': '12345678', 'PIN2': '1223', 'PUK2': '12345678',
+                                    'KI': 'e94d7fa6fb92375dae86744ff6ecef49', 'ADM1': '10101010',
+                                    'ADM2': '9999999999999999', 'KIC1': '79b4e39387c66253da68f653381ded44',
+                                    'KIC2': '560561b5dba89c1da8d1920049e5e4f7',
+                                    'KIC3': '79ff35e84e39305a119af8c79f84e8e5',
+                                    'KID1': '233baf89122159553d67545ecedcf8e0',
+                                    'KID2': '8fc2874164d7a8e40d72c968bc894ab8',
+                                    'KID3': '2e3320f0dda85054d261be920fbfa065',
+                                    'KIK1': 'd51b1b17630103d1672a3e9e0e4827ed',
+                                    'KIK2': 'd01edbc48be555139506b0d7982bf7ff',
+                                    'KIK3': 'a6487a5170849e8e0a03026afea91f5a',
+                                    'OPC': '6b0d19ef28bd12f2daac31828d426939'}, 'ICCID' :"8988211000000000002"},
+                     {'EXPECTED' : {'PIN1': '1234', 'PUK1': '12345678', 'PIN2': '1223', 'PUK2': '12345678',
+                                    'KI': '3cdec1552ef433a89f327905213c5a6e', 'ADM1': '10101010',
+                                    'ADM2': '9999999999999999', 'KIC1': '72986b13ce505e12653ad42df5cfca13',
+                                    'KIC2': '8f0d1e58b01e833773e5562c4940674d',
+                                    'KIC3': '9c72ba5a14d54f489edbffd3d8802f03',
+                                    'KID1': 'd23a42995df9ca83f74b2cfd22695526',
+                                    'KID2': '5c3a189d12aa1ac6614883d7de5e6c8c',
+                                    'KID3': 'a6ace0d303a2b38a96b418ab83c16725',
+                                    'KIK1': 'bf2319467d859c12527aa598430caef2',
+                                    'KIK2': '6a4c459934bea7e40787976b8881ab01',
+                                    'KIK3': '91cd02c38b5f68a98cc90a1f2299538f',
+                                    'OPC': '6df46814b1697daca003da23808bbbc3'}, 'ICCID' :"8988211000000000003"}]
+
+        for t in test_data:
+            result = card_key_provider_get(["PIN1","PUK1","PIN2","PUK2","KI","ADM1","ADM2","KIC1",
+                                            "KIC2","KIC3","KID1","KID2","KID3","KIK1","KIK2","KIK3","OPC"],
+                                           "ICCID", t.get('ICCID'))
+            self.assertEqual(result, t.get('EXPECTED'))
+            result = card_key_provider_get(["PIN1","puk1","PIN2","PUK2","KI","adm1","ADM2","KIC1",
+                                            "KIC2","kic3","KID1","KID2","KID3","kik1","KIK2","KIK3","OPC"],
+                                           "iccid", t.get('ICCID'))
+            self.assertEqual(result, t.get('EXPECTED'))
+
+
+    def test_card_key_provider_get_field(self):
+        test_data = [{'EXPECTED' : "3eb8567fa0b4b1e63bcab13bff5f2702", 'ICCID' :"8988211000000000001"},
+                     {'EXPECTED' : "79b4e39387c66253da68f653381ded44", 'ICCID' :"8988211000000000002"},
+                     {'EXPECTED' : "72986b13ce505e12653ad42df5cfca13", 'ICCID' :"8988211000000000003"}]
+
+        for t in test_data:
+            result = card_key_provider_get_field("KIC1", "ICCID", t.get('ICCID'))
+            self.assertEqual(result, t.get('EXPECTED'))
+        for t in test_data:
+            result = card_key_provider_get_field("kic1", "iccid", t.get('ICCID'))
+            self.assertEqual(result, t.get('EXPECTED'))
+
+
+class TestCardKeyFieldCryptor(unittest.TestCase):
+
+    def __init__(self, *args, **kwargs):
+        transport_keys = {"KI" : "000424252525535532532A0B0C0D0E0F",
+                          "OPC" : "000102030405065545645645645D0E0F",
+                          "KIC1" : "06410203546406456456450B0C0D0E0F",
+                          "UICC_SCP03" : "00040267840507667609045645645E0F"}
+        self.crypt = CardKeyFieldCryptor(transport_keys)
+        super().__init__(*args, **kwargs)
+
+    def test_encrypt_field(self):
+        test_data = [{'EXPECTED' : "0b1e1e56cd62645aeb4c2d72a7c98f27",
+                      'PLAINTEXT_VAL' : "000102030405060708090a0b0c0d0e0f", 'FIELDNAME' : "OPC"},
+                     {'EXPECTED' : "000102030405060708090a0b0c0d0e0f",
+                      'PLAINTEXT_VAL' : "000102030405060708090a0b0c0d0e0f", 'FIELDNAME' : "NOCRYPT"},
+                     {'EXPECTED' : "00248276d2734f108f9761e2f98e2a9d",
+                      'PLAINTEXT_VAL' : "000102030405060708090a0b0c0d0e0f", 'FIELDNAME' : "UICC_SCP03_KIC1"},
+                     {'EXPECTED' : "00248276d2734f108f9761e2f98e2a9d",
+                      'PLAINTEXT_VAL' : "000102030405060708090a0b0c0d0e0f", 'FIELDNAME' : "UICC_SCP03_KID1"},
+                     {'EXPECTED' : "00248276d2734f108f9761e2f98e2a9d",
+                      'PLAINTEXT_VAL' : "000102030405060708090a0b0c0d0e0f", 'FIELDNAME' : "UICC_SCP03_KIK1"},
+                     {'EXPECTED' : "0b1e1e56cd62645aeb4c2d72a7c98f27",
+                      'PLAINTEXT_VAL' : "000102030405060708090a0b0c0d0e0f", 'FIELDNAME' : "opc"},
+                     {'EXPECTED' : "000102030405060708090a0b0c0d0e0f",
+                      'PLAINTEXT_VAL' : "000102030405060708090a0b0c0d0e0f", 'FIELDNAME' : "nocrypt"},
+                     {'EXPECTED' : "00248276d2734f108f9761e2f98e2a9d",
+                      'PLAINTEXT_VAL' : "000102030405060708090a0b0c0d0e0f", 'FIELDNAME' : "uicc_scp03_kic1"},
+                     {'EXPECTED' : "00248276d2734f108f9761e2f98e2a9d",
+                      'PLAINTEXT_VAL' : "000102030405060708090a0b0c0d0e0f", 'FIELDNAME' : "uicc_scp03_kid1"},
+                     {'EXPECTED' : "00248276d2734f108f9761e2f98e2a9d",
+                      'PLAINTEXT_VAL' : "000102030405060708090a0b0c0d0e0f", 'FIELDNAME' : "uicc_scp03_kik1"}]
+
+        for t in test_data:
+            result = self.crypt.encrypt_field(t.get('FIELDNAME'), t.get('PLAINTEXT_VAL'))
+            self.assertEqual(result, t.get('EXPECTED'))
+
+    def test_decrypt_field(self):
+        test_data = [{'EXPECTED' : "000102030405060708090a0b0c0d0e0f",
+                      'ENCRYPTED_VAL' : "0b1e1e56cd62645aeb4c2d72a7c98f27", 'FIELDNAME' : "OPC"},
+                     {'EXPECTED' : "000102030405060708090a0b0c0d0e0f",
+                      'ENCRYPTED_VAL' : "000102030405060708090a0b0c0d0e0f", 'FIELDNAME' : "NOCRYPT"},
+                     {'EXPECTED' : "000102030405060708090a0b0c0d0e0f",
+                      'ENCRYPTED_VAL' : "00248276d2734f108f9761e2f98e2a9d", 'FIELDNAME' : "UICC_SCP03_KIC1"},
+                     {'EXPECTED' : "000102030405060708090a0b0c0d0e0f",
+                      'ENCRYPTED_VAL' : "00248276d2734f108f9761e2f98e2a9d", 'FIELDNAME' : "UICC_SCP03_KID1"},
+                     {'EXPECTED' : "000102030405060708090a0b0c0d0e0f",
+                      'ENCRYPTED_VAL' : "00248276d2734f108f9761e2f98e2a9d", 'FIELDNAME' : "UICC_SCP03_KIK1"},
+                     {'EXPECTED' : "000102030405060708090a0b0c0d0e0f",
+                      'ENCRYPTED_VAL' : "0b1e1e56cd62645aeb4c2d72a7c98f27", 'FIELDNAME' : "opc"},
+                     {'EXPECTED' : "000102030405060708090a0b0c0d0e0f",
+                      'ENCRYPTED_VAL' : "000102030405060708090a0b0c0d0e0f", 'FIELDNAME' : "nocrypt"},
+                     {'EXPECTED' : "000102030405060708090a0b0c0d0e0f",
+                      'ENCRYPTED_VAL' : "00248276d2734f108f9761e2f98e2a9d", 'FIELDNAME' : "uicc_scp03_kic1"},
+                     {'EXPECTED' : "000102030405060708090a0b0c0d0e0f",
+                      'ENCRYPTED_VAL' : "00248276d2734f108f9761e2f98e2a9d", 'FIELDNAME' : "uicc_scp03_kid1"},
+                     {'EXPECTED' : "000102030405060708090a0b0c0d0e0f",
+                      'ENCRYPTED_VAL' : "00248276d2734f108f9761e2f98e2a9d", 'FIELDNAME' : "uicc_scp03_kik1"}]
+
+        for t in test_data:
+            result = self.crypt.decrypt_field(t.get('FIELDNAME'), t.get('ENCRYPTED_VAL'))
+            self.assertEqual(result, t.get('EXPECTED'))
+
+
+if __name__ == "__main__":
+	unittest.main()

tests/unittests/test_log.py

@@ -0,0 +1,121 @@
+#!/usr/bin/env python3
+
+# (C) 2025 by Sysmocom s.f.m.c. GmbH
+# All Rights Reserved
+#
+# Author: Philipp Maier <pmaier@sysmocom.de>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import unittest
+import logging
+from pySim.log import PySimLogger
+import io
+import sys
+from inspect import currentframe, getframeinfo
+
+log = PySimLogger.get("TEST")
+
+TEST_MSG_DEBUG = "this is a debug message"
+TEST_MSG_INFO = "this is an info message"
+TEST_MSG_WARNING = "this is a warning message"
+TEST_MSG_ERROR = "this is an error message"
+TEST_MSG_CRITICAL = "this is a critical message"
+
+expected_message = None
+
+class PySimLogger_Test(unittest.TestCase):
+
+    def __test_01_safe_defaults_one(self, callback, message:str):
+        # When log messages are sent to an unconfigured PySimLogger class, we expect the unmodified message being
+        # logged to stdout, just as if it were printed via a normal print() statement.
+        log_output = io.StringIO()
+        sys.stdout = log_output
+        callback(message)
+        assert(log_output.getvalue().strip() == message)
+        sys.stdout = sys.__stdout__
+
+    def test_01_safe_defaults(self):
+        # When log messages are sent to an unconfigured PySimLogger class, we expect that all messages are logged,
+        # regardless of the logging level.
+        self.__test_01_safe_defaults_one(log.debug, TEST_MSG_DEBUG)
+        self.__test_01_safe_defaults_one(log.info, TEST_MSG_INFO)
+        self.__test_01_safe_defaults_one(log.warning, TEST_MSG_WARNING)
+        self.__test_01_safe_defaults_one(log.error, TEST_MSG_ERROR)
+        self.__test_01_safe_defaults_one(log.critical, TEST_MSG_CRITICAL)
+
+    @staticmethod
+    def _test_print_callback(message):
+        assert(message.strip() == expected_message)
+
+    def test_02_normal(self):
+        # When the PySimLogger is set up with its default values, we expect formatted log messages on all logging
+        # levels.
+        global expected_message
+        PySimLogger.setup(self._test_print_callback)
+        expected_message = "DEBUG: " + TEST_MSG_DEBUG
+        log.debug(TEST_MSG_DEBUG)
+        expected_message = "INFO: " + TEST_MSG_INFO
+        log.info(TEST_MSG_INFO)
+        expected_message = "WARNING: " + TEST_MSG_WARNING
+        log.warning(TEST_MSG_WARNING)
+        expected_message = "ERROR: " + TEST_MSG_ERROR
+        log.error(TEST_MSG_ERROR)
+        expected_message = "CRITICAL: " + TEST_MSG_CRITICAL
+        log.critical(TEST_MSG_CRITICAL)
+
+    def test_03_verbose(self):
+        # When the PySimLogger is set up with its default values, we expect verbose formatted log messages on all
+        # logging levels.
+        global expected_message
+        PySimLogger.setup(self._test_print_callback)
+        PySimLogger.set_verbose(True)
+        frame = currentframe()
+        expected_message = __name__ + "." + str(getframeinfo(frame).lineno + 1) + " -- TEST - DEBUG: " + TEST_MSG_DEBUG
+        log.debug(TEST_MSG_DEBUG)
+        expected_message = __name__ + "." + str(getframeinfo(frame).lineno + 1) + " -- TEST - INFO: " + TEST_MSG_INFO
+        log.info(TEST_MSG_INFO)
+        expected_message = __name__ + "." + str(getframeinfo(frame).lineno + 1) + " -- TEST - WARNING: " + TEST_MSG_WARNING
+        log.warning(TEST_MSG_WARNING)
+        expected_message = __name__ + "." + str(getframeinfo(frame).lineno + 1) + " -- TEST - ERROR: " + TEST_MSG_ERROR
+        log.error(TEST_MSG_ERROR)
+        expected_message = __name__ + "." + str(getframeinfo(frame).lineno + 1) + " -- TEST - CRITICAL: " + TEST_MSG_CRITICAL
+        log.critical(TEST_MSG_CRITICAL)
+
+    def test_04_level(self):
+        # When the PySimLogger is set up with its default values, we expect formatted log messages but since we will
+        # limit the log level to INFO, we should not see any messages of level DEBUG
+        global expected_message
+        PySimLogger.setup(self._test_print_callback)
+        PySimLogger.set_level(logging.INFO)
+
+        # We test this in non verbose mode, this will also confirm that disabeling the verbose mode works.
+        PySimLogger.set_verbose(False)
+
+        # Debug messages should not appear
+        expected_message = None
+        log.debug(TEST_MSG_DEBUG)
+
+        # All other messages should appear normally
+        expected_message = "INFO: " + TEST_MSG_INFO
+        log.info(TEST_MSG_INFO)
+        expected_message = "WARNING: " + TEST_MSG_WARNING
+        log.warning(TEST_MSG_WARNING)
+        expected_message = "ERROR: " + TEST_MSG_ERROR
+        log.error(TEST_MSG_ERROR)
+        expected_message = "CRITICAL: " + TEST_MSG_CRITICAL
+        log.critical(TEST_MSG_CRITICAL)
+
+if __name__ == '__main__':
+    unittest.main()