Mamtaj Akter

"""Per-method content extraction, hashing, and similarity computation.

Handles SHA-256 content hashing, ssdeep/TLSH fuzzy hashing, MinHash
computation, and obfuscation indicator detection for APK methods.
"""

import hashlib
import re
from typing import Dict, List, Optional

from redb.extractors.decompiler.apk.smali_normalization import (
    categorize_opcode,
    normalize_method_body,
)
from redb.extractors.decompiler.apk.smali_parser import SmaliParser


# ---------------------------------------------------------------------------
# Smali Prime Product — semantic primes matching Binary Ninja's LLIL primes
# ---------------------------------------------------------------------------
# Each Dalvik semantic category maps to the same prime its LLIL counterpart
# uses in cfg_features.py. This makes prime products semantically comparable
# for APK-vs-APK similarity (not numerically comparable to Binja values).

SMALI_OP_PRIMES = {
    "ALU": 37,       # ADD/SUB → same prime as LLIL_ADD
    "CONV": 131,     # Type conversions → same as LLIL_SX
    "CMP": 103,      # Comparisons → same as LLIL_CMP_E
    "MOV": 2,        # Register moves → same as LLIL_SET_REG
    "CONST": 2,      # Constants → SET_REG equivalent
    "LOAD": 5,       # Field/array reads → same as LLIL_LOAD
    "STORE": 7,      # Field/array writes → same as LLIL_STORE
    "CALL": 17,      # invoke-* → same as LLIL_CALL
    "BRANCH": 29,    # if-* → same as LLIL_IF
    "JMP": 31,       # goto → same as LLIL_GOTO
    "SWITCH": 151,   # switch → same as LLIL_JUMP_TO
    "RET": 23,       # return → same as LLIL_RET
    "ALLOC": 5,      # new-instance/new-array → LOAD-adjacent (heap access)
    "TYPE": 1,       # check-cast/instance-of → identity (metadata)
    "ARR": 5,        # array-length/fill-array → LOAD-adjacent
    "EXC": 23,       # throw → RET-adjacent (control transfer out)
    "SYNC": 1,       # monitor → identity (no LLIL equivalent)
    "OTHER": 1,      # Unknown → identity
}


def compute_prime_product_smali(smali_body: str) -> int:
    """Multiplicative hash of normalized Dalvik opcodes. Mod 2^64.

    Same algorithm as cfg_features.compute_prime_product but using
    Dalvik semantic categories instead of LLIL operation enums.
    """
    if not smali_body:
        return 0

    product = 1
    for line in smali_body.splitlines():
        stripped = line.strip()
        if not stripped or stripped.startswith((".", ":", "#")):
            continue
        opcode = stripped.split()[0].split("/")[0] if stripped else ""
        category = categorize_opcode(opcode)
        prime = SMALI_OP_PRIMES.get(category, 1)
        product = (product * prime) % (2**64)

    return product


def count_call_instructions(smali_body: str) -> int:
    """Count invoke-* instructions in a smali method body."""
    if not smali_body:
        return 0
    count = 0
    for line in smali_body.splitlines():
        stripped = line.strip()
        if stripped.startswith("invoke-"):
            count += 1
    return count


def compute_sha256(content: str) -> str:
    """Compute SHA-256 hash of normalized content."""
    return hashlib.sha256(content.encode("utf-8")).hexdigest()


def compute_ssdeep(content: str) -> Optional[str]:
    """Compute ssdeep fuzzy hash of content."""
    try:
        import ppdeep
        data = content.encode("utf-8")
        if len(data) < 50:
            return None
        result = ppdeep.hash(data)
        return result if result else None
    except (ImportError, Exception):
        return None


def compute_tlsh(content: str) -> Optional[str]:
    """Compute TLSH fuzzy hash of content."""
    try:
        import tlsh
        data = content.encode("utf-8")
        if len(data) < 50:
            return None
        result = tlsh.hash(data)
        return result if result else None
    except (ImportError, Exception):
        return None


