Hoernchen has uploaded this change for review. ( https://gerrit.osmocom.org/c/pysim/+/43547?usp=email )
Change subject: contrib: SCP81 TLS-PSK RAS for RAM/RFM over HTTPS ......................................................................
contrib: SCP81 TLS-PSK RAS for RAM/RFM over HTTPS
GP Amendment B "Remote Application Management over HTTP" v1.2 and ETSI TS 102 226 RFM/RAM server.
APDU command/response bodies use the Expanded Remote Application data format.
--mode {ram,rfm} selects what we want to do, make sure to target the right thing in the right mode, SD for RAM, AID for RFM.
This is hand-rolled on purpose: Klein/Twisted do not help us here, have annoying headers and settings, and don't work well at all if you want to import the file for tests and so on. I have wasted quite some time trying to make this prettier by re-using existing http handler parts because I thought well we import all kinds of packages already it can't be that hard, unfortunately it is.
Change-Id: I4f243aeca65d0f33ff34e6932e3f4e156446af8b --- A contrib/scp81_admin_server.py 1 file changed, 530 insertions(+), 0 deletions(-)
git pull ssh://gerrit.osmocom.org:29418/pysim refs/changes/47/43547/1
diff --git a/contrib/scp81_admin_server.py b/contrib/scp81_admin_server.py new file mode 100644 index 0000000..df1a7d0 --- /dev/null +++ b/contrib/scp81_admin_server.py @@ -0,0 +1,530 @@ +#!/usr/bin/env python3 +"""TLS-PSK HTTP Remote Administration Server for SCP81 / RAM over HTTP""" + +# (C) 2026 by sysmocom - s.f.m.c. GmbH info@sysmocom.de +# +# Author: Eric Wild +# +# 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/. + +# The card (specifically a SD supporting SCP81) is the TLS client: +# - it opens a TCP connection to this server, +# - performs a TLS handshake authenticated with a PSK, +# - and then drives the HTTP admin loop of to fetch remote APDU command strings +# - and posts back their responses. +# This program is the server side of that exchange, it: +# - accepts the PSK-TLS connection, +# - hands the card a queue of commands +# - and logs the decoded responses. +# +# The two TS 102 226 annex B figure B.1 administration modes are supported over the +# same session, selected with --mode: +# ram GP Amendment B RAM: +# command is handled by (--targeted-application) a SD +# rfm ETSI TS 102 226 RFM/RAM: +# command is routed to the Receiving/RFM Application specified by +# --targeted-application, for example UICC-filesystem/USIM-ADF RFM app. +# +# Remote APDU command/response bodies use the Expanded Remote Application +# data format. +# + +import ssl +import socket +import logging +import argparse +import threading +from pathlib import Path +from typing import List, Optional, Callable, Dict, Tuple + +from osmocom.utils import h2b, b2h + +from pySim.ota import encode_expanded_cmd, decode_expanded_resp + +logger = logging.getLogger(Path(__file__).stem) + +# Amendment B section 3.4 +ADMIN_PROTOCOL = 'globalplatform-remote-admin/1.0' +CT_COMMAND = 'application/vnd.globalplatform.card-content-mgt;version=1.0' +CT_RESPONSE = 'application/vnd.globalplatform.card-content-mgt-response;version=1.0' + +# TS 102 226 annex B, figure B.1 RFM/RAM over HTTPS content types +CT_RFM_COMMAND = 'application/vnd.etsi.scp.command-data;version=1.0' +CT_RFM_RESPONSE = 'application/vnd.etsi.scp.response-data;version=1.0' + +MODE_CONTENT_TYPE = { + 'ram': CT_COMMAND, # GP Amendment B RAM: target = a Security Domain + 'rfm': CT_RFM_COMMAND, # ETSI TS 102 226 RFM/RAM: target = an application +} + +# Amendment B Table 3-2, modern TLS 1.2 suites first. +# The deprecated TLS 1.0/1.1 suites (PSK-AES128-CBC-SHA) and 3DES/NULL are left +# out and can be enabled with --ciphers / --seclevel. +DEFAULT_CIPHERS = ':'.join([ + 'PSK-AES128-GCM-SHA256', # TLS_PSK_WITH_AES_128_GCM_SHA256 + 'PSK-AES256-GCM-SHA384', + 'PSK-AES128-CBC-SHA256', # TLS_PSK_WITH_AES_128_CBC_SHA256 (Table 3-2) + 'PSK-AES256-CBC-SHA384', + 'PSK-AES128-CBC-SHA', # TLS_PSK_WITH_AES_128_CBC_SHA (deprecated) + 'PSK-AES256-CBC-SHA', +]) + +TLS_VERSION_MAP = { + '1.0': ssl.TLSVersion.TLSv1, + '1.1': ssl.TLSVersion.TLSv1_1, + '1.2': ssl.TLSVersion.TLSv1_2, + '1.3': ssl.TLSVersion.TLSv1_3, +} + + +def format_aid(aid: str) -> str: + """AID -> //aid/<RID>/<PIX> for X-Admin-Targeted-Application from Amendment B section 3.4.2 + First 5 bytes RID, the PIX the remainder, string in //aid/ notation is passed through.""" + if aid.startswith('//aid/'): + return aid + aid = aid.replace(' ', '').lower() + if len(aid) < 10: + raise ValueError('AID must be at least 5 bytes?! (RID): %r' % aid) + rid, pix = aid[:10], aid[10:] + return '//aid/%s/%s' % (rid, pix) + + +def make_ssl_context(psk: bytes, identity: str, *, + ciphers: str = DEFAULT_CIPHERS, + min_tls: str = '1.2', max_tls: str = '1.3', + seclevel: Optional[int] = None, + identity_hint: Optional[str] = None, + allow_any_identity: bool = False, + extra_psks: Optional[Dict[str, bytes]] = None) -> ssl.SSLContext: + """PSK SSLContext, resolvesg the key from the client psk_identity + + extra_psks can carry additional identity->key mappings. + allow_any_identity can be used for debugging + """ + # The ssl PSK callback return must be immutable bytes but h2b() and + # similar helpers give us a mutable bytearray which F/$3/$ed me five times already + psk = bytes(psk) + keymap: Dict[str, bytes] = {identity: psk} + if extra_psks: + keymap.update({k: bytes(v) for k, v in extra_psks.items()}) + + ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + ctx.minimum_version = TLS_VERSION_MAP[min_tls] + ctx.maximum_version = TLS_VERSION_MAP[max_tls] + cipher_str = ciphers + if seclevel is not None: + # @SECLEVEL=0 is to enable NULL/3DES/legacy PSK suites + cipher_str = '%s:@SECLEVEL=%d' % (ciphers, seclevel) + if cipher_str: + ctx.set_ciphers(cipher_str) + + def psk_server_callback(client_identity: Optional[str]) -> bytes: + if allow_any_identity: + logger.info('PSK handshake: identity=%r (ACEPTING ANY!)', client_identity) + return psk + key = keymap.get(client_identity) + if key is None: + logger.warning('PSK handshake: unknown identity %r (known: %r) -> rejecting', + client_identity, list(keymap.keys())) + return b'' # empty PSK aborts handshake + logger.info('PSK handshake: identity=%r resolved', client_identity) + return key + + ctx.set_psk_server_callback(psk_server_callback, identity_hint=identity_hint) + return ctx + + +class HttpRequest: + """A parsed HTTP request (request line + headers + body).""" + __slots__ = ('method', 'uri', 'version', 'headers', 'body') + + def __init__(self, method: str, uri: str, version: str, + headers: Dict[str, str], body: bytes): + self.method = method + self.uri = uri + self.version = version + self.headers = headers # lower cased field names + self.body = body + + def get(self, name: str, default=None) -> Optional[str]: + return self.headers.get(name.lower(), default) + + +# http.client bound on a single HTTP line. +MAX_LINE = 65536 + + +def _read_line(rfile) -> bytes: + """readline() with a bound. reaching the bound without a + terminator means the card is out of sync somehow, not that the line is long.""" + line = rfile.readline(MAX_LINE) + if line and not line.endswith(b'\n'): + raise ValueError('HTTP line longer than %u bytes' % MAX_LINE) + return line + + +def _read_chunked_body(rfile) -> bytes: + """Read a Transfer-Encoding: chunked body. Amendment B section 3.4.1 lets + the card send its response string with either a Content-Length or chunked""" + out = bytearray() + while True: + size_line = _read_line(rfile) + if not size_line: + break + size = int(size_line.split(b';', 1)[0].strip() or b'0', 16) + if size == 0: + # consume trailer headers up to the terminating blank line + while _read_line(rfile) not in (b'\r\n', b'\n', b''): + pass + break + chunk = rfile.read(size) + if len(chunk) != size: + raise ValueError('chunked body ended after %u of %u bytes' % (len(chunk), size)) + out += chunk + _read_line(rfile) # trailing CRLF after the chunk data + return bytes(out) + + +def read_http_request(rfile, send: Optional[Callable[[bytes], None]] = None) -> Optional[HttpRequest]: + """Read one HTTP request from a buffered binary reader or None when closed""" + request_line = _read_line(rfile) + if not request_line: + return None + parts = request_line.rstrip(b'\r\n').decode('iso-8859-1').split(' ') + if len(parts) < 3: + raise ValueError('Malformed HTTP request line: %r' % request_line) + method, uri, version = parts[0], parts[1], parts[2] + + headers: Dict[str, str] = {} + while True: + line = _read_line(rfile) + if line in (b'\r\n', b'\n', b''): + break + name, _, value = line.rstrip(b'\r\n').decode('iso-8859-1').partition(':') + headers[name.strip().lower()] = value.strip() + + # Expect: 100-continue waits for the response before it sends the body, + # and RFC 2616 8.2.3 (Amendment B references RFC 2616 as [HTTP]) + # requires the server to send it. Amendment B 3.4.1 does not mention this + # header, tho, might be useless. + if send and '100-continue' in headers.get('expect', '').lower(): + logger.info('-> 100 Continue ') + send(b'HTTP/1.1 100 Continue\r\n\r\n') + + body = b'' + te = headers.get('transfer-encoding', '').lower() + if 'chunked' in te: + body = _read_chunked_body(rfile) + elif 'content-length' in headers: + n = int(headers['content-length']) + if n: + body = rfile.read(n) + return HttpRequest(method, uri, version, headers, body) + + +def build_http_response(status_line: str, headers: List[Tuple[str, str]], + body: bytes = b'') -> bytes: + """Serialise HTTP response. status_line 'HTTP/1.1 200 OK'.""" + lines = [status_line] + lines += ['%s: %s' % (name, value) for name, value in headers] + head = ('\r\n'.join(lines) + '\r\n\r\n').encode('iso-8859-1') + return head + body + + +def format_decoded_response(dec) -> str: + """hand over the data""" + bits = ['%u command(s) executed' % dec.number_of_commands] + for i, c in enumerate(dec.commands): + data = c.response_data or '-' + bits.append(' R-APDU[%u]: SW=%s data=%s' % (i, c.status_word, data)) + if dec.get('truncated'): + bits.append(' TRUNCATED: an R-APDU returned SW 62F1, so the card cut the response data ' + 'short and stopped executing the rest of the script ') # TS 102 226 5.2.1.1 + if dec.bad_format is not None: + bits.append(' bad-format: %s' % dec.bad_format) + if dec.immediate_action_response is not None: + bits.append(' immediate-action-response: %s' % dec.immediate_action_response) + if dec.script_chaining_response is not None: + bits.append(' script-chaining-response: %s' % dec.script_chaining_response) + return '\n'.join(bits) + + +class AdminSession: + + def __init__(self, command_bodies: List[bytes], + next_uri: Optional[str] = None, + targeted_application: Optional[str] = None, + on_response: Optional[Callable[[object], None]] = None, + content_type: str = CT_COMMAND): + self.pending: List[bytes] = list(command_bodies) + self.next_uri = next_uri # None -> echo the request URI + self.targeted_application = targeted_application + self.on_response = on_response + self.content_type = content_type # Content-Type for the command body + self.responses: List[object] = [] # decoded Containers, in order + + def record_response(self, dec) -> None: + self.responses.append(dec) + if self.on_response: + self.on_response(dec) + + +def run_admin_loop(rfile, send: Callable[[bytes], None], session: AdminSession) -> AdminSession: + """Drive the admin loop for one connection. + Just keep answering the card POST requests with the next queued command + (200 OK + Expanded command body) until the queue is empty, end session with 204 No Content.""" + while True: + req = read_http_request(rfile, send) + if req is None: + logger.info('connection closed by card') + return session + + if req.method != 'POST': + logger.warning('unexpected method %s %s -> 405', req.method, req.uri) + send(build_http_response('HTTP/1.1 405 Method Not Allowed', + [('X-Admin-Protocol', ADMIN_PROTOCOL), + ('Connection', 'close')])) + return session + + proto = req.get('x-admin-protocol') + if proto and proto != ADMIN_PROTOCOL: + logger.warning('card X-Admin-Protocol=%r (expected %r)', proto, ADMIN_PROTOCOL) + status = req.get('x-admin-script-status') + resume = req.get('x-admin-resume') + logger.info('POST %s from=%r status=%r resume=%r body=%uB', + req.uri, req.get('x-admin-from'), status, resume, len(req.body)) + + # section 3.4.1: + # - body with "X-Admin-Script-Status: ok" carries the previous command + # response string (Expanded Remote response format); + # - other status values carry no body (). + if req.body: + # Expanded Remote response: + # - GP Amd B 'card-content-mgt-response' + # - ETSI 'scp.response-data' + logger.debug(' response Content-Type=%r raw body (%uB): %s', + req.get('content-type'), len(req.body), b2h(req.body)) + if status in (None, 'ok'): + try: + dec = decode_expanded_resp(req.body) + session.record_response(dec) + logger.info('card response:\n%s', format_decoded_response(dec)) + except Exception as e: + logger.error('failed to decode response body %s: %s', b2h(req.body), e) + else: + logger.warning('body present with status=%r; ignoring', status) # section 3.4.1 + elif status and status != 'ok': + logger.info('card reported script-status=%r (no response body)', status) + + if session.pending: + body = session.pending.pop(0) + next_uri = session.next_uri or req.uri + headers = [('X-Admin-Protocol', ADMIN_PROTOCOL), + ('X-Admin-Next-URI', next_uri), + ('Content-Type', session.content_type)] + if session.targeted_application: + headers.append(('X-Admin-Targeted-Application', session.targeted_application)) + headers.append(('Content-Length', str(len(body)))) + logger.info('-> 200 OK, next command (%uB): %s', len(body), b2h(body)) + send(build_http_response('HTTP/1.1 200 OK', headers, body)) + else: + # section 3.4.2: No more commands, end session + # No Content-Type or body for 204 + logger.info('-> 204 No Content, ending administration session') + send(build_http_response('HTTP/1.1 204 No Content', + [('X-Admin-Protocol', ADMIN_PROTOCOL), + ('Connection', 'close')])) + return session + + +class Scp81AdminServer: + """Threaded TLS-PSK server that runs the Amendment B admin loop against each + connecting card.""" + + def __init__(self, host: str, port: int, ssl_ctx: ssl.SSLContext, + command_bodies: List[bytes], + next_uri: Optional[str] = None, + targeted_application: Optional[str] = None, + on_response: Optional[Callable[[object], None]] = None, + on_session_end: Optional[Callable[[AdminSession], None]] = None, + content_type: str = CT_COMMAND): + self.host = host + self.port = port + self.ssl_ctx = ssl_ctx + self.command_bodies = command_bodies + self.next_uri = next_uri + self.targeted_application = targeted_application + self.content_type = content_type + self.on_response = on_response + self.on_session_end = on_session_end + self._sock: Optional[socket.socket] = None + self._stop = threading.Event() + + def bind(self) -> int: + self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self._sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + self._sock.bind((self.host, self.port)) + self._sock.listen(5) + self._sock.settimeout(0.5) + self.port = self._sock.getsockname()[1] + return self.port + + def serve_forever(self) -> None: + if self._sock is None: + self.bind() + logger.info('SCP81 admin server listening on %s:%u (%u command(s) queued)', + self.host, self.port, len(self.command_bodies)) + while not self._stop.is_set(): + try: + conn, addr = self._sock.accept() + except socket.timeout: + continue + except OSError: + break + threading.Thread(target=self._handle, args=(conn, addr), daemon=True).start() + + def shutdown(self) -> None: + self._stop.set() + if self._sock is not None: + self._sock.close() + + def _handle(self, conn: socket.socket, addr) -> None: + try: + tls = self.ssl_ctx.wrap_socket(conn, server_side=True) + except (ssl.SSLError, OSError) as e: + logger.warning('TLS-PSK handshake with %s failed: %s', addr, e) + try: + conn.close() + except OSError: + pass + return + logger.info('TLS-PSK established with %s: %s / %s', addr, tls.version(), tls.cipher()) + session = AdminSession(self.command_bodies, next_uri=self.next_uri, + targeted_application=self.targeted_application, + on_response=self.on_response, + content_type=self.content_type) + try: + rfile = tls.makefile('rb') + run_admin_loop(rfile, tls.sendall, session) + except (ssl.SSLError, OSError, ValueError) as e: + logger.warning('session with %s aborted: %s', addr, e) + finally: + try: + tls.close() + except OSError: + pass + logger.info('session with %s ended: %u response(s) collected', addr, len(session.responses)) + if self.on_session_end: + self.on_session_end(session) + + +def build_command_bodies(apdus: List[bytes], batch: bool = False, + length_coding: str = 'definite') -> List[bytes]: + """wrpa apdus + each C-APDU -> one cmd message + one HTTP response per APDU + batch=True -> all C-APDUs in one Command Scripting template. + length_coding selects the definite or indefinite Command Scripting template.""" + if not apdus: + return [] + if batch: + return [encode_expanded_cmd(apdus, length_coding=length_coding)] + return [encode_expanded_cmd(a, length_coding=length_coding) for a in apdus] + + +def main(): + parser = argparse.ArgumentParser( + description='TLS-PSK HTTP Remote Administration Server for SCP81 / RAM over HTTP') + parser.add_argument('--host', default='0.0.0.0', help='Host/IP to bind to (default: 0.0.0.0)') + parser.add_argument('--port', type=int, default=8443, help='TCP port to bind to (default: 8443)') + parser.add_argument('--psk', required=True, + help='PSK TLS key, Amendment B key type 85 as hex') + parser.add_argument('--psk-identity', required=True, + help='Expected PSK identity string presented by the card') + parser.add_argument('--psk-identity-hint', default=None, + help='Optional PSK identity hint to send to the card (default: none)') + parser.add_argument('--allow-any-identity', action='store_true', + help='DEBUG: Accept any psk_identity') + parser.add_argument('--ciphers', default=DEFAULT_CIPHERS, + help='OpenSSL cipher string for TLS<=1.2') + parser.add_argument('--min-tls', default='1.2', choices=sorted(TLS_VERSION_MAP), + help='Minimum TLS version (default: 1.2)') + parser.add_argument('--max-tls', default='1.3', choices=sorted(TLS_VERSION_MAP), + help='Maximum TLS version (default: 1.3)') + parser.add_argument('--seclevel', type=int, default=None, + help='OpenSSL @SECLEVEL to force (0 to enable NULL/3DES/legacy PSK)') + parser.add_argument('--uri', default=None, + help='X-Admin-Next-URI to hand the card (default: request URI)') + parser.add_argument('--mode', choices=sorted(MODE_CONTENT_TYPE), default='ram', + help='"ram" = GP Amendment B RAM to a SD (default)' \ + '"rfm" = TS 102 226 RFM/RAM to the --targeted-application.') + parser.add_argument('--targeted-application', default=None, + help='X-Admin-Targeted-Application AID (hex). ' + 'Required by --mode rfm' + 'optional for --mode ram') + parser.add_argument('--length-coding', choices=('definite', 'indefinite'), default='definite', + help='Expanded format length coding "definite" "indefinite"') + parser.add_argument('--apdu', action='append', default=[], metavar='HEX', + help='one of many C-APDU (hex) to send, executed in order') + parser.add_argument('--apdu-file', default=None, + help='File with one C-APDU (hex) per line to push (# comments allowed)') + parser.add_argument('--batch', action='store_true', + help='All C-APDUs in one large command message') + parser.add_argument('--raw-cmd', action='append', default=[], metavar='HEX', + help='Debug, raw command') + parser.add_argument('-v', '--verbose', action='store_true', help='enable debug output') + args = parser.parse_args() + + logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO, + format='%(asctime)s %(levelname)s %(message)s', + datefmt='%Y-%m-%d %H:%M:%S') + + if args.mode == 'rfm' and not args.targeted_application: + parser.error('--mode rfm requires --targeted-application <RFM Application AID>') + content_type = MODE_CONTENT_TYPE[args.mode] + + apdus: List[bytes] = [h2b(a) for a in args.apdu] + if args.apdu_file: + for line in Path(args.apdu_file).read_text().splitlines(): + line = line.split('#', 1)[0].strip() + if line: + apdus.append(h2b(line)) + command_bodies = build_command_bodies(apdus, batch=args.batch, + length_coding=args.length_coding) + command_bodies += [h2b(r) for r in args.raw_cmd] + if not command_bodies: + logger.warning('no C-APDUs: the server will answer the first POST with 204...') + + targeted = format_aid(args.targeted_application) if args.targeted_application else None + logger.info('mode=%s content-type=%s length-coding=%s targeted-application=%s', + args.mode, content_type, args.length_coding, targeted or '(none)') + + ssl_ctx = make_ssl_context(h2b(args.psk), args.psk_identity, + ciphers=args.ciphers, + min_tls=args.min_tls, max_tls=args.max_tls, + seclevel=args.seclevel, + identity_hint=args.psk_identity_hint, + allow_any_identity=args.allow_any_identity) + + server = Scp81AdminServer(args.host, args.port, ssl_ctx, command_bodies, + next_uri=args.uri, targeted_application=targeted, + content_type=content_type) + try: + server.serve_forever() + except KeyboardInterrupt: + logger.info('shutting down') + server.shutdown() + + +if __name__ == '__main__': + main()