Hang Yin

"""Build a basic-block CFG from smali method bodies and compute graph metrics.

Handles both apktool smali (label-based branches like :cond_0) and
androguard fallback smali (offset-based branches like +005h).

Graph metrics match the Binary Ninja CFG pipeline for cross-platform
consistency: cyclomatic complexity (E - N + 2), loop count (back edges),
max BFS depth, max fan-out. Advanced features (topology hash, MD-index,
WL-MinHash, packed adjacency) reuse the generic cfg_features module.
"""

import logging
import re
from collections import deque
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple

from redb.extractors.decompiler.apk.smali_normalization import (
    categorize_opcode,
    CATEGORY_TO_ACFG_INDEX,
)
from redb.extractors.decompiler.bninja.analysis import cfg_features

logger = logging.getLogger(__name__)


# Instruction classification patterns
_IF_RE = re.compile(r"^if-\w+")
_GOTO_RE = re.compile(r"^goto(?:/\d+)?(?:\s|$)")
_RETURN_RE = re.compile(r"^return")
_THROW_RE = re.compile(r"^throw(?:\s|$)")
_SWITCH_RE = re.compile(r"^(?:packed|sparse)-switch\s")

# Label reference in apktool format: :cond_0, :goto_1, etc.
_LABEL_TARGET_RE = re.compile(r":[\w]+")

# Offset reference in androguard format: +005h, -003h
_OFFSET_TARGET_RE = re.compile(r"[+-]\w+h\b")

# Directives and labels
_SKIP_RE = re.compile(r"^\s*(?:\.|#|$)")
_LABEL_DEF_RE = re.compile(r"^\s*:([\w]+)")


@dataclass
class SmaliCFGMetrics:
    """CFG-derived metrics for a smali method."""
    block_count: int = 0
    edge_count: int = 0
    cyclomatic_complexity: int = 1
    loop_count: int = 0
    max_depth: int = 0
    max_fan_out: int = 0
    # Obfuscation scores (parity with code_binja_decompiled_functions_content)
    flattened_score: float = 0.0
    mba_score: float = 0.0
    # Per-block ACFG feature vectors (Gemini-style, same format as BNinja).
    # Each entry: [instr_count, arithmetic, logic, transfer, call,
    #              comparison, memory, successor_count]
    # Empty list if block features were not computed.
    block_features: List[List[int]] = field(default_factory=list)
    # Advanced CFG features (Phase 5 — parity with code_binja_cfg_functions)
    cfg_topology_hash: bytes = field(default_factory=lambda: b'\x00' * 16)
    md_index_topdown: int = 0
    md_index_bottomup: int = 0
    cfg_feature_tlsh: Optional[str] = None
    wl_minhash: List[int] = field(default_factory=lambda: [255] * 128)
    cfg_adjacency: List[int] = field(default_factory=list)


def compute_cfg_metrics(smali_body: str) -> SmaliCFGMetrics:
    """Compute CFG metrics from a smali method body.

    Works with both apktool label-based smali and androguard offset-based
    smali. Falls back to instruction-counting heuristic if CFG construction
    fails.
    """
    if not smali_body or not smali_body.strip():
        return SmaliCFGMetrics()

    lines = smali_body.split("\n")

    # Determine format: apktool (has labels) vs androguard (no labels)
    has_labels = any(_LABEL_DEF_RE.match(line) for line in lines)

    if has_labels:
        return _build_cfg_with_labels(lines)
    else:
        return _build_cfg_from_instructions(lines)


def _parse_instructions(lines: List[str]) -> List[Tuple[int, str]]:
    """Extract instruction lines, skipping directives, labels, blanks, comments.

    Returns list of (original_line_index, stripped_instruction).
    """
    instructions = []
    for i, line in enumerate(lines):
        stripped = line.strip()
        if not stripped or stripped.startswith(".") or stripped.startswith("#"):
            continue
        if stripped.startswith(":"):
            continue
        instructions.append((i, stripped))
    return instructions


