Hoernchen has uploaded this change for review.
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()
To view, visit change 43547. To unsubscribe, or for help writing mail filters, visit settings.