#!/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 ETH_P_ARP = 0x0806 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 describe_arp(pl): """Decode an ARP body far enough to be useful during bring-up.""" if len(pl) < 28: return "ARP, truncated" htype, ptype, hlen, plen, op = struct.unpack("!HHBBH", pl[:8]) if (htype, ptype, hlen, plen) != (1, ETH_P_IP, 6, 4): return (f"ARP with unexpected htype/ptype/hlen/plen " f"{htype}/{ptype:#06x}/{hlen}/{plen}") sha, spa = pl[8:14], socket.inet_ntoa(pl[14:18]) tha, tpa = pl[18:24], socket.inet_ntoa(pl[24:28]) kind = {1: "request", 2: "reply"}.get(op, f"op {op}") if op == 1 and spa == "0.0.0.0": return (f"ARP probe for {tpa} from {mac_str(sha)} -- lwIP checking the " f"DHCP lease is not already taken (normal)") if op == 1: return f"ARP request: who has {tpa}? tell {spa} ({mac_str(sha)})" return f"ARP reply: {spa} is at {mac_str(sha)} -> {tpa} ({mac_str(tha)})" def parse_dhcp(frame, want_mac): """Return (kind, info). kind is one of: 'dhcp' a DHCP message, info is the decoded fields 'info' a frame from the client that is simply not DHCP (ARP, etc.) 'malformed' it CLAIMED to be IPv4/UDP->67 and then failed to parse 'other' not from the client we care about ⚠️ 'info' and 'malformed' used to be the same thing, which made this tool report every ARP as a corrupted frame. A diagnostic that cries wolf on normal traffic is worse than no diagnostic: the whole point is that 'malformed' means something. ARP from the GC is expected -- lwIP probes its DHCP lease before accepting it, and the probe also happens to prove the TX path works for non-IP ethertypes. """ 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_ARP: return "info", describe_arp(frame[14:]) if etype != ETH_P_IP: return "info", f"ethertype {etype:#06x} (not IPv4 or ARP)" 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 == "info": print(f"[{ts}] {info} ({len(frame)} bytes)") if a.hex: print(" " + binascii.hexlify(frame[:64]).decode()) continue if kind == "malformed": # Reached only when the frame said IPv4/UDP->67 and then did not # hold up -- i.e. a genuinely damaged DHCP packet, which is what # a corrupted TX path would produce. print(f"[{ts}] ⚠️ DAMAGED DHCP frame from the client: {info}") print(f" {len(frame)} bytes -- a well-formed DISCOVER " f"would have parsed; suspect the TX path") 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")