git clone https://github.com/icarax/wordpress-rce-scanner.git
cd wordpress-rce-scanner

# Create a virtual env (recommended)
python -m venv .venv
source .venv/bin/activate   # Windows: .venv\Scripts\activate

# Install dependencies
pip install --upgrade pip
pip install requests beautifulsoup4 lxml tqdm python-dotenv

# Initialize npm project (if not already)
npm init -y

# Install core deps
npm install axios dotenv chalk progress

# Install TypeScript & typings (optional but recommended)
npm install --save-dev typescript @types/node ts-node nodemon
npx tsc --init   # creates tsconfig.json

wordpress-rce-scanner/
│
├─ python/
│   ├─ scanner.py
│   └─ .env.example
│
├─ js/
│   ├─ scanner.ts
│   └─ .env.example
│
└─ README.md

#!/usr/bin/env python3
"""
WordPress version scanner – defensive tool to spot sites running
known‑vulnerable WP core versions that could lead to unauthenticated RCE.

Usage:
    python scanner.py --targets targets.txt
"""

import argparse
import json
import os
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Tuple

import requests
from bs4 import BeautifulSoup
from dotenv import load_dotenv
from tqdm import tqdm

# ----------------------------------------------------------------------
# Load configuration (optional API key for WPScan DB, etc.)
# ----------------------------------------------------------------------
load_dotenv()  # reads .env file
WPSCAN_API_TOKEN = os.getenv("WPSCAN_API_TOKEN")  # optional, not required for basic version check

# ----------------------------------------------------------------------
# Constants
# ----------------------------------------------------------------------
WP_VERSION_CHECK_URL = "https://api.wordpress.org/core/version-check/1.7/"
KNOWN_VULNERABLE_FILE = os.path.join(os.path.dirname(__file__), "known_vulnerabilities.json")
TIMEOUT = 10  # seconds for each HTTP request
MAX_WORKERS = 10  # adjust based on your network/CPU

# ----------------------------------------------------------------------
# Helper functions
# ----------------------------------------------------------------------
def fetch_latest_wp_version() -> str:
    """
    Query wordpress.org for the latest stable release.
    Returns a version string like "6.5.2".
    """
    resp = requests.get(WP_VERSION_CHECK_URL, timeout=TIMEOUT)
    resp.raise_for_status()
    data = resp.json()
    # The API returns offers[]; the first offer is the current stable.
    return data["offers"][0]["current"]


def load_known_vulnerabilities() -> dict:
    """
    Expected JSON format:
    {
        "CVE-2021-29447": {"fixed_in": "5.8.1", "description": "..."},
        ...
    }
    """
    with open(KNOWN_VULNERABLE_FILE, "r", encoding="utf-8") as f:
        return json.load(f)


def get_wp_version_from_site(base_url: str) -> str | None:
    """
    Try to read the version from readme.html or license.txt.
    Returns None if version cannot be determined.
    """
    candidates = [
        f"{base_url.rstrip('/')}/readme.html",
        f"{base_url.rstrip('/')}/license.txt",
        f"{base_url.rstrip('/')}/wp-includes/version.php",  # fallback (may expose raw PHP)
    ]

    for url in candidates:
        try:
            r = requests.get(url, timeout=TIMEOUT, headers={"User-Agent": "WP-Scanner/1.0"})
            if r.status_code != 200:
                continue

            # Prefer HTML parsing for readme.html
            if url.endswith(".html"):
                soup = BeautifulSoup(r.text, "lxml")
                # Look for a line like "Version 6.5.2"
                for tag in soup.stripped_strings:
                    if tag.lower().startswith("version"):
                        # Extract numeric part
                        parts = tag.split()
                        if len(parts) >= 2 and parts[1].replace(".", "").isdigit():
                            return parts[1]
            else:
                # Plain text fallback – search for "Version X.Y.Z"
                import re
                match = re.search(r"Version\s+(\d+\.\d+\.\d+)", r.text, re.I)
                if match:
                    return match.group(1)
        except requests.RequestException:
            continue  # try next candidate
    return None


def version_is_vulnerable(version: str, vuln_db: dict) -> List[Tuple[str, str]]:
    """
    Return a list of (CVE, fixed_in) for which the supplied version is vulnerable.
    """
    vulnerable = []
    from packaging import version as pkg_version

    v = pkg_version.parse(version)
    for cve, info in vuln_db.items():
        fixed = pkg_version.parse(info["fixed_in"])
        if v < fixed:
            vulnerable.append((cve, info["fixed_in"]))
    return vulnerable


