# coding:utf-8 # 漏洞检测引擎 import urllib2 import thread import time import pymongo import sys import datetime import hashlib import json import re import uuid import os from kunpeng import kunpeng sys.path.append(sys.path[0] + '/vuldb') # 加载漏洞插件的目录 sys.path.append(sys.path[0] + "/../") from config import ProductionConfig db_conn = pymongo.MongoClient(ProductionConfig.DB, ProductionConfig.PORT) na_db = getattr(db_conn, ProductionConfig.DBNAME) na_db.authenticate(ProductionConfig.DBUSERNAME, ProductionConfig.DBPASSWORD) na_task = na_db.Task na_result = na_db.Result na_plugin = na_db.Plugin na_config = na_db.Config na_heart = na_db.Heartbeat na_update = na_db.Update lock = thread.allocate() PASSWORD_DIC = [] THREAD_COUNT = 50 TIMEOUT = 10 PLUGIN_DB = {} TASK_DATE_DIC = {} WHITE_LIST = [] kp = kunpeng() #巡风的漏洞扫描技术要先看初始化的方法，看完初始化的方法，再从main函数开始看，这样容易理解，分析完main函数之后，再从start函数慢慢看 class vulscan(): # 初始化操作 def __init__(self, task_id, task_netloc, task_plugin): self.task_id = task_id #任务id self.task_netloc = task_netloc #任务端口 self.task_plugin = task_plugin #任务插件 self.result_info = '' #任务结果 self.start() #任务开始 # start ---> 开始检测 def start(self): self.get_plugin_info() #获取插件列表，这个在80多行有个get_plugin_info方法，主要就是info = xxx。 #这个是获取.json格式的漏洞库，主要用来是探测，并不能起到exp的作用 if '.json' in self.plugin_info['filename']: # 标示符检测模式 self.load_json_plugin() # 读取漏洞标示 # 跟踪load_json_plugin()函数，读取的时候，是字符串。所以需要转换成json的形式。 self.set_request() # 标示符转换为请求 # 这个就是发送request的请求了 self.poc_check() # 检测 # 进行poc验证，这个地方需要是check()，里面有py,md5,json三种格式的请求，所以要分别的验证一下 # 如果时kunpeng的脚本开始验证，分为两种一种是web端的http,https，一种是除了web端。没咋懂，就这样吧，有空再看 elif 'KP-' in self.plugin_info['filename']: self.log(str(self.task_netloc) + 'call kunpeng - ' + self.plugin_info['filename']) kp.set_config(TIMEOUT, PASSWORD_DIC) if self.task_netloc[1] != 80: self.result_info = kp.check('service', '{}:{}'.format( self.task_netloc[0], self.task_netloc[1]), self.plugin_info['filename']) if not self.result_info: scheme = 'http' if self.task_netloc[1] == 443: scheme = 'https' self.result_info = kp.check('web', '{}://{}:{}'.format( scheme, self.task_netloc[0], self.task_netloc[1]), self.plugin_info['filename']) else: # 脚本检测模式，这个利用的py脚本的poc，还是比较有意思的，可能能说到说到的就是这一个了，这个和我们平常写的跑py脚本差不多。 plugin_filename = self.plugin_info['filename'] # log里面封装的是一个print方法 self.log(str(self.task_netloc) + 'call ' + self.task_plugin) if task_plugin not in PLUGIN_DB: plugin_res = __import__(plugin_filename) setattr(plugin_res, "PASSWORD_DIC", PASSWORD_DIC) # 给插件声明密码字典 PLUGIN_DB[plugin_filename] = plugin_res self.result_info = PLUGIN_DB[plugin_filename].check( str(self.task_netloc[0]), int(self.task_netloc[1]), TIMEOUT) self.save_request() # 保存结果 def get_plugin_info(self): info = na_plugin.find_one({"name": self.task_plugin}) self.plugin_info = info def load_json_plugin(self): json_plugin = open(sys.path[0] + '/vuldb/' + self.plugin_info['filename']).read() self.plugin_info['plugin'] = json.loads(json_plugin)['plugin'] def set_request(self): url = 'http://' + self.task_netloc[0] + ":" + str(self.task_netloc[1]) + self.plugin_info['plugin']['url'] if self.plugin_info['plugin']['method'] == 'GET': request = urllib2.Request(url) else: request = urllib2.Request(url, self.plugin_info['plugin']['data']) self.poc_request = request def get_code(self, header, html): try: m = re.search(r'<meta.*?charset=(.*?)"(>| |/)', html, flags=re.I) if m: return m.group(1).replace('"', '') except: pass try: if 'Content-Type' in header: Content_Type = header['Content-Type'] m = re.search(r'.*?charset=(.*?)(;|$)', Content_Type, flags=re.I) if m: return m.group(1) except: pass def poc_check(self): try: res = urllib2.urlopen(self.poc_request, timeout=30) res_html = res.read(204800) header = res.headers # res_code = res.code except urllib2.HTTPError, e: # res_code = e.code header = e.headers res_html = e.read(204800) except Exception, e: return try: html_code = self.get_code(header, res_html).strip() if html_code and len(html_code) < 12: res_html = res_html.decode(html_code).encode('utf-8') except: pass an_type = self.plugin_info['plugin']['analyzing'] vul_tag = self.plugin_info['plugin']['tag'] analyzingdata = self.plugin_info['plugin']['analyzingdata'] if an_type == 'keyword': # print poc['analyzingdata'].encode("utf-8") if analyzingdata.encode("utf-8") in res_html: self.result_info = vul_tag elif an_type == 'regex': if re.search(analyzingdata, res_html, re.I): self.result_info = vul_tag elif an_type == 'md5': md5 = hashlib.md5() md5.update(res_html) if md5.hexdigest() == analyzingdata: self.result_info = vul_tag def save_request(self): if self.result_info: time_ = datetime.datetime.now() self.log(str(self.task_netloc) + " " + self.result_info) v_count = na_result.find( {"ip": self.task_netloc[0], "port": self.task_netloc[1], "info": self.result_info}).count() if not v_count: na_plugin.update({"name": self.task_plugin}, {"$inc": {'count': 1}}) vulinfo = {"vul_name": self.plugin_info['name'], "vul_level": self.plugin_info['level'], "vul_type": self.plugin_info['type']} w_vul = {"task_id": self.task_id, "ip": self.task_netloc[0], "port": self.task_netloc[1], "vul_info": vulinfo, "info": self.result_info, "time": time_, "task_date": TASK_DATE_DIC[str(self.task_id)]} na_result.insert(w_vul) # self.wx_send(w_vul) # 自行定义漏洞提醒 def log(self, info): lock.acquire() try: time_str = time.strftime('%X', time.localtime(time.time())) print "[%s] %s" % (time_str, info) except: pass lock.release() def queue_get(): global TASK_DATE_DIC task_req = na_task.find_and_modify(query={"status": 0, "plan": 0}, update={ "$set": {"status": 1}}, sort={'time': 1}) if task_req: TASK_DATE_DIC[str(task_req['_id'])] = datetime.datetime.now() return task_req['_id'], task_req['plan'], task_req['target'], task_req['plugin'] else: task_req_row = na_task.find({"plan": {"$ne": 0}}) if task_req_row: for task_req in task_req_row: if (datetime.datetime.now() - task_req['time']).days / int(task_req['plan']) >= int(task_req['status']): if task_req['isupdate'] == 1: task_req['target'] = update_target( json.loads(task_req['query'])) na_task.update({"_id": task_req['_id']}, { "$set": {"target": task_req['target']}}) na_task.update({"_id": task_req['_id']}, { "$inc": {"status": 1}}) TASK_DATE_DIC[str(task_req['_id']) ] = datetime.datetime.now() return task_req['_id'], task_req['plan'], task_req['target'], task_req['plugin'] return '', '', '', '' def update_target(query): target_list = [] try: result_list = na_db.Info.find(query) for result in result_list: target = [result["ip"], result["port"]] target_list.append(target) except: pass return target_list def monitor(): global PASSWORD_DIC, THREAD_COUNT, TIMEOUT, WHITE_LIST while True: queue_count = na_task.find({"status": 0, "plan": 0}).count() if queue_count: load = 1 else: ac_count = thread._count() load = float(ac_count - 6) / THREAD_COUNT if load > 1: load = 1 if load < 0: load = 0 na_heart.update({"name": "load"}, { "$set": {"value": load, "up_time": datetime.datetime.now()}}) PASSWORD_DIC, THREAD_COUNT, TIMEOUT, WHITE_LIST = get_config() if load > 0: time.sleep(8) else: time.sleep(60) def get_config(): try: config_info = na_config.find_one({"type": "vulscan"}) pass_row = config_info['config']['Password_dic'] thread_row = config_info['config']['Thread'] timeout_row = config_info['config']['Timeout'] white_row = config_info['config']['White_list'] password_dic = pass_row['value'].split(' ') thread_count = int(thread_row['value']) timeout = int(timeout_row['value']) white_list = white_row['value'].split(' ') return password_dic, thread_count, timeout, white_list except Exception, e: print e def install_kunpeng_plugin(): time_ = datetime.datetime.now() for plugin in kp.get_plugin_list(): level_list = ['紧急','高危','中危','低危','提示'] plugin_info = { '_id': plugin['references']['kpid'], 'name': 'Kunpeng -' + plugin['name'], 'info': plugin['remarks'] + ' ' + plugin['references']['cve'], 'level': level_list[int(plugin['level'])], 'type': plugin['type'], 'author': plugin['author'], 'url': plugin['references']['url'], 'source': 1, 'keyword': '', 'add_time': time_, 'filename': plugin['references']['kpid'], 'count': 0 } na_plugin.insert(plugin_info) def init(): time_ = datetime.datetime.now() if na_plugin.find().count() >= 1: return script_plugin = [] json_plugin = [] print 'init plugins' file_list = os.listdir(sys.path[0] + '/vuldb') for filename in file_list: try: if filename.split('.')[1] == 'py': script_plugin.append(filename.split('.')[0]) if filename.split('.')[1] == 'json': json_plugin.append(filename) except: pass for plugin_name in script_plugin: try: res_tmp = __import__(plugin_name) plugin_info = res_tmp.get_plugin_info() plugin_info['add_time'] = time_ plugin_info['filename'] = plugin_name plugin_info['count'] = 0 na_plugin.insert(plugin_info) except: pass for plugin_name in json_plugin: try: json_text = open(sys.path[0] + '/vuldb/' + plugin_name, 'r').read() plugin_info = json.loads(json_text) plugin_info['add_time'] = time_ plugin_info['filename'] = plugin_name plugin_info['count'] = 0 del plugin_info['plugin'] na_plugin.insert(plugin_info) except: pass install_kunpeng_plugin() def kp_check(): while True: try: new_release = kp.check_version() print new_release if new_release: info = new_release['body'] if '###' in new_release['body']: info = new_release['body'].split('###')[1] row = { 'info': info, 'isInstall': 0, 'name': new_release['name'], 'author': new_release['author']['login'], 'pushtime': new_release['published_at'], 'location': "", 'unicode': new_release['tag_name'], 'coverage': 0, 'source': 'kunpeng' } na_update.insert(row) time.sleep(60 * 60 * 48) except Exception as e: print e time.sleep(60 * 30) def kp_update(): while True: try: row = na_update.find_one_and_delete( {'source': 'kunpeng', 'isInstall': 1}) if row: kp.update_version(row['unicode']) na_plugin.delete_many({'_id':re.compile('^KP')}) install_kunpeng_plugin() except Exception as e: print e time.sleep(10) if __name__ == '__main__': init() PASSWORD_DIC, THREAD_COUNT, TIMEOUT, WHITE_LIST = get_config() #从数据库里面获得配置参数，比如白名单，引擎的线程 thread.start_new_thread(monitor, ()) #巡风的心跳线程 thread.start_new_thread(kp_check, ()) #巡风的Kunpeng库，检查更新去情况，跟踪kp_check-->check_version-->_get_release_lates，这个函数发现的是与云上的库进行对比，如果有更新的话，就提示更新 # 如果没有更新的话，就不提示了 thread.start_new_thread(kp_update, ()) #kunpeng库更新，上面那个只是推送出来，但是还没有更新，这个是巡风更新的线程，就是如果你把kunpeng库确定更新了之后，然后才能是执行这个线程 while True: try: task_id, task_plan, task_target, task_plugin = queue_get() #获取队列，简单来说就是获得任务的参数 #如果status : 0 ，这个是未执行的状态，如果是status : 1就是正在执行的状态 if task_id == '': time.sleep(10) # 每个间隔10秒钟 continue if PLUGIN_DB: del sys.modules[PLUGIN_DB.keys()[0]] # 清理插件缓存，这个我也没太懂学长讲的怎么回事... PLUGIN_DB.clear() for task_netloc in task_target: while True: if int(thread._count()) < THREAD_COUNT: #如果任务的线程数<50 if task_netloc[0] in WHITE_LIST: #如果探测的资产是在白名单里面，那就break，结束 (^-^) break try: thread.start_new_thread( vulscan, (task_id, task_netloc, task_plugin)) except Exception as e: print e break else: time.sleep(2) if task_plan == 0: na_task.update({"_id": task_id}, {"$set": {"status": 2}}) except Exception as e: print e
巡风如果说作为一个检测内网的手段，真的是非常好用。另外分析下巡风的扫描，也能学到不少东西，比如我之前以为扫描摄像头的时候，以为是扫描的数据流，其实扫描的路径，拼接路径就可以了。还有巡风打poc的地方，也可以学习学习。不过自我感觉，还是巡风的资产探测比较好用，主要的还是要把巡风的资产探测学好，巡风的资产探测还是比较好的