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的CVE

0×03 参数：

 # author: JC0o0l  # GitHub: https://github.com/chroblert/  可选参数:    -h, --help           show this help message and exit    -u, --update-cve     更新CVEKB数据    -U, --update-exp     更新CVEEXP数据    -C, --check-EXP       检索具有EXP的CVE    -f FILE, --file FILE ps1脚本运行后产生的.json文件

0×04 示例：

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

 .\KBCollect.ps1

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

3. 安装一些python3模块

 python3 -m pip 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:

 function Get-CollectKB(){      # 1. 搜集所有的KB补丁      $KBArray = @()      $KBArray = Get-HotFix|ForEach-Object {$_.HotFixId}      $test = $KBArray|ConvertTo-Json      return $test  }  function Get-BasicInfo(){      # 1. 操作系统      # $windowsProductName = (Get-ComputerInfo).WindowsProductName      $windowsProductName = (Get-CimInstance Win32_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-File KB.json -encoding utf8

cve-check.py:

 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=""):      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()      for cve in intersectionList:          sql = "select CVEName from {} where CVEName == '{}' and hasPOC == 'True'".format(TableName,cve)          # print(sql)          con.execute(sql)          if len(con.fetchall()) != 0:              print("{} has EXP".format(cve))      # 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))          # 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          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文件")