docs/smpp-ota-tool: Add documentation/tutorial

We already have documentation that explains how to run pySim-smpp2sim.
With smpp-ota-tool we now have a counterpart for pySim-smpp2sim, so
let's add documentation for this tool as well.

Related: SYS#7881
Change-Id: If0d18a263f5a6dc035b90f5c5c6a942d46bbba49

.gitignore

@@ -1,5 +1,5 @@
 *.pyc
-.*.swp
+.*.sw?
 
 /docs/_*
 /docs/generated

contrib/analyze_simaResponse.py

@@ -0,0 +1,112 @@
+#!/usr/bin/env python3
+
+# A tool to analyze the eUICC simaResponse (series of EUICCResponse)
+#
+# (C) 2025 by sysmocom - s.f.m.c. GmbH
+# All Rights Reserved
+#
+# 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 argparse
+from osmocom.utils import h2b, b2h
+from osmocom.tlv import bertlv_parse_one, bertlv_encode_tag, bertlv_encode_len
+from pySim.esim.saip import *
+
+parser = argparse.ArgumentParser(description="""Utility program to analyze the contents of an eUICC simaResponse.""")
+parser.add_argument('SIMA_RESPONSE', help='Hexstring containing the simaResponse as received from the eUICC')
+
+def split_sima_response(sima_response):
+    """split an eUICC simaResponse field into a list of EUICCResponse fields"""
+
+    remainder = sima_response
+    result = []
+    while len(remainder):
+        tdict, l, v, next_remainder = bertlv_parse_one(remainder)
+        rawtag = bertlv_encode_tag(tdict)
+        rawlen = bertlv_encode_len(l)
+        result = result + [remainder[0:len(rawtag) + len(rawlen) + l]]
+        remainder = next_remainder
+    return result
+
+def analyze_status(status):
+    """
+    Convert a status code (integer) into a human readable string
+    (see eUICC Profile Package: Interoperable Format Technical Specification, section 8.11)
+    """
+
+    # SIMA status codes
+    string_values = {0 : 'ok',
+                     1 : 'pe-not-supported',
+                     2 : 'memory-failure',
+                     3 : 'bad-values',
+                     4 : 'not-enough-memory',
+                     5 : 'invalid-request-format',
+                     6 : 'invalid-parameter',
+                     7 : 'runtime-not-supported',
+                     8 : 'lib-not-supported',
+                     9 : 'template-not-supported ',
+                     10 : 'feature-not-supported',
+                     11 : 'pin-code-missing',
+                     31 : 'unsupported-profile-version'}
+
+    string_value = string_values.get(status, None)
+    if string_value is not None:
+        return "%d = %s (SIMA status code)" % (status, string_value)
+
+    # ISO 7816 status words
+    if status >= 24576 and status <= 28671:
+        return "%d = %04x (ISO7816 status word)" % (status, status)
+    elif status >= 36864 and status <= 40959:
+        return "%d = %04x (ISO7816 status word)" % (status, status)
+
+    # Proprietary status codes
+    elif status >= 40960 and status <= 65535:
+        return "%d = %04x (proprietary)" % (status, status)
+
+    # Unknown status codes
+    return "%d (unknown, proprietary?)" % status
+
+def analyze_euicc_response(euicc_response):
+    """Analyze and display the contents of an EUICCResponse"""
+
+    print(" EUICCResponse: %s" % b2h(euicc_response))
+    euicc_response_decoded = asn1.decode('EUICCResponse', euicc_response)
+
+    pe_status = euicc_response_decoded.get('peStatus')
+    print("  peStatus:")
+    for s in pe_status:
+        print("   status: %s" % analyze_status(s.get('status')))
+        print("   identification: %s" % str(s.get('identification', None)))
+        print("   additional-information: %s" % str(s.get('additional-information', None)))
+        print("   offset: %s" % str(s.get('offset', None)))
+
+    if euicc_response_decoded.get('profileInstallationAborted', False) is None:
+        # This type is defined as profileInstallationAborted NULL OPTIONAL, so when it is present it
+        # will have the value None, otherwise it is simply not present.
+        print("  profileInstallationAborted: True")
+    else:
+        print("  profileInstallationAborted: False")
+
+    status_message = euicc_response_decoded.get('statusMessage', None)
+    print("  statusMessage: %s" % str(status_message))
+
+if __name__ == '__main__':
+    opts = parser.parse_args()
+    sima_response = h2b(opts.SIMA_RESPONSE);
+
+    print("simaResponse: %s" % b2h(sima_response))
+    euicc_response_list = split_sima_response(sima_response)
+
+    for euicc_response in euicc_response_list:
+        analyze_euicc_response(euicc_response)

contrib/csv-to-pgsql.py

@@ -0,0 +1,304 @@
+#!/usr/bin/env python3
+
+# (C) 2025 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 argparse
+import logging
+import csv
+import sys
+import os
+import yaml
+import psycopg2
+from psycopg2.sql import Identifier, SQL
+from pathlib import Path
+from pySim.log import PySimLogger
+from packaging import version
+
+log = PySimLogger.get(Path(__file__).stem)
+
+class CardKeyDatabase:
+    def __init__(self, config_filename: str, table_name: str, create_table: bool = False, admin: bool = False):
+        """
+        Initialize database connection and set the table which shall be used as storage for the card key data.
+        In case the specified table does not exist yet it can be created using the create_table_type parameter.
+
+        New tables are always minimal tables which follow a pre-defined table scheme. The user may extend the table
+        with additional columns using the add_cols() later.
+
+        Args:
+                tablename : name of the database table to create.
+                create_table_type : type of the table to create ('UICC' or 'EUICC')
+        """
+
+        def user_from_config_file(config, role: str) -> tuple[str, str]:
+            db_users = config.get('db_users')
+            user = db_users.get(role)
+            if user is None:
+                raise ValueError("user for role '%s' not set up in config file." % role)
+            return user.get('name'), user.get('pass')
+
+        self.table = table_name.lower()
+        self.cols = None
+
+        # Depending on the table type, the table name must contain either the substring "uicc_keys" or "euicc_keys".
+        # This convention will allow us to deduct the table type from the table name.
+        if "euicc_keys" not in table_name and "uicc_keys" not in table_name:
+            raise ValueError("Table name (%s) should contain the substring \"uicc_keys\" or \"euicc_keys\"" % table_name)
+
+        # Read config file
+        log.info("Using config file: %s", config_filename)
+        with open(config_filename, "r") as cfg:
+            config = yaml.load(cfg, Loader=yaml.FullLoader)
+        host = config.get('host')
+        log.info("Database host: %s", host)
+        db_name = config.get('db_name')
+        log.info("Database name: %s", db_name)
+        table_names = config.get('table_names')
+        username_admin, password_admin = user_from_config_file(config, 'admin')
+        username_importer, password_importer = user_from_config_file(config, 'importer')
+        username_reader, _ = user_from_config_file(config, 'reader')
+
+        # Switch between admin and importer user
+        if admin:
+            username, password = username_admin, password_admin
+        else:
+            username, password = username_importer, password_importer
+
+        # Create database connection
+        log.info("Database user: %s", username)
+        self.conn = psycopg2.connect(dbname=db_name, user=username, password=password, host=host)
+        self.cur = self.conn.cursor()
+
+        # In the context of this tool it is not relevant if the table name is present in the config file. However,
+        # pySim-shell.py will require the table name to be configured properly to access the database table.
+        if self.table not in table_names:
+            log.warning("Specified table name (%s) is not yet present in config file (required for access from pySim-shell.py)",
+                        self.table)
+
+        # Create a new minimal database table of the specified table type.
+        if create_table:
+            if not admin:
+                raise ValueError("creation of new table refused, use option --admin and try again.")
+            if "euicc_keys" in self.table:
+                self.__create_table(username_reader, username_importer, ['EID'])
+            elif "uicc_keys" in self.table:
+                self.__create_table(username_reader, username_importer, ['ICCID', 'IMSI'])
+
+        # Ensure a table with the specified name exists
+        log.info("Database table: %s", self.table)
+        if self.get_cols() == []:
+            raise ValueError("Table name (%s) does not exist yet" % self.table)
+        log.info("Database table columns: %s", str(self.get_cols()))
+
+    def __create_table(self, user_reader:str, user_importer:str, cols:list[str]):
+        """
+        Initialize a new table. New tables are always minimal tables with one primary key and additional index columns.
+        Non index-columns may be added later using method _update_cols().
+        """
+
+        # Create table columns with primary key
+        query = SQL("CREATE TABLE {} ({} VARCHAR PRIMARY KEY").format(Identifier(self.table),
+                                                                      Identifier(cols[0].lower()))
+        for c in cols[1:]:
+            query += SQL(", {} VARCHAR").format(Identifier(c.lower()))
+        query += SQL(");")
+        self.cur.execute(query)
+
+        # Create indexes for all other columns
+        for c in cols[1:]:
+            self.cur.execute(query = SQL("CREATE INDEX {} ON {}({});").format(Identifier(c.lower()),
+                                                                              Identifier(self.table),
+                                                                              Identifier(c.lower())))
+
+        # Set permissions
+        self.cur.execute(SQL("GRANT INSERT ON {} TO {};").format(Identifier(self.table),
+                                                                 Identifier(user_importer)))
+        self.cur.execute(SQL("GRANT SELECT ON {} TO {};").format(Identifier(self.table),
+                                                                 Identifier(user_reader)))
+
+        log.info("New database table created: %s", self.table)
+
+    def get_cols(self) -> list[str]:
+        """
+        Get a list of all columns available in the current table scheme.
+
+        Returns:
+                list with column names (in uppercase) of the database table
+        """
+
+        # Return cached col list if present
+        if self.cols:
+            return self.cols
+
+        # Request a list of current cols from the database
+        self.cur.execute("SELECT column_name FROM information_schema.columns where table_name = %s;", (self.table,))
+
+        cols_result = self.cur.fetchall()
+        cols = []
+        for c in cols_result:
+            cols.append(c[0].upper())
+        self.cols = cols
+        return cols
+
+    def get_missing_cols(self, cols_expected:list[str]) -> list[str]:
+        """
+        Check if the current table scheme lacks any of the given expected columns.
+
+        Returns:
+                list with the missing columns.
+        """
+
+        cols_present = self.get_cols()
+        return list(set(cols_expected) - set(cols_present))
+
+    def add_cols(self, cols:list[str]):
+        """
+        Update the current table scheme with additional columns. In case the updated columns are already exist, the
+        table schema is not changed.
+
+        Args:
+                table : name of the database table to alter
+                cols : list with updated colum names to add
+        """
+
+        cols_missing = self.get_missing_cols(cols)
+
+        # Depending on the table type (see constructor), we either have a primary key 'ICCID' (for UICC data), or 'EID'
+        # (for eUICC data). Both table formats different types of data and have rather differen columns also. Let's
+        # prevent the excidentally mixing of both types.
+        if 'ICCID' in cols_missing:
+            raise ValueError("Table %s stores eUCCC key material, refusing to add UICC specific column 'ICCID'" % self.table)
+        if 'EID' in cols_missing:
+            raise ValueError("Table %s stores UCCC key material, refusing to add eUICC specific column 'EID'" % self.table)
+
+        # Add the missing columns to the table
+        self.cols = None
+        for c in cols_missing:
+            self.cur.execute(query = SQL("ALTER TABLE {} ADD {} VARCHAR;").format(Identifier(self.table),
+                                                                                  Identifier(c.lower())))
+
+    def insert_row(self, row:dict[str, str]):
+        """
+        Insert a new row into the database table.
+
+        Args:
+                row : dictionary with the colum names and their designated values
+        """
+
+        # Check if the row is compatible with the current table scheme
+        cols_expected = list(row.keys())
+        cols_missing = self.get_missing_cols(cols_expected)
+        if cols_missing != []:
+            raise ValueError("table %s has incompatible format, the row %s contains unknown cols %s" %
+                             (self.table, str(row), str(cols_missing)))
+
+        # Insert row into datbase table
+        row_keys = list(row.keys())
+        row_values = list(row.values())
+        query = SQL("INSERT INTO {} ").format(Identifier(self.table))
+        query += SQL("({} ").format(Identifier(row_keys[0].lower()))
+        for k in row_keys[1:]:
+            query += SQL(", {}").format(Identifier(k.lower()))
+        query += SQL(") VALUES (%s")
+        for v in row_values[1:]:
+            query += SQL(", %s")
+        query += SQL(");")
+        self.cur.execute(query, row_values)
+
+    def commit(self):
+        self.conn.commit()
+        log.info("Changes to table %s committed!", self.table)
+
+def open_csv(opts: argparse.Namespace):
+    log.info("CSV file: %s", opts.csv)
+    csv_file = open(opts.csv, 'r')
+    cr = csv.DictReader(csv_file)
+    if not cr:
+        raise RuntimeError("could not open DictReader for CSV-File '%s'" % opts.csv)
+    cr.fieldnames = [field.upper() for field in cr.fieldnames]
+    log.info("CSV file columns: %s", str(cr.fieldnames))
+    return cr
+
+def open_db(cr: csv.DictReader, opts: argparse.Namespace) -> CardKeyDatabase:
+    try:
+        db = CardKeyDatabase(os.path.expanduser(opts.pgsql), opts.table_name, opts.create_table, opts.admin)
+
+        # Check CSV format against table schema, add missing columns
+        cols_missing = db.get_missing_cols(cr.fieldnames)
+        if cols_missing != [] and (opts.update_columns or opts.create_table):
+            log.info("Adding missing columns: %s", str(cols_missing))
+            db.add_cols(cols_missing)
+            cols_missing = db.get_missing_cols(cr.fieldnames)
+
+        # Make sure the table schema has no missing columns
+        if cols_missing != []:
+            log.error("Database table lacks CSV file columns: %s -- import aborted!", cols_missing)
+            sys.exit(2)
+    except Exception as e:
+        log.error(str(e).strip())
+        log.error("Database initialization aborted due to error!")
+        sys.exit(2)
+
+    return db
+
+def import_from_csv(db: CardKeyDatabase, cr: csv.DictReader):
+    count = 0
+    for row in cr:
+        try:
+            db.insert_row(row)
+            count+=1
+            if count % 100 == 0:
+                log.info("CSV file import in progress, %d rows imported...", count)
+        except Exception as e:
+            log.error(str(e).strip())
+            log.error("CSV file import aborted due to error, no datasets committed!")
+            sys.exit(2)
+    log.info("CSV file import done, %d rows imported", count)
+
+if __name__ == '__main__':
+    option_parser = argparse.ArgumentParser(description='CSV importer for pySim-shell\'s PostgreSQL Card Key Provider',
+                                            formatter_class=argparse.ArgumentDefaultsHelpFormatter)
+    option_parser.add_argument("--verbose", help="Enable verbose logging", action='store_true', default=False)
+    option_parser.add_argument('--pgsql', metavar='FILE',
+                               default="~/.osmocom/pysim/card_data_pgsql.cfg",
+                               help='Read card data from PostgreSQL database (config file)')
+    option_parser.add_argument('--csv', metavar='FILE', help='input CSV file with card data', required=True)
+    option_parser.add_argument("--table-name", help="name of the card key table", type=str, required=True)
+    option_parser.add_argument("--update-columns", help="add missing table columns", action='store_true', default=False)
+    option_parser.add_argument("--create-table", action='store_true', help="create new card key table", default=False)
+    option_parser.add_argument("--admin", action='store_true', help="perform action as admin", default=False)
+    opts = option_parser.parse_args()
+
+    PySimLogger.setup(print, {logging.WARN: "\033[33m"})
+    if (opts.verbose):
+        PySimLogger.set_verbose(True)
+        PySimLogger.set_level(logging.DEBUG)
+
+    # Open CSV file
+    cr = open_csv(opts)
+
+    # Open database, create initial table, update column scheme
+    db = open_db(cr, opts)
+
+    # Progress with import
+    if not opts.admin:
+        import_from_csv(db, cr)
+
+    # Commit changes to the database
+    db.commit()

contrib/es2p_server.py

@@ -0,0 +1,100 @@
+#!/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 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 sys
+import argparse
+import logging
+import json
+import asn1tools
+import asn1tools.codecs.ber
+import asn1tools.codecs.der
+import pySim.esim.rsp as rsp
+import pySim.esim.saip as saip
+from pySim.esim.es2p import param, Es2pApiServerMno, Es2pApiServerHandlerMno
+from osmocom.utils import b2h
+from datetime import datetime
+from analyze_simaResponse import split_sima_response
+from pathlib import Path
+
+logger = logging.getLogger(Path(__file__).stem)
+
+parser = argparse.ArgumentParser(description="""
+Utility to receive and log requests against the ES2+ API of an SM-DP+ according to GSMA SGP.22.""")
+parser.add_argument("--host", help="Host/IP to bind HTTP(S) to", default="localhost")
+parser.add_argument("--port", help="TCP port to bind HTTP(S) to", default=443, type=int)
+parser.add_argument('--server-cert', help='X.509 server certificate used to provide the ES2+ HTTPs service')
+parser.add_argument('--client-ca-cert', help='X.509 CA certificates to authenticate the requesting client(s)')
+parser.add_argument("-v", "--verbose", help="enable debug output", action='store_true', default=False)
+
+def decode_sima_response(sima_response):
+    decoded = []
+    euicc_response_list = split_sima_response(sima_response)
+    for euicc_response in euicc_response_list:
+        decoded.append(saip.asn1.decode('EUICCResponse', euicc_response))
+    return decoded
+
+def decode_result_data(result_data):
+    return rsp.asn1.decode('PendingNotification', result_data)
+
+def decode(data, path="/"):
+    if data is None:
+        return 'none'
+    elif type(data) is datetime:
+        return data.isoformat()
+    elif type(data) is tuple:
+        return {str(data[0]) : decode(data[1], path + str(data[0]) + "/")}
+    elif type(data) is list:
+        new_data = []
+        for item in data:
+            new_data.append(decode(item, path))
+        return new_data
+    elif type(data) is bytes:
+        return b2h(data)
+    elif type(data) is dict:
+        new_data = {}
+        for key, item in data.items():
+            new_key = str(key)
+            if path == '/' and new_key == 'resultData':
+                new_item = decode_result_data(item)
+            elif (path == '/resultData/profileInstallationResult/profileInstallationResultData/finalResult/successResult/' \
+                  or path == '/resultData/profileInstallationResult/profileInstallationResultData/finalResult/errorResult/') \
+                  and new_key == 'simaResponse':
+                new_item = decode_sima_response(item)
+            else:
+                new_item = item
+            new_data[new_key] = decode(new_item, path + new_key + "/")
+        return new_data
+    else:
+        return data
+
+class Es2pApiServerHandlerForLogging(Es2pApiServerHandlerMno):
+    def call_handleDownloadProgressInfo(self, data: dict) -> (dict, str):
+        logging.info("ES2+:handleDownloadProgressInfo: %s" % json.dumps(decode(data)))
+        return {}, None
+
+if __name__ == "__main__":
+    args = parser.parse_args()
+
+    logging.basicConfig(level=logging.DEBUG if args.verbose else logging.WARNING,
+                        format='%(asctime)s %(levelname)s %(message)s',
+                        datefmt='%Y-%m-%d %H:%M:%S')
+
+    Es2pApiServerMno(args.port, args.host, Es2pApiServerHandlerForLogging(), args.server_cert, args.client_ca_cert)
+

contrib/es9p_client.py

@@ -126,14 +126,14 @@ class Es9pClient:
         if self.opts.iccid:
             ntf_metadata['iccid'] = h2b(swap_nibbles(self.opts.iccid))
 
-        if self.opts.operation == 'download':
+        if self.opts.operation == 'install':
             pird = {
-                'transactionId': self.opts.transaction_id,
+                'transactionId': h2b(self.opts.transaction_id),
                 'notificationMetadata': ntf_metadata,
                 'smdpOid': self.opts.smdpp_oid,
                 'finalResult': ('successResult', {
-                    'aid': self.opts.isdp_aid,
-                    'simaResponse': self.opts.sima_response,
+                    'aid': h2b(self.opts.isdp_aid),
+                    'simaResponse': h2b(self.opts.sima_response),
                     }),
             }
             pird_bin = rsp.asn1.encode('ProfileInstallationResultData', pird)

contrib/jenkins.sh

@@ -42,6 +42,9 @@ case "$JOB_TYPE" in
 
 	# Run pySim-shell integration tests (requires physical cards)
 	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

contrib/saip-tool.py

