How to Build a DNS Resolver in Python From Scratch, Walk From the Root Servers to an Answer, and Reject Forged Replies and Planted Records

Every time you open a website, send an email, or run git push, a small program first asks a question: “what is the network address of this name?” The system that answers is the Domain Name System, or DNS. Most of us never look inside it, which is a pity, because DNS is also one of the best protocols to learn from. A query can be as small as 29 bytes, the rules fit in a handful of RFCs, and many classic lessons about parsing untrusted input show up in it.

In this tutorial you will write a DNS resolver in Python from scratch, using only the standard library. When you finish, you will have a program that starts at the root servers, follows referrals down to the server that knows the answer, and prints the result the way dig does. Along the way you will attack your own code with hostile packets, forged replies, and planted records, and then add the defenses and write tests that prove each one works.

What you will build, and what you will learn

• A message builder and parser. You will build a DNS query byte by byte, then read a reply: the header, names with compression pointers, and the common record types.
• A careful network client. It sends over UDP, refuses forged replies, falls back to TCP when an answer is cut short, and uses EDNS(0) to ask for bigger answers.
• An iterative resolver. It walks from the root servers to the answer, following referrals, glue records, name servers without glue, and CNAME aliases, with a cache that remembers where zones live.
• A defense against planted records. You will see a poisoning attack work against a trusting resolver, and stop it with a bailiwick check.
• Proof. A 47-test suite, a fuzzer that throws 20,000 damaged messages at the parser, a mutation check that plants 22 bugs to see whether the tests notice, and a comparison with dnspython on 74 live replies.

Everything here is read-only against public servers, and the hostile traffic stays on your own machine: fake servers on loopback addresses, the reserved .test top-level domain, and the documentation address ranges from RFC 5737.

How DNS answers a question, in plain language

DNS is a distributed phone book. The names are arranged in a tree, read right to left. www.example.com sits under example.com, which sits under com, which sits under the unnamed root at the top. Each level is a zone, a part of the tree run by one party, and each zone is served by one or more name servers. A server that holds the real data for a zone is authoritative for it.

Three kinds of program take part, and it helps to keep them apart:

• A stub resolver is the small client inside your operating system. It asks one question and expects a finished answer.
• A recursive resolver does the legwork for the stub. Public ones include Cloudflare’s 1.1.1.1 and Google’s 8.8.8.8, and your router or your company usually runs another.
• An authoritative server knows one zone and answers only about that zone, or says who to ask next.

The recursive resolver’s job is the one you will build. It does not know the answer, so it asks the root, which does not know either but says who handles com. That reply is a referral. The resolver then asks a com server, which refers it to the servers for example.com, and the last one answers:

your program  -> root server              "where is www.example.com?"   -> "ask these servers for com"
your program  -> a .com server           "where is www.example.com?"   -> "ask these servers for example.com"
your program  -> an example.com server   "where is www.example.com?"   -> "172.66.147.243" (an A record)

A few more terms, defined once so they never surprise you later. A record is one fact, such as “this name has this IPv4 address” (an A record) or “this name is an alias for that one” (a CNAME). Every record carries a TTL, the number of seconds anyone may remember it. Glue is an address a server hands over along with a referral, so you can reach the next server without first looking up its name. NXDOMAIN means the name does not exist, and NODATA means the name exists but has no record of the type you asked for. Finally, DNS travels in UDP datagrams on port 53 and falls back to TCP for big answers.

Prerequisites

• Python. I ran everything on Python 3.13.14 on Windows 11. Other recent Python 3 versions should work, but I only tested that one.
• Basic Python. Functions, classes, and dictionaries. Sockets and bytes are explained as they appear.
• A network that allows outbound port 53. Steps 2, 4, 6, 7, and 9, and the checks at the end, talk to public DNS servers over UDP and TCP. If your network blocks that, the local steps (3, 5, 8, and the tests) still work.
• pytest for the tests, and dnspython for the cross-check in Step 10. I used pytest 9.2.0 and dnspython 2.9.2. No administrator rights are needed.
• Loopback addresses. Steps 5 and 8 and the tests bind 127.0.0.2 and 127.0.0.66 as well as 127.0.0.1. That worked on Windows 11; I did not run it on macOS or Linux.

Create a folder and install the two packages that the tests and the cross-check need. Every file in this tutorial goes into that folder.

mkdir dnslab
cd dnslab
python -m venv .venv
.venv\Scripts\activate
python -m pip install pytest dnspython

On macOS or Linux the activation line is source .venv/bin/activate. The files you will create, in the order you meet them, are dnswire.py (messages), dnsnet.py (sending), testkit.py (fake servers), resolver.py (the walk), fuzz.py, test_dns.py, mutations.py, dig.py, and a script per step. Be kind to public infrastructure: the live scripts send from a handful to about two hundred queries each (the survey in Step 7 is the largest), so run them, do not loop them.

Step 1: Build a DNS query by hand

A DNS message has a fixed 12-byte header followed by sections. A query holds one question: a name, a record type, and a class. The header says how many entries each section has, and it carries a 16-bit ID that the reply must repeat so you can match answers to questions.

Names are the interesting part. DNS does not send example.com as text. It sends each label with a length byte in front, and ends the name with a zero byte: 07 example 03 com 00. The limits come from RFC 1035, which lists “labels 63 octets or less” and “names 255 octets or less”. Start dnswire.py with the constants, a name encoder, and a function that builds a whole query:

"""dnswire.py: build and parse DNS messages (RFC 1035, RFC 6891) with only the standard library."""
import ipaddress
import secrets
import struct
from typing import NamedTuple

TYPES = {"A": 1, "NS": 2, "CNAME": 5, "SOA": 6, "PTR": 12, "MX": 15, "TXT": 16,
         "AAAA": 28, "OPT": 41, "DS": 43, "RRSIG": 46, "NSEC": 47, "DNSKEY": 48}
TYPE_NAMES = {number: name for name, number in TYPES.items()}
RCODES = {0: "NOERROR", 1: "FORMERR", 2: "SERVFAIL", 3: "NXDOMAIN", 4: "NOTIMP", 5: "REFUSED"}
CLASS_IN = 1


class DNSError(Exception):
    """Anything wrong with a DNS message: cut off, too long, looping, or contradicting itself."""


def encode_name(name):
    """Turn www.example.com into length-prefixed labels: 03 www 07 example 03 com 00."""
    name = name.strip().rstrip(".")
    if not name:
        return b"\x00"                                   # the root has no labels, only the final zero
    try:
        ascii_name = name.encode("idna").decode("ascii")  # non-ASCII names become xn-- labels
    except UnicodeError as exc:
        raise DNSError(f"cannot encode {name!r}: {exc}") from None
    encoded = bytearray()
    for label in ascii_name.split("."):
        if not 1 <= len(label) <= 63:
            raise DNSError(f"label {label!r} must be 1 to 63 bytes long")
        encoded.append(len(label))
        encoded += label.encode("ascii")
    if len(encoded) + 1 > 255:
        raise DNSError("a name takes at most 255 bytes on the wire")
    return bytes(encoded) + b"\x00"


def build_query(name, qtype="A", *, recursion_desired=True, edns_size=None, dnssec_ok=False, query_id=None):
    """Return (query_id, wire bytes) for a message with one question."""
    if query_id is None:
        query_id = secrets.randbits(16)                   # unpredictable, so a forger has to guess it
    flags = 0x0100 if recursion_desired else 0x0000       # the RD bit
    extra = 1 if edns_size else 0                         # one OPT record in the additional section
    header = struct.pack(">HHHHHH", query_id, flags, 1, 0, 0, extra)
    question = encode_name(name) + struct.pack(">HH", TYPES[qtype], CLASS_IN)
    opt = b""
    if edns_size:
        # OPT pseudo-record (RFC 6891): root name, type 41, "class" = our UDP size, "TTL" = flags (DO is the top bit)
        opt = b"\x00" + struct.pack(">HHIH", TYPES["OPT"], edns_size, 0x8000 if dnssec_ok else 0, 0)
    return query_id, header + question + opt

Three details are worth a pause. The ID comes from secrets, not random, because a forger who can guess the ID gets one step closer to fooling you; Step 5 shows how many guesses that costs. The encode("idna") call turns an internationalized name into the ASCII xn-- form that DNS actually carries, and it leaves ordinary ASCII names, including their capitalization and underscores, alone. And the optional OPT record, which only appears when you pass edns_size, is the EDNS(0) extension from RFC 6891 that lets a client say how large a UDP reply it can take. Step 6 uses it. Now run a script that prints the bytes of a query for example.com, with a fixed ID so the output is the same every time:

"""step01_build_query.py: build the 29 bytes that ask "what is the A record of example.com?"."""
from dnswire import build_query

query_id, wire = build_query("example.com", "A", query_id=0x1234)
print(f"{len(wire)} bytes in total")
print("header   :", wire[:12].hex(" "))
print("name     :", wire[12:25].hex(" "))
print("type, cls:", wire[25:].hex(" "))
print("the name as text:", wire[12:25])

python step01_build_query.py

29 bytes in total
header   : 12 34 01 00 00 01 00 00 00 00 00 00
name     : 07 65 78 61 6d 70 6c 65 03 63 6f 6d 00
type, cls: 00 01 00 01
the name as text: b'\x07example\x03com\x00'

Read the output like a map. The 12 header bytes are the ID (12 34), the flags (01 00, which sets only the “recursion desired” bit), and four counts: one question and nothing else. The name is 13 bytes: 07, the seven letters of example, 03, com, and the closing 00. The last four bytes are the type A (00 01) and the class IN, the Internet (00 01). That makes 29 bytes in total, and it is the whole question.

Common mistakes. Forgetting the final zero byte, using a label longer than 63 bytes, and passing a name with a trailing dot to code that does not expect one can produce queries that servers drop or answer with an error. How to verify: the test suite in Step 10 compares this query against a hand-assembled hex string.

Step 2: Send it and read the header of the reply

Now put those 29 bytes on the wire. A UDP socket needs no connection: you hand it a datagram and an address. You will ask a recursive resolver, so the “recursion desired” bit we set is meaningful. Any public resolver works; this one uses 1.1.1.1.

First add the header parser to dnswire.py. The flags field packs five booleans, a 4-bit opcode, and a 4-bit response code into 16 bits, so the parser uses shifts and masks to pull them apart:

# dnswire.py (continued)
class Header(NamedTuple):
    id: int
    qr: bool       # True for a response
    opcode: int
    aa: bool       # authoritative answer
    tc: bool       # truncated: the full answer did not fit
    rd: bool       # recursion desired
    ra: bool       # recursion available
    rcode: int
    counts: tuple  # questions, answers, authority records, additional records

    def flags_text(self):
        flags = [("qr", self.qr), ("aa", self.aa), ("tc", self.tc), ("rd", self.rd), ("ra", self.ra)]
        return " ".join(name for name, on in flags if on)


def parse_header(msg):
    if len(msg) < 12:
        raise DNSError(f"{len(msg)} bytes is too short for a DNS header (12 bytes)")
    ident, flags, qd, an, ns, ar = struct.unpack(">HHHHHH", msg[:12])
    return Header(ident, bool(flags & 0x8000), (flags >> 11) & 0xF, bool(flags & 0x0400), bool(flags & 0x0200),
                  bool(flags & 0x0100), bool(flags & 0x0080), flags & 0xF, (qd, an, ns, ar))

"""step02_send_query.py: send the query to a recursive resolver over UDP and read the 12 byte header of the reply."""
import socket

from dnswire import build_query, parse_header

query_id, wire = build_query("example.com", "A")
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
    sock.settimeout(3)
    sock.sendto(wire, ("1.1.1.1", 53))
    reply, sender = sock.recvfrom(4096)
print("sent", len(wire), "bytes, received", len(reply), "bytes from", sender[0])
header = parse_header(reply)
print("same id as our query:", header.id == query_id)
print("flags:", header.flags_text(), "| opcode:", header.opcode, "| rcode:", header.rcode)
print("questions, answers, authority, additional:", header.counts)
print("raw header:", reply[:12].hex(" "))

python step02_send_query.py

sent 29 bytes, received 61 bytes from 1.1.1.1
same id as our query: True
flags: qr rd ra | opcode: 0 | rcode: 0
questions, answers, authority, additional: (1, 2, 0, 0)
raw header: 6c b9 81 80 00 01 00 02 00 00 00 00

The reply is longer than the query, and its header repeats our ID. The flags qr rd ra mean: this is a response (qr), it repeats that we asked for recursion (rd), and the server offers recursion (ra). The response code 0 is NOERROR. The four counts say the reply echoes our one question and holds some answer records; the exact number depends on the name you ask about. If the script times out instead, your network is probably blocking UDP port 53, and the live steps will not work until that changes.

Step 3: Read names, including compressed ones, and survive hostile packets

To read the answer you must read names, and DNS names have a twist. RFC 1035 explains it: “In order to reduce the size of messages, the domain system utilizes a compression scheme which eliminates the repetition of domain names in a message.” Instead of repeating example.com in every record, a message can point back at an earlier copy. A pointer is two bytes: the top two bits are 11 (so a length byte of 0xC0 or higher means “pointer”), and the other 14 bits are an offset from the start of the message.

That makes name parsing the most security-sensitive code in a resolver. The packet comes from a stranger, and the pointers are instructions the stranger wrote. A pointer that points at itself, or two pointers that point at each other, can trap a naive reader in an endless loop. The reader below follows three rules, each of which stops a different trick:

1. A pointer must point backwards, and never into the 12-byte header. That kills self-pointers and forward loops.
2. A name may use at most 16 jumps. Legitimate names normally need one or two, and every live reply in Step 10 parsed under this limit, so it stops long chains that would make you do a lot of work for each name.
3. A decoded name may not exceed the 255-byte limit, which stops names built from many long labels.

It also prints names safely. DNS labels can hold any byte, including newlines and escape codes, and a name that you log or print is text an attacker chose. The escape helper turns anything unusual into a backslash and three digits, the same convention dig uses, so a hostile name cannot forge a log line. Add this to dnswire.py:

# dnswire.py (continued)
MAX_NAME = 255    # the longest name on the wire, counting every length byte and the final zero
MAX_JUMPS = 16    # compression pointers we follow while reading one name


def escape(raw, specials=b"\\", space_ok=False):
    """Make bytes safe to print: anything unusual becomes a backslash and three decimal digits, as dig does."""
    out = []
    for byte in raw:
        if byte in specials:
            out.append("\\" + chr(byte))
        elif 0x21 <= byte < 0x7F or (space_ok and byte == 0x20):
            out.append(chr(byte))
        else:
            out.append(f"\\{byte:03d}")
    return "".join(out)


def under(name, zone):
    """True when name is zone itself or lies below it. The root zone is the empty string."""
    return zone == "" or name == zone or name.endswith("." + zone)


def read_name(msg, pos):
    """Read a name that may use compression. Returns (name, position where the caller continues)."""
    labels = []
    resume = None     # set at the first pointer: the caller carries on right after the two pointer bytes
    size = 1          # the final zero byte
    jumps = 0
    while True:
        if pos >= len(msg):
            raise DNSError("a name runs past the end of the message")
        length = msg[pos]
        if length == 0:
            pos += 1
            break
        if length & 0xC0 == 0xC0:
            if pos + 1 >= len(msg):
                raise DNSError("a compression pointer is cut off")
            target = ((length & 0x3F) << 8) | msg[pos + 1]
            if target < 12:
                raise DNSError("a compression pointer points into the header")
            if target >= pos:
                raise DNSError("a compression pointer must point backwards")
            jumps += 1
            if jumps > MAX_JUMPS:
                raise DNSError(f"more than {MAX_JUMPS} compression pointers in one name")
            if resume is None:
                resume = pos + 2
            pos = target
            continue
        if length & 0xC0:
            raise DNSError("unknown label type")
        if pos + 1 + length > len(msg):
            raise DNSError("a label runs past the end of the message")
        size += 1 + length
        if size > MAX_NAME:
            raise DNSError(f"a name may not be longer than {MAX_NAME} bytes")
        labels.append(escape(msg[pos + 1:pos + 1 + length], b".\\"))
        pos += 1 + length
    return ".".join(labels), pos if resume is None else resume

Now try it on a tiny hand-made message, and then on hostile ones. The script also runs a deliberately naive reader, with none of the three rules, in a separate process with a one-second limit so that the loop cannot hang your terminal:

"""step03_names.py: read compressed names, then see what hostile packets do to a naive reader."""
import subprocess
import sys

from dnswire import DNSError, read_name

header = b"\x00" * 12
# "example.com" at offset 12, then "www" followed by a pointer back to offset 12
msg = header + b"\x07example\x03com\x00" + b"\x03www\xc0\x0c"
print("name at offset 12:", read_name(msg, 12))
print("name at offset 25:", read_name(msg, 25))

NAIVE = """
import sys
msg = bytes.fromhex(sys.argv[1])
pos, labels = 12, 0
while msg[pos]:
    if msg[pos] & 0xC0 == 0xC0:
        pos = ((msg[pos] & 0x3F) << 8) | msg[pos + 1]
    else:
        labels += 1
        pos += 1 + msg[pos]
print("finished after", labels, "labels")
"""

HOSTILE = {
    "a pointer to itself": b"\xc0\x0c",
    "a label, then a pointer back to it": b"\x01a\xc0\x0c",
    "a pointer that points ahead": b"\xc0\x0e\x00",
    "a pointer into the header": b"\xc0\x04",
    "a label type that does not exist": bytes([0x41]) + b"a" * 65 + b"\x00",
    "a label longer than the message": b"\x3fabc",
    "five 63 byte labels": (b"\x3f" + b"a" * 63) * 5 + b"\x00",
}
for title, tail in list(HOSTILE.items())[:2]:
    try:
        done = subprocess.run([sys.executable, "-c", NAIVE, (header + tail).hex()],
                              capture_output=True, text=True, timeout=1)
        print(f"naive reader, {title}: {done.stdout.strip()}")
    except subprocess.TimeoutExpired:
        print(f"naive reader, {title}: still running after 1 s, killed")
print()
for title, tail in HOSTILE.items():
    try:
        read_name(header + tail, 12)
        print(f"read_name, {title}: accepted")
    except DNSError as exc:
        print(f"read_name, {title}: refused ({exc})")

python step03_names.py

name at offset 12: ('example.com', 25)
name at offset 25: ('www.example.com', 31)
naive reader, a pointer to itself: still running after 1 s, killed
naive reader, a label, then a pointer back to it: still running after 1 s, killed

read_name, a pointer to itself: refused (a compression pointer must point backwards)
read_name, a label, then a pointer back to it: refused (more than 16 compression pointers in one name)
read_name, a pointer that points ahead: refused (a compression pointer must point backwards)
read_name, a pointer into the header: refused (a compression pointer points into the header)
read_name, a label type that does not exist: refused (unknown label type)
read_name, a label longer than the message: refused (a label runs past the end of the message)
read_name, five 63 byte labels: refused (a name may not be longer than 255 bytes)

The first two lines show compression working: the name at offset 25 is www plus a pointer to offset 12, and it decodes to www.example.com. The naive reader never finishes on either hostile packet, and the careful reader refuses all seven with a specific reason. Notice that the second case, a label followed by a pointer back to it, trips the jump limit rather than the pointer-direction rule, because that pointer does point backwards. The rules overlap on purpose. Step 10 plants a bug in each rule and checks that a test fails, which is how you learn that every rule is really covered.

Step 4: Read records and print them the way dig does

After the header and question come three sections of resource records: the answers, the authority records (usually the name servers or the SOA record of the zone), and the additional records (usually glue). Each record has an owner name, a type, a class, a TTL, and type-specific data. Add the record parser to dnswire.py:

# dnswire.py (continued)
class Question(NamedTuple):
    name: str
    qtype: int
    qclass: int


class Record(NamedTuple):
    name: str
    rtype: int
    rclass: int
    ttl: int
    data: str      # the record data as text, the way dig prints it


def format_rdata(msg, pos, end, rtype):
    """Turn the data of one record into text. Types we do not know become RFC 3597 hex."""
    size = end - pos
    if rtype == TYPES["A"]:
        if size != 4:
            raise DNSError("an A record holds exactly 4 bytes")
        return str(ipaddress.IPv4Address(msg[pos:end]))
    if rtype == TYPES["AAAA"]:
        if size != 16:
            raise DNSError("an AAAA record holds exactly 16 bytes")
        return str(ipaddress.IPv6Address(msg[pos:end]))
    if rtype in (TYPES["NS"], TYPES["CNAME"], TYPES["PTR"]):
        name, after = read_name(msg, pos)
        if after != end:
            raise DNSError("the name does not fill the record data")
        return name
    if rtype == TYPES["MX"]:
        if size < 3:
            raise DNSError("an MX record needs a preference and a name")
        name, after = read_name(msg, pos + 2)
        if after != end:
            raise DNSError("the name does not fill the record data")
        return f"{int.from_bytes(msg[pos:pos + 2], 'big')} {name or '.'}"
    if rtype == TYPES["TXT"]:
        strings = []
        while pos < end:
            length = msg[pos]
            if pos + 1 + length > end:
                raise DNSError("a TXT string runs past the record data")
            strings.append('"' + escape(msg[pos + 1:pos + 1 + length], b'\\"', space_ok=True) + '"')
            pos += 1 + length
        return " ".join(strings)
    if rtype == TYPES["SOA"]:
        primary, pos = read_name(msg, pos)
        mailbox, pos = read_name(msg, pos)
        if end - pos != 20:
            raise DNSError("an SOA record ends with five 32-bit numbers")
        numbers = struct.unpack(">IIIII", msg[pos:end])
        return f"{primary or '.'} {mailbox or '.'} " + " ".join(str(n) for n in numbers)
    return f"\\# {size} {msg[pos:end].hex()}"


def read_record(msg, pos):
    """Read one resource record. Returns (Record, position after it)."""
    name, pos = read_name(msg, pos)
    if pos + 10 > len(msg):
        raise DNSError("a record header is cut off")
    rtype, rclass, ttl, size = struct.unpack(">HHIH", msg[pos:pos + 10])
    pos += 10
    if pos + size > len(msg):
        raise DNSError("the record data runs past the end of the message")
    if ttl & 0x80000000 and rtype != TYPES["OPT"]:
        ttl = 0       # RFC 2181 section 8: a TTL with the top bit set counts as zero
    return Record(name, rtype, rclass, ttl, format_rdata(msg, pos, pos + size, rtype)), pos + size

Three decisions here are worth knowing. First, each record type has its own rule for what its data must look like, and the parser checks it: an A record is exactly 4 bytes, a name inside the data must end exactly where the data ends, and unknown types fall back to the hex form from RFC 3597 instead of crashing. Second, RFC 2181 says “Implementations should treat TTL values received with the most significant bit set as if the entire value received was zero”, so the parser does that. Third, escape is used again for TXT strings, which can contain arbitrary bytes. Now the message-level parser, which ties everything together and refuses a message with bytes left over:

# dnswire.py (continued)
class Message(NamedTuple):
    header: Header
    questions: list
    answers: list
    authority: list
    additional: list
    edns: object   # None, or (udp_size, extended_rcode, version, dnssec_ok)

    @property
    def rcode(self):
        return self.header.rcode | ((self.edns[1] << 4) if self.edns else 0)


def parse_message(msg):
    """Parse a whole message. Anything malformed raises DNSError, and nothing else."""
    header = parse_header(msg)
    pos = 12
    questions = []
    for _ in range(header.counts[0]):
        name, pos = read_name(msg, pos)
        if pos + 4 > len(msg):
            raise DNSError("a question is cut off")
        qtype, qclass = struct.unpack(">HH", msg[pos:pos + 4])
        questions.append(Question(name, qtype, qclass))
        pos += 4
    sections = []
    for count in header.counts[1:]:
        records = []
        for _ in range(count):
            record, pos = read_record(msg, pos)
            records.append(record)
        sections.append(records)
    if pos != len(msg):
        raise DNSError(f"{len(msg) - pos} unexpected bytes after the last record")
    edns = None
    for record in sections[2]:
        if record.rtype == TYPES["OPT"]:
            edns = (record.rclass, record.ttl >> 24, (record.ttl >> 16) & 0xFF, bool(record.ttl & 0x8000))
    return Message(header, questions, *sections, edns)


def type_name(rtype):
    return TYPE_NAMES.get(rtype, f"TYPE{rtype}")


def format_record(record):
    return f"{record.name + '.':<30} {record.ttl:<7} IN  {type_name(record.rtype):<6} {record.data}"

Notice the promise in the docstring: parse_message raises DNSError and nothing else. That is a contract you will test with a fuzzer in Step 10. Callers need only one except clause, and a malformed packet cannot crash the program with a stray IndexError. Now ask a recursive resolver seven questions and print the answers:

"""step04_dig.py: ask a recursive resolver seven questions and print the answers the way dig does."""
import socket

from dnswire import RCODES, build_query, format_record, parse_message

RESOLVER = "1.1.1.1"


def lookup(name, qtype):
    query_id, wire = build_query(name, qtype)
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
        sock.settimeout(3)
        sock.sendto(wire, (RESOLVER, 53))
        data, _ = sock.recvfrom(4096)
    message = parse_message(data)
    assert message.header.id == query_id
    return message, len(data)


QUESTIONS = [("example.com", "A"), ("example.com", "AAAA"), ("gmail.com", "MX"),
             ("example.com", "TXT"), ("example.com", "SOA"), ("no-such-name-8c1d.example.com", "A"),
             ("this-name-does-not-exist-8c1d.org", "A")]
for name, qtype in QUESTIONS:
    message, size = lookup(name, qtype)
    print(f";; {name} {qtype}: {RCODES[message.rcode]}, {size} bytes, flags: {message.header.flags_text()}")
    for record in message.answers + message.authority:
        print(format_record(record))
    print()

python step04_dig.py

;; example.com A: NOERROR, 61 bytes, flags: qr rd ra
example.com.                   25      IN  A      104.20.23.154
example.com.                   25      IN  A      172.66.147.243

;; example.com AAAA: NOERROR, 85 bytes, flags: qr rd ra
example.com.                   17      IN  AAAA   2606:4700:10::6814:179a
example.com.                   17      IN  AAAA   2606:4700:10::ac42:93f3

;; gmail.com MX: NOERROR, 150 bytes, flags: qr rd ra
gmail.com.                     755     IN  MX     20 alt2.gmail-smtp-in.l.google.com
gmail.com.                     755     IN  MX     30 alt3.gmail-smtp-in.l.google.com
gmail.com.                     755     IN  MX     10 alt1.gmail-smtp-in.l.google.com
gmail.com.                     755     IN  MX     40 alt4.gmail-smtp-in.l.google.com
gmail.com.                     755     IN  MX     5 gmail-smtp-in.l.google.com

;; example.com TXT: NOERROR, 98 bytes, flags: qr rd ra
example.com.                   244     IN  TXT    "v=spf1 -all"
example.com.                   244     IN  TXT    "_k2n1y4vw3qtb4skdx9e7dxt97qrmmq9"

;; example.com SOA: NOERROR, 91 bytes, flags: qr rd ra
example.com.                   1800    IN  SOA    elliott.ns.cloudflare.com dns.cloudflare.com 2418004112 10000 2400 604800 1800

;; no-such-name-8c1d.example.com A: NOERROR, 109 bytes, flags: qr rd ra
example.com.                   1800    IN  SOA    elliott.ns.cloudflare.com dns.cloudflare.com 2418004112 10000 2400 604800 1800

;; this-name-does-not-exist-8c1d.org A: NXDOMAIN, 122 bytes, flags: qr rd ra
org.                           3600    IN  SOA    v0n0.pir-ns.org hostmaster.donuts.email 1793997275 7200 900 1209600 3600

