#!/usr/bin/env python3 """Minimal DHCP server on a raw socket, for BBA bring-up. sudo ./scripts/mini-dhcpd.py enp0s31f6 Why raw (AF_PACKET) rather than a UDP socket on port 67: a UDP socket only sees frames the kernel accepts as well-formed IP/UDP. One of the things we are trying to find out is whether the GC's frames arrive INTACT -- libogc streams TX payload with EXI_Dma, and the GC's DMA-write engine is documented to have a 1-bit-shift bug. A corrupted DISCOVER would never reach a UDP socket, and the server would look "fine, just no client". On a raw socket we see every frame the adapter sends, parse it ourselves, and say exactly why it did not qualify. Replies go out as raw ethernet UNICAST to the client's MAC. That is also a test: the gateware filters RX to own-MAC + broadcast (the MACRAW MF bit), so a unicast reply landing is evidence the filter is set up correctly. Serves ONE address, logs everything, and does not touch the rest of the system. Not a real DHCP server -- no persistence, no conflict detection, no relay support. Stop with Ctrl-C. """ import argparse import binascii import socket import struct import sys import time ETH_P_ALL = 0x0003 ETH_P_IP = 0x0800 BOOTP_MIN = 236 # BOOTP header before the magic cookie MAGIC = b"\x63\x82\x53\x63" DHCP_DISCOVER, DHCP_OFFER, DHCP_REQUEST, DHCP_DECLINE, DHCP_ACK, DHCP_NAK = 1, 2, 3, 4, 5, 6 MSG_NAMES = {1: "DISCOVER", 2: "OFFER", 3: "REQUEST", 4: "DECLINE", 5: "ACK", 6: "NAK", 7: "RELEASE", 8: "INFORM"} def mac_str(b): return ":".join(f"{x:02x}" for x in b) def ip_csum(data): if len(data) % 2: data += b"\0" s = sum(struct.unpack("!%dH" % (len(data) // 2), data)) while s >> 16: s = (s & 0xFFFF) + (s >> 16) return (~s) & 0xFFFF def build_reply(src_mac, dst_mac, src_ip, dst_ip, payload): """Ethernet + IPv4 + UDP(67->68) around `payload`. UDP checksum 0, which IPv4 permits, so there is one less thing to get wrong.""" udp = struct.pack("!HHHH", 67, 68, 8 + len(payload), 0) + payload ihl_total = 20 + len(udp) ip = struct.pack("!BBHHHBBH4s4s", 0x45, 0, ihl_total, 0x1234, 0, 64, socket.IPPROTO_UDP, 0, socket.inet_aton(src_ip), socket.inet_aton(dst_ip)) ip = ip[:10] + struct.pack("!H", ip_csum(ip)) + ip[12:] return dst_mac + src_mac + struct.pack("!H", ETH_P_IP) + ip + udp def parse_dhcp(frame, want_mac): """Return (kind, info). kind is 'dhcp', 'other', or 'malformed' -- the last one is the interesting case, so it carries a reason.""" if len(frame) < 14: return "other", None dst, src, etype = frame[0:6], frame[6:12], struct.unpack("!H", frame[12:14])[0] if want_mac and src != want_mac: return "other", None if etype != ETH_P_IP: return "malformed", f"from {mac_str(src)} but ethertype {etype:#06x}, not IPv4" ip = frame[14:] if len(ip) < 20: return "malformed", "IPv4 header truncated" ihl = (ip[0] & 0x0F) * 4 if ip[0] >> 4 != 4: return "malformed", f"IP version {ip[0] >> 4}, not 4" if ip[9] != socket.IPPROTO_UDP: return "malformed", f"IP proto {ip[9]}, not UDP" udp = ip[ihl:] if len(udp) < 8: return "malformed", "UDP header truncated" sport, dport, ulen, _ = struct.unpack("!HHHH", udp[:8]) if dport != 67: return "other", None bootp = udp[8:] if len(bootp) < BOOTP_MIN + 4: return "malformed", f"BOOTP too short: {len(bootp)} bytes" if bootp[BOOTP_MIN:BOOTP_MIN + 4] != MAGIC: return "malformed", ("no DHCP magic cookie -- got " f"{binascii.hexlify(bootp[BOOTP_MIN:BOOTP_MIN+4]).decode()}") op, _, hlen, _ = bootp[0], bootp[1], bootp[2], bootp[3] xid = bootp[4:8] flags = struct.unpack("!H", bootp[10:12])[0] chaddr = bootp[28:28 + max(hlen, 6)][:6] opts, i, msg_type = {}, BOOTP_MIN + 4, None while i < len(bootp): code = bootp[i] if code == 255: break if code == 0: i += 1 continue if i + 2 > len(bootp): return "malformed", "option list truncated" ln = bootp[i + 1] opts[code] = bootp[i + 2:i + 2 + ln] i += 2 + ln if 53 in opts and len(opts[53]) == 1: msg_type = opts[53][0] return "dhcp", dict(op=op, xid=xid, flags=flags, chaddr=chaddr, msg_type=msg_type, opts=opts, src=src) def build_bootp(msg_type, xid, flags, chaddr, offer_ip, server_ip, mask, router, dns, lease): pkt = struct.pack("!BBBB4sHH", 2, 1, 6, 0, xid, 0, flags) pkt += socket.inet_aton("0.0.0.0") # ciaddr pkt += socket.inet_aton(offer_ip) # yiaddr pkt += socket.inet_aton(server_ip) # siaddr pkt += socket.inet_aton("0.0.0.0") # giaddr pkt += chaddr + b"\0" * (16 - len(chaddr)) # chaddr pkt += b"\0" * 64 + b"\0" * 128 # sname + file pkt += MAGIC def opt(c, v): return bytes([c, len(v)]) + v pkt += opt(53, bytes([msg_type])) pkt += opt(54, socket.inet_aton(server_ip)) pkt += opt(51, struct.pack("!I", lease)) pkt += opt(1, socket.inet_aton(mask)) if router: pkt += opt(3, socket.inet_aton(router)) if dns: pkt += opt(6, socket.inet_aton(dns)) pkt += b"\xff" if len(pkt) < 300: # pad to the BOOTP minimum pkt += b"\0" * (300 - len(pkt)) return pkt def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("iface") ap.add_argument("--server-ip", default="10.0.0.1") ap.add_argument("--offer-ip", default="10.0.0.64") ap.add_argument("--mask", default="255.255.255.0") ap.add_argument("--router", default=None, help="default: the server IP") ap.add_argument("--dns", default=None, help="default: the server IP") ap.add_argument("--lease", type=int, default=3600) ap.add_argument("--mac", default="00:09:bf:00:00:01", help="only react to this client; 'any' for all") ap.add_argument("--hex", action="store_true", help="hexdump every frame from the client") a = ap.parse_args() router = a.router or a.server_ip dns = a.dns or a.server_ip want = None if a.mac == "any" else bytes.fromhex(a.mac.replace(":", "")) s = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(ETH_P_ALL)) s.bind((a.iface, 0)) our_mac = s.getsockname()[4][:6] print(f"mini-dhcpd on {a.iface} ({mac_str(our_mac)})") print(f" server {a.server_ip} offering {a.offer_ip}/{a.mask}" f" router {router} dns {dns} lease {a.lease}s") print(f" client filter: {a.mac}") print(f" NOTE: give the interface an address in that subnet first, e.g.") print(f" sudo ip addr add {a.server_ip}/24 dev {a.iface}") print(" waiting for DISCOVER (Ctrl-C to stop)\n") seen_any = False while True: frame = s.recv(65535) kind, info = parse_dhcp(frame, want) if kind == "other": continue seen_any = True ts = time.strftime("%H:%M:%S") if kind == "malformed": print(f"[{ts}] ⚠️ frame from the client did NOT parse as DHCP: {info}") print(f" {len(frame)} bytes -- this is what a corrupted") print(f" TX frame looks like; compare against a known-good one") print(" " + binascii.hexlify(frame[:64]).decode()) continue name = MSG_NAMES.get(info["msg_type"], str(info["msg_type"])) print(f"[{ts}] {name} from {mac_str(info['chaddr'])} " f"xid={binascii.hexlify(info['xid']).decode()} " f"flags={info['flags']:#06x} len={len(frame)}") if a.hex: print(" " + binascii.hexlify(frame[:64]).decode()) if info["msg_type"] == DHCP_DISCOVER: reply_type = DHCP_OFFER elif info["msg_type"] == DHCP_REQUEST: reply_type = DHCP_ACK else: print(" (no reply needed)") continue bootp = build_bootp(reply_type, info["xid"], info["flags"], info["chaddr"], a.offer_ip, a.server_ip, a.mask, router, dns, a.lease) # Unicast to the client's MAC. The gateware filters RX to own-MAC + # broadcast, so this also proves the filter accepts unicast. out = build_reply(our_mac, info["chaddr"], a.server_ip, a.offer_ip, bootp) s.send(out) print(f" -> {MSG_NAMES[reply_type]} {a.offer_ip} " f"unicast to {mac_str(info['chaddr'])} ({len(out)} bytes)") if __name__ == "__main__": try: main() except PermissionError: sys.exit("needs root for AF_PACKET: run with sudo") except KeyboardInterrupt: print("\nstopped")