Foundationপ্রথম নীতি থেকে
LEVEL 7অ্যাডভান্সড~২২ ঘণ্টাPython

Packet Sniffer

Packet Sniffer

Ethernet frame থেকে TCP segment পর্যন্ত প্রতিটা layer-এর parser নিজ হাতে লিখে live traffic analyser বানানো — scapy শুধু raw capture-এ, বাকি সব hand-written। Bidirectional flow table, live top-talkers dashboard, Wireshark-এ খোলা যায় এমন pcap export, আর documented ethics gate দিয়ে শেষ।

মাইলস্টোন

আগে যা পড়া দরকার

কেন এই প্রজেক্ট

packet-capture লেসনে promiscuous mode আর BPF-এর মানচিত্র দেখেছি; ip-datagram আর tcp-handshake-এ header layout গুলো চেনা। কিন্তু Wireshark খুললে যা দেখি — প্রতিটা packet ভেঙে রঙিন tree-তে সাজানো — সেই magic আসলে একের পর এক struct.unpack: Ethernet header পড়ে EtherType দেখা, IHL পড়ে TCP-তে নামা, flag bit পড়ে handshake চেনা। নিজের sniffer লিখলে প্রমাণ হয় protocol মানে কোনো enchantment নয় — শুধু documented offset, byte order, আর সেগুলোর সঠিক ক্রম।

সঙ্গে দুটো engineering শিক্ষা। এক — streaming discipline: বাস্তব link-এ প্রতি সেকেন্ডে হাজারখানেক packet আসে, তাই O(1) flow lookup আর bounded memory না-ভাবলে sniffer-ই নেটওয়ার্কের সবচেয়ে বড় bottleneck হয়ে দাঁড়াবে। দুই — power ও responsibility: আপনার tool Wireshark-এর মতোই সব দেখতে পারবে, তাই ethics boundary এখানে feature নয়, requirement — আর সেটা এই প্রজেক্টে code দিয়ে enforce করবেন।

DNS Resolver-এ একটা flat binary format parse করেছি; এখানে সেই discipline layer-এ layer — প্রতিটা decoder-ের output পরেরটার input, আর একটা offset ভুল হলে নিচের সব স্তর একসঙ্গে অর্থহীন।

ধাপে ধাপে

১. Raw capture — scapy শুধু courier

from scapy.all import sniff

counters = {"captured": 0, "decoded": 0, "rejected": 0}

def on_packet(pkt):
    raw = bytes(pkt)               # এখানেই scapy-র দায়িত্ব শেষ
    counters["captured"] += 1
    try:
        decode_ethernet(raw)       # পরের ধাপের chain — সব আপনার লেখা
        counters["decoded"] += 1
    except DecodeError:
        counters["rejected"] += 1

sniff(iface="en0", prn=on_packet, store=False)

দুটো শর্ত। store=False — না দিলে scapy প্রতিটা packet memory-তে জমায় রাখে, দশ মিনিটেই process swap-এ যাবে। আর filter= parameter ব্যবহার নিষেধ — BPF নিজেও একটা packet parser, সেই কাজটা এই প্রজেক্টে আপনিই করবেন। Raw socket খোলার জন্য sudo/root লাগবে — permission denied পেলে অবাক হবেন না।

Checkpoint: এক terminal-এ sniffer, আরেকটায় ping -c 3 1.1.1.1 — captured counter বাড়বে কিন্তু decoded থাকবে ০, কারণ এখনো কোনো decoder নেই। এই gap-টাই পরের সাতটা ধাপে বন্ধ করবেন।

২. Ethernet frame decoder

import struct

class DecodeError(Exception): pass

def mac(b):
    return ":".join(f"{x:02x}" for x in b)

def decode_ethernet(frame):
    if len(frame) < 14:                        # runt guard
        raise DecodeError("short frame")
    dst, src, etype = struct.unpack("!6s6sH", frame[:14])
    if etype != 0x0800:                        # এই প্রজেক্টে IPv4-ই যথেষ্ট
        raise DecodeError(f"etype {etype:#06x}")
    return {"dst": mac(dst), "src": mac(src)}, frame[14:]

