rename OTAMan → SIMple (v2.3.0)
- package pysim_otaman_server → pysim_simple_server; distribution and console script pysim-otaman-server → pysim-simple-server - PWA branding SIMple: title, header, manifest, help EN/RU, GitHub links - localStorage keys otaman_* → simple_* (clean break, no migration) - export filenames simple-cards.json / simple-custom-files.json - service worker cache simple-v189 - README/docs/setup/start scripts updated; no 'otaman' left in the tree
This commit is contained in:
@@ -0,0 +1,252 @@
|
||||
import argparse
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import traceback
|
||||
from http.server import HTTPServer
|
||||
from pySim.card_handler import CardHandler
|
||||
from pySim.commands import SimCardCommands
|
||||
from pySim.exceptions import NoCardError
|
||||
from pySim.log import PySimLogger
|
||||
from pySim.cards import UiccCardBase
|
||||
|
||||
from .shell import load_pysim_app
|
||||
from . import fastinit
|
||||
from .server import PysimHandler, StderrApduTracer, _LoggingApduTracer, VERSION, _send_terminal_profile, _DefaultProactiveHandler, _handle_proactive_chain, _send_status, _init_proactive_session, _timing_on, _tlog, _set_menu_timeout, start_card_monitor, set_auto_equip, _read_iccid, _LineFilter
|
||||
|
||||
|
||||
_server_start = 0
|
||||
|
||||
def _log_stdout(msg):
|
||||
elapsed = time.time() - _server_start
|
||||
os.write(1, ('[%8.3f] %s\n' % (elapsed, msg)).encode())
|
||||
|
||||
|
||||
def _default_web_dir():
|
||||
# <repo>/frontend, whether run from source or an editable install.
|
||||
return os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'frontend')
|
||||
|
||||
|
||||
def main():
|
||||
global _server_start
|
||||
_server_start = time.time()
|
||||
mod = load_pysim_app()
|
||||
parser = mod.option_parser
|
||||
parser.description = 'pysim-simple-server — HTTP API for pysim'
|
||||
parser.add_argument('--http-host', default='127.0.0.1', help='Bind address (default: 127.0.0.1)')
|
||||
parser.add_argument('--http-port', type=int, default=8080, help='Bind port (default: 8080)')
|
||||
parser.add_argument('--web-dir', default=_default_web_dir(), metavar='PATH',
|
||||
help='Directory with the SIMple PWA static files to serve (default: <repo>/frontend)')
|
||||
parser.add_argument('--log-requests', action='store_true', default=False, help='Log request/response payloads to stderr')
|
||||
parser.add_argument('--sms-oa', default='12345', metavar='DIGITS',
|
||||
help='TP-Originating-Address (SMSC number) for the SMS-DELIVER TPDU (default: 12345)')
|
||||
parser.add_argument('--sms-sm-sc', default='12345678912', metavar='DIGITS',
|
||||
help='SM-SC address for SMS-SUBMIT routing in PoR-in-submit mode (default: 12345678912)')
|
||||
parser.add_argument('--terminal-profile',
|
||||
default='FFFFFFFF7F9F00DFFF03021FE2000000C3FB000704117800710100000038428003',
|
||||
metavar='HEX',
|
||||
help='TERMINAL PROFILE payload (default: the 33-byte profile of a real BIP-capable handset - the live card only starts HTTP OTA when BIP events/commands are advertised)')
|
||||
parser.add_argument('--poll-interval', type=int, default=30, metavar='SECS',
|
||||
help='Idle interval before automatic STATUS polling (1-255 seconds, default: 30). Disable with --poll-interval 0')
|
||||
parser.add_argument('--no-card-init', action='store_true', default=False,
|
||||
help='Skip pysim card initialization (preserve CAT session — no file manager)')
|
||||
parser.add_argument('--timing', action='store_true', default=False,
|
||||
help='Log phase durations, card resets and APDU counters with elapsed timestamps')
|
||||
parser.add_argument('--fast-init', action='store_true', help=argparse.SUPPRESS)
|
||||
parser.add_argument('--full-pysim-init', action='store_true', default=False,
|
||||
help="Use pysim's stock init_card/equip (multiple physical card resets) instead of the default reset-free fast init")
|
||||
parser.add_argument('--menu-timeout', type=int, default=60, metavar='SECS',
|
||||
help='Auto-send a timeout TERMINAL RESPONSE if a paused STK command is not answered (default: 60, 0 disables)')
|
||||
parser.add_argument('--no-auto-equip', action='store_true', default=False,
|
||||
help='Do not automatically initialize a card right after it is inserted (default: auto-equip on)')
|
||||
|
||||
opts = parser.parse_args()
|
||||
opts.skip_card_init = opts.no_card_init
|
||||
opts.fast_init = not opts.full_pysim_init
|
||||
if opts.timing:
|
||||
_timing_on()
|
||||
if opts.menu_timeout is not None:
|
||||
_set_menu_timeout(opts.menu_timeout)
|
||||
set_auto_equip(not opts.no_auto_equip and not opts.skip_card_init)
|
||||
sl = None
|
||||
scc = None
|
||||
card = None
|
||||
rs = None
|
||||
sim_menu = None
|
||||
event_list = None
|
||||
# Auto-detect PC/SC reader if none was explicitly specified.
|
||||
# Handles late pcscd startup and USB enumeration delays.
|
||||
if opts.pcsc_dev is None and opts.pcsc_regex is None:
|
||||
try:
|
||||
from smartcard.System import readers
|
||||
for attempt in range(3):
|
||||
r = readers()
|
||||
if r:
|
||||
sys.stderr.write('INIT: PC/SC reader detected: %s\n' % r[0])
|
||||
opts.pcsc_dev = 0
|
||||
break
|
||||
if attempt < 2:
|
||||
sys.stderr.write('INIT: no PC/SC readers found, retrying in 2s...\n')
|
||||
time.sleep(2)
|
||||
except Exception:
|
||||
pass # smartcard module not available or pcscd unreachable
|
||||
|
||||
try:
|
||||
kwargs = {}
|
||||
if opts.apdu_trace:
|
||||
kwargs['apdu_tracer'] = _LoggingApduTracer()
|
||||
t_phase = time.time()
|
||||
sl = mod.init_reader(opts, **kwargs)
|
||||
_tlog('init_reader: %.0fms' % ((time.time() - t_phase) * 1000))
|
||||
scc = SimCardCommands(sl)
|
||||
scc.cat_cla = '80' # UICC CLA default; overridden for SIM after init_card
|
||||
scc._tp.proactive_handler = _DefaultProactiveHandler()
|
||||
t_phase = time.time()
|
||||
if opts.fast_init:
|
||||
try:
|
||||
rs, card = fastinit.init_card_fast(sl, opts.skip_card_init, wait=True)
|
||||
except NoCardError:
|
||||
# Normal cardless start: there was no card in the reader (the
|
||||
# 3s wait timed out). The presence monitor auto-equips once a
|
||||
# card appears; nothing to recover, no traceback.
|
||||
sys.stderr.write('INIT: no card in the reader — server ready; insert a card or press Equip\n')
|
||||
except Exception:
|
||||
print("Warning: fast card initialization failed, falling back to pysim init:", file=sys.stderr)
|
||||
traceback.print_exc()
|
||||
try:
|
||||
rs, card = mod.init_card(sl, opts.skip_card_init)
|
||||
except NoCardError:
|
||||
# The fallback retried the cardless wait; still a normal
|
||||
# cardless start, not an initialization failure.
|
||||
sys.stderr.write('INIT: no card in the reader — server ready; insert a card or press Equip\n')
|
||||
else:
|
||||
try:
|
||||
sl.wait_for_card(3)
|
||||
rs, card = mod.init_card(sl, opts.skip_card_init)
|
||||
except NoCardError:
|
||||
sys.stderr.write('INIT: no card in the reader — server ready; insert a card or press Equip\n')
|
||||
_tlog('card_init: %.0fms' % ((time.time() - t_phase) * 1000))
|
||||
if card is not None:
|
||||
scc.cat_cla = '80' if isinstance(card, UiccCardBase) else 'a0'
|
||||
except Exception:
|
||||
print("Warning: reader/card initialization failed:", file=sys.stderr)
|
||||
traceback.print_exc()
|
||||
ch = CardHandler(sl) if sl else None
|
||||
t_phase = time.time()
|
||||
try:
|
||||
if card is not None:
|
||||
app = mod.PysimApp(verbose=opts.verbose, card=card, rs=rs, sl=sl, ch=ch)
|
||||
else:
|
||||
# Cardless start: pySim logs 'Waiting for card...' (its own retry
|
||||
# path) and pySim-shell prints 'pySim-shell not equipped!'; we
|
||||
# report both cases with our own single line above. A PysimApp
|
||||
# without a card would also retry the cardless wait, so install
|
||||
# the filter before constructing it and drop the two internals.
|
||||
saved_stdout = sys.stdout
|
||||
sys.stdout = _LineFilter(saved_stdout, ('Waiting for card...', 'pySim-shell not equipped!'))
|
||||
try:
|
||||
app = mod.PysimApp(verbose=opts.verbose, card=None, rs=None, sl=sl, ch=ch)
|
||||
finally:
|
||||
sys.stdout = saved_stdout
|
||||
except Exception:
|
||||
print("Warning: PysimApp creation failed:", file=sys.stderr)
|
||||
traceback.print_exc()
|
||||
app = None
|
||||
_tlog('pysim_app: %.0fms' % ((time.time() - t_phase) * 1000))
|
||||
if app is not None and opts.fast_init:
|
||||
fastinit.install(app)
|
||||
iccid = None
|
||||
if scc and card is not None and hasattr(scc, '_tp'):
|
||||
scc._tp.apdu_tracer = _LoggingApduTracer()
|
||||
try:
|
||||
_init_proactive_session()
|
||||
# Read EF.ICCID before the TERMINAL PROFILE starts the CAT session
|
||||
# (the PWA auto-selects the matching card preset from it).
|
||||
iccid = _read_iccid(app)
|
||||
if iccid:
|
||||
sys.stderr.write('INIT: ICCID %s\n' % iccid)
|
||||
t_phase = time.time()
|
||||
sys.stderr.write('INIT: sending TERMINAL PROFILE %s (CLA=%s)\n' % (opts.terminal_profile, scc.cat_cla))
|
||||
sm, el = _send_terminal_profile(scc, opts.terminal_profile)
|
||||
sys.stderr.write('INIT: TP done, menu=%s events=%s\n' % ('yes' if sm else 'no', 'yes' if el else 'no'))
|
||||
sim_menu = sm or sim_menu
|
||||
event_list = el or event_list
|
||||
for _ in range(3):
|
||||
st_data, st_sw = _send_status(scc)
|
||||
sys.stderr.write('INIT: drain STATUS -> %s\n' % st_sw)
|
||||
if not st_sw.startswith('91'):
|
||||
break
|
||||
_handle_proactive_chain(scc, st_sw)
|
||||
_tlog('terminal_profile_drain: %.0fms' % ((time.time() - t_phase) * 1000))
|
||||
except Exception:
|
||||
traceback.print_exc(file=sys.stderr)
|
||||
if app is not None and opts.apdu_trace:
|
||||
# PysimApp.__init__ routes PySimLogger through app.poutput() (app.stdout)
|
||||
# and drops the root level to INFO. Re-route pysim's own APDU trace logging
|
||||
# directly to fd 1 so it survives the app.stdout/StringIO redirection in the
|
||||
# HTTP handlers and the INFO level suppression.
|
||||
PySimLogger.setup(print_callback=_log_stdout)
|
||||
PySimLogger.set_level(logging.DEBUG)
|
||||
# PysimApp.__init__ and every `equip` wipe the transport apdu_tracer
|
||||
# (_onchange_apdu_trace sets it to None). Re-attach our tracer and make
|
||||
# sure it stays attached across equip/re-equip.
|
||||
tracer = _LoggingApduTracer()
|
||||
def _reattach_tracer():
|
||||
if app.card:
|
||||
app.card._scc._tp.apdu_tracer = tracer
|
||||
_reattach_tracer()
|
||||
orig_onchange = app._onchange_apdu_trace
|
||||
def _onchange_apdu_trace(param_name, old, new):
|
||||
orig_onchange(param_name, old, new)
|
||||
_reattach_tracer()
|
||||
app._onchange_apdu_trace = _onchange_apdu_trace
|
||||
server = HTTPServer((opts.http_host, opts.http_port), PysimHandler)
|
||||
server.sl = sl
|
||||
server.scc = scc
|
||||
server.card = card
|
||||
server.rs = rs
|
||||
server.app = app
|
||||
server.sms_oa = opts.sms_oa
|
||||
server.sms_sc = opts.sms_sm_sc
|
||||
server.log_requests = opts.log_requests
|
||||
server.terminal_profile = opts.terminal_profile
|
||||
server.cli_terminal_profile = opts.terminal_profile
|
||||
server.web_dir = opts.web_dir
|
||||
server.sim_menu = sim_menu
|
||||
server.event_list = event_list
|
||||
server.menu_active = False
|
||||
server.stk_pending = None
|
||||
server.card_present = card is not None
|
||||
server.card_session = 1 if card is not None else 0
|
||||
server.iccid = iccid
|
||||
server.equipping = False
|
||||
# Set server reference for polling timer and mark the card session state
|
||||
import pysim_simple_server.server
|
||||
pysim_simple_server.server._server_ref = server
|
||||
pysim_simple_server.server._CARD_CONNECTED = card is not None
|
||||
if opts.poll_interval is not None:
|
||||
pysim_simple_server.server._set_poll_interval(opts.poll_interval)
|
||||
# Auto-enable polling if card initialized successfully (unless interval is 0)
|
||||
if server.scc and server.card and opts.poll_interval != 0:
|
||||
pysim_simple_server.server._poll_enable()
|
||||
# Start presence monitoring only after the startup init: pyscard reports an
|
||||
# already-present card as "added" on the first pass, and we must not
|
||||
# auto-equip over a session we just initialized. If startup init failed,
|
||||
# that event triggers auto-equip instead — the desired retry.
|
||||
if sl is not None and getattr(sl, '_reader', None) is not None:
|
||||
start_card_monitor(str(sl._reader))
|
||||
print("─" * 70)
|
||||
print(" pysim-simple-server v%s listening on http://%s:%s" % (VERSION, opts.http_host, opts.http_port))
|
||||
print(" Open http://%s:%s in your browser for the SIMple UI (served by this server)."
|
||||
% (opts.http_host, opts.http_port))
|
||||
print("─" * 70)
|
||||
try:
|
||||
server.serve_forever()
|
||||
except KeyboardInterrupt:
|
||||
print("\nShutting down...")
|
||||
server.shutdown()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,176 @@
|
||||
"""Fast card initialization for pysim-simple-server.
|
||||
|
||||
pySim's ``init_card()`` performs several physical card resets: one per profile
|
||||
candidate tried by ``CardProfile.pick()`` plus one at the end of
|
||||
``RuntimeState.__init__``, and ``PysimApp.equip()`` resets yet again. On common
|
||||
readers each disconnect/connect costs around a second, so the stock path spends
|
||||
most of its time re-establishing a clean state (MF selected) that can also be
|
||||
restored in software.
|
||||
|
||||
This module mirrors ``pySim.app.init_card()`` with those resets removed: all
|
||||
profile probes run back-to-back on the same connection and the runtime state
|
||||
uses a software reset. It is the default init/equip path; ``--full-pysim-init``
|
||||
restores pysim's stock behavior, and the explicit ``equip``/``reset`` commands
|
||||
keep a real reconnect/physical reset.
|
||||
"""
|
||||
|
||||
import operator
|
||||
import sys
|
||||
|
||||
from pySim.cards import CardBase, SimCardBase, UiccCardBase, card_detect
|
||||
from pySim.commands import SimCardCommands
|
||||
from pySim.exceptions import ProtocolError, SwMatchError
|
||||
from pySim.filesystem import CardApplication, CardModel
|
||||
from pySim.profile import CardProfile
|
||||
from pySim.runtime import RuntimeState
|
||||
from pySim.ts_102_221 import CardProfileUICC
|
||||
from pySim.utils import all_subclasses
|
||||
|
||||
import pySim.euicc
|
||||
|
||||
from .server import _tlog
|
||||
|
||||
|
||||
class FastRuntimeState(RuntimeState):
|
||||
"""RuntimeState whose reset() restores software state (selects MF) instead
|
||||
of power-cycling the card. Use hard_reset() for an explicit reset."""
|
||||
|
||||
def reset(self, cmd_app=None):
|
||||
try:
|
||||
return self.soft_reset(cmd_app)
|
||||
except (SwMatchError, ProtocolError) as e:
|
||||
sys.stderr.write('FAST-RESET: soft reset failed (%s), falling back to physical reset\n' % e)
|
||||
return self.hard_reset(cmd_app)
|
||||
|
||||
def soft_reset(self, cmd_app=None):
|
||||
for lchan_nr in list(self.lchan.keys()):
|
||||
self.lchan[lchan_nr].scc.scp = None
|
||||
if lchan_nr == 0:
|
||||
continue
|
||||
del self.lchan[lchan_nr]
|
||||
self.adm_verified = False
|
||||
try:
|
||||
atr = self.card._scc.get_atr()
|
||||
except Exception:
|
||||
atr = None
|
||||
if cmd_app:
|
||||
cmd_app.lchan = self.lchan[0]
|
||||
self.lchan[0].select('MF', cmd_app)
|
||||
self.lchan[0].selected_adf = None
|
||||
self.identity['ATR'] = atr
|
||||
return atr
|
||||
|
||||
def hard_reset(self, cmd_app=None):
|
||||
return super().reset(cmd_app)
|
||||
|
||||
|
||||
def pick_profile_no_reset(scc):
|
||||
"""Like CardProfile.pick(), but without a physical reset between
|
||||
candidates. Each probe selects its own discriminating file, so a reset only
|
||||
costs a reconnect without changing the outcome."""
|
||||
original_reset = scc.reset_card
|
||||
scc.reset_card = lambda: None
|
||||
try:
|
||||
profiles = sorted(all_subclasses(CardProfile), key=operator.attrgetter('ORDER'))
|
||||
for p in profiles:
|
||||
if p.match_with_card(scc):
|
||||
return p()
|
||||
return None
|
||||
finally:
|
||||
scc.reset_card = original_reset
|
||||
|
||||
|
||||
def init_card_fast(sl, skip_card_init=False, wait=True):
|
||||
"""Replacement for pySim.app.init_card() that avoids redundant resets.
|
||||
|
||||
``wait`` performs the single disconnect/connect of this init (explicit
|
||||
equip passes True; startup already connects via wait_for_card). If probing
|
||||
leaves the card in a state the software reset cannot clear, retry once
|
||||
after a physical reset."""
|
||||
try:
|
||||
return _init_card_once(sl, skip_card_init, wait)
|
||||
except (SwMatchError, ProtocolError) as e:
|
||||
sys.stderr.write('FAST-INIT: %s; retrying after physical reset\n' % e)
|
||||
sl.reset_card()
|
||||
return _init_card_once(sl, skip_card_init, wait=False)
|
||||
|
||||
|
||||
def _init_card_once(sl, skip_card_init, wait):
|
||||
scc = SimCardCommands(transport=sl)
|
||||
if wait:
|
||||
sl.wait_for_card(3)
|
||||
if skip_card_init:
|
||||
return None, CardBase(scc)
|
||||
|
||||
generic_card = False
|
||||
card = card_detect(scc)
|
||||
if card is None:
|
||||
card = SimCardBase(scc)
|
||||
generic_card = True
|
||||
|
||||
profile = pick_profile_no_reset(scc)
|
||||
if profile is None:
|
||||
return None, card
|
||||
|
||||
if generic_card and isinstance(profile, CardProfileUICC):
|
||||
card._adm_chv_num = 0x0A
|
||||
|
||||
if isinstance(profile, CardProfileUICC):
|
||||
for app_cls in all_subclasses(CardApplication):
|
||||
if hasattr(app_cls, '_' + app_cls.__name__ + '__intermediate'):
|
||||
continue
|
||||
profile.add_application(app_cls())
|
||||
if generic_card:
|
||||
card = UiccCardBase(scc)
|
||||
|
||||
rs = FastRuntimeState(card, profile)
|
||||
|
||||
CardModel.apply_matching_models(scc, rs)
|
||||
|
||||
sl.set_sw_interpreter(rs)
|
||||
|
||||
isd_r = rs.mf.applications.get(pySim.euicc.AID_ISD_R.lower(), None)
|
||||
if isd_r:
|
||||
rs.lchan[0].select_file(isd_r)
|
||||
try:
|
||||
rs.identity['EID'] = pySim.euicc.CardApplicationISDR.get_eid(scc)
|
||||
except SwMatchError:
|
||||
pass
|
||||
finally:
|
||||
rs.soft_reset()
|
||||
|
||||
return rs, card
|
||||
|
||||
|
||||
def do_equip_fast(app):
|
||||
"""Explicit equip: one real reconnect (wait_for_card) then reset-free init.
|
||||
PysimApp.equip() unregisters the old command sets itself after the new init
|
||||
succeeds, so a failed init leaves the previous card state intact."""
|
||||
rs, card = init_card_fast(app.sl, wait=True)
|
||||
app.equip(card, rs)
|
||||
|
||||
|
||||
def do_reset_fast(app):
|
||||
"""Explicit reset: always a physical card reset."""
|
||||
if app.rs is None:
|
||||
app.card._scc.reset_card()
|
||||
atr = app.card._scc.get_atr()
|
||||
else:
|
||||
atr = app.rs.hard_reset(app)
|
||||
app.poutput('Card ATR: %s' % atr)
|
||||
|
||||
|
||||
def install(app):
|
||||
"""Route the pySim-shell equip/reset commands through the fast paths."""
|
||||
def _do_equip(statement):
|
||||
_tlog('do_equip_fast: start')
|
||||
do_equip_fast(app)
|
||||
_tlog('do_equip_fast: done')
|
||||
|
||||
def _do_reset(statement):
|
||||
_tlog('do_reset_fast: start')
|
||||
do_reset_fast(app)
|
||||
_tlog('do_reset_fast: done')
|
||||
|
||||
app.do_equip = _do_equip
|
||||
app.do_reset = _do_reset
|
||||
@@ -0,0 +1,483 @@
|
||||
"""HTTP OTA (SCP81 / GP RAM over HTTP) emulation.
|
||||
|
||||
Phase A: terminal-side BIP emulation (OPEN/SEND/RECEIVE/CLOSE CHANNEL) plus a
|
||||
raw TCP capture listener. In the default redirect mode the card's BIP channel
|
||||
is always redirected to the locally configured target (the future PSK TLS
|
||||
platform) and the address the card requested is only logged; in passthru mode
|
||||
the channel dials the destination the card requests in OPEN CHANNEL (TCP,
|
||||
UICC in client mode, remote connection).
|
||||
|
||||
Reference behavior (TS 102 223 8.52-8.56, GP v2.2 Amendment B) is taken from
|
||||
the captured real-terminal traces in samples/HTTP_OTA/traces:
|
||||
OPEN CHANNEL TR: result, Channel status (38), Bearer description (35), Buffer size (39)
|
||||
SEND DATA TR: result, Channel data length (37)
|
||||
RECEIVE DATA TR: result, Channel data (36), Channel data length (37)
|
||||
CLOSE CHANNEL TR: result
|
||||
"""
|
||||
|
||||
import socket
|
||||
import threading
|
||||
import time
|
||||
|
||||
MAX_LOG = 1000
|
||||
|
||||
|
||||
def ber_len_read(data, off):
|
||||
"""Read a BER-TLV length at data[off]; returns (length, next_offset)."""
|
||||
if off >= len(data):
|
||||
return 0, off
|
||||
b = data[off]
|
||||
if b < 0x80:
|
||||
return b, off + 1
|
||||
n = b & 0x7F
|
||||
if n == 0 or off + 1 + n > len(data):
|
||||
return 0, len(data)
|
||||
return int.from_bytes(data[off + 1:off + 1 + n], 'big'), off + 1 + n
|
||||
|
||||
|
||||
def proactive_tlvs(raw):
|
||||
"""Top-level TLV map {tag: value} of a D0 proactive command."""
|
||||
out = {}
|
||||
if not raw or raw[0] != 0xD0:
|
||||
return out
|
||||
ln, off = ber_len_read(raw, 1)
|
||||
end = min(len(raw), off + ln)
|
||||
while off + 1 < end:
|
||||
tag = raw[off]
|
||||
tlen, off2 = ber_len_read(raw, off + 1)
|
||||
val = raw[off2:off2 + tlen]
|
||||
off = off2 + tlen
|
||||
out.setdefault(tag, val)
|
||||
return out
|
||||
|
||||
|
||||
def parse_other_address(value):
|
||||
"""Decode an 'Other address' TLV (21=IPv4, 57=IPv6, F0=FQDN)."""
|
||||
if not value:
|
||||
return None
|
||||
t = value[0]
|
||||
if t == 0x21 and len(value) >= 5:
|
||||
return '.'.join(str(b) for b in value[1:5])
|
||||
if t == 0x57 and len(value) >= 17:
|
||||
return ':'.join('%x' % int.from_bytes(value[i:i + 2], 'big') for i in range(1, 17, 2))
|
||||
if t == 0xF0:
|
||||
return value[1:].decode('ascii', 'replace')
|
||||
return None
|
||||
|
||||
|
||||
def parse_transport_level(value):
|
||||
"""Decode an UICC/terminal interface transport level TLV -> (proto, port)."""
|
||||
if not value or len(value) < 3:
|
||||
return None, None
|
||||
return value[0], int.from_bytes(value[1:3], 'big')
|
||||
|
||||
|
||||
TAG_BEARER = 0x35
|
||||
TAG_CHANNEL_DATA = 0x36
|
||||
TAG_CHANNEL_DATA_LENGTH = 0x37
|
||||
TAG_CHANNEL_STATUS = 0x38
|
||||
TAG_BUFFER_SIZE = 0x39
|
||||
TAG_TRANSPORT_LEVEL = 0x3C
|
||||
TAG_OTHER_ADDRESS = 0x3E
|
||||
TAG_NAA = 0x47
|
||||
|
||||
|
||||
class BipChannel:
|
||||
def __init__(self, channel_id, sock, requested, target, buffer_size):
|
||||
self.id = channel_id
|
||||
self.sock = sock
|
||||
self.requested = requested
|
||||
self.target = target
|
||||
self.buffer_size = buffer_size or 512
|
||||
self.rx = bytearray()
|
||||
self.bytes_in = 0
|
||||
self.bytes_out = 0
|
||||
self.opened_at = time.time()
|
||||
self.peer_closed = False
|
||||
self.closed_reported = False
|
||||
self.notified_len = 0
|
||||
self.last_notify = 0.0
|
||||
|
||||
def pump(self, timeout=0.05):
|
||||
"""Move whatever the network has into the local buffer. Returns bytes moved."""
|
||||
if self.peer_closed:
|
||||
return 0
|
||||
moved = 0
|
||||
self.sock.settimeout(timeout)
|
||||
try:
|
||||
while True:
|
||||
chunk = self.sock.recv(self.buffer_size)
|
||||
if not chunk:
|
||||
self.peer_closed = True
|
||||
break
|
||||
self.rx.extend(chunk)
|
||||
self.bytes_in += len(chunk)
|
||||
moved += len(chunk)
|
||||
if len(chunk) < self.buffer_size:
|
||||
break
|
||||
except (socket.timeout, BlockingIOError):
|
||||
pass
|
||||
except OSError:
|
||||
self.peer_closed = True
|
||||
return moved
|
||||
|
||||
def send(self, data):
|
||||
self.sock.sendall(data)
|
||||
self.bytes_out += len(data)
|
||||
|
||||
def take(self, maxlen):
|
||||
self.pump()
|
||||
n = min(maxlen, len(self.rx), self.buffer_size)
|
||||
out = bytes(self.rx[:n])
|
||||
del self.rx[:n]
|
||||
return out
|
||||
|
||||
def available(self):
|
||||
self.pump()
|
||||
return len(self.rx)
|
||||
|
||||
def send_capacity(self):
|
||||
free = self.buffer_size - len(self.rx)
|
||||
return 0xFF if free > 0xFF else max(0, free)
|
||||
|
||||
def close(self):
|
||||
try:
|
||||
self.sock.shutdown(socket.SHUT_RDWR)
|
||||
except OSError:
|
||||
pass
|
||||
try:
|
||||
self.sock.close()
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
class BipTerminal:
|
||||
"""Terminal (device) side of BIP: channels to the configured target."""
|
||||
|
||||
def __init__(self):
|
||||
self.enabled = False
|
||||
self.mode = 'redirect'
|
||||
self.target = None
|
||||
self.channels = {}
|
||||
self.next_id = 1
|
||||
self.entries = []
|
||||
self.seq = 0
|
||||
self.lock = threading.Lock()
|
||||
self.pending_events = []
|
||||
self.on_data = None
|
||||
self._monitor = None
|
||||
|
||||
def log(self, kind, **fields):
|
||||
with self.lock:
|
||||
self.seq += 1
|
||||
entry = {'seq': self.seq, 't': time.time(), 'kind': kind}
|
||||
entry.update(fields)
|
||||
self.entries.append(entry)
|
||||
if len(self.entries) > MAX_LOG:
|
||||
del self.entries[:len(self.entries) - MAX_LOG]
|
||||
return entry
|
||||
|
||||
def _monitor_loop(self):
|
||||
"""Watch channels for incoming bytes and ask the card to fetch them.
|
||||
|
||||
The card only learns about server data through the Data available
|
||||
event (TS 102 223 7.5.10), so the socket must be pumped even while
|
||||
the card is idle."""
|
||||
while True:
|
||||
time.sleep(0.25)
|
||||
with self.lock:
|
||||
channels = list(self.channels.values())
|
||||
for ch in channels:
|
||||
try:
|
||||
ch.pump()
|
||||
except OSError:
|
||||
ch.peer_closed = True
|
||||
if ch.peer_closed and not ch.closed_reported and not ch.rx:
|
||||
# Report a dropped link (TS 102 223 7.5.11) only once the
|
||||
# buffered server data has been fetched: signalling the
|
||||
# drop while bytes are still waiting makes the card abort
|
||||
# the fetch and end the session prematurely.
|
||||
ch.closed_reported = True
|
||||
self.log('peer-close', channel=ch.id)
|
||||
self._queue_link_status(ch.id)
|
||||
if (self.on_data and ch.rx and not ch.peer_closed
|
||||
and (len(ch.rx) > ch.notified_len
|
||||
or time.time() - ch.last_notify > 2.0)):
|
||||
# Re-notify while data stays unfetched: the live card
|
||||
# sometimes needs the Data available event again to drain
|
||||
# a partially received TLS record.
|
||||
if self.on_data(ch):
|
||||
ch.notified_len = len(ch.rx)
|
||||
ch.last_notify = time.time()
|
||||
|
||||
def _start_monitor(self):
|
||||
if self._monitor is None or not self._monitor.is_alive():
|
||||
self._monitor = threading.Thread(target=self._monitor_loop,
|
||||
name='bip-monitor', daemon=True)
|
||||
self._monitor.start()
|
||||
|
||||
def enable(self, host=None, port=None, mode='redirect'):
|
||||
"""Enable the BIP terminal.
|
||||
|
||||
'redirect' (default) pins one target: every channel goes there whatever
|
||||
address the card requests. 'passthru' has no target at all: every
|
||||
channel dials the destination the card requested in OPEN CHANNEL."""
|
||||
self.mode = mode if mode in ('redirect', 'passthru') else 'redirect'
|
||||
self.target = (host, int(port)) if host and port not in (None, '') else None
|
||||
self.enabled = True
|
||||
self.log('enabled', mode=self.mode,
|
||||
target='%s:%d' % self.target if self.target else None)
|
||||
self._start_monitor()
|
||||
|
||||
def disable(self):
|
||||
self.enabled = False
|
||||
self.log('disabled')
|
||||
self.close_all(link_lost=True)
|
||||
self.target = None
|
||||
|
||||
def close_all(self, link_lost=False):
|
||||
for ch in list(self.channels.values()):
|
||||
self._close_channel(ch, link_lost=link_lost)
|
||||
|
||||
def _close_channel(self, ch, link_lost=False):
|
||||
ch.close()
|
||||
if self.channels.get(ch.id) is ch:
|
||||
del self.channels[ch.id]
|
||||
if link_lost:
|
||||
self._queue_link_status(ch.id)
|
||||
|
||||
def _queue_link_status(self, channel_id, status=None, info=0x05):
|
||||
"""Record a BIP link change that did not result from a proactive
|
||||
command (TS 102 223 7.5.11). The default is link not established +
|
||||
info 05 = link dropped; a successful background-mode OPEN CHANNEL
|
||||
reports link established instead. The server turns these into
|
||||
ENVELOPE (Channel status)."""
|
||||
with self.lock:
|
||||
if any(e['channel'] == channel_id for e in self.pending_events):
|
||||
return
|
||||
self.pending_events.append({
|
||||
'channel': channel_id,
|
||||
'status': channel_id & 0x07 if status is None else status,
|
||||
'info': info})
|
||||
|
||||
def take_pending_events(self):
|
||||
with self.lock:
|
||||
events, self.pending_events = self.pending_events, []
|
||||
return events
|
||||
|
||||
def _check_peer(self, ch):
|
||||
"""Notify once per channel when the peer closed the connection, after
|
||||
any buffered data has been fetched (see _monitor_loop)."""
|
||||
if ch.peer_closed and not ch.closed_reported and not ch.rx:
|
||||
ch.closed_reported = True
|
||||
self.log('peer-close', channel=ch.id)
|
||||
self._queue_link_status(ch.id)
|
||||
|
||||
def _alloc_id(self):
|
||||
for _ in range(7):
|
||||
cid = self.next_id
|
||||
self.next_id = 1 if cid >= 7 else cid + 1
|
||||
if cid not in self.channels:
|
||||
return cid
|
||||
return None
|
||||
|
||||
def open(self, requested_host, requested_port, buffer_size, proto=None):
|
||||
"""Open a channel.
|
||||
|
||||
Redirect modes connect to the pinned target; passthru dials the
|
||||
destination the card sent in OPEN CHANNEL (Other address + Transport
|
||||
level port). Returns (channel_id, error)."""
|
||||
if not self.enabled:
|
||||
return None, 'bip disabled'
|
||||
requested = '%s:%s' % (requested_host, requested_port)
|
||||
if self.mode == 'passthru':
|
||||
# Use the card's request as-is: TCP, UICC in client mode, remote
|
||||
# connection (TS 102 223 6.4.27.2 / 8.59). The specs define no
|
||||
# default port, so an incomplete or non-TCP request fails.
|
||||
host = (requested_host or '').strip()
|
||||
try:
|
||||
port = int(requested_port)
|
||||
except (TypeError, ValueError):
|
||||
port = 0
|
||||
if proto != 0x02:
|
||||
reason = 'card did not request TCP client transport (passthru)'
|
||||
elif not host or host == '-':
|
||||
reason = 'card did not request a destination address (passthru)'
|
||||
elif not 0 < port <= 0xFFFF:
|
||||
reason = 'card did not request a valid port (passthru)'
|
||||
else:
|
||||
reason = None
|
||||
if reason:
|
||||
self.log('open-fail', requested=requested, reason=reason)
|
||||
return None, reason
|
||||
target = (host, port)
|
||||
else:
|
||||
if not self.target:
|
||||
return None, 'bip disabled'
|
||||
target = self.target
|
||||
cid = self._alloc_id()
|
||||
if cid is None:
|
||||
self.log('open-fail', requested=requested, reason='no free channel')
|
||||
return None, 'no free channel'
|
||||
try:
|
||||
sock = socket.create_connection(target, timeout=2.0)
|
||||
except OSError as e:
|
||||
self.log('open-fail', requested=requested, target='%s:%d' % target, reason=str(e))
|
||||
return None, str(e)
|
||||
ch = BipChannel(cid, sock, requested, target, buffer_size)
|
||||
self.channels[cid] = ch
|
||||
self.log('open', channel=cid, requested=requested, target='%s:%d' % target,
|
||||
buffer_size=ch.buffer_size)
|
||||
return cid, None
|
||||
|
||||
def send(self, channel_id, data):
|
||||
ch = self.channels.get(channel_id)
|
||||
if not ch:
|
||||
return False
|
||||
try:
|
||||
ch.send(data)
|
||||
except OSError as e:
|
||||
self.log('send-fail', channel=channel_id, error=str(e))
|
||||
self._close_channel(ch, link_lost=True)
|
||||
return False
|
||||
self.log('send', channel=channel_id, bytes=len(data), hex=data.hex().upper()[:2000])
|
||||
return True
|
||||
|
||||
def receive(self, channel_id, maxlen):
|
||||
ch = self.channels.get(channel_id)
|
||||
if not ch:
|
||||
return None
|
||||
data = ch.take(maxlen)
|
||||
if data:
|
||||
self.log('receive', channel=channel_id, bytes=len(data), remaining=len(ch.rx),
|
||||
hex=data.hex().upper()[:2000])
|
||||
# The TR announced the remainder via the channel-data-length TLV,
|
||||
# but the live card still waits for a fresh Data available event
|
||||
# before fetching it - re-arm the notification for what is left.
|
||||
ch.notified_len = 0
|
||||
self._check_peer(ch)
|
||||
return data
|
||||
|
||||
def available(self, channel_id):
|
||||
ch = self.channels.get(channel_id)
|
||||
if not ch:
|
||||
return 0
|
||||
n = ch.available()
|
||||
self._check_peer(ch)
|
||||
return n
|
||||
|
||||
def send_capacity(self, channel_id):
|
||||
ch = self.channels.get(channel_id)
|
||||
if not ch:
|
||||
return 0
|
||||
n = ch.send_capacity()
|
||||
self._check_peer(ch)
|
||||
return n
|
||||
|
||||
def clear_log(self):
|
||||
with self.lock:
|
||||
self.entries = []
|
||||
|
||||
def close(self, channel_id):
|
||||
ch = self.channels.get(channel_id)
|
||||
if not ch:
|
||||
return False
|
||||
self.log('close', channel=channel_id, bytes_in=ch.bytes_in, bytes_out=ch.bytes_out)
|
||||
self._close_channel(ch)
|
||||
return True
|
||||
|
||||
def status(self):
|
||||
channels = []
|
||||
for ch in self.channels.values():
|
||||
channels.append({
|
||||
'id': ch.id,
|
||||
'requested': ch.requested,
|
||||
'target': '%s:%d' % ch.target,
|
||||
'buffer_size': ch.buffer_size,
|
||||
'bytes_in': ch.bytes_in,
|
||||
'bytes_out': ch.bytes_out,
|
||||
'pending': len(ch.rx),
|
||||
'peer_closed': ch.peer_closed,
|
||||
})
|
||||
return {
|
||||
'enabled': self.enabled,
|
||||
'mode': self.mode,
|
||||
'target': '%s:%d' % self.target if self.target else None,
|
||||
'channels': channels,
|
||||
'seq': self.seq,
|
||||
}
|
||||
|
||||
def entries_after(self, after=0):
|
||||
with self.lock:
|
||||
return [e for e in self.entries if e['seq'] > after]
|
||||
|
||||
|
||||
class TcpDumpServer:
|
||||
"""Plain TCP listener that logs whatever it receives (ClientHello capture).
|
||||
|
||||
Used as the BIP redirect target until the PSK TLS platform is brought up.
|
||||
"""
|
||||
|
||||
def __init__(self, host, port, on_rx=None, on_log=None):
|
||||
self.on_rx = on_rx
|
||||
self.on_log = on_log
|
||||
self.stopped = False
|
||||
self.conns = []
|
||||
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
self.sock.bind((host, int(port)))
|
||||
self.sock.listen(4)
|
||||
self.host, self.port = self.sock.getsockname()[:2]
|
||||
self.thread = threading.Thread(target=self._accept_loop, daemon=True)
|
||||
self.thread.start()
|
||||
if self.on_log:
|
||||
self.on_log('listener-start', host=self.host, port=self.port)
|
||||
|
||||
def _accept_loop(self):
|
||||
while not self.stopped:
|
||||
try:
|
||||
self.sock.settimeout(0.2)
|
||||
conn, addr = self.sock.accept()
|
||||
except socket.timeout:
|
||||
continue
|
||||
except OSError:
|
||||
break
|
||||
self.conns.append(conn)
|
||||
if self.on_log:
|
||||
self.on_log('conn', peer='%s:%d' % addr[:2])
|
||||
threading.Thread(target=self._conn_loop, args=(conn, addr), daemon=True).start()
|
||||
|
||||
def _conn_loop(self, conn, addr):
|
||||
try:
|
||||
while not self.stopped:
|
||||
conn.settimeout(0.2)
|
||||
try:
|
||||
data = conn.recv(4096)
|
||||
except socket.timeout:
|
||||
continue
|
||||
except OSError:
|
||||
break
|
||||
if not data:
|
||||
break
|
||||
if self.on_rx:
|
||||
self.on_rx('%s:%d' % addr[:2], data)
|
||||
finally:
|
||||
try:
|
||||
conn.close()
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
def stop(self):
|
||||
self.stopped = True
|
||||
if self.on_log:
|
||||
self.on_log('listener-stop', host=self.host, port=self.port)
|
||||
try:
|
||||
self.sock.close()
|
||||
except OSError:
|
||||
pass
|
||||
for conn in self.conns:
|
||||
try:
|
||||
conn.close()
|
||||
except OSError:
|
||||
pass
|
||||
self.conns = []
|
||||
@@ -0,0 +1,447 @@
|
||||
"""Phase B: PSK TLS server and HTTP administration session for SCP81.
|
||||
|
||||
Implements the Remote Administration Server side of GP RAM over HTTP
|
||||
(GPC v2.2 Amendment B):
|
||||
|
||||
- TLS 1.2 with the PSK cipher suites of clause 4.3.2. The handshake and
|
||||
record layer are handled by the stdlib ``ssl`` module through OpenSSL's
|
||||
PSK callbacks (identity -> PSK), so no TLS code lives here.
|
||||
- The HTTP dialog of clause 4.4: parse the Security Domain's POST
|
||||
(``X-Admin-*`` headers, optional body with the previous response string)
|
||||
and answer with 200 + a command string, or 204 No Content to close the
|
||||
administration session.
|
||||
|
||||
The card talks TLS *through* the BIP channel: this server listens on the
|
||||
local redirect target and the BIP terminal proxies the card's SEND/RECEIVE
|
||||
DATA records to it.
|
||||
"""
|
||||
|
||||
import socket
|
||||
import ssl
|
||||
import threading
|
||||
import time
|
||||
|
||||
MAX_HEAD = 32 * 1024
|
||||
MAX_BODY = 1 * 1024 * 1024
|
||||
|
||||
# TLS_PSK_* suites from GPC v2.2 Amendment B Table 4-2 / RFC 4279/4785/5487.
|
||||
PSK_CIPHERS = ':'.join([
|
||||
'PSK-AES128-CBC-SHA256', # TLS_PSK_WITH_AES_128_CBC_SHA256 (0x00AE)
|
||||
'PSK-AES128-CBC-SHA', # TLS_PSK_WITH_AES_128_CBC_SHA (0x008C)
|
||||
'PSK-AES256-CBC-SHA', # TLS_PSK_WITH_AES_256_CBC_SHA (0x008D)
|
||||
'PSK-3DES-EDE-CBC-SHA', # TLS_PSK_WITH_3DES_EDE_CBC_SHA (0x008B)
|
||||
'PSK-NULL-SHA256', # TLS_PSK_WITH_NULL_SHA256 (0x00B0)
|
||||
'PSK-NULL-SHA', # TLS_PSK_WITH_NULL_SHA (0x002C)
|
||||
])
|
||||
|
||||
GP_PROTOCOL = 'globalplatform-remote-admin/1.0'
|
||||
GP_CT_COMMAND = 'application/vnd.globalplatform.card-content-mgt;version=1.0'
|
||||
GP_CT_RESPONSE = 'application/vnd.globalplatform.card-content-mgt-response;version=1.0'
|
||||
|
||||
# OpenSSL SSL_OP_NO_ENCRYPT_THEN_MAC (not exposed by the ssl module). The live
|
||||
# card offers the encrypt_then_mac extension but aborts the session with
|
||||
# SSLV3_ALERT_UNEXPECTED_MESSAGE as soon as the server echoes it, so keep the
|
||||
# extension out of the ServerHello (verified live 2026-09-15).
|
||||
OP_NO_ENCRYPT_THEN_MAC = 0x00080000
|
||||
|
||||
TLS_VERSIONS = {
|
||||
'1.0': ssl.TLSVersion.TLSv1,
|
||||
'1.1': ssl.TLSVersion.TLSv1_1,
|
||||
'1.2': ssl.TLSVersion.TLSv1_2,
|
||||
}
|
||||
|
||||
|
||||
def _norm_identity(identity):
|
||||
"""Normalize a PSK identity to the str OpenSSL reports (CPython hands it
|
||||
to the PSK callback as a str; bytes are decoded byte-exact)."""
|
||||
if identity is None:
|
||||
return None
|
||||
if isinstance(identity, (bytes, bytearray)):
|
||||
return bytes(identity).decode('latin-1')
|
||||
return str(identity)
|
||||
|
||||
|
||||
def parse_http_request(data):
|
||||
"""Parse an HTTP/1.1 request head (bytes up to CRLFCRLF) into
|
||||
(method, target, headers dict with lower-case names)."""
|
||||
head = data.split(b'\r\n\r\n', 1)[0]
|
||||
lines = head.split(b'\r\n')
|
||||
parts = lines[0].split(b' ')
|
||||
if len(parts) < 3:
|
||||
raise ValueError('malformed request line')
|
||||
method, target = parts[0].decode('latin-1'), parts[1].decode('latin-1')
|
||||
headers = {}
|
||||
for line in lines[1:]:
|
||||
name, _, value = line.partition(b':')
|
||||
headers[name.strip().decode('latin-1').lower()] = value.strip().decode('latin-1')
|
||||
return method, target, headers
|
||||
|
||||
|
||||
def decode_chunked(body):
|
||||
"""Decode a chunked transfer body (RFC 2616 3.6.1)."""
|
||||
out = bytearray()
|
||||
while body:
|
||||
line, _, rest = body.partition(b'\r\n')
|
||||
try:
|
||||
size = int(line.split(b';')[0], 16)
|
||||
except ValueError:
|
||||
raise ValueError('bad chunk size %r' % line[:16])
|
||||
if size == 0:
|
||||
break
|
||||
out.extend(rest[:size])
|
||||
body = rest[size + 2:]
|
||||
return bytes(out)
|
||||
|
||||
|
||||
def build_http_response(status, reason, headers, body=b'', chunked=False,
|
||||
compact=False, connection=None):
|
||||
"""Build an HTTP response. With chunked=True the body is framed as 100-byte
|
||||
chunks (like the reference admin server); with compact=True header names
|
||||
and values are separated by ':' without whitespace, which keeps the whole
|
||||
response inside one card-sized TLS record (<= 256 bytes ciphertext).
|
||||
connection ('close'/'keep-alive') declares the connection fate: without
|
||||
it an HTTP/1.1 client assumes the connection persists and tries to reuse
|
||||
it for the next POST instead of dialing a new one (live card 2026-09-15)."""
|
||||
lines = ['HTTP/1.1 %d %s' % (status, reason)]
|
||||
sep = ':' if compact else ': '
|
||||
for name, value in headers.items():
|
||||
lines.append('%s%s%s' % (name, sep, value))
|
||||
if connection:
|
||||
lines.append('Connection%s%s' % (sep, connection))
|
||||
has_te = 'transfer-encoding' in [k.lower() for k in headers]
|
||||
if body and (chunked or has_te):
|
||||
if not has_te:
|
||||
lines.append('Transfer-Encoding: chunked')
|
||||
elif body and 'content-length' not in [k.lower() for k in headers]:
|
||||
lines.append('Content-Length%s%d' % (sep, len(body)))
|
||||
head = ('\r\n'.join(lines) + '\r\n\r\n').encode('latin-1')
|
||||
if not body:
|
||||
return head
|
||||
if not chunked:
|
||||
return head + body
|
||||
out = bytearray(head)
|
||||
for i in range(0, len(body), 100):
|
||||
piece = body[i:i + 100]
|
||||
out += ('%X\r\n' % len(piece)).encode('latin-1') + piece + b'\r\n'
|
||||
out += b'0\r\n\r\n'
|
||||
return bytes(out)
|
||||
|
||||
|
||||
class PskTlsServer:
|
||||
"""PSK TLS listener speaking the GP remote administration HTTP dialog."""
|
||||
|
||||
def __init__(self, host, port, psk=None, identity=None, on_log=None,
|
||||
responder=None, timeout=10.0, chunked=False, chunk_size=0,
|
||||
compact_headers=False, tls_version='auto',
|
||||
cipher=None, keylog=None,
|
||||
conn_header=None, half_close=False, answer_delay=0.0,
|
||||
psk_map=None):
|
||||
# PSK lookup table: identity -> key. With an explicit psk_map a
|
||||
# handshake is accepted only for a listed identity; the legacy
|
||||
# single-key form (psk + optional identity pin, pin None = accept any
|
||||
# identity) remains for scripts and tests.
|
||||
self.wildcard_psk = None
|
||||
self.psk_map = {}
|
||||
if psk_map is not None:
|
||||
self.psk_map = {_norm_identity(k): bytes(v)
|
||||
for k, v in dict(psk_map).items() if v}
|
||||
elif psk is not None:
|
||||
pin = _norm_identity(identity)
|
||||
if pin is None:
|
||||
self.wildcard_psk = psk
|
||||
else:
|
||||
self.psk_map = {pin: bytes(psk)}
|
||||
self.psk = psk
|
||||
self.identity = _norm_identity(identity)
|
||||
self.on_log = on_log
|
||||
self.responder = responder or self._default_responder
|
||||
self.timeout = timeout
|
||||
self.chunked = chunked
|
||||
# chunk_size 0 = one record for the whole response
|
||||
self.chunk_size = int(chunk_size)
|
||||
self.compact_headers = compact_headers
|
||||
# TLS is permissive by default: 'auto' accepts TLS 1.0-1.2 and lets
|
||||
# OpenSSL pick the highest the card offers. The '1.0'/'1.1'/'1.2'
|
||||
# pins are debugging aids for a card that offers 1.2 but mishandles
|
||||
# it; no setting is needed for normal use.
|
||||
self.tls_version = (tls_version if tls_version == 'auto'
|
||||
or tls_version in TLS_VERSIONS else 'auto')
|
||||
# Pin one cipher suite (e.g. PSK-AES128-CBC-SHA) if the card's SD only
|
||||
# maps a specific suite to a usable SCP81 security level.
|
||||
self.cipher = cipher or None
|
||||
# Debug aid: write the TLS traffic secrets to this file
|
||||
# (SSLKEYLOGFILE format), so captures of the PSK dialog can be
|
||||
# decrypted (tshark etc). Contains key material - use a temp path.
|
||||
self.keylog = keylog or None
|
||||
# Connection header value: None/'none' = omit the header (implicit
|
||||
# HTTP/1.1 keep-alive); 'keep-alive' adds it explicitly. The server
|
||||
# never closes mid-session - only the 204 ends the dialog.
|
||||
self.conn_header = conn_header or None
|
||||
# TLS half-close after a script body. NOTE (live 2026-09-16):
|
||||
# CPython's SSLSocket.unwrap() poisons the session when the peer does
|
||||
# not answer with its own close_notify in time, so this cannot be
|
||||
# implemented with the stdlib ssl module; the flag is kept for the
|
||||
# option surface and for cards that answer promptly (the exception
|
||||
# path leaves the session unusable, so it is off by default).
|
||||
self.half_close = half_close
|
||||
# Wait before answering a request (cards may need their BIP SEND DATA
|
||||
# conversation to settle before they accept the response; 0 = answer
|
||||
# immediately).
|
||||
self.answer_delay = float(answer_delay or 0)
|
||||
self.identity_seen = None
|
||||
self.identity_matched = None
|
||||
self.version_seen = None
|
||||
self.cipher_seen = None
|
||||
self.stopped = False
|
||||
self.conns = []
|
||||
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
# A quick Stop -> Start can race the previous listener's close (the
|
||||
# port stays busy for a moment); retry before giving up.
|
||||
last_error = None
|
||||
for _ in range(10):
|
||||
try:
|
||||
self.sock.bind((host, int(port)))
|
||||
last_error = None
|
||||
break
|
||||
except OSError as e:
|
||||
last_error = e
|
||||
time.sleep(0.3)
|
||||
if last_error is not None:
|
||||
self.sock.close()
|
||||
raise last_error
|
||||
self.sock.listen(4)
|
||||
self.host, self.port = self.sock.getsockname()[:2]
|
||||
self.ctx = self._make_context()
|
||||
if self.keylog:
|
||||
try:
|
||||
self.ctx.keylog_filename = self.keylog
|
||||
except (AttributeError, OSError):
|
||||
self.keylog = None
|
||||
self.thread = threading.Thread(target=self._accept_loop, daemon=True)
|
||||
self.thread.start()
|
||||
self.log('tls-listener-start', host=self.host, port=self.port)
|
||||
|
||||
def log(self, kind, **fields):
|
||||
if self.on_log:
|
||||
try:
|
||||
self.on_log(kind, **fields)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _make_context(self):
|
||||
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
|
||||
if self.tls_version == 'auto':
|
||||
# Accept everything the cards speak; OpenSSL negotiates the
|
||||
# highest common version.
|
||||
ctx.minimum_version = ssl.TLSVersion.TLSv1
|
||||
ctx.maximum_version = ssl.TLSVersion.TLSv1_2
|
||||
else:
|
||||
ver = TLS_VERSIONS[self.tls_version]
|
||||
ctx.minimum_version = ver
|
||||
ctx.maximum_version = ver
|
||||
ciphers = self.cipher or PSK_CIPHERS
|
||||
if self.tls_version in ('auto', '1.0', '1.1'):
|
||||
# OpenSSL 3.x disables the legacy protocol versions by default.
|
||||
ciphers += ':@SECLEVEL=0'
|
||||
ctx.set_ciphers(ciphers)
|
||||
# Prefer our (AES-first) order over the card's NULL-suite-first list.
|
||||
ctx.options |= ssl.OP_CIPHER_SERVER_PREFERENCE
|
||||
ctx.options |= OP_NO_ENCRYPT_THEN_MAC
|
||||
# No TLS session resumption: the live card aborts with
|
||||
# SSLV3_ALERT_UNEXPECTED_MESSAGE on the post-handshake
|
||||
# NewSessionTicket record (verified live 2026-09-15).
|
||||
ctx.options |= ssl.OP_NO_TICKET
|
||||
ctx.set_psk_server_callback(self._psk_cb)
|
||||
return ctx
|
||||
|
||||
def _psk_cb(self, identity):
|
||||
"""OpenSSL asks for the key of the identity the client sent.
|
||||
|
||||
The identity is looked up in the configured table (identity -> key);
|
||||
without a match the handshake fails on the Finished MAC check with a
|
||||
dummy key, and the attempt is logged as 'tls-psk-unknown'."""
|
||||
ident = _norm_identity(identity)
|
||||
self.identity_seen = ident
|
||||
key = self.psk_map.get(ident) if ident is not None else None
|
||||
if key is None:
|
||||
# Legacy single-key mode: no identity pin accepts any identity.
|
||||
key = self.wildcard_psk
|
||||
self.identity_matched = key is not None
|
||||
if key is None:
|
||||
self.log('tls-psk-unknown', identity=ident)
|
||||
return b'\x00' * 16
|
||||
return key
|
||||
|
||||
@property
|
||||
def psk_identities(self):
|
||||
"""Identities the listener looks up (keys are never exposed)."""
|
||||
return sorted(self.psk_map)
|
||||
|
||||
def set_psk_map(self, psk_map):
|
||||
"""Replace the identity -> key table of a running listener."""
|
||||
self.psk_map = {_norm_identity(k): bytes(v)
|
||||
for k, v in dict(psk_map).items() if v}
|
||||
self.wildcard_psk = None
|
||||
return self.psk_identities
|
||||
|
||||
@staticmethod
|
||||
def _default_responder(method, target, headers, body):
|
||||
"""No script configured: close the administration session (4.4.2)."""
|
||||
return 204, {'X-Admin-Protocol': GP_PROTOCOL}, b''
|
||||
|
||||
def _accept_loop(self):
|
||||
while not self.stopped:
|
||||
try:
|
||||
self.sock.settimeout(0.2)
|
||||
conn, addr = self.sock.accept()
|
||||
except socket.timeout:
|
||||
continue
|
||||
except OSError:
|
||||
break
|
||||
self.conns.append(conn)
|
||||
peer = '%s:%d' % addr[:2]
|
||||
threading.Thread(target=self._conn_loop, args=(conn, peer),
|
||||
daemon=True).start()
|
||||
|
||||
def _read_request(self, tls):
|
||||
buf = b''
|
||||
while b'\r\n\r\n' not in buf:
|
||||
chunk = tls.recv(4096)
|
||||
if not chunk:
|
||||
return None
|
||||
buf += chunk
|
||||
if len(buf) > MAX_HEAD:
|
||||
raise ValueError('request head too large')
|
||||
head, _, rest = buf.partition(b'\r\n\r\n')
|
||||
method, target, headers = parse_http_request(head + b'\r\n\r\n')
|
||||
body = rest
|
||||
if 'content-length' in headers:
|
||||
want = int(headers['content-length'])
|
||||
while len(body) < want:
|
||||
chunk = tls.recv(4096)
|
||||
if not chunk:
|
||||
break
|
||||
body += chunk
|
||||
body = body[:want]
|
||||
elif headers.get('transfer-encoding', '').lower() == 'chunked':
|
||||
while not body.endswith(b'0\r\n\r\n'):
|
||||
chunk = tls.recv(4096)
|
||||
if not chunk:
|
||||
break
|
||||
body += chunk
|
||||
body = decode_chunked(body)
|
||||
return method, target, headers, body
|
||||
|
||||
def _conn_loop(self, conn, peer):
|
||||
tls = None
|
||||
handshake_done = False
|
||||
try:
|
||||
tls = self.ctx.wrap_socket(conn, server_side=True)
|
||||
handshake_done = True
|
||||
self.version_seen = tls.version()
|
||||
self.cipher_seen = (tls.cipher() or (None,))[0]
|
||||
self.log('tls-handshake', peer=peer, cipher=self.cipher_seen,
|
||||
version=self.version_seen, identity=self.identity_seen,
|
||||
psk_match=self.identity_matched)
|
||||
while not self.stopped:
|
||||
req = self._read_request(tls)
|
||||
if req is None:
|
||||
break
|
||||
method, target, headers, body = req
|
||||
if self.answer_delay > 0:
|
||||
time.sleep(self.answer_delay)
|
||||
self.log('tls-request', peer=peer, method=method, uri=target,
|
||||
headers=headers,
|
||||
agent=headers.get('x-admin-from'),
|
||||
protocol=headers.get('x-admin-protocol'),
|
||||
script_status=headers.get('x-admin-script-status'),
|
||||
resume=headers.get('x-admin-resume'),
|
||||
content_type=headers.get('content-type'),
|
||||
bytes=len(body), body_hex=body.hex().upper()[:2000] or None)
|
||||
status, resp_headers, resp_body = self.responder(
|
||||
method, target, headers, body)
|
||||
reason = {200: 'OK', 204: 'No Content'}.get(status, 'Status')
|
||||
conn_hdr = None if self.conn_header in (None, 'none') else self.conn_header
|
||||
response = build_http_response(
|
||||
status, reason, resp_headers, resp_body,
|
||||
chunked=self.chunked, compact=self.compact_headers,
|
||||
connection=conn_hdr)
|
||||
# The card's HTTP client reads its response record-by-record:
|
||||
# the whole response must arrive in ONE TLS record (chunk_size
|
||||
# 0), otherwise a split head stalls it and a head-only record
|
||||
# followed by the body draws an unexpected_message alert. When
|
||||
# a chunk_size is given, the head goes in one record and the
|
||||
# body in pieces of that size.
|
||||
if self.chunk_size <= 0:
|
||||
tls.sendall(response)
|
||||
else:
|
||||
head, sep, rest = response.partition(b'\r\n\r\n')
|
||||
tls.sendall(head + sep if sep else head)
|
||||
for off in range(0, len(rest), self.chunk_size):
|
||||
tls.sendall(rest[off:off + self.chunk_size])
|
||||
self.log('tls-response', peer=peer, status=status,
|
||||
bytes=len(resp_body), chunked=self.chunked,
|
||||
response_hex=response.hex().upper()[:600],
|
||||
body_hex=resp_body.hex().upper()[:2000] or None)
|
||||
# Only the end of the dialog closes the connection: 204 (or
|
||||
# an empty body) ends the session; every other response
|
||||
# leaves the TLS connection open for the card's next POST.
|
||||
# Reusing it - or dialing a fresh one - is the card's call
|
||||
# (GP Am. B 4.3.1: the SD manages connection establishment).
|
||||
if status == 204 or not resp_body:
|
||||
# Clean TLS shutdown with the response still in the BIP
|
||||
# buffer: the card fetches the 204 and the close_notify
|
||||
# together, then the FIN. A bare close here makes the
|
||||
# card abort the session with a fatal alert.
|
||||
plain = None
|
||||
try:
|
||||
tls.settimeout(2.0)
|
||||
plain = tls.unwrap()
|
||||
tls = None
|
||||
except Exception:
|
||||
plain = None
|
||||
if plain is not None:
|
||||
try:
|
||||
plain.close()
|
||||
except OSError:
|
||||
pass
|
||||
break
|
||||
except ssl.SSLError as e:
|
||||
if handshake_done:
|
||||
self.log('tls-error', peer=peer, error=str(e))
|
||||
else:
|
||||
# No shared cipher / unsupported protocol version / card
|
||||
# alert: keep the handshake reason distinguishable from
|
||||
# post-handshake record errors.
|
||||
self.log('tls-handshake-failed', peer=peer, error=str(e))
|
||||
except (OSError, ValueError) as e:
|
||||
self.log('tls-error', peer=peer, error=str(e))
|
||||
finally:
|
||||
if tls is not None:
|
||||
try:
|
||||
tls.close()
|
||||
except OSError:
|
||||
pass
|
||||
else:
|
||||
try:
|
||||
conn.close()
|
||||
except OSError:
|
||||
pass
|
||||
self.log('tls-close', peer=peer)
|
||||
if conn in self.conns:
|
||||
self.conns.remove(conn)
|
||||
|
||||
def stop(self):
|
||||
self.stopped = True
|
||||
self.log('tls-listener-stop', host=self.host, port=self.port)
|
||||
try:
|
||||
self.sock.close()
|
||||
except OSError:
|
||||
pass
|
||||
for conn in list(self.conns):
|
||||
try:
|
||||
conn.close()
|
||||
except OSError:
|
||||
pass
|
||||
self.conns = []
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,22 @@
|
||||
import importlib.util
|
||||
import os
|
||||
import sys
|
||||
|
||||
|
||||
def load_pysim_app():
|
||||
import pySim
|
||||
pysim_dir = os.path.dirname(pySim.__file__)
|
||||
candidates = [
|
||||
os.path.join(os.path.dirname(pysim_dir), 'pySim-shell.py'),
|
||||
os.path.join(os.path.dirname(sys.executable), 'pySim-shell.py'),
|
||||
]
|
||||
for path in candidates:
|
||||
if os.path.exists(path):
|
||||
spec = importlib.util.spec_from_file_location("pySim_shell", path)
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(mod)
|
||||
return mod
|
||||
raise ImportError(
|
||||
"pySim-shell.py not found. Make sure pysim is installed "
|
||||
"(pip install pysim) and pySim-shell.py is on the PATH."
|
||||
)
|
||||
Reference in New Issue
Block a user