测试系统中有一项记录ssh登录日志,需要对此进行并发压力测试。
于是用多线程进行python并发记录
因为需要安装的一些依赖和模块比较麻烦,脚本完成后再用pyinstaller打成exe包分发给其他测试人员一起使用。
1.脚本编写
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 1# -*- coding: utf-8 -*-
2import paramiko
3import threading
4import time
5lt = []
6def ssh(a,xh,sp):
7 count = 0
8 for i in range(0,xh):
9 try:
10 ssh = paramiko.SSHClient()
11 ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
12 ssh.connect('ip地址',22,'用户名', '密码')
13 ssh.close()
14 print u"线程[%s]第[%s]次登录"%(a,i)
15 if sp != 0:
16 time.sleep(sp)
17 count += 1
18 except:
19 print u"线程异常,已处理"
20 lt.append(count)
21
22if __name__ == "__main__":
23 figlet = '''
24 _____ _____ _
25 | ___| | _ \ | |
26 | |__ | |_| | | |
27 | __| | _ { | |
28 | | | |_| | | |
29 |_| |_____/ |_|
30 Code by FBI.
31 '''
32 print figlet
33 print u"认证攻击次数=线程数*每个线程认证攻击次数"
34 print u"请输入线程数:"
35 xc = raw_input()
36 print u"请输入每个线程攻击次数:"
37 xh = raw_input()
38 print u"请输入每个线程延迟时间(秒),0为不休眠:"
39 sp = raw_input()
40 try:
41 print u"预计总共发送认证攻击%s次"%(int(xc)*int(xh))
42 threads = []
43 for j in range(int(xc)):
44 threads.append(threading.Thread(target=ssh,args=(j,int(xh),int(sp),)))
45 for t in threads:
46 t.start()
47 print t.name
48
49 t.join()
50
51 print lt
52 count = 0
53 for count in lt:
54 count += count
55 print u"程序执行完毕总共发送认证攻击【%s】次" % count
56 except ValueError,e:
57 print u"因为输入不规范导致程序出现错误,请输入数字"
58
59
2.pyinstaller制作exe程序
下载pyinstaller后
在根目录中cmd中执行python setup.py install安装pyinstaller
安装完成后执行命令打成exe文件
python pyinstaller.py -F 文件路径
3.执行效果
如图: