Ralf Bartenschlager

"""APK Code Analyzer — orchestrates androguard + JADX + apktool analysis.

This is the APK equivalent of BinaryNinjaDecompiler in
redb/extractors/decompiler/bninja/decompiler.py.
"""

import math
import os
import tempfile
from collections import Counter
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Any, Dict, List, Optional, Set

from redb.extractors.decompiler.apk.apktool_wrapper import ApktoolDisassembler
from redb.extractors.decompiler.apk.jadx_wrapper import JADXDecompiler
from redb.extractors.decompiler.apk.library_filter import LibraryFilter
from redb.extractors.decompiler.apk.method_extractor import (
    compute_minhash,
    compute_prime_product_smali,
    compute_sha256,
    compute_ssdeep,
    compute_tlsh,
    count_call_instructions,
    dalvik_to_java_class,
    dalvik_to_java_prototype,
    detect_obfuscation_indicators,
)
from redb.extractors.decompiler.apk.smali_normalization import normalize_method_body
from redb.extractors.decompiler.apk.smali_cfg import compute_cfg_metrics
from redb.extractors.decompiler.apk.smali_parser import SmaliParser


class APKCodeAnalyzer:
    """Orchestrates APK code analysis combining three tools.

    Combines androguard (call graphs, xrefs, method enumeration),
    JADX (Java decompilation), and apktool (smali disassembly).
    """

    def __init__(
        self,
        filepath: str,
        timeout: int = 600,
        log=None,
        decompile_modules: Set[str] = None,
    ):
        self.filepath = filepath
        self.timeout = timeout
        self.log = log
        self.decompile_modules = decompile_modules or {"all"}
        self.min_instructions = int(
            os.getenv("APK_MIN_METHOD_INSTRUCTIONS", "5")
        )
        self.library_filter = LibraryFilter()
        self.jadx = JADXDecompiler(log=log)
        self.apktool = ApktoolDisassembler(log=log)

        self._temp_dirs = []

    def extract(self) -> Dict[str, Any]:
        """Run full APK code analysis.

        Returns dict with keys:
            decompiled_content, decompiled_refs,
            smali_content, smali_refs,
            similarity_metrics, strings, analysis_errors
        """
        results = {
            "decompiled_content": [],
            "decompiled_refs": [],
            "smali_content": [],
            "smali_refs": [],
            "similarity_metrics": [],
            "cfg": [],
            "strings": [],
            "analysis_errors": [],
        }

        # Create temp directories
        jadx_dir = tempfile.mkdtemp(prefix="redb_jadx_")
        apktool_dir = tempfile.mkdtemp(prefix="redb_apktool_")
        self._temp_dirs.extend([jadx_dir, apktool_dir])

        # Step 1-3: Run JADX, apktool, and androguard in parallel
        jadx_success = False
        apktool_success = False
        androguard_result = None

        with ThreadPoolExecutor(max_workers=3) as executor:
            futures = {}
            futures[executor.submit(self.jadx.decompile, self.filepath, jadx_dir)] = "jadx"
            futures[executor.submit(self.apktool.disassemble, self.filepath, apktool_dir)] = "apktool"
            futures[executor.submit(self._run_androguard)] = "androguard"

            for future in as_completed(futures):
                tool = futures[future]
                try:
                    result = future.result()
                    if tool == "jadx":
                        jadx_success = result
                    elif tool == "apktool":
                        apktool_success = result
                    elif tool == "androguard":
                        androguard_result = result
                except Exception as e:
                    if self.log:
                        self.log.error(f"{tool} failed: {e}")
                    results["analysis_errors"].append({
                        "class_name": None,
                        "method_name": None,
                        "error_location": tool,
                        "error_message": str(e),
                        "error_type": type(e).__name__,
                    })

        if androguard_result is None:
            if self.log:
                self.log.error("Androguard analysis failed — cannot proceed")
            return results

        apk_obj, dexs, analysis = androguard_result

        # Step 4b: Extract strings from DEX files
        try:
            results["strings"] = self._extract_strings(dexs)
        except Exception as e:
            if self.log:
                self.log.error(f"String extraction failed: {e}")
            results["analysis_errors"].append({
                "class_name": None,
                "method_name": None,
                "error_location": "strings",
                "error_message": f"String extraction failed: {e}",
                "error_type": type(e).__name__,
            })

        # Step 5: Parse smali output
        smali_methods = {}
        if apktool_success:
            try:
                smali_dirs = self.apktool.get_smali_directories(apktool_dir)
                for sdir in smali_dirs:
                    smali_methods.update(SmaliParser.parse_smali_directory(sdir))
            except Exception as e:
                if self.log:
                    self.log.error(f"Smali parsing failed: {e}")
                results["analysis_errors"].append({
                    "class_name": None,
                    "method_name": None,
                    "error_location": "apktool",
                    "error_message": f"Smali parsing failed: {e}",
                    "error_type": type(e).__name__,
                })

        # Step 6: Parse Java output
        java_methods = {}
        if jadx_success:
            try:
                java_methods = self.jadx.parse_java_methods(jadx_dir)
            except Exception as e:
                if self.log:
                    self.log.error(f"Java parsing failed: {e}")
                results["analysis_errors"].append({
                    "class_name": None,
                    "method_name": None,
                    "error_location": "jadx",
                    "error_message": f"Java parsing failed: {e}",
                    "error_type": type(e).__name__,
                })

        if not apktool_success and self.log:
            self.log.info(
                "apktool failed — falling back to androguard disassembly"
            )

        # Step 7-8: Process each method from androguard
        seen_decompiled_hashes = set()
        seen_smali_hashes = set()

        try:
            for method in analysis.get_methods():
                try:
                    self._process_method(
                        method,
                        smali_methods,
                        java_methods,
                        results,
                        seen_decompiled_hashes,
                        seen_smali_hashes,
                    )
                except Exception as e:
                    method_name = "unknown"
                    class_name = "unknown"
                    try:
                        if not method.is_external():
                            enc = method.get_method()
                            class_name = enc.get_class_name()
                            method_name = enc.get_name()
                    except Exception:
                        pass
                    results["analysis_errors"].append({
                        "class_name": class_name,
                        "method_name": method_name,
                        "error_location": "analysis",
                        "error_message": str(e),
                        "error_type": type(e).__name__,
                    })
        except Exception as e:
            if self.log:
                self.log.error(f"Method enumeration failed: {e}")
            results["analysis_errors"].append({
                "class_name": None,
                "method_name": None,
                "error_location": "androguard",
                "error_message": f"Method enumeration failed: {e}",
                "error_type": type(e).__name__,
            })

        if self.log:
            stats = self.library_filter.get_filter_stats()
            self.log.info(
                f"APK analysis complete: {stats['user']} user methods, "
                f"{stats['library']} library methods filtered, "
                f"{len(results['decompiled_content'])} decompiled, "
                f"{len(results['smali_content'])} smali, "
                f"{len(results['strings'])} strings"
            )

        return results

    def _run_androguard(self):
        """Run androguard analysis on the APK."""
        from androguard.misc import AnalyzeAPK
        return AnalyzeAPK(self.filepath)

    def _process_method(
        self,
        method,
        smali_methods: Dict,
        java_methods: Dict,
        results: Dict,
        seen_decompiled: set,
        seen_smali: set,
    ):
        """Process a single method from androguard analysis."""
        # Skip external methods (no code body)
        if method.is_external():
            return

        encoded = method.get_method()
        class_name = encoded.get_class_name()
        method_name = encoded.get_name()
        descriptor = encoded.get_descriptor()

        # Build method key for cross-tool matching
        method_key = SmaliParser.make_method_key(class_name, method_name, descriptor)

        # Check if library
        is_lib = self.library_filter.is_library(class_name)
        method_type = "LIBRARY" if is_lib else "USER"

        # Skip library methods for content tables (but they're still in xrefs)
        if is_lib:
            return

        # Get xrefs
        # In androguard 4.x, xref tuples are (ClassAnalysis, MethodAnalysis, offset).
        # The MethodAnalysis wrapper doesn't expose get_class_name()/get_name()
        # directly — we need to unwrap via .get_method() first.
        callers = []
        callees = []
        try:
            for ref_class, ref_method, offset in method.get_xref_from():
                try:
                    enc = ref_method.get_method()
                    caller_key = f"{enc.get_class_name()}->{enc.get_name()}"
                except AttributeError:
                    # Fallback for older androguard where ref_method is EncodedMethod
                    caller_key = f"{ref_method.get_class_name()}->{ref_method.get_name()}"
                callers.append(caller_key)
        except Exception:
            pass

        try:
            for ref_class, ref_method, offset in method.get_xref_to():
                try:
                    enc = ref_method.get_method()
                    callee_key = f"{enc.get_class_name()}->{enc.get_name()}"
                except AttributeError:
                    callee_key = f"{ref_method.get_class_name()}->{ref_method.get_name()}"
                callees.append(callee_key)
        except Exception:
            pass

        # Look up smali body
        smali_method = smali_methods.get(method_key)
        smali_body = None
        smali_normalized = None
        smali_hash = None
        instruction_count = 0
        register_count = 0

        if smali_method:
            smali_body = smali_method.body
            instruction_count = smali_method.instruction_count
            register_count = smali_method.register_count
        else:
            # Fallback: use androguard's own disassembler
            smali_body, instruction_count, register_count = (
                self._disassemble_with_androguard(encoded)
            )

        # Apply minimum instruction filter
        if instruction_count < self.min_instructions:
            return

        if smali_body:
            smali_normalized = SmaliParser.normalize_smali_body(smali_body)
            if smali_normalized:
                smali_hash = compute_sha256(smali_normalized)

        if not smali_hash:
            return

        # Compute obfuscation indicators from smali
        obfuscation = detect_obfuscation_indicators(
            method_name, class_name, smali_body, instruction_count
        )

        # Compute CFG metrics early — needed for both content tables and cfg table
        cfg_metrics = None
        if smali_body:
            try:
                cfg_metrics = compute_cfg_metrics(smali_body)
            except Exception:
                pass

        # Smali content (deduplicated)
        if smali_hash and smali_hash not in seen_smali:
            seen_smali.add(smali_hash)
            results["smali_content"].append({
                "smali_method_hash": smali_hash,
                "smali_method": smali_body,
                "smali_method_type": method_type,
                "smali_instructions_count": instruction_count,
                "smali_register_count": register_count,
                "smali_has_string_encryption": obfuscation.get("has_string_encryption", False),
                "smali_has_reflection_calls": obfuscation.get("has_reflection_calls", False),
                "smali_excessive_goto_count": obfuscation.get("excessive_goto_count", False),
                "smali_flattened_score": cfg_metrics.flattened_score if cfg_metrics else 0.0,
                "smali_mba_score": cfg_metrics.mba_score if cfg_metrics else 0.0,
            })

        # Smali reference (per-binary)
        ssdeep_val = compute_ssdeep(smali_normalized) if smali_normalized else None
        tlsh_val = compute_tlsh(smali_normalized) if smali_normalized else None

        # Semantically normalized ssdeep/TLSH (analogous to Binja's ssdeep_llil/tlsh_llil)
        ssdeep_normalized_val = None
        tlsh_normalized_val = None
        if smali_normalized:
            normalized_lines = normalize_method_body(smali_normalized, level="opcode_api")
            if normalized_lines:
                normalized_text = "\n".join(normalized_lines)
                ssdeep_normalized_val = compute_ssdeep(normalized_text)
                tlsh_normalized_val = compute_tlsh(normalized_text)

        # Look up Java source
        java_class = dalvik_to_java_class(class_name)
        java_key_candidates = [
            f"{java_class}.{method_name}",
        ]
        java_source = None
        decompiled_hash = None

        for jk in java_key_candidates:
            for key, source in java_methods.items():
                if key.startswith(jk):
                    java_source = source
                    break
            if java_source:
                break

        if java_source:
            # Normalize and hash decompiled content
            normalized_java = _normalize_java(java_source)
            decompiled_hash = compute_sha256(normalized_java)

            # Decompiled content (deduplicated)
            if decompiled_hash not in seen_decompiled:
                seen_decompiled.add(decompiled_hash)
                results["decompiled_content"].append({
                    "decompiled_method_hash": decompiled_hash,
                    "decompiled_method": java_source,
                    "decompiled_method_type": method_type,
                    "decompiled_has_string_encryption": obfuscation.get("has_string_encryption", False),
                    "decompiled_has_reflection_calls": obfuscation.get("has_reflection_calls", False),
                    "decompiled_excessive_goto_count": obfuscation.get("excessive_goto_count", False),
                })

            # Decompiled reference
            method_prototype = dalvik_to_java_prototype(
                method_name, descriptor, class_name
            )
            results["decompiled_refs"].append({
                "decompiled_method_hash": decompiled_hash,
                "smali_method_hash": smali_hash,
                "decompiled_class_name": dalvik_to_java_class(class_name),
                "decompiled_method_name": method_name,
                "decompiled_method_signature": descriptor,
                "decompiled_method_prototype": method_prototype,
                "functions_caller": callers,
                "functions_call": callees,
            })

        # Smali reference
        if smali_hash:
            results["smali_refs"].append({
                "smali_method_hash": smali_hash,
                "decompiled_method_hash": decompiled_hash,
                "smali_class_name": dalvik_to_java_class(class_name),
                "smali_method_name": method_name,
                "smali_method_signature": descriptor,
                "ssdeep_smali": ssdeep_val,
                "tlsh_smali": tlsh_val,
            })

        # Similarity metrics (content-based fuzzy matching only)
        if smali_hash and smali_normalized:
            minhash_sig = compute_minhash(smali_normalized)
            sim_entry = {
                "smali_method_hash": smali_hash,
                "cyclomatic_complexity": cfg_metrics.cyclomatic_complexity if cfg_metrics else None,
                "ssdeep_smali": ssdeep_val,
                "tlsh_smali": tlsh_val,
                "ssdeep_smali_normalized": ssdeep_normalized_val,
                "tlsh_smali_normalized": tlsh_normalized_val,
                "minhash": minhash_sig or [],
            }
            results["similarity_metrics"].append(sim_entry)

        # CFG entry (structural/topological features)
        if smali_hash and cfg_metrics and smali_body:
            prime_product = compute_prime_product_smali(smali_body)
            call_count = count_call_instructions(smali_body)
            cfg_entry = {
                "smali_method_hash": smali_hash,
                "cfg_topology_hash": cfg_metrics.cfg_topology_hash,
                "block_count": cfg_metrics.block_count,
                "edge_count": cfg_metrics.edge_count,
                "cfg_instructions_count": instruction_count,
                "call_count": call_count,
                "cyclomatic_complexity": cfg_metrics.cyclomatic_complexity,
                "loop_count": cfg_metrics.loop_count,
                "max_depth": cfg_metrics.max_depth,
                "max_fan_out": cfg_metrics.max_fan_out,
                "md_index_topdown": cfg_metrics.md_index_topdown,
                "md_index_bottomup": cfg_metrics.md_index_bottomup,
                "prime_product_smali": prime_product,
                "cfg_feature_tlsh": cfg_metrics.cfg_feature_tlsh,
                "wl_minhash": cfg_metrics.wl_minhash,
                "bb_features": cfg_metrics.block_features,
                "cfg_adjacency": cfg_metrics.cfg_adjacency,
            }
            results["cfg"].append(cfg_entry)

    @staticmethod
    def _string_entropy(s: str) -> float:
        """Compute Shannon entropy of a string."""
        if not s:
            return 0.0
        freq = Counter(s)
        length = len(s)
        return -sum(
            (count / length) * math.log2(count / length)
            for count in freq.values()
        )

    def _extract_strings(self, dexs) -> List[Dict[str, Any]]:
        """Extract deduplicated strings from all DEX objects.

        Uses androguard's get_strings() on each DEX, deduplicates by value,
        and computes entropy — matching the Binja StringAnalysis output format
        so IOCExtractorFromResults can consume them identically.

        Returns list of dicts with keys:
            string, string_encoding, string_offset, string_length, string_entropy
        """
        seen = set()
        strings = []
        offset_counter = 0

        for dex in dexs:
            try:
                dex_strings = dex.get_strings()
            except Exception:
                continue

            if not dex_strings:
                continue

            for s in dex_strings:
                if not s or s in seen:
                    continue
                # DEX MUTF-8 strings may contain unpaired surrogates (e.g. \ud800)
                # that are invalid UTF-8 and will fail ClickHouse insert
                s = s.encode('utf-8', errors='replace').decode('utf-8')
                if not s or s in seen:
                    continue
                seen.add(s)

                strings.append({
                    "string": s,
                    "string_encoding": "UTF8",
                    "string_offset": offset_counter,
                    "string_length": len(s),
                    "string_entropy": self._string_entropy(s),
                })
                offset_counter += 1

        return strings

    @staticmethod
    def _disassemble_with_androguard(encoded):
        """Fallback disassembly using androguard when apktool fails.

        Returns (smali_body, instruction_count, register_count).
        """
        code = encoded.get_code()
        if not code:
            return None, 0, 0

        register_count = code.get_registers_size()
        lines = []
        instruction_count = 0

        try:
            bytecode = code.get_bc()
            if not bytecode:
                return None, 0, register_count

            for instruction in bytecode.get_instructions():
                op_name = instruction.get_name()
                output = instruction.get_output()
                if output:
                    output = _normalize_androguard_operands(op_name, output)
                    lines.append(f"    {op_name} {output}")
                else:
                    lines.append(f"    {op_name}")
                instruction_count += 1
        except Exception:
            return None, 0, register_count

        if not lines:
            return None, 0, register_count

        smali_body = "\n".join(lines)
        return smali_body, instruction_count, register_count

    def cleanup(self):
        """Remove temporary directories."""
        import shutil
        for d in self._temp_dirs:
            try:
                if os.path.isdir(d):
                    shutil.rmtree(d)
            except Exception:
                pass
        self._temp_dirs.clear()