def compute_minhash(
    content: str,
    n: int = 3,
    normalization_level: str = "opcode_api",
) -> Optional[List[int]]:
    """Compute MinHash signature from semantically normalized smali n-grams.

    Applies semantic normalization (analogous to Binary Ninja's LLIL) before
    computing the MinHash. This strips register allocation noise and
    instruction encoding variants while preserving operation semantics and
    API references.

    Uses the same algorithm and parameters as the Binary Ninja MinHasher
    (64 seeds from master seed 0xdeadbeef, 8-bit signature elements, mmh3)
    to ensure cross-platform similarity comparisons are compatible.

    Args:
        content: Raw smali method body.
        n: N-gram size (default 3).
        normalization_level: Normalization level for instructions.
            'opcode_api' (default) preserves API call/field references.
            'category' uses only semantic categories.
            'opcode' uses base opcodes without operands.
    """
    try:
        import mmh3
    except ImportError:
        return None

    HASH_MAX = 0xFFFFFFFF
    SIGNATURE_LENGTH = 64
    SIGNATURE_BITS = 8

    # Semantically normalize instructions (like LLIL for native code)
    lines = normalize_method_body(content, level=normalization_level)

    if len(lines) < n:
        return None

    # Build n-grams (tuples of normalized instruction strings)
    shingles = [tuple(lines[i:i + n]) for i in range(len(lines) - n + 1)]

    if not shingles:
        return None

    # Generate deterministic seeds matching the Binary Ninja pipeline
    import random
    rng = random.Random(0xDEADBEEF)
    seeds = [rng.randint(0, HASH_MAX) for _ in range(SIGNATURE_LENGTH)]

    # For each seed, hash all shingles and take the minimum
    signature = []
    for seed in seeds:
        min_val = HASH_MAX
        for shingle in shingles:
            text = "|".join(str(elem) for elem in shingle)
            h = mmh3.hash(text, seed) & HASH_MAX
            if h < min_val:
                min_val = h
        # Truncate to signature bits
        if SIGNATURE_BITS < 32:
            min_val %= (2 ** SIGNATURE_BITS)
        signature.append(min_val)

    return signature


def detect_obfuscation_indicators(
    method_name: str,
    class_name: str,
    smali_body: str,
    instruction_count: int,
) -> Dict[str, bool]:
    """Compute obfuscation indicators for a method.

    Returns dict with boolean indicators.
    """
    indicators = {}

    # Short method name (typical R8/ProGuard output)
    indicators["short_method_name"] = len(method_name) <= 2

    # Short class name — extract simple name from Dalvik descriptor
    simple_class = class_name
    if "/" in simple_class:
        simple_class = simple_class.rsplit("/", 1)[-1]
    simple_class = simple_class.rstrip(";")
    indicators["short_class_name"] = len(simple_class) <= 2

    # String encryption: const-string followed by decryption-pattern call
    indicators["has_string_encryption"] = _detect_string_encryption(smali_body)

    # Reflection calls
    indicators["has_reflection_calls"] = _detect_reflection_calls(smali_body)

    # Excessive goto count (control flow flattening)
    goto_count = _count_goto_instructions(smali_body)
    threshold = max(5, int(instruction_count * 0.15))
    indicators["excessive_goto_count"] = goto_count > threshold

    return indicators


def _detect_string_encryption(smali_body: str) -> bool:
    """Detect const-string followed by decryption-pattern calls."""
    lines = smali_body.split("\n")
    for i, line in enumerate(lines):
        stripped = line.strip()
        if stripped.startswith("const-string"):
            # Check the next 3 lines for invoke-* to potential decryption
            for j in range(i + 1, min(i + 4, len(lines))):
                next_line = lines[j].strip()
                if next_line.startswith("invoke-"):
                    # Common decryption patterns
                    if any(
                        pat in next_line
                        for pat in [
                            "decrypt",
                            "decode",
                            "Cipher",
                            "DES",
                            "AES",
                            "Base64",
                            "getBytes",
                        ]
                    ):
                        return True
    return False


