Python Scapy Packet Crafting — Deep Dive

Master Scapy for advanced packet crafting — custom protocol layers, traffic analysis, security testing, and real-world network forensics with Python.

System-level framing

Scapy is not just a packet crafter — it is a domain-specific language for network protocols, implemented as a Python library. Where most tools give you a fixed set of operations (scan ports, capture traffic, ping hosts), Scapy gives you primitive building blocks that combine into any operation you can imagine. It supports over 500 protocol layers out of the box, and you can define your own.

This deep dive covers advanced crafting, custom protocols, traffic analysis, and security testing patterns.

Installation

pip install scapy

On Linux, Scapy uses raw sockets (requires root). On macOS, it uses /dev/bpf. On Windows, it uses Npcap.

Packet anatomy and layer stacking

from scapy.all import Ether, IP, TCP, Raw

# Full stack: Ethernet → IP → TCP → Payload
pkt = (
    Ether(dst="ff:ff:ff:ff:ff:ff")
    / IP(dst="10.0.0.1", ttl=64)
    / TCP(dport=443, sport=12345, flags="S", seq=1000)
    / Raw(load=b"Hello")
)

# Inspect the packet
pkt.show()

# Access specific layers
print(pkt[IP].dst)       # "10.0.0.1"
print(pkt[TCP].flags)    # "S"
print(bytes(pkt))        # Raw bytes
print(len(pkt))          # Total packet size

Layer stacking with / triggers automatic field computation. When you convert to bytes, Scapy calculates checksums, lengths, and padding.

Sending and receiving

from scapy.all import IP, TCP, ICMP, sr1, sr, send, sendp

# Send a SYN and wait for a response
syn = IP(dst="scanme.nmap.org") / TCP(dport=80, flags="S")
response = sr1(syn, timeout=3)
if response and response.haslayer(TCP):
    if response[TCP].flags == "SA":  # SYN-ACK
        print("Port 80 is open")
    elif response[TCP].flags == "RA":  # RST-ACK
        print("Port 80 is closed")

# ICMP ping
ping = IP(dst="8.8.8.8") / ICMP()
reply = sr1(ping, timeout=2)
if reply:
    print(f"Reply from {reply[IP].src}, TTL={reply[IP].ttl}")

TCP SYN scan implementation

from scapy.all import IP, TCP, sr

def syn_scan(target: str, ports: range) -> dict:
    """Perform a SYN scan and categorize ports."""
    results = {"open": [], "closed": [], "filtered": []}

    # Send SYN to all ports at once
    packets = IP(dst=target) / TCP(dport=list(ports), flags="S")
    answered, unanswered = sr(packets, timeout=3, verbose=0)

    for sent, received in answered:
        port = sent[TCP].dport
        if received.haslayer(TCP):
            if received[TCP].flags == 0x12:  # SYN-ACK
                results["open"].append(port)
            elif received[TCP].flags == 0x14:  # RST-ACK
                results["closed"].append(port)

    # Unanswered = filtered (firewall dropped the packet)
    for sent in unanswered:
        results["filtered"].append(sent[TCP].dport)

    return results

# Usage (requires root)
# results = syn_scan("scanme.nmap.org", range(1, 1025))

Traceroute implementation

from scapy.all import IP, TCP, sr

def tcp_traceroute(target: str, max_hops: int = 30, dport: int = 80):
    """TCP-based traceroute (works through some firewalls that block ICMP)."""
    for ttl in range(1, max_hops + 1):
        pkt = IP(dst=target, ttl=ttl) / TCP(dport=dport, flags="S")
        reply = sr1(pkt, timeout=2, verbose=0)

        if reply is None:
            print(f"{ttl:3d}  *  (no reply)")
        elif reply.haslayer(TCP):
            print(f"{ttl:3d}  {reply[IP].src}  [REACHED - port {'open' if reply[TCP].flags == 0x12 else 'closed'}]")
            break
        else:
            print(f"{ttl:3d}  {reply[IP].src}")

Sniffing and analysis

from scapy.all import sniff, IP, TCP, wrpcap

# Capture 100 packets with a BPF filter
packets = sniff(count=100, filter="tcp port 80", iface="eth0")

# Analyze
for pkt in packets:
    if pkt.haslayer(IP) and pkt.haslayer(TCP):
        print(f"{pkt[IP].src}:{pkt[TCP].sport} → {pkt[IP].dst}:{pkt[TCP].dport}")

# Save to pcap
wrpcap("capture.pcap", packets)

# Async sniffing with callback
def process_packet(pkt):
    if pkt.haslayer(TCP) and pkt[TCP].dport == 80:
        if pkt.haslayer(Raw):
            payload = pkt[Raw].load.decode(errors="ignore")
            if "GET " in payload or "POST " in payload:
                print(f"HTTP request from {pkt[IP].src}: {payload.split(chr(13))[0]}")

sniff(filter="tcp port 80", prn=process_packet, store=0, count=50)

Custom protocol definition

Define your own protocol layer:

from scapy.all import Packet, ByteField, ShortField, IntField, StrLenField
from scapy.all import bind_layers, IP, UDP

class CustomProtocol(Packet):
    name = "CustomProtocol"
    fields_desc = [
        ByteField("version", 1),
        ByteField("msg_type", 0),
        ShortField("length", None),
        IntField("sequence", 0),
        StrLenField("payload", b"", length_from=lambda pkt: pkt.length - 8),
    ]

    def post_build(self, pkt, pay):
        if self.length is None:
            length = len(pkt) + len(pay)
            pkt = pkt[:2] + length.to_bytes(2, "big") + pkt[4:]
        return pkt + pay

