Module refinery.lib.scripts.js.deobfuscation.cff

Control-flow flattening recovery transforms.

"""
Control-flow flattening recovery transforms.
"""
from __future__ import annotations

from refinery.lib.scripts.js.deobfuscation.cff.sequential import JsControlFlowUnflattening
from refinery.lib.scripts.js.deobfuscation.cff.statemachine import JsGeneratorCFFUnflattening

__all__ = [
    'JsControlFlowUnflattening',
    'JsGeneratorCFFUnflattening',
]

Sub-modules

refinery.lib.scripts.js.deobfuscation.cff.sequential

Recover sequential code from control-flow-flattened dispatchers …

refinery.lib.scripts.js.deobfuscation.cff.statemachine

Recover original code from generator-based state-machine CFF dispatchers …

Classes

class JsControlFlowUnflattening

Detect and recover CFF dispatchers in function bodies and script-level code.

A dispatcher whose last dispatched case runs on with a bare continue exhausts its order array, matches no case, and completes undefined; a last case that instead leaves the switch reaches the loop's own trailing break and completes with that case's value. Under preserve_script_return the recovered straight-line code holds the script's completion at undefined only in the former case — decided per dispatcher by whether the last dispatched case loops back — where the dispatcher stood at the completion position and the recovered tail would otherwise answer with a value.

Expand source code Browse git

class JsControlFlowUnflattening(BodyProcessingTransformer):
    """
    Detect and recover CFF dispatchers in function bodies and script-level code.

    A dispatcher whose last dispatched case runs on with a bare `continue` exhausts its order array,
    matches no case, and completes `undefined`; a last case that instead leaves the switch reaches the
    loop's own trailing `break` and completes with that case's value. Under `preserve_script_return`
    the recovered straight-line code holds the script's completion at `undefined` only in the former
    case — decided per dispatcher by whether the last dispatched case loops back — where the dispatcher
    stood at the completion position and the recovered tail would otherwise answer with a value.
    """

    def __init__(self):
        super().__init__()
        self._root: JsScript | None = None

    def visit_JsScript(self, node: JsScript):
        self._root = node
        return super().visit_JsScript(node)

    def _process_body(self, parent: Node, body: list[Statement]) -> None:
        i = 0
        while i < len(body):
            stmt = body[i]
            if not isinstance(stmt, JsWhileStatement):
                i += 1
                continue
            match = _match_dispatcher(stmt)
            if match is None:
                i += 1
                continue
            order_info = _find_order_sequence(body, i, match.order_var, match.counter_var)
            if order_info is None:
                i += 1
                continue
            if not all(label in match.case_map for label in order_info.order_sequence):
                i += 1
                continue
            assert self._root is not None
            model = model_cache(self, self._root).model
            if not _consumed_only_by_dispatcher(model, match, order_info):
                i += 1
                continue
            recovered: list[Statement] = []
            for j in range(order_info.first_init_idx, i):
                decl_stmt = body[j]
                if isinstance(decl_stmt, JsVariableDeclaration):
                    remaining = [
                        d for d in decl_stmt.declarations
                        if all(d is not s for s in order_info.strip_declarators)
                    ]
                    if not remaining:
                        continue
                    if len(remaining) != len(decl_stmt.declarations):
                        set_child_list(decl_stmt, 'declarations', remaining)
                recovered.append(decl_stmt)
            for label in order_info.order_sequence:
                recovered.extend(match.case_map[label])
            if preserves_script_return(self.options):
                last_label = order_info.order_sequence[-1] if order_info.order_sequence else None
                recovered = preserve_script_end_value(
                    recovered,
                    returns_undefined=last_label is None or match.loops_back[last_label],
                    reaches_completion=reaches_script_completion(stmt, self._root),
                )
            replacement = body[:order_info.first_init_idx] + recovered + body[i + 1:]
            self._replace_body(parent, replacement)
            i = order_info.first_init_idx + len(recovered)

Ancestors

• BodyProcessingTransformer
• Transformer
• Visitor

Methods

def visit_JsScript(self, node)

Expand source code Browse git

def visit_JsScript(self, node: JsScript):
    self._root = node
    return super().visit_JsScript(node)

Inherited members

• BodyProcessingTransformer:
  • self_converging

class JsGeneratorCFFUnflattening

Recover original code from generator-based state-machine CFF dispatchers. Handles the pattern where a function body is replaced with a generator function containing a while/switch state machine driven by multiple state variables.

The pass holds the two models it reads, the semantic model and the intrinsic-write scan, for an invocation (PinnedTransformer). A recovery replaces the scaffolding of a machine with code made from copies of its states, and leaves the answers about the next machine no more permissive than rebuilt ones:

• Whether the scaffolding is private (_scaffolding_is_private) is asked of the bindings the recovery of that machine removes, which are its own: no other recovery declares or reaches them.
• Whether the recovered code declares only its own names (_declares_only_its_own_names) is asked of the function around the machine outside its scaffolding. Before deciding a machine whose body, or whose function outside the scaffolding, holds a body the invocation edited since the last renewal, the pass renews.
• Whether a plain object inherits a routing key (_provably_absent) is read off the scan, which a recovery changes only where it turns a store of a value the scan follows only through bindings into a binding store, where it removes the with statement between a name and a binding whose values denote a root the scan watches, or where its code declares or assigns a name the scan watches (_recovery_reveals). After such a recovery, the pass renews.

Expand source code Browse git

