wp2shell_check.py

#!/usr/bin/env python3
"""
wp2shell_check.py

Non-intrusive checker for CVE-2026-63030 / CVE-2026-60137 ("wp2shell"),
a pre-authentication RCE chain in WordPress core (REST API batch-route
confusion -> SQL injection -> RCE).

WHAT THIS SCRIPT DOES
  - Fingerprints the WordPress version from public, unauthenticated sources
    (readme.html, homepage <meta name="generator">, /feed/ generator tag).
  - Compares the detected version against the known vulnerable ranges.
  - Checks whether the vulnerable REST batch route is reachable
    (GET only, no payload) as a secondary signal / mitigation check.

WHAT THIS SCRIPT DELIBERATELY DOES NOT DO
  - It does NOT send any SQL injection or RCE payload. The exact exploit
    chain (REST batch route desync -> author__not_in SQLi -> RCE) has not
    been independently verified here, and firing untested injection
    payloads at production sites can cause real damage even when "only"
    testing. Use this script for safe triage; use WP-CLI (`wp core version`)
    or an authenticated scanner (e.g. vendor tooling) for ground truth.

Only run this against systems you are authorized to test.

Usage:
    python3 wp2shell_check.py https://internal-site.example.com
    python3 wp2shell_check.py -f targets.txt -o report.csv
    python3 wp2shell_check.py -f targets.txt --insecure --workers 8
"""

import argparse
import csv
import json
import re
import socket
import ssl
import sys
import urllib.error
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field, asdict
from typing import Optional

USER_AGENT = "wp2shell-version-check/1.0 (internal vulnerability triage)"
TIMEOUT = 10

# Known-affected / fixed ranges per public advisories (CVE-2026-63030, CVE-2026-60137)
FIXED_VERSIONS = {"6.8.6", "6.9.5", "7.0.2"}


def parse_version(v: str):
    parts = re.findall(r"\d+", v)
    return tuple(int(p) for p in parts[:3]) if parts else None


def in_range(v, lo, hi):
    def pad(t):
        return t + (0,) * (3 - len(t))
    v, lo, hi = pad(v), pad(lo), pad(hi)
    return lo <= v <= hi


def classify_version(version_str: str) -> str:
    """Return one of: VULNERABLE_RCE, VULNERABLE_SQLI_ONLY, PATCHED, UNKNOWN_RANGE"""
    v = parse_version(version_str)
    if not v:
        return "UNKNOWN_RANGE"
    if version_str in FIXED_VERSIONS:
        return "PATCHED"
    if in_range(v, (6, 9, 0), (6, 9, 4)) or in_range(v, (7, 0, 0), (7, 0, 1)) or in_range(v, (7, 1, 0), (7, 1, 0)):
        return "VULNERABLE_RCE"
    if in_range(v, (6, 8, 0), (6, 8, 5)):
        return "VULNERABLE_SQLI_ONLY"
    if v >= (7, 0, 2) or v >= (6, 9, 5) or (v[0] == 6 and v[1] == 8 and v[2] >= 6):
        return "PATCHED"
    return "UNKNOWN_RANGE"


@dataclass
class SiteResult:
    target: str
    reachable: bool = False
    detected_version: Optional[str] = None
    version_source: Optional[str] = None
    status: str = "UNKNOWN"
    batch_route_exposed: Optional[bool] = None
    notes: list = field(default_factory=list)
    error: Optional[str] = None


def build_opener(insecure: bool):
    if insecure:
        ctx = ssl.create_default_context()
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        handler = urllib.request.HTTPSHandler(context=ctx)
        return urllib.request.build_opener(handler)
    return urllib.request.build_opener()


def http_get(opener, url: str, timeout: int):
    req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
    with opener.open(req, timeout=timeout) as resp:
        body = resp.read(200_000).decode("utf-8", errors="replace")
        return resp.status, body


