Mallik Tatipamula

import logging
import re
from abc import ABCMeta, abstractmethod

from redb.extractors.extractor import Extractor
from redb.extractors.js_extractors.js_context import JSContext, _text_entropy

logger = logging.getLogger(__name__)

# ESM is recognised by line-anchored `import ... from "..."` / bare side-effect
# `import "..."` / top-level `export ...`. Anchored at line start to avoid
# matching the substring inside string literals or comments.
_ESM_PATTERN = re.compile(
    r'(?m)^\s*(?:'
    r'import\s+[^;\n]*?\bfrom\s+[\'"]'
    r'|import\s+[\'"][^\'"]+[\'"]'
    r'|export\s+(?:default\b|\{|\*|const\b|let\b|var\b|function\b|class\b|async\b)'
    r')'
)


@abstractmethod
class JSExtractor(Extractor, metaclass=ABCMeta):
    """Base class for JavaScript file extractors.

    Every JSExtractor reads its raw materials (bytes / decoded source / line
    list / scan_source results / pyjsparser AST / text entropy) from a shared
    `JSContext`. When workers.py drives the JS pipeline it builds one context
    per sample and threads it into every extractor via `context=`. When tests
    or other callers instantiate an extractor directly, the constructor builds
    a fresh context from `(filepath, source=...)`.

    All historical instance attributes (`self.binary`, `self.js_source`,
    `self.lines`) and helpers (`self._decode_source`, `self._parse_ast`,
    `self._calculate_text_entropy`) are preserved as thin delegators so
    existing extractor code keeps working unchanged.
    """

    def __init__(
        self,
        filepath,
        log,
        exporters=None,
        index_prefix=None,
        known_benign=False,
        known_malicious=False,
        source=None,
        context=None,
    ):
        if context is None:
            context = JSContext.from_path(filepath, log=log, source=source)
        elif source is not None and context.source != source:
            log.warning(
                "JSExtractor received both `source=` and `context=` with "
                "differing source; ignoring source kwarg"
            )
        self._context = context

        super().__init__(
            filepath,
            log,
            exporters,
            index_prefix,
            known_benign=known_benign,
            known_malicious=known_malicious,
        )

    @property
    def binary(self):
        return self._context.raw_bytes

    @property
    def js_source(self):
        return self._context.source

    @property
    def lines(self):
        return self._context.lines

    def _decode_source(self):
        """Back-compat shim — the context already decoded once at construction.

        Kept so any external caller using the historical method name keeps
        working without touching the underlying bytes again.
        """
        return self._context.source

    def _parse_ast(self):
        """Return the shared pyjsparser AST (or None if unavailable)."""
        return self._context.ast

    def _calculate_text_entropy(self, text):
        """Shannon text entropy for `text`.

        When `text` is the context's own source we read the cached value;
        otherwise we compute fresh. JSStringsExtractor calls this on arbitrary
        decoded substrings, so the fresh-compute path must remain available.
        """
        if text is self._context.source:
            return self._context.text_entropy
        return _text_entropy(text)

    def _detect_environment(self):
        """Detect the target JS runtime environment."""
        src = self.js_source
        if not src:
            return "unknown"

        # WScript/WSH indicators
        wscript_patterns = [
            'WScript.', 'WSH.', 'ActiveXObject', 'Scripting.FileSystemObject',
            'WScript.Shell', 'ADODB.Stream',
        ]
        for p in wscript_patterns:
            if p in src:
                return "wscript"

        # Browser-extension APIs — checked before generic browser/worker because
        # `chrome.*` and `browser.runtime` are distinctive of MV2/MV3 extensions
        extension_patterns = [
            'chrome.runtime', 'chrome.tabs', 'chrome.storage',
            'chrome.webRequest', 'browser.runtime', 'browser.tabs',
        ]
        for p in extension_patterns:
            if p in src:
                return "browser_extension"

        # Service / Web Workers — worker-only APIs that don't appear in regular
        # browser pages (a generic browser script would use `window.` or
        # `document.`, never `self.importScripts` or `caches.match`)
        worker_patterns = [
            "self.addEventListener('fetch'", 'self.addEventListener("fetch"',
            'self.importScripts', 'self.skipWaiting',
            'caches.match', 'caches.open',
        ]
        for p in worker_patterns:
            if p in src:
                return "service_worker"

        # Deno runtime
        if 'Deno.' in src:
            return "deno"

        # Node.js indicators
        node_patterns = [
            'require(', 'module.exports', 'process.env', '__dirname',
            '__filename', 'Buffer.', 'child_process',
        ]
        for p in node_patterns:
            if p in src:
                return "node"

        # Browser indicators
        browser_patterns = [
            'document.', 'window.', 'navigator.', 'localStorage',
            'sessionStorage', 'XMLHttpRequest', 'addEventListener',
        ]
        for p in browser_patterns:
            if p in src:
                return "browser"

        return "unknown"

    def _detect_script_type(self):
        """Detect the script type/format."""
        src = self.js_source
        if not src:
            return "unknown"

        stripped = src.lstrip()

        # JScript.Encode payload — must be checked first since the encoded
        # body can't be classified any other way
        if stripped.startswith('#@~^'):
            return "jse"

        # WSF / HTA live in the first few KB of an HTML-ish wrapper
        head_lower = stripped[:4096].lower()

        # Windows Script File — XML wrapper around one or more <script> blocks
        if ('<job' in head_lower or '<package' in head_lower) and '<script' in head_lower:
            return "wsf"

        # HTML Application — distinct from generic embedded_html because HTAs
        # run under mshta.exe with full WSH/ActiveX access
        if '<hta:application' in head_lower or 'application/hta' in head_lower:
            return "hta"

        if stripped.startswith('<!') or stripped.startswith('<html') or '<script' in stripped[:2000]:
            return "embedded_html"

        if 'WScript.' in src or 'WSH.' in src:
            return "wscript"

        if _ESM_PATTERN.search(src):
            return "esm"

        if 'require(' in src or 'module.exports' in src:
            return "node_module"

        return "standalone"