python windows系统漏洞扫描_Windows主机漏洞扫描工具

原标题：Windows主机漏洞扫描工具

0×00 说明：

这是一款基于主机的漏洞扫描工具，采用多线程确保可以快速的请求数据，采用线程锁可以在向sqlite数据库中写入数据避免database is locked的错误，采用md5哈希算法确保数据不重复插入。

本工具查找是否有公开exp的网站为shodan,该网站限制网络发包的速度，因而采用了单线程的方式，且耗时较长。

功能：

查找主机上具有的CVE

查找具有公开EXP的CVE

0×01 起因：

因为需要做一些主机漏洞扫描方面的工作，因而编写了这个简单的工具。之前也查找了几款类似的工具，如下：

vulmap:

vulmon开发的一款开源工具，原理是根据软件的名称和版本号来确定，是否有CVE及公开的EXP。这款Linux的工具挺好用，但是对于Windows系统层面不太适用。

windows-exp-suggester:

这款和本工具的原理一样，尝试使用了之后，发现它的CVEKB数据库只更新到2017年的，并且没有给出CVE是否有公开的EXP信息。

基于以上所以写了这个简单的工具,该项目在https://github.com/chroblert/WindowsVulnScan

0×02 原理：

1. 搜集CVE与KB的对应关系。首先在微软官网上收集CVE与KB对应的关系，然后存储进数据库中

2. 查找特定CVE网上是否有公开的EXP

3. 利用powershell脚本收集主机的一些系统版本与KB信息

4. 利用系统版本与KB信息搜寻主机上具有存在公开EXP的CVE0×03 参数：

# author: JC0o0l

# GitHub: https://github.com/chroblert/

可选参数:

-h, --helpshow this help message and exit

-u, --update-cve更新CVEKB数据

-U, --update-exp更新CVEEXP数据

-C, --check-EXP检索具有EXP的CVE

-fFILE, --fileFILE ps1脚本运行后产生的.json文件0×04 示例：

1. 首先运行powershell脚本KBCollect.ps收集一些信息

.KBCollect.ps1

2. 将运行后产生的KB.json文件移动到cve-check.py所在的目录

3. 安装一些python3模块

python3 -mpip install requirements.txt

4. 运行cve-check.py -u创建CVEKB数据库

5. 运行cve-check.py -U更新CVEKB数据库中的hasPOC字段

6. 运行cve-check.py -C -f KB.json查看具有公开EXP的CVE，如下：

0×05 源码：

KBCollect.ps1:

functionGet-CollectKB{

# 1. 搜集所有的KB补丁

$KBArray=@

$KBArray=Get-HotFix|ForEach-Object{$_.HotFixId}

$test=$KBArray|ConvertTo-Json

return$test

}

functionGet-BasicInfo{

# 1. 操作系统

# $windowsProductName = (Get-ComputerInfo).WindowsProductName

$windowsProductName=(Get-CimInstanceWin32_OperatingSystem).Caption

# 2. 操作系统版本

$windowsVersion=(Get-ComputerInfo).WindowsVersion

$basicInfo="{""windowsProductName"":""$windowsProductName"",""windowsVersion"":""$windowsVersion""}"

return$basicInfo

}

$basicInfo=Get-BasicInfo

$KBList=Get-CollectKB

$KBResult="{""basicInfo"":$basicInfo,""KBList"":$KBList}"

$KBResult|Out-FileKB.json-encodingutf8

cve-check.py:

importrequests

importsqlite3

importjson

importhashlib

importmath

importre

importthreading

importtime

importargparse

frompathlibimportPath

# 删除一些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

classCVEScanThread(threading.Thread):

def__init__(self,func,args,name="",):

threading.Thread.__init__(self)

self.func= func

self.args= args

self.name= name

self.result= None

defrun(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]))

classEXPScanThread(threading.Thread):

def__init__(self,func,args,name="",):

threading.Thread.__init__(self)

self.func= func

self.args= args

self.name= name

self.result= None

defrun(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]))

defget_result(self):

threading.Thread.join(self)

try:

returnself.result

exceptException:

return"Error"

defget_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']

returnmath.ceil(int(resultCount)/100)

defupdate_cvekb_database(num=1,pageSize=100):

pageCount= get_page_num

#for i in range(1,pageCount+1):

i= 1

pageCount=524

tmpCount= ThreadCount

whilei<= pageCount:

tmpCount= ThreadCountif(pageCount-i) >= ThreadCountelsepageCount-i

