public/pysim
1
0
Fork 1
mirror of https://gitea.osmocom.org/sim-card/pysim.git synced 2026-05-02 22:48:53 +03:00
Code Issues Packages Projects Releases Wiki Activity

Compare commits

base: public:pmaier/rcp
Branches Tags
public:master public:pmaier/rcp public:neels/suci public:neels/saip2 public:neels/wip public:neels/smsp_test public:pmaier/fixgr public:osmith/saip2 public:pmaier/putkey public:pmaier/otatest public:pmaier/otatestprep public:daniel/ota public:neels/saip public:pmaier/pgsql public:lynxis/esim-cccprofile public:osmith/wip public:ewild/ossl_verify public:ewild/ppk_tests public:laforge/ota public:laforge/wsrc public:laforge/wip public:laforge/scp public:laforge/smdp_asn1 public:fixeria/cmd2 public:chrysn/ota public:chrysn/for-29033 public:fixeria/run_gsm_alg public:fixeria/btsap public:sysmocom/factory public:users/daniel/csv-format public:fairwaves/sim public:zecke/tmp2 public:ccc public:zecke/hacks public:27c3
public:1.0
..
compare: public:pmaier/putkey
Branches Tags
public:pmaier/rcp public:master public:neels/suci public:neels/saip2 public:neels/wip public:neels/smsp_test public:pmaier/fixgr public:osmith/saip2 public:pmaier/putkey public:pmaier/otatest public:pmaier/otatestprep public:daniel/ota public:neels/saip public:pmaier/pgsql public:lynxis/esim-cccprofile public:osmith/wip public:ewild/ossl_verify public:ewild/ppk_tests public:laforge/ota public:laforge/wsrc public:laforge/wip public:laforge/scp public:laforge/smdp_asn1 public:fixeria/cmd2 public:chrysn/ota public:chrysn/for-29033 public:fixeria/run_gsm_alg public:fixeria/btsap public:sysmocom/factory public:users/daniel/csv-format public:fairwaves/sim public:zecke/tmp2 public:ccc public:zecke/hacks public:27c3
public:1.0

1 Commits
pmaier/rcp ... pmaier/put

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
72 changed files with 341 additions and 3683 deletions
Expand all files Collapse all files

1
.gitignore vendored
View File

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

2
README.md
View File

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

5
contrib/csv-to-pgsql.py
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)

34
contrib/jenkins.sh
View File

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

204
contrib/rcp/rcp_client.py
View File

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

387
contrib/rcp/rcp_module_utils.py
View File

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

361
contrib/rcp/rcp_server.py
View File

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

254
contrib/rcp/rcp_utils.py
View File

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

69
contrib/rcp/rcpc_rx_schema.json
View File

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

89
contrib/rcp/rcpc_tx_schema.json
View File

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

20
contrib/rcp/usage_example/certs/example_ssl_rcp_ca_cert.crt
View File

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

115
contrib/rcp/usage_example/certs/example_ssl_rcpc_rcps_cert.pem
View File

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

115
contrib/rcp/usage_example/certs/example_ssl_rcpm_rcps_cert.pem
View File

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

115
contrib/rcp/usage_example/certs/example_ssl_rcps_rcpm_cert.pem
View File

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

49
contrib/rcp/usage_example/certs/make_certs.sh
View File

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

36
contrib/rcp/usage_example/params.cfg
View File

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

86
contrib/rcp/usage_example/rcp_module.py
View File

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

14
contrib/rcp/usage_example/readme.txt
View File

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

29
contrib/rcp/usage_example/run_rcp_client.sh
View File

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

22
contrib/rcp/usage_example/run_rcp_client_cmd.sh
View File

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

6
contrib/rcp/usage_example/run_rcp_client_help.sh
View File

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

9
contrib/rcp/usage_example/run_rcp_client_help_cmd.sh
View File

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

22
contrib/rcp/usage_example/run_rcp_client_help_cmd_specific.sh
View File

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

11
contrib/rcp/usage_example/start_rcp_module.sh
View File

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

13
contrib/rcp/usage_example/start_rcp_server.sh
View File

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

6
docs/card-key-provider.rst
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.

26
docs/conf.py
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',
]