@@ -107,7 +107,7 @@ parser_esrv.add_argument('--output-file', required=True, help='Output file name'
 parser_esrv.add_argument('--add-flag', default=[], choices=esrv_flag_choices, action='append', help='Add flag to mandatory services list')
 parser_esrv.add_argument('--remove-flag', default=[], choices=esrv_flag_choices, action='append', help='Remove flag from mandatory services list')
 
-parser_info = subparsers.add_parser('tree', help='Display the filesystem tree')
+parser_tree = subparsers.add_parser('tree', help='Display the filesystem tree')
 
 def write_pes(pes: ProfileElementSequence, output_file:str):
     """write the PE sequence to a file"""

contrib/smpp-ota-tool.py

@@ -0,0 +1,240 @@
+#!/usr/bin/env python3
+
+# (C) 2026 by sysmocom - s.f.m.c. GmbH
+# All Rights Reserved
+#
+# Author: Harald Welte, 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 argparse
+import logging
+import smpplib.gsm
+import smpplib.client
+import smpplib.consts
+import time
+from pySim.ota import OtaKeyset, OtaDialectSms, OtaAlgoCrypt, OtaAlgoAuth, CNTR_REQ, RC_CC_DS, POR_REQ
+from pySim.utils import b2h, h2b, is_hexstr
+from pathlib import Path
+
+logger = logging.getLogger(Path(__file__).stem)
+
+option_parser = argparse.ArgumentParser(description='Tool to send OTA SMS RFM/RAM messages via SMPP',
+                                        formatter_class=argparse.ArgumentDefaultsHelpFormatter)
+option_parser.add_argument("--host", help="Host/IP of the SMPP server", default="localhost")
+option_parser.add_argument("--port", help="TCP port of the SMPP server", default=2775, type=int)
+option_parser.add_argument("--system-id", help="System ID to use to bind to the SMPP server", default="test")
+option_parser.add_argument("--password", help="Password to use to bind to the SMPP server", default="test")
+option_parser.add_argument("--verbose", help="Enable verbose logging", action='store_true', default=False)
+algo_crypt_choices = []
+algo_crypt_classes = OtaAlgoCrypt.__subclasses__()
+for cls in algo_crypt_classes:
+    algo_crypt_choices.append(cls.enum_name)
+option_parser.add_argument("--algo-crypt", choices=algo_crypt_choices, default='triple_des_cbc2',
+                           help="OTA crypt algorithm")
+algo_auth_choices = []
+algo_auth_classes = OtaAlgoAuth.__subclasses__()
+for cls in algo_auth_classes:
+    algo_auth_choices.append(cls.enum_name)
+option_parser.add_argument("--algo-auth", choices=algo_auth_choices, default='triple_des_cbc2',
+                           help="OTA auth algorithm")
+option_parser.add_argument('--kic', required=True, type=is_hexstr, help='OTA key (KIC)')
+option_parser.add_argument('--kic-idx', default=1, type=int, help='OTA key index (KIC)')
+option_parser.add_argument('--kid', required=True, type=is_hexstr, help='OTA key (KID)')
+option_parser.add_argument('--kid-idx', default=1, type=int, help='OTA key index (KID)')
+option_parser.add_argument('--cntr', default=0, type=int, help='replay protection counter')
+option_parser.add_argument('--tar', required=True, type=is_hexstr, help='Toolkit Application Reference')
+option_parser.add_argument("--cntr-req", choices=CNTR_REQ.decmapping.values(), default='no_counter',
+                           help="Counter requirement")
+option_parser.add_argument('--no-ciphering', action='store_true', default=False, help='Disable ciphering')
+option_parser.add_argument("--rc-cc-ds", choices=RC_CC_DS.decmapping.values(), default='cc',
+                           help="message check (rc=redundency check, cc=crypt. checksum, ds=digital signature)")
+option_parser.add_argument('--por-in-submit', action='store_true', default=False,
+                           help='require PoR to be sent via SMS-SUBMIT')
+option_parser.add_argument('--por-no-ciphering', action='store_true', default=False, help='Disable ciphering (PoR)')
+option_parser.add_argument("--por-rc-cc-ds", choices=RC_CC_DS.decmapping.values(), default='cc',
+                           help="PoR check (rc=redundency check, cc=crypt. checksum, ds=digital signature)")
+option_parser.add_argument("--por-req", choices=POR_REQ.decmapping.values(), default='por_required',
+                           help="Proof of Receipt requirements")
+option_parser.add_argument('--src-addr', default='12', type=str, help='SMS source address (MSISDN)')
+option_parser.add_argument('--dest-addr', default='23', type=str, help='SMS destination address (MSISDN)')
+option_parser.add_argument('--timeout', default=10, type=int, help='Maximum response waiting time')
+option_parser.add_argument('-a', '--apdu', action='append', required=True, type=is_hexstr, help='C-APDU to send')
+
+class SmppHandler:
+    client = None
+
+    def __init__(self, host: str, port: int,
+                 system_id: str, password: str,
+                 ota_keyset: OtaKeyset, spi: dict, tar: bytes):
+        """
+        Initialize connection to SMPP server and set static OTA SMS-TPDU ciphering parameters
+        Args:
+                host : Hostname or IPv4/IPv6 address of the SMPP server
+                port : TCP Port of the SMPP server
+                system_id: SMPP System-ID used by ESME (client) to bind
+                password: SMPP Password used by ESME (client) to bind
+                ota_keyset: OTA keyset to be used for SMS-TPDU ciphering
+                spi: Security Parameter Indicator (SPI) to be used for SMS-TPDU ciphering
+                tar: Toolkit Application Reference (TAR) of the targeted card application
+        """
+
+        # Create and connect SMPP client
+        client = smpplib.client.Client(host, port, allow_unknown_opt_params=True)
+        client.set_message_sent_handler(self.message_sent_handler)
+        client.set_message_received_handler(self.message_received_handler)
+        client.connect()
+        client.bind_transceiver(system_id=system_id, password=password)
+        self.client = client
+
+        # Setup static OTA parameters
+        self.ota_dialect = OtaDialectSms()
+        self.ota_keyset = ota_keyset
+        self.tar = tar
+        self.spi = spi
+
+    def __del__(self):
+        if self.client:
+            self.client.unbind()
+            self.client.disconnect()
+
+    def message_received_handler(self, pdu):
+        if pdu.short_message:
+            logger.info("SMS-TPDU received: %s", b2h(pdu.short_message))
+            try:
+                dec = self.ota_dialect.decode_resp(self.ota_keyset, self.spi, pdu.short_message)
+            except ValueError:
+                # Retry to decoding with ciphering disabled (in case the card has problems to decode the SMS-TDPU
+                # we have sent, the response will contain an unencrypted error message)
+                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)
+            logger.info("SMS-TPDU decoded: %s", dec)
+            self.response = dec
+        return None
+
+    def message_sent_handler(self, pdu):
+            logger.debug("SMS-TPDU sent: pdu_sequence=%s pdu_message_id=%s", pdu.sequence, pdu.message_id)
+
+    def transceive_sms_tpdu(self, tpdu: bytes, src_addr: str, dest_addr: str, timeout: int) -> tuple:
+        """
+        Transceive SMS-TPDU. This method sends the SMS-TPDU to the SMPP server, and waits for a response. The method
+        returns when the response is received.
+
+        Args:
+                tpdu : short message content (plaintext)
+                src_addr : short message source address
+                dest_addr : short message destination address
+                timeout : timeout after which this method should give up waiting for a response
+        Returns:
+                tuple containing the response (plaintext)
+        """
+
+        logger.info("SMS-TPDU sending: %s...", b2h(tpdu))
+
+        self.client.send_message(
+            # TODO: add parameters to switch source_addr_ton and dest_addr_ton between SMPP_TON_INTL and SMPP_NPI_ISDN
+            source_addr_ton=smpplib.consts.SMPP_TON_INTL,
+            source_addr=src_addr,
+            dest_addr_ton=smpplib.consts.SMPP_TON_INTL,
+            destination_addr=dest_addr,
+            short_message=tpdu,
+            # TODO: add parameters to set data_coding and esm_class
+            data_coding=smpplib.consts.SMPP_ENCODING_BINARY,
+            esm_class=smpplib.consts.SMPP_GSMFEAT_UDHI,
+            protocol_id=0x7f,
+            # TODO: add parameter to use registered delivery
+            # registered_delivery=True,
+        )
+
+        logger.info("SMS-TPDU sent, waiting for response...")
+        timestamp_sent=int(time.time())
+        self.response = None
+        while self.response is None:
+            self.client.poll()
+            if int(time.time()) - timestamp_sent > timeout:
+                raise ValueError("Timeout reached, no response SMS-TPDU received!")
+        return self.response
+
+    def transceive_apdu(self, apdu: bytes, src_addr: str, dest_addr: str, timeout: int) -> tuple[bytes, bytes]:
+        """
+        Transceive APDU. This method wraps the given APDU into an SMS-TPDU, sends it to the SMPP server and waits for
+        the response. When the response is received, the last response data and the last status word is extracted from
+        the response and returned to the caller.
+
+        Args:
+                apdu : one or more concatenated APDUs
+                src_addr : short message source address
+                dest_addr : short message destination address
+                timeout : timeout after which this method should give up waiting for a response
+        Returns:
+                tuple containing the last response data and the last status word as byte strings
+        """
+
+        logger.info("C-APDU sending: %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
+        response =  self.transceive_sms_tpdu(tpdu, src_addr, dest_addr, timeout)
+
+        # Extract last_response_data and last_status_word from the response
+        sw = None
+        resp = None
+        for container in response:
+            if container:
+                container_dict = dict(container)
+                resp = container_dict.get('last_response_data')
+                sw = container_dict.get('last_status_word')
+        if resp is None:
+            raise ValueError("Response does not contain any last_response_data, no R-APDU received!")
+        if sw is None:
+            raise ValueError("Response does not contain any last_status_word, no R-APDU received!")
+
+        logger.info("R-APDU received: %s %s", resp, sw)
+        return h2b(resp), h2b(sw)
+
+if __name__ == '__main__':
+    opts = option_parser.parse_args()
+
+    logging.basicConfig(level=logging.DEBUG if opts.verbose else logging.INFO,
+                        format='%(asctime)s %(levelname)s %(message)s',
+                        datefmt='%Y-%m-%d %H:%M:%S')
+
+    if opts.kic_idx != opts.kid_idx:
+        logger.warning("KIC index (%s) and KID index (%s) are different (security violation, card should reject message)",
+                       opts.kic_idx, opts.kid_idx)
+
+    ota_keyset = OtaKeyset(algo_crypt=opts.algo_crypt,
+                           kic_idx=opts.kic_idx,
+                           kic=h2b(opts.kic),
+                           algo_auth=opts.algo_auth,
+                           kid_idx=opts.kid_idx,
+                           kid=h2b(opts.kid),
+                           cntr=opts.cntr)
+    spi = {'counter' : opts.cntr_req,
+           'ciphering' : not opts.no_ciphering,
+           'rc_cc_ds': opts.rc_cc_ds,
+           'por_in_submit': opts.por_in_submit,
+           'por_shall_be_ciphered': not opts.por_no_ciphering,
+           'por_rc_cc_ds': opts.por_rc_cc_ds,
+           'por': opts.por_req}
+    apdu = h2b("".join(opts.apdu))
+
+    smpp_handler = SmppHandler(opts.host, opts.port, opts.system_id, opts.password, ota_keyset, spi, h2b(opts.tar))
+    resp, sw = smpp_handler.transceive_apdu(apdu, opts.src_addr, opts.dest_addr, opts.timeout)
+    print("%s %s" % (b2h(resp), b2h(sw)))

docs/card-key-provider.rst

@@ -1,4 +1,4 @@
-Retrieving card-individual keys via CardKeyProvider
+Retrieving card-individual keys via CardKeyProvider
 ===================================================
 
 When working with a batch of cards, or more than one card in general, it
@@ -20,9 +20,11 @@ example develop your own CardKeyProvider that queries some kind of
 database for the key material, or that uses a key derivation function to
 derive card-specific key material from a global master key.
 
-The only actual CardKeyProvider implementation included in pySim is the
-`CardKeyProviderCsv` which retrieves the key material from a
-[potentially encrypted] CSV file.
+pySim already includes two CardKeyProvider implementations. One to retrieve
+key material from a CSV file (`CardKeyProviderCsv`) and a second one that allows
+to retrieve the key material from a PostgreSQL database (`CardKeyProviderPgsql`).
+Both implementations equally implement a column encryption scheme that allows
+to protect sensitive columns using a *transport key*
 
 
 The CardKeyProviderCsv
@@ -40,11 +42,224 @@ of pySim-shell.  If you do not specify a CSV file, pySim will attempt to
 open a CSV file from the default location at
 `~/.osmocom/pysim/card_data.csv`, and use that, if it exists.
 
+The `CardKeyProviderCsv` is suitable to manage small amounts of key material
+locally. However, if your card inventory is very large and the key material
+must be made available on multiple sites, the `CardKeyProviderPgsql` is the
+better option.
+
+
+The CardKeyProviderPgsql
+------------------------
+
+With the `CardKeyProviderPgsql` you can use a PostgreSQL database as storage
+medium. The implementation comes with a CSV importer tool that consumes the
+same CSV files you would normally use with the `CardKeyProviderCsv`, so you
+can just use your existing CSV files and import them into the database.
+
+
+Requirements
+^^^^^^^^^^^^
+
+The `CardKeyProviderPgsql` uses the `Psycopg` PostgreSQL database adapter
+(https://www.psycopg.org). `Psycopg` is not part of the default requirements
+of pySim-shell and must be installed separately. `Psycopg` is available as
+Python package under the name `psycopg2-binary`.
+
+
+Setting up the database
+^^^^^^^^^^^^^^^^^^^^^^^
+
+From the perspective of the database, the `CardKeyProviderPgsql` has only
+minimal requirements. You do not have to create any tables in advance. An empty
+database and at least one user that may create, alter and insert into tables is
+sufficient. However, for increased reliability and as a protection against
+incorrect operation, the `CardKeyProviderPgsql` supports a hierarchical model
+with three users (or roles):
+
+* **admin**:
+  This should be the owner of the database. It is intended to be used for
+  administrative tasks like adding new tables or adding new columns to existing
+  tables. This user should not be used to insert new data into tables or to access
+  data from within pySim-shell using the `CardKeyProviderPgsql`
+
+* **importer**:
+  This user is used when feeding new data into an existing table. It should only
+  be able to insert new rows into existing tables. It should not be used for
+  administrative tasks or to access data from within pySim-shell using the
+  `CardKeyProviderPgsql`
+
+* **reader**:
+  To access data from within pySim shell using the `CardKeyProviderPgsql` the
+  reader user is the correct one to use. This user should have no write access
+  to the database or any of the tables.
+
+
+Creating a config file
+^^^^^^^^^^^^^^^^^^^^^^
+
+The default location for the config file is `~/.osmocom/pysim/card_data_pgsql.cfg`
+The file uses `yaml` syntax and should look like the example below:
+
+::
+
+   host: "127.0.0.1"
+   db_name: "my_database"
+   table_names:
+   - "uicc_keys"
+   - "euicc_keys"
+   db_users:
+      admin:
+         name: "my_admin_user"
+         pass: "my_admin_password"
+      importer:
+         name: "my_importer_user"
+         pass: "my_importer_password"
+      reader:
+         name: "my_reader_user"
+         pass: "my_reader_password"
+
+This file is used by pySim-shell and by the importer tool. Both expect the file
+in the aforementioned location. In case you want to store the file in a
+different location you may use the `--pgsql` commandline option to provide a
+custom config file path.
+
+The hostname and the database name for the PostgreSQL database is set with the
+`host` and `db_name` fields. The field `db_users` sets the user names and
+passwords for each of the aforementioned users (or roles). In case only a single
+admin user is used, all three entries may be populated with the same user name
+and password (not recommended)
+
+The field `table_names` sets the tables that the `CardKeyProviderPgsql` shall
+use to query to locate card key data. You can set up as many tables as you
+want, `CardKeyProviderPgsql` will query them in order, one by one until a
+matching entry is found.
+
+NOTE: In case you do not want to disclose the admin and the importer credentials
+to pySim-shell you may remove those lines. pySim-shell will only require the
+`reader` entry under `db_users`.
+
+
+Using the Importer
+^^^^^^^^^^^^^^^^^^
+
+Before data can be imported, you must first create a database table. Tables
+are created with the provided importer tool, which can be found under
+`contrib/csv-to-pgsql.py`. This tool is used to create the database table and
+read the data from the provided CSV file into the database.
+
+As mentioned before, all CSV file formats that work with `CardKeyProviderCsv`
+may be used. To demonstrate how the import process works, let's assume you want
+to import a CSV file format that looks like the following example. Let's also
+assume that you didn't get the Global Platform keys from your card vendor for
+this batch of UICC cards, so your CSV file lacks the columns for those fields.
+
+::
+
+   "id","imsi","iccid","acc","pin1","puk1","pin2","puk2","ki","opc","adm1"
+   "card1","999700000000001","8900000000000000001","0001","1111","11111111","0101","01010101","11111111111111111111111111111111","11111111111111111111111111111111","11111111"
+   "card2","999700000000002","8900000000000000002","0002","2222","22222222","0202","02020202","22222222222222222222222222222222","22222222222222222222222222222222","22222222"
+   "card3","999700000000003","8900000000000000003","0003","3333","22222222","0303","03030303","33333333333333333333333333333333","33333333333333333333333333333333","33333333"
+
+Since this is your first import, the database still lacks the table. To
+instruct the importer to create a new table, you may use the `--create-table`
+option. You also have to pick an appropriate name for the table. Any name may
+be chosen as long as it contains the string `uicc_keys` or `euicc_keys`,
+depending on the type of data (`UICC` or `eUICC`) you intend to store in the
+table. The creation of the table is an administrative task and can only be done
+with the `admin` user. The `admin` user is selected using the `--admin` switch.
+
+::
+
+   $ PYTHONPATH=../ ./csv-to-pgsql.py --csv ./csv-to-pgsql_example_01.csv --table-name uicc_keys --create-table --admin
+   INFO: CSV file: ./csv-to-pgsql_example_01.csv
+   INFO: CSV file columns: ['ID', 'IMSI', 'ICCID', 'ACC', 'PIN1', 'PUK1', 'PIN2', 'PUK2', 'KI', 'OPC', 'ADM1']
+   INFO: Using config file: /home/user/.osmocom/pysim/card_data_pgsql.cfg
+   INFO: Database host: 127.0.0.1
+   INFO: Database name: my_database
+   INFO: Database user: my_admin_user
+   INFO: New database table created: uicc_keys
+   INFO: Database table: uicc_keys
+   INFO: Database table columns: ['ICCID', 'IMSI']
+   INFO: Adding missing columns: ['PIN2', 'PUK1', 'PUK2', 'ACC', 'ID', 'PIN1', 'ADM1', 'KI', 'OPC']
+   INFO: Changes to table uicc_keys committed!
+
+The importer has created a new table with the name `uicc_keys`. The table is
+now ready to be filled with data.
+
+::
+
+   $ PYTHONPATH=../ ./csv-to-pgsql.py --csv ./csv-to-pgsql_example_01.csv --table-name uicc_keys
+   INFO: CSV file: ./csv-to-pgsql_example_01.csv
+   INFO: CSV file columns: ['ID', 'IMSI', 'ICCID', 'ACC', 'PIN1', 'PUK1', 'PIN2', 'PUK2', 'KI', 'OPC', 'ADM1']
+   INFO: Using config file: /home/user/.osmocom/pysim/card_data_pgsql.cfg
+   INFO: Database host: 127.0.0.1
+   INFO: Database name: my_database
+   INFO: Database user: my_importer_user
+   INFO: Database table: uicc_keys
+   INFO: Database table columns: ['ICCID', 'IMSI', 'PIN2', 'PUK1', 'PUK2', 'ACC', 'ID', 'PIN1', 'ADM1', 'KI', 'OPC']
+   INFO: CSV file import done, 3 rows imported
+   INFO: Changes to table uicc_keys committed!
+
+A quick `SELECT * FROM uicc_keys;` at the PostgreSQL console should now display
+the contents of the CSV file you have fed into the importer.
+
+Let's now assume that with your next batch of UICC cards your vendor includes
+the Global Platform keys so your CSV format changes. It may now look like this:
+
+::
+
+   "id","imsi","iccid","acc","pin1","puk1","pin2","puk2","ki","opc","adm1","scp02_dek_1","scp02_enc_1","scp02_mac_1"
+   "card4","999700000000004","8900000000000000004","0004","4444","44444444","0404","04040404","44444444444444444444444444444444","44444444444444444444444444444444","44444444","44444444444444444444444444444444","44444444444444444444444444444444","44444444444444444444444444444444"
+   "card5","999700000000005","8900000000000000005","0005","4444","55555555","0505","05050505","55555555555555555555555555555555","55555555555555555555555555555555","55555555","55555555555555555555555555555555","55555555555555555555555555555555","55555555555555555555555555555555"
+   "card6","999700000000006","8900000000000000006","0006","4444","66666666","0606","06060606","66666666666666666666666666666666","66666666666666666666666666666666","66666666","66666666666666666666666666666666","66666666666666666666666666666666","66666666666666666666666666666666"
+
+When importing data from an updated CSV format the database table also has
+to be updated. This is done using the `--update-columns` switch. Like when
+creating new tables, this operation also requires admin privileges, so the
+`--admin` switch is required again.
+
+::
+
+   $ PYTHONPATH=../ ./csv-to-pgsql.py --csv ./csv-to-pgsql_example_02.csv --table-name uicc_keys --update-columns --admin
+   INFO: CSV file: ./csv-to-pgsql_example_02.csv
+   INFO: CSV file columns: ['ID', 'IMSI', 'ICCID', 'ACC', 'PIN1', 'PUK1', 'PIN2', 'PUK2', 'KI', 'OPC', 'ADM1', 'SCP02_DEK_1', 'SCP02_ENC_1', 'SCP02_MAC_1']
+   INFO: Using config file: /home/user/.osmocom/pysim/card_data_pgsql.cfg
+   INFO: Database host: 127.0.0.1
+   INFO: Database name: my_database
+   INFO: Database user: my_admin_user
+   INFO: Database table: uicc_keys
+   INFO: Database table columns: ['ICCID', 'IMSI', 'PIN2', 'PUK1', 'PUK2', 'ACC', 'ID', 'PIN1', 'ADM1', 'KI', 'OPC']
+   INFO: Adding missing columns: ['SCP02_ENC_1', 'SCP02_MAC_1', 'SCP02_DEK_1']
+   INFO: Changes to table uicc_keys committed!
+
+When the new table columns are added, the import may be continued like the
+first one:
+
+::
+
+   $ PYTHONPATH=../ ./csv-to-pgsql.py --csv ./csv-to-pgsql_example_02.csv --table-name uicc_keys
+   INFO: CSV file: ./csv-to-pgsql_example_02.csv
+   INFO: CSV file columns: ['ID', 'IMSI', 'ICCID', 'ACC', 'PIN1', 'PUK1', 'PIN2', 'PUK2', 'KI', 'OPC', 'ADM1', 'SCP02_DEK_1', 'SCP02_ENC_1', 'SCP02_MAC_1']
+   INFO: Using config file: /home/user/.osmocom/pysim/card_data_pgsql.cfg
+   INFO: Database host: 127.0.0.1
+   INFO: Database name: my_database
+   INFO: Database user: my_importer_user
+   INFO: Database table: uicc_keys
+   INFO: Database table columns: ['ICCID', 'IMSI', 'PIN2', 'PUK1', 'PUK2', 'ACC', 'ID', 'PIN1', 'ADM1', 'KI', 'OPC', 'SCP02_ENC_1', 'SCP02_MAC_1', 'SCP02_DEK_1']
+   INFO: CSV file import done, 3 rows imported
+   INFO: Changes to table uicc_keys committed!
+
+On the PostgreSQL console a `SELECT * FROM uicc_keys;` should now show the
+imported data with the added columns. All important data should now also be
+available from within pySim-shell via the `CardKeyProviderPgsql`.
+
+
 Column-Level CSV encryption
-~~~~~~~~~~~~~~~~~~~~~~~~~~~
+---------------------------
 
 pySim supports column-level CSV encryption.  This feature will make sure
-that your key material is not stored in plaintext in the CSV file.
+that your key material is not stored in plaintext in the CSV file (or
+database).
 
 The encryption mechanism uses AES in CBC mode.  You can use any key
 length permitted by AES (128/192/256 bit).
@@ -72,6 +287,8 @@ by all columns of the set:
 * `SCP03_ISDA` is a group alias for `SCP03_ENC_ISDA`, `SCP03_MAC_ISDA`, `SCP03_DEK_ISDA`
 * `SCP03_ISDR` is a group alias for `SCP03_ENC_ISDR`, `SCP03_MAC_ISDR`, `SCP03_DEK_ISDR`
 
+NOTE: When using `CardKeyProviderPqsl`, the input CSV files must be encrypted
+before import.
 
 Field naming
 ------------
@@ -82,9 +299,9 @@ Field naming
 * For look-up of eUICC specific key material (like SCP03 keys for the
   ISD-R, ECASD), pySim uses the `EID` field as lookup key.
 
-As soon as the CardKeyProviderCsv finds a line (row) in your CSV where
-the ICCID or EID match, it looks for the column containing the requested
-data.
+As soon as the CardKeyProvider finds a line (row) in your CSV file
+(or database) where the ICCID or EID match, it looks for the column containing
+the requested data.
 
 
 ADM PIN

docs/index.rst

@@ -48,6 +48,7 @@ pySim consists of several parts:
    sim-rest
    suci-keytool
    saip-tool
+   smpp-ota-tool
 
 
 Indices and tables

docs/smpp-ota-tool.rst

@@ -0,0 +1,179 @@
+smpp-ota-tool
+=============
+
+The `smpp-ota-tool` allows users to send OTA SMS messages containing APDU scripts (RFM, RAM) via an SMPP server. The
+intended audience are developers who want to test/evaluate the OTA SMS interface of a SIM/UICC/eUICC. `smpp-ota-tool`
+is intended to be used as a companion tool for :ref:`pySim-smpp2sim`, however it should be usable on any other SMPP
+server (such as a production SMSC of a live cellular network) as well.
+
+From the technical perspective `smpp-ota-tool` takes the role of an SMPP ESME. It takes care of the encoding, encryption
+and checksumming (signing) of the RFM/RAM OTA SMS and eventually submits it to the SMPP server. The program then waits
+for a response. The response is automatically parsed and printed on stdout. This makes the program also suitable to be
+called from shell scripts.
+
+.. note:: In the following we will we will refer to `SIM` as one of the following: `SIM`, `USIM`, `ISIM`, `UICC`,
+	  `eUICC`, `eSIM`.
+
+Applying OTA keys
+~~~~~~~~~~~~~~~~~
+
+Depending on the `SIM` type you will receive one or more sets of keys which you can use to communicate with the `SIM`
+through a secure channel protocol. When using the OTA SMS method, the SCP80 protocol is used and it therefore crucial
+to use a keyset that is actually suitable for SCP80.
+
+A keyset usually consists of three keys:
+
+#. KIC: the key used for ciphering (encryption/decryption)
+#. KID: the key used to compute a cryptographic checksum (signing)
+#. KIK: the key used to encrypt/decrypt key material (key rotation, adding of new keys)
+
+From the transport security perspective, only KIC and KID are relevant. The KIK (also referenced as "Data Encryption
+Key", DEK) is only used when keys are rotated or new keys are added (see also ETSI TS 102 226, section 8.2.1.5).
+
+When the keyset is programmed into the security domain of the `SIM`, it is tied to a specific cryptographic algorithm
+(3DES, AES128 or AES256) and a so called Key Version Number (KVN). The term "Key Version Number" is misleading, since
+it is actually not a version number. It is a unique identifier of a certain keyset which also identifies for which
+secure channel protocol the keyset may be used. Keysets with a KVN from 1-15 (``0x01``-``0x0F``) are suitable for SCP80.
+This means that it is not only important to know just the KIC/KID/KIK keys. Also the related algorithms and the KVN
+numbers must be known.
+
+.. note:: SCP80 keysets typically start counting from 1 upwards. Typical configurations use a set of 3 keysets with
+	  KVN numbers 1-3.
+
+Addressing an Application
+~~~~~~~~~~~~~~~~~~~~~~~~~
+
+When communicating with a specific application on a `SIM` via SCP80, it is important to address that application with
+the correct parameters. The following two parameters must be known in advance:
+
+#. TAR: The Toolkit Application Reference (TAR) number is a three byte value that uniquely addresses an application
+   on the `SIM`. The exact values may vary (see also ETSI TS 101 220, Table D.1).
+#. MSL: The Minimum Security Level (MSL) is a bit-field that dictates which of the security measures encoded in the
+   SPI are mandatory (see also ETSI TS 102 225, section 5.1.1).
+
+A practical example
+~~~~~~~~~~~~~~~~~~~
+
+.. note:: This tutorial assumes that pySim-smpp2sim is running on the local machine with its default parameters.
+	  See also :ref:`pySim-smpp2sim`.
+
+Let's assume that an OTA SMS shall be sent to the SIM RFM application of an sysmoISIM-SJA2. What we want to do is to
+select DF.GSM and to get the select response back.
+
+We have received the following key material from the `SIM` vendor:
+
+::
+
+   KIC1: F09C43EE1A0391665CC9F05AF4E0BD10
+   KID1: 01981F4A20999F62AF99988007BAF6CA
+   KIK1: 8F8AEE5CDCC5D361368BC45673D99195
+   KIC2: 01022916E945B656FDE03F806A105FA2
+   KID2: D326CB69F160333CC5BD1495D448EFD6
+   KIK2: 08037E0590DFE049D4975FFB8652F625
+   KIC3: 2B22824D0D27A3A1CEEC512B312082B4
+   KID3: F1697766925A11F4458295590137B672
+   KIK3: C7EE69B2C5A1C8E160DD36A38EB517B3
+
+Those are three keysets. The enumeration is directly equal to the KVN used. All three keysets are 3DES keys, which
+means triple_des_cbc2 is the correct algorithm to use.
+
+.. note:: The key set configuration can be confirmed by retrieving the key configuration using
+	  `get_data key_information` from within an SCP02 session on ADF.ISD.
+
+In this example we intend to address the SIM RFM application on the `SIM`. Which according to the manual has TAR ``B00010``
+and MSL ``0x06``. When we hold ``0x06`` = ``0b00000110`` against the SPI coding chart (see also ETSI TS 102 225,
+section 5.1.1). We can deduct that Ciphering and Cryptographic Checksum are mandatory.
+
+.. note:: The MSL (see also ETSI TS 102 226, section 6.1) is assigned to an application by the `SIM` issuer. It is a
+	  custom decision and may vary with different `SIM` types/profiles. In the case of sysmoISIM-SJS1/SJA2/SJA5 the
+	  counter requirement has been waived to simplify lab/research type use. In productive environments, `SIM`
+	  applications should ideally use an MSL that makes the counter mandatory.
+
+In order to select DF.GSM (``0x7F20``) and to retrieve the select response, two APDUs are needed. The first APDU is the
+select command ``A0A40000027F20`` and the second is the related get-response command ``A0C0000016``. Those APDUs will be
+concatenated and are sent in a single message. The message containing the concatenated APDUs works as a script that
+is received by the SIM RFM application and then executed. This method poses some limitations that have to be taken into
+account when making requests like this (see also ETSI TS 102 226, section 5).
+
+With this information we may now construct a commandline for `smpp-ota-tool.py`. We will pass the KVN as kid_idx and
+kic_idx (see also ETSI TS 102 225, Table 2, fields `KIc` and `KID`). Both index values should refer to the same
+keyset/KVN as keysets should not be mixed. (`smpp-ota-tool` still provides separate parameters anyway to allow testing
+with invalid keyset combinations)
+
+::
+
+   $ PYTHONPATH=./ ./contrib/smpp-ota-tool.py --kic F09C43EE1A0391665CC9F05AF4E0BD10 --kid 01981F4A20999F62AF99988107BAF6CA --kid_idx 1 --kic_idx 1 --algo-crypt triple_des_cbc2 --algo-auth triple_des_cbc2 --tar B00010 --apdu A0A40000027F20 --apdu A0C0000016
+   2026-02-26 17:13:56 INFO Connecting to localhost:2775...
+   2026-02-26 17:13:56 INFO C-APDU sending: a0a40000027f20a0c0000016...
+   2026-02-26 17:13:56 INFO SMS-TPDU sending: 02700000281506191515b00010da1d6cbbd0d11ce4330d844c7408340943e843f67a6d7b0674730881605fd62d...
+   2026-02-26 17:13:56 INFO SMS-TPDU sent, waiting for response...
+   2026-02-26 17:13:56 INFO SMS-TPDU received: 027100002c12b000107ddf58d1780f771638b3975759f4296cf5c31efc87a16a1b61921426baa16da1b5ba1a9951d59a39
+   2026-02-26 17:13:56 INFO SMS-TPDU decoded: (Container(rpl=44, rhl=18, tar=b'\xb0\x00\x10', cntr=b'\x00\x00\x00\x00\x00', pcntr=0, response_status=uEnumIntegerString.new(0, 'por_ok'), cc_rc=b'\x8f\xea\xf5.\xf4\x0e\xc2\x14', secured_data=b'\x02\x90\x00\x00\x00\xff\xff\x7f \x02\x00\x00\x00\x00\x00\t\xb1\x065\x04\x00\x83\x8a\x83\x8a'), Container(number_of_commands=2, last_status_word=u'9000', last_response_data=u'0000ffff7f2002000000000009b106350400838a838a'))
+   2026-02-26 17:13:56 INFO R-APDU received: 0000ffff7f2002000000000009b106350400838a838a 9000
+   0000ffff7f2002000000000009b106350400838a838a 9000
+   2026-02-26 17:13:56 INFO Disconnecting...
+
+The result we see is the select response of DF.GSM and a status word indicating that the last command has been
+processed normally.
+
+As we can see, this mechanism now allows us to perform small administrative tasks remotely. We can read the contents of
+files remotely or make changes to files. Depending on the changes we make, there may be security issues arising from
+replay attacks. With the commandline above, the communication is encrypted and protected by a cryptographic checksum,
+so an adversary can neither read, nor alter the message. However, an adversary could still replay an intercepted
+message and the `SIM` would happily execute the contained APDUs again.
+
+To prevent this, we may include a replay protection counter within the message. In this case, the MSL indicates that a
+replay protection counter is not required. However, to extended the security of our messages, we may chose to use a
+counter anyway. In the following example, we will encode a counter value of 100. We will instruct the `SIM` to make sure
+that the value we send is higher than the counter value that is currently stored in the `SIM`.
+
+To add a replay connection counter we add the commandline arguments `--cntr-req` to set the counter requirement and
+`--cntr` to pass the counter value.
+
+::
+
+   $ PYTHONPATH=./ ./contrib/smpp-ota-tool.py --kic F09C43EE1A0391665CC9F05AF4E0BD10 --kid 01981F4A20999F62AF99988107BAF6CA --kid_idx 1 --kic_idx 1 --algo-crypt triple_des_cbc2 --algo-auth triple_des_cbc2 --tar B00010 --apdu A0A40000027F20 --apdu A0C0000016 --cntr-req counter_must_be_higher --cntr 100
+   2026-02-26 17:16:39 INFO Connecting to localhost:2775...
+   2026-02-26 17:16:39 INFO C-APDU sending: a0a40000027f20a0c0000016...
+   2026-02-26 17:16:39 INFO SMS-TPDU sending: 02700000281516191515b000103a4f599e94f2b5dcfbbda984761b7977df6514c57a580fb4844787c436d2eade...
+   2026-02-26 17:16:39 INFO SMS-TPDU sent, waiting for response...
+   2026-02-26 17:16:39 INFO SMS-TPDU received: 027100002c12b0001049fb0315f6c6401b553867f412cefaf9355b38271178edb342a3bc9cc7e670cdc1f45eea6ffcbb39
+   2026-02-26 17:16:39 INFO SMS-TPDU decoded: (Container(rpl=44, rhl=18, tar=b'\xb0\x00\x10', cntr=b'\x00\x00\x00\x00d', pcntr=0, response_status=uEnumIntegerString.new(0, 'por_ok'), cc_rc=b'\xa9/\xc7\xc9\x00"\xab5', secured_data=b'\x02\x90\x00\x00\x00\xff\xff\x7f \x02\x00\x00\x00\x00\x00\t\xb1\x065\x04\x00\x83\x8a\x83\x8a'), Container(number_of_commands=2, last_status_word=u'9000', last_response_data=u'0000ffff7f2002000000000009b106350400838a838a'))
+   2026-02-26 17:16:39 INFO R-APDU received: 0000ffff7f2002000000000009b106350400838a838a 9000
+   0000ffff7f2002000000000009b106350400838a838a 9000
+   2026-02-26 17:16:39 INFO Disconnecting...
+
+The `SIM` has accepted the message. The message got processed and the `SIM` has set its internal to 100. As an experiment,
+we may try to re-use the counter value:
+
+::
+
+   $ PYTHONPATH=./ ./contrib/smpp-ota-tool.py --kic F09C43EE1A0391665CC9F05AF4E0BD10 --kid 01981F4A20999F62AF99988107BAF6CA --kid_idx 1 --kic_idx 1 --algo-crypt triple_des_cbc2 --algo-auth triple_des_cbc2 --tar B00010 --apdu A0A40000027F20 --apdu A0C0000016 --cntr-req counter_must_be_higher --cntr 100
+   2026-02-26 17:16:43 INFO Connecting to localhost:2775...
+   2026-02-26 17:16:43 INFO C-APDU sending: a0a40000027f20a0c0000016...
+   2026-02-26 17:16:43 INFO SMS-TPDU sending: 02700000281516191515b000103a4f599e94f2b5dcfbbda984761b7977df6514c57a580fb4844787c436d2eade...
+   2026-02-26 17:16:43 INFO SMS-TPDU sent, waiting for response...
+   2026-02-26 17:16:43 INFO SMS-TPDU received: 027100000b0ab0001000000000000006
+   2026-02-26 17:16:43 INFO SMS-TPDU decoded: (Container(rpl=11, rhl=10, tar=b'\xb0\x00\x10', cntr=b'\x00\x00\x00\x00\x00', pcntr=0, response_status=uEnumIntegerString.new(6, 'undefined_security_error'), cc_rc=b'', secured_data=b''), None)
+   Traceback (most recent call last):
+   File "/home/user/work/git_master/pysim/./contrib/smpp-ota-tool.py", line 238, in <module>
+    resp, sw = smpp_handler.transceive_apdu(apdu, opts.src_addr, opts.dest_addr, opts.timeout)
+               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+   File "/home/user/work/git_master/pysim/./contrib/smpp-ota-tool.py", line 162, in transceive_apdu
+     raise ValueError("Response does not contain any last_response_data, no R-APDU received!")
+   ValueError: Response does not contain any last_response_data, no R-APDU received!
+   2026-02-26 17:16:43 INFO Disconnecting...
+
+As we can see, the `SIM` has rejected the message with an `undefined_security_error`. The replay-protection-counter
+ensures that a message can only be sent once.
+
+.. note:: The replay-protection-counter is implemented as a 5 byte integer value (see also ETSI TS 102 225, Table 3).
+	  When the counter has reached its maximum, it will not overflow nor can it be reset.
+
+smpp-ota-tool syntax
+~~~~~~~~~~~~~~~~~~~~
+
+.. argparse::
+   :module: contrib.smpp-ota-tool
+   :func: option_parser
+   :prog: contrib/smpp-ota-tool.py

docs/smpp2sim.rst

@@ -55,3 +55,5 @@ And once your external program is sending SMS to the simulated SMSC, it will log
   SMSPPDownload(DeviceIdentities({'source_dev_id': 'network', 'dest_dev_id': 'uicc'}),Address({'ton_npi': 0, 'call_number': '0123456'}),SMS_TPDU({'tpdu': '400290217ff6227052000000002d02700000281516191212b0000127fa28a5bac69d3c5e9df2c7155dfdde449c826b236215566530787b30e8be5d'}))
   INFO     root: ENVELOPE: d147820283818604001032548b3b400290217ff6227052000000002d02700000281516191212b0000127fa28a5bac69d3c5e9df2c7155dfdde449c826b236215566530787b30e8be5d
   INFO     root: SW 9000: 027100002412b000019a551bb7c28183652de0ace6170d0e563c5e949a3ba56747fe4c1dbbef16642c
+
+.. note:: for sending OTA SMS messages :ref:`smpp-ota-tool` may be used.

pySim-read.py

@@ -44,6 +44,7 @@ from pySim.exceptions import SwMatchError
 from pySim.legacy.cards import card_detect, SimCard, UsimCard, IsimCard
 from pySim.utils import dec_imsi, dec_iccid
 from pySim.legacy.utils import format_xplmn_w_act, dec_st, dec_msisdn
+from pySim.ts_51_011 import EF_SMSP
 
 option_parser = argparse.ArgumentParser(description='Legacy tool for reading some parts of a SIM card',
                                         formatter_class=argparse.ArgumentDefaultsHelpFormatter)
@@ -141,6 +142,15 @@ if __name__ == '__main__':
     (res, sw) = card.read_record('SMSP', 1)
     if sw == '9000':
         print("SMSP: %s" % (res,))
+        ef_smsp = EF_SMSP()
+        smsc_a = ef_smsp.decode_record_bin(h2b(res), 1).get('tp_sc_addr', {})
+        smsc_n = smsc_a.get('call_number', None)
+        if smsc_a.get('ton_npi', {}).get('type_of_number', None) == 'international' and smsc_n is not None:
+            smsc = '+' + smsc_n
+        else:
+            smsc = smsc_n
+        if smsc is not None:
+            print("SMSC: %s" % (smsc,))
     else:
         print("SMSP: Can't read, response code = %s" % (sw,))
 

pySim-shell.py

@@ -69,12 +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
+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.app import init_card
 
-log = PySimLogger.get("main")
+log = PySimLogger.get(Path(__file__).stem)
 
 class Cmd2Compat(cmd2.Cmd):
     """Backwards-compatibility wrapper around cmd2.Cmd to support older and newer
@@ -519,8 +519,17 @@ Online manual available at https://downloads.osmocom.org/docs/pysim/master/html/
     @cmd2.with_category(CUSTOM_CATEGORY)
     def do_version(self, opts):
         """Print the pySim software version."""
-        import pkg_resources
-        self.poutput(pkg_resources.get_distribution('pySim'))
+        from importlib.metadata import version as vsn
+        self.poutput("pyosmocom " + vsn('pyosmocom'))
+        import os
+        cwd = os.path.dirname(os.path.realpath(__file__))
+        if os.path.isdir(os.path.join(cwd, ".git")):
+            import subprocess
+            url = subprocess.check_output(['git', 'config', '--get', 'remote.origin.url']).decode('ascii').strip()
+            version = subprocess.check_output(['git', 'rev-parse', 'HEAD'], cwd=cwd).decode('ascii').strip()
+            self.poutput(os.path.basename(url) + " " + version)
+        else:
+            self.poutput("pySim " + vsn('pySim'))
 
 @with_default_category('pySim Commands')
 class PySimCommands(CommandSet):
@@ -1138,8 +1147,11 @@ global_group.add_argument("--verbose", help="Enable verbose logging",
 
 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",
+                            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',
@@ -1164,7 +1176,7 @@ if __name__ == '__main__':
 
     # Ensure that we are able to print formatted warnings from the beginning.
     PySimLogger.setup(print, {logging.WARN: YELLOW})
-    if (opts.verbose):
+    if opts.verbose:
         PySimLogger.set_verbose(True)
         PySimLogger.set_level(logging.DEBUG)
     else:
@@ -1177,8 +1189,10 @@ if __name__ == '__main__':
     for par in opts.column_key:
         name, key = par.split(':')
         column_keys[name] = key
-    if os.path.isfile(opts.csv):
-        card_key_provider_register(CardKeyProviderCsv(opts.csv, column_keys))
+    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 reader driver
     sl = init_reader(opts, proactive_handler = Proact())

pySim-trace.py

@@ -23,6 +23,7 @@ from pySim.apdu_source.gsmtap import GsmtapApduSource
 from pySim.apdu_source.pyshark_rspro import PysharkRsproPcap, PysharkRsproLive
 from pySim.apdu_source.pyshark_gsmtap import PysharkGsmtapPcap
 from pySim.apdu_source.tca_loader_log import TcaLoaderLogApduSource
+from pySim.apdu_source.stdin_hex import StdinHexApduSource
 
 from pySim.apdu.ts_102_221 import UiccSelect, UiccStatus
 
@@ -190,6 +191,10 @@ parser_tcaloader_log = subparsers.add_parser('tca-loader-log', help="""
 parser_tcaloader_log.add_argument('-f', '--log-file', required=True,
                                   help='Name of the log file to be read')
 
+parser_stdin_hex = subparsers.add_parser('stdin-hex', help="""
+    Read APDUs as hex-string from stdin.""")
+
+
 if __name__ == '__main__':
 
     opts = option_parser.parse_args()
@@ -205,6 +210,8 @@ if __name__ == '__main__':
         s = PysharkGsmtapPcap(opts.pcap_file)
     elif opts.source == 'tca-loader-log':
         s = TcaLoaderLogApduSource(opts.log_file)
+    elif opts.source == 'stdin-hex':
+        s = StdinHexApduSource()
     else:
         raise ValueError("unsupported source %s", opts.source)
 

pySim/apdu_source/pyshark_gsmtap.py

@@ -84,5 +84,5 @@ class PysharkGsmtapPcap(_PysharkGsmtap):
         Args:
             pcap_filename: File name of the pcap file to be opened
         """
-        pyshark_inst = pyshark.FileCapture(pcap_filename, display_filter='gsm_sim', use_json=True, keep_packets=False)
+        pyshark_inst = pyshark.FileCapture(pcap_filename, display_filter='gsm_sim || iso7816.atr', use_json=True, keep_packets=False)
         super().__init__(pyshark_inst)

pySim/apdu_source/stdin_hex.py

@@ -0,0 +1,39 @@
+# coding=utf-8
+
+# (C) 2024 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 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/>.
+
+
+from pySim.utils import h2b
+
+from pySim.apdu.ts_102_221 import ApduCommands as UiccApduCommands
+from pySim.apdu.ts_102_222 import ApduCommands as UiccAdmApduCommands
+from pySim.apdu.ts_31_102 import ApduCommands as UsimApduCommands
+from pySim.apdu.global_platform import ApduCommands as GpApduCommands
+
+from . import ApduSource, PacketType, CardReset
+
+ApduCommands = UiccApduCommands + UiccAdmApduCommands + UsimApduCommands + GpApduCommands
+
+class StdinHexApduSource(ApduSource):
+    """ApduSource for reading apdu hex-strings from stdin."""
+
+    def read_packet(self) -> PacketType:
+        while True:
+            command = input("C-APDU >")
+            if len(command) == 0:
+                continue
+            response = '9000'
+            return ApduCommands.parse_cmd_bytes(h2b(command) + h2b(response))

pySim/card_handler.py

@@ -7,7 +7,7 @@ there are also automatic card feeders.
 """
 
 #
-# (C) 2019 by Sysmocom s.f.m.c. GmbH
+# (C) 2019 by sysmocom - s.f.m.c. GmbH
 # All Rights Reserved
 #
 # This program is free software: you can redistribute it and/or modify

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-2025 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
@@ -36,8 +36,9 @@ from pySim.log import PySimLogger
 import abc
 import csv
 import logging
+import yaml
 
-log = PySimLogger.get("CARDKEY")
+log = PySimLogger.get(__name__)
 
 card_key_providers = []  # type: List['CardKeyProvider']
 
@@ -57,7 +58,7 @@ class CardKeyFieldCryptor:
             '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'],
-            'SCP03_ISDA': ['SCP03_ENC_ISDR', 'SCP03_MAC_ISDA', 'SCP03_DEK_ISDA'],
+            'SCP03_ISDA': ['SCP03_ENC_ISDA', 'SCP03_MAC_ISDA', 'SCP03_DEK_ISDA'],
             'SCP03_ECASD': ['SCP03_ENC_ECASD', 'SCP03_MAC_ECASD', 'SCP03_DEK_ECASD'],
     }
 
@@ -159,6 +160,7 @@ class CardKeyProviderCsv(CardKeyProvider):
                 csv_filename : file name (path) of CSV file containing card-individual key/data
                 transport_keys : (see class CardKeyFieldCryptor)
         """
+        log.info("Using CSV file as card key data source: %s" % csv_filename)
         self.csv_file = open(csv_filename, 'r')
         if not self.csv_file:
             raise RuntimeError("Could not open CSV file '%s'" % csv_filename)
@@ -186,6 +188,69 @@ class CardKeyProviderCsv(CardKeyProvider):
             return None
         return return_dict
 
+class CardKeyProviderPgsql(CardKeyProvider):
+    """Card key provider implementation that allows to query against a specified PostgreSQL database table."""
+
+    def __init__(self, config_filename: str, transport_keys: dict):
+        """
+        Args:
+                config_filename : file name (path) of CSV file containing card-individual key/data
+                transport_keys : (see class CardKeyFieldCryptor)
+        """
+        import psycopg2
+        log.info("Using SQL database as card key data source: %s" % config_filename)
+        with open(config_filename, "r") as cfg:
+            config = yaml.load(cfg, Loader=yaml.FullLoader)
+            log.info("Card key database name: %s" % config.get('db_name'))
+            db_users = config.get('db_users')
+            user = db_users.get('reader')
+            if user is None:
+                raise ValueError("user for role 'reader' not set up in config file.")
+            self.conn = psycopg2.connect(dbname=config.get('db_name'),
+                                         user=user.get('name'),
+                                         password=user.get('pass'),
+                                         host=config.get('host'))
+            self.tables = config.get('table_names')
+            log.info("Card key database tables: %s" % str(self.tables))
+            self.crypt = CardKeyFieldCryptor(transport_keys)
+
+    def get(self, fields: List[str], key: str, value: str) -> Dict[str, str]:
+        import psycopg2
+        from psycopg2.sql import Identifier, SQL
+        db_result = None
+        for t in self.tables:
+            self.conn.rollback()
+            cur = self.conn.cursor()
+
+            # Make sure that the database table and the key column actually exists. If not, move on to the next table
+            cur.execute("SELECT column_name FROM information_schema.columns where table_name = %s;", (t,))
+            cols_result = cur.fetchall()
+            if cols_result == []:
+                log.warning("Card Key database seems to lack table %s, check config file!" % t)
+                continue
+            if (key.lower(),) not in cols_result:
+                continue
+
+            # Query requested columns from database table
+            query = SQL("SELECT {}").format(Identifier(fields[0].lower()))
+            for f in fields[1:]:
+                query += SQL(", {}").format(Identifier(f.lower()))
+            query += SQL(" FROM {} WHERE {} = %s LIMIT 1;").format(Identifier(t.lower()),
+                                                                  Identifier(key.lower()))
+            cur.execute(query, (value,))
+            db_result = cur.fetchone()
+            cur.close()
+
+            if db_result:
+                break
+
+        if db_result is None:
+            return None
+        result = dict(zip(fields, db_result))
+
+        for k in result.keys():
+            result[k] = self.crypt.decrypt_field(k, result.get(k))
+        return result
 
 
 def card_key_provider_register(provider: CardKeyProvider, provider_list=card_key_providers):

pySim/esim/__init__.py

@@ -54,6 +54,8 @@ def compile_asn1_subdir(subdir_name:str, codec='der'):
     __ver = sys.version_info
     if (__ver.major, __ver.minor) >= (3, 9):
         for i in resources.files('pySim.esim').joinpath('asn1').joinpath(subdir_name).iterdir():
+            if not i.name.endswith('.asn'):
+                continue
             asn_txt += i.read_text()
             asn_txt += "\n"
     #else:

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
@@ -27,7 +33,7 @@ logger.setLevel(logging.DEBUG)
 
 class param:
     class Iccid(ApiParamString):
-        """String representation of 19 or 20 digits, where the 20th digit MAY optionally be the padding
+        """String representation of 18 to 20 digits, where the 20th digit MAY optionally be the padding
         character F."""
         @classmethod
         def _encode(cls, data):
@@ -40,7 +46,7 @@ class param:
 
         @classmethod
         def verify_encoded(cls, data):
-            if len(data) not in [19, 20]:
+            if len(data) not in (18, 19, 20):
                 raise ValueError('ICCID (%s) length (%u) invalid' % (data, len(data)))
 
         @classmethod
@@ -53,7 +59,7 @@ class param:
         @classmethod
         def verify_decoded(cls, data):
             data = str(data)
-            if len(data) not in [19, 20]:
+            if len(data) not in (18, 19, 20):
                 raise ValueError('ICCID (%s) length (%u) invalid' % (data, len(data)))
             if len(data) == 19:
                 decimal_part = data
@@ -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

@@ -21,6 +21,8 @@ import logging
 import json
 from typing import Optional
 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
@@ -149,7 +161,8 @@ class ApiError(Exception):
             'message': None,
         }
         actual_sec = func_ex_status.get('statusCodeData', None)