def scan_one(target: str, latest: str, vuln_db: dict) -> dict:
    """
    Scan a single WordPress site.
    Returns a dict with results suitable for JSON/pretty printing.
    """
    result = {
        "target": target,
        "wp_version": None,
        "latest_stable": latest,
        "vulnerable": [],
        "notes": [],
    }

    try:
        wp_ver = get_wp_version_from_site(target)
        if wp_ver is None:
            result["notes"].append("Could not determine WP version (maybe hidden or non‑WP).")
            return result

        result["wp_version"] = wp_ver

        # Version check against latest stable (informational)
        if wp_ver != latest:
            result["notes"].append(f"Running {wp_ver} while latest is {latest}.")

        # Vulnerability check
        vulns = version_is_vulnerable(wp_ver, vuln_db)
        if vulns:
            result["vulnerable"] = [{"cve": cve, "fixed_in": fix} for cve, fix in vulns]

    except Exception as exc:  # pragma: no cover – defensive
        result["notes"].append(f"Error during scan: {exc}")

    return result


def main() -> None:
    parser = argparse.ArgumentParser(description="Defensive WordPress version & vulnerability scanner")
    parser.add_argument(
        "-t",
        "--targets",
        required=True,
        help="Path to a file containing one WordPress base URL per line (e.g., https://example.com)",
    )
    parser.add_argument(
        "-o",
        "--output",
        default="scan_results.json",
        help="File to write JSON results (default: scan_results.json)",
    )
    parser.add_argument(
        "-w",
        "--workers",
        type=int,
        default=MAX_WORKERS,
        help=f"Number of concurrent workers (default: {MAX_WORKERS})",
    )
    args = parser.parse_args()

    # ------------------------------------------------------------------
    # Load data
    # ------------------------------------------------------------------
    try:
        latest_wp = fetch_latest_wp_version()
        print(f"[+] Latest WordPress stable version: {latest_wp}")
    except Exception as e:
        sys.exit(f"[-] Failed to fetch latest WP version: {e}")

    vuln_db = load_known_vulnerabilities()
    print(f"[+] Loaded {len(vuln_db)} known vulnerability entries.")

    # Read targets
    with open(args.targets, "r", encoding="utf-8") as f:
        targets = [line.strip() for line in f if line.strip() and not line.startswith("#")]

    if not targets:
        sys.exit("[-] No targets supplied.")

    # ------------------------------------------------------------------
    # Scan with thread pool
    # ------------------------------------------------------------------
    results = []
    with ThreadPoolExecutor(max_workers=args.workers) as executor:
        future_to_target = {
            executor.submit(scan_one, t, latest_wp, vuln_db): t for t in targets
        }
        for future in tqdm(
            as_completed(future_to_target),
            total=len(targets),
            desc="Scanning",
            unit="site",
        ):
            results.append(future.result())

    # ------------------------------------------------------------------
    # Output
    # ------------------------------------------------------------------
    with open(args.output, "w", encoding="utf-8") as f:
        json.dump(results, f, indent=2, ensure_ascii=False)

    print(f"\n[+] Scan complete. Results written to {args.output}")

    # Quick summary
    vulnerable_sites = [r for r in results if r["vulnerable"]]
    if vulnerable_sites:
        print(f"[!] {len(vulnerable_sites)} site(s) appear to run a vulnerable WP core:")
        for site in vulnerable_sites[:5]:  # show first 5
            print(
                f"    - {site['target']} (WP {site['wp_version']}) → "
                + ", ".join([v["cve"] for v in site["vulnerable"]])
            )
        if len(vulnerable_sites) > 5:
            print(f"    ... and {len(vulnerable_sites)-5} more.")
    else:
        print("[+] No known vulnerable versions detected among the scanned sites.")


if __name__ == "__main__":
    main()

#!/usr/bin/env node
/**
 * Defensive WordPress version scanner (Node/TS)
 * Mirrors the functionality of the Python scanner.
 *
 * Usage:
 *   npx ts-node scanner.ts --targets targets.txt
 *   # or after building:
 *   node dist/scanner.js --targets targets.txt
 */

import * as dotenv from "dotenv";
import * as fs from "fs";
import * as path from "path";
import axios from "axios";
import { version as semver } from "semver";
import { pipeline } from "stream";
import { promisify } from "util";
import * as readline from "readline";

dotenv.config();