class JsGeneratorCFFUnflattening(PinnedTransformer, BodyProcessingTransformer):
    """
    Recover original code from generator-based state-machine CFF dispatchers. Handles the pattern
    where a function body is replaced with a generator function containing a while/switch state
    machine driven by multiple state variables.

    The pass holds the two models it reads, the semantic model and the intrinsic-write scan, for an
    invocation (`refinery.lib.scripts.js.deobfuscation.helpers.PinnedTransformer`). A recovery
    replaces the scaffolding of a machine with code made from copies of its states, and leaves the
    answers about the next machine no more permissive than rebuilt ones:

    - Whether the scaffolding is private (`_scaffolding_is_private`) is asked of the bindings the
      recovery of that machine removes, which are its own: no other recovery declares or reaches
      them.
    - Whether the recovered code declares only its own names (`_declares_only_its_own_names`) is
      asked of the function around the machine outside its scaffolding. Before deciding a machine
      whose body, or whose function outside the scaffolding, holds a body the invocation edited
      since the last renewal, the pass renews.
    - Whether a plain object inherits a routing key (`_provably_absent`) is read off the scan, which
      a recovery changes only where it turns a store of a value the scan follows only through
      bindings into a binding store, where it removes the `with` statement between a name and a
      binding whose values denote a root the scan watches, or where its code declares or assigns a
      name the scan watches (`_recovery_reveals`). After such a recovery, the pass renews.
    """

    reads = ('_model', '_intrinsic_writes')

    def __init__(self):
        super().__init__()
        self._edited: dict[Node, list[Node]] = {}

    def visit_JsScript(self, node: JsScript):
        with self.holding_models(node):
            return super().visit_JsScript(node)

    def _forget_edits(self) -> None:
        self._edited = {}

    def _note_edit(self, body: Node) -> None:
        """
        Record that the invocation edited *body*, under the script and every function that holds it,
        which are the places `_edited_around` asks about.
        """
        cursor: Node | None = body
        while cursor is not None:
            if isinstance(cursor, (JsScript, *FUNCTION_NODES)):
                self._edited.setdefault(cursor, []).append(body)
            cursor = cursor.parent

    def _edited_around(self, parent: Node, removed: list[Statement]) -> bool:
        """
        Whether a body the invocation edited since the last renewal is *parent*, the body holding a
        machine, or lies in the function holding it outside the scaffolding *removed*.
        """
        function = parent if isinstance(parent, JsScript) else parent.parent
        if function is None:
            return False
        return any(
            body is parent or not any(body.is_descendant_of(statement) for statement in removed)
            for body in self._edited.get(function, ())
        )

    def _provably_absent(self, key: str) -> bool:
        """
        Whether a plain object that does not own *key* reads `undefined` under it, and a store there
        makes a property of data: no prototype the language installs holds the key, and the program
        writes no prototype of a plain object
        (`refinery.lib.scripts.js.analysis.intrinsics.IntrinsicWrites.chain_roots_unwritten`), or
        at least not this key
        (`refinery.lib.scripts.js.analysis.intrinsics.IntrinsicWrites.key_written`), as a file
        patching another key of `Object`, such as a polyfill, does not.
        """
        scan = self.held_models().intrinsic_writes
        return property_absent_from_written_chain(None, dict, key) and (
            scan.chain_roots_unwritten(dict) or not scan.key_written('Object', key)
        )

    def _process_body(self, parent: Node, body: list[Statement]) -> None:
        if not _is_a_function_body(parent):
            return
        i = 0
        while i < len(body):
            match = _match_generator_cff(body, i)
            if match is None:
                i += 1
                continue
            models = self.held_models()
            removed = body[match.gen_decl_index:match.scaffolding_end + 1]
            if self._edited_around(parent, removed):
                self.renew_models()
            model = models.model
            if not _scaffolding_is_private(model, body, match):
                i += 1
                continue
            if (
                match.routed
                and _spells_an_inherited_name_bare(match.switch_stmt, self._provably_absent)
            ):
                i += 1
                continue
            machine = _extract_state_blocks(match, self._provably_absent)
            if machine is None:
                i += 1
                continue
            slots = _initial_slots(
                machine.routing,
                match.scope_param_name,
                match.scope_default_inits,
                set(),
                {},
                self._provably_absent,
            )
            result = _execute_machine(machine, match, slots, main=True)
            if result is None:
                i += 1
                continue
            recovered, settled = result
            main = _Activation(
                None,
                match.scope_default_inits,
                props=match.scope_prop_names,
                routing=settled,
            )
            wrappers = _resolve_shared_wrappers(
                recovered, machine, match, main, self._provably_absent
            )
            if wrappers is None or not _bind_main_arguments(recovered, match):
                i += 1
                continue
            for activation in wrappers:
                if activation.function is not None:
                    _declare_wrapper_scope_vars(activation)
            recovered = _declare_recovered_scope_vars(recovered, main)
            if (
                _leaves_the_generator(recovered, match)
                or not _declares_only_its_own_names(model, body, match, recovered)
            ):
                i += 1
                continue
            settled = _settle_returns(
                recovered,
                match,
                body[match.scaffolding_end + 1:],
                bool(wrappers),
            )
            if settled is None:
                i += 1
                continue
            recovered = settled
            for s in recovered:
                s.parent = parent
            start = match.gen_decl_index
            end = match.scaffolding_end
            reveals = _recovery_reveals(model, models.intrinsic_writes, removed, recovered, match)
            replacement = body[:start] + recovered + body[end + 1:]
            self._replace_body(parent, replacement)
            self._note_edit(parent)
            if reveals:
                self.renew_models()
            i = start + len(recovered)

Ancestors

• PinnedTransformer
• BodyProcessingTransformer
• Transformer
• Visitor

Methods

def visit_JsScript(self, node)

Expand source code Browse git

def visit_JsScript(self, node: JsScript):
    with self.holding_models(node):
        return super().visit_JsScript(node)

Inherited members

• PinnedTransformer:
  • held_models
  • holding_models
  • reads
  • renew_models
  • self_converging