Kaicheng Yang

import inspect
from datetime import datetime, timezone
from typing import Any

from redb.extractors.enum import Tag
from redb.extractors.js_extractor import JSExtractor
from redb.extractors.js_extractors.js_patterns import FEATURE_PATTERNS
from redb.models.dataclasses import JSFeatures


_LONG_STRING_THRESHOLD = 256

# Comments still need a separate scan because the obfuscation metrics consume
# the matched text (to sum its length for comment_ratio), not just its count.
_COMMENT_RE = FEATURE_PATTERNS["comment"]


class JSFeaturesExtractor(JSExtractor):

    def __init__(
        self, filepath, log, exporters=None, index_prefix=None,
        known_benign=False, known_malicious=False, source=None, context=None,
    ):
        super().__init__(
            filepath, log, exporters, index_prefix,
            known_benign, known_malicious, source, context=context,
        )
        self.js_features = None
        self.log.debug(inspect.currentframe().f_code.co_name)

    def tag(self):
        return Tag.JS_FEATURES.value

    @staticmethod
    def _count(scan, name):
        entry = scan.get(name)
        return entry["count"] if entry else 0

    # Score tiers — see docs/js_analysis.md for the rationale behind each
    # threshold. Strong signals are ones that are unambiguous evidence of
    # obfuscation on their own (high encoding density, single-line packers,
    # 1-2 char identifiers). Weak signals are commonly seen in legitimate
    # code (eval, fromCharCode, mid-band entropy) and only count toward the
    # verdict when corroborated.
    _SCORE_STRONG_HEX_DENSITY = 0.05    # >5% of source is \xHH / \uHHHH escapes
    _SCORE_STRONG_MAX_LINE = 5000       # single line ≥5K chars (packer output)
    _SCORE_STRONG_MIN_ID = 2.0          # avg identifier length <2 chars
    _SCORE_STRONG_HIGH_ENTROPY = 5.0    # entropy >5.0 — encoded payload range
    _SCORE_STRONG_NON_ASCII = 0.30      # >30% non-ASCII codepoints in source
    _SCORE_STRONG_UNIQUE_LINES = 0.10   # <10% unique lines (with line_count >100)
    _MIN_LINES_FOR_REPETITION = 100     # below this, repetition isn't meaningful
    # Threshold is paired with the strong-signal gate in the verdict — a high
    # score alone is no longer enough, so the threshold serves as a noise
    # floor, not the false-positive prevention. The gate stops the original
    # "60/100 from weak ticks in clean code" failure mode regardless of where
    # this number sits; 40 keeps the score meaningful without re-banning
    # genuine obfuscation that lacks AST-derived signals (e.g. when
    # pyjsparser isn't installed and avg_identifier_length isn't available).
    _OBFUSCATED_THRESHOLD = 40

    def _score_obfuscation(self, metrics, scan):
        """Return `(score, strong_count, weak_count)` for the obfuscation
        heuristic. The verdict requires `score >= _OBFUSCATED_THRESHOLD` AND
        `strong_count >= 1` (or a js-x-ray hit, handled in the caller); a pile
        of weak signals alone is not enough.
        """
        score = 0
        strong = 0
        weak = 0

        src_len = len(self.js_source) if self.js_source else 1
        line_count = metrics.get('line_count', 1) or 1

        # --- Encoding density (strong / weak split by 1% vs 5%). The old
        # heuristic awarded the same +15 to a sample with 6 hex escapes in
        # 142 KB and to one that was 30% \xHH soup; this fixes that.
        hex_density = (metrics.get('hex_string_count', 0) * 4) / src_len
        unicode_density = (metrics.get('unicode_escape_count', 0) * 6) / src_len
        encoding_density = hex_density + unicode_density
        if encoding_density > self._SCORE_STRONG_HEX_DENSITY:
            score += 20
            strong += 1
        elif encoding_density > 0.01:
            score += 8
            weak += 1

        # --- Identifier length (strong: <2, weak: <3). Obfuscators rename
        # everything to `_0xNNNN` or single chars; legitimate code averages 6+.
        avg_id_len = metrics.get('avg_identifier_length', 10) or 10
        if 0 < avg_id_len < self._SCORE_STRONG_MIN_ID:
            score += 15
            strong += 1
        elif 0 < avg_id_len < 3.0:
            score += 6
            weak += 1

        # --- Single-line packers (strong: >10K, weak: >5K).
        max_line = metrics.get('max_line_length', 0)
        if max_line > 10000:
            score += 15
            strong += 1
        elif max_line > self._SCORE_STRONG_MAX_LINE:
            score += 8
            weak += 1

        # --- Text entropy (strong: >5.0). The old 4.5–5.0 weak band caught
        # jQuery/lodash and is dropped entirely.
        text_entropy = metrics.get('text_entropy', 0)
        if text_entropy > self._SCORE_STRONG_HIGH_ENTROPY:
            score += 15
            strong += 1

        # --- eval (weak; capped at +12). One eval is normal in templating,
        # AngularJS, and polyfills — it can no longer drive 30% of the verdict.
        eval_count = metrics.get('eval_count', 0)
        if eval_count > 0:
            score += min(eval_count * 4, 12)
            weak += 1

        # --- fromCharCode (weak). Common in legacy escapers but worth a tick.
        if metrics.get('fromcharcode_count', 0) > 0:
            score += 6
            weak += 1

        # --- String concatenation density (weak). >20 chains per 100 lines.
        concat_density = self._count(scan, "string_concat") / (line_count / 100)
        if concat_density > 20:
            score += 8
            weak += 1

        # --- Comment-stripped + few-line + large file (weak). Minifier tell.
        comment_ratio = metrics.get('comment_ratio', 0)
        if comment_ratio < 0.01 and line_count < 5 and src_len > 1000:
            score += 5
            weak += 1

        # --- Non-ASCII codepoint density (strong: >30%, weak: >10%). Real-world
        # JS averages <5% non-ASCII (mostly emoji or i18n string literals);
        # ≥30% almost always means a Unicode-codepoint payload (e.g. WSH
        # droppers that build a long string of non-ASCII chars and decode
        # them at runtime). Heavy localization files might cross 30% but
        # typically only score on this signal alone, which can't reach the
        # threshold by itself — the strong-signal gate prevents that
        # false-positive class while still surfacing the case where it
        # corroborates other signals.
        non_ascii_density = metrics.get('non_ascii_density', 0)
        if non_ascii_density > self._SCORE_STRONG_NON_ASCII:
            score += 20
            strong += 1
        elif non_ascii_density > 0.10:
            score += 8
            weak += 1

        # --- Line-uniqueness ratio (strong: <10%, weak: <30%, with
        # line_count >100). Hand-written or even minified code has near-1
        # line uniqueness; <10% means thousands of duplicate lines, which
        # is junk-padding / dead-code-injection used to bloat samples and
        # bury the actual payload. The line-count floor avoids
        # false-positives on tiny files that happen to repeat a few lines.
        unique_line_ratio = metrics.get('unique_line_ratio', 1.0)
        if line_count > self._MIN_LINES_FOR_REPETITION:
            if unique_line_ratio < self._SCORE_STRONG_UNIQUE_LINES:
                score += 15
                strong += 1
            elif unique_line_ratio < 0.30:
                score += 6
                weak += 1

        return min(score, 100), strong, weak

    def _detect_obfuscation_techniques(self, scan, src_len, metrics):
        """Tag the obfuscation techniques present in the source. Densities are
        computed against `src_len` so a handful of escapes in a large file
        does not get the same `hex_encoding` tag as a packed payload.

        Tags mirror the score's signals so the displayed reasoning matches
        the verdict. Three structural tags (`short_identifiers`,
        `packed_single_line`, `high_entropy`) cover the archetypes — minified
        single-line packers, renamed-identifier obfuscators, encoded-payload
        bodies — that the per-API tags below would otherwise miss entirely.
        """
        techniques = []
        if not self.js_source:
            return techniques

        if "eval" in scan:
            techniques.append("eval_usage")
        if "Function constructor" in scan:
            techniques.append("function_constructor")

        # hex_encoding / unicode_encoding by density — match the score's bar
        # so the displayed tags reflect what the score actually credited.
        hex_count = self._count(scan, "hex_escape")
        if hex_count > 5 and (hex_count * 4) / max(src_len, 1) > 0.001:
            techniques.append("hex_encoding")
        unicode_count = self._count(scan, "unicode_escape")
        if unicode_count > 5 and (unicode_count * 6) / max(src_len, 1) > 0.001:
            techniques.append("unicode_encoding")

        # charcode_encoding now requires a real cluster of calls, not one.
        if self._count(scan, "String.fromCharCode") > 3:
            techniques.append("charcode_encoding")
        if self._count(scan, "string_concat") > 10:
            techniques.append("string_concatenation")
        if "atob" in scan:
            techniques.append("base64_decoding")
        if "unescape" in scan:
            techniques.append("unescape_usage")
        if "array_function_call" in scan:
            techniques.append("array_function_calls")

        # Structural tags — surface the score's strong/weak signals so a
        # `is_likely_obfuscated: true` verdict never lands with an empty
        # techniques array (which is what happens on minified packer bodies
        # that don't match any per-API tag above).
        avg_id = metrics.get('avg_identifier_length', 0) or 0
        # Use the weak-tier bar (<3) so both strong (<2) and weak cases
        # surface — `0` means AST was unavailable from both pyjsparser and
        # js-x-ray, so we can't claim anything either way.
        if 0 < avg_id < 3.0:
            techniques.append("short_identifiers")

        # Use the weak-tier bar (>5000) so single-line packers surface even
        # below the strong 10K threshold — both cases credit the score, both
        # deserve a label.
        if metrics.get('max_line_length', 0) > 5000:
            techniques.append("packed_single_line")

        # Mirror the strong-tier entropy bar (>5.0) — the score's only
        # entropy band, since the old 4.5–4.8 weak band was dropped.
        if metrics.get('text_entropy', 0) > 5.0:
            techniques.append("high_entropy")

        # Non-ASCII codepoint payload — Unicode-character buffers that
        # decode at runtime (WSH dropper pattern). Tag at the weak bar
        # (>0.1) so any meaningful presence shows up in the techniques
        # list, even when it's not strong enough on its own.
        if metrics.get('non_ascii_density', 0) > 0.10:
            techniques.append("non_ascii_payload")

        # Repetitive padding — junk-filled bulk that buries the payload
        # under thousands of duplicate lines. Tag at the weak bar (<0.30
        # unique lines) provided the file has enough lines to make the
        # ratio meaningful.
        if (metrics.get('line_count', 0) > self._MIN_LINES_FOR_REPETITION
                and metrics.get('unique_line_ratio', 1.0) < 0.30):
            techniques.append("repetitive_padding")

        return techniques

    def _compute_ast_metrics(self, scan):
        """Compute AST-based metrics: function count, nesting depth, identifiers.

        Falls back to counts from the shared scan dict (function_decl / var_decl
        entries) when pyjsparser is unavailable.
        """
        ast = self._parse_ast()
        if not ast:
            return {
                'total_function_count': self._count(scan, "function_decl"),
                'total_variable_count': self._count(scan, "var_decl"),
                'max_nesting_depth': 0,
                'avg_identifier_length': 0.0,
            }

        counters = {'functions': 0, 'variables': 0}
        identifiers = []
        max_depth = [0]

        def walk(node, depth=0):
            if not isinstance(node, dict):
                return
            node_type = node.get('type', '')
            if node_type in ('FunctionDeclaration', 'FunctionExpression', 'ArrowFunctionExpression'):
                counters['functions'] += 1
            if node_type == 'VariableDeclaration':
                counters['variables'] += len(node.get('declarations', []))
            if node_type == 'Identifier':
                name = node.get('name', '')
                if name:
                    identifiers.append(name)

            # Track nesting depth for blocks/functions
            new_depth = depth
            if node_type in ('BlockStatement', 'FunctionDeclaration', 'FunctionExpression'):
                new_depth = depth + 1
                if new_depth > max_depth[0]:
                    max_depth[0] = new_depth

            for key, value in node.items():
                if key == 'type':
                    continue
                if isinstance(value, dict):
                    walk(value, new_depth)
                elif isinstance(value, list):
                    for item in value:
                        if isinstance(item, dict):
                            walk(item, new_depth)

        try:
            walk(ast)
        except RecursionError:
            self.log.warning(f"AST too deep for {self.hash.sha256}")

        avg_id = 0.0
        if identifiers:
            avg_id = round(sum(len(i) for i in identifiers) / len(identifiers), 2)

        return {
            'total_function_count': counters['functions'],
            'total_variable_count': counters['variables'],
            'max_nesting_depth': max_depth[0],
            'avg_identifier_length': avg_id,
        }

    def extract(self):
        src = self.js_source
        if not src:
            self.log.error(f"Empty JS source for {self.hash.sha256}")
            return None

        lines = self.lines
        line_count = len(lines)
        char_count = len(src)
        # text_entropy is over decoded characters (distinct from BasicProperties.file_entropy
        # over raw bytes); needed for non-ASCII sources where byte entropy is depressed by
        # encoding artefacts (e.g. UTF-16 nulls).
        text_entropy = self._calculate_text_entropy(src)

        line_lengths = [len(l) for l in lines] if lines else [0]
        max_line_length = max(line_lengths)
        avg_line_length = round(sum(line_lengths) / len(line_lengths), 2) if line_lengths else 0.0

        # Minification heuristic: few lines but large file, or very long average lines
        is_minified = (line_count < 5 and char_count > 500) or avg_line_length > 500

        # The shared per-sample scan dict (built once on the JSContext); every
        # count below reads from it, including _detect_obfuscation_techniques()
        # and _score_obfuscation().
        scan = self._context.scan
        # js-x-ray output is also cached on the context — same subprocess runs
        # at most once per sample regardless of how many extractors consult it.
        xray = self._context.xray

        eval_count = self._count(scan, "eval")
        function_constructor_count = self._count(scan, "Function constructor")
        settimeout_setinterval_count = self._count(scan, "settimeout_setinterval")
        document_write_count = self._count(scan, "document.write")
        innerhtml_count = self._count(scan, "innerHTML assignment")
        unescape_count = self._count(scan, "unescape")
        fromcharcode_count = self._count(scan, "String.fromCharCode")
        atob_count = self._count(scan, "atob")
        decodeuri_count = self._count(scan, "decodeURI")

        hex_string_count = self._count(scan, "hex_escape")
        unicode_escape_count = self._count(scan, "unicode_escape")
        long_string_count = self._count(scan, "long_string")
        base64_string_count = self._count(scan, "base64_string")

        # Comment ratio still needs the matched text (to sum its length), so
        # the comment regex is the one pattern we run separately.
        comments = _COMMENT_RE.findall(src)
        comment_chars = sum(len(c) for c in comments)
        comment_ratio = round(comment_chars / char_count, 4) if char_count else 0.0

        # Non-ASCII codepoint density and line-uniqueness ratio — both target
        # patterns the per-API/per-encoding signals miss: Unicode-codepoint
        # payloads (WSH droppers building runtime strings out of >0x7f chars)
        # and junk-padded bulk (thousands of duplicate lines hiding the actual
        # logic). Computed here so they ride alongside the existing metrics.
        non_ascii_count = sum(1 for c in src if ord(c) > 127)
        non_ascii_density = non_ascii_count / char_count if char_count else 0.0
        unique_lines = len({l for l in lines if l.strip()})
        unique_line_ratio = unique_lines / line_count if line_count else 1.0

        ast_metrics = self._compute_ast_metrics(scan)

        # pyjsparser is ES5.1-only — anything with destructuring, classes,
        # optional chaining, etc. fails parse and the AST path returns 0.0
        # for avg_identifier_length, which is exactly the strong signal the
        # heuristic needs to catch obfuscator.io's `_0xNNNN` renaming. js-x-ray
        # parses ES2015+ internally and reports the same statistic, so we use
        # it as the fallback when our own AST is missing.
        avg_id_length = ast_metrics['avg_identifier_length']
        if avg_id_length == 0.0 and xray.avg_identifier_length is not None:
            avg_id_length = xray.avg_identifier_length
            ast_metrics['avg_identifier_length'] = avg_id_length

        metrics = {
            'text_entropy': text_entropy,
            'eval_count': eval_count,
            'hex_string_count': hex_string_count,
            'unicode_escape_count': unicode_escape_count,
            'fromcharcode_count': fromcharcode_count,
            'line_count': line_count,
            'max_line_length': max_line_length,
            'comment_ratio': comment_ratio,
            'avg_identifier_length': avg_id_length,
            'non_ascii_density': non_ascii_density,
            'unique_line_ratio': unique_line_ratio,
        }

        obfuscation_techniques = self._detect_obfuscation_techniques(
            scan, char_count, metrics
        )
        obfuscation_score, strong_signals, _weak = self._score_obfuscation(metrics, scan)

        # Verdict: js-x-ray's recognised obfuscator family is authoritative.
        # Otherwise the heuristic must clear the threshold AND have at least
        # one strong signal — three weak ticks alone are no longer enough.
        obfuscator_name = xray.obfuscator
        is_likely_obfuscated = (
            obfuscator_name is not None
            or (obfuscation_score >= self._OBFUSCATED_THRESHOLD and strong_signals >= 1)
        )

        script_type = self._detect_script_type()
        detected_environment = self._detect_environment()

        self.js_features = JSFeatures(
            line_count=line_count,
            char_count=char_count,
            text_entropy=text_entropy,
            max_line_length=max_line_length,
            avg_line_length=avg_line_length,
            is_minified=is_minified,
            is_likely_obfuscated=is_likely_obfuscated,
            obfuscator_name=obfuscator_name,
            obfuscation_score=obfuscation_score,
            obfuscation_techniques=obfuscation_techniques,
            eval_count=eval_count,
            function_constructor_count=function_constructor_count,
            settimeout_setinterval_count=settimeout_setinterval_count,
            document_write_count=document_write_count,
            innerhtml_count=innerhtml_count,
            unescape_count=unescape_count,
            fromcharcode_count=fromcharcode_count,
            atob_count=atob_count,
            decodeuri_count=decodeuri_count,
            total_function_count=ast_metrics['total_function_count'],
            total_variable_count=ast_metrics['total_variable_count'],
            max_nesting_depth=ast_metrics['max_nesting_depth'],
            avg_identifier_length=ast_metrics['avg_identifier_length'],
            hex_string_count=hex_string_count,
            unicode_escape_count=unicode_escape_count,
            long_string_count=long_string_count,
            base64_string_count=base64_string_count,
            comment_ratio=comment_ratio,
            script_type=script_type,
            detected_environment=detected_environment,
        )
        return self.js_features

    def prepare_export_data(self, exporter_type: str) -> Any:
        if exporter_type == "ClickHouseExporter":
            if not self.js_features:
                return None

            f = self.js_features
            current_time = datetime.now(timezone.utc)
            data = [[
                self.sha256,
                f.line_count,
                f.char_count,
                f.text_entropy,
                f.max_line_length,
                f.avg_line_length,
                int(f.is_minified),
                int(f.is_likely_obfuscated),
                f.obfuscator_name or "",
                f.obfuscation_score,
                f.obfuscation_techniques,
                f.eval_count,
                f.function_constructor_count,
                f.settimeout_setinterval_count,
                f.document_write_count,
                f.innerhtml_count,
                f.unescape_count,
                f.fromcharcode_count,
                f.atob_count,
                f.decodeuri_count,
                f.total_function_count,
                f.total_variable_count,
                f.max_nesting_depth,
                f.avg_identifier_length,
                f.hex_string_count,
                f.unicode_escape_count,
                f.long_string_count,
                f.base64_string_count,
                f.comment_ratio,
                f.script_type,
                f.detected_environment,
                current_time,
            ]]

            column_names = [
                "sha256",
                "line_count", "char_count", "text_entropy",
                "max_line_length", "avg_line_length",
                "is_minified", "is_likely_obfuscated", "obfuscator_name",
                "obfuscation_score", "obfuscation_techniques",
                "eval_count", "function_constructor_count",
                "settimeout_setinterval_count", "document_write_count",
                "innerhtml_count", "unescape_count", "fromcharcode_count",
                "atob_count", "decodeuri_count",
                "total_function_count", "total_variable_count",
                "max_nesting_depth", "avg_identifier_length",
                "hex_string_count", "unicode_escape_count",
                "long_string_count", "base64_string_count",
                "comment_ratio",
                "script_type", "detected_environment",
                "analysis_date",
            ]

            column_type_names = [
                "FixedString(64)",
                "UInt32", "UInt64", "Float64",
                "UInt32", "Float64",
                "UInt8", "UInt8", "LowCardinality(String)",
                "UInt8", "Array(String)",
                "UInt32", "UInt32",
                "UInt32", "UInt32",
                "UInt32", "UInt32", "UInt32",
                "UInt32", "UInt32",
                "UInt32", "UInt32",
                "UInt16", "Float64",
                "UInt32", "UInt32",
                "UInt32", "UInt32",
                "Float64",
                "LowCardinality(String)", "LowCardinality(String)",
                "DateTime64(3, 'UTC')",
            ]

            return (data, column_names, column_type_names)

    def get_clickhouse_table(self) -> str:
        return "redb_js_features"