const WP_VERSION_CHECK_URL =
  "https://api.wordpress.org/core/version-check/1.7/";
const KNOWN_VULN_PATH = path.join(__dirname, "known_vulnerabilities.json");
const TIMEOUT = 10_000; // ms
const MAX_CONCURRENT = 10;

// ------------------------------------------------------------------
// Types
// ------------------------------------------------------------------
interface VulnInfo {
  fixed_in: string;
  description?: string;
}
interface KnownVulns {
  [cve: string]: VulnInfo;
}
interface ScanResult {
  target: string;
  wp_version?: string;
  latest_stable: string;
  vulnerable: Array<{ cve: string; fixed_in: string }>;
  notes: string[];
}

// ------------------------------------------------------------------
// Helper functions
// ------------------------------------------------------------------
async function fetchLatestWPVersion(): Promise<string> {
  const { data } = await axios.get(WP_VERSION_CHECK_URL, { timeout: TIMEOUT });
  // data.offers[0].current holds the latest stable version string
  return data.offers[0].current;
}

function loadKnownVulnerabilities(): KnownVulns {
  const raw = fs.readFileSync(KnownVulnPath, "utf8");
  return JSON.parse(raw);
}

/**
 * Try to read version from readme.html or license.txt.
 * Returns null if not found.
 */
async function getWPVersionFromSite(baseUrl: string): Promise<string | null> {
  const candidates = [
    `${baseUrl.replace(/\/+$/g, "")}/readme.html`,
    `${baseUrl.replace(/\/+$/g, "")}/license.txt`,
  ];

  for const url of candidates) {
    try {
      const resp = await axios.get(url, {
        timeout: TIMEOUT,
        headers: { "User-Agent": "WP-Scanner/1.0" },
        validateStatus: (status) => status < 500, // treat 4xx as OK (we'll check body)
      });

      const text = resp.data as string;

      // HTML case
      if (url.endsWith(".html")) {
        const versionMatch = text.match(/Version\s+(\d+\.\d+\.\d+)/i);
        if (versionMatch) {
          return versionMatch[1];
        }
      } else {
        // plain text fallback
        const versionMatch = text.match(/Version\s+(\d+\.\d+\.\d+)/i);
        if (versionMatch) {
          return versionMatch[1];
        }
      }
    } catch (err: any) {
      // ignore network errors and try next candidate
      if (err.code !== "ECONNABORTED" && err.response?.status !== 404) {
        // unexpected error – we could log it, but continue
      }
      continue;
    }
  }
  return null;
}

/**
 * Determine which CVEs affect the discovered version.
 */
function getVulnerableCVEs(
  version: string,
  vulnDB: KnownVulns
): Array<{ cve: string; fixed_in: string }> {
  const vulnerable: Array<{ cve: string; fixed_in: string }> = [];

  for (const [cve, info] of Object.entries(vulnDB)) {
    if (semver.lt(version, info.fixed_in)) {
      vulnerable.push({ cve, fixed_in: info.fixed_in });
    }
  }
  return vulnerable;
}

async function scanOne(
  target: string,
  latest: string,
  vulnDB: KnownVulns
): Promise<ScanResult> {
  const result: ScanResult = {
    target,
    latest_stable: latest,
    vulnerable: [],
    notes: [],
  };

  try {
    const wpVer = await getWPVersionFromSite(target);
    if (!wpVer) {
      result.notes.push("Could not determine WordPress version (maybe hidden or non‑WP).");
      return result;
    }

    result.wp_version = wpVer;

    if (wpVer !== latest) {
      result.notes.push(
        `Running ${wpVer} while latest stable is ${latest}.`
      );
    }

    const vulns = getVulnerableCVEs(wpVer, vulnDB);
    if (vulns.length) {
      result.vulnerable = vulns;
    }
  } catch (err: any) {
    result.notes.push(`Error during scan: ${err.message}`);
  }

  return result;
}

/**
 * Read a file line‑by‑line (targets) and return an array of non‑empty, non‑comment lines.
 */
function readTargets(filePath: string): Promise<string[]> {
  return new Promise((resolve, reject) => {
    const out: string[] = [];
    const rl = readline.createInterface({
      input: fs.createReadStream(filePath),
      crlfDelay: Infinity,
    });

    rl.on("line", (line) => {
      const trimmed = line.trim();
      if (trimmed && !trimmed.startsWith("#")) {
        out.push(trimmed);
      }
    });

    rl.on("close", () => resolve(out));
    rl.on("error", reject);
  });
}

