41. A DNS client from raw bytes with struct
DNS is a small binary protocol over UDP
(Net Ch 18).
Building a query by hand teaches struct, bytes manipulation, and why
parsers of network data must be defensive.
# labs/dnsq.py: query an A/AAAA record over UDP, parse the answer (with name compression)
import random
import socket
import struct
import sys
TYPES = {"A": 1, "NS": 2, "CNAME": 5, "MX": 15, "TXT": 16, "AAAA": 28}
def encode_name(name: str) -> bytes:
out = b""
for label in name.rstrip(".").split("."):
raw = label.encode("idna")
if not 0 < len(raw) < 64:
raise ValueError(f"bad label {label!r}")
out += bytes([len(raw)]) + raw
return out + b"\x00"
def build_query(name: str, qtype: int) -> tuple[int, bytes]:
qid = random.randint(0, 0xFFFF) # random ID: part of the defense against spoofed answers
header = struct.pack("!HHHHHH", qid, 0x0100, 1, 0, 0, 0) # flags: RD (recursion desired)
return qid, header + encode_name(name) + struct.pack("!HH", qtype, 1) # class IN
def read_name(msg: bytes, off: int, depth: int = 0) -> tuple[str, int]:
if depth > 10:
raise ValueError("compression loop") # a malicious packet can point to itself
labels = []
while True:
if off >= len(msg):
raise ValueError("truncated name")
n = msg[off]
if n & 0xC0 == 0xC0: # compression pointer: 2 bytes, 14-bit offset
ptr = struct.unpack_from("!H", msg, off)[0] & 0x3FFF
name, _ = read_name(msg, ptr, depth + 1)
labels.append(name)
return ".".join(l for l in labels if l), off + 2
if n == 0:
return ".".join(labels), off + 1
labels.append(msg[off + 1 : off + 1 + n].decode("ascii", "replace"))
off += 1 + n
def parse_response(msg: bytes, qid: int) -> list[tuple[str, str, int, str]]:
rid, flags, qd, an, _, _ = struct.unpack_from("!HHHHHH", msg, 0)
if rid != qid:
raise ValueError("ID mismatch: not our answer")
rcode = flags & 0xF
if rcode:
raise ValueError({3: "NXDOMAIN", 2: "SERVFAIL", 5: "REFUSED"}.get(rcode, f"rcode {rcode}"))
off = 12
for _ in range(qd): # skip the echoed question
_, off = read_name(msg, off)
off += 4
answers = []
for _ in range(an):
name, off = read_name(msg, off)
rtype, _cls, ttl, rdlen = struct.unpack_from("!HHIH", msg, off)
off += 10
rdata = msg[off : off + rdlen]
if rtype == 1:
value = socket.inet_ntop(socket.AF_INET, rdata)
elif rtype == 28:
value = socket.inet_ntop(socket.AF_INET6, rdata)
elif rtype in (2, 5):
value, _ = read_name(msg, off)
else:
value = rdata.hex()
kind = next((k for k, v in TYPES.items() if v == rtype), str(rtype))
answers.append((name, kind, ttl, value))
off += rdlen
return answers
def query(name: str, qtype: str = "A", server: str = "1.1.1.1") -> list[tuple[str, str, int, str]]:
qid, packet = build_query(name, TYPES[qtype])
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
s.settimeout(3)
s.sendto(packet, (server, 53))
msg, _ = s.recvfrom(4096)
return parse_response(msg, qid)
if __name__ == "__main__":
name = sys.argv[1] if len(sys.argv) > 1 else "www.github.com"
qtype = sys.argv[2] if len(sys.argv) > 2 else "A"
for rec in query(name, qtype):
print("%-30s %-5s ttl=%-6d %s" % rec)
$ python3 labs/dnsq.py www.github.com A
www.github.com CNAME ttl=3600 github.com
github.com A ttl=60 140.82.121.4
What to notice:
struct.pack("!HHHHHH", ...):!means network byte order (big-endian),Han unsigned 16-bit int,I32-bit. This is the same layout as the DNS header diagram in the networking guide.read_namelimits recursion depth. DNS compression pointers can form loops; a naive parser recurses until it crashes. Every field read from the network is attacker-controlled: lengths, offsets, counts.- This client does not handle truncation (
TCflag → retry over TCP), EDNS, or DNSSEC. In real code, usesocket.getaddrinfo(the system resolver) ordnspython.