js_runner.py

import json
import subprocess
import time
import os
import signal
import sys
from .base import BaseRunner, RunnerResult

# Atheris instruments Python, not V8. The bridge is a generated module whose
# Nth branch stands for the Nth distinct JS coverage item, so exercising new JS
# code shows up to libFuzzer as a new Python edge.
#
# COV_SLOTS must be >= the modulo used when assigning indices below. It used to
# be 2000 branches against a `% 10000` assignment, so every JS coverage item
# past the 2000th fell through the whole chain to the same `return -1` and
# became indistinguishable from every other one.
#
# COST WARNING: fake_coverage.py is a generated if-chain of COV_SLOTS branches
# and Atheris instruments EVERY branch, once per fuzzer process. The cost is
# superlinear: 512->1.4s, 1024->4.7s, 2000->17s, 10000->465s. At 10000 a whole
# corpus config spends >120s just instrumenting and gets SIGKILLed before it
# fuzzes anything (silent all-ERR campaign, 2026-09-17). Default is therefore
# 1024 -- a coverage item past the 1024th aliases (under-reports coverage, see
# the warning at the assignment below) but never crashes. Raise TOFU_JS_COV_SLOTS
# only for a single-target run where precise JS coverage matters.
COV_SLOTS = int(os.environ.get("TOFU_JS_COV_SLOTS", "1024"))

fake_cov_path = os.path.join(os.path.dirname(__file__), "fake_coverage.py")

def _generated_slots(path):
    """Branch count of an existing generated file, or -1 if absent/unreadable."""
    try:
        with open(path) as f:
            return sum(1 for line in f if line.startswith("    if idx == "))
    except OSError:
        return -1

if _generated_slots(fake_cov_path) != COV_SLOTS:
    # Regenerate whenever the size no longer matches, otherwise a stale 2000-slot
    # file silently caps the index space again.
    with open(fake_cov_path, "w") as f:
        f.write("# Generated for Atheris coverage mapping -- DO NOT EDIT.\n")
        f.write(f"# {COV_SLOTS} slots; regenerated by js_runner when TOFU_JS_COV_SLOTS changes.\n")
        f.write("def trigger(idx):\n")
        for i in range(COV_SLOTS):
            f.write(f"    if idx == {i}: return {i}\n")
        f.write("    return -1\n")

import socket
import uuid

def _inband_error(res_data):
    """The failure a server reported INSIDE a successful response, or None.

    Two shapes, both STRUCTURED -- no guessing at prose:

    1. a top-level truthy `error`/`errors` member. mcp-server-weread answers
       every input with {'error': {'message': '没有找到cookie...'}} and no
       isError, which made a credential-blocked tool look 100% executable.

    2. an MCP content block whose text parses as JSON carrying an explicit
       `"success": false` (or its own `error`). bhouston/mcp-server-text-editor
       answers refusals with
         {"content":[{"type":"text","text":
           "{\"success\":false,\"message\":\"Path must be absolute\"}"}]}
       Measured 2026-09-10 (gpt-4.1 arm): 426 alerts on that one tool, every
       one of them the tool CORRECTLY REFUSING -- "File not found:
       /etc/passwd\u0000", "Path must be absolute", "new_str parameter is
       required" -- while check_path_traversal reported a validation bypass
       because the oracle could not see the refusal.

    Still refuses to guess: only a parsed JSON object with an explicit boolean
    false `success`, or a truthy `error`, counts. Prose containing the word
    "error" does not.
    """
    if not isinstance(res_data, dict):
        return None
    for key in ("error", "errors"):
        val = res_data.get(key)
        if val:
            return val

    for block in (res_data.get("content") or []):
        if not isinstance(block, dict) or block.get("type") != "text":
            continue
        text = block.get("text")
        if not isinstance(text, str) or "{" not in text:
            continue
        try:
            payload = json.loads(text)
        except Exception:
            continue
        if not isinstance(payload, dict):
            continue
        if payload.get("success") is False:
            return payload.get("message") or payload.get("error") or "success:false"
        for key in ("error", "errors"):
            if payload.get(key):
                return payload[key]
    return None