Your addresses and TTLs will differ, because DNS data changes and TTLs count down, but the shape should match. Two things in my output are worth studying. There are two TXT records for example.com, each printed as a quoted string; a single TXT record can also hold several such strings, and the parser joins them with spaces. And the sixth question, for a name under example.com that does not exist, came back from 1.1.1.1 as NOERROR with an SOA record in the authority section. That is NODATA, not NXDOMAIN, even though the name really does not exist. The seventh question, under .org, returns the NXDOMAIN you would expect. Step 7 explains the difference.

Step 5: Send queries safely, and refuse forged replies

UDP has no handshake. Anyone who can send a datagram to your socket can claim to be the server, and your resolver has to decide which datagram to believe. RFC 5452, titled “DNS Resilience against Forged Answers”, lists what a forger has to get right: the question, the ID, the source address, and the destination port. Your job is to make each of those hard to guess or easy to check.

To practise this safely you need a server you control. testkit.py has two tools: encode_message, which writes DNS messages (with name compression, so it doubles as a lesson in how compression is produced), and FakeServer, a small UDP server on a loopback address that lets a handler function decide what to send back, including several replies at once:

"""testkit.py: write DNS messages (with name compression) and run tiny fake name servers on the loopback."""
import ipaddress
import socket
import struct
import threading

from dnswire import CLASS_IN, TYPE_NAMES, TYPES, DNSError, parse_message, under


def encode_message(ident, flags, questions, answers=(), authority=(), additional=(), compress=True):
    """questions: (name, type name). Records: (name, type name, ttl, data). Returns the wire bytes."""
    out = bytearray(struct.pack(">HHHHHH", ident, flags, len(questions), len(answers), len(authority), len(additional)))
    seen = {}                                   # name suffix (lower case labels) -> offset where it first appeared

    def put_name(name):
        labels = [part for part in name.rstrip(".").split(".") if part]
        for i in range(len(labels)):
            key = tuple(label.lower() for label in labels[i:])
            if compress and key in seen:
                out.extend(struct.pack(">H", 0xC000 | seen[key]))     # a pointer ends the name
                return
            if len(out) < 0x4000:                                      # a pointer holds 14 bits of offset
                seen.setdefault(key, len(out))
            raw = labels[i].encode("ascii")
            out.append(len(raw))
            out.extend(raw)
        out.append(0)

    for name, qtype in questions:
        put_name(name)
        out.extend(struct.pack(">HH", TYPES[qtype], CLASS_IN))
    for section in (answers, authority, additional):
        for name, rtype, ttl, data in section:
            put_name(name)
            out.extend(struct.pack(">HHI", TYPES[rtype], CLASS_IN, ttl))
            length_at = len(out)
            out.extend(b"\x00\x00")             # the data length, filled in below
            if rtype == "A":
                out.extend(ipaddress.IPv4Address(data).packed)
            elif rtype == "AAAA":
                out.extend(ipaddress.IPv6Address(data).packed)
            elif rtype in ("NS", "CNAME", "PTR"):
                put_name(data)
            elif rtype == "MX":
                out.extend(struct.pack(">H", data[0]))
                put_name(data[1])
            elif rtype == "TXT":
                for chunk in data:
                    raw = chunk.encode("latin-1")
                    out.append(len(raw))
                    out.extend(raw)
            elif rtype == "SOA":
                put_name(data[0])
                put_name(data[1])
                out.extend(struct.pack(">IIIII", *data[2:]))
            else:
                out.extend(data)
            struct.pack_into(">H", out, length_at, len(out) - length_at - 2)
    return bytes(out)


class FakeServer:
    """A UDP name server on ip:port. handler(query, transport, addr) returns bytes, a list of bytes, or None."""

    def __init__(self, ip, port, handler):
        self.handler = handler
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock.bind((ip, port))
        self.sock.settimeout(0.2)
        self.running = True
        self.thread = threading.Thread(target=self._loop, daemon=True)
        self.thread.start()

    def _loop(self):
        while self.running:
            try:
                data, addr = self.sock.recvfrom(4096)
            except socket.timeout:
                continue
            except ConnectionResetError:
                continue                         # Windows reports an earlier ICMP error on the next receive
            except OSError:
                return
            try:
                replies = self.handler(parse_message(data), "udp", addr)
            except DNSError:
                continue
            for wire in ([replies] if isinstance(replies, bytes) else replies or []):
                self.sock.sendto(wire, addr)

    def stop(self):
        self.running = False
        self.thread.join(1)
        self.sock.close()

Next comes dnsnet.py, the function that sends one question and accepts one reply. Read check_reply and ask_udp as a list of defenses, from the outside in:

• sock.connect(...) on a UDP socket does not open a connection, but it tells the kernel to discard datagrams from any other address or port. The Linux man page for connect(2) says that for a datagram socket the address is “the address to which datagrams are sent by default, and the only address from which datagrams are received.”
• A random 16-bit ID, plus a random source port that we pick ourselves in bind_random_port, so a forger has to guess both.
• check_reply then demands that the ID matches, that the message is a response and not a query, that it has exactly one question (RFC 9619 narrows QDCOUNT to a maximum of one), and that the question repeats ours.
• The loop keeps waiting after it discards a bad datagram, until a deadline. A forgery can therefore not end the exchange early; it can only be ignored.

"""dnsnet.py: send one DNS question to one server and accept only a reply that really answers it."""
import secrets
import socket
import struct
import time
from typing import NamedTuple

from dnswire import CLASS_IN, TYPES, DNSError, build_query, parse_message


class QueryFailed(Exception):
    """No acceptable reply: a timeout, a connection problem, or every datagram failed the checks."""


class Reply(NamedTuple):
    message: object
    size: int         # bytes on the wire
    transport: str    # "udp" or "tcp"
    seconds: float
    rejected: list    # why each datagram we discarded while waiting was discarded


def randomize_case(name):
    """0x20 encoding: random upper and lower case letters. An honest server echoes the question unchanged."""
    return "".join(c.upper() if c.isalpha() and secrets.randbits(1) else c.lower() for c in name)


def check_reply(message, query_id, name, qtype, exact_case=False):
    """Return the reason this message cannot be the answer to our question, or None if it can."""
    if message.header.id != query_id:
        return "wrong id"
    if not message.header.qr:
        return "not a response"
    if len(message.questions) != 1:
        return "expected exactly one question"
    question = message.questions[0]
    same_name = question.name == name if exact_case else question.name.lower() == name.lower()
    if not same_name or question.qtype != qtype or question.qclass != CLASS_IN:
        return "question does not match"
    return None


def bind_random_port(sock, family):
    """Pick our own source port in the 49152 to 65535 range instead of trusting the OS to randomize it."""
    for _ in range(20):
        port = 49152 + secrets.randbelow(65536 - 49152)
        try:
            sock.bind(("::" if family == socket.AF_INET6 else "0.0.0.0", port))
            return
        except OSError:
            continue                           # that port is busy: try another one
    # all twenty were busy: let connect() below choose


def prepare(name, qtype, recursion, edns_size, dnssec_ok, use_0x20):
    name = name.rstrip(".")
    sent_name = randomize_case(name) if use_0x20 else name
    query_id, wire = build_query(sent_name, qtype, recursion_desired=recursion,
                                 edns_size=edns_size, dnssec_ok=dnssec_ok)
    return sent_name, query_id, wire


def ask_udp(server, name, qtype="A", *, port=53, recursion=False, edns_size=None, dnssec_ok=False,
            use_0x20=False, timeout=2.0):
    """Ask one question over UDP and wait up to timeout seconds for a reply that passes check_reply."""
    sent_name, query_id, wire = prepare(name, qtype, recursion, edns_size, dnssec_ok, use_0x20)
    family = socket.AF_INET6 if ":" in server else socket.AF_INET
    rejected = []
    started = time.monotonic()
    try:
        with socket.socket(family, socket.SOCK_DGRAM) as sock:
            bind_random_port(sock, family)
            sock.connect((server, port))   # from now on the kernel drops datagrams from any other address or port
            sock.send(wire)
            while True:
                left = timeout - (time.monotonic() - started)
                if left <= 0:
                    detail = f" (discarded: {', '.join(rejected)})" if rejected else ""
                    raise QueryFailed(f"no acceptable reply from {server} within {timeout} s{detail}")
                sock.settimeout(left)
                try:
                    data = sock.recv(65535)
                except socket.timeout:
                    continue
                try:
                    message = parse_message(data)
                except DNSError as exc:
                    rejected.append(f"unparseable: {exc}")
                    continue
                problem = check_reply(message, query_id, sent_name, TYPES[qtype], use_0x20)
                if problem:
                    rejected.append(problem)
                    continue
                return Reply(message, len(data), "udp", time.monotonic() - started, rejected)
    except OSError as exc:
        raise QueryFailed(f"{server} port {port}: {exc}") from None

Now attack it. The script below starts a fake server whose handler is the attacker: for every query it sends a forgery from a second socket (right ID and question, wrong source port), then four more forgeries with a wrong ID, a wrong question, a query flag instead of a response flag, and plain garbage, and only then the real answer. It also runs a careless client that takes the first datagram it receives. Finally it measures where the operating system puts source ports:

"""step05_spoof.py: forge replies from the same machine and see which ones ask_udp throws away."""
import socket

from dnsnet import ask_udp, bind_random_port
from dnswire import build_query, parse_message
from testkit import FakeServer, encode_message

PORT = 15353
REAL, FORGED = "192.0.2.1", "6.6.6.6"
attacker = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)    # a second socket, so a different source port


def reply(query, ip, *, ident=None, flags=0x8400, name=None):
    asked = name or query.questions[0].name
    ident = query.header.id if ident is None else ident
    return encode_message(ident, flags, [(asked, "A")], [(asked, "A", 60, ip)])


def handler(query, transport, addr):
    attacker.sendto(reply(query, FORGED), addr)                  # everything right, but from the wrong port
    return [
        reply(query, FORGED, ident=(query.header.id + 1) & 0xFFFF),   # wrong id
        reply(query, FORGED, name="other.example"),                   # wrong question
        reply(query, FORGED, flags=0x0100),                           # a query, not a response
        b"\x00\x01 this is not DNS",
        reply(query, REAL),
    ]


server = FakeServer("127.0.0.1", PORT, handler)
try:
    got = ask_udp("127.0.0.1", "www.example.com", port=PORT)
    print("careful client accepted:", got.message.answers[0].data)
    for reason in got.rejected:
        print("  discarded:", reason)
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:     # a client with no checks at all
        sock.settimeout(2)
        sock.sendto(build_query("www.example.com", "A", recursion_desired=False)[1], ("127.0.0.1", PORT))
        first, _ = sock.recvfrom(4096)
    print("careless client accepted:", parse_message(first).answers[0].data)
finally:
    server.stop()


def source_ports(pick_ourselves):
    ports = []
    for _ in range(200):
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
            if pick_ourselves:
                bind_random_port(sock, socket.AF_INET)
            sock.connect(("127.0.0.1", PORT))
            ports.append(sock.getsockname()[1])
    return ports


for label, pick in (("the OS chooses", False), ("we choose", True)):
    ports = source_ports(pick)
    steps = sum(b == a + 1 for a, b in zip(ports, ports[1:]))
    print(f"{label}: 200 sockets, {len(set(ports))} different ports, "
          f"{steps} of 199 were the previous port plus one, range {min(ports)} to {max(ports)}")
letters = sum(c.isalpha() for c in "www.example.com")
span = 65535 - 49152 + 1
print(f"\nguesses for a forger: id only {2**16:,}")
print(f"  id and a source port out of {span:,}: {2**16 * span:,}")
print(f"  and 0x20 case bits on {letters} letters: {2**16 * span * 2**letters:,}")

python step05_spoof.py

careful client accepted: 192.0.2.1
  discarded: wrong id
  discarded: question does not match
  discarded: not a response
  discarded: unparseable: unknown label type
careless client accepted: 6.6.6.6
the OS chooses: 200 sockets, 200 different ports, 199 of 199 were the previous port plus one, range 61751 to 61950
we choose: 200 sockets, 199 different ports, 0 of 199 were the previous port plus one, range 49174 to 65423

guesses for a forger: id only 65,536
  id and a source port out of 16,384: 1,073,741,824
  and 0x20 case bits on 13 letters: 8,796,093,022,208

The careful client discarded all four bad datagrams that reached it, never saw the forgery from the second socket (the kernel dropped it because of the wrong source port), and returned the real answer, 192.0.2.1. The careless client accepted 6.6.6.6. That difference, which is only a few lines of checking, is a large part of why a resolver can work on an open network.

The last lines of output hold a surprise. On this Windows 11 machine, 200 sockets opened one after another got 200 consecutive source ports: each was the previous port plus one. A forger who learns one port, for example by getting you to query a server they control, can predict the ports of the queries that follow. That is why bind_random_port exists: it picks its own port from the 49152 to 65535 range, which is the dynamic range that netsh int ipv4 show dynamicport udp reports here (16,384 ports). RFC 5452 explains the arithmetic: “If multiple ports are used for sending queries, this enlarges the effective ID space by a factor equal to the number of ports used.” With 65,536 IDs and 16,384 ports, a blind forger faces 1,073,741,824 combinations instead of 65,536. Other operating systems may behave differently, so run the measurement on yours before you trust the default.

The last printed line mentions a third idea, 0x20 encoding, from an Internet-Draft by Vixie and others. The draft’s abstract says “The small (16-bit) size of the DNS transaction ID has made it a frequent target for forgery” and proposes randomizing the capitalization of the query name. Servers are supposed to echo the question unchanged, so a reply with different capitalization is suspect. ask_udp supports it through use_0x20, but it is off by default. Step 7 shows why.

Step 6: Handle truncated answers with TCP and EDNS

RFC 1035 says “Messages carried by UDP are restricted to 512 bytes (not counting the IP or UDP headers). Longer messages are truncated and the TC bit is set in the header.” Two mechanisms make larger answers possible. With EDNS(0) (RFC 6891), the client adds the OPT record you met in Step 1 to say how large a UDP reply it will accept; the RFC adds that “Values lower than 512 MUST be treated as equal to 512.” And when an answer still does not fit, the client retries over TCP, where every message is preceded by a two-byte length. RFC 7766 is blunt about it: “All general-purpose DNS implementations MUST support both UDP and TCP transport.”