def _build_cfg_with_labels(lines: List[str]) -> SmaliCFGMetrics:
    """Build CFG using apktool label-based format.

    Labels (e.g., :cond_0, :goto_1) define branch targets.
    Branch instructions reference labels directly.
    """
    # First pass: collect label positions and instructions
    # We track everything by instruction index (position in instruction list)
    labels: Dict[str, int] = {}  # label_name -> instruction_index
    instructions: List[str] = []
    # Map: line_index -> instruction_index (for label resolution)
    line_to_instr: Dict[int, int] = {}

    instr_idx = 0
    for i, line in enumerate(lines):
        stripped = line.strip()
        if not stripped or stripped.startswith(".") or stripped.startswith("#"):
            continue
        m = _LABEL_DEF_RE.match(stripped)
        if m:
            label_name = ":" + m.group(1)
            labels[label_name] = instr_idx  # next instruction after this label
            continue
        line_to_instr[i] = instr_idx
        instructions.append(stripped)
        instr_idx += 1

    n_instr = len(instructions)
    if n_instr == 0:
        return SmaliCFGMetrics()

    # Identify basic block start points
    block_starts = {0}

    for idx, instr in enumerate(instructions):
        next_idx = idx + 1

        if _IF_RE.match(instr):
            # Conditional branch: fall-through + branch target
            if next_idx < n_instr:
                block_starts.add(next_idx)
            target_label = _extract_label_target(instr)
            if target_label and target_label in labels:
                block_starts.add(labels[target_label])

        elif _GOTO_RE.match(instr):
            # Unconditional jump
            if next_idx < n_instr:
                block_starts.add(next_idx)
            target_label = _extract_label_target(instr)
            if target_label and target_label in labels:
                block_starts.add(labels[target_label])

        elif _RETURN_RE.match(instr) or _THROW_RE.match(instr):
            if next_idx < n_instr:
                block_starts.add(next_idx)

        elif _SWITCH_RE.match(instr):
            if next_idx < n_instr:
                block_starts.add(next_idx)

    # Also add all label targets as block starts
    for label, target_idx in labels.items():
        if target_idx < n_instr:
            block_starts.add(target_idx)

    # Build blocks: sorted list of start indices
    sorted_starts = sorted(block_starts)
    n_blocks = len(sorted_starts)

    # Map instruction index -> block index
    instr_to_block = {}
    for block_idx, start in enumerate(sorted_starts):
        end = sorted_starts[block_idx + 1] if block_idx + 1 < n_blocks else n_instr
        for i in range(start, end):
            instr_to_block[i] = block_idx

    # Collect per-block instruction lists for ACFG feature extraction
    block_instructions: List[List[str]] = []
    for block_idx in range(n_blocks):
        start = sorted_starts[block_idx]
        end = sorted_starts[block_idx + 1] if block_idx + 1 < n_blocks else n_instr
        block_instructions.append(instructions[start:end])

    # Build adjacency lists
    successors: List[List[int]] = [[] for _ in range(n_blocks)]

    for block_idx in range(n_blocks):
        start = sorted_starts[block_idx]
        end = sorted_starts[block_idx + 1] if block_idx + 1 < n_blocks else n_instr
        last_instr_idx = end - 1
        last_instr = instructions[last_instr_idx]

        if _IF_RE.match(last_instr):
            # Fall-through
            if block_idx + 1 < n_blocks:
                _add_edge(successors, block_idx, block_idx + 1)
            # Branch target
            target_label = _extract_label_target(last_instr)
            if target_label and target_label in labels:
                target_block = instr_to_block.get(labels[target_label])
                if target_block is not None:
                    _add_edge(successors, block_idx, target_block)

        elif _GOTO_RE.match(last_instr):
            # Only branch target, no fall-through
            target_label = _extract_label_target(last_instr)
            if target_label and target_label in labels:
                target_block = instr_to_block.get(labels[target_label])
                if target_block is not None:
                    _add_edge(successors, block_idx, target_block)

        elif _RETURN_RE.match(last_instr) or _THROW_RE.match(last_instr):
            # No successors
            pass

        elif _SWITCH_RE.match(last_instr):
            # Fall-through (default case)
            if block_idx + 1 < n_blocks:
                _add_edge(successors, block_idx, block_idx + 1)
            # Switch targets are defined in switch payload (.packed-switch/.sparse-switch)
            # which we can't easily parse from the body alone. The targets are labels
            # referenced in the switch data section. We handle them via label targets.
            _add_switch_targets(
                lines, last_instr, labels, instr_to_block,
                successors, block_idx
            )

        else:
            # Normal instruction at end of block — fall through
            if block_idx + 1 < n_blocks:
                _add_edge(successors, block_idx, block_idx + 1)

    return _compute_metrics_from_cfg(successors, n_blocks, block_instructions)