-        sec.update(actual_sec)
+        if actual_sec:
+            sec.update(actual_sec)
         self.subject_code = sec['subjectCode']
         self.reason_code = sec['reasonCode']
         self.subject_id = sec['subjectIdentifier']
@@ -160,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')
     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():
@@ -230,30 +296,167 @@ 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
+
+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 dict. Output data is returned as json-deserialized dict."""
+
+        # 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
+
+        # 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
+        url = self.url_prefix + self.api_func.path
         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:
-            return self.decode(response.json())
+            output = self.api_func.decode_client(response.json())
+            # In case the response contains a header, check it to make sure that the API call was executed successfully
+            # (the presence of the header field is checked by the decode_client method)
+            if 'header' in output:
+                if output['header']['functionExecutionStatus']['status'] not in ['Executed-Success','Executed-WithWarning']:
+                    raise ApiError(output['header']['functionExecutionStatus'])
+            return output
         return None
+
+class JsonHttpApiServer():
+    def __init__(self, api_func: JsonHttpApiFunction, call_handler = None):
+        """
+        Args:
+                api_func : API function definition (JsonHttpApiFunction)
+                call_handler : handler function to process the request. This function must accept the
+                               decoded request as a dictionary. The handler function must return a tuple consisting
+                               of the response in the form of a dictionary (may be empty), and a function execution
+                               status string ('Executed-Success', 'Executed-WithWarning', 'Failed' or 'Expired')
+        """
+        self.api_func = api_func
+        if call_handler:
+            self.call_handler = call_handler
+        else:
+            self.call_handler = self.default_handler
+
+    def default_handler(self, data: dict) -> (dict, str):
+        """default handler, used in case no call handler is provided."""
+        logger.error("no handler function for request: %s" % str(data))
+        return {}, 'Failed'
+
+    def call(self, request: Request) -> str:
+        """ Process an incoming request.
+        Args:
+                request : request object as received using twisted.web.server
+        Returns:
+                encoded JSON string (HTTP response code and headers are set by calling the appropriate methods on the
+                provided the request object)
+        """
+
+        # Make sure the request is done with the correct HTTP method
+        if (request.method.decode() != self.api_func.http_method):
+            raise ValueError('Wrong HTTP method %s!=%s' % (request.method.decode(), self.api_func.http_method))
+
+        # Decode the request
+        decoded_request = self.api_func.decode_server(json.loads(request.content.read()))
+
+        # Run call handler (see above)
+        data, fe_status = self.call_handler(decoded_request)
+
+        # In case a function execution status is returned, use it to generate and prepend the header field according to
+        # SGP.22, section 6.5.1.2 and 6.5.1.4 (the presence of the header filed is checked by the encode_server method)
+        if fe_status:
+            data = {'header' : {'functionExecutionStatus': {'status' : fe_status}}} | data
+
+        # Encode the message (the presence of mandatory fields is checked during encoding)
+        encoded = json.dumps(self.api_func.encode_server(data))
+
+        # Apply HTTP request headers according to SGP.22, section 6.5.1
+        res_headers = {
+            'Content-Type': 'application/json',
+            'X-Admin-Protocol': 'gsma/rsp/v2.5.0',
+        }
+        res_headers.update(self.api_func.extra_http_res_headers)
+        for header, value in res_headers.items():
+            request.setHeader(header, value)
+        request.setResponseCode(self.api_func.expected_http_status)
+
+        # Return the encoded result back to the caller for sending (using twisted/klein)
+        return encoded
+

pySim/esim/saip/__init__.py

@@ -21,6 +21,8 @@ import io
 import os
 from typing import Tuple, List, Optional, Dict, Union
 from collections import OrderedDict
+from difflib import SequenceMatcher, Match
+
 import asn1tools
 import zipfile
 from pySim import javacard
@@ -44,6 +46,29 @@ asn1 = compile_asn1_subdir('saip')
 
 logger = logging.getLogger(__name__)
 
+class NonMatch(Match):
+    """Representing a contiguous non-matching block of data; the opposite of difflib.Match"""
+    @classmethod
+    def from_matchlist(cls, l: List[Match], size:int) -> List['NonMatch']:
+        """Build a list of non-matching blocks of data from its inverse (list of matching blocks).
+        The caller must ensure that the input list is ordered, non-overlapping and only contains
+        matches at equal offsets in a and b."""
+        res = []
+        cur = 0
+        for match in l:
+            if match.a != match.b:
+                raise ValueError('only works for equal-offset matches')
+            assert match.a >= cur
+            nm_len = match.a - cur
+            if nm_len > 0:
+                # there's no point in generating zero-lenth non-matching sections
+                res.append(cls(a=cur, b=cur, size=nm_len))
+            cur = match.a + match.size
+        if size > cur:
+            res.append(cls(a=cur, b=cur, size=size-cur))
+
+        return res
+
 class Naa:
     """A class defining a Network Access Application (NAA)"""
     name = None
@@ -144,6 +169,9 @@ class File:
     def file_size(self) -> Optional[int]:
         """Return the size of the file in bytes."""
         if self.file_type in ['LF', 'CY']:
+            if self._file_size and self.nb_rec is None and self.rec_len:
+                self.nb_rec = self._file_size // self.rec_len
+
             return self.nb_rec * self.rec_len
         elif self.file_type in ['TR', 'BT']:
             return self._file_size
@@ -291,6 +319,10 @@ class File:
         dfName = fileDescriptor.get('dfName', None)
         if dfName:
             self.df_name = dfName
+        efFileSize = fileDescriptor.get('efFileSize', None)
+        if efFileSize:
+            self._file_size = self._decode_file_size(efFileSize)
+
         pefi = fileDescriptor.get('proprietaryEFInfo', {})
         securityAttributesReferenced = fileDescriptor.get('securityAttributesReferenced', None)
         if securityAttributesReferenced:
@@ -300,13 +332,11 @@ class File:
             fdb_dec = fd_dec['file_descriptor_byte']
             self.shareable = fdb_dec['shareable']
             if fdb_dec['file_type'] == 'working_ef':
-                efFileSize = fileDescriptor.get('efFileSize', None)
                 if fd_dec['num_of_rec']:
                     self.nb_rec = fd_dec['num_of_rec']
                 if fd_dec['record_len']:
                     self.rec_len = fd_dec['record_len']
                 if efFileSize:
-                    self._file_size = self._decode_file_size(efFileSize)
                     if self.rec_len and self.nb_rec == None:
                         # compute the number of records from file size and record length
                         self.nb_rec = self._file_size // self.rec_len
@@ -406,12 +436,40 @@ class File:
                 return ValueError("Unknown key '%s' in tuple list" % k)
         return stream.getvalue()
 
-    def file_content_to_tuples(self) -> List[Tuple]:
-        # FIXME: simplistic approach. needs optimization. We should first check if the content
-        # matches the expanded default value from the template. If it does, return empty list.
-        # Next, we should compute the diff between the default value and self.body, and encode
-        # that as a sequence of fillFileOffset and fillFileContent tuples.
-        return [('fillFileContent', self.body)]
+    def file_content_to_tuples(self, optimize:bool = False) -> List[Tuple]:
+        """Encode the file contents into a list of fillFileContent / fillFileOffset tuples that can be fed
+        into the asn.1 encoder.  If optimize is True, it will try to encode only the differences from the
+        fillFileContent of the profile template.  Otherwise, the entire file contents will be encoded
+        as-is."""
+        if not self.file_type in ['TR', 'LF', 'CY', 'BT']:
+            return []
+        if not optimize:
+            # simplistic approach: encode the full file, ignoring the template/default
+            return [('fillFileContent', self.body)]
+        # Try to 'compress' the file body, based on the default file contents.
+        if self.template:
+            default = self.template.expand_default_value_pattern(length=len(self.body))
+            if not default:
+                sm = SequenceMatcher(a=b'\xff'*len(self.body), b=self.body)
+            else:
+                if default == self.body:
+                    # 100% match: return an empty tuple list to make eUICC use the default
+                    return []
+                sm = SequenceMatcher(a=default, b=self.body)
+        else:
+            # no template at all: we can only remove padding
+            sm = SequenceMatcher(a=b'\xff'*len(self.body), b=self.body)
+        matching_blocks = sm.get_matching_blocks()
+        # we can only make use of matches that have the same offset in 'a' and 'b'
+        matching_blocks = [x for x in matching_blocks if x.size > 0 and x.a == x.b]
+        non_matching_blocks = NonMatch.from_matchlist(matching_blocks, self.file_size)
+        ret = []
+        cur = 0
+        for block in non_matching_blocks:
+            ret.append(('fillFileOffset', block.a - cur))
+            ret.append(('fillFileContent', self.body[block.a:block.a+block.size]))
+            cur += block.size
+        return ret
 
     def __str__(self) -> str:
         return "File(%s)" % self.pe_name
@@ -633,8 +691,15 @@ class FsProfileElement(ProfileElement):
                     self.pe_sequence.cur_df = pe_df
                 self.pe_sequence.cur_df = self.pe_sequence.cur_df.add_file(file)
 
+    def file2pe(self, file: File):
+        """Update the "decoded" member for the given file with the contents from the given File instance.
+        We expect that the File instance is part of self.files"""
+        if self.files[file.pe_name] != file:
+            raise ValueError("The file you passed is not part of this ProfileElement")
+        self.decoded[file.pe_name] = file.to_tuples()
+
     def files2pe(self):
-        """Update the "decoded" member with the contents of the "files" member."""
+        """Update the "decoded" member for each file with the contents of the "files" member."""
         for k, f in self.files.items():
             self.decoded[k] = f.to_tuples()
 
@@ -1020,6 +1085,7 @@ class ProfileElementSD(ProfileElement):
     def __init__(self, decoded: Optional[dict] = None, **kwargs):
         super().__init__(decoded, **kwargs)
         if decoded:
+            self._post_decode()
             return
         # provide some reasonable defaults for a MNO-SD
         self.decoded['instance'] = {
@@ -1738,8 +1804,7 @@ class ProfileElementSequence:
                 del hdr.decoded['eUICC-Mandatory-services'][service]
         # remove any associated mandatory filesystem templates
         for template in naa.templates:
-            if template in hdr.decoded['eUICC-Mandatory-GFSTEList']:
-                hdr.decoded['eUICC-Mandatory-GFSTEList'] = [x for x in hdr.decoded['eUICC-Mandatory-GFSTEList'] if not template.prefix_match(x)]
+            hdr.decoded['eUICC-Mandatory-GFSTEList'] = [x for x in hdr.decoded['eUICC-Mandatory-GFSTEList'] if not template.prefix_match(x)]
         # determine the ADF names (AIDs) of all NAA ADFs
         naa_adf_names = []
         if naa.pe_types[0] in self.pe_by_type:
@@ -1782,7 +1847,7 @@ class ProfileElementSequence:
             return None
 
     @staticmethod
-    def peclass_for_path(path: Path) -> Optional[ProfileElement]:
+    def peclass_for_path(path: Path) -> Tuple[Optional[ProfileElement], Optional[templates.FileTemplate]]:
         """Return the ProfileElement class that can contain a file with given path."""
         naa = ProfileElementSequence.naa_for_path(path)
         if naa:
@@ -1815,7 +1880,7 @@ class ProfileElementSequence:
                 return ProfileElementTelecom, ft
         return ProfileElementGFM, None
 
-    def pe_for_path(self, path: Path) -> Optional[ProfileElement]:
+    def pe_for_path(self, path: Path) -> Tuple[Optional[ProfileElement], Optional[templates.FileTemplate]]:
         """Return the ProfileElement instance that can contain a file with matching path. This will
         either be an existing PE within the sequence, or it will be a newly-allocated PE that is
         inserted into the sequence."""
@@ -1881,7 +1946,10 @@ class ProfileElementSequence:
 
 
 class FsNode:
-    """A node in the filesystem hierarchy."""
+    """A node in the filesystem hierarchy. Each node can have a parent node and any number of children.
+    Each node is identified uniquely within the parent by its numeric FID and its optional human-readable
+    name.  Each node usually is associated with an instance of the File class for the actual content of
+    the file.  FsNode is the base class used by more  specific nodes, such as FsNode{EF,DF,ADF,MF}."""
     def __init__(self, fid: int, parent: Optional['FsNode'], file: Optional[File] = None,
                  name: Optional[str] = None):
         self.fid = fid
@@ -1936,7 +2004,7 @@ class FsNode:
         return x
 
     def walk(self, fn, **kwargs):
-        """call 'fn(self, **kwargs) for the File."""
+        """call 'fn(self, ``**kwargs``) for the File."""
         return [fn(self, **kwargs)]
 
 class FsNodeEF(FsNode):