What size should you advertise? The DNS Flag Day 2020 recommendation is “An EDNS buffer size of 1232 bytes will avoid fragmentation on nearly all current networks.” Add the TCP code and a wrapper that falls back to it to dnsnet.py:

# dnsnet.py (continued)
def recv_exact(sock, count, deadline):
    """Read exactly count bytes from a TCP socket before the deadline (a time.monotonic() value)."""
    data = b""
    while len(data) < count:
        left = deadline - time.monotonic()
        if left <= 0:
            raise QueryFailed("timed out waiting for the TCP reply")
        sock.settimeout(left)
        chunk = sock.recv(count - len(data))
        if not chunk:
            raise QueryFailed("the server closed the connection early")
        data += chunk
    return data


def ask_tcp(server, name, qtype="A", *, port=53, recursion=False, edns_size=None, dnssec_ok=False,
            use_0x20=False, timeout=2.0):
    """Ask over TCP, where every message is preceded by a two byte length (RFC 7766)."""
    sent_name, query_id, wire = prepare(name, qtype, recursion, edns_size, dnssec_ok, use_0x20)
    started = time.monotonic()
    deadline = started + timeout
    try:
        with socket.create_connection((server, port), timeout=timeout) as sock:
            sock.sendall(struct.pack(">H", len(wire)) + wire)
            size = struct.unpack(">H", recv_exact(sock, 2, deadline))[0]
            data = recv_exact(sock, size, deadline)
    except OSError as exc:
        raise QueryFailed(f"{server} port {port}: {exc}") from None
    try:
        message = parse_message(data)
    except DNSError as exc:
        raise QueryFailed(f"{server} sent a broken TCP reply: {exc}") from None
    problem = check_reply(message, query_id, sent_name, TYPES[qtype], use_0x20)
    if problem:
        raise QueryFailed(f"{server} sent a bad TCP reply: {problem}")
    return Reply(message, len(data), "tcp", time.monotonic() - started, [])


def ask(server, name, qtype="A", *, allow_tcp=True, **options):
    """UDP first. If the server says the answer was cut short, ask again over TCP."""
    reply = ask_udp(server, name, qtype, **options)
    if reply.message.header.tc and allow_tcp:
        return ask_tcp(server, name, qtype, **options)
    return reply

Now an experiment. The root zone’s own DNSKEY records make a good test, because they are big. The script asks a root server for them five ways: without EDNS, with a 512-byte buffer, with 1232 bytes, and with the DO bit (“DNSSEC OK”), which asks the server to include signatures:

"""step06_truncation.py: ask a root server for its DNSKEY records with different EDNS settings."""
from dnsnet import ask, ask_udp
from dnswire import TYPES

SERVER = "198.41.0.4"    # a.root-servers.net
print(f"{'EDNS size':>10} {'DO':>4} {'UDP bytes':>10} {'TC':>4} {'answers':>8} {'RRSIG':>6}   with TCP fallback")
for edns_size, do in [(None, False), (512, False), (1232, False), (1232, True), (4096, True)]:
    udp = ask_udp(SERVER, ".", "DNSKEY", edns_size=edns_size, dnssec_ok=do)
    final = ask(SERVER, ".", "DNSKEY", edns_size=edns_size, dnssec_ok=do)
    signatures = sum(r.rtype == TYPES["RRSIG"] for r in udp.message.answers)
    print(f"{edns_size or 'none':>10} {'yes' if do else 'no':>4} {udp.size:>10} "
          f"{'yes' if udp.message.header.tc else 'no':>4} {len(udp.message.answers):>8} {signatures:>6}   "
          f"{final.size} bytes over {final.transport}, {len(final.message.answers)} answers")

python step06_truncation.py

 EDNS size   DO  UDP bytes   TC  answers  RRSIG   with TCP fallback
      none   no        292  yes        1      0   842 bytes over tcp, 3 answers
       512   no        303  yes        1      0   853 bytes over tcp, 3 answers
      1232   no        853   no        3      0   853 bytes over udp, 3 answers
      1232  yes       1139   no        4      1   1139 bytes over udp, 4 answers
      4096  yes       1139   no        4      1   1139 bytes over udp, 4 answers

In the first two rows the UDP reply is truncated: the TC column says yes, and the reply holds one of the three keys. A truncated reply may carry part of the data, so never use it; retry over TCP, as the last column does, and you get all three keys (842 bytes). The second row is 11 bytes longer than the first because it echoes our OPT record, which costs exactly 1 byte for the root name plus 10 fixed bytes. With a 1232-byte buffer the whole set fits in one UDP reply, and with the DO bit the reply grows to 1,139 bytes because it now carries a signature, still under the 1,232-byte limit. This is also why a resolver that only speaks UDP, or a firewall that drops TCP port 53, breaks in subtle ways.

Common mistake: treating the answer count of a truncated reply as the truth. Always check the TC flag before you look at the records.

Step 7: Walk from the root servers to the answer

You now have everything needed to write the resolver. The plan is RFC 1034’s resolver algorithm in miniature. Its steps include “Find the best servers to ask.” and “Send them queries until one returns a response.”, and when a reply “contains a better delegation to other servers”, the resolver remembers it and goes back to looking for the best servers. In plain words: ask a server that knows the closest zone, and either take its answer or follow its referral to a closer zone, until you arrive. The code has four parts. First, the root server addresses and a few small helpers:

"""resolver.py: an iterative DNS resolver. It starts at the root servers and follows referrals downward."""
import time
from typing import NamedTuple

from dnsnet import QueryFailed, ask
from dnswire import RCODES, TYPES, under

ROOT_SERVERS = {   # IPv4 addresses from the IANA root hints file, https://www.internic.net/domain/named.root
    "a.root-servers.net": "198.41.0.4", "b.root-servers.net": "170.247.170.2",
    "c.root-servers.net": "192.33.4.12", "d.root-servers.net": "199.7.91.13",
    "e.root-servers.net": "192.203.230.10", "f.root-servers.net": "192.5.5.241",
    "g.root-servers.net": "192.112.36.4", "h.root-servers.net": "198.97.190.53",
    "i.root-servers.net": "192.36.148.17", "j.root-servers.net": "192.58.128.30",
    "k.root-servers.net": "193.0.14.129", "l.root-servers.net": "199.7.83.42",
    "m.root-servers.net": "202.12.27.33",
}
MAX_HOPS = 16     # referrals followed for one name
MAX_CNAMES = 8    # CNAME links followed for one lookup
MAX_DEPTH = 4     # nested lookups for the address of a name server


class ResolveError(Exception):
    """The lookup could not be completed."""


def tidy(record):
    """Lower-case the names in a record. 0x20 case randomization leaks into names that compression copies from the question."""
    data = record.data.lower() if record.rtype in (TYPES["CNAME"], TYPES["NS"], TYPES["PTR"]) else record.data
    return record._replace(name=record.name.lower(), data=data)


class Result(NamedTuple):
    status: str     # "ok", "nodata" or "nxdomain"
    answers: list   # the CNAME chain followed by the records that were asked for
    queries: int    # packets sent for this lookup, nested lookups included
    trace: list     # one line for every reply we used

Those thirteen addresses are the root hints, the one piece of knowledge every resolver starts with. A hard-coded table can go stale, so check it against the official file. This short script downloads IANA’s root hints and compares them with the table:

"""check_roots.py: compare the root server table in resolver.py with the root hints file that IANA publishes."""
import re
import urllib.request

from resolver import ROOT_SERVERS

request = urllib.request.Request("https://www.internic.net/domain/named.root", headers={"User-Agent": "dns-tutorial/1.0"})
text = urllib.request.urlopen(request, timeout=20).read().decode("ascii")
published = {m.group(1).lower(): m.group(2)
             for m in re.finditer(r"^([A-Z]\.ROOT-SERVERS\.NET)\.\s+\d+\s+A\s+(\S+)", text, re.M)}
print("IANA root hints, last update:", re.search(r"last update:\s*(.+)", text).group(1).strip())
print(f"our table matches the file: {ROOT_SERVERS == published} ({len(published)} servers)")

python check_roots.py

IANA root hints, last update: November 03, 2026
our table matches the file: True (13 servers)

The second part is the cache. Every referral carries a TTL, so a resolver can remember where a zone’s name servers are and skip the top of the tree next time. closest finds the deepest zone above a name that is still remembered:

# resolver.py (continued)
class Cache:
    """Remember where a zone's name servers live, until the TTL on the NS records runs out."""

    def __init__(self, clock=time.monotonic):
        self.clock = clock
        self.zones = {}   # zone -> (expiry time, [name server addresses])

    def put(self, zone, addresses, ttl):
        self.zones[zone] = (self.clock() + ttl, list(addresses))

    def closest(self, name):
        """The deepest zone at or above name that we still remember: ('example.com', [addresses]) or None."""
        labels = name.split(".") if name else []
        for i in range(len(labels)):
            zone = ".".join(labels[i:])
            entry = self.zones.get(zone)
            if entry and entry[0] > self.clock():
                return zone, entry[1]
        return None

The third part is the walk itself. Read _walk as a loop with three outcomes after each reply. If the reply holds the records we asked for, we are done. If it holds a CNAME, we start over with the alias target (a CNAME can point into a completely different zone). If it holds a referral, we move to the child zone and ask its servers. The two checks _referral applies, which you will see in part four, are what keep a lying server from sending us in circles. Add this to resolver.py, indented as shown so that the methods stay inside the class:

# resolver.py (continued)
class Resolver:
    def __init__(self, roots=None, *, port=53, timeout=2.0, use_0x20=False, strict=True, cache=None):
        self.roots = list((roots or ROOT_SERVERS).values())
        self.port, self.timeout, self.use_0x20, self.strict = port, timeout, use_0x20, strict
        self.cache = cache if cache is not None else Cache()
        self.queries = 0
        self.trace = []

    def resolve(self, name, qtype="A", _depth=0):
        """Look name up from the root down. Returns a Result, or raises ResolveError."""
        name = name.rstrip(".").lower()
        first_query, first_line = self.queries, len(self.trace)
        chain = []
        for _ in range(MAX_CNAMES + 1):
            kind, found = self._walk(name, qtype, _depth)
            if kind == "cname":
                chain.append(found)
                name = found.data.lower()
                continue
            answers = chain + (found if kind == "ok" else [])
            return Result(kind, answers, self.queries - first_query, self.trace[first_line:])
        raise ResolveError(f"more than {MAX_CNAMES} CNAMEs in a row")

    def _walk(self, name, qtype, depth):
        start = self.cache.closest(name)
        zone, servers = start if start else ("", self.roots)
        spare, ttl = [], None    # name servers known only by name, and the TTL of the referral that named them
        for _ in range(MAX_HOPS):
            ip, reply = self._ask_any(zone, servers, spare, name, qtype, depth)
            if ttl and ip not in servers:                 # we got here through a name server we had to look up
                servers = servers + [ip]
                self.cache.put(zone, servers, ttl)
            message = reply.message
            self._learn(zone, message)
            wanted = [tidy(r) for r in message.answers if r.name.lower() == name]
            direct = [r for r in wanted if r.rtype == TYPES[qtype]]
            cnames = [r for r in wanted if r.rtype == TYPES["CNAME"]]
            if direct:
                self._note(zone, ip, name, qtype, reply, f"answer, {len(direct)} record(s)")
                return "ok", direct
            if cnames and qtype != "CNAME":
                self._note(zone, ip, name, qtype, reply, f"CNAME to {cnames[0].data}")
                return "cname", cnames[0]
            if message.rcode == 3:
                self._note(zone, ip, name, qtype, reply, "NXDOMAIN")
                return "nxdomain", None
            referral = self._referral(zone, name, message)
            if referral is None:
                if message.header.aa:
                    self._note(zone, ip, name, qtype, reply, "NODATA (the name exists, not this type)")
                    return "nodata", None
                raise ResolveError(f"{ip} gave neither an answer nor a usable referral for {name}")
            child, hosts, ttl = referral
            glue = self._glue(zone, message)
            servers = [address for host in hosts for address in glue.get(host, [])]
            spare = [host for host in hosts if host not in glue]
            self._note(zone, ip, name, qtype, reply, f"referral to {child} "
                       f"({len(hosts)} name servers, {len(hosts) - len(spare)} with addresses)")
            if servers:
                self.cache.put(child, servers, ttl)
            zone = child
        raise ResolveError(f"more than {MAX_HOPS} referrals while looking up {name}")

The fourth part holds the helpers. Four of them deserve attention. _referral accepts only NS records whose owner is below the zone we asked and at or above the name we want, which guarantees progress. _glue and _learn decide which extra records to believe; Step 8 puts them under attack. _candidates yields the addresses we know first and looks up the others lazily, because a referral often names more servers than it supplies glue for. And _lookup_hosts resolves a name server’s own name when no glue came with the referral:

# resolver.py (continued)
    def _candidates(self, servers, spare, depth):
        """Addresses worth trying: the ones we know, then those of name servers we must look up first."""
        yield from servers
        for host in spare:
            yield from self._lookup_hosts([host], depth)

    def _ask_any(self, zone, servers, spare, name, qtype, depth):
        """Ask the servers one after another until one gives a usable reply."""
        problems = []
        for ip in self._candidates(servers, spare, depth):
            self.queries += 1
            try:
                reply = ask(ip, name, qtype, port=self.port, use_0x20=self.use_0x20,
                            timeout=self.timeout, edns_size=1232)
            except QueryFailed as exc:
                problems.append(str(exc))
                continue
            if reply.message.rcode not in (0, 3):
                problems.append(f"{ip} answered {RCODES.get(reply.message.rcode, reply.message.rcode)}")
                continue
            return ip, reply
        detail = "; ".join(problems) or "no name server has a known address"
        raise ResolveError(f"no server for {zone or '.'} could answer {name}: {detail}")

    def _referral(self, zone, name, message):
        """The NS records that move us closer: owner strictly below the zone we asked, at or above the name."""
        records = [r for r in message.authority
                   if r.rtype == TYPES["NS"] and r.name.lower() != zone
                   and under(r.name.lower(), zone) and under(name, r.name.lower())]
        if not records:
            return None
        child = records[0].name.lower()
        same = [r for r in records if r.name.lower() == child]
        return child, [r.data.lower() for r in same], min(r.ttl for r in same)

    def _glue(self, zone, message):
        """Addresses the reply supplies for name servers. Strict mode keeps only names the asked zone owns."""
        glue = {}
        for r in message.additional:
            if r.rtype == TYPES["A"] and (not self.strict or under(r.name.lower(), zone)):
                glue.setdefault(r.name.lower(), []).append(r.data)
        return glue

    def _learn(self, zone, message):
        """Cache every delegation the reply carries. Strict mode ignores zones the asked zone does not own."""
        glue = self._glue(zone, message)
        found = {}
        for r in message.answers + message.authority:
            owner = r.name.lower()
            if r.rtype != TYPES["NS"] or (self.strict and not under(owner, zone)):
                continue
            entry = found.setdefault(owner, {"addresses": [], "ttl": r.ttl})
            entry["addresses"] += glue.get(r.data.lower(), [])
            entry["ttl"] = min(entry["ttl"], r.ttl)
        for owner, entry in found.items():
            if entry["addresses"]:
                self.cache.put(owner, entry["addresses"], entry["ttl"])

    def _lookup_hosts(self, hosts, depth):
        """No glue came with the referral, so find a name server's address by resolving its name ourselves."""
        if depth >= MAX_DEPTH:
            return []
        for host in hosts:
            try:
                result = self.resolve(host, "A", depth + 1)
            except ResolveError:
                continue
            addresses = [r.data for r in result.answers if r.rtype == TYPES["A"]]
            if addresses:
                return addresses
        return []

    def _note(self, zone, ip, name, qtype, reply, what):
        self.trace.append(f"{zone or '.':<14} {ip:<16} {name} {qtype}: {what} "
                          f"({reply.size} bytes, {reply.seconds * 1000:.0f} ms, {reply.transport})")

Run it on four names. The first is an ordinary site, the second is a site that may be an alias, and the last two do not exist:

"""step07_walk.py: resolve names from the root servers down and show every server we asked."""
from dnswire import format_record
from resolver import Resolver

resolver = Resolver()
for name, qtype in [("www.example.com", "A"), ("www.github.com", "A"), ("no-such-name-8c1d.example.com", "A"), ("this-name-does-not-exist-8c1d.org", "A")]:
    result = resolver.resolve(name, qtype)
    print(f"=== {name} {qtype}: {result.status}, {result.queries} queries")
    for line in result.trace:
        print(line)
    for record in result.answers:
        print(format_record(record))
    print()

python step07_walk.py

=== www.example.com A: ok, 3 queries
.              198.41.0.4       www.example.com A: referral to com (13 name servers, 13 with addresses) (840 bytes, 20 ms, udp)
com            192.41.162.30    www.example.com A: referral to example.com (2 name servers, 2 with addresses) (363 bytes, 57 ms, udp)
example.com    108.162.192.162  www.example.com A: answer, 2 record(s) (65 bytes, 4 ms, udp)
www.example.com.               122     IN  A      172.66.147.243
www.example.com.               122     IN  A      104.20.23.154

=== www.github.com A: ok, 3 queries
com            192.41.162.30    www.github.com A: referral to github.com (8 name servers, 1 with addresses) (310 bytes, 49 ms, udp)
github.com     205.251.193.165  www.github.com A: CNAME to github.com (296 bytes, 20 ms, udp)
github.com     205.251.193.165  github.com A: answer, 1 record(s) (278 bytes, 17 ms, udp)
www.github.com.                3600    IN  CNAME  github.com
github.com.                    60      IN  A      140.82.114.4

=== no-such-name-8c1d.example.com A: nxdomain, 1 queries
example.com    108.162.192.162  no-such-name-8c1d.example.com A: NXDOMAIN (120 bytes, 25 ms, udp)

=== this-name-does-not-exist-8c1d.org A: nxdomain, 2 queries
.              198.41.0.4       this-name-does-not-exist-8c1d.org A: referral to org (6 name servers, 6 with addresses) (477 bytes, 20 ms, udp)
org            199.19.56.1      this-name-does-not-exist-8c1d.org A: NXDOMAIN (133 bytes, 46 ms, udp)

Each trace line shows the zone we asked about, the server’s address, what we asked, what the server said, and the reply size and time. The first lookup took three queries: the root referred us to com (13 name servers, all with addresses), a com server referred us to example.com, and an example.com server answered. The second lookup began at com instead of the root, because the cache remembered it. The GitHub name was a CNAME when I ran this, but it was a plain A record a few minutes earlier in another run, so your output may differ; a steadier alias example is www.wikipedia.org, which crosses from the wikipedia.org zone into wikimedia.org, and the end-to-end section runs it. Finally, notice that GitHub’s referral listed 8 name servers but supplied an address for only 1. Seven of the names, such as ns-520.awsdns-01.net and dns1.p08.nsone.net, sit under net, org, and co.uk, so the com servers supplied no glue for them, and only ns-421.awsdns-52.com comes with an address. That is the situation _candidates exists for.

Why one root reply was 493 bytes and another 840

The trace shows the first root reply at 840 bytes. That is larger than the old 512-byte UDP limit because our queries carry the OPT record that advertises 1232 bytes. Without EDNS, the same root server has to squeeze its referral into 512 bytes, and name compression is what makes that possible. The script below asks without EDNS and then re-encodes the same records with and without compression, using the encoder from testkit.py:

"""step07b_compression.py: how many bytes does name compression save in a real referral from a root server?"""
from dnsnet import ask_udp
from dnswire import TYPE_NAMES, TYPES, type_name
from testkit import encode_message

reply = ask_udp("198.41.0.4", "www.example.com", "A")
message = reply.message


def rewrite(records):
    return [(r.name, TYPE_NAMES[r.rtype], r.ttl, r.data) for r in records if r.rtype != TYPES["OPT"]]


for compress in (True, False):
    wire = encode_message(message.header.id, 0x8000, [("www.example.com", "A")], rewrite(message.answers),
                          rewrite(message.authority), rewrite(message.additional), compress=compress)
    print(f"re-encoded with compression {'on ' if compress else 'off'}: {len(wire)} bytes")
print(f"the server sent {reply.size} bytes: {len(message.authority)} NS records, {len(message.additional)} glue records")
kinds = sorted({type_name(r.rtype) for r in message.additional})
print("glue record types:", ", ".join(f"{sum(type_name(r.rtype) == k for r in message.additional)} {k}" for k in kinds))

python step07b_compression.py

re-encoded with compression on : 493 bytes
re-encoded with compression off: 922 bytes
the server sent 493 bytes: 13 NS records, 11 glue records
glue record types: 6 A, 5 AAAA

Compression shrank the referral from 922 bytes to 493, a saving of 429 bytes or about 46.5 percent. Even so, the 512-byte reply had room for only 11 glue records, 6 addresses of the A type and 5 of the AAAA type, whereas the 1232-byte reply carried all 13 IPv4 addresses. A resolver that does not speak EDNS gets worse referrals and more round trips.

NXDOMAIN, NODATA, and a server that fibs on purpose

Back in Step 4 the made-up name under example.com returned NOERROR, but our walk in Step 7 got NXDOMAIN for the same name. The difference is in who asked and how. This script asks the authoritative server directly, twice, with the DO bit clear and then set:

"""step07c_denial.py: ask the same name server about a name that does not exist, with and without the DO bit."""
from dnsnet import ask_udp
from dnswire import RCODES, type_name
from resolver import Resolver

NAME = "no-such-name-8c1d.example.com"
trace = Resolver().resolve(NAME, "A").trace
server = trace[-1].split()[1]
print(f"the authoritative server for example.com that answered: {server}")
for do in (False, True):
    reply = ask_udp(server, NAME, "A", edns_size=1232, dnssec_ok=do)
    kinds = sorted({type_name(r.rtype) for r in reply.message.authority})
    print(f"DO bit {'set  ' if do else 'clear'}: {RCODES[reply.message.rcode]:<8} {reply.size:>4} bytes, "
          f"answers {len(reply.message.answers)}, authority records: {', '.join(kinds)}")

python step07c_denial.py

the authoritative server for example.com that answered: 108.162.192.162
DO bit clear: NXDOMAIN  120 bytes, answers 0, authority records: SOA
DO bit set  : NOERROR   398 bytes, answers 0, authority records: NSEC, RRSIG, SOA

The same server says NXDOMAIN to a plain query and NOERROR, with an NSEC record, to a query that sets the DO bit. That is a technique called compact denial of existence, which Cloudflare calls “black lies”. Its 2016 post says “Instead of white lies, we do black lies” and explains that this lets the server build a minimal signed answer on the fly, without searching its database or precomputing anything; RFC 9824 now standardizes the idea as “a technique to generate a signed DNS response on demand for a nonexistent name by claiming that the name exists but doesn’t have any data for the queried record type.” A validating resolver sets the DO bit, so it is plausible that this is why 1.1.1.1 told us NODATA; I did not capture the resolver’s own upstream queries, so treat that link as my inference. The practical lesson stands either way: do not test NXDOMAIN handling against a DNSSEC-signed zone that does this, and do not assume NXDOMAIN always arrives for a missing name.

Should you turn on 0x20 encoding?

Now the evidence promised in Step 5. The script below asks the root servers, the com servers, and the example.com servers for a name with random capitalization and counts how many echo the case back unchanged. It then repeats the example.com test with a new name every time:

"""step07d_case_echo.py: do real name servers echo a mixed-case question back unchanged (0x20 encoding)?"""
import secrets
import time

from dnsnet import QueryFailed, ask_udp
from resolver import ROOT_SERVERS, Resolver

resolver = Resolver()
resolver.resolve("www.example.com")            # fills the cache with the .com and example.com servers
groups = {
    "root servers": list(ROOT_SERVERS.values()),
    ".com servers": resolver.cache.zones["com"][1],
    "example.com servers": resolver.cache.zones["example.com"][1],
}
TRIES = 5
groups["example.com, new names"] = groups["example.com servers"]
for label, addresses in groups.items():
    kept = changed = silent = 0
    probe_name = (lambda: "t" + secrets.token_hex(4) + ".example.com") if "new" in label else (lambda: "www.example.com")
    for address in addresses:
        for _ in range(TRIES):
            try:
                ask_udp(address, probe_name(), "A", use_0x20=True, timeout=1.0)
                kept += 1                      # with use_0x20=True, ask_udp only returns a reply that echoed our random case exactly
            except QueryFailed as exc:
                if "question does not match" in str(exc):
                    changed += 1
                else:
                    silent += 1
            time.sleep(0.05)
    total = len(addresses) * TRIES
    print(f"{label:<20} {len(addresses):>2} servers, {total:>3} queries: "
          f"case kept {kept}, case changed {changed}, no usable reply {silent}")

python step07d_case_echo.py

root servers         13 servers,  65 queries: case kept 60, case changed 0, no usable reply 5
.com servers         13 servers,  65 queries: case kept 65, case changed 0, no usable reply 0
example.com servers   6 servers,  30 queries: case kept 0, case changed 30, no usable reply 0
example.com, new names  6 servers,  30 queries: case kept 30, case changed 0, no usable reply 0

The root servers and com servers kept the case (five of the root queries got no usable reply, and I did not investigate why). The example.com servers lowercased the question in most replies when I repeated the same name, but kept the case every time when each name was new. A cached or precomputed reply keyed without regard to case would explain that, but that is my inference; I did not look inside the servers. In four runs the number of unchanged replies for the repeated name ranged from 0 to 6 out of 30. A resolver that insists on exact case would fail against such servers; my first version had it on by default and could not resolve www.example.com at all. Unbound, a widely used resolver, documents use-caps-for-id as “an experimental implementation of draft dns-0x20” with the default “no”, and offers a caps-exempt list for domains that “do not support 0x20”. So your resolver keeps the three strong defenses from Step 5 on, and treats 0x20 as an option.

Step 8: Do not believe records a server had no right to send

A reply can carry more than you asked for. The additional section exists so that a server can hand over glue, and a malicious or compromised server can use it to plant records for names it does not own. RFC 5452 section 6 is titled “Accepting Only In-Domain Records” and warns: “If accepted uncritically, the resolver stands the chance of accepting data from an untrusted source.” Its advice: “One very simple way to achieve this is to only accept data if it is part of the domain for which the query was intended.” RFC 2181 section 5.4.1 adds a ranking of how much to trust each kind of data, and says that “additional information from a reply will be ignored if the cache contains data from an authoritative answer or a zone file.” The check that enforces this is called a bailiwick check: a server may only speak for names at or below the zone it was asked about.

To see the attack, you need a small world to run it in. This second part of testkit.py adds Zone, a tiny authoritative server that serves records, referrals, NODATA, and NXDOMAIN, and builds five fake servers on loopback addresses. The .test top-level domain is reserved for testing (RFC 6761), and the addresses are from the documentation ranges (RFC 5737), so nothing here can touch a real name. The server for evil.test adds a planted delegation for bank.test, pointing at the attacker’s server at 127.0.0.66, to every answer it gives:

# testkit.py (continued)
class Zone:
    """A tiny authoritative server: answers, referrals to child zones, NODATA and NXDOMAIN."""

    def __init__(self, origin, records=(), delegations=None, planted_authority=(), planted_additional=()):
        self.origin = origin.rstrip(".").lower()
        self.records = [(n.rstrip(".").lower(), t, ttl, d) for n, t, ttl, d in records]
        self.delegations = {k.lower(): v for k, v in (delegations or {}).items()}   # child -> [(host, address or None)]
        self.planted_authority = list(planted_authority)      # extra records we sneak into positive answers
        self.planted_additional = list(planted_additional)

    def handle(self, query, transport="udp", addr=None):
        asked = query.questions[0]
        name, qtype = asked.name.lower(), TYPE_NAMES.get(asked.qtype, "A")
        flags = 0x8000 | (0x0100 if query.header.rd else 0)          # a response, recursion desired echoed
        question = [(asked.name, qtype)]
        for child, servers in self.delegations.items():
            if under(name, child):                                   # a referral: NS records plus glue
                authority = [(child, "NS", 172800, host) for host, _ in servers]
                additional = [(host, "A", 172800, ip) for host, ip in servers if ip]
                return encode_message(query.header.id, flags, question, [], authority, additional)
        here = [r for r in self.records if r[0] == name]
        answers = [r for r in here if r[1] == qtype] or [r for r in here if r[1] == "CNAME"]
        if answers:
            owned = [(asked.name, t, ttl, d) for _, t, ttl, d in answers]
            return encode_message(query.header.id, flags | 0x0400, question, owned,
                                  self.planted_authority, self.planted_additional)
        soa = [(self.origin, "SOA", 300, ("ns." + self.origin, "hostmaster." + self.origin, 1, 3600, 600, 86400, 300))]
        rcode = 0 if here else 3                                     # NODATA if the name exists, else NXDOMAIN
        return encode_message(query.header.id, flags | 0x0400 | rcode, question, [], soa)


def start_world(port, zones):
    """zones maps a loopback address to a Zone. Returns the running servers; call stop() on each when done."""
    return [FakeServer(ip, port, zone.handle) for ip, zone in zones.items()]


WORLD_IPS = {"root": "127.0.0.1", "test": "127.0.0.2", "evil": "127.0.0.3", "bank": "127.0.0.4", "attacker": "127.0.0.66"}


def make_world(poisoned):
    """root -> test. -> evil.test and bank.test. The evil server may plant a bank.test delegation in its answers."""
    ips = WORLD_IPS
    authority = [("bank.test", "NS", 300, "ns.bank.test")] if poisoned else []
    additional = [("ns.bank.test", "A", 300, ips["attacker"])] if poisoned else []
    return {
        ips["root"]: Zone("", delegations={"test": [("a.nic.test", ips["test"])]}),
        ips["test"]: Zone("test", delegations={"evil.test": [("ns.evil.test", ips["evil"])],
                                               "bank.test": [("ns.bank.test", ips["bank"])]}),
        ips["evil"]: Zone("evil.test", [("www.evil.test", "A", 300, "203.0.113.66"),
                                        ("mail.evil.test", "A", 300, "203.0.113.67")],
                          planted_authority=authority, planted_additional=additional),
        ips["bank"]: Zone("bank.test", [("www.bank.test", "A", 300, "192.0.2.10")]),
        ips["attacker"]: Zone("bank.test", [("www.bank.test", "A", 300, "198.51.100.66")]),
    }

"""step08_poisoning.py: let one server plant a delegation for someone else's zone and see who believes it."""
from resolver import Resolver
from testkit import WORLD_IPS, make_world, start_world

PORT = 15360
servers = start_world(PORT, make_world(poisoned=True))
try:
    for strict in (False, True):
        resolver = Resolver(roots={"root": WORLD_IPS["root"]}, port=PORT, timeout=1.0, strict=strict)
        resolver.resolve("www.evil.test")                       # an ordinary lookup of the attacker's own name
        result = resolver.resolve("www.bank.test")              # later, someone looks up the bank
        asked = result.trace[-1].split()[1]
        who = "the attacker" if asked == WORLD_IPS["attacker"] else "the real bank"
        print(f"strict={strict}:")
        print(f"  cached delegation for bank.test: {resolver.cache.zones.get('bank.test', (0, None))[1]}")
        print(f"  www.bank.test is {result.answers[0].data}, learned from {who} at {asked} ({result.queries} queries)")
finally:
    for server in servers:
        server.stop()

python step08_poisoning.py

strict=False:
  cached delegation for bank.test: ['127.0.0.66']
  www.bank.test is 198.51.100.66, learned from the attacker at 127.0.0.66 (1 queries)
strict=True:
  cached delegation for bank.test: ['127.0.0.4']
  www.bank.test is 192.0.2.10, learned from the real bank at 127.0.0.4 (2 queries)

Both resolvers do the same two lookups. The first is an innocent-looking name in the attacker’s own zone. The second is the bank, much later. The trusting resolver (strict=False) cached the planted delegation, so the bank lookup went straight to 127.0.0.66 and returned the attacker’s address. The strict resolver threw the planted records away at the moment they arrived, because bank.test is not at or below evil.test, and it found the real bank. In the code, that is the pair of under(...) checks inside _glue and _learn. Either one alone would stop this particular attack, so Step 10 removes both together to prove the test notices.

Step 9: Cache delegations and watch the packet count fall

The cache from Step 7 has been working all along. This script resolves six names in a row and prints how many packets each needed, which server it started with, and how long the two top delegations stay valid:

"""step09_cache.py: count the packets for lookups that share name servers, with the cache doing its job."""
import time

from resolver import Resolver

resolver = Resolver()
print(f"{'lookup':<28}{'queries':>8}   first server asked")
for name in ["www.example.com", "example.com", "www.example.com", "www.iana.org", "gmail.com", "www.gmail.com"]:
    result = resolver.resolve(name, "A")
    first = result.trace[0].split()[0]
    print(f"{name:<28}{result.queries:>8}   {first}")
print("\ncached zones:", ", ".join(sorted(resolver.cache.zones)))
for zone in ("com", "example.com"):
    seconds = resolver.cache.zones[zone][0] - time.monotonic()
    print(f"the {zone} delegation stays cached for {seconds:,.0f} more seconds ({seconds / 86400:.1f} days)")

python step09_cache.py

lookup                       queries   first server asked
www.example.com                    3   .
example.com                        1   example.com
www.example.com                    1   example.com
www.iana.org                       6   .
gmail.com                          2   com
www.gmail.com                      1   gmail.com

cached zones: cloudflare.net, com, example.com, gmail.com, iana.org, net, org
the com delegation stays cached for 172,800 more seconds (2.0 days)
the example.com delegation stays cached for 172,800 more seconds (2.0 days)

A cold lookup of www.example.com cost three queries. The next lookup in the same zone cost one, because the cache already knew where example.com lives. A lookup under gmail.com began at com and cost two, and the one after it cost one. The www.iana.org lookup cost six because it is an alias into another zone: its CNAME points at www.iana.org.cdn.cloudflare.net, so the resolver had to walk net and cloudflare.net as well, which you can see in the cached-zones list. The last two lines come from the TTL on the referral: both delegations stay cached for two days, which is why a resolver with a warm cache rarely needs to ask the root. This cache stores only delegations. A production resolver also caches answers and negative answers (RFC 2308), which is the first item in the next-steps list.

Step 10: Prove it with tests, a fuzzer, and planted bugs

You have a program that works on the happy path. This step shows how to find out whether it works on the unhappy one. There are three layers.

Layer 1: a fuzzer for the parser

Recall the promise that parse_message raises DNSError and nothing else. A fuzzer tests promises like that by brute force. It takes four valid messages, damages them at random (overwriting bytes, truncating, inserting bytes, lying about section counts, planting compression pointers), and counts what comes out. Any exception other than DNSError would be a bug:

"""fuzz.py: feed damaged copies of valid messages to parse_message and count what comes out."""
import random
import time

from dnswire import DNSError, parse_message
from testkit import encode_message

QUESTION = [("www.example.com", "A")]
SOA = ("ns1.example.com", "hostmaster.example.com", 2026110601, 7200, 900, 1209600, 300)
SEEDS = [
    # a referral: two NS records plus glue
    encode_message(0x1111, 0x8000, QUESTION, [],
                   [("example.com", "NS", 172800, "ns1.example.net"), ("example.com", "NS", 172800, "ns2.example.net")],
                   [("ns1.example.net", "A", 172800, "192.0.2.53"), ("ns2.example.net", "AAAA", 172800, "2001:db8::53")]),
    # an answer that follows a CNAME
    encode_message(0x2222, 0x8400, QUESTION,
                   [("www.example.com", "CNAME", 300, "edge.example.net"), ("edge.example.net", "A", 60, "192.0.2.80")],
                   [("example.net", "NS", 300, "ns1.example.net")]),
    # MX, TXT and SOA data
    encode_message(0x3333, 0x8400, [("example.com", "MX")],
                   [("example.com", "MX", 300, (10, "mail.example.com")),
                    ("example.com", "TXT", 300, ["v=spf1 -all", "second string"])],
                   [("example.com", "SOA", 300, SOA)]),
    # NXDOMAIN with an SOA
    encode_message(0x4444, 0x8403, QUESTION, [], [("example.com", "SOA", 300, SOA)]),
]


def damage(rng, data):
    data = bytearray(data)
    kind = rng.randrange(5)
    if kind == 0:                                    # overwrite a few bytes
        for _ in range(rng.randrange(1, 5)):
            data[rng.randrange(len(data))] = rng.randrange(256)
    elif kind == 1:                                  # cut the message short
        del data[rng.randrange(len(data)):]
    elif kind == 2:                                  # insert random bytes
        at = rng.randrange(len(data))
        data[at:at] = bytes(rng.randrange(256) for _ in range(rng.randrange(1, 8)))
    elif kind == 3:                                  # lie about a section count
        data[rng.choice([4, 6, 8, 10])] = 0xFF
    else:                                            # plant a compression pointer somewhere
        at = rng.randrange(12, len(data) - 1)
        data[at], data[at + 1] = 0xC0 | rng.randrange(0x40), rng.randrange(256)
    return bytes(data)


def run_fuzz(count, seed):
    rng = random.Random(seed)
    outcome = {"parsed": 0, "refused": 0, "other": [], "slowest_ms": 0.0}
    for _ in range(count):
        packet = damage(rng, rng.choice(SEEDS))
        started = time.perf_counter()
        try:
            parse_message(packet)
            outcome["parsed"] += 1
        except DNSError:
            outcome["refused"] += 1
        except Exception as exc:                     # anything else is a bug in the parser
            outcome["other"].append(f"{type(exc).__name__}: {exc}")
        outcome["slowest_ms"] = max(outcome["slowest_ms"], (time.perf_counter() - started) * 1000)
    return outcome


if __name__ == "__main__":
    result = run_fuzz(20000, seed=2026)
    print(f"20,000 damaged messages: {result['parsed']} still parsed, {result['refused']} refused with DNSError, "
          f"{len(result['other'])} raised something else; slowest {result['slowest_ms']:.2f} ms")

python fuzz.py

20,000 damaged messages: 4523 still parsed, 15477 refused with DNSError, 0 raised something else; slowest 0.18 ms

Out of 20,000 damaged messages, 4,523 still happened to be valid and 15,477 were refused with DNSError. None raised anything else, and the slowest took well under a millisecond, so none of those inputs made the parser do runaway work.

Layer 2: the test suite

The tests follow the structure of this tutorial: the wire format (including every hostile name from Step 3, each checked for the right refusal reason), the reply checks (using fake servers on loopback), and the resolver (using the fake world from Step 8). Each test names the behavior it protects. First the wire format:

"""test_dns.py: tests for the wire format, the reply checks and the resolver. Run with: python -m pytest -q"""
import socket
import struct
import threading
import time

import pytest

from dnsnet import QueryFailed, ask, ask_udp, bind_random_port
from dnswire import DNSError, build_query, encode_name, parse_message, read_name
from fuzz import SEEDS, run_fuzz
from resolver import Cache, ResolveError, Resolver
from testkit import WORLD_IPS, FakeServer, Zone, encode_message, make_world, start_world

HEADER = b"\x00" * 12
CHAIN = HEADER + b"\x01a\x00" + b"".join(
    struct.pack(">H", 0xC000 | (12 if i == 0 else 13 + 2 * i)) for i in range(40))   # forty pointers, each to the one before


# ---------------------------------------------------------------- the wire format
def test_query_bytes_match_a_hand_assembled_message():
    _, wire = build_query("example.com", "A", query_id=0x1234)
    assert wire.hex() == "123401000001000000000000076578616d706c6503636f6d0000010001"


def test_response_with_a_compression_pointer():
    packet = (struct.pack(">HHHHHH", 0x1234, 0x8180, 1, 1, 0, 0) + encode_name("example.com") + struct.pack(">HH", 1, 1)
              + b"\xc0\x0c" + struct.pack(">HHIH", 1, 1, 300, 4) + bytes([93, 184, 216, 34]))
    message = parse_message(packet)
    assert message.answers[0] == ("example.com", 1, 1, 300, "93.184.216.34")
    assert message.header.flags_text() == "qr rd ra"


def test_edns_settings_are_read_back():
    _, wire = build_query("example.com", "A", edns_size=1232, dnssec_ok=True, query_id=1)
    assert parse_message(wire).edns == (1232, 0, 0, True)


def test_name_limits():
    encode_name("a" * 63 + ".com")
    for bad in ("a" * 64 + ".com", "a..b", ".".join(["a" * 60] * 5)):
        with pytest.raises(DNSError):
            encode_name(bad)


def test_case_and_underscores_survive_a_round_trip():
    assert read_name(HEADER + encode_name("_DmArC.ExAmPle.com"), 12)[0] == "_DmArC.ExAmPle.com"


HOSTILE = [
    ("pointer to itself", HEADER + b"\xc0\x0c", 12, "backwards"),
    ("pointer ahead", HEADER + b"\xc0\x0e\x00", 12, "backwards"),
    ("pointer into the header", HEADER + b"\xc0\x04", 12, "into the header"),
    ("label then pointer back", HEADER + b"\x01a\xc0\x0c", 12, "compression pointers in one name"),
    ("forty pointers in a row", CHAIN, 93, "compression pointers in one name"),
    ("label type that does not exist", HEADER + bytes([0x41]) + b"a" * 65 + b"\x00", 12, "unknown label type"),
    ("pointer cut off", HEADER + b"\xc0", 12, "cut off"),
    ("label runs past the end", HEADER + b"\x3fabc", 12, "label runs past"),
    ("name over 255 bytes", HEADER + (b"\x3f" + b"a" * 63) * 5 + b"\x00", 12, "longer than 255"),
    ("name without a final zero", HEADER + b"\x03abc", 12, "name runs past"),
]


