wpScanner.py

#!/usr/bin/env python3

import argparse
import subprocess
import sys
import os
import csv
import socket
import shutil
import importlib
import contextlib
import io
import re
import time
from datetime import datetime, timezone
from concurrent.futures import ThreadPoolExecutor, as_completed

import urllib3
import requests
from requests.exceptions import ChunkedEncodingError
from rich.console import Console
from rich.progress import Progress, SpinnerColumn, TextColumn

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

console = Console()
os.system('cls' if os.name == 'nt' else 'clear')

USER_AGENT = 'WP2Shell-Scanner (defensive version check)'

# --- WP2Shell (CVE-2026-63030 + CVE-2026-60137) affected version ranges -------
# Vulnerable: WordPress 6.9.0-6.9.4 and 7.0.0-7.0.1
# Fixed:      6.9.5, 7.0.2, 7.1+
VULN_RANGES = [((6, 9, 0), (6, 9, 4)), ((7, 0, 0), (7, 0, 1))]
FIXED_VERSIONS = "6.9.5, 7.0.2, or 7.1+"


def module_installed(name):
    buf = io.StringIO()
    try:
        with contextlib.redirect_stdout(buf), contextlib.redirect_stderr(buf):
            importlib.import_module(name)
        return True
    except ImportError:
        return False


# sublist3r is only needed for active subdomain enumeration; requests + rich are core.
required = ['requests', 'rich']
missing = [m for m in required if not module_installed(m)]
if missing:
    console.print(f"[bold red]Missing dependencies:[/bold red] {', '.join(missing)}")
    if console.input("Install now? [Y/n] ").strip().lower() in ('', 'y', 'yes'):
        subprocess.check_call([sys.executable, '-m', 'pip', 'install'] + missing)
        console.print("[green]Installed - please re-run the script.[/green]")
    sys.exit(1)

session = requests.Session()
session.trust_env = False


def load_splash():
    p = os.path.join(os.path.dirname(__file__), 'splash.txt')
    if os.path.isfile(p):
        return open(p, 'r', encoding='utf-8', errors='replace').read()
    return ""


# ---------------------------------------------------------------------------
# Subdomain enumeration
# ---------------------------------------------------------------------------

def get_crtsh_subdomains(domain):
    url = f"https://crt.sh/?q=%25.{domain}"
    try:
        resp = session.get(url, timeout=30, verify=False)
        html = resp.text
        pattern = re.compile(rf"[\w\.-]+\.{re.escape(domain)}", re.IGNORECASE)
        names = set(pattern.findall(html))
        return sorted(names)
    except Exception:
        return []


def get_subfinder_subdomains(domain):
    """Run ProjectDiscovery's subfinder if the binary is available on PATH."""
    if not shutil.which('subfinder'):
        console.print("[yellow]subfinder not found on PATH - skipping. Install from "
                      "https://github.com/projectdiscovery/subfinder[/yellow]")
        return []
    try:
        proc = subprocess.run(
            ['subfinder', '-d', domain, '-silent'],
            stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
            text=True, timeout=300, check=False
        )
        return [l.strip().lower() for l in proc.stdout.splitlines() if l.strip()]
    except Exception:
        return []


def get_subdomains(domain, use_subfinder=False):
    s3r = []
    if module_installed('sublist3r'):
        out = 'subdomains.txt'
        try:
            subprocess.run(
                [sys.executable, '-m', 'sublist3r',
                 '-e', 'bing,google,yahoo,netcraft',
                 '-d', domain, '-o', out, '-t', '50'],
                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=True
            )
        except subprocess.CalledProcessError:
            subprocess.run(
                [sys.executable, '-m', 'sublist3r', '-d', domain, '-o', out],
                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False
            )
        if os.path.isfile(out):
            with open(out) as f:
                s3r = [l.strip().lower() for l in f if l.strip()]
    else:
        console.print("[yellow]sublist3r not installed - using crt.sh only for enumeration.[/yellow]")

    crt = get_crtsh_subdomains(domain)
    sf = get_subfinder_subdomains(domain) if use_subfinder else []

    combined = []
    for name in s3r + sf + crt:
        if name not in combined:
            combined.append(name)
    if domain not in combined:
        combined.insert(0, domain)
    return combined


