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.

Expand source code Browse git

class JsGeneratorCFFUnflattening(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.
    """

    def _process_body(self, parent: Node, body: list[Statement]) -> None:
        is_script = isinstance(parent, JsScript)
        i = 0
        while i < len(body):
            match = _match_generator_cff(body, i)
            if match is None:
                i += 1
                continue
            machine = _extract_state_blocks(match)
            if machine is None:
                i += 1
                continue
            result = _execute_machine(machine, match)
            if result is None:
                i += 1
                continue
            recovered, outer_state = result
            if match.arg_var_name is not None:
                recovered = _resolve_shared_wrappers(recovered, machine, match, outer_state)
            recovered = _declare_recovered_scope_vars(recovered, match)
            if match.scope_default_props:
                recovered = _emit_scope_namespace_declarations(match, recovered) + recovered
            if match.arg_params:
                recovered = _emit_arg_param_declarations(match) + recovered
            if is_script:
                sanitized = sanitize_inlined_body(recovered)
                if sanitized is None:
                    i += 1
                    continue
                recovered = sanitized
            for s in recovered:
                s.parent = parent
            start = match.gen_decl_index
            end = match.scaffolding_end
            replacement = body[:start] + recovered + body[end + 1:]
            self._replace_body(parent, replacement)
            i = start + len(recovered)

Ancestors

• BodyProcessingTransformer
• Transformer
• Visitor

Inherited members

• BodyProcessingTransformer:
  • self_converging