#!/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 . import sys import os 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, load_json_schema, JsonValidator from pySim.transport import init_reader, argparse_add_reader_args, LinkBase from packaging.version import Version SERVER_TIMEOUT = 10 # The RCP Client software version shall be incremented when there are changes to the RCP Client (this module) or changes # to other related modules, which affect the RCP Client. The RCP Client software version is also disclosed towards the # RCP Server. RCPC_VERSION_SOFTWARE = "1.0.0" # The RCP Client protocol version refers to the protocol spoken between RCP Client and RCP Server. The protocol version # shall be incremented when there are changes to the protocol (JSON Schema and/or application logic, see also # RCPC_VERSION_PROTOCOL in rcp_server.py). RCPC_VERSION_PROTOCOL = "1.0.0" 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 check_version(self): """ Send the Protocol and Software version of this RCP Client to the RCP Server. The RCP Server will then check if this client is (still) compatible. If an incompatibility is detected, the connection will be closed. """ log.info("Checking version ...") tx_json = {'rcpc_version': {'software' : RCPC_VERSION_SOFTWARE, 'protocol' : RCPC_VERSION_PROTOCOL}} log.info("RCP Client version: software=%s, protocol=%s", RCPC_VERSION_SOFTWARE, RCPC_VERSION_PROTOCOL) rx_json = await self._transact(tx_json) rcps_version_software = Version(rx_json['rcpc_version']['software']) rcps_version_protocol = Version(rx_json['rcpc_version']['protocol']) rcps_version_info = str(rx_json['rcpc_version'].get('info')) if rcps_version_info: log.info("RCP Server version: software=%s, protocol=%s", rcps_version_software, rcps_version_protocol) else: log.info("RCP Server version: software=%s, protocol=%s, %s", rcps_version_software, rcps_version_protocol, rcps_version_info) 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: rcpc_to_rcps_schema = load_json_schema(os.path.join(Path(__file__).parent.resolve(), "rcpc_to_rcps_schema.json")) rcps_to_rcpc_schema = load_json_schema(os.path.join(Path(__file__).parent.resolve(), "rcps_to_rcpc_schema.json")) json_validator = JsonValidator(rcps_to_rcpc_schema, rcpc_to_rcps_schema) client = RcpcCltConnHdlr(sl, websocket, SERVER_TIMEOUT, json_validator) # Check software and protocol version await client.check_version() # Retrieve module description module_descrs = await client.describe({"atr" : card_atr}) # Complete the commandline 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 help screen 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 help screen. 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)