def _detect_reflection_calls(smali_body: str) -> bool:
    """Detect use of Java reflection APIs."""
    reflection_patterns = [
        "Ljava/lang/reflect/",
        "Ljava/lang/Class;->forName",
        "Ljava/lang/Class;->getMethod",
        "Ljava/lang/Class;->getDeclaredMethod",
        "Ljava/lang/Class;->getField",
        "Ljava/lang/Class;->getDeclaredField",
    ]
    for pattern in reflection_patterns:
        if pattern in smali_body:
            return True
    return False


def _count_goto_instructions(smali_body: str) -> int:
    """Count goto/goto_16/goto_32 instructions."""
    count = 0
    for line in smali_body.split("\n"):
        stripped = line.strip()
        if stripped.startswith(("goto ", "goto/16 ", "goto/32 ")):
            count += 1
        elif stripped in ("goto", "goto/16", "goto/32"):
            count += 1
    return count


def dalvik_to_java_class(descriptor: str) -> str:
    """Convert Dalvik class descriptor to Java dot notation.

    Lcom/example/Foo; -> com.example.Foo
    """
    if descriptor.startswith("L") and descriptor.endswith(";"):
        return descriptor[1:-1].replace("/", ".")
    return descriptor.replace("/", ".")


def dalvik_type_to_java(type_desc: str) -> str:
    """Convert a Dalvik type descriptor to Java type name."""
    type_map = {
        "V": "void",
        "Z": "boolean",
        "B": "byte",
        "S": "short",
        "C": "char",
        "I": "int",
        "J": "long",
        "F": "float",
        "D": "double",
    }

    if not type_desc:
        return "void"

    if type_desc in type_map:
        return type_map[type_desc]

    if type_desc.startswith("["):
        return dalvik_type_to_java(type_desc[1:]) + "[]"

    if type_desc.startswith("L") and type_desc.endswith(";"):
        full = type_desc[1:-1].replace("/", ".")
        # Return simple name
        return full.rsplit(".", 1)[-1] if "." in full else full

    return type_desc


def dalvik_to_java_prototype(
    method_name: str, signature: str, class_name: str = ""
) -> str:
    """Convert Dalvik method signature to Java-style prototype.

    Input: method_name='onCreate', signature='(Landroid/os/Bundle;)V'
    Output: 'void onCreate(Bundle)'
    """
    # Parse return type and param types from signature
    if not signature or not signature.startswith("("):
        return f"void {method_name}()"

    close_paren = signature.find(")")
    if close_paren == -1:
        return f"void {method_name}()"

    params_str = signature[1:close_paren]
    return_type_str = signature[close_paren + 1:]

    return_type = dalvik_type_to_java(return_type_str)
    params = _parse_dalvik_params(params_str)
    param_java = ", ".join(dalvik_type_to_java(p) for p in params)

    return f"{return_type} {method_name}({param_java})"


def _parse_dalvik_params(params_str: str) -> List[str]:
    """Parse Dalvik parameter descriptor string into individual types."""
    params = []
    i = 0
    while i < len(params_str):
        ch = params_str[i]
        if ch in "VZBSCIJFD":
            params.append(ch)
            i += 1
        elif ch == "[":
            # Array — find the base type
            array_prefix = "["
            i += 1
            while i < len(params_str) and params_str[i] == "[":
                array_prefix += "["
                i += 1
            if i < len(params_str):
                if params_str[i] == "L":
                    end = params_str.find(";", i)
                    if end != -1:
                        params.append(array_prefix + params_str[i : end + 1])
                        i = end + 1
                    else:
                        break
                else:
                    params.append(array_prefix + params_str[i])
                    i += 1
        elif ch == "L":
            end = params_str.find(";", i)
            if end != -1:
                params.append(params_str[i : end + 1])
                i = end + 1
            else:
                break
        else:
            i += 1
    return params