python多線程在滲透測試中的應用

難易程度：★★★
閱讀點：python; web安全;
文章做者：xiaoye
文章來源：i春秋
關鍵字：網絡滲透技術

前言

python是門簡單易學的語言，強大的第三方庫讓咱們在編程中事半功倍，今天，咱們來談談python多線程在滲透測試中的應用，本文，咱們將編寫一個簡易c段存活主機掃描腳本，以及一個python版本的多線程御劍--目錄掃描工具

1、python 多線程
python多線程有幾種寫法
1. thread模塊
python的一個多線程模塊，小腳本能夠用，可是有瑕疵，好比不穩定，線程數很差控制

下方貼出一個c段存活主機掃描腳本，這個腳本i春秋ado老師也有教過哦
思想：輸入一個ip，通過字符拆分，獲取c段，多線程地ping -c 2 ip ，根據返回的信息來判斷主機是否存活
demo ping_thread.py:

'''
Created on 2017-2-27
 
@author: xiaoye
'''
#coding: utf-8
import thread
import time
from subprocess import Popen,PIPE
 
def scan_ip(ip):
    process = Popen('ping -c 2 ' + ip, stdin=PIPE, stdout=PIPE, shell=True)
    data = process.stdout.read()
    if 'ttl' in data:
        print '%s is live ,now time is %s' % (ip, time.strftime('%H:%M:%S'))
     
if __name__ == '__main__':
    #scan_ip('111.13.147.229')
    ips = raw_input()
    ip_header = '.'.join(ips.split('.')[:3])
    for i in range(1,255):
        ip = ip_header + '.' + str(i)
        #print ip
        thread.start_new_thread(scan_ip, (ip,))
        time.sleep(0.1)

 

 

 

 

運行狀況：
                                                                           
速度還行，穩定性通常

thread模塊，核心在：

[Python] 純文本查看 複製代碼

?

1 2	thread.start_new_thread(scan_ip, (ip,))          time.sleep( 0.1 )

scan_ip是要執行的函數，(ip,)是傳入的參數，記得sleep一下

2.threading模塊用法：
demo:

'''
Created on 2017-2-28
@author: xiaoye
'''
#coding: utf-8
import threading
import time
 
def test(th):
    print 'i am doing %s %s' % (th, time.strftime('%H:%M:%S'))
     
def main():
    thread = []
    keys = ['movie_th','swim_th','listen_th','learn_th','movie_th','swim_th','listen_th','learn_th','movie_th','swim_th','listen_th','learn_th','movie_th','swim_th','listen_th','learn_th']
    thread_count = len(keys)
    #print thread_count
    for i in range(thread_count):
        t = threading.Thread(target=test, args=(keys[i],))
        thread.append(t)
    for i in range(thread_count):
        thread[i].start()
    for i in range(thread_count):
        thread[i].join()
 
if __name__ == '__main__':
    main()

 

 

運行狀況：

能夠看到，基本是同時運行的，threading.Thread模塊的一種用法就是這樣：

for i in range(thread_count):
        t = threading.Thread(target=test, args=(keys[i],))
        thread.append(t)
    for i in range(thread_count):
        thread[i].start()
    for i in range(thread_count):
        thread[i].join()

 

 

模式1.：一個列表存放全部線程，start()執行列表中線程，join()等待運行完畢

模式1？，還有模式2嗎？
固然，模式2就是從threading.Thread繼承一個子類class，重寫父類run方法，實現多線程運行run函數，而這種也是很是良好的寫法
demo:

# -*- coding: utf-8 -*-
import threading
 
class T(threading.Thread):
 
    def __init__(self):
        threading.Thread.__init__(self)
 
    def run(self):                    #繼承,threading.Thread子類, 重寫run方法, run方法在start()後自動執行
        print 'i love you'
 
 
def main():
    thread = []
    for i in range(10):
      thread.append(T())
    for i in thread:
        i.start()
    for i in thread:
        i.join()
 
if __name__ == '__main__':
    main()

 

 

 

 

運行狀況：


2、線程間的數據守護
Queue絕對是保護線程間數據安全的好選擇，有關於Queue，你們能夠自行百度其用法，我發出一點常常用的：

Queue.qsize() 返回隊列的大小
Queue.empty() 若是隊列爲空，返回True,反之False
Queue.full() 若是隊列滿了，返回True,反之False
Queue.full 與 maxsize 大小對應
Queue.get([block[, timeout]]) 獲取隊列，timeout等待時間
Queue.get_nowait() 至關Queue.get(False)
非阻塞 Queue.put(item) 寫入隊列，timeout等待時間
Queue.put_nowait(item) 至關Queue.put(item, False)
Queue.task_done() 在完成一項工做以後，Queue.task_done() 函數向任務已經完成的隊列發送一個信號
Queue.join() 實際上意味着等到隊列爲空，再執行別的操做


3、多線程threading.Thread+Queue實現滲透測試工具編寫
腳本放出來：
1.多線程c段存活主機掃描：

'''
Created on 2017-2-28
@author: xiaoye
'''
#coding: utf-8
import time
import sys
import threading
import Queue
from subprocess import Popen,PIPE
 
 
class Quethread(threading.Thread):
    def __init__(self, que):
        threading.Thread.__init__(self)
        self._que = que
     
    def run(self):
        while not self._que.empty():
            ip = self._que.get()
            process = Popen('ping -c 2 ' + ip, stdin=PIPE, stdout=PIPE, shell=True)
            data = process.stdout.read()
            if 'ttl' in data:
                sys.stdout.write('%s is live %s\n' % (ip, time.strftime('%H:%M:%S')))
     
     
def main():
    que = Queue.Queue()
    ips = raw_input()
    thread = []
    thread_count = 200
    ip_head = '.'.join(ips.split('.')[:3])
    #print ip_head
    for i in range(1, 255):
        que.put(ip_head + '.' + str(i))
    '''for i in range(1,255):
        print que.get()'''
     
    for i in range(thread_count):
        thread.append(Quethread(que))
         
    for i in thread:
        i.start()
         
    for i in thread:
        i.join()
         
     
if __name__ == '__main__':
    main()

 

 

 

 

ubuntu下運行成功，win下須要修改Popen裏的命令等，截圖：

速度很快，穩定性較強
c段主機存活腳本：https://github.com/xiaoyecent/ping_threading_Queue

2.py版多線程御劍--目錄掃描--支持自定義字典、輸出文件位置以及自定義線程數：

'''
@author: xiaoye
'''
#coding: utf-8
import requests
import sys
import threading
#import time
import Queue
from optparse import OptionParser
 
reload(sys)
sys.setdefaultencoding('utf8')
 
class Doscan(threading.Thread):
    def __init__(self, que):
        threading.Thread.__init__(self)
        self._que = que
         
    def run(self):
        while not self._que.empty():
            d = self._que.get()
            try:
                r = requests.get(url + d, headers=headers, timeout=3)
                sys.stdout.write(d + ' is scan  status:' + str(r.status_code) + '\n')
                if r.status_code == 200:
                    with open(option.outfile, 'a') as f:
                        f.write(url + d + '\n')
            except:
                pass
     
def main():
    thread = []
    thread_count = option.threadcount
    que = Queue.Queue()
     
    with open(option.dictname, 'r') as f:
        for d in f.readlines():
            d = d.strip('\n')
            que.put(d)
     
    for i in range(thread_count):
        thread.append(Doscan(que))
     
    for i in thread:
        i.start()
     
    for i in thread:
        i.join()
         
if __name__ == '__main__':
    parse = OptionParser()
    parse.add_option('-u', '--url', dest='input_url', type='string', help='the url you wan to scan dir')
    parse.add_option('-o', '--out', dest='outfile', type='string', help='output filename', default='result.txt')
    parse.add_option('-s', '--speed', dest='threadcount', type='int', default=60, help='the thread_count')
    parse.add_option('-d', '--dict', dest='dictname', type='string', help='dict filename')
    (option, args) = parse.parse_args()
    headers = {'User-Agent':'Mozilla/5.0 (Windows NT 10.0; WOW64; rv:50.0) Gecko/20100101 Firefox/50.0'}
    url = option.input_url
    main()

Usage: scan_dir.py [options]
 
Options:
  -h, --help            show this help message and exit
  -u INPUT_URL, --url=INPUT_URL
                        the url you wan to scan dir
  -o OUTFILE, --out=OUTFILE
                        output filename
  -s THREADCOUNT, --speed=THREADCOUNT
                        the thread_count
  -d DICTNAME, --dict=DICTNAME
                        dict filename

 

 

參數用法貼出來
運行狀況
舉個例子：
-u http://localhost -s 30 -d d://PHP.txt -o d://ichunqiu.txt：


結果：


運行速度取決於線程數（默認60）和實際環境
源碼：https://github.com/xiaoyecent/scan_dir

4、總結
多線程加隊列實現線程間的數據保護是很好的搭配，threading.Thread+Queue的用法但願你們可以掌握，另外，繼承threading.Thread寫出子類，重寫父類run方法來實現多線程的寫法也值得借鑑