DNS Resolver
DNS Resolver
DNS wire format-এর query builder আর compression-pointer parser লিখে, hardcoded root hints থেকে root→TLD→authoritative iterative resolution — recursion নিষিদ্ধ। TTL cache, CNAME chasing, NXDOMAIN/SERVFAIL error path, আর ১০টা real domain resolve করা integration test দিয়ে শেষ।
কেন এই প্রজেক্ট
Dns লেসনে hierarchy আর record type-এর মানচিত্র দেখেছি; udp লেসনে
connectionless datagram। কিন্তু dig চালালে যা ঘটে তার গভীরতম অংশ
দেখা যায় না — iterative resolution। System resolver recursive
query পাঠায় (RD=1) আর উপরের দিকে বসে থাকে; নিজের resolver লিখলে
আপনিই হাঁটবেন root nameserver থেকে TLD থেকে authoritative server —
প্রতি step-এ referral NS record পড়ে, glue A record বের করে, next hop
ধরে।
সঙ্গে binary protocol-এর পুরো শৃঙ্খলা: DNS message fixed header, variable labels, আর সবচেয়ে চটকদার — compression pointer (0xC0 prefix), যা packet-এর আগের অংশে jump করে domain name repeat বাঁচায়। এই pointer-কে ভুল handle করলে infinite loop — এটাই প্রজেক্টের সবচেয়ে জনপ্রিয় bug।
ধাপে ধাপে
১. Query builder
import os, struct
def build_query(name, qtype=1): # 1 = A
qid = os.urandom(2)
flags = 0x0000 # RD=0 — iterative!
header = struct.pack('>HHHHHH', int.from_bytes(qid, 'big'),
flags, 1, 0, 0, 0)
qname = b''
for label in name.split('.'):
qname += bytes([len(label)]) + label.encode()
qname += b'\x00'
return qid + header[2:] + qname + struct.pack('>HH', qtype, 1)Header layout: ID(16) | Flags(16) | QDCOUNT | ANCOUNT | NSCOUNT |
ARCOUNT — সব big-endian। Flags-এ QR=0 (query), Opcode 0, RD=0। RD
বিটটাই এই প্রজেক্টের thesis: ওটা set করলে আপনি stub client, clear
রাখলে আপনি নিজেই walker। যাচাই প্রথম দিনেই: dig +norecurse @198.41.0.4 example.com আপনার builder-এর output mirror করে — দুটো packet
side-by-side hexdump মিলিয়ে দেখুন।
২. Response parser + compression pointers
def read_name(pkt, off):
labels, end, jumps = [], None, 0
while True:
length = pkt[off]
if length & 0xC0 == 0xC0: # compression pointer!
if end is None:
end = off + 2 # ফিরে আসার জায়গা মনে রাখুন
ptr = ((length & 0x3F) << 8) | pkt[off + 1]
jumps += 1
if jumps > 128 or ptr >= len(pkt): # self/forward pointer guard
raise ValueError('bad pointer')
off = ptr
continue
off += 1
if length == 0:
break
labels.append(pkt[off:off+length].decode())
off += length
return '.'.join(labels), (end if end else off)
def parse_rr(pkt, off):
name, off = read_name(pkt, off)
rtype, _cls, ttl, rdlen = struct.unpack('>HHIH', pkt[off:off+10])
off += 10
rdata = pkt[off:off+rdlen]
return (name, rtype, ttl, rdata), off + rdlen
def parse_message(pkt):
_, flags, qd, an, ns, ar = struct.unpack('>HHHHHH', pkt[:12])
rcode = flags & 0xF
off = 12
for _ in range(qd): # question skip
_, off = read_name(pkt, off); off += 4
sections = []
for count in (an, ns, ar): # answer, authority, additional
recs, off_now = [], off
for _ in range(count):
rr, off_now = parse_rr(pkt, off_now)
recs.append(rr)
sections.append(recs)Pointer rule: প্রথম ২ বিট 11 মানে বাকি 14 বিট offset — message-এর শুরু থেকে গণনা, বর্তমান position থেকে নয়। Loop guard: pointer chain-এ visited count cap (যেমন 128) — malicious packet-এ self-pointing pointer থাকতেই পারে। Resource record: NAME | TYPE(2) | CLASS(2) | TTL(4) | RDLENGTH(2) | RDATA — A-র RDATA ঠিক ৪ byte IPv4।
Checkpoint: dig +noedns example.com A response-এর raw bytes
(dig ... +qr +noedns দিয়ে capture) আপনার parser-এ ঢোকান — answer
count, TTL, IP তিনটাই dig report-এর সঙ্গে মিলবে।
৩. Iterative resolution — root থেকে authoritative
Root hints hardcoded (a.root-servers.net … m.root-servers.net-এর IP),
যেমন 198.41.0.4, 192.33.4.12, 199.7.91.13:
def resolve(name):
servers = ROOT_HINTS[:]
while True:
resp = udp_query(build_query(name), servers)
ans = answers_for(resp, name)
if ans: return ans
ns_ips = []
for rr in resp.authority:
if rr.type == 2: # NS
ns_names.append(rr.rdata)
for rr in resp.additional:
if rr.type == 1 and rr.name in ns_names:
ns_ips.append(rr.rdata) # glue record
if not ns_ips: # glue নেই — NS address নিজে resolve
ns_ips = [resolve(ns_names[0])]
servers = ns_ips # next hop: আরও নিচের স্তরপ্রতিটা non-answer response-এ তিনটা section আলাদা করুন: answer (খালি),
authority (NS referrals), additional (glue A records)। Root আপনাকে
.com TLD-র NS বলবে, TLD বলবে example.com-এর authoritative NS,
authoritative শেষে A record দেবে। Log-এ প্রতি hop print করুন:
root 198.41.0.4 → referral a.gtld-servers.net (192.5.6.30)
tld 192.5.6.30 → referral ns1.example.com (93.184.216.34)
auth 93.184.216.34 → A 93.184.216.34 ttl=3600Glue-missing edge case আসল: authoritative NS-এর নিজের name যদি তার নিচের zone-এ থাকে, additional section খালি — তখন NS hostname-টাই recurse resolve করতে হয় (cyclic dependency এড়াতে parent-এর glue-ই আসল উত্তর; না পেলে child-এ নামিয়ে দেখুন)।
৪. TTL cache
import time
class Cache:
def __init__(self):
self.store = {} # (name, type) -> (expires_at, records)
def get(self, name, rtype):
hit = self.store.get((name, rtype))
if hit and time.monotonic() < hit[0]:
return hit[1]
return NoneKey (name, type); expiry time.monotonic() + ttl — wall clock নয়,
NTP jump-এ cache ভেঙে যাবে। Negative caching-ও রাখুন: NXDOMAIN
response-এ SOA record-এর MINIMUM field = negative TTL — “এই নাম নেই”
জানাটাও cache-worthy। Cache hit-এ remaining TTL কমছে সেটা log করুন;
hit rate stats রাখলে ablation (cache on/off latency) টেবিল ready।
৫. CNAME chasing
www.github.com query করলে authoritative প্রথমে CNAME দেয়, A নয়:
www.github.com CNAME github.com ttl=3600
github.com A 140.82.x.x ttl=60Rule: answer section-এ requested type না পেয়ে CNAME পেলে — নতুন target
query শুরু (iterative walk আবার), alias chain ধরে রাখুন। Guard:
chase depth cap ৮ — CNAME loop (misconfigured zone) এ আটকে যাবেন না।
Final result-এ full chain report করুন, dig-এর >> chain-এর মতো।
৬. Error paths
| RCODE | মানে | আপনার আচরণ |
|---|---|---|
| 0 NOERROR + empty answer | নাম আছে, type নেই | NODATA — cache negative |
| 3 NXDOMAIN | নামই নেই | user-কে স্পষ্ট error, negative cache |
| 2 SERVFAIL | server ব্যর্থ | alternate server retry |
| timeout | packet হারায়নি তো? | retry ×2, তারপর next server |
UDP-র জগৎ: response আসতেই হবে না, duplicate আসতে পারে, ভিন্ন source থেকে spoof-ও আসতে পারে। Minimum defense: socket connect করা specific server-এ (unrelated packet ফেলে দিতে), transaction ID match check — spoofed response ID mismatch-এ discard। এটাই DNS spoofing-এর বিরুদ্ধে প্রাচীনতম ঢাল, আপনি এখন নিজে বুঝবেন কেন আধুনিক resolver-এ আরো স্তর (DNSSEC, random port) দরকার।
৭. Integration test — real domains
DOMAINS = ["wikipedia.org", "github.com", "google.com", "cloudflare.com",
"amazon.co.jp", "bbc.co.uk", "mit.edu", "python.org",
"kernel.org", "archive.org"]
for d in DOMAINS:
mine = resolver.resolve(d)
theirs = {sock[4][0] for sock in socket.getaddrinfo(d, 443, proto=socket.IPPROTO_TCP)}
assert any(ip in theirs for ip in mine), dIP set exact-match চাইবেন না — CDN geo-response ভিন্ন হয়; membership check-ই যথেষ্ট। সঙ্গে মাপুন: cold (cache-empty) total time vs warm repeat — iterative walk কত RTT খায় তার টেবিল।
নিজেকে চ্যালেঞ্জ করুন
- AAAA (IPv6) আর MX record support — RDATA parsing টাইপভেদে ভিন্ন
- Truncation handling — TC flag দেখে TCP fallback (port 53 stream mode)
- EDNS0 OPT record — 4096-byte payload advertise করে বড় response
- Parallel queries — সব authoritative NS-কে একসঙ্গে জিজ্ঞেস, fastest wins (Happy Eyeballs spirit)
- Local cache-only mode — offline test fixture packets replay করে parser unit-test suite
- DNS-over-UDP spoofing demo — lab network-এ forged response ঢোকানো, ID check কেন অপর্যাপ্ত তার case study (ethics boundary মেনে)
যা এখানে শিখলেন, পরে যেখানে লাগবে
| এখানে | পরে |
|---|---|
| Binary wire-format craft | Packet Sniffer প্রজেক্ট — একই struct.unpack discipline |
| Compression pointer + loop guard | সব binary format parser-এ reference handling |
| Iterative delegation model | Certificate chain validation, routing table walk — একই trust ladder |
| TTL + monotonic cache | HTTP cache, ARP cache — সব freshness protocol |
| Timeout/retry/spoof-check | যেকোনো UDP service design-এর minimum hygiene |