# ---------------------------------------------------------------------------
# Safe detection helpers
# ---------------------------------------------------------------------------

def _get(url, method='GET'):
    """Single safe request. Never sends attack payloads."""
    return session.request(
        method, url,
        headers={'User-Agent': USER_AGENT},
        timeout=10, verify=False, allow_redirects=True
    )


def get_home(domain):
    """Return (base_url, response) trying HTTPS then HTTP."""
    last_err = None
    for scheme in ('https', 'http'):
        base = f"{scheme}://{domain}"
        try:
            r = _get(base)
            return base, r, None
        except Exception as e:
            last_err = str(e)
    return None, None, last_err


def is_wordpress(base, home_html):
    """Heuristic WordPress fingerprint."""
    if home_html and re.search(r'wp-content|wp-includes|name=["\']generator["\'][^>]*WordPress',
                               home_html, re.IGNORECASE):
        return True
    # REST API index confirms WordPress even when markup is customised/cached.
    try:
        r = _get(base + '/wp-json/')
        if r.status_code == 200 and ('wp/v2' in r.text or '"namespaces"' in r.text):
            return True
    except Exception:
        pass
    return False


def parse_version(s):
    if not s:
        return None
    m = re.match(r'(\d+)(?:\.(\d+))?(?:\.(\d+))?', s.strip())
    if not m:
        return None
    return tuple(int(x) if x else 0 for x in m.groups())


def gather_versions(base, home_html):
    """Collect (source, version) from every safe, read-only fingerprint.

    Returns a list so the caller can corroborate across sources and spot a stale
    cache or a spoofed generator tag. No exploitation, no payloads.
    """
    found = []

    # 1) HTML meta generator tag
    if home_html:
        m = re.search(r'name=["\']generator["\']\s+content=["\']WordPress\s+([0-9.]+)',
                      home_html, re.IGNORECASE)
        if m:
            found.append(('meta generator', m.group(1)))

    # 2) RSS feed generator
    for feed in ('/feed/', '/?feed=rss2'):
        try:
            r = _get(base + feed)
            m = re.search(r'<generator>https?://wordpress\.org/\?v=([0-9.]+)', r.text, re.IGNORECASE)
            if m:
                found.append(('RSS feed', m.group(1)))
                break
        except Exception:
            pass

    # 3) readme.html (often present on default installs)
    try:
        r = _get(base + '/readme.html')
        if r.status_code == 200:
            # Require major.minor so we don't match "...License version 2" text.
            m = re.search(r'[Vv]ersion\s+([0-9]+\.[0-9]+(?:\.[0-9]+)?)', r.text)
            if m:
                found.append(('readme.html', m.group(1)))
    except Exception:
        pass

    # 4) OPML export generator
    try:
        r = _get(base + '/wp-links-opml.php')
        m = re.search(r'generator=["\']WordPress/([0-9.]+)', r.text, re.IGNORECASE)
        if m:
            found.append(('opml', m.group(1)))
    except Exception:
        pass

    # 5) Core asset ?ver= query string (least authoritative)
    if home_html:
        m = re.search(r'wp-includes/[^"\']+\?ver=([0-9]+\.[0-9]+(?:\.[0-9]+)?)',
                      home_html, re.IGNORECASE)
        if m:
            found.append(('asset ver (approx)', m.group(1)))

    return found


# Authoritative sources first, approximate last.
_SOURCE_PRIORITY = ['meta generator', 'RSS feed', 'readme.html', 'opml', 'asset ver (approx)']


def pick_primary(found):
    for src in _SOURCE_PRIORITY:
        for s, v in found:
            if s == src:
                return v, s
    return None, None