@@ -2026,7 +2094,7 @@ class FsNodeDF(FsNode):
         return cur
 
     def walk(self, fn, **kwargs):
-        """call 'fn(self, **kwargs) for the DF and recursively for all children."""
+        """call 'fn(self, ``**kwargs``) for the DF and recursively for all children."""
         ret = super().walk(fn, **kwargs)
         for c in self.children.values():
             ret += c.walk(fn, **kwargs)

pySim/esim/saip/personalization.py

@@ -22,6 +22,7 @@ from typing import List, Tuple
 from osmocom.tlv import camel_to_snake
 from pySim.utils import enc_iccid, enc_imsi, h2b, rpad, sanitize_iccid
 from pySim.esim.saip import ProfileElement, ProfileElementSequence
+from pySim.ts_51_011 import EF_SMSP
 
 def remove_unwanted_tuples_from_list(l: List[Tuple], unwanted_keys: List[str]) -> List[Tuple]:
     """In a list of tuples, remove all tuples whose first part equals 'unwanted_key'."""
@@ -46,56 +47,293 @@ class ClassVarMeta(abc.ABCMeta):
         return x
 
 class ConfigurableParameter(abc.ABC, metaclass=ClassVarMeta):
-    """Base class representing a part of the eSIM profile that is configurable during the
-    personalization process (with dynamic data from elsewhere)."""
-    def __init__(self, input_value):
+    r"""Base class representing a part of the eSIM profile that is configurable during the
+    personalization process (with dynamic data from elsewhere).
+
+    This class is abstract, you will only use subclasses in practice.
+
+    Subclasses have to implement the apply_val() classmethods, and may choose to override the default validate_val()
+    implementation.
+    The default validate_val() is a generic validator that uses the following class members (defined in subclasses) to
+    configure the validation; if any of them is None, it means that the particular validation is skipped:
+
+    allow_types: a list of types permitted as argument to validate_val(); allow_types = (bytes, str,)
+    allow_chars: if val is a str, accept only these characters; allow_chars = "0123456789"
+    strip_chars: if val is a str, remove these characters; strip_chars = ' \t\r\n'
+    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)
+
+    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().
+
+    Usage examples, by example of Iccid:
+
+    1) use a ConfigurableParameter instance, with .input_value and .value state::
+
+          iccid = Iccid()
+          try:
+            iccid.input_value = '123456789012345678'
+            iccid.validate()
+          except ValueError:
+            print(f"failed to validate {iccid.name} == {iccid.input_value}")
+
+          pes = ProfileElementSequence.from_der(der_data_from_file)
+          try:
+            iccid.apply(pes)
+          except ValueError:
+            print(f"failed to apply {iccid.name} := {iccid.input_value}")
+
+          changed_der = pes.to_der()
+
+    2) use a ConfigurableParameter class, without state::
+
+          cls = Iccid
+          input_val = '123456789012345678'
+
+          try:
+            clean_val = cls.validate_val(input_val)
+          except ValueError:
+            print(f"failed to validate {cls.get_name()} = {input_val}")
+
+          pes = ProfileElementSequence.from_der(der_data_from_file)
+          try:
+            cls.apply_val(pes, clean_val)
+          except ValueError:
+            print(f"failed to apply {cls.get_name()} = {input_val}")
+
+          changed_der = pes.to_der()
+
+    """
+
+    # A subclass can set an explicit string as name (like name = "PIN1").
+    # If name is left None, then __init__() will set self.name to a name derived from the python class name (like
+    # "pin1"). See also the get_name() classmethod when you have no instance at hand.
+    name = None
+    allow_types = (str, int, )
+    allow_chars = None
+    strip_chars = None
+    min_len = None
+    max_len = None
+    allow_len = None # a list of specific lengths
+    example_input = None
+
+    def __init__(self, input_value=None):
         self.input_value = input_value # the raw input value as given by caller
         self.value = None # the processed input value (e.g. with check digit) as produced by validate()
 
