#!/usr/bin/env python3 # sclink.py -- bring a caller into Skull Crash over SCLink, from any board that # runs a STDIO door (WWIV, Talisman, Synchronet, Mystic, ...) or a DOOR32 door # that hands the caller's socket to the door (EleBBS/Win32 and other Windows # boards that write DOOR32.SYS). # # Design: the game's Vault, "Remote BBS Door Design 2026-10". The board proves # itself with an HMAC over a one-time challenge, gets a single-use ticket, and # this script then connects the caller -- over TLS at level "tls" (the # default) -- and pipes the caller's bytes to the game and back. # # sclink.py [rows] [cols] STDIO door # sclink.py --door32 [rows] [cols] DOOR32 door # sclink.py --selftest # # STDIO: the caller is the door's stdin and stdout; the board has already done # the telnet work. # # DOOR32 (Python 3 only): DOOR32.SYS line 1 must be 2 (telnet), line 2 is the # caller's socket handle, inherited from the board, and line 5 is the user # record number, which becomes the SCLink user number. The socket is the # board's: it is never shut down here, and when the game is over only OUR # handle to it is closed (close(), never detach(): on Windows a handle left # open at exit gets the connection reset), so the caller returns to the board. # The bytes on it are still telnet-framed, so the game's 0xFF (a CP437 glyph) # is doubled on the way out and the client's IAC sequences are taken out on the # way in; negotiation is left to the board. The socket is used as found: never # set blocking or non-blocking (a board may have WSAEventSelect on it), always # select first. # # config.json holds the values the game's owner issued you: # # { "board": "DMB", # "secret": "", # "host": "game.example.org", # "auth_port": 4443, # "play_port": 4514, # "level": "tls" } # # Optional: "ca" (a CA file to trust instead of the system's), "verify": false # (last resort), "caller_ip" is taken from $SCLINK_CALLER if the board sets it. # # The secret is a password: keep the config private. Python 2.7.9+ or 3 for a # STDIO door; Python 3 for a DOOR32 door. import errno import hashlib import hmac import json import os import re import select import socket import sys import threading try: import http.client as httplib except ImportError: # Python 2 import httplib SCLINK_CLIENT = "generic-py/1" # One of SCLink.vectors.json: every optional field empty, sha256. VECTOR = { "secret": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", "fields": { "alg": "sha256", "board": "DMB", "user": "1234", "challenge": "00112233445566778899aabbccddeeff", "rows": "", "cols": "", "term": "", "caller": "", }, "mac": "f33b142e92a2f05c5dea0a5c8e47c1783d9c9a8a86af5172946298949d668a9d", } SIGNED = ["alg", "board", "user", "challenge", "rows", "cols", "term", "caller"] NOT_SET_UP = b"\r\nSkull Crash is not set up on this board yet. Tell your sysop.\r\n" # Python 2's socket errors are IOError, not OSError. GONE = (OSError, IOError, socket.error, ValueError) def signed_string(fields): """The lines, in order, joined with \\n, ending in END: never a trailing newline.""" parts = ["SCLINK1"] for key in SIGNED: value = fields.get(key) parts.append("" if value is None else str(value)) parts.append("END") return "\n".join(parts) def sclink_mac(secret, fields): """The key is the ASCII TEXT of the 64-hex secret, not the bytes it encodes.""" alg = fields.get("alg", "sha256") return hmac.new( secret.encode("ascii"), signed_string(fields).encode("ascii"), getattr(hashlib, alg) ).hexdigest() def self_test(): return sclink_mac(VECTOR["secret"], VECTOR["fields"]) == VECTOR["mac"] def sanitise(text): """What the caller may be shown: printable ASCII, at most 60 characters.""" return re.sub(r"[^\x20-\x7e]", "", "" if text is None else str(text))[:60] def parse_reply(body): """'OK ' -> (value, None); 'ERR ' -> (None, text).""" line = (body or "").splitlines()[0] if body else "" if line.startswith("OK "): return line[3:].strip(), None match = re.match(r"^ERR \S+ (.*)$", line) return None, sanitise(match.group(1) if match else "Skull Crash refused the connection.") def _optional(value, pattern): text = "" if value is None else str(value) return text if re.match(pattern, text) else "" def _dimension(value): try: n = int(value) except (TypeError, ValueError): return "" return str(n) if 10 <= n <= 255 else "" def ticket_fields(cfg, caller, challenge): level = cfg.get("level", "tls") return { "alg": cfg.get("alg", "sha256"), "board": str(cfg["board"]), "user": str(caller["number"]), "challenge": challenge, "rows": _dimension(caller.get("rows")), "cols": _dimension(caller.get("cols")), "term": _optional(str(caller.get("term") or "").lower(), r"^[a-z0-9-]{1,16}$"), # The caller's address only ever travels encrypted. "caller": "" if level == "plain" else _optional(caller.get("ip"), r"^[0-9a-fA-F:.]{2,45}$"), } def ticket_headers(cfg, fields): headers = { "X-SC-Board": fields["board"], "X-SC-User": fields["user"], "X-SC-Challenge": fields["challenge"], "X-SC-Alg": fields["alg"], "X-SC-Mac": sclink_mac(cfg["secret"], fields), "X-SC-Client": SCLINK_CLIENT, } for key in ("rows", "cols", "term", "caller"): if fields[key]: headers["X-SC-" + key.capitalize()] = fields[key] return headers def config_problem(cfg, levels=("tls", "tls-auth", "plain")): if not re.match(r"^[A-Z0-9]{2,6}$", str(cfg.get("board", ""))): return "board must be 2-6 of A-Z / 0-9" if not re.match(r"^[0-9a-f]{64}$", str(cfg.get("secret", ""))): return "secret must be the 64 hex characters issued" if not cfg.get("host"): return "host is required" if not cfg.get("auth_port") or not cfg.get("play_port"): return "auth_port and play_port are required" if cfg.get("level", "tls") not in levels: return "level must be one of " + ", ".join(levels) if cfg.get("alg", "sha256") not in ("sha256", "sha1"): return "alg must be sha256 or sha1" return None def tls_context(cfg): import ssl context = ssl.create_default_context(cafile=cfg.get("ca")) if cfg.get("verify") is False: context.check_hostname = False context.verify_mode = ssl.CERT_NONE return context def http_get(cfg, path, headers): secure = cfg.get("level", "tls") != "plain" timeout = cfg.get("timeout", 10) if secure: conn = httplib.HTTPSConnection( cfg["host"], int(cfg["auth_port"]), timeout=timeout, context=tls_context(cfg) ) else: conn = httplib.HTTPConnection(cfg["host"], int(cfg["auth_port"]), timeout=timeout) try: conn.request("GET", path, headers=headers) return conn.getresponse().read(1024).decode("latin-1") finally: conn.close() def fetch_ticket(cfg, caller): """Steps 1 and 2. Returns (ticket, None) or (None, text for the caller).""" challenge, error = parse_reply(http_get(cfg, "/sclink/v1/challenge?board=" + cfg["board"], {})) if error: return None, error if not re.match(r"^[0-9a-f]{32}$", challenge): return None, "Connection error. Tell your sysop." fields = ticket_fields(cfg, caller, challenge) ticket, error = parse_reply(http_get(cfg, "/sclink/v1/ticket", ticket_headers(cfg, fields))) if error: return None, error if not re.match(r"^[0-9a-f]{32}$", ticket): return None, "Connection error. Tell your sysop." return ticket, None # --------------------------------------------------------------------------- # The caller's side: a STDIO door, or the board's own socket (DOOR32) # --------------------------------------------------------------------------- class StdioCaller(object): """The door's stdin and stdout. The board has done the telnet work.""" def __init__(self, stdin_fd, out): self.stdin_fd = stdin_fd self.out = out def read(self, size): return os.read(self.stdin_fd, size) def write(self, data): self.out.write(data) def flush(self): self.out.flush() def stop(self): pass def release(self): pass # Telnet (RFC 854): what the caller's client may still send on a DOOR32 socket. IAC, SE, SB, WILL, WONT, DO, DONT = 255, 240, 250, 251, 252, 253, 254 class TelnetFilter(object): """Takes the telnet framing off the caller's bytes; keeps its state across reads. IAC IAC -> 0xFF; IAC WILL/WONT/DO/DONT , IAC SB ... IAC SE and every other two-byte IAC command are dropped (the board negotiated, not us); CR NUL -> CR. A sequence may be split across any read. """ DATA, AFTER_IAC, AFTER_OPTION_CMD, IN_SB, IN_SB_IAC = range(5) # A subnegotiation nobody ends would otherwise swallow the caller's every # keystroke for good; past this many bytes it is given up on. SB_LIMIT = 512 def __init__(self): self.state = self.DATA # The last data byte was CR, so a NUL next is telnet's, not the # caller's -- even with an IAC command in between. self.after_cr = False self.sb_length = 0 def feed(self, data): out = bytearray() state = self.state after_cr = self.after_cr for byte in bytearray(data): if state == self.DATA: if byte == IAC: state = self.AFTER_IAC elif after_cr and byte == 0: after_cr = False else: out.append(byte) after_cr = byte == 13 elif state == self.AFTER_IAC: if byte == IAC: out.append(IAC) after_cr = False state = self.DATA elif byte in (WILL, WONT, DO, DONT): state = self.AFTER_OPTION_CMD elif byte == SB: state = self.IN_SB self.sb_length = 0 else: state = self.DATA # NOP, GA, AYT, a stray SE, ... elif state == self.AFTER_OPTION_CMD: state = self.DATA elif state == self.IN_SB: if byte == IAC: state = self.IN_SB_IAC else: self.sb_length += 1 if self.sb_length > self.SB_LIMIT: state = self.DATA elif state == self.IN_SB_IAC: # SE ends the subnegotiation; IAC IAC is a 0xFF inside it; # anything else is malformed and stays inside. state = self.DATA if byte == SE else self.IN_SB self.state = state self.after_cr = after_cr return bytes(out) def escape_iac(data): """0xFF is a CP437 glyph in the game's output and an IAC on a telnet wire.""" return bytes(data).replace(b"\xff", b"\xff\xff") WOULD_BLOCK = set( getattr(errno, name) for name in ("EWOULDBLOCK", "EAGAIN", "WSAEWOULDBLOCK") if hasattr(errno, name) ) def would_block(err): return getattr(err, "errno", None) in WOULD_BLOCK class SocketCaller(object): """The caller's telnet socket, inherited from a DOOR32 board. The socket is the BOARD's. It is never shut down here: when the game is over our handle is closed (see release()) and the caller is back on the board. Its blocking mode is left exactly as found (setblocking() can fail on a socket a Windows board has WSAEventSelect on), so every read and write selects first and treats would-block as "not yet", never as a hang-up. """ POLL = 0.25 # seconds between looks at the stop flag def __init__(self, sock): self.sock = sock self.filter = TelnetFilter() self.stopped = False def read(self, size): """Filtered caller bytes; b'' when the caller hung up or the door is stopping.""" while not self.stopped: sock = self.sock # released in another thread: then we are done if sock is None: break readable, _, _ = select.select([sock], [], [], self.POLL) if not readable: continue try: chunk = sock.recv(size) except GONE as err: if would_block(err): continue return b"" if not chunk: return b"" data = self.filter.feed(chunk) if data: return data return b"" def write(self, data): data = escape_iac(data) while data and not self.stopped: sock = self.sock if sock is None: break _, writable, _ = select.select([], [sock], [], self.POLL) if not writable: continue try: sent = sock.send(data) except GONE as err: if would_block(err): continue raise data = data[sent:] def flush(self): pass def stop(self): """Lets a read() blocked in another thread return.""" self.stopped = True def release(self): """Hand the socket back to the board, with its connection intact. close() here is closesocket()/close() on OUR handle only: the board holds another to the same socket, so the connection stays up. It is never shutdown(), which would act on the connection itself. And it is not detach(): a Windows process that exits with a Winsock handle still open has that socket RESET at Winsock cleanup (measured 2026-11-06, Python 3.13 / Windows 11: detach-and-exit gave the board WSAECONNABORTED; close-and-exit left its connection working). """ self.stopped = True if self.sock is not None: self.sock.close() self.sock = None class Door32Error(Exception): pass MAX_HANDLE = 2 ** 63 - 1 # past this socket.socket(fileno=) overflows instead of refusing def read_door32(path): """DOOR32.SYS -> {'comm_type', 'handle', 'user'}; Door32Error on anything else.""" try: with open(path, "rb") as handle: lines = handle.read().decode("latin-1").splitlines() except (IOError, OSError) as err: raise Door32Error("cannot read DOOR32.SYS at %s: %s" % (path, err)) lines = [line.strip() for line in lines] if len(lines) < 5: raise Door32Error("DOOR32.SYS at %s has %d lines; it needs at least 5" % (path, len(lines))) def number(index, what): if not re.match(r"^[0-9]+$", lines[index]): raise Door32Error("DOOR32.SYS line %d (%s) is not a number: %r" % (index + 1, what, lines[index])) return int(lines[index]) comm_type = number(0, "comm type") if comm_type != 2: raise Door32Error( "DOOR32.SYS line 1 is comm type %d (%s); SCLink needs a telnet socket, comm type 2" % (comm_type, {0: "local", 1: "serial"}.get(comm_type, "unknown")) ) handle = number(1, "socket handle") if handle == 0: raise Door32Error("DOOR32.SYS line 2 has no socket handle (0)") if handle > MAX_HANDLE: raise Door32Error("DOOR32.SYS line 2: %d is not a possible socket handle" % handle) return {"comm_type": comm_type, "handle": handle, "user": str(number(4, "user number"))} def inherited_socket(handle): """The board's socket, by the handle it wrote; Door32Error when it is not one.""" try: return socket.socket(fileno=handle) except GONE + (OverflowError,) as err: raise Door32Error("DOOR32.SYS line 2: %d is not an open socket handle (%s)" % (handle, err)) # --------------------------------------------------------------------------- # The play connection: RLOGIN framing, then a byte pump # --------------------------------------------------------------------------- def play(cfg, user_number, ticket, caller): """Connect, present the ticket, and pipe until either side ends. `caller` has read(size) -> bytes (b'' on hang-up), write(bytes), flush(), stop() to free a read in another thread, and release() when the game is over. """ sock = socket.create_connection((cfg["host"], int(cfg["play_port"])), cfg.get("timeout", 10)) if cfg.get("level", "tls") == "tls": sock = tls_context(cfg).wrap_socket(sock, server_hostname=cfg["host"]) sock.sendall(("\0%s\0%s\0ansi\0" % (user_number, ticket)).encode("latin-1")) first = sock.recv(4096) if not first or first[:1] != b"\0": lines = first[1:].decode("latin-1").splitlines() if first else [] sock.close() return sanitise(lines[0] if lines else "") or "Skull Crash refused the connection." sock.settimeout(None) def caller_to_game(): try: while True: data = caller.read(4096) if not data: break sock.sendall(data) except GONE: pass # The caller hung up (or their input broke): end the game session too, # rather than leave it open until the board kills this process. try: sock.shutdown(socket.SHUT_RDWR) except GONE: pass pump = threading.Thread(target=caller_to_game) pump.daemon = True pump.start() data = first[1:] while True: if data: try: caller.write(data) caller.flush() except GONE: break # the caller is gone: the pump ends the game side try: data = sock.recv(4096) except GONE: break if not data: break sock.close() # The game is over. The pump may be waiting on the caller: let it go, so a # DOOR32 socket is handed back to the board rather than closed under it. caller.stop() pump.join(2.0) return None def usage(): sys.stderr.write( "usage: sclink.py [rows] [cols]\n" " sclink.py --door32 [rows] [cols]\n" " sclink.py --selftest\n" ) return 2 def main(argv): if "--selftest" in argv: ok = self_test() print("sclink.py self-test: " + ("PASS" if ok else "FAIL")) return 0 if ok else 1 if len(argv) < 3: return usage() out = getattr(sys.stdout, "buffer", sys.stdout) door32 = argv[2] == "--door32" if door32: if len(argv) < 4: return usage() if sys.version_info[0] < 3: sys.stderr.write( "sclink.py: --door32 needs Python 3: it opens the board's socket handle with " "socket.socket(fileno=...), which Python 2 cannot do on Windows.\n" ) out.write(NOT_SET_UP) out.flush() return 2 try: drop = read_door32(argv[3]) caller = SocketCaller(inherited_socket(drop["handle"])) except Door32Error as err: # No socket to speak on: the console is the only output path left. sys.stderr.write("sclink.py: %s\n" % err) out.write(NOT_SET_UP) out.flush() return 2 number = drop["user"] dims = argv[4:6] else: caller = StdioCaller(sys.stdin.fileno(), out) number = argv[2] dims = argv[3:5] try: return run(argv[1], number, dims, caller) finally: caller.release() def not_set_up(caller, reason): """The board's own config is the problem: say what on stderr, tell the caller.""" sys.stderr.write("sclink.py: %s\n" % reason) try: caller.write(NOT_SET_UP) caller.flush() except GONE: pass return 1 def run(config_path, number, dims, caller): try: with open(config_path) as handle: cfg = json.load(handle) if not isinstance(cfg, dict): raise ValueError("not a JSON object") except (IOError, OSError, ValueError) as err: # json's decode error is a ValueError return not_set_up(caller, "cannot read config %s: %s" % (config_path, err)) problem = config_problem(cfg) if problem or not self_test(): return not_set_up(caller, problem or "the built-in MAC self-test failed") who = { "number": number, "rows": dims[0] if len(dims) > 0 else None, "cols": dims[1] if len(dims) > 1 else None, "ip": os.environ.get("SCLINK_CALLER"), } try: ticket, error = fetch_ticket(cfg, who) if not error: error = play(cfg, who["number"], ticket, caller) except Exception as err: # noqa: BLE001 -- anything here means the game is unreachable sys.stderr.write("sclink.py: %s\n" % err) error = "Skull Crash is unreachable right now." if error: try: caller.write(("\r\n" + sanitise(error) + "\r\n").encode("latin-1")) caller.flush() except GONE: pass return 1 return 0 if __name__ == "__main__": sys.exit(main(sys.argv))