Fixed 14-byte header: dst MAC(6) | src MAC(6) | EtherType(2), সব big-endian (!)। EtherType 0x0800 IPv4, 0x0806 ARP, 0x86DD IPv6 — বাকিগুলো ফেলে দিলেও একটা counter-এ রাখুন; আপনার LAN-এ যা যা ঘুরছে তার census মজার।

Checkpoint: decoded frame-এর dst MAC আপনার NIC-এর MAC-এর সঙ্গে মিলান (ifconfig en0) — সঙ্গে কয়েকটা ff:ff:ff:ff:ff:ff broadcast frame-ও দেখবেন, ARP-গুলো ওভাবেই আসে।

৩. IPv4 — variable header আর checksum

def ip(b):
    return ".".join(str(x) for x in b)

def ipv4_checksum(hdr):
    if len(hdr) % 2:
        hdr += b"\x00"
    total = sum(struct.unpack(f"!{len(hdr)//2}H", hdr))
    total = (total >> 16) + (total & 0xFFFF)   # carry fold
    total += total >> 16
    return (~total) & 0xFFFF                   # ones' complement

def decode_ipv4(data):
    if len(data) < 20:
        raise DecodeError("short IP")
    v_ihl, _tos, total_len, _id, _flg, ttl, proto, csum, src, dst = \
        struct.unpack("!BBHHHBBH4s4s", data[:20])
    if v_ihl >> 4 != 4:
        raise DecodeError("not IPv4")
    ihl = (v_ihl & 0xF) * 4                    # options-সহ আসল header length
    if ihl < 20 or ihl > len(data):
        raise DecodeError("bad IHL")
    if ipv4_checksum(data[:ihl]) != 0:         # valid হলে sum ঠিক শূন্য
        raise DecodeError("IP checksum mismatch")
    end = min(total_len, len(data))            # snaplen-এ কাটা পড়তে পারে
    return {"src": ip(src), "dst": ip(dst), "ttl": ttl}, proto, data[ihl:end]

তিনটা subtlety। IHL নিচের ৪ বিট, ৩২-bit word এককে — options থাকলে header ২০-এর বেশি, তাই payload data[20:] নয়, data[ihl:] থেকে। Checksum verify-র শর্টকাট: header-সহ পুরোটার sum ঠিক 0xFFFF হয়, complement নিলে 0x0000 — তাই আলাদা করে stored checksum বের করা লাগে না। আর total_len বনাম আসল captured length — capture snapshot ছোট হলে (truncation) packet-এর শেষাংশই নাও থাকতে পারে, min()-টা সেজন্য।

Checkpoint: নিজের machine-এর IP উৎস হিসেবে থাকা packets ছাপিয়ে tcpdump -ni en0 -c 5-এর সঙ্গে side-by-side মিলান। তারপর saboteur মোড: একটা byte ইচ্ছে করে flip করে rejection counter বাড়ছে দেখুন — checksum-এর জগতে এক বিট বদলালেই পুরো header অচল।

৪. TCP/UDP decoder chain

FLAG_BITS = [("FIN", 0x01), ("SYN", 0x02), ("RST", 0x04),
             ("PSH", 0x08), ("ACK", 0x10)]

def decode_tcp(seg):
    sport, dport, seq, ack, off_flags, win = struct.unpack("!HHIIHH", seg[:16])
    offset = (off_flags >> 12) * 4             # data offset, ৪-byte এককে
    flags = [name for name, bit in FLAG_BITS if off_flags & bit]
    return {"sport": sport, "dport": dport, "seq": seq, "ack": ack,
            "flags": "+".join(flags) or "-"}, seg[offset:]

def decode_udp(seg):
    sport, dport, ulen, _csum = struct.unpack("!HHHH", seg[:8])
    return {"sport": sport, "dport": dport}, seg[8:ulen]

DECODERS = {6: decode_tcp, 17: decode_udp}     # IP protocol number

TCP fixed header 20 byte, কিন্তু আমাদের দরকার প্রথম 16-টাই — data offset-এর উঁচু ৪ বিট বলে দেয় options-সহ header কতটুকু, payload তার পরেই শুরু। Flag গুলো bitmap: off_flags & 0x02 মানে SYN — এই একটা expression-ই tcp-handshake লেসনের পুরো গল্প পড়তে পারে। UDP মাত্র ৮ byte, আর তার length field-এ payload-সহ মোট দৈর্ঘ্য।

