#!/usr/bin/env python3 """gateway.py - host-side "link cable adapter" for gbos networking. Wraps the emulator, owns its link-port serial (stdio), speaks SLIP, and does the real network work the Game Boy can't. Modes: echo - bounce every frame back (loopback test) http - treat each frame as a URL, HTTP GET it, stream the body back chat - a simple echo-bot peer (+ a few async messages) irc - bridge to a real IRC server over TLS; the GB is a chat client and the gateway does the crypto/TCP/IRC protocol Reply frames are typed for http (b'D'+chunk .. b'E'); chat/irc frames are the raw display line. usage: tools/gateway.py [--mode ...] [--cmd "chat"] [--nick gameboy] [--channel '#sl0p'] [--keys STR] [--rom PATH] """ import os, sys, subprocess, threading, time, argparse, urllib.request import socket, ssl, re, struct EMU = os.path.expanduser("~/dev/gbc/build/sl0pboy") END, ESC, ESC_END, ESC_ESC = 0xC0, 0xDB, 0xDC, 0xDD CHUNK, LIMIT = 200, 800 def slip_encode(payload): out = bytearray([END]) for c in payload: if c == END: out += bytes([ESC, ESC_END]) elif c == ESC: out += bytes([ESC, ESC_ESC]) else: out.append(c) out.append(END) return bytes(out) def http_fetch(payload): url = bytes(payload).decode("latin1", "replace").strip() if not url.startswith("http"): url = "http://" + url sys.stderr.write("\n[gw] GET %s\n" % url) try: req = urllib.request.Request(url, headers={"User-Agent": "gbos/0.1"}) with urllib.request.urlopen(req, timeout=8) as r: body = r.read(LIMIT) except Exception as e: body = ("ERR: " + str(e)).encode() sys.stderr.write("[gw] %d bytes\n" % len(body)) return [b"D" + body[i:i+CHUNK] for i in range(0, len(body), CHUNK)] + [b"E"] _IRC_FMT = re.compile(r'(\x03\d{0,2}(,\d{1,2})?|[\x00-\x1f])') def strip_irc(s): return _IRC_FMT.sub('', s) class IRC: """Minimal TLS IRC client. The gateway does all the crypto/protocol.""" def __init__(self, host, port, nick, chan, to_gb): self.nick, self.chan, self.to_gb, self.buf = nick, chan, to_gb, b"" ctx = ssl.create_default_context() ctx.check_hostname = False; ctx.verify_mode = ssl.CERT_NONE self.s = ctx.wrap_socket(socket.create_connection((host, port), timeout=10), server_hostname=host) self._send("NICK " + nick) self._send("USER %s 0 * :gbos on a link cable" % nick) def _send(self, x): self.s.sendall((x + "\r\n").encode("latin1", "replace")) def privmsg(self, text): self._send("PRIVMSG %s :%s" % (self.chan, text[:400])) def loop(self): while True: try: d = self.s.recv(4096) except Exception: return if not d: return self.buf += d while b"\r\n" in self.buf: line, self.buf = self.buf.split(b"\r\n", 1) self._handle(line.decode("latin1", "replace")) def _handle(self, line): if line.startswith("PING"): self._send("PONG" + line[4:]); return p = line.split(" ", 3) if len(p) < 2: return cmd = p[1] if cmd == "001": self._send("JOIN " + self.chan) self.to_gb("* connected to %s, joining %s" % (p[0].lstrip(":"), self.chan)) elif cmd == "433": self.nick += "_"; self._send("NICK " + self.nick) elif cmd == "JOIN" and len(p) >= 3: who = p[0].lstrip(":").split("!")[0] if who == self.nick: self.to_gb("* joined %s" % self.chan) elif cmd == "PRIVMSG" and len(p) >= 4 and p[2].lower() == self.chan.lower(): who = p[0].lstrip(":").split("!")[0] self.to_gb("%s: %s" % (who, strip_irc(p[3].lstrip(":")))) def reader(out, inp, on_frame): buf, in_frame, esc = bytearray(), False, False while True: b = out.read(1) if not b: return c = b[0] if c == END: if in_frame and buf: for f in on_frame(bytes(buf)): inp.write(slip_encode(f)); inp.flush() buf, in_frame, esc = bytearray(), False, False else: buf, in_frame, esc = bytearray(), True, False elif in_frame: if esc: c = {ESC_END: END, ESC_ESC: ESC}.get(c, c); esc = False; buf.append(c) elif c == ESC: esc = True else: buf.append(c) else: sys.stderr.write(chr(c) if 32 <= c < 127 else ("\n" if c == 10 else ".")) sys.stderr.flush() GB_IP, GW_IP = bytes([10,0,0,2]), bytes([10,0,0,1]) def inet_cksum(data): if len(data) % 2: data += b"\x00" s = sum(struct.unpack("!%dH" % (len(data)//2), data)) while s >> 16: s = (s & 0xffff) + (s >> 16) return (~s) & 0xffff def icmp_echo(seq, payload=b"gbos-ping-0123456789"): icmp = struct.pack("!BBHHH", 8, 0, 0, 0xBEEF, seq) + payload icmp = icmp[:2] + struct.pack("!H", inet_cksum(icmp)) + icmp[4:] ip = struct.pack("!BBHHHBBH", 0x45, 0, 20+len(icmp), seq, 0, 64, 1, 0) + GW_IP + GB_IP ip = ip[:10] + struct.pack("!H", inet_cksum(ip)) + ip[12:] return ip + icmp def main(): ap = argparse.ArgumentParser() ap.add_argument("--mode", default="echo", choices=["echo", "http", "chat", "irc", "ping"]) ap.add_argument("--cmd", default="necho") ap.add_argument("--nick", default="gameboy") ap.add_argument("--channel", default="#sl0p") ap.add_argument("--host", default="sl0p.foo") ap.add_argument("--port", type=int, default=6697) ap.add_argument("--keys", default=None) ap.add_argument("--seconds", type=float, default=6.0) ap.add_argument("--rom", default=os.path.join(os.path.dirname(__file__), "..", "gbos.gb")) a = ap.parse_args() cmd = [EMU, "--headless", "--uncapped"] if a.keys: cmd += ["--keys", a.keys] cmd.append(a.rom) p = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, bufsize=0) irc_holder = [None] if a.mode == "irc": def on_frame(payload): msg = bytes(payload).decode("latin1", "replace") if irc_holder[0] is None: return [] # chat not wired to IRC yet sys.stderr.write("\n[gw] GB -> IRC: %s\n" % msg) irc_holder[0].privmsg(msg) return [] elif a.mode == "http": on_frame = http_fetch elif a.mode == "chat": def on_frame(payload): m = bytes(payload).decode("latin1", "replace") sys.stderr.write("\n[gw] GB says: %s\n" % m) return [(" " + m).encode()] elif a.mode == "ping": def on_frame(payload): p = bytes(payload) if len(p) >= 28 and (p[0] >> 4) == 4 and p[9] == 1: ihl = (p[0] & 0x0f) * 4 if p[ihl] == 0: # ICMP echo reply seq = struct.unpack("!H", p[ihl+6:ihl+8])[0] ok = inet_cksum(p[:ihl]) == 0 and inet_cksum(p[ihl:]) == 0 sys.stderr.write("\n[gw] reply from %d.%d.%d.%d icmp_seq=%d cksum=%s\n" % (p[12],p[13],p[14],p[15],seq,"ok" if ok else "BAD")) return [] else: on_frame = lambda payload: [bytes(payload)] threading.Thread(target=reader, args=(p.stdout, p.stdin, on_frame), daemon=True).start() time.sleep(1.0) p.stdin.write((a.cmd + "\n").encode()); p.stdin.flush() if a.mode == "irc": time.sleep(1.5) # let `chat` come up first def to_gb(text): try: p.stdin.write(slip_encode(text.encode("latin1", "replace"))); p.stdin.flush() except Exception: pass irc_holder[0] = IRC(a.host, a.port, a.nick, a.channel, to_gb) threading.Thread(target=irc_holder[0].loop, daemon=True).start() if a.mode == "ping": time.sleep(1.0) for seq in range(1, 5): p.stdin.write(slip_encode(icmp_echo(seq))); p.stdin.flush() sys.stderr.write("[gw] ping 10.0.0.2 icmp_seq=%d\n" % seq) time.sleep(0.8) if a.mode == "chat": time.sleep(1.0) for m in [b"welcome to sl0pchat", b" hey gameboy!", b" nice link cable"]: p.stdin.write(slip_encode(m)); p.stdin.flush(); time.sleep(0.6) time.sleep(a.seconds) try: p.stdin.write(b"exit\n"); p.stdin.flush() p.wait(timeout=3) except Exception: p.kill() if __name__ == "__main__": main()