IP Datagram — best-effort delivery-র contract

লক্ষ্য: RFC 791-এর প্রতিটা rule একবার executable করা — parse, verify, build, fragment, reassemble — পাঁচ দক্ষতা এক file-এ।

#!/usr/bin/env python3
"""ip_toolkit.py — IPv4 header parse/build/checksum/fragmentation।"""
import struct, random

def checksum(data: bytes) -> int:
    """RFC 1071 ones'-complement sum (16-bit words)।"""
    if len(data) % 2: data += b"\x00"
    total = sum(struct.unpack(f"!{len(data)//2}H", data))
    while total >> 16:                       # carry fold-back
        total = (total & 0xFFFF) + (total >> 16)
    return (~total) & 0xFFFF

def parse(ip: bytes) -> dict:
    ver_ihl, tos, total_len, ident, fl_off, ttl, proto, \
        cks, src, dst = struct.unpack("!BBHHHBBH4s4s", ip[:20])
    ver, ihl = ver_ihl >> 4, ver_ihl & 0xF
    df, mf, off = bool(fl_off & 0x4000), bool(fl_off & 0x2000), (fl_off & 0x1FFF) * 8
    ip_str = lambda b: ".".join(map(str, b))
    return dict(ver=ver, ihl_words=ihl, dscp=tos >> 2, ecn=tos & 3,
                total=total_len, ident=ident, df=df, mf=mf, offset=off,
                ttl=ttl, proto={1:"ICMP",6:"TCP",17:"UDP"}.get(proto, str(proto)),
                cks_ok=(checksum(ip[:10] + b"\x00\x00" + ip[12:20]) == cks),
                src=ip_str(src), dst=ip_str(dst),
                opts_bytes=(ihl - 5) * 4)

def build(src, dst, proto=17, ttl=64, ident=None, df=True, payload=b""):
    ident = random.randint(0, 65535) if ident is None else ident
    flags_off = (0x4000 if df else 0)          # DF set, offset 0
    hdr = struct.pack("!BBHHHBBH4s4s",
                      0x45, 0, 20 + len(payload), ident, flags_off,
                      ttl, proto, 0,
                      bytes(int(x) for x in src.split(".")),
                      bytes(int(x) for x in dst.split(".")))
    hdr = hdr[:10] + struct.pack("!H", checksum(hdr)) + hdr[12:]
    return hdr + payload

def fragment(datagram: bytes, mtu: int):
    ihl = (datagram[0] & 0xF) * 4
    hdr, data = datagram[:ihl], datagram[ihl:]
    chunk = ((mtu - ihl) // 8) * 8              # 8-byte align!
    frags, off = [], 0
    while off < len(data):
        piece = data[off : off + chunk]
        more = off + len(piece) < len(data)
        fo = struct.unpack("!H", hdr[6:8])[0] & 0xE000
        fo |= (off // 8) | (0x2000 if more else 0)
        fh = hdr[:6] + struct.pack("!H", fo) + hdr[8:]
        fh = fh[:10] + struct.pack("!H", checksum(fh)) + fh[12:]
        frags.append(fh + piece)
        off += len(piece)
    return frags

def reassemble(frags: list[bytes]) -> bytes | None:
    pieces, ident = {}, None
    for f in frags:
        ihl = (f[0] & 0xF) * 4
        ident = struct.unpack("!H", f[4:6])[0]
        mf, off = bool(int.from_bytes(f[6:8],"big") & 0x2000), \
                  (int.from_bytes(f[6:8], "big") & 0x1FFF) * 8
        pieces[off] = f[ihl:]                    # overlap silently wins — real stacks detect!
    if any(mf for f in frags):                   # last frag missing?
        if not any((struct.unpack("!H", f[6:8])[0] & 0x2000) == 0
                   for f in frags): pass
    expected, got = 0, sorted(pieces)
    for o in got:                                # gap check
        if o != expected: return None
        expected += len(pieces[o])
    print(f"  reassembled id={ident}: {expected} bytes")
    return b"".join(pieces[o] for o in got)

payload = bytes(random.getrandbits(8) for _ in range(300))
dgram = build("192.168.1.20", "93.184.216.34", ident=777, payload=payload)
info = parse(dgram)
print("parsed:", {k: info[k] for k in ("ver","ihl_words","total",
      "ident","df","proto","cks_ok","src","dst")})
frags = fragment(dgram, mtu=128)
print(f"fragmented into {len(frags)} pieces @MTU128:")
for i, f in enumerate(frags):
    p = parse(f)
    print(f"  #{i} len={p['total']:4d} MF={int(p['mf'])} off={p['offset']}")
back = reassemble(frags)
assert back == payload and back is not None
print("roundtrip OK: fragments == original payload")

প্রত্যাশিত আউটপুর skeleton:

parsed: {'ver': 4, 'ihl_words': 5, 'total': 320, 'ident': 777,
         'df': True, 'proto': 'UDP', 'cks_ok': True,
         'src': '192.168.1.20', 'dst': '93.184.216.34'}
fragmented into 3 pieces @MTU128:
  #0 len= 124 MF=1 off=0
  #1 len= 124 MF=1 off=112
  #2 len=  96 MF=0 off=224
  reassembled id=777: 300 bytes
roundtrip OK: fragments == original payload

সূক্ষ্ম শিক্ষা: fragment()-এ // 8 * 8 — offset-alignment-এর contract না-মানলে real receiver আপনার খণ্ড ফেলে দেবে; reassemble()- এ gap-detect — হারানো খণ্ডে None ফেরা বাধ্যতামূলক behavior।

নিজে বাড়ান:

1. reassemble()-এ overlap-detection যোগ করুন (teardrop-attack-এর defense) — দুই খণ্ড same-range দিলে reject
2. DF-set datagram fragment করতে চাইলে exception ছুড়ুন — router-এর আসল behavior
3. parse()-কে options-aware করুন: IHL>5 হলে option-TLV পড়ুন (record-route খুঁজুন)