def detect_version(opener, base: str, timeout: int, result: SiteResult):
    # 1) readme.html - standard core file, usually present, contains "Version X.Y.Z"
    try:
        status, body = http_get(opener, base + "/readme.html", timeout)
        if status == 200:
            m = re.search(r"[Vv]ersion\s+(\d+\.\d+(?:\.\d+)?)", body)
            if m:
                result.detected_version = m.group(1)
                result.version_source = "readme.html"
                return
    except Exception:
        pass

    # 2) homepage <meta name="generator" content="WordPress X.Y.Z">
    try:
        status, body = http_get(opener, base + "/", timeout)
        result.reachable = True
        m = re.search(r'<meta\s+name=["\']generator["\']\s+content=["\']WordPress\s+([\d.]+)', body, re.I)
        if m:
            result.detected_version = m.group(1)
            result.version_source = "homepage generator meta tag"
            return
    except Exception as e:
        result.error = f"homepage fetch failed: {e}"

    # 3) /feed/ generator tag
    try:
        status, body = http_get(opener, base + "/feed/", timeout)
        m = re.search(r"<generator>https?://wordpress\.org/\?v=([\d.]+)</generator>", body)
        if m:
            result.detected_version = m.group(1)
            result.version_source = "feed generator tag"
            return
    except Exception:
        pass


def check_batch_route(opener, base: str, timeout: int, result: SiteResult):
    """
    Informational only: checks whether the REST API batch route is
    registered/reachable. No payload is sent. A 404/403 here typically
    means the route is blocked (WAF rule, REST API disabled for anon
    users, or a WP version where the route doesn't exist) - a useful
    mitigation signal even without confirming the patch version.
    """
    for path in ("/wp-json/batch/v1", "/?rest_route=/batch/v1"):
        try:
            status, _ = http_get(opener, base + path, timeout)
            if status not in (404,):
                result.batch_route_exposed = True
                result.notes.append(f"{path} -> HTTP {status} (route reachable)")
                return
            else:
                result.notes.append(f"{path} -> HTTP {status}")
        except urllib.error.HTTPError as e:
            # WordPress REST API commonly returns 400/401/403 for malformed/blocked
            # batch calls rather than a clean exception; still informative.
            if e.code != 404:
                result.batch_route_exposed = True
                result.notes.append(f"{path} -> HTTP {e.code} (route reachable)")
                return
            result.notes.append(f"{path} -> HTTP {e.code}")
        except Exception as e:
            result.notes.append(f"{path} -> error: {e}")
    if result.batch_route_exposed is None:
        result.batch_route_exposed = False


def normalize(target: str) -> str:
    target = target.strip()
    if not target:
        return target
    if not re.match(r"^https?://", target):
        target = "https://" + target
    return target.rstrip("/")


def scan_one(target: str, insecure: bool, timeout: int) -> SiteResult:
    base = normalize(target)
    result = SiteResult(target=base)
    if not base:
        result.error = "empty target"
        return result

    opener = build_opener(insecure)

    try:
        detect_version(opener, base, timeout, result)
    except Exception as e:
        result.error = str(e)

    try:
        check_batch_route(opener, base, timeout, result)
    except Exception as e:
        result.notes.append(f"batch route check failed: {e}")

    if result.detected_version:
        result.status = classify_version(result.detected_version)
    else:
        result.status = "VERSION_NOT_DETECTED"
        result.notes.append(
            "Could not fingerprint WP version remotely (generator tag/readme.html "
            "may be hidden). Verify manually, e.g. via `wp core version` on the host."
        )

    return result


def load_targets(args) -> list:
    targets = []
    if args.targets:
        targets.extend(args.targets)
    if args.file:
        with open(args.file, "r", encoding="utf-8") as f:
            for line in f:
                line = line.split("#", 1)[0].strip()
                if line:
                    targets.append(line)
    return targets