def _build_cfg_from_instructions(lines: List[str]) -> SmaliCFGMetrics:
    """Build CFG from androguard offset-based format.

    Without labels, we use instruction counting to build a basic CFG.
    Branch targets are hex offsets (e.g., +005h) which we resolve by
    tracking instruction positions.
    """
    instructions = _parse_instructions(lines)
    n_instr = len(instructions)
    if n_instr == 0:
        return SmaliCFGMetrics()

    # Identify basic block starts
    block_starts = {0}

    for idx, (_, instr) in enumerate(instructions):
        next_idx = idx + 1

        if _IF_RE.match(instr):
            if next_idx < n_instr:
                block_starts.add(next_idx)
            # Try to resolve offset target to instruction index
            target = _resolve_offset_target(instr, idx, n_instr)
            if target is not None:
                block_starts.add(target)

        elif _GOTO_RE.match(instr):
            if next_idx < n_instr:
                block_starts.add(next_idx)
            target = _resolve_offset_target(instr, idx, n_instr)
            if target is not None:
                block_starts.add(target)

        elif _RETURN_RE.match(instr) or _THROW_RE.match(instr):
            if next_idx < n_instr:
                block_starts.add(next_idx)

    sorted_starts = sorted(block_starts)
    n_blocks = len(sorted_starts)

    # Map instruction index -> block index
    instr_to_block = {}
    for block_idx, start in enumerate(sorted_starts):
        end = sorted_starts[block_idx + 1] if block_idx + 1 < n_blocks else n_instr
        for i in range(start, end):
            instr_to_block[i] = block_idx

    # Collect per-block instruction lists for ACFG features
    block_instructions: List[List[str]] = []
    for block_idx in range(n_blocks):
        start = sorted_starts[block_idx]
        end = sorted_starts[block_idx + 1] if block_idx + 1 < n_blocks else n_instr
        block_instructions.append(
            [instructions[i][1] for i in range(start, end)]
        )

    # Build adjacency
    successors: List[List[int]] = [[] for _ in range(n_blocks)]

    for block_idx in range(n_blocks):
        start = sorted_starts[block_idx]
        end = sorted_starts[block_idx + 1] if block_idx + 1 < n_blocks else n_instr
        last_idx = end - 1
        _, last_instr = instructions[last_idx]

        if _IF_RE.match(last_instr):
            if block_idx + 1 < n_blocks:
                _add_edge(successors, block_idx, block_idx + 1)
            target = _resolve_offset_target(last_instr, last_idx, n_instr)
            if target is not None:
                target_block = instr_to_block.get(target)
                if target_block is not None:
                    _add_edge(successors, block_idx, target_block)

        elif _GOTO_RE.match(last_instr):
            target = _resolve_offset_target(last_instr, last_idx, n_instr)
            if target is not None:
                target_block = instr_to_block.get(target)
                if target_block is not None:
                    _add_edge(successors, block_idx, target_block)

        elif _RETURN_RE.match(last_instr) or _THROW_RE.match(last_instr):
            pass

        else:
            if block_idx + 1 < n_blocks:
                _add_edge(successors, block_idx, block_idx + 1)

    return _compute_metrics_from_cfg(successors, n_blocks, block_instructions)


def _resolve_offset_target(instr: str, current_idx: int, n_instr: int) -> Optional[int]:
    """Resolve androguard hex offset to an instruction index.

    Androguard offsets (e.g., +005h, -003h) are in 16-bit code units relative
    to the branch instruction. Since most Dalvik instructions are 1-3 code
    units, we approximate: each instruction ≈ 1 code unit for offset
    resolution. This gives an approximate but usable CFG.

    For better accuracy, we treat the offset as an instruction count
    (which is correct for 1-unit instructions and approximate for larger ones).
    """
    m = _OFFSET_TARGET_RE.search(instr)
    if not m:
        return None

    offset_str = m.group(0)
    try:
        # Parse hex offset: +005h -> 5, -003h -> -3
        offset_val = int(offset_str.rstrip("h"), 16)
    except ValueError:
        return None

    target = current_idx + offset_val
    if 0 <= target < n_instr:
        return target
    return None


def _extract_label_target(instr: str) -> Optional[str]:
    """Extract the label target from a branch/goto instruction.

    E.g., 'if-eqz v0, :cond_0' -> ':cond_0'
          'goto :goto_1' -> ':goto_1'
    """
    m = _LABEL_TARGET_RE.search(instr)
    return m.group(0) if m else None


def _add_edge(successors: List[List[int]], src: int, dst: int):
    """Add edge if not duplicate."""
    if dst not in successors[src]:
        successors[src].append(dst)


def _add_switch_targets(
    lines: List[str],
    switch_instr: str,
    labels: Dict[str, int],
    instr_to_block: Dict[int, int],
    successors: List[List[int]],
    block_idx: int,
):
    """Try to resolve switch case targets.

    Switch payloads in apktool smali are defined as:
      .packed-switch 0x0
        :pswitch_0
        :pswitch_1
      .end packed-switch

    We scan the body for label references in switch payload sections.
    """
    # Find the switch payload target label
    target_label = _extract_label_target(switch_instr)
    if not target_label:
        return

    # Scan for packed-switch/sparse-switch payload sections
    in_switch = False
    for line in lines:
        stripped = line.strip()
        if stripped.startswith(".packed-switch") or stripped.startswith(".sparse-switch"):
            in_switch = True
            continue
        if stripped.startswith(".end packed-switch") or stripped.startswith(".end sparse-switch"):
            in_switch = False
            continue
        if in_switch:
            # Lines in switch payload are label references
            m = _LABEL_TARGET_RE.search(stripped)
            if m:
                case_label = m.group(0)
                if case_label in labels:
                    target_block = instr_to_block.get(labels[case_label])
                    if target_block is not None:
                        _add_edge(successors, block_idx, target_block)


def _build_block_features(
    block_instructions: List[List[str]],
    successors: List[List[int]],
    n: int,
) -> List[List[int]]:
    """Build Gemini-style ACFG feature vectors per block from smali instructions.

    Same 8-element format as Binary Ninja's build_block_features:
    [instr_count, arithmetic, logic, transfer, call, comparison, memory, successor_count]

    Uses semantic opcode categorization (analogous to LLIL operation categories)
    to map each Dalvik instruction to one of 7 category bins.
    """
    features = []
    for i in range(n):
        cats = [0, 0, 0, 0, 0, 0, 0]  # 7 categories
        instrs = block_instructions[i] if i < len(block_instructions) else []
        for instr in instrs:
            opcode = instr.split(None, 1)[0] if instr else ""
            category = categorize_opcode(opcode)
            acfg_idx = CATEGORY_TO_ACFG_INDEX.get(category, 6)
            cats[acfg_idx] += 1

        features.append([
            min(len(instrs), 65535),
            min(cats[0], 65535),  # arithmetic
            min(cats[1], 65535),  # logic
            min(cats[2], 65535),  # transfer
            min(cats[3], 65535),  # call
            min(cats[4], 65535),  # comparison
            min(cats[5], 65535),  # memory
            min(len(successors[i]), 65535),
        ])
    return features


