1 Commits

Author SHA1 Message Date
Philipp Maier
4d9f16b3ac docs/put_key: add tutorial that explains how to manage global platform keys
With the increased interest in using GlobalPlatform features of
UICC and eUICCs (OTA-SMS, applets, etc.), also comes an increased
interest in how the related GlobalPlatform keys can be managed
(key rotation, adding/removing keysets from/to a Security Domain).

Unfortunately, many aspects of this topic are not immediately
obvious for the average user. Let's add a tutorial that contains
some practical examples to shine some light on the topic.

Related: SYS#7881
Change-Id: I163dfedca3df572cb8442e9a4a280e6c5b00327e
2026-03-13 18:10:09 +01:00
51 changed files with 703 additions and 5288 deletions

1
.gitignore vendored
View File

@@ -3,7 +3,6 @@
/docs/_*
/docs/generated
/docs/filesystem.rst
/.cache
/.local
/build

View File

@@ -285,7 +285,10 @@ if __name__ == '__main__':
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"}, opts.verbose)
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)

View File

@@ -305,16 +305,16 @@ the requested data.
ADM PIN
^^^^^^^
~~~~~~~
The `verify_adm` command will attempt to look up the `ADM1` column
indexed by the ICCID of the SIM/UICC.
SCP02 / SCP03
^^^^^^^^^^^^^
~~~~~~~~~~~~~
SCP02 and SCP03 each use key triplets consisting of ENC, MAC and DEK
SCP02 and SCP03 each use key triplets consisting if ENC, MAC and DEK
keys. For more details, see the applicable GlobalPlatform
specifications.

View File

@@ -13,7 +13,6 @@
import os
import sys
sys.path.insert(0, os.path.abspath('..'))
sys.path.insert(0, os.path.abspath('.')) # for local extensions (pysim_fs_sphinx, ...)
# -- Project information -----------------------------------------------------
@@ -40,8 +39,7 @@ extensions = [
"sphinx.ext.autodoc",
"sphinxarg.ext",
"sphinx.ext.autosectionlabel",
"sphinx.ext.napoleon",
"pysim_fs_sphinx",
"sphinx.ext.napoleon"
]
# Add any paths that contain templates here, relative to this directory.
@@ -66,25 +64,3 @@ html_theme = 'alabaster'
html_static_path = ['_static']
autoclass_content = 'both'
# Mock optional server-side deps of es2p and http_json_api/es9p,
# so that autodoc can import and document those modules.
autodoc_mock_imports = ['klein', 'twisted']
# Workaround for duplicate label warnings:
# https://github.com/sphinx-doc/sphinx-argparse/issues/14
#
# sphinxarg.ext generates generic sub-headings ("Named arguments",
# "Positional arguments", "Sub-commands", "General options", ...) for every
# argparse command/tool. These repeat across many files and trigger tons
# of autosectionlabel duplicate-label warnings - suppress them.
autosectionlabel_maxdepth = 3
suppress_warnings = [
'autosectionlabel.filesystem',
'autosectionlabel.saip-tool',
'autosectionlabel.shell',
'autosectionlabel.smpp2sim',
'autosectionlabel.smpp-ota-tool',
'autosectionlabel.suci-keytool',
'autosectionlabel.trace',
]

View File

@@ -39,7 +39,6 @@ pySim consists of several parts:
:caption: Contents:
shell
filesystem
trace
legacy
smpp2sim

View File

@@ -205,7 +205,7 @@ Specifically, pySim-read will dump the following:
* DF.GSM
* EF.IMSI
* EF,IMSI
* EF.GID1
* EF.GID2
* EF.SMSP

View File

@@ -1,7 +1,7 @@
Guide: Managing GP Keys
=======================
Most of today's smartcards follow the GlobalPlatform Card Specification and the included Security Domain model.
Most of todays smartcards follow the GlobalPlatform Card Specification and the included Security Domain model.
UICCs and eUCCCs are no exception here.
The Security Domain acts as an on-card representative of a card authority or administrator. It is used to perform tasks
@@ -13,7 +13,7 @@ In this tutorial, we will show how to work with the key material (keysets) store
rotate (replace) existing keys. We will also show how to provision new keys.
.. warning:: Making changes to keysets requires extreme caution as misconfigured keysets may lock you out permanently.
It's also strongly recommended to maintain at least one backup keyset that you can use as fallback in case
It also strongly recommended to maintain at least one backup keyset that you can use as fallback in case
the primary keyset becomes unusable for some reason.
@@ -34,61 +34,69 @@ any other file.
}
}
When working with eUICCs, multiple Security Domains are involved. The model is fundamentally different from the classic
model with one primary Security Domain (ISD). In the case of eUICCs, an ISD-R (Issuer Security Domain - Root) and an
ISD-P (Issuer Security Domain - Profile) exist (see also: GSMA SGP.02, section 2.2.1).
When working with eUICCs, multiple Security Domains are involved. The model is slightly different from the classic
model with one primary ISD. In the case of eUICCs, an ISD-R and an ISD-P exists.
The ISD-P is established by the ISD-R during the profile installation and serves as a secure container for an eSIM
profile. Within the ISD-P the eSIM profile establishes a dedicated Security Domain called `MNO-SD` (see also GSMA
SGP.02, section 2.2.4). This `MNO-SD` is comparable to the Issuer Security Domain (ISD) we find on UICCs. The AID of
`MNO-SD` is either the default AID for the Issuer Security Domain (see also GlobalPlatform, section H.1.3) or a
different value specified by the provider of the eSIM profile.
The ISD-R (Issuer Security Domain - Root) is indeed the primary ISD. Its purpose is to handle the installation of new
profiles and to manage the already installed profiles. The ISD-R shows up as a `ADF.ISD-R` and can be selected normally
(see above) The key material that allows access to the ISD-R is usually only known to the eUICC manufacturer.
Since the AID of the `MNO-SD` is not a fixed value, it is not known by `pySim-shell`. This means there will be no
`ADF.ISD` file shown in the file system, but we can simply select the `ADF.ISD-R` first and then select the `MNO-SD`
using a raw APDU. In the following example we assume that the default AID (``a000000151000000``) is used The APDU
would look like this: ``00a4040408`` + ``a000000151000000`` + ``00``
The ISD-P (Issuer Security Domain - Profile) is the primary ISD of the currently enabled profile. The ISD-P is
comparable to the ISD we find on a UICC. The key material for the ISD-P should be known known to the ISP, which
is the owner of the installed profile.
Since the AID of the ISD-P is allocated during the profile installation and different for each profile, it is not known
by pySim-shell. This means there will no `ADF.ISD-P` file show up in the file system, but we can simply select the
ISD-R, request the AID of the ISD-P and switch over to that ISD-P using a raw APDU:
``00a4040410`` + ``a0000005591010ffffffff8900001000`` + ``00``
::
pySIM-shell (00:MF)> select ADF.ISD-R
{
"application_id": "a0000005591010ffffffff8900000100",
"proprietary_data": {
"maximum_length_of_data_field_in_command_message": 255
},
"isdr_proprietary_application_template": {
"supported_version_number": "020300"
}
}
pySIM-shell (00:MF/ADF.ISD-R)> apdu 00a4040408a00000015100000000
SW: 9000, RESP: 6f108408a000000151000000a5049f6501ff
pySIM-shell (00:MF)> select ADF.ISD-R
{
"application_id": "a0000005591010ffffffff8900000100",
"proprietary_data": {
"maximum_length_of_data_field_in_command_message": 255
},
"isdr_proprietary_application_template": {
"supported_version_number": "020300"
}
}
pySIM-shell (00:MF/ADF.ISD-R)> get_profiles_info
{
"profile_info_seq": {
"profile_info": {
"iccid": "8949449999999990023",
"isdp_aid": "a0000005591010ffffffff8900001000",
"profile_state": "enabled",
"service_provider_name": "OsmocomSPN",
"profile_name": "TS48V1-A-UNIQUE",
"profile_class": "operational"
}
}
}
pySIM-shell (00:MF/ADF.ISD-R)> apdu 00a4040410a0000005591010ffffffff890000100000
SW: 9000, RESP: 6f188410a0000005591010ffffffff8900001000a5049f6501ff
pySIM-shell (00:MF/ADF.ISD-R)>
After that, the prompt will still show the `ADF.ISD-R`, but we are actually in `ADF.ISD` and the standard GlobalPlatform
operations like `establish_scpXX`, `get_data`, and `put_key` should work. By doing this, we simply have tricked
`pySim-shell` into making the GlobalPlatform related commands available for some other Security Domain we are not
interested in. With the raw APDU we then have swapped out the Security Domain under the hood. The same workaround can
be applied to any Security Domain, provided that the AID is known to the user.
After that, the prompt will still show the ADF.ISD-R, but we are actually in ADF.ISD-P and the standard GlobalPlatform
operations like `establish_scpXX`, `get_data`, and `put_key` should work. The same workaround can also be applied to any
Supplementary Security Domain as well, provided that the AID is known to the user.
Establishing a secure channel
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Before we can make changes to the keysets in the currently selected Security Domain we must first establish a secure
channel with that Security Domain. In the following examples we will use `SCP02` (see also GlobalPlatform Card
Specification, section E.1.1) and `SCP03` (see also GlobalPlatform Card Specification Amendment D) to establish the
secure channel. `SCP02` is slightly older than `SCP03`. The main difference between the two is that `SCP02` uses 3DES
while `SCP03` is based on AES.
Before we can make changes to the keysets in the currently selected Security Domain we must first establish a secure channel
with that Security Domain. The secure channel protocols commonly used for this are `SCP02` (see also GlobalPlatform Card
Specification, section E.1.1) and `SCP03` (see also GlobalPlatform Card Specification Amendment D). `SCP02` is slightly
older and commonly used on UICCs. The more modern `SCP03` is commonly used on eUICCs. The main difference between the
two is that `SCP02` uses 3DES while `SCP03` is based on AES.
.. warning:: Secure channel protocols like `SCP02` and `SCP03` may manage an error counter to count failed login
attempts. This means attempting to establish a secure channel with a wrong keyset multiple times may lock
you out permanently. Double check the applied keyset before attempting to establish a secure channel.
.. warning:: The key values used in the following examples are random key values used for illustration purposes only.
Each UICC or eSIM profile is shipped with individual keys, which means that the keys used below will not
work with your UICC or eSIM profile. You must replace the key values with the values you have received
from your UICC vendor or eSIM profile provider.
Example: `SCP02`
----------------
@@ -111,7 +119,7 @@ establish a secure channel using the SCP02 Secure Channel Protocol.
::
pySIM-shell (00:MF/ADF.ISD)> establish_scp02 --key-enc F09C43EE1A0391665CC9F05AF4E0BD10 --key-mac 01981F4A20999F62AF99988007BAF6CA --key-dek 8F8AEE5CDCC5D361368BC45673D99195 --key-ver 112 --security-level 3
pySIM-shell (00:MF/ADF.ISD)> establish_scp02 --key-enc F09C43EE1A0391665CC9F05AF4E0BD10 --key-mac 01981F4A20999F62AF99988007BAF6CA --key-dek 8F8AEE5CDCC5D361368BC45673D99195 --security-level 3
Successfully established a SCP02[03] secure channel
@@ -119,34 +127,32 @@ Example: `SCP03`
----------------
The establishment of a secure channel via SCP03 works just the same. In the following example we will establish a
secure channel to the `MNO-SD` of an eSIM profile. The SCP03 keyset we use is tied to KVN 48 and looks like this:
secure channel to the ISD-R of an eUICC. The SCP03 keyset we use is tied to KVN 50 and looks like this:
+---------+------------------------------------------------------------------+
| Keyname | Keyvalue |
+=========+==================================================================+
| ENC/KIC | 63af517c29ad6ac6fcadfe6ac8a3c8a041d8141c7eb845ef1cba6112a325e430 |
| ENC/KIC | 620ff456b0c0328b68dc0d7d5eb24e07dd749aa86c9ff1836a7263e1d8896510 |
+---------+------------------------------------------------------------------+
| MAC/KID | 54b9ad6713ae922f54014ed762132e7b59bdcd2a2a6beba98fb9afe6b4df27e1 |
| MAC/KID | b38116a2c85f2c8f46bbdc0081d6e8a04b0a58087d0ce5ee0ccc4c945e4aeda6 |
+---------+------------------------------------------------------------------+
| DEK/KIK | cbb933ba2389da93c86c112739cd96389139f16c6f80f7d16bf3593e407ca893 |
| DEK/KIK | d409486cbcb8092a8592ee46d8668dfa97bea5eb7ce9c2b5a3f3bb1db358a153 |
+---------+------------------------------------------------------------------+
We assume that the `MNO-SD` is already selected (see above). We may now establish the SCP03 secure channel:
We assume that ADF.ISD-R is already selected. We may now establish the SCP03 secure channel:
::
pySIM-shell (00:MF/ADF.ISD-R)> establish_scp03 --key-enc 63af517c29ad6ac6fcadfe6ac8a3c8a041d8141c7eb845ef1cba6112a325e430 --key-mac 54b9ad6713ae922f54014ed762132e7b59bdcd2a2a6beba98fb9afe6b4df27e1 --key-dek cbb933ba2389da93c86c112739cd96389139f16c6f80f7d16bf3593e407ca893 --key-ver 48 --security-level 3
pySIM-shell (00:MF/ADF.ISD-R)> establish_scp03 --key-enc 620ff456b0c0328b68dc0d7d5eb24e07dd749aa86c9ff1836a7263e1d8896510 --key-mac b38116a2c85f2c8f46bbdc0081d6e8a04b0a58087d0ce5ee0ccc4c945e4aeda6 --key-dek d409486cbcb8092a8592ee46d8668dfa97bea5eb7ce9c2b5a3f3bb1db358a153 --key-ver 50 --security-level 3
Successfully established a SCP03[03] secure channel
Understanding Keysets
~~~~~~~~~~~~~~~~~~~~~
Before making any changes to keysets, it is recommended to check the status of the currently installed keysets. To do
so, we use the `get_data` command to retrieve the `key_information`. This command does not require the establishment of
a secure channel. We also cannot read back the key values themselves, but we get a summary of the installed keys
together with their KVN numbers, IDs, algorithm and key length values.
so, we use the `get_data` command to retrieve the `key_information`. We cannot read back the key values themselves, but
we get a summary of the installed keys together with their KVN numbers, IDs, algorithm and key length values.
Example: `key_information` from a `sysmoISIM-SJA5`:
@@ -398,18 +404,18 @@ used with which Secure Channel Protocol.
+-----------+-------------------------------------------------------+
| 48-63 | reserved for `SCP03` |
+-----------+-------------------------------------------------------+
| 64-79 | reserved for `SCP81` (GSMA SGP.02, section 2.2.5.1) |
+-----------+-------------------------------------------------------+
| 112 | Token key (RSA public or DES, also used with `SCP02`) |
+-----------+-------------------------------------------------------+
| 113 | Receipt key (DES) |
+-----------+-------------------------------------------------------+
| 115 | DAP verification key (RS public or DES) |
| 115 | DAP verifiation key (RS public or DES) |
+-----------+-------------------------------------------------------+
| 116 | reserved for CASD |
+-----------+-------------------------------------------------------+
| 117 | 16-byte DES key for Ciphered Load File Data Block |
+-----------+-------------------------------------------------------+
| 129-143 | reserved for `SCP81` |
+-----------+-------------------------------------------------------+
| 255 | reserved for ISD with SCP02 without SCP80 support |
+-----------+-------------------------------------------------------+
@@ -444,7 +450,7 @@ In this case, all three keys share the same length and are used with the same al
to implicitly select sub-types of an algorithm. (e.g. a 16 byte key of type `aes` is associated with `AES128`, where a 32
byte key would be associated with `AES256`).
The second example shows that different schemes are possible. The `SCP80` keyset from the second example uses a scheme
That different schemes are possible shows the second example. The `SCP80` keyset from the second example uses a scheme
that works with two keys:
+----------------+---------+---------------------------------------+
@@ -493,7 +499,7 @@ keys in the `--key-data` arguments. It is also important that each `--key-data`
argument that sets the algorithm correctly (`des` in this case).
Finally we have to target the keyset we want to rotate by its KVN. The `--old-key-version-nr` argument is set to 112
as this identifies the keyset we want to rotate. The `--key-version-nr` is also set to 112 as we do not want
as this is identifies the keyset we want to rotate. The `--key-version-nr` is also set to 112 as we do not want to the
KVN to be changed in this example. Changing the KVN while rotating a keyset is possible. In case the KVN has to change
for some reason, the new KVN must be selected carefully to keep the key usable with the associated Secure Channel
Protocol.
@@ -538,7 +544,7 @@ Adding a keyset
In the following we will discuss how to add an entirely new keyset. The procedure is almost identical with the key
rotation procedure we have already discussed and it is assumed that all details about the key rotation are understood.
In this section we will go into more detail and illustrate how to provision new 3DES, `AES128` and `AES256` keysets.
In this section we will go into more detail and and illustrate how to provision new 3DES, `AES128` and `AES256` keysets.
It is important to keep in mind that storage space on smartcard is a precious resource. In many cases the amount of
keysets that a Security Domain can store is limited. In some situations you may be forced to sacrifice one of your

