Hang Fan

"""Subprocess wrapper for the bundled js-x-ray Node bridge.

Mirrors `js_deobfuscator.py`: shell out to a Node script with a per-sample
timeout, kill the process group on hang, demote `FileNotFoundError` to debug
(missing tool is routine — the host either has Node + the bundled package
installed or it doesn't), and return a structured result on success.

The bridge lives at `redb/extractors/js_extractors/scripts/js-xray-runner.js`.
Operators install the JS dependency once with `npm install` in that directory
(or override the path with `JS_XRAY_RUNNER_PATH`).

Configuration (env vars):
    JS_XRAY_RUNNER_PATH   Path to the Node bridge script (default: bundled).
    JS_XRAY_TIMEOUT       Seconds before the subprocess is killed. Default: 30.

`run(source, log)` returns `XRayResult(obfuscator, warnings)` on a successful
analysis, or `XRayResult(None, [])` for any non-success path (binary missing,
timeout, parse failure, etc.). The two unsuccessful states are
indistinguishable to the caller on purpose — they all collapse to "no
js-x-ray verdict, fall back to heuristic".
"""

from __future__ import annotations

import json
import os
import signal
import subprocess
import tempfile
from dataclasses import dataclass, field
from typing import List, Optional

# Bundled bridge: redb/extractors/js_extractors/scripts/js-xray-runner.js
_DEFAULT_RUNNER = os.path.join(
    os.path.dirname(__file__), "scripts", "js-xray-runner.js"
)
_DEFAULT_NODE = "node"
_DEFAULT_TIMEOUT_SECS = 30


@dataclass
class XRayResult:
    """Parsed js-x-ray output. `obfuscator` is the recognised family name
    (e.g. "jsfuck", "obfuscator.io") or None when js-x-ray did not flag the
    code. `warnings` carries every {kind, value} pair the analyser produced;
    the heuristic uses it as a corroborating signal. `avg_identifier_length`
    is js-x-ray's own AST-derived figure — used as a fallback for the
    heuristic's `avg_identifier_length<2` strong signal when pyjsparser
    can't parse the source (anything ES2015+ trips it)."""

    obfuscator: Optional[str] = None
    warnings: List[dict] = field(default_factory=list)
    avg_identifier_length: Optional[float] = None

    @property
    def flagged(self) -> bool:
        return self.obfuscator is not None


def _empty() -> XRayResult:
    return XRayResult(obfuscator=None, warnings=[])


def run(source: str, log) -> XRayResult:
    if not source:
        return _empty()

    runner = os.getenv("JS_XRAY_RUNNER_PATH", _DEFAULT_RUNNER)
    node_bin = os.getenv("JS_XRAY_NODE_BIN", _DEFAULT_NODE)
    timeout = int(os.getenv("JS_XRAY_TIMEOUT", str(_DEFAULT_TIMEOUT_SECS)))

    if not os.path.exists(runner):
        log.debug(f"js-x-ray runner not found at {runner}")
        return _empty()

    # Skip the subprocess entirely when the JS dependency isn't installed.
    # Without this, every call to a host that has `node` but never ran
    # `npm install` next to the runner would still fork node, get a require
    # error, and exit nonzero — wasted ~50–200ms per JS sample (and per test).
    runner_dir = os.path.dirname(runner)
    if not os.path.isdir(os.path.join(runner_dir, "node_modules", "@nodesecure", "js-x-ray")):
        log.debug(f"@nodesecure/js-x-ray not installed in {runner_dir}")
        return _empty()

    tmp_path = None
    try:
        with tempfile.NamedTemporaryFile(
            suffix=".js", mode="w", delete=False, encoding="utf-8"
        ) as tmp:
            tmp.write(source)
            tmp_path = tmp.name

        try:
            process = subprocess.Popen(
                [node_bin, runner, tmp_path],
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                preexec_fn=os.setsid,
            )
            try:
                stdout, stderr = process.communicate(timeout=timeout)
            except subprocess.TimeoutExpired:
                # Kill the whole process group so any node helpers die too.
                try:
                    os.killpg(os.getpgid(process.pid), signal.SIGTERM)
                    process.wait(timeout=5)
                except Exception:
                    try:
                        os.killpg(os.getpgid(process.pid), signal.SIGKILL)
                    except Exception:
                        pass
                log.warning(f"js-x-ray timed out after {timeout}s")
                return _empty()

            if process.returncode != 0:
                err = stderr.decode("utf-8", errors="replace").strip()
                log.debug(f"js-x-ray exited {process.returncode}: {err}")
                return _empty()

            text = stdout.decode("utf-8", errors="replace").strip()
            if not text:
                return _empty()

            try:
                payload = json.loads(text)
            except json.JSONDecodeError as e:
                log.warning(f"js-x-ray emitted non-JSON output: {e}")
                return _empty()

            obfuscator = payload.get("obfuscator")
            warnings = payload.get("warnings") or []
            if not isinstance(warnings, list):
                warnings = []

            ids_avg = payload.get("idsLengthAvg")
            if not isinstance(ids_avg, (int, float)):
                ids_avg = None

            return XRayResult(
                obfuscator=obfuscator,
                warnings=warnings,
                avg_identifier_length=ids_avg,
            )
        finally:
            if tmp_path:
                try:
                    os.unlink(tmp_path)
                except Exception:
                    pass
    except FileNotFoundError:
        log.debug(f"node binary not found at {node_bin}")
        return _empty()
    except Exception as e:
        log.error(f"js-x-ray subprocess error: {e}")
        return _empty()