def _compute_metrics_from_cfg(
    successors: List[List[int]],
    n: int,
    block_instructions: Optional[List[List[str]]] = None,
) -> SmaliCFGMetrics:
    """Compute all graph metrics from the adjacency list."""
    if n == 0:
        return SmaliCFGMetrics()

    edge_count = sum(len(s) for s in successors)

    # Cyclomatic complexity: E - N + 2
    cc = edge_count - n + 2
    if cc < 1:
        cc = 1

    # Loop count: back edges via iterative DFS
    loop_count = _count_back_edges(successors, n)

    # Max BFS depth from entry
    max_depth = _bfs_max_depth(successors, n)

    # Max fan-out
    max_fan_out = max(len(s) for s in successors) if successors else 0

    # Per-block ACFG features
    bb_features = []
    if block_instructions is not None:
        bb_features = _build_block_features(block_instructions, successors, n)

    # Obfuscation scores
    flattened = _compute_flattened_score(successors, n)
    mba = (
        _compute_mba_score(block_instructions, n)
        if block_instructions is not None
        else 0.0
    )

    # Advanced CFG features — reuse generic cfg_features module
    # Build predecessors from successors
    predecessors = [[] for _ in range(n)]
    for src, targets in enumerate(successors):
        for tgt in targets:
            predecessors[tgt].append(src)

    try:
        bfs = cfg_features.bfs_order(successors, n)
        topology_hash = cfg_features.compute_topology_hash(successors, bfs, n)
        md_topdown = cfg_features.compute_md_index_topdown(
            successors, predecessors, bfs
        )
        md_bottomup = cfg_features.compute_md_index_bottomup(
            successors, predecessors, n
        )
        cfg_tlsh = (
            cfg_features.compute_cfg_feature_tlsh(bb_features, bfs)
            if bb_features
            else None
        )
        wl_minhash = (
            cfg_features.compute_wl_minhash(
                successors, predecessors, bb_features, n
            )
            if bb_features
            else [255] * 128
        )
        adjacency = cfg_features.pack_adjacency(successors)
    except Exception as e:
        logger.debug("Advanced CFG features failed: %s", e)
        topology_hash = b'\x00' * 16
        md_topdown = 0
        md_bottomup = 0
        cfg_tlsh = None
        wl_minhash = [255] * 128
        adjacency = []

    return SmaliCFGMetrics(
        block_count=n,
        edge_count=edge_count,
        cyclomatic_complexity=cc,
        loop_count=loop_count,
        max_depth=max_depth,
        max_fan_out=max_fan_out,
        flattened_score=flattened,
        mba_score=mba,
        block_features=bb_features,
        cfg_topology_hash=topology_hash,
        md_index_topdown=md_topdown,
        md_index_bottomup=md_bottomup,
        cfg_feature_tlsh=cfg_tlsh,
        wl_minhash=wl_minhash,
        cfg_adjacency=adjacency,
    )


def _compute_dominators(successors: List[List[int]], n: int) -> List[int]:
    """Compute immediate dominators using iterative dataflow algorithm.

    Returns idom[i] = immediate dominator of block i.  idom[0] = -1 (entry).
    """
    if n == 0:
        return []

    # Build predecessors
    preds: List[List[int]] = [[] for _ in range(n)]
    for src, targets in enumerate(successors):
        for tgt in targets:
            preds[tgt].append(src)

    # Initialize: dom[0] = {0}, dom[i] = all blocks
    all_blocks = set(range(n))
    dom = [all_blocks.copy() for _ in range(n)]
    dom[0] = {0}

    changed = True
    while changed:
        changed = False
        for i in range(1, n):
            if not preds[i]:
                new_dom = {i}
            else:
                new_dom = all_blocks.copy()
                for p in preds[i]:
                    new_dom &= dom[p]
                new_dom.add(i)
            if new_dom != dom[i]:
                dom[i] = new_dom
                changed = True

    # Extract immediate dominators from dominator sets
    idom = [-1] * n
    for i in range(1, n):
        # idom[i] = the dominator of i (other than i itself) that is
        # dominated by all other dominators of i
        doms_of_i = dom[i] - {i}
        if not doms_of_i:
            continue
        for candidate in doms_of_i:
            # candidate is idom if it is dominated by all other dominators
            if all(candidate in dom[other] for other in doms_of_i):
                # candidate dominates no other dominator besides itself
                # (i.e., it's the closest dominator)
                if all(
                    other == candidate or candidate not in dom[other]
                    for other in doms_of_i
                ):
                    pass  # not the closest
                else:
                    continue
            else:
                continue
        # Simpler approach: idom is the element in doms_of_i with the
        # largest dominator set (closest to i in the dominator tree)
        idom[i] = max(doms_of_i, key=lambda d: len(dom[d]))

    return idom


def _compute_flattened_score(
    successors: List[List[int]], n: int
) -> float:
    """Detect control flow flattening — same heuristic as Binary Ninja's
    ObfuscationScores.flattened_score (Tim Blazytko).

    Walks over all basic blocks, finds those with back edges (loop headers),
    and computes the ratio of blocks dominated by them to total blocks.
    """
    if n <= 1:
        return 0.0

    idom = _compute_dominators(successors, n)

    # Build dominator tree children from idom
    dom_children: List[List[int]] = [[] for _ in range(n)]
    for i in range(1, n):
        if idom[i] >= 0:
            dom_children[idom[i]].append(i)

    max_ratio = 0.0

    for block in range(n):
        # Get all blocks dominated by this block (reachable in dominator tree)
        dominated = set()
        worklist = [block]
        while worklist:
            b = worklist.pop()
            dominated.add(b)
            worklist.extend(dom_children[b])

        # Check for a back edge: any predecessor of block is in dominated set
        has_back_edge = False
        for src, targets in enumerate(successors):
            if block in targets and src in dominated:
                has_back_edge = True
                break

        if not has_back_edge:
            continue

        ratio = len(dominated) / n
        if ratio > max_ratio:
            max_ratio = ratio

    return max_ratio


def _compute_mba_score(block_instructions: List[List[str]], n: int) -> float:
    """Compute mixed boolean-arithmetic score for a smali method.

    Same concept as Binary Ninja's ObfuscationScores.MBA_score: ratio of
    instructions that mix arithmetic and logic operations.

    At the smali level, we check each instruction's opcode:
    - Arithmetic: add, sub, mul, div, rem, neg
    - Logic: and, or, xor, shl, shr, ushr, not

    Since Dalvik instructions are single operations (unlike x86 complex
    instructions or HLIL expression trees), we check per-instruction whether
    the method mixes both categories. The score is the fraction of
    instructions belonging to the minority category when both are present.
    """
    ARITHMETIC_OPS = {"add", "sub", "mul", "div", "rem", "neg"}
    LOGIC_OPS = {"and", "or", "xor", "shl", "shr", "ushr", "not"}

    arithmetic_count = 0
    logic_count = 0
    total_instructions = 0

    for block in block_instructions[:n]:
        for instr in block:
            opcode = instr.split(None, 1)[0] if instr else ""
            # Strip type suffix: add-int/2addr -> add
            base = opcode.split("-")[0] if "-" in opcode else opcode
            total_instructions += 1
            if base in ARITHMETIC_OPS:
                arithmetic_count += 1
            elif base in LOGIC_OPS:
                logic_count += 1

    if total_instructions == 0:
        return 0.0

    # MBA is present when both arithmetic and logic operations co-exist.
    # Score = min(arith, logic) / total — measures how much mixing occurs.
    if arithmetic_count == 0 or logic_count == 0:
        return 0.0

    return min(arithmetic_count, logic_count) / total_instructions


def _count_back_edges(successors: List[List[int]], n: int) -> int:
    """Count natural loops via iterative DFS back-edge detection.

    Same algorithm as bninja/analysis/cfg_features.py:count_back_edges.
    """
    if n == 0:
        return 0

    WHITE, GRAY, BLACK = 0, 1, 2
    color = [WHITE] * n
    back_edges = 0

    stack = [(0, iter(successors[0]))]
    color[0] = GRAY

    while stack:
        u, children = stack[-1]
        try:
            v = next(children)
            if color[v] == GRAY:
                back_edges += 1
            elif color[v] == WHITE:
                color[v] = GRAY
                stack.append((v, iter(successors[v])))
        except StopIteration:
            color[u] = BLACK
            stack.pop()

    return back_edges


def _bfs_max_depth(successors: List[List[int]], n: int) -> int:
    """Maximum BFS depth from entry block.

    Same algorithm as bninja/analysis/cfg_features.py:bfs_max_depth.
    """
    if n == 0:
        return 0

    depth = {0: 0}
    max_d = 0
    queue = deque([0])

    while queue:
        node = queue.popleft()
        for s in successors[node]:
            if s not in depth:
                depth[s] = depth[node] + 1
                if depth[s] > max_d:
                    max_d = depth[s]
                queue.append(s)

    return max_d