print("i:{},pageCount-i:{},ThreadCount:{},PageCount:{}".format(i,pageCount-i,ThreadCount,pageCount))

time.sleep(0.5)

threads= []

print("===============================")

forjinrange(1,tmpCount+1):

print("更新第{}页".format(i+j-1))

t= CVEScanThread(update_onepage_cvedb_database,(i+j-1,j,),str(j))

threads.append(t)

fortinthreads:

t.start

fortinthreads:

t.join

# update_onepage_cvedb_database(num=i)

i= i+tmpCount

conn= sqlite3.connect(DBFileName)

forsqlininsertSQL:

conn.execute(sql)

conn.commit

conn.close

iftmpCount!= ThreadCount:

break

defcheck_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)

exceptExceptionase:

print("Error,{}".format(CVEName))

print(e)

return"Error"

if"Total Results"notinresponse.text:

return"False"

else:

return"True"

defupdate_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

forresultinresultList:

KBName= result['articleTitle1'] +";"if(result['articleTitle1'] != None) andresult['articleTitle1'].isdigitelse""

KBName+= result['articleTitle2'] +";"if(result['articleTitle2'] != None) andresult['articleTitle2'].isdigitelse""

KBName+= result['articleTitle3'] +";"if(result['articleTitle3'] != None) andresult['articleTitle3'].isdigitelse""

KBName+= result['articleTitle4'] +";"if(result['articleTitle4'] != None) andresult['articleTitle4'].isdigitelse""

ifKBName== "":

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+"')"

withlock:

globalinsertSQL

insertSQL.append(sql)

# conn.execute(sql)

# conn.commit

# conn.close

# pass

defselect_CVE(tmpList=[],windowsProductName="",windowsVersion=""):

conn= sqlite3.connect(DBFileName)

con= conn.cursor

intersectionList= []

count= 0

foriintmpList:

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= []

forcveincon.execute(sql):

cveList.append(cve[0])

ifcount== 0:

intersectionList= cveList.copy

count+=1

intersectionList= list(set(intersectionList).intersection(set(cveList)))

intersectionList.sort

forcveinintersectionList:

sql= "select CVEName from {} where CVEName == '{}' and hasPOC == 'True'".format(TableName,cve)

# print(sql)

con.execute(sql)

iflen(con.fetchall) != 0:

print("{} has EXP".format(cve))

# print(intersectionList)

defupdate_hasPOC(key= "Empty"):

conn= sqlite3.connect(DBFileName)

con= conn.cursor

ifkey== "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

whilei

print("|=========={}============|".format(i))

# tmpCount = ThreadCount if (len(cveNameList) - i) >= ThreadCount else len(cveNameList) - i

# threads = []

# for j in range(1,tmpCount+1):

# t = EXPScanThread(check_POC_every_CVE,(cveNameList[i+j][0],j,i+j,),str(j))

# threads.append(t)

# for t in threads:

# t.start

# for t in threads:

# t.join

# j = 1

# for t in threads:

# hasPOC = t.get_result

# print(hasPOC)

# update_sql = "UPDATE "+TableName+" set hasPOC = '" + hasPOC + "' WHERE cveName == '" + cveNameList[i+j][0] +"';"

# conn.execute(update_sql)

# print("[+] update:{}".format(update_sql))

# j += 1

# i=i+ThreadCount

# conn.commit

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

ifcount== 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(notargs.check_EXP) and(notargs.update_cve) and(notargs.update_exp) andargs.fileisNone:

parser.print_help

ifargs.update_cve:

update_cvekb_database

ifargs.update_exp:

dbfile=Path(DBFileName)

ifdbfile.exists:

update_hasPOC(key="Empty")

else:

print("请先使用-u 创建数据库")

parser.print_help

ifargs.check_EXP:

dbfile=Path(DBFileName)

ifnotdbfile.exists:

print("请先使用-u 创建数据库，之后使用 -U 更新是否有EXP")

parser.print_help

exit

ifargs.file:

withopen(args.file,"r",encoding="utf-8") asf:

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= []

forKBintmpKBList:

KBList.append(KB.replace("KB",""))

print("KB信息如下:")

print(KBList)

print("EXP信息如下:")

select_CVE(tmpList=KBList,windowsProductName=windowsProductName,windowsVersion=windowsVersion)

else:

print("请输入.json文件")

*本文作者：jerrybird，转载请注明来自FreeBuf.COM 返回搜狐，查看更多

责任编辑：