diff --git a/pysim_otaman_server/httpota.py b/pysim_otaman_server/httpota.py new file mode 100644 index 0000000..04e6c56 --- /dev/null +++ b/pysim_otaman_server/httpota.py @@ -0,0 +1,356 @@ +"""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. The card's BIP channel is always redirected to the +locally configured target (the future PSK TLS platform); the address the card +requested is only logged. + +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 + + 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.target = None + self.channels = {} + self.next_id = 1 + self.entries = [] + self.seq = 0 + self.lock = threading.Lock() + + 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 enable(self, host, port): + self.target = (host, int(port)) + self.enabled = True + self.log('enabled', target='%s:%d' % self.target) + + def disable(self): + self.enabled = False + self.log('disabled') + self.close_all() + self.target = None + + def close_all(self): + for ch in list(self.channels.values()): + self._close_channel(ch) + + def _close_channel(self, ch): + ch.close() + if self.channels.get(ch.id) is ch: + del self.channels[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): + """Open a channel to the redirect target. Returns (channel_id, error).""" + if not self.enabled or not self.target: + return None, 'bip disabled' + target = self.target + requested = '%s:%s' % (requested_host, requested_port) + 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) + 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]) + return data + + def available(self, channel_id): + ch = self.channels.get(channel_id) + return ch.available() if ch else 0 + + def send_capacity(self, channel_id): + ch = self.channels.get(channel_id) + return ch.send_capacity() if ch else 0 + + 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, + '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 = [] diff --git a/pysim_otaman_server/server.py b/pysim_otaman_server/server.py index cd2c82c..effb61e 100644 --- a/pysim_otaman_server/server.py +++ b/pysim_otaman_server/server.py @@ -11,6 +11,7 @@ from http.server import HTTPServer, BaseHTTPRequestHandler from io import StringIO from pySim.transport import ApduTracer, ProactiveHandler from pySim.cards import UiccCardBase +from pysim_otaman_server import httpota from smartcard.CardMonitoring import CardMonitor, CardObserver import gsm0338 # registers 'gsm03.38' codec @@ -753,6 +754,8 @@ PROACTIVE_TYPE_NAMES = { 0x24: 'SELECT ITEM', 0x25: 'SET UP MENU', 0x26: 'PROVIDE LOCAL INFORMATION', 0x15: 'LAUNCH BROWSER', 0x70: 'ACTIVATE', + 0x40: 'OPEN CHANNEL', 0x41: 'CLOSE CHANNEL', + 0x42: 'RECEIVE DATA', 0x43: 'SEND DATA', 0x44: 'GET CHANNEL STATUS', } PLI_QUALIFIER_NAMES = { @@ -782,6 +785,9 @@ PLI_QUALIFIER_NAMES = { _PLI_DATA = {q: '' for q in PLI_QUALIFIER_NAMES} +_BIP = httpota.BipTerminal() +_SCP81_LISTENER = None + _POLL_ENABLED = False _POLL_INTERVAL = 30 _POLL_TIMER = None @@ -989,6 +995,8 @@ def _decode_cmd(cmd_type, raw, qualifier): if items: return [{'label': 'Items', 'value': ', '.join('%s. %s' % (it['id'], it['text']) for it in items)}] return [] + if cmd_type in (0x40, 0x42, 0x43): + return _decode_bip_cmd(cmd_type, raw) if cmd_type == 0x26 and qualifier is not None: name = PLI_QUALIFIER_NAMES.get(qualifier, 'Unknown') return [{'label': 'Qualifier', 'value': '%s (0x%02X)' % (name, qualifier)}] @@ -1058,6 +1066,143 @@ def _decode_tr(type_hex, qual_hex, tr_hex): return [{'label': 'Data', 'value': h}] +def _bip_channel_id(dev_dst): + return (dev_dst & 0x07) if 0x21 <= dev_dst <= 0x27 else None + + +def _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst, result=0x00, info=None, extra=b''): + """TERMINAL RESPONSE payload for a BIP command. + + Tags follow the captured real-terminal traces (comprehension TLVs 01/02/03) + and the result precedes the optional Channel status / data TLVs. + """ + tr = bytes([0x01, 0x03, cmd_num & 0xFF, cmd_type & 0xFF, (cmd_qual if cmd_qual is not None else 0) & 0xFF, + 0x02, 0x02, 0x82, 0x81]) + if info is None: + tr += bytes([0x03, 0x01, result & 0xFF]) + else: + tr += bytes([0x03, 0x02, result & 0xFF, info & 0xFF]) + return tr + extra + + +def _decode_bip_cmd(cmd_type, raw): + tlvs = httpota.proactive_tlvs(raw) + out = [] + if cmd_type == 0x40: + bearer = tlvs.get(httpota.TAG_BEARER, b'') + if bearer: + out.append({'label': 'Bearer', 'value': '0x%02X' % bearer[0]}) + bs = tlvs.get(httpota.TAG_BUFFER_SIZE, b'') + if len(bs) >= 2: + out.append({'label': 'Buffer size', 'value': str(int.from_bytes(bs[:2], 'big'))}) + naa = tlvs.get(httpota.TAG_NAA, b'') + if naa: + out.append({'label': 'APN', 'value': naa[1:].decode('ascii', 'replace')}) + addr = httpota.parse_other_address(tlvs.get(httpota.TAG_OTHER_ADDRESS, b'')) + if addr: + out.append({'label': 'Destination', 'value': addr}) + proto, port = httpota.parse_transport_level(tlvs.get(httpota.TAG_TRANSPORT_LEVEL, b'')) + if port is not None: + out.append({'label': 'Transport', 'value': '%s port %d' % ({0x02: 'TCP client'}.get(proto, 'proto 0x%02X' % (proto or 0)), port)}) + elif cmd_type == 0x42: + req = tlvs.get(httpota.TAG_CHANNEL_DATA_LENGTH, b'') + if req: + out.append({'label': 'Requested bytes', 'value': str(req[0])}) + elif cmd_type == 0x43: + data = tlvs.get(httpota.TAG_CHANNEL_DATA, b'') + out.append({'label': 'Data bytes', 'value': str(len(data))}) + if data: + out.append({'label': 'Data', 'value': data.hex()[:120]}) + return out + + +def _handle_bip_command(scc, cmd_num, cmd_type, cmd_qual, raw, dev_src, dev_dst): + """Handle a BIP proactive command. Returns TR payload bytes, or None for the generic path.""" + tlvs = httpota.proactive_tlvs(raw) + channel = _bip_channel_id(dev_dst) + if cmd_type == 0x40: + bs = tlvs.get(httpota.TAG_BUFFER_SIZE, b'\x02\x00') + buffer_size = int.from_bytes(bs[:2], 'big') if len(bs) >= 2 else 0x0200 + bearer = tlvs.get(httpota.TAG_BEARER, b'\x03') + extra = bytes([httpota.TAG_BEARER, len(bearer)]) + bearer + extra += bytes([httpota.TAG_BUFFER_SIZE, 0x02]) + buffer_size.to_bytes(2, 'big') + if not _BIP.enabled: + _BIP.log('open-unavailable', reason='BIP not enabled') + return _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst, 0x3A, 0x00, extra) + addr = httpota.parse_other_address(tlvs.get(httpota.TAG_OTHER_ADDRESS, b'')) + proto, port = httpota.parse_transport_level(tlvs.get(httpota.TAG_TRANSPORT_LEVEL, b'')) + if not addr or port is None: + _BIP.log('open-unavailable', reason='missing destination/transport', address=addr, port=port) + return _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst, 0x3A, 0x00, extra) + cid, err = _BIP.open(addr, port, buffer_size) + if cid is None: + return _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst, 0x3A, 0x00, extra) + status = bytes([httpota.TAG_CHANNEL_STATUS, 0x02, 0x80 | (cid & 0x07), 0x00]) + return _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst, 0x00, None, status + extra) + if cmd_type == 0x43: + data = tlvs.get(httpota.TAG_CHANNEL_DATA, b'') + if channel is None or not _BIP.send(channel, data): + return _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst, 0x3A, 0x00) + length = _BIP.send_capacity(channel) + return _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst, 0x00, None, + bytes([httpota.TAG_CHANNEL_DATA_LENGTH, 0x01, length & 0xFF])) + if cmd_type == 0x42: + req = tlvs.get(httpota.TAG_CHANNEL_DATA_LENGTH, b'\x00') + n = req[0] if req else 0 + if channel is None: + return _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst, 0x3A, 0x00) + data = _BIP.receive(channel, n) + if data is None: + return _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst, 0x3A, 0x00) + remaining = _BIP.available(channel) + extra = bytes([httpota.TAG_CHANNEL_DATA, len(data)]) + data if data else b'' + extra += bytes([httpota.TAG_CHANNEL_DATA_LENGTH, 0x01, 0xFF if remaining > 0xFF else remaining]) + return _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst, 0x00, None, extra) + if cmd_type == 0x41: + if channel is None or not _BIP.close(channel): + return _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst, 0x3A, 0x00) + return _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst, 0x00) + if cmd_type == 0x44: + extra = b'' + for cid in sorted(_BIP.channels): + extra += bytes([httpota.TAG_CHANNEL_STATUS, 0x02, 0x80 | (cid & 0x07), 0x00]) + return _bip_tr(cmd_num, cmd_type, cmd_qual, dev_src, dev_dst, 0x00, None, extra) + return None + + +def _scp81_listener_status(): + if not _SCP81_LISTENER: + return None + return {'mode': 'dump', 'host': _SCP81_LISTENER.host, 'port': _SCP81_LISTENER.port} + + +def _scp81_bip_control(body): + global _SCP81_LISTENER + body = body or {} + action = body.get('action', 'start') + if action == 'stop': + if _SCP81_LISTENER: + _SCP81_LISTENER.stop() + _SCP81_LISTENER = None + _BIP.disable() + return {'ok': True, 'bip': _BIP.status(), 'listener': None} + host = body.get('host') or '127.0.0.1' + port = int(body.get('port') or 8443) + mode = body.get('mode', 'dump') + if _SCP81_LISTENER: + _SCP81_LISTENER.stop() + _SCP81_LISTENER = None + _BIP.disable() + if mode != 'dump': + return {'ok': False, 'error': 'unsupported mode: %s' % mode} + _SCP81_LISTENER = httpota.TcpDumpServer( + host, port, + on_rx=lambda peer, data: _BIP.log('dump-rx', peer=peer, bytes=len(data), hex=data.hex().upper()[:2000]), + on_log=lambda kind, **fields: _BIP.log(kind, **fields)) + _BIP.enable(host, _SCP81_LISTENER.port) + return {'ok': True, 'bip': _BIP.status(), 'listener': _scp81_listener_status()} + + def _build_tr(scc, cmd_num, cmd_type, dev_src, dev_dst, cmd_qual): """Build the TERMINAL RESPONSE TLV payload for a fetched command (includes PLI dictionary data for PROVIDE LOCAL INFORMATION).""" @@ -1555,7 +1700,11 @@ def _handle_proactive_chain(scc, sw91, on_fetch=None): if on_fetch: action = on_fetch(raw, cmd_num, cmd_type, dev_src, dev_dst) if action != 'pause': - tr_tlv = _build_tr(scc, cmd_num, cmd_type, dev_src, dev_dst, cmd_qual) + tr_tlv = None + if raw and cmd_type in (0x40, 0x41, 0x42, 0x43, 0x44): + tr_tlv = _handle_bip_command(scc, cmd_num, cmd_type, cmd_qual, raw, dev_src, dev_dst) + if tr_tlv is None: + tr_tlv = _build_tr(scc, cmd_num, cmd_type, dev_src, dev_dst, cmd_qual) tr_rv = scc._tp.send_apdu('%s140000%02x%s' % (scc.cat_cla, len(tr_tlv), tr_tlv.hex())) sys.stderr.write('TR: cmd=%02x type=%02x -> %s %s\n' % (cmd_num, cmd_type, tr_rv[1], ('(%d bytes)' % len(tr_tlv)))) _record_tr(entry, tr_tlv, tr_rv[1]) @@ -1913,6 +2062,24 @@ class PysimHandler(BaseHTTPRequestHandler): 'pending_type': self.server.stk_pending['type'] if self.server.stk_pending else None} self._send_json(resp) self._log_resp(resp) + elif self.path == '/api/scp81/status': + self._log_req() + resp = {'bip': _BIP.status(), 'listener': _scp81_listener_status()} + self._send_json(resp) + self._log_resp(resp) + elif self.path == '/api/scp81/log' or self.path.startswith('/api/scp81/log?'): + self._log_req() + after = 0 + if '?' in self.path: + for kv in self.path.split('?', 1)[1].split('&'): + if kv.startswith('after='): + try: + after = int(kv[6:]) + except ValueError: + after = 0 + resp = {'seq': _BIP.seq, 'entries': _BIP.entries_after(after)[-200:]} + self._send_json(resp) + self._log_resp(resp) elif self.path.startswith('/api/'): self._send_json({'error': _err('not_found', lang)}, 404) self._log_resp({'error': _err('not_found', lang)}) @@ -2692,6 +2859,22 @@ class PysimHandler(BaseHTTPRequestHandler): err = {'success': False, 'error': str(e)} self._send_json(err, 500) self._log_resp(err) + elif self.path == '/api/scp81/bip': + body = self._read_body() + self._log_req(body) + try: + resp = _scp81_bip_control(body) + except Exception as e: + resp = {'ok': False, 'error': str(e)} + self._send_json(resp) + self._log_resp(resp) + elif self.path == '/api/scp81/log-clear': + body = self._read_body() + self._log_req(body) + _BIP.clear_log() + resp = {'ok': True, 'seq': _BIP.seq} + self._send_json(resp) + self._log_resp(resp) else: self._send_json({'error': _err('not_found', lang)}, 404) self._log_resp({'error': _err('not_found', lang)}) diff --git a/tests/test_httpota.py b/tests/test_httpota.py new file mode 100644 index 0000000..e0444d4 --- /dev/null +++ b/tests/test_httpota.py @@ -0,0 +1,234 @@ +#!/usr/bin/env python3 +"""Unit tests for the HTTP OTA (SCP81) BIP terminal emulation (Phase A). + +TR byte vectors come from the captured real-terminal traces in +samples/HTTP_OTA/traces (OPEN CHANNEL success/failure, SEND/RECEIVE/CLOSE). +No live card or live card data is used here. +""" + +import socket +import sys +import threading +import time +import unittest +from pathlib import Path + +PROJECTS = Path(__file__).resolve().parents[2] +PY_SIM = PROJECTS / 'pysim' +if str(PY_SIM) not in sys.path: + sys.path.insert(0, str(PY_SIM)) + +from pysim_otaman_server import httpota +import pysim_otaman_server.server as server + + +OPEN_LOCALHOST = bytes.fromhex( + 'd02b010301400102028182050035010339020200470b076d656761666f6e2e7275' + '3c03021f903e05217f000001') + +TR_OPEN_OK = '0103014001020282810301003802810035010339020200' +TR_OPEN_FAIL = '01030140010202828103023a0035010339020200' + + +class PeerServer(threading.Thread): + """Tiny TCP peer: accepts one connection, greets, records what it receives.""" + + def __init__(self, greeting=b''): + super().__init__(daemon=True) + self.sock = socket.socket() + self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + self.sock.bind(('127.0.0.1', 0)) + self.sock.listen(1) + self.port = self.sock.getsockname()[1] + self.greeting = greeting + self.received = b'' + self.conn = None + self.ready = threading.Event() + self.done = threading.Event() + + def run(self): + self.sock.settimeout(3) + try: + self.conn, _ = self.sock.accept() + except OSError: + return + self.ready.set() + if self.greeting: + self.conn.sendall(self.greeting) + self.conn.settimeout(2) + deadline = time.time() + 3 + try: + while time.time() < deadline: + try: + data = self.conn.recv(4096) + except socket.timeout: + break + if not data: + break + self.received += data + except OSError: + pass + self.done.set() + + def stop(self): + self.sock.close() + + +def open_cmd(host, port, buffer_size=512): + ip = bytes(int(x) for x in host.split('.')) + tlvs = (b'\x81\x03\x01\x40\x01' + b'\x82\x02\x81\x82' + b'\x35\x01\x03' + b'\x39\x02' + buffer_size.to_bytes(2, 'big') + + b'\x3c\x03\x02' + port.to_bytes(2, 'big') + + b'\x3e\x05\x21' + ip) + return b'\xd0' + bytes([len(tlvs)]) + tlvs + + +def channel_cmd(cmd_type, qualifier, data_tlvs=b''): + tlvs = (bytes([0x81, 0x03, 0x01, cmd_type, qualifier]) + + b'\x82\x02\x81\x21' + data_tlvs) + return b'\xd0' + bytes([len(tlvs)]) + tlvs + + +class TlvTest(unittest.TestCase): + def test_proactive_tlvs_open_channel(self): + tlvs = httpota.proactive_tlvs(OPEN_LOCALHOST) + self.assertEqual(tlvs[httpota.TAG_BEARER], b'\x03') + self.assertEqual(tlvs[httpota.TAG_BUFFER_SIZE], b'\x02\x00') + self.assertEqual(tlvs[httpota.TAG_NAA], b'\x07megafon.ru') + self.assertEqual(httpota.parse_transport_level(tlvs[httpota.TAG_TRANSPORT_LEVEL]), (0x02, 8080)) + self.assertEqual(httpota.parse_other_address(tlvs[httpota.TAG_OTHER_ADDRESS]), '127.0.0.1') + + +class TrVectorTest(unittest.TestCase): + def test_open_channel_success_vector(self): + extra = bytes([0x38, 0x02, 0x81, 0x00]) + bytes([0x35, 0x01, 0x03]) + bytes([0x39, 0x02, 0x02, 0x00]) + tr = server._bip_tr(1, 0x40, 0x01, 0x81, 0x82, 0x00, None, extra) + self.assertEqual(tr.hex(), TR_OPEN_OK) + + def test_open_channel_failure_vector(self): + extra = bytes([0x35, 0x01, 0x03]) + bytes([0x39, 0x02, 0x02, 0x00]) + tr = server._bip_tr(1, 0x40, 0x01, 0x81, 0x82, 0x3A, 0x00, extra) + self.assertEqual(tr.hex(), TR_OPEN_FAIL) + + def test_disabled_bip_fails_open_channel(self): + old = server._BIP + try: + server._BIP = httpota.BipTerminal() + tr = server._handle_bip_command(None, 1, 0x40, 0x01, OPEN_LOCALHOST, 0x81, 0x82) + self.assertEqual(tr.hex(), TR_OPEN_FAIL) + finally: + server._BIP = old + + +class BipTerminalTest(unittest.TestCase): + def test_redirect_and_roundtrip(self): + peer = PeerServer(greeting=b'SERVERHELLO') + peer.start() + bip = httpota.BipTerminal() + bip.enable('127.0.0.1', peer.port) + cid, err = bip.open('10.9.9.9', 1234, 512) + self.assertIsNone(err) + self.assertEqual(bip.channels[cid].requested, '10.9.9.9:1234') + self.assertEqual(bip.channels[cid].target, ('127.0.0.1', peer.port)) + self.assertTrue(bip.send(cid, b'CLIENTHELLO')) + data = b'' + for _ in range(20): + data = bip.receive(cid, 100) + if data: + break + time.sleep(0.05) + self.assertEqual(data, b'SERVERHELLO') + self.assertTrue(bip.close(cid)) + peer.done.wait(3) + self.assertEqual(peer.received, b'CLIENTHELLO') + kinds = [e['kind'] for e in bip.entries_after(0)] + self.assertIn('open', kinds) + self.assertIn('send', kinds) + self.assertIn('receive', kinds) + self.assertIn('close', kinds) + peer.stop() + + def test_disabled_terminal_refuses_open(self): + bip = httpota.BipTerminal() + cid, err = bip.open('127.0.0.1', 1, 512) + self.assertIsNone(cid) + self.assertIn('disabled', err) + + def test_dump_server_logs_received_bytes(self): + received = [] + dump = httpota.TcpDumpServer('127.0.0.1', 0, on_rx=lambda peer, data: received.append(data)) + c = socket.create_connection(('127.0.0.1', dump.port), timeout=2) + c.sendall(b'HELLOCARD') + deadline = time.time() + 2 + while time.time() < deadline and not received: + time.sleep(0.02) + c.close() + dump.stop() + self.assertEqual(b''.join(received), b'HELLOCARD') + + +def parse_tr(tr): + """Parse a BIP TERMINAL RESPONSE payload into {tag: value}.""" + out = {} + off = 0 + while off + 1 < len(tr): + tag, ln = tr[off], tr[off + 1] + out[tag] = tr[off + 2:off + 2 + ln] + off += 2 + ln + return out + + +class BipCommandFlowTest(unittest.TestCase): + def setUp(self): + self.peer = PeerServer(greeting=b'SERVERHELLO') + self.peer.start() + self.old = server._BIP + self.bip = httpota.BipTerminal() + self.bip.enable('127.0.0.1', self.peer.port) + server._BIP = self.bip + + def tearDown(self): + server._BIP = self.old + self.peer.stop() + + def test_open_send_receive_close_flow(self): + tr = server._handle_bip_command(None, 1, 0x40, 0x01, open_cmd('127.0.0.1', self.peer.port), 0x81, 0x82) + self.assertEqual(tr.hex(), TR_OPEN_OK) + + tr = server._handle_bip_command(None, 1, 0x43, 0x01, + channel_cmd(0x43, 0x01, bytes([0x36, 0x08]) + b'CLIENTHE'), + 0x81, 0x21) + tlvs = parse_tr(tr) + self.assertEqual(tlvs[0x01].hex(), '014301') + self.assertEqual(tlvs[0x02].hex(), '8281') + self.assertEqual(tlvs[0x03], b'\x00') + self.assertEqual(tlvs[0x37], b'\xff') + + data = b'' + for _ in range(20): + tr = server._handle_bip_command(None, 1, 0x42, 0x00, channel_cmd(0x42, 0x00, bytes([0x37, 0x01, 0x64])), 0x81, 0x21) + tlvs = parse_tr(tr) + if 0x36 in tlvs and tlvs[0x36]: + data += tlvs[0x36] + break + time.sleep(0.05) + self.assertEqual(data, b'SERVERHELLO') + self.assertEqual(tlvs[0x37], b'\x00') + + tr = server._handle_bip_command(None, 1, 0x41, 0x00, channel_cmd(0x41, 0x00), 0x81, 0x21) + self.assertEqual(tr.hex(), '010301410002028281030100') + self.peer.done.wait(3) + self.assertEqual(self.peer.received, b'CLIENTHE') + + def test_send_without_channel_fails(self): + tr = server._handle_bip_command(None, 1, 0x43, 0x01, + channel_cmd(0x43, 0x01, bytes([0x36, 0x01]) + b'X'), + 0x81, 0x21) + tlvs = parse_tr(tr) + self.assertEqual(tlvs[0x03].hex(), '3a00') + + +if __name__ == '__main__': + unittest.main()