def print_report(results: list):
    order = {
        "VULNERABLE_RCE": 0,
        "VULNERABLE_SQLI_ONLY": 1,
        "VERSION_NOT_DETECTED": 2,
        "UNKNOWN_RANGE": 3,
        "PATCHED": 4,
    }
    results = sorted(results, key=lambda r: order.get(r.status, 99))

    labels = {
        "VULNERABLE_RCE": "\033[91mVULNERABLE (RCE chain)\033[0m",
        "VULNERABLE_SQLI_ONLY": "\033[93mVULNERABLE (SQLi only, CVE-2026-60137)\033[0m",
        "PATCHED": "\033[92mPATCHED\033[0m",
        "VERSION_NOT_DETECTED": "\033[96mVERSION NOT DETECTED\033[0m",
        "UNKNOWN_RANGE": "\033[96mUNKNOWN\033[0m",
    }

    print(f"\n{'TARGET':45} {'STATUS':35} {'VERSION':10} {'BATCH ROUTE':12}")
    print("-" * 110)
    for r in results:
        v = r.detected_version or "-"
        route = "exposed" if r.batch_route_exposed else ("blocked/absent" if r.batch_route_exposed is False else "?")
        print(f"{r.target:45} {labels.get(r.status, r.status):35} {v:10} {route:12}")
        for note in r.notes:
            print(f"    - {note}")
        if r.error:
            print(f"    ! error: {r.error}")
    print()

    vulnerable = [r for r in results if r.status in ("VULNERABLE_RCE", "VULNERABLE_SQLI_ONLY")]
    unknown = [r for r in results if r.status in ("VERSION_NOT_DETECTED", "UNKNOWN_RANGE")]
    print(f"Summary: {len(vulnerable)} likely vulnerable, {len(unknown)} need manual verification, "
          f"{len(results) - len(vulnerable) - len(unknown)} patched, out of {len(results)} total.\n")


def write_csv(results: list, path: str):
    with open(path, "w", newline="", encoding="utf-8") as f:
        w = csv.writer(f)
        w.writerow(["target", "status", "detected_version", "version_source",
                    "batch_route_exposed", "notes", "error"])
        for r in results:
            w.writerow([r.target, r.status, r.detected_version, r.version_source,
                        r.batch_route_exposed, " | ".join(r.notes), r.error])


def main():
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("targets", nargs="*", help="One or more site URLs/hostnames")
    ap.add_argument("-f", "--file", help="File with one target URL/hostname per line")
    ap.add_argument("-o", "--output", help="Write results to this CSV file")
    ap.add_argument("--json", help="Write results to this JSON file")
    ap.add_argument("--workers", type=int, default=4, help="Parallel workers (default: 4)")
    ap.add_argument("--timeout", type=int, default=TIMEOUT, help="Per-request timeout in seconds")
    ap.add_argument("--insecure", action="store_true",
                     help="Skip TLS certificate verification (for internal self-signed certs)")
    args = ap.parse_args()

    targets = load_targets(args)
    if not targets:
        ap.error("No targets given. Pass URLs directly or use -f targets.txt")

    results = []
    with ThreadPoolExecutor(max_workers=args.workers) as pool:
        futures = {pool.submit(scan_one, t, args.insecure, args.timeout): t for t in targets}
        for fut in as_completed(futures):
            results.append(fut.result())

    print_report(results)

    if args.output:
        write_csv(results, args.output)
        print(f"CSV report written to {args.output}")
    if args.json:
        with open(args.json, "w", encoding="utf-8") as f:
            json.dump([asdict(r) for r in results], f, indent=2)
        print(f"JSON report written to {args.json}")

    if any(r.status in ("VULNERABLE_RCE", "VULNERABLE_SQLI_ONLY") for r in results):
        sys.exit(2)
    if any(r.status in ("VERSION_NOT_DETECTED", "UNKNOWN_RANGE") for r in results):
        sys.exit(1)
    sys.exit(0)


if __name__ == "__main__":
    main()