-    def validate(self):
-        """Optional validation method. Can be used by derived classes to perform validation
-        of the input value (self.value).  Will raise an exception if validation fails."""
-        # default implementation: simply copy input_value over to value
-        self.value = self.input_value
+        # if there is no explicit name string set, use the class name
+        self.name = self.get_name()
+
+    @classmethod
+    def get_name(cls):
+        """Return cls.name when it is set, otherwise return the python class name converted from 'CamelCase' to
+        'snake_case'.
+        When using class *instances*, you can just use my_instance.name.
+        When using *classes*, cls.get_name() returns the same name a class instance would have.
+        """
+        if cls.name:
+            return cls.name
+        return camel_to_snake(cls.__name__)
+
+    def validate(self):
+        """Validate self.input_value and place the result in self.value.
+        This is also called implicitly by apply(), if self.value is still None.
+        To override validation in a subclass, rather re-implement the classmethod validate_val()."""
+        try:
+            self.value = self.__class__.validate_val(self.input_value)
+        except (TypeError, ValueError, KeyError) as e:
+            raise ValueError(f'{self.name}: {e}') from e
 
-    @abc.abstractmethod
     def apply(self, pes: ProfileElementSequence):
+        """Place self.value into the ProfileElementSequence at the right place.
+        If self.value is None, this implicitly calls self.validate() first, to generate a sanitized self.value from
+        self.input_value.
+        To override apply() in a subclass, rather override the classmethod apply_val()."""
+        if self.value is None:
+            self.validate()
+            assert self.value is not None
+        try:
+            self.__class__.apply_val(pes, self.value)
+        except (TypeError, ValueError, KeyError) as e:
+            raise ValueError(f'{self.name}: {e}') from e
+
+    @classmethod
+    def validate_val(cls, val):
+        """This is a default implementation, with the behavior configured by subclasses' allow_types...max_len settings.
+        subclasses may override this function:
+        Validate the contents of val, and raise ValueError on validation errors.
+        Return a sanitized version of val, that is ready for cls.apply_val().
+        """
+
+        if cls.allow_types is not None:
+            if not isinstance(val, cls.allow_types):
+                raise ValueError(f'input value must be one of {cls.allow_types}, not {type(val)}')
+        elif val is None:
+            raise ValueError('there is no value (val is None)')
+
+        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)
+            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}")
+        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 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 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 len(val) > cls.max_len:
+                raise ValueError(f'length must be at most {cls.max_len}, not {len(val)}: {val!r}')
+        return val
+
+    @classmethod
+    def apply_val(cls, pes: ProfileElementSequence, val):
+        """This is what subclasses implement: store a value in a decoded profile package.
+        Write the given val in the right format in all the right places in pes."""
         pass
 
-class Iccid(ConfigurableParameter):
-    """Configurable ICCID.  Expects the value to be a string of decimal digits.
-    If the string of digits is only 18 digits long, a Luhn check digit will be added."""
+    @classmethod
+    def get_len_range(cls):
+        """considering all of min_len, max_len and allow_len, get a tuple of the resulting (min, max) of permitted
+        value length. For example, if an input value is an int, which needs to be represented with a minimum nr of
+        digits, this function is useful to easily get that minimum permitted length.
+        """
+        vals = []
+        if cls.allow_len is not None:
+            if isinstance(cls.allow_len, (tuple, list)):
+                vals.extend(cls.allow_len)
+            else:
+                vals.append(cls.allow_len)
+        if cls.min_len is not None:
+            vals.append(cls.min_len)
+        if cls.max_len is not None:
+            vals.append(cls.max_len)
+        if not vals:
+            return (None, None)
+        return (min(vals), max(vals))
 
-    def validate(self):
-        # convert to string as it might be an integer
-        iccid_str = str(self.input_value)
-        if len(iccid_str) < 18 or len(iccid_str) > 20:
-            raise ValueError('ICCID must be 18, 19 or 20 digits long')
-        if not iccid_str.isdecimal():
-            raise ValueError('ICCID must only contain decimal digits')
-        self.value = sanitize_iccid(iccid_str)
 
-    def apply(self, pes: ProfileElementSequence):
+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'
+
+    @classmethod
+    def validate_val(cls, val):
+        if isinstance(val, int):
+            min_len, max_len = cls.get_len_range()
+            l = min_len or 1
+            val = '%0*d' % (l, val)
+        return super().validate_val(val)
+
+
+class DecimalHexParam(DecimalParam):
+    """The input value is decimal digits. The decimal value is stored such that each hexadecimal digit represents one
+    decimal digit, useful for various PIN type parameters.
+
+    Optionally, the value is stored with padding, for example: rpad = 8 would store '123' as '123fffff'. This is also
+    common in PIN type parameters.
+    """
+    rpad = None
+    rpad_char = 'f'
+
+    @classmethod
+    def validate_val(cls, val):
+        val = super().validate_val(val)
+        val = ''.join('%02x' % ord(x) for x in val)
+        if cls.rpad is not None:
+            c = cls.rpad_char
+            val = rpad(val, cls.rpad, c)
+        # a DecimalHexParam subclass expects the apply_val() input to be a bytes instance ready for the pes
+        return h2b(val)
+
+class IntegerParam(ConfigurableParameter):
+    allow_types = (str, int)
+    allow_chars = '0123456789'
+
+    # two integers, if the resulting int should be range limited
+    min_val = None
+    max_val = None
+
+    @classmethod
+    def validate_val(cls, val):
+        val = super().validate_val(val)
+        val = int(val)
+        exceeds_limits = False
+        if cls.min_val is not None:
+            if val < cls.min_val:
+                exceeds_limits = True
+        if cls.max_val is not None:
+            if val > cls.max_val:
+                exceeds_limits = True
+        if exceeds_limits:
+            raise ValueError(f'Value {val} is out of range, must be [{cls.min_val}..{cls.max_val}]')
+        return val
+
+class BinaryParam(ConfigurableParameter):
+    allow_types = (str, io.BytesIO, bytes, bytearray)
+    allow_chars = '0123456789abcdefABCDEF'
+    strip_chars = ' \t\r\n'
+
+    @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, str):
+            if cls.strip_chars is not None:
+                val = ''.join(c for c in val if c not in cls.strip_chars)
+            if len(val) & 1:
+                raise ValueError('Invalid hexadecimal string, must have even number of digits:'
+                                 f' {val!r} {len(val)=}')
+            try:
+                val = h2b(val)
+            except ValueError as e:
+                raise ValueError(f'Invalid hexadecimal string: {val!r} {len(val)=}') from e
+
+        val = super().validate_val(val)
+        return bytes(val)
+
+
+class Iccid(DecimalParam):
+    """ICCID Parameter. Input: string of decimal digits.
+    If the string of digits is only 18 digits long, add a Luhn check digit."""
+    name = 'ICCID'
+    min_len = 18
+    max_len = 20
+    example_input = '998877665544332211'
+
+    @classmethod
+    def validate_val(cls, val):
+        iccid_str = super().validate_val(val)
+        return sanitize_iccid(iccid_str)
+
+    @classmethod
+    def apply_val(cls, pes: ProfileElementSequence, val):
         # patch the header
-        pes.get_pe_for_type('header').decoded['iccid'] = h2b(rpad(self.value, 20))
+        pes.get_pe_for_type('header').decoded['iccid'] = h2b(rpad(val, 20))
         # patch MF/EF.ICCID
-        file_replace_content(pes.get_pe_for_type('mf').decoded['ef-iccid'], h2b(enc_iccid(self.value)))
+        file_replace_content(pes.get_pe_for_type('mf').decoded['ef-iccid'], h2b(enc_iccid(val)))
 
-class Imsi(ConfigurableParameter):
+class Imsi(DecimalParam):
     """Configurable IMSI. Expects value to be a string of digits. Automatically sets the ACC to
     the last digit of the IMSI."""
 
-    def validate(self):
-        # convert to string as it might be an integer
-        imsi_str = str(self.input_value)
-        if len(imsi_str) < 6 or len(imsi_str) > 15:
-            raise ValueError('IMSI must be 6..15 digits long')
-        if not imsi_str.isdecimal():
-            raise ValueError('IMSI must only contain decimal digits')
-        self.value = imsi_str
+    name = 'IMSI'
+    min_len = 6
+    max_len = 15
+    example_input = '00101' + ('0' * 10)
 
-    def apply(self, pes: ProfileElementSequence):
-        imsi_str = self.value
+    @classmethod
+    def apply_val(cls, pes: ProfileElementSequence, val):
+        imsi_str = val
         # we always use the least significant byte of the IMSI as ACC
         acc = (1 << int(imsi_str[-1]))
         # patch ADF.USIM/EF.IMSI
@@ -104,45 +342,93 @@ class Imsi(ConfigurableParameter):
             file_replace_content(pe.decoded['ef-acc'], acc.to_bytes(2, 'big'))
         # TODO: DF.GSM_ACCESS if not linked?
 
+class SmspTpScAddr(ConfigurableParameter):
+    """Configurable SMSC (SMS Service Centre) TP-SC-ADDR. Expects to be a phone number in national or
+    international format (designated by a leading +). Automatically sets the NPI to E.164 and the TON based on
+    presence or absence of leading +."""
 
-class SdKey(ConfigurableParameter, metaclass=ClassVarMeta):
+    name = 'SMSP-TP-SC-ADDR'
+    allow_chars = '+0123456789'
+    strip_chars = ' \t\r\n'
+    max_len = 21 # '+' and 20 digits
+    min_len = 1
+    example_input = '+49301234567'
+
+    @classmethod
+    def validate_val(cls, val):
+        val = super().validate_val(val)
+        addr_str = str(val)
+        if addr_str[0] == '+':
+            digits = addr_str[1:]
+            international = True
+        else:
+            digits = addr_str
+            international = False
+        if len(digits) > 20:
+            raise ValueError(f'TP-SC-ADDR must not exceed 20 digits: {digits!r}')
+        if not digits.isdecimal():
+            raise ValueError(f'TP-SC-ADDR must only contain decimal digits: {digits!r}')
+        return (international, digits)
+
+    @classmethod
+    def apply_val(cls, pes: ProfileElementSequence, val):
+        """val must be a tuple (international[bool], digits[str]).
+        For example, an input of "+1234" corresponds to (True, "1234");
+        An input of "1234" corresponds to (False, "1234")."""
+        international, digits = val
+        for pe in pes.get_pes_for_type('usim'):
+            # obtain the File instance from the ProfileElementUSIM
+            f_smsp = pe.files['ef-smsp']
+            #print("SMSP (orig): %s" % f_smsp.body)
+            # instantiate the pySim.ts_51_011.EF_SMSP class for decode/encode
+            ef_smsp = EF_SMSP()
+            # decode the existing file body
+            ef_smsp_dec = ef_smsp.decode_record_bin(f_smsp.body, 1)
+            # patch the actual number
+            ef_smsp_dec['tp_sc_addr']['call_number'] = digits
+            # patch the NPI to isdn_e164
+            ef_smsp_dec['tp_sc_addr']['ton_npi']['numbering_plan_id'] = 'isdn_e164'
+            # patch the TON to international or unknown depending on +
+            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
+            # 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)
+            # re-generate the pe.decoded member from the File instance
+            pe.file2pe(f_smsp)
+
+class SdKey(BinaryParam, metaclass=ClassVarMeta):
     """Configurable Security Domain (SD) Key.  Value is presented as bytes."""
-    # these will be set by derived classes
+    # these will be set by subclasses
     key_type = None
     key_id = None
     kvn = None
     key_usage_qual = None
-    permitted_len = []
 
-    def validate(self):
-        if not isinstance(self.input_value, (io.BytesIO, bytes, bytearray)):
-            raise ValueError('Value must be of bytes-like type')
-        if self.permitted_len:
-            if len(self.input_value) not in self.permitted_len:
-                raise ValueError('Value length must be %s' % self.permitted_len)
-        self.value = self.input_value
-
-    def _apply_sd(self, pe: ProfileElement):
+    @classmethod
+    def _apply_sd(cls, pe: ProfileElement, value):
         assert pe.type == 'securityDomain'
         for key in pe.decoded['keyList']:
-            if key['keyIdentifier'][0] == self.key_id and key['keyVersionNumber'][0] == self.kvn:
+            if key['keyIdentifier'][0] == cls.key_id and key['keyVersionNumber'][0] == cls.kvn:
                 assert len(key['keyComponents']) == 1
-                key['keyComponents'][0]['keyData'] = self.value
+                key['keyComponents'][0]['keyData'] = value
                 return
         # Could not find matching key to patch, create a new one
         key = {
-            'keyUsageQualifier': bytes([self.key_usage_qual]),
-            'keyIdentifier': bytes([self.key_id]),
-            'keyVersionNumber': bytes([self.kvn]),
+            'keyUsageQualifier': bytes([cls.key_usage_qual]),
+            'keyIdentifier': bytes([cls.key_id]),
+            'keyVersionNumber': bytes([cls.kvn]),
             'keyComponents': [
-                { 'keyType': bytes([self.key_type]), 'keyData': self.value },
+                { 'keyType': bytes([cls.key_type]), 'keyData': value },
             ]
         }
         pe.decoded['keyList'].append(key)
 
-    def apply(self, pes: ProfileElementSequence):
+    @classmethod
+    def apply_val(cls, pes: ProfileElementSequence, value):
         for pe in pes.get_pes_for_type('securityDomain'):
-            self._apply_sd(pe)
+            cls._apply_sd(pe, value)
 
 class SdKeyScp80_01(SdKey, kvn=0x01, key_type=0x88, permitted_len=[16,24,32]): # AES key type
     pass
@@ -199,6 +485,9 @@ class SdKeyScp03_32Dek(SdKeyScp03_32, key_id=0x03, key_usage_qual=0x48):
 
 
 
+def obtain_all_pe_from_pelist(l: List[ProfileElement], wanted_type: str) -> ProfileElement:
+    return (pe for pe in l if pe.type == wanted_type)
+
 def obtain_singleton_pe_from_pelist(l: List[ProfileElement], wanted_type: str) -> ProfileElement:
     filtered = list(filter(lambda x: x.type == wanted_type, l))
     assert len(filtered) == 1
@@ -208,117 +497,173 @@ def obtain_first_pe_from_pelist(l: List[ProfileElement], wanted_type: str) -> Pr
     filtered = list(filter(lambda x: x.type == wanted_type, l))
     return filtered[0]
 
-class Puk(ConfigurableParameter, metaclass=ClassVarMeta):
+class Puk(DecimalHexParam):
     """Configurable PUK (Pin Unblock Code). String ASCII-encoded digits."""
+    allow_len = 8
+    rpad = 16
     keyReference = None
-    def validate(self):
-        if isinstance(self.input_value, int):
-            self.value = '%08d' % self.input_value
-        else:
-            self.value = self.input_value
-        # FIXME: valid length?
-        if not self.value.isdecimal():
-            raise ValueError('PUK must only contain decimal digits')
+    example_input = '0' * allow_len
 
-    def apply(self, pes: ProfileElementSequence):
-        puk = ''.join(['%02x' % (ord(x)) for x in self.value])
-        padded_puk = rpad(puk, 16)
+    @classmethod
+    def apply_val(cls, pes: ProfileElementSequence, val):
+        val_bytes = val
         mf_pes = pes.pes_by_naa['mf'][0]
         pukCodes = obtain_singleton_pe_from_pelist(mf_pes, 'pukCodes')
         for pukCode in pukCodes.decoded['pukCodes']:
-            if pukCode['keyReference'] == self.keyReference:
-                pukCode['pukValue'] = h2b(padded_puk)
+            if pukCode['keyReference'] == cls.keyReference:
+                pukCode['pukValue'] = val_bytes
                 return
-        raise ValueError('cannot find pukCode')
-class Puk1(Puk, keyReference=0x01):
-    pass
-class Puk2(Puk, keyReference=0x81):
-    pass
+        raise ValueError("input template UPP has unexpected structure:"
+                         f" cannot find pukCode with keyReference={cls.keyReference}")
 
-class Pin(ConfigurableParameter, metaclass=ClassVarMeta):
+class Puk1(Puk):
+    name = 'PUK1'
+    keyReference = 0x01
+
+class Puk2(Puk):
+    name = 'PUK2'
+    keyReference = 0x81
+
+class Pin(DecimalHexParam):
     """Configurable PIN (Personal Identification Number).  String of digits."""
+    rpad = 16
+    min_len = 4
+    max_len = 8
+    example_input = '0' * max_len
     keyReference = None
-    def validate(self):
-        if isinstance(self.input_value, int):
-            self.value = '%04d' % self.input_value
-        else:
-            self.value = self.input_value
-        if len(self.value) < 4 or len(self.value) > 8:
-            raise ValueError('PIN mus be 4..8 digits long')
-        if not self.value.isdecimal():
-            raise ValueError('PIN must only contain decimal digits')
-    def apply(self, pes: ProfileElementSequence):
-        pin = ''.join(['%02x' % (ord(x)) for x in self.value])
-        padded_pin = rpad(pin, 16)
-        mf_pes = pes.pes_by_naa['mf'][0]
-        pinCodes = obtain_first_pe_from_pelist(mf_pes, 'pinCodes')
-        if pinCodes.decoded['pinCodes'][0] != 'pinconfig':
-            return
-        for pinCode in pinCodes.decoded['pinCodes'][1]:
-            if pinCode['keyReference'] == self.keyReference:
-                 pinCode['pinValue'] = h2b(padded_pin)
-                 return
-        raise ValueError('cannot find pinCode')
-class AppPin(ConfigurableParameter, metaclass=ClassVarMeta):
-    """Configurable PIN (Personal Identification Number).  String of digits."""
-    keyReference = None
-    def validate(self):
-        if isinstance(self.input_value, int):
-            self.value = '%04d' % self.input_value
-        else:
-            self.value = self.input_value
-        if len(self.value) < 4 or len(self.value) > 8:
-            raise ValueError('PIN mus be 4..8 digits long')
-        if not self.value.isdecimal():
-            raise ValueError('PIN must only contain decimal digits')
-    def _apply_one(self, pe: ProfileElement):
-        pin = ''.join(['%02x' % (ord(x)) for x in self.value])
-        padded_pin = rpad(pin, 16)
-        pinCodes = obtain_first_pe_from_pelist(pe, 'pinCodes')
-        if pinCodes.decoded['pinCodes'][0] != 'pinconfig':
-            return
-        for pinCode in pinCodes.decoded['pinCodes'][1]:
-            if pinCode['keyReference'] == self.keyReference:
-                pinCode['pinValue'] = h2b(padded_pin)
-                return
-        raise ValueError('cannot find pinCode')
-    def apply(self, pes: ProfileElementSequence):
+
+    @staticmethod
+    def _apply_pinvalue(pe: ProfileElement, keyReference, val_bytes):
+        for pinCodes in obtain_all_pe_from_pelist(pe, 'pinCodes'):
+            if pinCodes.decoded['pinCodes'][0] != 'pinconfig':
+                continue
+
+            for pinCode in pinCodes.decoded['pinCodes'][1]:
+                if pinCode['keyReference'] == keyReference:
+                     pinCode['pinValue'] = val_bytes
+                     return True
+        return False
+
+    @classmethod
+    def apply_val(cls, pes: ProfileElementSequence, val):
+        val_bytes = val
+        if not cls._apply_pinvalue(pes.pes_by_naa['mf'][0], cls.keyReference, val_bytes):
+            raise ValueError('input template UPP has unexpected structure:'
+                             + f' {cls.get_name()} cannot find pinCode with keyReference={cls.keyReference}')
+
+class Pin1(Pin):
+    name = 'PIN1'
+    example_input = '0' * 4  # PIN are usually 4 digits
+    keyReference = 0x01
+
+class Pin2(Pin1):
+    name = 'PIN2'
+    keyReference = 0x81
+
+    @classmethod
+    def apply_val(cls, pes: ProfileElementSequence, val):
+        val_bytes = val
+        # PIN2 is special: telecom + usim + isim + csim
         for naa in pes.pes_by_naa:
             if naa not in ['usim','isim','csim','telecom']:
                 continue
             for instance in pes.pes_by_naa[naa]:
-                self._apply_one(instance)
-class Pin1(Pin, keyReference=0x01):
-    pass
-# PIN2 is special: telecom + usim + isim + csim
-class Pin2(AppPin, keyReference=0x81):
-    pass
-class Adm1(Pin, keyReference=0x0A):
-    pass
-class Adm2(Pin, keyReference=0x0B):
-    pass
+                if not cls._apply_pinvalue(instance, cls.keyReference, val_bytes):
+                    raise ValueError('input template UPP has unexpected structure:'
+                            + f' {cls.get_name()} cannot find pinCode with keyReference={cls.keyReference} in {naa=}')
 
+class Adm1(Pin):
+    name = 'ADM1'
+    keyReference = 0x0A
 
-class AlgoConfig(ConfigurableParameter, metaclass=ClassVarMeta):
-    """Configurable Algorithm parameter."""
-    key = None
-    def validate(self):
-        if not isinstance(self.input_value, (io.BytesIO, bytes, bytearray)):
-            raise ValueError('Value must be of bytes-like type')
-        self.value = self.input_value
-    def apply(self, pes: ProfileElementSequence):
+class Adm2(Adm1):
+    name = 'ADM2'
+    keyReference = 0x0B
+
+class AlgoConfig(ConfigurableParameter):
+    algo_config_key = None
+
+    @classmethod
+    def apply_val(cls, pes: ProfileElementSequence, val):
+        found = 0
         for pe in pes.get_pes_for_type('akaParameter'):
             algoConfiguration = pe.decoded['algoConfiguration']
             if algoConfiguration[0] != 'algoParameter':
                 continue
-            algoConfiguration[1][self.key] = self.value
+            algoConfiguration[1][cls.algo_config_key] = val
+            found += 1
+        if not found:
+            raise ValueError('input template UPP has unexpected structure:'
+                             f' {cls.__name__} cannot find algoParameter with key={cls.algo_config_key}')
 
-class K(AlgoConfig, key='key'):
-    pass
-class Opc(AlgoConfig, key='opc'):
-    pass
-class AlgorithmID(AlgoConfig, key='algorithmID'):
-    def validate(self):
-        if self.input_value not in [1, 2, 3]:
-            raise ValueError('Invalid algorithmID %s' % (self.input_value))
-        self.value = self.input_value
+class AlgorithmID(DecimalParam, AlgoConfig):
+    algo_config_key = 'algorithmID'
+    allow_len = 1
+    example_input = 1  # Milenage
+
+    @classmethod
+    def validate_val(cls, val):
+        val = super().validate_val(val)
+        val = int(val)
+        valid = (1, 2, 3)
+        if val not in valid:
+            raise ValueError(f'Invalid algorithmID {val!r}, must be one of {valid}')
+        return val
+
+class K(BinaryParam, AlgoConfig):
+    """use validate_val() from BinaryParam, and apply_val() from AlgoConfig"""
+    name = 'K'
+    algo_config_key = 'key'
+    allow_len = (128 // 8, 256 // 8) # length in bytes (from BinaryParam); TUAK also allows 256 bit
+    example_input = '00' * allow_len[0]
+
+class Opc(K):
+    name = 'OPc'
+    algo_config_key = 'opc'
+
+class MilenageRotationConstants(BinaryParam, AlgoConfig):
+    """rotation constants r1,r2,r3,r4,r5 of Milenage, Range 0..127. See 3GPP TS 35.206 Sections 2.3 + 5.3.
+    Provided as octet-string concatenation of all 5 constants.  Expects a bytes-like object of length 5, with
+    each byte in the range of 0..127.  The default value by 3GPP is '4000204060' (hex notation)"""
+    name = 'MilenageRotation'
+    algo_config_key = 'rotationConstants'
+    allow_len = 5 # length in bytes (from BinaryParam)
+    example_input = '40 00 20 40 60'
+
+    @classmethod
+    def validate_val(cls, val):
+        "allow_len checks the length, this in addition checks the value range"
+        val = super().validate_val(val)
+        assert isinstance(val, bytes)
+        if any(r > 127 for r in val):
+            raise ValueError('r values must be in the range 0..127')
+        return val
+
+class MilenageXoringConstants(BinaryParam, AlgoConfig):
+    """XOR-ing constants c1,c2,c3,c4,c5 of Milenage, 128bit each. See 3GPP TS 35.206 Sections 2.3 + 5.3.
+    Provided as octet-string concatenation of all 5 constants. The default value by 3GPP is the concetenation
+    of::
+
+     00000000000000000000000000000000
+     00000000000000000000000000000001
+     00000000000000000000000000000002
+     00000000000000000000000000000004
+     00000000000000000000000000000008
+
+    """
+    name = 'MilenageXOR'
+    algo_config_key = 'xoringConstants'
+    allow_len = 80 # length in bytes (from BinaryParam)
+    example_input = ('00000000000000000000000000000000'
+                     ' 00000000000000000000000000000001'
+                     ' 00000000000000000000000000000002'
+                     ' 00000000000000000000000000000004'
+                     ' 00000000000000000000000000000008')
+
+class TuakNumberOfKeccak(IntegerParam, AlgoConfig):
+    """Number of iterations of Keccak-f[1600] permutation as recomended by Section 7.2 of 3GPP TS 35.231"""
+    name = 'KECCAK-N'
+    algo_config_key = 'numberOfKeccak'
+    min_val = 1
+    max_val = 255
+    example_input = '1'

pySim/esim/saip/templates.py

@@ -673,7 +673,7 @@ class FilesUsimDf5GS(ProfileTemplate):
         FileTemplate(0x4f06, 'EF.UAC_AIC',           'TR', None,    4,   2, 0x06, None, True, ass_serv=[126]),
         FileTemplate(0x4f07, 'EF.SUCI_Calc_Info',    'TR', None, None,   2, 0x07, 'FF...FF', False, ass_serv=[124]),
         FileTemplate(0x4f08, 'EF.OPL5G',             'LF', None,   10,  10, 0x08, 'FF...FF', False, ['nb_rec'], ass_serv=[129]),
-        FileTemplate(0x4f09, 'EF.SUPI_NAI',          'TR', None, None,   2, 0x09, None, True, ['size'], ass_serv=[130]),
+        FileTemplate(0x4f09, 'EF.SUPI_NAI',          'TR', None, None,   2, 0x09, None, True, ['size'], ass_serv=[130], pe_name='ef-supinai'),
         FileTemplate(0x4f0a, 'EF.Routing_Indicator', 'TR', None,    4,   2, 0x0a, 'F0FFFFFF', False, ass_serv=[124]),
     ]
 