// ------------------------------------------------------------------
// Main
// ------------------------------------------------------------------
(async () => {
  const args = process.argv.slice(2);
  const argMap: Record<string, string> = {};
  for (let i = 0; i < args.length; i += 2) {
    const key = args[i].replace(/^--/, "");
    const value = args[i + 1];
    argMap[key] = value;
  }

  const targetsFile = argMap.targets;
  if (!targetsFile) {
    console.error("Error: --targets <file> is required.");
    process.exit(1);
  }

  const outputFile = argMap.output || "scan_results.json";
  const workers = Number(argMap.workers) || MAX_CONCURRENT;

  try {
    const [latestWP, vulnDB, targets] = await Promise.all([
      fetchLatestWPVersion(),
      Promise.resolve(loadKnownVulnerabilities()),
      readTargets(targetsFile),
    ]);

    console.log(`[+] Latest WordPress stable: ${latestWP}`);
    console.log(`[+] Loaded ${Object.keys(vulnDB).length} known vulnerability entries.`);
    console.log(`[+] Scanning ${targets.length} target(s) with ${workers} workers...\n`);

    // Simple worker pool using Promise.all with chunking
    const chunkSize = workers;
    const results: ScanResult[] = [];

    for (let i = 0; i < targets.length; i += chunkSize) {
      const chunk = targets.slice(i, i + chunkSize);
      const promises = chunk.map((t) => scanOne(t, latestWP, vulnDB));
      const chunkResults = await Promise.all(promises);
      results.push(...chunkResults);

      // Optional progress indicator
      process.stdout.write(
        `\r[+] Processed ${Math.min(i + chunkSize, targets.length)}/${targets.length} sites`
      );
    }
    console.log("\n");

    // Write results
    fs.writeFileSync(outputFile, JSON.stringify(results, null, 2), "utf8");
    console.log(`[+] Scan complete. Results written to ${outputFile}`);

    // Summary
    const vulnerable = results.filter((r) => r.vulnerable.length > 0);
    if (vulnerable.length) {
      console.warn(
        `[!] ${vulnerable.length} site(s) appear to run a vulnerable WP core:`
      );
      vulnerable.slice(0, 5).forEach((r) => {
        console.log(
          `    - ${r.target} (WP ${r.wp_version}) →`,
          r.vulnerable.map((v) => v.cve).join(", ")
        );
      });
      if (vulnerable.length > 5) {
        console.log(`    ... and ${vulnerable.length - 5} more.`);
      }
    } else {
      console.log("[+] No known vulnerable versions detected.");
    }
  } catch (err: any) {
    console.error("[-] Fatal error:", err.message);
    process.exit(1);
  }
})();

# Optional: If you want to use a proxy for outbound requests (e.g., Burp, ZAP)
HTTP_PROXY=http://127.0.0.1:8080
HTTPS_PROXY=http://127.0.0.1:8080

# Optional: WPScan API token (if you later extend the scanner to query their vulnerability DB)
WPSCAN_API_TOKEN=your_wpscan_token_here

# Optional: Adjust timeout (seconds) – overrides default in code
REQUEST_TIMEOUT=15

{
  "CVE-2021-29447": {
    "fixed_in": "5.8.1",
    "description": "Unauthenticated PHP Object Injection in WP Core leading to RCE via certain plugins/themes."
  },
  "CVE-2022-21663": {
    "fixed_in": "6.0.2",
    "description": "Unauthenticated REST API endpoint allowing arbitrary file upload → RCE."
  },
  "CVE-2022-21661": {
    "fixed_in": "6.0.2",
    "description": "Unauthenticated Cross‑Site Scripting (XSS) that can be chained to RCE via admin‑ajax."
  },
  "CVE-2023-XXXX": {
    "fixed_in": "6.5.0",
    "description": "Placeholder for future CVE – replace with real data when disclosed."
  }
}

def http_get(url: str, **kwargs) -> requests.Response:
    """Centralized GET with timeout, retry, and user‑agent."""
    headers = kwargs.pop("headers", {})
    headers.setdefault("User-Agent", "WP-Scanner/1.0")
    return requests.get(
        url,
        timeout=kwargs.pop("timeout", TIMEOUT),
        headers=headers,
        **kwargs,
    )

from packaging import version as pkg_ver

def normalize(v: str) -> str:
    """Strip any non‑numeric suffixes (e.g., '6.5.2‑rc1' → '6.5.2')."""
    return pkg_ver.parse(v).base_version