DNS — phonebook যা internet ধরে রাখে

লক্ষ্য: আজকের পুরো তত্ত্ব — header flags, RR ছাঁচ, 0xC0 pointer — একটা ~১১০ লাইনের সৎ parser-এ, কোনো DNS library ছাড়া (dnspython-র নামও নেই)। এটাই গত লেসনের sockets API-র UDP রূপের প্রথম বাস্তব কাজে লাগানো।

# dnsq.py — python3 dnsq.py www.example.com [resolver-ip]
import random
import socket
import struct
import sys


def encode_name(name: str) -> bytes:
    """'www.example.com' → 03www07example03com00"""
    wire = b""
    for label in name.rstrip(".").split("."):
        part = label.encode()
        wire += bytes([len(part)]) + part
    return wire + b"\x00"


def build_query(name: str, qtype: int = 1):
    """A query — RD=1 (recursion চাইছি), QCLASS=IN"""
    qid = random.randint(0, 0xFFFF)
    flags = 0x0100                      # RD bit
    header = struct.pack("!HHHHHH", qid, flags, 1, 0, 0, 0)
    return qid, header + encode_name(name) + struct.pack("!HH", qtype, 1)


def read_name(msg: bytes, off: int):
    """wire-থেকে নাম; 0xC0 pointer follow করে।
    return: (নাম, পরের field কোথা থেকে শুরু)"""
    labels, jumped, next_off = [], False, off
    while True:
        length = msg[off]
        if length & 0xC0 == 0xC0:            # compression pointer
            target = ((length & 0x3F) << 8) | msg[off + 1]
            if not jumped:
                next_off = off + 2           # ফেরার ঠিকানা একবারই
            off = target; jumped = True
        elif length == 0:                    # root label — নাম শেষ
            if not jumped:
                next_off = off + 1
            return ".".join(labels), next_off
        else:                                # সাধারণ label
            labels.append(msg[off + 1 : off + 1 + length].decode())
            off += 1 + length


TYPES = {1: "A", 2: "NS", 5: "CNAME", 6: "SOA", 15: "MX", 28: "AAAA"}
RCODES = {0: "NOERROR", 2: "SERVFAIL", 3: "NXDOMAIN", 5: "REFUSED"}


def parse_response(msg: bytes):
    rid, flags, qd, an, ns, ar = struct.unpack("!HHHHHH", msg[:12])
    fl = {"QR": (flags >> 15) & 1, "AA": (flags >> 10) & 1,
          "TC": (flags >> 9) & 1, "RD": (flags >> 8) & 1, "RA": (flags >> 7) & 1}
    print("id={} {}".format(rid, " ".join(f"{k}={v}" for k, v in fl.items())))
    print("RCODE={}  QD={} AN={} NS={} AR={}".format(
        RCODES.get(flags & 0xF), qd, an, ns, ar))

    off = 12
    for _ in range(qd):                      # question section পেরোনো
        _, off = read_name(msg, off)
        off += 4                             # QTYPE + QCLASS

    print("{:<26} {:>7} {:<7} RDATA".format("NAME", "TTL", "TYPE"))
    for count in (an, ns, ar):               # তিন RR section-ই এক ছাঁচে
        for _ in range(count):
            name, off = read_name(msg, off)
            rtype, _cls, ttl, rdlen = struct.unpack("!HHIH", msg[off:off + 10])
            off += 10
            end = off + rdlen
            if rtype == 1:                   # A — ৪ byte IPv4
                rdata = ".".join(str(b) for b in msg[off:end])
            elif rtype in (2, 5):            # NS / CNAME — RDATA নামই
                rdata, _ = read_name(msg, off)
            elif rtype == 15:                # MX — 2-byte pref + নাম
                pref = struct.unpack("!H", msg[off:off + 2])[0]
                host, _ = read_name(msg, off + 2)
                rdata = "{} {}".format(pref, host)
            elif rtype == 28:                # AAAA — 16 byte IPv6
                rdata = ":".join(msg[i:i + 2].hex() for i in range(off, end, 2))
            else:                            # SOA/TXT/OPT — আপাতত raw
                rdata = msg[off:end].hex()
            print("{:<26} {:>7} {:<7} {}".format(
                name, ttl, TYPES.get(rtype, rtype), rdata))
            off = end


def main():
    name = sys.argv[1] if len(sys.argv) > 1 else "www.example.com"
    server = (sys.argv[2] if len(sys.argv) > 2 else "1.1.1.1", 53)
    qid, query = build_query(name)
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)  # UDP-first!
    s.settimeout(3)
    s.sendto(query, server)
    reply, _ = s.recvfrom(4096)
    assert struct.unpack("!H", reply[:2])[0] == qid, "ID mismatch!"
    parse_response(reply)


main()

চালিয়ে দেখুন:

$ python3 dnsq.py www.example.com
id=39172 QR=1 AA=0 TC=0 RD=1 RA=1
RCODE=NOERROR  QD=1 AN=2 NS=0 AR=1
NAME                        TTL    TYPE    RDATA
www.example.com.             297   CNAME   example.com.
example.com.                 359   A       93.184.216.34
00fb01c00102030405...

তিনটা জিনিস লক্ষ্য করুন। এক — উত্তরের NAME কলামে পুরো নাম, অথচ wire-এ সেখানে মাত্র C0 0C ছিল: read_name pointer follow করে নাম পুনর্গঠন করেছে। দুই — TTL ৩৬০ না, ৩৫৯-ঘেঁঁটে: কপিটা cache-এ কিছুক্ষণ আগেই এসেছে (experiment-র countdown এখানেও)। তিন — শেষ row hex-এ, কারণ সেটা OPT (rtype 41) — EDNS0-র pseudo-record, hood-র 1232-byte কথার স্বাক্ষর; আসল RR না বলে decoder raw-ই রেখেছে।

নিজে বাড়ান:

1. TC=1 দেখলে একই query TCP-তে আবার পাঠানোর fallback লিখুন — বড় NS তালিকাওলা নাম দিয়ে পরীক্ষা করুন
2. read_name-এ visited-offset set যোগ করে pointer-loop guard দিন, তারপর হাতে C0 00 C0 00-জাতীয় দুষ্টু packet বানিয়ে guard যাচাই করুন
3. SOA আর TXT-র decode যোগ করুন — python3 dnsq.py example.com NS @a.gtld-servers.net চালিয়ে delegation-র NS তালিকা নিজের parser দিয়েই দেখুন