#!/usr/bin/env python3 """gbhub - a virtual switch/router + DHCP for several Game Boys on one network. Spawns N emulators, leases each a distinct IP (10.0.0.2, 10.0.0.3, ...), routes packets between them (so they can ping/connect to *each other*), floods broadcasts, and NATs external traffic out a TUN (so they also reach the internet). No kernel changes: the GB just sends to a dst IP over its link and the hub decides where it goes. sudo tools/gbhub 2 # spawn 2 headless GBs, one network sudo tools/gbhub 2 netd "ping 10.0.0.2" # scripted: GB0 serves, GB1 pings it sudo tools/gbhub --daemon # just the hub; join windowed GBs with # tools/gbjoin (in other terminals) In spawn mode each positional arg after N is a console command auto-typed into that GB once it has a lease (a demo hook; interactively you'd use the OSK). Requires root. """ import os, sys, socket, threading, subprocess, struct, fcntl, time, signal, atexit, shutil HOME = os.path.expanduser("~" + (os.environ.get("SUDO_USER") or "")) EMU = os.path.join(HOME, "dev/gbc/build/sl0pboy") ROM = os.path.join(HOME, "dev/gbos/gbos.gb") SOCK = "/tmp/gbhub.sock" CTL = "/tmp/gbhub.ctl" # control socket: external shell-input injection (gbtype) IFACE, GW, MTU = "gbtun0", b"\x0a\x00\x00\x01", 320 END, ESC, BCAST = 0xC0, 0xDB, b"\xff\xff\xff\xff" def slip(pkt): o = bytearray([END]) for c in pkt: o += bytes([ESC,0xDC]) if c==END else bytes([ESC,0xDD]) if c==ESC else bytes([c]) o.append(END); return bytes(o) def _ck(d): s = 0 for i in range(0, len(d)-1, 2): s += (d[i]<<8)|d[i+1] if len(d)%2: s += d[-1]<<8 while s>>16: s = (s&0xFFFF)+(s>>16) return (~s)&0xFFFF def _dmt(dh): p = 240 while p < len(dh) and dh[p] != 255: if dh[p]==0: p += 1; continue if dh[p]==53: return dh[p+2] p += 2 + dh[p+1] return 0 def _dreply(req, mt, gbip): d = bytearray(240); d[0]=2; d[1]=1; d[2]=6; d[4:8]=req[4:8] d[16:20]=gbip; d[20:24]=GW; d[28:44]=req[28:44]; d[236:240]=b"\x63\x82\x53\x63" d += bytes([53,1,mt,54,4])+GW+bytes([255]) udp = bytearray(8); udp[1]=67; udp[3]=68; ul=8+len(d); udp[4]=ul>>8; udp[5]=ul&0xFF; udp += d tot = 20+len(udp); ip = bytearray(20); ip[0]=0x45; ip[2]=tot>>8; ip[3]=tot&0xFF ip[8]=64; ip[9]=17; ip[12:16]=GW; ip[16:20]=b"\xff\xff\xff\xff" ck=_ck(ip); ip[10]=ck>>8; ip[11]=ck&0xFF return bytes(ip)+bytes(udp) def sh(cmd): subprocess.run(cmd, shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) conns = {} # ip(bytes) -> GB pool = list(range(2, 40)) # free host octets (10.0.0.2 .. .39) lock = threading.Lock() tun = None class GB: def __init__(self, octet, conn): self.octet, self.idx, self.conn = octet, octet - 2, conn self.ip = bytes([10,0,0,octet]) self.wlock = threading.Lock() self.line = "" def send(self, data): with self.wlock: try: self.conn.sendall(data) except Exception: pass def frame(self, pkt): self.send(slip(pkt)) def type(self, text): self.send(text.encode()) # non-framed = console input def route(gb, fr): if len(fr) >= 28 and fr[9] == 17: # DHCP -> answer with this GB's IP ihl = (fr[0]&0x0f)*4 if ((fr[ihl+2]<<8)|fr[ihl+3]) == 67: mt = _dmt(fr[ihl+8:]) if mt == 1: gb.frame(_dreply(fr[ihl+8:], 2, gb.ip)) elif mt == 3: gb.frame(_dreply(fr[ihl+8:], 5, gb.ip)) return if len(fr) < 20: return dst = fr[16:20] if dst == BCAST: for g in conns.values(): if g is not gb: g.frame(fr) elif dst in conns: conns[dst].frame(fr) # Game Boy -> Game Boy elif tun is not None: try: os.write(tun, fr) # -> internet via NAT except Exception: pass def gb_reader(gb): buf, inf, esc = bytearray(), False, False while True: why = "eof" try: chunk = gb.conn.recv(2048) except Exception as e: chunk = None; why = "%s: %s" % (type(e).__name__, e) if not chunk: with lock: # GB left: free its address conns.pop(gb.ip, None) if gb.octet not in pool: pool.append(gb.octet); pool.sort() sys.stderr.write("[gbhub] %s GB%d (10.0.0.%d) left (%s)\n" % (time.strftime("%H:%M:%S"), gb.idx, gb.octet, why)) return for c in chunk: if c == END: if inf and buf: route(gb, bytes(buf)); buf=bytearray(); inf=False else: buf=bytearray(); inf=True; esc=False elif inf: if esc: c={0xDC:END,0xDD:ESC}.get(c,c); esc=False; buf.append(c) elif c==ESC: esc=True else: buf.append(c) else: # console output -> log per GB if c == 10: sys.stderr.write("[GB%d] %s\n" % (gb.idx, gb.line)); sys.stderr.flush(); gb.line="" elif 32 <= c < 127: gb.line += chr(c) def tun_reader(): while True: try: pkt = os.read(tun, 2048) except Exception: return if len(pkt) >= 20 and pkt[16:20] in conns: conns[pkt[16:20]].frame(pkt) def net_setup(): global tun tun = os.open("/dev/net/tun", os.O_RDWR) try: fcntl.ioctl(tun, 0x400454ca, struct.pack("16sH", IFACE.encode(), 0x0001|0x1000)) except OSError as e: sys.exit("gbhub: cannot claim %s (%s) - another gbhub is already running?\n" " (sudo pkill -f tools/gbhub to stop it)" % (IFACE, e.strerror)) wan = subprocess.check_output("ip route show default|awk '{print $5;exit}'", shell=True).decode().strip() or "eth0" sh("ip addr add 10.0.0.1/24 dev %s" % IFACE); sh("ip link set %s mtu %d up" % (IFACE, MTU)) sh("sysctl -w net.ipv4.ip_forward=1") sh("iptables -t nat -C POSTROUTING -s 10.0.0.0/24 -o %s -j MASQUERADE || iptables -t nat -A POSTROUTING -s 10.0.0.0/24 -o %s -j MASQUERADE" % (wan, wan)) sh("iptables -C FORWARD -i %s -j ACCEPT || iptables -I FORWARD -i %s -j ACCEPT" % (IFACE, IFACE)) sh("iptables -C FORWARD -o %s -j ACCEPT || iptables -I FORWARD -o %s -j ACCEPT" % (IFACE, IFACE)) def cleanup(): sh("iptables -t nat -D POSTROUTING -s 10.0.0.0/24 -o %s -j MASQUERADE" % wan) sh("iptables -D FORWARD -i %s -j ACCEPT" % IFACE); sh("iptables -D FORWARD -o %s -j ACCEPT" % IFACE) for p in (SOCK, CTL): try: os.unlink(p) except Exception: pass atexit.register(cleanup) return wan def register(conn): with lock: if not pool: conn.close(); return None octet = pool.pop(0) gb = GB(octet, conn); conns[gb.ip] = gb threading.Thread(target=gb_reader, args=(gb,), daemon=True).start() sys.stderr.write("[gbhub] %s GB%d joined -> 10.0.0.%d\n" % (time.strftime("%H:%M:%S"), gb.idx, octet)) return gb # ----------------------------------------------------------------------------- # Control socket: inject shell input into a GB from the host (see tools/gbtype). # The link port is the GB's console when nothing SLIP-frames the bytes, so text # written with GB.type() lands on the shell's stdin - the OSK, remotely. # ----------------------------------------------------------------------------- def _find_gb(idx): gbs = sorted(conns.values(), key=lambda g: g.idx) if not gbs: return None if idx is None: return gbs[0] # default: lowest-index GB for g in gbs: if g.idx == idx: return g return None def _type_paced(gb, data): # The kernel console ring is 64 bytes and one net_pump drains a whole burst # into it at once, so feed in small chunks with a brief gap - the shell # keeps up easily, and arbitrarily long scripts inject without overflow. for i in range(0, len(data), 32): gb.send(data[i:i+32]); time.sleep(0.04) def _control_client(conn): try: conn.settimeout(5) data = b"" while True: # read header + payload try: d = conn.recv(4096) # until the client except Exception: break # half-closes (SHUT_WR) if not d: break data += d head, _, payload = data.partition(b"\n") parts = head.decode(errors="replace").split() cmd = parts[0].lower() if parts else "" if cmd == "list": gbs = sorted(conns.values(), key=lambda g: g.idx) msg = "".join("GB%d 10.0.0.%d\n" % (g.idx, g.octet) for g in gbs) conn.sendall(msg.encode() or b"(no Game Boys connected)\n") elif cmd == "type": idx = None if len(parts) >= 2: try: idx = int(parts[1]) except ValueError: idx = None gb = _find_gb(idx) if gb is None: conn.sendall(b"err: no such Game Boy (try 'list')\n") else: _type_paced(gb, payload) conn.sendall(("ok: %d bytes -> GB%d\n" % (len(payload), gb.idx)).encode()) else: conn.sendall(b"err: commands are 'type [idx]' (payload follows newline) or 'list'\n") except Exception: pass finally: try: conn.close() except Exception: pass def control_server(): try: os.unlink(CTL) except Exception: pass c = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) c.bind(CTL); c.listen(8) try: os.chmod(CTL, 0o777) # let non-root gbtype connect except Exception: pass while True: try: conn, _ = c.accept() except Exception: return threading.Thread(target=_control_client, args=(conn,), daemon=True).start() def main(): if os.geteuid() != 0: sys.exit("gbhub: run as root") args = sys.argv[1:] daemon = bool(args) and args[0] in ("--daemon", "-d") wan = net_setup() try: os.unlink(SOCK) except Exception: pass srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM); srv.bind(SOCK); srv.listen(8) threading.Thread(target=tun_reader, daemon=True).start() threading.Thread(target=control_server, daemon=True).start() # gbtype input if daemon: os.chmod(SOCK, 0o777) # let non-root gbjoin connect sys.stderr.write("[gbhub] up on 10.0.0.0/24, NAT via %s. " "join Game Boys in other terminals: tools/gbjoin\n" "[gbhub] inject shell input without the OSK: tools/gbtype 'ls'\n" % wan) try: while True: conn, _ = srv.accept(); register(conn) except KeyboardInterrupt: pass return # spawn mode: launch N headless GBs ourselves (self-contained demo) n = int(args[0]) if args else 2 cmds = args[1:1+n] emus = [] for i in range(n): rom = "/tmp/gbhub_%d.gb" % i # per-GB ROM so saves don't clash shutil.copyfile(ROM, rom) err = open("/tmp/gbhub_%d.err" % i, "ab", buffering=0) # keep emulator stderr for postmortems emus.append(subprocess.Popen([EMU, "--headless", "--uncapped", "--serial-sock", SOCK, rom], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=err, start_new_session=True)) err.close() sys.stderr.write("[gbhub] %d Game Boys on 10.0.0.0/24 (10.0.0.2..%d), NAT via %s\n" % (n, 1+n, wan)) srv.settimeout(12) for i in range(n): conn, _ = srv.accept(); register(conn) time.sleep(2.0) # let DHCP settle for i in range(n): if i < len(cmds) and cmds[i]: g = conns.get(bytes([10,0,0,2+i])) if g: g.type(cmds[i] + "\n") def stop(*a): for e in emus: try: os.killpg(os.getpgid(e.pid), signal.SIGKILL) except Exception: pass sys.exit(0) signal.signal(signal.SIGTERM, stop) try: while all(e.poll() is None for e in emus): time.sleep(0.3) except KeyboardInterrupt: pass finally: stop() if __name__ == "__main__": main()