WindowsVulnScan | 一款基于主机的漏洞扫描工具

database is locked

md5

shodan

vulmap

windows-exp-suggester

<pre class="brush:php;toolbar:false;"># 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文件

KBCollect.ps

<pre class="brush:php;toolbar:false;">.\KBCollect.ps1

KB.json

cve-check.py

<pre class="brush:php;toolbar:false;">python3 -m pip install requirements.txt

cve-check.py -u

cve-check.py -u

hasPOC

cve-check.py -C -f KB.json

KBCollect.ps1

<pre class="brush:php;toolbar:false;">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

<pre class="brush:php;toolbar:false;">import requestsimport sqlite3import jsonimport hashlibimport mathimport reimport threadingimport timeimport argparsefrom pathlib import Path# 删除一些ssl 认证的warnging信息requests.packages.urllib3.disable_warnings()ThreadCount=20DBFileName="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:            breakdef 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()    # passdef 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文件")