1
docs/index.rst
View File

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

2
docs/legacy.rst
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

138
docs/put_key-tutorial.rst
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.
@@ -25,7 +25,7 @@ When working with those cards, the ISD will show up in the UICC filesystem tree
any other file.
::
pySIM-shell (00:MF)> select ADF.ISD
{
"application_id": "a000000003000000",
@@ -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,40 +127,38 @@ 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`:
::
pySIM-shell (SCP02[03]:00:MF/ADF.ISD)> get_data key_information
pySIM-shell (SCP02[03]:00:MF/ADF.ISD)> get_data key_information
{
"key_information": [
{
@@ -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:
+----------------+---------+---------------------------------------+
@@ -458,7 +464,7 @@ that works with two keys:
It should also be noted that the order in which keysets and keys appear is an implementation detail of the UICC/eUICC
O/S. The order has no influence on how a keyset is interpreted. Only the Key Version Number (KVN) and the Key Identifier
matter.
Rotating a keyset
~~~~~~~~~~~~~~~~~
@@ -493,14 +499,14 @@ 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.
The commandline that matches the keyset we had laid out above looks like this:
::
pySIM-shell (SCP02[03]:00:MF/ADF.ISD)> put_key --key-id 1 --key-type des --key-data 542C37A6043679F2F9F71116418B1CD5 --key-type des --key-data 34F11BAC8E5390B57F4E601372339E3C --key-type des --key-data 5524F4BECFE96FB63FC29D6BAAC6058B --old-key-version-nr 112 --key-version-nr 112
After executing this put_key commandline, the keyset identified by KVN 122 is equipped with new keys. We can use
@@ -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
@@ -583,7 +589,7 @@ the one from the key rotation example where we were rotating a 3DES key. The onl
an old KVN number and that we have chosen a different KVN.
::
pySIM-shell (SCP02[03]:00:MF/ADF.ISD)> put_key --key-id 1 --key-type des --key-data 542C37A6043679F2F9F71116418B1CD5 --key-type des --key-data 34F11BAC8E5390B57F4E601372339E3C --key-type des --key-data 5524F4BECFE96FB63FC29D6BAAC6058B --key-version-nr 46
In case of success, the keyset should appear in the `key_information` among the other keysets that are already present.
@@ -729,11 +735,11 @@ still unused.
With that we can go ahead and make up the following commandline:
::
pySIM-shell (SCP02[03]:00:MF/ADF.ISD)> put_key --key-id 1 --key-type aes --key-data 542C37A6043679F2F9F71116418B1CD5542C37A6043679F2F9F71116418B1CD5 --key-type aes --key-data 34F11BAC8E5390B57F4E601372339E3C34F11BAC8E5390B57F4E601372339E3C --key-type aes --key-data 5524F4BECFE96FB63FC29D6BAAC6058B5524F4BECFE96FB63FC29D6BAAC6058B --key-version-nr 51
In case of success, we should see the keyset in the `key_information`
::
pySIM-shell (SCP02[03]:00:MF/ADF.ISD)> get_data key_information

267
docs/pysim_fs_sphinx.py
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}

2
docs/saip-tool.rst
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
::

8
docs/shell.rst
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

14
pySim-prog.py
View File

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

12
pySim-read.py
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)

65
pySim-shell.py
View File

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

16
pySim/ara_m.py
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):

46
pySim/card_key_provider.py
View File

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

2
pySim/cdma_ruim.py
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),

38
pySim/esim/http_json_api.py
View File

@@ -19,7 +19,7 @@ import abc
import requests
import logging
import json
from typing import Optional, Tuple
from typing import Optional
import base64
from twisted.web.server import Request
@@ -180,7 +180,7 @@ class JsonHttpApiFunction(abc.ABC):
# receives from the a requesting client. The same applies vice versa to class variables that have an "output_"
# prefix.
# path of the API function (e.g. '/gsma/rsp2/es2plus/confirmOrder', see also method rewrite_url).
# path of the API function (e.g. '/gsma/rsp2/es2plus/confirmOrder')
path = None
# dictionary of input parameters. key is parameter name, value is ApiParam class
@@ -336,22 +336,6 @@ class JsonHttpApiFunction(abc.ABC):
output[p] = p_class.decode(v)
return output
def rewrite_url(self, data: dict, url: str) -> Tuple[dict, str]:
"""
Rewrite a static URL using information passed in the data dict. This method may be overloaded by a derived
class to allow fully dynamic URLs. The input parameters required for the URL rewriting may be passed using
data parameter. In case those parameters are additional parameters that are not intended to be passed to
the encode_client method later, they must be removed explcitly.
Args:
data: (see JsonHttpApiClient and JsonHttpApiServer)
url: statically generated URL string (see comment in JsonHttpApiClient)
"""
# This implementation is a placeholder in which we do not perform any URL rewriting. We just pass through data
# and url unmodified.
return data, url
class JsonHttpApiClient():
def __init__(self, api_func: JsonHttpApiFunction, url_prefix: str, func_req_id: Optional[str],
session: requests.Session):
@@ -368,16 +352,8 @@ class JsonHttpApiClient():
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 fields. `data` may also contain additional parameters required for URL rewriting (see
rewrite_url in class JsonHttpApiFunction). Output data is returned as json-deserialized dict.
Args:
data: Input data required to perform the request.
func_call_id: Function Call Identifier, if present a header field is generated automatically.
timeout: Maximum amount of time to wait for the request to complete.
"""
"""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
@@ -386,11 +362,6 @@ class JsonHttpApiClient():
data = {'header' : {'functionRequesterIdentifier': self.func_req_id,
'functionCallIdentifier': func_call_id}} | data
# The URL used for the HTTP request (see below) normally consists of the initially given url_prefix
# concatenated with the path defined by the JsonHttpApiFunction definition. This static URL path may be
# rewritten by rewrite_url method defined in the JsonHttpApiFunction.
data, url = self.api_func.rewrite_url(data, self.url_prefix + self.api_func.path)
# Encode the message (the presence of mandatory fields is checked during encoding)
encoded = json.dumps(self.api_func.encode_client(data))
@@ -402,6 +373,7 @@ class JsonHttpApiClient():
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.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))

9
pySim/esim/saip/__init__.py
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'

120
pySim/esim/saip/batch.py
View File

@@ -1,120 +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
from typing import Generator
from pySim.esim.saip.personalization import ConfigurableParameter
from pySim.esim.saip import param_source
from pySim.esim.saip import ProfileElementSequence
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 = open('some_file', 'rb').read()
pes = ProfileElementSequence.from_der(der_input)
p = 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):
if isinstance(param, type):
self.param_cls = param
else:
self.param_cls = param.__class__
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 generator (e.g. iter(list_of_rows)) producing all CSV rows one at a time, starting with a row
containing the column headers. This is compatible with the python csv.reader. Each row gets passed to
ParamSource.get_next(), such that ParamSource implementations can access the row items. See
param_source.CsvSource.
"""
self.n = n
self.params = params or []
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, 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_cls.validate_val(input_value)
p.param_cls.apply_val(pes, value)
except Exception as e:
raise ValueError(f'{p.param_cls.get_name()} fed by {p.src.name}: {e}') from e
yield pes

203
pySim/esim/saip/param_source.py
View File

@@ -1,203 +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 random
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
def __init__(self, input_str:str):
"""Subclasses should call super().__init__(input_str) before evaluating self.input_str. Each subclass __init__()
may in turn manipulate self.input_str to apply expansions or decodings."""
self.input_str = input_str
def get_next(self, csv_row:dict=None):
"""Subclasses implement this: return the next value from the parameter source.
When there are no more values from the source, raise a ParamSourceExhaustedExn.
This default implementation is an empty source."""
raise ParamSourceExhaustedExn()
@classmethod
def from_str(cls, input_str:str):
"""compatibility with earlier version of ParamSource. Just use the constructor."""
return cls(input_str)
class ConstantSource(ParamSource):
"""one value for all"""
name = "constant"
def get_next(self, csv_row:dict=None):
return self.input_str
class InputExpandingParamSource(ParamSource):
def __init__(self, input_str:str):
super().__init__(input_str)
self.input_str = self.expand_input_str(self.input_str)
@classmethod
def expand_input_str(cls, input_str:str):
# user convenience syntax '0*32' becomes '00000000000000000000000000000000'
if "*" not in input_str:
return input_str
# re: "XX * 123" with optional spaces
tokens = re.split(r"([^ \t]+)[ \t]*\*[ \t]*([0-9]+)", input_str)
if len(tokens) < 3:
return input_str
parts = []
for unchanged, snippet, repeat_str in zip(tokens[0::3], tokens[1::3], tokens[2::3]):
parts.append(unchanged)
repeat = int(repeat_str)
parts.append(snippet * repeat)
return "".join(parts)
class DecimalRangeSource(InputExpandingParamSource):
"""abstract: decimal numbers with a value range"""
numeric_base = 10
def __init__(self, input_str:str=None, num_digits:int=None, first_value:int=None, last_value:int=None):
"""Constructor to set up values from a (user entered) string: DecimalRangeSource(input_str).
Constructor to set up values directly: DecimalRangeSource(num_digits=3, first_value=123, last_value=456)
num_digits produces leading zeros when first_value..last_value are shorter.
"""
assert ((input_str is not None and (num_digits, first_value, last_value) == (None, None, None))
or (input_str is None and None not in (num_digits, first_value, last_value)))
if input_str is not None:
super().__init__(input_str)
input_str = self.input_str
if ".." in input_str:
first_str, last_str = input_str.split('..')
first_str = first_str.strip()
last_str = last_str.strip()
else:
first_str = input_str.strip()
last_str = None
num_digits = len(first_str)
first_value = int(first_str)
last_value = int(last_str if last_str is not None else "9" * num_digits)
assert num_digits > 0
assert first_value <= last_value
self.num_digits = num_digits
self.first_value = first_value
self.last_value = last_value
def val_to_digit(self, val:int):
return "%0*d" % (self.num_digits, val) # pylint: disable=consider-using-f-string
class RandomDigitSource(DecimalRangeSource):
"""return a different sequence of random decimal digits each"""
name = "random decimal digits"
def get_next(self, csv_row:dict=None):
val = random.randint(self.first_value, self.last_value) # TODO secure random source?
return self.val_to_digit(val)
class RandomHexDigitSource(InputExpandingParamSource):
"""return a different sequence of random hexadecimal digits each"""
name = "random hexadecimal digits"
numeric_base = 16
def __init__(self, input_str:str):
super().__init__(input_str)
input_str = self.input_str
num_digits = len(input_str.strip())
if num_digits < 1:
raise ValueError("zero number of digits")
# hex digits always come in two
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):
val = random.randbytes(self.num_digits // 2) # TODO secure random source?
return b2h(val)
class IncDigitSource(DecimalRangeSource):
"""incrementing sequence of digits"""
name = "incrementing decimal digits"
def __init__(self, input_str:str=None, num_digits:int=None, first_value:int=None, last_value:int=None):
"""input_str: the range of values to iterate. Format: 'FIRST..LAST' (e.g. '0001..9999') or
just 'FIRST' (iterates to the maximum value for the given digit width). Leading zeros in
FIRST determine the digit width and are preserved in returned values."""
super().__init__(input_str, num_digits, first_value, last_value)
self.next_val = None
self.reset()
def reset(self):
"""Restart from the first value of the defined range passed to __init__()."""
self.next_val = self.first_value
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.last_value:
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, input_str:str):
"""input_str: the CSV column name to read values from.
The caller passes the current CSV row to get_next(), from which CsvSource picks the column matching
this name."""
super().__init__(input_str)
self.csv_column = self.input_str
def get_next(self, csv_row:dict=None):
val = None
if csv_row:
val = csv_row.get(self.csv_column)
if val is None:
raise ParamSourceUndefinedExn(f"no value for CSV column {self.csv_column!r}")
return val

372
pySim/esim/saip/personalization.py
View File

@@ -16,22 +16,13 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import abc
import enum
import io
import re
from typing import List, Tuple, Generator, Optional
from typing import List, Tuple
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.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'."""
@@ -126,7 +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
def __init__(self, input_value=None):
self.input_value = input_value # the raw input value as given by caller
@@ -209,29 +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
@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
@@ -252,13 +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
class DecimalParam(ConfigurableParameter):
"""Decimal digits. The input value may be a string of decimal digits like '012345', or an int. The output of
@@ -289,7 +249,6 @@ class DecimalHexParam(DecimalParam):
@classmethod
def validate_val(cls, val):
val = super().validate_val(val)
assert isinstance(val, str)
val = ''.join('%02x' % ord(x) for x in val)
if cls.rpad is not None:
c = cls.rpad_char
@@ -297,17 +256,6 @@ 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'
@@ -331,19 +279,10 @@ 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)
allow_chars = '0123456789abcdefABCDEF'
strip_chars = ' \t\r\n'
default_source = param_source.RandomHexDigitSource
@classmethod
def validate_val(cls, val):
@@ -362,82 +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):
"""ConfigurableParameter for named integer enumeration values.
Subclasses must define a nested enum.IntEnum named 'Values' listing all valid names and their
integer codes. apply_val() and get_values_from_pes() are not implemented here and this must
be inherited from another mixin."""
class Values(enum.IntEnum):
pass # subclasses override this
@classmethod
def validate_val(cls, val) -> int:
if isinstance(val, int):
try:
return int(cls.Values(val))
except ValueError:
pass
elif isinstance(val, str):
member = cls.map_name_to_val(val, strict=False)
if member is not None:
return member
valid = ', '.join(m.name for m in cls.Values)
raise ValueError(f"{cls.get_name()}: invalid argument: {val!r}. Valid arguments are: {valid}")
@classmethod
def map_name_to_val(cls, name: str, strict=True) -> int:
"""Return the integer value for a given enum member name. Performs an exact match first,
then falls back to fuzzy matching (case-insensitive, punctuation-insensitive)."""
try:
return int(cls.Values[name])
except KeyError:
pass
clean = cls.clean_name_str(name)
for member in cls.Values:
if cls.clean_name_str(member.name) == clean:
return int(member)
if strict:
valid = ', '.join(m.name for m in cls.Values)
raise ValueError(f"{cls.get_name()}: {name!r} is not a known value. Known values are: {valid}")
return None
@classmethod
def map_val_to_name(cls, val, strict=False) -> str:
"""Return the enum member name for a given integer value."""
try:
return cls.Values(val).name
except ValueError:
if strict:
raise ValueError(f"{cls.get_name()}: {val!r} ({type(val).__name__}) is not a known value.")
return None
@classmethod
def name_normalize(cls, name: str) -> str:
"""Map a (possibly fuzzy) name to its canonical enum member name."""
return cls.Values(cls.map_name_to_val(name)).name
@classmethod
def clean_name_str(cls, val: str) -> str:
"""Strip punctuation and case for fuzzy name comparison.
Treats hyphens and underscores as equivalent (both removed)."""
return re.sub('[^0-9A-Za-z]', '', val).lower()
class Iccid(DecimalParam):
"""ICCID Parameter. Input: string of decimal digits.
@@ -446,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):
@@ -460,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."""
@@ -479,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):
@@ -492,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
@@ -515,41 +353,22 @@ class SmspTpScAddr(ConfigurableParameter):
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):
@@ -579,32 +398,6 @@ 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['ef-smsp']
ef_smsp = EF_SMSP()
ef_smsp_dec = ef_smsp.decode_record_bin(f_smsp.body, 1)
tp_sc_addr = ef_smsp_dec.get('tp_sc_addr', None)
if not tp_sc_addr:
continue
digits = tp_sc_addr.get('call_number', None)
if not digits:
continue
ton_npi = tp_sc_addr.get('ton_npi', None)
if not ton_npi:
continue
international = ton_npi.get('type_of_number', None)
if international is None:
continue
international = (international == 'international')
yield { cls.name: cls.tuple_to_str((international, digits)) }
class SdKey(BinaryParam, metaclass=ClassVarMeta):
"""Configurable Security Domain (SD) Key. Value is presented as bytes."""
# these will be set by subclasses
@@ -614,40 +407,28 @@ class SdKey(BinaryParam, metaclass=ClassVarMeta):
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=KeyUsageQualifier.build(cls.key_usage_qual),
key_components=set_components,
)
pe.add_key(key)
else:
key.key_components = set_components
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)
@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) }
def apply_val(cls, pes: ProfileElementSequence, value):
for pe in pes.get_pes_for_type('securityDomain'):
cls._apply_sd(pe, value)
class SdKeyScp80_01(SdKey, kvn=0x01, key_type=0x88, permitted_len=[16,24,32]): # AES key type
pass
@@ -721,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):
@@ -736,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
@@ -757,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
@@ -780,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):
@@ -816,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
@@ -848,59 +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"
class AlgorithmID(DecimalParam, AlgoConfig):
algo_config_key = 'algorithmID'
example_input = "Milenage"
default_source = param_source.ConstantSource
# as in pySim/esim/asn1/saip/PE_Definitions-3.3.1.asn
class Values(enum.IntEnum):
Milenage = 1
TUAK = 2
usim_test = 3 # input 'usim-test' also accepted via fuzzy matching
# EnumParam.validate_val() returns the int values from Values
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'
@@ -914,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):
@@ -945,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"""
@@ -954,4 +667,3 @@ class TuakNumberOfKeccak(IntegerParam, AlgoConfig):
min_val = 1
max_val = 255
example_input = '1'
default_source = param_source.ConstantSource

102
pySim/filesystem.py
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:

58
pySim/global_platform/__init__.py
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)

30
pySim/global_platform/install_param.py
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())

2
pySim/global_platform/scp.py
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.

1
pySim/global_platform/uicc.py
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

3
pySim/legacy/cards.py
View File

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

12
pySim/log.py
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):
"""

4
pySim/ota.py
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:

10
pySim/sms.py
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)

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

21
pySim/transport/pcsc.py
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:

6
pySim/ts_31_102.py
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):

94
pySim/ts_51_011.py
View File

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

4
pySim/utils.py
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

2
requirements.txt
View File

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

2
setup.py
View File

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

4
tests/pySim-prog_test/Fairwaves-SIM.ok
View File

@@ -1,11 +1,11 @@
INFO: Using PC/SC reader interface
Using PC/SC reader interface
Reading ...
Autodetected card type: Fairwaves-SIM
ICCID: 8988219000000117833
IMSI: 001010000000111
GID1: ffffffffffffffff
GID2: ffffffffffffffff
SMSP: ffffffffffffffffffffffffffffe1ffffffffffffffffffffffff0581005155f5ffffffffffff000000
SMSP: e1ffffffffffffffffffffffff0581005155f5ffffffffffff000000ffffffffffffffffffffffffffff
SMSC: 0015555
SPN: Fairwaves
Show in HPLMN: False

4
tests/pySim-prog_test/Wavemobile-SIM.ok
View File

@@ -1,11 +1,11 @@
INFO: Using PC/SC reader interface
Using PC/SC reader interface
Reading ...
Autodetected card type: Wavemobile-SIM
ICCID: 89445310150011013678
IMSI: 001010000000102
GID1: Can't read file -- SW match failed! Expected 9000 and got 6a82.
GID2: Can't read file -- SW match failed! Expected 9000 and got 6a82.
SMSP: ffffffffffffffffffffffffffffe1ffffffffffffffffffffffff0581005155f5ffffffffffff000000
SMSP: e1ffffffffffffffffffffffff0581005155f5ffffffffffff000000ffffffffffffffffffffffffffff
SMSC: 0015555
SPN: wavemobile
Show in HPLMN: False

2
tests/pySim-prog_test/fakemagicsim.ok
View File

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

2
tests/pySim-prog_test/sysmoISIM-SJA2.ok
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

2
tests/pySim-prog_test/sysmoISIM-SJA5.ok
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

2
tests/pySim-prog_test/sysmoUSIM-SJS1.ok
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

2
tests/pySim-prog_test/sysmosim-gr1.ok
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

14
tests/pySim-shell_test/apdu/test_apdu.script
View File

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

2
tests/pySim-smpp2sim_test/pySim-smpp2sim_test.sh
View File

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

13
tests/unittests/test_files.py
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):

144
tests/unittests/test_fs_coverage.py
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()

2
tests/unittests/test_globalplatform.py
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__":