1"""Chapter 4 practice: a toy recursive DNS resolver with a TTL cache (Level 3 Build)
2and the 'moved servers with a 1-day TTL' migration bug (Level 4 Debug).
3No real network calls: the DNS hierarchy is simulated with dictionaries.
4Run: python3 ch04_dns.py"""
5
6# Each "server" maps a question to an answer.
7# ("NS", zone) -> referral to the next server ("A", name) -> (ip, ttl)
8# ("CNAME", name) -> (target, ttl)
9ROOT = {"com.": "tld-com"}
10SERVERS = {
11 "tld-com": {"example.com.": "ns1.example.com"},
12 "ns1.example.com": {
13 ("A", "example.com."): ("203.0.113.10", 300),
14 ("CNAME", "www.example.com."): ("example.com.", 300),
15 ("A", "shop.example.com."): ("203.0.113.20", 300),
16 },
17}
18
19
20class Resolver:
21 """A recursive resolver: walks root -> TLD -> authoritative, and caches answers."""
22
23 def __init__(self, servers=SERVERS, root=ROOT, negative_ttl=60):
24 self.servers = servers
25 self.root = root
26 self.cache = {} # name -> (ip or None, expires_at)
27 self.queries = 0 # how many times we had to ask other servers
28 self.negative_ttl = negative_ttl
29
30 def _ask(self, server, question):
31 self.queries += 1
32 return self.servers[server].get(question)
33
34 def resolve(self, name, now=0, depth=0):
35 if depth > 8:
36 raise RuntimeError("CNAME loop")
37 hit = self.cache.get(name)
38 if hit and hit[1] > now:
39 return hit[0] # answered from cache
40
41 tld = name.rstrip(".").split(".")[-1] + "."
42 self.queries += 1 # ask the root
43 tld_server = self.root.get(tld)
44 zone = ".".join(name.rstrip(".").split(".")[-2:]) + "."
45 auth = self._ask(tld_server, zone) if tld_server else None
46 if auth is None:
47 self.cache[name] = (None, now + self.negative_ttl) # NXDOMAIN, cached briefly
48 return None
49
50 a = self._ask(auth, ("A", name))
51 if a:
52 ip, ttl = a
53 self.cache[name] = (ip, now + ttl)
54 return ip
55 cname = self._ask(auth, ("CNAME", name))
56 if cname:
57 target, ttl = cname
58 ip = self.resolve(target, now, depth + 1)
59 if ip:
60 self.cache[name] = (ip, now + ttl)
61 return ip
62 self.cache[name] = (None, now + self.negative_ttl)
63 return None
64
65
66# ---- Level 4: the migration ---------------------------------------------------------
67
68DAY = 86_400
69
70
71def migrate(old_ttl, lower_first):
72 """Simulate moving shop.example.com from the old IP to a new IP at time 0.
73 Returns what a resolver that cached the record just before the move answers
74 one hour after the move."""
75 record = {"ip": "203.0.113.20", "ttl": old_ttl}
76 cache = {}
77
78 def lookup(now):
79 hit = cache.get("shop")
80 if hit and hit[1] > now:
81 return hit[0]
82 cache["shop"] = (record["ip"], now + record["ttl"])
83 return record["ip"]
84
85 if lower_first:
86 record["ttl"] = 300 # 2 days before the move: lower the TTL
87 lookup(-2 * DAY + 10) # (old 1-day copies expire well before the move)
88 lookup(-60) # a user visits one minute before the move
89 record["ip"] = "198.51.100.7" # the move: new server, old server switched off
90 return lookup(3600) # an hour later
91
92
93def run_checks():
94 r = Resolver()
95 assert r.resolve("shop.example.com.", now=0) == "203.0.113.20"
96 first = r.queries
97 assert first == 3 # root, TLD, authoritative
98
99 assert r.resolve("shop.example.com.", now=100) == "203.0.113.20"
100 assert r.queries == first # cache hit: no new queries
101
102 assert r.resolve("shop.example.com.", now=301) == "203.0.113.20"
103 assert r.queries == first + 3 # TTL expired: asked again
104
105 assert r.resolve("www.example.com.", now=400) == "203.0.113.10" # followed the CNAME
106
107 before = r.queries
108 assert r.resolve("nope.example.com.", now=500) is None
109 after_miss = r.queries
110 assert r.resolve("nope.example.com.", now=510) is None
111 assert r.queries == after_miss > before # negative answer was cached
112
113 assert r.resolve("example.org.", now=600) is None # unknown TLD
114
115 # Level 4
116 assert migrate(old_ttl=DAY, lower_first=False) == "203.0.113.20" # still the OLD server
117 assert migrate(old_ttl=DAY, lower_first=True) == "198.51.100.7" # new server
118
119 print("All checks pass. Without lowering the TTL, an hour after the move users still get",
120 migrate(DAY, False))
121
122
123if __name__ == "__main__":
124 run_checks()