# Bind to UDP port 9999
bind_layers(UDP, CustomProtocol, dport=9999)

# Now Scapy auto-decodes packets on port 9999
pkt = IP(dst="10.0.0.1") / UDP(dport=9999) / CustomProtocol(
    msg_type=1, sequence=42, payload=b"hello"
)
pkt.show()

ARP network discovery

from scapy.all import Ether, ARP, srp

def discover_network(network: str = "192.168.1.0/24") -> list[dict]:
    """Discover active hosts on the local network via ARP."""
    arp = Ether(dst="ff:ff:ff:ff:ff:ff") / ARP(pdst=network)
    answered, _ = srp(arp, timeout=3, verbose=0)

    hosts = []
    for sent, received in answered:
        hosts.append({
            "ip": received[ARP].psrc,
            "mac": received[ARP].hwsrc,
        })

    return sorted(hosts, key=lambda h: h["ip"])

DNS query crafting

from scapy.all import IP, UDP, DNS, DNSQR, sr1

def dns_query(domain: str, server: str = "8.8.8.8", qtype: str = "A") -> list:
    """Craft and send a DNS query."""
    pkt = (
        IP(dst=server)
        / UDP(dport=53)
        / DNS(rd=1, qd=DNSQR(qname=domain, qtype=qtype))
    )

    response = sr1(pkt, timeout=3, verbose=0)
    if response is None:
        return []

    results = []
    for i in range(response[DNS].ancount):
        rr = response[DNS].an[i] if response[DNS].ancount == 1 else response[DNS].an[i]
        results.append(rr.rdata)

    return results

Packet manipulation and replay

from scapy.all import rdpcap, wrpcap, IP, TCP

# Load a pcap file
packets = rdpcap("original.pcap")

# Modify packets — change destination IP
modified = []
for pkt in packets:
    if pkt.haslayer(IP):
        pkt[IP].dst = "10.0.0.99"
        del pkt[IP].chksum  # Force recalculation
        if pkt.haslayer(TCP):
            del pkt[TCP].chksum
    modified.append(pkt)

# Save modified packets
wrpcap("modified.pcap", modified)

# Replay at original timing
from scapy.all import sendp
sendp(modified, inter=0.01, iface="eth0")

Deleting checksums before sending forces Scapy to recalculate them, which is necessary after modifying any header field.

Performance optimization

Scapy’s pure Python implementation is slow for high packet rates. Strategies:

from scapy.all import conf

# Use libpcap for faster sniffing
conf.use_pcap = True

# Disable automatic route lookups for crafted packets
conf.route.resync()

# For bulk sending, use sendpfast (requires tcpreplay)
from scapy.all import sendpfast
packets = [IP(dst="10.0.0.1")/TCP(dport=p, flags="S") for p in range(1, 1025)]
sendpfast(packets, pps=10000, iface="eth0")

For truly high-speed work (millions of packets per second), generate packets with Scapy but send them through a C-based tool or use Scapy only for analysis.

Security testing patterns

Firewall rule testing

def test_firewall_rules(target: str, ports: list[int]) -> dict:
    """Test which ports pass through a firewall."""
    results = {}
    for port in ports:
        # Test TCP
        tcp_pkt = IP(dst=target) / TCP(dport=port, flags="S")
        tcp_resp = sr1(tcp_pkt, timeout=2, verbose=0)
        results[f"tcp/{port}"] = "open" if tcp_resp else "filtered"

        # Test UDP
        udp_pkt = IP(dst=target) / UDP(dport=port) / Raw(load=b"\x00")
        udp_resp = sr1(udp_pkt, timeout=2, verbose=0)
        results[f"udp/{port}"] = "open/filtered" if not udp_resp else "closed"

    return results

IP spoofing detection

from scapy.all import IP, ICMP, sr1

def test_spoof_filtering(target: str, spoofed_src: str) -> str:
    """Test if the network filters spoofed source addresses."""
    pkt = IP(src=spoofed_src, dst=target) / ICMP()
    resp = sr1(pkt, timeout=3, verbose=0)
    if resp:
        return f"WARNING: Spoofed packet reached {target} and got a reply"
    else:
        return "Good: No reply (packet likely filtered by BCP38/ingress filtering)"

Tradeoffs

Approach	Pros	Cons
Scapy	Maximum flexibility, interactive, 500+ protocols	Slow (Python), requires root
nmap	Fast scanning, OS detection, script engine	Fixed operations, less customizable
hping3	Fast packet crafting, C-based	No Python API, fewer protocols
trafgen	Wire-speed packet generation	Low-level, no protocol decoding
dpkt	Fast packet parsing (C-optimized)	Read-only, no sending

Legal and ethical considerations

Only test on networks you own or have written authorization to test. Unauthorized packet injection is a criminal offense in most countries.
Document your testing scope before starting. Include IP ranges, time windows, and approved techniques.
Never use spoofed packets on production networks without explicit approval and safeguards.
ARP spoofing, even for testing, can disrupt production traffic. Use isolated lab networks.
Keep records of all tests for compliance and audit purposes.

One thing to remember: Scapy turns Python into a network protocol laboratory. You can build any packet, test any hypothesis, and analyze any traffic. The power is in combining simple primitives — layer stacking, send/receive, sniff — into arbitrarily complex network experiments.

pythonnetworkingsecurity