card_key_provider: add PostgreSQL support

The Card Key Provider currently only has support for CSV files
as input. Unfortunately using CSV files does not scale very well
when the card inventory is very large and continously updated.
In this case a centralized storage in the form of a database
is the more suitable approach.

This patch adds PostgreSQL support next to the existing CSV
file support. It also adds an importer tool to import existing
CSV files into the database.

Change-Id: Icba625c02a60d7e1f519b506a46bda5ded0537d3
Related: SYS#7725

README.md

@@ -100,6 +100,7 @@ Please install the following dependencies:
  - pyyaml >= 5.1
  - smpp.pdu (from `github.com/hologram-io/smpp.pdu`)
  - termcolor
+ - psycopg2-binary
 
 Example for Debian:
 ```sh

contrib/csv-encrypt-columns.py

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

contrib/csv-to-pgsql.py

@@ -0,0 +1,286 @@
+#!/usr/bin/env python3
+
+import argparse
+import logging
+import csv
+import sys
+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("CSV2PGQSL")
+
+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')
+
+        log = PySimLogger.get("PQSQL")
+        self.table = table_name
+        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.lower()),
+                                                                      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.lower()),
+                                                                              Identifier(c.lower())))
+
+        # Set permissions
+        self.cur.execute(SQL("GRANT INSERT ON {} TO {};").format(Identifier(self.table.lower()),
+                                                                 Identifier(user_importer)))
+        self.cur.execute(SQL("GRANT SELECT ON {} TO {};").format(Identifier(self.table.lower()),
+                                                                 Identifier(user_reader)))
+
+        log.info("New database table created: %s", str(self.table.lower()))
+
+    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.lower(),))
+
+        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.lower()),
+                                                                                  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.lower()))
+        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(opts.pqsql, 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('--pqsql', metavar='FILE',
+                               default=str(Path.home()) + "/.osmocom/pysim/card_data_pqsql.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/esim_gen_metadata.py

@@ -0,0 +1,40 @@
+#!/usr/bin/env python3
+
+# (C) 2025 by Harald Welte <laforge@osmocom.org>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import argparse
+from osmocom.utils import h2b, swap_nibbles
+from pySim.esim.es8p import ProfileMetadata
+
+parser = argparse.ArgumentParser(description="""Utility program to generate profile metadata in the
+StoreMetadataRequest format based on input values from the command line.""")
+parser.add_argument('--iccid', required=True, help="ICCID of eSIM profile");
+parser.add_argument('--spn', required=True, help="Service Provider Name");
+parser.add_argument('--profile-name', required=True, help="eSIM Profile Name");
+parser.add_argument('--profile-class', choices=['test', 'operational', 'provisioning'],
+                    default='operational', help="Profile Class");
+parser.add_argument('--outfile', required=True, help="Output File Name");
+
+if __name__ == '__main__':
+    opts = parser.parse_args()
+
+    iccid_bin = h2b(swap_nibbles(opts.iccid))
+    pmd = ProfileMetadata(iccid_bin, spn=opts.spn, profile_name=opts.profile_name,
+                          profile_class=opts.profile_class)
+
+    with open(opts.outfile, 'wb') as f:
+        f.write(pmd.gen_store_metadata_request())
+    print("Written StoreMetadataRequest to '%s'" % opts.outfile)

contrib/saip-tool.py

@@ -329,7 +329,7 @@ def do_info(pes: ProfileElementSequence, opts):
         print("Security domain Instance AID: %s" % b2h(sd.decoded['instance']['instanceAID']))
         # FIXME: 'applicationSpecificParametersC9' parsing to figure out enabled SCP
         for key in sd.keys:
-            print("\tKVN=0x%02x, KID=0x%02x, %s" % (key.key_version_number, key.key_identifier, key.key_components))
+            print("\t%s" % repr(key))
 
     # RFM
     print()

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,215 @@ 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 CardKeyProviderPqsql
+------------------------
+
+With the `CardKeyProviderPsql` 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.
+
+
+Setting up the database
+^^^^^^^^^^^^^^^^^^^^^^^
+
+From the perspective of the database, the `CardKeyProviderPsql` 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 `CardKeyProviderPsql` 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 `CardKeyProviderPsql`
+
+* **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
+  `CardKeyProviderPsql`
+
+* **reader**:
+  To access data from within pySim shell using the `CardKeyProviderPsql` 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_pqsql.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 `CardKeyProviderPsql` shall
+use to query to locate card key data. You can set up as many tables as you
+want, `CardKeyProviderPsql` 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_pqsql.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_pqsql.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_pqsql.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_pqsql.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 +278,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 +290,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/conf.py

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

docs/osmo-smdpp.rst

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

osmo-smdpp.py

@@ -861,10 +861,10 @@ class SmDppHttpServer:
 
 def main(argv):
     parser = argparse.ArgumentParser()
-    parser.add_argument("-H", "--host", help="Host/IP to bind HTTP to", default="localhost")
-    parser.add_argument("-p", "--port", help="TCP port to bind HTTP to", default=8000)
+    parser.add_argument("-H", "--host", help="Host/IP to bind HTTP(S) to", default="localhost")
+    parser.add_argument("-p", "--port", help="TCP port to bind HTTP(S) to", default=443)
     parser.add_argument("-c", "--certdir", help=f"cert subdir relative to {DATA_DIR}", default="certs")
-    parser.add_argument("-s", "--nossl", help="do NOT use ssl", action='store_true', default=False)
+    parser.add_argument("-s", "--nossl", help="disable built in SSL/TLS support", action='store_true', default=False)
     parser.add_argument("-v", "--verbose", help="dump more raw info", action='store_true', default=False)
     parser.add_argument("-b", "--brainpool", help="Use Brainpool curves instead of NIST",
                         action='store_true', default=False)

pySim-shell.py

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

pySim/card_key_provider.py

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

pySim/commands.py

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

pySim/esim/es8p.py

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

pySim/esim/saip/__init__.py

@@ -183,7 +183,7 @@ class File:
         self.file_type = template.file_type
         self.fid = template.fid
         self.sfi = template.sfi
-        self.arr = template.arr.to_bytes(1)
+        self.arr = template.arr.to_bytes(1, 'big')
         if hasattr(template, 'rec_len'):
             self.rec_len = template.rec_len
         else:
@@ -227,7 +227,7 @@ class File:
             fileDescriptor['shortEFID'] = bytes([self.sfi])
         if self.df_name:
             fileDescriptor['dfName'] = self.df_name
-        if self.arr and self.arr != self.template.arr.to_bytes(1):
+        if self.arr and self.arr != self.template.arr.to_bytes(1, 'big'):
             fileDescriptor['securityAttributesReferenced'] = self.arr
         if self.file_type in ['LF', 'CY']:
             fdb_dec['file_type'] = 'working_ef'
@@ -264,7 +264,7 @@ class File:
         if self.read_and_update_when_deact:
             spfi |= 0x40 # TS 102 222 Table 5
         if spfi != 0x00:
-            pefi['specialFileInformation'] = spfi.to_bytes(1)
+            pefi['specialFileInformation'] = spfi.to_bytes(1, 'big')
         if self.fill_pattern:
             if not self.fill_pattern_repeat:
                 pefi['fillPattern'] = self.fill_pattern
@@ -985,9 +985,9 @@ class SecurityDomainKey:
         self.key_components = key_components
 
     def __repr__(self) -> str:
-        return 'SdKey(KVN=0x%02x, ID=0x%02x, Usage=%s, Comp=%s)' % (self.key_version_number,
+        return 'SdKey(KVN=0x%02x, ID=0x%02x, Usage=0x%x, Comp=%s)' % (self.key_version_number,
                                                                     self.key_identifier,
-                                                                    self.key_usage_qualifier,
+                                                                    build_construct(KeyUsageQualifier, self.key_usage_qualifier)[0],
                                                                     repr(self.key_components))
 
     @classmethod

pySim/log.py

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

pySim/runtime.py

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

pySim/ts_102_221.py

@@ -750,7 +750,7 @@ class EF_ARR(LinFixedEF):
         @cmd2.with_argparser(LinFixedEF.ShellCommands.read_rec_dec_parser)
         def do_read_arr_record(self, opts):
             """Read one EF.ARR record in flattened, human-friendly form."""
-            (data, _sw) = self._cmd.lchan.read_record_dec(opts.record_nr)
+            (data, _sw) = self._cmd.lchan.read_record_dec(opts.RECORD_NR)
             data = self._cmd.lchan.selected_file.flatten(data)
             self._cmd.poutput_json(data, opts.oneline)
 

pySim/ts_51_011.py

@@ -267,11 +267,11 @@ class EF_SMSP(LinFixedEF):
                 raise ValueError
         def _encode(self, obj, context, path):
             if obj <= 12*60:
-                return obj/5 - 1
+                return obj // 5 - 1
             elif obj <= 24*60:
                 return 143 + ((obj - (12 * 60)) // 30)
             elif obj <= 30 * 24 * 60:
-                return 166 + (obj / (24 * 60))
+                return 166 + (obj // (24 * 60))
             elif obj <= 63 * 7 * 24 * 60:
                 return 192 + (obj // (7 * 24 * 60))
             else:
@@ -280,7 +280,7 @@ class EF_SMSP(LinFixedEF):
     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))))
-        self._construct = Struct('alpha_id'/COptional(GsmStringAdapter(Rpad(Bytes(this._.total_len-28)))),
+        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)),
                                  'tp_dest_addr'/ScAddr,

requirements.txt

@@ -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
+psycopg2-binary

setup.py

@@ -21,7 +21,7 @@ setup(
         "pyscard",
         "pyserial",
         "pytlv",
-        "cmd2 >= 1.5.0",
+        "cmd2 >= 1.5.0, < 3.0",
         "jsonpath-ng",
         "construct >= 2.10.70",
         "bidict",
@@ -34,6 +34,7 @@ setup(
         "smpp.pdu @ git+https://github.com/hologram-io/smpp.pdu",
         "asn1tools",
         "smpp.twisted3 @ git+https://github.com/jookies/smpp.twisted",
+        "psycopg2-binary"
     ],
     scripts=[
         'pySim-prog.py',

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

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

tests/unittests/test_card_key_provider.csv

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

tests/unittests/test_card_key_provider.py

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

tests/unittests/test_log.py

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