def _normalize_androguard_operands(op_name: str, output: str) -> str:
    """Normalize androguard instruction output to match apktool smali format.

    Androguard's get_output() differs from apktool in two key ways:
    1. invoke-* operands lack {braces} around registers:
       androguard: 'v0, v1, Lcom/Foo;->bar()V'
       apktool:    '{v0, v1}, Lcom/Foo;->bar()V'
    2. Field instructions use a space instead of colon between name and type:
       androguard: 'v0, v1, LA;->field Ljava/lang/String;'
       apktool:    'v0, v1, LA;->field:Ljava/lang/String;'
    """
    # invoke-* instructions: wrap register args in {braces}
    if op_name.startswith("invoke-"):
        # Find the class/method reference (starts with L or [)
        # Split on ', ' and find where the reference begins
        parts = output.split(", ")
        reg_parts = []
        ref_idx = None
        for i, part in enumerate(parts):
            stripped = part.strip()
            if stripped.startswith("L") or stripped.startswith("["):
                ref_idx = i
                break
            reg_parts.append(part)

        if ref_idx is not None and reg_parts:
            regs = ", ".join(reg_parts)
            ref = ", ".join(parts[ref_idx:])
            return "{" + regs + "}, " + ref
        elif reg_parts:
            # No reference found — just wrap all as registers
            return "{" + ", ".join(reg_parts) + "}"

    # Field instructions: fix 'field Ltype;' → 'field:Ltype;'
    if op_name.startswith(("iget", "iput", "sget", "sput")):
        # Pattern: '... ClassName;->fieldName Ltype;' or '... ClassName;->fieldName [Ltype;'
        # The space between fieldName and the type descriptor should be a colon
        arrow_idx = output.find("->")
        if arrow_idx != -1:
            after_arrow = output[arrow_idx + 2:]
            # Find the space before the type descriptor
            space_idx = after_arrow.find(" ")
            if space_idx != -1:
                remaining = after_arrow[space_idx + 1:]
                # Check that what follows is a type descriptor
                if remaining.startswith(("L", "[", "Z", "B", "S", "C",
                                         "I", "J", "F", "D")):
                    after_arrow = after_arrow[:space_idx] + ":" + remaining
                    output = output[:arrow_idx + 2] + after_arrow

    return output


def _normalize_java(source: str) -> str:
    """Normalize Java source for consistent hashing.

    Strip leading/trailing whitespace, normalize indentation.
    """
    lines = []
    for line in source.split("\n"):
        stripped = line.strip()
        if stripped:
            lines.append(stripped)
    return "\n".join(lines)