View File

@@ -1,267 +0,0 @@
"""
Sphinx extension: auto-generate docs/filesystem.rst from the pySim EF class hierarchy.
Hooked into Sphinx's ``builder-inited`` event so the file is always regenerated
from the live Python classes before Sphinx reads any source files.
The table of root objects to document is in SECTIONS near the top of this file.
EXCLUDED lists CardProfile/CardApplication subclasses intentionally omitted from
SECTIONS, with reasons. Both tables are read by tests/unittests/test_fs_coverage.py
to ensure every class with EF/DF content is accounted for.
"""
import importlib
import inspect
import json
import os
import sys
import textwrap
# Ensure pySim is importable when this module is loaded as a Sphinx extension
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
from pySim.filesystem import (CardApplication, CardDF, CardMF, CardEF, # noqa: E402
TransparentEF, TransRecEF, LinFixedEF, CyclicEF, BerTlvEF)
from pySim.profile import CardProfile # noqa: E402
# Generic EF base classes whose docstrings describe the *type* of file
# (Transparent, LinFixed, ...) rather than a specific file's content.
# Suppress those boilerplate texts in the per-EF entries; they are only
# useful once, at the top of the document or in a dedicated glossary.
_EF_BASE_TYPES = frozenset([TransparentEF,
TransRecEF,
LinFixedEF,
CyclicEF,
BerTlvEF])
# ---------------------------------------------------------------------------
# Sections: (heading, module, class-name)
# The class must be either a CardProfile (uses .files_in_mf) or a CardDF
# subclass (uses .children).
# ---------------------------------------------------------------------------
SECTIONS = [
('MF / TS 102 221 (UICC)',
'pySim.ts_102_221', 'CardProfileUICC'),
('ADF.USIM / TS 31.102',
'pySim.ts_31_102', 'ADF_USIM'),
('ADF.ISIM / TS 31.103',
'pySim.ts_31_103', 'ADF_ISIM'),
('ADF.HPSIM / TS 31.104',
'pySim.ts_31_104', 'ADF_HPSIM'),
('DF.GSM + DF.TELECOM / TS 51.011 (SIM)',
'pySim.ts_51_011', 'CardProfileSIM'),
('CDMA / IS-820 (RUIM)',
'pySim.cdma_ruim', 'CardProfileRUIM'),
('DF.EIRENE / GSM-R',
'pySim.gsm_r', 'DF_EIRENE'),
('DF.SYSTEM / sysmocom SJA2+SJA5',
'pySim.sysmocom_sja2', 'DF_SYSTEM'),
]
# ---------------------------------------------------------------------------
# Excluded: {(module, class-name)}
# CardProfile and CardApplication subclasses that have EF/DF children but are
# intentionally absent from SECTIONS. Keeping this list explicit lets
# test_fs_coverage.py detect newly added classes that the developer forgot to
# add to either table.
# ---------------------------------------------------------------------------
EXCLUDED = {
# eUICC profiles inherit files_in_mf verbatim from CardProfileUICC; the
# eUICC-specific content lives in ISD-R / ISD-P applications, not in MF.
('pySim.euicc', 'CardProfileEuiccSGP02'),
('pySim.euicc', 'CardProfileEuiccSGP22'),
('pySim.euicc', 'CardProfileEuiccSGP32'),
# CardApplication* classes are thin wrappers that embed an ADF_* instance.
# The ADF contents are already documented via the corresponding ADF_* entry
# in SECTIONS above.
('pySim.ts_31_102', 'CardApplicationUSIM'),
('pySim.ts_31_102', 'CardApplicationUSIMnonIMSI'),
('pySim.ts_31_103', 'CardApplicationISIM'),
('pySim.ts_31_104', 'CardApplicationHPSIM'),
}
# RST underline characters ordered by nesting depth
_HEADING_CHARS = ['=', '=', '-', '~', '^', '"']
# Level 0 uses '=' with overline (page title).
# Level 1 uses '=' without overline (major sections).
# Levels 2+ use the remaining characters for DFs.
# ---------------------------------------------------------------------------
# RST formatting helpers
# ---------------------------------------------------------------------------
def _heading(title: str, level: int) -> str:
"""Return an RST heading string. Level 0 gets an overline."""
char = _HEADING_CHARS[level]
rule = char * len(title)
if level == 0:
return f'{rule}\n{title}\n{rule}\n\n'
return f'{title}\n{rule}\n\n'
def _json_default(obj):
"""Fallback serialiser: bytes -> hex, anything else -> repr."""
if isinstance(obj, (bytes, bytearray)):
return obj.hex()
return repr(obj)
def _examples_block(cls) -> str:
"""Return RST code-block examples (one per vector), or '' if none exist.
Each example is rendered as a ``json5`` code-block with the hex-encoded
binary as a ``// comment`` on the first line, followed by the decoded JSON.
``json5`` is used instead of ``json`` so that Pygments does not flag the
``//`` comment as a syntax error.
"""
vectors = []
for attr in ('_test_de_encode', '_test_decode'):
v = getattr(cls, attr, None)
if v:
vectors.extend(v)
if not vectors:
return ''
lines = ['**Examples**\n\n']
for t in vectors:
# 2-tuple: (encoded, decoded)
# 3-tuple: (encoded, record_nr, decoded) — LinFixedEF / CyclicEF
if len(t) >= 3:
encoded, record_nr, decoded = t[0], t[1], t[2]
comment = f'record {record_nr}: {encoded.lower()}'
else:
encoded, decoded = t[0], t[1]
comment = f'file: {encoded.lower()}'
json_str = json.dumps(decoded, default=_json_default, indent=2)
json_indented = textwrap.indent(json_str, ' ')
lines.append('.. code-block:: json5\n\n')
lines.append(f' // {comment}\n')
lines.append(json_indented + '\n')
lines.append('\n')
return ''.join(lines)
def _document_ef(ef: CardEF) -> str:
"""Return RST for a single EF. Uses ``rubric`` to stay out of the TOC."""
cls = type(ef)
parts = [ef.fully_qualified_path_str()]
if ef.fid:
parts.append(f'({ef.fid.upper()})')
if ef.desc:
parts.append(f'\u2014 {ef.desc}') # em-dash
title = ' '.join(parts)
lines = [f'.. rubric:: {title}\n\n']
# Only show a docstring if it is specific to this class. EFs that are
# direct instances of a base type (TransparentEF, LinFixedEF, ...) carry
# only the generic "what is a TransparentEF" boilerplate; named subclasses
# without their own __doc__ have cls.__dict__['__doc__'] == None. Either
# way, suppress the text here - it belongs at the document level, not
# repeated for every single EF entry.
doc = None if cls in _EF_BASE_TYPES else cls.__dict__.get('__doc__')
if doc:
lines.append(inspect.cleandoc(doc) + '\n\n')
examples = _examples_block(cls)
if examples:
lines.append(examples)
return ''.join(lines)
def _document_df(df: CardDF, level: int) -> str:
"""Return RST for a DF section and all its children, recursively."""
parts = [df.fully_qualified_path_str()]
if df.fid:
parts.append(f'({df.fid.upper()})')
if df.desc:
parts.append(f'\u2014 {df.desc}') # em-dash
title = ' '.join(parts)
lines = [_heading(title, level)]
cls = type(df)
doc = None if cls in (CardDF, CardMF) else cls.__dict__.get('__doc__')
if doc:
lines.append(inspect.cleandoc(doc) + '\n\n')
for child in df.children.values():
if isinstance(child, CardDF):
lines.append(_document_df(child, level + 1))
elif isinstance(child, CardEF):
lines.append(_document_ef(child))
return ''.join(lines)
# ---------------------------------------------------------------------------
# Top-level generator
# ---------------------------------------------------------------------------
def generate_filesystem_rst() -> str:
"""Walk all registered sections and return the full RST document as a string."""
out = [
'.. This file is auto-generated by docs/pysim_fs_sphinx.py — do not edit.\n\n',
_heading('Card Filesystem Reference', 0),
'This page documents all Elementary Files (EFs) and Dedicated Files (DFs) '
'implemented in pySim, organised by their location in the card filesystem.\n\n',
]
# Track already-documented classes so that DFs/EFs shared between profiles
# (e.g. DF.TELECOM / DF.GSM present in both CardProfileSIM and CardProfileRUIM)
# are only emitted once.
seen_types: set = set()
for section_title, module_path, class_name in SECTIONS:
module = importlib.import_module(module_path)
cls = getattr(module, class_name)
obj = cls()
if isinstance(obj, CardProfile):
files = obj.files_in_mf
elif isinstance(obj, CardApplication):
files = list(obj.adf.children.values())
elif isinstance(obj, CardDF):
files = list(obj.children.values())
else:
continue
# Filter out files whose class was already documented in an earlier section.
files = [f for f in files if type(f) not in seen_types]
if not files:
continue
out.append(_heading(section_title, 1))
for f in files:
seen_types.add(type(f))
if isinstance(f, CardDF):
out.append(_document_df(f, level=2))
elif isinstance(f, CardEF):
out.append(_document_ef(f))
return ''.join(out)
# ---------------------------------------------------------------------------
# Sphinx integration
# ---------------------------------------------------------------------------
def _on_builder_inited(app):
output_path = os.path.join(app.srcdir, 'filesystem.rst')
with open(output_path, 'w') as fh:
fh.write(generate_filesystem_rst())
def setup(app):
app.connect('builder-inited', _on_builder_inited)
return {'version': '0.1', 'parallel_read_safe': True}

View File

@@ -67,7 +67,7 @@ Inspecting applications
To inspect the application PE contents of an existing profile package, sub-command `info` with parameter '--apps' can
be used. This command lists out all application and their parameters in detail. This allows an application developer
to check if the applet insertion was carried out as expected.
to check if the applet insertaion was carried out as expected.
Example: Listing applications and their parameters
::

View File

@@ -602,8 +602,8 @@ This allows for easy interactive modification of records.
If this command fails before the editor is spawned, it means that the current record contents is not decodable,
and you should use the :ref:`update_record_decoded` or :ref:`update_record` command.
If this command fails after making your modifications in the editor, it means that the new file contents is not
encodable; please check your input and/or use the raw :ref:`update_record` command.
If this command fails after making your modificatiosn in the editor, it means that the new file contents is not
encodable; please check your input and/or us the raw :ref:`update_record` comamdn.
decode_hex
@@ -708,8 +708,8 @@ This allows for easy interactive modification of file contents.
If this command fails before the editor is spawned, it means that the current file contents is not decodable,
and you should use the :ref:`update_binary_decoded` or :ref:`update_binary` command.
If this command fails after making your modifications in the editor, it means that the new file contents is not
encodable; please check your input and/or use the raw :ref:`update_binary` command.
If this command fails after making your modificatiosn in the editor, it means that the new file contents is not
encodable; please check your input and/or us the raw :ref:`update_binary` comamdn.
decode_hex

View File

@@ -44,11 +44,6 @@ from pySim.legacy.ts_51_011 import EF
from pySim.card_handler import *
from pySim.utils import *
from pathlib import Path
import logging
from pySim.log import PySimLogger
log = PySimLogger.get(Path(__file__).stem)
def parse_options():
@@ -190,7 +185,6 @@ def parse_options():
default=False, action="store_true")
parser.add_argument("--card_handler", dest="card_handler_config", metavar="FILE",
help="Use automatic card handling machine")
parser.add_argument("--verbose", help="Enable verbose logging", action='store_true', default=False)
options = parser.parse_args()
@@ -776,9 +770,6 @@ if __name__ == '__main__':
# Parse options
opts = parse_options()
# Setup logger
PySimLogger.setup(print, {logging.WARN: "\033[33m"}, opts.verbose)
# Init card reader driver
sl = init_reader(opts)

View File

@@ -25,6 +25,7 @@
import hashlib
import argparse
import os
import random
import re
import sys
@@ -45,17 +46,11 @@ from pySim.utils import dec_imsi, dec_iccid
from pySim.legacy.utils import format_xplmn_w_act, dec_st, dec_msisdn
from pySim.ts_51_011 import EF_SMSP
from pathlib import Path
import logging
from pySim.log import PySimLogger
log = PySimLogger.get(Path(__file__).stem)
option_parser = argparse.ArgumentParser(description='Legacy tool for reading some parts of a SIM card',
formatter_class=argparse.ArgumentDefaultsHelpFormatter)
option_parser.add_argument("--verbose", help="Enable verbose logging", action='store_true', default=False)
argparse_add_reader_args(option_parser)
def select_app(adf: str, card: SimCard):
"""Select application by its AID"""
sw = 0
@@ -80,9 +75,6 @@ if __name__ == '__main__':
# Parse options
opts = option_parser.parse_args()
# Setup logger
PySimLogger.setup(print, {logging.WARN: "\033[33m"}, opts.verbose)
# Init card reader driver
sl = init_reader(opts)

View File