def endpoint_exposure(base):
    """Safely assess whether the batch REST route is actually reachable or shielded.

    Read-only (GET / OPTIONS only - never a batch payload). Distinguishes a live
    attack surface from one blocked by a WAF/proxy or a REST-lockdown mitigation, so
    a vulnerable *version* behind a mitigation is not reported as outright vulnerable.
    Returns one of: 'reachable', 'blocked', 'rest_disabled', 'unknown'.
    """
    idx_text = ''
    try:
        idx = _get(base + '/wp-json/')
        idx_text = idx.text
        # REST API itself locked down (common hardening mitigation).
        if idx.status_code in (401, 403, 451):
            return 'rest_disabled'
    except Exception:
        pass

    # Probe the specific batch route. OPTIONS returns the route schema and
    # triggers no batch processing.
    try:
        r = _get(base + '/wp-json/batch/v1', method='OPTIONS')
        if r.status_code in (401, 403, 406, 429, 451) or 500 <= r.status_code < 600:
            return 'blocked'
        if r.status_code == 200 and ('methods' in r.text or 'endpoints' in r.text
                                     or '"batch' in r.text):
            return 'reachable'
    except Exception:
        return 'unknown'

    # Fall back to the index listing if OPTIONS was inconclusive.
    if 'batch/v1' in idx_text:
        return 'reachable'
    return 'unknown'


def classify_version(vt):
    if vt is None:
        return 'UNKNOWN'
    if vt < (6, 9, 0):
        return 'NOT_AFFECTED'
    for lo, hi in VULN_RANGES:
        if lo <= vt <= hi:
            return 'VULNERABLE'
    return 'PATCHED'


def assess(domain):
    """Return a result dict for a single host. Purely diagnostic - never exploits."""
    ts = datetime.now(timezone.utc).isoformat()
    rec = {'domain': domain, 'ip': '', 'is_wordpress': False, 'wp_version': '',
           'version_source': '', 'version_sources': '', 'endpoint_status': '',
           'verdict': 'ERROR', 'detail': '', 'timestamp': ts}
    try:
        rec['ip'] = socket.gethostbyname(domain)
    except Exception:
        rec['ip'] = ''

    base, home, err = get_home(domain)
    if base is None:
        rec['verdict'] = 'ERROR'
        rec['detail'] = err or 'Host unreachable'
        return rec

    home_html = home.text if home is not None else ''
    if not is_wordpress(base, home_html):
        rec['verdict'] = 'NOT_WORDPRESS'
        rec['detail'] = 'No WordPress fingerprint found'
        return rec

    rec['is_wordpress'] = True
    found = gather_versions(base, home_html)
    primary_ver, primary_src = pick_primary(found)
    rec['version_sources'] = '; '.join(f"{s}={v}" for s, v in found)

    # Corroborate across authoritative sources (the approximate asset-ver often reports a
    # bundled library version, so it is excluded).
    auth = [(s, v) for s, v in found if s != 'asset ver (approx)']
    auth_vt = [parse_version(v) for _, v in auth if parse_version(v) is not None]
    disagree = len(set(auth_vt)) > 1

    # On disagreement, classify on the LOWEST (most vulnerable) authoritative version so a
    # stale cache or spoofed generator can never produce a falsely reassuring verdict.
    if disagree and auth_vt:
        low = min(auth_vt)
        class_ver = next((v for s, v in auth if parse_version(v) == low), primary_ver)
        class_src = 'lowest of disagreeing sources'
    else:
        class_ver, class_src = primary_ver, primary_src

    rec['wp_version'] = class_ver or ''
    rec['version_source'] = class_src or ''

    exposure = endpoint_exposure(base)
    rec['endpoint_status'] = exposure

    vclass = classify_version(parse_version(class_ver))

    if vclass == 'VULNERABLE':
        if exposure in ('blocked', 'rest_disabled'):
            rec['verdict'] = 'LIKELY_MITIGATED'
            rec['detail'] = (f"WordPress {class_ver} is a vulnerable version, but the batch "
                             f"endpoint appears {exposure.replace('_', ' ')} (a WAF or REST "
                             "lockdown seems to be in front). Patch core to be certain.")
        else:
            rec['verdict'] = 'VULNERABLE'
            rec['detail'] = (f"WordPress {class_ver} is within the WP2Shell affected range "
                             f"and the batch endpoint is {exposure}")
    elif vclass in ('PATCHED', 'NOT_AFFECTED'):
        rec['verdict'] = 'PATCHED'
        rec['detail'] = (f"WordPress {class_ver} is patched against WP2Shell"
                         if vclass == 'PATCHED'
                         else f"WordPress {class_ver} predates the affected releases")
    else:
        rec['verdict'] = 'VERSION_UNKNOWN'
        rec['detail'] = ("WordPress detected but version could not be read; batch "
                         f"endpoint is {exposure}. Confirm the version manually")

    if disagree:
        rec['detail'] += (f" | NOTE: sources disagree ({rec['version_sources']}); classified "
                          "on the lowest to stay safe - verify with `wp core version`")
    return rec


