aboutsummaryrefslogtreecommitdiffstats
path: root/tools
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--tools/gateway.py169
1 files changed, 134 insertions, 35 deletions
diff --git a/tools/gateway.py b/tools/gateway.py
index 3b18024..590526e 100644
--- a/tools/gateway.py
+++ b/tools/gateway.py
@@ -2,21 +2,26 @@
"""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 (DNS/TCP/HTTP). Modes:
+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: b'D'+chunk for data, b'E' to end the stream.
+Reply frames are typed for http (b'D'+chunk .. b'E'); chat/irc frames are the
+raw display line.
-usage: tools/gateway.py [--mode echo|http] [--cmd "wget example.com"] [--rom PATH]
+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 = 200
-LIMIT = 800 # cap the fetched body (link is ~1-32 KB/s)
+CHUNK, LIMIT = 200, 800
def slip_encode(payload):
out = bytearray([END])
@@ -29,8 +34,7 @@ def slip_encode(payload):
def http_fetch(payload):
url = bytes(payload).decode("latin1", "replace").strip()
- if not url.startswith("http"):
- url = "http://" + url
+ 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"})
@@ -39,29 +43,61 @@ def http_fetch(payload):
except Exception as e:
body = ("ERR: " + str(e)).encode()
sys.stderr.write("[gw] %d bytes\n" % len(body))
- frames = [b"D" + body[i:i+CHUNK] for i in range(0, len(body), CHUNK)]
- frames.append(b"E")
- return frames
+ return [b"D" + body[i:i+CHUNK] for i in range(0, len(body), CHUNK)] + [b"E"]
-def handle(payload, mode):
- if mode == "http":
- return http_fetch(payload)
- if mode == "chat":
- msg = bytes(payload).decode("latin1", "replace")
- sys.stderr.write("\n[gw] GB says: %s\n" % msg)
- return [("<bot> " + msg).encode()] # a simple echo-bot peer
- return [bytes(payload)] # echo
+_IRC_FMT = re.compile(r'(\x03\d{0,2}(,\d{1,2})?|[\x00-\x1f])')
+def strip_irc(s): return _IRC_FMT.sub('', s)
-def reader(out, inp, mode):
+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
+ if not b: return
c = b[0]
if c == END:
if in_frame and buf:
- for f in handle(buf, mode):
+ for f in on_frame(bytes(buf)):
inp.write(slip_encode(f)); inp.flush()
buf, in_frame, esc = bytearray(), False, False
else:
@@ -69,37 +105,100 @@ def reader(out, inp, mode):
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)
+ 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"])
+ ap.add_argument("--mode", default="echo", choices=["echo", "http", "chat", "irc", "ping"])
ap.add_argument("--cmd", default="necho")
- ap.add_argument("--rom", default=os.path.join(os.path.dirname(__file__), "..", "gbos.gb"))
+ 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)
- threading.Thread(target=reader, args=(p.stdout, p.stdin, a.mode), daemon=True).start()
+ 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 [("<bot> " + 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"<alice> hey gameboy!", b"<bob> nice link cable"]:
p.stdin.write(slip_encode(m)); p.stdin.flush(); time.sleep(0.6)
- time.sleep(4.0)
- p.stdin.write(b"exit\n"); p.stdin.flush()
- try: p.wait(timeout=3)
- except Exception: p.kill()
+ 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()
generated by cgit v1.3.1-sl0p (git 2.54.0) at 2026-07-21 23:18:30 +0000 — sl0p fork