Checkpoint: sniffer চালু রেখে curl -s http://example.com > /dev/null — ঠিক এই ক্রমে তিনটা packet: SYN, SYN+ACK, ACK, তারপর PSH+ACK বহন করছে HTTP request। লেসনের diagram এবার আপনার terminal-এ বেঁচে উঠল।

৫. Bidirectional flow table

from collections import defaultdict

def flow_key(ip_hdr, tp_hdr):
    fwd = (ip_hdr["src"], tp_hdr["sport"], ip_hdr["dst"], tp_hdr["dport"])
    rev = (ip_hdr["dst"], tp_hdr["dport"], ip_hdr["src"], tp_hdr["sport"])
    return min(fwd, rev)                       # direction-agnostic key

flows = defaultdict(lambda: {"packets": 0, "bytes": 0,
                             "syn_seen": False, "established": False})

def account(ip_hdr, tp_hdr, nbytes):
    f = flows[flow_key(ip_hdr, tp_hdr)]
    f["packets"] += 1
    f["bytes"] += nbytes
    flags = tp_hdr["flags"]
    if "SYN" in flags and "ACK" not in flags:
        f["syn_seen"] = True                   # connection opener
    elif "SYN" in flags and f["syn_seen"]:
        f["established"] = True

মূল ভাবনা canonical key: একই conversation-এর A→B আর B→A packet দুই দিক থেকে আসে, কিন্তু min() tuple-দুটোর মধ্যে ছোটটা বেছে নেয় — তাই দুই direction-ই একই entry-তে পড়ে, আলাদা comparison code ছাড়াই। State machine-টা ইচ্ছাকৃতভাবে ন্যূনতম: SYN দেখা → SYN+ACK দেখা = established; FIN/RST handling আর idle eviction challenge হিসেবে রাখুন।

একটা সংখ্যা মনে রাখুন: dict-এ প্রতি flow এক entry, আর ব্যস্ত host-এ flow লক্ষ ছুঁয়ে ফেলে। Unbounded growth এখানে theoretical সমস্যা নয় — ঘণ্টাখানেক চালালেই বাস্তব।

৬. Live top-talkers dashboard

import sys
from collections import Counter

talkers = Counter()                            # (src, dst) -> bytes

def render():
    top = talkers.most_common(5)
    total = sum(talkers.values()) or 1
    sys.stdout.write("\x1b[2J\x1b[H")          # screen clear, cursor home
    sys.stdout.write(f"flows={len(flows):>6}  captured={counters['captured']}\n\n")
    for (src, dst), nbytes in top:
        bar = "#" * max(1, int(50 * nbytes / total))
        sys.stdout.write(f"{src:>15} -> {dst:<15} {nbytes:>12,d} {bar}\n")
    sys.stdout.flush()

while True:
    try:
        sniff(iface="en0", prn=on_packet, store=False, timeout=1)
        render()
    except KeyboardInterrupt:
        break
render()
print("\ndone — final table above stays on screen")

Hot path-এ শুধু O(1) accounting (Counter update), rendering আলাদা — timeout=1 দিয়ে sniff-কে প্রতি সেকেন্ডে ফিরিয়ে এনে মূল loop থেকেই draw করা যায়, threading ছাড়া। প্রতি packet-এ print করার লোভ করবেন না — terminal-ই bottleneck হয়ে throughput ডুবিয়ে দেবে।

Checkpoint: একদিকে video streaming (download-heavy flow উপরে উঠবে), অন্যদিকে নিজের server-এ বড় ফাইল scp (upload direction-এর flow) — leaderboard-এ দুই ধরনের traffic আলাদা করে চোখে পড়বে।

৭. pcap writer — Wireshark যা পড়ে

import time

PCAP_GLOBAL = struct.pack("<IHHiIII",
                          0xA1B2C3D4,   # magic — endianness নিজেই বলে দেয়
                          2, 4,         # version 2.4
                          0, 0,         # thiszone, sigfigs
                          65535,        # snaplen
                          1)            # network = Ethernet linktype