def assess_with_retries(domain, retries):
    rec = None
    for _ in range(retries):
        rec = assess(domain)
        if rec['verdict'] != 'ERROR':
            return rec
        time.sleep(1)
    return rec


# ---------------------------------------------------------------------------

REMEDIATION = """
[bold]How to protect yourself against WP2Shell[/bold]
  1. Update WordPress core to a fixed release: {fixed}.
     (Dashboard > Updates, or `wp core update` via WP-CLI.)
  2. Ensure auto-updates for core are enabled so forced fixes land automatically.
  3. If you cannot patch immediately, block /wp-json/batch/v1 at your WAF/reverse
     proxy and restrict anonymous REST API access.
  4. After patching, review access logs for prior anomalous requests to
     ?rest_route=/batch/v1 or /wp-json/batch/v1, and check for unfamiliar admin
     accounts, plugins, or PHP files (post-compromise indicators).

References:
  - https://www.rapid7.com/blog/post/etr-cve-2026-63030-wp2shell-a-critical-remote-code-execution-vulnerability-in-wordpress-core/
  - https://www.wiz.io/blog/wp2shell-cve-2026-63030-cve-2026-60137
""".format(fixed=FIXED_VERSIONS)


def main():
    parser = argparse.ArgumentParser(description='WP2Shell Scanner by DanSec - '
                                                 'CVE-2026-63030 / CVE-2026-60137 exposure check')
    parser.add_argument('domain', help='Root domain or host to check (e.g. example.com)')
    parser.add_argument('-o', '--output', default='WP2Shell_output.csv',
                        help='CSV output file')
    parser.add_argument('--passive', action='store_true',
                        help='Skip subdomain enumeration (check the given host only)')
    parser.add_argument('--subfinder', action='store_true',
                        help='Also enumerate subdomains with ProjectDiscovery subfinder '
                             '(must be installed and on PATH)')
    parser.add_argument('--threads', type=int, default=1,
                        help='Concurrent workers (default: 1)')
    parser.add_argument('--retries', type=int, default=1,
                        help='Probe retries on failure (default: 1)')
    parser.add_argument('--rate-limit', type=float, default=0,
                        help='Max requests/sec in single-threaded mode (0=no limit)')
    args = parser.parse_args()

    splash = load_splash()
    if splash:
        console.print(splash, style="bold red")

    console.print(
        "[yellow]Note:[/yellow] This tool only checks the WordPress version and REST "
        "exposure. It never exploits the vulnerability.\n", style="white"
    )
    console.print(
        f"Target: [bold]{args.domain}[/bold]  Threads: [bold]{args.threads}[/bold]  "
        f"Retries: [bold]{args.retries}[/bold]  Rate-limit: [bold]{args.rate_limit}[/bold]\n"
    )

    if args.passive:
        subs = [args.domain]
        console.print("[cyan]Passive mode:[/cyan] checking the given host only\n")
    else:
        try:
            with console.status("[cyan]Enumerating subdomains...[/cyan]", spinner="dots"):
                subs = get_subdomains(args.domain, use_subfinder=args.subfinder)
        except Exception as e:
            console.print(f"[yellow]Warning:[/yellow] {e} - switching to passive mode\n")
            subs = [args.domain]

    if not subs:
        console.print("[yellow]No subdomains found - passive mode[/yellow]\n")
        subs = [args.domain]

    console.print(f"Will check {len(subs)} host(s): {', '.join(subs)}\n")

    results = []

    def scan_one(domain):
        if args.threads == 1 and args.rate_limit > 0:
            time.sleep(1.0 / args.rate_limit)
        return assess_with_retries(domain, args.retries)

    with Progress(SpinnerColumn(), TextColumn("{task.description}"), transient=True) as progress:
        task = progress.add_task("Checking...", total=len(subs))
        if args.threads == 1:
            for d in subs:
                rec = scan_one(d)
                results.append(rec)
                progress.update(task, description=f"{d}: {rec['verdict']}")
                progress.advance(task)
        else:
            with ThreadPoolExecutor(max_workers=args.threads) as pool:
                futures = {pool.submit(scan_one, d): d for d in subs}
                for fut in as_completed(futures):
                    d = futures[fut]
                    rec = fut.result()
                    results.append(rec)
                    progress.update(task, description=f"{d}: {rec['verdict']}")
                    progress.advance(task)

    out = os.path.abspath(args.output)
    with open(out, 'w', newline='', encoding='utf-8') as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames=[
            'domain', 'ip', 'is_wordpress', 'wp_version', 'version_source',
            'version_sources', 'endpoint_status', 'verdict', 'detail', 'timestamp'
        ])
        writer.writeheader()
        writer.writerows(results)

    total = len(results)
    vuln = sum(1 for r in results if r['verdict'] == 'VULNERABLE')
    mitigated = sum(1 for r in results if r['verdict'] == 'LIKELY_MITIGATED')
    patched = sum(1 for r in results if r['verdict'] == 'PATCHED')
    unknown = sum(1 for r in results if r['verdict'] == 'VERSION_UNKNOWN')
    not_wp = sum(1 for r in results if r['verdict'] == 'NOT_WORDPRESS')
    errors = sum(1 for r in results if r['verdict'] == 'ERROR')

    console.print(f"\nResults saved to [bold]{out}[/bold]\n")
    for r in results:
        colour = {'VULNERABLE': 'bold red', 'LIKELY_MITIGATED': 'yellow',
                  'PATCHED': 'green', 'VERSION_UNKNOWN': 'yellow',
                  'NOT_WORDPRESS': 'cyan', 'ERROR': 'magenta'}.get(r['verdict'], 'white')
        ver = f" (v{r['wp_version']})" if r['wp_version'] else ""
        console.print(f"  [{colour}]{r['verdict']:<16}[/{colour}] {r['domain']}{ver} - {r['detail']}")

    console.print("")
    console.print(f"[bold red]VULNERABLE[/bold red]:       {vuln}/{total}")
    console.print(f"[yellow]LIKELY MITIGATED[/yellow]: {mitigated}/{total}")
    console.print(f"[green]PATCHED/SAFE[/green]:     {patched}/{total}")
    console.print(f"[yellow]VERSION UNKNOWN[/yellow]:  {unknown}/{total}")
    console.print(f"[cyan]NOT WORDPRESS[/cyan]:    {not_wp}/{total}")
    console.print(f"[magenta]ERRORS[/magenta]:          {errors}/{total}\n")

    if vuln or mitigated or unknown:
        console.print(REMEDIATION)


if __name__ == '__main__':
    main()