@@ -818,7 +818,7 @@ class FilesIsimOptional(ProfileTemplate):
     base_path = Path('ADF.ISIM')
     extends = FilesIsimMandatory
     files = [
-        FileTemplate(0x6f09, 'EF.P-CSCF',      'LF',    1, None,   2, None, None, True, ['size'], ass_serv=[1,5]),
+        FileTemplate(0x6f09, 'EF.P-CSCF',      'LF',    1, None,   2, None, None, True, ['size'], ass_serv=[1,5], pe_name='ef-pcscf'),
         FileTemplate(0x6f3c, 'EF.SMS',         'LF',   10,  176,   5, None, '00FF...FF', False, ass_serv=[6,8]),
         FileTemplate(0x6f42, 'EF.SMSP',        'LF',    1,   38,   5, None, 'FF...FF', False, ass_serv=[8]),
         FileTemplate(0x6f43, 'EF.SMSS',        'TR', None,    2,   5, None, 'FFFF', False, ass_serv=[6,8]),

pySim/esim/saip/validation.py

@@ -103,6 +103,26 @@ class CheckBasicStructure(ProfileConstraintChecker):
         if 'profile-a-p256' in m_svcs and not ('usim' in m_svcs or 'isim' in m_svcs):
             raise ProfileError('profile-a-p256 mandatory, but no usim or isim')
 
+    def check_mandatory_services_aka(self, pes: ProfileElementSequence):
+        """Ensure that no unnecessary authentication related services are marked as mandatory but not
+        actually used within the profile"""
+        m_svcs = pes.get_pe_for_type('header').decoded['eUICC-Mandatory-services']
+        # list of tuples (algo_id, key_len_in_octets) for all the akaParameters in the PE Sequence
+        algo_id_klen = [(x.decoded['algoConfiguration'][1]['algorithmID'],
+                         len(x.decoded['algoConfiguration'][1]['key'])) for x in pes.get_pes_for_type('akaParameter')]
+        # just a plain list of algorithm IDs in akaParameters
+        algorithm_ids = [x[0] for x in algo_id_klen]
+        if 'milenage' in m_svcs and not 1 in algorithm_ids:
+            raise ProfileError('milenage mandatory, but no related algorithm_id in akaParameter')
+        if 'tuak128' in m_svcs and not (2, 128/8) in algo_id_klen:
+            raise ProfileError('tuak128 mandatory, but no related algorithm_id in akaParameter')
+        if 'cave' in m_svcs and not pes.get_pe_for_type('cdmaParameter'):
+            raise ProfileError('cave mandatory, but no related cdmaParameter')
+        if 'tuak256' in m_svcs and (2, 256/8) in algo_id_klen:
+            raise ProfileError('tuak256 mandatory, but no related algorithm_id in akaParameter')
+        if 'usim-test-algorithm' in m_svcs and not 3 in algorithm_ids:
+            raise ProfileError('usim-test-algorithm mandatory, but no related algorithm_id in akaParameter')
+
     def check_identification_unique(self, pes: ProfileElementSequence):
         """Ensure that each PE has a unique identification value."""
         id_list = [pe.header['identification'] for pe in pes.pe_list if pe.header]

pySim/euicc.py

@@ -181,7 +181,7 @@ class SeqNumber(BER_TLV_IE, tag=0x80):
 class NotificationAddress(BER_TLV_IE, tag=0x0c):
     _construct = Utf8Adapter(GreedyBytes)
 class Iccid(BER_TLV_IE, tag=0x5a):
-    _construct = BcdAdapter(GreedyBytes)
+    _construct = PaddedBcdAdapter(GreedyBytes)
 class NotificationMetadata(BER_TLV_IE, tag=0xbf2f, nested=[SeqNumber, ProfileMgmtOperation,
                                                            NotificationAddress, Iccid]):
     pass

pySim/global_platform/install_param.py

@@ -1,6 +1,6 @@
 # GlobalPlatform install parameter generator
 #
-# (C) 2024 by Sysmocom s.f.m.c. GmbH
+# (C) 2024 by sysmocom - s.f.m.c. GmbH
 # All Rights Reserved
 #
 # This program is free software: you can redistribute it and/or modify

pySim/javacard.py

@@ -1,6 +1,6 @@
 # JavaCard related utilities
 #
-# (C) 2024 by Sysmocom s.f.m.c. GmbH
+# (C) 2024 by sysmocom - s.f.m.c. GmbH
 # All Rights Reserved
 #
 # This program is free software: you can redistribute it and/or modify

pySim/log.py

@@ -4,7 +4,7 @@
 """
 
 #
-# (C) 2025 by Sysmocom s.f.m.c. GmbH
+# (C) 2025 by sysmocom - s.f.m.c. GmbH
 # All Rights Reserved
 #
 # Author: Philipp Maier <pmaier@sysmocom.de>
@@ -44,7 +44,7 @@ class PySimLogger:
     """
 
     LOG_FMTSTR = "%(levelname)s: %(message)s"
-    LOG_FMTSTR_VERBOSE = "%(module)s.%(lineno)d -- %(name)s - " + LOG_FMTSTR
+    LOG_FMTSTR_VERBOSE = "%(module)s.%(lineno)d -- " + LOG_FMTSTR
     __formatter = logging.Formatter(LOG_FMTSTR)
     __formatter_verbose = logging.Formatter(LOG_FMTSTR_VERBOSE)
 
@@ -108,7 +108,10 @@ class PySimLogger:
                 formatted_message = logging.Formatter.format(PySimLogger.__formatter, record)
             color = PySimLogger.colors.get(record.levelno)
             if color:
-                PySimLogger.print_callback(style(formatted_message, fg = color))
+                if isinstance(color, str):
+                    PySimLogger.print_callback(color + formatted_message + "\033[0m")
+                else:
+                    PySimLogger.print_callback(style(formatted_message, fg = color))
             else:
                 PySimLogger.print_callback(formatted_message)
 

pySim/ota.py

@@ -57,12 +57,13 @@ CompactRemoteResp = Struct('number_of_commands'/Int8ub,
                            'last_response_data'/HexAdapter(GreedyBytes))
 
 RC_CC_DS = Enum(BitsInteger(2), no_rc_cc_ds=0, rc=1, cc=2, ds=3)
+CNTR_REQ = Enum(BitsInteger(2), no_counter=0, counter_no_replay_or_seq=1, counter_must_be_higher=2, counter_must_be_lower=3)
+POR_REQ = Enum(BitsInteger(2), no_por=0, por_required=1, por_only_when_error=2)
 
 # TS 102 225 Section 5.1.1 + TS 31.115 Section 4.2
 SPI = BitStruct(  # first octet
     Padding(3),
-    'counter'/Enum(BitsInteger(2), no_counter=0, counter_no_replay_or_seq=1,
-                   counter_must_be_higher=2, counter_must_be_lower=3),
+    'counter'/CNTR_REQ,
     'ciphering'/Flag,
     'rc_cc_ds'/RC_CC_DS,
     # second octet
@@ -70,8 +71,7 @@ SPI = BitStruct(  # first octet
     'por_in_submit'/Flag,
     'por_shall_be_ciphered'/Flag,
     'por_rc_cc_ds'/RC_CC_DS,
-    'por'/Enum(BitsInteger(2), no_por=0,
-               por_required=1, por_only_when_error=2)
+    'por'/POR_REQ
 )
 
 # TS 102 225 Section 5.1.2

pySim/profile.py

@@ -4,7 +4,7 @@
 """
 
 #
-# (C) 2021 by Sysmocom s.f.m.c. GmbH
+# (C) 2021 by sysmocom - s.f.m.c. GmbH
 # All Rights Reserved
 #
 # This program is free software: you can redistribute it and/or modify

pySim/runtime.py

@@ -1,4 +1,5 @@
 # coding=utf-8
+
 """Representation of the runtime state of an application like pySim-shell.
 """
 
@@ -25,7 +26,7 @@ from pySim.exceptions import *
 from pySim.filesystem import *
 from pySim.log import PySimLogger
 
-log = PySimLogger.get("RUNTIME")
+log = PySimLogger.get(__name__)
 
 def lchan_nr_from_cla(cla: int) -> int:
     """Resolve the logical channel number from the CLA byte."""
@@ -115,7 +116,7 @@ class RuntimeState:
             for a in aids_unknown:
                 log.info(" unknown: %s (EF.DIR)" % a)
         else:
-            log.warn("EF.DIR seems to be empty!")
+            log.warning("EF.DIR seems to be empty!")
 
         # Some card applications may not be registered in EF.DIR, we will actively
         # probe for those applications
@@ -556,8 +557,8 @@ class RuntimeLchan:
             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))
+            log.warning("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.

pySim/transport/__init__.py

@@ -3,18 +3,6 @@
 """ pySim: PCSC reader transport link base
 """
 
-import os
-import abc
-import argparse
-from typing import Optional, Tuple
-from construct import Construct
-from osmocom.utils import b2h, h2b, i2h, Hexstr
-
-from pySim.exceptions import *
-from pySim.utils import SwHexstr, SwMatchstr, ResTuple, sw_match, parse_command_apdu
-from pySim.cat import ProactiveCommand, CommandDetails, DeviceIdentities, Result
-
-#
 # Copyright (C) 2009-2010  Sylvain Munaut <tnt@246tNt.com>
 # Copyright (C) 2021-2023 Harald Welte <laforge@osmocom.org>
 #
@@ -30,8 +18,20 @@ from pySim.cat import ProactiveCommand, CommandDetails, DeviceIdentities, Result
 #
 # 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 abc
+import argparse
+from typing import Optional, Tuple
+from construct import Construct
+from osmocom.utils import b2h, h2b, i2h, Hexstr
+
+from pySim.exceptions import *
+from pySim.utils import SwHexstr, SwMatchstr, ResTuple, sw_match, parse_command_apdu
+from pySim.cat import ProactiveCommand, CommandDetails, DeviceIdentities, Result
+from pySim.log import PySimLogger
+
+log = PySimLogger.get(__name__)
 
 class ApduTracer:
     def trace_command(self, cmd):
@@ -46,11 +46,11 @@ class ApduTracer:
 class StdoutApduTracer(ApduTracer):
     """Minimalistic APDU tracer, printing commands to stdout."""
     def trace_response(self, cmd, sw, resp):
-        print("-> %s %s" % (cmd[:10], cmd[10:]))
-        print("<- %s: %s" % (sw, resp))
+        log.info("-> %s %s", cmd[:10], cmd[10:])
+        log.info("<- %s: %s", sw, resp)
 
     def trace_reset(self):
-        print("-- RESET")
+        log.info("-- RESET")
 
 class ProactiveHandler(abc.ABC):
     """Abstract base class representing the interface of some code that handles
@@ -177,7 +177,7 @@ class LinkBase(abc.ABC):
             if self.apdu_strict:
                 raise ValueError(exeption_str)
             else:
-                print('Warning: %s' % exeption_str)
+                log.warning(exeption_str)
 
         return (data, sw)
 
@@ -211,7 +211,7 @@ class LinkBase(abc.ABC):
             # parse the proactive command
             pcmd = ProactiveCommand()
             parsed = pcmd.from_tlv(h2b(fetch_rv[0]))
-            print("FETCH: %s (%s)" % (fetch_rv[0], type(parsed).__name__))
+            log.info("FETCH: %s (%s)", fetch_rv[0], type(parsed).__name__)
             if self.proactive_handler:
                 # Extension point: If this does return a list of TLV objects,
                 # they could be appended after the Result; if the first is a
@@ -361,13 +361,13 @@ def init_reader(opts, **kwargs) -> LinkBase:
         from pySim.transport.modem_atcmd import ModemATCommandLink
         sl = ModemATCommandLink(opts, **kwargs)
     else:  # Serial reader is default
-        print("No reader/driver specified; falling back to default (Serial reader)")
+        log.warning("No reader/driver specified; falling back to default (Serial reader)")
         from pySim.transport.serial import SerialSimLink
         sl = SerialSimLink(opts, **kwargs)
 
     if os.environ.get('PYSIM_INTEGRATION_TEST') == "1":
-        print("Using %s reader interface" % (sl.name))
+        log.info("Using %s reader interface" % (sl.name))
     else:
-        print("Using reader %s" % sl)
+        log.info("Using reader %s" % sl)
 
     return sl

pySim/transport/modem_atcmd.py

@@ -166,7 +166,7 @@ class ModemATCommandLink(LinkBaseTpdu):
 
         # Make sure that the response has format: b'+CSIM: %d,\"%s\"'
         try:
-            result = re.match(b'\+CSIM: (\d+),\"([0-9A-F]+)\"', rsp)
+            result = re.match(rb'\+CSIM: (\d+),\"([0-9A-F]+)\"', rsp)
             (_rsp_tpdu_len, rsp_tpdu) = result.groups()
         except Exception as exc:
             raise ReaderError('Failed to parse response from modem: %s' % rsp) from exc

pySim/ts_31_102.py

@@ -486,17 +486,17 @@ class EF_UST(EF_UServiceTable):
 # TS 31.103 Section 4.2.7 - *not* the same as DF.GSM/EF.ECC!
 class EF_ECC(LinFixedEF):
     _test_de_encode = [
-        ( '19f1ff01', { "call_code": "911f",
+        ( '19f1ff01', { "call_code": "911",
                         "service_category": { "police": True, "ambulance": False, "fire_brigade": False,
                                               "marine_guard": False, "mountain_rescue": False,
                                               "manual_ecall": False, "automatic_ecall": False } } ),
-        ( '19f3ff02', { "call_code": "913f",
+        ( '19f3ff02', { "call_code": "913",
                         "service_category": { "police": False, "ambulance": True, "fire_brigade": False,
                                               "marine_guard": False, "mountain_rescue": False,
                                               "manual_ecall": False, "automatic_ecall": False } } ),
     ]
     _test_no_pad = True
-    cc_construct = BcdAdapter(Rpad(Bytes(3)))
+    cc_construct = PaddedBcdAdapter(Rpad(Bytes(3)))
     category_construct = FlagsEnum(Byte, police=1, ambulance=2, fire_brigade=3, marine_guard=4,
                                    mountain_rescue=5, manual_ecall=6, automatic_ecall=7)
     alpha_construct = GsmOrUcs2Adapter(Rpad(GreedyBytes))
@@ -596,7 +596,7 @@ class EF_ICI(CyclicEF):
         self._construct = Struct('alpha_id'/Bytes(this._.total_len-28),
                                  'len_of_bcd_contents'/Int8ub,
                                  'ton_npi'/Int8ub,
-                                 'call_number'/BcdAdapter(Bytes(10)),
+                                 'call_number'/PaddedBcdAdapter(Rpad(Bytes(10))),
                                  'cap_cfg2_record_id'/Int8ub,
                                  'ext5_record_id'/Int8ub,
                                  'date_and_time'/BcdAdapter(Bytes(7)),
@@ -612,7 +612,7 @@ class EF_OCI(CyclicEF):
         self._construct = Struct('alpha_id'/Bytes(this._.total_len-27),
                                  'len_of_bcd_contents'/Int8ub,
                                  'ton_npi'/Int8ub,
-                                 'call_number'/BcdAdapter(Bytes(10)),
+                                 'call_number'/PaddedBcdAdapter(Rpad(Bytes(10))),
                                  'cap_cfg2_record_id'/Int8ub,
                                  'ext5_record_id'/Int8ub,
                                  'date_and_time'/BcdAdapter(Bytes(7)),
@@ -1118,7 +1118,7 @@ class EF_Routing_Indicator(TransparentEF):
         # responsibility of home network operator but BCD coding shall be used. If a network
         # operator decides to assign less than 4 digits to Routing Indicator, the remaining digits
         # shall be coded as "1111" to fill the 4 digits coding of Routing Indicator
-        self._construct = Struct('routing_indicator'/Rpad(BcdAdapter(Bytes(2)), 'f', 2),
+        self._construct = Struct('routing_indicator'/PaddedBcdAdapter(Rpad(Bytes(2))),
                                  'rfu'/Bytes(2))
 
 # TS 31.102 Section 4.4.11.13 (Rel 16)

pySim/ts_51_011.py

@@ -40,6 +40,7 @@ from osmocom.utils import *
 from osmocom.construct import *
 
 from pySim.utils import dec_iccid, enc_iccid, dec_imsi, enc_imsi, dec_plmn, enc_plmn, dec_xplmn_w_act
+from pySim.utils import bytes_for_nibbles
 from pySim.profile import CardProfile, CardProfileAddon
 from pySim.filesystem import *
 from pySim.ts_31_102_telecom import DF_PHONEBOOK, DF_MULTIMEDIA, DF_MCS, DF_V2X
@@ -151,7 +152,7 @@ class EF_ADN(LinFixedEF):
         self._construct = Struct('alpha_id'/COptional(GsmOrUcs2Adapter(Rpad(Bytes(this._.total_len-14)))),
                                  'len_of_bcd'/Int8ub,
                                  'ton_npi'/TonNpi,
-                                 'dialing_nr'/ExtendedBcdAdapter(BcdAdapter(Rpad(Bytes(10)))),
+                                 'dialing_nr'/ExtendedBcdAdapter(PaddedBcdAdapter(Rpad(Bytes(10)))),
                                  'cap_conf_id'/Int8ub,
                                  ext_name/Int8ub)
 
@@ -192,11 +193,11 @@ class EF_MSISDN(LinFixedEF):
         ( 'ffffffffffffffffffffffffffffffffffffffff04b12143f5ffffffffffffffffff',
             {"alpha_id": "", "len_of_bcd": 4, "ton_npi": {"ext": True, "type_of_number": "network_specific",
                                                           "numbering_plan_id": "isdn_e164"},
-             "dialing_nr": "12345f"}),
+             "dialing_nr": "12345"}),
         ( '456967656e65205275666e756d6d6572ffffffff0891947172199181f3ffffffffff',
             {"alpha_id": "Eigene Rufnummer", "len_of_bcd": 8, "ton_npi": {"ext": True, "type_of_number": "international",
                                                                           "numbering_plan_id": "isdn_e164"},
-             "dialing_nr": "4917279119183f"}),
+             "dialing_nr": "4917279119183"}),
     ]
 
     # Ensure deprecated representations still work
@@ -214,7 +215,7 @@ class EF_MSISDN(LinFixedEF):
         self._construct = Struct('alpha_id'/COptional(GsmOrUcs2Adapter(Rpad(Bytes(this._.total_len-14)))),
                                  'len_of_bcd'/Int8ub,
                                  'ton_npi'/TonNpi,
-                                 'dialing_nr'/ExtendedBcdAdapter(BcdAdapter(Rpad(Bytes(10)))),
+                                 'dialing_nr'/ExtendedBcdAdapter(PaddedBcdAdapter(Rpad(Bytes(10)))),
                                   Padding(2, pattern=b'\xff'))
 
     # Maintain compatibility with deprecated representations
@@ -239,11 +240,20 @@ class EF_MSISDN(LinFixedEF):
 
 # TS 51.011 Section 10.5.6
 class EF_SMSP(LinFixedEF):
-    # FIXME: re-encode fails / missing alpha_id at start of output
-    _test_decode = [
+    _test_de_encode = [
+        ( '534d5343ffffffffffffffffffffffffe1ffffffffffffffffffffffff0891945197109099f9ffffff0000a9',
+          { "alpha_id": "SMSC", "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": True },
+                                                               "tp_pid": True, "tp_dcs": True, "tp_vp": False },
             "tp_dest_addr": { "length": 255, "ton_npi": { "ext": True, "type_of_number": "reserved_for_extension",
                                                           "numbering_plan_id": "reserved_for_extension" },
                               "call_number": "" },
@@ -277,12 +287,26 @@ class EF_SMSP(LinFixedEF):
             else:
                 raise ValueError
 
+    @staticmethod
+    def sc_addr_len(ctx):
+        """Compute the length field for an address field (like TP-DestAddr or TP-ScAddr)."""
+        if not hasattr(ctx, 'call_number') or len(ctx.call_number) == 0:
+            return 0xff
+        else:
+            return bytes_for_nibbles(len(ctx.call_number)) + 1
+
     def __init__(self, fid='6f42', sfid=None, name='EF.SMSP', desc='Short message service parameters', **kwargs):
         super().__init__(fid, sfid=sfid, name=name, desc=desc, rec_len=(28, None), **kwargs)
-        ScAddr = Struct('length'/Int8ub, 'ton_npi'/TonNpi, 'call_number'/BcdAdapter(Rpad(Bytes(10))))
+        ScAddr = Struct('length'/Rebuild(Int8ub, lambda ctx: EF_SMSP.sc_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)))),
-                                 'parameter_indicators'/InvertAdapter(FlagsEnum(Byte, tp_dest_addr=1, tp_sc_addr=2,
-                                                                                tp_pid=3, tp_dcs=4, tp_vp=5)),
+                                 'parameter_indicators'/InvertAdapter(BitStruct(
+                                                                        Const(7, BitsInteger(3)),
+                                                                        'tp_vp'/Flag,
+                                                                        'tp_dcs'/Flag,
+                                                                        'tp_pid'/Flag,
+                                                                        'tp_sc_addr'/Flag,
+                                                                        'tp_dest_addr'/Flag)),
                                  'tp_dest_addr'/ScAddr,
                                  'tp_sc_addr'/ScAddr,
 
@@ -637,12 +661,12 @@ class EF_AD(TransparentEF):
 # TS 51.011 Section 10.3.20 / 10.3.22
 class EF_VGCS(TransRecEF):
     _test_de_encode = [
-            ( "92f9ffff", "299fffff" ),
+            ( "92f9ffff", "299" ),
         ]
     def __init__(self, fid='6fb1', sfid=None, name='EF.VGCS', size=(4, 200), rec_len=4,
                  desc='Voice Group Call Service', **kwargs):
         super().__init__(fid, sfid=sfid, name=name, desc=desc, size=size, rec_len=rec_len, **kwargs)
-        self._construct = BcdAdapter(Bytes(4))
+        self._construct = PaddedBcdAdapter(Rpad(Bytes(4)))
 
 # TS 51.011 Section 10.3.21 / 10.3.23
 class EF_VGCSS(TransparentEF):

pySim/utils.py

@@ -526,6 +526,13 @@ def expand_hex(hexstring, length):
     # no change
     return hexstring
 
+def bytes_for_nibbles(num_nibbles: int) -> int:
+    """compute the number of bytes needed to store the given number of nibbles."""
+    n_bytes = num_nibbles // 2
+    if num_nibbles & 1:
+        n_bytes += 1
+    return n_bytes
+
 
 def boxed_heading_str(heading, width=80):
     """Generate a string that contains a boxed heading."""
@@ -624,15 +631,17 @@ def decomposeATR(atr_txt):
         Returns:
             dictionary of field and values
 
-        >>> decomposeATR("3B A7 00 40 18 80 65 A2 08 01 01 52")
-    { 'T0': {'value': 167},
-      'TB': {1: {'value': 0}},
-      'TC': {2: {'value': 24}},
-      'TD': {1: {'value': 64}},
-      'TS': {'value': 59},
-      'atr': [59, 167, 0, 64, 24, 128, 101, 162, 8, 1, 1, 82],
-      'hb': {'value': [128, 101, 162, 8, 1, 1, 82]},
-      'hbn': 7}
+        Example::
+
+          >>> decomposeATR("3B A7 00 40 18 80 65 A2 08 01 01 52")
+            { 'T0': {'value': 167},
+              'TB': {1: {'value': 0}},
+              'TC': {2: {'value': 24}},
+              'TD': {1: {'value': 64}},
+              'TS': {'value': 59},
+              'atr': [59, 167, 0, 64, 24, 128, 101, 162, 8, 1, 1, 82],
+              'hb': {'value': [128, 101, 162, 8, 1, 1, 82]},
+              'hbn': 7}
     """
     ATR_PROTOCOL_TYPE_T0 = 0
     atr_txt = normalizeATR(atr_txt)

requirements.txt

@@ -5,7 +5,7 @@ cmd2>=2.6.2,<3.0
 jsonpath-ng
 construct>=2.10.70
 bidict
-pyosmocom>=0.0.9
+pyosmocom>=0.0.12
 pyyaml>=5.1
 termcolor
 colorlog
@@ -15,3 +15,4 @@ git+https://github.com/osmocom/asn1tools
 packaging
 git+https://github.com/hologram-io/smpp.pdu
 smpp.twisted3 @ git+https://github.com/jookies/smpp.twisted
+smpplib

setup.py

@@ -25,7 +25,7 @@ setup(
         "jsonpath-ng",
         "construct >= 2.10.70",
         "bidict",
-        "pyosmocom >= 0.0.9",
+        "pyosmocom >= 0.0.12",
         "pyyaml >= 5.1",
         "termcolor",
         "colorlog",
@@ -55,6 +55,10 @@ setup(
             "service-identity",
             "pyopenssl",
             "requests",
+            "smpplib",
+        ],
+        "CardKeyProviderPgsql": [
+            "psycopg2-binary",
         ]
     },
 )

tests/card_sanitizer/card_backup_3b9f96801f878031e073fe211b674a357530350265f8_8949440000001155314.script

@@ -2200,9 +2200,9 @@ update_record 6 fe0112ffb53e96e5ff99731d51ad7beafd0e23ffffffffffffffffffffffffff
 update_record 7 fe02101da012f436d06824ecdd15050419ff9affffffffffffffffffffffffffffffff
 update_record 8 fe02116929a373388ac904aff57ff57f6b3431ffffffffffffffffffffffffffffffff
 update_record 9 fe0212a99245a5dc814e2f4c1aa908e9946e03ffffffffffffffffffffffffffffffff
-update_record 10 fe0310521312c05a9aea93d70d44405172a580ffffffffffffffffffffffffffffffff
-update_record 11 fe0311a9e45c72d45abde7db74261ee0c11b1bffffffffffffffffffffffffffffffff
-update_record 12 fe0312867ba36b5873d60ea8b2cdcf3c0ddddaffffffffffffffffffffffffffffffff
+update_record 10 fe03601111111111111111111111111111111111111111111111111111111111111111
+update_record 11 fe03612222222222222222222222222222222222222222222222222222222222222222
+update_record 12 fe03623333333333333333333333333333333333333333333333333333333333333333
 #
 ################################################################################
 # MF/DF.SYSTEM/EF.SIM_AUTH_COUNTER                                             #

tests/pySim-prog_test/Fairwaves-SIM.ok

@@ -6,6 +6,7 @@ IMSI: 001010000000111
 GID1: ffffffffffffffff
 GID2: ffffffffffffffff
 SMSP: e1ffffffffffffffffffffffff0581005155f5ffffffffffff000000ffffffffffffffffffffffffffff
+SMSC: 0015555
 SPN: Fairwaves
 Show in HPLMN: False
 Hide in OPLMN: False

tests/pySim-prog_test/Wavemobile-SIM.ok

@@ -6,6 +6,7 @@ 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
+SMSC: 0015555
 SPN: wavemobile
 Show in HPLMN: False
 Hide in OPLMN: False

tests/pySim-prog_test/fakemagicsim.ok

@@ -6,6 +6,7 @@ IMSI: 001010000000102
 GID1: Can't read file -- SW match failed! Expected 9000 and got 9404.
 GID2: Can't read file -- SW match failed! Expected 9000 and got 9404.
 SMSP: ffffffffffffffffffffffffe1ffffffffffffffffffffffff0581005155f5ffffffffffff000000
+SMSC: 0015555
 SPN: Magic
 Show in HPLMN: True
 Hide in OPLMN: False

tests/pySim-prog_test/pySim-prog_test.sh

@@ -2,7 +2,7 @@
 
 # Utility to verify the functionality of pySim-prog.py
 #
-# (C) 2018 by Sysmocom s.f.m.c. GmbH
+# (C) 2018 by sysmocom - s.f.m.c. GmbH
 # All Rights Reserved
 #
 # Author: Philipp Maier

tests/pySim-prog_test/sysmoISIM-SJA2.ok

@@ -6,6 +6,7 @@ IMSI: 001010000000102
 GID1: ffffffffffffffffffff
 GID2: ffffffffffffffffffff
 SMSP: ffffffffffffffffffffffffffffffffffffffffffffffffe1ffffffffffffffffffffffff0581005155f5ffffffffffff000000
+SMSC: 0015555
 SPN: Magic
 Show in HPLMN: True
 Hide in OPLMN: True

tests/pySim-prog_test/sysmoISIM-SJA5.ok

@@ -6,6 +6,7 @@ IMSI: 001010000000102
 GID1: ffffffffffffffffffff
 GID2: ffffffffffffffffffff
 SMSP: ffffffffffffffffffffffffffffffffffffffffffffffffe1ffffffffffffffffffffffff0581005155f5ffffffffffff000000
+SMSC: 0015555
 SPN: Magic
 Show in HPLMN: True
 Hide in OPLMN: True

tests/pySim-prog_test/sysmoUSIM-SJS1.ok

@@ -6,6 +6,7 @@ IMSI: 001010000000102
 GID1: ffffffffffffffffffff
 GID2: ffffffffffffffffffff
 SMSP: ffffffffffffffffffffffffffffffffffffffffffffffffe1ffffffffffffffffffffffff0581005155f5ffffffffffff000000
+SMSC: 0015555
 SPN: Magic
 Show in HPLMN: True
 Hide in OPLMN: True

tests/pySim-prog_test/sysmosim-gr1.ok

@@ -6,6 +6,7 @@ IMSI: 001010000000102
 GID1: Can't read file -- SW match failed! Expected 9000 and got 9404.
 GID2: Can't read file -- SW match failed! Expected 9000 and got 9404.
 SMSP: ffffffffffffffffffffffffe1ffffffffffffffffffffffff0581005155f5ffffffffffff000000
+SMSC: 0015555
 SPN: Not available
 Show in HPLMN: False
 Hide in OPLMN: False

tests/pySim-shell_test/euicc/get_profiles_info.ok

@@ -15,7 +15,7 @@
         },
         {
             "profile_info": {
-                "iccid": "8949449999999990031f",
+                "iccid": "8949449999999990031",
                 "isdp_aid": "a0000005591010ffffffff8900001200",
                 "profile_state": "disabled",
                 "service_provider_name": "OsmocomSPN",

tests/pySim-shell_test/euicc/test.py

@@ -23,7 +23,7 @@ import os
 import json
 from utils import *
 
-# This testcase requires a sysmoEUICC1-C2T with the test prfile TS48V1-B-UNIQUE (ICCID 8949449999999990031f)
+# This testcase requires a sysmoEUICC1-C2T with the test prfile TS48V1-B-UNIQUE (ICCID 8949449999999990031)
 # installed, and in disabled state. Also the profile must be installed in such a way that notifications are
 # generated when the profile is disabled or enabled (ProfileMetadata)
 
@@ -56,7 +56,7 @@ class test_case(UnittestUtils):
         self.runPySimShell(cardname, "test_enable_disable_profile.script")
         self.assertEqualFiles("enable_disable_profile.tmp")
 
-    def test_enable_disable_profile(self):
+    def test_set_nickname(self):
         cardname = 'sysmoEUICC1-C2T'
 
         self.runPySimShell(cardname, "test_set_nickname.script")

tests/pySim-shell_test/euicc/test_enable_disable_profile.script

@@ -3,6 +3,9 @@ set echo true
 
 select ADF.ISD-R
 
+# Ensure that the test-profile we intend to test with is actually enabled
+enable_profile --iccid 89000123456789012341
+
 # by ICCID (pre-installed test profile on sysmoEUICC1-C2T)
 disable_profile --iccid 89000123456789012341 > enable_disable_profile.tmp
 enable_profile --iccid 89000123456789012341 >> enable_disable_profile.tmp

tests/pySim-shell_test/euicc/test_gen_notif.script

@@ -3,6 +3,11 @@ set echo true
 
 select ADF.ISD-R
 
-# Generate two (additional) notifications by quickly enabeling the test profile
-enable_profile --iccid 8949449999999990031f
+# Ensure that the test-profile is actually enabled. (In case te test-profile
+# was disabled, a notification may be generated. The testcase should tolerate
+# that)
+enable_profile --iccid 89000123456789012341
+
+# Generate two (additional) notifications by quickly enabeling the test profile
+enable_profile --iccid 8949449999999990031
 enable_profile --iccid 89000123456789012341

tests/pySim-shell_test/euicc/test_get_profiles_info.script

@@ -1,5 +1,10 @@
 set debug true
 set echo true
 
+# The output of get_profiles_info will also include the "profile_state", which
+# can be either "enabled" or "disabled". Ensure that the correct profile is
+# enabled.
+enable_profile --iccid 89000123456789012341
+
 select ADF.ISD-R
 get_profiles_info > get_profiles_info.tmp

tests/pySim-shell_test/file_content/test_record_uicc.ok

@@ -19,7 +19,7 @@
             "type_of_number": "reserved_for_extension",
             "numbering_plan_id": "reserved_for_extension"
         },
-        "dialing_nr": "123456",
+        "dialing_nr": "1234567",
         "cap_conf_id": 42,
         "ext4_record_id": 23
     },
@@ -67,7 +67,7 @@
             "type_of_number": "reserved_for_extension",
             "numbering_plan_id": "reserved_for_extension"
         },
-        "dialing_nr": "123456",
+        "dialing_nr": "1234567",
         "cap_conf_id": 42,
         "ext4_record_id": 23
     },
@@ -127,7 +127,7 @@
             "type_of_number": "reserved_for_extension",
             "numbering_plan_id": "reserved_for_extension"
         },
-        "dialing_nr": "123456",
+        "dialing_nr": "1234567",
         "cap_conf_id": 42,
         "ext4_record_id": 23
     }
@@ -140,7 +140,7 @@
         "type_of_number": "reserved_for_extension",
         "numbering_plan_id": "reserved_for_extension"
     },
-    "dialing_nr": "123456",
+    "dialing_nr": "1234567",
     "cap_conf_id": 42,
     "ext4_record_id": 23
 }

tests/pySim-smpp2sim_test/pySim-smpp2sim_test.sh

@@ -0,0 +1,216 @@
+#!/bin/bash
+
+# Utility to verify the functionality of pySim-trace.py
+#
+# (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/>.
+
+PYSIM_SHELL=./pySim-shell.py
+PYSIM_SHELL_LOG=./pySim-shell.log
+PYSIM_SMPP2SIM=./pySim-smpp2sim.py
+PYSIM_SMPP2SIM_LOG=./pySim-smpp2sim.log
+PYSIM_SMPP2SIM_PORT=2775
+PYSIM_SMPP2SIM_TIMEOUT=10
+PYSIM_SMPPOTATOOL=./contrib/smpp-ota-tool.py
+PYSIM_SMPPOTATOOL_LOG=./smpp-ota-tool.log
+
+function dump_logs {
+    echo ""
+    echo "$PYSIM_SMPPOTATOOL_LOG"
+    echo "------------8<------------"
+    cat $PYSIM_SMPPOTATOOL_LOG
+    echo "------------8<------------"
+    echo ""
+    echo "$PYSIM_SMPP2SIM_LOG"
+    echo "------------8<------------"
+    cat $PYSIM_SMPP2SIM_LOG
+    echo "------------8<------------"
+}
+
+function send_test_request {
+    echo ""
+    echo "Sending request to SMPP server:"
+    C_APDU=$1
+    R_APDU_EXPECTED=$2
+
+    echo "Sending: $C_APDU"
+    COMMANDLINE="$PYSIM_SMPPOTATOOL --verbose --port $PYSIM_SMPP2SIM_PORT --kic $KIC --kid $KID --kic-idx $KEY_INDEX --kid-idx $KEY_INDEX --algo-crypt $ALGO_CRYPT --algo-auth $ALGO_AUTH --tar $TAR --apdu $C_APDU"
+    echo "Commandline: $COMMANDLINE"
+    R_APDU=`$COMMANDLINE 2> $PYSIM_SMPPOTATOOL_LOG`
+    if [ $? -ne 0 ]; then
+	echo "Unable to send request! -- failed!"
+	dump_logs
+	exit 1
+    fi
+    echo ""
+    echo "Got response from SMPP server:"
+    echo "Sent: $C_APDU"
+    echo "Received: $R_APDU"
+    echo "Expected: $R_APDU_EXPECTED"
+    if [ "$R_APDU" != "$R_APDU_EXPECTED" ]; then
+	echo "Response does not match the expected response! -- failed!"
+	dump_logs
+	exit 1
+    fi
+    echo "Response matches the expected response -- success!"
+}
+
+function start_smpp_server {
+    PCSC_READER=$1
+    echo ""
+    echo "Starting SMPP server:"
+
+    # Start the SMPP server
+    COMMANDLINE="$PYSIM_SMPP2SIM -p $PCSC_READER --smpp-bind-port $PYSIM_SMPP2SIM_PORT --apdu-trace"
+    echo "Commandline: $COMMANDLINE"
+    $COMMANDLINE > $PYSIM_SMPP2SIM_LOG 2>&1 &
+    PYSIM_SMPP2SIM_PID=$!
+    trap 'kill $PYSIM_SMPP2SIM_PID' EXIT
+    echo "SMPP server started (PID=$PYSIM_SMPP2SIM_PID)"
+
+    # Wait until the SMPP server is reachable
+    RC=1
+    RETRY_COUNT=0
+    while [ $RC -ne 0 ]; do
+	nc -z localhost $PYSIM_SMPP2SIM_PORT
+	RC=$?
+	((RETRY_COUNT++))
+	if [ $RETRY_COUNT -gt $PYSIM_SMPP2SIM_TIMEOUT ]; then
+	    echo "SMPP server not reachable (port=$PYSIM_SMPP2SIM_PORT) -- abort"
+	    dump_logs
+	    exit 1
+	fi
+	sleep 1
+    done
+    echo "SMPP server reachable (port=$PYSIM_SMPP2SIM_PORT)"
+}
+
+function stop_smpp_server {
+    echo ""
+    echo "Stopping SMPP server:"
+    kill $PYSIM_SMPP2SIM_PID
+    echo "SMPP server stopped (PID=$PYSIM_SMPP2SIM_PID)"
+    trap EXIT
+}
+
+function find_card_by_iccid_or_eid {
+    ICCID=$1
+    EID=$2
+    echo ""
+    echo "Searching for card:"
+    echo "ICCID: \"$ICCID\""
+    if [ -n "$EID" ]; then
+	echo "EID: \"$EID\""
+    fi
+
+    # Determine number of available PCSC readers
+    PCSC_READER_COUNT=`pcsc_scan -rn | wc -l`
+
+    # In case an EID is set, search for a card with that EID first
+    if [ -n "$EID" ]; then
+	for PCSC_READER in $(seq 0 $(($PCSC_READER_COUNT-1))); do
+	    echo "probing card (eID) in reader $PCSC_READER ..."
+	    RESULT_JSON=`$PYSIM_SHELL -p $PCSC_READER --noprompt -e "select ADF.ISD-R" -e "get_eid" 2> /dev/null | tail -3`
+	    echo $RESULT_JSON | grep $EID > /dev/null
+	    if [ $? -eq 0 ]; then
+		echo "Found card (eID) in reader $PCSC_READER"
+		return $PCSC_READER
+	    fi
+	done
+    fi
+
+    # Search for card with the given ICCID
+    if [ -z "$ICCID" ]; then
+	echo "invalid ICCID, zero length ICCID is not allowed! -- abort"
+	exit 1
+    fi
+    for PCSC_READER in $(seq 0 $(($PCSC_READER_COUNT-1))); do
+	echo "probing card (ICCID) in reader $PCSC_READER ..."
+	RESULT_JSON=`$PYSIM_SHELL -p $PCSC_READER --noprompt -e "select EF.ICCID" -e "read_binary_decoded" 2> /dev/null | tail -3`
+	echo $RESULT_JSON | grep $ICCID > /dev/null
+	if [ $? -eq 0 ]; then
+	    echo "Found card (by ICCID) in reader $PCSC_READER"
+	    return $PCSC_READER
+	fi
+    done
+
+    echo "Card not found -- abort"
+    exit 1
+}
+
+function enable_profile {
+    PCSC_READER=$1
+    ICCID=$2
+    EID=$3
+    if [ -z "$EID" ]; then
+	# This is no eUICC, nothing to enable
+	return 0
+    fi
+
+    # Check if the profile is already enabled
+    RESULT_JSON=`$PYSIM_SHELL -p $PCSC_READER --noprompt -e "select EF.ICCID" -e "read_binary_decoded" 2> /dev/null | tail -3`
+    ICCID_ENABLED=`echo $RESULT_JSON | jq -r '.iccid'`
+    if [ $ICCID != $ICCID_ENABLED ]; then
+	# Disable the currentle enabled profile
+	echo ""
+	echo "Disabeling currently enabled profile:"
+	echo "ICCID: \"$ICCID\""
+	RESULT_JSON=`$PYSIM_SHELL -p $PCSC_READER --noprompt -e "select ADF.ISD-R" -e "disable_profile --iccid $ICCID_ENABLED" 2> /dev/null | tail -3`
+	echo $RESULT_JSON | grep "ok" > /dev/null
+	if [ $? -ne 0 ]; then
+	    echo "unable to disable profile with \"$ICCID_ENABLED\""
+	    exit 1
+	fi
+	echo "profile disabled"
+
+	# Enable the profile we intend to test with
+	echo ""
+	echo "Enabeling profile:"
+	echo "ICCID: \"$ICCID\""
+	RESULT_JSON=`$PYSIM_SHELL -p $PCSC_READER --noprompt -e "select ADF.ISD-R" -e "enable_profile --iccid $ICCID" 2> /dev/null | tail -3`
+	echo $RESULT_JSON | grep "ok\|profileNotInDisabledState" > /dev/null
+	if [ $? -ne 0 ]; then
+	    echo "unable to enable profile with \"$ICCID\""
+	    exit 1
+	fi
+	echo "profile enabled"
+    fi
+}
+
+export PYTHONPATH=./
+
+echo "pySim-smpp2sim_test - a test program to test pySim-smpp2sim.py"
+echo "=============================================================="
+
+TESTCASE_DIR=`dirname $0`
+for TEST_CONFIG_FILE in $TESTCASE_DIR/testcase_*.cfg ; do
+    echo ""
+    echo "running testcase: $TEST_CONFIG_FILE"
+    . $TEST_CONFIG_FILE
+    find_card_by_iccid_or_eid $ICCID $EID
+    PCSC_READER=$?
+    enable_profile $PCSC_READER $ICCID $EID
+    start_smpp_server $PCSC_READER
+    send_test_request $APDU "$EXPECTED_RESPONSE"
+    stop_smpp_server
+    echo ""
+    echo "testcase ok"
+    echo "--------------------------------------------------------------"
+done
+
+echo "done."

tests/pySim-smpp2sim_test/testcase_3des_cbc2_rfm.cfg

@@ -0,0 +1,17 @@
+# Preparation:
+# This testcase executes against a sysmoISIM-SJA5 card. For the testcase, the
+# key configuration on the card may be used as it is.
+
+# Card parameter:
+ICCID="8949440000001155314" # <-- change to the ICCID of your card!
+EID=""
+KIC='51D4FC44BCBA7C4589DFADA3297720AF' # <-- change to the KIC1 of your card!
+KID='0449699C472CE71E2FB7B56245EF7684' # <-- change to the KID1 of your card!
+KEY_INDEX=1
+ALGO_CRYPT=triple_des_cbc2
+ALGO_AUTH=triple_des_cbc2
+TAR='B00010'
+
+# Testcase: Send OTA-SMS that selects DF.GSM and returns the select response
+APDU='A0A40000027F20A0C0000016'
+EXPECTED_RESPONSE='0000ffff7f2002000000000009b106350400838a838a 9000'

tests/pySim-smpp2sim_test/testcase_aes128_cbc_cmac_rfm.cfg

@@ -0,0 +1,19 @@
+# Preparation:
+# This testcase executes against a sysmoEUICC1-C2T, which is equipped with the
+# TS48V1-B-UNIQUE test profile from https://test.rsp.sysmocom.de/ (Activation
+# code: 1$smdpp.test.rsp.sysmocom.de$TS48V1-B-UNIQUE). This testprofile must be
+# present on the eUICC before this testcase can be executed.
+
+# Card parameter:
+ICCID="8949449999999990031"
+EID="89049044900000000000000000102355" # <-- change to the EID of your card!
+KIC='66778899aabbccdd1122334455eeff10'
+KID='112233445566778899aabbccddeeff10'
+KEY_INDEX=2
+ALGO_CRYPT=aes_cbc
+ALGO_AUTH=aes_cmac
+TAR='b00120'
+
+# Testcase: Send OTA-SMS that selects DF.ICCID and returns the select response
+APDU='00a40004022fe200C000001d'
+EXPECTED_RESPONSE='621b8202412183022fe2a503d001408a01058b032f06038002000a8800 9000'

tests/pySim-smpp2sim_test/testcase_aes256_cbc_cmac_rfm.cfg

@@ -0,0 +1,28 @@
+# Preparation:
+# This testcase executes against a sysmoISIM-SJA5 card. Since this card model is
+# shipped with a classic DES key configuration, it is necessary to provision
+# AES128 test keys before this testcase may be executed. The the following
+# pySim-shell command sequence may be used:
+#
+# verify_adm 34173960 # <-- change to the ADM key of your card!
+# select /DF.SYSTEM/EF.0348_KEY
+# update_record 10 fe03601111111111111111111111111111111111111111111111111111111111111111
+# update_record 11 fe03612222222222222222222222222222222222222222222222222222222222222222
+# update_record 12 fe03623333333333333333333333333333333333333333333333333333333333333333
+#
+# This overwrites one of the already existing 3DES SCP02 key (KVN 47) and replaces it
+# with an AES256 SCP80 key (KVN 3).
+
+# Card parameter:
+ICCID="8949440000001155314" # <-- change to the ICCID of your card!
+EID=""
+KIC='1111111111111111111111111111111111111111111111111111111111111111'
+KID='2222222222222222222222222222222222222222222222222222222222222222'
+KEY_INDEX=3
+ALGO_CRYPT=aes_cbc
+ALGO_AUTH=aes_cmac
+TAR='B00010'
+
+# Testcase: Send OTA-SMS that selects DF.GSM and returns the select response
+APDU='A0A40000027F20A0C0000016'
+EXPECTED_RESPONSE='0000ffff7f2002000000000009b106350400838a838a 9000'

tests/pySim-trace_test/pySim-trace_test.sh

@@ -2,7 +2,7 @@
 
 # Utility to verify the functionality of pySim-trace.py
 #
-# (C) 2023 by Sysmocom s.f.m.c. GmbH
+# (C) 2023 by sysmocom - s.f.m.c. GmbH
 # All Rights Reserved
 #
 # Author: Philipp Maier

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

@@ -143,7 +143,7 @@ CardReset(3b9f96801f878031e073fe211b674a4c753034054ba9)
 ===============================
 00 SEARCH RECORD    MF/ADF.USIM/EF.SMSP                 01       9000 {"cmd": {"file": "currently_selected_ef", "mode": "forward_search", "record_number": 1, "search_string": "ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff"}, "rsp": {"body": [2], "sw": "9000"}}
 ===============================
-00 READ RECORD      MF/ADF.USIM/EF.SMSP                 01       9000 {"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": 5, "ton_npi": {"ext": true, "type_of_number": "unknown", "numbering_plan_id": "isdn_e164"}, "call_number": "0015555f"}, "tp_pid": "00", "tp_dcs": "00", "tp_vp_minutes": 5}
+00 READ RECORD      MF/ADF.USIM/EF.SMSP                 01       9000 {"alpha_id": "", "parameter_indicators": {"tp_vp": true, "tp_dcs": true, "tp_pid": true, "tp_sc_addr": true, "tp_dest_addr": false}, "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": 5, "ton_npi": {"ext": true, "type_of_number": "unknown", "numbering_plan_id": "isdn_e164"}, "call_number": "0015555"}, "tp_pid": "00", "tp_dcs": "00", "tp_vp_minutes": 5}
 ===============================
 00 SEARCH RECORD    MF/ADF.USIM/EF.SMS                  01       9000 {"cmd": {"file": "currently_selected_ef", "mode": "forward_search", "record_number": 1, "search_string": "00ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff"}, "rsp": {"body": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30], "sw": "9000"}}
 ===============================

tests/unittests/test_esim_saip.py

@@ -90,5 +90,34 @@ class OidTest(unittest.TestCase):
         self.assertTrue(oid.OID('1.0.1') > oid.OID('1.0'))
         self.assertTrue(oid.OID('1.0.2') > oid.OID('1.0.1'))
 
+class NonMatchTest(unittest.TestCase):
+    def test_nonmatch(self):
+        # non-matches before, in between and after matches
+        match_list = [Match(a=10, b=10, size=5), Match(a=20, b=20, size=4)]
+        nm_list = NonMatch.from_matchlist(match_list, 26)
+        self.assertEqual(nm_list, [NonMatch(a=0, b=0, size=10), NonMatch(a=15, b=15, size=5),
+                                   NonMatch(a=24, b=24, size=2)])
+
+    def test_nonmatch_beg(self):
+        # single match at beginning
+        match_list = [Match(a=0, b=0, size=5)]
+        nm_list = NonMatch.from_matchlist(match_list, 20)
+        self.assertEqual(nm_list, [NonMatch(a=5, b=5, size=15)])
+
+    def test_nonmatch_end(self):
+        # single match at end
+        match_list = [Match(a=19, b=19, size=5)]
+        nm_list = NonMatch.from_matchlist(match_list, 24)
+        self.assertEqual(nm_list, [NonMatch(a=0, b=0, size=19)])
+
+    def test_nonmatch_none(self):
+        # no match at all
+        match_list = []
+        nm_list = NonMatch.from_matchlist(match_list, 24)
+        self.assertEqual(nm_list, [NonMatch(a=0, b=0, size=24)])
+
+
+
+
 if __name__ == "__main__":
 	unittest.main()

tests/unittests/test_files.py

@@ -96,7 +96,7 @@ class LinFixed_Test(unittest.TestCase):
                         inst = c()
                         encoded, rec_num, decoded = self._parse_t(t)
                         logging.debug("Testing encode of %s", name)
-                        re_enc = inst.encode_record_hex(decoded, rec_num)
+                        re_enc = inst.encode_record_hex(decoded, rec_num, len(encoded)//2)
                         self.assertEqual(encoded.upper(), re_enc.upper())
 
     def test_de_encode_record(self):
@@ -122,7 +122,7 @@ class LinFixed_Test(unittest.TestCase):
                         self.assertEqual(decoded, re_dec)
                         # re-encode the decoded data
                         logging.debug("Testing re-encode of %s", name)
-                        re_enc = inst.encode_record_hex(re_dec, rec_num)
+                        re_enc = inst.encode_record_hex(re_dec, rec_num, len(encoded)//2)
                         self.assertEqual(encoded.upper(), re_enc.upper())
                     if hasattr(c, '_test_no_pad') and c._test_no_pad:
                         continue
@@ -196,7 +196,7 @@ class TransRecEF_Test(unittest.TestCase):
                         self.assertEqual(decoded, re_dec)
                         # re-encode the decoded data
                         logging.debug("Testing re-encode of %s", name)
-                        re_enc = inst.encode_record_hex(re_dec)
+                        re_enc = inst.encode_record_hex(re_dec, len(encoded)//2)
                         self.assertEqual(encoded.upper(), re_enc.upper())
                     # there's no point in testing padded input, as TransRecEF have a fixed record
                     # size and we cannot ever receive more input data than that size.
@@ -256,8 +256,8 @@ class TransparentEF_Test(unittest.TestCase):
                         encoded = t[0]
                         decoded = t[1]
                         logging.debug("Testing encode of %s", name)
-                        re_dec = inst.decode_hex(encoded)
-                        self.assertEqual(decoded, re_dec)
+                        re_enc = inst.encode_hex(decoded, len(encoded)//2)
+                        self.assertEqual(encoded, re_enc)
 
     def test_de_encode_file(self):
         """Test the decoder and encoder for a transparent EF.  Performs first a decoder
@@ -280,7 +280,7 @@ class TransparentEF_Test(unittest.TestCase):
                         self.assertEqual(decoded, re_dec)
                         logging.debug("Testing re-encode of %s", name)
                         re_dec = inst.decode_hex(encoded)
-                        re_enc = inst.encode_hex(re_dec)
+                        re_enc = inst.encode_hex(re_dec, len(encoded)//2)
                         self.assertEqual(encoded.upper(), re_enc.upper())
                     if hasattr(c, '_test_no_pad') and c._test_no_pad:
                         continue

tests/unittests/test_log.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 
-# (C) 2025 by Sysmocom s.f.m.c. GmbH
+# (C) 2025 by sysmocom - s.f.m.c. GmbH
 # All Rights Reserved
 #
 # Author: Philipp Maier <pmaier@sysmocom.de>
@@ -25,7 +25,7 @@ import io
 import sys
 from inspect import currentframe, getframeinfo
 
-log = PySimLogger.get("TEST")
+log = PySimLogger.get(__name__)
 
 TEST_MSG_DEBUG = "this is a debug message"
 TEST_MSG_INFO = "this is an info message"
@@ -82,15 +82,15 @@ class PySimLogger_Test(unittest.TestCase):
         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
+        expected_message = __name__ + "." + str(getframeinfo(frame).lineno + 1) + " -- DEBUG: " + TEST_MSG_DEBUG
         log.debug(TEST_MSG_DEBUG)
-        expected_message = __name__ + "." + str(getframeinfo(frame).lineno + 1) + " -- TEST - INFO: " + TEST_MSG_INFO
+        expected_message = __name__ + "." + str(getframeinfo(frame).lineno + 1) + " -- INFO: " + TEST_MSG_INFO
         log.info(TEST_MSG_INFO)
-        expected_message = __name__ + "." + str(getframeinfo(frame).lineno + 1) + " -- TEST - WARNING: " + TEST_MSG_WARNING
+        expected_message = __name__ + "." + str(getframeinfo(frame).lineno + 1) + " -- WARNING: " + TEST_MSG_WARNING
         log.warning(TEST_MSG_WARNING)
-        expected_message = __name__ + "." + str(getframeinfo(frame).lineno + 1) + " -- TEST - ERROR: " + TEST_MSG_ERROR
+        expected_message = __name__ + "." + str(getframeinfo(frame).lineno + 1) + " -- ERROR: " + TEST_MSG_ERROR
         log.error(TEST_MSG_ERROR)
-        expected_message = __name__ + "." + str(getframeinfo(frame).lineno + 1) + " -- TEST - CRITICAL: " + TEST_MSG_CRITICAL
+        expected_message = __name__ + "." + str(getframeinfo(frame).lineno + 1) + " -- CRITICAL: " + TEST_MSG_CRITICAL
         log.critical(TEST_MSG_CRITICAL)
 
     def test_04_level(self):

tests/unittests/test_utils.py

@@ -79,6 +79,30 @@ class DecTestCase(unittest.TestCase):
 	def testDecMNCfromPLMN_unused_str(self):
 		self.assertEqual(utils.dec_mnc_from_plmn_str("00f0ff"), "")
 
+	def testEncImsi(self):
+		#Type IMSI, odd number of identity digits
+		self.assertEqual(utils.enc_imsi("228062800000208"), "082982608200002080")
+		self.assertEqual(utils.enc_imsi("001010000123456"), "080910100000214365")
+		self.assertEqual(utils.enc_imsi("0010100001234"), "0709101000002143ff")
+
+		#Type IMSI, even number of identity digits
+		self.assertEqual(utils.enc_imsi("22806280000028"), "0821826082000020f8")
+		self.assertEqual(utils.enc_imsi("00101000012345"), "0801101000002143f5")
+		self.assertEqual(utils.enc_imsi("001010000123"), "07011010000021f3ff")
+
+	def testDecImsi(self):
+		#Type IMSI, odd number of identity digits
+		self.assertEqual(utils.dec_imsi("082982608200002080"), "228062800000208")
+		self.assertEqual(utils.dec_imsi("080910100000214365"), "001010000123456")
+		self.assertEqual(utils.dec_imsi("0709101000002143ff"), "0010100001234")
+		self.assertEqual(utils.dec_imsi("0709101000002143"), "0010100001234")
+
+		#Type IMSI, even number of identity digits
+		self.assertEqual(utils.dec_imsi("0821826082000020f8"), "22806280000028")
+		self.assertEqual(utils.dec_imsi("0801101000002143f5"), "00101000012345")
+		self.assertEqual(utils.dec_imsi("07011010000021f3ff"), "001010000123")
+		self.assertEqual(utils.dec_imsi("07011010000021f3"), "001010000123")
+
 	def test_enc_plmn(self):
 		with self.subTest("2-digit MCC"):
 			self.assertEqual(utils.enc_plmn("001", "01F"), "00F110")