diff --git a/scripts/mini-dhcpd.py b/scripts/mini-dhcpd.py new file mode 100755 index 0000000..0edb049 --- /dev/null +++ b/scripts/mini-dhcpd.py @@ -0,0 +1,221 @@ +#!/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")