Roflin
0c5a7bf9db
Swiss got a DHCP lease on hardware (10.0.0.64/24), but the script flagged
every ARP from the GC as "frame did NOT parse as DHCP ... this is what a
corrupted TX frame looks like". That was wrong and actively misleading: the
frames are valid lwIP duplicate-address probes (sender 0.0.0.0, target =
the offered address), sent to check the lease is free before accepting it.
The parser conflated "not DHCP" with "corrupted". A diagnostic that cries
wolf on normal traffic is worse than none, because 'malformed' then means
nothing. Now:
'info' from the client but not DHCP -- ARP decoded properly (probe vs
request vs reply, with addresses), other ethertypes named
'malformed' CLAIMED IPv4/UDP->67 and then failed to parse, i.e. what a
genuinely damaged TX frame would look like
So the corruption diagnostic still works -- that was the point of using a raw
socket -- but it only fires when it means something.
scripts/test-mini-dhcpd.py keeps the self-test, now including the exact
60-byte frame from the console (T6), a damaged DHCP packet that must still be
flagged (T7), and a raw ethertype (T8).
One note for anyone extending that test: the real frame goes in as ONE string
literal. Splitting the hex across two adjacent literals is how the first
version of T6 silently lost a byte and blamed the parser for decoding the
target as 0.10.0.0 -- the parser was correct.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
265 lines
11 KiB
Python
Executable File
265 lines
11 KiB
Python
Executable File
#!/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")
|