-
Notifications
You must be signed in to change notification settings - Fork 98
/
Copy pathwindowsVulnScanUtil.py
288 lines (274 loc) · 12 KB
/
windowsVulnScanUtil.py
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
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
import requests
import sqlite3
import json
import hashlib
import math
import re
import threading
import time
import argparse
from pathlib import Path
# 删除一些ssl 认证的warnging信息
requests.packages.urllib3.disable_warnings()
ThreadCount=20
DBFileName="CVEKB.db"
TableName="CVEKB"
insertSQL = []
updateSQL = []
lock = threading.Lock()
KBResult = {}
# parser = argparse.ArgumentParser()
# parser.add_argument("-u","--update-cve",help="更新CVEKB数据",action="store_true")
# parser.add_argument("-U","--update-exp",help="更新CVEEXP数据",action="store_true")
# parser.add_argument("-C","--check-EXP",help="检索具有EXP的CVE",action="store_true")
# parser.add_argument("-f","--file",help="ps1脚本运行后产生的.json文件")
# args = parser.parse_args()
class CVEScanThread(threading.Thread):
def __init__(self,func,args,name="",):
threading.Thread.__init__(self)
self.func = func
self.args = args
self.name = name
self.result = None
def run(self):
print("thread:{} :start scan page {}".format(self.args[1],self.args[0]))
self.result = self.func(self.args[0],)
print("thread:{} :stop scan page {}".format(self.args[1],self.args[0]))
class EXPScanThread(threading.Thread):
def __init__(self,func,args,name="",):
threading.Thread.__init__(self)
self.func = func
self.args = args
self.name = name
self.result = None
def run(self):
print("thread:{} :start scan CVE: {},xuehao:{}".format(self.args[1],self.args[0],self.args[2]))
self.result = self.func(self.args[0],)
print("thread:{} :stop scan CVE: {}".format(self.args[1],self.args[0]))
def get_result(self):
threading.Thread.join(self)
try:
return self.result
except Exception:
return "Error"
def get_page_num(num=1,pageSize=100):
url = "https://portal.msrc.microsoft.com/api/security-guidance/en-us"
payload = "{\"familyIds\":[],\"productIds\":[],\"severityIds\":[],\"impactIds\":[],\"pageNumber\":" + str(num) + ",\"pageSize\":" + str(pageSize) + ",\"includeCveNumber\":true,\"includeSeverity\":false,\"includeImpact\":false,\"orderBy\":\"publishedDate\",\"orderByMonthly\":\"releaseDate\",\"isDescending\":true,\"isDescendingMonthly\":true,\"queryText\":\"\",\"isSearch\":false,\"filterText\":\"\",\"fromPublishedDate\":\"01/01/1998\",\"toPublishedDate\":\"03/02/2020\"}"
headers = {
'origin': "https//portal.msrc.microsoft.com",
'referer': "https//portal.msrc.microsoft.com/en-us/security-guidance",
'accept-language': "zh-CN",
'user-agent': "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36 Edge/16.16299",
'accept': "application/json, text/plain, */*",
'accept-encoding': "gzip, deflate",
'host': "portal.msrc.microsoft.com",
'connection': "close",
'cache-control': "no-cache",
'content-type': "application/json",
}
response = requests.request("POST", url, data=payload, headers=headers, verify = False)
resultCount = json.loads(response.text)['count']
return math.ceil(int(resultCount)/100)
def update_cvekb_database(num=1,pageSize=100):
pageCount = get_page_num()
#for i in range(1,pageCount+1):
i = 1
# pageCount=524
tmpCount = ThreadCount
while i <= pageCount:
tmpCount = ThreadCount if (pageCount - i) >= ThreadCount else pageCount - i
print("i:{},pageCount-i:{},ThreadCount:{},PageCount:{}".format(i,pageCount-i,ThreadCount,pageCount))
time.sleep(0.5)
threads = []
print("===============================")
for j in range(1,tmpCount + 1):
print("更新第{}页".format(i+j-1))
t = CVEScanThread(update_onepage_cvedb_database,(i+j-1,j,),str(j))
threads.append(t)
for t in threads:
t.start()
for t in threads:
t.join()
# update_onepage_cvedb_database(num=i)
i = i + tmpCount
conn = sqlite3.connect(DBFileName)
for sql in insertSQL:
conn.execute(sql)
conn.commit()
conn.close()
if tmpCount != ThreadCount:
break
def check_POC_every_CVE(CVEName=""):
#apiKey = ""
#url = "https://exploits.shodan.io/api/search?query=" + CVEName + "&key=" + apiKey
url = "https://exploits.shodan.io/?q=" + CVEName
try:
response = requests.request("GET",url=url,verify=False,timeout=10)
#total = json.loads(response.text)
except Exception as e:
print("Error,{}".format(CVEName))
print(e)
return "Error"
if "Total Results" not in response.text:
return "False"
else:
return "True"
def update_onepage_cvedb_database(num=1,pageSize=100):
url = "https://portal.msrc.microsoft.com/api/security-guidance/en-us"
payload = "{\"familyIds\":[],\"productIds\":[],\"severityIds\":[],\"impactIds\":[],\"pageNumber\":" + str(num) + ",\"pageSize\":" + str(pageSize) + ",\"includeCveNumber\":true,\"includeSeverity\":false,\"includeImpact\":false,\"orderBy\":\"publishedDate\",\"orderByMonthly\":\"releaseDate\",\"isDescending\":true,\"isDescendingMonthly\":true,\"queryText\":\"\",\"isSearch\":false,\"filterText\":\"\",\"fromPublishedDate\":\"01/01/1998\",\"toPublishedDate\":\"03/02/2020\"}"
headers = {
'origin': "https//portal.msrc.microsoft.com",
'referer': "https//portal.msrc.microsoft.com/en-us/security-guidance",
'accept-language': "zh-CN",
'user-agent': "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36 Edge/16.16299",
'accept': "application/json, text/plain, */*",
'accept-encoding': "gzip, deflate",
'host': "portal.msrc.microsoft.com",
'connection': "close",
'cache-control': "no-cache",
'content-type': "application/json",
}
try:
response = requests.request("POST", url, data=payload, headers=headers, verify = False)
resultList = json.loads(response.text)['details']
except :
print(response.text)
conn = sqlite3.connect(DBFileName)
create_sql = """Create Table IF NOT EXISTS {} (
hash TEXT UNIQUE,
name TEXT,
KBName TEXT,
CVEName TEXT,
impact TEXT,
hasPOC TEXT)""".format(TableName)
conn.execute(create_sql)
conn.commit()
conn.close()
for result in resultList:
KBName = result['articleTitle1'] + ";" if (result['articleTitle1'] != None) and result['articleTitle1'].isdigit() else ""
KBName += result['articleTitle2'] + ";" if (result['articleTitle2'] != None) and result['articleTitle2'].isdigit() else ""
KBName += result['articleTitle3'] + ";" if (result['articleTitle3'] != None) and result['articleTitle3'].isdigit() else ""
KBName += result['articleTitle4'] + ";" if (result['articleTitle4'] != None) and result['articleTitle4'].isdigit() else ""
if KBName == "":
continue
h1 = hashlib.md5()
metaStr = result['name'] + KBName + result['cveNumber'] + result['impact']
h1.update(metaStr.encode('utf-8'))
#hasPOC = check_POC_every_CVE(result['cveNumber'])
# 收集到所有的KB后再搜索有没有公开的EXP
hasPOC = ""
sql = "INSERT OR IGNORE INTO "+TableName+" VALUES ('" + h1.hexdigest() + "','" + result['name'] + "','" + KBName + "','" + result['cveNumber'] + "','" + result['impact'] + "','" + hasPOC+"')"
with lock:
global insertSQL
insertSQL.append(sql)
# conn.execute(sql)
# conn.commit()
# conn.close()
# pass
def select_CVE(tmpList=[],windowsProductName="",windowsVersion=""):
if not Path("CVEKB.db").exists():
return []
conn = sqlite3.connect(DBFileName)
con = conn.cursor()
intersectionList = []
count = 0
for i in tmpList:
sql = 'select distinct(CVEName) from '+ TableName+' where (name like "'+ windowsProductName+'%'+ windowsVersion + '%") and ("'+ i +'" not in (select KBName from '+ TableName +' where name like "'+ windowsProductName+'%'+windowsVersion+'%")); '
cveList = []
for cve in con.execute(sql):
cveList.append(cve[0])
if count == 0:
intersectionList = cveList.copy()
count +=1
intersectionList = list(set(intersectionList).intersection(set(cveList)))
intersectionList.sort()
# print(intersectionList)
# return []
CVEExpList = []
for cve in intersectionList:
sql = "select distinct(CVEName),KBName,impact,name from {} where CVEName == '{}' and hasPOC == 'True' and name like '{}' limit 1".format(TableName,cve,windowsProductName+'%'+ windowsVersion + '%')
# print(sql)
con.execute(sql)
resList=con.fetchall()
if len(resList) != 0:
for row in resList:
print("{} has EXP".format(cve))
resDict={}
resDict['CVEName']=row[0]
resDict['KBName']=row[1]
resDict['impact']=row[2]
resDict['name']=row[3]
CVEExpList.append(resDict)
return CVEExpList
# print(intersectionList)
def update_hasPOC(key = "Empty"):
conn = sqlite3.connect(DBFileName)
con = conn.cursor()
if key == "All":
sql = "select distinct(CVEName) from {}".format(TableName)
else:
sql = "select distinct(CVEName) from {} where (hasPOC IS NULL) OR (hasPOC == '')".format(TableName)
con.execute(sql)
cveNameList = con.fetchall()
i = 0
count = 1
while i < len(cveNameList):
print("|=========={}============|".format(i))
hasPOC = check_POC_every_CVE(cveNameList[i][0])
time.sleep(0.3)
update_sql = "UPDATE "+TableName+" set hasPOC = '" + hasPOC + "' WHERE cveName == '" + cveNameList[i][0] +"';"
conn.execute(update_sql)
print(update_sql)
count += 1
i += 1
if count == 10:
conn.commit()
print("[+]update")
count = 1
conn.commit()
conn.close()
print("Over")
# if __name__ == "__main__":
# banner = """
# ========CVE-EXP-Check===============
# | author:JC0o0l |
# | wechat:JC_SecNotes |
# | version:1.0 |
# =====================================
# """
# print(banner)
# if (not args.check_EXP ) and (not args.update_cve) and (not args.update_exp) and args.file is None:
# parser.print_help()
# if args.update_cve:
# update_cvekb_database()
# if args.update_exp:
# dbfile=Path(DBFileName)
# if dbfile.exists():
# update_hasPOC(key="Empty")
# else:
# print("请先使用-u 创建数据库")
# parser.print_help()
# if args.check_EXP:
# dbfile=Path(DBFileName)
# if not dbfile.exists():
# print("请先使用-u 创建数据库，之后使用 -U 更新是否有EXP")
# parser.print_help()
# exit()
# if args.file:
# with open(args.file,"r",encoding="utf-8") as f:
# KBResult = json.load(f)
# windowsProductName = KBResult['basicInfo']['windowsProductName']
# windowsProductName = ((re.search("\w[\w|\s]+\d+[\s|$]",windowsProductName).group()).strip()).replace("Microsoft","").strip()
# windowsVersion = KBResult['basicInfo']['windowsVersion']
# print("系统信息如下:")
# print("{} {}".format(windowsProductName,windowsVersion))
# tmpKBList = KBResult['KBList']
# KBList = []
# for KB in tmpKBList:
# KBList.append(KB.replace("KB",""))
# print("KB信息如下:")
# print(KBList)
# print("EXP信息如下:")
# select_CVE(tmpList=KBList,windowsProductName=windowsProductName,windowsVersion=windowsVersion)
# else:
# print("请输入.json文件")