@pytest.mark.parametrize("title,packet,start,reason", HOSTILE, ids=[h[0] for h in HOSTILE])
def test_hostile_names_are_refused_for_the_right_reason(title, packet, start, reason):
    with pytest.raises(DNSError, match=reason):
        read_name(packet, start)


def test_counts_that_lie_end_in_dns_error():
    packet = (struct.pack(">HHHHHH", 1, 0x8180, 1, 65535, 65535, 65535) + encode_name("a.com") + struct.pack(">HH", 1, 1))
    with pytest.raises(DNSError):
        parse_message(packet)


def test_bytes_after_the_last_record_are_refused():
    _, wire = build_query("a.com", "A", query_id=1)
    with pytest.raises(DNSError, match="unexpected bytes"):
        parse_message(wire + b"\x00")


def test_ttl_with_the_top_bit_counts_as_zero():
    wire = encode_message(1, 0x8400, [("a.test", "A")],
                          [("a.test", "A", 0xFFFFFFFF, "192.0.2.1"), ("a.test", "A", 300, "192.0.2.2")])
    assert [r.ttl for r in parse_message(wire).answers] == [0, 300]


def test_an_ns_name_must_fill_the_record_data():
    rdata = encode_name("ns.a.test")
    packet = (struct.pack(">HHHHHH", 1, 0x8400, 1, 1, 0, 0) + encode_name("a.test") + struct.pack(">HH", 2, 1)
              + encode_name("a.test") + struct.pack(">HHIH", 2, 1, 60, len(rdata) + 1) + rdata + b"\x00")
    with pytest.raises(DNSError, match="does not fill"):
        parse_message(packet)


def test_txt_with_control_bytes_prints_safely():
    wire = encode_message(1, 0x8400, [("a.test", "TXT")], [("a.test", "TXT", 60, ["line one\nline two", 'say "hi"'])])
    assert parse_message(wire).answers[0].data == '"line one\\010line two" "say \\"hi\\""'


def test_a_null_mx_prints_the_root_as_a_dot():
    wire = encode_message(1, 0x8400, [("a.test", "MX")], [("a.test", "MX", 60, (0, "."))])
    assert parse_message(wire).answers[0].data == "0 ."


@pytest.mark.parametrize("seed", range(len(SEEDS)))
def test_valid_messages_parse_and_the_counts_add_up(seed):
    message = parse_message(SEEDS[seed])
    counts = (len(message.questions), len(message.answers), len(message.authority), len(message.additional))
    assert message.header.counts == counts


def test_damaged_messages_only_ever_raise_dns_error():
    outcome = run_fuzz(3000, seed=7)
    assert outcome["other"] == []
    assert outcome["parsed"] + outcome["refused"] == 3000

Then the reply checks. Each test starts a fake server that misbehaves in one specific way:

# test_dns.py (continued)
# ---------------------------------------------------------------- the reply checks
def answer(query, ip="192.0.2.1", *, ident=None, flags=0x8400, name=None):
    asked = name or query.questions[0].name
    ident = query.header.id if ident is None else ident
    return encode_message(ident, flags, [(asked, "A")], [(asked, "A", 60, ip)])


@pytest.fixture
def serve():
    running = []

    def start(handler, port):
        running.append(FakeServer("127.0.0.1", port, handler))

    yield start
    for server in running:
        server.stop()


def test_a_matching_reply_is_accepted(serve):
    serve(lambda q, t, a: answer(q), 15401)
    got = ask_udp("127.0.0.1", "www.example.com", port=15401)
    assert got.message.answers[0].data == "192.0.2.1" and got.rejected == []


def test_a_wrong_id_is_ignored_and_the_real_reply_still_wins(serve):
    serve(lambda q, t, a: [answer(q, "6.6.6.6", ident=(q.header.id + 1) & 0xFFFF), answer(q)], 15402)
    got = ask_udp("127.0.0.1", "www.example.com", port=15402)
    assert got.message.answers[0].data == "192.0.2.1" and got.rejected == ["wrong id"]


def test_a_reply_to_another_question_is_rejected(serve):
    serve(lambda q, t, a: answer(q, "6.6.6.6", name="other.example"), 15403)
    with pytest.raises(QueryFailed, match="question does not match"):
        ask_udp("127.0.0.1", "www.example.com", port=15403, timeout=0.4)


def test_a_query_is_not_accepted_as_a_response(serve):
    serve(lambda q, t, a: answer(q, flags=0x0100), 15404)
    with pytest.raises(QueryFailed, match="not a response"):
        ask_udp("127.0.0.1", "www.example.com", port=15404, timeout=0.4)


def test_0x20_demands_the_exact_case(serve):
    serve(lambda q, t, a: answer(q, name=q.questions[0].name.lower()), 15405)
    name = "www.a-name-with-plenty-of-letters.example.com"
    assert ask_udp("127.0.0.1", name, port=15405, use_0x20=False).rejected == []
    with pytest.raises(QueryFailed, match="question does not match"):
        ask_udp("127.0.0.1", name, port=15405, use_0x20=True, timeout=0.4)


def test_a_forgery_from_another_port_never_arrives(serve):
    attacker = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

    def handler(query, transport, addr):
        attacker.sendto(answer(query, "6.6.6.6"), addr)
        return answer(query)

    serve(handler, 15406)
    got = ask_udp("127.0.0.1", "www.example.com", port=15406)
    attacker.close()
    assert got.message.answers[0].data == "192.0.2.1" and got.rejected == []


def test_silence_ends_in_a_timeout(serve):
    serve(lambda q, t, a: None, 15407)
    with pytest.raises(QueryFailed, match="no acceptable reply"):
        ask_udp("127.0.0.1", "www.example.com", port=15407, timeout=0.3)


def test_a_closed_port_fails_fast():
    started = time.monotonic()
    with pytest.raises(QueryFailed):
        ask_udp("127.0.0.1", "www.example.com", port=15408, timeout=3.0)
    assert time.monotonic() - started < 2.5


def test_a_reply_with_two_questions_is_rejected(serve):
    two = [("www.example.com", "A"), ("www.example.com", "A")]
    serve(lambda q, t, a: encode_message(q.header.id, 0x8400, two, [("www.example.com", "A", 60, "6.6.6.6")]), 15410)
    with pytest.raises(QueryFailed, match="exactly one question"):
        ask_udp("127.0.0.1", "www.example.com", port=15410, timeout=0.4)


def test_we_pick_unpredictable_source_ports():
    ports = []
    for _ in range(50):
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
            bind_random_port(sock, socket.AF_INET)
            ports.append(sock.getsockname()[1])
    assert all(49152 <= port <= 65535 for port in ports)
    assert sum(b == a + 1 for a, b in zip(ports, ports[1:])) < 5


def test_a_truncated_udp_reply_triggers_tcp(serve):
    port = 15409
    serve(lambda q, t, a: encode_message(q.header.id, 0x8600, [(q.questions[0].name, "A")]), port)   # TC bit, no answers
    listener = socket.socket()
    listener.bind(("127.0.0.1", port))
    listener.listen(1)

    def tcp_side():
        conn, _ = listener.accept()
        with conn:
            size = struct.unpack(">H", conn.recv(2))[0]
            wire = answer(parse_message(conn.recv(size)))
            conn.sendall(struct.pack(">H", len(wire)) + wire)
        listener.close()

    threading.Thread(target=tcp_side, daemon=True).start()
    got = ask("127.0.0.1", "www.example.com", port=port)
    assert got.transport == "tcp" and got.message.answers[0].data == "192.0.2.1"

Last, the resolver, including the glue-less delegation, the referral loop, the poisoning attack from Step 8, and cache expiry with a fake clock:

# test_dns.py (continued)
# ---------------------------------------------------------------- the resolver
WORLD_PORT = 15500


def add_extras(zones):
    """A CNAME into another zone, a delegation without glue, and a server that refers to itself."""
    ips = WORLD_IPS
    zones[ips["evil"]].records.append(("alias.evil.test", "CNAME", 300, "www.bank.test"))
    zones[ips["test"]].delegations.update({
        "shop.test": [("ns.hosting.test", None)],
        "hosting.test": [("ns.hosting.test", "127.0.0.5")],
        "loop.test": [("ns.loop.test", "127.0.0.7")]})
    zones["127.0.0.5"] = Zone("hosting.test", [("ns.hosting.test", "A", 300, "127.0.0.6")])
    zones["127.0.0.6"] = Zone("shop.test", [("www.shop.test", "A", 300, "203.0.113.5")])
    zones["127.0.0.7"] = Zone("loop.test", delegations={"loop.test": [("ns.loop.test", "127.0.0.7")]})


@pytest.fixture
def world():
    running = []

    def start(poisoned=False, extras=False):
        zones = make_world(poisoned)
        if extras:
            add_extras(zones)
        running.extend(start_world(WORLD_PORT, zones))

    yield start
    for server in running:
        server.stop()


def resolver(**options):
    return Resolver(roots={"root": WORLD_IPS["root"]}, port=WORLD_PORT, timeout=1.0, **options)


class Clock:
    now = 0.0

    def __call__(self):
        return self.now


def test_the_walk_goes_root_then_tld_then_zone(world):
    world()
    result = resolver().resolve("www.evil.test")
    assert result.status == "ok" and result.queries == 3
    assert [r.data for r in result.answers] == ["203.0.113.66"]
    assert [line.split()[0] for line in result.trace] == [".", "test", "evil.test"]


def test_a_second_lookup_in_a_known_zone_costs_one_query(world):
    world()
    r = resolver()
    r.resolve("www.evil.test")
    assert r.resolve("mail.evil.test").queries == 1


def test_names_in_results_are_lower_case_even_with_0x20(world):
    world()
    assert resolver(use_0x20=True).resolve("www.evil.test").answers[0].name == "www.evil.test"


def test_a_cname_into_another_zone_is_followed(world):
    world(extras=True)
    result = resolver().resolve("alias.evil.test")
    assert [r.rtype for r in result.answers] == [5, 1] and result.answers[-1].data == "192.0.2.10"


def test_nodata_and_nxdomain_are_told_apart(world):
    world()
    r = resolver()
    assert r.resolve("www.bank.test", "MX").status == "nodata"
    assert r.resolve("nope.bank.test").status == "nxdomain"


def test_a_name_server_without_glue_is_looked_up(world):
    world(extras=True)
    r = resolver()
    result = r.resolve("www.shop.test")
    assert result.answers[0].data == "203.0.113.5" and result.queries == 5
    assert r.cache.zones["shop.test"][1] == ["127.0.0.6"]


def test_a_server_that_refers_to_itself_is_cut_off(world):
    world(extras=True)
    r = resolver()
    with pytest.raises(ResolveError, match="neither an answer nor a usable referral"):
        r.resolve("www.loop.test")
    assert r.queries <= 4


def test_a_strict_resolver_ignores_a_planted_delegation(world):
    world(poisoned=True)
    r = resolver(strict=True)
    r.resolve("www.evil.test")
    assert "bank.test" not in r.cache.zones
    assert r.resolve("www.bank.test").answers[0].data == "192.0.2.10"


def test_a_trusting_resolver_is_poisoned(world):
    world(poisoned=True)
    r = resolver(strict=False)
    r.resolve("www.evil.test")
    assert r.cache.zones["bank.test"][1] == [WORLD_IPS["attacker"]]
    assert r.resolve("www.bank.test").answers[0].data == "198.51.100.66"


def test_cached_delegations_expire_with_their_ttl(world):
    world()
    clock = Clock()
    r = resolver(cache=Cache(clock))
    assert r.resolve("www.evil.test").queries == 3
    assert r.resolve("mail.evil.test").queries == 1
    clock.now = 172801
    assert r.resolve("mail.evil.test").queries == 3

python -m pytest -q test_dns.py

...............................................                                                                                                                    [100%]
47 passed in 14.11s

Layer 3: plant bugs and see whether the tests notice

A passing suite proves little if the tests would also pass on broken code. A mutation check tests the tests: it copies the code, plants one bug at a time, runs the suite, and expects a failure. This script plants 22 bugs, such as letting pointers point forwards, accepting any reply ID, leaving the UDP socket unconnected, never expiring the cache, and removing the bailiwick checks:

"""mutations.py: plant one bug at a time in a copy of the code and check that test_dns.py notices."""
import os
import pathlib
import shutil
import subprocess
import sys
import tempfile

HERE = pathlib.Path(__file__).parent
MUTANTS = [
    ("pointers may point forwards", "dnswire.py", [("if target >= pos:", "if target >= len(msg):")]),
    ("pointers may point into the header", "dnswire.py", [("if target < 12:", "if target < 0:")]),
    ("no limit on compression jumps", "dnswire.py", [("if jumps > MAX_JUMPS:", "if jumps > 10 ** 6:")]),
    ("no limit on name length", "dnswire.py", [("if size > MAX_NAME:", "if size > 10 ** 6:")]),
    ("reserved label types accepted", "dnswire.py", [("if length & 0xC0:\n", "if False:\n")]),
    ("top bit of the TTL kept", "dnswire.py", [('if ttl & 0x80000000 and rtype != TYPES["OPT"]:', "if False:")]),
    ("NS name may leave bytes over", "dnswire.py",
     [("name, after = read_name(msg, pos)\n        if after != end:", "name, after = read_name(msg, pos)\n        if False:")]),
    ("bytes after the last record accepted", "dnswire.py", [("if pos != len(msg):", "if False:")]),
    ("control bytes printed raw", "dnswire.py",
     [("elif 0x21 <= byte < 0x7F or (space_ok and byte == 0x20):", "elif True:")]),
    ("reply id not checked", "dnsnet.py", [("if message.header.id != query_id:", "if False:")]),
    ("only one question not required", "dnsnet.py", [("if len(message.questions) != 1:", "if False:")]),
    ("0x20 case compared loosely", "dnsnet.py",
     [("same_name = question.name == name if exact_case else question.name.lower() == name.lower()",
       "same_name = question.name.lower() == name.lower()")]),
    ("UDP socket not connected", "dnsnet.py",
     [("sock.connect((server, port))", "pass"), ("sock.send(wire)", "sock.sendto(wire, (server, port))")]),
    ("source port left to the OS", "dnsnet.py", [("port = 49152 + secrets.randbelow(65536 - 49152)", "port = 0")]),
    ("truncated replies not retried over TCP", "dnsnet.py", [("if reply.message.header.tc and allow_tcp:", "if False:")]),
    ("bailiwick checks removed", "resolver.py",
     [('if r.rtype != TYPES["NS"] or (self.strict and not under(owner, zone)):', 'if r.rtype != TYPES["NS"]:'),
      ("if r.rtype == TYPES[\"A\"] and (not self.strict or under(r.name.lower(), zone)):",
       'if r.rtype == TYPES["A"]:')]),
    ("referral to the same zone accepted", "resolver.py", [(" and r.name.lower() != zone", "")]),
    ("cache never expires", "resolver.py", [("if entry and entry[0] > self.clock():", "if entry:")]),
    ("names not lower-cased", "resolver.py",
     [("wanted = [tidy(r) for r in message.answers if r.name.lower() == name]",
       "wanted = [r for r in message.answers if r.name.lower() == name]")]),
    ("name servers without glue never looked up", "resolver.py",
     [("yield from self._lookup_hosts([host], depth)", "pass")]),
    ("looked-up name server not cached", "resolver.py", [("if ttl and ip not in servers:", "if False:")]),
    ("CNAME links dropped from the answer", "resolver.py", [("chain.append(found)", "pass")]),
]


def run(title, filename, edits):
    work = pathlib.Path(tempfile.mkdtemp(prefix="mutant_"))
    try:
        for source in HERE.glob("*.py"):
            shutil.copy(source, work)
        text = (work / filename).read_text(encoding="utf-8")
        for old, new in edits:
            assert text.count(old) == 1, (title, old, text.count(old))
            text = text.replace(old, new)
        (work / filename).write_text(text, encoding="utf-8", newline="\n")
        env = dict(os.environ, PYTHONDONTWRITEBYTECODE="1", COLUMNS="170")
        try:
            done = subprocess.run([sys.executable, "-m", "pytest", "-x", "-q", "test_dns.py"], cwd=work, env=env,
                                  capture_output=True, text=True, timeout=180)
        except subprocess.TimeoutExpired:
            return "hang"
        failed = [ln for ln in done.stdout.splitlines() if ln.startswith("FAILED ")]
        return failed[0][len("FAILED test_dns.py::"):].split(" - ")[0] if failed else None
    finally:
        shutil.rmtree(work, ignore_errors=True)


caught = 0
for title, filename, edits in MUTANTS:
    result = run(title, filename, edits)
    caught += result is not None
    print(f"{'CAUGHT  ' if result else 'SURVIVED'} {title:<44} {result or ''}")
print(f"\n{caught} of {len(MUTANTS)} planted bugs made the tests fail")

python mutations.py

CAUGHT   pointers may point forwards                  test_hostile_names_are_refused_for_the_right_reason[pointer to itself]
CAUGHT   pointers may point into the header           test_hostile_names_are_refused_for_the_right_reason[pointer into the header]
CAUGHT   no limit on compression jumps                test_hostile_names_are_refused_for_the_right_reason[label then pointer back]
CAUGHT   no limit on name length                      test_hostile_names_are_refused_for_the_right_reason[name over 255 bytes]
CAUGHT   reserved label types accepted                test_hostile_names_are_refused_for_the_right_reason[label type that does not exist]
CAUGHT   top bit of the TTL kept                      test_ttl_with_the_top_bit_counts_as_zero
CAUGHT   NS name may leave bytes over                 test_an_ns_name_must_fill_the_record_data
CAUGHT   bytes after the last record accepted         test_bytes_after_the_last_record_are_refused
CAUGHT   control bytes printed raw                    test_txt_with_control_bytes_prints_safely
CAUGHT   reply id not checked                         test_a_wrong_id_is_ignored_and_the_real_reply_still_wins
CAUGHT   only one question not required               test_a_reply_with_two_questions_is_rejected
CAUGHT   0x20 case compared loosely                   test_0x20_demands_the_exact_case
CAUGHT   UDP socket not connected                     test_a_forgery_from_another_port_never_arrives
CAUGHT   source port left to the OS                   test_we_pick_unpredictable_source_ports
CAUGHT   truncated replies not retried over TCP       test_a_truncated_udp_reply_triggers_tcp
CAUGHT   bailiwick checks removed                     test_a_strict_resolver_ignores_a_planted_delegation
CAUGHT   referral to the same zone accepted           test_a_server_that_refers_to_itself_is_cut_off
CAUGHT   cache never expires                          test_cached_delegations_expire_with_their_ttl
CAUGHT   names not lower-cased                        test_names_in_results_are_lower_case_even_with_0x20
CAUGHT   name servers without glue never looked up    test_a_name_server_without_glue_is_looked_up
CAUGHT   looked-up name server not cached             test_a_name_server_without_glue_is_looked_up
CAUGHT   CNAME links dropped from the answer          test_a_cname_into_another_zone_is_followed

22 of 22 planted bugs made the tests fail

All 22 planted bugs made a test fail, and each failure names the test that caught it. That also settles a doubt from Step 3: each of the three name rules has its own test with its own expected reason, so removing any single rule is caught. If a mutant had survived, the fix would be a new test, not a shrug.

A second opinion: compare with dnspython

The last check asks whether your parser agrees with an independent one. This script sends 15 questions to five servers (two recursive resolvers, a root server, a com server, and an org server), parses every reply with your code and with dnspython, and compares each record in each section:

"""crosscheck.py: parse live replies with our parser and with dnspython, and compare every record. Needs: pip install dnspython"""
import io
import socket

import dns.message
import dns.version

from dnswire import TYPE_NAMES, build_query, escape, parse_message

SERVERS = [("1.1.1.1", True), ("8.8.8.8", True), ("198.41.0.4", False), ("192.5.6.30", False), ("199.19.56.1", False)]
QUESTIONS = [("example.com", "A"), ("example.com", "AAAA"), ("example.com", "TXT"), ("example.com", "SOA"),
             ("example.com", "NS"), ("gmail.com", "MX"), ("gmail.com", "TXT"), ("www.wikipedia.org", "CNAME"),
             ("iana.org", "MX"), ("google.com", "TXT"), (".", "DNSKEY"), (".", "NS"), ("org", "DNSKEY"),
             ("example.org", "DS"), ("this-name-does-not-exist-8c1d.org", "A")]


def name_text(name):
    return name.to_text().rstrip(".").lower()


def their_text(rdata, rtype):
    """dnspython's view of the record data, written in the same form our parser prints."""
    kind = TYPE_NAMES.get(rtype)
    if kind in ("A", "AAAA"):
        return rdata.address
    if kind in ("NS", "CNAME", "PTR"):
        return name_text(rdata.target)
    if kind == "MX":
        return f"{rdata.preference} {name_text(rdata.exchange) or '.'}"
    if kind == "TXT":
        return " ".join('"' + escape(s, b'\\"', space_ok=True) + '"' for s in rdata.strings)
    if kind == "SOA":
        return (f"{name_text(rdata.mname) or '.'} {name_text(rdata.rname) or '.'} {rdata.serial} {rdata.refresh} "
                f"{rdata.retry} {rdata.expire} {rdata.minimum}")
    wire = io.BytesIO()
    rdata.to_wire(wire)
    return f"\\# {len(wire.getvalue())} {wire.getvalue().hex()}"


replies = compared = differences = 0
for server, recursion in SERVERS:
    for name, qtype in QUESTIONS:
        _, wire = build_query(name, qtype, recursion_desired=recursion, edns_size=1232, dnssec_ok=True)
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
            sock.settimeout(2)
            sock.connect((server, 53))
            sock.send(wire)
            try:
                data = sock.recv(65535)
            except OSError:
                continue
        replies += 1
        ours, theirs = parse_message(data), dns.message.from_wire(data)
        sections = ((ours.answers, theirs.answer), (ours.authority, theirs.authority), (ours.additional, theirs.additional))
        for our_records, their_rrsets in sections:
            want = sorted((name_text(rrset.name), rrset.rdtype, their_text(rdata, rrset.rdtype))
                          for rrset in their_rrsets for rdata in rrset)
            have = sorted((r.name.lower(), r.rtype, r.data) for r in our_records if r.rtype != 41)   # 41 is the OPT record
            compared += len(have)
            if have != want:
                differences += 1
                print("DIFFERENT:", server, name, qtype)
        if ours.rcode != theirs.rcode() or ours.header.id != theirs.id:
            differences += 1
            print("DIFFERENT header:", server, name, qtype)
print(f"{replies} live replies, {compared} records compared against dnspython {dns.version.version}, {differences} differences")

python crosscheck.py

74 live replies, 876 records compared against dnspython 2.9.2, 0 differences

The replies included DNSSEC record types: DNSKEY, RRSIG, DS, and NSEC3 (type 50, which our parser shows as TYPE50). Our parser prints these as hex, and the comparison found no differences. In my first, larger version of this check, 104 replies and 1,223 records also matched. Agreement with a second implementation is not proof of correctness, but it would catch a whole class of bugs that your own tests were written around.

Run it end to end

The last file is a command-line tool that ties the modules together. It resolves a name from the root servers down and prints the trace and the answer, and its exit code reports the outcome:

"""dig.py: look a name up the long way, from the root servers down, and print what we find."""
import argparse
import sys

from dnswire import format_record
from resolver import ResolveError, Resolver


def main():
    parser = argparse.ArgumentParser(description="Resolve a name iteratively, starting at the root servers.")
    parser.add_argument("name")
    parser.add_argument("qtype", nargs="?", default="A")
    parser.add_argument("--trace", action="store_true", help="show every server we asked")
    args = parser.parse_args()
    try:
        result = Resolver().resolve(args.name, args.qtype.upper())
    except ResolveError as exc:
        print(f";; resolution failed: {exc}")
        return 2
    if args.trace:
        for line in result.trace:
            print(f";; {line}")
    print(f";; status: {result.status}, {result.queries} queries")
    for record in result.answers:
        print(format_record(record))
    return 0 if result.status == "ok" else 1


if __name__ == "__main__":
    sys.exit(main())

python dig.py www.wikipedia.org --trace

;; .              198.41.0.4       www.wikipedia.org A: referral to org (6 name servers, 6 with addresses) (461 bytes, 18 ms, udp)
;; org            199.19.56.1      www.wikipedia.org A: referral to wikipedia.org (3 name servers, 3 with addresses) (242 bytes, 43 ms, udp)
;; wikipedia.org  208.80.154.238   www.wikipedia.org A: CNAME to dyna.wikimedia.org (75 bytes, 47 ms, udp)
;; org            199.19.56.1      dyna.wikimedia.org A: referral to wikimedia.org (3 name servers, 3 with addresses) (233 bytes, 48 ms, udp)
;; wikimedia.org  208.80.153.231   dyna.wikimedia.org A: answer, 1 record(s) (63 bytes, 55 ms, udp)
;; status: ok, 5 queries
www.wikipedia.org.             86400   IN  CNAME  dyna.wikimedia.org
dyna.wikimedia.org.            180     IN  A      208.80.154.224

That is a complete resolution: the root refers to org, an org server refers to wikipedia.org, a wikipedia.org server returns a CNAME to a name in a different zone, so the resolver goes back to org to find wikimedia.org, and the answer arrives on the fifth query. To confirm that everything works, run these four commands. Each should finish without an error:

python step01_build_query.py
python -m pytest -q test_dns.py
python dig.py example.com MX
python crosscheck.py

The exit code of dig.py is 0 for an answer, 1 for NODATA or NXDOMAIN, and 2 if the resolution failed. One more command shows a detail that the simulation of this very tutorial caught in my own code:

python dig.py example.com MX

;; status: ok, 3 queries
example.com.                   300     IN  MX     0 .

The answer is a null MX: preference 0 and the root name, which is written as a dot. RFC 7505 says that this record “formalizes the existing mechanism by which a domain announces that it accepts no mail, without having to provide a mail server”. My first version printed the root name as an empty string, so the line ended in a bare 0 and looked broken. Inside the parser the root name is the empty string, so the printing code now writes a dot, as dig does, and a test covers it.

Common mistakes and how to fix them

Where to go next

• Cache answers and failures. Store final answers by their TTL, and cache negative answers using the SOA record, as RFC 2308 describes.
• Validate DNSSEC. Your parser already reads RRSIG and DNSKEY records. Our post on the root key rollover explains why trust anchors matter and what happens when they change.
• Ask less. Query name minimisation, RFC 9156, sends each server only the part of the name it needs. DNS cookies, RFC 7873, add another cheap defense against forged replies.
• Encrypt the transport. DNS over TLS, RFC 7858, and DNS over HTTPS, RFC 8484, reuse your message code with a different socket.
• Pick servers by speed, and retry sensibly. Our timeouts and deadlines tutorial covers budgets and retries that apply here too.
• Use DNS from the other side. The SPF, DKIM, and DMARC audit reads TXT records the way Step 4 does, and the decompression bomb tutorial applies the same “hard limits on hostile input” thinking you used in Step 3.

Sources and further reading

• RFC 1035, Domain Names: Implementation and Specification (message format, compression, UDP limit), and RFC 1034, Domain Names: Concepts and Facilities.
• RFC 2181, Clarifications to the DNS Specification (TTL rule, ranking of data), RFC 5452, Measures for Making DNS More Resilient against Forged Answers, and RFC 9619, In the DNS, QDCOUNT Is (Usually) One.
• RFC 6891, Extension Mechanisms for DNS (EDNS(0)), and RFC 7766, DNS Transport over TCP.
• DNS Flag Day 2020 (the 1232-byte recommendation) and the 0x20 Internet-Draft, with Unbound’s configuration reference for how a production resolver treats it.
• RFC 7505, A “Null MX” No Service Resource Record for Domains That Accept No Mail.
• RFC 9824, Compact Denial of Existence in DNSSEC, and Cloudflare’s Making DNSSEC Answers Cheap post.
• RFC 6761, Special-Use Domain Names (the .test domain), and RFC 5737, IPv4 Address Blocks Reserved for Documentation.
• IANA’s root hints file and the Linux manual page for connect(2).