@@ -107,12 +107,12 @@ Online manual available at https://downloads.osmocom.org/docs/pysim/master/html/
kwargs = {'include_ipy': True}
self.verbose = verbose
PySimLogger.setup(self.poutput, {logging.WARN: YELLOW})
self._onchange_verbose('verbose', False, self.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
@@ -136,7 +136,8 @@ Online manual available at https://downloads.osmocom.org/docs/pysim/master/html/
self.add_settable(Settable2Compat('apdu_trace', bool, 'Trace and display APDUs exchanged with card', self,
onchange_cb=self._onchange_apdu_trace))
self.add_settable(Settable2Compat('apdu_strict', bool,
'Strictly apply APDU format according to ISO/IEC 7816-3, table 12', self))
'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))
@@ -217,6 +218,13 @@ Online manual available at https://downloads.osmocom.org/docs/pysim/master/html/
else:
self.card._scc._tp.apdu_tracer = None
def _onchange_apdu_strict(self, param_name, old, new):
if self.card:
if new == True:
self.card._scc._tp.apdu_strict = True
else:
self.card._scc._tp.apdu_strict = False
def _onchange_verbose(self, param_name, old, new):
PySimLogger.set_verbose(new)
if new == True:
@@ -273,7 +281,7 @@ Online manual available at https://downloads.osmocom.org/docs/pysim/master/html/
apdu_cmd_parser.add_argument('--expect-sw', help='expect a specified status word', type=str, default=None)
apdu_cmd_parser.add_argument('--expect-response-regex', help='match response against regex', type=str, default=None)
apdu_cmd_parser.add_argument('--raw', help='Bypass the logical channel (and secure channel)', action='store_true')
apdu_cmd_parser.add_argument('APDU', type=is_hexstr, help='APDU as hex string (see also: ISO/IEC 7816-3, section 12.1')
apdu_cmd_parser.add_argument('APDU', type=is_hexstr, help='APDU as hex string')
@cmd2.with_argparser(apdu_cmd_parser)
def do_apdu(self, opts):
@@ -282,23 +290,14 @@ Online manual available at https://downloads.osmocom.org/docs/pysim/master/html/
tracked. Depending on the raw APDU sent, pySim-shell may not continue to work as expected if you e.g. select
a different file."""
if not hasattr(self, 'apdu_strict_warning_displayed') and self.apdu_strict is False:
self.poutput("Warning: The default for the setable parameter `apdu_strict` will be changed from")
self.poutput(" `False` to `True` in future pySim-shell releases. In case you are using")
self.poutput(" the `apdu` command from a script that still mixes APDUs with TPDUs, consider")
self.poutput(" fixing or adding a `set apdu_strict false` line at the beginning.")
self.apdu_strict_warning_displayed = True;
# When sending raw APDUs we access the scc object through _scc member of the card object. It should also be
# noted that the apdu command plays an exceptional role since it is the only card accessing command that
# can be executed without the presence of a runtime state (self.rs) object. However, this also means that
# self.lchan is also not present (see method equip).
self.card._scc._tp.apdu_strict = self.apdu_strict
if opts.raw or self.lchan is None:
data, sw = self.card._scc.send_apdu(opts.APDU, apply_lchan = False)
else:
data, sw = self.lchan.scc.send_apdu(opts.APDU, apply_lchan = False)
self.card._scc._tp.apdu_strict = True
if data:
self.poutput("SW: %s, RESP: %s" % (sw, data))
else:
@@ -1176,7 +1175,13 @@ if __name__ == '__main__':
opts = option_parser.parse_args()
# Ensure that we are able to print formatted warnings from the beginning.
PySimLogger.setup(print, {logging.WARN: YELLOW}, opts.verbose)
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

View File

@@ -72,10 +72,10 @@ class ApduArDO(BER_TLV_IE, tag=0xd0):
if do[0] == 0x01:
self.decoded = {'generic_access_rule': 'always'}
return self.decoded
raise ValueError('Invalid 1-byte generic APDU access rule')
return ValueError('Invalid 1-byte generic APDU access rule')
else:
if len(do) % 8:
raise ValueError('Invalid non-modulo-8 length of APDU filter: %d' % len(do))
return ValueError('Invalid non-modulo-8 length of APDU filter: %d' % len(do))
self.decoded = {'apdu_filter': []}
offset = 0
while offset < len(do):
@@ -90,19 +90,19 @@ class ApduArDO(BER_TLV_IE, tag=0xd0):
return b'\x00'
if self.decoded['generic_access_rule'] == 'always':
return b'\x01'
raise ValueError('Invalid 1-byte generic APDU access rule')
return ValueError('Invalid 1-byte generic APDU access rule')
else:
if not 'apdu_filter' in self.decoded:
raise ValueError('Invalid APDU AR DO')
return ValueError('Invalid APDU AR DO')
filters = self.decoded['apdu_filter']
res = b''
for f in filters:
if not 'header' in f or not 'mask' in f:
raise ValueError('APDU filter must contain header and mask')
return ValueError('APDU filter must contain header and mask')
header_b = h2b(f['header'])
mask_b = h2b(f['mask'])
if len(header_b) != 4 or len(mask_b) != 4:
raise ValueError('APDU filter header and mask must each be 4 bytes')
return ValueError('APDU filter header and mask must each be 4 bytes')
res += header_b + mask_b
return res
@@ -269,7 +269,7 @@ class ADF_ARAM(CardADF):
cmd_do_enc = cmd_do.to_ie()
cmd_do_len = len(cmd_do_enc)
if cmd_do_len > 255:
raise ValueError('DO > 255 bytes not supported yet')
return ValueError('DO > 255 bytes not supported yet')
else:
cmd_do_enc = b''
cmd_do_len = 0
@@ -361,7 +361,7 @@ class ADF_ARAM(CardADF):
ar_do_content += [{'apdu_ar_do': {'generic_access_rule': 'always'}}]
elif opts.apdu_filter:
if len(opts.apdu_filter) % 16:
raise ValueError(f'Invalid non-modulo-16 length of APDU filter: {len(opts.apdu_filter)}')
return ValueError('Invalid non-modulo-16 length of APDU filter: %d' % len(do))
offset = 0
apdu_filter = []
while offset < len(opts.apdu_filter):

View File

@@ -131,7 +131,7 @@ class EF_AD(TransparentEF):
desc='Administrative Data', size=(3, None), **kwargs):
super().__init__(fid, sfid=sfid, name=name, desc=desc, size=size, **kwargs)
self._construct = Struct(
# Byte 1: MS operation mode
# Byte 1: Display Condition
'ms_operation_mode'/Enum(Byte, self.OP_MODE),
# Bytes 2-3: Additional information
'additional_info'/Bytes(2),

View File

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

View File

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

View File

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

View File

@@ -441,7 +441,7 @@ class File:
elif k == 'fillFileContent':
stream.write(v)
else:
raise ValueError("Unknown key '%s' in tuple list" % k)
return ValueError("Unknown key '%s' in tuple list" % k)
return stream.getvalue()
def file_content_to_tuples(self, optimize:bool = False) -> List[Tuple]:
@@ -1079,13 +1079,6 @@ class SecurityDomainKey:
'keyVersionNumber': bytes([self.key_version_number]),
'keyComponents': [k.to_saip_dict() for k in self.key_components]}
def get_key_component(self, key_type):
for kc in self.key_components:
if kc.key_type == key_type:
return kc.key_data
return None
class ProfileElementSD(ProfileElement):
"""Class representing a securityDomain ProfileElement."""
type = 'securityDomain'

View File

@@ -1,366 +0,0 @@
"""Implementation of Personalization of eSIM profiles in SimAlliance/TCA Interoperable Profile:
Run a batch of N personalizations"""
# (C) 2025-2026 by sysmocom - s.f.m.c. GmbH <info@sysmocom.de>
#
# Author: nhofmeyr@sysmocom.de
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import copy
import pprint
import logging
import traceback
import inspect
from typing import List, Generator
from pySim.esim.saip.personalization import ConfigurableParameter
from pySim.esim.saip import param_source
from pySim.esim.saip import ProfileElementSequence, ProfileElementSD
from pySim.global_platform import KeyUsageQualifier
from osmocom.utils import b2h
logger = logging.getLogger(__name__)
def _func_():
return inspect.currentframe().f_back.f_code.co_name
class BatchPersonalization:
"""Produce a series of eSIM profiles from predefined parameters.
Personalization parameters are derived from pysim.esim.saip.param_source.ParamSource.
Usage example:
der_input = some_file.open('rb').read()
pes = ProfileElementSequence.from_der(der_input)
p = pers.BatchPersonalization(
n=10,
src_pes=pes,
csv_rows=get_csv_reader())
p.add_param_and_src(
personalization.Iccid(),
param_source.IncDigitSource(
num_digits=18,
first_value=123456789012340001,
last_value=123456789012340010))
# add more parameters here, using ConfigurableParameter and ParamSource subclass instances to define the profile
# ...
# generate all 10 profiles (from n=10 above)
for result_pes in p.generate_profiles():
upp = result_pes.to_der()
store_upp(upp)
"""
class ParamAndSrc:
'tie a ConfigurableParameter to a source of actual values'
def __init__(self, param: ConfigurableParameter, src: param_source.ParamSource):
self.param = param
self.src = src
def __init__(self,
n: int,
src_pes: ProfileElementSequence,
params: list[ParamAndSrc]=None,
csv_rows: Generator=None,
):
"""
n: number of eSIM profiles to generate.
src_pes: a decoded eSIM profile as ProfileElementSequence, to serve as template. This is not modified, only
copied.
params: list of ParamAndSrc instances, defining a ConfigurableParameter and corresponding ParamSource to fill in
profile values.
csv_rows: A list or generator producing all CSV rows one at a time, starting with a row containing the column
headers. This is compatible with the python csv.reader. Each row gets passed to
ParamSource.get_next(), such that ParamSource implementations can access the row items.
See param_source.CsvSource.
"""
self.n = n
self.params = params or []
self.src_pes = src_pes
self.csv_rows = csv_rows
def add_param_and_src(self, param:ConfigurableParameter, src:param_source.ParamSource):
self.params.append(BatchPersonalization.ParamAndSrc(param=param, src=src))
def generate_profiles(self):
# get first row of CSV: column names
csv_columns = None
if self.csv_rows:
try:
csv_columns = next(self.csv_rows)
except StopIteration as e:
raise ValueError('the input CSV file appears to be empty') from e
for i in range(self.n):
csv_row = None
if self.csv_rows and csv_columns:
try:
csv_row_list = next(self.csv_rows)
except StopIteration as e:
raise ValueError(f'not enough rows in the input CSV for eSIM nr {i+1} of {self.n}') from e
csv_row = dict(zip(csv_columns, csv_row_list))
pes = copy.deepcopy(self.src_pes)
for p in self.params:
try:
input_value = p.src.get_next(csv_row=csv_row)
assert input_value is not None
value = p.param.__class__.validate_val(input_value)
p.param.__class__.apply_val(pes, value)
except Exception as e:
print(traceback.format_exc())
logger.error('during %s: %r', _func_(), e)
raise ValueError(f'{p.param.name} fed by {p.src.name}: {e!r}') from e
yield pes
class UppAudit(dict):
"""
Key-value pairs collected from a single UPP DER or PES.
UppAudit itself is a dict, callers may use the standard python dict API to access key-value pairs read from the UPP.
"""
@classmethod
def from_der(cls, der: bytes, params: List, der_size=False, additional_sd_keys=False):
'''return a dict of parameter name and set of selected parameter values found in a DER encoded profile. Note:
some ConfigurableParameter implementations return more than one key-value pair, for example, Imsi returns
both 'IMSI' and 'IMSI-ACC' parameters.
e.g.
UppAudit.from_der(my_der, [Imsi, ])
--> {'IMSI': '001010000000023', 'IMSI-ACC': '5'}
(where 'IMSI' == Imsi.name)
Read all parameters listed in params. params is a list of either ConfigurableParameter classes or
ConfigurableParameter class instances. This calls only classmethods, so each entry in params can either be the
class itself, or a class-instance of, a (non-abstract) ConfigurableParameter subclass.
For example, params = [Imsi, ] is equivalent to params = [Imsi(), ].
For der_size=True, also include a {'der_size':12345} entry.
For additional_sd_keys=True, output also all Security Domain KVN that there are *no* ConfigurableParameter
subclasses for. For example, SCP80 has reserved kvn 0x01..0x0f, but we offer only Scp80Kvn01, Scp80Kvn02,
Scp80Kvn03. So we would not show kvn 0x04..0x0f in an audit. additional_sd_keys=True includes audits of all SD
key KVN there may be in the UPP. This helps to spot SD keys that may already be present in a UPP template, with
unexpected / unusual kvn.
'''
# make an instance of this class
upp_audit = cls()
if der_size:
upp_audit['der_size'] = set((len(der), ))
pes = ProfileElementSequence.from_der(der)
for param in params:
try:
for valdict in param.get_values_from_pes(pes):
upp_audit.add_values(valdict)
except Exception as e:
raise ValueError(f'Error during audit for parameter {param}: {e}') from e
if not additional_sd_keys:
return upp_audit
# additional_sd_keys
for pe in pes.pe_list:
if pe.type != 'securityDomain':
continue
assert isinstance(pe, ProfileElementSD)
for key in pe.keys:
audit_key = f'SdKey_KVN{key.key_version_number:02x}_ID{key.key_identifier:02x}'
kuq_bin = KeyUsageQualifier.build(key.key_usage_qualifier).hex()
audit_val = f'{key.key_components=!r} key_usage_qualifier=0x{kuq_bin}={key.key_usage_qualifier!r}'
upp_audit[audit_key] = set((audit_val, ))
return upp_audit
def get_single_val(self, key, validate=True, allow_absent=False, absent_val=None):
"""
Return the audit's value for the given audit key (like 'IMSI' or 'IMSI-ACC').
Any kind of value may occur multiple times in a profile. When all of these agree to the same unambiguous value,
return that value. When they do not agree, raise a ValueError.
"""
# key should be a string, but if someone passes a ConfigurableParameter, just use its default name
if ConfigurableParameter.is_super_of(key):
key = key.get_name()
assert isinstance(key, str)
v = self.get(key)
if v is None and allow_absent:
return absent_val
if not isinstance(v, set):
raise ValueError(f'audit value should be a set(), got {v!r}')
if len(v) != 1:
raise ValueError(f'expected a single value for {key}, got {v!r}')
v = tuple(v)[0]
return v
@staticmethod
def audit_val_to_str(v):
"""
Usually, we want to see a single value in an audit. Still, to be able to collect multiple ambiguous values,
audit values are always python sets. Turn it into a nice string representation: only the value when it is
unambiguous, otherwise a list of the ambiguous values.
A value may also be completely absent, then return 'not present'.
"""
def try_single_val(w):
'change single-entry sets to just the single value'
if isinstance(w, set):
if len(w) == 1:
return tuple(w)[0]
if len(w) == 0:
return None
return w
v = try_single_val(v)
if isinstance(v, bytes):
v = bytes_to_hexstr(v)
if v is None:
return 'not present'
return str(v)
def get_val_str(self, key):
"""Return a string of the value stored for the given key"""
return UppAudit.audit_val_to_str(self.get(key))
def add_values(self, src:dict):
"""self and src are both a dict of sets.
For example from
self == { 'a': set((123,)) }
and
src == { 'a': set((456,)), 'b': set((789,)) }
then after this function call:
self == { 'a': set((123, 456,)), 'b': set((789,)) }
"""
assert isinstance(src, dict)
for key, srcvalset in src.items():
dstvalset = self.get(key)
if dstvalset is None:
dstvalset = set()
self[key] = dstvalset
dstvalset.add(srcvalset)
def __str__(self):
return '\n'.join(f'{key}: {self.get_val_str(key)}' for key in sorted(self.keys()))
class BatchAudit(list):
"""
Collect UppAudit instances for a batch of UPP, for example from a personalization.BatchPersonalization.
Produce an output CSV.
Usage example:
ba = BatchAudit(params=(personalization.Iccid, ))
for upp_der in upps:
ba.add_audit(upp_der)
print(ba.summarize())
with open('output.csv', 'wb') as csv_data:
csv_str = io.TextIOWrapper(csv_data, 'utf-8', newline='')
csv.writer(csv_str).writerows( ba.to_csv_rows() )
csv_str.flush()
BatchAudit itself is a list, callers may use the standard python list API to access the UppAudit instances.
"""
def __init__(self, params:List):
assert params
self.params = params
def add_audit(self, upp_der:bytes):
audit = UppAudit.from_der(upp_der, self.params)
self.append(audit)
return audit
def summarize(self):
batch_audit = UppAudit()
audits = self
if len(audits) > 2:
val_sep = ', ..., '
else:
val_sep = ', '
first_audit = None
last_audit = None
if len(audits) >= 1:
first_audit = audits[0]
if len(audits) >= 2:
last_audit = audits[-1]
if first_audit:
if last_audit:
for key in first_audit.keys():
first_val = first_audit.get_val_str(key)
last_val = last_audit.get_val_str(key)
if first_val == last_val:
val = first_val
else:
val_sep_with_newline = f"{val_sep.rstrip()}\n{' ' * (len(key) + 2)}"
val = val_sep_with_newline.join((first_val, last_val))
batch_audit[key] = val
else:
batch_audit.update(first_audit)
return batch_audit
def to_csv_rows(self, headers=True, sort_key=None):
'''generator that yields all audits' values as rows, useful feed to a csv.writer.'''
columns = set()
for audit in self:
columns.update(audit.keys())
columns = tuple(sorted(columns, key=sort_key))
if headers:
yield columns
for audit in self:
yield (audit.get_single_val(col, allow_absent=True, absent_val="") for col in columns)
def bytes_to_hexstr(b:bytes, sep=''):
return sep.join(f'{x:02x}' for x in b)
def esim_profile_introspect(upp):
pes = ProfileElementSequence.from_der(upp.read())
d = {}
d['upp'] = repr(pes)
def show_bytes_as_hexdump(item):
if isinstance(item, bytes):
return bytes_to_hexstr(item)
if isinstance(item, list):
return list(show_bytes_as_hexdump(i) for i in item)
if isinstance(item, tuple):
return tuple(show_bytes_as_hexdump(i) for i in item)
if isinstance(item, dict):
d = {}
for k, v in item.items():
d[k] = show_bytes_as_hexdump(v)
return d
return item
l = list((pe.type, show_bytes_as_hexdump(pe.decoded)) for pe in pes)
d['pp'] = pprint.pformat(l, width=120)
return d

View File

@@ -1,228 +0,0 @@
# Implementation of SimAlliance/TCA Interoperable Profile handling: parameter sources for batch personalization.
#
# (C) 2025 by sysmocom - s.f.m.c. GmbH <info@sysmocom.de>
#
# Author: nhofmeyr@sysmocom.de
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import secrets
import re
from osmocom.utils import b2h
class ParamSourceExn(Exception):
pass
class ParamSourceExhaustedExn(ParamSourceExn):
pass
class ParamSourceUndefinedExn(ParamSourceExn):
pass
class ParamSource:
"""abstract parameter source. For usage, see personalization.BatchPersonalization."""
# This name should be short but descriptive, useful for a user interface, like 'random decimal digits'.
name = "none"
numeric_base = None # or 10 or 16
@classmethod
def from_str(cls, s:str):
"""Subclasses implement this:
if a parameter source defines some string input magic, override this function.
For example, a RandomDigitSource derives the number of digits from the string length,
so the user can enter '0000' to get a four digit random number."""
return cls(s)
def get_next(self, csv_row:dict=None):
"""Subclasses implement this: return the next value from the parameter source.
When there are no more values from the source, raise a ParamSourceExhaustedExn.
This default implementation is an empty source."""
raise ParamSourceExhaustedExn()
class ConstantSource(ParamSource):
"""one value for all"""
name = "constant"
def __init__(self, val:str):
self.val = val
def get_next(self, csv_row:dict=None):
return self.val
class InputExpandingParamSource(ParamSource):
@classmethod
def expand_str(cls, s:str):
# user convenience syntax '0*32' becomes '00000000000000000000000000000000'
if "*" not in s:
return s
tokens = re.split(r"([^ \t]+)[ \t]*\*[ \t]*([0-9]+)", s)
if len(tokens) < 3:
return s
parts = []
for unchanged, snippet, repeat_str in zip(tokens[0::3], tokens[1::3], tokens[2::3]):
parts.append(unchanged)
repeat = int(repeat_str)
parts.append(snippet * repeat)
return "".join(parts)
@classmethod
def from_str(cls, s:str):
return cls(cls.expand_str(s))
class DecimalRangeSource(InputExpandingParamSource):
"""abstract: decimal numbers with a value range"""
numeric_base = 10
def __init__(self, num_digits, first_value, last_value):
"""
See also from_str().
All arguments are integer values, and are converted to int if necessary, so a string of an integer is fine.
num_digits: fixed number of digits (possibly with leading zeros) to generate.
first_value, last_value: the decimal range in which to provide digits.
"""
num_digits = int(num_digits)
first_value = int(first_value)
last_value = int(last_value)
assert num_digits > 0
assert first_value <= last_value
self.num_digits = num_digits
self.val_first_last = (first_value, last_value)
def val_to_digit(self, val:int):
return "%0*d" % (self.num_digits, val) # pylint: disable=consider-using-f-string
@classmethod
def from_str(cls, s:str):
s = cls.expand_str(s)
if ".." in s:
first_str, last_str = s.split('..')
first_str = first_str.strip()
last_str = last_str.strip()
else:
first_str = s.strip()
last_str = None
first_value = int(first_str)
last_value = int(last_str) if last_str is not None else "9" * len(first_str)
return cls(num_digits=len(first_str), first_value=first_value, last_value=last_value)
class RandomSourceMixin:
random_impl = secrets.SystemRandom()
class RandomDigitSource(DecimalRangeSource, RandomSourceMixin):
"""return a different sequence of random decimal digits each"""
name = "random decimal digits"
used_keys = set()
def get_next(self, csv_row:dict=None):
# try to generate random digits that are always different from previously produced random bytes
attempts = 10
while True:
val = self.random_impl.randint(*self.val_first_last)
if val in RandomDigitSource.used_keys:
attempts -= 1
if attempts:
continue
RandomDigitSource.used_keys.add(val)
break
return self.val_to_digit(val)
class RandomHexDigitSource(InputExpandingParamSource, RandomSourceMixin):
"""return a different sequence of random hexadecimal digits each"""
name = "random hexadecimal digits"
numeric_base = 16
used_keys = set()
def __init__(self, num_digits):
"""see from_str()"""
num_digits = int(num_digits)
if num_digits < 1:
raise ValueError("zero number of digits")
# hex digits always come in two
if (num_digits & 1) != 0:
raise ValueError(f"hexadecimal value should have even number of digits, not {num_digits}")
self.num_digits = num_digits
def get_next(self, csv_row:dict=None):
# try to generate random bytes that are always different from previously produced random bytes
attempts = 10
while True:
val = self.random_impl.randbytes(self.num_digits // 2)
if val in RandomHexDigitSource.used_keys:
attempts -= 1
if attempts:
continue
RandomHexDigitSource.used_keys.add(val)
break
return b2h(val)
@classmethod
def from_str(cls, s:str):
s = cls.expand_str(s)
return cls(num_digits=len(s.strip()))
class IncDigitSource(DecimalRangeSource):
"""incrementing sequence of digits"""
name = "incrementing decimal digits"
def __init__(self, num_digits, first_value, last_value):
super().__init__(num_digits, first_value, last_value)
self.next_val = None
self.reset()
def reset(self):
"""Restart from the first value of the defined range passed to __init__()."""
self.next_val = self.val_first_last[0]
def get_next(self, csv_row:dict=None):
val = self.next_val
if val is None:
raise ParamSourceExhaustedExn()
returnval = self.val_to_digit(val)
val += 1
if val > self.val_first_last[1]:
self.next_val = None
else:
self.next_val = val
return returnval
class CsvSource(ParamSource):
"""apply a column from a CSV row, as passed in to ParamSource.get_next(csv_row)"""
name = "from CSV"
def __init__(self, csv_column):
"""
csv_column: column name indicating the column to use for this parameter.
This name is used in get_next(): the caller passes the current CSV row to get_next(), from which
CsvSource picks the column with the name matching csv_column.
"""
self.csv_column = csv_column
def get_next(self, csv_row:dict=None):
val = None
if csv_row:
val = csv_row.get(self.csv_column)
if not val:
raise ParamSourceUndefinedExn(f"no value for CSV column {self.csv_column!r}")
return val

View File

@@ -17,24 +17,12 @@
import abc
import io
import os
import re
import pprint
from typing import List, Tuple, Generator, Optional
from typing import List, Tuple
from construct.core import StreamError
from osmocom.tlv import camel_to_snake
from osmocom.utils import hexstr
from pySim.utils import enc_iccid, dec_iccid, enc_imsi, dec_imsi, h2b, b2h, rpad, sanitize_iccid
from pySim.ts_31_102 import EF_AD
from pySim.utils import enc_iccid, enc_imsi, h2b, rpad, sanitize_iccid
from pySim.esim.saip import ProfileElement, ProfileElementSequence
from pySim.ts_51_011 import EF_SMSP
from pySim.esim.saip import param_source
from pySim.esim.saip import ProfileElement, ProfileElementSD, ProfileElementSequence
from pySim.esim.saip import SecurityDomainKey, SecurityDomainKeyComponent
from pySim.global_platform import KeyUsageQualifier, KeyType
def unrpad(s: hexstr, c='f') -> hexstr:
return hexstr(s.rstrip(c))
def remove_unwanted_tuples_from_list(l: List[Tuple], unwanted_keys: List[str]) -> List[Tuple]:
"""In a list of tuples, remove all tuples whose first part equals 'unwanted_key'."""
@@ -55,6 +43,7 @@ class ClassVarMeta(abc.ABCMeta):
x = super().__new__(metacls, name, bases, namespace)
for k, v in kwargs.items():
setattr(x, k, v)
setattr(x, 'name', camel_to_snake(name))
return x
class ConfigurableParameter(abc.ABC, metaclass=ClassVarMeta):
@@ -74,7 +63,6 @@ class ConfigurableParameter(abc.ABC, metaclass=ClassVarMeta):
min_len: minimum length of an input str; min_len = 4
max_len: maximum length of an input str; max_len = 8
allow_len: permit only specific lengths; allow_len = (8, 16, 32)
numeric_base: indicate hex / decimal, if any; numeric_base = None; numeric_base = 10; numeric_base = 16
Subclasses may change the meaning of these by overriding validate_val(), for example that the length counts
resulting bytes instead of a hexstring length. Most subclasses will be covered by the default validate_val().
@@ -129,8 +117,6 @@ class ConfigurableParameter(abc.ABC, metaclass=ClassVarMeta):
max_len = None
allow_len = None # a list of specific lengths
example_input = None
default_source = None # a param_source.ParamSource subclass
numeric_base = None # or 10 or 16
def __init__(self, input_value=None):
self.input_value = input_value # the raw input value as given by caller
@@ -192,28 +178,19 @@ class ConfigurableParameter(abc.ABC, metaclass=ClassVarMeta):
if cls.allow_chars is not None:
if any(c not in cls.allow_chars for c in val):
raise ValueError(f"invalid characters in input value {val!r}, valid chars are {cls.allow_chars}")
elif isinstance(val, io.BytesIO):
val = val.getvalue()
if hasattr(val, '__len__'):
val_len = len(val)
else:
# e.g. int length
val_len = len(str(val))
if cls.allow_len is not None:
l = cls.allow_len
# cls.allow_len could be one int, or a tuple of ints. Wrap a single int also in a tuple.
if not isinstance(l, (tuple, list)):
l = (l,)
if val_len not in l:
raise ValueError(f'length must be one of {cls.allow_len}, not {val_len}: {val!r}')
if len(val) not in l:
raise ValueError(f'length must be one of {cls.allow_len}, not {len(val)}: {val!r}')
if cls.min_len is not None:
if val_len < cls.min_len:
raise ValueError(f'length must be at least {cls.min_len}, not {val_len}: {val!r}')
if len(val) < cls.min_len:
raise ValueError(f'length must be at least {cls.min_len}, not {len(val)}: {val!r}')
if cls.max_len is not None:
if val_len > cls.max_len:
raise ValueError(f'length must be at most {cls.max_len}, not {val_len}: {val!r}')
if len(val) > cls.max_len:
raise ValueError(f'length must be at most {cls.max_len}, not {len(val)}: {val!r}')
return val
@classmethod
@@ -222,49 +199,6 @@ class ConfigurableParameter(abc.ABC, metaclass=ClassVarMeta):
Write the given val in the right format in all the right places in pes."""
pass
@classmethod
def get_value_from_pes(cls, pes: ProfileElementSequence):
"""Same as get_values_from_pes() but expecting a single value.
get_values_from_pes() may return values like this:
[{ 'AlgorithmID': 'Milenage' }, { 'AlgorithmID': 'Milenage' }]
This ensures that all these entries are identical and would return only
{ 'AlgorithmID': 'Milenage' }.
This is relevant for any profile element that may appear multiple times in the same PES (only a few),
where each occurrence should reflect the same value (all currently known parameters).
"""
val = None
for v in cls.get_values_from_pes(pes):
if val is None:
val = v
elif val != v:
raise ValueError(f'get_value_from_pes(): got distinct values: {val!r} != {v!r}')
return val
@classmethod
@abc.abstractmethod
def get_values_from_pes(cls, pes: ProfileElementSequence) -> Generator:
"""This is what subclasses implement: yield all values from a decoded profile package.
Find all values in the pes, and yield them decoded to a valid cls.input_value format.
Should be a generator function, i.e. use 'yield' instead of 'return'.
Yielded value must be a dict(). Usually, an implementation will return only one key, like
{ "ICCID": "1234567890123456789" }
Some implementations have more than one value to return, like
{ "IMSI": "00101012345678", "IMSI-ACC" : "5" }
Implementation example:
for pe in pes:
if my_condition(pe):
yield { cls.name: b2h(my_bin_value_from(pe)) }
"""
pass
@classmethod
def get_len_range(cls):
"""considering all of min_len, max_len and allow_len, get a tuple of the resulting (min, max) of permitted
@@ -285,20 +219,6 @@ class ConfigurableParameter(abc.ABC, metaclass=ClassVarMeta):
return (None, None)
return (min(vals), max(vals))
@classmethod
def get_typical_input_len(cls):
'''return a good length to use as the visible width of a user interface input field.
May be overridden by subclasses.
This default implementation returns the maximum allowed value length -- a good fit for most subclasses.
'''
return cls.get_len_range()[1] or 16
@classmethod
def is_super_of(cls, other_class):
try:
return issubclass(other_class, cls)
except TypeError:
return False
class DecimalParam(ConfigurableParameter):
"""Decimal digits. The input value may be a string of decimal digits like '012345', or an int. The output of
@@ -306,7 +226,6 @@ class DecimalParam(ConfigurableParameter):
"""
allow_types = (str, int)
allow_chars = '0123456789'
numeric_base = 10
@classmethod
def validate_val(cls, val):
@@ -337,21 +256,9 @@ class DecimalHexParam(DecimalParam):
# a DecimalHexParam subclass expects the apply_val() input to be a bytes instance ready for the pes
return h2b(val)
@classmethod
def decimal_hex_to_str(cls, val):
'useful for get_values_from_pes() implementations of subclasses'
if isinstance(val, bytes):
val = b2h(val)
assert isinstance(val, hexstr)
if cls.rpad is not None:
c = cls.rpad_char or 'f'
val = unrpad(val, c)
return val.to_bytes().decode('ascii')
class IntegerParam(ConfigurableParameter):
allow_types = (str, int)
allow_chars = '0123456789'
numeric_base = 10
# two integers, if the resulting int should be range limited
min_val = None
@@ -372,28 +279,14 @@ class IntegerParam(ConfigurableParameter):
raise ValueError(f'Value {val} is out of range, must be [{cls.min_val}..{cls.max_val}]')
return val
@classmethod
def get_values_from_pes(cls, pes: ProfileElementSequence):
for valdict in super().get_values_from_pes(pes):
for key, val in valdict.items():
if isinstance(val, int):
valdict[key] = str(val)
yield valdict
class BinaryParam(ConfigurableParameter):
allow_types = (str, io.BytesIO, bytes, bytearray, int)
allow_types = (str, io.BytesIO, bytes, bytearray)
allow_chars = '0123456789abcdefABCDEF'
strip_chars = ' \t\r\n'
numeric_base = 16
default_source = param_source.RandomHexDigitSource
@classmethod
def validate_val(cls, val):
# take care that min_len and max_len are applied to the binary length by converting to bytes first
if isinstance(val, int):
min_len, _max_len = cls.get_len_range()
val = '%0*d' % (min_len, val)
if isinstance(val, str):
if cls.strip_chars is not None:
val = ''.join(c for c in val if c not in cls.strip_chars)
@@ -408,80 +301,6 @@ class BinaryParam(ConfigurableParameter):
val = super().validate_val(val)
return bytes(val)
@classmethod
def get_typical_input_len(cls):
# override to return twice the length, because of hex digits.
min_len, max_len = cls.get_len_range()
if max_len is None:
return None
# two hex characters per value octet.
# (maybe *3 to also allow for spaces?)
return max_len * 2
class EnumParam(ConfigurableParameter):
value_map = {
# For example:
#'Meaningful label for value 23': 0x23,
# Where 0x23 is a valid value to use for apply_val().
}
_value_map_reverse = None
@classmethod
def validate_val(cls, val):
orig_val = val
enum_val = None
if isinstance(val, str):
enum_name = val
enum_val = cls.map_name_to_val(enum_name)
# if the str is not one of the known value_map.keys(), is it maybe one of value_map.keys()?
if enum_val is None and val in cls.value_map.values():
enum_val = val
if enum_val not in cls.value_map.values():
raise ValueError(f"{cls.get_name()}: invalid argument: {orig_val!r}. Valid arguments are:"
f" {', '.join(cls.value_map.keys())}")
return enum_val
@classmethod
def map_name_to_val(cls, name:str, strict=True):
val = cls.value_map.get(name)
if val is not None:
return val
clean_name = cls.clean_name_str(name)
for k, v in cls.value_map.items():
if clean_name == cls.clean_name_str(k):
return v
if strict:
raise ValueError(f"Problem in {cls.get_name()}: {name!r} is not a known value."
f" Known values are: {cls.value_map.keys()!r}")
return None
@classmethod
def map_val_to_name(cls, val, strict=False) -> str:
if cls._value_map_reverse is None:
cls._value_map_reverse = dict((v, k) for k, v in cls.value_map.items())
name = cls._value_map_reverse.get(val)
if name:
return name
if strict:
raise ValueError(f"Problem in {cls.get_name()}: {val!r} ({type(val)}) is not a known value."
f" Known values are: {cls.value_map.values()!r}")
return None
@classmethod
def name_normalize(cls, name:str) -> str:
return cls.map_val_to_name(cls.map_name_to_val(name))
@classmethod
def clean_name_str(cls, val):
return re.sub('[^0-9A-Za-z-_]', '', val).lower()
class Iccid(DecimalParam):
"""ICCID Parameter. Input: string of decimal digits.
@@ -490,7 +309,6 @@ class Iccid(DecimalParam):
min_len = 18
max_len = 20
example_input = '998877665544332211'
default_source = param_source.IncDigitSource
@classmethod
def validate_val(cls, val):
@@ -504,17 +322,6 @@ class Iccid(DecimalParam):
# patch MF/EF.ICCID
file_replace_content(pes.get_pe_for_type('mf').decoded['ef-iccid'], h2b(enc_iccid(val)))
@classmethod
def get_values_from_pes(cls, pes: ProfileElementSequence):
padded = b2h(pes.get_pe_for_type('header').decoded['iccid'])
iccid = unrpad(padded)
yield { cls.name: iccid }
for pe in pes.get_pes_for_type('mf'):
iccid_f = pe.files.get('ef-iccid', None)
if iccid_f is not None:
yield { cls.name: dec_iccid(b2h(iccid_f.body)) }
class Imsi(DecimalParam):
"""Configurable IMSI. Expects value to be a string of digits. Automatically sets the ACC to
the last digit of the IMSI."""
@@ -523,7 +330,6 @@ class Imsi(DecimalParam):
min_len = 6
max_len = 15
example_input = '00101' + ('0' * 10)
default_source = param_source.IncDigitSource
@classmethod
def apply_val(cls, pes: ProfileElementSequence, val):
@@ -536,18 +342,6 @@ class Imsi(DecimalParam):
file_replace_content(pe.decoded['ef-acc'], acc.to_bytes(2, 'big'))
# TODO: DF.GSM_ACCESS if not linked?
@classmethod
def get_values_from_pes(cls, pes: ProfileElementSequence):
for pe in pes.get_pes_for_type('usim'):
imsi_f = pe.files.get('ef-imsi', None)
acc_f = pe.files.get('ef-acc', None)
y = {}
if imsi_f:
y[cls.name] = dec_imsi(b2h(imsi_f.body))
if acc_f:
y[cls.name + '-ACC'] = b2h(acc_f.body)
yield y
class SmspTpScAddr(ConfigurableParameter):
"""Configurable SMSC (SMS Service Centre) TP-SC-ADDR. Expects to be a phone number in national or
international format (designated by a leading +). Automatically sets the NPI to E.164 and the TON based on
@@ -556,45 +350,25 @@ class SmspTpScAddr(ConfigurableParameter):
name = 'SMSP-TP-SC-ADDR'
allow_chars = '+0123456789'
strip_chars = ' \t\r\n'
numeric_base = 10
max_len = 21 # '+' and 20 digits
min_len = 1
example_input = '+49301234567'
default_source = param_source.ConstantSource
@staticmethod
def str_to_tuple(addr_str):
@classmethod
def validate_val(cls, val):
val = super().validate_val(val)
addr_str = str(val)
if addr_str[0] == '+':
digits = addr_str[1:]
international = True
else:
digits = addr_str
international = False
return (international, digits)
@staticmethod
def tuple_to_str(addr_tuple):
international, digits = addr_tuple
if international:
ret = '+'
else:
ret = ''
ret += digits
return ret
@classmethod
def validate_val(cls, val):
val = super().validate_val(val)
addr_tuple = cls.str_to_tuple(str(val))
international, digits = addr_tuple
if len(digits) > 20:
raise ValueError(f'TP-SC-ADDR must not exceed 20 digits: {digits!r}')
if not digits.isdecimal():
raise ValueError(f'TP-SC-ADDR must only contain decimal digits: {digits!r}')
return addr_tuple
return (international, digits)
@classmethod
def apply_val(cls, pes: ProfileElementSequence, val):
@@ -624,303 +398,92 @@ class SmspTpScAddr(ConfigurableParameter):
# re-generate the pe.decoded member from the File instance
pe.file2pe(f_smsp)
@classmethod
def get_values_from_pes(cls, pes: ProfileElementSequence):
for pe in pes.get_pes_for_type('usim'):
f_smsp = pe.files.get('ef-smsp', None)
if f_smsp is None:
continue
try:
ef_smsp = EF_SMSP()
ef_smsp_dec = ef_smsp.decode_record_bin(f_smsp.body, 1)
except IndexError:
continue
tp_sc_addr = ef_smsp_dec.get('tp_sc_addr', None)
digits = tp_sc_addr.get('call_number', None)
ton_npi = tp_sc_addr.get('ton_npi', None)
international = ton_npi.get('type_of_number', None)
international = (international == 'international')
yield { cls.name: cls.tuple_to_str((international, digits)) }
class MncLen(ConfigurableParameter):
"""MNC length. Must be either 2 or 3. Sets only the MNC length field in EF-AD (Administrative Data)."""
name = 'MNC-LEN'
allow_chars = '23'
strip_chars = ' \t\r\n'
numeric_base = 10
max_len = 1
min_len = 1
example_input = '2'
default_source = param_source.ConstantSource
@classmethod
def validate_val(cls, val):
val = super().validate_val(val)
val = int(val)
if val not in (2, 3):
raise ValueError(f"MNC-LEN must be either 2 or 3, not {val!r}")
return val
@classmethod
def apply_val(cls, pes: ProfileElementSequence, val):
"""val must be an int: either 2 or 3"""
for pe in pes.get_pes_for_type('usim'):
if not hasattr(pe, 'files'):
continue
f_ad = pe.files.get('ef-ad')
if not f_ad:
continue
# decode existing values
if not f_ad.body:
continue
try:
ef_ad = EF_AD()
ef_ad_dec = ef_ad.decode_bin(f_ad.body)
except StreamError:
continue
if 'mnc_len' not in ef_ad_dec:
continue
# change mnc_len
ef_ad_dec['mnc_len'] = val
# re-encode into the File body
f_ad.body = ef_ad.encode_bin(ef_ad_dec)
pe.file2pe(f_ad)
@classmethod
def get_values_from_pes(cls, pes: ProfileElementSequence):
for pe in pes.get_pes_for_type('usim'):
if not hasattr(pe, 'files'):
continue
f_ad = pe.files.get('ef-ad', None)
if f_ad is None:
continue
try:
ef_ad = EF_AD()
ef_ad_dec = ef_ad.decode_bin(f_ad.body)
except StreamError:
continue
mnc_len = ef_ad_dec.get('mnc_len', None)
if mnc_len is None:
continue
yield { cls.name: str(mnc_len) }
class SdKey(BinaryParam):
"""Configurable Security Domain (SD) Key. Value is presented as bytes.
Non-abstract implementations are generated in SdKey.generate_sd_key_classes"""
class SdKey(BinaryParam, metaclass=ClassVarMeta):
"""Configurable Security Domain (SD) Key. Value is presented as bytes."""
# these will be set by subclasses
key_type = None
kvn = None
reserved_kvn = tuple() # tuple of all reserved kvn for a given SCPxx
key_id = None
kvn = None
key_usage_qual = None
@classmethod
def apply_val(cls, pes: ProfileElementSequence, val):
set_components = [ SecurityDomainKeyComponent(cls.key_type, val) ]
for pe in pes.pe_list:
if pe.type != 'securityDomain':
continue
assert isinstance(pe, ProfileElementSD)
key = pe.find_key(key_version_number=cls.kvn, key_id=cls.key_id)
if not key:
# Could not find matching key to patch, create a new one
key = SecurityDomainKey(
key_version_number=cls.kvn,
key_id=cls.key_id,
key_usage_qualifier=cls.key_usage_qual,
key_components=set_components,
)
pe.add_key(key)
else:
# A key of this KVN and ID already exists in the profile.
# Keep the key_usage_qualifier as it was in the profile, so skip this here:
# key.key_usage_qualifier = cls.key_usage_qual
key.key_components = set_components
@classmethod
def get_values_from_pes(cls, pes: ProfileElementSequence):
for pe in pes.pe_list:
if pe.type != 'securityDomain':
continue
assert isinstance(pe, ProfileElementSD)
key = pe.find_key(key_version_number=cls.kvn, key_id=cls.key_id)
if not key:
continue
kc = key.get_key_component(cls.key_type)
if kc:
yield { cls.name: b2h(kc) }
NO_OP = (('', {}))
LEN_128 = (16,)
LEN_128_192_256 = (16, 24, 32)
LEN_128_256 = (16, 32)
DES = ('DES', dict(key_type=KeyType.des, allow_len=LEN_128) )
AES = ('AES', dict(key_type=KeyType.aes, allow_len=LEN_128_192_256) )
ENC = ('ENC', dict(key_id=0x01, key_usage_qual=0x18) )
MAC = ('MAC', dict(key_id=0x02, key_usage_qual=0x14) )
DEK = ('DEK', dict(key_id=0x03, key_usage_qual=0x48) )
TLSPSK_PSK = ('TLSPSK', dict(key_type=KeyType.tls_psk, key_id=0x01, key_usage_qual=0x3c, allow_len=LEN_128_192_256) )
TLSPSK_DEK = ('DEK', dict(key_id=0x02, key_usage_qual=0x48) )
# THIS IS THE LIST that controls which SdKeyXxx subclasses exist:
SD_KEY_DEFS = (
# name KVN x variants x variants
('SCP02', (0x20, 0x21, 0x22, 0xff), (AES, ), (ENC, MAC, DEK) ),
('SCP03', (0x30, 0x31, 0x32), (AES, ), (ENC, MAC, DEK) ),
('SCP80', (0x01, 0x02, 0x03), (DES, AES), (ENC, MAC, DEK) ),
# key_id=1
('SCP81', (0x40, 0x41, 0x42), (TLSPSK_PSK, ), ),
# key_id=2
('SCP81', (0x40, 0x41, 0x42), (DES, AES), (TLSPSK_DEK, ) ),
)
all_implementations = None
@classmethod
def generate_sd_key_classes(cls, sd_key_defs=SD_KEY_DEFS):
'''This generates python classes to be exported in this module, as subclasses of class SdKey.
We create SdKey subclasses dynamically from a list.
You can list all of them via:
from pySim.esim.saip.personalization import SdKey
SdKey.all_implementations
or
print('\n'.join(sorted(f'{x.__name__}\t{x.name}' for x in SdKey.all_implementations)))
at time of writing this comment, this prints:
SdKeyScp02Kvn20AesDek SCP02-KVN20-AES-DEK
SdKeyScp02Kvn20AesEnc SCP02-KVN20-AES-ENC
SdKeyScp02Kvn20AesMac SCP02-KVN20-AES-MAC
SdKeyScp02Kvn21AesDek SCP02-KVN21-AES-DEK
SdKeyScp02Kvn21AesEnc SCP02-KVN21-AES-ENC
SdKeyScp02Kvn21AesMac SCP02-KVN21-AES-MAC
SdKeyScp02Kvn22AesDek SCP02-KVN22-AES-DEK
SdKeyScp02Kvn22AesEnc SCP02-KVN22-AES-ENC
SdKeyScp02Kvn22AesMac SCP02-KVN22-AES-MAC
SdKeyScp02KvnffAesDek SCP02-KVNff-AES-DEK
SdKeyScp02KvnffAesEnc SCP02-KVNff-AES-ENC
SdKeyScp02KvnffAesMac SCP02-KVNff-AES-MAC
SdKeyScp03Kvn30AesDek SCP03-KVN30-AES-DEK
SdKeyScp03Kvn30AesEnc SCP03-KVN30-AES-ENC
SdKeyScp03Kvn30AesMac SCP03-KVN30-AES-MAC
SdKeyScp03Kvn31AesDek SCP03-KVN31-AES-DEK
SdKeyScp03Kvn31AesEnc SCP03-KVN31-AES-ENC
SdKeyScp03Kvn31AesMac SCP03-KVN31-AES-MAC
SdKeyScp03Kvn32AesDek SCP03-KVN32-AES-DEK
SdKeyScp03Kvn32AesEnc SCP03-KVN32-AES-ENC
SdKeyScp03Kvn32AesMac SCP03-KVN32-AES-MAC
SdKeyScp80Kvn01AesDek SCP80-KVN01-AES-DEK
SdKeyScp80Kvn01AesEnc SCP80-KVN01-AES-ENC
SdKeyScp80Kvn01AesMac SCP80-KVN01-AES-MAC
SdKeyScp80Kvn01DesDek SCP80-KVN01-DES-DEK
SdKeyScp80Kvn01DesEnc SCP80-KVN01-DES-ENC
SdKeyScp80Kvn01DesMac SCP80-KVN01-DES-MAC
SdKeyScp80Kvn02AesDek SCP80-KVN02-AES-DEK
SdKeyScp80Kvn02AesEnc SCP80-KVN02-AES-ENC
SdKeyScp80Kvn02AesMac SCP80-KVN02-AES-MAC
SdKeyScp80Kvn02DesDek SCP80-KVN02-DES-DEK
SdKeyScp80Kvn02DesEnc SCP80-KVN02-DES-ENC
SdKeyScp80Kvn02DesMac SCP80-KVN02-DES-MAC
SdKeyScp80Kvn03AesDek SCP80-KVN03-AES-DEK
SdKeyScp80Kvn03AesEnc SCP80-KVN03-AES-ENC
SdKeyScp80Kvn03AesMac SCP80-KVN03-AES-MAC
SdKeyScp80Kvn03DesDek SCP80-KVN03-DES-DEK
SdKeyScp80Kvn03DesEnc SCP80-KVN03-DES-ENC
SdKeyScp80Kvn03DesMac SCP80-KVN03-DES-MAC
SdKeyScp81Kvn40AesDek SCP81-KVN40-AES-DEK
SdKeyScp81Kvn40DesDek SCP81-KVN40-DES-DEK
SdKeyScp81Kvn40Tlspsk SCP81-KVN40-TLSPSK
SdKeyScp81Kvn41AesDek SCP81-KVN41-AES-DEK
SdKeyScp81Kvn41DesDek SCP81-KVN41-DES-DEK
SdKeyScp81Kvn41Tlspsk SCP81-KVN41-TLSPSK
SdKeyScp81Kvn42AesDek SCP81-KVN42-AES-DEK
SdKeyScp81Kvn42DesDek SCP81-KVN42-DES-DEK
SdKeyScp81Kvn42Tlspsk SCP81-KVN42-TLSPSK
'''
SdKey.all_implementations = []
def camel(s):
return s[:1].upper() + s[1:].lower()
def do_variants(name, kvn, remaining_variants, labels=[], attrs={}):
'recurse to unfold as many variants as there may be'
if remaining_variants:
# not a leaf node, collect more labels and attrs
variants = remaining_variants[0]
remaining_variants = remaining_variants[1:]
for label, valdict in variants:
# pass copies to recursion
inner_labels = list(labels)
inner_attrs = dict(attrs)
inner_labels.append(label)
inner_attrs.update(valdict)
do_variants(name, kvn, remaining_variants,
labels=inner_labels,
attrs=inner_attrs)
def _apply_sd(cls, pe: ProfileElement, value):
assert pe.type == 'securityDomain'
for key in pe.decoded['keyList']:
if key['keyIdentifier'][0] == cls.key_id and key['keyVersionNumber'][0] == cls.kvn:
assert len(key['keyComponents']) == 1
key['keyComponents'][0]['keyData'] = value
return
# Could not find matching key to patch, create a new one
key = {
'keyUsageQualifier': bytes([cls.key_usage_qual]),
'keyIdentifier': bytes([cls.key_id]),
'keyVersionNumber': bytes([cls.kvn]),
'keyComponents': [
{ 'keyType': bytes([cls.key_type]), 'keyData': value },
]
}
pe.decoded['keyList'].append(key)
# leaf node. create a new class with all the accumulated vals
parts = [name, f'KVN{kvn:02x}',] + labels
cls_label = '-'.join(p for p in parts if p)
@classmethod
def apply_val(cls, pes: ProfileElementSequence, value):
for pe in pes.get_pes_for_type('securityDomain'):
cls._apply_sd(pe, value)
parts = ['Sd', 'Key', name, f'Kvn{kvn:02x}'] + labels
clsname = ''.join(camel(p) for p in parts)
class SdKeyScp80_01(SdKey, kvn=0x01, key_type=0x88, permitted_len=[16,24,32]): # AES key type
pass
class SdKeyScp80_01Kic(SdKeyScp80_01, key_id=0x01, key_usage_qual=0x18): # FIXME: ordering?
pass
class SdKeyScp80_01Kid(SdKeyScp80_01, key_id=0x02, key_usage_qual=0x14):
pass
class SdKeyScp80_01Kik(SdKeyScp80_01, key_id=0x03, key_usage_qual=0x48):
pass
max_key_len = attrs.get('allow_len')[-1]
class SdKeyScp81_01(SdKey, kvn=0x81): # FIXME
pass
class SdKeyScp81_01Psk(SdKeyScp81_01, key_id=0x01, key_type=0x85, key_usage_qual=0x3C):
pass
class SdKeyScp81_01Dek(SdKeyScp81_01, key_id=0x02, key_type=0x88, key_usage_qual=0x48):
pass
attrs.update({
'name' : cls_label,
'kvn': kvn,
'example_input': f'00*{max_key_len}',
})
class SdKeyScp02_20(SdKey, kvn=0x20, key_type=0x88, permitted_len=[16,24,32]): # AES key type
pass
class SdKeyScp02_20Enc(SdKeyScp02_20, key_id=0x01, key_usage_qual=0x18):
pass
class SdKeyScp02_20Mac(SdKeyScp02_20, key_id=0x02, key_usage_qual=0x14):
pass
class SdKeyScp02_20Dek(SdKeyScp02_20, key_id=0x03, key_usage_qual=0x48):
pass
# below line is like
# class SdKeyScpNNKvnXXYyyZzz(SdKey):
# <set attrs>
cls_def = type(clsname, (cls,), attrs)
class SdKeyScp03_30(SdKey, kvn=0x30, key_type=0x88, permitted_len=[16,24,32]): # AES key type
pass
class SdKeyScp03_30Enc(SdKeyScp03_30, key_id=0x01, key_usage_qual=0x18):
pass
class SdKeyScp03_30Mac(SdKeyScp03_30, key_id=0x02, key_usage_qual=0x14):
pass
class SdKeyScp03_30Dek(SdKeyScp03_30, key_id=0x03, key_usage_qual=0x48):
pass
# for some unknown reason, subclassing from abc.ABC makes cls_def.__module__ == 'abc',
# but we don't want 'abc.SdKeyScp03Kvn32AesEnc'.
# Make sure it is 'pySim.esim.saip.personalization.SdKeyScp03Kvn32AesEnc'
cls_def.__module__ = __name__
class SdKeyScp03_31(SdKey, kvn=0x31, key_type=0x88, permitted_len=[16,24,32]): # AES key type
pass
class SdKeyScp03_31Enc(SdKeyScp03_31, key_id=0x01, key_usage_qual=0x18):
pass
class SdKeyScp03_31Mac(SdKeyScp03_31, key_id=0x02, key_usage_qual=0x14):
pass
class SdKeyScp03_31Dek(SdKeyScp03_31, key_id=0x03, key_usage_qual=0x48):
pass
globals()[clsname] = cls_def
SdKey.all_implementations.append(cls_def)
class SdKeyScp03_32(SdKey, kvn=0x32, key_type=0x88, permitted_len=[16,24,32]): # AES key type
pass
class SdKeyScp03_32Enc(SdKeyScp03_32, key_id=0x01, key_usage_qual=0x18):
pass
class SdKeyScp03_32Mac(SdKeyScp03_32, key_id=0x02, key_usage_qual=0x14):
pass
class SdKeyScp03_32Dek(SdKeyScp03_32, key_id=0x03, key_usage_qual=0x48):
pass
for items in sd_key_defs:
name, kvns = items[:2]
variants = items[2:]
for kvn in kvns:
do_variants(name, kvn, variants)
# this creates all of the classes named like SdKeyScp02Kvn20AesDek to be published in this python module:
SdKey.generate_sd_key_classes()
def obtain_all_pe_from_pelist(l: List[ProfileElement], wanted_type: str) -> ProfileElement:
return (pe for pe in l if pe.type == wanted_type)
@@ -939,8 +502,7 @@ class Puk(DecimalHexParam):
allow_len = 8
rpad = 16
keyReference = None
example_input = f'0*{allow_len}'
default_source = param_source.RandomDigitSource
example_input = '0' * allow_len
@classmethod
def apply_val(cls, pes: ProfileElementSequence, val):
@@ -954,14 +516,6 @@ class Puk(DecimalHexParam):
raise ValueError("input template UPP has unexpected structure:"
f" cannot find pukCode with keyReference={cls.keyReference}")
@classmethod
def get_values_from_pes(cls, pes: ProfileElementSequence):
mf_pes = pes.pes_by_naa['mf'][0]
for pukCodes in obtain_all_pe_from_pelist(mf_pes, 'pukCodes'):
for pukCode in pukCodes.decoded['pukCodes']:
if pukCode['keyReference'] == cls.keyReference:
yield { cls.name: cls.decimal_hex_to_str(pukCode['pukValue']) }
class Puk1(Puk):
name = 'PUK1'
keyReference = 0x01
@@ -975,8 +529,7 @@ class Pin(DecimalHexParam):
rpad = 16
min_len = 4
max_len = 8
example_input = f'0*{max_len}'
default_source = param_source.RandomDigitSource
example_input = '0' * max_len
keyReference = None
@staticmethod
@@ -998,24 +551,9 @@ class Pin(DecimalHexParam):
raise ValueError('input template UPP has unexpected structure:'
+ f' {cls.get_name()} cannot find pinCode with keyReference={cls.keyReference}')
@classmethod
def _read_all_pinvalues_from_pe(cls, pe: ProfileElement):
"This is a separate function because subclasses may feed different pe arguments."
for pinCodes in obtain_all_pe_from_pelist(pe, 'pinCodes'):
if pinCodes.decoded['pinCodes'][0] != 'pinconfig':
continue
for pinCode in pinCodes.decoded['pinCodes'][1]:
if pinCode['keyReference'] == cls.keyReference:
yield { cls.name: cls.decimal_hex_to_str(pinCode['pinValue']) }
@classmethod
def get_values_from_pes(cls, pes: ProfileElementSequence):
yield from cls._read_all_pinvalues_from_pe(pes.pes_by_naa['mf'][0])
class Pin1(Pin):
name = 'PIN1'
example_input = '0*4' # PIN are usually 4 digits
example_input = '0' * 4 # PIN are usually 4 digits
keyReference = 0x01
class Pin2(Pin1):
@@ -1034,14 +572,6 @@ class Pin2(Pin1):
raise ValueError('input template UPP has unexpected structure:'
+ f' {cls.get_name()} cannot find pinCode with keyReference={cls.keyReference} in {naa=}')
@classmethod
def get_values_from_pes(cls, pes: ProfileElementSequence):
for naa in pes.pes_by_naa:
if naa not in ['usim','isim','csim','telecom']:
continue
for pe in pes.pes_by_naa[naa]:
yield from cls._read_all_pinvalues_from_pe(pe)
class Adm1(Pin):
name = 'ADM1'
keyReference = 0x0A
@@ -1066,61 +596,26 @@ class AlgoConfig(ConfigurableParameter):
raise ValueError('input template UPP has unexpected structure:'
f' {cls.__name__} cannot find algoParameter with key={cls.algo_config_key}')
@classmethod
def get_values_from_pes(cls, pes: ProfileElementSequence):
for pe in pes.get_pes_for_type('akaParameter'):
algoConfiguration = pe.decoded['algoConfiguration']
if len(algoConfiguration) < 2:
continue
if algoConfiguration[0] != 'algoParameter':
continue
if not algoConfiguration[1]:
continue
val = algoConfiguration[1].get(cls.algo_config_key, None)
if val is None:
continue
if isinstance(val, bytes):
val = b2h(val)
# if it is an int (algorithmID), just pass thru as int
yield { cls.name: val }
class AlgorithmID(EnumParam, AlgoConfig):
'''use validate_val() from EnumParam, and apply_val() from AlgoConfig.
In get_values_from_pes(), return enum value names, not raw values.'''
name = "Algorithm"
# as in pySim/esim/asn1/saip/PE_Definitions-3.3.1.asn
value_map = {
"Milenage" : 1,
"TUAK" : 2,
"usim-test" : 3,
}
example_input = "Milenage"
default_source = param_source.ConstantSource
class AlgorithmID(DecimalParam, AlgoConfig):
algo_config_key = 'algorithmID'
# EnumParam.validate_val() returns the int values from value_map
allow_len = 1
example_input = 1 # Milenage
@classmethod
def get_values_from_pes(cls, pes: ProfileElementSequence):
# return enum names, not raw values.
# use of super(): this intends to call AlgoConfig.get_values_from_pes() so that the cls argument is this cls
# here (AlgorithmID); i.e. AlgoConfig.get_values_from_pes(pes) doesn't work, because AlgoConfig needs to look up
# cls.algo_config_key.
for d in super(cls, cls).get_values_from_pes(pes):
if cls.name in d:
# convert int to value string
val = d[cls.name]
d[cls.name] = cls.map_val_to_name(val, strict=True)
yield d
def validate_val(cls, val):
val = super().validate_val(val)
val = int(val)
valid = (1, 2, 3)
if val not in valid:
raise ValueError(f'Invalid algorithmID {val!r}, must be one of {valid}')
return val
class K(BinaryParam, AlgoConfig):
"""use validate_val() from BinaryParam, and apply_val() from AlgoConfig"""
name = 'K'
algo_config_key = 'key'
allow_len = (128 // 8, 256 // 8) # length in bytes (from BinaryParam); TUAK also allows 256 bit
example_input = f'00*{allow_len[0]}'
example_input = '00' * allow_len[0]
class Opc(K):
name = 'OPc'
@@ -1134,7 +629,6 @@ class MilenageRotationConstants(BinaryParam, AlgoConfig):
algo_config_key = 'rotationConstants'
allow_len = 5 # length in bytes (from BinaryParam)
example_input = '40 00 20 40 60'
default_source = param_source.ConstantSource
@classmethod
def validate_val(cls, val):
@@ -1165,7 +659,6 @@ class MilenageXoringConstants(BinaryParam, AlgoConfig):
' 00000000000000000000000000000002'
' 00000000000000000000000000000004'
' 00000000000000000000000000000008')
default_source = param_source.ConstantSource
class TuakNumberOfKeccak(IntegerParam, AlgoConfig):
"""Number of iterations of Keccak-f[1600] permutation as recomended by Section 7.2 of 3GPP TS 35.231"""
@@ -1174,4 +667,3 @@ class TuakNumberOfKeccak(IntegerParam, AlgoConfig):
min_val = 1
max_val = 255
example_input = '1'
default_source = param_source.ConstantSource

View File

@@ -30,7 +30,6 @@ import tempfile
import json
import abc
import inspect
import os
import cmd2
from cmd2 import CommandSet, with_default_category
@@ -553,85 +552,6 @@ class CardADF(CardDF):
return lchan.selected_file.application.export(as_json, lchan)
class JsonEditor:
"""Context manager for editing a JSON-encoded EF value in an external editor.
Writes the current JSON value (plus encode/decode examples as //-comments)
to a temporary file, opens the user's editor, then reads the result back
(stripping comment lines) and returns it as the context variable::
with JsonEditor(self._cmd, orig_json, ef) as edited_json:
if edited_json != orig_json:
...write back...
"""
def __init__(self, cmd, orig_json, ef):
self._cmd = cmd
self._orig_json = orig_json
self._ef = ef
self._file = None
@staticmethod
def _strip_comments(text: str) -> str:
"""Strip //-comment lines from text before JSON parsing."""
# TODO: also strip inline comments?
return '\n'.join(line for line in text.splitlines() if not line.lstrip().startswith('//'))
def _append_examples_as_comments(self, text_file) -> None:
"""Append encode/decode test vectors as //-comment lines to an open file.
The examples are taken from _test_de_encode and _test_decode class
attributes (same source as the auto-generated filesystem documentation).
The comment block is intentionally ignored on read-back by _strip_comments."""
vectors = []
for attr in ('_test_de_encode', '_test_decode'):
v = getattr(type(self._ef), attr, None)
if v:
vectors.extend(v)
if not vectors:
return
ef = self._ef
parts = [ef.fully_qualified_path_str()]
if ef.fid:
parts.append(f'({ef.fid.upper()})')
if ef.desc:
parts.append(f'- {ef.desc}')
text_file.write(f'\n\n// {" ".join(parts)}\n')
text_file.write('// Examples (ignored on save):\n')
for t in vectors:
if len(t) >= 3:
encoded, record_nr, decoded = t[0], t[1], t[2]
text_file.write(f'// record {record_nr}: {encoded}\n')
else:
encoded, decoded = t[0], t[1]
text_file.write(f'// file: {encoded}\n')
for line in json.dumps(decoded, indent=4, cls=JsonEncoder).splitlines():
text_file.write(f'// {line}\n')
def __enter__(self) -> object:
"""Write JSON + examples to a temp file, run the editor, return parsed result.
On JSONDecodeError the user is offered the option to re-open the file
and fix the mistake interactively. The temp file is removed by __exit__()
on success, or when the user declines to retry."""
self._file = tempfile.NamedTemporaryFile(prefix='pysim_', suffix='.json',
mode='w', delete=False)
json.dump(self._orig_json, self._file, indent=4, cls=JsonEncoder)
self._append_examples_as_comments(self._file)
self._file.close()
while True:
self._cmd.run_editor(self._file.name)
try:
with open(self._file.name, 'r') as f:
return json.loads(self._strip_comments(f.read()))
except json.JSONDecodeError as e:
self._cmd.perror(f'Invalid JSON: {e}')
answer = self._cmd.read_input('Re-open file for editing? [y]es/[n]o: ')
if answer not in ('y', 'yes'):
return self._orig_json
def __exit__(self, *args):
os.unlink(self._file.name)
class CardEF(CardFile):
"""EF (Entry File) in the smart card filesystem"""
@@ -737,8 +657,15 @@ class TransparentEF(CardEF):
def do_edit_binary_decoded(self, _opts):
"""Edit the JSON representation of the EF contents in an editor."""
(orig_json, _sw) = self._cmd.lchan.read_binary_dec()
ef = self._cmd.lchan.selected_file
with JsonEditor(self._cmd, orig_json, ef) as edited_json:
with tempfile.TemporaryDirectory(prefix='pysim_') as dirname:
filename = '%s/file' % dirname
# write existing data as JSON to file
with open(filename, 'w') as text_file:
json.dump(orig_json, text_file, indent=4, cls=JsonEncoder)
# run a text editor
self._cmd.run_editor(filename)
with open(filename, 'r') as text_file:
edited_json = json.load(text_file)
if edited_json == orig_json:
self._cmd.poutput("Data not modified, skipping write")
else:
@@ -1032,8 +959,15 @@ class LinFixedEF(CardEF):
def do_edit_record_decoded(self, opts):
"""Edit the JSON representation of one record in an editor."""
(orig_json, _sw) = self._cmd.lchan.read_record_dec(opts.RECORD_NR)
ef = self._cmd.lchan.selected_file
with JsonEditor(self._cmd, orig_json, ef) as edited_json:
with tempfile.TemporaryDirectory(prefix='pysim_') as dirname:
filename = '%s/file' % dirname
# write existing data as JSON to file
with open(filename, 'w') as text_file:
json.dump(orig_json, text_file, indent=4, cls=JsonEncoder)
# run a text editor
self._cmd.run_editor(filename)
with open(filename, 'r') as text_file:
edited_json = json.load(text_file)
if edited_json == orig_json:
self._cmd.poutput("Data not modified, skipping write")
else:

View File

@@ -276,7 +276,7 @@ class ListOfSupportedOptions(BER_TLV_IE, tag=0x81):
class SupportedKeysForScp03(BER_TLV_IE, tag=0x82):
_construct = FlagsEnum(Byte, aes128=0x01, aes192=0x02, aes256=0x04)
class SupportedTlsCipherSuitesForScp81(BER_TLV_IE, tag=0x83):
_construct = GreedyRange(Int16ub)
_consuruct = GreedyRange(Int16ub)
class ScpInformation(BER_TLV_IE, tag=0xa0, nested=[ScpType, ListOfSupportedOptions, SupportedKeysForScp03,
SupportedTlsCipherSuitesForScp81]):
pass
@@ -319,7 +319,7 @@ class CurrentSecurityLevel(BER_TLV_IE, tag=0xd3):
# GlobalPlatform v2.3.1 Section 11.3.3.1.3
class ApplicationAID(BER_TLV_IE, tag=0x4f):
_construct = GreedyBytes
class ApplicationTemplate(BER_TLV_IE, tag=0x61, nested=[ApplicationAID]):
class ApplicationTemplate(BER_TLV_IE, tag=0x61, ntested=[ApplicationAID]):
pass
class ListOfApplications(BER_TLV_IE, tag=0x2f00, nested=[ApplicationTemplate]):
pass
@@ -562,14 +562,14 @@ class ADF_SD(CardADF):
@cmd2.with_argparser(store_data_parser)
def do_store_data(self, opts):
"""Perform the GlobalPlatform STORE DATA command in order to store some card-specific data.
See GlobalPlatform CardSpecification v2.3 Section 11.11 for details."""
"""Perform the GlobalPlatform GET DATA command in order to store some card-specific data.
See GlobalPlatform CardSpecification v2.3Section 11.11 for details."""
response_permitted = opts.response == 'may_be_returned'
self.store_data(h2b(opts.DATA), opts.data_structure, opts.encryption, response_permitted)
def store_data(self, data: bytes, structure:str = 'none', encryption:str = 'none', response_permitted: bool = False) -> bytes:
"""Perform the GlobalPlatform STORE DATA command in order to store some card-specific data.
See GlobalPlatform CardSpecification v2.3 Section 11.11 for details."""
"""Perform the GlobalPlatform GET DATA command in order to store some card-specific data.
See GlobalPlatform CardSpecification v2.3Section 11.11 for details."""
max_cmd_len = self._cmd.lchan.scc.max_cmd_len
# Table 11-89 of GP Card Specification v2.3
remainder = data
@@ -585,7 +585,7 @@ class ADF_SD(CardADF):
data, _sw = self._cmd.lchan.scc.send_apdu_checksw(hdr + b2h(chunk) + "00")
block_nr += 1
response += data
return h2b(response)
return data
put_key_parser = argparse.ArgumentParser()
put_key_parser.add_argument('--old-key-version-nr', type=auto_uint8, default=0, help='Old Key Version Number')
@@ -859,28 +859,22 @@ class ADF_SD(CardADF):
_rsp_hex, _sw = self._cmd.lchan.scc.send_apdu_checksw(cmd_hex)
self._cmd.poutput("Loaded a total of %u bytes in %u blocks. Don't forget install_for_install (and make selectable) now!" % (total_size, block_nr))
install_cap_parser = argparse.ArgumentParser(usage='%(prog)s FILE [--install-parameters | --install-parameters-*]')
install_cap_parser = argparse.ArgumentParser()
install_cap_parser.add_argument('cap_file', type=str, metavar='FILE',
help='JAVA-CARD CAP file to install')
# Ideally, the parser should enforce that:
# * either the `--install-parameters` is given alone,
# * or distinct `--install-parameters-*` are optionally given instead.
# We tried to achieve this using mutually exclusive groups (add_mutually_exclusive_group).
# However, group nesting was never supported, often failed to work correctly, and was unintentionally
# exposed through inheritance. It has been deprecated since version 3.11, removed in version 3.14.
# Hence, we have to implement the enforcement manually.
install_cap_parser_inst_prm_grp = install_cap_parser.add_argument_group('Install Parameters')
install_cap_parser_inst_prm_grp.add_argument('--install-parameters', type=is_hexstr, default=None,
help='install Parameters (GPC_SPE_034, section 11.5.2.3.7, table 11-49)')
install_cap_parser_inst_prm_grp.add_argument('--install-parameters-volatile-memory-quota',
type=int, default=None,
help='volatile memory quota (GPC_SPE_034, section 11.5.2.3.7, table 11-49)')
install_cap_parser_inst_prm_grp.add_argument('--install-parameters-non-volatile-memory-quota',
type=int, default=None,
help='non volatile memory quota (GPC_SPE_034, section 11.5.2.3.7, table 11-49)')
install_cap_parser_inst_prm_grp.add_argument('--install-parameters-stk',
type=is_hexstr, default=None,
help='Load Parameters (ETSI TS 102 226, section 8.2.1.3.2.1)')
install_cap_parser_inst_prm_g = install_cap_parser.add_mutually_exclusive_group()
install_cap_parser_inst_prm_g.add_argument('--install-parameters', type=is_hexstr, default=None,
help='install Parameters (GPC_SPE_034, section 11.5.2.3.7, table 11-49)')
install_cap_parser_inst_prm_g_grp = install_cap_parser_inst_prm_g.add_argument_group()
install_cap_parser_inst_prm_g_grp.add_argument('--install-parameters-volatile-memory-quota',
type=int, default=None,
help='volatile memory quota (GPC_SPE_034, section 11.5.2.3.7, table 11-49)')
install_cap_parser_inst_prm_g_grp.add_argument('--install-parameters-non-volatile-memory-quota',
type=int, default=None,
help='non volatile memory quota (GPC_SPE_034, section 11.5.2.3.7, table 11-49)')
install_cap_parser_inst_prm_g_grp.add_argument('--install-parameters-stk',
type=is_hexstr, default=None,
help='Load Parameters (ETSI TS 102 226, section 8.2.1.3.2.1)')
@cmd2.with_argparser(install_cap_parser)
def do_install_cap(self, opts):
@@ -894,17 +888,9 @@ class ADF_SD(CardADF):
load_file_aid = cap.get_loadfile_aid()
module_aid = cap.get_applet_aid()
application_aid = module_aid
if opts.install_parameters is not None:
# `--install-parameters` and `--install-parameters-*` are mutually exclusive
# make sure that none of `--install-parameters-*` is given; abort otherwise
if any(p is not None for p in [opts.install_parameters_non_volatile_memory_quota,
opts.install_parameters_volatile_memory_quota,
opts.install_parameters_stk]):
self.install_cap_parser.error('arguments --install-parameters-* are '
'not allowed with --install-parameters')
if opts.install_parameters:
install_parameters = opts.install_parameters;
else:
# `--install-parameters-*` are all optional
install_parameters = gen_install_parameters(opts.install_parameters_non_volatile_memory_quota,
opts.install_parameters_volatile_memory_quota,
opts.install_parameters_stk)

View File

@@ -17,8 +17,6 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
from typing import Optional
from osmocom.construct import *
from osmocom.utils import *
from osmocom.tlv import *
@@ -48,9 +46,7 @@ class InstallParams(TLV_IE_Collection, nested=[AppSpecificParams, SystemSpecific
# GPD_SPE_013, table 11-49
pass
def gen_install_parameters(non_volatile_memory_quota: Optional[int] = None,
volatile_memory_quota: Optional[int] = None,
stk_parameter: Optional[str] = None):
def gen_install_parameters(non_volatile_memory_quota:int, volatile_memory_quota:int, stk_parameter:str):
# GPD_SPE_013, table 11-49
@@ -58,17 +54,19 @@ def gen_install_parameters(non_volatile_memory_quota: Optional[int] = None,
install_params = InstallParams()
install_params_dict = [{'app_specific_params': None}]
# Collect system specific parameters (optional)
system_specific_params = []
if non_volatile_memory_quota is not None:
system_specific_params.append({'non_volatile_memory_quota': non_volatile_memory_quota})
if volatile_memory_quota is not None:
system_specific_params.append({'volatile_memory_quota': volatile_memory_quota})
if stk_parameter is not None:
system_specific_params.append({'stk_parameter': stk_parameter})
# Add system specific parameters to the install parameters, if any
if system_specific_params:
install_params_dict.append({'system_specific_params': system_specific_params})
#Conditional
if non_volatile_memory_quota and volatile_memory_quota and stk_parameter:
system_specific_params = []
#Optional
if non_volatile_memory_quota:
system_specific_params += [{'non_volatile_memory_quota': non_volatile_memory_quota}]
#Optional
if volatile_memory_quota:
system_specific_params += [{'volatile_memory_quota': volatile_memory_quota}]
#Optional
if stk_parameter:
system_specific_params += [{'stk_parameter': stk_parameter}]
install_params_dict += [{'system_specific_params': system_specific_params}]
install_params.from_dict(install_params_dict)
return b2h(install_params.to_bytes())

View File

@@ -438,7 +438,7 @@ class Scp03SessionKeys:
"""Obtain the ICV value computed as described in 6.2.6.
This method has two modes:
* is_response=False for computing the ICV for C-ENC. Will pre-increment the counter.
* is_response=True for computing the ICV for R-DEC."""
* is_response=False for computing the ICV for R-DEC."""
if not is_response:
self.block_nr += 1
# The binary value of this number SHALL be left padded with zeroes to form a full block.

View File

@@ -91,7 +91,6 @@ class UiccSdInstallParams(TLV_IE_Collection, nested=[UiccScp, AcceptExtradAppsAn
# Key Usage:
# KVN 0x01 .. 0x0F reserved for SCP80
# KVN 0x81 .. 0x8f reserved for SCP81
# KVN 0x11 reserved for DAP specified in ETSI TS 102 226
# KVN 0x20 .. 0x2F reserved for SCP02
# KID 0x01 = ENC; 0x02 = MAC; 0x03 = DEK

View File

@@ -63,7 +63,7 @@ class PySimLogger:
raise RuntimeError('static class, do not instantiate')
@staticmethod
def setup(print_callback = None, colors:dict = {}, verbose_debug:bool = False):
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.
@@ -72,20 +72,10 @@ class PySimLogger:
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})
verbose_debug: Enable verbose logging and set the loglevel DEBUG when set to true. Otherwise the
non-verbose logging is used and the loglevel is set to INFO. This setting can be changed
using the set_verbose and set_level methods at any time.
"""
PySimLogger.print_callback = print_callback
PySimLogger.colors = colors
if (verbose_debug):
PySimLogger.set_verbose(True)
PySimLogger.set_level(logging.DEBUG)
else:
PySimLogger.set_verbose(False)
PySimLogger.set_level(logging.INFO)
@staticmethod
def set_verbose(verbose:bool = False):
"""

View File

@@ -221,12 +221,12 @@ class OtaAlgoCrypt(OtaAlgo, abc.ABC):
for subc in cls.__subclasses__():
if subc.enum_name == otak.algo_crypt:
return subc(otak)
raise ValueError('No implementation for crypt algorithm %s' % otak.algo_crypt)
raise ValueError('No implementation for crypt algorithm %s' % otak.algo_auth)
class OtaAlgoAuth(OtaAlgo, abc.ABC):
def __init__(self, otak: OtaKeyset):
if self.enum_name != otak.algo_auth:
raise ValueError('Cannot use algorithm %s with key for %s' % (self.enum_name, otak.algo_auth))
raise ValueError('Cannot use algorithm %s with key for %s' % (self.enum_name, otak.algo_crypt))
super().__init__(otak)
def sign(self, data:bytes) -> bytes:

View File

@@ -169,14 +169,8 @@ class SMS_TPDU(abc.ABC):
class SMS_DELIVER(SMS_TPDU):
"""Representation of a SMS-DELIVER T-PDU. This is the Network to MS/UE (downlink) direction."""
flags_construct = BitStruct('tp_rp'/Flag,
'tp_udhi'/Flag,
'tp_sri'/Flag,
Padding(1),
'tp_lp'/Flag,
'tp_mms'/Flag,
'tp_mti'/BitsInteger(2))
flags_construct = BitStruct('tp_rp'/Flag, 'tp_udhi'/Flag, 'tp_rp'/Flag, 'tp_sri'/Flag,
Padding(1), 'tp_mms'/Flag, 'tp_mti'/BitsInteger(2))
def __init__(self, **kwargs):
kwargs['tp_mti'] = 0
super().__init__(**kwargs)

View File

@@ -90,7 +90,7 @@ class LinkBase(abc.ABC):
self.sw_interpreter = sw_interpreter
self.apdu_tracer = apdu_tracer
self.proactive_handler = proactive_handler
self.apdu_strict = True
self.apdu_strict = False
@abc.abstractmethod
def __str__(self) -> str:

View File

@@ -26,7 +26,6 @@ from smartcard.CardRequest import CardRequest
from smartcard.Exceptions import NoCardException, CardRequestTimeoutException, CardConnectionException
from smartcard.System import readers
from smartcard.ExclusiveConnectCardConnection import ExclusiveConnectCardConnection
from smartcard.ATR import ATR
from osmocom.utils import h2i, i2h, Hexstr
@@ -81,25 +80,23 @@ class PcscSimLink(LinkBaseTpdu):
def connect(self):
try:
# To avoid leakage of resources, make sure the reader is disconnected
# To avoid leakage of resources, make sure the reader
# is disconnected
self.disconnect()
# Make card connection and select a suitable communication protocol
# (Even though pyscard provides an automatic protocol selection, we will make an independent decision
# based on the ATR. There are two reasons for that:
# 1) In case a card supports T=0 and T=1, we perfer to use T=0.
# 2) The automatic protocol selection may be unreliabe on some platforms
# see also: https://osmocom.org/issues/6952)
self._con.connect()
atr = ATR(self._con.getATR())
if atr.isT0Supported():
self._con.setProtocol(CardConnection.T0_protocol)
supported_protocols = self._con.getProtocol();
self.disconnect()
if (supported_protocols & CardConnection.T0_protocol):
protocol = CardConnection.T0_protocol
self.set_tpdu_format(0)
elif atr.isT1Supported():
self._con.setProtocol(CardConnection.T1_protocol)
elif (supported_protocols & CardConnection.T1_protocol):
protocol = CardConnection.T1_protocol
self.set_tpdu_format(1)
else:
raise ReaderError('Unsupported card protocol')
self._con.connect(protocol)
except CardConnectionException as exc:
raise ProtocolError() from exc
except NoCardException as exc:

View File

@@ -1058,7 +1058,7 @@ class EF_OCSGL(LinFixedEF):
# TS 31.102 Section 4.4.11.2 (Rel 15)
class EF_5GS3GPPLOCI(TransparentEF):
def __init__(self, fid='4f01', sfid=0x01, name='EF.5GS3GPPLOCI', size=(20, 20),
desc='5GS 3GPP location information', **kwargs):
desc='5S 3GP location information', **kwargs):
super().__init__(fid, sfid=sfid, name=name, desc=desc, size=size, **kwargs)
upd_status_constr = Enum(
Byte, updated=0, not_updated=1, roaming_not_allowed=2)
@@ -1326,7 +1326,7 @@ class EF_5G_PROSE_UIR(TransparentEF):
pass
class FiveGDdnmfCtfAddrForUploading(BER_TLV_IE, tag=0x97):
pass
class ProSeConfigDataForUsageInfoReporting(BER_TLV_IE, tag=0xa0,
class ProSeConfigDataForUeToNetworkRelayUE(BER_TLV_IE, tag=0xa0,
nested=[EF_5G_PROSE_DD.ValidityTimer,
CollectionPeriod, ReportingWindow,
ReportingIndicators,
@@ -1336,7 +1336,7 @@ class EF_5G_PROSE_UIR(TransparentEF):
desc='5G ProSe configuration data for usage information reporting', **kwargs):
super().__init__(fid, sfid=sfid, name=name, desc=desc, **kwargs)
# contains TLV structure despite being TransparentEF, not BER-TLV ?!?
self._tlv = EF_5G_PROSE_UIR.ProSeConfigDataForUsageInfoReporting
self._tlv = EF_5G_PROSE_UIR.ProSeConfigDataForUeToNetworkRelayUE
# TS 31.102 Section 4.4.13.8 (Rel 18)
class EF_5G_PROSE_U2URU(TransparentEF):

View File

@@ -261,26 +261,6 @@ class EF_SMSP(LinFixedEF):
"numbering_plan_id": "reserved_for_extension" },
"call_number": "" },
"tp_pid": b"\x00", "tp_dcs": b"\x00", "tp_vp_minutes": 1440 } ),
( 'fffffffffffffffffffffffffffffffffffffffffffffffffdffffffffffffffffffffffff07919403214365f7ffffffffffffff',
{ "alpha_id": "", "parameter_indicators": { "tp_dest_addr": False, "tp_sc_addr": True,
"tp_pid": False, "tp_dcs": False, "tp_vp": False },
"tp_dest_addr": { "length": 255, "ton_npi": { "ext": True, "type_of_number": "reserved_for_extension",
"numbering_plan_id": "reserved_for_extension" },
"call_number": "" },
"tp_sc_addr": { "length": 7, "ton_npi": { "ext": True, "type_of_number": "international",
"numbering_plan_id": "isdn_e164" },
"call_number": "49301234567" },
"tp_pid": b"\xff", "tp_dcs": b"\xff", "tp_vp_minutes": 635040 } ),
( 'fffffffffffffffffffffffffffffffffffffffffffffffffc0b919403214365f7ffffffff07919403214365f7ffffffffffffff',
{ "alpha_id": "", "parameter_indicators": { "tp_dest_addr": True, "tp_sc_addr": True,
"tp_pid": False, "tp_dcs": False, "tp_vp": False },
"tp_dest_addr": { "length": 11, "ton_npi": { "ext": True, "type_of_number": "international",
"numbering_plan_id": "isdn_e164" },
"call_number": "49301234567" },
"tp_sc_addr": { "length": 7, "ton_npi": { "ext": True, "type_of_number": "international",
"numbering_plan_id": "isdn_e164" },
"call_number": "49301234567" },
"tp_pid": b"\xff", "tp_dcs": b"\xff", "tp_vp_minutes": 635040 } ),
]
_test_no_pad = True
class ValidityPeriodAdapter(Adapter):
@@ -309,28 +289,16 @@ class EF_SMSP(LinFixedEF):
@staticmethod
def sc_addr_len(ctx):
"""Compute the length field for an address field (see also: 3GPP TS 24.011, section 8.2.5.2)."""
"""Compute the length field for an address field (like TP-DestAddr or TP-ScAddr)."""
if not hasattr(ctx, 'call_number') or len(ctx.call_number) == 0:
return 0xff
else:
# octets required for the call_number + one octet for ton_npi
return bytes_for_nibbles(len(ctx.call_number)) + 1
@staticmethod
def dest_addr_len(ctx):
"""Compute the length field for an address field (see also: 3GPP TS 23.040, section 9.1.2.5)."""
if not hasattr(ctx, 'call_number') or len(ctx.call_number) == 0:
return 0xff
else:
# number of call_number digits
return len(ctx.call_number)
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'/Rebuild(Int8ub, lambda ctx: EF_SMSP.sc_addr_len(ctx)),
'ton_npi'/TonNpi, 'call_number'/PaddedBcdAdapter(Rpad(Bytes(10))))
DestAddr = Struct('length'/Rebuild(Int8ub, lambda ctx: EF_SMSP.dest_addr_len(ctx)),
'ton_npi'/TonNpi, 'call_number'/PaddedBcdAdapter(Rpad(Bytes(10))))
self._construct = Struct('alpha_id'/COptional(GsmOrUcs2Adapter(Rpad(Bytes(this._.total_len-28)))),
'parameter_indicators'/InvertAdapter(BitStruct(
Const(7, BitsInteger(3)),
@@ -339,8 +307,9 @@ class EF_SMSP(LinFixedEF):
'tp_pid'/Flag,
'tp_sc_addr'/Flag,
'tp_dest_addr'/Flag)),
'tp_dest_addr'/DestAddr,
'tp_dest_addr'/ScAddr,
'tp_sc_addr'/ScAddr,
'tp_pid'/Bytes(1),
'tp_dcs'/Bytes(1),
'tp_vp_minutes'/EF_SMSP.ValidityPeriodAdapter(Byte))
@@ -420,7 +389,7 @@ class DF_TELECOM(CardDF):
# TS 51.011 Section 10.3.1
class EF_LP(TransRecEF):
_test_de_encode = [
( "24", ["24"] ),
( "24", "24"),
]
def __init__(self, fid='6f05', sfid=None, name='EF.LP', size=(1, None), rec_len=1,
desc='Language Preference'):
@@ -477,8 +446,8 @@ class EF_IMSI(TransparentEF):
# TS 51.011 Section 10.3.4
class EF_PLMNsel(TransRecEF):
_test_de_encode = [
( "22F860", [{ "mcc": "228", "mnc": "06" }] ),
( "330420", [{ "mcc": "334", "mnc": "020" }] ),
( "22F860", { "mcc": "228", "mnc": "06" } ),
( "330420", { "mcc": "334", "mnc": "020" } ),
]
def __init__(self, fid='6f30', sfid=None, name='EF.PLMNsel', desc='PLMN selector',
size=(24, None), rec_len=3, **kwargs):
@@ -692,7 +661,7 @@ class EF_AD(TransparentEF):
# TS 51.011 Section 10.3.20 / 10.3.22
class EF_VGCS(TransRecEF):
_test_de_encode = [
( "92f9ffff", ["299"] ),
( "92f9ffff", "299" ),
]
def __init__(self, fid='6fb1', sfid=None, name='EF.VGCS', size=(4, 200), rec_len=4,
desc='Voice Group Call Service', **kwargs):
@@ -828,9 +797,9 @@ class EF_LOCIGPRS(TransparentEF):
# TS 51.011 Section 10.3.35..37
class EF_xPLMNwAcT(TransRecEF):
_test_de_encode = [
( '62F2104000', [{ "mcc": "262", "mnc": "01", "act": [ "E-UTRAN NB-S1", "E-UTRAN WB-S1" ] }] ),
( '62F2108000', [{ "mcc": "262", "mnc": "01", "act": [ "UTRAN" ] }] ),
( '62F220488C', [{ "mcc": "262", "mnc": "02", "act": ['E-UTRAN NB-S1', 'E-UTRAN WB-S1', 'EC-GSM-IoT', 'GSM', 'NG-RAN'] }] ),
( '62F2104000', { "mcc": "262", "mnc": "01", "act": [ "E-UTRAN NB-S1", "E-UTRAN WB-S1" ] } ),
( '62F2108000', { "mcc": "262", "mnc": "01", "act": [ "UTRAN" ] } ),
( '62F220488C', { "mcc": "262", "mnc": "02", "act": ['E-UTRAN NB-S1', 'E-UTRAN WB-S1', 'EC-GSM-IoT', 'GSM', 'NG-RAN'] } ),
]
def __init__(self, fid='1234', sfid=None, name=None, desc=None, size=(40, None), rec_len=5, **kwargs):
super().__init__(fid, sfid=sfid, name=name, desc=desc, size=size, rec_len=rec_len, **kwargs)
@@ -1065,10 +1034,9 @@ class EF_ICCID(TransparentEF):
# TS 102 221 Section 13.3 / TS 31.101 Section 13 / TS 51.011 Section 10.1.2
class EF_PL(TransRecEF):
_test_de_encode = [
( '6465', ["de"] ),
( '656e', ["en"] ),
( 'ffff', [None] ),
( '656e64657275ffffffff', ["en", "de", "ru", None, None] ),
( '6465', "de" ),
( '656e', "en" ),
( 'ffff', None ),
]
def __init__(self, fid='2f05', sfid=0x05, name='EF.PL', desc='Preferred Languages'):
@@ -1149,8 +1117,8 @@ class DF_GSM(CardDF):
EF_MBI(),
EF_MWIS(),
EF_CFIS(),
EF_EXT('6fc8', None, 'EF.EXT6', desc='Extension6 (MBDN)'),
EF_EXT('6fcc', None, 'EF.EXT7', desc='Extension7 (CFIS)'),
EF_EXT('6fc8', None, 'EF.EXT6', desc='Externsion6 (MBDN)'),
EF_EXT('6fcc', None, 'EF.EXT7', desc='Externsion7 (CFIS)'),
EF_SPDI(),
EF_MMSN(),
EF_EXT('6fcf', None, 'EF.EXT8', desc='Extension8 (MMSN)'),

View File

@@ -139,6 +139,7 @@ def enc_plmn(mcc: Hexstr, mnc: Hexstr) -> Hexstr:
def dec_plmn(threehexbytes: Hexstr) -> dict:
res = {'mcc': "0", 'mnc': "0"}
dec_mcc_from_plmn_str(threehexbytes)
res['mcc'] = dec_mcc_from_plmn_str(threehexbytes)
res['mnc'] = dec_mnc_from_plmn_str(threehexbytes)
return res
@@ -910,8 +911,7 @@ class DataObjectCollection:
def encode(self, decoded) -> bytes:
res = bytearray()
for i in decoded:
name = i[0]
obj = self.members_by_name[name]
obj = self.members_by_name(i[0])
res.append(obj.to_tlv())
return res

View File

@@ -1,4 +1,4 @@
INFO: Using PC/SC reader interface
Using PC/SC reader interface
Reading ...
Autodetected card type: Fairwaves-SIM
ICCID: 8988219000000117833

View File

@@ -1,4 +1,4 @@
INFO: Using PC/SC reader interface
Using PC/SC reader interface
Reading ...
Autodetected card type: Wavemobile-SIM
ICCID: 89445310150011013678

View File

@@ -1,4 +1,4 @@
INFO: Using PC/SC reader interface
Using PC/SC reader interface
Reading ...
Autodetected card type: fakemagicsim
ICCID: 1122334455667788990

View File

@@ -1,4 +1,4 @@
INFO: Using PC/SC reader interface
Using PC/SC reader interface
Reading ...
Autodetected card type: sysmoISIM-SJA2
ICCID: 8988211000000467343

View File

@@ -1,4 +1,4 @@
INFO: Using PC/SC reader interface
Using PC/SC reader interface
Reading ...
Autodetected card type: sysmoISIM-SJA5
ICCID: 8949440000001155314

View File

@@ -1,4 +1,4 @@
INFO: Using PC/SC reader interface
Using PC/SC reader interface
Reading ...
Autodetected card type: sysmoUSIM-SJS1
ICCID: 8988211320300000028

View File

@@ -1,4 +1,4 @@
INFO: Using PC/SC reader interface
Using PC/SC reader interface
Reading ...
Autodetected card type: sysmosim-gr1
ICCID: 2222334455667788990

View File

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

View File

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

View File

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

View File

@@ -176,11 +176,12 @@ class TransRecEF_Test(unittest.TestCase):
def test_de_encode_record(self):
"""Test the decoder and encoder for a transparent record-oriented EF at the whole-file
level. Performs first a decode test, then re-encodes and compares with the input.
"""Test the decoder and encoder for a transparent record-oriented EF. Performs first a decoder
test, and then re-encodes the decoded data, comparing the re-encoded data with the
initial input data.
Requires the given TransRecEF subclass to have a '_test_de_encode' attribute,
containing a list of 2-tuples (hexstring, decoded_list).
containing a list of tuples. Each tuple has to be a 2-tuple (hexstring, decoded_dict).
"""
for c in self.classes:
name = get_qualified_name(c)
@@ -191,12 +192,14 @@ class TransRecEF_Test(unittest.TestCase):
encoded = t[0]
decoded = t[1]
logging.debug("Testing decode of %s", name)
re_dec = inst.decode_hex(encoded)
re_dec = inst.decode_record_hex(encoded)
self.assertEqual(decoded, re_dec)
# re-encode the decoded data
logging.debug("Testing re-encode of %s", name)
re_enc = inst.encode_hex(re_dec, len(encoded)//2)
re_enc = inst.encode_record_hex(re_dec, len(encoded)//2)
self.assertEqual(encoded.upper(), re_enc.upper())
# there's no point in testing padded input, as TransRecEF have a fixed record
# size and we cannot ever receive more input data than that size.
class TransparentEF_Test(unittest.TestCase):

View File

@@ -1,144 +0,0 @@
#!/usr/bin/env python3
# (C) 2026 by sysmocom - s.f.m.c. GmbH <info@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/>.
"""Verify that every CardProfile / CardApplication subclass with EF/DF content,
and every standalone CardDF subclass (one not reachable as a child of any profile
or application), is either listed in docs/pysim_fs_sphinx.py::SECTIONS or
explicitly EXCLUDED."""
import unittest
import importlib
import inspect
import pkgutil
import sys
import os
# Make docs/pysim_fs_sphinx.py importable without a full Sphinx build.
_DOCS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', '..', 'docs')
sys.path.insert(0, os.path.abspath(_DOCS_DIR))
import pySim # noqa: E402
from pySim.filesystem import CardApplication, CardDF, CardMF, CardADF # noqa: E402
from pySim.profile import CardProfile # noqa: E402
from pysim_fs_sphinx import EXCLUDED, SECTIONS # noqa: E402
class TestFsCoverage(unittest.TestCase):
"""Ensure SECTIONS + EXCLUDED together account for all classes with content."""
# Base CardDF types that are not concrete filesystem objects on their own.
_DF_BASE_TYPES = frozenset([CardDF, CardMF, CardADF])
@staticmethod
def _collect_reachable_df_types(obj) -> set:
"""Return the set of all CardDF *types* reachable as children of *obj*."""
result = set()
if isinstance(obj, CardProfile):
children = obj.files_in_mf
elif isinstance(obj, CardApplication):
result.add(type(obj.adf))
children = list(obj.adf.children.values())
elif isinstance(obj, CardDF):
children = list(obj.children.values())
else:
return result
queue = list(children)
while queue:
child = queue.pop()
if isinstance(child, CardDF):
result.add(type(child))
queue.extend(child.children.values())
return result
@staticmethod
def _has_content(obj) -> bool:
"""Return True if *obj* owns any EFs/DFs."""
if isinstance(obj, CardProfile):
return bool(obj.files_in_mf)
if isinstance(obj, CardApplication):
return bool(obj.adf.children)
return False
def test_all_profiles_and_apps_covered(self):
# build a set of (module, class-name) pairs that are already accounted for
covered = {(mod, cls) for (_, mod, cls) in SECTIONS}
accounted_for = covered | EXCLUDED
uncovered = []
reachable_df_types = set()
loaded_modules = {}
for modinfo in pkgutil.walk_packages(pySim.__path__, prefix='pySim.'):
modname = modinfo.name
try:
module = importlib.import_module(modname)
except Exception: # skip inport errors, if any
continue
loaded_modules[modname] = module
for name, cls in inspect.getmembers(module, inspect.isclass):
# skip classes that are merely imported by this module
if cls.__module__ != modname:
continue
# examine only subclasses of CardProfile and CardApplication
if not issubclass(cls, (CardProfile, CardApplication)):
continue
# skip the abstract base classes themselves
if cls in (CardProfile, CardApplication):
continue
# classes that require constructor arguments cannot be probed
try:
obj = cls()
except Exception:
continue
# collect all CardDF types reachable from this profile/application
# (used below to identify standalone DFs)
reachable_df_types |= self._collect_reachable_df_types(obj)
if self._has_content(obj) and (modname, name) not in accounted_for:
uncovered.append((modname, name))
# check standalone CardDFs (such as DF.EIRENE or DF.SYSTEM)
for modname, module in loaded_modules.items():
for name, cls in inspect.getmembers(module, inspect.isclass):
if cls.__module__ != modname:
continue
if not issubclass(cls, CardDF):
continue
if cls in self._DF_BASE_TYPES:
continue
if cls in reachable_df_types:
continue
try:
obj = cls()
except Exception:
continue
if obj.children and (modname, name) not in accounted_for:
uncovered.append((modname, name))
if uncovered:
lines = [
'The following classes have EFs/DFs, but not listed in SECTIONS or EXCLUDED:',
*(f' {modname}.{name}' for modname, name in sorted(uncovered)),
'Please modify docs/pysim_fs_sphinx.py accordingly',
]
self.fail('\n'.join(lines))
if __name__ == '__main__':
unittest.main()

View File

@@ -295,7 +295,7 @@ class Install_param_Test(unittest.TestCase):
load_parameters = gen_install_parameters(256, 256, '010001001505000000000000000000000000')
self.assertEqual(load_parameters, 'c900ef1cc8020100c7020100ca12010001001505000000000000000000000000')
load_parameters = gen_install_parameters()
load_parameters = gen_install_parameters(None, None, '')
self.assertEqual(load_parameters, 'c900')
if __name__ == "__main__":

View File

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

File diff suppressed because it is too large Load Diff

View File

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