# Credentials that must never reach a fuzzed target. Exact names plus a
# suffix pattern; applied to the INHERITED environment only, before any
# per-target overrides are merged (so deliberate placeholders survive).
# Definitions live in runners/secrets.py so PythonRunner (in-process: must
# scrub os.environ itself) uses the identical name set. Re-exported under the
# original names; tests import _scrub_secrets from here.
from .secrets import (  # noqa: E402
    SECRET_EXACT as _SECRET_EXACT,
    SECRET_SUFFIX as _SECRET_SUFFIX,
    scrub_secrets as _scrub_secrets,
)
from .procreap import pdeathsig_prefix, reap_tagged  # noqa: E402
import atexit  # noqa: E402
import weakref  # noqa: E402

# Every JSRunner whose server may still be alive. atexit stops them on any
# Python-level exit (sys.exit, uncaught exception). It does NOT run when
# libFuzzer ends the run with its C-level exit() -- that path is covered by the
# wrapper's orphan guard, PDEATHSIG and run_eval's tag sweep (see procreap.py).
_LIVE_RUNNERS = weakref.WeakSet()


@atexit.register
def _stop_live_runners():
    for r in list(_LIVE_RUNNERS):
        try:
            r.stop()
        except Exception:
            pass

class JSRunner(BaseRunner):
    """
    A runner for Node.js/JavaScript MCP servers that uses a Unix domain socket
    for zero-latency IPC and V8 coverage extraction.
    """
    def __init__(self, command: str, args: list, target_tool: str, custom_env: dict = None,
                 cwd: str = None):
        super().__init__(command, args, target_tool)
        # BUG 21. Without this the target inherits ToFu's cwd (the repo root),
        # so "../../../../tmp/x" resolves to /tmp instead of failing inside the
        # target's own directory -- and the path-traversal oracle scores the
        # escape as a finding about the TARGET. Measured 2026-09-11:
        # excel-reader-mcp::export_to_json returned ENOENT under its own cwd and
        # wrote /tmp/evil.json under ToFu's. 14 alerts, all artefact.
        self.cwd = cwd
        self.process = None
        self.sock = None
        self.sock_path = f"/tmp/mcp_fuzz_{uuid.uuid4().hex[:8]}.sock"
        self.request_id = 1
        self.cov_map = {}
        self.next_cov_idx = 0
        self.already_covered = set()   # cumulative, for reporting only
        self.cov_exhausted = False
        self.custom_env = custom_env or {}

    def start(self):
        if self.process is not None:
            # Restart after a crash (execute() calls start() when the server
            # died): reap what the previous server left in its group first.
            self.stop()
            self.process = None
        try:
            from . import fake_coverage
        except ImportError:
            pass

        if self.command.endswith(".js"):
            target_script = self.command
            target_args = self.args
        else:
            target_script = None
            target_args = []
            for idx, arg in enumerate(self.args):
                if arg.endswith(".js"):
                    target_script = arg
                    target_args = self.args[:idx] + self.args[idx+1:]
                    break
            if not target_script:
                target_script = self.command
                target_args = self.args

        target_script = os.path.abspath(target_script)
        wrapper_path = os.path.join(os.path.dirname(__file__), "coverage_wrapper.js")

        # LEAK FIX (2026-09-23): pdeathsig_prefix() arms PR_SET_PDEATHSIG on
        # Linux so the server tree dies with the fuzzer even when libFuzzer's
        # C exit() skips stop(). See procreap.py for the 663-orphan incident.
        cmd = pdeathsig_prefix() + ["node", wrapper_path, target_script] + target_args
        # BUG 23 (2026-09-12): this was a bare os.environ.copy(). run_arm_*.sh
        # sources .env, so every fuzzed server -- untrusted third-party code --
        # inherited our real OPENAI_API_KEY / OLLAMA_CLOUD_API_KEY and could
        # have exfiltrated them. Strip credentials from the inherited env FIRST,
        # then apply custom_env, so per-target PLACEHOLDER credentials (see
        # prepare_targets.TARGET_ENV_HINTS) still reach the target.
        env = _scrub_secrets(os.environ.copy())
        env.update(self.custom_env)
        env["FUZZER_SOCKET_PATH"] = self.sock_path

        # Clean up stale socket file if exists
        if os.path.exists(self.sock_path):
            try:
                os.unlink(self.sock_path)
            except Exception:
                pass

        self.process = subprocess.Popen(
            cmd,
            cwd=self.cwd,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
            text=True,
            bufsize=1,
            env=env,
            # Own process group so the watchdog can SIGKILL the WHOLE tree. Some
            # servers spawn child shells (win-cli spawns cmd/ssh) that inherit
            # stdout; killing only the node parent leaves the pipe open and the
            # SOCKET_READY readline never gets EOF.
            start_new_session=True,
        )
        _LIVE_RUNNERS.add(self)

        # Wait for socket ready signal on stdout, BOUNDED. Many corpus node
        # servers hang on startup (wait on a backend/API) and never print
        # SOCKET_READY; this readline loop would then block until the run_eval
        # hard kill (~120s). A watchdog SIGKILLs the server after
        # TOFU_STARTUP_TIMEOUT so readline() returns EOF and we fail fast.
        import threading
        _deadline = float(os.environ.get("TOFU_STARTUP_TIMEOUT", "25"))
        _ready = {"seen": False}

        def _watchdog():
            end = time.time() + _deadline
            while time.time() < end:
                if _ready["seen"]:
                    return
                time.sleep(0.2)
            if not _ready["seen"] and self.process and self.process.poll() is None:
                # Kill the whole group (node + any child shells holding stdout),
                # so the readline below hits EOF. Fall back to killing the parent.
                try:
                    os.killpg(os.getpgid(self.process.pid), signal.SIGKILL)
                except Exception:
                    try:
                        self.process.kill()
                    except Exception:
                        pass

        threading.Thread(target=_watchdog, daemon=True).start()
        while True:
            line = self.process.stdout.readline()
            if not line:
                break
            if "SOCKET_READY:" in line:
                _ready["seen"] = True
                break
        if not _ready["seen"]:
            self.stop()
            raise ConnectionError(
                f"JS server did not signal SOCKET_READY within {_deadline}s "
                f"(startup hang; command={self.command} {self.args})")

        # Connect socket, BOUNDED by TOFU_STARTUP_TIMEOUT total. Some servers
        # print SOCKET_READY (so the watchdog above passes) but never accept() on
        # the socket -- the old settimeout(5.0) x 20 retries could block ~100s
        # here (win-cli-mcp-server). Cap the whole retry window and fail fast.
        self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        self.sock.settimeout(2.0)
        connected = False
        _connect_deadline = time.time() + float(os.environ.get("TOFU_STARTUP_TIMEOUT", "25"))
        while time.time() < _connect_deadline:
            try:
                self.sock.connect(self.sock_path)
                connected = True
                break
            except Exception:
                time.sleep(0.2)
        if not connected:
            self.stop()
            raise ConnectionError(
                f"JS server signalled ready but did not accept on {self.sock_path} "
                f"within TOFU_STARTUP_TIMEOUT (startup hang; command={self.command} {self.args})")
            
        self.sock_file = self.sock.makefile('r')

    def stop(self):
        if self.sock:
            try:
                self.sock.close()
            except Exception:
                pass
            self.sock = None
        if self.process:
            # The server runs in its own session (start_new_session=True), so
            # reap the WHOLE group -- some servers spawn child shells that would
            # otherwise orphan when only the node parent is killed. The pgid is
            # the leader's pid; do not ask os.getpgid() for it, which fails
            # once the leader is gone while its children are still in the group.
            pgid = self.process.pid
            def _killpg(sig):
                try:
                    os.killpg(pgid, sig)
                    return True
                except Exception:
                    return False
            if not _killpg(signal.SIGTERM):
                try: self.process.terminate()
                except Exception: pass
            try:
                self.process.wait(timeout=2.0)
            except subprocess.TimeoutExpired:
                pass
            # Unconditional: a child that ignores SIGTERM outlives its leader.
            if not _killpg(signal.SIGKILL) and self.process.poll() is None:
                try: self.process.kill()
                except Exception: pass
            try:
                self.process.wait(timeout=2.0)
            except Exception:
                pass
            # Anything the server started OUTSIDE its group (a grandchild that
            # called setsid, a daemonised helper) still inherits the unique
            # FUZZER_SOCKET_PATH, so sweep by that too.
            try:
                reap_tagged("FUZZER_SOCKET_PATH", self.sock_path)
            except Exception:
                pass
            _LIVE_RUNNERS.discard(self)
        if os.path.exists(self.sock_path):
            try:
                os.unlink(self.sock_path)
            except Exception:
                pass

    def execute(self, arguments: dict) -> RunnerResult:
        if not self.process or self.process.poll() is not None or not self.sock:
            try:
                self.start()
            except Exception as e:
                return RunnerResult(success=False, coverage=set(), error=f"JS runner process is not running and failed to restart: {e}")

        start_time = time.perf_counter()
        
        req = {
            "jsonrpc": "2.0",
            "method": "tools/call",
            "params": {
                "name": self.target_tool,
                "arguments": arguments
            },
            "id": self.request_id
        }
        
        try:
            self.sock.sendall((json.dumps(req) + "\n").encode('utf-8'))
            response_line = self.sock_file.readline()
        except Exception as e:
            print(f"[JSRunner] Socket Communication Error: {e}", file=sys.stderr)
            self.stop()
            return RunnerResult(success=False, coverage=set(), error=f"Socket error: {e}")

        response_obj = None
        coverage_set = set()

        if response_line:
            try:
                payload = json.loads(response_line.strip())
                cov_data = payload.get("coverage", [])
                coverage_set.update(cov_data)
                
                resp_str = payload.get("response", "")
                if resp_str.startswith("{"):
                    response_obj = json.loads(resp_str)
            except Exception:
                pass

        try:
            from . import fake_coverage
        except ImportError:
            fake_coverage = None

        if fake_coverage:
            # coverage_set is now the set of ranges executed by THIS call
            # (coverage_wrapper.js reports count deltas, not cumulative counts),
            # so every reported item is triggered on every execution. That is
            # what libFuzzer needs: the edge set must be a function of the input.
            #
            # The mapping item -> slot is stable for the life of the runner, so
            # the same JS lines always produce the same Python edges.
            for cov_item in coverage_set:
                if cov_item not in self.cov_map:
                    if self.next_cov_idx >= COV_SLOTS:
                        if not self.cov_exhausted:
                            self.cov_exhausted = True
                            print(f"[JSRunner] WARNING: JS coverage index space exhausted "
                                  f"({COV_SLOTS} slots). Further coverage items alias onto "
                                  f"existing slots and coverage is UNDER-reported from here. "
                                  f"Raise TOFU_JS_COV_SLOTS and re-run before using this run's "
                                  f"coverage numbers.", file=sys.stderr)
                        # Alias rather than silently collapsing everything onto
                        # the chain's fall-through path.
                        self.cov_map[cov_item] = hash(cov_item) % COV_SLOTS
                    else:
                        self.cov_map[cov_item] = self.next_cov_idx
                        self.next_cov_idx += 1
                fake_coverage.trigger(self.cov_map[cov_item])
            self.already_covered |= coverage_set

        end_time = time.perf_counter()
        execution_time_ms = (end_time - start_time) * 1000.0
        self.request_id += 1

        if not response_obj:
            stderr_msg = ""
            if self.process.poll() is not None:
                try:
                    stderr_msg = self.process.stderr.read().strip()
                except Exception:
                    pass
            err_detail = "Did not receive a matching response from JS process"
            if stderr_msg:
                err_detail += f". Stderr: {stderr_msg}"
            print(f"[JSRunner] JS Process Exception/Crash: {err_detail}", file=sys.stderr)
            return RunnerResult(
                success=False,
                coverage=coverage_set,
                error=err_detail,
                execution_time_ms=execution_time_ms
            )

        # Check for JSON-RPC error or tool execution success
        success = True
        error_msg = None
        result_val = None
        
        if "error" in response_obj:
            success = False
            error_msg = response_obj["error"].get("message", "JSON-RPC error")
        elif "result" in response_obj:
            res_data = response_obj["result"]
            if isinstance(res_data, dict) and res_data.get("isError"):
                success = False
                error_msg = str(res_data.get("content", "Tool reported error"))
            elif _inband_error(res_data):
                # IN-BAND ERROR. Some servers answer a failed call with a
                # NORMAL MCP result whose payload is an error object rather
                # than setting isError. Measured 2026-09-10 on
                # freestylefly/mcp-server-weread, which has no cookie
                # configured and answers EVERY input with
                #   {'error': {'message': '没有找到cookie...'}}
                # Treated as success, that tool looked 100% executable and the
                # outcome oracle concluded the target "executed successfully
                # without raising" -- 404 alerts across two tools, every one a
                # false positive, on a tool that never ran its own logic.
                #
                # sec4_design.tex:270 says the oracle evaluates "structured
                # error types and schema rejections", which is exactly this.
                success = False
                error_msg = f"in-band error payload: {str(_inband_error(res_data))[:200]}"
            else:
                result_val = res_data

        return RunnerResult(
            success=success,
            coverage=coverage_set,
            result=result_val,
            error=error_msg,
            execution_time_ms=execution_time_ms
        )