class PcapWriter:
    def __init__(self, path):
        self.f = open(path, "wb")
        self.f.write(PCAP_GLOBAL)

    def add(self, raw):
        now = time.time()
        sec, frac = int(now), int((now - now // 1) * 1_000_000)
        self.f.write(struct.pack("<IIII", sec, frac, len(raw), len(raw)))
        self.f.write(raw)                      # decoder-এ যাওয়া সেই raw-ই

    def close(self):
        self.f.close()

Format দুই স্তরের: একবার ২৪-byte global header, তারপর প্রতি packet-এ ১৬-byte record header (ts_sec, ts_usec, incl_len, orig_len) + exact frame bytes। incl_len আর orig_len আলাদা field — truncated capture-তে ভিন্ন হয়; আপনার এখানে সমান থাকবে, কিন্তু field-দুটো আলাদা করে লেখা মানে format-টা বুঝে লেখা। Magic 0xA1B2C3D4 little-endian-এ লিখলে reader endianness নিজেই detect করে।

Checkpoint: নিজের লেখা out.pcap খুলুন — wireshark out.pcap বা tshark -r out.pcap -c 5। আপনার hand-written bytes-কে Wireshark tree-তে সাজিয়ে দেখাবে; পুরো প্রজেক্টের সবচেয়ে satisfying moment এটা। সঙ্গে cross-check: Wireshark-এর frame count আপনার decoded counter-এর সঙ্গে মিলবে।

৮. Ethics checkpoint — gate হিসেবে agreement

এই tool Wireshark-এর মতোই powerful; পার্থক্য শুধু আপনি কোথায় চালাচ্ছেন। Rule: শুধু নিজের device-এ, নিজের control-এর network-এ — বাসার wifi, নিজের VM, lab-এর isolated segment। Campus, office, cafe-র shared network-এ “passive”-ও consent ছাড়া নয়; payload-এ password/email পড়ে যাওয়া এখানে accident নয়, design failure।

তাই agreement-টা document নয়, code: repo-তে USAGE.md না থাকলে sniffer শুরুই হবে না —

# Capture usage agreement
operator: <your name>
scope: home wlan, 192.168.1.0/24, own devices only
window: single session, max 60 minutes
payload: headers only — body inspection disabled
after: summary printed, pcap kept under captures/
import os, sys

if not os.path.exists("USAGE.md"):
    sys.exit("ethics gate: USAGE.md নেই — আগে scope লিখে commit করুন")

Startup-এ এই check, সঙ্গে capture-চলবস্থায় প্রতি মিনিটে scope reminder এক লাইন log-এ। Ethics-কে enforceable constraint বানানোই এই milestone-এর আসল deliverable — পরে যেকোনো security tooling-এ একই pattern বারবার লাগবে।

Checkpoint: USAGE.md rename করে দেখুন — sniffer refuse করবে; ফেরত আনলে চলবে। Gate-টা test suite-এ case হিসেবেও রাখুন।

নিজেকে চ্যালেঞ্জ করুন

  1. VLAN tag (EtherType 0x8100) unwrap করে ভেতরের আসল EtherType পড়া
  2. IPv4 fragmentation reassembly — MF flag + fragment offset, overlap handling সহ (কঠিন, সত্যিই)
  3. ICMP echo decode — নিজের sniffer দিয়ে ping-এর RTT মাপা
  4. Mini BPF — "tcp port 443" পড়ে decoder-gate-এ compile করা
  5. Idle eviction — flow table-এ ৫ মিনিট নিঃশব্দ flow বাদ, LRU-style
  6. pcap rotation — 100 MB পার হলে নতুন file, timestamp-এ নাম
  7. ARP listener — 0x0806 frame থেকে live ip↔mac map বানানো

যা এখানে শিখলেন, পরে যেখানে লাগবে

এখানেপরে
Layer-by-layer binary decodeDNS Resolver-এর wire-format craft-এর ধারাবাহিকতা — যেকোনো protocol dissector
Canonical flow keyNAT, connection tracking, stateful firewall
Streaming + bounded memoryLevel 8+ — log pipeline, metrics collector
pcap formatreplay-based testing, IDS dataset building
Ethics gate-as-codeসব security tooling-এ consent/scope enforcement